adjust chunk size if needed

risingwavelabs · Jan 18, 2024 · 28721f3 · 28721f3
1 parent 955e46e
commit 28721f3
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/src/stream/src/from_proto/stream_cdc_scan.rs b/src/stream/src/from_proto/stream_cdc_scan.rs
@@ -88,6 +88,12 @@ impl ExecutorBuilder for StreamCdcScanExecutorBuilder {
         let state_table =
             StateTable::from_table_catalog(node.get_state_table()?, state_store, vnodes).await;
 
+        let chunk_size = params.env.config().developer.chunk_size;
+        let backfill_chunk_size = node
+            .rate_limit
+            .map(|x| std::cmp::min(x as usize, chunk_size))
+            .unwrap_or(chunk_size);
+
         let executor = CdcBackfillExecutor::new(
             params.actor_context.clone(),
             params.info,
@@ -97,7 +103,7 @@ impl ExecutorBuilder for StreamCdcScanExecutorBuilder {
             None,
             params.executor_stats,
             state_table,
-            params.env.config().developer.chunk_size,
+            backfill_chunk_size,
         )
         .boxed();