fix: Change logic for batch writing to write when batch size is reached, not exceeded

[maaarcelino]

The current logic for batch writing prevents us from cleanly testing the DeleteRecord handling in the plugin. Currently we have to set BatchSize: 1 and send two DeleteRecord messages in the test, which is not the cleanest logic. This PR changes the logic so the batch is flushed when the batch size is reached, not exceeded (so for BatchSize: 1 it will flush after one DeleteRecord is received).

This also matches the logic we have in https://github.com/cloudquery/plugin-sdk/blob/main/internal/batch/cap.go#L7.

[murarustefaan]

Can't the described behavior be fixed by running tests in destinations with lower batch sizes / timeouts, rather than configuring the same delete message to be sent more times than it's needed?

[maaarcelino]

@murarustefaan Yeah I guess that's a fair point. I updated the PR to change the logic for batch writers to use >= instead of > for comparisons against the batch size. This is to enable running the WriterTestSuiteTests test suite which sends a single DeleteRecord message to the plugin. We need to set the batch size to 1 in the test and to test that we have to allow for using >= in the logic. Does that make sense?

[murarustefaan]

Yeah I don't think this will impact anything, it will flush at 1 instead of 2 for batch_size: 1and at 5000 instead of 5001 at batch_size: 5000

[murarustefaan]

Could you change the title and description of this PR, @maaarcelino ?

[erezrokah]

Agree with the change, seems like a left over from when we refactored the code to support multi row batches.

I would first try to update a few destinations (e.g. Postgres, S3 and MySQL) to this branch sha to see if this triggers any test failures, as existing tests might rely on this behavior

[erezrokah]

Suggested change

	if w.batchSize > 0 && l >= w.batchSize {
	limit := batch.CappedAt(0, w.batchSize)
	limit.AddRows(l)
	if limit.ReachedLimit() {

I think this is more aligned with the rest of the codebase e.g.

plugin-sdk/writers/streamingbatchwriter/streamingbatchwriter.go

Line 406 in c82d006

if len(toFlush) > 0 || rest != nil || s.limit.ReachedLimit() {

same comment for other places where we check the limit

[maaarcelino]

Done in 05b4212. Tested and it works correctly