diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index cd8ac61bf4d..fa6fae91792 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -165,6 +165,25 @@ For a custom S3-compatible endpoint, configure the catalog with the endpoint, pa These `s3.*` storage properties are not specific to the Hive catalog shown here. When `s3.access-key-id` / `s3.secret-access-key` are omitted, credentials come from the standard AWS chain (environment variables, instance profiles, and so on). `client.region` is auto-detected for AWS but should be set for non-AWS endpoints. If your REST catalog vends temporary credentials, the native reader does not consume them automatically, and wiring that requires the credential provider bridge. See Iceberg's [S3 FileIO](https://iceberg.apache.org/docs/latest/aws/#s3-fileio) docs for the full property list, and [S3 Credential Providers](s3-credential-providers.md) for vended or per-request credentials. +### Object store configuration (HDFS) + +`hdfs://` tables are read and written through iceberg-rust's `hdfs-native` backend, a pure-Rust HDFS RPC client. This is **not** the libhdfs/JNI client that the plain-Parquet native scan uses for `spark.hadoop.fs.comet.libhdfs.schemes`: the two clients live in the same process but connect independently, so an Iceberg table and a plain Parquet file on the same cluster each open their own connections. The Rust client still reads `core-site.xml` / `hdfs-site.xml` from `$HADOOP_CONF_DIR` (or `$HADOOP_HOME`), and Kerberos works through the system `libgssapi_krb5` and the ambient credential cache — it does not reuse the JVM's Kerberos subject. + +The NameNode endpoints are the one thing the Rust client cannot infer from the Hadoop XML. The underlying OpenDAL builder connects to the endpoints given in the `hdfs.name-node` property (comma-separated for HA failover) and falls back to the authority written in the table location when that property is absent. A single-NameNode cluster therefore needs no configuration, because `hdfs://nn.example.com:8020/...` is already a routable address. An HA cluster does: its locations read `hdfs:///...`, and a nameservice is not a host. + +Comet resolves this automatically from the session Hadoop configuration — it reads `dfs.ha.namenodes.` and each `dfs.namenode.rpc-address..` and hands iceberg-rust the same failover list the JVM client would use. Nothing needs to be set as long as the standard HDFS client configuration is on the classpath. To override it (or to supply endpoints Spark's configuration does not carry), set the property on the catalog: + +```shell + --conf spark.sql.catalog.hdfs_cat=org.apache.iceberg.spark.SparkCatalog \ + --conf spark.sql.catalog.hdfs_cat.type=hadoop \ + --conf spark.sql.catalog.hdfs_cat.warehouse=hdfs://nameservice1/warehouse \ + --conf spark.sql.catalog.hdfs_cat.hdfs.name-node=hdfs://nn1.example.com:8020,hdfs://nn2.example.com:8020 +``` + +An explicit catalog property always wins over the values derived from the Hadoop configuration. Individual HDFS client settings can also be forwarded with `hadoop.`-prefixed catalog properties (for example `spark.sql.catalog.hdfs_cat.hadoop.dfs.client.failover.random.order=true`), which override the values loaded from `$HADOOP_CONF_DIR`. + +A location with no authority at all (`hdfs:///warehouse/...`) falls back to the JVM reader: the scheme gate runs before the catalog properties are assembled, so Comet declines rather than assume a NameNode. + ### Current limitations The following scenarios will fall back to the JVM Iceberg reader: diff --git a/native/Cargo.lock b/native/Cargo.lock index fabb56e2ec8..d9d70830ccf 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -34,10 +34,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" dependencies = [ "cfg-if", - "cipher", + "cipher 0.4.4", "cpufeatures 0.2.17", ] +[[package]] +name = "aes" +version = "0.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35f0f96ce78e38c3dc6d8948aa8163d06385be74000f3c7a95bf1eef35d3ea32" +dependencies = [ + "cipher 0.5.2", + "cpubits", + "cpufeatures 0.3.1", +] + [[package]] name = "aes-gcm" version = "0.10.3" @@ -45,9 +56,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1" dependencies = [ "aead", - "aes", - "cipher", - "ctr", + "aes 0.8.4", + "cipher 0.4.4", + "ctr 0.9.2", "ghash", "subtle", ] @@ -1165,6 +1176,15 @@ dependencies = [ "generic-array", ] +[[package]] +name = "block-padding" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "710f1dd022ef4e93f8a438b4ba958de7f64308434fa6a87104481645cc30068b" +dependencies = [ + "hybrid-array", +] + [[package]] name = "blocking" version = "1.7.0" @@ -1296,7 +1316,16 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "26b52a9543ae338f279b96b0b9fed9c8093744685043739079ce85cd58f289a6" dependencies = [ - "cipher", + "cipher 0.4.4", +] + +[[package]] +name = "cbc" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce2dc9ee5f88d11e0beb842c88b33c8a5cf0d1329c4b19494af42b07dbfe8896" +dependencies = [ + "cipher 0.5.2", ] [[package]] @@ -1398,7 +1427,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ "crypto-common 0.1.7", - "inout", + "inout 0.1.4", +] + +[[package]] +name = "cipher" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8cf2a2c93cd704877c0858356ed03480ff301ee950b43f1cbe4573b088bfa6c" +dependencies = [ + "block-buffer 0.12.1", + "crypto-common 0.2.2", + "inout 0.2.2", ] [[package]] @@ -1591,6 +1631,12 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "cpubits" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15b85f9c39137c3a891689859392b1bd49812121d0d61c9caf00d46ed5ce06ae" + [[package]] name = "cpufeatures" version = "0.2.17" @@ -1609,6 +1655,21 @@ dependencies = [ "libc", ] +[[package]] +name = "crc" +version = "3.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d" +dependencies = [ + "crc-catalog", +] + +[[package]] +name = "crc-catalog" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" + [[package]] name = "crc-fast" version = "1.10.0" @@ -1769,7 +1830,16 @@ version = "0.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835" dependencies = [ - "cipher", + "cipher 0.4.4", +] + +[[package]] +name = "ctr" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baaca1c4b237092596f64d571e9db6ce4109c4ef9742e27590f1709594461f21" +dependencies = [ + "cipher 0.5.2", ] [[package]] @@ -2820,6 +2890,15 @@ dependencies = [ "unicode-xid", ] +[[package]] +name = "des" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "916a94e407b54f9034d71dd748234cd1e516ced6284009906ae246f177eafe5a" +dependencies = [ + "cipher 0.5.2", +] + [[package]] name = "destructure_traitobject" version = "0.2.0" @@ -2876,6 +2955,18 @@ dependencies = [ "const-random", ] +[[package]] +name = "dns-lookup" +version = "3.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e39034cee21a2f5bbb66ba0e3689819c4bb5d00382a282006e802a7ffa6c41d" +dependencies = [ + "cfg-if", + "libc", + "socket2", + "windows-sys 0.60.2", +] + [[package]] name = "dunce" version = "1.0.5" @@ -3174,6 +3265,34 @@ dependencies = [ "slab", ] +[[package]] +name = "g2gen" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c5a7e0eb46f83a20260b850117d204366674e85d3a908d90865c78df9a6b1dfc" +dependencies = [ + "g2poly", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "g2p" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "539e2644c030d3bf4cd208cb842d2ce2f80e82e6e8472390bcef83ceba0d80ad" +dependencies = [ + "g2gen", + "g2poly", +] + +[[package]] +name = "g2poly" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "312d2295c7302019c395cfb90dacd00a82a2eabd700429bba9c7a3f38dbbe11b" + [[package]] name = "generic-array" version = "0.14.7" @@ -3193,7 +3312,7 @@ dependencies = [ "cfg-if", "js-sys", "libc", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", "wasm-bindgen", ] @@ -3320,6 +3439,47 @@ dependencies = [ "foldhash 0.2.0", ] +[[package]] +name = "hdfs-native" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd181084003308224efddf737417832186839ce2882d2468ddad0114cfb3551" +dependencies = [ + "aes 0.9.3", + "base64 0.22.1", + "bitflags 2.13.2", + "bumpalo", + "bytes", + "cbc 0.2.1", + "chrono", + "cipher 0.5.2", + "crc", + "ctr 0.10.1", + "des", + "dns-lookup", + "futures", + "g2p", + "hex", + "hmac 0.13.0", + "libc", + "libloading 0.9.0", + "log", + "md-5 0.11.0", + "num-traits", + "once_cell", + "prost", + "prost-types", + "rand 0.10.2", + "regex", + "roxmltree", + "socket2", + "thiserror 2.0.20", + "tokio", + "url", + "uuid", + "whoami", +] + [[package]] name = "hdfs-sys" version = "0.3.0" @@ -3543,7 +3703,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=665c64e48e8d33797ecb1a421f327edd9b024879#665c64e48e8d33797ecb1a421f327edd9b024879" +source = "git+https://github.com/mixermt/iceberg-rust?rev=9d7d2d89e8391245863ebbcbad753b99a0b3b4cc#9d7d2d89e8391245863ebbcbad753b99a0b3b4cc" dependencies = [ "aes-gcm", "anyhow", @@ -3602,7 +3762,7 @@ dependencies = [ [[package]] name = "iceberg-property-macro" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=665c64e48e8d33797ecb1a421f327edd9b024879#665c64e48e8d33797ecb1a421f327edd9b024879" +source = "git+https://github.com/mixermt/iceberg-rust?rev=9d7d2d89e8391245863ebbcbad753b99a0b3b4cc#9d7d2d89e8391245863ebbcbad753b99a0b3b4cc" dependencies = [ "proc-macro2", "quote", @@ -3612,7 +3772,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=665c64e48e8d33797ecb1a421f327edd9b024879#665c64e48e8d33797ecb1a421f327edd9b024879" +source = "git+https://github.com/mixermt/iceberg-rust?rev=9d7d2d89e8391245863ebbcbad753b99a0b3b4cc#9d7d2d89e8391245863ebbcbad753b99a0b3b4cc" dependencies = [ "anyhow", "async-trait", @@ -3785,10 +3945,20 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" dependencies = [ - "block-padding", + "block-padding 0.3.3", "generic-array", ] +[[package]] +name = "inout" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4250ce6452e92010fdf7268ccc5d14faa80bb12fc741938534c58f16804e03c7" +dependencies = [ + "block-padding 0.4.2", + "hybrid-array", +] + [[package]] name = "inventory" version = "0.3.24" @@ -3945,7 +4115,7 @@ dependencies = [ "java-locator", "jni-macros", "jni-sys 0.4.1", - "libloading", + "libloading 0.8.9", "log", "simd_cesu8", "thiserror 2.0.20", @@ -4112,6 +4282,16 @@ dependencies = [ "windows-link", ] +[[package]] +name = "libloading" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "754ca22de805bb5744484a5b151a9e1a8e837d5dc232c2d7d8c2e3492edc8b60" +dependencies = [ + "cfg-if", + "windows-link", +] + [[package]] name = "liblzma" version = "0.4.8" @@ -4147,6 +4327,15 @@ dependencies = [ "cc", ] +[[package]] +name = "libredox" +version = "0.1.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6480ccc157a1389bb2e4891b24751b0f798ba640d22386f23143fbcc89da195a" +dependencies = [ + "libc", +] + [[package]] name = "link-section" version = "0.19.3" @@ -4320,7 +4509,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8" dependencies = [ "libc", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", "windows-sys 0.61.2", ] @@ -4489,6 +4678,24 @@ dependencies = [ "libm", ] +[[package]] +name = "objc2-core-foundation" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" +dependencies = [ + "bitflags 2.13.2", +] + +[[package]] +name = "objc2-system-configuration" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7216bd11cbda54ccabcab84d523dc93b858ec75ecfb3a7d89513fa22464da396" +dependencies = [ + "objc2-core-foundation", +] + [[package]] name = "object" version = "0.37.3" @@ -4596,6 +4803,7 @@ dependencies = [ "opendal-service-fs", "opendal-service-gcs", "opendal-service-hdfs", + "opendal-service-hdfs-native", "opendal-service-oss", "opendal-service-s3", ] @@ -4764,6 +4972,20 @@ dependencies = [ "tokio", ] +[[package]] +name = "opendal-service-hdfs-native" +version = "0.58.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8aebbf956ea9e64d70fc8d1f25381604c02528805c23bd55443b8e924b438222" +dependencies = [ + "bytes", + "futures", + "hdfs-native", + "log", + "opendal-core", + "serde", +] + [[package]] name = "opendal-service-oss" version = "0.58.2" @@ -5087,8 +5309,8 @@ version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e847e2c91a18bfa887dd028ec33f2fe6f25db77db3619024764914affe8b69a6" dependencies = [ - "aes", - "cbc", + "aes 0.8.4", + "cbc 0.1.2", "der", "pbkdf2", "scrypt", @@ -5856,6 +6078,15 @@ dependencies = [ "byteorder", ] +[[package]] +name = "roxmltree" +version = "0.21.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1964b10c76125c36f8afe190065a4bf9a87bf324842c05701330bba9f1cacbb" +dependencies = [ + "memchr", +] + [[package]] name = "rsa" version = "0.9.10" @@ -6015,7 +6246,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213" dependencies = [ - "cipher", + "cipher 0.4.4", ] [[package]] @@ -7069,6 +7300,15 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "wasi" +version = "0.14.7+wasi-0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "883478de20367e224c0090af9cf5f9fa85bed63a95c1abf3afc5c083ebc06e8c" +dependencies = [ + "wasip2", +] + [[package]] name = "wasip2" version = "1.0.4+wasi-0.2.12" @@ -7078,6 +7318,15 @@ dependencies = [ "wit-bindgen", ] +[[package]] +name = "wasite" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66fe902b4a6b8028a753d5424909b764ccf79b7a209eac9bf97e59cda9f71a42" +dependencies = [ + "wasi 0.14.7+wasi-0.2.4", +] + [[package]] name = "wasm-bindgen" version = "0.2.128" @@ -7188,6 +7437,19 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "whoami" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "626c4bac6755d76ffc12cb01b2eac751db1996b9e0041de9aa02c8c211ddc82c" +dependencies = [ + "libc", + "libredox", + "objc2-system-configuration", + "wasite", + "web-sys", +] + [[package]] name = "winapi" version = "0.3.9" @@ -7296,6 +7558,15 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "windows-sys" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb" +dependencies = [ + "windows-targets 0.53.5", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -7329,13 +7600,30 @@ dependencies = [ "windows_aarch64_gnullvm 0.52.6", "windows_aarch64_msvc 0.52.6", "windows_i686_gnu 0.52.6", - "windows_i686_gnullvm", + "windows_i686_gnullvm 0.52.6", "windows_i686_msvc 0.52.6", "windows_x86_64_gnu 0.52.6", "windows_x86_64_gnullvm 0.52.6", "windows_x86_64_msvc 0.52.6", ] +[[package]] +name = "windows-targets" +version = "0.53.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3" +dependencies = [ + "windows-link", + "windows_aarch64_gnullvm 0.53.1", + "windows_aarch64_msvc 0.53.1", + "windows_i686_gnu 0.53.1", + "windows_i686_gnullvm 0.53.1", + "windows_i686_msvc 0.53.1", + "windows_x86_64_gnu 0.53.1", + "windows_x86_64_gnullvm 0.53.1", + "windows_x86_64_msvc 0.53.1", +] + [[package]] name = "windows_aarch64_gnullvm" version = "0.42.2" @@ -7348,6 +7636,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" + [[package]] name = "windows_aarch64_msvc" version = "0.42.2" @@ -7360,6 +7654,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" +[[package]] +name = "windows_aarch64_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" + [[package]] name = "windows_i686_gnu" version = "0.42.2" @@ -7372,12 +7672,24 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" +[[package]] +name = "windows_i686_gnu" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3" + [[package]] name = "windows_i686_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" +[[package]] +name = "windows_i686_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" + [[package]] name = "windows_i686_msvc" version = "0.42.2" @@ -7390,6 +7702,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" +[[package]] +name = "windows_i686_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" + [[package]] name = "windows_x86_64_gnu" version = "0.42.2" @@ -7402,6 +7720,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" +[[package]] +name = "windows_x86_64_gnu" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" + [[package]] name = "windows_x86_64_gnullvm" version = "0.42.2" @@ -7414,6 +7738,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" + [[package]] name = "windows_x86_64_msvc" version = "0.42.2" @@ -7426,6 +7756,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "windows_x86_64_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" + [[package]] name = "wit-bindgen" version = "0.57.1" diff --git a/native/Cargo.toml b/native/Cargo.toml index bde2ba432c0..6277dbd6994 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -61,8 +61,11 @@ object_store = { version = "0.13.2", features = ["gcp", "azure", "aws", "http"] url = "2.2" aws-config = "1.8.18" aws-credential-types = "1.2.13" -iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "665c64e48e8d33797ecb1a421f327edd9b024879" } -iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "665c64e48e8d33797ecb1a421f327edd9b024879", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } +# apache/iceberg-rust#3111 (HDFS via opendal services-hdfs-native), pinned to the PR head on the +# author's fork. This also advances iceberg-rust 29 commits past the previous pin (665c64e). +# Retarget at apache/iceberg-rust once #3111 merges. +iceberg = { git = "https://github.com/mixermt/iceberg-rust", rev = "9d7d2d89e8391245863ebbcbad753b99a0b3b4cc" } +iceberg-storage-opendal = { git = "https://github.com/mixermt/iceberg-rust", rev = "9d7d2d89e8391245863ebbcbad753b99a0b3b4cc", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls", "opendal-hdfs-native"] } reqsign-core = "3" [profile.release] @@ -81,3 +84,4 @@ codegen-units = 16 # Parallel codegen (faster compile, slightly larger binary) debug-assertions = true panic = "unwind" # Allow panics to be caught and logged across FFI boundary # overflow-checks inherited as false from release + diff --git a/native/core/src/execution/operators/iceberg_common.rs b/native/core/src/execution/operators/iceberg_common.rs index 509e3b1f986..7c30a38f378 100644 --- a/native/core/src/execution/operators/iceberg_common.rs +++ b/native/core/src/execution/operators/iceberg_common.rs @@ -36,7 +36,11 @@ const ICEBERG_PROVIDER_CLASS_PROPERTY: &str = "s3.comet.credential.provider.clas /// Key prefixes forwarded to iceberg-rust's `FileIO`. The full unfiltered catalog bag (catalog /// URI, OAuth tokens, credentials.uri, tenant-id, etc.) is kept upstream so /// `CometS3CredentialBridge` can read whatever the vendor needs. -const STORAGE_PROPERTY_PREFIXES: &[&str] = &["s3.", "gcs.", "adls.", "client."]; +/// +/// `hdfs.` carries the NameNode list and `hadoop.` the HDFS client overrides; dropping them would +/// leave an HA table with only its nameservice authority, which is not a routable host (see +/// `CometIcebergNativeScan.hadoopToIcebergHdfsProperties`). +const STORAGE_PROPERTY_PREFIXES: &[&str] = &["s3.", "gcs.", "adls.", "client.", "hdfs.", "hadoop."]; /// Pick an OpenDAL storage backend from a URI's scheme. `file` (or no scheme) falls through to /// the local file system. `memory` is used by the write path to assemble manifest bytes that @@ -59,6 +63,10 @@ pub(crate) fn storage_factory_for( "file" => Ok(Arc::new(OpenDalStorageFactory::Fs)), "memory" => Ok(Arc::new(OpenDalStorageFactory::Memory)), "gs" => Ok(Arc::new(OpenDalStorageFactory::Gcs)), + // iceberg-rust's pure-Rust `hdfs-native` backend -- NOT the libhdfs/JNI client the + // plain-Parquet path uses (`fs.comet.libhdfs.schemes`). Both link into the same + // `libcomet`, but they are separate clients with separate connections and Kerberos state. + "hdfs" => Ok(Arc::new(OpenDalStorageFactory::HdfsNative)), // Reads keep the OSS backend they have always had (CometScanRule admits `oss` scan // locations through HadoopFileIO). Writes fail closed: Comet does not forward `oss.*` // properties into the FileIO and no test covers the write path, so OSS-specific @@ -269,13 +277,52 @@ mod tests { #[test] fn unknown_scheme_is_rejected() { - let err = factory_result("hdfs://nn/db/table", AccessMode::Read).unwrap_err(); + // object_store recognizes abfss, but iceberg-rust's OpenDAL storage factory has no arm + // for it, so the JVM gate must decline rather than fail here at execution time. + let err = factory_result("abfss://c@acct/db/table", AccessMode::Read).unwrap_err(); assert!( err.contains("Unsupported storage scheme"), "unexpected error: {err}" ); } + #[test] + fn hdfs_scheme_resolves_for_both_modes() { + // Unlike `oss`, writes are admitted: `hdfs.`/`hadoop.` properties are forwarded, so + // nothing is silently dropped. + for mode in [AccessMode::Read, AccessMode::Write] { + assert!(factory_result("hdfs://nn:8020/warehouse/db/t", mode).is_ok()); + assert!(factory_result("hdfs://nameservice1/warehouse/db/t", mode).is_ok()); + } + } + + #[test] + fn hdfs_properties_reach_the_file_io() { + // If the prefix filter drops these, an HA table connects to its nameservice as if it were + // a host and fails only once a task opens a file. + let props = HashMap::from([ + ( + "hdfs.name-node".to_string(), + "hdfs://nn1:8020,hdfs://nn2:8020".to_string(), + ), + ( + "hadoop.dfs.client.failover.random.order".to_string(), + "true".to_string(), + ), + // Must not survive the narrowing: the unfiltered bag also carries catalog identity + // and OAuth material that iceberg-rust's FileIO has no business seeing. + ("uri".to_string(), "thrift://metastore:9083".to_string()), + ]); + + let forwarded: Vec<&String> = props + .keys() + .filter(|k| STORAGE_PROPERTY_PREFIXES.iter().any(|p| k.starts_with(p))) + .collect(); + + assert_eq!(forwarded.len(), 2, "forwarded: {forwarded:?}"); + assert!(load_file_io(&props, "hdfs://nameservice1/db/t", "cat", AccessMode::Read).is_ok()); + } + #[test] fn scheme_of_extracts_scheme_from_all_uri_forms() { // Host-bearing and hostless/opaque vendor forms must resolve to the same scheme, so an diff --git a/pom.xml b/pom.xml index 6766979652d..d667d206f29 100644 --- a/pom.xml +++ b/pom.xml @@ -85,7 +85,7 @@ under the License. 3.25.5 1.16.0 provided - 3.3.4 + 3.4.2 18.3.0 1.9.13 2.43.0 @@ -667,6 +667,7 @@ under the License. 2.12.17 2.12 3.4.3 + 3.3.4 3.4 1.13.1 4.8.8 @@ -686,6 +687,7 @@ under the License. 2.12.18 2.12 3.5.9 + 3.3.4 3.5 1.13.1 4.8.8 @@ -705,6 +707,7 @@ under the License. 2.13.16 2.13 4.0.4 + 3.4.1 4.0 1.15.2 4.13.6 @@ -728,6 +731,7 @@ under the License. 2.13.17 2.13 4.1.3 + 3.4.2 4.1 1.16.0 4.13.6 @@ -748,6 +752,7 @@ under the License. 2.13.18 2.13 4.2.0 + 3.5.0 4.2 1.17.0 4.13.6 diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala index f7e0fe5cb98..534d4b3ccfb 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -516,6 +516,21 @@ case class CometScanRule(session: SparkSession) } val icebergDataBucket: Option[String] = taskValidation.dataFileBuckets.headOption + // The HDFS analogue of the multi-bucket check above: one `hdfs.name-node` per scan, and + // it wins over every path authority, so a second nameservice would be read from the + // first one's NameNode at the same relative path. + if (taskValidation.dataFileHdfsAuthorities.size > 1) { + fallbackReasons += + "Iceberg scan reads data/delete files across multiple HDFS authorities " + + s"(${taskValidation.dataFileHdfsAuthorities.toSeq.sorted.mkString(", ")}); " + + "Comet's native reader resolves a single NameNode per scan" + return withFallbackReasons(scanExec, fallbackReasons.toSet) + } + // The DATA authority, which Iceberg allows to differ from the metadata location + // (`write.data.path`); None for an empty scan or a non-HDFS table. + val icebergDataHdfsAuthority: Option[String] = + taskValidation.dataFileHdfsAuthorities.headOption + // Extract all Iceberg metadata once using reflection. // If any required reflection fails, this returns None, and we fall back to Spark. // First get metadataLocation and catalogProperties which are needed by the factory. @@ -584,7 +599,16 @@ case class CometScanRule(session: SparkSession) // iceberg-rust's FileIO, so the alias key never reaches FileIO -- but it is still // handed, unfiltered, to CometS3CredentialBridge, so a custom credential provider sees // it. That is intended: the provider gets the full property bag. - val catalogProperties = hadoopDerivedProperties ++ fileIOProperties ++ + // Resolved against the DATA authority when the tasks yielded one, else the metadata + // location. Before `fileIOProperties` so an explicit catalog `hdfs.name-node` wins. + val hdfsAuthorityUri = icebergDataHdfsAuthority + .map(authority => new java.net.URI(s"hdfs://$authority/")) + .getOrElse(effectiveUri) + val hadoopDerivedHdfsProperties = + CometIcebergNativeScan.hadoopToIcebergHdfsProperties(hdfsAuthorityUri, hadoopConf) + + val catalogProperties = hadoopDerivedProperties ++ hadoopDerivedHdfsProperties ++ + fileIOProperties ++ hadoopS3Options .get(COMET_S3_COMPLIANT_SCHEMES_KEY) .map(COMET_S3_COMPLIANT_SCHEMES_KEY -> _) @@ -1222,15 +1246,18 @@ object CometScanRule extends Logging { * NOT delegated to `isNativelyReadableScheme`: object_store recognizes schemes (http/https, * azure, memory) that iceberg-rust's OpenDAL storage factory cannot build, and admitting them * here turns a clean JVM fallback into a native runtime "Unsupported storage scheme" error. Add - * here what you add to `storage_factory_for` (currently Aliyun `oss` and GCS `gs`). + * here what you add to `storage_factory_for` (currently Aliyun `oss`, GCS `gs` and `hdfs`). * S3-compliant aliases like `blob` are opt-in via `fs.comet.s3Compliant.schemes` (see * `isIcebergReadableScheme`), not hardcoded, since the native planner opens them via S3. The * write path keeps its own list (`CometIcebergNativeWrite.SupportedStorageSchemes`), which * differs deliberately: it excludes `oss` (fails closed, see `storage_factory_for`) and * includes `memory`. + * + * `hdfs` routes to iceberg-rust's pure-Rust `hdfs-native` backend, not the libhdfs/JNI client + * the plain-Parquet path uses; a libhdfs alias scheme has no iceberg-rust arm and stays out. */ private val icebergReadableSchemes: Set[String] = - Set("file", "s3", "s3a", "gs", "oss") + Set("file", "s3", "s3a", "gs", "oss", "hdfs") /** * "Supported schemes: ..." suffix shared by the Iceberg scheme-fallback messages. Lists the @@ -1266,6 +1293,10 @@ object CometScanRule extends Logging { * The one exception is an opt-in S3-compliant alias, which the native reader opens by promoting * the bucket from the first path segment (`s3_blob_fs_support.rs`), so a hostless * `blob:///bucket/key.parquet` IS openable when it carries a promotable bucket segment. + * + * `hdfs` follows the general rule: the authority is the NameNode. A hostless `hdfs:///path` + * could be opened from a configured `hdfs.name-node`, but this gate runs before the catalog + * properties are assembled, so it declines rather than guess. */ private[rules] def hasOpenableAuthority(uri: URI, s3CompliantSchemes: Set[String]): Boolean = { val scheme = NativeConfig.lowerScheme(uri) @@ -1310,6 +1341,9 @@ object CometScanRule extends Logging { // Buckets the native FileIO must read (data + delete files), for the single-config check in // CometScanRule; only S3-family locations contribute (see NativeConfig.bucketForUri). val dataFileBuckets = mutable.Set[String]() + // Distinct `hdfs://` authorities across data and delete files, for the single-NameNode check + // in CometScanRule. + val dataFileHdfsAuthorities = mutable.Set[String]() // First data/delete location with a readable scheme but no URL host (see // hasOpenableAuthority); non-empty => decline. One example suffices for the message. var hostlessLocation: Option[String] = None @@ -1332,6 +1366,10 @@ object CometScanRule extends Logging { unsupportedSchemes += lower } else if (!hasOpenableAuthority(uri, s3CompliantSchemes)) { if (hostlessLocation.isEmpty) hostlessLocation = Some(rawPath) + } else if (lower == "hdfs") { + // RAW authority, not getHost, which answers null for a nameservice carrying an underscore + // and would quietly weaken the single-NameNode check. + Option(uri.getRawAuthority).filter(_.nonEmpty).foreach(dataFileHdfsAuthorities += _) } else { // bucketForUri yields None for non-S3-family URIs, so no scheme re-check is needed here. NativeConfig.bucketForUri(uri, s3CompliantSchemes).foreach(dataFileBuckets += _) @@ -1391,6 +1429,7 @@ object CometScanRule extends Logging { nonIdentityTransform, deleteFiles, dataFileBuckets.toSet, + dataFileHdfsAuthorities.toSet, hostlessLocation) } } @@ -1404,4 +1443,5 @@ case class IcebergTaskValidationResult( nonIdentityTransform: Option[String], deleteFiles: java.util.List[_], dataFileBuckets: Set[String], + dataFileHdfsAuthorities: Set[String], hostlessLocation: Option[String]) diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala index ac2538288e2..ff094679531 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala @@ -594,6 +594,54 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit } } + /** + * Resolves the `hdfs.name-node` property iceberg-rust's `hdfs-native` backend needs, from the + * session Hadoop configuration. + * + * opendal's `HdfsNativeBuilder` never dials the path authority: it builds one client against a + * synthetic authority and synthesizes the HA config from the comma-separated `name_node` value + * (`init_hdfs_config` in `opendal-service-hdfs-native`). iceberg-rust falls back to the path + * authority only when this property is absent, which is correct just for a real `host:port`. An + * HA location reads `hdfs:///...`, and a nameservice is not a routable host, so + * without this mapping every HA table fails to connect at execution time -- after the planner + * has already committed to the native scan. + * + * A non-HA authority yields nothing: the path authority is already correct, and a property + * would only pin the scan to one endpoint. Call sites order this before the catalog properties, + * so an explicit `spark.sql.catalog..hdfs.name-node` still wins. + * + * @param uri + * the metadata (scan) or data (write) location whose authority names the nameservice + */ + def hadoopToIcebergHdfsProperties( + uri: java.net.URI, + hadoopConf: org.apache.hadoop.conf.Configuration): Map[String, String] = { + if (!NativeConfig.lowerScheme(uri).contains("hdfs")) return Map.empty + // The RAW authority, not `getHost`, which answers null for a nameservice carrying an + // underscore. + val nameservice = Option(uri.getRawAuthority).filter(_.nonEmpty).getOrElse(return Map.empty) + + // Absent for a plain `host:port` authority, which needs no mapping. + val nnIds = Option(hadoopConf.getTrimmedStrings(s"dfs.ha.namenodes.$nameservice")) + .map(_.toSeq) + .getOrElse(Seq.empty) + .filter(_.nonEmpty) + + val endpoints = nnIds.flatMap { nnId => + Option(hadoopConf.getTrimmed(s"dfs.namenode.rpc-address.$nameservice.$nnId")) + .filter(_.nonEmpty) + .map(addr => if (addr.startsWith("hdfs://")) addr else s"hdfs://$addr") + } + + // All-or-nothing: a partially resolved list would silently drop a NameNode, turning a + // failover into an outage. + if (endpoints.nonEmpty && endpoints.size == nnIds.size) { + Map("hdfs.name-node" -> endpoints.mkString(",")) + } else { + Map.empty + } + } + /** * Transforms Hadoop S3A configuration keys to Iceberg FileIO property keys. * diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala index 4aee4123ff3..ec7c123bc5a 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala @@ -84,8 +84,10 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] { // `oss` is deliberately absent: iceberg-rust has an OSS backend, but Comet does not forward // `oss.*` catalog properties to it and no functional test covers the path, so an OSS write // could silently drop endpoint/credential configuration. Fail closed until it is covered. + // + // `hdfs` IS present: its NameNode endpoints are forwarded below, so nothing is dropped. private val SupportedStorageSchemes: Set[String] = - Set("file", "memory", "s3", "s3a", "gs") + Set("file", "memory", "s3", "s3a", "gs", "hdfs") private val MinUnsupportedFormatVersion = 3 private val ParquetWritePropertyPrefix = "write.parquet." private val ParquetMrPropertyPrefix = "parquet." @@ -651,7 +653,11 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] { val hadoopDerivedProperties = CometIcebergNativeScan.hadoopToIcebergS3Properties( NativeConfig.extractObjectStoreOptions(writeHadoopConf, dataUri), dataBucket) - val catalogProperties = hadoopDerivedProperties ++ fileIOProperties + // Before `fileIOProperties` so an explicit catalog `hdfs.name-node` wins, as on the scan path. + val hadoopDerivedHdfsProperties = + CometIcebergNativeScan.hadoopToIcebergHdfsProperties(dataUri, writeHadoopConf) + val catalogProperties = + hadoopDerivedProperties ++ hadoopDerivedHdfsProperties ++ fileIOProperties val common = IcebergWriteProtoTranslation.buildCommon( catalogProperties = catalogProperties, diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergHdfsSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergHdfsSuite.scala new file mode 100644 index 00000000000..845c93f3601 --- /dev/null +++ b/spark/src/test/scala/org/apache/comet/CometIcebergHdfsSuite.scala @@ -0,0 +1,169 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet + +import java.util.UUID + +import org.apache.spark.sql.CometTestBase +import org.apache.spark.sql.comet.CometIcebergNativeScanExec +import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper + +import org.apache.comet.iceberg.IcebergReflection + +/** + * End-to-end coverage of the native Iceberg scan against an `hdfs://` warehouse, backed by an + * in-process `MiniDFSCluster`. + * + * This is the only test that exercises iceberg-rust's `hdfs-native` backend for real. It matters + * because that backend is a second, independent HDFS client: the plain-Parquet native scan + * reaches HDFS through libhdfs/JNI (`fs.comet.libhdfs.schemes`), while an Iceberg table on HDFS + * is opened by a pure-Rust RPC client that shares nothing with it but the `$HADOOP_CONF_DIR` XML. + * A unit test over the scheme allowlists cannot tell whether that client actually connects. + * + * The cluster is a single NameNode, so table locations carry a real `host:port` authority and + * iceberg-rust needs no `hdfs.name-node` property; the HA translation that supplies one is + * covered by `CometIcebergNativeScanSuite`. + */ +class CometIcebergHdfsSuite + extends CometTestBase + with AdaptiveSparkPlanHelper + with CometIcebergTestBase + with WithHdfsCluster { + + /** + * MiniDFSCluster cannot start when the `hadoop-client-minicluster` pinned in `pom.xml` (3.3.4) + * is older than the `hadoop-client-api`/`runtime` Spark supplies: `HttpServer2` then resolves a + * shaded Jetty class the older jar does not carry, and the NameNode web server dies. That is + * true on the Spark 4.x profiles today. Record the failure and skip rather than fail, so this + * suite reports honestly on the profiles where the fixture works and stays quiet elsewhere. + */ + private var hdfsClusterAvailable = false + + override def beforeAll(): Unit = { + super.beforeAll() + try { + startHdfsCluster() + hdfsClusterAvailable = true + } catch { + case e: Throwable => + logWarning(s"Skipping ${getClass.getSimpleName}: MiniDFSCluster failed to start", e) + } + } + + override def afterAll(): Unit = { + try if (hdfsClusterAvailable) stopHdfsCluster() + finally super.afterAll() + } + + private def assumeHdfs(): Unit = { + assume(icebergAvailable, "Iceberg not available in classpath") + assume(hdfsClusterAvailable, "MiniDFSCluster unavailable in this dependency set") + } + + /** `hdfs://localhost:` -- the authority iceberg-rust dials as the NameNode. */ + private def hdfsUri: String = s"hdfs://localhost:$getDFSPort" + + private def assertSingleNativeScan(cometPlan: SparkPlan): Unit = { + val scans = collect(cometPlan) { case scan: CometIcebergNativeScanExec => scan } + assert( + scans.length == 1, + s"Expected exactly 1 CometIcebergNativeScanExec but found ${scans.length}. " + + s"Plan:\n$cometPlan") + } + + /** + * Runs `f` with a Hadoop-catalog Iceberg warehouse rooted on the MiniDFS cluster. The catalog + * name is unique per test so Spark's catalog cache cannot hand back a warehouse from an earlier + * test. + */ + private def withHdfsIcebergCatalog(f: String => Unit): Unit = { + val catalog = s"hdfs_cat_${UUID.randomUUID().toString.replace("-", "")}" + val warehouse = s"$hdfsUri/warehouse/${UUID.randomUUID()}" + withSQLConf( + s"spark.sql.catalog.$catalog" -> "org.apache.iceberg.spark.SparkCatalog", + s"spark.sql.catalog.$catalog.type" -> "hadoop", + s"spark.sql.catalog.$catalog.warehouse" -> warehouse, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + f(catalog) + } + } + + test("native Iceberg scan reads a table stored on HDFS") { + assumeHdfs() + + withHdfsIcebergCatalog { catalog => + spark.sql(s"CREATE TABLE $catalog.db.t (id INT, name STRING, value DOUBLE) USING iceberg") + spark.sql( + s"INSERT INTO $catalog.db.t VALUES (1, 'Alice', 10.5), (2, 'Bob', 20.3), " + + "(3, 'Charlie', 30.7)") + + // The data location must actually be on HDFS, or this suite would silently degrade into a + // duplicate of the local-filesystem coverage. + val dataLocation = IcebergReflection + .getDataLocation(loadIcebergTable(spark, catalog, "db", "t")) + .getOrElse(fail("could not resolve the Iceberg data location")) + assert( + dataLocation.startsWith("hdfs://"), + s"expected an hdfs:// data location, got $dataLocation") + + val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $catalog.db.t ORDER BY id") + assertSingleNativeScan(cometPlan) + + spark.sql(s"DROP TABLE $catalog.db.t") + } + } + + test("native Iceberg scan on HDFS applies a pushed-down filter") { + assumeHdfs() + + withHdfsIcebergCatalog { catalog => + spark.sql(s"CREATE TABLE $catalog.db.f (id INT, name STRING) USING iceberg") + spark.sql( + s"INSERT INTO $catalog.db.f VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'), (5, 'e')") + + val (_, cometPlan) = + checkSparkAnswer(s"SELECT id, name FROM $catalog.db.f WHERE id > 3 ORDER BY id") + assertSingleNativeScan(cometPlan) + + spark.sql(s"DROP TABLE $catalog.db.f") + } + } + + test("native Iceberg scan reads a partitioned table across multiple HDFS data files") { + assumeHdfs() + + withHdfsIcebergCatalog { catalog => + spark.sql( + s"CREATE TABLE $catalog.db.p (id INT, part STRING) USING iceberg PARTITIONED BY (part)") + // Separate inserts so each partition lands in its own data file: a single-file read would + // not prove the operator cache serves more than one path from one NameNode. + spark.sql(s"INSERT INTO $catalog.db.p VALUES (1, 'x'), (2, 'x')") + spark.sql(s"INSERT INTO $catalog.db.p VALUES (3, 'y'), (4, 'y')") + + val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $catalog.db.p ORDER BY id") + assertSingleNativeScan(cometPlan) + + spark.sql(s"DROP TABLE $catalog.db.p") + } + } +} diff --git a/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala b/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala index 5a14f9f592e..77344010f3e 100644 --- a/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala +++ b/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala @@ -129,7 +129,10 @@ class CometScanSchemeFallbackSuite extends CometTestBase { "s3://bucket/key.parquet", "s3a://bucket/key.parquet", "gs://bucket/key.parquet", - "oss://bucket/key.parquet").foreach { u => + "oss://bucket/key.parquet", + // hdfs-native backend, not the libhdfs/JNI client the plain-Parquet path uses. + "hdfs://nn:8020/warehouse/db/t/key.parquet", + "hdfs://nameservice1/warehouse/db/t/key.parquet").foreach { u => assert( CometScanRule.isIcebergReadableScheme(new URI(u), Set.empty), s"$u must be iceberg-readable; icebergReadableSchemes has regressed") @@ -216,6 +219,13 @@ class CometScanSchemeFallbackSuite extends CometTestBase { assert( !openable("blob:///bucket/k.parquet", Set.empty), "without opt-in, blob gets no bucket promotion, so a hostless blob location is unopenable") + // hdfs: the authority is the NameNode, and the gate runs before the catalog properties that + // could supply one, so a hostless location declines rather than guess. + assert(openable("hdfs://nn:8020/warehouse/db/t/k.parquet")) + assert(openable("hdfs://nameservice1/warehouse/db/t/k.parquet")) + assert( + !openable("hdfs:///warehouse/db/t/k.parquet"), + "authorityless hdfs:/// carries no NameNode and gets no promotion") } test("native scan claims hdfs:// when libhdfs.schemes is unset (native-default lockstep)") { diff --git a/spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala b/spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala index e11f8bd3477..216bc8543bf 100644 --- a/spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala +++ b/spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala @@ -137,4 +137,55 @@ class CometIcebergNativeScanSuite extends AnyFunSuite with Matchers { out("s3.endpoint") shouldBe "https://global.example.com" out.values.toSet should not contain "https://some.example.com" } + + // --- hadoopToIcebergHdfsProperties ------------------------------------------------------- + // + // These pin the HA translation that makes `hdfs:///...` reachable; see that + // method's scaladoc for why the property is required rather than optional. + + private def hdfsProps(location: String, conf: Map[String, String]): Map[String, String] = { + val hadoopConf = new org.apache.hadoop.conf.Configuration(false) + conf.foreach { case (k, v) => hadoopConf.set(k, v) } + CometIcebergNativeScan.hadoopToIcebergHdfsProperties(new java.net.URI(location), hadoopConf) + } + + test("HA nameservice resolves to the comma-separated NameNode list, in declaration order") { + val out = hdfsProps( + "hdfs://nameservice1/warehouse/db/t/metadata.json", + Map( + "dfs.ha.namenodes.nameservice1" -> "nn1,nn2", + "dfs.namenode.rpc-address.nameservice1.nn1" -> "host-a.example.com:8020", + "dfs.namenode.rpc-address.nameservice1.nn2" -> "host-b.example.com:8020")) + + out shouldBe Map( + "hdfs.name-node" -> "hdfs://host-a.example.com:8020,hdfs://host-b.example.com:8020") + } + + test("a plain host:port authority needs no mapping") { + // A property would only pin the scan to one endpoint. + hdfsProps("hdfs://nn.example.com:8020/warehouse/db/t", Map.empty) shouldBe Map.empty + } + + test("a partially resolved HA list yields nothing rather than a short failover list") { + // Dropping nn2 would silently turn a failover into an outage. + hdfsProps( + "hdfs://nameservice1/warehouse", + Map( + "dfs.ha.namenodes.nameservice1" -> "nn1,nn2", + "dfs.namenode.rpc-address.nameservice1.nn1" -> "host-a.example.com:8020")) shouldBe Map.empty + } + + test("rpc-address already carrying the hdfs:// prefix is not double-prefixed") { + hdfsProps( + "hdfs://ns/warehouse", + Map( + "dfs.ha.namenodes.ns" -> "nn1", + "dfs.namenode.rpc-address.ns.nn1" -> "hdfs://host-a.example.com:8020")) shouldBe + Map("hdfs.name-node" -> "hdfs://host-a.example.com:8020") + } + + test("non-hdfs and authority-less locations are ignored") { + hdfsProps("s3://bucket/key", Map("dfs.ha.namenodes.bucket" -> "nn1")) shouldBe Map.empty + hdfsProps("hdfs:///warehouse/db/t", Map.empty) shouldBe Map.empty + } }