mixermt commented on code in PR #5898:
URL: https://github.com/apache/datafusion-comet/pull/5898#discussion_r4477884862
##########
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:
Right: `SparkSession.builder().config(...)` on an existing context, or
`SET`, puts the key in the SQLConf without the executors seeing it.
- The gate now reads only `sparkContext.getConf`, the configuration the
executors were launched with; the SQLConf is not consulted.
- The planner tests set the live conf (`SparkEnv.get.conf`, restored
afterwards). New case: the Spark user in the SQLConf and another user on the
context is declined; the reverse is claimed, since in local mode the JVM's user
applies, which is also why "SQLConf names the Spark user, live conf names
nobody" cannot fall back there.
- The docs say the setting has to be given when the application is submitted.
##########
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:
Right, a purged table re-created inside a new zone at the same path would
hit the stale listing. Two changes:
- Listings are keyed by table UUID as well: the re-created table is a new
UUID, so its first scan lists again, at no periodic cost.
- A listing older than an hour is taken again when
`spark.sql.metadataCacheTTLSeconds` 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.
Both are in `CometIcebergNativeScanSuite`, with the injected clock and a
second UUID over the same directories.
--
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]
