Fix nil relationMessageMapping not updating original input field (#1340)

Introduced hastily in #1335, `state.SyncFlowOptions.RelationMessageMapping = state.RelationMessageMapping` needs to happen before constructing arguments for activity
PeerDB-io · Feb 20, 2024 · 37d88f3 · 37d88f3
1 parent 7228412
commit 37d88f3
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 11 deletions.
diff --git a/flow/activities/flowable.go b/flow/activities/flowable.go
@@ -252,6 +252,10 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
 
 	errGroup, errCtx := errgroup.WithContext(ctx)
 	errGroup.Go(func() error {
+		if input.RelationMessageMapping == nil {
+			input.RelationMessageMapping = make(map[uint32]*protos.RelationMessage)
+		}
+
 		return srcConn.PullRecords(errCtx, a.CatalogPool, &model.PullRecordsRequest{
 			FlowJobName:           flowName,
 			SrcTableIDNameMapping: input.SrcTableIdNameMapping,

diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go
@@ -520,11 +520,7 @@ func (p *PostgresCDCSource) processMessage(
 		p.logger.Debug(fmt.Sprintf("RelationMessage => RelationID: %d, Namespace: %s, RelationName: %s, Columns: %v",
 			msg.RelationID, msg.Namespace, msg.RelationName, msg.Columns))
 
-		if p.relationMessageMapping == nil {
-			p.relationMessageMapping = map[uint32]*protos.RelationMessage{
-				msg.RelationID: convertRelationMessageToProto(msg),
-			}
-		} else if p.relationMessageMapping[msg.RelationID] == nil {
+		if p.relationMessageMapping[msg.RelationID] == nil {
 			p.relationMessageMapping[msg.RelationID] = convertRelationMessageToProto(msg)
 		} else {
 			// RelationMessages don't contain an LSN, so we use current clientXlogPos instead.
@@ -855,11 +851,7 @@ func (p *PostgresCDCSource) processRelationMessage(
 		}
 	}
 
-	if p.relationMessageMapping == nil {
-		p.relationMessageMapping = map[uint32]*protos.RelationMessage{currRel.RelationId: currRel}
-	} else {
-		p.relationMessageMapping[currRel.RelationId] = currRel
-	}
+	p.relationMessageMapping[currRel.RelationId] = currRel
 	rec := &model.RelationRecord{
 		TableSchemaDelta: schemaDelta,
 		CheckpointID:     int64(lsn),

diff --git a/flow/workflows/cdc_flow.go b/flow/workflows/cdc_flow.go
@@ -492,6 +492,7 @@ func CDCFlowWorkflowWithConfig(
 			WaitForCancellation: true,
 		})
 
+		state.SyncFlowOptions.RelationMessageMapping = state.RelationMessageMapping
 		startFlowInput := &protos.StartFlowInput{
 			FlowConnectionConfigs:  cfg,
 			SyncFlowOptions:        state.SyncFlowOptions,
@@ -501,7 +502,6 @@ func CDCFlowWorkflowWithConfig(
 		}
 		w.logger.Info("executing sync flow", slog.String("flowName", cfg.FlowJobName))
 		fStartFlow := workflow.ExecuteActivity(startFlowCtx, flowable.StartFlow, startFlowInput)
-		state.SyncFlowOptions.RelationMessageMapping = state.RelationMessageMapping
 
 		var syncDone bool
 		var normalizeSignalError error