Fix: Filter data files by snapshot to handle complete row deletion (#32)

Author: shefeek-jinnah

src/metadata_provider.rs

@@ -43,7 +43,9 @@ pub const SQL_GET_DATA_FILES: &str = "
         AND del.table_id = ?
         AND ? >= del.begin_snapshot
         AND (? < del.end_snapshot OR del.end_snapshot IS NULL)
-    WHERE data.table_id = ?";
+    WHERE data.table_id = ?
+      AND ? >= data.begin_snapshot
+      AND (? < data.end_snapshot OR data.end_snapshot IS NULL)";
 
 pub const SQL_GET_DATA_PATH: &str =
     "SELECT value FROM ducklake_metadata WHERE key = 'data_path' AND scope IS NULL";
@@ -155,6 +157,8 @@ pub const SQL_LIST_ALL_FILES: &str = "
       AND (? < s.end_snapshot OR s.end_snapshot IS NULL)
       AND ? >= t.begin_snapshot
       AND (? < t.end_snapshot OR t.end_snapshot IS NULL)
+      AND ? >= data.begin_snapshot
+      AND (? < data.end_snapshot OR data.end_snapshot IS NULL)
     ORDER BY s.schema_name, t.table_name, data.path";
 
 /// Metadata for a snapshot in the DuckLake catalog

src/metadata_provider_duckdb.rs

@@ -165,40 +165,43 @@ impl MetadataProvider for DuckdbMetadataProvider {
         let mut stmt = conn.prepare(SQL_GET_DATA_FILES)?;
 
         let files = stmt
-            .query_map([table_id, snapshot_id, snapshot_id, table_id], |row| {
-                // Parse data file (columns 0-5)
-                let _data_file_id: i64 = row.get(0)?;
-                let data_file = DuckLakeFileData {
-                    path: row.get(1)?,
-                    path_is_relative: row.get(2)?,
-                    file_size_bytes: row.get(3)?,
-                    footer_size: row.get(4)?,
-                    encryption_key: row.get(5)?,
-                };
-
-                // Parse delete file (columns 6-12) if exists
-                let delete_file = if let Ok(Some(_)) = row.get::<_, Option<i64>>(6) {
-                    Some(DuckLakeFileData {
-                        path: row.get(7)?,
-                        path_is_relative: row.get(8)?,
-                        file_size_bytes: row.get(9)?,
-                        footer_size: row.get(10)?,
-                        encryption_key: row.get(11)?,
+            .query_map(
+                [table_id, snapshot_id, snapshot_id, table_id, snapshot_id, snapshot_id],
+                |row| {
+                    // Parse data file (columns 0-5)
+                    let _data_file_id: i64 = row.get(0)?;
+                    let data_file = DuckLakeFileData {
+                        path: row.get(1)?,
+                        path_is_relative: row.get(2)?,
+                        file_size_bytes: row.get(3)?,
+                        footer_size: row.get(4)?,
+                        encryption_key: row.get(5)?,
+                    };
+
+                    // Parse delete file (columns 6-12) if exists
+                    let delete_file = if let Ok(Some(_)) = row.get::<_, Option<i64>>(6) {
+                        Some(DuckLakeFileData {
+                            path: row.get(7)?,
+                            path_is_relative: row.get(8)?,
+                            file_size_bytes: row.get(9)?,
+                            footer_size: row.get(10)?,
+                            encryption_key: row.get(11)?,
+                        })
+                    } else {
+                        None
+                    };
+
+                    let _delete_count: Option<i64> = row.get(12)?;
+
+                    Ok(DuckLakeTableFile {
+                        file: data_file,
+                        delete_file,
+                        row_id_start: None,
+                        snapshot_id: Some(snapshot_id),
+                        max_row_count: None, // Set to None until we have actual row count from data file metadata
                     })
-                } else {
-                    None
-                };
-
-                let _delete_count: Option<i64> = row.get(12)?;
-
-                Ok(DuckLakeTableFile {
-                    file: data_file,
-                    delete_file,
-                    row_id_start: None,
-                    snapshot_id: Some(snapshot_id),
-                    max_row_count: None, // Set to None until we have actual row count from data file metadata
-                })
-            })?
+                },
+            )?
             .collect::<Result<Vec<_>, _>>()?;
 
         Ok(files)
@@ -332,6 +335,8 @@ impl MetadataProvider for DuckdbMetadataProvider {
                     snapshot_id,
                     snapshot_id,
                     snapshot_id,
+                    snapshot_id,
+                    snapshot_id,
                     snapshot_id
                 ],
                 |row| {

src/metadata_provider_postgres.rs

@@ -32,6 +32,17 @@ macro_rules! bind_repeat {
             .bind($value)
             .bind($value)
     };
+    ($query:expr, $value:expr, 8) => {
+        $query
+            .bind($value)
+            .bind($value)
+            .bind($value)
+            .bind($value)
+            .bind($value)
+            .bind($value)
+            .bind($value)
+            .bind($value)
+    };
 }
 
 /// PostgreSQL-based metadata provider for DuckLake catalogs.
@@ -211,12 +222,16 @@ impl MetadataProvider for PostgresMetadataProvider {
                     AND del.table_id = $1
                     AND $2 >= del.begin_snapshot
                     AND ($3 < del.end_snapshot OR del.end_snapshot IS NULL)
-                WHERE data.table_id = $4",
+                WHERE data.table_id = $4
+                  AND $5 >= data.begin_snapshot
+                  AND ($6 < data.end_snapshot OR data.end_snapshot IS NULL)",
             )
             .bind(table_id)
             .bind(snapshot_id)
             .bind(snapshot_id)
             .bind(table_id)
+            .bind(snapshot_id)
+            .bind(snapshot_id)
             .fetch_all(&self.pool)
             .await?;
 
@@ -442,6 +457,8 @@ impl MetadataProvider for PostgresMetadataProvider {
                   AND ($4 < s.end_snapshot OR s.end_snapshot IS NULL)
                   AND $5 >= t.begin_snapshot
                   AND ($6 < t.end_snapshot OR t.end_snapshot IS NULL)
+                  AND $7 >= data.begin_snapshot
+                  AND ($8 < data.end_snapshot OR data.end_snapshot IS NULL)
                 ORDER BY s.schema_name, t.table_name, data.path",
             )
             .bind(snapshot_id)
@@ -450,6 +467,8 @@ impl MetadataProvider for PostgresMetadataProvider {
             .bind(snapshot_id)
             .bind(snapshot_id)
             .bind(snapshot_id)
+            .bind(snapshot_id)
+            .bind(snapshot_id)
             .fetch_all(&self.pool)
             .await?;
 

tests/delete_filter_tests.rs

@@ -325,6 +325,58 @@ mod integration_tests {
         Ok(())
     }
 
+    /// Test that a table with all rows deleted returns 0 rows
+    ///
+    /// This is the bug scenario from https://github.com/hotdata-dev/datafusion-ducklake/issues/30
+    /// When all rows are deleted from a table, querying should return 0 rows.
+    #[tokio::test]
+    async fn test_table_with_all_rows_deleted() -> DataFusionResult<()> {
+        let temp_dir = TempDir::new()
+            .map_err(|e| datafusion::error::DataFusionError::External(Box::new(e)))?;
+        let catalog_path = temp_dir.path().join("all_deleted.ducklake");
+
+        // Generate test data - inserts a row, then deletes it
+        common::create_catalog_empty_table(&catalog_path).map_err(common::to_datafusion_error)?;
+
+        let catalog = create_catalog(&catalog_path.to_string_lossy())?;
+
+        let ctx = SessionContext::new();
+        ctx.register_catalog("all_deleted", catalog);
+
+        // Query the table - should return 0 rows since all data was deleted
+        let df = ctx.sql("SELECT * FROM all_deleted.main.tbl").await?;
+        let results = df.collect().await?;
+
+        let total_rows: usize = results.iter().map(|b| b.num_rows()).sum();
+        assert_eq!(
+            total_rows, 0,
+            "Table with all rows deleted should return 0 rows, but got {}",
+            total_rows
+        );
+
+        // Also verify COUNT(*) returns 0
+        let df = ctx
+            .sql("SELECT COUNT(*) as cnt FROM all_deleted.main.tbl")
+            .await?;
+        let results = df.collect().await?;
+
+        assert!(!results.is_empty());
+        let batch = &results[0];
+        let counts = batch
+            .column(0)
+            .as_any()
+            .downcast_ref::<Int64Array>()
+            .unwrap();
+
+        assert_eq!(
+            counts.value(0),
+            0,
+            "COUNT(*) should be 0 after all rows deleted"
+        );
+
+        Ok(())
+    }
+
     /// Test filter pushdown correctness with delete files
     ///
     /// This test verifies that WHERE filters are applied AFTER delete filtering,