[adapters] Delta output: enable checkpoints.

ryzhyk · ryzhyk · commit e20b7835b037 · 2026-04-08T08:10:31.000-07:00
The Delta output connector did not create periodic checkpoints. While this is in itself problematic, it also meant that the connector became slow over time, due to this delta-rs bug, which causes the `update_incremental` function to scan the entire transaction log on every commit: delta-io/delta-kernel-rs#2103. This commit: - Introduces the `checkpoint_interval` option, which tells the connector to configure checkpoint interval when creating the table. - Creates a CommitBuilder that is actually setup to create checkpoints. Without this fix the time to create a trivial delta commit increases from 1.5s to 6s after ~1000 commits. With the fix it remains constant at ~2s. Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
diff --git a/crates/adapters/benches/delta_encoder.rs b/crates/adapters/benches/delta_encoder.rs
@@ -19,6 +19,7 @@ fn create_indexed_writer(threads: usize, table_uri: &str) -> DeltaTableWriter {
         max_retries: Some(0),
         threads: Some(threads),
         object_store_config: Default::default(),
+        checkpoint_interval: None,
     };
     let key_schema = Some(BenchKeyStruct::relation_schema());
     let mut value_schema = BenchTestStruct::relation_schema();
diff --git a/crates/adapters/src/integrated/delta_table/output.rs b/crates/adapters/src/integrated/delta_table/output.rs
@@ -16,7 +16,7 @@ use dbsp::circuit::tokio::TOKIO;
 use delta_kernel::engine::arrow_conversion::TryFromArrow;
 use delta_kernel::table_properties::DataSkippingNumIndexedCols;
 use deltalake::DeltaTable;
-use deltalake::kernel::transaction::{CommitBuilder, TableReference};
+use deltalake::kernel::transaction::{CommitBuilder, CommitProperties, TableReference};
 use deltalake::kernel::{Action, Add, DataType, StructField};
 use deltalake::logstore::ObjectStoreRef;
 use deltalake::operations::create::CreateBuilder;
@@ -281,11 +281,20 @@ impl WriterTask {
             let mut operation_timeout: Duration = Duration::from_secs(60);
 
             loop {
+                let checkpoint_interval = match inner.config.checkpoint_interval {
+                    Some(0) => None,
+                    Some(interval) => Some(interval.to_string()),
+                    None => Some("10".to_string()),
+                };
                 let create_future = CreateBuilder::new()
                     .with_location(inner.config.uri.clone())
                     .with_save_mode(save_mode)
                     .with_storage_options(storage_options.clone())
-                    .with_columns(inner.struct_fields.clone());
+                    .with_columns(inner.struct_fields.clone())
+                    .with_configuration_property(
+                        deltalake::TableProperty::CheckpointInterval,
+                        checkpoint_interval,
+                    );
 
                 match tokio::time::timeout(operation_timeout, create_future).await {
                     Ok(Ok(table)) => break table,
@@ -357,7 +366,11 @@ impl WriterTask {
                 ))
             })?;
 
-        CommitBuilder::default()
+        // `CommitBuilder::default()` leaves `post_commit_hook` unset, so delta-rs skips the
+        // post-commit hook entirely and never writes `_last_checkpoint` / `*.checkpoint.parquet`,
+        // regardless of `delta.checkpointInterval`. Use default commit properties so checkpoint
+        // creation runs when `(version + 1) % checkpoint_interval == 0`.
+        CommitBuilder::from(CommitProperties::default())
             .with_actions(
                 actions
                     .iter()
@@ -988,6 +1001,7 @@ mod parallel {
                 max_retries: Some(0),
                 threads: Some(threads),
                 object_store_config: Default::default(),
+                checkpoint_interval: None,
             },
             &key_schema,
             &value_relation(),
@@ -1268,6 +1282,7 @@ mod parallel {
                 max_retries: Some(0),
                 threads: Some(4),
                 object_store_config: Default::default(),
+                checkpoint_interval: None,
             },
             &key_schema,
             &value_relation(),
@@ -1405,6 +1420,7 @@ mod parallel {
                 max_retries: Some(1),
                 threads: Some(1),
                 object_store_config: Default::default(),
+                checkpoint_interval: None,
             },
             &key_schema,
             &value_relation(),
@@ -1433,6 +1449,7 @@ mod parallel {
             max_retries: Some(0),
             threads: Some(0),
             object_store_config: Default::default(),
+            checkpoint_interval: None,
         };
         assert!(config.validate().is_err());
     }
diff --git a/crates/adapters/src/integrated/delta_table/test.rs b/crates/adapters/src/integrated/delta_table/test.rs
@@ -646,8 +646,14 @@ fn delta_table_output_test(
 
         let mut output_records = Vec::with_capacity(data.len());
         for parquet_file in parquet_files {
-            let mut records: Vec<DeltaTestStruct> = load_parquet_file(&parquet_file);
-            output_records.append(&mut records);
+            if !parquet_file
+                .display()
+                .to_string()
+                .ends_with(".checkpoint.parquet")
+            {
+                let mut records: Vec<DeltaTestStruct> = load_parquet_file(&parquet_file);
+                output_records.append(&mut records);
+            }
         }
 
         output_records.sort();
diff --git a/crates/feldera-types/src/transport/delta_table.rs b/crates/feldera-types/src/transport/delta_table.rs
@@ -9,6 +9,8 @@ use utoipa::ToSchema;
 #[derive(Default, Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)]
 pub enum DeltaTableWriteMode {
     /// New updates will be appended to the existing table at the target location.
+    ///
+    /// If the table doesn't exist, it is created.
     #[default]
     #[serde(rename = "append")]
     Append,
@@ -17,6 +19,8 @@ pub enum DeltaTableWriteMode {
     ///
     /// The connector truncates the table by outputting delete actions for all
     /// files in the latest snapshot of the table.
+    ///
+    /// If the table doesn't exist, it is created.
     #[serde(rename = "truncate")]
     Truncate,
 
@@ -35,6 +39,18 @@ pub struct DeltaTableWriterConfig {
     #[serde(default)]
     pub mode: DeltaTableWriteMode,
 
+    /// Checkpoint interval (i.e., the number of commits after which a new checkpoint should be created) for newly created Delta tables.
+    ///
+    /// The option is only available when creating the Delta table (`mode = append` and there
+    /// is no existing table at the target location or `mode = truncate`). It configures the `checkpointInterval`
+    /// table property, which determines the number of commits after which a new checkpoint should be created.
+    ///
+    /// 0 means no checkpoints are created.
+    ///
+    /// Default: 10.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub checkpoint_interval: Option<u32>,
+
     /// Maximum number of retries for failed operations.
     ///
     /// The connector performs retries on several levels: individual S3 operations, Delta Lake transaction commits,
diff --git a/docs.feldera.com/docs/connectors/sinks/delta.md b/docs.feldera.com/docs/connectors/sinks/delta.md
@@ -73,9 +73,10 @@ MERGE INTO {target_table} AS target
 |------------|------------|
 | `uri`*     | Table URI, e.g., `"s3://feldera-fraud-detection-data/feature_train"`. |
 | `mode`*    | Determines how the Delta table connector handles an existing table at the target location. Options: |
-|            | - `append`: New updates will be appended to the existing table at the target location. |
+|            | - `append`: New updates will be appended to the existing table at the target location. If the table doesn't exist, it will be created. |
 |            | - `truncate`: Existing table at the specified location will be truncated. The connector achieves this by outputting delete actions for all files in the latest snapshot of the table. |
 |            | - `error_if_exists`: If a table exists at the specified location, the operation will fail. |
+| `checkpoint_interval` | <p>Checkpoint interval (i.e., the number of commits after which a new checkpoint should be created) for newly created Delta tables.</p><p>The option is only available when creating the Delta table (`mode = append` and there is no existing table at the target location or `mode = truncate`). It configures the `checkpointInterval` table property, which determines the number of commits after which a new checkpoint should be created.</p><p>0 means no checkpoints are created.</p><p>Default: 10.</p>|
 | `max_retries`|<p>Maximum number of retries for failed Delta Lake operations like writing Parquet files and committing transactions.</p><p>The connector performs retries on several levels: individual S3 operations, Delta Lake transaction commits, and overall operation retries. This setting controls the overall operation retries. When a write to the table fails, because of an S3 timeout or any other reason that was not resolved by lower-level retries, the connector will retry the entire operation.</p><p>When not specified, the connector performs infinite retries. When set to 0, the connector doesn't retry failed operations.</p>|
 | `threads` | Number of parallel threads used by the connector. Increasing this value can improve Delta Lake write throughput by enabling concurrent writes. Default: `1`. |
 
diff --git a/openapi.json b/openapi.json
@@ -7997,6 +7997,13 @@
           "uri"
         ],
         "properties": {
+          "checkpoint_interval": {
+            "type": "integer",
+            "format": "int32",
+            "description": "Checkpoint interval (i.e., the number of commits after which a new checkpoint should be created) for newly created Delta tables.\n\nThe option is only available when creating the Delta table (`mode = append` and there\nis no existing table at the target location or `mode = truncate`). It configures the `checkpointInterval`\ntable property, which determines the number of commits after which a new checkpoint should be created.\n\n0 means no checkpoints are created.\n\nDefault: 10.",
+            "nullable": true,
+            "minimum": 0
+          },
           "max_retries": {
             "type": "integer",
             "format": "int32",
