fix: BatchWriter: Ensure flushed batch has at least one record (#2489)

#### Summary

Ensure only flushing batches that have more than 0 records.

Issue originally was discovered in cloudquery/cloudquery#22750

Author: bbernays

writers/batchwriter/batchwriter.go

@@ -161,8 +161,9 @@ func (w *BatchWriter) worker(ctx context.Context, tableName string, ch <-chan *m
 				limit.AddSlice(add)
 			}
 			if len(toFlush) > 0 || rest != nil || limit.ReachedLimit() {
-				// flush current batch
-				send()
+				if limit.Rows() > 0 {
+					send()
+				}
 				ticker.Reset(w.batchTimeout)
 			}
 			for _, sliceToFlush := range toFlush {

writers/batchwriter/batchwriter_test.go

@@ -233,6 +233,57 @@ func TestBatchUpserts(t *testing.T) {
 	}
 }
 
+// noEmptyBatchClient wraps testBatchClient and fails the test if WriteTableBatch
+// is called with an empty messages slice.
+type noEmptyBatchClient struct {
+	testBatchClient
+	t *testing.T
+}
+
+func (c *noEmptyBatchClient) WriteTableBatch(ctx context.Context, name string, messages message.WriteInserts) error {
+	if len(messages) == 0 {
+		// Use t.Error (not t.Fatal) because this may be called from a worker goroutine.
+		c.t.Error("WriteTableBatch called with empty messages slice")
+		return nil
+	}
+	return c.testBatchClient.WriteTableBatch(ctx, name, messages)
+}
+
+// TestBatchNoEmptyFlush is a regression test ensuring WriteTableBatch is never called with
+// an empty messages slice. This can happen when batchSizeBytes is so small that no row fits
+// in the initial batch (SliceRecord returns add==nil while toFlush is non-empty), which
+// previously caused send() to call WriteTableBatch with an empty resources slice.
+func TestBatchNoEmptyFlush(t *testing.T) {
+	ctx := context.Background()
+
+	testClient := &noEmptyBatchClient{t: t}
+	// batchSizeBytes=1 ensures that no single row fits in the initial batch:
+	// SliceRecord returns add==nil, toFlush=[one record per row], rest=nil.
+	wr, err := New(testClient, WithBatchSizeBytes(1))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	table := schema.Table{Name: "table1", Columns: []schema.Column{{Name: "id", Type: arrow.PrimitiveTypes.Int64}}}
+	const numRows = 5
+	record := getRecord(table.ToArrowSchema(), numRows)
+	if err := wr.writeAll(ctx, []message.WriteMessage{&message.WriteInsert{Record: record}}); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := wr.Flush(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if err := wr.Close(ctx); err != nil {
+		t.Fatal(err)
+	}
+
+	// All rows must have been written via at least one non-empty batch.
+	if testClient.InsertsLen() == 0 {
+		t.Fatalf("expected at least 1 insert message, got 0")
+	}
+}
+
 func getRecord(sc *arrow.Schema, rows int) arrow.RecordBatch {
 	builder := array.NewRecordBuilder(memory.DefaultAllocator, sc)
 	defer builder.Release()