[TEST][VL] Forward Hadoop filesystem configuration per runtime

Collect user-provided filesystem settings once per Spark session and carry them through whole-stage RDDs and native writer paths without adding credentials to the Substrait plan.

Create per-runtime Velox connector configuration, use credential-safe runtime identity, and redact sensitive filesystem and UGI settings. Preserve legacy overloads and constructors for backend compatibility.

Author: jackylee-ch

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala

@@ -18,7 +18,7 @@ package org.apache.gluten.backendsapi.velox
 
 import org.apache.gluten.GlutenBuildInfo._
 import org.apache.gluten.backendsapi._
-import org.apache.gluten.config.{GlutenConfig, VeloxConfig}
+import org.apache.gluten.config.{GlutenConfig, HadoopConfContributor, VeloxConfig}
 import org.apache.gluten.exception.GlutenNotSupportException
 import org.apache.gluten.execution.ValidationResult
 import org.apache.gluten.execution.WriteFilesExecTransformer
@@ -52,7 +52,7 @@ import org.apache.hadoop.fs.Path
 
 import scala.util.control.Breaks.breakable
 
-class VeloxBackend extends SubstraitBackend {
+class VeloxBackend extends SubstraitBackend with HadoopConfContributor {
   import VeloxBackend._
 
   override def name(): String = VeloxBackend.BACKEND_NAME
@@ -72,6 +72,7 @@ class VeloxBackend extends SubstraitBackend {
   override def settings(): BackendSettingsApi = VeloxBackendSettings
   override def convFuncOverride(): ConventionFunc.Override = new ConvFunc()
   override def costers(): Seq[LongCoster] = Seq(LegacyCoster, RoughCoster)
+  override def interestedPrefixes(): Set[String] = Set("fs.")
 }
 
 object VeloxBackend {

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala

@@ -202,6 +202,30 @@ class VeloxIteratorApi extends IteratorApi with Logging {
       inputIterators: Seq[Iterator[ColumnarBatch]] = Seq(),
       enableCudf: Boolean = false,
       wsContext: WholeStageTransformContext = null): Iterator[ColumnarBatch] = {
+    genFirstStageIterator(
+      inputPartition,
+      context,
+      pipelineTime,
+      updateInputMetrics,
+      updateNativeMetrics,
+      partitionIndex,
+      inputIterators,
+      enableCudf,
+      wsContext,
+      Map.empty)
+  }
+
+  override def genFirstStageIterator(
+      inputPartition: BaseGlutenPartition,
+      context: TaskContext,
+      pipelineTime: SQLMetric,
+      updateInputMetrics: InputMetricsWrapper => Unit,
+      updateNativeMetrics: IMetrics => Unit,
+      partitionIndex: Int,
+      inputIterators: Seq[Iterator[ColumnarBatch]],
+      enableCudf: Boolean,
+      wsContext: WholeStageTransformContext,
+      fsConf: Map[String, String]): Iterator[ColumnarBatch] = {
     assert(
       inputPartition.isInstanceOf[GlutenPartition],
       "Velox backend only accept GlutenPartition.")
@@ -210,7 +234,9 @@ 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 = VeloxIteratorApi
+      .buildExtraConf(fsConf, enableCudf, supportsValueStreamDynamicFilter = true)
+      .asJava
     val transKernel = NativePlanEvaluator.create(BackendsApiManager.getBackendName, extraConf)
 
     val splitInfoByteArray = inputPartition
@@ -262,11 +288,36 @@ class VeloxIteratorApi extends IteratorApi with Logging {
       materializeInput: Boolean,
       enableCudf: Boolean = false,
       supportsValueStreamDynamicFilter: Boolean = true): Iterator[ColumnarBatch] = {
-    val extraConfMap = mutable.Map(GlutenConfig.COLUMNAR_CUDF_ENABLED.key -> enableCudf.toString)
-    if (!supportsValueStreamDynamicFilter) {
-      extraConfMap(VeloxConfig.VALUE_STREAM_DYNAMIC_FILTER_ENABLED.key) = "false"
-    }
-    val extraConf = extraConfMap.asJava
+    genFinalStageIterator(
+      context,
+      inputIterators,
+      sparkConf,
+      rootNode,
+      pipelineTime,
+      updateNativeMetrics,
+      partitionIndex,
+      materializeInput,
+      enableCudf,
+      supportsValueStreamDynamicFilter,
+      Map.empty
+    )
+  }
+
+  override def genFinalStageIterator(
+      context: TaskContext,
+      inputIterators: Seq[Iterator[ColumnarBatch]],
+      sparkConf: SparkConf,
+      rootNode: PlanNode,
+      pipelineTime: SQLMetric,
+      updateNativeMetrics: IMetrics => Unit,
+      partitionIndex: Int,
+      materializeInput: Boolean,
+      enableCudf: Boolean,
+      supportsValueStreamDynamicFilter: Boolean,
+      fsConf: Map[String, String]): Iterator[ColumnarBatch] = {
+    val extraConf = VeloxIteratorApi
+      .buildExtraConf(fsConf, enableCudf, supportsValueStreamDynamicFilter)
+      .asJava
     val transKernel = NativePlanEvaluator.create(BackendsApiManager.getBackendName, extraConf)
     val columnarNativeIterator =
       inputIterators.map {
@@ -303,6 +354,21 @@ class VeloxIteratorApi extends IteratorApi with Logging {
 }
 
 object VeloxIteratorApi {
+  private[gluten] def buildExtraConf(
+      fsConf: Map[String, String],
+      enableCudf: Boolean,
+      supportsValueStreamDynamicFilter: Boolean): Map[String, String] = {
+    val dynamicFilterConf =
+      if (supportsValueStreamDynamicFilter) {
+        Map.empty[String, String]
+      } else {
+        Map(VeloxConfig.VALUE_STREAM_DYNAMIC_FILTER_ENABLED.key -> "false")
+      }
+    val controlConf =
+      Map(GlutenConfig.COLUMNAR_CUDF_ENABLED.key -> enableCudf.toString) ++ dynamicFilterConf
+    fsConf.filter { case (key, _) => key.startsWith("spark.hadoop.fs.") } ++ controlConf
+  }
+
   // lookup table to translate '0' -> 0 ... 'F'/'f' -> 15
   private val unhexDigits = {
     val array = Array.fill[Byte](128)(-1)

backends-velox/src/test/scala/org/apache/gluten/backendsapi/velox/VeloxHadoopConfTransportSuite.scala

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.backendsapi.velox
+
+import org.apache.gluten.config.{GlutenConfig, VeloxConfig}
+
+import org.scalatest.funsuite.AnyFunSuite
+
+class VeloxHadoopConfTransportSuite extends AnyFunSuite {
+  private val cudfKey = GlutenConfig.COLUMNAR_CUDF_ENABLED.key
+  private val dynamicFilterKey = VeloxConfig.VALUE_STREAM_DYNAMIC_FILTER_ENABLED.key
+
+  test("extra conf keeps filesystem credentials, drops other settings, overrides control keys") {
+    val fsConf = Map(
+      "spark.hadoop.fs.s3a.access.key" -> "s3-access-key",
+      "spark.hadoop.fs.gs.auth.service.account.private.key" -> "gcs-key",
+      "spark.hadoop.fs.oss.endpoint" -> "unknown-scheme-endpoint",
+      "spark.sql.session.timeZone" -> "UTC",
+      "spark.gluten.ugi.tokens" -> "delegation-token",
+      cudfKey -> "true",
+      dynamicFilterKey -> "true"
+    )
+
+    val merged = VeloxIteratorApi.buildExtraConf(
+      fsConf,
+      enableCudf = false,
+      supportsValueStreamDynamicFilter = false)
+
+    assert(merged("spark.hadoop.fs.s3a.access.key") == "s3-access-key")
+    assert(merged("spark.hadoop.fs.gs.auth.service.account.private.key") == "gcs-key")
+    assert(merged("spark.hadoop.fs.oss.endpoint") == "unknown-scheme-endpoint")
+    assert(!merged.contains("spark.sql.session.timeZone"))
+    assert(!merged.contains("spark.gluten.ugi.tokens"))
+    assert(merged(cudfKey) == "false")
+    assert(merged(dynamicFilterKey) == "false")
+  }
+
+  test("extra conf enables CUDF without injecting an enabled dynamic filter") {
+    val merged = VeloxIteratorApi.buildExtraConf(
+      Map.empty,
+      enableCudf = true,
+      supportsValueStreamDynamicFilter = true)
+
+    assert(merged(cudfKey) == "true")
+    assert(!merged.contains(dynamicFilterKey))
+    assert(!merged.keys.exists(_.startsWith("spark.hadoop.fs.")))
+  }
+}

backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHadoopConfCollectorSuite.scala

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.execution
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.test.SharedSparkSession
+
+class VeloxHadoopConfCollectorSuite extends SparkFunSuite with SharedSparkSession {
+  test("collect caches the first Hadoop configuration snapshot for each Spark session") {
+    val key = "spark.hadoop.fs.s3a.gluten.collector.cached"
+    val session = spark.newSession()
+    session.conf.set(key, "first-snapshot")
+
+    val firstCollected = HadoopConfCollector.collect(session)
+    session.conf.set(key, "changed-after-first-collect")
+    val secondCollected = HadoopConfCollector.collect(session)
+
+    assert(secondCollected eq firstCollected)
+    assert(secondCollected(key) == "first-snapshot")
+  }
+
+  test("collect keeps configuration values isolated between Spark sessions") {
+    val key = "spark.hadoop.fs.s3a.gluten.collector.session"
+    val first = spark.newSession()
+    val second = spark.newSession()
+    first.conf.set(key, "first-session")
+    second.conf.set(key, "second-session")
+
+    val firstCollected = HadoopConfCollector.collect(first)
+    val secondCollected = HadoopConfCollector.collect(second)
+
+    assert(firstCollected(key) == "first-session")
+    assert(secondCollected(key) == "second-session")
+  }
+}

backends-velox/src/test/scala/org/apache/gluten/runtime/RuntimesSuite.scala

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.runtime
+
+import org.apache.spark.SparkFunSuite
+
+import java.util
+
+class RuntimesSuite extends SparkFunSuite {
+  test("resource ID neither exposes nor fingerprints filesystem configuration values") {
+    val secrets = linkedMap(
+      "spark.hadoop.fs.s3a.access.key" -> "AKIA-super-secret-access-key",
+      "spark.hadoop.fs.s3a.secret.key" -> "super-secret-credential-value")
+    val resourceId = Runtimes.resourceId("velox", "context", secrets)
+
+    assert(!resourceId.contains("AKIA-super-secret-access-key"))
+    assert(!resourceId.contains("super-secret-credential-value"))
+
+    // Rotating only the credential values must not change the runtime identity.
+    val rotated = linkedMap(
+      "spark.hadoop.fs.s3a.access.key" -> "different-access-key",
+      "spark.hadoop.fs.s3a.secret.key" -> "different-credential-value")
+    assert(Runtimes.resourceId("velox", "context", rotated) == resourceId)
+  }
+
+  test("resource ID is independent of map insertion order") {
+    val first = linkedMap("alpha" -> "one", "beta" -> "two")
+    val reversed = linkedMap("beta" -> "two", "alpha" -> "one")
+
+    assert(Runtimes.resourceId("velox", "context", first) ==
+      Runtimes.resourceId("velox", "context", reversed))
+  }
+
+  test("resource ID distinguishes non-filesystem configuration values") {
+    val first = linkedMap("spark.gluten.sql.columnar.backend.velox.cudf" -> "false")
+    val second = linkedMap("spark.gluten.sql.columnar.backend.velox.cudf" -> "true")
+
+    assert(Runtimes.resourceId("velox", "context", first) !=
+      Runtimes.resourceId("velox", "context", second))
+  }
+
+  test("resource ID uses unambiguous key and value encoding") {
+    val delimiterInKey = linkedMap("alpha=beta" -> "gamma")
+    val delimiterInValue = linkedMap("alpha" -> "beta=gamma")
+
+    assert(Runtimes.resourceId("velox", "context", delimiterInKey) !=
+      Runtimes.resourceId("velox", "context", delimiterInValue))
+  }
+
+  private def linkedMap(entries: (String, String)*): util.Map[String, String] = {
+    val result = new util.LinkedHashMap[String, String]()
+    entries.foreach { case (key, value) => result.put(key, value) }
+    result
+  }
+}