ClickHouse: Segregate rename with and without softdelete (#3540)

ClickHouse connectors do not actually need the table schemas of source
tables. This PR introduces a RenameWithSoftDelete connector which takes
in the schema map and remove the schema map from the existing
RenameTables connector.
RenameWithSoftDelete connector members are snowflake, BQ and PG

Author: Amogh-Bharadwaj

flow/activities/flowable.go

@@ -1262,13 +1262,39 @@ func (a *FlowableActivity) RenameTables(ctx context.Context, config *protos.Rena
 	})
 	defer shutdown()
 
+	var renameOutput *protos.RenameTablesOutput
 	ctx = context.WithValue(ctx, shared.FlowNameKey, config.FlowJobName)
-	conn, err := connectors.GetByNameAs[connectors.RenameTablesConnector](ctx, nil, a.CatalogPool, config.PeerName)
+	renameWithSoftDeleteConn, err := connectors.GetByNameAs[connectors.RenameTablesWithSoftDeleteConnector](
+		ctx, nil, a.CatalogPool, config.PeerName)
 	if err != nil {
-		return nil, a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to get connector: %w", err))
+		if err == errors.ErrUnsupported {
+			// Rename without soft-delete
+			renameConn, renameErr := connectors.GetByNameAs[connectors.RenameTablesConnector](ctx, nil, a.CatalogPool, config.PeerName)
+			if renameErr != nil {
+				return nil, a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to get rename connector: %w", renameErr))
+			}
+			defer connectors.CloseConnector(ctx, renameConn)
+
+			a.Alerter.LogFlowInfo(ctx, config.FlowJobName, "Renaming tables for resync")
+			renameOutput, err = renameConn.RenameTables(ctx, config)
+			if err != nil {
+				return nil, a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to rename tables: %w", err))
+			}
+
+			err = a.updateTableSchemaMappingForResync(ctx, config.RenameTableOptions, config.FlowJobName)
+			if err != nil {
+				return nil, a.Alerter.LogFlowError(ctx, config.FlowJobName,
+					fmt.Errorf("failed to update table_schema_mapping after resync: %w", err))
+			}
+
+			a.Alerter.LogFlowInfo(ctx, config.FlowJobName, "Resync completed for all tables")
+			return renameOutput, nil
+		}
+		return nil, a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to get rename with soft-delete connector: %w", err))
 	}
-	defer connectors.CloseConnector(ctx, conn)
+	defer connectors.CloseConnector(ctx, renameWithSoftDeleteConn)
 
+	// Rename with soft-delete
 	tableNameSchemaMapping := make(map[string]*protos.TableSchema, len(config.RenameTableOptions))
 	for _, option := range config.RenameTableOptions {
 		schema, err := internal.LoadTableSchemaFromCatalog(
@@ -1282,44 +1308,56 @@ func (a *FlowableActivity) RenameTables(ctx context.Context, config *protos.Rena
 		}
 		tableNameSchemaMapping[option.CurrentName] = schema
 	}
-
-	renameOutput, err := conn.RenameTables(ctx, config, tableNameSchemaMapping)
+	a.Alerter.LogFlowInfo(ctx, config.FlowJobName, "Renaming tables for resync with soft-delete")
+	renameOutput, err = renameWithSoftDeleteConn.RenameTables(ctx, config, tableNameSchemaMapping)
 	if err != nil {
 		return nil, a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to rename tables: %w", err))
 	}
 
+	err = a.updateTableSchemaMappingForResync(ctx, config.RenameTableOptions, config.FlowJobName)
+	if err != nil {
+		return nil, a.Alerter.LogFlowError(ctx, config.FlowJobName,
+			fmt.Errorf("failed to update table_schema_mapping after resync with soft-delete: %w", err))
+	}
+
+	a.Alerter.LogFlowInfo(ctx, config.FlowJobName, "Resync with soft-delete completed for all tables")
+	return renameOutput, nil
+}
+
+func (a *FlowableActivity) updateTableSchemaMappingForResync(
+	ctx context.Context,
+	renameOptions []*protos.RenameTableOption,
+	flowJobName string,
+) error {
 	tx, err := a.CatalogPool.Begin(ctx)
 	if err != nil {
-		return nil, fmt.Errorf("failed to begin updating table_schema_mapping: %w", err)
+		return fmt.Errorf("failed to begin updating table_schema_mapping: %w", err)
 	}
-	logger := log.With(internal.LoggerFromCtx(ctx), slog.String(string(shared.FlowNameKey), config.FlowJobName))
+	logger := log.With(internal.LoggerFromCtx(ctx), slog.String(string(shared.FlowNameKey), flowJobName))
 	defer shared.RollbackTx(tx, logger)
 
-	for _, option := range config.RenameTableOptions {
+	for _, option := range renameOptions {
 		if option.NewName != option.CurrentName {
 			if _, err := tx.Exec(
 				ctx,
 				"delete from table_schema_mapping where flow_name = $1 and table_name = $2",
-				config.FlowJobName,
+				flowJobName,
 				option.NewName,
 			); err != nil {
-				return nil, fmt.Errorf("failed to update table_schema_mapping: %w", err)
+				return fmt.Errorf("failed to delete _resync entries in table_schema_mapping: %w", err)
 			}
 		}
 		if _, err := tx.Exec(
 			ctx,
 			"update table_schema_mapping set table_name = $3 where flow_name = $1 and table_name = $2",
-			config.FlowJobName,
+			flowJobName,
 			option.CurrentName,
 			option.NewName,
 		); err != nil {
-			return nil, fmt.Errorf("failed to update table_schema_mapping: %w", err)
+			return fmt.Errorf("failed to update table_schema_mapping: %w", err)
 		}
 	}
-
-	a.Alerter.LogFlowInfo(ctx, config.FlowJobName, "Resync completed for all tables")
-
-	return renameOutput, tx.Commit(ctx)
+	return tx.Commit(ctx)
 }
 
 func (a *FlowableActivity) DeleteMirrorStats(ctx context.Context, flowName string) error {

flow/connectors/clickhouse/cdc.go

@@ -226,7 +226,6 @@ func (c *ClickHouseConnector) ReplayTableSchemaDeltas(
 func (c *ClickHouseConnector) RenameTables(
 	ctx context.Context,
 	req *protos.RenameTablesInput,
-	tableNameSchemaMapping map[string]*protos.TableSchema,
 ) (*protos.RenameTablesOutput, error) {
 	onCluster := c.onCluster()
 	for _, renameRequest := range req.RenameTableOptions {

flow/connectors/core.go

@@ -290,6 +290,12 @@ type RawTableConnector interface {
 type RenameTablesConnector interface {
 	Connector
 
+	RenameTables(context.Context, *protos.RenameTablesInput) (*protos.RenameTablesOutput, error)
+}
+
+type RenameTablesWithSoftDeleteConnector interface {
+	Connector
+
 	RenameTables(context.Context, *protos.RenameTablesInput, map[string]*protos.TableSchema) (*protos.RenameTablesOutput, error)
 }
 
@@ -593,10 +599,10 @@ var (
 	_ QRepConsolidateConnector = &connsnowflake.SnowflakeConnector{}
 	_ QRepConsolidateConnector = &connclickhouse.ClickHouseConnector{}
 
-	_ RenameTablesConnector = &connsnowflake.SnowflakeConnector{}
-	_ RenameTablesConnector = &connbigquery.BigQueryConnector{}
-	_ RenameTablesConnector = &connpostgres.PostgresConnector{}
-	_ RenameTablesConnector = &connclickhouse.ClickHouseConnector{}
+	_ RenameTablesWithSoftDeleteConnector = &connsnowflake.SnowflakeConnector{}
+	_ RenameTablesWithSoftDeleteConnector = &connbigquery.BigQueryConnector{}
+	_ RenameTablesWithSoftDeleteConnector = &connpostgres.PostgresConnector{}
+	_ RenameTablesConnector               = &connclickhouse.ClickHouseConnector{}
 
 	_ RawTableConnector = &connclickhouse.ClickHouseConnector{}
 	_ RawTableConnector = &connbigquery.BigQueryConnector{}