[mixermt]

Which issue does this PR close?

Closes #5894.

Rationale for this change

Comet's native Iceberg scan had no hdfs arm, so every Iceberg table on HDFS fell back to the JVM reader, while the plain-Parquet native scan already reads HDFS through libhdfs. On-premise Iceberg deployments are commonly HDFS-backed and got no native acceleration.

apache/iceberg-rust now ships a pure-Rust HDFS backend in iceberg-storage-opendal behind the opendal-hdfs-native feature (OpenDAL's services-hdfs-native: no JNI, no libhdfs). This PR is the Comet side of it: native reads of Iceberg tables on HDFS. Writes to HDFS keep Iceberg's own writer.

Two facts about that backend shape the design:

• It dials an authority with a port as written, but a portless one (hdfs://ns1/..., an HA nameservice) only through a declaration in the FileIO properties: hdfs.name-node.ns1, or Hadoop's own dfs.ha.namenodes.ns1 keys passed as hadoop.*. It never resolves a nameservice from $HADOOP_CONF_DIR, has no default port, and caches one client per effective NameNode.
• OpenDAL builds every client for a synthetic nameservice, and the client looks delegation tokens up by that name. An executor on a Kerberos cluster holds tokens issued for ha-hdfs:<ns> or a NameNode address and no TGT, so it cannot authenticate. Kerberos clusters therefore fall back.

Worth stating plainly: this is a second, independent HDFS client in the same process. The plain-Parquet path reaches HDFS through libhdfs/JNI; an Iceberg table is opened over pure-Rust RPC. Both read the same $HADOOP_CONF_DIR XML on the executor, but they hold separate connections and separate credentials.

What changes are included in this PR?

• Moves the iceberg-rust pin to ca21f158, which contains the HDFS backend, and enables opendal-hdfs-native. hdfs-native resolves to 0.14.7; releases before 0.14.5 fail every read on a cluster whose dfs.checksum.type is CRC32 (hdfs-native's CRC32 checksum verification fix shipped in 0.14.5).
• Native (iceberg_common.rs): storage_factory_for routes hdfs to OpenDalStorageFactory::HdfsNative; hdfs joins the read list of builtin_storage_schemes and stays out of the write list; hdfs. and hadoop. join STORAGE_PROPERTY_PREFIXES, so the declarations reach the FileIO.
• JVM: validateIcebergFileScanTasks collects the HDFS authorities of the data and delete files. CometIcebergNativeScan.hadoopToIcebergHdfsProperties declares each portless one as hdfs.name-node.<ns> from the session's dfs.ha.namenodes.<ns> and dfs.namenode.rpc-address.<ns>.<id>, all-or-nothing per nameservice (a partial list would turn a failover into an outage), with the NameNodes in declaration order. It is merged before the catalog's properties, so a catalog's own declaration wins. The scan falls back at planning, instead of failing on an executor, when a portless authority stays undeclared or when hadoop.security.authentication is not simple.
• JVM: a scan whose data or delete files are not all on the same storage backend as the table metadata falls back. The native scan builds one FileIO from the metadata location and opens every file through it, so an hdfs:// data file under local metadata (or a local file under HDFS metadata, or S3 data under local metadata, which was already claimed before this PR) failed at open time. The planner now compares the storage family of every data/delete file (file, s3/s3a, each S3-compliant alias, gs, oss, hdfs) with the metadata location's.
• JVM: HDFS files in an encryption zone fall back. hdfs-native is built without its kms feature and rejects every encrypted file. When the NameNode advertises a key provider (one client-cached getServerDefaults call) or the session configures one, the planner checks each distinct data/delete directory with FileStatus.isEncrypted(), which the NameNode sets for directories inside a zone too; clusters without a key provider pay nothing further.
• JVM: a scan of a table with a name mapping (schema.name-mapping.default, which migrate, snapshot and add_files set) that reads a list or map column falls back. iceberg-rust at this pin reads a file without field ids through the name mapping, but cannot open the list layouts Hive and Spark's writeLegacyFormat write that way: the read schema it supplies keeps a field id on the list's element field and the Parquet reader rejects it as incompatible. The layout is only known from a file's footer, so the scan declines at planning. Iceberg's own TestCreateActions migrates such files and reads them back with Comet enabled; without this gate its Hive-style and two-level list cases fail.
• pom.xml: hadoop.version, which pins the hadoop-client-minicluster the HDFS test fixture needs, now follows each Spark profile's Hadoop line, as hadoop-azure.version already does. With the global 3.3.4, MiniDFSCluster cannot start on the Spark 4.x profiles: its HttpServer2 resolves shaded Jetty classes against Spark's newer hadoop-client-runtime, which left every WithHdfsCluster suite unrunnable there. This is pre-existing and not specific to Iceberg.
• Docs: an HDFS section in the Iceberg guide, the fallback cases, and the gaps below.

How are these changes tested?

• CometIcebergHdfsSuite (new): end to end against in-process MiniDFSClusters, on Spark 3.4 through 4.1 (4.2 has no Iceberg runtime; the suite skips there and says so). A plain read, a pushed-down filter, and a partitioned table across several data files, each asserting a single CometIcebergNativeScanExec and result parity with Spark, and the first that the data location really is hdfs://. Then a two-NameNode HA cluster addressed by nameservice, read before and after a failover and again after the first NameNode is shut down, asserting on the planned scan that the declaration Comet derived is exactly what the client was given. Then a cluster with a jceks key provider and an encryption zone: a table inside the zone falls back with the zone named and matches Spark, a table outside it on the same NameNode still scans natively.
• CometScanSchemeFallbackSuite: planner-level cases that drive the real CometScanRule over a real Iceberg table (its warehouse on a fake hdfs:// filesystem, so metadata and data share a storage) whose manifests name hdfs:// data files, so nothing executes: a declared nameservice is claimed with its NameNode list; a NameNode without an rpc-address, a nameservice listed only in dfs.nameservices, and a portless plain host are declined; a host:port location needs no declaration; two declared nameservices in one scan are both declared; a catalog's hdfs.name-node.<ns> wins over the derived one; a nameservice declared only by the catalog's hadoop.* keys is claimed; any HDFS location is declined on a Kerberos cluster; HDFS data under local metadata, a local file under HDFS metadata, and S3 data under local metadata are each declined for mixing storage. Plus the scheme lists (hdfs readable, not writable) and the hostless hdfs:/// form.
• CometIcebergNativeSuite: a table imported from legacy-format Parquet with a name mapping falls back for its list column and stays native for the primitives.
• CometIcebergNativeScanSuite: declaration order, a NameNode without a port, an undeclared nameservice, an authority with a port, a catalog's own declaration, and bracketed IPv6 authorities.
• Rust (iceberg_common): hdfs resolves for reads and is rejected for writes; hdfs. and hadoop. properties reach the FileIO; the FileIO cache key separates catalogs that differ only in their HDFS declaration or client overrides; and a contract test that the pinned iceberg-rust expands hdfs.name-node.<ns> into Hadoop's own keys and rejects a portless entry when the storage is built.
• The pin also changes how iceberg-rust matches nested fields in data files that have neither field ids nor a name mapping: a read of only some of a struct's fields now fails instead of matching by position. The test for such files expects that, and the Iceberg guide documents it with the workaround (spark.comet.scan.icebergNative.nestedSchemaPruning.enabled=false).
• Locally: cargo test -p datafusion-comet --lib (961 tests), cargo clippy --all-targets --workspace -- -D warnings, cargo fmt --check; on Spark 4.1, with scalastyle: CometIcebergHdfsSuite, CometIcebergNativeScanSuite, CometScanSchemeFallbackSuite, CometIcebergWriteDetectionSuite, CometIcebergNativeSuite, CometIcebergWriteActionSuite, CometIcebergResidualPushdownSuite, CometIcebergEncryptionSuite, CometIcebergRewriteActionSuite and CometFuzzIcebergSuite, all passing (405 tests, 2 assume cancels); on Spark 3.5: the HDFS, planner and CometIcebergNativeSuite suites, 183 passing (10 version-gated cancels); spotless, dev/ci/check-suites.py and dev/ci/check-ci-config.py. dev/local-ci.sh iceberg (Iceberg 1.11 on Spark 4.1, the merge-queue target): all four shards, the extensions suite and the runtime integration test pass. Without the name-mapping fallback above, shard-4 failed TestCreateActions' Hive-style and two-level list cases.

Known gaps

• Kerberos clusters fall back, as described above. Fixing that needs OpenDAL to build the client for the real nameservice, or hdfs-native to find tokens by NameNode address.
• Without Kerberos, the client connects as HADOOP_USER_NAME, or as the executor's operating system user when that is unset, rather than as the Spark user.
• Files in an HDFS encryption zone fall back: OpenDAL does not enable hdfs-native's kms feature. Enabling it is not enough on its own: hdfs-native reads the KMS URI only from hadoop.security.key.provider.path, not from the NameNode's default, and rejects any non-kms:// provider when the client is built.
• Data and delete files on a storage backend other than the metadata's fall back; routing each file to its own backend is a follow-up.
• Tables with a name mapping lose native reads of their list and map columns, standard layouts included, until iceberg-rust can open legacy list layouts through a name mapping.
• A pruned read of some fields of a struct in a data file with neither field ids nor a name mapping fails (see above); reading the whole struct, or disabling nested schema pruning, avoids it.
• A portless single-NameNode location (hdfs://namenode/...) falls back: iceberg-rust has no default port, so to it that is an undeclared nameservice. Write the port, or declare it as hdfs.name-node.namenode on the catalog.
• The HDFS clients run on iceberg-rust's own Tokio runtime, one per process, and the FileIO cache keys on the metadata location, so each new snapshot of a table builds a new client.

AI Disclosure

Developed with AI assistance (Claude Code): drafting the implementation, tests, and this description. I reviewed the changes and ran all verification locally.

🤖 Generated with Claude Code

[comphead]

Sounds good, thanks @mixermt

[mixermt]

Found that CometIcebergNativeScanExec does not bridge bytes read or records read into Spark's standard TaskMetrics.inputMetrics, As a result, the Spark UI reports Input: 0 B / 0 records for stages dominated by Comet native scans.
Fixing this

[comphead]

@mixermt congrats on landing apache/iceberg-rust#3111. I opened #7157 as a draft that wires the merged backend into Comet, building on your design here.

It follows the merged semantics: each portless authority is declared as hdfs.name-node.<ns> from the session's Hadoop config, and there is no single-NameNode gate. It also falls back on Kerberos clusters. opendal's hdfs-native service connects every client to a synthetic nameservice, so hdfs-native looks up an ha-hdfs:nameservice token and never finds an executor's delegation tokens.

Reviews welcome, and happy to coordinate if you would rather carry this in your PR.

[mixermt]

@mixermt congrats on landing apache/iceberg-rust#3111. I opened #7157 as a draft that wires the merged backend into Comet, building on your design here.

It follows the merged semantics: each portless authority is declared as hdfs.name-node.<ns> from the session's Hadoop config, and there is no single-NameNode gate. It also falls back on Kerberos clusters. opendal's hdfs-native service connects every client to a synthetic nameservice, so hdfs-native looks up an ha-hdfs:nameservice token and never finds an executor's delegation tokens.

Reviews welcome, and happy to coordinate if you would rather carry this in your PR.

Personally, I prefer to continue this from my PR 🙏 as I've invested not trivial time in actual testing this on high-scale workloads.

[sunchao]

Reviewed the full 16-file diff at b493b627ecb16d5b2814cfa6b349becef9ba557d against 69111f68a5ff62904342910c445d191f53615aff. The PR remains non-draft. Found three introduced P2 issues.

Read all four conversation comments. There are no submitted reviews, inline comments, or review threads. Routed skills: review-comet-pr, review-comet-expression-pr for the serde, and review-comet-iceberg-write-pr for the dependency pin and shared storage implementation.

Summary

• Prior state and problem: HDFS-backed Iceberg scans previously fell back to Iceberg’s JVM reader. This PR adds native HDFS reads while retaining JVM writes.
• Design approach: Enable iceberg-rust’s OpenDAL HDFS backend, translate HA nameservice declarations from Hadoop configuration, and check eligibility during planning.
• Correctness: The HA declaration handling and name-mapped container fallback have targeted coverage. Three failure cases remain: pruned structs without field IDs, HDFS files paired with metadata on another filesystem, and HDFS encryption-zone files. Checked Spark configuration handling against sources for 3.4.3, 3.5.9, 4.0.4, and 4.1.3, and Iceberg reader behavior across 1.5.2, 1.8.1, 1.10.0, and 1.11.0.
• Compatibility analysis: Compared against branch-1.2 at 889856dc5af821de1f7c6ca2e9538bd663431d04. Its affected production sources and iceberg-rust pin match the supplied base. Native HDFS routing and the name-mapped container fallback are intended changes. The three reported execution failures are regressions. Spark 4.2 has no configured Iceberg runtime, so the new integration suite skips it.
• Key design decisions: HDFS remains read-only, complete HA declarations preserve NameNode order, catalog declarations override derived declarations, and Kerberos scans fall back. These are sensible boundaries.
• Implementation sketch: The change extends the native scheme/property lists, collects HDFS authorities during existing task validation, adds two Scala configuration helpers, aligns MiniDFSCluster dependencies with Spark profiles, and registers the new suite on Linux and macOS.
• Performance: Authority collection reuses the existing task traversal. The backend caches clients, although snapshot-specific FileIO keys can create additional clients and the backend owns a separate runtime. No benchmark was run, and no separate evidence-backed P1/P2 performance finding was identified.
• Design: Reusing the existing planner gate and storage factory keeps the implementation understandable. Eligibility must also account for the single backend selected from the metadata URI and for unsupported HDFS encryption.
• Abstraction & complexity: The small nameservice helpers and existing property forwarding avoid unnecessary protocol changes. A conservative filesystem-consistency guard would address the mixed-filesystem failure without requiring a new general routing abstraction.
• Behavioral changes worth calling out: Native Iceberg scanning is enabled by default, so eligible HDFS tables automatically change readers. Name-mapped list/map projections now fall back, while primitive projections can remain native. The dependency bump also changes some previously successful nested projections into exceptions.
• Suggested improvements: Preserve Spark-compatible handling of pruned files without field IDs, decline incompatible metadata/data filesystem combinations, and retain JVM reads for encryption-zone files until native KMS support is wired and validated. Add parity coverage for these specific triggers.

Exact-head CI: Comet CI and CodeQL both report action_required. Only the labeling workflow succeeded. There is no exact-head build or test verdict.

Validation: Ran standalone reproductions against both exact iceberg-rust revisions with Arrow/Parquet 59.3.1 and OpenDAL 0.58.2. The old pin returns two NULL rows for the pruned-struct case, while the new pin errors. The FileIO reproduction confirms that HDFS paths are incorrectly opened through the local backend when selected from local metadata. check-suites.py and check-ci-config.py passed. Full Comet JVM/native suites and live HDFS/HA/KMS tests were not run. The encryption-zone finding is verified from the pinned dependency’s feature configuration and unconditional error path. The author’s reported integration runs were not independently reproduced. Project code is unchanged.

[sunchao]

[P2] Preserve compatible reads of pruned structs when updating this pin. Import a Parquet file containing s STRUCT<a: INT, b: INT> without field IDs or a name mapping, then run SELECT s.b with the default nested pruning enabled. The new revision throws cannot reconcile struct fields by id; Spark completes the query. With b containing only NULLs, the base also returns the same NULL rows as Spark, isolating a newly introduced failure. The changed test now expects this exception instead of preventing it. Please add compatible handling or a JVM fallback for this case and retain a Spark-parity assertion.

Evidence: Executed standalone ArrowReader reproductions with iceberg-rust 1f3bc34058e3f3e45336bc5428f37b92649a608b and ca21f1587c6caaeb88dd3a9e1cc245747e1aba08, both using Arrow/Parquet 59.3.1. For rows s=(10,NULL),(20,NULL), the pruned base read succeeds with two NULLs and the head read errors. Logs: /tmp/5898-b493-base-exact-arrow.log and /tmp/5898-b493-head-exact-arrow-online.log. Iceberg Java’s fallback schema assigns only top-level IDs and its optional-field reader supplies NULLs. The PR’s CometIcebergNativeSuite.scala:2814 expects the same new failure.

[mixermt]

Pushing back on this one, under these assumptions: (a) files written by Iceberg writers carry field ids, so this only concerns tables imported from pre-existing Parquet without a name mapping (migrate/snapshot/add_files set one; only the low-level importSparkTable does not); (b) Spark returns NULL for every id-less nested field (#6794), so the base read of s.b that returned the file's values was not parity either — the repro agrees only because b is all-NULL; (c) the pin has to pass apache/iceberg-rust#3255 to reach the HDFS backend, and nothing at planning time tells id-less files apart. Parity (NULL, pruned or not) lands with apache/iceberg-rust#3437; once the pin passes it, this test becomes a checkSparkAnswer assertion. Until then the documented knob applies.