Remove InitializeTableSchema (#1078)

Split from #893

flow/activities/flowable.go

@@ -197,11 +197,6 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
 	}
 	defer connectors.CloseConnector(dstConn)
 
-	slog.InfoContext(ctx, "initializing table schema...")
-	err = dstConn.InitializeTableSchema(input.FlowConnectionConfigs.TableNameSchemaMapping)
-	if err != nil {
-		return nil, fmt.Errorf("failed to initialize table schema: %w", err)
-	}
 	activity.RecordHeartbeat(ctx, "initialized table schema")
 	slog.InfoContext(ctx, "pulling records...")
 	tblNameMapping := make(map[string]model.NameAndExclude)
@@ -398,17 +393,12 @@ func (a *FlowableActivity) StartNormalize(
 	})
 	defer shutdown()
 
-	slog.InfoContext(ctx, "initializing table schema...")
-	err = dstConn.InitializeTableSchema(input.FlowConnectionConfigs.TableNameSchemaMapping)
-	if err != nil {
-		return nil, fmt.Errorf("failed to initialize table schema: %w", err)
-	}
-
 	res, err := dstConn.NormalizeRecords(&model.NormalizeRecordsRequest{
-		FlowJobName:       input.FlowConnectionConfigs.FlowJobName,
-		SoftDelete:        input.FlowConnectionConfigs.SoftDelete,
-		SoftDeleteColName: input.FlowConnectionConfigs.SoftDeleteColName,
-		SyncedAtColName:   input.FlowConnectionConfigs.SyncedAtColName,
+		FlowJobName:            input.FlowConnectionConfigs.FlowJobName,
+		SoftDelete:             input.FlowConnectionConfigs.SoftDelete,
+		SoftDeleteColName:      input.FlowConnectionConfigs.SoftDeleteColName,
+		SyncedAtColName:        input.FlowConnectionConfigs.SyncedAtColName,
+		TableNameSchemaMapping: input.FlowConnectionConfigs.TableNameSchemaMapping,
 	})
 	if err != nil {
 		a.Alerter.LogFlowError(ctx, input.FlowConnectionConfigs.FlowJobName, err)

flow/connectors/bigquery/bigquery.go

@@ -58,14 +58,13 @@ type BigQueryServiceAccount struct {
 
 // BigQueryConnector is a Connector implementation for BigQuery.
 type BigQueryConnector struct {
-	ctx                    context.Context
-	bqConfig               *protos.BigqueryConfig
-	client                 *bigquery.Client
-	storageClient          *storage.Client
-	tableNameSchemaMapping map[string]*protos.TableSchema
-	datasetID              string
-	catalogPool            *pgxpool.Pool
-	logger                 slog.Logger
+	ctx           context.Context
+	bqConfig      *protos.BigqueryConfig
+	client        *bigquery.Client
+	storageClient *storage.Client
+	datasetID     string
+	catalogPool   *pgxpool.Pool
+	logger        slog.Logger
 }
 
 // Create BigQueryServiceAccount from BigqueryConfig
@@ -211,12 +210,6 @@ func (c *BigQueryConnector) NeedsSetupMetadataTables() bool {
 	return err != nil
 }
 
-// InitializeTableSchema initializes the schema for a table, implementing the Connector interface.
-func (c *BigQueryConnector) InitializeTableSchema(req map[string]*protos.TableSchema) error {
-	c.tableNameSchemaMapping = req
-	return nil
-}
-
 func (c *BigQueryConnector) waitForTableReady(datasetTable *datasetTable) error {
 	table := c.client.Dataset(datasetTable.dataset).Table(datasetTable.table)
 	maxDuration := 5 * time.Minute
@@ -596,7 +589,7 @@ func (c *BigQueryConnector) NormalizeRecords(req *model.NormalizeRecordsRequest)
 			},
 			dstTableName:          tableName,
 			dstDatasetTable:       dstDatasetTable,
-			normalizedTableSchema: c.tableNameSchemaMapping[tableName],
+			normalizedTableSchema: req.TableNameSchemaMapping[tableName],
 			syncBatchID:           batchIDs.SyncBatchID,
 			normalizeBatchID:      batchIDs.NormalizeBatchID,
 			peerdbCols: &protos.PeerDBColumns{

flow/connectors/core.go

@@ -68,9 +68,6 @@ type CDCSyncConnector interface {
 	// GetLastSyncBatchID gets the last batch synced to the destination from the metadata table
 	GetLastSyncBatchID(jobName string) (int64, error)
 
-	// InitializeTableSchema initializes the table schema of all the destination tables for the connector.
-	InitializeTableSchema(req map[string]*protos.TableSchema) error
-
 	// CreateRawTable creates a raw table for the connector with a given name and a fixed schema.
 	CreateRawTable(req *protos.CreateRawTableInput) (*protos.CreateRawTableOutput, error)
 
@@ -89,9 +86,6 @@ type CDCSyncConnector interface {
 type CDCNormalizeConnector interface {
 	Connector
 
-	// InitializeTableSchema initializes the table schema of all the destination tables for the connector.
-	InitializeTableSchema(req map[string]*protos.TableSchema) error
-
 	// NormalizeRecords merges records pushed earlier into the destination table.
 	// This method should be idempotent, and should be able to be called multiple times with the same request.
 	NormalizeRecords(req *model.NormalizeRecordsRequest) (*model.NormalizeResponse, error)

flow/connectors/eventhub/eventhub.go

@@ -18,13 +18,12 @@ import (
 )
 
 type EventHubConnector struct {
-	ctx          context.Context
-	config       *protos.EventHubGroupConfig
-	pgMetadata   *metadataStore.PostgresMetadataStore
-	tableSchemas map[string]*protos.TableSchema
-	creds        *azidentity.DefaultAzureCredential
-	hubManager   *EventHubManager
-	logger       slog.Logger
+	ctx        context.Context
+	config     *protos.EventHubGroupConfig
+	pgMetadata *metadataStore.PostgresMetadataStore
+	creds      *azidentity.DefaultAzureCredential
+	hubManager *EventHubManager
+	logger     slog.Logger
 }
 
 // NewEventHubConnector creates a new EventHubConnector.
@@ -83,11 +82,6 @@ func (c *EventHubConnector) ConnectionActive() error {
 	return nil
 }
 
-func (c *EventHubConnector) InitializeTableSchema(req map[string]*protos.TableSchema) error {
-	c.tableSchemas = req
-	return nil
-}
-
 func (c *EventHubConnector) NeedsSetupMetadataTables() bool {
 	return c.pgMetadata.NeedsSetupMetadata()
 }

flow/connectors/postgres/postgres.go

@@ -28,7 +28,6 @@ type PostgresConnector struct {
 	pool               *SSHWrappedPostgresPool
 	replConfig         *pgxpool.Config
 	replPool           *SSHWrappedPostgresPool
-	tableSchemaMapping map[string]*protos.TableSchema
 	customTypesMapping map[uint32]string
 	metadataSchema     string
 	logger             slog.Logger
@@ -465,7 +464,7 @@ func (c *PostgresConnector) NormalizeRecords(req *model.NormalizeRecordsRequest)
 		normalizeStmtGen := &normalizeStmtGenerator{
 			rawTableName:          rawTableIdentifier,
 			dstTableName:          destinationTableName,
-			normalizedTableSchema: c.tableSchemaMapping[destinationTableName],
+			normalizedTableSchema: req.TableNameSchemaMapping[destinationTableName],
 			unchangedToastColumns: unchangedToastColsMap[destinationTableName],
 			peerdbCols: &protos.PeerDBColumns{
 				SoftDeleteColName: req.SoftDeleteColName,
@@ -701,12 +700,6 @@ func (c *PostgresConnector) SetupNormalizedTables(req *protos.SetupNormalizedTab
 	}, nil
 }
 
-// InitializeTableSchema initializes the schema for a table, implementing the Connector interface.
-func (c *PostgresConnector) InitializeTableSchema(req map[string]*protos.TableSchema) error {
-	c.tableSchemaMapping = req
-	return nil
-}
-
 // ReplayTableSchemaDelta changes a destination table to match the schema at source
 // This could involve adding or dropping multiple columns.
 func (c *PostgresConnector) ReplayTableSchemaDeltas(flowJobName string,

flow/connectors/s3/s3.go

@@ -86,11 +86,6 @@ func (c *S3Connector) CreateRawTable(req *protos.CreateRawTableInput) (*protos.C
 	return nil, nil
 }
 
-func (c *S3Connector) InitializeTableSchema(req map[string]*protos.TableSchema) error {
-	c.logger.Info("InitializeTableSchema for S3 is a no-op")
-	return nil
-}
-
 func (c *S3Connector) Close() error {
 	c.logger.Debug("Closing metadata store connection")
 	return c.pgMetadata.Close()

flow/connectors/snowflake/snowflake.go

@@ -80,11 +80,10 @@ const (
 )
 
 type SnowflakeConnector struct {
-	ctx                context.Context
-	database           *sql.DB
-	tableSchemaMapping map[string]*protos.TableSchema
-	metadataSchema     string
-	logger             slog.Logger
+	ctx            context.Context
+	database       *sql.DB
+	metadataSchema string
+	logger         slog.Logger
 }
 
 // creating this to capture array results from snowflake.
@@ -149,11 +148,10 @@ func NewSnowflakeConnector(ctx context.Context,
 
 	flowName, _ := ctx.Value(shared.FlowNameKey).(string)
 	return &SnowflakeConnector{
-		ctx:                ctx,
-		database:           database,
-		tableSchemaMapping: nil,
-		metadataSchema:     metadataSchema,
-		logger:             *slog.With(slog.String(string(shared.FlowNameKey), flowName)),
+		ctx:            ctx,
+		database:       database,
+		metadataSchema: metadataSchema,
+		logger:         *slog.With(slog.String(string(shared.FlowNameKey), flowName)),
 	}, nil
 }
 
@@ -448,11 +446,6 @@ func (c *SnowflakeConnector) SetupNormalizedTables(
 	}, nil
 }
 
-func (c *SnowflakeConnector) InitializeTableSchema(req map[string]*protos.TableSchema) error {
-	c.tableSchemaMapping = req
-	return nil
-}
-
 // ReplayTableSchemaDeltas changes a destination table to match the schema at source
 // This could involve adding or dropping multiple columns.
 func (c *SnowflakeConnector) ReplayTableSchemaDeltas(flowJobName string,
@@ -641,7 +634,7 @@ func (c *SnowflakeConnector) NormalizeRecords(req *model.NormalizeRecordsRequest
 				dstTableName:          tableName,
 				syncBatchID:           batchIDs.SyncBatchID,
 				normalizeBatchID:      batchIDs.NormalizeBatchID,
-				normalizedTableSchema: c.tableSchemaMapping[tableName],
+				normalizedTableSchema: req.TableNameSchemaMapping[tableName],
 				unchangedToastColumns: tableNametoUnchangedToastCols[tableName],
 				peerdbCols: &protos.PeerDBColumns{
 					SoftDelete:        req.SoftDelete,

flow/model/model.go

@@ -505,10 +505,11 @@ type SyncRecordsRequest struct {
 }
 
 type NormalizeRecordsRequest struct {
-	FlowJobName       string
-	SoftDelete        bool
-	SoftDeleteColName string
-	SyncedAtColName   string
+	FlowJobName            string
+	SoftDelete             bool
+	SoftDeleteColName      string
+	SyncedAtColName        string
+	TableNameSchemaMapping map[string]*protos.TableSchema
 }
 
 type SyncResponse struct {