parses max_batch_size

PeerDB-io · Sep 21, 2023 · 3bdf024 · 3bdf024
1 parent 06a53ed
commit 3bdf024
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 0 deletions.
diff --git a/nexus/analyzer/src/lib.rs b/nexus/analyzer/src/lib.rs
@@ -249,6 +249,12 @@ impl StatementAnalyzer for PeerDDLAnalyzer {
                             _ => None,
                         };
 
+                        let max_batch_size: Option<u32> = match raw_options.remove("max_batch_size")
+                        {
+                            Some(sqlparser::ast::Value::Number(n, _)) => Some(n.parse::<u32>()?),
+                            _ => None,
+                        };
+
                         let flow_job = FlowJob {
                             name: cdc.mirror_name.to_string().to_lowercase(),
                             source_peer: cdc.source_peer.to_string().to_lowercase(),
@@ -268,6 +274,7 @@ impl StatementAnalyzer for PeerDDLAnalyzer {
                             replication_slot_name,
                             push_batch_size,
                             push_parallelism,
+                            max_batch_size,
                         };
 
                         // Error reporting

diff --git a/nexus/flow-rs/src/grpc.rs b/nexus/flow-rs/src/grpc.rs
@@ -164,6 +164,7 @@ impl FlowGrpcClient {
             replication_slot_name: replication_slot_name.unwrap_or_default(),
             push_batch_size: job.push_batch_size.unwrap_or_default(),
             push_parallelism: job.push_parallelism.unwrap_or_default(),
+            max_batch_size: job.max_batch_size.unwrap_or_default(),
             ..Default::default()
         };
 

diff --git a/nexus/pt/src/flow_model.rs b/nexus/pt/src/flow_model.rs
@@ -75,6 +75,7 @@ pub struct FlowJob {
     pub replication_slot_name: Option<String>,
     pub push_parallelism: Option<i64>,
     pub push_batch_size: Option<i64>,
+    pub max_batch_size: Option<u32>,
 }
 
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]