diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml index 2d9bd07f343..b176bc829ad 100644 --- a/.github/workflows/pr_build_linux.yml +++ b/.github/workflows/pr_build_linux.yml @@ -535,6 +535,7 @@ jobs: org.apache.comet.CometVariantProjectionSuite org.apache.spark.sql.CometVariantShreddingSuite org.apache.comet.CometIcebergNativeSuite + org.apache.comet.CometIcebergHdfsSuite org.apache.comet.CometBroadcastBuildSideSuite org.apache.comet.CometIcebergEncryptionSuite org.apache.comet.CometIcebergRewriteActionSuite diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml index 1f70acdeb2b..a9b1a12832d 100644 --- a/.github/workflows/pr_build_macos.yml +++ b/.github/workflows/pr_build_macos.yml @@ -173,6 +173,7 @@ jobs: org.apache.comet.CometVariantProjectionSuite org.apache.spark.sql.CometVariantShreddingSuite org.apache.comet.CometIcebergNativeSuite + org.apache.comet.CometIcebergHdfsSuite org.apache.comet.CometBroadcastBuildSideSuite org.apache.comet.CometIcebergEncryptionSuite org.apache.comet.CometIcebergRewriteActionSuite diff --git a/docs/source/contributor-guide/roadmap.md b/docs/source/contributor-guide/roadmap.md index 06bff3435dd..2724910e85f 100644 --- a/docs/source/contributor-guide/roadmap.md +++ b/docs/source/contributor-guide/roadmap.md @@ -68,9 +68,8 @@ Comet's native Iceberg scans read V3 tables, including encrypted tables ([#4991] ([#5853]). We want to add the remaining V3 features so that these scans don't fall back to Spark: row lineage metadata columns, column default values, and the new V3 types (`variant`, `geometry`, `geography`, and `unknown`). The work is tracked in [#3376], and upstream `iceberg-rust` support in [iceberg-rust #2411]. Native Iceberg scans -also don't support HDFS-backed tables today: Comet's native Iceberg storage layer handles only local files, S3 and -S3-compatible stores, GCS, and OSS, and would need an HDFS `StorageFactory` upstream in `iceberg-storage-opendal`. -We're scoping what that work would take. +read HDFS-backed tables through iceberg-rust's experimental `hdfs-native` backend, but fall back to Spark on clusters +with Kerberos, because that backend cannot authenticate with delegation tokens yet. [#3376]: https://github.com/apache/datafusion-comet/issues/3376 [#4991]: https://github.com/apache/datafusion-comet/pull/4991 diff --git a/docs/source/contributor-guide/s3-credential-provider-design.md b/docs/source/contributor-guide/s3-credential-provider-design.md index a30548566bd..416e2f8e50f 100644 --- a/docs/source/contributor-guide/s3-credential-provider-design.md +++ b/docs/source/contributor-guide/s3-credential-provider-design.md @@ -142,7 +142,7 @@ The 5-minute fallback on the Iceberg path bounds how long reqsign reuses a crede ## Property-bag handling on the Iceberg path -The full unfiltered FileIO property bag crosses JNI as `catalog_properties`. The storage-prefix filter (`s3.`/`gcs.`/`adls.`/`client.`/`opendal.`) is applied native-side in `iceberg_common.rs::build_file_io` immediately before `FileIOBuilder.with_prop`. This means the bridge sees `credentials.uri`, OAuth tokens, and any vendor-custom keys with no parallel field on the operator and no driver-side broadcast. Vendors set their own keys on the catalog config and read them back inside `initialize(Map)`. +The full unfiltered FileIO property bag crosses JNI as `catalog_properties`. The storage-prefix filter (`s3.`/`gcs.`/`adls.`/`client.`/`opendal.`/`hdfs.`/`hadoop.`) is applied native-side in `iceberg_common.rs::build_file_io` immediately before `FileIOBuilder.with_prop`. This means the bridge sees `credentials.uri`, OAuth tokens, and any vendor-custom keys with no parallel field on the operator and no driver-side broadcast. Vendors set their own keys on the catalog config and read them back inside `initialize(Map)`. `IcebergScanExec` derives a redacting `Debug`, and the `FileIO` cache key's `Debug` omits the property bag, so plan dumps and tracing do not leak it. diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index 7c98b3fb93a..fdde7d9cd17 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -166,8 +166,8 @@ attaches its own worker threads, so a worker that has read from HDFS can crash t the HDFS read itself, typically while an unrelated query is running. ``` -Native Iceberg scans do not support HDFS-backed tables; those scans fall back to Spark. See the -[Iceberg](iceberg.md) page. +Native Iceberg scans read HDFS through a different client, iceberg-rust's `hdfs-native` backend. See +the [Iceberg](iceberg.md#hdfs) page. ### Building Comet with HDFS support diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index 6a9d767c25e..4c603f79b86 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -138,8 +138,7 @@ safe partial pruning is tracked in [#5883](https://github.com/apache/datafusion- - Local filesystem - S3-compatible storage (AWS S3, MinIO) - Google Cloud Storage (`gs`) and Alibaba Cloud OSS (`oss`) - -HDFS-backed tables are not supported by the native Iceberg reader and fall back to Spark. +- HDFS (`hdfs`), experimental (see [HDFS](#hdfs)) ### REST Catalog @@ -194,6 +193,47 @@ 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). The region is not auto-detected: when neither the catalog (`client.region` or `s3.region`) nor the executor environment (`AWS_REGION` or `AWS_DEFAULT_REGION`) supplies one, Comet uses `us-east-1`, so set it for AWS buckets in any other region. 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. +### HDFS + +The native reader reads HDFS through iceberg-rust's experimental `hdfs-native` backend, a Rust HDFS +client that uses neither libhdfs nor the JVM. The client reads `core-site.xml` and `hdfs-site.xml` +from `$HADOOP_CONF_DIR`, or from `$HADOOP_HOME/etc/hadoop`, on each executor. Of the session's +Hadoop configuration (the driver's XML and `spark.hadoop.*`), Comet forwards the settings the +client reads: `dfs.client.use.datanode.hostname`, `dfs.client.socket-timeout`, +`dfs.client.failover.max.attempts`, `dfs.client.failover.sleep.base.millis` and +`dfs.client.failover.sleep.max.millis` when the session sets them, and the NameNodes of an HA +nameservice. A location such as `hdfs://ns1/warehouse` has no port, and Comet passes the client +the NameNodes that the session's Hadoop configuration declares for `ns1` with +`dfs.ha.namenodes.ns1` and `dfs.namenode.rpc-address.ns1.`. A portless location that the +configuration does not declare is a single NameNode on Hadoop's default port, 8020, as it is to +the JVM client. A catalog property under `hadoop.` sets any client key directly and wins over the +forwarded value. The client does not read the JVM client's other settings, among them short-circuit +and hedged reads, `dfs.client.retry.*`, `ipc.client.*`, `hadoop.rpc.protection` and the +per-nameservice `dfs.client.failover.*.` keys. Data transfer is encrypted when the NameNode +requires it (`dfs.encrypt.data.transfer`), as for the JVM client. + +A scan of HDFS files falls back to Spark when `hadoop.security.authentication` is not `simple`, +because the client cannot authenticate with the delegation tokens that Spark executors use on a +Kerberos cluster. It also falls back when a file location has no port and names a nameservice that +the Hadoop configuration does not declare, or when it has no authority (`hdfs:///path`), and when a +data or delete file lies in an HDFS encryption zone: the client is built without KMS support and +cannot decrypt it. Comet checks that at planning, on clusters whose NameNode advertises a key +provider, with one directory listing per parent of the data directories. A listing is kept per +table for an hour, or for `spark.sql.metadataCacheTTLSeconds` when that is set, so a table +re-created at the same path lists again on its first scan, and a new partition lists its parent +again; a scan that would list more than a million directory entries falls back, and later scans +continue from the listings kept, which hold a million entries at most. + +On a cluster without Kerberos, the client connects as `HADOOP_USER_NAME`, or as the executor's +operating system user when that variable is unset, rather than as the Spark user. Comet therefore +claims an HDFS scan only when the executors are known to read as the Spark user: set +`spark.executorEnv.HADOOP_USER_NAME` to the user that submits the job, when the application is +submitted (Comet reads it from the SparkContext's configuration; a value set on a session +afterwards never reaches the executors). In local mode, the JVM's `HADOOP_USER_NAME` or operating +system user must be that user. Otherwise the scan falls back with a reason that names the setting. +To read HDFS tables with Spark's Iceberg reader and keep the native scan for other storage, set +`spark.comet.scan.icebergNative.hdfs.enabled=false`. Writes to HDFS use Iceberg's own writer. + ### Current limitations The following scenarios will fall back to the JVM Iceberg reader: @@ -210,6 +250,12 @@ The following scenarios will fall back to the JVM Iceberg reader: - Queries that use `input_file_name()`, `input_file_block_start()` or `input_file_block_length()` - Scans whose data or delete files span more than one S3 bucket (the native reader uses one object-store configuration per scan) +- HDFS scans on a cluster with Kerberos, or whose executors are not known to read as the Spark + user; HDFS file locations that name a nameservice without NameNodes or that have no authority; + and HDFS files in an encryption zone (see [HDFS](#hdfs)) +- Scans whose data or delete files are on a different storage backend than the table metadata, for + example a local or REST-served metadata location with `write.data.path` on S3 or HDFS. The native + reader opens every file of a scan through one storage, chosen from the metadata location - Scans where Iceberg reports an output ordering (for example with `spark.sql.iceberg.planning.preserve-data-ordering=true`). Spark may then skip the sort above the scan, and the native reader does not preserve file order @@ -788,16 +834,19 @@ the [S3](datasources.md#s3-limitations) and [Azure](datasources.md#azure-limitat Which entries apply depends on which native paths are enabled: -| Native path | Setting | Default | -| -------------- | ---------------------------------------- | ------- | -| Iceberg scan | `spark.comet.scan.icebergNative.enabled` | `true` | -| Iceberg writer | `spark.comet.write.iceberg.enabled` | `false` | +| Native path | Setting | Default | +| -------------------- | --------------------------------------------- | ------- | +| Iceberg scan | `spark.comet.scan.icebergNative.enabled` | `true` | +| Iceberg scan of HDFS | `spark.comet.scan.icebergNative.hdfs.enabled` | `true` | +| Iceberg writer | `spark.comet.write.iceberg.enabled` | `false` | ### Reads that differ from Spark -- **Data files with neither field ids nor a name mapping.** Spark reads every nested field of such - a file as NULL, and the native scan returns the values stored in the file. A read of only some - of a struct's fields fails instead +- **Data files with neither field ids nor a name mapping.** Neither reader can match such a + file's nested fields to the table's, so a struct's fields read as NULL in both. The containers + differ: Spark's reader returns NULL for an optional struct itself, where the native scan + returns a struct of NULL fields, and an empty list for a list of structs, where the native scan + returns the list with each element's fields NULL ([#6794](https://github.com/apache/datafusion-comet/issues/6794)). - **`VERSION AS OF` reads of a column renamed after the snapshot.** The renamed column reads as NULL, and two columns that swapped names return each other's values diff --git a/native/Cargo.lock b/native/Cargo.lock index eb93db1252b..aa6af5f3079 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", ] @@ -1168,6 +1179,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" @@ -1299,7 +1319,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]] @@ -1395,7 +1424,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.5", ] [[package]] @@ -1603,6 +1643,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" @@ -1621,6 +1667,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" @@ -1781,7 +1842,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]] @@ -2017,7 +2087,7 @@ dependencies = [ "jni", "lazy_static", "libc", - "libloading", + "libloading 0.8.9", "log", "log4rs", "mimalloc", @@ -2876,6 +2946,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" @@ -2926,6 +3005,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" @@ -3214,6 +3305,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" @@ -3233,7 +3352,7 @@ dependencies = [ "cfg-if", "js-sys", "libc", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", "wasm-bindgen", ] @@ -3360,6 +3479,48 @@ dependencies = [ "foldhash 0.2.0", ] +[[package]] +name = "hdfs-native" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5141c4b745cab57a0018daa9c2759c92308bcf4e41e0c617d05fbf9639c16754" +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.3", + "regex", + "roxmltree", + "socket2", + "thiserror", + "tokio", + "tokio-util", + "url", + "uuid", + "whoami", +] + [[package]] name = "hdfs-sys" version = "0.3.1" @@ -3568,7 +3729,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=b154603a710d730d2f43f8af4db257cc137237ef#b154603a710d730d2f43f8af4db257cc137237ef" +source = "git+https://github.com/apache/iceberg-rust?rev=de0a6ac8471da43f63382d66fac27d7e56fd273e#de0a6ac8471da43f63382d66fac27d7e56fd273e" dependencies = [ "aes-gcm", "anyhow", @@ -3626,7 +3787,7 @@ dependencies = [ [[package]] name = "iceberg-property-macro" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=b154603a710d730d2f43f8af4db257cc137237ef#b154603a710d730d2f43f8af4db257cc137237ef" +source = "git+https://github.com/apache/iceberg-rust?rev=de0a6ac8471da43f63382d66fac27d7e56fd273e#de0a6ac8471da43f63382d66fac27d7e56fd273e" dependencies = [ "proc-macro2", "quote", @@ -3636,7 +3797,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=b154603a710d730d2f43f8af4db257cc137237ef#b154603a710d730d2f43f8af4db257cc137237ef" +source = "git+https://github.com/apache/iceberg-rust?rev=de0a6ac8471da43f63382d66fac27d7e56fd273e#de0a6ac8471da43f63382d66fac27d7e56fd273e" dependencies = [ "anyhow", "async-trait", @@ -3649,6 +3810,8 @@ dependencies = [ "reqsign-aws-v4", "reqsign-core", "serde", + "tokio", + "tracing", "typetag", "url", ] @@ -3810,10 +3973,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.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b5cf4372d022092674a83b8febf5865f7a573cc0c5baec5e49c879b6a20978f" +dependencies = [ + "block-padding 0.4.2", + "hybrid-array", +] + [[package]] name = "inventory" version = "0.3.25" @@ -3955,7 +4128,7 @@ dependencies = [ "java-locator", "jni-macros", "jni-sys", - "libloading", + "libloading 0.8.9", "log", "simd_cesu8", "thiserror", @@ -4113,6 +4286,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.9" @@ -4148,6 +4331,15 @@ dependencies = [ "cc", ] +[[package]] +name = "libredox" +version = "0.1.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61ff90caf6077a803a240f62fdbe88645a890bbca49ef8174c3cb0404362171d" +dependencies = [ + "libc", +] + [[package]] name = "link-section" version = "0.19.3" @@ -4321,7 +4513,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" dependencies = [ "libc", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", "windows-sys 0.61.2", ] @@ -4490,6 +4682,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 +4807,7 @@ dependencies = [ "opendal-service-fs", "opendal-service-gcs", "opendal-service-hdfs", + "opendal-service-hdfs-native", "opendal-service-oss", "opendal-service-s3", ] @@ -4765,6 +4976,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" @@ -5112,8 +5337,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", @@ -5961,6 +6186,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" @@ -6120,7 +6354,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213" dependencies = [ - "cipher", + "cipher 0.4.4", ] [[package]] @@ -7179,6 +7413,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" @@ -7188,6 +7431,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.129" @@ -7299,6 +7551,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" @@ -7395,7 +7660,16 @@ version = "0.52.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" dependencies = [ - "windows-targets", + "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]] @@ -7413,14 +7687,31 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" dependencies = [ - "windows_aarch64_gnullvm", - "windows_aarch64_msvc", - "windows_i686_gnu", - "windows_i686_gnullvm", - "windows_i686_msvc", - "windows_x86_64_gnu", - "windows_x86_64_gnullvm", - "windows_x86_64_msvc", + "windows_aarch64_gnullvm 0.52.6", + "windows_aarch64_msvc 0.52.6", + "windows_i686_gnu 0.52.6", + "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]] @@ -7429,48 +7720,96 @@ 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.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.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.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.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.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.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 0e936ba733e..10f5934beb9 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -86,8 +86,8 @@ aws-sdk-sts = { version = "1.114.0", default-features = false } # Pinned to a revision rather than a release because no published iceberg-rust # version is on our arrow: 0.10.1 requires arrow/parquet ^58. Move to a plain # version requirement once a release ships on arrow 59. Bump policy: #5645. -iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "b154603a710d730d2f43f8af4db257cc137237ef" } -iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "b154603a710d730d2f43f8af4db257cc137237ef", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } +iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "de0a6ac8471da43f63382d66fac27d7e56fd273e" } +iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "de0a6ac8471da43f63382d66fac27d7e56fd273e", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls", "opendal-hdfs-native"] } reqsign-core = "3" [profile.release] diff --git a/native/core/src/execution/operators/iceberg_common.rs b/native/core/src/execution/operators/iceberg_common.rs index a17eb076e0d..f3f2286bb8a 100644 --- a/native/core/src/execution/operators/iceberg_common.rs +++ b/native/core/src/execution/operators/iceberg_common.rs @@ -45,8 +45,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. `opendal.` carries -/// iceberg-storage-opendal's own settings, such as `opendal.io-timeout-ms`. -const STORAGE_PROPERTY_PREFIXES: &[&str] = &["s3.", "gcs.", "adls.", "client.", "opendal."]; +/// iceberg-storage-opendal's own settings, such as `opendal.io-timeout-ms`. `hdfs.` and `hadoop.` +/// carry the HDFS backend's NameNode declarations and Hadoop client settings. +const STORAGE_PROPERTY_PREFIXES: &[&str] = &[ + "s3.", "gcs.", "adls.", "client.", "opendal.", "hdfs.", "hadoop.", +]; /// Pick an OpenDAL storage backend for a URI whose scheme `builtin_storage_schemes` lists for /// `access_mode`, or that is an opted-in S3-compliant alias written in lowercase; anything else is @@ -79,6 +82,7 @@ pub(crate) fn storage_factory_for( "memory" => Ok((Arc::new(OpenDalStorageFactory::Memory), true)), "gs" => Ok((Arc::new(OpenDalStorageFactory::Gcs), true)), "oss" => Ok((Arc::new(OpenDalStorageFactory::Oss), true)), + "hdfs" => Ok((Arc::new(OpenDalStorageFactory::HdfsNative), true)), // s3, s3a, and any opted-in s3-compliant alias (e.g. blob) route to the S3 backend. Listed // last so the built-in backends above stay authoritative even if one of their schemes is // also named in `fs.comet.s3Compliant.schemes`. An alias additionally gets a wrapper that @@ -302,9 +306,10 @@ fn cached_file_io( /// mode; the JVM read and write gates load these lists over JNI. `memory` is write-only: an OpenDAL /// memory backend is a fresh empty in-process store the write path assembles manifests in, so a /// read finds nothing. `oss` is read-only: no `oss.*` property is forwarded and no test covers it. +/// `hdfs` is read-only: the write path has not been tested against it. pub(crate) fn builtin_storage_schemes(access_mode: AccessMode) -> &'static [&'static str] { match access_mode { - AccessMode::Read => &["file", "gs", "oss", "s3", "s3a"], + AccessMode::Read => &["file", "gs", "hdfs", "oss", "s3", "s3a"], AccessMode::Write => &["file", "memory", "gs", "s3", "s3a"], } } @@ -847,6 +852,158 @@ mod tests { ); } + #[test] + fn hdfs_scheme_is_readable_but_not_writable() { + assert!(factory_result("hdfs://nn:8020/db/table", AccessMode::Read).is_ok()); + let err = factory_result("hdfs://nn:8020/db/table", AccessMode::Write).unwrap_err(); + assert!( + err.contains("Unsupported storage scheme: hdfs"), + "unexpected error: {err}" + ); + } + + /// The JVM declares a nameservice as `hdfs.name-node.`, and a catalog can declare one with + /// Hadoop's own keys under `hadoop.`. + #[test] + fn hdfs_properties_reach_the_file_io() { + let props = HashMap::from([ + ("hdfs.name-node.ns1".to_string(), "nn1:8020".to_string()), + ("hadoop.dfs.ha.namenodes.ns2".to_string(), "a".to_string()), + ( + "hadoop.dfs.namenode.rpc-address.ns2.a".to_string(), + "nn2:8020".to_string(), + ), + ]); + let (file_io, _) = + build_file_io(&props, "hdfs://ns2/warehouse", "", AccessMode::Read).unwrap(); + for (key, value) in &props { + assert_eq!(file_io.config().get(key), Some(value), "{key}"); + } + } + + /// The cache is shared per executor, and an HDFS `FileIO` keeps its NameNode clients for life. + /// Two catalogs that differ only in how they reach HDFS must therefore never share a key, or + /// the second would be served by the first one's clients. A regression pin on the key, which + /// hashes every property. + #[test] + fn cache_key_separates_hdfs_catalogs() { + let table = "hdfs://nameservice1/warehouse/db/t"; + let props = |name_node: &str, random_order: &str| { + HashMap::from([ + // What the JVM declares for an `hdfs://nameservice1/...` table. + ( + "hdfs.name-node.nameservice1".to_string(), + name_node.to_string(), + ), + ( + "hadoop.dfs.client.failover.random.order".to_string(), + random_order.to_string(), + ), + ]) + }; + let key = |props: &HashMap| { + FileIoCacheKey::new(props, table, "cat", AccessMode::Read) + }; + let base = props("nn1:8020,nn2:8020", "true"); + + // Control: the same properties always give the same key, so the inequalities below are + // down to the one property that changed. + assert!(key(&base).is_some()); + assert_eq!(key(&base), key(&base.clone())); + + // Only the declaration differs: another NameNode, another list, the same list reordered + // (which changes the failover order), and no list at all. + for other in [ + props("nn1:8020,nn3:8020", "true"), + props("nn1:8020", "true"), + props("nn2:8020,nn1:8020", "true"), + ] { + assert_ne!(key(&base), key(&other), "{other:?}"); + } + let mut without_name_node = base.clone(); + without_name_node.remove("hdfs.name-node.nameservice1"); + assert_ne!(key(&base), key(&without_name_node)); + + // Only the `hadoop.dfs.client.failover.*` overrides differ: a changed value, an added + // override and a removed one. + let mut other_failover = base.clone(); + other_failover.insert( + "hadoop.dfs.client.failover.proxy.provider.nameservice1".to_string(), + "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider".to_string(), + ); + let mut without_failover = base.clone(); + without_failover.remove("hadoop.dfs.client.failover.random.order"); + for other in [ + props("nn1:8020,nn2:8020", "false"), + other_failover, + without_failover, + ] { + assert_ne!(key(&base), key(&other), "{other:?}"); + } + } + + /// Builds the real HDFS storage from one property the way a `FileIO` does on first use. + /// Parsing is all it does: an operator, and with it a NameNode connection, is only made by + /// the first storage call, so this needs neither a NameNode nor a runtime. The storage is + /// inspected through its serde form, so this is a contract test on the pinned iceberg-rust + /// revision and its config field names. + fn build_hdfs_storage(key: &str, value: &str) -> Result { + let config = + StorageConfig::from_props(HashMap::from([(key.to_string(), value.to_string())])); + let storage = OpenDalStorageFactory::HdfsNative + .build(&config) + .map_err(|e| e.to_string())?; + Ok(serde_json::to_value(&*storage).expect("the storage serializes")) + } + + /// One Hadoop option of the parsed storage config. + fn parsed_option<'a>(storage: &'a serde_json::Value, key: &str) -> Option<&'a str> { + storage + .pointer("/HdfsNative/config/options") + .and_then(|options| options.get(key)) + .and_then(serde_json::Value::as_str) + } + + /// The contract `CometIcebergNativeScan.hadoopToIcebergHdfsProperties` relies on: it declares + /// an HA nameservice as `hdfs.name-node.`, and iceberg-rust expands that into + /// Hadoop's own `dfs.ha.namenodes.` / `dfs.namenode.rpc-address..`, the keys its + /// resolver reads for a portless path authority. A portless entry is refused when the + /// storage is built, naming the key and the entry. + #[test] + fn hdfs_nameservice_declaration_is_parsed_by_the_real_storage() { + for input in [ + "nn1:8020,nn2:8020", + // Whitespace around entries, a trailing slash and the `hdfs://` prefix are tolerated. + " hdfs://nn1:8020/ , nn2:8020 ", + ] { + let storage = build_hdfs_storage("hdfs.name-node.ns1", input) + .unwrap_or_else(|e| panic!("`{input}` should be accepted: {e}")); + assert_eq!( + parsed_option(&storage, "dfs.ha.namenodes.ns1"), + Some("nn0,nn1"), + "input `{input}`: {storage}" + ); + assert_eq!( + parsed_option(&storage, "dfs.namenode.rpc-address.ns1.nn0"), + Some("nn1:8020"), + "input `{input}`: {storage}" + ); + assert_eq!( + parsed_option(&storage, "dfs.namenode.rpc-address.ns1.nn1"), + Some("nn2:8020"), + "input `{input}`: {storage}" + ); + } + + // A portless entry would only fail at the first I/O, so it is refused here, and the + // error names the offending key and entry rather than the whole list. + let err = build_hdfs_storage("hdfs.name-node.ns1", "nn1:8020,nn2").unwrap_err(); + assert!( + err.contains("hdfs.name-node.ns1") && err.contains("nn2") && !err.contains("nn1:8020"), + "unexpected error: {err}" + ); + } + #[test] fn memory_scheme_is_writable_but_not_readable() { // The write path assembles manifests in a fresh in-process memory store. A read against @@ -948,7 +1105,7 @@ mod tests { #[test] fn unknown_scheme_is_rejected() { - let err = factory_result("hdfs://nn/db/table", AccessMode::Read).unwrap_err(); + let err = factory_result("viewfs://cluster/db/table", AccessMode::Read).unwrap_err(); assert!( err.contains("Unsupported storage scheme"), "unexpected error: {err}" @@ -997,7 +1154,7 @@ mod tests { #[test] fn unlisted_schemes_are_rejected_for_both_modes() { let unlisted = [ - "hdfs", "abfs", "abfss", "wasb", "wasbs", "gcs", "http", "https", "azure", + "viewfs", "abfs", "abfss", "wasb", "wasbs", "gcs", "http", "https", "azure", ]; for mode in [AccessMode::Read, AccessMode::Write] { for scheme in unlisted { diff --git a/native/core/src/execution/operators/iceberg_partition_path.rs b/native/core/src/execution/operators/iceberg_partition_path.rs index c9ead454b0f..dba99bc2b6f 100644 --- a/native/core/src/execution/operators/iceberg_partition_path.rs +++ b/native/core/src/execution/operators/iceberg_partition_path.rs @@ -152,7 +152,7 @@ const NULL: &str = "null"; /// | `timestamp` | `1969-12-31T23:59:58.5` | `1969-12-31 23:59:58.500` | /// | `timestamptz` | `1969-12-31T23:59:58.5+00:00` | panics for a negative value with a sub-second part; otherwise `1969-12-31 23:59:58.500 UTC` | /// | `binary` / `fixed` | base64 | uppercase hex | -/// | `float` / `double` | `1.0`, `1.0E20` | `1`, `100000000000000000000` | +/// | `float` / `double` | `1.0`, `1.0E20` | `1.0`, `1.0E20` since #3358; `Double::MAX` still renders as 309 digits (see below) | /// /// The nanosecond timestamp types get the same treatment for the same reason. Spark cannot plan a /// write to them today, and `CometIcebergNativeWrite`'s column-type rule declines them as well; the diff --git a/pom.xml b/pom.xml index c3d3912b537..93d99760ea1 100644 --- a/pom.xml +++ b/pom.xml @@ -89,7 +89,10 @@ 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 @@ -108,7 +111,7 @@ under the License. + it down to 3.3.x (SDK v1). --> 3.4.1 + + org.bouncycastle + bcprov-jdk18on + 1.78.1 + test + diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index e67a4fc225b..02a0c82425f 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -121,6 +121,16 @@ object CometConf extends ShimCometConf { .booleanConf .createWithDefault(true) + val COMET_ICEBERG_NATIVE_HDFS_ENABLED: ConfigEntry[Boolean] = + conf("spark.comet.scan.icebergNative.hdfs.enabled") + .category(CATEGORY_SCAN) + .doc( + "Whether the native Iceberg scan reads tables on HDFS, through iceberg-rust's " + + "experimental `hdfs-native` backend. When false, a scan that reads HDFS files falls " + + "back to Spark's Iceberg reader, and scans of other storage are unaffected.") + .booleanConf + .createWithDefault(true) + val COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.write.iceberg.splitOperator.enabled") .category(CATEGORY_TESTING) diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala index ff318e0dfb6..58278b488aa 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -991,6 +991,19 @@ object IcebergReflection extends Logging { } } + /** + * The table's UUID, from `Table.uuid()` or its current metadata. A table dropped and created + * again at the same location gets a new one. + */ + def getTableUuid(table: Any): Option[String] = { + def uuidOf(target: Any): Option[String] = + findMethodInHierarchy(target.getClass, "uuid").flatMap { method => + try Option(method.invoke(target)).map(_.toString) + catch { case NonFatal(_) => None } + } + uuidOf(table).orElse(getTableMetadata(table).flatMap(uuidOf)) + } + /** * Gets the metadata file location from an Iceberg table. * 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 955db40ce42..61f7c265e20 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -41,7 +41,7 @@ import org.apache.spark.sql.execution.{FileSourceScanExec, InSubqueryExec, Spark import org.apache.spark.sql.execution.datasources.HadoopFsRelation import org.apache.spark.sql.execution.datasources.v2.BatchScanExec import org.apache.spark.sql.execution.datasources.v2.csv.CSVScan -import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf} import org.apache.spark.sql.types._ import org.apache.comet.{CometConf, DataTypeSupport, NativeBase} @@ -640,6 +640,13 @@ case class CometScanRule(session: SparkSession) hadoopS3Options, icebergDataBucket) + // Declares the HDFS nameservices the data/delete files name, and forwards the client + // settings the session set. Merged before the catalog's properties, so a catalog's own + // declaration or setting wins. + val hdfsProperties = CometIcebergNativeScan.hadoopToIcebergHdfsProperties( + hadoopConf, + taskValidation.hdfsAuthorities) + // Forward the full FileIO property bag (including credentials.uri, OAuth tokens, // tenant-id, etc.) so a CometS3CredentialProvider can see everything LoadTableResponse // returned. The storage-prefix narrowing happens native-side just before @@ -655,10 +662,18 @@ 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 ++ - hadoopS3Options - .get(COMET_S3_COMPLIANT_SCHEMES_KEY) - .map(COMET_S3_COMPLIANT_SCHEMES_KEY -> _) + // A nameservice the catalog declares with Hadoop's own keys is not also given a + // default-port host declaration. + val derivedHdfsProperties = hdfsProperties.filterNot { case (key, _) => + key.startsWith("hdfs.name-node.") && + fileIOProperties.contains( + s"hadoop.dfs.ha.namenodes.${key.stripPrefix("hdfs.name-node.")}") + } + val catalogProperties = + hadoopDerivedProperties ++ derivedHdfsProperties ++ fileIOProperties ++ + hadoopS3Options + .get(COMET_S3_COMPLIANT_SCHEMES_KEY) + .map(COMET_S3_COMPLIANT_SCHEMES_KEY -> _) val result = CometIcebergNativeScanMetadata.extract( scanExec.scan, @@ -703,6 +718,22 @@ case class CometScanRule(session: SparkSession) false } + // The native scan opens every data and delete file through one storage, built from the + // metadata location's scheme (`load_file_io` in iceberg_scan.rs), so a file on another + // backend fails at open time: an `hdfs://` file handed to the local storage, a local path + // to the S3 one. Declined here instead; the multi-bucket check above is the same idea one + // level down. + val metadataFamily = CometScanRule.storageFamily(metadataUri.getScheme) + val foreignFamilies = + taskValidation.fileSchemes.map(CometScanRule.storageFamily) - metadataFamily + val storageFamiliesConsistent = foreignFamilies.isEmpty || { + fallbackReasons += "Iceberg scan reads data/delete files on storage " + + s"${foreignFamilies.toSeq.sorted.map(f => s"'$f'").mkString(", ")} while the table " + + s"metadata is on '$metadataFamily'; Comet's native Iceberg reader opens every file " + + "of a scan through one storage backend, chosen from the metadata location" + false + } + // Now perform all validation using the pre-extracted metadata // Check if table uses a FileIO implementation compatible with iceberg-rust. // Comet's native reader bypasses Iceberg Java's FileIO entirely and reads through @@ -924,6 +955,76 @@ case class CometScanRule(session: SparkSession) case None => true } + // iceberg-rust's HDFS client cannot use an executor's delegation tokens: opendal builds + // every client for a synthetic nameservice, and the client looks tokens up by that name. + // A portless authority must be a declared nameservice (see hadoopToIcebergHdfsProperties). + def hdfsScanSupported(): Boolean = { + val authentication = hadoopConf.getTrimmed("hadoop.security.authentication", "simple") + val secure = !authentication.equalsIgnoreCase("simple") + if (secure) { + fallbackReasons += "Iceberg scan reads HDFS with " + + s"hadoop.security.authentication=$authentication, but Comet's native Iceberg " + + "reader cannot authenticate to HDFS with delegation tokens" + } + val undeclared = CometIcebergNativeScan.undeclaredHdfsNameservice( + taskValidation.hdfsAuthorities, + metadata.catalogProperties) + undeclared.foreach { ns => + fallbackReasons += s"Iceberg scan reads HDFS authority '$ns', which has no port and " + + "which the Hadoop configuration does not declare as a nameservice " + + s"(dfs.ha.namenodes.$ns, and a host:port dfs.namenode.rpc-address.$ns. for " + + "each NameNode)" + } + // The client connects as HADOOP_USER_NAME, or as the executor's OS user, never as the + // Spark user the JVM reader runs as (opendal exposes no user setting). Claimed only + // when the executors are known to read as the Spark user: HADOOP_USER_NAME set to it + // in their environment, or, in local mode, the JVM's own user. The setting is read + // from the SparkContext's configuration, which launched the executors: a session can + // hold the key in its SQLConf (`SET`, or a builder option on an existing context) + // without the executors ever seeing it. + val sparkUser = session.sparkContext.sparkUser + val executorEnvKey = "spark.executorEnv.HADOOP_USER_NAME" + val executorUser = Option(session.sparkContext.getConf.get(executorEnvKey, null)) + .orElse { + if (session.sparkContext.isLocal) { + Some(sys.env.getOrElse("HADOOP_USER_NAME", System.getProperty("user.name"))) + } else { + None + } + } + val identityKnown = executorUser.contains(sparkUser) + if (!identityKnown) { + fallbackReasons += "Iceberg scan reads HDFS as the executors' user " + + s"${executorUser.map(u => s"'$u'").getOrElse("(unknown)")} rather than the Spark " + + s"user '$sparkUser'; set $executorEnvKey=$sparkUser so Comet's native Iceberg " + + "reader connects as the Spark user" + } + // The client is built without KMS support and rejects every file of an encryption + // zone. Probed only once the scan is otherwise eligible, so an undeclared nameservice + // is not also dialed. + val encryptionZoneFallback = + if (secure || undeclared.nonEmpty || !identityKnown) None + else { + CometIcebergNativeScan.hdfsEncryptionZoneFallback( + hadoopConf, + taskValidation.hdfsDirectories, + tableUuid = tableOpt.flatMap(IcebergReflection.getTableUuid), + ttlSeconds = session.sessionState.conf + .getConf(StaticSQLConf.METADATA_CACHE_TTL_SECONDS)) + } + encryptionZoneFallback.foreach(fallbackReasons += _) + !secure && undeclared.isEmpty && identityKnown && encryptionZoneFallback.isEmpty + } + val hdfsSupported = taskValidation.hdfsAuthorities.isEmpty || { + if (COMET_ICEBERG_NATIVE_HDFS_ENABLED.get()) { + hdfsScanSupported() + } else { + fallbackReasons += "Iceberg scan reads HDFS files, and " + + s"${COMET_ICEBERG_NATIVE_HDFS_ENABLED.key} is false" + false + } + } + if (!taskValidation.allParquet) { fallbackReasons += "Iceberg scan contains non-Parquet files (ORC or Avro). " + "Comet only supports Parquet files in Iceberg tables" @@ -1135,9 +1236,11 @@ case class CometScanRule(session: SparkSession) if (schemaSupported && fileIOCompatible && formatVersionSupported && defaultValuesSupported && schemaTypesSupported && encryptionKeyLengthSupported && taskValidation.allParquet && allSupportedFilesystems && allLocationsOpenable && - metadataSchemeSupported && partitionTypesSupported && unifiedPartitionTypeSupported && + metadataSchemeSupported && storageFamiliesConsistent && partitionTypesSupported && + unifiedPartitionTypeSupported && transformFunctionsSupported && deleteFileTypesSupported && dppSubqueriesSupported && - nestedFieldsSupported && taskSchemaNamesSupported && nestedNameMappingSupported) { + nestedFieldsSupported && taskSchemaNamesSupported && hdfsSupported && + nestedNameMappingSupported) { CometBatchScanExec( scanExec.clone().asInstanceOf[BatchScanExec], runtimeFilters = scanExec.runtimeFilters, @@ -1475,6 +1578,13 @@ 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]() + // Schemes of the data/delete files the native FileIO opens, for the storage-family check in + // CometScanRule; a schemeless path is local, as iceberg-rust's `scheme_of` reads it. + val fileSchemes = mutable.Set[String]() + // HDFS authorities and directories of the data/delete files, for the HDFS checks in + // CometScanRule. + val hdfsAuthorities = mutable.Set[String]() + val hdfsDirectories = 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 @@ -1488,14 +1598,24 @@ object CometScanRule extends Logging { catch { case _: java.net.URISyntaxException => return } // A schemeless local path routes to iceberg-rust's LocalFs and needs no host. val scheme = uri.getScheme - if (scheme == null) return + if (scheme == null) { + fileSchemes += "file" + return + } if (!isAdmittedIcebergScheme(scheme, s3CompliantSchemes)) { unsupportedSchemes += scheme } else if (!hasOpenableAuthority(uri, s3CompliantSchemes)) { if (hostlessLocation.isEmpty) hostlessLocation = Some(rawPath) } else { - // bucketForUri yields None for non-S3-family URIs, so no scheme re-check is needed here. - NativeConfig.bucketForUri(uri, s3CompliantSchemes).foreach(dataFileBuckets += _) + fileSchemes += scheme + if (scheme == "hdfs") { + hdfsAuthorities += uri.getRawAuthority + val slash = rawPath.lastIndexOf('/') + if (slash > 0) hdfsDirectories += rawPath.substring(0, slash) + } else { + // bucketForUri yields None for non-S3-family URIs, so no scheme re-check is needed. + NativeConfig.bucketForUri(uri, s3CompliantSchemes).foreach(dataFileBuckets += _) + } } } @@ -1567,9 +1687,23 @@ object CometScanRule extends Logging { nonIdentityTransform, deleteFiles, dataFileBuckets.toSet, + fileSchemes.toSet, + hdfsAuthorities.toSet, + hdfsDirectories.toSet, hostlessLocation, requiredFieldIds) } + + /** + * The native storage backend a data/delete file scheme routes to (`storage_factory_for` in + * iceberg_common.rs): `s3` and `s3a` share one, a schemeless path is local, and every other + * scheme, an S3-compliant alias included, is its own. + */ + private[rules] def storageFamily(scheme: String): String = scheme match { + case null | "file" => "file" + case "s3" | "s3a" => "s3" + case other => other + } } /** @@ -1581,5 +1715,8 @@ case class IcebergTaskValidationResult( nonIdentityTransform: Option[String], deleteFiles: java.util.List[_], dataFileBuckets: Set[String], + fileSchemes: Set[String], + hdfsAuthorities: Set[String], + hdfsDirectories: Set[String], hostlessLocation: Option[String], requiredFieldIds: Set[Int]) 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 4070d785367..ef70389caa1 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 @@ -26,10 +26,14 @@ import java.util.{Base64, UUID} import scala.collection.mutable import scala.jdk.CollectionConverters._ +import scala.util.control.NonFatal import org.json4s.JsonDSL._ import org.json4s.jackson.JsonMethods._ +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.Path +import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys import org.apache.spark.internal.Logging import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.comet.{CometBatchScanExec, CometNativeExec} @@ -655,6 +659,239 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit global.result() ++ promoted.result() } + /** + * The client settings iceberg-rust's HDFS backend reads from its configuration and the JVM + * client reads too (hdfs-native 0.14, `common/config.rs`). The client otherwise reads only the + * executors' `$HADOOP_CONF_DIR`, so a value the session sets would never reach it. Not + * forwarded: the per-nameservice `dfs.client.failover.*.` keys, which the client would look + * up under a nameservice name of opendal's own, and `dfs.data.transfer.protection`, on which + * the client negotiates SASL whenever it is set where the JVM client does so only on a secure + * cluster, which falls back. + */ + private[comet] val forwardedHdfsClientKeys: Seq[String] = Seq( + "dfs.client.use.datanode.hostname", + "dfs.client.socket-timeout", + "dfs.client.failover.max.attempts", + "dfs.client.failover.sleep.base.millis", + "dfs.client.failover.sleep.max.millis") + + /** + * The HDFS properties a scan of `authorities` needs from the Hadoop configuration, for + * iceberg-rust's HDFS backend: nameservice declarations, and the client settings of + * [[forwardedHdfsClientKeys]] that the session set (a value from Hadoop's own `*-default.xml` + * is not a setting, and forwarded would override the executors' XML), as `hadoop.`. + * + * The backend dials an authority with a port as written, but resolves a portless one + * (`hdfs://ns1/...`) only through a nameservice declared in the FileIO properties, never + * through `$HADOOP_CONF_DIR`. A nameservice the Hadoop configuration declares in full, with + * `dfs.ha.namenodes.` and a `host:port` `dfs.namenode.rpc-address..` for each + * NameNode, becomes `hdfs.name-node.` with its NameNodes in declaration order. An authority + * the configuration knows nothing of is a single NameNode on Hadoop's default port, as + * `DFSUtilClient.getNNAddress` dials it for the JVM client. A name in `dfs.nameservices` + * without NameNodes, or with one that has no port, is left out, and + * [[undeclaredHdfsNameservice]] then declines the scan. + */ + def hadoopToIcebergHdfsProperties( + hadoopConf: Configuration, + authorities: Set[String]): Map[String, String] = { + val clientSettings = if (authorities.isEmpty) { + Map.empty[String, String] + } else { + forwardedHdfsClientKeys.flatMap { key => + val sources = Option(hadoopConf.getPropertySources(key)).getOrElse(Array.empty[String]) + val set = sources.nonEmpty && !sources.exists(_.endsWith("-default.xml")) + if (set) Option(hadoopConf.getTrimmed(key)).map(s"hadoop.$key" -> _) else None + }.toMap + } + val logicalNames = hadoopConf.getTrimmedStringCollection("dfs.nameservices").asScala.toSet + val declarations = hdfsNameservices(authorities).flatMap { ns => + val nameNodes = hadoopConf + .getTrimmedStrings(s"dfs.ha.namenodes.$ns") + .map(id => hadoopConf.getTrimmed(s"dfs.namenode.rpc-address.$ns.$id")) + if (nameNodes.nonEmpty) { + if (nameNodes.forall(hasPort)) Some(s"hdfs.name-node.$ns" -> nameNodes.mkString(",")) + else None + } else if (logicalNames.contains(ns)) { + None + } else { + Some( + s"hdfs.name-node.$ns" -> s"$ns:${HdfsClientConfigKeys.DFS_NAMENODE_RPC_PORT_DEFAULT}") + } + }.toMap + declarations ++ clientSettings + } + + /** + * The first portless HDFS authority among `authorities` that `properties` declares neither with + * iceberg-rust's `hdfs.name-node.` nor with Hadoop's `dfs.ha.namenodes.`. + */ + def undeclaredHdfsNameservice( + authorities: Set[String], + properties: Map[String, String]): Option[String] = + hdfsNameservices(authorities).find { ns => + !properties.contains(s"hdfs.name-node.$ns") && + !properties.contains(s"hadoop.dfs.ha.namenodes.$ns") + } + + private def hdfsNameservices(authorities: Set[String]): Seq[String] = + authorities.filterNot(hasPort).toSeq.sorted + + /** + * Listings taken for the encryption-zone check, by filesystem URI, table UUID and parent + * directory: when each was taken, and the encrypted flag of every child, by name. A data + * directory cannot become a zone root once it holds files, and a table dropped and created + * again at the same path, inside a new zone or not, has a new UUID, so a listing is reused + * until it is older than `spark.sql.metadataCacheTTLSeconds`, or than + * [[DefaultEncryptionZoneListingAgeSeconds]] when that is unset. Every child is kept, not only + * the directories one scan reads, so a later scan of other partitions of the table lists + * nothing; what is kept is bounded to [[MaxKeptEncryptionZoneEntries]] children in all, the + * least recently used parents leaving first. Accessed under the map's own lock: a read moves + * the parent to the front. + */ + private val listedParents = + new java.util.LinkedHashMap[(String, String, String), (Long, Map[String, Boolean])]( + 64, + 0.75f, + true) + private var keptEntryCount = 0 + + private def keptListing(key: (String, String, String)): Option[(Long, Map[String, Boolean])] = + listedParents.synchronized(Option(listedParents.get(key))) + + private def keepListing( + key: (String, String, String), + listing: (Long, Map[String, Boolean]), + maxKeptEntries: Int): Unit = listedParents.synchronized { + Option(listedParents.put(key, listing)).foreach(old => keptEntryCount -= old._2.size) + keptEntryCount += listing._2.size + val eldestFirst = listedParents.entrySet().iterator() + while (keptEntryCount > maxKeptEntries && eldestFirst.hasNext) { + keptEntryCount -= eldestFirst.next().getValue._2.size + eldestFirst.remove() + } + } + + /** Children the kept listings hold, in all. */ + private[comet] def keptEncryptionZoneEntries: Int = listedParents.synchronized(keptEntryCount) + + /** + * Directory entries one scan may list for the encryption-zone check before it is declined + * instead. At HDFS's default page of 1000 entries (`dfs.ls.limit`) that is a thousand NameNode + * round trips, and a directory holds at most about a million entries by default + * (`dfs.namenode.fs-limits.max-directory-items`). + */ + private[comet] val MaxEncryptionZoneEntriesPerScan = 1000000 + + /** Children the kept listings may hold in all, on the order of 100 bytes each. */ + private[comet] val MaxKeptEncryptionZoneEntries = 1000000 + + /** How long a listing is kept when `spark.sql.metadataCacheTTLSeconds` is unset. */ + private[comet] val DefaultEncryptionZoneListingAgeSeconds = 3600L + + /** + * Why a scan of `directories`, HDFS directories of data and delete files, must not be claimed + * on account of HDFS encryption zones, or None. iceberg-rust's HDFS client is built without KMS + * support and rejects every file of a zone. The NameNode flags each path inside a zone as + * encrypted, directories included, so one listing of a parent tells the flag of every data + * directory under it: the partitions of a table cost one call, not one each. A listing is kept + * per table (see [[listedParents]]), and a directory missing from a kept listing, a new + * partition, lists its parent again. A cluster that cannot hold a zone at all, one whose + * NameNode advertises no key provider (a client-cached call) and whose session configures none, + * is not listed. A scan that would list more than `maxEntries` directory entries is declined + * once that many are listed; the parents listed whole are kept, so a later scan continues from + * them. + */ + def hdfsEncryptionZoneFallback( + hadoopConf: Configuration, + directories: Set[String], + tableUuid: Option[String] = None, + ttlSeconds: Long = -1L, + maxEntries: Int = MaxEncryptionZoneEntriesPerScan, + maxKeptEntries: Int = MaxKeptEncryptionZoneEntries, + nowMillis: Long = System.currentTimeMillis()): Option[String] = { + val configuredProvider = + Seq("hadoop.security.key.provider.path", "dfs.encryption.key.provider.uri") + .exists(key => hadoopConf.getTrimmed(key, "").nonEmpty) + val maxAgeMillis = + (if (ttlSeconds > 0) ttlSeconds else DefaultEncryptionZoneListingAgeSeconds) * 1000 + def fresh(listedAt: Long): Boolean = nowMillis - listedAt < maxAgeMillis + // A directory at the root of the filesystem has no parent and is listed itself. + val byParent = directories.toSeq.sorted.map(new Path(_)).groupBy { dir => + Option(dir.getParent).getOrElse(dir) + } + var entries = 0 + def budgetReason(parent: Path): String = + s"Iceberg scan reads HDFS directories under '$parent' and beyond, which the check for " + + s"encryption zones did not reach within its budget of $maxEntries directory entries; " + + "a later scan continues from the listings kept" + byParent.toSeq + .sortBy(_._1.toString) + .iterator + .map { case (parent, dirs) => + try { + val fs = parent.getFileSystem(hadoopConf) + val key = (fs.getUri.toString, tableUuid.getOrElse(""), parent.toString) + // Children are kept by name: every directory here is a child of `parent`. + val kept = keptListing(key).collect { + case (listedAt, flags) + if fresh(listedAt) && dirs.forall(dir => flags.contains(dir.getName)) => + flags + } + def encryptedReason(flags: Map[String, Boolean]): Option[String] = + dirs.find(dir => flags.getOrElse(dir.getName, false)).map { dir => + s"Iceberg scan reads HDFS directory '$dir', which is in an encryption zone; " + + "Comet's native Iceberg reader cannot decrypt HDFS encryption zones" + } + kept match { + case Some(flags) => encryptedReason(flags) + case None => + val canHoldZones = configuredProvider || + Option(fs.getServerDefaults(parent).getKeyProviderUri).exists(_.trim.nonEmpty) + if (!canHoldZones) { + None + } else if (entries >= maxEntries) { + Some(budgetReason(parent)) + } else { + // Paged on HDFS, so a parent beyond the budget costs the budget, no more. + val listing = fs.listStatusIterator(parent) + val listed = Map.newBuilder[String, Boolean] + while (entries < maxEntries && listing.hasNext) { + val status = listing.next() + listed += status.getPath.getName -> status.isEncrypted + entries += 1 + } + if (listing.hasNext) { + Some(budgetReason(parent)) + } else { + val children = listed.result() + val missing = dirs + .filterNot(dir => children.contains(dir.getName)) + .map(dir => dir.getName -> fs.getFileStatus(dir).isEncrypted) + val flags = children ++ missing + keepListing(key, (nowMillis, flags), maxKeptEntries) + encryptedReason(flags) + } + } + } + } catch { + case NonFatal(e) => + logWarning(s"Could not tell whether HDFS directories under $parent are encrypted", e) + Some( + s"Iceberg scan reads HDFS directories under '$parent', whose status could not be " + + s"read (${e.getClass.getSimpleName}: ${e.getMessage}); Comet's native Iceberg " + + "reader cannot decrypt HDFS encryption zones, so the scan is not claimed") + } + } + .collectFirst { case Some(reason) => reason } + } + + private def hasPort(address: String): Boolean = + address != null && { + val colon = address.lastIndexOf(':') + colon > address.lastIndexOf(']') && colon < address.length - 1 && + address.substring(colon + 1).forall(_.isDigit) + } + /** iceberg-rust's per-I/O-call timeout, set from `spark.comet.iceberg.ioTimeout`. */ def ioTimeoutProperty(): (String, String) = "opendal.io-timeout-ms" -> CometConf.COMET_ICEBERG_IO_TIMEOUT.get().toString 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..8addcd2e9a7 --- /dev/null +++ b/spark/src/test/scala/org/apache/comet/CometIcebergHdfsSuite.scala @@ -0,0 +1,623 @@ +/* + * 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.io.File +import java.net.URI +import java.nio.file.Files +import java.security.PrivilegedExceptionAction +import java.util.{EnumSet, UUID} + +import org.scalactic.source.Position +import org.scalatest.Tag + +import org.apache.commons.io.FileUtils +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.crypto.key.{KeyProvider, KeyProviderFactory} +import org.apache.hadoop.fs.{FileSystem, Path} +import org.apache.hadoop.fs.permission.FsPermission +import org.apache.hadoop.hdfs.{MiniDFSCluster, MiniDFSNNTopology} +import org.apache.hadoop.hdfs.client.{CreateEncryptionZoneFlag, HdfsAdmin} +import org.apache.hadoop.security.UserGroupInformation +import org.apache.hadoop.util.VersionInfo +import org.apache.iceberg.{DataFiles, FileFormat, Table} +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.CometSparkSessionExtensions.isSpark42Plus +import org.apache.comet.iceberg.IcebergReflection + +/** + * End-to-end coverage of the native Iceberg scan against an `hdfs://` warehouse, backed by + * in-process `MiniDFSCluster`s. + * + * 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 NameNode endpoints + * Comet hands it. A unit test over the scheme allowlists cannot tell whether that client actually + * connects and reads. Writes to HDFS use Iceberg's own writer and are not covered here. + * + * What is covered: + * - scans of tables on a single-NameNode cluster, whose locations carry a real `host:port` + * authority, so iceberg-rust needs no NameNode declaration; + * - a two-NameNode HA cluster addressed by nameservice (`hdfs:///...`), which is not a + * host, so the scan only connects through the `hdfs.name-node.` declaration Comet derives + * from the session Hadoop configuration (iceberg-rust resolves a portless authority only + * through it). It is read before and after a failover, and once the first NameNode is shut + * down. + * + * Which profiles run it. The cluster needs the `hadoop-client-minicluster` that `pom.xml` pins + * per Spark profile (`hadoop.version`: 3.5 -> 3.3.4, 4.0 -> 3.4.1, 4.1 -> 3.4.2, 4.2 -> 3.5.0) to + * match the `hadoop-client-api`/`runtime` Spark supplies. Measured by running the scan tests on + * every profile: + * - Spark 3.5, 4.0 and 4.1: the cluster starts and every test runs, none canceled. A cluster + * that fails to start there is a broken fixture, so `beforeAll` fails the suite with the + * cause instead of cancelling the tests. On Spark 4.0 the encryption-zone test needs + * BouncyCastle on the test class path, which that profile adds: Hadoop 3.4.1's `KeyProvider` + * links against it. + * - Spark 4.2: nothing runs (all tests are cancelled), for two independent reasons. There is no + * Iceberg runtime for 4.2 (`icebergAvailable` is false there), and the cluster cannot start + * either: `MiniDFSCluster.Builder` throws `NoClassDefFoundError: + * org/junit/jupiter/api/Assertions`, because Hadoop 3.5.0's `GenericTestUtils` asserts with + * JUnit 5, which the shaded minicluster leaves out and the test classpath does not carry. + * With junit-jupiter-api 5.10.3 and its three small dependencies put on the JVM class path + * the cluster starts, so a test-scoped `junit-jupiter-api` dependency on the 4.2 profile is + * the fix once an Iceberg runtime exists. Until then this suite skips there and starts + * nothing. + */ +class CometIcebergHdfsSuite + extends CometTestBase + with AdaptiveSparkPlanHelper + with CometIcebergTestBase + with WithHdfsCluster { + + /** + * The only profile on which this suite does not run (see the class docstring): no Iceberg + * runtime, and no MiniDFSCluster. Anywhere else the Iceberg runtime and the cluster are + * required, and their absence fails the suite rather than cancelling it. + */ + private def profileSkipped: Boolean = isSpark42Plus + + override protected def test(testName: String, testTags: Tag*)(testFun: => Any)(implicit + pos: Position): Unit = { + super.test(testName, testTags: _*) { + assume(!profileSkipped, "Spark 4.2 has no Iceberg runtime and no usable MiniDFSCluster") + testFun + } + } + + /** + * The nameservice of the HA cluster. Not the `ns1` of `simpleHATopology`, to be unmistakable. + */ + private val haNameservice = "cometha" + + private var haCluster: MiniDFSCluster = _ + private var haBaseDir: File = _ + + override def beforeAll(): Unit = { + super.beforeAll() + if (!profileSkipped) { + assert(icebergAvailable, "the Iceberg runtime is missing from the test classpath") + try startHdfsCluster() + catch { + case e @ (_: Exception | _: LinkageError) => + throw new IllegalStateException( + "MiniDFSCluster failed to start, so none of the HDFS tests can run. " + + s"Spark ${org.apache.spark.SPARK_VERSION}, hadoop-client-api " + + s"${VersionInfo.getVersion}, hadoop-client-minicluster loaded from " + + s"${classOf[MiniDFSCluster].getProtectionDomain.getCodeSource.getLocation}: $e", + e) + } + } + } + + override def afterAll(): Unit = { + try stopHaCluster() + finally { + try stopEncryptedCluster() + finally { + try stopPermissionsCluster() + finally { + try stopHdfsCluster() + finally super.afterAll() + } + } + } + } + + /** `hdfs://localhost:` -- the authority iceberg-rust dials as the NameNode. */ + private def hdfsUri: String = s"hdfs://localhost:$getDFSPort" + + private def assertSingleNativeScan(cometPlan: SparkPlan): CometIcebergNativeScanExec = { + 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") + scans.head + } + + /** + * Runs `f` with a Hadoop-catalog Iceberg warehouse rooted at `baseUri`, which is the + * single-NameNode cluster by default. The catalog name is unique per test so Spark's catalog + * cache cannot hand back a warehouse from an earlier test, and the warehouse directory is + * removed afterwards (`cleanup`). + */ + private def withHdfsIcebergCatalog(baseUri: String = hdfsUri, cleanup: Boolean = true)( + f: String => Unit): Unit = { + val catalog = s"hdfs_cat_${UUID.randomUUID().toString.replace("-", "")}" + val warehouse = s"$baseUri/warehouse/${UUID.randomUUID()}" + try { + 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) + } + } finally { + if (cleanup) getFileSystem.delete(new Path(warehouse), true) + } + } + + private def tableLocation(catalog: String, table: String): String = { + val icebergTable = loadIcebergTable(spark, catalog, "db", table) + icebergTable.getClass.getMethod("location").invoke(icebergTable).toString + } + + test("native Iceberg scan reads a table stored on HDFS") { + 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") { + 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") + } + } + + // Delete files are opened through the same FileIO as the data files, so a merge-on-read table + // is the one read where the HDFS backend serves two kinds of file of one task. + test("native Iceberg scan on HDFS applies position deletes") { + withHdfsIcebergCatalog() { catalog => + spark.sql(s""" + CREATE TABLE $catalog.db.mor (id INT, name STRING) USING iceberg + TBLPROPERTIES ('format-version' = '2', 'write.delete.mode' = 'merge-on-read') + """) + // One data file, so the delete cannot drop whole files through metadata and must write a + // position-delete file on HDFS. + spark.sql(s""" + INSERT INTO $catalog.db.mor + SELECT /*+ COALESCE(1) */ id, name FROM VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd') AS t(id, name) + """) + spark.sql(s"DELETE FROM $catalog.db.mor WHERE id IN (2, 4)") + val summary = + loadIcebergTable(spark, catalog, "db", "mor").asInstanceOf[Table].currentSnapshot.summary + assert(summary.get("added-delete-files") == "1", summary) + + val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $catalog.db.mor ORDER BY id") + assertSingleNativeScan(cometPlan) + assert(spark.sql(s"SELECT count(*) FROM $catalog.db.mor").collect().head.getLong(0) == 2L) + } + } + + test("native Iceberg scan reads a partitioned table across multiple HDFS data files") { + 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") + } + } + + private def stopHaCluster(): Unit = { + try if (haCluster != null) haCluster.shutdown(true) + finally { + haCluster = null + if (haBaseDir != null) FileUtils.deleteQuietly(haBaseDir) + haBaseDir = null + } + } + + /** + * Starts a two-NameNode MiniDFSCluster with `nn1` active, and returns the client configuration + * a Spark job needs to address it by nameservice: exactly the keys a production `hdfs-site.xml` + * carries, so that Comet's derivation of `hdfs.name-node.` is what lets the native client + * dial. + */ + private def startHaCluster(): Seq[(String, String)] = { + val conf = new Configuration() + conf.set("dfs.namenode.metrics.logger.period.seconds", "0") + conf.set("dfs.datanode.metrics.logger.period.seconds", "0") + conf.setIfUnset("dfs.namenode.rpc-bind-host", "localhost") + // A second cluster in this JVM must not format the first one's `target/test/data/dfs`. + haBaseDir = Files.createTempDirectory("comet_ha_dfs_").toFile + conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, haBaseDir.getAbsolutePath) + + val topology = new MiniDFSNNTopology().addNameservice( + new MiniDFSNNTopology.NSConf(haNameservice) + .addNN(new MiniDFSNNTopology.NNConf("nn1")) + .addNN(new MiniDFSNNTopology.NNConf("nn2"))) + haCluster = new MiniDFSCluster.Builder(conf).nnTopology(topology).numDataNodes(1).build() + haCluster.transitionToActive(0) + + def rpcAddress(nn: String): String = { + val key = s"dfs.namenode.rpc-address.$haNameservice.$nn" + val address = haCluster.getConfiguration(0).get(key) + assert(address != null && address.nonEmpty, s"the HA cluster did not configure $key") + address + } + Seq( + "dfs.nameservices" -> haNameservice, + s"dfs.ha.namenodes.$haNameservice" -> "nn1,nn2", + s"dfs.namenode.rpc-address.$haNameservice.nn1" -> rpcAddress("nn1"), + s"dfs.namenode.rpc-address.$haNameservice.nn2" -> rpcAddress("nn2"), + s"dfs.client.failover.proxy.provider.$haNameservice" -> + "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider") + } + + test("native Iceberg scan resolves an HA nameservice and follows a failover") { + val haClientConf = startHaCluster() + val logicalUri = new URI(s"hdfs://$haNameservice") + val haConf = haClientConf.toMap + // What Comet declares for the nameservice: its NameNodes in `dfs.ha.namenodes` order, as + // their rpc-addresses read. + val expectedNameNodes = Seq("nn1", "nn2") + .map(nn => haConf(s"dfs.namenode.rpc-address.$haNameservice.$nn")) + .mkString(",") + + val hadoopConf = new Configuration() + haClientConf.foreach { case (k, v) => hadoopConf.set(k, v) } + try { + // The nameservice keys reach the session Hadoop configuration (planning, the Iceberg + // catalog, the commit) the way `spark.hadoop.*` settings would in a deployment. + withSQLConf(haClientConf: _*) { + withHdfsIcebergCatalog(baseUri = logicalUri.toString, cleanup = false) { catalog => + // A location with no host:port, which iceberg-rust cannot dial by itself. + spark.sql(s"CREATE TABLE $catalog.db.ha (id INT, name STRING) USING iceberg") + spark.sql(s"INSERT INTO $catalog.db.ha VALUES (1, 'a'), (2, 'b'), (3, 'c')") + val location = tableLocation(catalog, "ha") + assert( + location.startsWith(s"hdfs://$haNameservice/"), + s"expected a nameservice location, got $location") + + // Succeeding is not enough: the declaration Comet derived is what the client was given. + // The per-nameservice key, since iceberg-rust reads the global `hdfs.name-node` only + // for authority-less paths. + def assertDeclared(properties: Map[String, String]): Unit = { + assert( + properties.get(s"hdfs.name-node.$haNameservice") == Some(expectedNameNodes), + s"$properties") + assert(!properties.contains("hdfs.name-node"), s"$properties") + } + + def scanAndCheck(expectedRows: Long): Unit = { + val query = s"SELECT * FROM $catalog.db.ha ORDER BY id" + // Checked on the planned scan before anything runs, so a lost declaration fails here + // with the properties in the message, not as a NameNode resolution error on an + // executor after the native client's retries. + assertDeclared( + assertSingleNativeScan( + spark + .sql(query) + .queryExecution + .executedPlan).nativeIcebergScanMetadata.catalogProperties) + val (_, cometPlan) = checkSparkAnswer(query) + assertDeclared( + assertSingleNativeScan(cometPlan).nativeIcebergScanMetadata.catalogProperties) + assert( + spark.sql(s"SELECT count(*) FROM $catalog.db.ha").collect().head.getLong(0) == + expectedRows) + } + + scanAndCheck(3L) + + // Fail over: nn1 goes standby and nn2 takes over. The first NameNode on the native + // client's list is now a standby, and the list must carry it to the new active. + haCluster.transitionToStandby(0) + haCluster.transitionToActive(1) + assert( + haCluster.getNameNode(0).isStandbyState && haCluster.getNameNode(1).isActiveState) + scanAndCheck(3L) + + // A NameNode that is gone rather than standby: connections to nn1 are now refused, and + // the cached native client must still reach nn2, for the files it knows and for one + // Iceberg's own writer adds through the HA client meanwhile. + haCluster.shutdownNameNode(0) + spark.sql(s"INSERT INTO $catalog.db.ha VALUES (4, 'd')") + scanAndCheck(4L) + } + } + } finally { + // Evict the cached client of the nameservice before its cluster goes away. + try FileSystem.get(logicalUri, hadoopConf).close() + catch { case _: Exception => () } + stopHaCluster() + } + } + + private var ezCluster: MiniDFSCluster = _ + private var ezBaseDir: File = _ + + private def stopEncryptedCluster(): Unit = { + try if (ezCluster != null) ezCluster.shutdown(true) + finally { + ezCluster = null + if (ezBaseDir != null) FileUtils.deleteQuietly(ezBaseDir) + ezBaseDir = null + } + } + + /** + * Starts a cluster whose NameNode has a key provider (a `jceks` keystore on the local disk, as + * Hadoop's own encryption tests use), with one key and an encryption zone at `/ez`. The key is + * created before the NameNode starts, since it loads the keystore once. + */ + private def startEncryptedCluster(): Unit = { + val conf = new Configuration() + conf.set("dfs.namenode.metrics.logger.period.seconds", "0") + conf.set("dfs.datanode.metrics.logger.period.seconds", "0") + conf.setIfUnset("dfs.namenode.rpc-bind-host", "localhost") + ezBaseDir = Files.createTempDirectory("comet_ez_dfs_").toFile + conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, ezBaseDir.getAbsolutePath) + val providerUri = s"jceks://file${new File(ezBaseDir, "ez.jceks").toURI.getPath}" + conf.set("hadoop.security.key.provider.path", providerUri) + val provider = KeyProviderFactory.get(new URI(providerUri), conf) + provider.createKey("ez-key", new KeyProvider.Options(conf)) + provider.flush() + + ezCluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build() + val fs = ezCluster.getFileSystem + fs.mkdirs(new Path("/ez")) + new HdfsAdmin(fs.getUri, ezCluster.getConfiguration(0)) + .createEncryptionZone( + new Path("/ez"), + "ez-key", + EnumSet.noneOf(classOf[CreateEncryptionZoneFlag])) + } + + test("native Iceberg scan falls back for a table in an HDFS encryption zone") { + startEncryptedCluster() + try { + val ezUri = s"hdfs://localhost:${ezCluster.getNameNodePort}" + // Iceberg's own writer encrypts through the key provider the NameNode advertises; the + // native reader cannot decrypt, so the scan must be declined at planning, not fail on an + // executor. + withHdfsIcebergCatalog(baseUri = s"$ezUri/ez", cleanup = false) { catalog => + // Partitioned, so the probe meets several data directories under one parent. + spark.sql( + s"CREATE TABLE $catalog.db.enc (id INT, name STRING) USING iceberg " + + "PARTITIONED BY (name)") + spark.sql(s"INSERT INTO $catalog.db.enc VALUES (1, 'a'), (2, 'b')") + val dataLocation = IcebergReflection + .getDataLocation(loadIcebergTable(spark, catalog, "db", "enc")) + .getOrElse(fail("could not resolve the Iceberg data location")) + // The fixture really is a zone: the NameNode flags the data directory as encrypted. + assert( + ezCluster.getFileSystem.getFileStatus(new Path(dataLocation)).isEncrypted, + s"$dataLocation is not in the encryption zone") + + val (_, cometPlan) = checkSparkAnswerAndFallbackReason( + s"SELECT * FROM $catalog.db.enc ORDER BY id", + "encryption zone") + val scans = collect(cometPlan) { case scan: CometIcebergNativeScanExec => scan } + assert( + scans.isEmpty, + s"an encryption-zone table must fall back, but was claimed:\n$cometPlan") + } + // The gate is per location, not per cluster: a table outside the zone on the same + // NameNode still scans natively. + withHdfsIcebergCatalog(baseUri = ezUri, cleanup = false) { catalog => + spark.sql(s"CREATE TABLE $catalog.db.plain (id INT) USING iceberg") + spark.sql(s"INSERT INTO $catalog.db.plain VALUES (1), (2)") + val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $catalog.db.plain ORDER BY id") + assertSingleNativeScan(cometPlan) + } + } finally stopEncryptedCluster() + } + + test("one native scan reads data files on two NameNodes") { + val haClientConf = startHaCluster() + val logicalUri = new URI(s"hdfs://$haNameservice") + val hadoopConf = new Configuration() + haClientConf.foreach { case (k, v) => hadoopConf.set(k, v) } + try { + withSQLConf(haClientConf: _*) { + // Metadata and the first data file on the single-NameNode cluster. + withHdfsIcebergCatalog() { catalog => + spark.sql(s"CREATE TABLE $catalog.db.two (id INT, name STRING) USING iceberg") + spark.sql(s"INSERT INTO $catalog.db.two VALUES (1, 'a')") + + // A second data file, written by Iceberg on the HA cluster and registered in the first + // table's manifest. Same DDL, so the field ids line up. + withHdfsIcebergCatalog(baseUri = logicalUri.toString, cleanup = false) { haCatalog => + spark.sql(s"CREATE TABLE $haCatalog.db.src (id INT, name STRING) USING iceberg") + spark.sql(s"INSERT INTO $haCatalog.db.src VALUES (2, 'b')") + val src = loadIcebergTable(spark, haCatalog, "db", "src").asInstanceOf[Table] + val srcFile = src.newScan().planFiles().iterator().next().file() + val srcLocation = IcebergReflection + .extractFileLocation( + IcebergReflection.loadClass(IcebergReflection.ClassNames.CONTENT_FILE), + srcFile) + .getOrElse(fail("could not read the source data file location")) + assert(srcLocation.startsWith(s"hdfs://$haNameservice/"), srcLocation) + + val table = loadIcebergTable(spark, catalog, "db", "two").asInstanceOf[Table] + table + .newAppend() + .appendFile( + DataFiles + .builder(table.spec()) + .withPath(srcLocation) + .withFormat(FileFormat.PARQUET) + .withFileSizeInBytes(srcFile.fileSizeInBytes()) + .withRecordCount(srcFile.recordCount()) + .build()) + .commit() + spark.sql(s"REFRESH TABLE $catalog.db.two") + } + + // One FileIO serves both: the host:port authority dialed as written, the nameservice + // through the declaration. + val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $catalog.db.two ORDER BY id") + val scan = assertSingleNativeScan(cometPlan) + assert( + scan.nativeIcebergScanMetadata.catalogProperties + .contains(s"hdfs.name-node.$haNameservice"), + s"${scan.nativeIcebergScanMetadata.catalogProperties}") + assert( + spark.sql(s"SELECT count(*) FROM $catalog.db.two").collect().head.getLong(0) == 2L) + } + } + } finally { + try FileSystem.get(logicalUri, hadoopConf).close() + catch { case _: Exception => () } + stopHaCluster() + } + } + + private var permCluster: MiniDFSCluster = _ + private var permBaseDir: File = _ + + private def stopPermissionsCluster(): Unit = { + try if (permCluster != null) permCluster.shutdown(true) + finally { + permCluster = null + if (permBaseDir != null) FileUtils.deleteQuietly(permBaseDir) + permBaseDir = null + } + } + + /** + * Starts a cluster as `superUser`, which makes that user the NameNode's superuser instead of + * this JVM's user. `dfs.permissions.enabled` is on by default; what makes every check pass on + * the other clusters here is that the JVM's user started their NameNodes. To this one it is an + * ordinary user, which owns `/perm`. + */ + private def startPermissionsCluster(superUser: UserGroupInformation): Unit = { + val conf = new Configuration() + conf.set("dfs.namenode.metrics.logger.period.seconds", "0") + conf.set("dfs.datanode.metrics.logger.period.seconds", "0") + conf.setIfUnset("dfs.namenode.rpc-bind-host", "localhost") + permBaseDir = Files.createTempDirectory("comet_perm_dfs_").toFile + conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, permBaseDir.getAbsolutePath) + val owner = spark.sparkContext.sparkUser + permCluster = superUser.doAs(new PrivilegedExceptionAction[MiniDFSCluster] { + override def run(): MiniDFSCluster = { + val cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build() + val fs = cluster.getFileSystem + fs.mkdirs(new Path("/perm")) + fs.setOwner(new Path("/perm"), owner, null) + cluster + } + }) + } + + // The planner claims an HDFS scan on the premise that the native client connects as the Spark + // user (in local mode, this JVM's user). With permissions enforced by a NameNode that does not + // treat that user as its superuser, a read of a directory only that user may enter proves it. + test("native Iceberg scan reads HDFS as the Spark user when permissions are enforced") { + val superUser = UserGroupInformation.createRemoteUser("hdfs-super") + startPermissionsCluster(superUser) + try { + val sparkUser = spark.sparkContext.sparkUser + val permUri = s"hdfs://localhost:${permCluster.getNameNodePort}" + withHdfsIcebergCatalog(baseUri = s"$permUri/perm", cleanup = false) { catalog => + spark.sql(s"CREATE TABLE $catalog.db.owned (id INT, name STRING) USING iceberg") + spark.sql(s"INSERT INTO $catalog.db.owned VALUES (1, 'a'), (2, 'b')") + val dataDir = new Path( + IcebergReflection + .getDataLocation(loadIcebergTable(spark, catalog, "db", "owned")) + .getOrElse(fail("could not resolve the Iceberg data location"))) + // Outside `doAs` the filesystem is this JVM's user, an ordinary user to this NameNode. + val fs = permCluster.getFileSystem + assert( + fs.getFileStatus(dataDir).getOwner == sparkUser, + s"$dataDir is owned by ${fs.getFileStatus(dataDir).getOwner}, not $sparkUser") + fs.setPermission(dataDir, new FsPermission("700")) + + // Only the owner may enter the data directory, so the native read succeeds only if the + // client connected as that user. + val query = s"SELECT * FROM $catalog.db.owned ORDER BY id" + val (_, cometPlan) = checkSparkAnswer(query) + assertSingleNativeScan(cometPlan) + + // The check is live for this client: once the directory belongs to someone else, the + // native read is denied as Spark's is. + superUser.doAs(new PrivilegedExceptionAction[Unit] { + override def run(): Unit = + permCluster.getFileSystem.setOwner(dataDir, "someone-else", null) + }) + Seq(true, false).foreach { cometEnabled => + withSQLConf(CometConf.COMET_ENABLED.key -> cometEnabled.toString) { + val e = intercept[Exception](spark.sql(query).collect()) + val messages = Iterator + .iterate(e: Throwable)(_.getCause) + .takeWhile(_ != null) + .map(_.getMessage) + .mkString("\n") + assert( + messages.contains("AccessControlException") || messages.contains( + "Permission denied"), + s"comet=$cometEnabled failed for another reason: $messages") + } + } + } + } finally stopPermissionsCluster() + } +} diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 629f2458842..3f2bd95b6be 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.iceberg.expressions.Expressions import org.apache.iceberg.spark.Spark3Util import org.apache.parquet.hadoop.ParquetFileReader import org.apache.parquet.hadoop.util.HadoopInputFile -import org.apache.spark.{CometListenerBusUtils, SparkException} +import org.apache.spark.CometListenerBusUtils import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} import org.apache.spark.sql.{CometTestBase, DataFrame, Row} import org.apache.spark.sql.catalyst.expressions.{Ascending, DynamicPruningExpression, SortOrder} @@ -3017,11 +3017,12 @@ class CometIcebergNativeSuite } } - // Data files written without Iceberg field ids and read with no name mapping: iceberg-rust then - // projects whole top-level columns by position, and a by-name struct cast narrows each one to - // the pruned type. Spark's own reader returns NULL nested fields for such a table, so the pruned - // read is checked against the native read of every nested field instead. - test("nested schema pruning on data files without field ids") { + // Data files written without Iceberg field ids and read with no name mapping: neither reader + // can match such a file's nested fields to the table's, so a struct's fields read as NULL. + // Spark's reader gives them fallback readers; iceberg-rust, since apache/iceberg-rust#3437, + // treats them as absent, whether the struct is read whole or pruned. Top-level columns still + // read. A list of structs is the one shape the readers still disagree on. + test("nested fields of data files without field ids read as NULL, as in Spark") { assume(icebergAvailable, "Iceberg not available in classpath") withTempIcebergDir { warehouseDir => @@ -3081,30 +3082,35 @@ class CometIcebergNativeSuite spark.sql( s"ALTER TABLE $table UNSET TBLPROPERTIES IF EXISTS ('schema.name-mapping.default')") - def nativeRows(query: String, pruneNestedFields: Boolean): Seq[Row] = { - var rows = Seq.empty[Row] + // Both readers agree, with the struct read whole and read pruned. + Seq( + s"SELECT id, s FROM $table ORDER BY id", + s"SELECT id, s.b FROM $table ORDER BY id", + s"SELECT count(*) FROM $table WHERE s.a IS NULL", + s"SELECT count(*) FROM $table WHERE s IS NULL").foreach(checkPrunedAndFullReads) + // And NULL is what they agree on, not the file's values. + val rows = spark.sql(s"SELECT id, s.b FROM $table ORDER BY id").collect() + assert(rows.length == 100 && rows.forall(_.isNullAt(1)), rows.mkString(",")) + // A list of structs: Spark's reader drops the elements and returns an empty list, the + // native read keeps each element with its fields NULL. + val listQuery = s"SELECT items.x FROM $table ORDER BY id" + Seq(true, false).foreach { pruneNestedFields => withSQLConf( CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key -> pruneNestedFields.toString) { - val df = spark.sql(query) - rows = df.collect().toSeq + val df = spark.sql(listQuery) + val native = df.collect() assertSingleNativeScan(df.queryExecution.executedPlan) + assert( + native.length == 100 && native.forall(_.getSeq[Any](0) == Seq(null)), + native.take(3).mkString(",")) } - rows } - // Without field ids or a name mapping, iceberg-rust matches a struct's fields to the - // table's by position, and only when their counts are equal. A read of the whole struct - // gets the file's values, as it did before. A pruned read of some of its fields fails - // rather than reading a different field, since `s.b` is the struct's last field. - Seq( - s"SELECT id, s.b FROM $table ORDER BY id", - s"SELECT id, items.x FROM $table ORDER BY id").foreach { query => - val full = nativeRows(query, pruneNestedFields = false) - assert(full.length == 100 && full.forall(!_.isNullAt(1)), s"$query: $full") - val e = intercept[SparkException](nativeRows(query, pruneNestedFields = true)) + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val vanilla = spark.sql(listQuery).collect() assert( - e.getMessage.contains("cannot reconcile struct fields by id"), - s"$query: ${e.getMessage}") + vanilla.length == 100 && vanilla.forall(_.getSeq[Any](0).isEmpty), + vanilla.take(3).mkString(",")) } spark.sql(s"DROP TABLE $table") @@ -3117,6 +3123,102 @@ class CometIcebergNativeSuite } } + // A table migrated from Spark-written Parquet reads its files, which have no field ids, through + // its name mapping. iceberg-rust cannot open the list layout that `spark.sql.parquet.writeLegacyFormat` + // writes (a two-level list) through a name mapping: the read schema it supplies keeps a field + // id on the list's element field and the Parquet reader rejects it. The layout is only known + // from the file, so the scan falls back for the list column and stays native for the primitives. + test("name-mapped table with a list column from a legacy Parquet layout falls back") { + 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") { + + val dataPath = s"${warehouseDir.getAbsolutePath}/legacy_list" + // Non-null elements and a non-null struct field: with `writeLegacyFormat` Spark then writes + // the two-level layout, `repeated group array { required int32 col2 }`, the list's + // element being the repeated group itself. + val sparkSchema = StructType( + Seq( + org.apache.spark.sql.types.StructField("id", IntegerType, nullable = false), + org.apache.spark.sql.types.StructField( + "col1", + ArrayType( + StructType(Seq( + org.apache.spark.sql.types.StructField("col2", IntegerType, nullable = false))), + containsNull = false)))) + val jsonRows = (0 until 10).map(i => s"""{"id": $i, "col1": [{"col2": ${i * 2}}]}""") + withSQLConf("spark.sql.parquet.writeLegacyFormat" -> "true") { + spark.read + .schema(sparkSchema) + .json(spark.createDataset(jsonRows)(org.apache.spark.sql.Encoders.STRING)) + .coalesce(1) + .write + .parquet(dataPath) + } + spark.sql("CREATE NAMESPACE IF NOT EXISTS test_cat.db") + val table = "test_cat.db.legacy_list" + spark.sql(s"CREATE TABLE $table (id INT, col1 ARRAY>) USING iceberg") + try { + val tableUtilClass = Class.forName("org.apache.iceberg.spark.SparkTableUtil") + val icebergTable = spark.sessionState.catalogManager + .catalog("test_cat") + .asInstanceOf[org.apache.iceberg.spark.SparkCatalog] + .loadTable( + org.apache.spark.sql.connector.catalog.Identifier.of(Array("db"), "legacy_list")) + .asInstanceOf[org.apache.iceberg.spark.source.SparkTable] + .table() + // What the migrate and snapshot actions set for a table they create: `importSparkTable` + // reads a mapping that is already there and does not create one. + icebergTable + .updateProperties() + .set( + "schema.name-mapping.default", + org.apache.iceberg.mapping.NameMappingParser + .toJson(org.apache.iceberg.mapping.MappingUtil.create(icebergTable.schema()))) + .commit() + spark.sql(s"CREATE TABLE parquet_legacy_list USING parquet LOCATION '$dataPath'") + tableUtilClass + .getMethod( + "importSparkTable", + classOf[org.apache.spark.sql.SparkSession], + classOf[org.apache.spark.sql.catalyst.TableIdentifier], + classOf[org.apache.iceberg.Table], + classOf[String]) + .invoke( + null, + spark, + new org.apache.spark.sql.catalyst.TableIdentifier("parquet_legacy_list"), + icebergTable, + s"${warehouseDir.getAbsolutePath}/staging") + + // The list column: declined at planning, and Spark reads the file's values. + val (_, listPlan) = checkSparkAnswerAndFallbackReason( + s"SELECT id, col1 FROM $table ORDER BY id", + "Nested columns with Iceberg name mapping") + assert( + collect(listPlan) { case s: CometIcebergNativeScanExec => s }.isEmpty, + s"$listPlan") + val read = spark.sql(s"SELECT col1[0].col2 FROM $table ORDER BY id").collect() + assert(read.map(_.getInt(0)).toSeq == (0 until 10).map(_ * 2), read.mkString(",")) + + // The primitives alone stay native. + val (_, idPlan) = checkSparkAnswer(s"SELECT id FROM $table ORDER BY id") + assertSingleNativeScan(idPlan) + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table") + spark.sql("DROP TABLE IF EXISTS parquet_legacy_list") + } + } + } + } + test("UUID type - native Iceberg UUID column (reproduces type mismatch)") { assume(icebergAvailable, "Iceberg not available in classpath") @@ -4705,27 +4807,31 @@ class CometIcebergNativeSuite distinctCount == numRows, s"Expected $numRows distinct IDs but got $distinctCount") - // Spark's Iceberg reader returns null for INT96 timestamps inside structs, - // so we can't use checkIcebergNativeScan (which compares against Spark) for - // ts_struct. Instead, compare Comet's read against the raw Parquet source. checkIcebergNativeScan( "SELECT id, ts, value, ts_array, ts_map FROM test_cat.db.int96_test ORDER BY id") checkIcebergNativeScan("SELECT id, ts FROM test_cat.db.int96_test ORDER BY id") checkIcebergNativeScan("SELECT id, ts_array FROM test_cat.db.int96_test ORDER BY id") checkIcebergNativeScan("SELECT id, ts_map FROM test_cat.db.int96_test ORDER BY id") - - // Validate ts_struct against raw Parquet since Spark's Iceberg reader can't read it - val icebergStructDf = spark - .sql("SELECT id, ts_struct FROM test_cat.db.int96_test ORDER BY id") - .collect() - val parquetStructDf = spark.read - .parquet(dataPath) - .select("id", "ts_struct") - .orderBy("id") - .collect() + // The imported file has no field ids, so neither reader can read the struct's fields + // (apache/iceberg-rust#3437): the native scan returns the struct with NULL fields, and + // Spark's reader, which cannot tell an optional struct's own nullness without a field + // to read, returns NULL for the struct itself. The native read still decodes the + // INT96 leaf, since such a file is read by whole columns, so an overflow would + // surface here. + val structQuery = "SELECT id, ts_struct FROM test_cat.db.int96_test ORDER BY id" + val structDf = spark.sql(structQuery) + val nativeStructs = structDf.collect() + assertSingleNativeScan(structDf.queryExecution.executedPlan) assert( - icebergStructDf.sameElements(parquetStructDf), - "ts_struct mismatch between Comet Iceberg read and raw Parquet") + nativeStructs.length == numRows && + nativeStructs.forall(_.getStruct(1) == Row(null, null)), + nativeStructs.take(3).mkString(",")) + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val sparkStructs = spark.sql(structQuery).collect() + assert( + sparkStructs.length == numRows && sparkStructs.forall(_.isNullAt(1)), + sparkStructs.take(3).mkString(",")) + } spark.sql("DROP TABLE test_cat.db.int96_test") spark.sql("DROP TABLE parquet_temp") 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 8cb1765fa39..5a3722f545d 100644 --- a/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala +++ b/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala @@ -28,16 +28,19 @@ import java.util.concurrent.atomic.AtomicLong import scala.util.Random import org.apache.commons.io.FileUtils +import org.apache.iceberg.{DataFiles, FileFormat, Table} import org.apache.spark.SparkConf +import org.apache.spark.SparkEnv import org.apache.spark.paths.SparkPath import org.apache.spark.sql.{CometTestBase, SaveMode} import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.comet.{CometIcebergNativeScanExec, CometIcebergWriteExec, CometScanExec, CometScanUtils} +import org.apache.spark.sql.comet.{CometBatchScanExec, CometIcebergNativeScanExec, CometIcebergWriteExec, CometScanExec, CometScanUtils} import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan} import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile} +import org.apache.spark.sql.execution.datasources.v2.BatchScanExec import org.apache.spark.sql.internal.SQLConf -import org.apache.comet.{CometConf, CometIcebergTestBase, ExtendedExplainInfo, NativeBase} +import org.apache.comet.{CometConf, CometExplainInfo, CometIcebergTestBase, ExtendedExplainInfo, NativeBase} import org.apache.comet.hadoop.fs.{FakeHdfsAuthorityFileSystem, FakeHDFSFileSystem, FakeHdfsSchemeFileSystem, FakeWasbSchemeFileSystem} import org.apache.comet.iceberg.IcebergStorageSchemes import org.apache.comet.objectstore.NativeConfig @@ -154,6 +157,10 @@ class CometScanSchemeFallbackSuite extends CometTestBase with CometIcebergTestBa assert( !write.contains("oss"), s"oss must not be in the native write schemes ${write.toSeq.sorted}") + assert(read.contains("hdfs"), s"hdfs must be in the native read schemes ${read.toSeq.sorted}") + assert( + !write.contains("hdfs"), + s"hdfs must not be in the native write schemes ${write.toSeq.sorted}") assert( write.contains("memory"), s"memory must be in the native write schemes ${write.toSeq.sorted}") @@ -181,7 +188,8 @@ class CometScanSchemeFallbackSuite extends CometTestBase with CometIcebergTestBa "s3://bucket/key.parquet", "s3a://bucket/key.parquet", "gs://bucket/key.parquet", - "oss://bucket/key.parquet").foreach { u => + "oss://bucket/key.parquet", + "hdfs://nn:8020/key.parquet").foreach { u => assert( CometScanRule.isIcebergReadableScheme(new URI(u), Set.empty), s"$u must be iceberg-readable; icebergReadableSchemes has regressed") @@ -284,6 +292,9 @@ class CometScanSchemeFallbackSuite extends CometTestBase with CometIcebergTestBa assert( !openable("s3:///bucket/k.parquet"), "authorityless s3:/// has no host and no alias promotion") + assert( + !openable("hdfs:///warehouse/k.parquet"), + "authorityless hdfs:/// needs a default NameNode, which Comet does not pass to native") // blob not opted in: no promotion, so the hostless form is not openable either. assert( !openable("blob:///bucket/k.parquet", Set.empty), @@ -740,6 +751,345 @@ class CometScanSchemeFallbackSuite extends CometTestBase with CometIcebergTestBa 2 -> Seq("s3a://b/4.csv"))) assert(split.flatMap(_.files.map(_.partitionValues.getInt(0))) == Seq(1, 3, 2, 4)) } + + // --- Iceberg scans over HDFS data files --------------------------------------------------- + // + // Planner-level: a real CometScanRule over a real Iceberg table whose data files are registered + // in the manifests with the `hdfs://` paths the test names. Planning reads manifests, never the + // data files, so nothing here needs a NameNode and nothing is executed. + + /** + * Plans `SELECT id` over an Iceberg table whose data files sit at `dataFiles`, applies + * `CometScanRule` with native Iceberg scans enabled and `hadoopConf` set on the session, and + * returns the scans the rule claimed plus the fallback reasons it left on the ones it declined. + * `catalogProps` are set on the catalog before its first use, so its FileIO is built with them. + * The warehouse, and so the table metadata, sits on the fake `hdfs://` filesystem + * (`metadataPrefix`): the native reader opens every file through one storage chosen from the + * metadata location, so `hdfs://` data files are claimed only under `hdfs://` metadata. An + * empty prefix leaves the metadata on the local filesystem. + */ + private def planIcebergScanOverHdfs( + dataFiles: Seq[String], + hadoopConf: Seq[(String, String)] = Seq.empty, + catalogProps: Seq[(String, String)] = Seq.empty, + metadataPrefix: String = FakeHdfsSchemeFileSystem.PREFIX, + sparkConf: Seq[(String, String)] = Seq.empty): (Seq[CometBatchScanExec], Set[String]) = { + // A catalog of its own, so no test reads another's cached catalog or warehouse. + val catalog = s"hdfs_gate_${UUID.randomUUID().toString.take(8)}" + var claimed = Seq.empty[CometBatchScanExec] + var reasons = Set.empty[String] + val catalogConfs = catalogProps.map { case (k, v) => s"spark.sql.catalog.$catalog.$k" -> v } + withTempIcebergDir { dir => + val confs = Seq( + s"spark.sql.catalog.$catalog" -> "org.apache.iceberg.spark.SparkCatalog", + s"spark.sql.catalog.$catalog.type" -> "hadoop", + s"spark.sql.catalog.$catalog.warehouse" -> + s"$metadataPrefix${dir.getAbsolutePath}/warehouse", + // Iceberg fetches block locations of `hdfs://` data files when the table is on HDFS; + // the files here exist only in the manifests. + "spark.sql.iceberg.locality.enabled" -> "false") ++ catalogConfs + withSQLConf(confs: _*) { + spark.sql(s"CREATE TABLE $catalog.db.t (id INT) USING iceberg") + val table = loadIcebergTable(spark, catalog, "db", "t").asInstanceOf[Table] + val append = table.newAppend() + dataFiles.foreach { path => + append.appendFile( + DataFiles + .builder(table.spec()) + .withPath(path) + .withFormat(FileFormat.PARQUET) + .withFileSizeInBytes(1024) + .withRecordCount(1) + .build()) + } + append.commit() + + // The Spark plan first, with Comet off, then the rule alone: no execution. + var sparkPlan: SparkPlan = null + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + sparkPlan = spark.sql(s"SELECT id FROM $catalog.db.t").queryExecution.executedPlan + } + val confs = Seq( + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") ++ hadoopConf + withSQLConf(confs: _*) { + val transformed = withLiveSparkConf(sparkConf) { + CometScanRule(spark).apply(stripAQEPlan(sparkPlan)) + } + claimed = transformed.collect { case s: CometBatchScanExec => s } + reasons = transformed + .collect { case s: BatchScanExec => + s.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty[String]) + } + .flatten + .toSet + } + } + } + (claimed, reasons) + } + + /** + * Sets `entries` on the live SparkConf for `f`: the configuration the SparkContext launched its + * executors with, which `CometScanRule` reads for `spark.executorEnv.*`, as opposed to the + * session's SQLConf that `withSQLConf` sets. + */ + private def withLiveSparkConf[T](entries: Seq[(String, String)])(f: => T): T = { + val conf = SparkEnv.get.conf + val previous = entries.map { case (key, _) => key -> conf.getOption(key) } + entries.foreach { case (key, value) => + conf.set(key, value) + assert(spark.sparkContext.getConf.get(key) == value, s"$key did not reach the context") + } + try f + finally + previous.foreach { + case (key, Some(value)) => conf.set(key, value) + case (key, None) => conf.remove(key) + } + } + + private def nameNodeProperties(scan: CometBatchScanExec): Map[String, String] = + scan.nativeIcebergScanMetadata.get.catalogProperties.filter { case (k, _) => + k.startsWith("hdfs.name-node") + } + + private val declaredNs1 = Seq( + "dfs.nameservices" -> "ns1", + "dfs.ha.namenodes.ns1" -> "nn1,nn2", + "dfs.namenode.rpc-address.ns1.nn1" -> "nn1.example.com:9000", + "dfs.namenode.rpc-address.ns1.nn2" -> "nn2.example.com:8020") + + private val undeclaredReason = "which has no port and which the Hadoop configuration does not" + + private val mixedStorageReason = "through one storage backend" + + test("iceberg scan: an HDFS nameservice the Hadoop configuration declares is claimed") { + assume(icebergAvailable, "Iceberg not available in classpath") + val (claimed, reasons) = + planIcebergScanOverHdfs(Seq("hdfs://ns1/warehouse/db/t/a.parquet"), declaredNs1) + assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons") + // The fixture must really put the metadata on hdfs://, or these tests would exercise the + // mixed-storage gate instead of the nameservice declaration. + assert( + claimed.head.nativeIcebergScanMetadata.get.metadataLocation + .startsWith(FakeHdfsSchemeFileSystem.PREFIX), + s"metadata location: ${claimed.head.nativeIcebergScanMetadata.get.metadataLocation}") + // The per-nameservice key, with the NameNodes in `dfs.ha.namenodes` order; iceberg-rust + // reads the global `hdfs.name-node` only for authority-less paths. + assert( + nameNodeProperties(claimed.head) == + Map("hdfs.name-node.ns1" -> "nn1.example.com:9000,nn2.example.com:8020"), + s"unexpected properties: ${claimed.head.nativeIcebergScanMetadata.get.catalogProperties}") + } + + test("iceberg scan: a nameservice with a NameNode missing its rpc-address is declined") { + assume(icebergAvailable, "Iceberg not available in classpath") + // nn2 has no rpc-address, so the declaration is all-or-nothing empty and the executor would + // be left to resolve `ns1` as if it were a host. + val (claimed, reasons) = planIcebergScanOverHdfs( + Seq("hdfs://ns1/warehouse/db/t/a.parquet"), + declaredNs1.filterNot(_._1 == "dfs.namenode.rpc-address.ns1.nn2")) + assert(claimed.isEmpty, "an undeclared nameservice must not be claimed natively") + assert( + reasons.exists(r => r.contains("HDFS authority 'ns1'") && r.contains(undeclaredReason)), + s"expected the nameservice fall-back reason, got: $reasons") + } + + test("iceberg scan: a nameservice listed only in dfs.nameservices is declined") { + assume(icebergAvailable, "Iceberg not available in classpath") + val (claimed, reasons) = planIcebergScanOverHdfs( + Seq("hdfs://ns1/warehouse/db/t/a.parquet"), + Seq("dfs.nameservices" -> "ns1,ns2")) + assert(claimed.isEmpty, "a nameservice without NameNodes must not be claimed") + assert( + reasons.exists(r => r.contains("HDFS authority 'ns1'") && r.contains(undeclaredReason)), + s"expected the nameservice fall-back reason, got: $reasons") + } + + test("iceberg scan: a portless plain HDFS host is claimed on Hadoop's default port") { + assume(icebergAvailable, "Iceberg not available in classpath") + // iceberg-rust has no default NameNode port, so `hdfs://nn.example.com/...` would be a + // nameservice name to it; Comet declares it on 8020, as the JVM client dials it. + val (claimed, reasons) = + planIcebergScanOverHdfs(Seq("hdfs://nn.example.com/warehouse/db/t/a.parquet")) + assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons") + assert( + nameNodeProperties(claimed.head) == + Map("hdfs.name-node.nn.example.com" -> "nn.example.com:8020")) + } + + test( + "iceberg scan: HDFS client settings of the session reach the FileIO, behind the catalog's") { + val (claimed, reasons) = planIcebergScanOverHdfs( + Seq("hdfs://nn.example.com:8020/warehouse/t/data/f.parquet"), + hadoopConf = + Seq("dfs.client.use.datanode.hostname" -> "true", "dfs.client.socket-timeout" -> "30000"), + catalogProps = Seq("hadoop.dfs.client.socket-timeout" -> "5000")) + assert(reasons.isEmpty, reasons) + val properties = claimed.head.nativeIcebergScanMetadata.get.catalogProperties + assert(properties.get("hadoop.dfs.client.use.datanode.hostname").contains("true"), properties) + assert(properties.get("hadoop.dfs.client.socket-timeout").contains("5000"), properties) + } + + test("iceberg scan: HDFS files are declined when executors would read as another user") { + assume(icebergAvailable, "Iceberg not available in classpath") + // The native client connects as HADOOP_USER_NAME or the executor's OS user, never as the + // Spark user the JVM reader uses; the scan is claimed only when those are known to agree. + val (claimed, reasons) = planIcebergScanOverHdfs( + Seq("hdfs://nn.example.com:8020/warehouse/db/t/a.parquet"), + sparkConf = Seq("spark.executorEnv.HADOOP_USER_NAME" -> "someone-else")) + assert(claimed.isEmpty, "a scan read as another user must not be claimed") + assert( + reasons.exists(r => + r.contains("'someone-else'") && r.contains("spark.executorEnv.HADOOP_USER_NAME")), + s"expected the identity fall-back reason, got: $reasons") + } + + test("iceberg scan: HDFS files are claimed when HADOOP_USER_NAME is the Spark user") { + assume(icebergAvailable, "Iceberg not available in classpath") + val (claimed, reasons) = planIcebergScanOverHdfs( + Seq("hdfs://nn.example.com:8020/warehouse/db/t/a.parquet"), + sparkConf = Seq("spark.executorEnv.HADOOP_USER_NAME" -> spark.sparkContext.sparkUser)) + assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons") + } + + test("iceberg scan: HDFS files are declined when icebergNative.hdfs.enabled is false") { + assume(icebergAvailable, "Iceberg not available in classpath") + val key = CometConf.COMET_ICEBERG_NATIVE_HDFS_ENABLED.key + val (declined, reasons) = planIcebergScanOverHdfs( + Seq("hdfs://nn.example.com:8020/warehouse/db/t/a.parquet"), + hadoopConf = Seq(key -> "false")) + assert(declined.isEmpty, "the switch must decline the scan") + assert(reasons.exists(_.contains(s"$key is false")), reasons) + // The switch is the one HDFS reason given: the other HDFS checks are not run. + assert(!reasons.exists(_.contains("HADOOP_USER_NAME")), reasons) + } + + test( + "iceberg scan: HADOOP_USER_NAME in the session's SQLConf does not stand for the executors'") { + assume(icebergAvailable, "Iceberg not available in classpath") + val key = "spark.executorEnv.HADOOP_USER_NAME" + val dataFile = "hdfs://nn.example.com:8020/warehouse/db/t/a.parquet" + // A session can hold the key without the executors having seen it, so the SQLConf value + // must not stand in for the context's: the Spark user there and another user on the + // context is declined. + val (declined, reasons) = planIcebergScanOverHdfs( + Seq(dataFile), + hadoopConf = Seq(key -> spark.sparkContext.sparkUser), + sparkConf = Seq(key -> "someone-else")) + assert(declined.isEmpty, "the SQLConf value must not be read") + assert(reasons.exists(_.contains("'someone-else'")), reasons) + // The other way round the SQLConf is ignored too; in local mode the JVM's user then counts. + val (claimed, _) = + planIcebergScanOverHdfs(Seq(dataFile), hadoopConf = Seq(key -> "someone-else")) + assert(claimed.size == 1, "the SQLConf value must not be read") + } + + test("iceberg scan: a host:port HDFS location is claimed with no NameNode declaration") { + assume(icebergAvailable, "Iceberg not available in classpath") + val (claimed, reasons) = + planIcebergScanOverHdfs(Seq("hdfs://nn.example.com:8020/warehouse/db/t/a.parquet")) + assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons") + assert(nameNodeProperties(claimed.head).isEmpty) + } + + test("iceberg scan: data files on two declared nameservices are claimed, each declared") { + assume(icebergAvailable, "Iceberg not available in classpath") + // iceberg-rust resolves each path's authority on its own and keeps one client per NameNode, + // so a scan may span clusters as long as every nameservice is declared. + val (claimed, reasons) = planIcebergScanOverHdfs( + Seq("hdfs://ns1/warehouse/db/t/a.parquet", "hdfs://ns2/warehouse/db/t/b.parquet"), + declaredNs1 ++ Seq( + "dfs.ha.namenodes.ns2" -> "nn1", + "dfs.namenode.rpc-address.ns2.nn1" -> "nn3.example.com:8020")) + assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons") + assert( + nameNodeProperties(claimed.head) == Map( + "hdfs.name-node.ns1" -> "nn1.example.com:9000,nn2.example.com:8020", + "hdfs.name-node.ns2" -> "nn3.example.com:8020")) + } + + test("iceberg scan: a catalog hdfs.name-node. wins over the Hadoop-derived declaration") { + assume(icebergAvailable, "Iceberg not available in classpath") + val (claimed, reasons) = planIcebergScanOverHdfs( + Seq("hdfs://ns1/warehouse/db/t/a.parquet"), + declaredNs1, + Seq("hdfs.name-node.ns1" -> "nn9.example.com:8020")) + assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons") + assert( + nameNodeProperties(claimed.head) == Map("hdfs.name-node.ns1" -> "nn9.example.com:8020"), + "the catalog value must reach native") + } + + test("iceberg scan: a nameservice declared only by the catalog's hadoop.* keys is claimed") { + assume(icebergAvailable, "Iceberg not available in classpath") + // Comet derives declarations from the session configuration alone, but iceberg-rust reads + // Hadoop's own keys from the properties too, which a catalog forwards as `hadoop.*`. + val (claimed, reasons) = planIcebergScanOverHdfs( + Seq("hdfs://nsb/warehouse/db/t/a.parquet"), + catalogProps = Seq( + "hadoop.dfs.nameservices" -> "nsb", + "hadoop.dfs.ha.namenodes.nsb" -> "nn1,nn2", + "hadoop.dfs.namenode.rpc-address.nsb.nn1" -> "a.example.com:8020", + "hadoop.dfs.namenode.rpc-address.nsb.nn2" -> "b.example.com:8020")) + assert(claimed.size == 1, s"expected the scan to be claimed, fell back with: $reasons") + val properties = claimed.head.nativeIcebergScanMetadata.get.catalogProperties + assert(properties.get("hadoop.dfs.ha.namenodes.nsb") == Some("nn1,nn2"), s"$properties") + assert(!properties.contains("hdfs.name-node.nsb"), s"$properties") + } + + test("iceberg scan: HDFS data files of a table whose metadata is local are declined") { + assume(icebergAvailable, "Iceberg not available in classpath") + // The native scan builds its one storage from the metadata location, a local path here, and + // would open the `hdfs://` file through it; declined at planning rather than failing at open. + val (claimed, reasons) = planIcebergScanOverHdfs( + Seq("hdfs://nn.example.com:8020/warehouse/db/t/a.parquet"), + metadataPrefix = "") + assert(claimed.isEmpty, "a local-metadata table with HDFS data files must not be claimed") + assert( + reasons.exists(r => r.contains("'hdfs'") && r.contains(mixedStorageReason)), + s"expected the mixed-storage fall-back reason, got: $reasons") + } + + test("iceberg scan: a local data file of a table whose metadata is on HDFS is declined") { + assume(icebergAvailable, "Iceberg not available in classpath") + val (claimed, reasons) = + planIcebergScanOverHdfs(Seq(s"file://${fakeRootDir.getAbsolutePath}/t/a.parquet")) + assert(claimed.isEmpty, "an HDFS-metadata table with a local data file must not be claimed") + assert( + reasons.exists(r => r.contains("'file'") && r.contains(mixedStorageReason)), + s"expected the mixed-storage fall-back reason, got: $reasons") + } + + test("iceberg scan: S3 data files of a table whose metadata is local are declined") { + assume(icebergAvailable, "Iceberg not available in classpath") + // Not an HDFS matter: the single storage fails the same way for an object store. + val (claimed, reasons) = + planIcebergScanOverHdfs(Seq("s3://bucket/warehouse/db/t/a.parquet"), metadataPrefix = "") + assert(claimed.isEmpty, "a local-metadata table with S3 data files must not be claimed") + assert( + reasons.exists(r => r.contains("'s3'") && r.contains(mixedStorageReason)), + s"expected the mixed-storage fall-back reason, got: $reasons") + } + + test("iceberg scan: HDFS data files on a Kerberos cluster are declined") { + assume(icebergAvailable, "Iceberg not available in classpath") + // The native client cannot use an executor's delegation tokens, so any HDFS location falls + // back once the cluster is secured, declared nameservice or not. + for (location <- Seq( + "hdfs://ns1/warehouse/db/t/a.parquet", + "hdfs://nn.example.com:8020/warehouse/db/t/a.parquet")) { + val (claimed, reasons) = planIcebergScanOverHdfs( + Seq(location), + declaredNs1 :+ ("hadoop.security.authentication" -> "kerberos")) + assert(claimed.isEmpty, s"$location must not be claimed on a Kerberos cluster") + assert( + reasons.exists(_.contains("cannot authenticate to HDFS with delegation tokens")), + s"expected the Kerberos fall-back reason for $location, got: $reasons") + } + } } /** A store key that counts the calls to its `equals`. */ 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 6ff5daf27fe..afb3de2a55c 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 @@ -19,11 +19,17 @@ package org.apache.comet.serde.operator +import java.io.File import java.lang.reflect.InvocationTargetException +import java.net.URI +import java.nio.file.Files +import java.util.concurrent.atomic.AtomicInteger import org.scalatest.funsuite.AnyFunSuite import org.scalatest.matchers.should.Matchers +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.{FileStatus, Path, RawLocalFileSystem} import org.apache.iceberg.{DeleteFile, FileContent} import org.apache.iceberg.expressions.Expressions import org.apache.spark.sql.catalyst.expressions.AttributeReference @@ -240,6 +246,207 @@ class CometIcebergNativeScanSuite extends AnyFunSuite with Matchers { out.values.toSet should not contain "https://some.example.com" } + test("HDFS nameservices are declared from the Hadoop configuration") { + val conf = new Configuration(false) + conf.set("dfs.ha.namenodes.ns1", "nn2, nn1") + conf.set("dfs.namenode.rpc-address.ns1.nn1", "host1:8020") + conf.set("dfs.namenode.rpc-address.ns1.nn2", "host2:8020") + // A NameNode without a port cannot be dialed, so ns2 is not declared. + conf.set("dfs.ha.namenodes.ns2", "nn1") + conf.set("dfs.namenode.rpc-address.ns2.nn1", "host3") + val authorities = Set("ns1", "ns2", "ns3", "host4:8020") + + val properties = CometIcebergNativeScan.hadoopToIcebergHdfsProperties(conf, authorities) + + // NameNodes keep their declaration order, an authority with a port needs no declaration, and + // one the configuration knows nothing of is a host on the default port. + properties shouldBe Map( + "hdfs.name-node.ns1" -> "host2:8020,host1:8020", + "hdfs.name-node.ns3" -> "ns3:8020") + CometIcebergNativeScan.undeclaredHdfsNameservice(authorities, properties) shouldBe Some("ns2") + // A catalog's own declaration counts in either form. + val catalog = Map("hdfs.name-node.ns2" -> "host3:8020", "hadoop.dfs.ha.namenodes.ns3" -> "a") + CometIcebergNativeScan.undeclaredHdfsNameservice( + authorities, + properties ++ catalog) shouldBe None + } + + test("HDFS authorities with a port need no declaration, IPv6 literals included") { + val conf = new Configuration(false) + conf.set("dfs.ha.namenodes.ns1", "nn1") + conf.set("dfs.namenode.rpc-address.ns1.nn1", "[::1]:8020") + val authorities = Set("[::1]:8020", "[2001:db8::1]:9000", "[::1]", "ns1") + + val properties = CometIcebergNativeScan.hadoopToIcebergHdfsProperties(conf, authorities) + + // The colons inside the brackets are not a port separator: a bracketed rpc-address keeps + // its port, and a bracketed literal without one is a portless host, so it gets the default + // port like any other undeclared host. + properties shouldBe Map( + "hdfs.name-node.ns1" -> "[::1]:8020", + "hdfs.name-node.[::1]" -> "[::1]:8020") + CometIcebergNativeScan.undeclaredHdfsNameservice(authorities, properties) shouldBe None + } + + test("an undeclared portless HDFS host is declared on Hadoop's default NameNode port") { + val conf = new Configuration(false) + conf.set("dfs.nameservices", "ns1") + val authorities = Set("nn.example.com", "ns1") + + val properties = CometIcebergNativeScan.hadoopToIcebergHdfsProperties(conf, authorities) + + // What DFSUtilClient.getNNAddress does for the JVM client: a host without a port is dialed on + // 8020. A name in dfs.nameservices is logical to Hadoop too; without NameNodes it stays + // undeclared rather than becoming a host. + properties shouldBe Map("hdfs.name-node.nn.example.com" -> "nn.example.com:8020") + CometIcebergNativeScan.undeclaredHdfsNameservice(authorities, properties) shouldBe Some("ns1") + } + + test("HDFS client settings the session sets are forwarded under hadoop.") { + val conf = new Configuration(false) + conf.set("dfs.client.use.datanode.hostname", "true") + conf.set("dfs.client.socket-timeout", "30000") + // Hadoop's own default is not a setting: forwarded, it would override the executors' XML. + conf.set("dfs.client.failover.max.attempts", "15", "hdfs-default.xml") + // What the client does not read, or must not be given, stays behind. + conf.set("dfs.client.read.shortcircuit", "true") + conf.set("dfs.data.transfer.protection", "privacy") + conf.set("dfs.client.failover.random.order.ns1", "true") + + CometIcebergNativeScan.hadoopToIcebergHdfsProperties(conf, Set("nn1:8020")) shouldBe Map( + "hadoop.dfs.client.use.datanode.hostname" -> "true", + "hadoop.dfs.client.socket-timeout" -> "30000") + // A scan with no HDFS files needs none of it. + CometIcebergNativeScan.hadoopToIcebergHdfsProperties(conf, Set.empty) shouldBe Map.empty + } + + test("the encryption-zone probe lists each parent directory once and keeps the listing") { + val root = Files.createTempDirectory("comet_ez_probe").toFile + def dir(relative: String): String = { + val dir = new File(root, relative) + assert(dir.mkdirs()) + s"${CountingHdfsFileSystem.PREFIX}${dir.getAbsolutePath}" + } + val dirs = (1 to 3).map(i => dir(s"data/p=$i")).toSet + val conf = new Configuration(false) + conf.set("fs.hdfs.impl", classOf[CountingHdfsFileSystem].getName) + conf.set("fs.hdfs.impl.disable.cache", "true") + // A configured key provider makes the probe run; the local filesystem has no zones. + conf.set("hadoop.security.key.provider.path", "kms://http@kms.example.com:9600/kms") + CountingHdfsFileSystem.reset() + + CometIcebergNativeScan.hdfsEncryptionZoneFallback(conf, dirs) shouldBe None + // The three partition directories share a parent: one listing, no per-directory status. + CountingHdfsFileSystem.listStatusCalls.get() shouldBe 1 + CountingHdfsFileSystem.getFileStatusCalls.get() shouldBe 0 + + CometIcebergNativeScan.hdfsEncryptionZoneFallback(conf, dirs) shouldBe None + CountingHdfsFileSystem.listStatusCalls.get() shouldBe 1 + + // A directory the kept listing does not have, a new partition, lists its parent again. + CometIcebergNativeScan.hdfsEncryptionZoneFallback(conf, dirs + dir("data/p=4")) shouldBe None + CountingHdfsFileSystem.listStatusCalls.get() shouldBe 2 + + // An expired listing is taken again: after the TTL when one is set, after an hour otherwise. + // A listing carries the clock it was taken with, so the second clock counts from the first. + val later = System.currentTimeMillis() + 61 * 1000 + CometIcebergNativeScan.hdfsEncryptionZoneFallback( + conf, + dirs, + ttlSeconds = 60, + nowMillis = later) shouldBe None + CountingHdfsFileSystem.listStatusCalls.get() shouldBe 3 + CometIcebergNativeScan.hdfsEncryptionZoneFallback( + conf, + dirs, + nowMillis = later + 3601 * 1000) shouldBe None + CountingHdfsFileSystem.listStatusCalls.get() shouldBe 4 + + // Another table at the same directories, say one re-created after a purge, has its own + // listing. + CometIcebergNativeScan.hdfsEncryptionZoneFallback( + conf, + dirs, + tableUuid = Some("re-created")) shouldBe None + CountingHdfsFileSystem.listStatusCalls.get() shouldBe 5 + CometIcebergNativeScan.hdfsEncryptionZoneFallback( + conf, + dirs, + tableUuid = Some("re-created")) shouldBe None + CountingHdfsFileSystem.listStatusCalls.get() shouldBe 5 + + // More entries than the budget: the scan is declined once that many are listed, the parents + // listed whole are kept, and the next scan continues from them. + val spread = (1 to 3).map(i => dir(s"spread/t=$i/p=1")).toSet + CountingHdfsFileSystem.reset() + val declined = + CometIcebergNativeScan.hdfsEncryptionZoneFallback(conf, spread, maxEntries = 2) + assert(declined.exists(_.contains("2 directory entries")), declined) + CountingHdfsFileSystem.listStatusCalls.get() shouldBe 2 + CometIcebergNativeScan.hdfsEncryptionZoneFallback(conf, spread, maxEntries = 2) shouldBe None + CountingHdfsFileSystem.listStatusCalls.get() shouldBe 3 + + // A directory at the filesystem root has no parent to list. + CometIcebergNativeScan.hdfsEncryptionZoneFallback( + conf, + Set(s"${CountingHdfsFileSystem.PREFIX}/")) shouldBe None + } + + test("the encryption-zone probe lists a parent of many partitions once, in linear time") { + val conf = new Configuration(false) + conf.set("fs.hdfs.impl", classOf[CountingHdfsFileSystem].getName) + conf.set("fs.hdfs.impl.disable.cache", "true") + conf.set("hadoop.security.key.provider.path", "kms://http@kms.example.com:9600/kms") + val prefix = CountingHdfsFileSystem.PREFIX + def partitions(parent: String): Set[String] = + (1 to 50000).map(i => s"$prefix/$parent/p=$i").toSet + // Listings of these parents are made up, not read from disk. + CountingHdfsFileSystem.synthetic = Map(s"$prefix/wide" -> 50000, s"$prefix/wider" -> 50000) + CountingHdfsFileSystem.reset() + try { + val wide = partitions("wide") + val keptBefore = CometIcebergNativeScan.keptEncryptionZoneEntries + val started = System.nanoTime() + CometIcebergNativeScan.hdfsEncryptionZoneFallback( + conf, + wide, + tableUuid = Some("wide")) shouldBe None + val elapsedMillis = (System.nanoTime() - started) / 1000000 + // One listing, no per-directory status, and nothing quadratic in the partitions: a lookup + // that scanned the listing per directory took minutes here. + CountingHdfsFileSystem.listStatusCalls.get() shouldBe 1 + CountingHdfsFileSystem.getFileStatusCalls.get() shouldBe 0 + assert(elapsedMillis < 10000, s"the probe took $elapsedMillis ms over 50000 partitions") + CometIcebergNativeScan.keptEncryptionZoneEntries - keptBefore shouldBe 50000 + + // The budget counts entries, not listings: a parent beyond it is declined and not kept. + val declined = CometIcebergNativeScan.hdfsEncryptionZoneFallback( + conf, + wide, + tableUuid = Some("over-budget"), + maxEntries = 10) + assert(declined.exists(_.contains("10 directory entries")), declined) + CountingHdfsFileSystem.listStatusCalls.get() shouldBe 2 + CometIcebergNativeScan.keptEncryptionZoneEntries - keptBefore shouldBe 50000 + + // What is kept is bounded in entries too, the least recently used parents leaving first: + // the first table's listing goes, and its next scan lists again. + CometIcebergNativeScan.hdfsEncryptionZoneFallback( + conf, + partitions("wider"), + tableUuid = Some("wider"), + maxKeptEntries = 60000) shouldBe None + assert(CometIcebergNativeScan.keptEncryptionZoneEntries <= 60000) + CometIcebergNativeScan.hdfsEncryptionZoneFallback( + conf, + wide, + tableUuid = Some("wide")) shouldBe None + CountingHdfsFileSystem.listStatusCalls.get() shouldBe 4 + } finally { + CountingHdfsFileSystem.synthetic = Map.empty + } + } + private def keyMetadataMethod(clazz: Class[_]) = clazz.getMethod("keyMetadata") private def serializeDeleteFile(file: AnyRef) = @@ -395,3 +602,44 @@ class CometIcebergNativeScanSuite extends AnyFunSuite with Matchers { def keyMetadata(): java.nio.ByteBuffer = null } } + +/** + * A local-disk `hdfs://` filesystem that counts the NameNode calls the planner would make. The + * local `listStatus` stats each child itself, which a NameNode listing does not, so those inner + * calls are not counted. + */ +class CountingHdfsFileSystem extends RawLocalFileSystem { + RawLocalFileSystem.useStatIfAvailable() + private var listing = false + override def getScheme: String = "hdfs" + override def getUri: URI = URI.create(CountingHdfsFileSystem.PREFIX) + override def listStatus(path: Path): Array[FileStatus] = { + CountingHdfsFileSystem.listStatusCalls.incrementAndGet() + CountingHdfsFileSystem.synthetic.get(path.toString) match { + case Some(children) => + Array.tabulate(children)(i => + new FileStatus(0, true, 0, 0, 0, new Path(path, s"p=${i + 1}"))) + case None => + listing = true + try super.listStatus(path) + finally listing = false + } + } + override def getFileStatus(path: Path): FileStatus = { + if (!listing) CountingHdfsFileSystem.getFileStatusCalls.incrementAndGet() + super.getFileStatus(path) + } +} + +object CountingHdfsFileSystem { + val PREFIX = "hdfs://counting" + val listStatusCalls = new AtomicInteger() + val getFileStatusCalls = new AtomicInteger() + + /** Parents whose listing is made up of that many `p=` directories, by path. */ + @volatile var synthetic: Map[String, Int] = Map.empty + def reset(): Unit = { + listStatusCalls.set(0) + getFileStatusCalls.set(0) + } +}