Implements Phase 1 of ducklake_table_changes() table function

Author: shefeek-jinnah

src/lib.rs

@@ -43,6 +43,7 @@ pub mod metadata_provider;
 pub mod path_resolver;
 pub mod schema;
 pub mod table;
+pub mod table_changes;
 pub mod table_functions;
 pub mod types;
 

src/metadata_provider.rs

@@ -67,6 +67,24 @@ pub const SQL_TABLE_EXISTS: &str = "SELECT EXISTS(
          AND (? < end_snapshot OR end_snapshot IS NULL)
      )";
 
+// Queries for table_changes (CDC) - files added/removed between snapshots
+
+pub const SQL_GET_DATA_FILES_ADDED_BETWEEN_SNAPSHOTS: &str = "
+    SELECT data.begin_snapshot
+    FROM ducklake_data_file AS data
+    WHERE data.table_id = ?
+      AND data.begin_snapshot > ?
+      AND data.begin_snapshot <= ?
+    ORDER BY data.begin_snapshot";
+
+pub const SQL_GET_DELETE_FILES_ADDED_BETWEEN_SNAPSHOTS: &str = "
+    SELECT del.begin_snapshot
+    FROM ducklake_delete_file AS del
+    WHERE del.table_id = ?
+      AND del.begin_snapshot > ?
+      AND del.begin_snapshot <= ?
+    ORDER BY del.begin_snapshot";
+
 // Bulk queries for information_schema (avoids N+1 query problem)
 
 pub const SQL_LIST_ALL_TABLES: &str = "
@@ -270,6 +288,18 @@ impl DuckLakeTableFile {
     }
 }
 
+// Change tracking structures for table_changes (CDC) functionality
+
+#[derive(Debug, Clone)]
+pub struct DataFileChange {
+    pub begin_snapshot: i64,
+}
+
+#[derive(Debug, Clone)]
+pub struct DeleteFileChange {
+    pub begin_snapshot: i64,
+}
+
 pub trait MetadataProvider: Send + Sync + std::fmt::Debug {
     /// Get the current snapshot ID (dynamic, not cached)
     fn get_current_snapshot(&self) -> Result<i64>;
@@ -323,6 +353,28 @@ pub trait MetadataProvider: Send + Sync + std::fmt::Debug {
 
     /// List all files across all tables for a snapshot
     fn list_all_files(&self, snapshot_id: i64) -> Result<Vec<FileWithTable>>;
+
+    // Change tracking methods for table_changes (CDC) functionality
+
+    /// Get data files added between two snapshots (exclusive start, inclusive end)
+    /// Returns files where begin_snapshot > start_snapshot AND begin_snapshot <= end_snapshot
+    /// These represent INSERT changes - new rows added to the table
+    fn get_data_files_added_between_snapshots(
+        &self,
+        table_id: i64,
+        start_snapshot: i64,
+        end_snapshot: i64,
+    ) -> Result<Vec<DataFileChange>>;
+
+    /// Get delete files added between two snapshots (exclusive start, inclusive end)
+    /// Returns delete files where begin_snapshot > start_snapshot AND begin_snapshot <= end_snapshot
+    /// These represent DELETE changes - rows removed from the table
+    fn get_delete_files_added_between_snapshots(
+        &self,
+        table_id: i64,
+        start_snapshot: i64,
+        end_snapshot: i64,
+    ) -> Result<Vec<DeleteFileChange>>;
 }
 
 #[cfg(feature = "metadata-postgres")]

src/metadata_provider_duckdb.rs

@@ -1,10 +1,12 @@
 use crate::DuckLakeError;
 use crate::metadata_provider::{
-    ColumnWithTable, DuckLakeFileData, DuckLakeTableColumn, DuckLakeTableFile, FileWithTable,
-    MetadataProvider, SQL_GET_DATA_FILES, SQL_GET_DATA_PATH, SQL_GET_LATEST_SNAPSHOT,
-    SQL_GET_SCHEMA_BY_NAME, SQL_GET_TABLE_BY_NAME, SQL_GET_TABLE_COLUMNS, SQL_LIST_ALL_COLUMNS,
-    SQL_LIST_ALL_FILES, SQL_LIST_ALL_TABLES, SQL_LIST_SCHEMAS, SQL_LIST_SNAPSHOTS, SQL_LIST_TABLES,
-    SQL_TABLE_EXISTS, SchemaMetadata, SnapshotMetadata, TableMetadata, TableWithSchema,
+    ColumnWithTable, DataFileChange, DeleteFileChange, DuckLakeFileData, DuckLakeTableColumn,
+    DuckLakeTableFile, FileWithTable, MetadataProvider, SQL_GET_DATA_FILES,
+    SQL_GET_DATA_FILES_ADDED_BETWEEN_SNAPSHOTS, SQL_GET_DATA_PATH,
+    SQL_GET_DELETE_FILES_ADDED_BETWEEN_SNAPSHOTS, SQL_GET_LATEST_SNAPSHOT, SQL_GET_SCHEMA_BY_NAME,
+    SQL_GET_TABLE_BY_NAME, SQL_GET_TABLE_COLUMNS, SQL_LIST_ALL_COLUMNS, SQL_LIST_ALL_FILES,
+    SQL_LIST_ALL_TABLES, SQL_LIST_SCHEMAS, SQL_LIST_SNAPSHOTS, SQL_LIST_TABLES, SQL_TABLE_EXISTS,
+    SchemaMetadata, SnapshotMetadata, TableMetadata, TableWithSchema,
 };
 use duckdb::AccessMode::ReadOnly;
 use duckdb::{Config, Connection, params};
@@ -378,4 +380,44 @@ impl MetadataProvider for DuckdbMetadataProvider {
 
         Ok(files)
     }
+
+    fn get_data_files_added_between_snapshots(
+        &self,
+        table_id: i64,
+        start_snapshot: i64,
+        end_snapshot: i64,
+    ) -> crate::Result<Vec<DataFileChange>> {
+        let conn = self.open_connection()?;
+        let mut stmt = conn.prepare(SQL_GET_DATA_FILES_ADDED_BETWEEN_SNAPSHOTS)?;
+
+        let files = stmt
+            .query_map(params![table_id, start_snapshot, end_snapshot], |row| {
+                Ok(DataFileChange {
+                    begin_snapshot: row.get(0)?,
+                })
+            })?
+            .collect::<Result<Vec<_>, _>>()?;
+
+        Ok(files)
+    }
+
+    fn get_delete_files_added_between_snapshots(
+        &self,
+        table_id: i64,
+        start_snapshot: i64,
+        end_snapshot: i64,
+    ) -> crate::Result<Vec<DeleteFileChange>> {
+        let conn = self.open_connection()?;
+        let mut stmt = conn.prepare(SQL_GET_DELETE_FILES_ADDED_BETWEEN_SNAPSHOTS)?;
+
+        let files = stmt
+            .query_map(params![table_id, start_snapshot, end_snapshot], |row| {
+                Ok(DeleteFileChange {
+                    begin_snapshot: row.get(0)?,
+                })
+            })?
+            .collect::<Result<Vec<_>, _>>()?;
+
+        Ok(files)
+    }
 }

src/metadata_provider_postgres.rs

@@ -2,8 +2,9 @@
 
 use crate::Result;
 use crate::metadata_provider::{
-    ColumnWithTable, DuckLakeFileData, DuckLakeTableColumn, DuckLakeTableFile, FileWithTable,
-    MetadataProvider, SchemaMetadata, SnapshotMetadata, TableMetadata, TableWithSchema, block_on,
+    ColumnWithTable, DataFileChange, DeleteFileChange, DuckLakeFileData, DuckLakeTableColumn,
+    DuckLakeTableFile, FileWithTable, MetadataProvider, SchemaMetadata, SnapshotMetadata,
+    TableMetadata, TableWithSchema, block_on,
 };
 use sqlx::Row;
 use sqlx::postgres::{PgPool, PgPoolOptions};
@@ -485,4 +486,66 @@ impl MetadataProvider for PostgresMetadataProvider {
                 .collect()
         })
     }
+
+    fn get_data_files_added_between_snapshots(
+        &self,
+        table_id: i64,
+        start_snapshot: i64,
+        end_snapshot: i64,
+    ) -> Result<Vec<DataFileChange>> {
+        block_on(async {
+            let rows = sqlx::query(
+                "SELECT data.begin_snapshot
+                FROM ducklake_data_file AS data
+                WHERE data.table_id = $1
+                  AND data.begin_snapshot > $2
+                  AND data.begin_snapshot <= $3
+                ORDER BY data.begin_snapshot",
+            )
+            .bind(table_id)
+            .bind(start_snapshot)
+            .bind(end_snapshot)
+            .fetch_all(&self.pool)
+            .await?;
+
+            rows.into_iter()
+                .map(|row| {
+                    Ok(DataFileChange {
+                        begin_snapshot: row.try_get(0)?,
+                    })
+                })
+                .collect()
+        })
+    }
+
+    fn get_delete_files_added_between_snapshots(
+        &self,
+        table_id: i64,
+        start_snapshot: i64,
+        end_snapshot: i64,
+    ) -> Result<Vec<DeleteFileChange>> {
+        block_on(async {
+            let rows = sqlx::query(
+                "SELECT del.begin_snapshot
+                FROM ducklake_delete_file AS del
+                WHERE del.table_id = $1
+                  AND del.begin_snapshot > $2
+                  AND del.begin_snapshot <= $3
+                ORDER BY del.begin_snapshot",
+            )
+            .bind(table_id)
+            .bind(start_snapshot)
+            .bind(end_snapshot)
+            .fetch_all(&self.pool)
+            .await?;
+
+            rows.into_iter()
+                .map(|row| {
+                    Ok(DeleteFileChange {
+                        begin_snapshot: row.try_get(0)?,
+                    })
+                })
+                .collect()
+        })
+    }
 }

src/table_changes.rs

@@ -0,0 +1,133 @@
+//! Table changes (CDC) functionality for DuckLake
+
+use std::any::Any;
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, Int64Array, StringArray};
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use arrow::record_batch::RecordBatch;
+use datafusion::catalog::Session;
+use datafusion::common::Result as DataFusionResult;
+use datafusion::datasource::memory::MemTable;
+use datafusion::datasource::{TableProvider, TableType};
+use datafusion::physical_plan::ExecutionPlan;
+
+use crate::metadata_provider::MetadataProvider;
+
+#[derive(Debug)]
+pub struct TableChangesTable {
+    provider: Arc<dyn MetadataProvider>,
+    table_id: i64,
+    start_snapshot: i64,
+    end_snapshot: i64,
+    schema: SchemaRef,
+}
+
+impl TableChangesTable {
+    pub fn new(
+        provider: Arc<dyn MetadataProvider>,
+        table_id: i64,
+        start_snapshot: i64,
+        end_snapshot: i64,
+    ) -> Self {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("snapshot_id", DataType::Int64, false),
+            Field::new("change_type", DataType::Utf8, false),
+        ]));
+        Self {
+            provider,
+            table_id,
+            start_snapshot,
+            end_snapshot,
+            schema,
+        }
+    }
+
+    fn query_changes(&self) -> DataFusionResult<RecordBatch> {
+        // Get data files added (INSERT changes)
+        let data_files = self
+            .provider
+            .get_data_files_added_between_snapshots(
+                self.table_id,
+                self.start_snapshot,
+                self.end_snapshot,
+            )
+            .map_err(|e| datafusion::error::DataFusionError::External(Box::new(e)))?;
+
+        // Get delete files added (DELETE changes)
+        let delete_files = self
+            .provider
+            .get_delete_files_added_between_snapshots(
+                self.table_id,
+                self.start_snapshot,
+                self.end_snapshot,
+            )
+            .map_err(|e| datafusion::error::DataFusionError::External(Box::new(e)))?;
+
+        // Collect all changes into a sortable structure
+        struct ChangeRecord {
+            snapshot_id: i64,
+            change_type: &'static str,
+        }
+
+        let mut changes: Vec<ChangeRecord> =
+            Vec::with_capacity(data_files.len() + delete_files.len());
+
+        // Add INSERT changes (data files added)
+        for data_file in &data_files {
+            changes.push(ChangeRecord {
+                snapshot_id: data_file.begin_snapshot,
+                change_type: "insert",
+            });
+        }
+
+        // Add DELETE changes (delete files added)
+        for delete_file in &delete_files {
+            changes.push(ChangeRecord {
+                snapshot_id: delete_file.begin_snapshot,
+                change_type: "delete",
+            });
+        }
+
+        // Sort by snapshot_id for deterministic output
+        changes.sort_by_key(|c| c.snapshot_id);
+
+        // Build arrays from sorted changes
+        let snapshot_ids: ArrayRef = Arc::new(Int64Array::from(
+            changes.iter().map(|c| c.snapshot_id).collect::<Vec<_>>(),
+        ));
+        let change_types: ArrayRef = Arc::new(StringArray::from(
+            changes.iter().map(|c| c.change_type).collect::<Vec<_>>(),
+        ));
+
+        RecordBatch::try_new(self.schema.clone(), vec![snapshot_ids, change_types])
+            .map_err(|e| datafusion::error::DataFusionError::ArrowError(Box::new(e), None))
+    }
+}
+
+#[async_trait::async_trait]
+impl TableProvider for TableChangesTable {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn schema(&self) -> SchemaRef {
+        self.schema.clone()
+    }
+
+    fn table_type(&self) -> TableType {
+        TableType::View
+    }
+
+    async fn scan(
+        &self,
+        state: &dyn Session,
+        projection: Option<&Vec<usize>>,
+        filters: &[datafusion::prelude::Expr],
+        limit: Option<usize>,
+    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+        let batch = self.query_changes()?;
+        let mem_table = MemTable::try_new(self.schema.clone(), vec![vec![batch]])?;
+        mem_table.scan(state, projection, filters, limit).await
+    }
+}