fix(arrow): read nested struct fields as null for files without ids or name mapping - #3437

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 <[email protected]>

Parquet can only read a map when both its key and value are projected.
When a name-mapped or embedded-id map value is a struct whose leaves
match no table field, only the key was selected and the read failed
with "Map field must have exactly 2 fields" (apache#3557).

- Convert selected leaves one at a time instead of converting a
  pre-projected schema that can contain a key-only map.
- When a map key is read but no value leaf is, also read the value
  leaves so the value comes back as a struct of NULLs (apache#3437).
- Only rescue struct values: a primitive value is matched by position
  and would be cast to the table type.
- Never rescue leaves with an invalid type promotion; fail with
  DataInvalid if no value leaf is left to read.
- Keep rejecting duplicate field ids in the file.

…-rust's hdfs-native backend (apache#5898)

* feat: read and write Iceberg tables on HDFS via iceberg-rust's hdfs-native 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) <[email protected]>

* docs: trim duplicated commentary around the HDFS NameNode resolution

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) <[email protected]>

* test: skip the Iceberg HDFS suite when MiniDFSCluster cannot start

`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) <[email protected]>

* fix: match hadoop-client-minicluster to the hadoop each Spark profile 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) <[email protected]>

* fix: report task input metrics for fused and shuffled native Iceberg 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 <[email protected]>

* fix: register the native shuffle input bridge ahead of nested inputs

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 <[email protected]>

* feat: show Iceberg planning metrics on the native scan node in the SQL 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 <[email protected]>

* style: wrap the resolve-needed fallback reason under the 100-column scalastyle 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 <[email protected]>

* chore: move the iceberg-rust pin to the latest head of apache/iceberg-rust#3111

ef451b66 merges apache/iceberg-rust main into the PR: typed NULL arrays for
schema-added nested columns on read (apache#2635), an azdls relative-path fix
(apache#3311) and dependency bumps. The HDFS resolver is unchanged, so the
hdfs.name-node.<nameservice> declaration Comet emits still applies, and the
head still contains main's pin (af1da4c5).

Co-Authored-By: Claude Opus 5.5 <[email protected]>

* feat(iceberg): read Iceberg tables on HDFS natively via iceberg-rust's hdfs-native backend

Moves the iceberg-rust pin to apache/iceberg-rust ca21f158, which carries the
HDFS backend (opendal-hdfs-native, a pure-Rust HDFS client), and routes hdfs
to it for reads. Writes keep Iceberg's writer.

That backend dials an authority with a port as written and resolves a
portless one, an HA nameservice, only through a declaration in the FileIO
properties, never through $HADOOP_CONF_DIR. The scan rule now collects the
HDFS authorities of the data and delete files and declares each portless one
as hdfs.name-node.<ns> from the session's dfs.ha.namenodes.<ns> and
dfs.namenode.rpc-address.<ns>.<id>, merged before the catalog's properties so
a catalog's own declaration wins. A nameservice that stays undeclared, and any
HDFS read on a cluster whose hadoop.security.authentication is not simple
(the client cannot use an executor's delegation tokens), fall back at planning
instead of failing on an executor.

hadoop.version, which pins the hadoop-client-minicluster the HDFS test fixture
needs, now follows each Spark profile's Hadoop line: with a global 3.3.4,
MiniDFSCluster cannot start on the Spark 4.x profiles. CometIcebergHdfsSuite
reads tables on a single-NameNode cluster and on a two-NameNode HA cluster
addressed by nameservice, before and after a failover and with the first
NameNode shut down. CometScanSchemeFallbackSuite drives the real scan rule
over Iceberg tables whose manifests name hdfs:// data files.

The pin also changes how iceberg-rust matches nested fields in data files that
have neither field ids nor a name mapping: a read of only some of a struct's
fields now fails rather than guessing by position. The test for such files
expects that, and the Iceberg guide documents it.

Co-Authored-By: Claude Fable 5.1 <[email protected]>

* fix(iceberg): fall back for name-mapped tables that read list or map columns

iceberg-rust at the pinned revision reads a data file that has no field ids
through the table's name mapping, but cannot open the list layouts Hive and
Spark's writeLegacyFormat write (a two-level list, or a three-level one with
bag/array_element groups) that way: the read schema it supplies keeps a field
id on the list's element field, and the Parquet reader rejects it as
incompatible. A file's layout is only known from its footer, so a scan of a
table with schema.name-mapping.default now falls back whenever it reads a
list or map column, at any depth. Iceberg's TestCreateActions migrates such
files and reads them back with Comet enabled; its Hive-style and two-level
list cases failed without this.

CometIcebergNativeSuite imports a legacy-format file under a name mapping and
checks that the list column falls back while the primitives stay native. The
limitation is listed in the Iceberg guide.

Co-Authored-By: Claude Fable 5.1 <[email protected]>

* fix(iceberg): decline scans that mix storage backends or read HDFS encryption 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 <[email protected]>

* fix(iceberg): claim HDFS scans only as the Spark user, batch the encryption-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 <[email protected]>

* feat(iceberg): move the iceberg-rust pin past apache/iceberg-rust#3437

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 <[email protected]>

* chore(iceberg): drop the iceberg-rust#3437 note from the pin comment

Co-Authored-By: Claude Fable 5.1 <[email protected]>

* test(iceberg): create the HDFS encryption zone with the non-deprecated 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 <[email protected]>

* test(iceberg): prove the native HDFS client reads as the Spark user

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 <[email protected]>

* feat(iceberg): forward the HDFS client settings the session sets to the 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 <[email protected]>

* fix(iceberg): keep the HDFS encryption-zone listings per parent, with 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 <[email protected]>

* test(iceberg): read a merge-on-read table's position deletes over HDFS

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 <[email protected]>

* fix(iceberg): read the executors' HADOOP_USER_NAME from the SparkContext, not the session

The HDFS identity gate read spark.executorEnv.HADOOP_USER_NAME from the
session's SQLConf first. A session built on an existing context, or a SET,
can hold the key there without the executors ever seeing it, and the scan
would be claimed while the native client connects as another user, with no
fallback. The gate now reads only the SparkContext's configuration, which
launched the executors. The planner tests set the live conf instead of the
SQLConf, and one pins that the SQLConf value does not stand in.

Co-Authored-By: Claude Fable 5.1 <[email protected]>

* fix(iceberg): keep HDFS encryption-zone listings per table and for an hour by default

With spark.sql.metadataCacheTTLSeconds unset, a listing lived for the process,
so a table dropped with purge and created again at the same path inside a new
encryption zone was still claimed from the old listing and failed on the
executors. Listings are now keyed by the table's UUID as well, which the
re-created table does not share, so its first scan lists again; and a listing
older than an hour is taken again when the TTL 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.

Co-Authored-By: Claude Fable 5.1 <[email protected]>

* fix(iceberg): give getMetadataLocation its Scaladoc back

getTableUuid was inserted between the doc block of getMetadataLocation and
its definition, so the paragraph on the metadata-table gate documented the
new method and getMetadataLocation had none. Move getTableUuid, with its own
doc, above that block.

Co-Authored-By: Claude Fable 5.1 <[email protected]>

* fix(iceberg): probe encryption zones in linear time and bound what is kept by entries

The probe looked each requested directory up with a scan of the parent's
listing, rebuilding Path.toString per comparison: directories times children,
billions of comparisons on the driver for a flat table of tens of thousands of
partitions, repeated on every cold listing. The lookup is now a map lookup, and
children are kept by name rather than full path.

Every child of a parent is still kept, so a later scan of other partitions
lists nothing, but the whole is now bounded to a million children across
tables, least recently used parents evicted first, instead of ten thousand
parents of any size. The per-scan budget counts directory entries rather than
listStatus calls, a million, which at HDFS's default page of a thousand is the
thousand round trips the old budget meant; the listing is paged, so a parent
beyond the budget costs the budget and is not kept.

The suite probes 50k partitions under one parent: one listing, no per-directory
status, under ten seconds, 50k entries kept; the budget declines without
keeping, and the kept bound evicts the older table.

Co-Authored-By: Claude Fable 5.1 <[email protected]>

* feat(iceberg): add spark.comet.scan.icebergNative.hdfs.enabled

The hdfs-native backend is experimental, and the only way back to Spark's
Iceberg reader for HDFS tables was to turn every native Iceberg scan off. The
new switch, true by default, declines a scan that reads HDFS files with the
key in the reason, skips the other HDFS checks and the encryption-zone probe,
and leaves scans of other storage alone.

Co-Authored-By: Claude Fable 5.1 <[email protected]>

* test(iceberg): drop the removed Spark 3.4 profile from the HDFS suite's Scaladoc

Main no longer has a spark-3.4 profile, so the suite's note on which profiles run it lists
3.5 through 4.2 only.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

* test(iceberg): put BouncyCastle on the Spark 4.0 test class path for the HDFS suite

On the Spark 4.0 profile the encryption-zone test in CometIcebergHdfsSuite
aborted the whole ScalaTest run with

  NoClassDefFoundError: org/bouncycastle/jce/provider/BouncyCastleProvider

Spark 4.0 uses Hadoop 3.4.1, whose KeyProvider constructor links against
BouncyCastleProvider, so the class is loaded as soon as a key provider is
created. Hadoop 3.4.2 and later load it reflectively and 3.3.4 does not
reference it, so the other profiles were unaffected.

Add bcprov-jdk18on 1.78.1 (the version Hadoop 3.4.1 builds with) as a
test-scoped dependency of the spark-4.0 profile, and note it in the suite's
Scaladoc.

Co-Authored-By: Claude <[email protected]>

---------

Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
Co-authored-by: Cursor <[email protected]>

Parquet can only read a map when both its key and value are projected.
When a name-mapped or embedded-id map value is a struct whose leaves
match no table field, only the key was selected and the read failed
with "Map field must have exactly 2 fields" (apache#3557).

- Convert selected leaves one at a time instead of converting a
  pre-projected schema that can contain a key-only map.
- When a map key is read but no value leaf is, also read the value
  leaves so the value comes back as a struct of NULLs (apache#3437).
- Only rescue struct values: a primitive value is matched by position
  and would be cast to the table type.
- Never rescue leaves with an invalid type promotion; fail with
  DataInvalid if no value leaf is left to read.
- Reject duplicate field ids in the projected subtree, not only in the
  whole file schema.