Copilot commented on code in PR #12483:
URL: https://github.com/apache/gluten/pull/12483#discussion_r3557963963
##########
gluten-substrait/src/main/scala/org/apache/gluten/execution/BatchScanExecTransformer.scala:
##########
@@ -113,6 +113,8 @@ abstract class BatchScanExecTransformerBase(
postDriverMetrics()
}
+ override def readerOptions: Map[String, String] = Map.empty
Review Comment:
`BatchScanExecTransformerBase` currently hard-codes `readerOptions` to
`Map.empty`, which means DSv2 `FileScan.options` are never included in the
merged `fsConf` (even though the comment in `WholeStageTransformer` mentions
DSv2).
If FS credentials are provided via `DataFrameReader.option(...)` for DSv2
sources, they still won’t reach native. Consider wiring `FileScan.options`
through `readerOptions` here.
##########
gluten-substrait/src/main/scala/org/apache/gluten/execution/GlutenWholeStageColumnarRDD.scala:
##########
@@ -35,6 +35,50 @@ trait BaseGlutenPartition extends Partition with
InputPartition {
def plan: Array[Byte]
}
+/**
+ * Wraps Hadoop/Velox filesystem credential key-value pairs (fs.azure.*,
fs.s3a.*, fs.gs.*) so they
+ * cannot be accidentally exposed through logging, toString, exception
messages, or other debug
+ * paths that might print an arbitrary object.
+ *
+ * `toString` is deliberately overridden to redact values - only key NAMES are
shown (these are not
+ * secret; only the bearer values like access keys, secret keys, and OAuth
client secrets are). This
+ * makes accidental leaks structurally harder: a future `logDebug(s"...
$fsConfHolder")` or similar
+ * call will print "FsCredentialConf(3 keys: [fs.azure.account.auth.type,
...], values redacted)"
+ * instead of the literal secret strings.
+ *
+ * The underlying map is also `private` so it cannot be reached by name from
outside this file
+ * without going through `.unsafeValue`, which is named to discourage casual
use.
+ */
+final case class FsCredentialConf private (private val raw: Map[String,
String]) {
+
+ /** Number of credential entries held. Safe to log. */
+ def size: Int = raw.size
+
+ def isEmpty: Boolean = raw.isEmpty
+ def nonEmpty: Boolean = raw.nonEmpty
+
+ /**
+ * Returns the underlying map with real values. Named `unsafeValue` to make
call sites grep-able
+ * and to discourage passing the result to logging code. Only the native JNI
boundary (extraConf
+ * for NativePlanEvaluator) should call this.
+ */
+ def unsafeValue: Map[String, String] = raw
+
+ override def toString: String = {
+ if (raw.isEmpty) {
+ "FsCredentialConf(empty)"
+ } else {
+ s"FsCredentialConf(${raw.size} keys:
[${raw.keys.toSeq.sorted.mkString(", ")}], " +
+ "values redacted)"
+ }
+ }
+}
Review Comment:
`FsCredentialConf` is declared as a **case class**, which means Scala will
generate `unapply`/`Product` accessors that can expose the underlying `raw` map
(including secret values) without going through `unsafeValue`. That undermines
the stated goal of making value access grep-able and structurally harder to
leak.
Consider making this a plain `final class` (not a case class) so the only
intentional escape hatch is `unsafeValue`, while keeping the redacting
`toString`.
##########
gluten-arrow/src/main/scala/org/apache/gluten/runtime/Runtimes.scala:
##########
@@ -18,10 +18,48 @@ package org.apache.gluten.runtime
import org.apache.spark.task.TaskResources
+import java.security.MessageDigest
import java.util
object Runtimes {
+ /**
+ * Produce a stable, value-free cache key for a (backendName, name,
extraConf) triple.
+ *
+ * Two problems with the old `s"$backendName:$name:$extraConf"` key:
+ *
+ * 1. **Credential leakage** – `Map.toString` embeds secret values (e.g.
`fs.s3a.secret.key`,
+ * `fs.azure.account.oauth2.client.secret`) in a plain heap string that
can appear in logs,
+ * thread dumps, and heap snapshots.
+ * 2. **Nondeterminism** – Scala `Map.toString` does not guarantee
insertion order, so two maps
+ * with identical entries can produce different strings, causing
spurious duplicate
+ * `VeloxRuntime` registrations within a task.
+ *
+ * Fix: sort keys, hash them with SHA-256, and use only the hex digest as
the key. Values are
+ * intentionally excluded from the digest – distinct configs (different
credentials for the same
+ * key set) that need separate runtimes are already separated at the task
level through
+ * `GlutenPartition.fsConf`. Within a single task the key set is stable, so
the digest is stable.
+ */
+ private def stableKey(
+ backendName: String,
+ name: String,
+ extraConf: util.Map[String, String]): String = {
+ val digest = MessageDigest.getInstance("SHA-256")
+ digest.update(backendName.getBytes("UTF-8"))
+ digest.update(0.toByte) // field separator
+ digest.update(name.getBytes("UTF-8"))
+ digest.update(0.toByte)
+ // Sort keys for determinism; hash only keys, not values, to avoid leaking
secrets.
+ val sortedKeys = new java.util.ArrayList(extraConf.keySet)
+ java.util.Collections.sort(sortedKeys)
+ sortedKeys.forEach {
+ k =>
+ digest.update(k.getBytes("UTF-8"))
+ digest.update(0.toByte)
+ }
Review Comment:
`stableKey` hashes only the *keys* of `extraConf`. Since `RuntimeImpl`
merges `extraConf` **values** into the serialized native config (see
`Runtime.scala`), two different `extraConf` maps that share the same key set
(but different values/credentials) will collide and can cause `TaskResources`
to incorrectly reuse a `Runtime` configured with the wrong credentials/settings.
Also, the scaladoc rationale references `GlutenPartition.fsConf`, but
`GlutenPartition` does not carry `fsConf` in this PR (it’s stored on the RDD
instead), so the “values intentionally excluded” justification is no longer
accurate.
Hash sorted (key,value) pairs so the cache key stays deterministic and does
not embed secrets as a string, while still separating distinct configurations.
##########
gluten-substrait/src/main/scala/org/apache/gluten/execution/WholeStageTransformer.scala:
##########
@@ -367,6 +367,56 @@ case class WholeStageTransformer(child: SparkPlan,
materializeInput: Boolean = f
allSplitInfos,
leafTransformers)
+ // Compute fsConf once here on the driver from the leaf transformers'
session
+ // Hadoop configuration. It is stored as a plain field on the RDD and
+ // serialized once per executor via the existing task closure - no
per-partition
+ // copies, no Broadcast overhead (no BlockManager registration, no HTTP
fetch,
+ // no master round-trip). GlutenPartition never carries fsConf.
+ //
+ // Two sources merged (getPropsWithPrefix avoids full conf iteration):
+ // 1. sparkContext.hadoopConfiguration - spark.hadoop.* and
sc.hadoopConf.set()
+ // 2. SQLConf - spark.conf.set("fs.*", ...) keys that live only in
SQLConf
+ val fsConf: Map[String, String] = {
+ import scala.collection.JavaConverters._
+ val fsPrefixes = Seq("fs.azure.", "fs.s3a.", "fs.gs.")
+
+ // Source 1: sparkContext.hadoopConfiguration (spark.hadoop.* and
sc.hadoopConf.set()).
+ // Read directly from the mutable base Configuration via
getPropsWithPrefix
+ // rather than sessionState.newHadoopConf(), which would make a full copy
+ // of the entire Hadoop config (hundreds of entries from core-site.xml,
+ // hdfs-site.xml, etc.) just to look up a handful of fs.* keys.
+ // getPropsWithPrefix reads directly from the internal map and returns
only
+ // matching entries - cost proportional to matches, not total config
size.
+ // scalastyle:off hadoopconfiguration
+ val baseHadoopConf = sparkContext.hadoopConfiguration
+ // scalastyle:on hadoopconfiguration
+ val fromHadoop: Map[String, String] = fsPrefixes.flatMap {
+ prefix =>
+ baseHadoopConf.getPropsWithPrefix(prefix).asScala
+ .map { case (suffix, v) => (prefix + suffix) -> v }
+ }.toMap
+
+ // Source 2: SQLConf (spark.conf.set("fs.*", ...) at session level).
+ val fromSqlConf: Map[String, String] =
+ org.apache.spark.sql.internal.SQLConf.get.getAllConfs.filter {
+ case (k, _) => fsPrefixes.exists(k.startsWith)
+ }
Review Comment:
`fromSqlConf` only picks up keys that start with `fs.azure.`, `fs.s3a.`, or
`fs.gs.`. In Spark, session-level Hadoop FS configs are very commonly set as
`spark.hadoop.fs.*` via `spark.conf.set(...)`; those will be missed here, so
the PR still won’t pass many session-level FS settings to native.
Consider normalizing `spark.hadoop.*` keys from `SQLConf` by stripping the
prefix and then applying the same `fs.*` prefix filter, so both `fs.*` and
`spark.hadoop.fs.*` session configs are propagated.
--
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]
