Merge branch 'JanKaul:main' into main

Author: gruuya

Cargo.lock

@@ -35,10 +35,12 @@ derive_builder = "0.20"
 futures = "0.3.31"
 getrandom = { version = "0.3.1", features = ["std"] }
 itertools = "0.14.0"
+lazy_static = "1.5.0"
 lru = "0.16.0"
 object_store = { version = "0.12", features = ["aws", "gcp"] }
 parquet = { version = "55", features = ["async", "object_store"] }
 pin-project-lite = "0.2"
+regex = "1.11.1"
 serde = "^1.0"
 serde_derive = "^1.0"
 serde_json = "^1.0"

Cargo.toml

@@ -885,10 +885,7 @@ pub mod tests {
             .await
             .unwrap();
 
-        assert_eq!(
-            std::str::from_utf8(&version_hint).unwrap(),
-            "s3://warehouse/tpch/lineitem/metadata/v1.metadata.json"
-        );
+        assert_eq!(std::str::from_utf8(&version_hint).unwrap(), "1");
     }
 
     #[tokio::test]

catalogs/iceberg-file-catalog/src/lib.rs

@@ -781,9 +781,9 @@ pub mod tests {
     use testcontainers_modules::{localstack::LocalStack, postgres::Postgres};
     use tokio::time::sleep;
 
-    use std::{sync::Arc, time::Duration};
-
     use crate::SqlCatalog;
+    use iceberg_rust::object_store::store::version_hint_content;
+    use std::{sync::Arc, time::Duration};
 
     #[tokio::test]
     async fn test_create_update_drop_table() {
@@ -1004,8 +1004,10 @@ pub mod tests {
             .await
             .unwrap();
 
-        assert!(std::str::from_utf8(&version_hint)
-            .unwrap()
-            .ends_with(".metadata.json"));
+        let cache = iceberg_catalog.cache.read().unwrap();
+        let keys = cache.values().collect::<Vec<_>>();
+        let version = version_hint_content(&keys[0].clone().0);
+
+        assert_eq!(std::str::from_utf8(&version_hint).unwrap(), version);
     }
 }

catalogs/iceberg-sql-catalog/src/lib.rs

@@ -21,7 +21,7 @@ iceberg-rust = { path = "../iceberg-rust", version = "0.8.0" }
 itertools = { workspace = true }
 object_store = { workspace = true }
 pin-project-lite = "0.2.16"
-regex = "1.11.1"
+regex = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { version = "1.43", features = ["rt-multi-thread"] }

datafusion_iceberg/Cargo.toml

@@ -204,26 +204,14 @@ 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])
+        .query_arrow([table_dir])
         .unwrap()
         .collect();
     assert_batches_eq!(expected, &duckdb_batches);

datafusion_iceberg/tests/equality_delete.rs

@@ -21,10 +21,12 @@ futures = { workspace = true }
 getrandom = { workspace = true }
 iceberg-rust-spec = { path = "../iceberg-rust-spec", version = "0.8.0" }
 itertools = { workspace = true }
+lazy_static = { workspace = true }
 lru = { workspace = true }
 object_store = { workspace = true }
 parquet = { workspace = true }
 pin-project-lite = { workspace = true }
+regex = { workspace = true }
 serde = { workspace = true }
 serde_derive = { workspace = true }
 serde_json = { workspace = true }
@@ -39,6 +41,7 @@ uuid = { workspace = true }
 
 
 [dev-dependencies]
+rstest = "0.23.0"
 chrono = { workspace = true }
 iceberg-sql-catalog = { path = "../catalogs/iceberg-sql-catalog" }
 

iceberg-rust/Cargo.toml

@@ -9,6 +9,8 @@ use object_store::{Attributes, ObjectStore, PutOptions, TagSet};
 
 use crate::error::Error;
 use flate2::read::GzDecoder;
+use lazy_static::lazy_static;
+use regex::Regex;
 use std::io::Read;
 
 /// Simplify interaction with iceberg files
@@ -59,7 +61,7 @@ impl<T: ObjectStore> IcebergStore for T {
                     "Path for version-hint for {location}"
                 )))?
                 .into(),
-            location.to_string().into(),
+            version_hint_content(location).into(),
             PutOptions {
                 mode: object_store::PutMode::Overwrite,
                 tags: TagSet::default(),
@@ -83,6 +85,39 @@ fn version_hint_path(original: &str) -> Option<String> {
     )
 }
 
+lazy_static! {
+    static ref SUPPORTED_METADATA_FILE_FORMATS: Vec<Regex> = vec![
+        // The standard metastore format https://iceberg.apache.org/spec/#metastore-tables
+        Regex::new(
+            r"^(?<version>[0-9]{5}-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}).(?:gz.)?metadata.json$"
+        )
+        .unwrap(),
+        // The legacy file-system format https://iceberg.apache.org/spec/#file-system-tables
+        Regex::new(r"^v(?<version>[0-9]+).metadata.json$").unwrap(),
+    ];
+}
+
+/// Given a full path to a metadata file, extract an appropriate version hint that other readers
+/// without access to the catalog can parse.
+pub fn version_hint_content(original: &str) -> String {
+    original
+        .split("/")
+        .last()
+        .and_then(|filename| {
+            SUPPORTED_METADATA_FILE_FORMATS
+                .iter()
+                .filter_map(|regex| {
+                    regex.captures(filename).and_then(|capture| {
+                        capture
+                            .name("version")
+                            .and_then(|m| m.as_str().parse().ok())
+                    })
+                })
+                .next()
+        })
+        .unwrap_or(original.to_string())
+}
+
 fn parse_metadata(location: &str, bytes: &[u8]) -> Result<TabularMetadata, Error> {
     if location.ends_with(".gz.metadata.json") {
         let mut decoder = GzDecoder::new(bytes);
@@ -99,6 +134,7 @@ fn parse_metadata(location: &str, bytes: &[u8]) -> Result<TabularMetadata, Error
 #[cfg(test)]
 mod tests {
     use super::*;
+    use rstest::rstest;
     use std::io::Write;
 
     #[test]
@@ -142,6 +178,21 @@ mod tests {
         assert_eq!(version_hint_path(input), Some(expected.to_string()));
     }
 
+    #[rstest]
+    #[case::file_format("/path/to/metadata/v2.metadata.json", "2")]
+    #[case::metastore_format_no_gzip(
+        "/path/to/metadata/00004-3f569e94-5601-48f3-9199-8d71df4ea7b0.metadata.json",
+        "00004-3f569e94-5601-48f3-9199-8d71df4ea7b0"
+    )]
+    #[case::metastore_format_with_gzip(
+        "/path/to/metadata/00004-3f569e94-5601-48f3-9199-8d71df4ea7b0.gz.metadata.json",
+        "00004-3f569e94-5601-48f3-9199-8d71df4ea7b0"
+    )]
+    #[test]
+    fn test_version_hint_content(#[case] input: &str, #[case] expected: &str) {
+        assert_eq!(version_hint_content(input), expected);
+    }
+
     #[test]
     fn test_parse_metadata_table_plain_json() {
         let location = "/path/to/metadata/v1.metadata.json";