Repository navigation
feat(iceberg): native Iceberg scans of HDFS-backed tables via iceberg-rust's hdfs-native backend - #5898
feat(iceberg): native Iceberg scans of HDFS-backed tables via iceberg-rust's hdfs-native backend#5898mixermt wants to merge 35 commits into
Conversation
…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>
… 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>
|
Sounds good, thanks @mixermt |
Co-authored-by: Cursor <cursoragent@cursor.com>
|
Found that |
…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>
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>
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>
sunchao
left a comment
There was a problem hiding this comment.
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.2at889856dc5af821de1f7c6ca2e9538bd663431d04. 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.
…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>
…feat/iceberg-hdfs-native
andygrove
left a comment
There was a problem hiding this comment.
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.hdfsEncryptedDirectorydoes one sequentialgetFileStatusper 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 alogWarning.- 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 singleFileIOserves 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
mainand CI has not run (action_required). After mergingmain, could you applyrun-spark-3.5-tests,run-spark-4.0-tests,run-spark-4.1-testsandrun-iceberg-tests, or rundev/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?
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>
|
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>
|
Update on 1 and 5: apache/iceberg-rust#3437 merged, so the pin moved to apache main at |
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
andygrove
left a comment
There was a problem hiding this comment.
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.
- The pin now includes apache/iceberg-rust#3437, and the pruned and whole struct reads of id-less files both return NULL fields, with
CometIcebergNativeSuiteasserting it. - The PR no longer carries its own name-mapping gate and inherits the one on
mainfrom #7085. - 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?
Findings addressed in later commits (pin past iceberg-rust#3437, identity gate). See the follow-up review.
|
@mixermt the new review is from my agent - feel free to disagree on any of these points. |
|
@comphead could you take a look as well? |
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>
|
Answers below; each change is its own commit, plus one fix for the fork CI (the encryption-zone test used the deprecated two-argument 1. Identity test — added.
2. Driver-only settings — forwarded.
3. Probe at scale — both points right; changed.
4. Pin audit —
|
Which issue does this PR close?
Closes #5894.
Rationale for this change
Comet's native Iceberg scan has no
hdfsarm, 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-opendalbehind theopendal-hdfs-nativefeature (OpenDAL'sservices-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:
hdfs://ns1/...) only through a declaration in the FileIO properties,hdfs.name-node.ns1or Hadoop's owndfs.ha.namenodes.ns1keys passed ashadoop.*. It never reads a nameservice from$HADOOP_CONF_DIR, has no default port, and keeps one client per NameNode.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.kmsfeature 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_DIRXML, with separate connections and credentials.What changes are included in this PR?
Native:
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, withopendal-hdfs-nativeenabled;hdfs-nativeresolves to 0.14.7 (releases before 0.14.5 fail every read on adfs.checksum.type=CRC32cluster). What the bump changes for reads is under "Reads that change" below.storage_factory_forrouteshdfstoOpenDalStorageFactory::HdfsNative;hdfsis in the read list ofbuiltin_storage_schemesand not in the write list;hdfs.andhadoop.joinSTORAGE_PROPERTY_PREFIXES.Planner (
CometScanRule,CometIcebergNativeScan), all decided at planning so that no task fails on an executor:validateIcebergFileScanTaskscollects 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 ahost:portdfs.namenode.rpc-address.<ns>.<id>for each NameNode) becomeshdfs.name-node.<ns>, all-or-nothing per nameservice and in declaration order; one the configuration knows nothing of becomeshost:8020, asDFSUtilClient.getNNAddressdials it for the JVM client; a name indfs.nameserviceswithout NameNodes, or an authority-lesshdfs:///location, falls back. A catalog's ownhdfs.name-node.<ns>orhadoop.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 underhadoop., behind the catalog's own; Hadoop's defaults are not, since the client reads the executors' XML underneath.spark.executorEnv.HADOOP_USER_NAMEset to it (in local mode, the JVM'sHADOOP_USER_NAMEor OS user). Otherwise it falls back with a reason naming the setting. Carrying the user into the client needs an OpenDALuseroption and an iceberg-rusthdfs.userpassthrough; hdfs-native already hasClientBuilder::with_user.hadoop.security.authenticationother thansimplefalls back.isEncryptedflag the NameNode sets on every path inside a zone: onelistStatusper parent directory (the partitions of a table cost one call), kept per filesystem and parent untilspark.sql.metadataCacheTTLSecondsexpires 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.FileIOfrom 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 withs3://data files.Build and docs:
pom.xml:hadoop.version(thehadoop-client-miniclusterthe HDFS fixture needs) follows each Spark profile's Hadoop line, ashadoop-azure.versiondoes; with the global 3.3.4,MiniDFSClustercannot start on the Spark 4.x profiles.Reads that change with the pin
CometIcebergNativeSuiteasserts 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.migrate,snapshot,add_files): nested columns fall back onmainsince 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 andwriteLegacyFormatwrite; 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-processMiniDFSClusters 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 ajcekskey 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 realCometScanRuleover a real Iceberg table whose warehouse sits on a fakehdfs://filesystem and whose manifests namehdfs://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 indfs.nameservicesare declined; a portless host is claimed on 8020; ahost:portlocation needs no declaration; two declared nameservices are both declared; a catalog'shdfs.name-node.<ns>wins; a nameservice declared only by the catalog'shadoop.*keys is claimed; a Kerberos cluster is declined; executors reading as another user are declined andHADOOP_USER_NAMEset 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 localhdfs://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.iceberg_common):hdfsresolves for reads and is rejected for writes;hdfs.andhadoop.properties reach the FileIO; the FileIO cache key separates catalogs that differ in their HDFS declaration; a contract test that the pinned iceberg-rust expandshdfs.name-node.<ns>into Hadoop's keys and rejects a portless entry.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
hdfs.user.kmsfeature alone is not enough (it reads the KMS URI only fromhadoop.security.key.provider.pathand rejects any non-kms://provider at client construction).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