Aligning it as per ducklake spec

Author: Shefeek Jinnah

src/information_schema.rs

@@ -725,9 +725,6 @@ impl TableChangesTable {
         let schema = Arc::new(Schema::new(vec![
             Field::new("snapshot_id", DataType::Int64, false),
             Field::new("change_type", DataType::Utf8, false),
-            Field::new("file_path", DataType::Utf8, false),
-            Field::new("file_size_bytes", DataType::Int64, false),
-            Field::new("row_count", DataType::Int64, true),
         ]));
         Self {
             provider,
@@ -763,9 +760,6 @@ impl TableChangesTable {
         struct ChangeRecord {
             snapshot_id: i64,
             change_type: &'static str,
-            file_path: String,
-            file_size_bytes: i64,
-            row_count: Option<i64>,
         }
 
         let mut changes: Vec<ChangeRecord> =
@@ -776,9 +770,6 @@ impl TableChangesTable {
             changes.push(ChangeRecord {
                 snapshot_id: data_file.begin_snapshot,
                 change_type: "insert",
-                file_path: data_file.file.path.clone(),
-                file_size_bytes: data_file.file.file_size_bytes,
-                row_count: None,
             });
         }
 
@@ -787,9 +778,6 @@ impl TableChangesTable {
             changes.push(ChangeRecord {
                 snapshot_id: delete_file.begin_snapshot,
                 change_type: "delete",
-                file_path: delete_file.delete_file.path.clone(),
-                file_size_bytes: delete_file.delete_file.file_size_bytes,
-                row_count: delete_file.delete_count,
             });
         }
 
@@ -803,27 +791,9 @@ impl TableChangesTable {
         let change_types: ArrayRef = Arc::new(StringArray::from(
             changes.iter().map(|c| c.change_type).collect::<Vec<_>>(),
         ));
-        let file_paths: ArrayRef = Arc::new(StringArray::from(
-            changes
-                .iter()
-                .map(|c| c.file_path.as_str())
-                .collect::<Vec<_>>(),
-        ));
-        let file_sizes: ArrayRef = Arc::new(Int64Array::from(
-            changes
-                .iter()
-                .map(|c| c.file_size_bytes)
-                .collect::<Vec<_>>(),
-        ));
-        let row_counts: ArrayRef = Arc::new(Int64Array::from(
-            changes.iter().map(|c| c.row_count).collect::<Vec<_>>(),
-        ));
 
-        RecordBatch::try_new(
-            self.schema.clone(),
-            vec![snapshot_ids, change_types, file_paths, file_sizes, row_counts],
-        )
-        .map_err(|e| datafusion::error::DataFusionError::ArrowError(Box::new(e), None))
+        RecordBatch::try_new(self.schema.clone(), vec![snapshot_ids, change_types])
+            .map_err(|e| datafusion::error::DataFusionError::ArrowError(Box::new(e), None))
     }
 }
 

tests/table_changes_tests.rs

@@ -92,7 +92,9 @@ mod integration_tests {
 
         // Query changes from snapshot 0 to current (should include all inserts)
         let df = ctx
-            .sql("SELECT change_type, file_path FROM ducklake_table_changes('main.events', 0, 10)")
+            .sql(
+                "SELECT snapshot_id, change_type FROM ducklake_table_changes('main.events', 0, 10)",
+            )
             .await?;
 
         let batches: Vec<RecordBatch> = df.collect().await?;
@@ -101,9 +103,9 @@ mod integration_tests {
         let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
         assert!(total_rows > 0, "Should have some changes");
 
-        // Check that we have insert changes
+        // Check that we have insert changes (change_type is column 1)
         for batch in &batches {
-            let change_types = get_string_column(batch, 0);
+            let change_types = get_string_column(batch, 1);
             for change_type in change_types {
                 assert!(
                     change_type == "insert" || change_type == "delete",
@@ -130,7 +132,7 @@ mod integration_tests {
         // Query only delete changes (from last snapshot which has the delete)
         // We need to find the delete files by querying a range that includes the delete operation
         let df = ctx
-            .sql("SELECT change_type, row_count FROM ducklake_table_changes('main.events', 0, 100) WHERE change_type = 'delete'")
+            .sql("SELECT snapshot_id, change_type FROM ducklake_table_changes('main.events', 0, 100) WHERE change_type = 'delete'")
             .await?;
 
         let batches: Vec<RecordBatch> = df.collect().await?;
@@ -187,7 +189,7 @@ mod integration_tests {
 
         let schema = df.schema();
 
-        // Verify the expected columns are present
+        // Verify the expected columns are present (Phase 1: only spec columns)
         let field_names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
 
         assert!(
@@ -198,18 +200,6 @@ mod integration_tests {
             field_names.contains(&"change_type"),
             "Schema should contain 'change_type'"
         );
-        assert!(
-            field_names.contains(&"file_path"),
-            "Schema should contain 'file_path'"
-        );
-        assert!(
-            field_names.contains(&"file_size_bytes"),
-            "Schema should contain 'file_size_bytes'"
-        );
-        assert!(
-            field_names.contains(&"row_count"),
-            "Schema should contain 'row_count'"
-        );
 
         Ok(())
     }