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

backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/GlutenParquetFileFormat.scala

@@ -62,7 +62,7 @@ class GlutenParquetFileFormat
     conf.set(ParquetOutputFormat.COMPRESSION, parquetOptions.compressionCodecClassName)
     val nativeConf =
       GlutenFormatFactory("parquet")
-        .nativeConf(options, parquetOptions.compressionCodecClassName)
+        .nativeConf(sparkSession, options, parquetOptions.compressionCodecClassName)
 
     new OutputWriterFactory {
       override def getFileExtension(context: TaskAttemptContext): String = {

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/main/scala/org/apache/gluten/datasource/VeloxDataSourceUtil.scala

@@ -31,25 +31,60 @@ import java.util
 
 object VeloxDataSourceUtil {
   def readSchema(files: Seq[FileStatus]): Option[StructType] = {
+    readSchema(files, new util.HashMap[String, String]())
+  }
+
+  def readSchema(
+      files: Seq[FileStatus],
+      fsConf: util.Map[String, String]): Option[StructType] = {
     if (files.isEmpty) {
       throw new IllegalArgumentException("No input file specified")
     }
-    readSchema(files.toList.head)
+    readSchema(files.toList.head, fsConf)
   }
 
   def readSchema(file: FileStatus): Option[StructType] = {
+    readSchema(file, new util.HashMap[String, String]())
+  }
+
+  def readSchema(
+      file: FileStatus,
+      fsConf: util.Map[String, String]): Option[StructType] = {
     val allocator = ArrowBufferAllocators.contextInstance()
-    val runtime = Runtimes.contextInstance(BackendsApiManager.getBackendName, "VeloxWriter")
+    val runtime =
+      Runtimes.contextInstance(BackendsApiManager.getBackendName, "VeloxWriter", fsConf)
     val datasourceJniWrapper = VeloxDataSourceJniWrapper.create(runtime)
-    val dsHandle =
-      datasourceJniWrapper.init(file.getPath.toString, -1, new util.HashMap[String, String]())
-    val cSchema = ArrowSchema.allocateNew(allocator)
-    datasourceJniWrapper.inspectSchema(dsHandle, cSchema.memoryAddress())
+    withDataSourceResource[ArrowSchema, Option[StructType]](
+      () => datasourceJniWrapper.init(file.getPath.toString, -1, fsConf),
+      () => ArrowSchema.allocateNew(allocator),
+      (dsHandle, cSchema) => {
+        datasourceJniWrapper.inspectSchema(dsHandle, cSchema.memoryAddress())
+        Option(SparkSchemaUtil.fromArrowSchema(ArrowAbiUtil.importToSchema(allocator, cSchema)))
+      },
+      datasourceJniWrapper.close
+    )
+  }
+
+  private[datasource] def withDataSourceResource[R <: AutoCloseable, T](
+      initialize: () => Long,
+      allocate: () => R,
+      use: (Long, R) => T,
+      closeHandle: Long => Unit): T = {
+    var dsHandle: Option[Long] = None
+    var resource: R = null.asInstanceOf[R]
     try {
-      Option(SparkSchemaUtil.fromArrowSchema(ArrowAbiUtil.importToSchema(allocator, cSchema)))
+      val handle = initialize()
+      dsHandle = Some(handle)
+      resource = allocate()
+      use(handle, resource)
     } finally {
-      cSchema.close()
-      datasourceJniWrapper.close(dsHandle)
+      try {
+        if (resource != null) {
+          resource.close()
+        }
+      } finally {
+        dsHandle.foreach(closeHandle)
+      }
     }
   }
 }

backends-velox/src/main/scala/org/apache/spark/sql/execution/datasources/velox/VeloxFormatWriterInjects.scala

@@ -20,7 +20,7 @@ import org.apache.gluten.backendsapi.BackendsApiManager
 import org.apache.gluten.columnarbatch.ColumnarBatches
 import org.apache.gluten.datasource.{VeloxDataSourceJniWrapper, VeloxDataSourceUtil}
 import org.apache.gluten.exception.GlutenException
-import org.apache.gluten.execution.BatchCarrierRow
+import org.apache.gluten.execution.{BatchCarrierRow, HadoopConfCollector}
 import org.apache.gluten.execution.datasource.GlutenRowSplitter
 import org.apache.gluten.memory.arrow.alloc.ArrowBufferAllocators
 import org.apache.gluten.runtime.Runtimes
@@ -33,12 +33,33 @@ import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql.utils.SparkArrowUtil
 import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.task.TaskResources
 
 import org.apache.arrow.c.ArrowSchema
 import org.apache.hadoop.fs.{FileStatus, Path}
 import org.apache.hadoop.mapreduce.TaskAttemptContext
 
 import java.io.IOException
+import java.util
+
+import scala.collection.JavaConverters._
+
+object VeloxFormatWriterInjects {
+  private val SparkHadoopFsPrefix = "spark.hadoop.fs."
+
+  private[velox] def runtimeFsConf(
+      nativeConf: util.Map[String, String]): util.Map[String, String] = {
+    nativeConf.asScala.filter {
+      case (key, _) => key.startsWith(SparkHadoopFsPrefix)
+    }.asJava
+  }
+
+  private[velox] def runWithTaskResources[T](sparkSession: SparkSession)(body: => T): T = {
+    sparkSession.withActive {
+      TaskResources.runUnsafe(body)
+    }
+  }
+}
 
 trait VeloxFormatWriterInjects extends GlutenFormatWriterInjectsBase {
   def createOutputWriter(
@@ -60,7 +81,10 @@ trait VeloxFormatWriterInjects extends GlutenFormatWriterInjectsBase {
       SparkArrowUtil.toArrowSchema(dataSchema, SQLConf.get.sessionLocalTimeZone)
     val cSchema = ArrowSchema.allocateNew(ArrowBufferAllocators.contextInstance())
     var dsHandle = -1L
-    val runtime = Runtimes.contextInstance(BackendsApiManager.getBackendName, "VeloxWriter")
+    val runtime = Runtimes.contextInstance(
+      BackendsApiManager.getBackendName,
+      "VeloxWriter",
+      VeloxFormatWriterInjects.runtimeFsConf(nativeConf))
     val datasourceJniWrapper = VeloxDataSourceJniWrapper.create(runtime)
     val allocator = ArrowBufferAllocators.contextInstance()
     try {
@@ -103,7 +127,9 @@ trait VeloxFormatWriterInjects extends GlutenFormatWriterInjectsBase {
       sparkSession: SparkSession,
       options: Map[String, String],
       files: Seq[FileStatus]): Option[StructType] = {
-    VeloxDataSourceUtil.readSchema(files)
+    VeloxFormatWriterInjects.runWithTaskResources(sparkSession) {
+      VeloxDataSourceUtil.readSchema(files, HadoopConfCollector.collect(sparkSession).asJava)
+    }
   }
 }
 

backends-velox/src/main/scala/org/apache/spark/sql/execution/datasources/velox/VeloxParquetWriterInjects.scala

@@ -17,9 +17,13 @@
 package org.apache.spark.sql.execution.datasources.velox
 
 import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.HadoopConfCollector
 
+import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.internal.SQLConf
 
+import java.util
+
 import scala.collection.JavaConverters.mapAsJavaMapConverter
 import scala.collection.mutable
 
@@ -60,5 +64,17 @@ class VeloxParquetWriterInjects extends VeloxFormatWriterInjects {
     sparkOptions.asJava
   }
 
+  override def nativeConf(
+      sparkSession: SparkSession,
+      options: Map[String, String],
+      compressionCodec: String): java.util.Map[String, String] = {
+    val sparkOptions = new util.HashMap[String, String](
+      sparkSession.withActive {
+        nativeConf(options, compressionCodec)
+      })
+    sparkOptions.putAll(HadoopConfCollector.collect(sparkSession).asJava)
+    sparkOptions
+  }
+
   override val formatName: String = "parquet"
 }

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

@@ -0,0 +1,88 @@
+/*
+ * 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 {
+
+  test("extra conf preserves filesystem credentials and overrides control keys") {
+    val cudfKey = GlutenConfig.COLUMNAR_CUDF_ENABLED.key
+    val dynamicFilterKey = VeloxConfig.VALUE_STREAM_DYNAMIC_FILTER_ENABLED.key
+    val fsConf = Map(
+      "spark.hadoop.fs.s3a.access.key" -> "s3-access-key",
+      "spark.hadoop.fs.azure.account.key.container.dfs.core.windows.net" -> "azure-key",
+      "spark.hadoop.fs.gs.auth.service.account.private.key" -> "gcs-key",
+      "spark.hadoop.fs.oss.endpoint" -> "unknown-scheme-endpoint",
+      cudfKey -> "true",
+      dynamicFilterKey -> "true"
+    )
+    val original = fsConf
+
+    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.azure.account.key.container.dfs.core.windows.net") == "azure-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(cudfKey) == "false")
+    assert(merged(dynamicFilterKey) == "false")
+    assert(fsConf == original)
+    assert(fsConf(cudfKey) == "true")
+    assert(fsConf(dynamicFilterKey) == "true")
+  }
+
+  test("extra conf enables CUDF without injecting an enabled dynamic filter") {
+    val merged = VeloxIteratorApi.buildExtraConf(
+      Map.empty,
+      enableCudf = true,
+      supportsValueStreamDynamicFilter = true)
+
+    assert(merged(GlutenConfig.COLUMNAR_CUDF_ENABLED.key) == "true")
+    assert(!merged.contains(VeloxConfig.VALUE_STREAM_DYNAMIC_FILTER_ENABLED.key))
+  }
+
+  test("empty fsConf does not add filesystem keys to extra conf") {
+    val merged = VeloxIteratorApi.buildExtraConf(
+      Map.empty,
+      enableCudf = false,
+      supportsValueStreamDynamicFilter = false)
+
+    assert(!merged.keys.exists(_.startsWith("spark.hadoop.fs.")))
+  }
+
+  test("extra conf drops non-filesystem settings at the native boundary") {
+    val merged = VeloxIteratorApi.buildExtraConf(
+      Map(
+        "spark.hadoop.fs.s3a.endpoint" -> "s3-endpoint",
+        "spark.sql.session.timeZone" -> "UTC",
+        "spark.gluten.ugi.tokens" -> "delegation-token"),
+      enableCudf = false,
+      supportsValueStreamDynamicFilter = true
+    )
+
+    assert(merged("spark.hadoop.fs.s3a.endpoint") == "s3-endpoint")
+    assert(!merged.contains("spark.sql.session.timeZone"))
+    assert(!merged.contains("spark.gluten.ugi.tokens"))
+  }
+}