diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index 0b98a39356..3561447b64 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -170,6 +170,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: @@ -244,3 +263,16 @@ the expression fall back to Spark. The native Iceberg reader populates Spark's task-level `inputMetrics.bytesRead` (visible in the Spark UI Stages tab) using the `bytes_read` counter from iceberg-rust's `ScanMetrics`. This counter includes bytes read from both data files and delete files. Iceberg Java does not explicitly report `bytesRead` to Spark's task input metrics. On the iceberg Java path, any `bytesRead` value comes from Hadoop's filesystem-level I/O counters, not from Iceberg itself. Because Comet's native reader and the Hadoop filesystem use different counting mechanisms, the exact byte counts will differ between the two paths. + +### SQL tab metrics + +`CometIcebergNativeScan` reports Iceberg's planning metrics (manifests and data files scanned or +skipped, planning duration, total data and delete file sizes) under their Iceberg names, posted +from the driver for each execution once the scan's partitions are planned, and the native read time +as `scan time`. + +iceberg-rust applies the residual predicate Comet hands it as a row filter inside the scan, so +`number of output rows`, and with it the task-level `recordsRead`, count the rows that pass it. They +are therefore lower than the `BatchScan` figures on the Iceberg Java path, where every row leaves +the scan and is filtered by the `Filter` above it. `number of row deletes applied` has no native +counterpart: iceberg-rust's `ScanMetrics` exposes bytes read only. diff --git a/native/Cargo.lock b/native/Cargo.lock index d4c592e1f8..613d463cc7 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]] @@ -2821,6 +2891,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" @@ -2877,6 +2956,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" @@ -3175,6 +3266,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" @@ -3194,7 +3313,7 @@ dependencies = [ "cfg-if", "js-sys", "libc", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", "wasm-bindgen", ] @@ -3321,6 +3440,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" @@ -3544,7 +3704,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", @@ -3603,7 +3763,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", @@ -3613,7 +3773,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", @@ -3786,10 +3946,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" @@ -3946,7 +4116,7 @@ dependencies = [ "java-locator", "jni-macros", "jni-sys 0.4.1", - "libloading", + "libloading 0.8.9", "log", "simd_cesu8", "thiserror 2.0.20", @@ -4113,6 +4283,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" @@ -4148,6 +4328,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" @@ -4321,7 +4510,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", ] @@ -4490,6 +4679,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" @@ -4597,6 +4804,7 @@ dependencies = [ "opendal-service-fs", "opendal-service-gcs", "opendal-service-hdfs", + "opendal-service-hdfs-native", "opendal-service-oss", "opendal-service-s3", ] @@ -4765,6 +4973,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" @@ -5088,8 +5310,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", @@ -5857,6 +6079,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" @@ -6016,7 +6247,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213" dependencies = [ - "cipher", + "cipher 0.4.4", ] [[package]] @@ -7070,6 +7301,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" @@ -7079,6 +7319,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" @@ -7189,6 +7438,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" @@ -7297,6 +7559,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" @@ -7330,13 +7601,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" @@ -7349,6 +7637,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" @@ -7361,6 +7655,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" @@ -7373,12 +7673,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" @@ -7391,6 +7703,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" @@ -7403,6 +7721,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" @@ -7415,6 +7739,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" @@ -7427,6 +7757,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 6cf35f8676..aa0fce05b9 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -63,8 +63,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] @@ -83,3 +86,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 509e3b1f98..7c30a38f37 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 6766979652..d667d206f2 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 d6a5223bee..5e6b3df121 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -515,6 +515,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. @@ -583,7 +598,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 -> _) @@ -1193,15 +1217,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 @@ -1237,6 +1264,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) @@ -1281,6 +1312,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 @@ -1303,6 +1337,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 += _) @@ -1362,6 +1400,7 @@ object CometScanRule extends Logging { nonIdentityTransform, deleteFiles, dataFileBuckets.toSet, + dataFileHdfsAuthorities.toSet, hostlessLocation) } } @@ -1375,4 +1414,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 9b1c6b72e8..351dc16f78 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 @@ -595,6 +595,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 94f6b13927..8769f9a34d 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,9 +84,11 @@ 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. // `gs` is additionally gated on the resolved FileIO (`requireGcsFileIOForGcsDataLocation`). 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." @@ -684,7 +686,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/main/scala/org/apache/spark/sql/comet/CometExecRDD.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometExecRDD.scala index 1d876dfb83..be108aa6a9 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometExecRDD.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometExecRDD.scala @@ -99,9 +99,12 @@ private[spark] class CometExecRDD( override def compute(split: Partition, context: TaskContext): Iterator[ColumnarBatch] = { // Must precede resolveInputObjects and the CometExecIterator: completion listeners run in - // reverse registration order, so registering first means this listener runs last, after + // reverse registration order, so registering first means these listeners run last, after // nested native blocks and the iterator have published their final metric values. - Option(context).foreach(nativeMetrics.reportSpillMetrics) + Option(context).foreach { ctx => + nativeMetrics.reportSpillMetrics(ctx) + nativeMetrics.reportScanInputMetrics(ctx) + } val partition = split.asInstanceOf[CometExecPartition] diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala index 65cbe30038..8946f0eea0 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala @@ -19,9 +19,10 @@ package org.apache.spark.sql.comet +import java.util.concurrent.ConcurrentHashMap + import scala.jdk.CollectionConverters._ -import org.apache.spark.{Partition, TaskContext} import org.apache.spark.rdd.RDD import org.apache.spark.sql.SparkSession import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, SortOrder} @@ -69,6 +70,31 @@ case class CometIcebergNativeScanExec( override val nodeName: String = "CometIcebergNativeScan" + /** + * `originalPlan` rebuilt with the current top-level `runtimeFilters`. Spark's + * PlanAdaptiveDynamicPruningFilters and our transformExpressionsUp passes rewrite the top-level + * `runtimeFilters` (visible via productIterator), but `originalPlan` is @transient and not + * touched by transformAllExpressions. serializePartitions reads runtime filters via `inputRDD + * -> filteredPartitions`, so an out-of-sync originalPlan would re-translate the original + * (unresolved) InSubqueryExec and throw "no subquery result". This makes the top-level + * runtimeFilters the single source of truth at serialization time. Iceberg reports its planning + * metrics on the instance whose `inputRDD` ran, so [[LazyIcebergMetric]] reads them from here. + */ + @transient private lazy val plannedOriginalPlan: BatchScanExec = { + // Canonicalized instances set originalPlan = null and are not meant to be executed. + // If we ever reach this lazy val on a canonicalized form, fail loud rather than NPE + // deep inside originalPlan.inputRDD. + assert( + originalPlan != null, + "plan data accessed on a canonicalized CometIcebergNativeScanExec; " + + "this lazy val should only execute on non-canonical instances") + if (originalPlan.runtimeFilters != runtimeFilters) { + originalPlan.copy(runtimeFilters = runtimeFilters) + } else { + originalPlan + } + } + /** * Lazy partition serialization, deferred until execution time. Triggered from `commonData` / * `perPartitionData` (via `PlanDataInjector.findAllPlanData`) and from @@ -84,33 +110,11 @@ case class CometIcebergNativeScanExec( * by every construction site), so values resolved through `waitForSubqueries` are visible on * both sides. */ - @transient private lazy val serializedPartitionData: (Array[Byte], Array[Array[Byte]]) = { - // Canonicalized instances set originalPlan = null and are not meant to be executed. - // If we ever reach this lazy val on a canonicalized form, fail loud rather than NPE - // deep inside originalPlan.inputRDD. - assert( - originalPlan != null, - "serializedPartitionData accessed on a canonicalized CometIcebergNativeScanExec; " + - "this lazy val should only execute on non-canonical instances") - // Rebuild originalPlan with the current top-level runtimeFilters before serializing. - // Spark's PlanAdaptiveDynamicPruningFilters and our transformExpressionsUp passes rewrite - // the top-level `runtimeFilters` (visible via productIterator), but `originalPlan` is - // @transient and not touched by transformAllExpressions. serializePartitions reads runtime - // filters via originalPlan.inputRDD -> filteredPartitions, so an out-of-sync originalPlan - // would re-translate the original (unresolved) InSubqueryExec and throw "no subquery - // result". This makes the top-level runtimeFilters the single source of truth at - // serialization time. - val effectiveOriginalPlan = - if (originalPlan.runtimeFilters != runtimeFilters) { - originalPlan.copy(runtimeFilters = runtimeFilters) - } else { - originalPlan - } + @transient private lazy val serializedPartitionData: (Array[Byte], Array[Array[Byte]]) = CometIcebergNativeScan.serializePartitions( - effectiveOriginalPlan, + plannedOriginalPlan, output, nativeIcebergScanMetadata) - } def commonData: Array[Byte] = serializedPartitionData._1 @@ -162,8 +166,9 @@ case class CometIcebergNativeScanExec( * and throw at executeCollect(). Lazy value access ensures planning runs only when the value is * actually needed, by which time CometPlanAdaptiveDynamicPruningFilters has converted the SAB. * - * Overrides merge/reset because executor accumulator updates carry 0 (these are driver-side - * planning metrics) and would zero out the resolved value at end of stage. + * Overrides merge/reset so nothing can zero out the resolved value: these are driver-side + * planning metrics, and [[CometMetricNode.fromCometPlan]] keeps them out of the tree that tasks + * update. */ private class LazyIcebergMetric(metricType: String, metricName: String) extends SQLMetric(metricType, 0) { @@ -174,7 +179,7 @@ case class CometIcebergNativeScanExec( // and inputRDD -> filteredPartitions skips DPP, caching an unfiltered result. ensureSubqueriesResolved() val _ = serializedPartitionData - originalPlan.metrics.get(metricName).map(_.value).getOrElse(0L) + plannedOriginalPlan.metrics.get(metricName).map(_.value).getOrElse(0L) } override def merge(other: AccumulatorV2[Long, Long]): Unit = {} @@ -223,6 +228,10 @@ case class CometIcebergNativeScanExec( baseMetrics ++ icebergPlanningMetrics + ("num_splits" -> numSplitsMetric) } + /** The metrics native execution updates; the planning metrics are set on the driver only. */ + private[comet] def runtimeMetrics: Map[String, SQLMetric] = + metrics -- icebergPlanningMetrics.keys + /** * Posts the Iceberg planning metrics (data/delete file and manifest counts, file sizes, and * total planning duration) to the SQL UI as driver metrics. Iceberg-Java produces these during @@ -237,19 +246,24 @@ case class CometIcebergNativeScanExec( * for a pushed predicate), the parent runs the whole subtree as one RDD and this node's * doExecuteColumnar is never invoked. CometNativeExec.findAllPlanData walks the subtree at * execution time and reaches every leaf scan (calling this leaf lifecycle hook alongside - * ensureSubqueriesResolved), so it calls this too. Re-posting the same values is harmless. + * ensureSubqueriesResolved), so it calls this too. Posted once per SQL execution: Spark appends + * driver updates to a list, so a second post within one execution would double the displayed + * totals, while a re-executed Dataset reuses this plan and needs its own post. */ + @transient private lazy val planningMetricsPostedTo = ConcurrentHashMap.newKeySet[String]() + override def sendDriverMetrics(): Unit = { - if (icebergPlanningMetrics.isEmpty) { - return - } - // Force planning so originalPlan.metrics are populated; LazyIcebergMetric.value reads them. - val _ = serializedPartitionData val executionId = sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY) - SQLMetrics.postDriverMetricUpdates( - sparkContext, - executionId, - icebergPlanningMetrics.values.toSeq) + if (icebergPlanningMetrics.nonEmpty && executionId != null && + planningMetricsPostedTo.add(executionId)) { + // Force planning so plannedOriginalPlan.metrics are populated; LazyIcebergMetric.value + // reads them. + val _ = serializedPartitionData + SQLMetrics.postDriverMetricUpdates( + sparkContext, + executionId, + icebergPlanningMetrics.values.toSeq) + } } /** Executes using CometExecRDD - planning data is computed lazily on first access. */ @@ -271,13 +285,7 @@ case class CometIcebergNativeScanExec( defaultNumPartitions = perPartitionData.length, numOutputCols = output.length, nativeMetrics = nativeMetrics, - subqueries = Seq.empty) { - override def compute(split: Partition, context: TaskContext): Iterator[ColumnarBatch] = { - val res = super.compute(split, context) - Option(context).foreach(nativeMetrics.reportScanInputMetrics) - res - } - } + subqueries = Seq.empty) } /** diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala index 806c9d00ed..c08f472932 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala @@ -22,6 +22,7 @@ package org.apache.spark.sql.comet import java.util.IdentityHashMap import java.util.concurrent.ConcurrentHashMap +import scala.collection.mutable import scala.jdk.CollectionConverters._ import org.apache.spark.{SparkContext, TaskContext} @@ -105,24 +106,41 @@ case class CometMetricNode(metrics: Map[String, SQLMetric], children: Seq[CometM }) /** - * Reports aggregated scan input metrics (bytesRead, recordsRead) to Spark's task metrics. - * Aggregates across all scan leaf nodes to handle plans with multiple scans (e.g., joins). Must - * be called in a TaskCompletionListener after the iterator is fully consumed. + * Reports the scan leaves' bytes and rows (output rows plus rows pruned by pushed-down + * predicates) to Spark's task-level [[org.apache.spark.executor.InputMetrics]], so the Spark UI + * Stages tab Input column shows them. Scan leaves are the leaf nodes carrying `bytes_scanned`. + * + * Must be registered on the task thread before [[org.apache.comet.CometExecIterator]] so its + * completion listener publishes final SQL metrics before this listener runs, including when the + * iterator was not drained. Increments rather than sets so several scan trees in one task, and + * input reported by other sources, all add up; as in [[reportSpillMetrics]], each accumulator + * is claimed once per task so overlapping trees do not double count. A coalesced task reports + * what its last native plan left in the shared accumulators, as the SQL metrics themselves do. */ def reportScanInputMetrics(ctx: TaskContext): Unit = { + val seenMetrics = CometMetricNode.taskSeenMetrics(ctx) ctx.addTaskCompletionListener[Unit] { _ => val scanLeaves = leafNodes.filter(_.metrics.contains("bytes_scanned")) - if (scanLeaves.nonEmpty) { - val totalBytes = scanLeaves.map(_.metrics("bytes_scanned").value).sum - val totalRows = scanLeaves.map { leaf => - val outputRows = - leaf.metrics.get("output_rows").map(_.value).getOrElse(0L) - val prunedRows = - leaf.metrics.get("pushdown_rows_pruned").map(_.value).getOrElse(0L) - outputRows + prunedRows - }.sum - ctx.taskMetrics().inputMetrics.setBytesRead(totalBytes) - ctx.taskMetrics().inputMetrics.setRecordsRead(totalRows) + def sumUnclaimed(metricName: String): Long = + scanLeaves + .flatMap(_.metrics.get(metricName)) + .map { metric => + if (seenMetrics(metricName).put(metric, java.lang.Boolean.TRUE) == null) { + math.max(metric.value, 0L) + } else { + 0L + } + } + .sum + + val bytesRead = sumUnclaimed("bytes_scanned") + if (bytesRead > 0L) { + ctx.taskMetrics().inputMetrics.incBytesRead(bytesRead) + } + + val recordsRead = sumUnclaimed("output_rows") + sumUnclaimed("pushdown_rows_pruned") + if (recordsRead > 0L) { + ctx.taskMetrics().inputMetrics.incRecordsRead(recordsRead) } } } @@ -161,14 +179,15 @@ case class CometMetricNode(metrics: Map[String, SQLMetric], children: Seq[CometM * per-task registry, so each accumulator is counted once while disjoint trees still all report. */ def reportSpillMetrics(ctx: TaskContext): Unit = { - val seenMetrics = CometMetricNode.taskSeenSpillMetrics(ctx) + val seenMetrics = CometMetricNode.taskSeenMetrics(ctx) ctx.addTaskCompletionListener[Unit] { _ => - val diskBytesSpilled = sumMetricValues("spilled_bytes", seenMetrics.disk) + val diskBytesSpilled = sumMetricValues("spilled_bytes", seenMetrics("spilled_bytes")) if (diskBytesSpilled > 0L) { ctx.taskMetrics().incDiskBytesSpilled(diskBytesSpilled) } - val memoryBytesSpilled = sumMetricValues("memory_spilled_bytes", seenMetrics.memory) + val memoryBytesSpilled = + sumMetricValues("memory_spilled_bytes", seenMetrics("memory_spilled_bytes")) if (memoryBytesSpilled > 0L) { ctx.taskMetrics().incMemoryBytesSpilled(memoryBytesSpilled) } @@ -225,26 +244,31 @@ object CometMetricNode { private val aggregateMetricNames = Set("spill_count", "spilled_bytes", "spilled_rows", "peak_mem_used") - private case class SeenSpillMetrics( - disk: IdentityHashMap[SQLMetric, java.lang.Boolean], - memory: IdentityHashMap[SQLMetric, java.lang.Boolean]) + /** The accumulators a task's reporting listeners have claimed, one identity set per metric. */ + private class SeenMetrics { + private val byName = + mutable.HashMap.empty[String, IdentityHashMap[SQLMetric, java.lang.Boolean]] + + def apply(metricName: String): IdentityHashMap[SQLMetric, java.lang.Boolean] = + byName.getOrElseUpdate(metricName, new IdentityHashMap()) + } - // Per running task attempt: the spill accumulators already claimed by a reporting listener, - // one identity set per metric name (see reportSpillMetrics). The first registration installs - // a cleanup listener ahead of every reporting listener, so it runs last (reverse registration - // order) and removes the entry. - private val seenSpillMetricsByTask = new ConcurrentHashMap[Long, SeenSpillMetrics]() + // Per running task attempt: the accumulators already claimed by a reporting listener (see + // reportSpillMetrics and reportScanInputMetrics). The first registration installs a cleanup + // listener ahead of every reporting listener, so it runs last (reverse registration order) and + // removes the entry. + private val seenMetricsByTask = new ConcurrentHashMap[Long, SeenMetrics]() - private def taskSeenSpillMetrics(ctx: TaskContext): SeenSpillMetrics = { + private def taskSeenMetrics(ctx: TaskContext): SeenMetrics = { val attemptId = ctx.taskAttemptId() - val existing = seenSpillMetricsByTask.get(attemptId) + val existing = seenMetricsByTask.get(attemptId) if (existing != null) { existing } else { // The task thread is the only registrant for its attempt id, so there is no put race. - val created = SeenSpillMetrics(new IdentityHashMap(), new IdentityHashMap()) - seenSpillMetricsByTask.put(attemptId, created) - ctx.addTaskCompletionListener[Unit](_ => seenSpillMetricsByTask.remove(attemptId)) + val created = new SeenMetrics + seenMetricsByTask.put(attemptId, created) + ctx.addTaskCompletionListener[Unit](_ => seenMetricsByTask.remove(attemptId)) created } } @@ -522,6 +546,9 @@ object CometMetricNode { */ def fromCometPlan(cometPlan: SparkPlan): CometMetricNode = { val nodeMetrics = cometPlan match { + // Driver-only planning metrics stay out of the tasks' accumulators, which would otherwise + // report zeros into the SQL UI's per-task statistics. + case scan: CometIcebergNativeScanExec => scan.runtimeMetrics case _: CometPlan => cometPlan.metrics case _ => try cometPlan.metrics diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala index a4365c0075..4d869f2ecf 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala @@ -19,7 +19,6 @@ package org.apache.spark.sql.comet -import org.apache.spark.{Partition, TaskContext} import org.apache.spark.rdd.RDD import org.apache.spark.sql.SparkSession import org.apache.spark.sql.catalyst._ @@ -276,16 +275,7 @@ case class CometNativeScanExec( Seq.empty, broadcastedHadoopConfForEncryption, encryptedFilePaths, - perPartitionFilePaths = perPartitionFilePaths) { - override def compute(split: Partition, context: TaskContext): Iterator[ColumnarBatch] = { - val res = super.compute(split, context) - - // Report scan input metrics after the iterator is fully consumed. - Option(context).foreach(nativeMetrics.reportScanInputMetrics) - - res - } - } + perPartitionFilePaths = perPartitionFilePaths) } override def doCanonicalize(): CometNativeScanExec = { diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDD.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDD.scala index bced55b34d..c8912574b9 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDD.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDD.scala @@ -39,7 +39,7 @@ private[shuffle] class CometNativeShuffleInputRDD( var inputRDDs: Seq[RDD[_]], numPartitionsParam: Int, shuffleScanIndices: Set[Int], - spillMetricNode: CometMetricNode, + taskMetricNode: CometMetricNode, @transient perPartitionByKey: Map[String, Array[Array[Byte]]] = Map.empty) extends RDD[Product2[Int, ColumnarBatch]]( sc, @@ -56,7 +56,7 @@ private[shuffle] class CometNativeShuffleInputRDD( inputRDDs, numPartitionsParam, shuffleScanIndices, - spillMetricNode, + taskMetricNode, perPartitionByKey) override protected def getPartitions: Array[Partition] = @@ -78,7 +78,11 @@ private[shuffle] class CometNativeShuffleInputRDD( override def compute( split: Partition, context: TaskContext): Iterator[Product2[Int, ColumnarBatch]] = { - spillMetricNode.reportSpillMetrics(context) + // Registered before the input producers and the writer's iterator so these listeners run + // after every nested native block and the writer plan have published their final metrics. + // The leaf scans run inside the writer's plan, so no CometExecRDD reports them. + taskMetricNode.reportSpillMetrics(context) + taskMetricNode.reportScanInputMetrics(context) val partition = split.asInstanceOf[CometNativeShuffleInputPartition] val (inputObjects, shuffleBlockIters) = CometExecRDD.resolveInputObjects( diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala index fce4291deb..f25ace2887 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala @@ -177,13 +177,6 @@ class CometNativeShuffleWriter[K, V]( // breakdown matches what the split-driver flow showed. val nativeMetrics = CometMetricNode(shuffleWriterSQLMetrics, Seq(spec.childMetricNode)) - // The leaf scans execute inside this writer's single plan rather than a separate native - // stage RDD, so the usual CometExecRDD.compute() bridge (operators.scala) never runs for - // them. Report their bytes/rows to the task's input metrics here instead. - if (ctx.hasScanInput) { - Option(context).foreach(nativeMetrics.reportScanInputMetrics) - } - val cometIter = new CometExecIterator( CometExec.newIterId, inputObjects, diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala index 2a53964375..662e4119b7 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala @@ -777,7 +777,7 @@ object CometShuffleExchangeExec Seq(streamRDD), rdd.getNumPartitions, shuffleScanIndices = Set.empty, - spillMetricNode = CometMetricNode(metrics, Seq(childMetricNode))) + taskMetricNode = CometMetricNode(metrics, Seq(childMetricNode))) val ctx = NativeExecContext( inputs = Seq(streamRDD), @@ -787,8 +787,7 @@ object CometShuffleExchangeExec encryptedFilePaths = Seq.empty, commonByKey = Map.empty, perPartitionByKey = Map.empty, - shuffleScanIndices = Set.empty, - hasScanInput = false) + shuffleScanIndices = Set.empty) // The Scan placeholder has no per-operator metrics, so the metric tree for the unified plan // is `shuffleWriterMetrics` at the root with one empty leaf for the Scan child. diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala index fe1a2e637a..cd49c7f8eb 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala @@ -25,7 +25,6 @@ import scala.collection.mutable import scala.collection.mutable.ArrayBuffer import scala.jdk.CollectionConverters._ -import org.apache.spark.{Partition, TaskContext} import org.apache.spark.broadcast.Broadcast import org.apache.spark.internal.Logging import org.apache.spark.rdd.RDD @@ -766,8 +765,7 @@ private[comet] case class NativeExecContext( // slice, never this map. Keeping it off the wire stops it from bloating the broadcast task // binary when this context rides on the non-transient CometShuffleDependency.nativeShuffleSpec. @transient perPartitionByKey: Map[String, Array[Array[Byte]]], - shuffleScanIndices: Set[Int], - hasScanInput: Boolean) { + shuffleScanIndices: Set[Int]) { // Catch shape divergence (e.g. broadcast scans with different partition counts after DPP // filtering) at construction so consumers don't trip ArrayIndexOutOfBoundsException at // partition idx access time. @@ -846,15 +844,7 @@ abstract class CometNativeExec extends CometExec { ctx.subqueries, ctx.broadcastedHadoopConfForEncryption, ctx.encryptedFilePaths, - ctx.shuffleScanIndices) { - override def compute(split: Partition, context: TaskContext): Iterator[ColumnarBatch] = { - val res = super.compute(split, context) - if (ctx.hasScanInput) { - Option(context).foreach(nativeMetrics.reportScanInputMetrics) - } - res - } - } + ctx.shuffleScanIndices) } /** @@ -1046,8 +1036,7 @@ abstract class CometNativeExec extends CometExec { encryptedFilePaths = encryptedFilePaths, commonByKey = commonByKey, perPartitionByKey = perPartitionByKey, - shuffleScanIndices = shuffleScanIndices, - hasScanInput = sparkPlans.exists(_.isInstanceOf[CometNativeScanExec])) + shuffleScanIndices = shuffleScanIndices) } /** 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 0000000000..845c93f360 --- /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/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 04dbd1f0a1..4ab016f588 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -34,7 +34,7 @@ import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} import org.apache.spark.sql.{CometTestBase, DataFrame, Row} import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression import org.apache.spark.sql.comet._ -import org.apache.spark.sql.comet.execution.shuffle.CometShuffleExchangeExec +import org.apache.spark.sql.comet.execution.shuffle.{CometNativeShuffle, CometShuffleExchangeExec} import org.apache.spark.sql.execution.{InSubqueryExec, ReusedSubqueryExec, SparkPlan, SubqueryExec} import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AdaptiveSparkPlanHelper, BroadcastQueryStageExec} import org.apache.spark.sql.execution.exchange.{ReusedExchangeExec, ShuffleExchangeExec} @@ -2247,6 +2247,13 @@ class CometIcebergNativeSuite uiValues.contains(nameToAccId("scan time")), "scan time should have a value in the SQL UI store") + // Planning metrics stay out of the executor metric tree, so a size metric renders as the + // driver's single value instead of statistics over per-task zeros. + val totalDataFileSize = uiValues(nameToAccId("totalDataFileSize")) + assert( + !totalDataFileSize.contains("0.0 B"), + s"totalDataFileSize should not carry task zeros, got $totalDataFileSize") + spark.sql("DROP TABLE test_cat.db.driver_metrics_test") } } @@ -3825,6 +3832,12 @@ class CometIcebergNativeSuite s"Expected CometIcebergNativeScanExec but found none. Plan:\n$cometPlan") val numPartitions = icebergScans.head.numPartitions assert(numPartitions == 1, s"Expected DPP to prune to 1 partition but got $numPartitions") + // Planning ran on the copy carrying the resolved DPP filters; the metrics are read from + // that copy rather than from originalPlan, whose accumulators stay at zero. + val resultDataFiles = icebergScans.head.metrics("resultDataFiles").value + assert( + resultDataFiles > 0, + s"Expected the planning metrics of the DPP scan, got resultDataFiles=$resultDataFiles") // Verify AQE DPP used CometSubqueryBroadcastExec with broadcast reuse if (isSpark35Plus) { @@ -4159,6 +4172,96 @@ class CometIcebergNativeSuite } } + test("task-level input metrics cover Iceberg scans fused below other native operators") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + spark.sql(""" + CREATE TABLE test_cat.db.fused_metrics_test ( + id INT, + value DOUBLE + ) USING iceberg + """) + + spark + .range(10000) + .selectExpr("CAST(id AS INT)", "CAST(id * 1.5 AS DOUBLE) as value") + .repartition(5) + .write + .format("iceberg") + .mode("append") + .saveAsTable("test_cat.db.fused_metrics_test") + + val bytesReadValues = mutable.ArrayBuffer.empty[Long] + val recordsReadValues = mutable.ArrayBuffer.empty[Long] + + val listener = new SparkListener { + override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = { + val im = taskEnd.taskMetrics.inputMetrics + bytesReadValues.synchronized { + bytesReadValues += im.bytesRead + recordsReadValues += im.recordsRead + } + } + } + spark.sparkContext.addSparkListener(listener) + + def collectInputMetrics(df: DataFrame): (Long, Long) = { + CometListenerBusUtils.waitUntilEmpty(spark.sparkContext) + bytesReadValues.clear() + recordsReadValues.clear() + df.collect() + CometListenerBusUtils.waitUntilEmpty(spark.sparkContext) + (bytesReadValues.sum, recordsReadValues.sum) + } + + try { + // Iceberg returns every pushed filter as a post-scan filter, so the WHERE keeps a + // CometFilterExec above the scan and the scan runs inside the filter's native block. + val filtered = spark.sql("SELECT * FROM test_cat.db.fused_metrics_test WHERE id >= 0") + val (filteredBytes, filteredRecords) = collectInputMetrics(filtered) + val filteredPlan = filtered.queryExecution.executedPlan + assert( + find(filteredPlan)(_.isInstanceOf[CometFilterExec]).isDefined && + collectIcebergNativeScans(filteredPlan).nonEmpty, + s"Expected a CometFilterExec above a CometIcebergNativeScanExec:\n$filteredPlan") + assert( + filteredRecords == 10000, + s"recordsRead below a native filter should be 10000, got $filteredRecords") + assert(filteredBytes > 0, "bytesRead below a native filter should be > 0") + + // The scan runs inside the native shuffle writer's plan on the map side. + val scanned = spark.sql("SELECT * FROM test_cat.db.fused_metrics_test") + val shuffled = scanned.repartition(4, scanned("id")) + val (shuffledBytes, shuffledRecords) = collectInputMetrics(shuffled) + val shuffledPlan = shuffled.queryExecution.executedPlan + assert( + find(shuffledPlan) { + case exchange: CometShuffleExchangeExec => + exchange.shuffleType == CometNativeShuffle + case _ => false + }.isDefined && collectIcebergNativeScans(shuffledPlan).nonEmpty, + s"Expected a native shuffle above a CometIcebergNativeScanExec:\n$shuffledPlan") + assert( + shuffledRecords == 10000, + s"recordsRead below a native shuffle should be 10000, got $shuffledRecords") + assert(shuffledBytes > 0, "bytesRead below a native shuffle should be > 0") + } finally { + spark.sparkContext.removeSparkListener(listener) + spark.sql("DROP TABLE test_cat.db.fused_metrics_test") + } + } + } + } + test("exchange reuse must not collapse scans with different pushed filters (#4774)") { assume(icebergAvailable, "Iceberg not available") 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 5a14f9f592..77344010f3 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 4fe2642117..ab27ede598 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 @@ -162,4 +162,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 + } } diff --git a/spark/src/test/scala/org/apache/spark/sql/comet/CometTaskMetricsSuite.scala b/spark/src/test/scala/org/apache/spark/sql/comet/CometTaskMetricsSuite.scala index 4679125541..bd74a0c829 100644 --- a/spark/src/test/scala/org/apache/spark/sql/comet/CometTaskMetricsSuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/comet/CometTaskMetricsSuite.scala @@ -127,6 +127,50 @@ class CometTaskMetricsSuite extends CometTestBase with AdaptiveSparkPlanHelper { } } + test("overlapping scan trees registered on one task report input metrics once") { + val nestedBytes = new SQLMetric("nestedBytes", -1L) + val nestedRows = new SQLMetric("nestedRows") + val nestedPruned = new SQLMetric("nestedPruned") + val filterRows = new SQLMetric("filterRows") + val siblingBytes = new SQLMetric("siblingBytes", -1L) + val siblingRows = new SQLMetric("siblingRows") + val nestedScan = CometMetricNode( + Map( + "bytes_scanned" -> nestedBytes, + "output_rows" -> nestedRows, + "pushdown_rows_pruned" -> nestedPruned)) + // Operators above the scan carry output_rows too; only scan leaves feed recordsRead. + val nestedTree = CometMetricNode(Map("output_rows" -> filterRows), Seq(nestedScan)) + val outerTree = CometMetricNode(Map.empty, Seq(nestedTree)) + val siblingTree = + CometMetricNode(Map("bytes_scanned" -> siblingBytes, "output_rows" -> siblingRows)) + + Seq(None, Some(new IllegalStateException("failed native stage"))).foreach { failure => + val ctx = TaskContext.empty() + // Input already reported by another source in the same task, such as a JVM scan. + ctx.taskMetrics.inputMetrics.incBytesRead(100L) + ctx.taskMetrics.inputMetrics.incRecordsRead(1L) + outerTree.reportScanInputMetrics(ctx) + nestedTree.reportScanInputMetrics(ctx) + nestedTree.reportScanInputMetrics(ctx) + siblingTree.reportScanInputMetrics(ctx) + // Registered last so it runs first, like native iterators publishing final metric values + // as they close at task completion. + ctx.addTaskCompletionListener[Unit] { _ => + nestedBytes.set(5L) + nestedRows.set(7L) + nestedPruned.set(11L) + filterRows.set(1000L) + siblingBytes.set(13L) + siblingRows.set(17L) + } + ctx.markTaskCompleted(failure) + + assert(ctx.taskMetrics.inputMetrics.bytesRead == 100L + 5L + 13L) + assert(ctx.taskMetrics.inputMetrics.recordsRead == 1L + 7L + 11L + 17L) + } + } + test("native sort in a non-shuffle stage reports task-level disk spill metrics") { val expectedRecords = 20000L val compressibleValue = "non-shuffle-sort-spill-metrics-" * 8 diff --git a/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDDSuite.scala b/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDDSuite.scala index 99d9fe93e0..8a2267d6ab 100644 --- a/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDDSuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleInputRDDSuite.scala @@ -112,14 +112,20 @@ class CometNativeShuffleInputRDDSuite extends CometTestBase { } } - test("spill reporting is registered before native shuffle input producers") { + test("task metric reporting is registered before native shuffle input producers") { Seq(None, Some(new IllegalStateException("failed native shuffle"))).foreach { failure => val writerDisk = new SQLMetric("writerDisk") val writerMemory = new SQLMetric("writerMemory") val childDisk = new SQLMetric("childDisk") val childMemory = new SQLMetric("childMemory") - val childMetrics = - CometMetricNode(Map("spilled_bytes" -> childDisk, "memory_spilled_bytes" -> childMemory)) + val childBytes = new SQLMetric("childBytes", -1L) + val childRows = new SQLMetric("childRows") + val childMetrics = CometMetricNode( + Map( + "spilled_bytes" -> childDisk, + "memory_spilled_bytes" -> childMemory, + "bytes_scanned" -> childBytes, + "output_rows" -> childRows)) val taskContext = TaskContext.empty() val nestedInput = new RDD[AnyRef](spark.sparkContext, Nil) { override protected def getPartitions: Array[Partition] = Array(new Partition { @@ -130,6 +136,8 @@ class CometNativeShuffleInputRDDSuite extends CometTestBase { context.addTaskCompletionListener[Unit] { _ => childDisk.set(19L) childMemory.set(37L) + childBytes.set(53L) + childRows.set(59L) } Iterator.single(null) } @@ -144,16 +152,6 @@ class CometNativeShuffleInputRDDSuite extends CometTestBase { CometMetricNode(writerMetrics, Seq(childMetrics))) inputRDD.iterator(inputRDD.partitions.head, taskContext) - new CometNativeShuffleWriter[Int, Any]( - NativeShuffleSpec(null, childMetrics, null), - null, - Nil, - writerMetrics, - 1, - 0, - 0L, - taskContext, - null) taskContext.addTaskCompletionListener[Unit] { _ => writerDisk.set(23L) writerMemory.set(41L) @@ -162,6 +160,8 @@ class CometNativeShuffleInputRDDSuite extends CometTestBase { assert(taskContext.taskMetrics.diskBytesSpilled == 42L) assert(taskContext.taskMetrics.memoryBytesSpilled == 78L) + assert(taskContext.taskMetrics.inputMetrics.bytesRead == 53L) + assert(taskContext.taskMetrics.inputMetrics.recordsRead == 59L) } } @@ -186,7 +186,7 @@ class CometNativeShuffleInputRDDSuite extends CometTestBase { inputRDDs = Seq.empty, numPartitionsParam = numPartitions, shuffleScanIndices = Set.empty, - spillMetricNode = CometMetricNode(writerMetrics, Seq(childMetricNode)), + taskMetricNode = CometMetricNode(writerMetrics, Seq(childMetricNode)), perPartitionByKey = perPartitionByKey) val execContext = NativeExecContext( inputs = Seq.empty, @@ -196,8 +196,7 @@ class CometNativeShuffleInputRDDSuite extends CometTestBase { encryptedFilePaths = Seq.empty, commonByKey = Map.empty, perPartitionByKey = perPartitionByKey, - shuffleScanIndices = Set.empty, - hasScanInput = false) + shuffleScanIndices = Set.empty) val spec = NativeShuffleSpec(Operator.getDefaultInstance, childMetricNode, execContext) val dep = new CometShuffleDependency[Int, ColumnarBatch, ColumnarBatch]( _rdd = rdd,