Re: [PR] feat(iceberg): native Iceberg scans of HDFS-backed tables via iceberg-rust's hdfs-native backend [datafusion-comet]

["sunchao (via GitHub)" <gi...@apache.org>]

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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org