fix: Skip incremental tables when performing delete-stale (#13482)

hermanschaaf · web-flow · commit 2ce73877e367 · 2023-08-30T16:11:58.000Z
This fixes a bug introduced with the migration to SDK v4 / Protocol v3; when `overwrite-delete-stale` write mode is used, stale entries should not be removed from incremental tables when a state backend is active.
diff --git a/cli/cmd/sync_v3.go b/cli/cmd/sync_v3.go
@@ -34,7 +34,7 @@ type v3destination struct {
 func syncConnectionV3(ctx context.Context, source v3source, destinations []v3destination, backend *v3destination, uid string, noMigrate bool) error {
 	var mt metrics.Metrics
 	var exitReason = ExitReasonStopped
-	tables := make(map[string]bool, 0)
+	tablesForDeleteStale := make(map[string]bool, 0)
 
 	sourceSpec := source.spec
 	sourceClient := source.client
@@ -200,6 +200,8 @@ func syncConnectionV3(ctx context.Context, source v3source, destinations []v3des
 		}
 	}()
 
+	isStateBackendEnabled := sourceSpec.BackendOptions != nil && sourceSpec.BackendOptions.TableName != ""
+
 	// Read from the sync stream and write to all destinations.
 	totalResources := 0
 	for {
@@ -242,7 +244,10 @@ func syncConnectionV3(ctx context.Context, source v3source, destinations []v3des
 				return err
 			}
 			tableName := tableNameFromSchema(sc)
-			tables[tableName] = true
+
+			if !isStateBackendEnabled || !tableIsIncremental(sc) {
+				tablesForDeleteStale[tableName] = true
+			}
 			if noMigrate {
 				continue
 			}
@@ -274,7 +279,7 @@ func syncConnectionV3(ctx context.Context, source v3source, destinations []v3des
 
 	for i := range destinationsClients {
 		if destinationSpecs[i].WriteMode == specs.WriteModeOverwriteDeleteStale {
-			if err := deleteStale(writeClients[i], tables, sourceName, syncTime); err != nil {
+			if err := deleteStale(writeClients[i], tablesForDeleteStale, sourceName, syncTime); err != nil {
 				return err
 			}
 		}
@@ -313,6 +318,11 @@ func tableNameFromSchema(sc *arrow.Schema) string {
 	return tableName
 }
 
+func tableIsIncremental(sc *arrow.Schema) bool {
+	inc, _ := sc.Metadata().GetValue("cq:extension:incremental")
+	return inc == "true"
+}
+
 func deleteStale(client plugin.Plugin_WriteClient, tables map[string]bool, sourceName string, syncTime time.Time) error {
 	for tableName := range tables {
 		if err := client.Send(&plugin.Write_Request{
