From 17216aa98217f1380e9214ef40b5ba26ecbfcb7b Mon Sep 17 00:00:00 2001 From: Yuan Date: Fri, 3 Jul 2026 14:06:08 +0100 Subject: [PATCH] [VL] allow to pass hadoop options to native backend Signed-off-by: Yuan Revert "[VL] allow to pass hadoop options to native backend" This reverts commit 1a969763e07987afa9e8c42edb5cf6f0cd51c96e. fix Signed-off-by: Yuan fix format Signed-off-by: Yuan fix Signed-off-by: Yuan --- .../clickhouse/CHIteratorApi.scala | 3 +- .../backendsapi/velox/VeloxIteratorApi.scala | 6 +- .../org/apache/gluten/runtime/Runtimes.scala | 40 ++++++++++++- .../gluten/backendsapi/IteratorApi.scala | 3 +- .../execution/BasicScanExecTransformer.scala | 10 ++++ .../execution/BatchScanExecTransformer.scala | 2 + .../FileSourceScanExecTransformer.scala | 2 + .../GlutenWholeStageColumnarRDD.scala | 57 ++++++++++++++++++- .../execution/WholeStageTransformer.scala | 53 ++++++++++++++++- 9 files changed, 168 insertions(+), 8 deletions(-) diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHIteratorApi.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHIteratorApi.scala index 22e7a75aedf..81e1a560504 100644 --- a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHIteratorApi.scala +++ b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHIteratorApi.scala @@ -285,7 +285,8 @@ class CHIteratorApi extends IteratorApi with Logging with LogLevelUtil { partitionIndex: Int, inputIterators: Seq[Iterator[ColumnarBatch]] = Seq(), enableCudf: Boolean = false, - wsContext: WholeStageTransformContext = null + wsContext: WholeStageTransformContext = null, + fsConf: Map[String, String] = Map.empty ): Iterator[ColumnarBatch] = { require( diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala index f81b78c71f5..78ae782a809 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala @@ -201,7 +201,8 @@ class VeloxIteratorApi extends IteratorApi with Logging { partitionIndex: Int, inputIterators: Seq[Iterator[ColumnarBatch]] = Seq(), enableCudf: Boolean = false, - wsContext: WholeStageTransformContext = null): Iterator[ColumnarBatch] = { + wsContext: WholeStageTransformContext = null, + fsConf: Map[String, String] = Map.empty): Iterator[ColumnarBatch] = { assert( inputPartition.isInstanceOf[GlutenPartition], "Velox backend only accept GlutenPartition.") @@ -210,7 +211,8 @@ class VeloxIteratorApi extends IteratorApi with Logging { iter => new ColumnarBatchInIterator(BackendsApiManager.getBackendName, iter.asJava) } - val extraConf = Map(GlutenConfig.COLUMNAR_CUDF_ENABLED.key -> enableCudf.toString).asJava + val extraConf = + (fsConf + (GlutenConfig.COLUMNAR_CUDF_ENABLED.key -> enableCudf.toString)).asJava val transKernel = NativePlanEvaluator.create(BackendsApiManager.getBackendName, extraConf) val splitInfoByteArray = inputPartition diff --git a/gluten-arrow/src/main/scala/org/apache/gluten/runtime/Runtimes.scala b/gluten-arrow/src/main/scala/org/apache/gluten/runtime/Runtimes.scala index b4c344ca989..8fb5ae2af30 100644 --- a/gluten-arrow/src/main/scala/org/apache/gluten/runtime/Runtimes.scala +++ b/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) + } + digest.digest().map("%02x".format(_)).mkString + } + def contextInstance( backendName: String, name: String, @@ -30,7 +68,7 @@ object Runtimes { throw new IllegalStateException("This method must be called in a Spark task.") } TaskResources.addResourceIfNotRegistered( - s"$backendName:$name:$extraConf", + stableKey(backendName, name, extraConf), () => Runtime(backendName, name, extraConf)) } diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/IteratorApi.scala b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/IteratorApi.scala index 2cb45b97a82..8b85e41cc1f 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/IteratorApi.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/IteratorApi.scala @@ -69,7 +69,8 @@ trait IteratorApi { partitionIndex: Int, inputIterators: Seq[Iterator[ColumnarBatch]] = Seq(), enableCudf: Boolean = false, - wsContext: WholeStageTransformContext = null + wsContext: WholeStageTransformContext = null, + fsConf: Map[String, String] = Map.empty ): Iterator[ColumnarBatch] /** diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala index c765ed333da..c176e625317 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala @@ -73,6 +73,16 @@ trait BasicScanExecTransformer extends LeafTransformSupport with BaseDataSource */ def pushDownFilters: Option[Seq[Expression]] + /** + * Returns the options passed via DataFrameReader.option() or DataStreamReader.option(). For DSv1 + * scans these are stored in HadoopFsRelation.options and are NOT propagated to + * sparkContext.hadoopConfiguration or SQLConf, so they must be collected here explicitly and + * transported to the native backend via GlutenWholeStageColumnarRDD.fsConf. + * + * The default implementation returns Map.empty (used for DSv2 and non-file scans). + */ + def readerOptions: Map[String, String] = Map.empty + /** Copy the scan with filters that pushed by filterNode. */ def withNewPushdownFilters(filters: Seq[Expression]): BasicScanExecTransformer = { throw new UnsupportedOperationException( diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BatchScanExecTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BatchScanExecTransformer.scala index a0c3bb87575..d269a6814b8 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BatchScanExecTransformer.scala +++ b/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 + override def scanFilters: Seq[Expression] = scan match { case fileScan: FileScan => fileScan.dataFilters case _ => diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala index 83023f027ac..2250a72cbca 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala @@ -128,6 +128,8 @@ abstract class FileSourceScanExecTransformerBase( override def scanFilters: Seq[Expression] = dataFilters + override def readerOptions: Map[String, String] = relation.options + override def getMetadataColumns(): Seq[AttributeReference] = metadataColumns override def getPartitions: Seq[Partition] = { diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/GlutenWholeStageColumnarRDD.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/GlutenWholeStageColumnarRDD.scala index afec9cc10f6..1c2e6baadde 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/execution/GlutenWholeStageColumnarRDD.scala +++ b/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)" + } + } +} + +object FsCredentialConf { + val empty: FsCredentialConf = FsCredentialConf(Map.empty) + def apply(raw: Map[String, String]): FsCredentialConf = new FsCredentialConf(raw) +} + case class GlutenPartition( index: Int, plan: Array[Byte], @@ -61,9 +105,17 @@ class GlutenWholeStageColumnarRDD( updateInputMetrics: InputMetricsWrapper => Unit, updateNativeMetrics: IMetrics => Unit, enableCudf: Boolean = false, - wsContext: WholeStageTransformContext = null) + wsContext: WholeStageTransformContext = null, + private val fsConf: FsCredentialConf = FsCredentialConf.empty) extends RDD[ColumnarBatch](sc, rdds.getDependencies) { + // Override toString so that fsConf credential values are never exposed in + // Spark logs, DAG visualization, toDebugString, or the UI. Delegates to + // FsCredentialConf.toString, which shows only key NAMES (redacted values) - + // see FsCredentialConf's doc comment for the full rationale. + override def toString: String = + s"GlutenWholeStageColumnarRDD[$id] fsConf=$fsConf" + override def compute(split: Partition, context: TaskContext): Iterator[ColumnarBatch] = { GlutenTimeMetric.millis(pipelineTime) { _ => @@ -84,7 +136,8 @@ class GlutenWholeStageColumnarRDD( split.index, inputIterators, enableCudf, - wsContext + wsContext, + fsConf.unsafeValue ) } } diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/WholeStageTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/WholeStageTransformer.scala index 1f3ae4d753c..c424a25565a 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/execution/WholeStageTransformer.scala +++ b/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) + } + + // Source 3: DataFrameReader.option() / DataStreamReader.option() passed to + // each scan. These are stored in HadoopFsRelation.options (DSv1) or + // FileScan.options (DSv2) and are NOT in hadoopConfiguration or SQLConf. + // Collect from all leaf scan transformers and merge (last writer wins, so + // later leaves can override earlier ones for the same key). + val fromReaderOptions: Map[String, String] = + leafTransformers + .collect { case s: BasicScanExecTransformer => s.readerOptions } + .flatMap(_.filter { case (k, _) => fsPrefixes.exists(k.startsWith) }) + .toMap + + // Precedence: readerOptions (most specific) > SQLConf > hadoopConfiguration + fromHadoop ++ fromSqlConf ++ fromReaderOptions + } + val rdd = new GlutenWholeStageColumnarRDD( sparkContext, inputPartitions, @@ -380,7 +430,8 @@ case class WholeStageTransformer(child: SparkPlan, materializeInput: Boolean = f wsCtx.substraitContext.registeredAggregationParams ), wsCtx.enableCudf, - wsCtx + wsCtx, + FsCredentialConf(fsConf) ) val allInputPartitions = leafTransformers.map(_.getPartitions)