andygrove commented on code in PR #5898:
URL: https://github.com/apache/datafusion-comet/pull/5898#discussion_r4443143875
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala:
##########
@@ -655,6 +659,193 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
global.result() ++ promoted.result()
}
+ /**
+ * The client settings iceberg-rust's HDFS backend reads from its
configuration and the JVM
+ * client reads too (hdfs-native 0.14, `common/config.rs`). The client
otherwise reads only the
+ * executors' `$HADOOP_CONF_DIR`, so a value the session sets would never
reach it. Not
+ * forwarded: the per-nameservice `dfs.client.failover.*.<ns>` keys, which
the client would look
+ * up under a nameservice name of opendal's own, and
`dfs.data.transfer.protection`, on which
+ * the client negotiates SASL whenever it is set where the JVM client does
so only on a secure
+ * cluster, which falls back.
+ */
+ private[comet] val forwardedHdfsClientKeys: Seq[String] = Seq(
+ "dfs.client.use.datanode.hostname",
+ "dfs.client.socket-timeout",
+ "dfs.client.failover.max.attempts",
+ "dfs.client.failover.sleep.base.millis",
+ "dfs.client.failover.sleep.max.millis")
+
+ /**
+ * The HDFS properties a scan of `authorities` needs from the Hadoop
configuration, for
+ * iceberg-rust's HDFS backend: nameservice declarations, and the client
settings of
+ * [[forwardedHdfsClientKeys]] that the session set (a value from Hadoop's
own `*-default.xml`
+ * is not a setting, and forwarded would override the executors' XML), as
`hadoop.<key>`.
+ *
+ * The backend dials an authority with a port as written, but resolves a
portless one
+ * (`hdfs://ns1/...`) only through a nameservice declared in the FileIO
properties, never
+ * through `$HADOOP_CONF_DIR`. A nameservice the Hadoop configuration
declares in full, with
+ * `dfs.ha.namenodes.<ns>` and a `host:port`
`dfs.namenode.rpc-address.<ns>.<id>` for each
+ * NameNode, becomes `hdfs.name-node.<ns>` with its NameNodes in declaration
order. An authority
+ * the configuration knows nothing of is a single NameNode on Hadoop's
default port, as
+ * `DFSUtilClient.getNNAddress` dials it for the JVM client. A name in
`dfs.nameservices`
+ * without NameNodes, or with one that has no port, is left out, and
+ * [[undeclaredHdfsNameservice]] then declines the scan.
+ */
+ def hadoopToIcebergHdfsProperties(
+ hadoopConf: Configuration,
+ authorities: Set[String]): Map[String, String] = {
+ val clientSettings = if (authorities.isEmpty) {
+ Map.empty[String, String]
+ } else {
+ forwardedHdfsClientKeys.flatMap { key =>
+ val sources =
Option(hadoopConf.getPropertySources(key)).getOrElse(Array.empty[String])
+ val set = sources.nonEmpty &&
!sources.exists(_.endsWith("-default.xml"))
+ if (set) Option(hadoopConf.getTrimmed(key)).map(s"hadoop.$key" -> _)
else None
+ }.toMap
+ }
+ val logicalNames =
hadoopConf.getTrimmedStringCollection("dfs.nameservices").asScala.toSet
+ val declarations = hdfsNameservices(authorities).flatMap { ns =>
+ val nameNodes = hadoopConf
+ .getTrimmedStrings(s"dfs.ha.namenodes.$ns")
+ .map(id => hadoopConf.getTrimmed(s"dfs.namenode.rpc-address.$ns.$id"))
+ if (nameNodes.nonEmpty) {
+ if (nameNodes.forall(hasPort)) Some(s"hdfs.name-node.$ns" ->
nameNodes.mkString(","))
+ else None
+ } else if (logicalNames.contains(ns)) {
+ None
+ } else {
+ Some(
+ s"hdfs.name-node.$ns" ->
s"$ns:${HdfsClientConfigKeys.DFS_NAMENODE_RPC_PORT_DEFAULT}")
+ }
+ }.toMap
+ declarations ++ clientSettings
+ }
+
+ /**
+ * The first portless HDFS authority among `authorities` that `properties`
declares neither with
+ * iceberg-rust's `hdfs.name-node.<ns>` nor with Hadoop's
`dfs.ha.namenodes.<ns>`.
+ */
+ def undeclaredHdfsNameservice(
+ authorities: Set[String],
+ properties: Map[String, String]): Option[String] =
+ hdfsNameservices(authorities).find { ns =>
+ !properties.contains(s"hdfs.name-node.$ns") &&
+ !properties.contains(s"hadoop.dfs.ha.namenodes.$ns")
+ }
+
+ private def hdfsNameservices(authorities: Set[String]): Seq[String] =
+ authorities.filterNot(hasPort).toSeq.sorted
+
+ /**
+ * Listings taken for the encryption-zone check, by filesystem URI and
parent directory: when
+ * each was taken, and the encrypted flag of every child. A data directory
cannot become a zone
+ * root once it holds files, so a listing is reused until
`spark.sql.metadataCacheTTLSeconds`
+ * expires it, if set; bounded so a long session over many tables stays
small.
+ */
+ private val listedParents: java.util.Map[(String, String), (Long,
Map[String, Boolean])] =
+ java.util.Collections.synchronizedMap(
+ new java.util.LinkedHashMap[(String, String), (Long, Map[String,
Boolean])](
+ 64,
+ 0.75f,
+ true) {
+ override def removeEldestEntry(
+ eldest: java.util.Map.Entry[(String, String), (Long, Map[String,
Boolean])])
+ : Boolean =
+ size() > 10000
+ })
+
+ /** Listings one scan may take for the encryption-zone check before it is
declined instead. */
+ private[comet] val MaxEncryptionZoneListings = 1000
+
+ /**
+ * Why a scan of `directories`, HDFS directories of data and delete files,
must not be claimed
+ * on account of HDFS encryption zones, or None. iceberg-rust's HDFS client
is built without KMS
+ * support and rejects every file of a zone. The NameNode flags each path
inside a zone as
+ * encrypted, directories included, so one listing of a parent tells the
flag of every data
+ * directory under it: the partitions of a table cost one call, not one
each. A listing is kept
+ * (see [[listedParents]]), and a directory missing from a kept listing, a
new partition, lists
+ * its parent again. A cluster that cannot hold a zone at all, one whose
NameNode advertises no
+ * key provider (a client-cached call) and whose session configures none, is
not listed. A scan
+ * that would take more than `maxListings` listings is declined after that
many, which are kept,
+ * so a later scan continues from them.
+ */
+ def hdfsEncryptionZoneFallback(
+ hadoopConf: Configuration,
+ directories: Set[String],
+ ttlSeconds: Long = -1L,
+ maxListings: Int = MaxEncryptionZoneListings,
+ nowMillis: Long = System.currentTimeMillis()): Option[String] = {
+ val configuredProvider =
+ Seq("hadoop.security.key.provider.path",
"dfs.encryption.key.provider.uri")
+ .exists(key => hadoopConf.getTrimmed(key, "").nonEmpty)
+ def fresh(listedAt: Long): Boolean =
Review Comment:
With `spark.sql.metadataCacheTTLSeconds` unset, which is its default,
`fresh` is always true, so a cached "not encrypted" listing lives as long as
the process. The argument that a directory holding files cannot become a zone
root is right for the same directory. It does not cover a table that is dropped
with purge and recreated at the same path inside a new encryption zone. In a
long-lived driver such as a Thrift server, the scan would still be claimed and
the tasks would fail on the encrypted files.
Could the entries get a default maximum age when the TTL is unset, say a few
minutes? The function already takes `nowMillis`, so a test with an injected
clock would cover it.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -924,6 +955,67 @@ 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)"
+ }
+ // The client connects as HADOOP_USER_NAME, or as the executor's OS
user, never as the
+ // Spark user the JVM reader runs as (opendal exposes no user
setting). Claimed only
+ // when the executors are known to read as the Spark user:
HADOOP_USER_NAME set to it
+ // in their environment, or, in local mode, the JVM's own user.
+ val sparkUser = session.sparkContext.sparkUser
+ val executorEnvKey = "spark.executorEnv.HADOOP_USER_NAME"
+ val executorUser = Option(
Review Comment:
Thanks for adding the identity test and the fallback reason. One question on
the gate itself. It reads `spark.executorEnv.HADOOP_USER_NAME` from the session
SQLConf first. A session built with
`SparkSession.builder().config(...).getOrCreate()` on an existing context can
hold that key in its SQLConf without the executors ever seeing it. The planner
would then believe the executors read as the Spark user while the native client
connects as their OS user. The result is an `AccessControlException`, or a read
under an identity the Spark user did not have, with no fallback.
Would it make sense to read only the live conf here (`SparkEnv.get.conf` or
`sparkContext.getConf`)? The planner tests set the value with `withSQLConf`, so
they would need to set it on the live conf and restore it. A case where the
SQLConf names the Spark user and the live conf names nobody, which must fall
back, would pin it.
--
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]
