Validate mirror: no need to get source schemas for resync (#3325)

We get source schemas for validating destination tables' schemas, which
in the case of resync we do not do. Hence no need to fetch schemas from
source in the case of resync

Author: Amogh-Bharadwaj

flow/cmd/validate_mirror.go

@@ -12,6 +12,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/connectors"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
+	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 )
 
@@ -20,6 +21,7 @@ var CustomColumnTypeRegex = regexp.MustCompile(`^$|^[a-zA-Z][a-zA-Z0-9(),]*$`)
 func (h *FlowRequestHandler) ValidateCDCMirror(
 	ctx context.Context, req *protos.CreateCDCFlowRequest,
 ) (*protos.ValidateCDCMirrorResponse, error) {
+	ctx = context.WithValue(ctx, shared.FlowNameKey, req.ConnectionConfigs.FlowJobName)
 	underMaintenance, err := internal.PeerDBMaintenanceModeEnabled(ctx, nil)
 	if err != nil {
 		slog.Error("unable to check maintenance mode", slog.Any("error", err))
@@ -82,13 +84,16 @@ func (h *FlowRequestHandler) ValidateCDCMirror(
 	}
 	defer connectors.CloseConnector(ctx, dstConn)
 
-	res, err := srcConn.GetTableSchema(ctx, req.ConnectionConfigs.Env, req.ConnectionConfigs.Version,
-		req.ConnectionConfigs.System, req.ConnectionConfigs.TableMappings)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get source table schema: %w", err)
+	var tableSchemaMap map[string]*protos.TableSchema
+	if !req.ConnectionConfigs.Resync {
+		var getTableSchemaError error
+		tableSchemaMap, getTableSchemaError = srcConn.GetTableSchema(ctx, req.ConnectionConfigs.Env, req.ConnectionConfigs.Version,
+			req.ConnectionConfigs.System, req.ConnectionConfigs.TableMappings)
+		if getTableSchemaError != nil {
+			return nil, fmt.Errorf("failed to get source table schema: %w", getTableSchemaError)
+		}
 	}
-
-	if err := dstConn.ValidateMirrorDestination(ctx, req.ConnectionConfigs, res); err != nil {
+	if err := dstConn.ValidateMirrorDestination(ctx, req.ConnectionConfigs, tableSchemaMap); err != nil {
 		return nil, err
 	}
 

flow/connectors/clickhouse/validate.go

@@ -24,10 +24,10 @@ func (c *ClickHouseConnector) ValidateMirrorDestination(
 		}
 	}
 
-	peerDBColumns := []string{signColName, versionColName}
-	if cfg.SyncedAtColName != "" {
-		peerDBColumns = append(peerDBColumns, strings.ToLower(cfg.SyncedAtColName))
+	if cfg.Resync {
+		return nil // no need to validate schema for resync, as we will create or replace the tables
 	}
+
 	// this is for handling column exclusion, processed schema does that in a step
 	processedMapping := internal.BuildProcessedSchemaMapping(cfg.TableMappings, tableNameSchemaMapping, c.logger)
 	dstTableNames := slices.Collect(maps.Keys(processedMapping))
@@ -46,7 +46,7 @@ func (c *ClickHouseConnector) ValidateMirrorDestination(
 		return err
 	}
 
-	if !(cfg.Resync || sourceSchemaAsDestinationColumn) {
+	if !sourceSchemaAsDestinationColumn {
 		if err := chvalidate.CheckIfTablesEmptyAndEngine(ctx, c.logger, c.database,
 			dstTableNames, cfg.DoInitialSnapshot, internal.PeerDBOnlyClickHouseAllowed(), initialLoadAllowNonEmptyTables,
 		); err != nil {
@@ -59,6 +59,11 @@ func (c *ClickHouseConnector) ValidateMirrorDestination(
 		return err
 	}
 
+	peerDBColumns := []string{signColName, versionColName}
+	if cfg.SyncedAtColName != "" {
+		peerDBColumns = append(peerDBColumns, strings.ToLower(cfg.SyncedAtColName))
+	}
+
 	for _, tableMapping := range cfg.TableMappings {
 		dstTableName := tableMapping.DestinationTableIdentifier
 		if _, ok := processedMapping[dstTableName]; !ok {
@@ -70,14 +75,12 @@ func (c *ClickHouseConnector) ValidateMirrorDestination(
 			continue
 		}
 
-		if !cfg.Resync {
-			// for resync, we don't need to check the content or structure of the original tables;
-			// they'll anyways get swapped out with the _resync tables which we CREATE OR REPLACE
-			if err := c.processTableComparison(dstTableName, processedMapping[dstTableName],
-				chTableColumnsMapping[dstTableName], peerDBColumns, tableMapping,
-			); err != nil {
-				return err
-			}
+		// for resync, we don't need to check the content or structure of the original tables;
+		// they'll anyways get swapped out with the _resync tables which we CREATE OR REPLACE
+		if err := c.processTableComparison(dstTableName, processedMapping[dstTableName],
+			chTableColumnsMapping[dstTableName], peerDBColumns, tableMapping,
+		); err != nil {
+			return err
 		}
 	}
 	return nil