feat: Extend error handling to allow conflict resolution

[MetalBlueberry]

No description provided.

[arajkumar]

first pass!

[arajkumar]

IMHO, having next handler complicates the API. Do you think it would be useful? if so, why not just use array of batch error handlers.?

[MetalBlueberry]

The current design is a decorator pattern. A next handler is something is called after your function finishes and allows to chain options.

Now that you mention it, next it is provably not correct working, as the actual goal here is to fallback in case of not having a conflict error.

Having an array doesn't solve the problem. because if you do that, how the array should behave? execute all the handlers? just the last one? how do you combine outputs? .... it becomes hard to manage.

If we rename it, do you have any ideas for the name? 😅 I can't think of any good one.

[arajkumar]

There is no copy involved with net.Buffers., should we retain that property to avoid excessive garbage?

[MetalBlueberry]

There was a copy, but done in scan.go function

here exactly, I hope the link works

https://github.com/timescale/timescaledb-parallel-copy/pull/124/files#diff-50cd4d562d776a354f7d300bd42b001d9736e761941da821c9e14344d51a78e8L200

[arajkumar]

Is there a use-case for SeekCurrent and SeekEnd? I think we could have just stayed with net.Buffers and buf[0][0] might be good enough?

[MetalBlueberry]

net buffer discards data on read.

here is the implementation. on Read it consumes the bytes by resizing the slice. That is why this implementation uses the moving indexes instead. It allows to just reset it to come back to the start.

// Read from the buffers.
//
// Read implements [io.Reader] for [Buffers].
//
// Read modifies the slice v as well as v[i] for 0 <= i < len(v),
// but does not modify v[i][j] for any i, j.
func (v *Buffers) Read(p []byte) (n int, err error) {
	for len(p) > 0 && len(*v) > 0 {
		n0 := copy(p, (*v)[0])
		v.consume(int64(n0))
		p = p[n0:]
		n += n0
	}
	if len(*v) == 0 {
		err = io.EOF
	}
	return
}

func (v *Buffers) consume(n int64) {
	for len(*v) > 0 {
		ln0 := int64(len((*v)[0]))
		if ln0 > n {
			(*v)[0] = (*v)[0][n:]
			return
		}
		n -= ln0
		(*v)[0] = nil
		*v = (*v)[1:]
	}
}

[arajkumar]

I failed to understand the usefulness of adding worker id into the ctx. Also, why adding worker id into the COPY statements will be useful?

[MetalBlueberry]

1. why adding worker id to the context
   Being in the context, it is automatically propagated across interfaces. This means you can implement the BatchErrorHandler and you will still receive the worker id over the context. Then you can use it to log which worker executed the handler. The main advantage is that if you want, you can forget it exists.

2. why adding worker id to copy statements
   I hit a situation where there was a dead lock and the error statements display the entire query. having the id in there allows me to correlate the error with an specific worker. So it is easy for me to link app logs with postgres logs.

[arajkumar]

I failed to understand the usefulness of adding worker id to the copy command.

[MetalBlueberry]

why adding worker id to copy statements
I hit a situation where there was a dead lock and the error statements display the entire query. having the id in there allows me to correlate the error with an specific worker. So it is easy for me to link app logs with postgres logs.