Fixes Postgres insert identity with OnConflict bug + fixes flaky postgres integration test (#3408)

Author: alishakawaguchi

internal/integration-tests/worker/workflow/workflow-integration_test.go

@@ -112,7 +112,7 @@ func Test_Workflow(t *testing.T) {
 				t.Parallel()
 				test_postgres_schema_reconciliation(t, ctx, postgres, neosyncApi, dbManagers, accountId, sourceConn, destConn, true)
 			})
-			t.Run("no_truncate", func(t *testing.T) {
+			t.Run("retain_data", func(t *testing.T) {
 				t.Parallel()
 				test_postgres_schema_reconciliation(t, ctx, postgres, neosyncApi, dbManagers, accountId, sourceConn, destConn, false)
 			})

internal/testutil/testdata/postgres/schema-init/alter-job-mappings.go

@@ -24,8 +24,8 @@ func GetAlteredSyncJobMappings(schema string) []*mgmtv1alpha1.JobMapping {
 			Column: "region_number",
 			Transformer: &mgmtv1alpha1.JobMappingTransformer{
 				Config: &mgmtv1alpha1.TransformerConfig{
-					Config: &mgmtv1alpha1.TransformerConfig_GenerateDefaultConfig{
-						GenerateDefaultConfig: &mgmtv1alpha1.GenerateDefault{},
+					Config: &mgmtv1alpha1.TransformerConfig_PassthroughConfig{
+						PassthroughConfig: &mgmtv1alpha1.Passthrough{},
 					},
 				},
 			},

worker/pkg/query-builder/insert-query-builder.go

@@ -134,8 +134,19 @@ type PostgresDriver struct {
 }
 
 func (d *PostgresDriver) BuildInsertQuery(rows []map[string]any) (query string, queryargs []any, err error) {
-	goquRows := toGoquRecords(rows)
+	insertQuery, args, err := d.buildInsertQuery(rows)
+	if err != nil {
+		return "", nil, fmt.Errorf("failed to build postgres insert query: %w", err)
+	}
+
+	if d.options.shouldOverrideColumnDefault {
+		insertQuery = sqlmanager_postgres.BuildPgInsertIdentityAlwaysSql(insertQuery)
+	}
+	return insertQuery, args, nil
+}
 
+func (d *PostgresDriver) buildInsertQuery(rows []map[string]any) (sql string, args []any, err error) {
+	goquRows := toGoquRecords(rows)
 	if d.options.conflictConfig.onConflictDoUpdate != nil {
 		if len(rows) == 0 {
 			return "", []any{}, errors.New("no rows to insert")
@@ -147,9 +158,6 @@ func (d *PostgresDriver) BuildInsertQuery(rows []map[string]any) (query string,
 			if err != nil {
 				return "", nil, fmt.Errorf("failed to build insert query on conflict do nothing fallback: %w", err)
 			}
-			if d.options.shouldOverrideColumnDefault {
-				insertQuery = sqlmanager_postgres.BuildPgInsertIdentityAlwaysSql(insertQuery)
-			}
 			return insertQuery, args, nil
 		}
 
@@ -168,9 +176,6 @@ func (d *PostgresDriver) BuildInsertQuery(rows []map[string]any) (query string,
 	if err != nil {
 		return "", nil, fmt.Errorf("failed to build insert query: %w", err)
 	}
-	if d.options.shouldOverrideColumnDefault {
-		insertQuery = sqlmanager_postgres.BuildPgInsertIdentityAlwaysSql(insertQuery)
-	}
 	return insertQuery, args, nil
 }