[adapters] Fix delta connector stuck in the barrier state.

Leonid Ryzhyk · ryzhyk · commit 2dbdf68e22c9 · 2025-07-10T05:07:15.000Z
Make sure that the delta connector doesn't get stuck in the barrier
state indefinitely even if the transaction log of the table is empty.

Signed-off-by: Leonid Ryzhyk &lt;leonid@feldera.com&gt;
diff --git a/crates/adapters/src/integrated/delta_table/input.rs b/crates/adapters/src/integrated/delta_table/input.rs
@@ -845,6 +845,13 @@ impl DeltaTableInputEndpointInner {
             )
         })?;
 
+        // If we are about to follow the table, set resume state to the current table version, otherwise
+        // the connector will remain in the barrier state until at least one transaction is added to the log.
+        if !self.config.snapshot() {
+            *self.last_resume_status.lock().unwrap() =
+                Some(DeltaResumeInfo::new(delta_table.version()));
+        }
+
         // Register object store with datafusion, so it will recognize individual parquet
         // file URIs when processing transaction log.  The `object_store_url` function
         // generates a unique URL, which only makes sense to datafusion.  We must append
@@ -1002,14 +1009,14 @@ impl DeltaTableInputEndpointInner {
             .table("snapshot")
             .await
             .map_err(|e| {
-                ControllerError::invalid_transport_configuration(&self.endpoint_name, &format!("internal error when compiling the 'cdc_delete_filter' expession '{delete_filter}'; {REPORT_ERROR}: table 'snapshot' not found"))
+                ControllerError::invalid_transport_configuration(&self.endpoint_name, &format!("internal error when compiling the 'cdc_delete_filter' expression '{delete_filter}'; {REPORT_ERROR}: table 'snapshot' not found"))
             })?
             .schema()
             .clone();
 
         let physical_expr = DefaultPhysicalPlanner::default()
             .create_physical_expr(&filter_expr, &schema, &self.datafusion.state())
-            .map_err(|e| ControllerError::invalid_transport_configuration(&self.endpoint_name, &format!("internal error when compiling the 'cdc_delete_filter' expession '{delete_filter}'; {REPORT_ERROR}: error generating physical plan: {e}")))?;
+            .map_err(|e| ControllerError::invalid_transport_configuration(&self.endpoint_name, &format!("internal error when compiling the 'cdc_delete_filter' expression '{delete_filter}'; {REPORT_ERROR}: error generating physical plan: {e}")))?;
 
         Ok(Some(physical_expr))
     }
@@ -1035,7 +1042,7 @@ impl DeltaTableInputEndpointInner {
             .downcast_ref::<BooleanArray>()
             .ok_or_else(|| {
                 anyhow!(
-                    "internal error converting the result of the delete filter exptession to BooleanArray; {REPORT_ERROR}: expected Boolean, found {:?}", array.data_type()
+                    "internal error converting the result of the delete filter expression to BooleanArray; {REPORT_ERROR}: expected Boolean, found {:?}", array.data_type()
                 )
             })?;
 
diff --git a/crates/adapters/src/integrated/delta_table/test.rs b/crates/adapters/src/integrated/delta_table/test.rs
@@ -1048,11 +1048,12 @@ async fn test_cdc(
 
     let mut total_count = 0;
 
-    // Write data in 10 chunks, wait for it to show up in the output view.
-    for chunk in data.chunks(std::cmp::max(data.len() / 10, 1)) {
-        write_updates_as_json(input_file.as_file_mut(), chunk, true);
-        total_count += chunk.len();
+    let mut chunks = data.chunks(std::cmp::max(data.len() / 10, 1));
 
+    // Write data in 10 chunks, wait for it to show up in the output view.
+    loop {
+        // Perform the first suspend while the table is empty to test that the delta lake connector
+        // can successfully suspend when reading an empty table.
         if suspend {
             println!("start suspend");
             let (sender, mut receiver) = mpsc::channel(1);
@@ -1103,6 +1104,13 @@ async fn test_cdc(
                 .collect::<Vec<_>>(),
         )
         .await;
+
+        let Some(chunk) = chunks.next() else {
+            break;
+        };
+
+        write_updates_as_json(input_file.as_file_mut(), chunk, true);
+        total_count += chunk.len();
     }
 
     // Modify all records by negating the `boolean` field.
