Tenanative first go at removing tableMappings & storing in DB

Author: alexstoick

flow/activities/flowable.go

@@ -1822,3 +1822,34 @@ func (a *FlowableActivity) ReportStatusMetric(ctx context.Context, status protos
 	)))
 	return nil
 }
+
+func (a *FlowableActivity) MigrateTableMappingsToCatalog(
+	ctx context.Context,
+	flowJobName string, tableMappings []*protos.TableMapping, version uint32,
+) error {
+	logger := internal.LoggerFromCtx(ctx)
+	tx, err := a.CatalogPool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to begin transaction to migrate table mappings to catalog: %w", err)
+	}
+	defer shared.RollbackTx(tx, logger)
+
+	tableMappingsBytes := [][]byte{}
+	for _, tableMapping := range tableMappings {
+		tableMappingBytes, err := proto.Marshal(tableMapping)
+		if err != nil {
+			return fmt.Errorf("failed to marshal table mapping to migrate to catalog: %w", err)
+		}
+		tableMappingsBytes = append(tableMappingsBytes, tableMappingBytes)
+	}
+
+	stmt := `INSERT INTO table_mappings (flow_name, version, table_mapping) VALUES ($1, $2, $3)
+	 ON CONFLICT (flow_name, version) DO UPDATE SET table_mapping = EXCLUDED.table_mapping`
+	_, err = tx.Exec(ctx, stmt, flowJobName, version, tableMappingsBytes)
+
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("failed to commit transaction to migrate table mappings to catalog: %w", err)
+	}
+
+	return nil
+}

flow/activities/flowable_core.go

@@ -55,6 +55,33 @@ func heartbeatRoutine(
 	)
 }
 
+func (a *FlowableActivity) getTableMappings(ctx context.Context, flowName string, version uint) ([]*protos.TableMapping, error) {
+	rows, err := a.CatalogPool.Query(ctx, "select table_mapping from table_mappings where flow_name = $1 AND version = $2", flowName, version)
+	if err != nil {
+		return nil, err
+	}
+
+	var tableMappingsBytes [][]byte
+	var tableMappings []*protos.TableMapping = []*protos.TableMapping{}
+
+	err = rows.Scan([]any{&tableMappingsBytes})
+	if err != nil {
+		return nil, fmt.Errorf("failed to deserialize table mapping schema proto: %w", err)
+	}
+
+	for _, tableMappingBytes := range tableMappingsBytes {
+
+		var tableMapping *protos.TableMapping
+		if err := proto.Unmarshal(tableMappingBytes, tableMapping); err != nil {
+			return nil, err
+		}
+
+		tableMappings = append(tableMappings, tableMapping)
+	}
+
+	return tableMappings, nil
+}
+
 func (a *FlowableActivity) getTableNameSchemaMapping(ctx context.Context, flowName string) (map[string]*protos.TableSchema, error) {
 	rows, err := a.CatalogPool.Query(ctx, "select table_name, table_schema from table_schema_mapping where flow_name = $1", flowName)
 	if err != nil {

flow/internal/flow_configuration_helpers.go

@@ -11,6 +11,22 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/shared"
 )
 
+func TableNameMapping(tableMappings []*protos.TableMapping, resync bool) map[string]string {
+	tblNameMapping := make(map[string]string, len(tableMappings))
+	if resync {
+		for _, mapping := range tableMappings {
+			if mapping.Engine != protos.TableEngine_CH_ENGINE_NULL {
+				mapping.DestinationTableIdentifier += "_resync"
+			}
+		}
+	}
+	for _, v := range tableMappings {
+		tblNameMapping[v.SourceTableIdentifier] = v.DestinationTableIdentifier
+	}
+
+	return tblNameMapping
+}
+
 func FetchConfigFromDB(ctx context.Context, catalogPool shared.CatalogPool, flowName string) (*protos.FlowConnectionConfigsCore, error) {
 	var configBytes sql.RawBytes
 	if err := catalogPool.QueryRow(ctx,

flow/workflows/cdc_flow.go

@@ -123,7 +123,7 @@ func updateFlowConfigWithLatestSettings(
 	cloneCfg := proto.CloneOf(cfg)
 	cloneCfg.MaxBatchSize = state.SyncFlowOptions.BatchSize
 	cloneCfg.IdleTimeoutSeconds = state.SyncFlowOptions.IdleTimeoutSeconds
-	cloneCfg.TableMappings = state.SyncFlowOptions.TableMappings
+	cloneCfg.TableMappingVersion = state.SyncFlowOptions.TableMappingVersion
 	if state.SnapshotNumRowsPerPartition > 0 {
 		cloneCfg.SnapshotNumRowsPerPartition = state.SnapshotNumRowsPerPartition
 	}
@@ -221,6 +221,24 @@ func processCDCFlowConfigUpdate(
 		}
 	}
 
+	// MIGRATION: Move `tableMapping` to the DB
+	if len(state.SyncFlowOptions.TableMappings) > 0 {
+		migrateCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
+			StartToCloseTimeout: 5 * time.Minute,
+		})
+		if err := workflow.ExecuteActivity(
+			migrateCtx,
+			flowable.MigrateTableMappingsToCatalog,
+			cfg.FlowJobName,
+			state.SyncFlowOptions.TableMappings,
+			1, //hardcoding version as 1 for migration purposes.
+		).Get(migrateCtx, nil); err != nil {
+			return fmt.Errorf("failed to migrate TableMappings: %w", err)
+		}
+		state.SyncFlowOptions.TableMappings = nil
+		state.SyncFlowOptions.TableMappingVersion = 1
+	}
+
 	syncStateToConfigProtoInCatalog(ctx, cfg, state)
 	return nil
 }
@@ -296,7 +314,7 @@ func processTableAdditions(
 	alterPublicationAddAdditionalTablesFuture := workflow.ExecuteActivity(
 		alterPublicationAddAdditionalTablesCtx,
 		flowable.AddTablesToPublication,
-		internal.MinimizeFlowConfiguration(cfg), flowConfigUpdate.AdditionalTables)
+		cfg, flowConfigUpdate.AdditionalTables)
 
 	var res *CDCFlowWorkflowResult
 	var addTablesFlowErr error
@@ -338,7 +356,6 @@ func processTableAdditions(
 			childAddTablesCDCFlowFuture := workflow.ExecuteChildWorkflow(
 				childAddTablesCDCFlowCtx,
 				CDCFlowWorkflow,
-				// TODO: `additonalTableCfg` this cannot be minimized in the main branch; but the limitation is minimal.
 				additionalTablesCfg,
 				nil,
 			)
@@ -413,7 +430,7 @@ func processTableRemovals(
 		rawTableCleanupFuture := workflow.ExecuteActivity(
 			removeTablesCtx,
 			flowable.RemoveTablesFromRawTable,
-			internal.MinimizeFlowConfiguration(cfg), state.FlowConfigUpdate.RemovedTables)
+			cfg, state.FlowConfigUpdate.RemovedTables)
 		removeTablesSelector.AddFuture(rawTableCleanupFuture, func(f workflow.Future) {
 			if err := f.Get(ctx, nil); err != nil {
 				logger.Error("failed to clean up raw table for removed tables", slog.Any("error", err))
@@ -425,7 +442,7 @@ func processTableRemovals(
 			removeTablesFromCatalogFuture := workflow.ExecuteActivity(
 				removeTablesCtx,
 				flowable.RemoveTablesFromCatalog,
-				internal.MinimizeFlowConfiguration(cfg), state.FlowConfigUpdate.RemovedTables)
+				cfg, state.FlowConfigUpdate.RemovedTables)
 			removeTablesSelector.AddFuture(removeTablesFromCatalogFuture, func(f workflow.Future) {
 				if err := f.Get(ctx, nil); err != nil {
 					logger.Error("failed to clean up raw table for removed tables", slog.Any("error", err))
@@ -595,7 +612,7 @@ func CDCFlowWorkflow(
 
 		logger.Info("mirror resumed", slog.Duration("after", time.Since(startTime)))
 		state.updateStatus(ctx, logger, protos.FlowStatus_STATUS_RUNNING)
-		return state, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, internal.MinimizeFlowConfiguration(cfg), state)
+		return state, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, cfg, state)
 	}
 
 	originalRunID := workflow.GetInfo(ctx).OriginalRunID
@@ -677,7 +694,7 @@ func CDCFlowWorkflow(
 		// Resync will rely rely on the `cfg.Resync` flag to rename the tables
 		// during the snapshot process. This is how we're able to also remove the need
 		// to sync the config back into the DB / not rely on the `state.TableMappings`.
-		setupFlowFuture := workflow.ExecuteChildWorkflow(setupFlowCtx, SetupFlowWorkflow, internal.MinimizeFlowConfiguration(cfg))
+		setupFlowFuture := workflow.ExecuteChildWorkflow(setupFlowCtx, SetupFlowWorkflow, cfg)
 
 		var setupFlowOutput *protos.SetupFlowOutput
 		var setupFlowError error
@@ -813,7 +830,7 @@ func CDCFlowWorkflow(
 			logger.Info("executed setup flow and snapshot flow, start running")
 			state.updateStatus(ctx, logger, protos.FlowStatus_STATUS_RUNNING)
 		}
-		return state, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, internal.MinimizeFlowConfiguration(cfg), state)
+		return state, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, cfg, state)
 	}
 
 	var finished bool
@@ -825,7 +842,7 @@ func CDCFlowWorkflow(
 		RetryPolicy:         &temporal.RetryPolicy{MaximumAttempts: 1},
 	}))
 	state.SyncFlowOptions.TableMappings = []*protos.TableMapping{}
-	syncFlowFuture := workflow.ExecuteActivity(syncCtx, flowable.SyncFlow, internal.MinimizeFlowConfiguration(cfg), state.SyncFlowOptions)
+	syncFlowFuture := workflow.ExecuteActivity(syncCtx, flowable.SyncFlow, cfg, state.SyncFlowOptions)
 
 	mainLoopSelector := workflow.NewNamedSelector(ctx, "MainLoop")
 	mainLoopSelector.AddReceive(ctx.Done(), func(_ workflow.ReceiveChannel, _ bool) {
@@ -946,7 +963,7 @@ func CDCFlowWorkflow(
 			if state.ActiveSignal == model.TerminateSignal || state.ActiveSignal == model.ResyncSignal {
 				return state, workflow.NewContinueAsNewError(ctx, DropFlowWorkflow, state.DropFlowInput)
 			}
-			return state, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, internal.MinimizeFlowConfiguration(cfg), state)
+			return state, workflow.NewContinueAsNewError(ctx, CDCFlowWorkflow, cfg, state)
 		}
 	}
 }

nexus/catalog/migrations/V49__table_mappings_table.sql

@@ -0,0 +1,6 @@
+CREATE TABLE IF NOT EXISTS table_mappings (
+    flow_name varchar(255) not null,Expand commentComment on line R2ResolvedCode has comments. Press enter to view.
+    version int32 not null default 1,
+    table_mappings []bytea not null,
+    primary key (flow_name, version)
+);

protos/flow.proto

@@ -41,6 +41,7 @@ message TableMapping {
   string sharding_key = 7;
   string policy_name = 8;
   string partition_by_expr = 9;
+  uint32 version = 10;
 }
 
 message SetupInput {
@@ -52,12 +53,12 @@ message SetupInput {
 // FlowConnectionConfigs is for external use by the API, maintaining backwards compatibility
 // When adding fields here, add them to FlowConnectionConfigsCore too
 message FlowConnectionConfigs {
-  reserved 2,3,17;
+  reserved 2,3,17,4;
   string flow_job_name = 1;
 
   // config for the CDC flow itself
   // currently, TableMappings, MaxBatchSize and IdleTimeoutSeconds are dynamic via Temporal signals
-  repeated TableMapping table_mappings = 4;
+  //repeatable TableMapping table_mappings = 4;
   uint32 max_batch_size = 5;
   uint64 idle_timeout_seconds = 6;
   string cdc_staging_path = 7;
@@ -91,6 +92,7 @@ message FlowConnectionConfigs {
 
   map<string, string> env = 24;
   uint32 version = 25;
+  uint32 table_mapping_version = 27;
 }
 
 // FlowConnectionConfigsCore is used internally in the codebase, it is safe to remove (mark reserved) fields from it
@@ -181,6 +183,7 @@ message SyncFlowOptions {
   map<uint32, string> src_table_id_name_mapping = 4;
   repeated TableMapping table_mappings = 6;
   int32 number_of_syncs = 7;
+  uint32 table_mapping_version = 8;
 }
 
 message EnsurePullabilityBatchInput {