[GLUTEN-10113][VL] Pass hadoop fs related session level configurations to native

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(

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

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))
   }
 

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]
 
   /**

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(

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 _ =>

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] = {

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
         )
     }
   }

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)