Skip to content

feat(iceberg): native Iceberg scans of HDFS-backed tables via iceberg-rust's hdfs-native backend - #5898

Open
mixermt wants to merge 35 commits into
apache:mainfrom
mixermt:feat/iceberg-hdfs-native
Open

mixermt wants to merge 35 commits into
apache:mainfrom
mixermt:feat/iceberg-hdfs-native

Conversation

@mixermt

@mixermt mixermt commented Sep 13, 2026 •

Copy link
Copy Markdown
Contributor

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:

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 (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 and parent until spark.sql.metadataCacheTTLSeconds expires it; a directory missing from a kept listing lists its parent again, and a scan that would take more than 1000 listings falls back with the listings taken kept. 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

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; 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 or after the TTL, and declines past the listing budget (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

mixermt and others added 3 commits September 13, 2026 13:30
…ative backend

Adds `hdfs://` to the native Iceberg scan and write paths, backed by
iceberg-rust's pure-Rust `hdfs-native` OpenDAL backend (apache/iceberg-rust#3111).
This is a second, independent HDFS client: the plain-Parquet native scan reaches
HDFS through libhdfs/JNI (`fs.comet.libhdfs.schemes`), while an Iceberg table is
opened over pure-Rust RPC. The two share only the `$HADOOP_CONF_DIR` XML and
authenticate separately.

The scheme arm alone is not sufficient on an HA cluster. opendal's
`HdfsNativeBuilder` never dials the authority written in the path: it builds its
client against a synthetic authority and synthesizes
`dfs.ha.namenodes.<synthetic>` / `dfs.namenode.rpc-address.<synthetic>.nnN` from
the comma-separated `hdfs.name-node` value (`init_hdfs_config`). iceberg-rust
falls back to the path authority only when that property is absent, which is
correct just for a single-NameNode cluster. An HA table location reads
`hdfs://<nameservice>/...`, and a nameservice is not a routable host, so without
a NameNode list every HA table would fail to connect at execution time -- after
the planner had already committed to the native scan.

`hadoopToIcebergHdfsProperties` therefore resolves the endpoints from the session
Hadoop configuration (`dfs.ha.namenodes.<ns>` plus each
`dfs.namenode.rpc-address.<ns>.<nn>`), joined in declaration order, so a standard
HDFS client configuration needs no extra settings. An explicit catalog
`hdfs.name-node` still wins. A partially resolved list yields nothing rather than
a short failover list, which would silently turn a failover into an outage.

Because one `hdfs.name-node` overrides the authority of every path the FileIO
opens, a scan whose data/delete files span more than one HDFS authority now falls
back: the second nameservice would otherwise be read from the first one's
NameNode at the same relative path, returning wrong data rather than an error.

Changes:
- `storage_factory_for`: `hdfs` arm; `hdfs.`/`hadoop.` added to
  `STORAGE_PROPERTY_PREFIXES` so the NameNode list and client overrides reach FileIO
- `CometScanRule.icebergReadableSchemes` and
  `CometIcebergNativeWrite.SupportedStorageSchemes`: admit `hdfs`
- `IcebergTaskValidationResult.dataFileHdfsAuthorities` + multi-authority fallback
- `native/Cargo.toml`: enable `opendal-hdfs-native`

Testing:
- `CometIcebergHdfsSuite`: end-to-end reads against an in-process MiniDFSCluster
  (plain read, pushdown filter, partitioned table across multiple data files),
  asserting the data location is genuinely `hdfs://`
- 5 unit tests pinning the HA NameNode translation, 2 pinning the native scheme
  and property forwarding
- Full Rust workspace suite (1442 tests) and the three affected JVM suites pass

Note: writes are gated and property-forwarded but have no functional test; only
reads are covered end to end. The HA translation is unit-tested only, as
MiniDFSCluster is single-NameNode.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The opendal synthetic-authority mechanic -- that `hdfs.name-node` overrides the
authority of every path the FileIO opens -- was spelled out at seven sites. Keep
it once, in `hadoopToIcebergHdfsProperties`'s scaladoc, and leave short pointers
at the rest.

Also drops a paragraph that narrated the eight lines of code beneath it, a
`@param` restating its own signature, and a sentence duplicated four lines apart
inside `hadoopToIcebergHdfsProperties`.

Retained the reasoning that cannot be recovered from the code: why opendal
ignores the path authority, why the NameNode list is all-or-nothing, why
`getRawAuthority` rather than `getHost`, and that hdfs-native and libhdfs/JNI are
two separate clients.

Comments only; no behaviour change.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`hadoop-client-minicluster` is pinned at 3.3.4 while Spark supplies
`hadoop-client-api`/`runtime` 3.4.2 on the Spark 4.x profiles, so `HttpServer2`
resolves a shaded Jetty class the older jar does not carry and the NameNode web
server fails to start. Record that in `beforeAll` and cancel the tests instead of
aborting the suite, so CI stays green on profiles where the fixture cannot run.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
@github-actions github-actions Bot added enhancement New feature or request area:writer Native Parquet writer area:scan Parquet scan / data reading area:Iceberg labels Sep 13, 2026
mixermt and others added 3 commits September 13, 2026 17:12
… supplies

`hadoop.version` was a single global 3.3.4, but only Spark 3.4/3.5 ship hadoop
3.3.x. On the Spark 4.x profiles `HttpServer2` comes from Spark's newer
hadoop-client-runtime and resolves shaded Jetty classes the 3.3.4 minicluster does
not carry, so MiniDFSCluster's NameNode web server fails to start and any suite
using `WithHdfsCluster` cannot run.

Set `hadoop.version` per Spark profile, as `parquet.version` already is:
3.4 -> 3.3.4, 3.5 -> 3.3.4, 4.0 -> 3.4.1, 4.1 -> 3.4.2, 4.2 -> 3.5.0. The 3.x
profiles are unchanged; a global bump would only have moved the skew onto them.

`CometIcebergHdfsSuite` now runs on the default profile instead of cancelling.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
@comphead

Copy link
Copy Markdown
Contributor

Sounds good, thanks @mixermt

Co-authored-by: Cursor <cursoragent@cursor.com>
@mixermt

mixermt commented Sep 19, 2026

Copy link
Copy Markdown
Contributor Author

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

mixermt and others added 3 commits September 19, 2026 21:26
…scans

The scan input bridge only ran when the native block's root was the scan
itself. `hasScanInput` recognised `CometNativeScanExec` alone, so an Iceberg
scan fused below a filter, project or aggregate, or feeding the native
shuffle writer, never reported bytesRead/recordsRead and the Spark UI showed
"Input: 0 B / 0 records" for those stages. The listener was also registered
after `CometExecIterator`, so it ran before the final native metric flush
whenever the iterator was not drained.

Register the bridge in `CometExecRDD.compute` ahead of the iterator, drop the
per-exec `compute` overrides and the `hasScanInput` gate (the bridge is a
no-op without `bytes_scanned` leaves), and increment rather than set the task
input metrics, claiming each accumulator once per task through the registry
the spill reporter already uses so overlapping or repeated registrations
count once while other input sources still add up.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
CometNativeShuffleWriter registered reportScanInputMetrics after
CometNativeShuffleInputRDD had resolved its inputs, so for a nested native
block that a failed or undrained writer plan left unflushed, the writer's
listener claimed the nested scan leaves before their iterator published
final values and the nested block's own listener then added nothing.

Register scan input reporting next to spill reporting in
CometNativeShuffleInputRDD.compute, before the input producers and the
writer's iterator, and drop the writer's registration. Rename the RDD's
metric node to taskMetricNode since it now feeds both task metrics, and
state that a coalesced task's repeated registrations are deduplicated
rather than summed.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…L UI

The SQL UI aggregates accumulator updates from listener events and never
reads driver-side values. Iceberg posts its planning metrics (manifests and
files scanned or skipped, planning duration, file sizes) for the original
BatchScanExec's accumulators, whose ids are not in the Comet plan graph, and
CometIcebergNativeScanExec never posted its own, so every planning metric
showed 0. Under DPP the driver-side values were 0 as well, because planning
runs on a copy of the original plan while the lazy metrics read the
original's accumulators.

Post the planning metrics as a driver metric update once per SQL execution
from the first execution-time read of commonData, read them from the plan
copy that actually planned, register them under Iceberg's descriptions so
the labels match BatchScan, and keep them out of the executor metric tree so
tasks do not report zeros into the per-task statistics. Document that output
rows and recordsRead count rows after iceberg-rust's row filter, and that
row deletes applied has no native counterpart.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
@github-actions github-actions Bot added the area:shuffle Shuffle (JVM and native) label Sep 19, 2026
Reconciles the Iceberg planning metrics work with apache#6027,
which posts them through the new CometLeafExec.sendDriverMetrics hook: keep the
hook and Iceberg's metric names, and layer on the per-execution post guard, the
DPP-correct plannedOriginalPlan, and the exclusion of planning metrics from the
executor metric tree.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
mixermt and others added 3 commits October 7, 2026 00:09
Brings in origin/main up to 8c93fbb and moves the iceberg-rust pin to the
head of apache/iceberg-rust#3111 (22db8e6f, HDFS via opendal hdfs-native).

The branch's own Iceberg task-input and planning metrics commits are dropped
in favour of main's equivalents (apache#5880, apache#6085, apache#5265).

The pin also carries iceberg-rust main changes, so two Comet adaptations:
FileWrite::close now returns FileMetadata (LocationScopedFileWrite) and
UnboundPartitionField is built through its builder (planner tests).

Also: CometIcebergHdfsSuite is registered in the CI workflows, the HDFS case
in CometIcebergWriteDetectionSuite no longer treats hdfs as an unsupported
scheme, and the Iceberg HDFS docs are corrected.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Brings in origin/main up to f7eb8aa and moves the iceberg-rust pin to the
current head of apache/iceberg-rust#3111 (b9624086), which contains main's
own pin (af1da4c5) plus the HDFS commits.

That head changes how HDFS NameNodes resolve: a portless path authority is a
logical nameservice resolved only through its declaration,
hdfs.name-node.<nameservice> (or Hadoop's dfs.ha.namenodes keys passed as
hadoop.*); the global hdfs.name-node now serves authority-less paths only,
and there is no default port. Comet's HA translation emitted the global key,
so the HA MiniDFS test failed with "logical nameservice ... is not declared".

Comet now:
- declares HA nameservices as hdfs.name-node.<nameservice>, derived from the
  session Hadoop configuration overlaid with the table FileIO's dfs.* keys;
- declares a portless plain host on port 8020, as the JVM client dials it,
  and no longer invents 8020 for a portless rpc-address, which the JVM
  rejects;
- falls back at planning, instead of failing on an executor, for an
  undeclared nameservice, a NameNode entry that is not host:port in any
  hdfs.name-node* key, DNS-resolved NameNodes (resolve-needed), malformed
  hdfs.host/hdfs.port or a bare hadoop. key, and a declaration of opendal's
  synthetic nameservice name.

The HA test now also checks the declaration before executing and keeps
reading and writing after the first NameNode is shut down. The
pre-epoch partition tripwire follows iceberg-rust apache#3323, which the pin
carries; Comet's own values are unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…calastyle limit

The reason for an unresolved nameservice now names the Hadoop configuration
the declaration is derived from (session overlaid with the table FileIO's)
rather than the session alone.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@mixermt
mixermt marked this pull request as ready for review October 23, 2026 19:19

@sunchao sunchao left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment thread native/Cargo.toml Resolved
mixermt and others added 3 commits October 24, 2026 00:38
…cryption zones

The native scan builds one FileIO from the metadata location and opens every data and
delete file through it, so a file on another backend (hdfs:// data under local metadata,
a local file under HDFS metadata, S3 data under local metadata) failed at open time.
CometScanRule now declines a scan whose data/delete files are not all in the metadata
location's storage family (file, s3/s3a, each S3-compliant alias, gs, oss, hdfs).

hdfs-native is built without its kms feature and rejects every file of an HDFS encryption
zone. When the NameNode advertises a key provider or the session configures one, the rule
checks each distinct data/delete directory with FileStatus.isEncrypted() and falls back.

The planner tests keep their warehouse on the fake hdfs:// filesystem so metadata and data
share a storage; CometIcebergHdfsSuite gains a cluster with a jceks key provider and an
encryption zone.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

@andygrove andygrove left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 and others added 2 commits October 27, 2026 23:06
Keeps main's name-mapping gate (every nested column of a name-mapped table falls back, apache#6790)
and drops this branch's narrower list/map one, which it subsumes; the legacy-layout test now
expects main's reason.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…yption-zone probe, default the NameNode port

The native client connects as HADOOP_USER_NAME or the executor's OS user, never as the Spark
user the JVM reader runs as, and OpenDAL exposes no user setting. The planner now claims an
HDFS scan 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 user) and falls back
naming the setting otherwise.

The encryption-zone probe groups data directories by parent and lists each parent once (the
NameNode returns every child's encrypted flag), caching the flags per filesystem and directory,
so a partitioned table costs one call instead of one per partition.

An undeclared portless authority is a single NameNode on Hadoop's default port, 8020, as
DFSUtilClient.getNNAddress dials it for the JVM client; a name in dfs.nameservices without
NameNodes still falls back, and a catalog's own hadoop.dfs.ha.namenodes.<ns> keeps precedence.

Tests: an end-to-end scan over data files on two NameNodes, the encryption-zone table is
partitioned, the probe's listing and caching are counted on a local hdfs:// filesystem, and the
planner cases for the identity gate and the default port.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
@mixermt

mixermt commented Oct 27, 2026

Copy link
Copy Markdown
Contributor Author
  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.

A struct's fields in a data file with neither field ids nor a name mapping now read as NULL,
as Spark reads them, whether the struct is read whole or pruned, instead of by position or
failing with "cannot reconcile struct fields by id". The test asserts parity with Spark for
both reads. What still differs is documented: Spark 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. The INT96 migration test
asserts those shapes instead of comparing the struct with raw Parquet.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
@mixermt

mixermt commented Oct 27, 2026

Copy link
Copy Markdown
Contributor Author

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).

mixermt and others added 2 commits October 28, 2026 10:18
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
@mixermt
mixermt requested a review from andygrove October 30, 2026 20:51
@andygrove andygrove added run-iceberg-tests run-spark-3.5-tests Run the Spark 3.5 SQL tests on this PR run-spark-4.0-tests run-spark-4.1-tests Run the Spark 4.1 SQL tests on this pull request instead of waiting for the merge queue labels Oct 30, 2026

@andygrove andygrove left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Thanks @mixermt for the thorough follow-up. The three regressions I raised earlier are addressed, so I am dismissing my request for changes.

  1. The pin now includes apache/iceberg-rust#3437, and the pruned and whole struct reads of id-less files both return NULL fields, with CometIcebergNativeSuite asserting it.
  2. The PR no longer carries its own name-mapping gate and inherits the one on main from #7085.
  3. HDFS scans are claimed only when the executors are known to read as the Spark user, and the fallback reason names the setting.

The batched encryption probe, the default port for a plain host, the test with metadata and data on different NameNodes, and the rewritten description all look good too. I compared against branch-1.2 and found no unintended change in behavior. I have added run-iceberg-tests, run-spark-3.5-tests, run-spark-4.0-tests and run-spark-4.1-tests, so we will have a verdict from the Iceberg and Spark suites before this goes to the merge queue.

A few questions I would like answered.

Is the identity claim tested? The gate rests on the native client connecting as HADOOP_USER_NAME, or as the OS user when it is unset. The planner tests check the decision, but MiniDFSCluster runs with permissions off, so nothing shows the client really reads as that user. Could you add a case with dfs.permissions.enabled=true and a table directory that only the test user can read?

What about settings that exist only on the driver? The secure-cluster, nameservice and encryption checks use the session Hadoop conf, including spark.hadoop.*. The native client reads only the executors' $HADOOP_CONF_DIR XML, as the new docs say. A setting such as spark.hadoop.dfs.client.use.datanode.hostname=true works for the JVM reader but never reaches the native client, so a read could fail on an executor with no fallback. Which connectivity settings are worth forwarding under the new hadoop. prefix, and which should the docs call out as unsupported?

How does the encryption probe behave at scale? On a cluster with a key provider, hdfsEncryptedDirectory makes one sequential listStatus per uncached parent. The cache holds 10000 entries as an LRU, so a scan of tens of thousands of partition directories would re-list on every query. Would it make sense to cap the number of listings and decline past the cap? Also, a cached "not encrypted" answer lives for the whole process. If a table is purged and re-created inside a new zone at the same path, would the scan still be claimed? Is a TTL worth adding?

Could you refresh the pin audit against the current main pin? The list of behavior changes in the discussion covers 21 commits. Against b154603a the range is 59 commits. It includes the NOT rewrite before pruning (#3416), the widening of promoted partition values and bounds (#3483), the new delete row selection (#3454) and float JSON parsing (#3476). Which existing suites exercise position-delete reads for #3454?

@andygrove
andygrove dismissed their stale review October 30, 2026 21:01

Findings addressed in later commits (pin past iceberg-rust#3437, identity gate). See the follow-up review.

@andygrove

Copy link
Copy Markdown
Member

@mixermt the new review is from my agent - feel free to disagree on any of these points.

@andygrove

Copy link
Copy Markdown
Member

@comphead could you take a look as well?

mixermt and others added 6 commits October 31, 2026 21:17
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…d overload

The two-argument HdfsAdmin.createEncryptionZone is deprecated in Hadoop 3.3 and
3.4, and the strict-warnings build on the Spark 3.5 profile rejects it. The
overload with an empty CreateEncryptionZoneFlag set is what it delegates to.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The planner claims an HDFS scan on the premise that iceberg-rust's client
connects as the Spark user. MiniDFSCluster enforces permissions by default,
but the JVM's user is the superuser of every NameNode it starts, so no test
showed it. A cluster started under another user's UGI makes the JVM's user an
ordinary one: with the table's data directory at 700 the native scan reads it,
and once the directory belongs to someone else the native read is denied with
AccessControlException, as Spark's is.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…he native reader

iceberg-rust's HDFS client reads the executors' $HADOOP_CONF_DIR XML, overridden
by the hadoop.* properties of the FileIO, so a client setting the driver holds
(spark.hadoop.* or its own XML) never reached it. Of what hdfs-native reads from
its configuration, the planner now forwards dfs.client.use.datanode.hostname,
dfs.client.socket-timeout and the three dfs.client.failover pacing keys when the
session sets them, behind the catalog's own hadoop.* properties. Hadoop's own
defaults are not forwarded, since they would override the executors' XML.

Not forwarded: the per-nameservice dfs.client.failover.*.<ns> keys, which the
client would look up under opendal's own nameservice name, and
dfs.data.transfer.protection, on which the client negotiates SASL whenever it is
set while the JVM client does so only on a secure cluster, which falls back.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
… a TTL and a budget

The check cached one flag per data directory in a 10000-entry LRU, so a table
with more data directories than that listed its parents again on every query,
and a cached answer lived for the process: a table purged and re-created inside
a new zone at the same path would be claimed and fail on the executors.

The listings are now kept per filesystem and parent directory, with every
child's flag, so entries scale with parents rather than partitions, and a
directory missing from a kept listing, a new partition, lists its parent again.
A listing expires with spark.sql.metadataCacheTTLSeconds, Spark's TTL for its
own partition file metadata cache (unset by default). A scan that would take
more than 1000 listings is declined after that many, which are kept, so a later
scan continues from them. The fallback reason now says which of the three
happened.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Delete files are opened through the same FileIO as the data files, and no HDFS
test read any; this one writes a position-delete file on HDFS and reads it in
one native scan.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
@mixermt

mixermt commented Oct 31, 2026

Copy link
Copy Markdown
Contributor Author

Answers below; each change is its own commit, plus one fix for the fork CI (the encryption-zone test used the deprecated two-argument createEncryptionZone, which the strict-warnings job rejects).

1. Identity test — added.

  • MiniDFSCluster enforces permissions by default. The check was vacuous because the JVM's user is the superuser of every NameNode it starts.
  • The new test starts a cluster under another user's UGI, so the JVM's user is an ordinary one: with the data directory at 700 the native scan reads it; once the directory belongs to someone else, the native read is denied with AccessControlException, as Spark's is.
  • HADOOP_USER_NAME itself is process environment the client reads at connect, which a JVM test cannot set; hdfs-native covers that branch.

2. Driver-only settings — forwarded.

  • The client's config is the executors' XML overridden by what OpenDAL passes in, which iceberg-rust fills from hadoop.*. Comet already let those through, so far only for nameservices.
  • hdfs-native reads few client keys. Now forwarded under hadoop. when the session sets them (not Hadoop's defaults, which would override the executors' XML), behind the catalog's own: dfs.client.use.datanode.hostname, dfs.client.socket-timeout, dfs.client.failover.{max.attempts,sleep.base.millis,sleep.max.millis}.
  • Not forwardable: the per-nameservice dfs.client.failover.*.<ns> keys (OpenDAL dials every NameNode list as a nameservice of its own name, so the client never looks them up) and dfs.data.transfer.protection (the client negotiates SASL whenever it is set; the JVM client only on a secure cluster, which falls back).
  • The docs now list what the client ignores: short-circuit and hedged reads, dfs.client.retry.*, ipc.client.*, hadoop.rpc.protection.

3. Probe at scale — both points right; changed.

  • Listings are kept per parent (filesystem, parent → child flags), so entries scale with parents, not partitions; a directory missing from a kept listing, a new partition, re-lists only its parent.
  • Expiry honors spark.sql.metadataCacheTTLSeconds, Spark's TTL for its own partition file metadata cache (default none). That bounds the purge-and-recreate case.
  • A scan that would take more than 1000 listings is declined with that in the reason; the listings taken are kept, so later scans converge.
  • listEncryptionZones is superuser-only and getEZForPath is one call per path, so a listing per parent stays the cheapest honest check for a plain user.

4. Pin audit — b154603a..de0a6ac8, 59 commits.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area:Iceberg area:scan Parquet scan / data reading area:shuffle Shuffle (JVM and native) area:writer Native Parquet writer enhancement New feature or request run-iceberg-tests run-spark-3.5-tests Run the Spark 3.5 SQL tests on this PR run-spark-4.0-tests run-spark-4.1-tests Run the Spark 4.1 SQL tests on this pull request instead of waiting for the merge queue

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Support Iceberg tables on HDFS in the native scan

4 participants