Shadow deploy applySchemaDeltasV2 with validations (#3770)

This pull request introduces a more robust approach for applying schema
deltas to the catalog. It's a follow-up to
#3768 (and addresses the race
condition described in the PR description), with e2e test coverage, and
with the new approach shadow-applied for validation before it's fully
enabled, given this PR introduces some non-trivial changes db state that
is hard to debug/rollback.

Changes:
- Moved previous `applySchemaDeltas` logic to `applySchemaDeltasV1`, and
introduce a separate `applySchemaDeltasV2`. Note that
`applySchemaDeltasV2` applies change in-memory, and works in conjunction
with a `ReadModifyWriteTableSchemasToCatalog` method to support
transaction with ReadModifyWrite patterns.
- Added a feature flag (`PEERDB_APPLY_SCHEMA_DELTA_TO_CATALOG`) and
logic to choose between the legacy (v1) and new (v2) approaches for
applying schema deltas to the catalog. For now the FF is **disabled** by
default, which means the old approach continue to be used as source of
truth.
- The new v2 method is shadow-applied. This is done by adding temporary
validation utilities to compare the results of v1 and v2 schema delta
approaches, logging discrepancies and reporting metrics for monitoring.
- Improved end-to-end tests to cover a race condition where columns
added without subsequent DML operations could be lost in the destination
schema, verifying catalog correctness after each sync.

Author: jgao54

flow/activities/flowable_core.go

@@ -67,6 +67,64 @@ func (a *FlowableActivity) applySchemaDeltas(
 	config *protos.FlowConnectionConfigsCore,
 	options *protos.SyncFlowOptions,
 	schemaDeltas []*protos.TableSchemaDelta,
+) error {
+	logger := internal.LoggerFromCtx(ctx)
+
+	dstTableNamesInDeltas := make([]string, 0, len(schemaDeltas))
+	for _, schemaDelta := range schemaDeltas {
+		dstTableNamesInDeltas = append(dstTableNamesInDeltas, schemaDelta.DstTableName)
+	}
+
+	applyV2, err := internal.PeerDBApplySchemaDeltaToCatalogEnabled(ctx, config.Env)
+	if err != nil {
+		logger.Warn("failed to check if schema delta v2 is enabled, defaulting to false", slog.Any("error", err))
+		applyV2 = false
+	}
+
+	if applyV2 {
+		err := internal.ReadModifyWriteTableSchemasToCatalog(
+			ctx,
+			a.CatalogPool,
+			logger,
+			config.FlowJobName,
+			dstTableNamesInDeltas,
+			// use a closure to keep ReadModifyWriteTableSchemasToCatalog's `modifyFn` flexible
+			func(schemas map[string]*protos.TableSchema) (map[string]*protos.TableSchema, error) {
+				return applySchemaDeltaV2(ctx, schemas, schemaDeltas)
+			},
+		)
+		if err != nil {
+			return fmt.Errorf("failed to update table schemas in catalog: %w", err)
+		}
+		return nil
+	} else {
+		skipValidate := false
+		baseSchema, err := internal.LoadTableSchemasFromCatalog(ctx, a.CatalogPool, config.FlowJobName, dstTableNamesInDeltas)
+		if err != nil {
+			logger.Warn("skipping v2 validation: cannot load base schemas", slog.Any("error", err))
+			skipValidate = true
+		}
+
+		if err := a.applySchemaDeltasV1(ctx, config, options, schemaDeltas); err != nil {
+			return err
+		}
+
+		if !skipValidate {
+			validateV2AgainstV1(ctx, a.CatalogPool, config.FlowJobName, baseSchema, schemaDeltas, dstTableNamesInDeltas)
+		}
+	}
+	return nil
+}
+
+// existing approach to applying schemaDeltas. `applySchemaDeltas` is actually
+// a bit misleading, as we are fetching the latest schema from the source database.
+// schemaDeltas is only used to identify matching tables.
+// This approach has a race condition where schema deltas do not get correctly
+// applied because the latest schema from source db includes the added column.
+func (a *FlowableActivity) applySchemaDeltasV1(ctx context.Context,
+	config *protos.FlowConnectionConfigsCore,
+	options *protos.SyncFlowOptions,
+	schemaDeltas []*protos.TableSchemaDelta,
 ) error {
 	filteredTableMappings := make([]*protos.TableMapping, 0, len(schemaDeltas))
 	for _, tableMapping := range options.TableMappings {
@@ -93,6 +151,49 @@ func (a *FlowableActivity) applySchemaDeltas(
 	return nil
 }
 
+// this is the updated approach of applying schema deltas to catalog. Unlike v1,
+// we use `table_schema_mapping` from catalog as base, and add new columns from
+// schemaDeltas that are not in the existing mapping. This function returns
+// a copy of schemasInCatalog with schemaDeltas applied.
+func applySchemaDeltaV2(
+	ctx context.Context,
+	schemasInCatalog map[string]*protos.TableSchema,
+	schemaDeltas []*protos.TableSchemaDelta,
+) (map[string]*protos.TableSchema, error) {
+	logger := internal.LoggerFromCtx(ctx)
+
+	// deep copy to avoid mutating input
+	schemasInCatalogCopy := make(map[string]*protos.TableSchema, len(schemasInCatalog))
+	for tableName, schema := range schemasInCatalog {
+		if schema == nil {
+			return nil, fmt.Errorf("failed to deep copy table schema from catalog: table %s has nil schema", tableName)
+		}
+		schemasInCatalogCopy[tableName] = proto.CloneOf(schema)
+	}
+
+	for _, schemaDelta := range schemaDeltas {
+		if schema, exists := schemasInCatalogCopy[schemaDelta.DstTableName]; exists {
+			columnNames := make(map[string]struct{}, len(schema.GetColumns()))
+			for _, col := range schema.GetColumns() {
+				columnNames[col.Name] = struct{}{}
+			}
+			for _, newCol := range schemaDelta.GetAddedColumns() {
+				// only add columns that don't already exist
+				if _, exists := columnNames[newCol.Name]; !exists {
+					schema.Columns = append(schema.Columns, newCol)
+					columnNames[newCol.Name] = struct{}{}
+				} else {
+					logger.Warn(fmt.Sprintf("skip adding duplicated column '%s' (type '%s') in table %s",
+						newCol.Name, newCol.Type, schemaDelta.DstTableName))
+				}
+			}
+		} else {
+			logger.Warn(fmt.Sprintf("skip adding columns for table '%s' because it's not in catalog", schemaDelta.DstTableName))
+		}
+	}
+	return schemasInCatalogCopy, nil
+}
+
 func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncConnectorCore, Items model.Items](
 	ctx context.Context,
 	a *FlowableActivity,

flow/activities/schema_delta_validation_utils.go

@@ -0,0 +1,187 @@
+// TODO: This file contains temporary validation code used during the migration from v1 to v2 schema delta approaches.
+// Once the v2 approach is fully rolled out and v1 is removed, this entire file should be deleted.
+// The validation ensures that v2 produces identical results to v1 during the transition period.
+//
+// Related cleanup tasks when deleting this file:
+// - Remove applySchemaDeltasV1() function once v1 is deprecated
+// - Clean up any references to this validation logic
+// - Clean up `PEERDB_APPLY_SCHEMA_DELTA_TO_CATALOG` in dynconf.go
+package activities
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"slices"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+
+	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/internal"
+	"github.com/PeerDB-io/peerdb/flow/otel_metrics"
+	"github.com/PeerDB-io/peerdb/flow/shared"
+)
+
+// validateV2AgainstV1 compares v1 and v2 schema delta approaches for validation.
+// Failures in this validation are logged but don't affect production behavior.
+func validateV2AgainstV1(
+	ctx context.Context,
+	pool shared.CatalogPool,
+	flowName string,
+	baseSchema map[string]*protos.TableSchema,
+	schemaDeltas []*protos.TableSchemaDelta,
+	affectedTables []string,
+) {
+	logger := internal.LoggerFromCtx(ctx)
+
+	schemaAfterV1, err := internal.LoadTableSchemasFromCatalog(ctx, pool, flowName, affectedTables)
+	if err != nil {
+		logger.Warn("skipping v1/v2 validation: cannot load post-v1 schemas", slog.Any("error", err))
+		return
+	}
+
+	schemaAfterV2, err := applySchemaDeltaV2(ctx, baseSchema, schemaDeltas)
+	if err != nil {
+		logger.Warn("skipping v1/v2 validation: applySchemaDeltaV2 failed", slog.Any("error", err))
+		return
+	}
+
+	reportUnexpectedSchemaDiffs(ctx, flowName, baseSchema, schemaAfterV1, schemaAfterV2, schemaDeltas)
+}
+
+func reportUnexpectedSchemaDiffs(
+	ctx context.Context,
+	flowName string,
+	baseSchemas map[string]*protos.TableSchema,
+	schemasAfterV1 map[string]*protos.TableSchema,
+	schemasAfterV2 map[string]*protos.TableSchema,
+	schemaDeltas []*protos.TableSchemaDelta,
+) {
+	logger := internal.LoggerFromCtx(ctx)
+
+	for _, schemaDelta := range schemaDeltas {
+		tableName := schemaDelta.DstTableName
+		baseSchema := baseSchemas[tableName]
+		schemaV1 := schemasAfterV1[tableName]
+		schemaV2 := schemasAfterV2[tableName]
+
+		if baseSchema == nil || schemaV1 == nil || schemaV2 == nil {
+			logger.Warn("skipping validation for table due to missing schema",
+				slog.String("table", tableName),
+				slog.Bool("hasBase", baseSchema != nil),
+				slog.Bool("hasV1", schemaV1 != nil),
+				slog.Bool("hasV2", schemaV2 != nil))
+			continue
+		}
+
+		var issues []string
+
+		v1ColumnMap := make(map[string]*protos.FieldDescription)
+		for _, col := range schemaV1.Columns {
+			v1ColumnMap[col.Name] = col
+		}
+
+		v2ColumnMap := make(map[string]*protos.FieldDescription)
+		for _, col := range schemaV2.Columns {
+			v2ColumnMap[col.Name] = col
+		}
+
+		// Validate existing columns match in v1 and v2
+		for _, baseCol := range baseSchema.Columns {
+			v1Col, existsInV1 := v1ColumnMap[baseCol.Name]
+			v2Col, existsInV2 := v2ColumnMap[baseCol.Name]
+
+			// Column missing in v1 is expected:
+			// - in applySchemaDeltaV1, we were syncing catalog with schema from source.
+			//   DROP COLUMN DDLs are ignored and do not trigger a schema fetch,
+			//   but subsequent ADD COLUMN DDLs would fetch the latest schema from source
+			//   with dropped columns removed.
+			// - in applySchemaDeltaV2, we never fetch schema from source, therefore it's
+			//   expected that deleted columns are in v2, but not in v1.
+			// Only report error if column exists in v1 but not in v2.
+			if existsInV1 && !existsInV2 {
+				issues = append(issues, fmt.Sprintf("existing column '%s' missing in v2", baseCol.Name))
+			}
+
+			if existsInV1 && existsInV2 && !fieldDescriptionEqual(v1Col, v2Col) {
+				issues = append(issues, fmt.Sprintf("existing column '%s' differs: v1=%+v, v2=%+v",
+					baseCol.Name, v1Col, v2Col))
+			}
+		}
+
+		// Validate new columns match in v1 and v2
+		for _, newCol := range schemaDelta.AddedColumns {
+			v1Col, existsInV1 := v1ColumnMap[newCol.Name]
+			v2Col, existsInV2 := v2ColumnMap[newCol.Name]
+
+			if !existsInV1 {
+				issues = append(issues, fmt.Sprintf("new column '%s' missing in v1", newCol.Name))
+			}
+			if !existsInV2 {
+				issues = append(issues, fmt.Sprintf("new column '%s' missing in v2", newCol.Name))
+			}
+
+			if existsInV1 && existsInV2 && !fieldDescriptionEqual(v1Col, v2Col) {
+				issues = append(issues, fmt.Sprintf("new column '%s' differs: v1=%+v, v2=%+v",
+					newCol.Name, v1Col, v2Col))
+			}
+		}
+
+		// Note: v1 may have extra columns due to race condition (fetching latest schema from source)
+		// This is expected and not considered an error.
+
+		// Check all other TableSchema fields match
+		if schemaV1.TableIdentifier != schemaV2.TableIdentifier {
+			issues = append(issues, fmt.Sprintf("table_identifier differs: v1='%s', v2='%s'",
+				schemaV1.TableIdentifier, schemaV2.TableIdentifier))
+		}
+		if !slices.Equal(schemaV1.PrimaryKeyColumns, schemaV2.PrimaryKeyColumns) {
+			issues = append(issues, fmt.Sprintf("primary_key_columns differs: v1=%v, v2=%v",
+				schemaV1.PrimaryKeyColumns, schemaV2.PrimaryKeyColumns))
+		}
+		if schemaV1.IsReplicaIdentityFull != schemaV2.IsReplicaIdentityFull {
+			issues = append(issues, fmt.Sprintf("is_replica_identity_full differs: v1=%v, v2=%v",
+				schemaV1.IsReplicaIdentityFull, schemaV2.IsReplicaIdentityFull))
+		}
+		if schemaV1.System != schemaV2.System {
+			issues = append(issues, fmt.Sprintf("system differs: v1=%v, v2=%v",
+				schemaV1.System, schemaV2.System))
+		}
+		if schemaV1.NullableEnabled != schemaV2.NullableEnabled {
+			issues = append(issues, fmt.Sprintf("nullable_enabled differs: v1=%v, v2=%v",
+				schemaV1.NullableEnabled, schemaV2.NullableEnabled))
+		}
+		if schemaV1.TableOid != schemaV2.TableOid {
+			issues = append(issues, fmt.Sprintf("table_oid differs: v1=%v, v2=%v",
+				schemaV1.TableOid, schemaV2.TableOid))
+		}
+
+		// summarize the findings in logs
+		if len(issues) > 0 {
+			logger.Warn("schema validation issues found: v1 and v2 differ",
+				slog.String("flowName", flowName),
+				slog.String("table", tableName),
+				slog.Int("issueCount", len(issues)),
+				slog.Any("issues", issues))
+
+			// if there is a discrepancy, report this as a metric to code_notification
+			otel_metrics.CodeNotificationCounter.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
+				attribute.String("message", fmt.Sprintf("Schema delta v2 validation failed for flow=%s table=%s: %d issues",
+					flowName, tableName, len(issues))),
+				attribute.String("flowName", flowName),
+				attribute.String("tableName", tableName),
+			)))
+		}
+	}
+}
+
+func fieldDescriptionEqual(a, b *protos.FieldDescription) bool {
+	if a == nil || b == nil {
+		return a == b
+	}
+	return a.Name == b.Name &&
+		a.Type == b.Type &&
+		a.TypeModifier == b.TypeModifier &&
+		a.Nullable == b.Nullable
+}