[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. Of the client settings hdfs-native reads from its configuration, the ones the session sets (dfs.client.use.datanode.hostname, dfs.client.socket-timeout, dfs.client.failover.{max.attempts,sleep.base.millis,sleep.max.millis}) are forwarded under hadoop., behind the catalog's own; Hadoop's defaults are not, since the client reads the executors' XML underneath.
• 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 on the SparkContext's configuration, the one that launched the executors (a value only in a session's SQLConf is ignored); 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), kept per filesystem, table UUID and parent for an hour, or for spark.sql.metadataCacheTTLSeconds when set, so a table re-created at the same path lists again; a directory missing from a kept listing lists its parent again, and a scan that would list more than a million directory entries falls back with the parents listed whole kept. The listings kept hold a million children at most, least recently used parents evicted first. An encrypted directory, or one whose status cannot be read, falls back.
• Switch. spark.comet.scan.icebergNative.hdfs.enabled (default true) turns HDFS reads off on their own: a scan that reads HDFS files falls back with the key in the reason, the other HDFS checks and the probe are skipped, and other storage is unaffected.
• 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; a merge-on-read table whose position deletes are read natively; and a cluster started as another superuser, where the native scan reads a data directory only the Spark user may enter and is denied, as Spark is, once it belongs to someone else.
• 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, both on the live SparkConf, and the SQLConf value does not stand in; the switch set to false declines with the key as the one HDFS reason; the client settings the session sets reach the FileIO behind the catalog's own; 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, keeps the listing, lists again for a new directory, another table UUID, or after the TTL or the default hour, and declines past the entry budget; over 50k partitions under one parent it lists once in linear time, keeps 50k entries, and evicts the older table past the kept bound (counted on a local hdfs:// filesystem); the client settings the session sets are forwarded and Hadoop's defaults are not.
• 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

[andygrove]

This review was prepared with AI assistance (Claude Code) and checked by the reviewer before posting.

Thanks @mixermt for the detailed answers and the follow-up commits. The identity test, the forwarded client settings, the batched and capped encryption probe, and the pin audit all look good. I compared the touched paths against branch-1.2 and found no unintended change in behavior. The storage-family fallback, native HDFS reads and the id-less struct NULLs are all intended changes.

I have two questions on the new HDFS gates, left inline. Both are cases where the scan would be claimed and then fail on an executor instead of falling back.

On CI, the label runs from 2026-10-30 ran at 81d5e291, before the latest push. I removed and re-added run-iceberg-tests, run-spark-3.5-tests, run-spark-4.0-tests and run-spark-4.1-tests so they run again at 4bf68ae5, and I approved them. There is nothing for you to run by hand. I will look at the results when they finish.

[andygrove]

Thanks for adding the identity test and the fallback reason. One question on the gate itself. It reads spark.executorEnv.HADOOP_USER_NAME from the session SQLConf first. A session built with SparkSession.builder().config(...).getOrCreate() on an existing context can hold that key in its SQLConf without the executors ever seeing it. The planner would then believe the executors read as the Spark user while the native client connects as their OS user. The result is an AccessControlException, or a read under an identity the Spark user did not have, with no fallback.

Would it make sense to read only the live conf here (SparkEnv.get.conf or sparkContext.getConf)? The planner tests set the value with withSQLConf, so they would need to set it on the live conf and restore it. A case where the SQLConf names the Spark user and the live conf names nobody, which must fall back, would pin it.

[mixermt]

Right: SparkSession.builder().config(...) on an existing context, or SET, puts the key in the SQLConf without the executors seeing it.

• The gate now reads only sparkContext.getConf, the configuration the executors were launched with; the SQLConf is not consulted.
• The planner tests set the live conf (SparkEnv.get.conf, restored afterwards). New case: the Spark user in the SQLConf and another user on the context is declined; the reverse is claimed, since in local mode the JVM's user applies, which is also why "SQLConf names the Spark user, live conf names nobody" cannot fall back there.
• The docs say the setting has to be given when the application is submitted.

[andygrove]

With spark.sql.metadataCacheTTLSeconds unset, which is its default, fresh is always true, so a cached "not encrypted" listing lives as long as the process. The argument that a directory holding files cannot become a zone root is right for the same directory. It does not cover a table that is dropped with purge and recreated at the same path inside a new encryption zone. In a long-lived driver such as a Thrift server, the scan would still be claimed and the tasks would fail on the encrypted files.

Could the entries get a default maximum age when the TTL is unset, say a few minutes? The function already takes nowMillis, so a test with an injected clock would cover it.

[mixermt]

Right, a purged table re-created inside a new zone at the same path would hit the stale listing. Two changes:

• Listings are keyed by table UUID as well: the re-created table is a new UUID, so its first scan lists again, at no periodic cost.
• A listing older than an hour is taken again when spark.sql.metadataCacheTTLSeconds is unset, as a bound for anything else. A few minutes would re-list every table of a key-provider cluster constantly and push tables with more than 1000 parents over the listing budget after each expiry.

Both are in CometIcebergNativeScanSuite, with the injected clock and a second UUID over the same directories.

[andygrove]

This review was prepared with AI assistance (Claude Code) and checked by the reviewer before posting.

Thanks @mixermt for the careful work and for the detailed answers in the earlier rounds. The identity gate now reads the SparkContext conf, the listings are keyed per table with an expiry, and the pin audit is thorough. I compared the touched paths against branch-1.2 and found no unintended change in behavior. Native HDFS reads, the storage-family fallback and the NULL fields for id-less structs are all intended. I did not build the branch, so the points below come from reading the code.

I am requesting changes for one thing, a planning-time performance regression for HDFS tables on clusters that advertise a key provider. Those tables fell back to the JVM reader on the release at no planning cost.

1. The encryption-zone probe does quadratic work and keeps every child of a parent.

In hdfsEncryptionZoneFallback, missing calls listed.exists(_._1 == dir.toString) for each requested directory, and listed is an Array. The requested directories are a subset of the listed children, so the cost is directories times children. For a partitioned table the parent is data/ and the children are the partitions. With 50k partitions under one data/ directory that is on the order of billions of string comparisons on the driver while planning. It repeats on every cold listing and again after the one-hour expiry.

flags also holds an entry for every child of the parent, not only the directories the scan reads. The 10000-entry cap counts parents, not entries, so a large partitioned table stays resident for an hour. The 1000-listing budget counts listStatus calls too, and one call on a very large directory is many paged RPCs.

Could the lookup use a Set of the listed paths and store only the flags for the requested directories? Would it also make sense for the budget to count entries rather than calls? A test with a few thousand partitions that checks the planning time and the size of the kept map would settle it.

2. The Scaladoc for getMetadataLocation is now attached to getTableUuid.

getTableUuid was inserted between the doc block of getMetadataLocation and its definition. The paragraph that explains the metadata-table gate and the @return line now document the new method, and getMetadataLocation has no doc. Could you move getTableUuid above that doc block?

3. Is there a way to turn off native HDFS reads on their own?

The docs call the hdfs-native backend experimental. It is claimed by default in local mode, or when spark.executorEnv.HADOOP_USER_NAME is set. Today the only way back to the JVM reader is to disable every native Iceberg scan. Would a supported config for HDFS alone be reasonable while the backend matures?

[mixermt]

Answers below; each change is its own commit.

1. Probe at scale — the lookup was quadratic; fixed, and both bounds now count entries.

• missing is a map lookup, not a scan of the listing per directory (which also rebuilt Path.toString per comparison).
• Every child of a parent is still kept, by name now rather than full path. Keeping only the directories a scan reads would list the parent again for every scan that touches a partition not seen before, a paged round trip per 1000 children each time; keeping all of them costs one listing per table per hour. Spark's own FileStatusCache keeps a status per file of a partitioned file-source table, 250 MB by default; this holds a flag per directory.
• What is kept is bounded to a million children in all, on the order of 100 MB at worst, least recently used parents evicted first; the 10000-parent cap is gone.
• The per-scan budget counts entries: a million, which at HDFS's default page of 1000 (dfs.ls.limit) is the thousand round trips the old budget meant for small parents. The listing is paged (listStatusIterator), so a parent beyond the budget costs the budget and is not kept.
• On a key-provider cluster the planning cost is one paged listing per parent per table per hour; spark.comet.scan.icebergNative.hdfs.enabled=false (3) removes it.
• Test in CometIcebergNativeScanSuite: 50k partitions under one parent, one listStatus, no getFileStatus, asserted under ten seconds and running in well under one (the old lookup took minutes), 50k entries kept; the budget declines without keeping; the kept bound evicts the older table, whose next scan lists again.

2. Scaladoc — getTableUuid moved above the block.

3. HDFS switch — added spark.comet.scan.icebergNative.hdfs.enabled, default true. When false, a scan that reads HDFS files falls back with the key in the reason, the other HDFS checks and the probe are skipped, and other storage is unaffected. Planner test in CometScanSchemeFallbackSuite; documented in the HDFS section.

[andygrove]

This review was prepared with AI assistance (Claude Code) and checked by the reviewer before posting.

Thanks @mixermt for the quick turnaround. The three points from my last review are addressed, so I am approving.

1. The encryption-zone probe is now linear. missing is a map lookup on child names, and the kept listings and the per-scan budget both count entries. The new test in CometIcebergNativeScanSuite probes 50k partitions under one parent and asserts a single listing, no per-directory status and a ten second bound.
2. getTableUuid now sits above the getMetadataLocation doc block, so each method has its own Scaladoc.
3. spark.comet.scan.icebergNative.hdfs.enabled gives HDFS its own switch. The planner test checks that the key is the only HDFS reason when it is false, and the HDFS section of the Iceberg guide names it.

I compared the touched paths against branch-1.2 at 3d73b4cfe. That branch now carries the name-mapping fallback from #7085, so the gate this PR keeps is not a difference any more. I found no unintended change in behavior. Native HDFS reads behind the gates, the storage-family fallback and the NULL fields for id-less structs are all intended. I did not build the branch, so this comes from reading the code and the diff.