mixermt commented on code in PR #5898:
URL: https://github.com/apache/datafusion-comet/pull/5898#discussion_r4366281622
##########
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:
Pushing back on this one, under these assumptions: (a) files written by
Iceberg writers carry field ids, so this only concerns tables imported from
pre-existing Parquet without a name mapping (`migrate`/`snapshot`/`add_files`
set one; only the low-level `importSparkTable` does not); (b) Spark returns
NULL for every id-less nested field (#6794), so the base read of `s.b` that
returned the file's values was not parity either — the repro agrees only
because `b` is all-NULL; (c) the pin has to pass apache/iceberg-rust#3255 to
reach the HDFS backend, and nothing at planning time tells id-less files apart.
Parity (NULL, pruned or not) lands with apache/iceberg-rust#3437; once the pin
passes it, this test becomes a `checkSparkAnswer` assertion. Until then the
documented knob applies.
##########
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:
Agreed: one `FileIO` is built from the metadata location, so a data/delete
file on another backend fails at open. The same holds today for local or REST
metadata with `s3://`/`gs://` data, so the gate is general: the planner
declines a scan whose data/delete files are not all in the metadata location's
storage family (`file`, `s3`/`s3a`, aliases, `gs`, `oss`, `hdfs`). The
nameservice planner tests moved their warehouse onto the fake `hdfs://`
filesystem; decline tests cover local+hdfs, hdfs+local and local+s3;
`CometIcebergHdfsSuite` covers all-hdfs end to end.
##########
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:
Agreed. The planner now declines instead: when the NameNode's server
defaults or the session conf name a key provider, each distinct data/delete
directory is checked with `FileStatus.isEncrypted()` (set for directories
inside a zone; works through viewfs/HA); clusters without a key provider pay
one cached `getServerDefaults` call. Not enabling `kms` here: hdfs-native reads
the KMS URI only from `hadoop.security.key.provider.path` and rejects any
non-`kms://` provider at client construction, which would break unencrypted
reads on `jceks://` clusters; MiniKMS is also absent from the minicluster jar.
`CometIcebergHdfsSuite` gains an encryption-zone case (jceks provider; the
table inside the zone falls back and matches Spark, a sibling outside stays
native); the docs entry moves to the fallback list.
--
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]
