chore: add checkpoint mechanism

Author: benjamin-awd

src/sinks/delta_lake/service.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use std::task::{Context, Poll};
 
 use deltalake::DeltaTableError;
+use deltalake::checkpoints::create_checkpoint;
 use deltalake::datafusion::datasource::TableProvider;
 use deltalake::operations::write::SchemaMode;
 use deltalake::protocol::SaveMode;
@@ -13,6 +14,11 @@ use crate::sinks::prelude::*;
 
 use super::request_builder::{DeltaLakeRequest, SharedSchema};
 
+/// Checkpoint interval - create a checkpoint every N commits.
+/// Checkpoints consolidate the transaction log into a single parquet file,
+/// dramatically improving read performance for tables with many versions.
+const CHECKPOINT_INTERVAL: i64 = 10;
+
 /// Response from Delta Lake write operations.
 ///
 /// Contains metrics about the write operation including the number of files
@@ -203,6 +209,30 @@ impl Service<DeltaLakeRequest> for DeltaLakeService {
                             shared_schema.store(new_schema);
                         }
 
+                        // Create checkpoint if we've hit the interval.
+                        // Checkpoints consolidate the transaction log into a single parquet file,
+                        // which dramatically reduces memory usage when opening the table.
+                        if let Some(version) = new_table.version() {
+                            if version % CHECKPOINT_INTERVAL == 0 {
+                                match create_checkpoint(&new_table, None).await {
+                                    Ok(()) => {
+                                        info!(
+                                            message = "Created checkpoint",
+                                            version = version,
+                                        );
+                                    }
+                                    Err(e) => {
+                                        // Log but don't fail the write - checkpoint is best-effort
+                                        warn!(
+                                            message = "Failed to create checkpoint",
+                                            version = version,
+                                            error = %e,
+                                        );
+                                    }
+                                }
+                            }
+                        }
+
                         // Get the byte size from the request (Arrow in-memory size)
                         let bytes_written = request.byte_size;
 

src/sinks/mod.rs

@@ -47,6 +47,8 @@ pub mod databend;
 pub mod datadog;
 #[cfg(feature = "sinks-delta_lake")]
 pub mod delta_lake;
+#[cfg(feature = "sinks-iceberg")]
+pub mod iceberg;
 #[cfg(feature = "sinks-elasticsearch")]
 pub mod elasticsearch;
 #[cfg(feature = "sinks-file")]