diff --git a/Project.toml b/Project.toml index a0d7a06..240c32e 100644 --- a/Project.toml +++ b/Project.toml @@ -1,6 +1,6 @@ name = "RustyIceberg" uuid = "390bdf5b-b624-43dc-a846-0ef7a3405804" -version = "0.9.4" +version = "0.9.5" [deps] Arrow = "69666777-d1a9-59fb-9406-91d4454c9d45" diff --git a/iceberg_rust_ffi/Cargo.lock b/iceberg_rust_ffi/Cargo.lock index 5d04ce0..7d0c377 100644 --- a/iceberg_rust_ffi/Cargo.lock +++ b/iceberg_rust_ffi/Cargo.lock @@ -108,7 +108,7 @@ dependencies = [ "digest 0.10.7", "log", "miniz_oxide", - "num-bigint", + "num-bigint 0.4.8", "quad-rand", "rand 0.9.5", "regex-lite", @@ -131,9 +131,9 @@ checksum = "3d62b7694a562cdf5a74227903507c56ab2cc8bdd1f781ed5cb4cf9c9f810bfc" [[package]] name = "arrow-arith" -version = "58.4.0" +version = "59.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0a41203398f0eaa6f7ec8e62c0da742a21abf282c148fc157f6c35c90e29981a" +checksum = "ce2961626677665b2195eb59242af4c7befe7b8737ca2050295389362380104e" dependencies = [ "arrow-array", "arrow-buffer", @@ -145,9 +145,9 @@ dependencies = [ [[package]] name = "arrow-array" -version = "58.4.0" +version = "59.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ae33dad492b7df00a217563a7b0ef2874df68a0deea1b1a3acf628152f7f7a69" +checksum = "1e5f6adeffdf587d7a31db5d2266189624b526730cd3627f9ff9fedae97ad584" dependencies = [ "ahash", "arrow-buffer", @@ -156,6 +156,7 @@ dependencies = [ "chrono", "half", "hashbrown 0.17.1", + "libc", "num-complex", "num-integer", "num-traits", @@ -163,21 +164,21 @@ dependencies = [ [[package]] name = "arrow-buffer" -version = "58.4.0" +version = "59.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b9552f96391c005e6ab449fa941420935e7e062489b12b8b1b08879b2163f5b5" +checksum = "097d193003ce7995d5d087089069ec2a6e0187faf5a6f8c9f38af2645d987182" dependencies = [ "bytes", "half", - "num-bigint", + "num-bigint 0.5.1", "num-traits", ] [[package]] name = "arrow-cast" -version = "58.4.0" +version = "59.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a8a327c9649f30d8406995f27642b68df354713cca3baaaf100f076f18d5f34" +checksum = "635c9c635668ad26adf76cce8fb276c4be7cf06e63bd516de7da514f9680ee53" dependencies = [ "arrow-array", "arrow-buffer", @@ -186,7 +187,7 @@ dependencies = [ "arrow-schema", "arrow-select", "atoi", - "base64 0.22.1", + "base64 0.23.1", "chrono", "half", "lexical-core", @@ -196,9 +197,9 @@ dependencies = [ [[package]] name = "arrow-data" -version = "58.4.0" +version = "59.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2b24852db04738907e06c04ea61e42fe7fda962a34513022dc0d0e754fb7976b" +checksum = "9ba2f832eaeca24b8f26143dba750e42ee4ab51cf7d65e701ca9607cfda9f358" dependencies = [ "arrow-buffer", "arrow-schema", @@ -209,9 +210,9 @@ dependencies = [ [[package]] name = "arrow-ipc" -version = "58.4.0" +version = "59.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "29a908a11fcfb3fb2f6730f4ac15e367bc644e419155e96238f68cf3adde572b" +checksum = "dcc41681ea80f521df14c36725b74d4c60702c47f0793af2be469c04527e2599" dependencies = [ "arrow-array", "arrow-buffer", @@ -223,9 +224,9 @@ dependencies = [ [[package]] name = "arrow-ord" -version = "58.4.0" +version = "59.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "63a083ec750f5c043f02946b4baf05fcdbb55f4560a3277055caca5cc99f3eb0" +checksum = "2c900759f3bd8354fd4196bc4403eee846894dc2adf66b4225472006a0bf18c5" dependencies = [ "arrow-array", "arrow-buffer", @@ -236,18 +237,18 @@ dependencies = [ [[package]] name = "arrow-schema" -version = "58.4.0" +version = "59.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "21ca356ad6425cecb6eb7b28e4f659f1ee7880fbb1a16127de7dd62901efee9e" +checksum = "10fab8d4563491417ba801fab29d205104d20d4bdf37bda6cd1cf425cff598cd" dependencies = [ "bitflags 2.13.1", ] [[package]] name = "arrow-select" -version = "58.4.0" +version = "59.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c58da39eb3d8350ad4a549e5c2bc49284dac554016c69829310350f1731b0aad" +checksum = "fc58569193c2525915f3cc6310edba3792f1200f65d6e9ed330aa33e691493b8" dependencies = [ "ahash", "arrow-array", @@ -259,9 +260,9 @@ dependencies = [ [[package]] name = "arrow-string" -version = "58.4.0" +version = "59.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6789b388467525e3271326b6b4915666ecfdf5142aef09779445c954b67543c" +checksum = "2e0813f3c35c1cfea65e14c20a953440f7783c088b7ad2d0db162ccdeefcec14" dependencies = [ "arrow-array", "arrow-buffer", @@ -427,7 +428,7 @@ checksum = "4d6867f1565b3aad85681f1015055b087fcfd840d6aeee6eee7f2da317603695" dependencies = [ "autocfg", "libm", - "num-bigint", + "num-bigint 0.4.8", "num-integer", "num-traits", "serde", @@ -1785,8 +1786,8 @@ dependencies = [ [[package]] name = "iceberg" -version = "0.10.0" -source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=9eb8921f941582b0165954dcd2e28ef3a4c3d56a#9eb8921f941582b0165954dcd2e28ef3a4c3d56a" +version = "0.10.1" +source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=ec07ed0f7c08390702d841298e4331976df93b45#ec07ed0f7c08390702d841298e4331976df93b45" dependencies = [ "aes-gcm", "anyhow", @@ -1820,13 +1821,9 @@ dependencies = [ "moka", "murmur3", "once_cell", - "opendal", - "ordered-float 4.6.0", + "ordered-float", "parquet", "rand 0.9.5", - "reqsign", - "reqsign-aws-v4", - "reqsign-core", "reqwest 0.12.28", "roaring", "serde", @@ -1848,8 +1845,8 @@ dependencies = [ [[package]] name = "iceberg-catalog-rest" -version = "0.10.0" -source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=9eb8921f941582b0165954dcd2e28ef3a4c3d56a#9eb8921f941582b0165954dcd2e28ef3a4c3d56a" +version = "0.10.1" +source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=ec07ed0f7c08390702d841298e4331976df93b45#ec07ed0f7c08390702d841298e4331976df93b45" dependencies = [ "async-trait", "chrono", @@ -1868,17 +1865,52 @@ dependencies = [ [[package]] name = "iceberg-property-macro" -version = "0.10.0" -source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=9eb8921f941582b0165954dcd2e28ef3a4c3d56a#9eb8921f941582b0165954dcd2e28ef3a4c3d56a" +version = "0.10.1" +source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=ec07ed0f7c08390702d841298e4331976df93b45#ec07ed0f7c08390702d841298e4331976df93b45" dependencies = [ "proc-macro2", "quote", "syn 2.0.119", ] +[[package]] +name = "iceberg-storage-opendal" +version = "0.10.1" +source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=ec07ed0f7c08390702d841298e4331976df93b45#ec07ed0f7c08390702d841298e4331976df93b45" +dependencies = [ + "anyhow", + "async-trait", + "bytes", + "cfg-if", + "futures", + "iceberg", + "opendal", + "reqsign-aws-v4", + "reqsign-core", + "serde", + "typetag", + "url", +] + +[[package]] +name = "iceberg-storage-refreshable" +version = "0.10.1" +source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=ec07ed0f7c08390702d841298e4331976df93b45#ec07ed0f7c08390702d841298e4331976df93b45" +dependencies = [ + "async-trait", + "bytes", + "futures", + "iceberg", + "iceberg-storage-opendal", + "serde", + "serde_json", + "tokio", + "typetag", +] + [[package]] name = "iceberg_rust_ffi" -version = "0.9.4" +version = "0.9.5" dependencies = [ "anyhow", "arrow-array", @@ -1892,6 +1924,8 @@ dependencies = [ "home", "iceberg", "iceberg-catalog-rest", + "iceberg-storage-opendal", + "iceberg-storage-refreshable", "libc", "object_store_ffi", "once_cell", @@ -2057,12 +2091,6 @@ dependencies = [ "cfg-if", ] -[[package]] -name = "integer-encoding" -version = "3.0.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8bb03732005da905c88227371639bf1ad885cc712789c011c31c5fb3ab3ccf02" - [[package]] name = "inventory" version = "0.3.24" @@ -2371,9 +2399,9 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" [[package]] name = "lz4_flex" -version = "0.13.1" +version = "0.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ef0d4ed8669f8f8826eb00dc878084aa8f253506c4fd5e8f58f5bce72ddb97e" +checksum = "ecbdfe44b1bd960b68170b417450a628c43f7cf56bb3c5317e61cb230ee7f226" dependencies = [ "twox-hash", ] @@ -2446,7 +2474,7 @@ dependencies = [ "indexmap 2.14.0", "metrics", "num_cpus", - "ordered-float 4.6.0", + "ordered-float", "quanta", "radix_trie", "sketches-ddsketch", @@ -2537,6 +2565,16 @@ dependencies = [ "serde", ] +[[package]] +name = "num-bigint" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-bigint-dig" version = "0.8.6" @@ -2930,15 +2968,6 @@ dependencies = [ "vcpkg", ] -[[package]] -name = "ordered-float" -version = "2.10.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68f19d67e5a2795c94e73e0bb1cc1a7edeb2e28efd39e2e1c9b7a40c1108b11c" -dependencies = [ - "num-traits", -] - [[package]] name = "ordered-float" version = "4.6.0" @@ -2989,9 +3018,9 @@ dependencies = [ [[package]] name = "parquet" -version = "58.4.0" +version = "59.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d298093b2dec60289dce0684c986d0f7679e9dd15771c2c65406e1aaf604a704" +checksum = "ff322f54b1a0f9288e614ed1f2d329b380af5476420db19f46ffb865e1163d73" dependencies = [ "ahash", "arrow-array", @@ -3000,7 +3029,7 @@ dependencies = [ "arrow-ipc", "arrow-schema", "arrow-select", - "base64 0.22.1", + "base64 0.23.1", "brotli", "bytes", "chrono", @@ -3009,26 +3038,18 @@ dependencies = [ "half", "hashbrown 0.17.1", "lz4_flex", - "num-bigint", + "num-bigint 0.5.1", "num-integer", "num-traits", - "paste", "ring", "seq-macro", "simdutf8", "snap", - "thrift", "tokio", "twox-hash", "zstd", ] -[[package]] -name = "paste" -version = "1.0.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" - [[package]] name = "pbkdf2" version = "0.12.2" @@ -3491,35 +3512,6 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" -[[package]] -name = "reqsign" -version = "0.16.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43451dbf3590a7590684c25fb8d12ecdcc90ed3ac123433e500447c7d77ed701" -dependencies = [ - "anyhow", - "async-trait", - "base64 0.22.1", - "chrono", - "form_urlencoded", - "getrandom 0.2.17", - "hex", - "hmac 0.12.1", - "home", - "http", - "log", - "percent-encoding", - "quick-xml 0.37.5", - "rand 0.8.8", - "reqwest 0.12.28", - "rust-ini", - "serde", - "serde_json", - "sha1 0.10.7", - "sha2 0.10.9", - "tokio", -] - [[package]] name = "reqsign-aws-core" version = "3.1.1" @@ -3538,7 +3530,7 @@ dependencies = [ "serde", "serde_json", "serde_urlencoded", - "sha1 0.11.0", + "sha1", ] [[package]] @@ -3574,7 +3566,7 @@ dependencies = [ "rsa", "serde", "serde_json", - "sha1 0.11.0", + "sha1", ] [[package]] @@ -3597,7 +3589,7 @@ dependencies = [ "rsa", "serde", "serde_json", - "sha1 0.11.0", + "sha1", "sha2 0.11.0", "windows-sys 0.61.2", ] @@ -4105,17 +4097,6 @@ dependencies = [ "syn 2.0.119", ] -[[package]] -name = "sha1" -version = "0.10.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8" -dependencies = [ - "cfg-if", - "cpufeatures 0.2.17", - "digest 0.10.7", -] - [[package]] name = "sha1" version = "0.11.0" @@ -4457,17 +4438,6 @@ dependencies = [ "cfg-if", ] -[[package]] -name = "thrift" -version = "0.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7e54bc85fc7faa8bc175c4bab5b92ba8d9a3ce893d0e9f42cc455c8ab16a9e09" -dependencies = [ - "byteorder", - "integer-encoding", - "ordered-float 2.10.1", -] - [[package]] name = "time" version = "0.3.55" diff --git a/iceberg_rust_ffi/Cargo.toml b/iceberg_rust_ffi/Cargo.toml index 3d8b6e0..93b18b7 100644 --- a/iceberg_rust_ffi/Cargo.toml +++ b/iceberg_rust_ffi/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "iceberg_rust_ffi" -version = "0.9.4" +version = "0.9.5" edition = "2021" [lib] @@ -12,19 +12,21 @@ default = ["julia"] julia = [] [dependencies] -iceberg = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "9eb8921f941582b0165954dcd2e28ef3a4c3d56a", features = ["storage-azdls"] } -iceberg-catalog-rest = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "9eb8921f941582b0165954dcd2e28ef3a4c3d56a" } +iceberg = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "ec07ed0f7c08390702d841298e4331976df93b45" } +iceberg-catalog-rest = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "ec07ed0f7c08390702d841298e4331976df93b45" } +iceberg-storage-opendal = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "ec07ed0f7c08390702d841298e4331976df93b45", features = ["opendal-azdls"] } +iceberg-storage-refreshable = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "ec07ed0f7c08390702d841298e4331976df93b45" } object_store_ffi = { git = "https://github.com/RelationalAI/object_store_ffi", rev = "bbbbd2f0ab0d609416014cef0d124efa99a515f3", default-features = false } tokio = { version = "1.0", features = ["full"] } futures = "0.3" libc = "0.2" anyhow = "1.0" -arrow-array = { version = "58.1", features = ["ffi"] } -arrow-buffer = "58.1" -arrow-cast = "58.1" -arrow-ipc = "58.1" -arrow-schema = "58.1" -parquet = "58.1" +arrow-array = { version = "59.2", features = ["ffi"] } +arrow-buffer = "59.2" +arrow-cast = "59.2" +arrow-ipc = "59.2" +arrow-schema = "59.2" +parquet = "59.2" tracing-subscriber = "0.3" tracing = "0.1" once_cell = "1.19" diff --git a/iceberg_rust_ffi/src/catalog.rs b/iceberg_rust_ffi/src/catalog.rs index 4ce16fb..6d01a9e 100644 --- a/iceberg_rust_ffi/src/catalog.rs +++ b/iceberg_rust_ffi/src/catalog.rs @@ -8,13 +8,14 @@ use crate::{unexpected, IcebergTable}; use anyhow::Result; use async_trait::async_trait; use iceberg::io::{ - LocalFsStorageFactory, MemoryStorageFactory, OpenDalRoutingStorageFactory, - OpenDalStorageFactory, RefreshableStorageFactory, StorageCredential, StorageCredentialsLoader, + LocalFsStorageFactory, MemoryStorageFactory, StorageCredential, StorageCredentialsLoader, StorageFactory, }; use iceberg::memory::{MemoryCatalogBuilder, MEMORY_CATALOG_WAREHOUSE}; use iceberg::{Catalog, CatalogBuilder, Error, ErrorKind, NamespaceIdent, TableIdent}; use iceberg_catalog_rest::{CustomAuthenticator, RestCatalog, RestCatalogBuilder}; +use iceberg_storage_opendal::{OpenDalResolvingStorageFactory, OpenDalStorageFactory}; +use iceberg_storage_refreshable::RefreshableStorageFactory; use std::collections::HashMap; use std::ffi::{c_char, c_void}; use std::sync::{Arc, Mutex, OnceLock, Weak}; @@ -244,7 +245,7 @@ impl IcebergCatalog { let _ = loader_ref.catalog.set(Arc::downgrade(&catalog_arc)); catalog_arc } else { - let factory: Arc = Arc::new(OpenDalRoutingStorageFactory); + let factory: Arc = Arc::new(OpenDalResolvingStorageFactory::new()); Arc::new( builder .with_storage_factory(factory) @@ -277,17 +278,11 @@ impl IcebergCatalog { ) -> Result { let factory: Arc = if warehouse.starts_with("s3://") || warehouse.starts_with("s3a://") { - // OpenDalRoutingStorageFactory requires PROP_METADATA_LOCATION in the + // OpenDalResolvingStorageFactory requires PROP_METADATA_LOCATION in the // StorageConfig at build() time, but MemoryCatalog builds its FileIO once // at construction time (before any table exists). Use the pre-configured // S3 variant instead, which reads credentials from catalog props directly. - let scheme = if warehouse.starts_with("s3a://") { - "s3a" - } else { - "s3" - }; Arc::new(OpenDalStorageFactory::S3 { - configured_scheme: scheme.to_string(), customized_credential_load: None, }) } else if warehouse == "memory" { diff --git a/iceberg_rust_ffi/src/error_codes.rs b/iceberg_rust_ffi/src/error_codes.rs index d4addc7..84eab8f 100644 --- a/iceberg_rust_ffi/src/error_codes.rs +++ b/iceberg_rust_ffi/src/error_codes.rs @@ -207,6 +207,15 @@ fn classify_data_invalid(detail: &str) -> (IcebergErrorCode, String) { "Snapshot not found".into(), ); } + // LocalFsStorage (and other backends) can report a missing file as + // DataInvalid rather than a dedicated not-found kind; match the same text + // classify_message uses for its own not-found fallback. + if lower.contains("nosuchkey") || lower.contains("no such file") { + return ( + IcebergErrorCode::NOT_FOUND_METADATA, + "Metadata file not found".into(), + ); + } // JSON parse errors contain positional text like "at line N column M" — check // for json parsing first so "column M" doesn't trigger DATA_SCHEMA_MISMATCH. if lower.contains("parse json") diff --git a/iceberg_rust_ffi/src/full_pipeline.rs b/iceberg_rust_ffi/src/full_pipeline.rs index 78da5e2..c885e87 100644 --- a/iceberg_rust_ffi/src/full_pipeline.rs +++ b/iceberg_rust_ffi/src/full_pipeline.rs @@ -48,7 +48,7 @@ pub async fn create_full_scan_pipeline( ) -> IcebergFileScanStream { let files = tasks.map_ok(move |task: FileScanTask| { let filename = task.data_file_path().to_string(); - let record_count = task.record_count.unwrap_or(0) as i64; + let record_count = task.record_count().unwrap_or(0) as i64; let reader = build_reader(file_io.clone(), batch_size); FileToScan { filename, @@ -172,26 +172,18 @@ mod tests { .build() .unwrap(), ); - let task = FileScanTask { - data_file_path: path.to_string(), - file_size_in_bytes: file_size, - start: 0, - length: file_size, - record_count: Some(3), - data_file_format: DataFileFormat::Parquet, - schema: iceberg_schema, - project_field_ids: vec![1], - predicate: None, - deletes: vec![], - partition: None, - partition_spec: None, - name_mapping: None, - unified_partition_type: None, - first_row_id: None, - data_sequence_number: None, - key_metadata: None, - case_sensitive: false, - }; + let task = FileScanTask::builder() + .with_data_file_path(path.to_string()) + .with_file_size_in_bytes(file_size) + .with_start(0) + .with_length(file_size) + .with_record_count(Some(3)) + .with_data_file_format(DataFileFormat::Parquet) + .with_schema(iceberg_schema) + .with_project_field_ids(vec![1]) + .with_case_sensitive(false) + .build() + .unwrap(); // ── 5. Run the full pipeline ─────────────────────────────────────── // Exercises: create_full_scan_pipeline → spawn_file_task → process_file @@ -309,26 +301,20 @@ mod tests { .write(parquet_bytes.clone()) .await .unwrap(); - tasks.push(FileScanTask { - data_file_path: path, - file_size_in_bytes: file_size, - start: 0, - length: file_size, - record_count: Some(3), - data_file_format: DataFileFormat::Parquet, - schema: iceberg_schema.clone(), - project_field_ids: vec![1], - predicate: None, - deletes: vec![], - partition: None, - partition_spec: None, - name_mapping: None, - unified_partition_type: None, - first_row_id: None, - data_sequence_number: None, - key_metadata: None, - case_sensitive: false, - }); + tasks.push( + FileScanTask::builder() + .with_data_file_path(path) + .with_file_size_in_bytes(file_size) + .with_start(0) + .with_length(file_size) + .with_record_count(Some(3)) + .with_data_file_format(DataFileFormat::Parquet) + .with_schema(iceberg_schema.clone()) + .with_project_field_ids(vec![1]) + .with_case_sensitive(false) + .build() + .unwrap(), + ); } // Run the flat pipeline. batch_size=1 forces 3 batches per file so @@ -434,26 +420,20 @@ mod tests { .write(parquet_bytes.clone()) .await .unwrap(); - tasks.push(FileScanTask { - data_file_path: path, - file_size_in_bytes: file_size, - start: 0, - length: file_size, - record_count: Some(3), - data_file_format: DataFileFormat::Parquet, - schema: iceberg_schema.clone(), - project_field_ids: vec![1], - predicate: None, - deletes: vec![], - partition: None, - partition_spec: None, - name_mapping: None, - unified_partition_type: None, - first_row_id: None, - data_sequence_number: None, - key_metadata: None, - case_sensitive: false, - }); + tasks.push( + FileScanTask::builder() + .with_data_file_path(path) + .with_file_size_in_bytes(file_size) + .with_start(0) + .with_length(file_size) + .with_record_count(Some(3)) + .with_data_file_format(DataFileFormat::Parquet) + .with_schema(iceberg_schema.clone()) + .with_project_field_ids(vec![1]) + .with_case_sensitive(false) + .build() + .unwrap(), + ); } let prefetch_depth = 2; diff --git a/iceberg_rust_ffi/src/table.rs b/iceberg_rust_ffi/src/table.rs index 1f0a507..5090058 100644 --- a/iceberg_rust_ffi/src/table.rs +++ b/iceberg_rust_ffi/src/table.rs @@ -4,10 +4,11 @@ use crate::response::IcebergBoxedResponse; /// Table and streaming support for iceberg_rust_ffi use crate::{CResult, Context, RawResponse}; use arrow_array::ffi::{FFI_ArrowArray, FFI_ArrowSchema}; -use iceberg::io::{FileIOBuilder, OpenDalRoutingStorageFactory}; +use iceberg::io::{FileIOBuilder, LocalFsStorageFactory}; use iceberg::table::StaticTable; use iceberg::table::Table; use iceberg::TableIdent; +use iceberg_storage_opendal::OpenDalResolvingStorageFactory; use std::ffi::{c_char, c_void}; use std::ptr; use tokio::sync::Mutex as AsyncMutex; @@ -301,12 +302,24 @@ export_runtime_op!( async { let (full_metadata_path, props) = result_tuple; - // Create file IO using routing factory that infers scheme from metadata location - let factory = std::sync::Arc::new(OpenDalRoutingStorageFactory); - let file_io = FileIOBuilder::new(factory) - .with_props(props) - .with_prop("iceberg.internal.metadata-location", &full_metadata_path) - .build(); + // OpenDalResolvingStorageFactory resolves storage by URL scheme, so it rejects a + // bare local filesystem path (e.g. "/tmp/table/metadata.json") with a URL-parse + // error rather than treating it as a local file, unlike this crate's own + // conventions elsewhere (e.g. LocalFsStorageFactory-backed memory catalogs, which + // accept bare paths). Route schemeless paths to LocalFsStorageFactory directly so + // reading local metadata files keeps working the same way. + let file_io = if full_metadata_path.contains("://") { + let factory = std::sync::Arc::new(OpenDalResolvingStorageFactory::new()); + FileIOBuilder::new(factory) + .with_props(props) + .with_prop("iceberg.internal.metadata-location", &full_metadata_path) + .build() + } else { + FileIOBuilder::new(std::sync::Arc::new(LocalFsStorageFactory)) + .with_props(props) + .with_prop("iceberg.internal.metadata-location", &full_metadata_path) + .build() + }; // Create table identifier let table_ident = TableIdent::from_strs(["default", "table"])?; diff --git a/test/scan_tests.jl b/test/scan_tests.jl index df473e7..1962718 100644 --- a/test/scan_tests.jl +++ b/test/scan_tests.jl @@ -12,7 +12,7 @@ using Tables println(" Snapshot path: $snapshot_path") # Open table - table = RustyIceberg.table_open(snapshot_path) + table = RustyIceberg.table_open(snapshot_path; properties=s3_path_style_properties()) @test table != C_NULL println("✅ Table opened successfully") @@ -88,7 +88,7 @@ using Tables # Select first two columns for testing selected_columns = names(first_df)[1:min(2, length(names(first_df)))] - table2 = RustyIceberg.table_open(snapshot_path) + table2 = RustyIceberg.table_open(snapshot_path; properties=s3_path_style_properties()) scan2 = RustyIceberg.new_scan(table2, RustyIceberg.IcebergPerfConfig(batch_size=8)) RustyIceberg.select_columns!(scan2, selected_columns) stream2 = RustyIceberg.scan!(scan2) @@ -143,7 +143,7 @@ end println("Testing reading nations table...") - table = RustyIceberg.table_open(nations_snapshot_path) + table = RustyIceberg.table_open(nations_snapshot_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=5)) stream = RustyIceberg.scan!(scan) @@ -214,7 +214,7 @@ end @testset "foreach_arrow_batch" begin nations_snapshot_path = "s3://warehouse/tpch.sf01/nation/metadata/00001-44f668fe-3688-49d5-851f-36e75d143321.metadata.json" - table = RustyIceberg.table_open(nations_snapshot_path) + table = RustyIceberg.table_open(nations_snapshot_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=5)) stream = RustyIceberg.scan!(scan) @@ -246,7 +246,7 @@ end test_snapshot_path = "s3://warehouse/incremental/test1/metadata/00003-359e8bb8-1e5d-46d2-bcde-fdaeaa41114f.metadata.json" # Open the table - table = RustyIceberg.table_open(test_snapshot_path) + table = RustyIceberg.table_open(test_snapshot_path; properties=s3_path_style_properties()) @test table != C_NULL println("✅ Table opened successfully") @@ -500,7 +500,7 @@ end to_snapshot_id = Int64(6832180054960511692) @testset "select_columns! - Full Scan" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024)) # Select specific columns @@ -530,7 +530,7 @@ end end @testset "select_columns! - Incremental Scan" begin - table = RustyIceberg.table_open(incremental_path) + table = RustyIceberg.table_open(incremental_path; properties=s3_path_style_properties()) scan = new_incremental_scan(table, from_snapshot_id, to_snapshot_id, RustyIceberg.IcebergPerfConfig()) RustyIceberg.select_columns!(scan, ["n"]) @@ -560,7 +560,7 @@ end end @testset "batch_size via IcebergPerfConfig - Full Scan" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) # Set small batch size via perf config scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=10)) stream = RustyIceberg.scan!(scan) @@ -587,7 +587,7 @@ end end @testset "batch_size via IcebergPerfConfig - Incremental Scan" begin - table = RustyIceberg.table_open(incremental_path) + table = RustyIceberg.table_open(incremental_path; properties=s3_path_style_properties()) scan = new_incremental_scan(table, from_snapshot_id, to_snapshot_id, RustyIceberg.IcebergPerfConfig(batch_size=10)) inserts_stream, deletes_stream = RustyIceberg.scan!(scan) @@ -614,7 +614,7 @@ end end @testset "full scan basic" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024)) stream = RustyIceberg.scan!(scan) @@ -633,7 +633,7 @@ end end @testset "incremental scan basic" begin - table = RustyIceberg.table_open(incremental_path) + table = RustyIceberg.table_open(incremental_path; properties=s3_path_style_properties()) scan = new_incremental_scan(table, from_snapshot_id, to_snapshot_id, RustyIceberg.IcebergPerfConfig()) inserts_stream, deletes_stream = RustyIceberg.scan!(scan) @@ -654,7 +654,7 @@ end end @testset "manifest_entry_concurrency_limit via IcebergPerfConfig - Full Scan" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) # Set concurrency limit via perf config scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024, manifest_entry_concurrency_limit=4)) stream = RustyIceberg.scan!(scan) @@ -674,7 +674,7 @@ end end @testset "manifest_file_concurrency_limit via IcebergPerfConfig - Full Scan" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) # Set concurrency limit via perf config scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024, manifest_file_concurrency_limit=4)) stream = RustyIceberg.scan!(scan) @@ -694,7 +694,7 @@ end end @testset "manifest_entry_concurrency_limit via IcebergPerfConfig - Incremental Scan" begin - table = RustyIceberg.table_open(incremental_path) + table = RustyIceberg.table_open(incremental_path; properties=s3_path_style_properties()) scan = new_incremental_scan(table, from_snapshot_id, to_snapshot_id, RustyIceberg.IcebergPerfConfig(manifest_entry_concurrency_limit=4)) inserts_stream, deletes_stream = RustyIceberg.scan!(scan) @@ -715,7 +715,7 @@ end end @testset "Combined Builder Methods - Full Scan" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=5, manifest_entry_concurrency_limit=2, serialization_concurrency_limit=2)) # Combine multiple builder methods @@ -748,7 +748,7 @@ end end @testset "Combined Builder Methods - Incremental Scan" begin - table = RustyIceberg.table_open(incremental_path) + table = RustyIceberg.table_open(incremental_path; properties=s3_path_style_properties()) scan = new_incremental_scan(table, from_snapshot_id, to_snapshot_id, RustyIceberg.IcebergPerfConfig(batch_size=5, manifest_file_concurrency_limit=2, manifest_entry_concurrency_limit=2, serialization_concurrency_limit=2)) # Combine multiple builder methods @@ -781,7 +781,7 @@ end end @testset "select_columns! with with_file_column! - Full Scan" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024)) # Select specific columns AND include file metadata @@ -824,7 +824,7 @@ end end @testset "select_columns! with with_file_column! - Incremental Scan" begin - table = RustyIceberg.table_open(incremental_path) + table = RustyIceberg.table_open(incremental_path; properties=s3_path_style_properties()) scan = new_incremental_scan(table, from_snapshot_id, to_snapshot_id, RustyIceberg.IcebergPerfConfig()) # Select specific column AND include file metadata for incremental scan @@ -865,7 +865,7 @@ end end @testset "select_columns! with FILE_COLUMN constant" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024)) # Select columns including FILE_COLUMN constant @@ -906,7 +906,7 @@ end end @testset "select_columns! with with_pos_column! - Full Scan" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024)) # Select specific columns AND include pos metadata @@ -952,7 +952,7 @@ end end @testset "select_columns! with with_pos_column! - Incremental Scan" begin - table = RustyIceberg.table_open(incremental_path) + table = RustyIceberg.table_open(incremental_path; properties=s3_path_style_properties()) scan = new_incremental_scan(table, from_snapshot_id, to_snapshot_id, RustyIceberg.IcebergPerfConfig()) # Select specific column AND include pos metadata for incremental scan @@ -1042,7 +1042,7 @@ end end @testset "select_columns! with POS_COLUMN constant" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024)) # Select columns including POS_COLUMN constant @@ -1086,7 +1086,7 @@ end end @testset "with_file_column! and with_pos_column! combined" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024)) # Select columns and include both file and pos metadata @@ -1135,7 +1135,7 @@ end end @testset "with_snapshot_id! - Full Scan via Builder Method" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) # Use the correct snapshot ID for the customer table customer_snapshot_id = Int64(3441867730092225551) @@ -1192,7 +1192,7 @@ end end @testset "file_prefetch_depth via IcebergPerfConfig" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024, file_prefetch_depth=4)) stream = RustyIceberg.scan!(scan) @@ -1218,7 +1218,7 @@ end end @testset "file_prefetch_depth via IcebergPerfConfig - Incremental Scan" begin - table = RustyIceberg.table_open(incremental_path) + table = RustyIceberg.table_open(incremental_path; properties=s3_path_style_properties()) scan = new_incremental_scan(table, from_snapshot_id, to_snapshot_id, RustyIceberg.IcebergPerfConfig(batch_size=1024, file_prefetch_depth=4)) append_stream, delete_stream = RustyIceberg.scan_incremental_nested!(scan) @@ -1251,7 +1251,7 @@ end end @testset "Combined with_snapshot_id! and other builder methods" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=5)) # Use the correct snapshot ID for the customer table @@ -1307,7 +1307,7 @@ end @testset "nested_arrow_stream called directly after build!" begin # scan_nested! = build! + nested_arrow_stream; exercise the two-step path. - table = RustyIceberg.table_open(nations_path) + table = RustyIceberg.table_open(nations_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024)) RustyIceberg.build!(scan) outer = RustyIceberg.nested_arrow_stream(scan) @@ -1341,7 +1341,7 @@ end end @testset "Basic iteration — filenames and record counts" begin - table = RustyIceberg.table_open(nations_path) + table = RustyIceberg.table_open(nations_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024)) outer = RustyIceberg.scan_nested!(scan) @test outer != C_NULL @@ -1385,7 +1385,7 @@ end @testset "Row counts match flat scan" begin # Collect total rows via nested pipeline. - table_n = RustyIceberg.table_open(customer_path) + table_n = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan_n = RustyIceberg.new_scan(table_n, RustyIceberg.IcebergPerfConfig(batch_size=1024)) outer = RustyIceberg.scan_nested!(scan_n) @@ -1413,7 +1413,7 @@ end end # Collect total rows via flat pipeline. - table_f = RustyIceberg.table_open(customer_path) + table_f = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan_f = RustyIceberg.new_scan(table_f, RustyIceberg.IcebergPerfConfig(batch_size=1024)) stream_f = RustyIceberg.scan!(scan_f) @@ -1440,7 +1440,7 @@ end end @testset "Correct data — nations table" begin - table = RustyIceberg.table_open(nations_path) + table = RustyIceberg.table_open(nations_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024)) outer = RustyIceberg.scan_nested!(scan) @@ -1479,7 +1479,7 @@ end end @testset "Builder methods — select_columns! and batch_size via IcebergPerfConfig" begin - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=10)) RustyIceberg.select_columns!(scan, ["c_custkey", "c_name"]) outer = RustyIceberg.scan_nested!(scan) @@ -1512,7 +1512,7 @@ end @testset "Safe early drop of FileScan" begin # Drop a FileScan after reading only the first batch — must not crash or hang. - table = RustyIceberg.table_open(customer_path) + table = RustyIceberg.table_open(customer_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1)) outer = RustyIceberg.scan_nested!(scan) @@ -1539,7 +1539,7 @@ end @testset "print_pipeline_stats and reset_pipeline_stats" begin # Run a full scan so the pipeline stats are populated. - table = RustyIceberg.table_open(nations_path) + table = RustyIceberg.table_open(nations_path; properties=s3_path_style_properties()) scan = RustyIceberg.new_scan(table, RustyIceberg.IcebergPerfConfig(batch_size=1024)) stream = RustyIceberg.scan!(scan) try diff --git a/test/test_config.jl b/test/test_config.jl index 7a6cf39..7d79aeb 100644 --- a/test/test_config.jl +++ b/test/test_config.jl @@ -60,6 +60,18 @@ function get_s3_config() ) end +""" + s3_path_style_properties() -> Dict{String, String} + +Minimal properties for `table_open` calls that authenticate purely through +AWS_* environment variables and only need S3 addressing style fixed for this +repo's local MinIO. See the comment on `get_catalog_properties`'s +`"s3.path-style-access"` entry for why this is needed. +""" +function s3_path_style_properties() + return Dict("s3.path-style-access" => "true") +end + """ get_warehouse_name() -> String @@ -87,6 +99,14 @@ function get_catalog_properties() "s3.endpoint" => s3["endpoint"], "s3.access-key-id" => s3["access_key_id"], "s3.secret-access-key" => s3["secret_access_key"], + # Iceberg's own spec default is virtual-host-style (matching Java's + # S3FileIOProperties.PATH_STYLE_ACCESS_DEFAULT = false), which this + # repo's local MinIO can't satisfy from outside the docker network: + # MinIO's virtual-host mode is configured for the `minio` domain + # (MINIO_DOMAIN=minio in docker-compose.yml), not `localhost`, so a + # virtual-hosted request for `warehouse.localhost:9000` 404s with + # NoSuchBucket. Force path-style for this test environment. + "s3.path-style-access" => "true", "s3.region" => s3["region"] ) end