Merge pull request JanKaul#228 from JanKaul/fix-equalitly-deletes

use separate manifest files for delete files

Author: JanKaul

.github/workflows/rust.yml

@@ -20,7 +20,7 @@ jobs:
         android: true
         dotnet: true
         haskell: true
-        large-packages: false
+        large-packages: true
         docker-images: false
         swap-storage: false
     - uses: actions/checkout@v3

Cargo.lock

@@ -7,16 +7,16 @@ test-iceberg-rust:
 	cargo test -p iceberg-rust --lib
 
 test-datafusion_iceberg:
-	cargo test -p datafusion_iceberg --tests -j 2 
+	cargo test -p datafusion_iceberg --tests -j 2 && cargo clean -p datafusion_iceberg
 
 test-rest-catalog:
-	cargo test -p iceberg-rest-catalog --lib 
+	cargo test -p iceberg-rest-catalog --lib && cargo clean -p iceberg-rest-catalog
 
 test-file-catalog:
-	cargo test -p iceberg-file-catalog --lib 
+	cargo test -p iceberg-file-catalog --lib && cargo clean -p iceberg-file-catalog
 
 test-sql-catalog:
-	cargo test -p iceberg-sql-catalog --lib 
+	cargo test -p iceberg-sql-catalog --lib && cargo clean -p iceberg-sql-catalog
 clippy:
 	cargo clippy --all-targets --all-features -- -D warnings
 fmt:

Makefile

@@ -29,6 +29,7 @@ url = { workspace = true }
 uuid = { workspace = true }
 
 [dev-dependencies]
+duckdb = { version = "1.3.2", features = ["bundled"] }
 iceberg-rest-catalog = { path = "../catalogs/iceberg-rest-catalog" }
 iceberg-sql-catalog = { path = "../catalogs/iceberg-sql-catalog" }
 reqwest = "0.12"

datafusion_iceberg/Cargo.toml

@@ -1,5 +1,10 @@
-use datafusion::{arrow::error::ArrowError, assert_batches_eq, prelude::SessionContext};
+use datafusion::{
+    arrow::{error::ArrowError, record_batch::RecordBatch},
+    assert_batches_eq,
+    prelude::SessionContext,
+};
 use datafusion_iceberg::catalog::catalog::IcebergCatalog;
+use duckdb::Connection;
 use futures::stream;
 use iceberg_rust::catalog::identifier::Identifier;
 use iceberg_rust::catalog::tabular::Tabular;
@@ -16,11 +21,15 @@ use iceberg_rust::{
     table::Table,
 };
 use iceberg_sql_catalog::SqlCatalog;
+use object_store::local::LocalFileSystem;
 use std::sync::Arc;
+use tempfile::TempDir;
 
 #[tokio::test]
 pub async fn test_equality_delete() {
-    let object_store = ObjectStoreBuilder::memory();
+    let temp_dir = TempDir::new().unwrap();
+    let table_dir = format!("{}/test/orders", temp_dir.path().to_str().unwrap());
+    let object_store = ObjectStoreBuilder::Filesystem(Arc::new(LocalFileSystem::new()));
 
     let catalog: Arc<dyn Catalog> = Arc::new(
         SqlCatalog::new("sqlite://", "warehouse", object_store.clone())
@@ -79,7 +88,7 @@ pub async fn test_equality_delete() {
 
     let table = Table::builder()
         .with_name("orders")
-        .with_location("/test/orders")
+        .with_location(&table_dir)
         .with_schema(schema)
         .with_partition_spec(partition_spec)
         .build(&["test".to_owned()], catalog.clone())
@@ -194,4 +203,28 @@ pub async fn test_equality_delete() {
         "+----+-------------+------------+------------+--------+",
     ];
     assert_batches_eq!(expected, &batches);
+
+    let latest_version = table
+        .object_store()
+        .get(&object_store::path::Path::from(format!(
+            "{table_dir}/metadata/version-hint.text"
+        )))
+        .await
+        .unwrap()
+        .bytes()
+        .await
+        .unwrap();
+    let latest_version_str = std::str::from_utf8(&latest_version).unwrap();
+
+    let conn = Connection::open_in_memory().unwrap();
+    conn.execute("install iceberg", []).unwrap();
+    conn.execute("load iceberg", []).unwrap();
+
+    let duckdb_batches: Vec<RecordBatch> = conn
+        .prepare("select * from iceberg_scan(?) order by id")
+        .unwrap()
+        .query_arrow([latest_version_str])
+        .unwrap()
+        .collect();
+    assert_batches_eq!(expected, &duckdb_batches);
 }

datafusion_iceberg/tests/equality_delete.rs

@@ -94,7 +94,7 @@ pub struct FieldSummary {
     pub upper_bound: Option<Value>,
 }
 
-#[derive(Debug, Serialize_repr, Deserialize_repr, PartialEq, Eq, Clone)]
+#[derive(Debug, Serialize_repr, Deserialize_repr, PartialEq, Eq, Clone, Copy)]
 #[repr(u8)]
 /// Type of content stored by the data file.
 pub enum Content {

iceberg-rust-spec/src/spec/manifest_list.rs

@@ -182,6 +182,7 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> {
         snapshot_id: i64,
         schema: &'schema AvroSchema,
         table_metadata: &'metadata TableMetadata,
+        content: manifest_list::Content,
         branch: Option<&str>,
     ) -> Result<Self, Error> {
         let mut writer = AvroWriter::new(schema, Vec::new());
@@ -229,14 +230,20 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> {
             serde_json::to_string(&spec_id)?,
         )?;
 
-        writer.add_user_metadata("content".to_string(), "data")?;
+        writer.add_user_metadata(
+            "content".to_string(),
+            match content {
+                manifest_list::Content::Data => "data",
+                manifest_list::Content::Deletes => "deletes",
+            },
+        )?;
 
         let manifest = ManifestListEntry {
             format_version: table_metadata.format_version,
             manifest_path: manifest_location.to_owned(),
             manifest_length: 0,
             partition_spec_id: table_metadata.default_spec_id,
-            content: manifest_list::Content::Data,
+            content,
             sequence_number: table_metadata.last_sequence_number + 1,
             min_sequence_number: table_metadata.last_sequence_number + 1,
             added_snapshot_id: snapshot_id,
@@ -331,7 +338,13 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> {
             serde_json::to_string(&spec_id)?,
         )?;
 
-        writer.add_user_metadata("content".to_string(), "data")?;
+        writer.add_user_metadata(
+            "content".to_string(),
+            match manifest.content {
+                manifest_list::Content::Data => "data",
+                manifest_list::Content::Deletes => "deletes",
+            },
+        )?;
 
         writer.extend(
             manifest_reader
@@ -455,7 +468,13 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> {
             serde_json::to_string(&spec_id)?,
         )?;
 
-        writer.add_user_metadata("content".to_string(), "data")?;
+        writer.add_user_metadata(
+            "content".to_string(),
+            match manifest.content {
+                manifest_list::Content::Data => "data",
+                manifest_list::Content::Deletes => "deletes",
+            },
+        )?;
 
         writer.extend(manifest_reader.filter_map(|entry| {
             let mut entry = entry