fixed review comments pt.1

Author: heavycrystal

flow/activities/flowable.go

@@ -234,9 +234,9 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
 			SrcTableIDNameMapping: input.SrcTableIdNameMapping,
 			TableNameMapping:      tblNameMapping,
 			LastOffset:            input.LastSyncState.Checkpoint,
-			MaxBatchSize:          uint32(input.SyncFlowOptions.BatchSize),
+			MaxBatchSize:          input.SyncFlowOptions.BatchSize,
 			IdleTimeout: peerdbenv.PeerDBCDCIdleTimeoutSeconds(
-				int(input.FlowConnectionConfigs.IdleTimeoutSeconds),
+				int(input.SyncFlowOptions.IdleTimeoutSeconds),
 			),
 			TableNameSchemaMapping:      input.TableNameSchemaMapping,
 			OverridePublicationName:     input.FlowConnectionConfigs.PublicationName,

flow/cmd/handler.go

@@ -131,10 +131,9 @@ func (h *FlowRequestHandler) CreateCDCFlow(
 		},
 	}
 
-	maxBatchSize := int(cfg.MaxBatchSize)
+	maxBatchSize := cfg.MaxBatchSize
 	if maxBatchSize == 0 {
 		maxBatchSize = 1_000_000
-		cfg.MaxBatchSize = uint32(maxBatchSize)
 	}
 
 	limits := &peerflow.CDCFlowLimits{

flow/workflows/cdc_flow.go

@@ -26,7 +26,7 @@ type CDCFlowLimits struct {
 	// This is typically non-zero for testing purposes.
 	TotalSyncFlows int
 	// Maximum number of rows in a sync flow batch.
-	MaxBatchSize int
+	MaxBatchSize uint32
 	// Rows synced after which we can say a test is done.
 	ExitAfterRecords int
 }
@@ -48,14 +48,15 @@ type CDCFlowWorkflowState struct {
 	// Needed to support schema changes.
 	RelationMessageMapping model.RelationMessageMapping
 	// current workflow state
-	CurrentFlowState       protos.FlowStatus
+	CurrentFlowState protos.FlowStatus
+	// moved from config here, set by SetupFlow
 	SrcTableIdNameMapping  map[uint32]string
 	TableNameSchemaMapping map[string]*protos.TableSchema
 }
 
 type SignalProps struct {
-	BatchSize   int32
-	IdleTimeout int64
+	BatchSize   uint32
+	IdleTimeout uint64
 }
 
 // returns a new empty PeerFlowState
@@ -155,13 +156,12 @@ func CDCFlowWorkflowWithConfig(
 	limits *CDCFlowLimits,
 	state *CDCFlowWorkflowState,
 ) (*CDCFlowWorkflowResult, error) {
-	if state == nil {
-		state = NewCDCFlowWorkflowState(len(cfg.TableMappings))
-	}
-
 	if cfg == nil {
 		return nil, fmt.Errorf("invalid connection configs")
 	}
+	if state == nil {
+		state = NewCDCFlowWorkflowState(len(cfg.TableMappings))
+	}
 
 	w := NewCDCFlowWorkflowExecution(ctx)
 
@@ -304,7 +304,7 @@ func CDCFlowWorkflowWithConfig(
 	}
 
 	syncFlowOptions := &protos.SyncFlowOptions{
-		BatchSize:              int32(limits.MaxBatchSize),
+		BatchSize:              limits.MaxBatchSize,
 		IdleTimeoutSeconds:     0,
 		SrcTableIdNameMapping:  state.SrcTableIdNameMapping,
 		TableNameSchemaMapping: state.TableNameSchemaMapping,
@@ -319,14 +319,12 @@ func CDCFlowWorkflowWithConfig(
 	cdcPropertiesSelector.AddReceive(cdcPropertiesSignalChannel, func(c workflow.ReceiveChannel, more bool) {
 		var cdcSignal SignalProps
 		c.Receive(ctx, &cdcSignal)
+		// only modify for options since SyncFlow uses it
 		if cdcSignal.BatchSize > 0 {
 			syncFlowOptions.BatchSize = cdcSignal.BatchSize
-			cfg.MaxBatchSize = uint32(cdcSignal.BatchSize)
-			limits.MaxBatchSize = int(cdcSignal.BatchSize)
 		}
 		if cdcSignal.IdleTimeout > 0 {
 			syncFlowOptions.IdleTimeoutSeconds = cdcSignal.IdleTimeout
-			cfg.IdleTimeoutSeconds = cdcSignal.IdleTimeout
 		}
 
 		slog.Info("CDC Signal received. Parameters on signal reception:", slog.Int("BatchSize", int(cfg.MaxBatchSize)),

nexus/analyzer/src/lib.rs

@@ -332,7 +332,7 @@ impl<'a> StatementAnalyzer for PeerDDLAnalyzer<'a> {
                             resync,
                             soft_delete_col_name,
                             synced_at_col_name,
-                            initial_copy_only,
+                            initial_snapshot_only: initial_copy_only,
                         };
 
                         if initial_copy_only && !do_initial_copy {

nexus/flow-rs/src/grpc.rs

@@ -143,7 +143,7 @@ impl FlowGrpcClient {
             });
         });
 
-        let do_initial_copy = job.do_initial_copy;
+        let do_initial_snapshot = job.do_initial_copy;
         let publication_name = job.publication_name.clone();
         let replication_slot_name = job.replication_slot_name.clone();
         let snapshot_num_rows_per_partition = job.snapshot_num_rows_per_partition;
@@ -155,7 +155,7 @@ impl FlowGrpcClient {
             destination: Some(dst),
             flow_job_name: job.name.clone(),
             table_mappings,
-            do_initial_snapshot: do_initial_copy,
+            do_initial_snapshot,
             publication_name: publication_name.unwrap_or_default(),
             snapshot_num_rows_per_partition: snapshot_num_rows_per_partition.unwrap_or(0),
             snapshot_max_parallel_workers: snapshot_max_parallel_workers.unwrap_or(0),
@@ -168,7 +168,7 @@ impl FlowGrpcClient {
             resync: job.resync,
             soft_delete_col_name: job.soft_delete_col_name.clone().unwrap_or_default(),
             synced_at_col_name: job.synced_at_col_name.clone().unwrap_or_default(),
-            initial_snapshot_only: job.initial_copy_only,
+            initial_snapshot_only: job.initial_snapshot_only,
             ..Default::default()
         };
 

nexus/pt/src/flow_model.rs

@@ -70,7 +70,7 @@ pub struct FlowJob {
     pub resync: bool,
     pub soft_delete_col_name: Option<String>,
     pub synced_at_col_name: Option<String>,
-    pub initial_copy_only: bool,
+    pub initial_snapshot_only: bool,
 }
 
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]

protos/flow.proto

@@ -45,7 +45,7 @@ message FlowConnectionConfigs {
   // currently, MaxBatchSize and IdleTimeoutSeconds are dynamic via Temporal signals
   repeated TableMapping table_mappings = 4;
   uint32 max_batch_size = 5;
-  int64 idle_timeout_seconds = 6;
+  uint64 idle_timeout_seconds = 6;
   string cdc_staging_path = 7;
   string publication_name = 8;
   string replication_slot_name = 9;
@@ -98,9 +98,9 @@ message CreateTablesFromExistingOutput {
 }
 
 message SyncFlowOptions {
-  int32 batch_size = 1;
+  uint32 batch_size = 1;
   map<uint32, RelationMessage> relation_message_mapping = 2;
-  int64 idle_timeout_seconds = 3;
+  uint64 idle_timeout_seconds = 3;
   map<uint32, string> src_table_id_name_mapping = 4;
   map<string, TableSchema> table_name_schema_mapping = 5;
 }
@@ -111,7 +111,6 @@ message NormalizeFlowOptions {
 
 message LastSyncState {
   int64 checkpoint = 1;
-  google.protobuf.Timestamp last_synced_at = 2;
 }
 
 message StartFlowInput {