Merge branch 'main' into remove-reflect-methods

Author: kodiakhq[bot]

.release-please-manifest.json

@@ -1,3 +1,3 @@
 {
-  ".": "4.80.1"
+  ".": "4.80.3"
 }

CHANGELOG.md

@@ -5,6 +5,21 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [4.80.3](https://github.com/cloudquery/plugin-sdk/compare/v4.80.2...v4.80.3) (2025-05-19)
+
+
+### Bug Fixes
+
+* Pass correct value for plugin version ([#2156](https://github.com/cloudquery/plugin-sdk/issues/2156)) ([37b4157](https://github.com/cloudquery/plugin-sdk/commit/37b41572165f96c8d0c67390bded2b20815a506d))
+
+## [4.80.2](https://github.com/cloudquery/plugin-sdk/compare/v4.80.1...v4.80.2) (2025-05-19)
+
+
+### Bug Fixes
+
+* Change logic for batch writing to write when batch size is reached, not exceeded ([#2153](https://github.com/cloudquery/plugin-sdk/issues/2153)) ([58c8a1e](https://github.com/cloudquery/plugin-sdk/commit/58c8a1e35d8d77f7cb1ae1c73e70e4a21b23e0a7))
+* Flush DeleteRecord messages when batch writer is flushed ([#2154](https://github.com/cloudquery/plugin-sdk/issues/2154)) ([791c865](https://github.com/cloudquery/plugin-sdk/commit/791c8658a0b0224f080dce8ea0cf734dbc9ce911))
+
 ## [4.80.1](https://github.com/cloudquery/plugin-sdk/compare/v4.80.0...v4.80.1) (2025-05-12)
 
 

examples/simple_plugin/go.mod

@@ -6,7 +6,7 @@ toolchain go1.24.1
 
 require (
 	github.com/apache/arrow-go/v18 v18.2.0
-	github.com/cloudquery/plugin-sdk/v4 v4.80.1
+	github.com/cloudquery/plugin-sdk/v4 v4.80.2
 	github.com/rs/zerolog v1.34.0
 )
 

serve/package.go

@@ -129,7 +129,7 @@ func (s *PluginServe) build(pluginDirectory string, target plugin.BuildTarget, d
 	if target.IncludeSymbols {
 		stripSymbols = ""
 	}
-	ldFlags := fmt.Sprintf("%[1]s -w -X %[2]s/plugin.Version=%[3]s -X %[2]s/resources/plugin.Version=%[2]s", stripSymbols, importPath, pluginVersion)
+	ldFlags := fmt.Sprintf("%[1]s -w -X %[2]s/plugin.Version=%[3]s -X %[2]s/resources/plugin.Version=%[3]s", stripSymbols, importPath, pluginVersion)
 	args := []string{"build", "-o", pluginPath}
 	args = append(args, "-buildmode=exe")
 	args = append(args, "-ldflags", ldFlags)

writers/batchwriter/batchwriter.go

@@ -107,7 +107,10 @@ func (w *BatchWriter) Flush(ctx context.Context) error {
 	if err := w.flushMigrateTables(ctx); err != nil {
 		return err
 	}
-	return w.flushDeleteStaleTables(ctx)
+	if err := w.flushDeleteStaleTables(ctx); err != nil {
+		return err
+	}
+	return w.flushDeleteRecordTables(ctx)
 }
 
 func (w *BatchWriter) Close(context.Context) error {
@@ -278,7 +281,7 @@ func (w *BatchWriter) Write(ctx context.Context, msgs <-chan message.WriteMessag
 			w.deleteStaleMessages = append(w.deleteStaleMessages, m)
 			l := int64(len(w.deleteStaleMessages))
 			w.deleteStaleLock.Unlock()
-			if w.batchSize > 0 && l > w.batchSize {
+			if w.isLimitReached(l) {
 				if err := w.flushDeleteStaleTables(ctx); err != nil {
 					return err
 				}
@@ -298,7 +301,7 @@ func (w *BatchWriter) Write(ctx context.Context, msgs <-chan message.WriteMessag
 			w.deleteRecordMessages = append(w.deleteRecordMessages, m)
 			l := int64(len(w.deleteRecordMessages))
 			w.deleteRecordLock.Unlock()
-			if w.batchSize > 0 && l > w.batchSize {
+			if w.isLimitReached(l) {
 				if err := w.flushDeleteRecordTables(ctx); err != nil {
 					return err
 				}
@@ -310,6 +313,9 @@ func (w *BatchWriter) Write(ctx context.Context, msgs <-chan message.WriteMessag
 			if err := w.flushDeleteStaleTables(ctx); err != nil {
 				return err
 			}
+			if err := w.flushDeleteRecordTables(ctx); err != nil {
+				return err
+			}
 			if err := w.startWorker(ctx, m); err != nil {
 				return err
 			}
@@ -322,7 +328,7 @@ func (w *BatchWriter) Write(ctx context.Context, msgs <-chan message.WriteMessag
 			w.migrateTableMessages = append(w.migrateTableMessages, m)
 			l := int64(len(w.migrateTableMessages))
 			w.migrateTableLock.Unlock()
-			if w.batchSize > 0 && l > w.batchSize {
+			if w.isLimitReached(l) {
 				if err := w.flushMigrateTables(ctx); err != nil {
 					return err
 				}
@@ -332,6 +338,12 @@ func (w *BatchWriter) Write(ctx context.Context, msgs <-chan message.WriteMessag
 	return nil
 }
 
+func (w *BatchWriter) isLimitReached(rowCount int64) bool {
+	limit := batch.CappedAt(0, w.batchSize)
+	limit.AddRows(rowCount)
+	return limit.ReachedLimit()
+}
+
 func (w *BatchWriter) startWorker(_ context.Context, msg *message.WriteInsert) error {
 	w.workersLock.RLock()
 	md := msg.Record.Schema().Metadata()