ref(defaults): Make batch size parameters optional (#344)

iambriccardo · web-flow · commit 15f1894e9536 · 2025-09-09T11:55:13.000+02:00
diff --git a/Cargo.toml b/Cargo.toml
@@ -72,3 +72,6 @@ utoipa = { version = "5.4.0", default-features = false }
 utoipa-swagger-ui = { version = "9.0.2", default-features = false }
 uuid = { version = "1.17.0", default-features = false }
 x509-cert = { version = "0.2.2", default-features = false }
+
+[profile.bench]
+debug = true
diff --git a/etl-api/src/configs/pipeline.rs b/etl-api/src/configs/pipeline.rs
@@ -1,19 +1,32 @@
-use crate::configs::store::Store;
 use etl_config::shared::{BatchConfig, PgConnectionConfig, PipelineConfig};
 use serde::{Deserialize, Serialize};
 use utoipa::ToSchema;
 
-const DEFAULT_BATCH_MAX_SIZE: usize = 1000000;
+use crate::configs::store::Store;
+
+const DEFAULT_BATCH_MAX_SIZE: usize = 100000;
 const DEFAULT_BATCH_MAX_FILL_MS: u64 = 10000;
 const DEFAULT_TABLE_ERROR_RETRY_DELAY_MS: u64 = 10000;
 const DEFAULT_MAX_TABLE_SYNC_WORKERS: u16 = 4;
 
+/// Batch processing configuration for pipelines.
+#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub struct ApiBatchConfig {
+    /// Maximum number of items in a batch for table copy and event streaming.
+    #[schema(example = 1000)]
+    pub max_size: Option<usize>,
+    /// Maximum time, in milliseconds, to wait for a batch to fill before processing.
+    #[schema(example = 1000)]
+    pub max_fill_ms: Option<u64>,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct FullApiPipelineConfig {
     #[schema(example = "my_publication")]
     pub publication_name: String,
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub batch: Option<BatchConfig>,
+    pub batch: Option<ApiBatchConfig>,
     #[schema(example = 1000)]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub table_error_retry_delay_ms: Option<u64>,
@@ -26,7 +39,10 @@ impl From<StoredPipelineConfig> for FullApiPipelineConfig {
     fn from(value: StoredPipelineConfig) -> Self {
         Self {
             publication_name: value.publication_name,
-            batch: Some(value.batch),
+            batch: Some(ApiBatchConfig {
+                max_size: Some(value.batch.max_size),
+                max_fill_ms: Some(value.batch.max_fill_ms),
+            }),
             table_error_retry_delay_ms: Some(value.table_error_retry_delay_ms),
             max_table_sync_workers: Some(value.max_table_sync_workers),
         }
@@ -40,7 +56,7 @@ pub struct PartialApiPipelineConfig {
     pub publication_name: Option<String>,
     #[schema(example = r#"{"max_size": 1000000, "max_fill_ms": 10000}"#)]
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub batch: Option<BatchConfig>,
+    pub batch: Option<ApiBatchConfig>,
     #[schema(example = 1000)]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub table_error_retry_delay_ms: Option<u64>,
@@ -78,8 +94,13 @@ impl StoredPipelineConfig {
             self.publication_name = value;
         }
 
-        if let Some(value) = partial.batch {
-            self.batch = value;
+        if let Some(value) = partial.batch
+            && let (Some(max_size), Some(max_fill_ms)) = (value.max_size, value.max_fill_ms)
+        {
+            self.batch = BatchConfig {
+                max_size,
+                max_fill_ms,
+            };
         }
 
         if let Some(value) = partial.table_error_retry_delay_ms {
@@ -96,12 +117,20 @@ impl Store for StoredPipelineConfig {}
 
 impl From<FullApiPipelineConfig> for StoredPipelineConfig {
     fn from(value: FullApiPipelineConfig) -> Self {
-        Self {
-            publication_name: value.publication_name,
-            batch: value.batch.unwrap_or(BatchConfig {
+        let batch = value
+            .batch
+            .map(|b| BatchConfig {
+                max_size: b.max_size.unwrap_or(DEFAULT_BATCH_MAX_SIZE),
+                max_fill_ms: b.max_fill_ms.unwrap_or(DEFAULT_BATCH_MAX_FILL_MS),
+            })
+            .unwrap_or(BatchConfig {
                 max_size: DEFAULT_BATCH_MAX_SIZE,
                 max_fill_ms: DEFAULT_BATCH_MAX_FILL_MS,
-            }),
+            });
+
+        Self {
+            publication_name: value.publication_name,
+            batch,
             table_error_retry_delay_ms: value
                 .table_error_retry_delay_ms
                 .unwrap_or(DEFAULT_TABLE_ERROR_RETRY_DELAY_MS),
@@ -196,9 +225,9 @@ mod tests {
 
         let partial = PartialApiPipelineConfig {
             publication_name: Some("new_publication".to_string()),
-            batch: Some(BatchConfig {
-                max_size: 2000,
-                max_fill_ms: 8000,
+            batch: Some(ApiBatchConfig {
+                max_size: Some(1000),
+                max_fill_ms: Some(8000),
             }),
             table_error_retry_delay_ms: Some(5000),
             max_table_sync_workers: None,
@@ -207,7 +236,7 @@ mod tests {
         stored.merge(partial);
 
         assert_eq!(stored.publication_name, "new_publication");
-        assert_eq!(stored.batch.max_size, 2000);
+        assert_eq!(stored.batch.max_size, 1000);
         assert_eq!(stored.batch.max_fill_ms, 8000);
         assert_eq!(stored.table_error_retry_delay_ms, 5000);
         assert_eq!(stored.max_table_sync_workers, 2);
diff --git a/etl-api/tests/pipelines.rs b/etl-api/tests/pipelines.rs
@@ -1,11 +1,12 @@
+use etl_api::configs::pipeline::ApiBatchConfig;
 use etl_api::routes::pipelines::{
     CreatePipelineRequest, CreatePipelineResponse, GetPipelineReplicationStatusResponse,
     GetPipelineVersionResponse, ReadPipelineResponse, ReadPipelinesResponse,
     RollbackTableStateRequest, RollbackTableStateResponse, RollbackType,
     SimpleTableReplicationState, UpdatePipelineConfigRequest, UpdatePipelineConfigResponse,
     UpdatePipelineRequest, UpdatePipelineVersionRequest,
 };
-use etl_config::shared::{BatchConfig, PgConnectionConfig};
+use etl_config::shared::PgConnectionConfig;
 use etl_postgres::sqlx::test_utils::drop_pg_database;
 use etl_telemetry::tracing::init_test_tracing;
 use reqwest::StatusCode;
@@ -755,10 +756,12 @@ async fn pipeline_config_can_be_updated() {
 
     // Act
     let update_request = UpdatePipelineConfigRequest {
-        config: partially_updated_optional_pipeline_config(ConfigUpdateType::Batch(BatchConfig {
-            max_size: 10_000,
-            max_fill_ms: 100,
-        })),
+        config: partially_updated_optional_pipeline_config(ConfigUpdateType::Batch(
+            ApiBatchConfig {
+                max_size: Some(10_000),
+                max_fill_ms: Some(100),
+            },
+        )),
     };
     let response = app
         .update_pipeline_config(&tenant_id, pipeline_id, &update_request)
diff --git a/etl-api/tests/snapshots/destinations_pipelines__an_existing_destination_and_pipeline_can_be_updated-2.snap b/etl-api/tests/snapshots/destinations_pipelines__an_existing_destination_and_pipeline_can_be_updated-2.snap
@@ -5,9 +5,13 @@ expression: response.config
 FullApiPipelineConfig {
     publication_name: "updated_publication",
     batch: Some(
-        BatchConfig {
-            max_size: 2000,
-            max_fill_ms: 10,
+        ApiBatchConfig {
+            max_size: Some(
+                2000,
+            ),
+            max_fill_ms: Some(
+                10,
+            ),
         },
     ),
     table_error_retry_delay_ms: Some(
diff --git a/etl-api/tests/snapshots/destinations_pipelines__destination_and_pipeline_can_be_created-2.snap b/etl-api/tests/snapshots/destinations_pipelines__destination_and_pipeline_can_be_created-2.snap
@@ -5,9 +5,13 @@ expression: response.config
 FullApiPipelineConfig {
     publication_name: "publication",
     batch: Some(
-        BatchConfig {
-            max_size: 1000,
-            max_fill_ms: 5,
+        ApiBatchConfig {
+            max_size: Some(
+                1000,
+            ),
+            max_fill_ms: Some(
+                5,
+            ),
         },
     ),
     table_error_retry_delay_ms: Some(
diff --git a/etl-api/tests/snapshots/pipelines__all_pipelines_can_be_read-2.snap b/etl-api/tests/snapshots/pipelines__all_pipelines_can_be_read-2.snap
@@ -5,9 +5,13 @@ expression: pipeline.config
 FullApiPipelineConfig {
     publication_name: "updated_publication",
     batch: Some(
-        BatchConfig {
-            max_size: 2000,
-            max_fill_ms: 10,
+        ApiBatchConfig {
+            max_size: Some(
+                2000,
+            ),
+            max_fill_ms: Some(
+                10,
+            ),
         },
     ),
     table_error_retry_delay_ms: Some(
diff --git a/etl-api/tests/snapshots/pipelines__all_pipelines_can_be_read.snap b/etl-api/tests/snapshots/pipelines__all_pipelines_can_be_read.snap
@@ -5,9 +5,13 @@ expression: pipeline.config
 FullApiPipelineConfig {
     publication_name: "publication",
     batch: Some(
-        BatchConfig {
-            max_size: 1000,
-            max_fill_ms: 5,
+        ApiBatchConfig {
+            max_size: Some(
+                1000,
+            ),
+            max_fill_ms: Some(
+                5,
+            ),
         },
     ),
     table_error_retry_delay_ms: Some(
diff --git a/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_read.snap b/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_read.snap
@@ -5,9 +5,13 @@ expression: response.config
 FullApiPipelineConfig {
     publication_name: "publication",
     batch: Some(
-        BatchConfig {
-            max_size: 1000,
-            max_fill_ms: 5,
+        ApiBatchConfig {
+            max_size: Some(
+                1000,
+            ),
+            max_fill_ms: Some(
+                5,
+            ),
         },
     ),
     table_error_retry_delay_ms: Some(
diff --git a/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_updated.snap b/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_updated.snap
@@ -5,9 +5,13 @@ expression: response.config
 FullApiPipelineConfig {
     publication_name: "updated_publication",
     batch: Some(
-        BatchConfig {
-            max_size: 2000,
-            max_fill_ms: 10,
+        ApiBatchConfig {
+            max_size: Some(
+                2000,
+            ),
+            max_fill_ms: Some(
+                10,
+            ),
         },
     ),
     table_error_retry_delay_ms: Some(
diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-2.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-2.snap
@@ -5,9 +5,13 @@ expression: response.config
 FullApiPipelineConfig {
     publication_name: "publication",
     batch: Some(
-        BatchConfig {
-            max_size: 10000,
-            max_fill_ms: 100,
+        ApiBatchConfig {
+            max_size: Some(
+                10000,
+            ),
+            max_fill_ms: Some(
+                100,
+            ),
         },
     ),
     table_error_retry_delay_ms: Some(
diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-3.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-3.snap
@@ -5,9 +5,13 @@ expression: response.config
 FullApiPipelineConfig {
     publication_name: "publication",
     batch: Some(
-        BatchConfig {
-            max_size: 10000,
-            max_fill_ms: 100,
+        ApiBatchConfig {
+            max_size: Some(
+                10000,
+            ),
+            max_fill_ms: Some(
+                100,
+            ),
         },
     ),
     table_error_retry_delay_ms: Some(
diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated.snap
@@ -5,9 +5,13 @@ expression: response.config
 FullApiPipelineConfig {
     publication_name: "publication",
     batch: Some(
-        BatchConfig {
-            max_size: 10000,
-            max_fill_ms: 100,
+        ApiBatchConfig {
+            max_size: Some(
+                10000,
+            ),
+            max_fill_ms: Some(
+                100,
+            ),
         },
     ),
     table_error_retry_delay_ms: Some(
diff --git a/etl-api/tests/support/mocks.rs b/etl-api/tests/support/mocks.rs
@@ -7,7 +7,7 @@ use etl_api::routes::destinations::{CreateDestinationRequest, CreateDestinationR
 use etl_api::routes::images::{CreateImageRequest, CreateImageResponse};
 use etl_api::routes::pipelines::{CreatePipelineRequest, CreatePipelineResponse};
 use etl_api::routes::sources::{CreateSourceRequest, CreateSourceResponse};
-use etl_config::{SerializableSecretString, shared::BatchConfig};
+use etl_config::SerializableSecretString;
 
 use crate::support::test_app::TestApp;
 
@@ -176,14 +176,15 @@ pub mod tenants {
 /// Pipeline config helpers.
 pub mod pipelines {
     use super::*;
+    use etl_api::configs::pipeline::ApiBatchConfig;
 
     /// Returns a default pipeline config.
     pub fn new_pipeline_config() -> FullApiPipelineConfig {
         FullApiPipelineConfig {
             publication_name: "publication".to_owned(),
-            batch: Some(BatchConfig {
-                max_size: 1000,
-                max_fill_ms: 5,
+            batch: Some(ApiBatchConfig {
+                max_size: Some(1000),
+                max_fill_ms: Some(5),
             }),
             table_error_retry_delay_ms: Some(10000),
             max_table_sync_workers: Some(2),
@@ -194,9 +195,9 @@ pub mod pipelines {
     pub fn updated_pipeline_config() -> FullApiPipelineConfig {
         FullApiPipelineConfig {
             publication_name: "updated_publication".to_owned(),
-            batch: Some(BatchConfig {
-                max_size: 2000,
-                max_fill_ms: 10,
+            batch: Some(ApiBatchConfig {
+                max_size: Some(2000),
+                max_fill_ms: Some(10),
             }),
             table_error_retry_delay_ms: Some(20000),
             max_table_sync_workers: Some(4),
@@ -205,7 +206,7 @@ pub mod pipelines {
 
     /// Partial config update variants used in tests.
     pub enum ConfigUpdateType {
-        Batch(BatchConfig),
+        Batch(ApiBatchConfig),
         TableErrorRetryDelayMs(u64),
         MaxTableSyncWorkers(u16),
     }
@@ -242,9 +243,9 @@ pub mod pipelines {
     pub fn updated_optional_pipeline_config() -> PartialApiPipelineConfig {
         PartialApiPipelineConfig {
             publication_name: None,
-            batch: Some(BatchConfig {
-                max_size: 1_000_000,
-                max_fill_ms: 100,
+            batch: Some(ApiBatchConfig {
+                max_size: Some(1_000_000),
+                max_fill_ms: Some(100),
             }),
             table_error_retry_delay_ms: Some(10000),
             max_table_sync_workers: Some(8),
diff --git a/etl-benchmarks/Cargo.toml b/etl-benchmarks/Cargo.toml
@@ -26,6 +26,3 @@ tracing = { workspace = true, default-features = true }
 [[bench]]
 name = "table_copies"
 harness = false
-
-[profile.bench]
-debug = true
