remove table name schema migration (#2375)

migration introduced in 0.18.1

flow/activities/flowable.go

@@ -137,38 +137,6 @@ func (a *FlowableActivity) CreateRawTable(
 	return res, nil
 }
 
-func (a *FlowableActivity) MigrateTableSchema(
-	ctx context.Context,
-	flowName string,
-	schemas map[string]*protos.TableSchema,
-) error {
-	logger := activity.GetLogger(ctx)
-	tx, err := a.CatalogPool.BeginTx(ctx, pgx.TxOptions{})
-	if err != nil {
-		return err
-	}
-	defer shared.RollbackTx(tx, logger)
-
-	for k, v := range schemas {
-		processedBytes, err := proto.Marshal(v)
-		if err != nil {
-			return err
-		}
-		if _, err := tx.Exec(
-			ctx,
-			"insert into table_schema_mapping(flow_name, table_name, table_schema) values ($1, $2, $3) "+
-				"on conflict (flow_name, table_name) do update set table_schema = $3",
-			flowName,
-			k,
-			processedBytes,
-		); err != nil {
-			return err
-		}
-	}
-
-	return tx.Commit(ctx)
-}
-
 // SetupTableSchema populates table_schema_mapping
 func (a *FlowableActivity) SetupTableSchema(
 	ctx context.Context,

flow/workflows/cdc_flow.go

@@ -368,22 +368,6 @@ func CDCFlowWorkflow(
 
 	originalRunID := workflow.GetInfo(ctx).OriginalRunID
 
-	// MIGRATION TableNameSchemaMapping moved to catalog
-	if state.SyncFlowOptions.TableNameSchemaMapping != nil {
-		migrateCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
-			StartToCloseTimeout: 5 * time.Minute,
-		})
-		if err := workflow.ExecuteActivity(
-			migrateCtx,
-			flowable.MigrateTableSchema,
-			cfg.FlowJobName,
-			state.SyncFlowOptions.TableNameSchemaMapping,
-		).Get(migrateCtx, nil); err != nil {
-			return state, fmt.Errorf("failed to migrate TableNameSchemaMapping: %w", err)
-		}
-		state.SyncFlowOptions.TableNameSchemaMapping = nil
-	}
-
 	// we cannot skip SetupFlow if SnapshotFlow did not complete in cases where Resync is enabled
 	// because Resync modifies TableMappings before Setup and also before Snapshot
 	// for safety, rely on the idempotency of SetupFlow instead

protos/flow.proto

@@ -109,10 +109,10 @@ message CreateTablesFromExistingOutput {
 }
 
 message SyncFlowOptions {
+  reserved 5;
   uint32 batch_size = 1;
   uint64 idle_timeout_seconds = 3;
   map<uint32, string> src_table_id_name_mapping = 4;
-  map<string, TableSchema> table_name_schema_mapping = 5;
   repeated TableMapping table_mappings = 6;
   int32 number_of_syncs = 7;
 }