[mixermt]

Which issue does this PR close?

Closes #5894.

Rationale for this change

Comet's native Iceberg scan has no hdfs arm, so every Iceberg table on HDFS falls 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 get no native acceleration.

apache/iceberg-rust 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: native reads of Iceberg tables on HDFS. Writes to HDFS keep Iceberg's own writer.

The backend has three properties that shape the design:

• It dials an authority with a port as written, and resolves a portless one (hdfs://ns1/...) 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 reads a nameservice from $HADOOP_CONF_DIR, has no default port, and keeps one client per NameNode.
• It connects as HADOOP_USER_NAME, or as the process's operating system user, and OpenDAL exposes no user setting; the JVM reader runs as the Spark user.
• It is built without hdfs-native's kms feature and rejects every file of an encryption zone, and it cannot authenticate with the delegation tokens executors hold on a Kerberos cluster.

This is a second, independent HDFS client in the executor process: the plain-Parquet path uses libhdfs/JNI, an Iceberg table is read over pure-Rust RPC. Both read the same $HADOOP_CONF_DIR XML, with separate connections and credentials.

What changes are included in this PR?

Native:

• The iceberg-rust pin moves to de0a6ac8, apache main past the HDFS backend (feat(storage): Add HDFS support via opendal services-hdfs-native iceberg-rust#3111) and past fix(arrow): read nested struct fields as null for files without ids or name mapping iceberg-rust#3437, with opendal-hdfs-native enabled; hdfs-native resolves to 0.14.7 (releases before 0.14.5 fail every read on a dfs.checksum.type=CRC32 cluster). What the bump changes for reads is under "Reads that change" below.
• storage_factory_for routes hdfs to OpenDalStorageFactory::HdfsNative; hdfs is in the read list of builtin_storage_schemes and not in the write list; hdfs. and hadoop. join STORAGE_PROPERTY_PREFIXES.

Planner (CometScanRule, CometIcebergNativeScan), all decided at planning so that no task fails on an executor:

• NameNodes. validateIcebergFileScanTasks collects the HDFS authorities of the data and delete files. A portless one that the session's Hadoop configuration declares as a nameservice (dfs.ha.namenodes.<ns> with a host:port dfs.namenode.rpc-address.<ns>.<id> for each NameNode) becomes hdfs.name-node.<ns>, all-or-nothing per nameservice and in declaration order; one the configuration knows nothing of becomes host:8020, as DFSUtilClient.getNNAddress dials it for the JVM client; a name in dfs.nameservices without NameNodes, or an authority-less hdfs:/// location, falls back. A catalog's own hdfs.name-node.<ns> or hadoop.dfs.ha.namenodes.<ns> wins over the derived value.
• Identity. The scan is claimed only when the executors are known to read as the Spark user: spark.executorEnv.HADOOP_USER_NAME set to it (in local mode, the JVM's HADOOP_USER_NAME or OS user). Otherwise it falls back with a reason naming the setting. Carrying the user into the client needs an OpenDAL user option and an iceberg-rust hdfs.user passthrough; hdfs-native already has ClientBuilder::with_user.
• Kerberos. hadoop.security.authentication other than simple falls back.
• Encryption zones. On a cluster whose NameNode advertises a key provider, or whose session configures one, each distinct data/delete directory is checked with the isEncrypted flag the NameNode sets on every path inside a zone: one listStatus per parent directory (the partitions of a table cost one call), cached per filesystem and directory for the process. An encrypted directory, or one whose status cannot be read, falls back.
• Storage families. The native scan builds one FileIO from the metadata location and opens every data and delete file through it, so a scan whose files are not all in the metadata location's storage family (file, s3/s3a, each S3-compliant alias, gs, oss, hdfs) falls back. This also covers combinations that were claimed and failed before, such as local metadata with s3:// data files.

Build and docs:

• pom.xml: hadoop.version (the hadoop-client-minicluster the HDFS fixture needs) follows each Spark profile's Hadoop line, as hadoop-azure.version does; with the global 3.3.4, MiniDFSCluster cannot start on the Spark 4.x profiles.
• The Iceberg guide gains an HDFS section, the fallback cases, and the reads that change.

Reads that change with the pin

• A data file with neither field ids nor a name mapping: a struct's fields now read as NULL, as in Spark (fix(arrow): read nested struct fields as null for files without ids or name mapping iceberg-rust#3437), whether the struct is read whole or pruned; the release returned the file's values natively (Native Iceberg scan returns nested values that Spark reads as NULL in files without field ids or a name mapping #6794). CometIcebergNativeSuite asserts parity for both reads. The containers still differ, and are documented: 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 keeps the elements with NULL fields.
• A name-mapped table (migrate, snapshot, add_files): nested columns fall back on main since fix: match Spark nullness for missing nested Parquet fields #7085, and this PR keeps that gate. At this pin iceberg-rust reads name-mapped structs and standard lists and maps correctly, but not the list layouts Hive and writeLegacyFormat write; narrowing the gate is tracked in Bump iceberg-rust past apache/iceberg-rust#3255 and narrow the nested schema evolution fallback #6934.

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, a partitioned table across several data files; a two-NameNode HA cluster addressed by nameservice, read before and after a failover and after the first NameNode is shut down, asserting on the planned scan that the derived declaration is exactly what the client was given; one scan over data files on two NameNodes (the host:port cluster and the HA nameservice); and a cluster with a jceks key provider and an encryption zone, where a partitioned table inside the zone falls back with the zone named and matches Spark while a table outside it still scans natively.
• CometScanSchemeFallbackSuite: planner-level cases that drive the real CometScanRule over a real Iceberg table whose warehouse sits on a fake hdfs:// filesystem and whose manifests name hdfs:// data files, so nothing executes: a declared nameservice is claimed with its NameNode list; a NameNode without an rpc-address and a nameservice listed only in dfs.nameservices are declined; a portless host is claimed on 8020; a host:port location needs no declaration; two declared nameservices are both declared; a catalog's hdfs.name-node.<ns> wins; a nameservice declared only by the catalog's hadoop.* keys is claimed; a Kerberos cluster is declined; executors reading as another user are declined and HADOOP_USER_NAME set to the Spark user is claimed; HDFS data under local metadata, a local file under HDFS metadata and S3 data under local metadata are declined for mixing storage.
• CometIcebergNativeScanSuite: declaration order, a NameNode without a port, the default port for an undeclared host, bracketed IPv6 authorities, a catalog's own declaration; the encryption-zone probe lists each parent once and caches the answer (counted on a local hdfs:// filesystem).
• CometIcebergNativeSuite: a table imported from legacy-format Parquet with a name mapping falls back for its list column and stays native for the primitives; nested fields of a table imported without field ids or a name mapping read as NULL in both readers, pruned and whole.
• 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 in their HDFS declaration; a contract test that the pinned iceberg-rust expands hdfs.name-node.<ns> into Hadoop's keys and rejects a portless entry.
• Locally: cargo test -p datafusion-comet --lib, clippy and fmt; on Spark 4.1 the Iceberg suites (CometIcebergHdfsSuite, CometScanSchemeFallbackSuite, CometIcebergNativeScanSuite, CometIcebergNativeSuite, CometIcebergWriteDetectionSuite, CometIcebergWriteActionSuite, CometIcebergResidualPushdownSuite, CometIcebergEncryptionSuite, CometIcebergRewriteActionSuite, CometFuzzIcebergSuite); on Spark 3.5 the HDFS, planner, serde and native suites; spotless and scalastyle. dev/local-ci.sh iceberg (Iceberg 1.11 on Spark 4.1) passed on the previous revision of this branch.

Known gaps

• Kerberos clusters fall back: OpenDAL builds the client for a synthetic nameservice, so delegation tokens never match.
• The Spark user is not carried into the client; the executor identity requirement above stands until OpenDAL and iceberg-rust expose hdfs.user.
• Files in an HDFS encryption zone fall back; enabling hdfs-native's kms feature alone is not enough (it reads the KMS URI only from hadoop.security.key.provider.path and rejects any non-kms:// provider at client construction).
• Data and delete files on a storage backend other than the metadata's fall back; per-file routing is a follow-up.
• Nested columns of name-mapped tables fall back (fix: match Spark nullness for missing nested Parquet fields #7085, Bump iceberg-rust past apache/iceberg-rust#3255 and narrow the nested schema evolution fallback #6934).
• 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.

[andygrove]

Thanks @mixermt for the careful work here. The design notes on the HDFS backend, the all-or-nothing nameservice declaration, and the Kerberos fallback are clear, and the planner suites cover a lot of ground. I read the diff and the existing threads. I did not build the branch, so the points below come from reading the code and comparing against branch-1.2.

I am requesting changes because three things look like regressions against the latest release.

1. A pruned read of an id-less struct goes from a result to an error.
After the pin bump, a read of only some fields of a struct in a Parquet file with neither field ids nor a name mapping throws cannot reconcile struct fields by id. On branch-1.2 the same query returned the file's values. Spark returns NULL here (#6794), so the release was not at parity either, but raising an error where Spark returns a result is a different failure, and the workaround only exists in the docs. The planner cannot see footers, but could it fall back when the table has no name mapping and might hold id-less files? Or could the pin wait for apache/iceberg-rust#3437, which is still open?

2. List and map columns of name-mapped tables leave the native path.
The new fallback in CometScanRule fires for any scan that outputs a list or map at any depth when schema.name-mapping.default is set. That includes files with standard layouts, and those columns ran natively on the release. I understand TestCreateActions needs it with this pin. Do we know how many users migrate, snapshot and add_files tables cover? Would it make sense to land the HDFS wiring separately from the bump past apache/iceberg-rust#3255, which #6934 already tracks, so the two regressions above wait for the upstream fixes?

3. Native HDFS reads run as a different user than Spark.
The new docs say that on a simple-auth cluster the client connects as HADOOP_USER_NAME, or as the executor's OS user, rather than as the Spark user. Before this PR those tables were read through the JVM under the Spark user. With HDFS permissions on, a table the Spark user can read may now fail on an executor, with no fallback. It could also be read as a different identity than the query's. Is there a way to carry the Spark user into the client? If not, should native HDFS be behind a supported config that is off by default until it can?

Other things I would like answered before merge:

• CometIcebergNativeScan.hdfsEncryptedDirectory does one sequential getFileStatus per distinct data and delete directory at planning. How does that behave on a table with thousands of partitions, and does the rule re-run per query or per AQE pass? Could the probe be batched, capped or cached? Any exception, including a permission error, also turns the scan into a silent fallback with only a logWarning.
• The pin moves 52 iceberg-rust commits. You found two behavior changes. Could you list any others on default paths, using the open items in #6934 as a starting point?
• The storage-family check compares schemes only. Is there an end-to-end test where data files live on a different hdfs:// authority or nameservice than the metadata? The planner suite checks the property declarations, but not that a single FileIO serves both.
• A portless single-NameNode location (hdfs://namenode/...) falls back, and that is a common form. Is a default port reasonable here?
• The branch conflicts with main and CI has not run (action_required). After merging main, could you apply run-spark-3.5-tests, run-spark-4.0-tests, run-spark-4.1-tests and run-iceberg-tests, or run dev/local-ci.sh iceberg? Those suites otherwise first report in the merge queue.
• The description reads as an accumulated list. Could you rewrite it as the current state before merge?

[mixermt]

1. I'll hold the bump on fix(arrow): read nested struct fields as null for files without ids or name mapping iceberg-rust#3437 (approved) and move past it once it merges, so the pruned read returns NULL like Spark.
2. Since fix: match Spark nullness for missing nested Parquet fields #7085, main already falls back for every nested column of a name-mapped table; after syncing, this PR inherits that gate and drops its own narrower one. Narrowing it to structs is Bump iceberg-rust past apache/iceberg-rust#3255 and narrow the nested schema evolution fallback #6934's follow-up. feat: bump iceberg-rust past apache/iceberg-rust#3255 and narrow the nested schema evolution fallback #7028 carries the same bump; I can sequence this PR after it so the diff is HDFS-only.
3. hdfs-native has with_user, but opendal doesn't expose it and iceberg-rust ignores hdfs.user. Until that is wired upstream, the planner claims native HDFS only when the executors' identity is known to be the Spark user (spark.executorEnv.HADOOP_USER_NAME set to it; in local mode the JVM's user), and falls back naming the setting otherwise.
4. The probe now groups directories by parent and uses one listStatus per parent (the NameNode returns each child's encrypted flag), caches per filesystem and directory for the session, and the rule runs once per query (AQE re-plans unexecuted scans only). Errors keep declining; the reason is in EXPLAIN, not only the log.
5. Other changes in the range on default paths: [Incompatibility] Document arrays_overlap null handling differences #3364 fixes task failures on short BYTE_ARRAY decimal bounds; chore: add MacOS file and event trace log to gitignore #3070 makes an equality delete on case-colliding names error instead of picking one; fix: [df52] Fix/dictionary unpack tests #3473 reads no columns for an empty projection (COUNT(*) suites pass); fix: [iceberg] Mark nativeIcebergScanMetadata @transient #2930 adds arrow.uuid field metadata (uuid suites pass). The rest is API, catalog or CI.
6. Added an end-to-end test: metadata on one NameNode, a data file on the HA nameservice, one native scan reads both.
7. An undeclared portless host now resolves to port 8020, as DFSUtilClient.getNNAddress does for the JVM client; a name listed only in dfs.nameservices still falls back.
8. Description rewritten as current state.

[mixermt]

Update on 1 and 5: apache/iceberg-rust#3437 merged, so the pin moved to apache main at de0a6ac8. A pruned or whole read of a struct in a file without field ids or a name mapping now returns NULL fields like Spark, and CometIcebergNativeSuite asserts that parity instead of the error; what still differs (an optional struct itself, and a list of structs) is in the docs. Of the 21 commits between the pins, two touch Comet's paths, both fixes: a row group whose min/max is NaN is now might-match for an inequality residual (#3502), and the native writer drops NaN from column bounds (#3539). The rest is catalog, transaction, CI, or source-compatible signature changes (impl Into<Type> in NestedField constructors, #3527).