chore: add max_versions_per_poll

Author: benjamin-awd

src/sources/delta_lake_cdf/config.rs

@@ -96,6 +96,19 @@ pub struct DeltaLakeCdfConfig {
     #[serde(default)]
     pub ending_version: Option<i64>,
 
+    /// Maximum number of versions to process per poll cycle.
+    ///
+    /// When catching up on a large backlog of versions, processing too many
+    /// versions at once can cause the query planner to hang while listing
+    /// thousands of CDF files. This option limits how many versions are
+    /// processed in each poll cycle, allowing incremental catchup.
+    ///
+    /// If not specified, all available versions are processed at once.
+    #[serde(default)]
+    #[configurable(metadata(docs::examples = 100))]
+    #[configurable(metadata(docs::examples = 1000))]
+    pub max_versions_per_poll: Option<i64>,
+
     /// Directory for storing checkpoints.
     ///
     /// The source persists its current position (version) to disk
@@ -177,6 +190,7 @@ impl Default for DeltaLakeCdfConfig {
             include_data: default_include_data(),
             change_types: Vec::new(),
             ending_version: None,
+            max_versions_per_poll: None,
             data_dir: None,
             log_namespace: None,
         }
@@ -214,6 +228,7 @@ impl GenerateConfig for DeltaLakeCdfConfig {
             include_data: default_include_data(),
             change_types: Vec::new(),
             ending_version: None,
+            max_versions_per_poll: None,
             data_dir: None,
             log_namespace: None,
         })
@@ -482,4 +497,25 @@ mod tests {
         );
         assert_eq!(ChangeType::from_cdf_string("unknown"), None);
     }
+
+    #[test]
+    fn test_config_max_versions_per_poll() {
+        let config_str = r#"
+            table_uri = "s3://bucket/table"
+            max_versions_per_poll = 100
+        "#;
+
+        let config: DeltaLakeCdfConfig = toml::from_str(config_str).expect("Config should parse");
+        assert_eq!(config.max_versions_per_poll, Some(100));
+    }
+
+    #[test]
+    fn test_config_max_versions_per_poll_default() {
+        let config_str = r#"
+            table_uri = "s3://bucket/table"
+        "#;
+
+        let config: DeltaLakeCdfConfig = toml::from_str(config_str).expect("Config should parse");
+        assert_eq!(config.max_versions_per_poll, None);
+    }
 }

src/sources/delta_lake_cdf/source.rs

@@ -104,16 +104,35 @@ pub async fn run_cdf_source(
                 }
 
                 // Determine the end version for this batch
+                // Apply max_versions_per_poll limit to prevent query planner from
+                // hanging when catching up on large backlogs
+                let max_end_from_batch_limit = config
+                    .max_versions_per_poll
+                    .map(|max| current_version.saturating_add(max).saturating_sub(1))
+                    .unwrap_or(i64::MAX);
+
                 let end_version = config
                     .ending_version
                     .map(|e| e.min(latest_version))
-                    .unwrap_or(latest_version);
+                    .unwrap_or(latest_version)
+                    .min(max_end_from_batch_limit);
 
-                debug!(
-                    message = "Processing CDF versions",
-                    start_version = current_version,
-                    end_version = end_version,
-                );
+                // Log at info level when catching up with batching, debug otherwise
+                let versions_behind = latest_version.saturating_sub(end_version);
+                if versions_behind > 0 && config.max_versions_per_poll.is_some() {
+                    info!(
+                        message = "Processing CDF versions (catching up)",
+                        start_version = current_version,
+                        end_version = end_version,
+                        versions_behind = versions_behind,
+                    );
+                } else {
+                    debug!(
+                        message = "Processing CDF versions",
+                        start_version = current_version,
+                        end_version = end_version,
+                    );
+                }
 
                 // Create streaming CDF reader for the version range
                 match create_cdf_stream(&ctx, &table, current_version, end_version).await {