feat(table_changes): return actual row data from Parquet files

Author: shefeek-jinnah

src/metadata_provider.rs

@@ -70,7 +70,12 @@ pub const SQL_TABLE_EXISTS: &str = "SELECT EXISTS(
 // Queries for table_changes (CDC) - files added/removed between snapshots
 
 pub const SQL_GET_DATA_FILES_ADDED_BETWEEN_SNAPSHOTS: &str = "
-    SELECT data.begin_snapshot
+    SELECT
+        data.begin_snapshot,
+        data.path,
+        data.path_is_relative,
+        data.file_size_bytes,
+        data.footer_size
     FROM ducklake_data_file AS data
     WHERE data.table_id = ?
       AND data.begin_snapshot > ?
@@ -293,6 +298,10 @@ impl DuckLakeTableFile {
 #[derive(Debug, Clone)]
 pub struct DataFileChange {
     pub begin_snapshot: i64,
+    pub path: String,
+    pub path_is_relative: bool,
+    pub file_size_bytes: i64,
+    pub footer_size: Option<i64>,
 }
 
 #[derive(Debug, Clone)]

src/metadata_provider_duckdb.rs

@@ -394,6 +394,10 @@ impl MetadataProvider for DuckdbMetadataProvider {
             .query_map(params![table_id, start_snapshot, end_snapshot], |row| {
                 Ok(DataFileChange {
                     begin_snapshot: row.get(0)?,
+                    path: row.get(1)?,
+                    path_is_relative: row.get(2)?,
+                    file_size_bytes: row.get(3)?,
+                    footer_size: row.get(4)?,
                 })
             })?
             .collect::<Result<Vec<_>, _>>()?;

src/metadata_provider_postgres.rs

@@ -495,7 +495,12 @@ impl MetadataProvider for PostgresMetadataProvider {
     ) -> Result<Vec<DataFileChange>> {
         block_on(async {
             let rows = sqlx::query(
-                "SELECT data.begin_snapshot
+                "SELECT
+                    data.begin_snapshot,
+                    data.path,
+                    data.path_is_relative,
+                    data.file_size_bytes,
+                    data.footer_size
                 FROM ducklake_data_file AS data
                 WHERE data.table_id = $1
                   AND data.begin_snapshot > $2
@@ -512,6 +517,10 @@ impl MetadataProvider for PostgresMetadataProvider {
                 .map(|row| {
                     Ok(DataFileChange {
                         begin_snapshot: row.try_get(0)?,
+                        path: row.try_get(1)?,
+                        path_is_relative: row.try_get(2)?,
+                        file_size_bytes: row.try_get(3)?,
+                        footer_size: row.try_get(4)?,
                     })
                 })
                 .collect()