sunchao commented on code in PR #5898:
URL: https://github.com/apache/datafusion-comet/pull/5898#discussion_r4363594959
##########
native/Cargo.toml:
##########
@@ -85,8 +85,8 @@ aws-sdk-sts = { version = "1.114.0", default-features = false
}
# Pinned to a revision rather than a release because no published iceberg-rust
# version is on our arrow: 0.10.1 requires arrow/parquet ^58. Move to a plain
# version requirement once a release ships on arrow 59. Bump policy: #5645.
-iceberg = { git = "https://github.com/apache/iceberg-rust";, rev =
"1f3bc34058e3f3e45336bc5428f37b92649a608b" }
-iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust";,
rev = "1f3bc34058e3f3e45336bc5428f37b92649a608b", features = ["opendal-memory",
"opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
+iceberg = { git = "https://github.com/apache/iceberg-rust";, rev =
"ca21f1587c6caaeb88dd3a9e1cc245747e1aba08" }
Review Comment:
[P2] Preserve compatible reads of pruned structs when updating this pin.
Import a Parquet file containing `s STRUCT<a: INT, b: INT>` without field IDs
or a name mapping, then run `SELECT s.b` with the default nested pruning
enabled. The new revision throws `cannot reconcile struct fields by id`; Spark
completes the query. With `b` containing only NULLs, the base also returns the
same NULL rows as Spark, isolating a newly introduced failure. The changed test
now expects this exception instead of preventing it. Please add compatible
handling or a JVM fallback for this case and retain a Spark-parity assertion.
Evidence: Executed standalone ArrowReader reproductions with iceberg-rust
`1f3bc34058e3f3e45336bc5428f37b92649a608b` and
`ca21f1587c6caaeb88dd3a9e1cc245747e1aba08`, both using Arrow/Parquet 59.3.1.
For rows s=(10,NULL),(20,NULL), the pruned base read succeeds with two NULLs
and the head read errors. Logs: `/tmp/5898-b493-base-exact-arrow.log` and
`/tmp/5898-b493-head-exact-arrow-online.log`. Iceberg Java’s fallback schema
assigns only top-level IDs and its optional-field reader supplies NULLs. The
PR’s `CometIcebergNativeSuite.scala:2814` expects the same new failure.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -898,6 +905,52 @@ case class CometScanRule(session: SparkSession)
case None => true
}
+ // iceberg-rust's HDFS client cannot use an executor's delegation
tokens: opendal builds
+ // every client for a synthetic nameservice, and the client looks
tokens up by that name.
+ // A portless authority must be a declared nameservice (see
hadoopToIcebergHdfsProperties).
+ val hdfsSupported = taskValidation.hdfsAuthorities.isEmpty || {
+ val authentication =
hadoopConf.getTrimmed("hadoop.security.authentication", "simple")
+ val secure = !authentication.equalsIgnoreCase("simple")
+ if (secure) {
+ fallbackReasons += "Iceberg scan reads HDFS with " +
+ s"hadoop.security.authentication=$authentication, but Comet's
native Iceberg " +
+ "reader cannot authenticate to HDFS with delegation tokens"
+ }
+ val undeclared = CometIcebergNativeScan.undeclaredHdfsNameservice(
+ taskValidation.hdfsAuthorities,
+ metadata.catalogProperties)
+ undeclared.foreach { ns =>
+ fallbackReasons += s"Iceberg scan reads HDFS authority '$ns',
which has no port and " +
+ "which the Hadoop configuration does not declare as a
nameservice " +
+ s"(dfs.ha.namenodes.$ns, and a host:port
dfs.namenode.rpc-address.$ns.<id> for " +
+ "each NameNode)"
+ }
+ !secure && undeclared.isEmpty
Review Comment:
[P2] Retain JVM fallback for HDFS encryption-zone files until native
decryption is supported. On a simple-auth cluster with a configured KMS, an
authorized read of a table inside an encryption zone passes this
authentication/nameservice gate. The selected client is built without the `kms`
feature and rejects every encrypted file, whereas Hadoop’s reader decrypts it.
Because native Iceberg scans default to enabled, this changes previously
successful reads into failed tasks. Please add a conservative fallback or wire
and validate KMS support before admitting these scans.
Evidence: The lockfile pins hdfs-native 0.14.7. OpenDAL 0.58.2 depends on it
without enabling `kms`, and Comet adds no feature activation. In hdfs-native
`client.rs`, `read()` passes `file_encryption_info` to `build_crypto_codec`,
whose non-KMS branch unconditionally returns `UnsupportedFeature` for an
encrypted file. Hadoop 3.3.4 and 3.4.2 `DFSClient.createWrappedInputStream`
instead invoke `HdfsKMSUtil` when that information is present. This failure
path was source-verified; a live encryption-zone reproduction was not run.
##########
native/core/src/execution/operators/iceberg_common.rs:
##########
@@ -302,9 +306,10 @@ fn cached_file_io(
/// mode; the JVM read and write gates load these lists over JNI. `memory` is
write-only: an OpenDAL
/// memory backend is a fresh empty in-process store the write path assembles
manifests in, so a
/// read finds nothing. `oss` is read-only: no `oss.*` property is forwarded
and no test covers it.
+/// `hdfs` is read-only: the write path has not been tested against it.
pub(crate) fn builtin_storage_schemes(access_mode: AccessMode) -> &'static
[&'static str] {
match access_mode {
- AccessMode::Read => &["file", "gs", "oss", "s3", "s3a"],
+ AccessMode::Read => &["file", "gs", "hdfs", "oss", "s3", "s3a"],
Review Comment:
[P2] Check filesystem consistency before admitting HDFS files. A
Hadoop-catalog table can have local metadata and
`write.data.path=hdfs://nn:8020/warehouse/data`. This allowlist change makes
that scan eligible, but `IcebergScanExec` still selects its single FileIO
backend from `metadata_location`, choosing `Fs`. Reading the HDFS file then
attempts a local `dfs:/nn:8020/...` path and fails. Previously the HDFS scheme
triggered JVM fallback, whose HadoopFileIO resolves each file’s filesystem.
Please decline these combinations until the native FileIO can route every
data/delete location correctly.
Evidence: The new planner tests explicitly claim scans with local catalog
metadata and HDFS data paths. Traced `IcebergScanExec` through
`load_file_io(metadata_location)` and `storage_factory_for`. A reproduction
using the exact head dependency and OpenDAL 0.58.2 calls an Fs-backed FileIO
with `hdfs://nn:8020/warehouse/data.parquet` and receives `NotFound`, `service:
fs`, `path: dfs:/nn:8020/warehouse/data.parquet`. Iceberg Java’s
`HadoopInputFile.fromLocation` resolves `FileSystem` from each supplied
location. No live HDFS connection was needed to demonstrate the incorrect
dispatch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
