CDCRecordStream: remove RelationMessageMapping (#1129)

RelationMessageMapping of input is mutated, thus it can be added to sync response from input at end
PeerDB-io · Jan 23, 2024 · 7042307 · 7042307
1 parent ddca548
commit 7042307
Show file tree

Hide file tree

Showing 8 changed files with 6 additions and 15 deletions.
diff --git a/flow/activities/flowable.go b/flow/activities/flowable.go
@@ -286,8 +286,8 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
 		}
 
 		return &model.SyncResponse{
-			RelationMessageMapping: <-recordBatch.RelationMessageMapping,
 			TableSchemaDeltas:      tableSchemaDeltas,
+			RelationMessageMapping: input.RelationMessageMapping,
 		}, nil
 	}
 
@@ -298,6 +298,7 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
 		TableMappings: input.FlowConnectionConfigs.TableMappings,
 		StagingPath:   input.FlowConnectionConfigs.CdcStagingPath,
 	})
+	res.RelationMessageMapping = input.RelationMessageMapping
 	if err != nil {
 		slog.Warn("failed to push records", slog.Any("error", err))
 		a.Alerter.LogFlowError(ctx, flowName, err)

diff --git a/flow/connectors/bigquery/qrep_avro_sync.go b/flow/connectors/bigquery/qrep_avro_sync.go
@@ -119,7 +119,6 @@ func (s *QRepAvroSyncMethod) SyncRecords(
 		CurrentSyncBatchID:     syncBatchID,
 		TableNameRowsMapping:   tableNameRowsMapping,
 		TableSchemaDeltas:      tableSchemaDeltas,
-		RelationMessageMapping: <-req.Records.RelationMessageMapping,
 	}, nil
 }
 

diff --git a/flow/connectors/eventhub/eventhub.go b/flow/connectors/eventhub/eventhub.go
@@ -261,7 +261,6 @@ func (c *EventHubConnector) SyncRecords(req *model.SyncRecordsRequest) (*model.S
 		NumRecordsSynced:       rowsSynced,
 		TableNameRowsMapping:   make(map[string]uint32),
 		TableSchemaDeltas:      req.Records.WaitForSchemaDeltas(req.TableMappings),
-		RelationMessageMapping: <-req.Records.RelationMessageMapping,
 	}, nil
 }
 

diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go
@@ -236,7 +236,6 @@ func (p *PostgresCDCSource) consumeStream(
 		if cdcRecordsStorage.IsEmpty() {
 			records.SignalAsEmpty()
 		}
-		records.RelationMessageMapping <- p.relationMessageMapping
 		p.logger.Info(fmt.Sprintf("[finished] PullRecords streamed %d records", cdcRecordsStorage.Len()))
 		err := cdcRecordsStorage.Close()
 		if err != nil {

diff --git a/flow/connectors/postgres/postgres.go b/flow/connectors/postgres/postgres.go
@@ -417,7 +417,6 @@ func (c *PostgresConnector) SyncRecords(req *model.SyncRecordsRequest) (*model.S
 		CurrentSyncBatchID:     syncBatchID,
 		TableNameRowsMapping:   tableNameRowsMapping,
 		TableSchemaDeltas:      tableSchemaDeltas,
-		RelationMessageMapping: <-req.Records.RelationMessageMapping,
 	}, nil
 }
 

diff --git a/flow/connectors/s3/s3.go b/flow/connectors/s3/s3.go
@@ -229,7 +229,6 @@ func (c *S3Connector) SyncRecords(req *model.SyncRecordsRequest) (*model.SyncRes
 		NumRecordsSynced:       int64(numRecords),
 		TableNameRowsMapping:   tableNameRowsMapping,
 		TableSchemaDeltas:      req.Records.WaitForSchemaDeltas(req.TableMappings),
-		RelationMessageMapping: <-req.Records.RelationMessageMapping,
 	}, nil
 }
 

diff --git a/flow/connectors/snowflake/snowflake.go b/flow/connectors/snowflake/snowflake.go
@@ -569,7 +569,6 @@ func (c *SnowflakeConnector) syncRecordsViaAvro(
 		CurrentSyncBatchID:     syncBatchID,
 		TableNameRowsMapping:   tableNameRowsMapping,
 		TableSchemaDeltas:      tableSchemaDeltas,
-		RelationMessageMapping: <-req.Records.RelationMessageMapping,
 	}, nil
 }
 

diff --git a/flow/model/model.go b/flow/model/model.go
@@ -425,8 +425,6 @@ type CDCRecordStream struct {
 	records chan Record
 	// Schema changes from the slot
 	SchemaDeltas chan *protos.TableSchemaDelta
-	// Relation message mapping
-	RelationMessageMapping chan RelationMessageMapping
 	// Indicates if the last checkpoint has been set.
 	lastCheckpointSet bool
 	// lastCheckPointID is the last ID of the commit that corresponds to this batch.
@@ -440,11 +438,10 @@ func NewCDCRecordStream() *CDCRecordStream {
 	return &CDCRecordStream{
 		records: make(chan Record, channelBuffer),
 		// TODO (kaushik): more than 1024 schema deltas can cause problems!
-		SchemaDeltas:           make(chan *protos.TableSchemaDelta, 1<<10),
-		emptySignal:            make(chan bool, 1),
-		RelationMessageMapping: make(chan RelationMessageMapping, 1),
-		lastCheckpointSet:      false,
-		lastCheckPointID:       atomic.Int64{},
+		SchemaDeltas:      make(chan *protos.TableSchemaDelta, 1<<10),
+		emptySignal:       make(chan bool, 1),
+		lastCheckpointSet: false,
+		lastCheckPointID:  atomic.Int64{},
 	}
 }
 
@@ -515,7 +512,6 @@ func (r *CDCRecordStream) Close() {
 	close(r.emptySignal)
 	close(r.records)
 	close(r.SchemaDeltas)
-	close(r.RelationMessageMapping)
 	r.lastCheckpointSet = true
 }