From cb57a6bfcdd06a725b6fa0473e4e2bdcea7fa2e5 Mon Sep 17 00:00:00 2001 From: jackylee Date: Fri, 10 Jul 2026 16:55:44 +0800 Subject: [PATCH] [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. --- .../backendsapi/velox/VeloxBackend.scala | 5 +- .../backendsapi/velox/VeloxIteratorApi.scala | 78 +++++++- .../velox/VeloxHadoopConfTransportSuite.scala | 62 +++++++ .../VeloxHadoopConfCollectorSuite.scala | 49 +++++ .../apache/gluten/runtime/RuntimesSuite.scala | 69 +++++++ .../VeloxWriteHadoopConfForwardingSuite.scala | 102 +++++++++++ cpp/velox/compute/VeloxBackend.cc | 37 +++- cpp/velox/compute/VeloxBackend.h | 20 +++ cpp/velox/compute/VeloxRuntime.cc | 16 +- cpp/velox/tests/RuntimeTest.cc | 81 ++++++++- cpp/velox/utils/ConfigExtractor.cc | 15 ++ cpp/velox/utils/ConfigExtractor.h | 5 + .../org/apache/gluten/runtime/Runtimes.scala | 25 ++- .../gluten/config/HadoopConfContributor.scala | 21 +++ .../gluten/backendsapi/IteratorApi.scala | 49 +++++ .../GlutenWholeStageColumnarRDD.scala | 35 ++-- .../execution/HadoopConfCollector.scala | 124 +++++++++++++ .../execution/WholeStageTransformer.scala | 12 +- .../WholeStageZippedPartitionsRDD.scala | 43 +++-- .../execution/HadoopConfCollectorSuite.scala | 88 +++++++++ .../execution/HadoopConfTransportSuite.scala | 168 ++++++++++++++++++ 21 files changed, 1053 insertions(+), 51 deletions(-) create mode 100644 backends-velox/src/test/scala/org/apache/gluten/backendsapi/velox/VeloxHadoopConfTransportSuite.scala create mode 100644 backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHadoopConfCollectorSuite.scala create mode 100644 backends-velox/src/test/scala/org/apache/gluten/runtime/RuntimesSuite.scala create mode 100644 backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxWriteHadoopConfForwardingSuite.scala create mode 100644 gluten-core/src/main/scala/org/apache/gluten/config/HadoopConfContributor.scala create mode 100644 gluten-substrait/src/main/scala/org/apache/gluten/execution/HadoopConfCollector.scala create mode 100644 gluten-ut/test/src/test/scala/org/apache/gluten/execution/HadoopConfCollectorSuite.scala create mode 100644 gluten-ut/test/src/test/scala/org/apache/gluten/execution/HadoopConfTransportSuite.scala diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala index 14150196817..53d3c91b1f0 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala +++ b/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 { 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..a47d7dce080 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 @@ -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) diff --git a/backends-velox/src/test/scala/org/apache/gluten/backendsapi/velox/VeloxHadoopConfTransportSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/backendsapi/velox/VeloxHadoopConfTransportSuite.scala new file mode 100644 index 00000000000..42705b94d47 --- /dev/null +++ b/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."))) + } +} diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHadoopConfCollectorSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHadoopConfCollectorSuite.scala new file mode 100644 index 00000000000..ae6c30ce722 --- /dev/null +++ b/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") + } +} diff --git a/backends-velox/src/test/scala/org/apache/gluten/runtime/RuntimesSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/runtime/RuntimesSuite.scala new file mode 100644 index 00000000000..6cc37c8278c --- /dev/null +++ b/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 + } +} diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxWriteHadoopConfForwardingSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxWriteHadoopConfForwardingSuite.scala new file mode 100644 index 00000000000..637cf30c56e --- /dev/null +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxWriteHadoopConfForwardingSuite.scala @@ -0,0 +1,102 @@ +/* + * 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.spark.sql.execution + +import org.apache.gluten.config.GlutenConfig +import org.apache.gluten.execution.{GlutenWholeStageColumnarRDD, VeloxWholeStageTransformerSuite, WholeStageZippedPartitionsRDD} + +import org.apache.spark.SparkConf +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.util.QueryExecutionListener + +import scala.collection.mutable + +class VeloxWriteHadoopConfForwardingSuite extends VeloxWholeStageTransformerSuite with WriteUtils { + + override protected val resourcePath: String = "" + override protected val fileFormat: String = "parquet" + + private val fsConfKey = "spark.hadoop.fs.s3a.gluten.write.access.key" + private val fsConfValue = "gluten-write-access-key" + + override protected def sparkConf: SparkConf = { + super.sparkConf + .set(GlutenConfig.NATIVE_WRITER_ENABLED.key, "true") + .set(fsConfKey, fsConfValue) + } + + // The write exec feeds `child.executeColumnar()` straight into the native writer, so the fsConf + // collected by WholeStageTransformer must live on one of the Gluten RDDs in that lineage. Walk + // dependencies in case the RDD is ever wrapped above the whole-stage RDD. + private def findFsConf(rdd: RDD[_]): Option[Map[String, String]] = { + val visited = mutable.Set.empty[Int] + def search(r: RDD[_]): Option[Map[String, String]] = { + if (!visited.add(r.id)) { + None + } else { + r match { + case g: GlutenWholeStageColumnarRDD => Some(g.fsConf) + case z: WholeStageZippedPartitionsRDD => Some(z.fsConf) + case other => + other.dependencies.iterator.map(dep => search(dep.rdd)).collectFirst { + case Some(conf) => conf + } + } + } + } + search(rdd) + } + + test("native write forwards spark.hadoop.fs.* config to the write RDD") { + // The columnar write files exec only exists on the Spark 3.4+ native write code path. + assume(isSparkVersionGE("3.4")) + + spark.range(0, 10, 1, 1).createOrReplaceTempView("gluten_write_fs_source") + + // The native write runs as a nested execution whose plan carries the write exec, mirroring how + // WriteUtils.checkNativeWrite detects native writes. + var writeExec: Option[ColumnarWriteFilesExec] = None + val listener = new QueryExecutionListener { + override def onFailure(f: String, qe: QueryExecution, e: Exception): Unit = {} + override def onSuccess(funcName: String, qe: QueryExecution, duration: Long): Unit = { + if (writeExec.isEmpty) { + writeExec = qe.executedPlan.collectFirst { case w: ColumnarWriteFilesExec => w } + } + } + } + + withTempPath { + path => + spark.listenerManager.register(listener) + try { + spark.sql(s""" + |INSERT OVERWRITE DIRECTORY USING PARQUET + |OPTIONS ('path' '${path.getCanonicalPath}') + |SELECT * FROM gluten_write_fs_source + |""".stripMargin) + spark.sparkContext.listenerBus.waitUntilEmpty() + } finally { + spark.listenerManager.unregister(listener) + } + } + + assert(writeExec.isDefined, "expected a native ColumnarWriteFilesExec on the write plan") + val fsConf = findFsConf(writeExec.get.child.executeColumnar()) + assert(fsConf.isDefined, "native write child RDD must carry a fsConf-bearing Gluten RDD") + assert(fsConf.get.get(fsConfKey).contains(fsConfValue)) + } +} diff --git a/cpp/velox/compute/VeloxBackend.cc b/cpp/velox/compute/VeloxBackend.cc index 0d45559f0e2..e33100db73f 100644 --- a/cpp/velox/compute/VeloxBackend.cc +++ b/cpp/velox/compute/VeloxBackend.cc @@ -384,20 +384,42 @@ void VeloxBackend::initCache() { std::shared_ptr VeloxBackend::createHiveConnector( const std::string& connectorId, folly::Executor* ioExecutor) const { - return std::make_shared(connectorId, hiveConnectorConfig_, ioExecutor); + return createHiveConnector(connectorId, ioExecutor, hiveConnectorConfig_); +} + +std::shared_ptr VeloxBackend::createHiveConnector( + const std::string& connectorId, + folly::Executor* ioExecutor, + std::shared_ptr hiveConnectorConfig) const { + return std::make_shared( + connectorId, std::move(hiveConnectorConfig), ioExecutor); } std::shared_ptr VeloxBackend::createDeltaConnector( const std::string& connectorId, folly::Executor* ioExecutor) const { - return std::make_shared(connectorId, hiveConnectorConfig_, ioExecutor); + return createDeltaConnector(connectorId, ioExecutor, hiveConnectorConfig_); +} + +std::shared_ptr VeloxBackend::createDeltaConnector( + const std::string& connectorId, + folly::Executor* ioExecutor, + std::shared_ptr hiveConnectorConfig) const { + return std::make_shared(connectorId, std::move(hiveConnectorConfig), ioExecutor); } std::shared_ptr VeloxBackend::createIcebergConnector( const std::string& connectorId, folly::Executor* ioExecutor) const { + return createIcebergConnector(connectorId, ioExecutor, hiveConnectorConfig_); +} + +std::shared_ptr VeloxBackend::createIcebergConnector( + const std::string& connectorId, + folly::Executor* ioExecutor, + std::shared_ptr hiveConnectorConfig) const { return std::make_shared( - connectorId, hiveConnectorConfig_, ioExecutor); + connectorId, std::move(hiveConnectorConfig), ioExecutor); } std::shared_ptr VeloxBackend::createValueStreamConnector( @@ -410,8 +432,15 @@ std::shared_ptr VeloxBackend::createValue std::shared_ptr VeloxBackend::createCudfHiveConnector( const std::string& connectorId, folly::Executor* ioExecutor) const { + return createCudfHiveConnector(connectorId, ioExecutor, hiveConnectorConfig_); +} + +std::shared_ptr VeloxBackend::createCudfHiveConnector( + const std::string& connectorId, + folly::Executor* ioExecutor, + std::shared_ptr hiveConnectorConfig) const { facebook::velox::cudf_velox::connector::hive::CudfHiveConnectorFactory factory; - return factory.newConnector(connectorId, hiveConnectorConfig_, ioExecutor); + return factory.newConnector(connectorId, std::move(hiveConnectorConfig), ioExecutor); } #endif diff --git a/cpp/velox/compute/VeloxBackend.h b/cpp/velox/compute/VeloxBackend.h index 09fad04dc4a..7ecb0e2591f 100644 --- a/cpp/velox/compute/VeloxBackend.h +++ b/cpp/velox/compute/VeloxBackend.h @@ -77,14 +77,29 @@ class VeloxBackend { const std::string& connectorId, folly::Executor* ioExecutor) const; + std::shared_ptr createHiveConnector( + const std::string& connectorId, + folly::Executor* ioExecutor, + std::shared_ptr hiveConnectorConfig) const; + std::shared_ptr createIcebergConnector( const std::string& connectorId, folly::Executor* ioExecutor) const; + std::shared_ptr createIcebergConnector( + const std::string& connectorId, + folly::Executor* ioExecutor, + std::shared_ptr hiveConnectorConfig) const; + std::shared_ptr createDeltaConnector( const std::string& connectorId, folly::Executor* ioExecutor) const; + std::shared_ptr createDeltaConnector( + const std::string& connectorId, + folly::Executor* ioExecutor, + std::shared_ptr hiveConnectorConfig) const; + std::shared_ptr createValueStreamConnector( const std::string& connectorId, bool dynamicFilterEnabled) const; @@ -93,6 +108,11 @@ class VeloxBackend { std::shared_ptr createCudfHiveConnector( const std::string& connectorId, folly::Executor* ioExecutor) const; + + std::shared_ptr createCudfHiveConnector( + const std::string& connectorId, + folly::Executor* ioExecutor, + std::shared_ptr hiveConnectorConfig) const; #endif void tearDown(); diff --git a/cpp/velox/compute/VeloxRuntime.cc b/cpp/velox/compute/VeloxRuntime.cc index 45649c05f8e..81bcae1463d 100644 --- a/cpp/velox/compute/VeloxRuntime.cc +++ b/cpp/velox/compute/VeloxRuntime.cc @@ -284,23 +284,25 @@ void VeloxRuntime::initializeExecutors() { void VeloxRuntime::registerConnectors() { auto* backend = VeloxBackend::get(); - connectorIds_.hiveRegistered = - velox::connector::registerConnector(backend->createHiveConnector(connectorIds_.hive, ioExecutor_.get())); + auto perRuntimeHiveConfig = + createHiveConnectorConfig(mergeFileSystemConfigs(backend->getBackendConf(), veloxCfg_), FileSystemType::kAll); + connectorIds_.hiveRegistered = velox::connector::registerConnector( + backend->createHiveConnector(connectorIds_.hive, ioExecutor_.get(), perRuntimeHiveConfig)); GLUTEN_CHECK(connectorIds_.hiveRegistered, "Failed to register scoped hive connector: " + connectorIds_.hive); GLUTEN_CHECK( velox::connector::hasConnector(connectorIds_.hive), "Scoped hive connector not found after registration: " + connectorIds_.hive); - connectorIds_.icebergRegistered = - velox::connector::registerConnector(backend->createIcebergConnector(connectorIds_.iceberg, ioExecutor_.get())); + connectorIds_.icebergRegistered = velox::connector::registerConnector( + backend->createIcebergConnector(connectorIds_.iceberg, ioExecutor_.get(), perRuntimeHiveConfig)); GLUTEN_CHECK( connectorIds_.icebergRegistered, "Failed to register scoped Iceberg connector: " + connectorIds_.iceberg); GLUTEN_CHECK( velox::connector::hasConnector(connectorIds_.iceberg), "Scoped Iceberg connector not found after registration: " + connectorIds_.iceberg); - connectorIds_.deltaRegistered = - velox::connector::registerConnector(backend->createDeltaConnector(connectorIds_.delta, ioExecutor_.get())); + connectorIds_.deltaRegistered = velox::connector::registerConnector( + backend->createDeltaConnector(connectorIds_.delta, ioExecutor_.get(), perRuntimeHiveConfig)); GLUTEN_CHECK(connectorIds_.deltaRegistered, "Failed to register scoped delta connector: " + connectorIds_.delta); GLUTEN_CHECK( velox::connector::hasConnector(connectorIds_.delta), @@ -320,7 +322,7 @@ void VeloxRuntime::registerConnectors() { if (veloxCfg_->get(kCudfEnableTableScan, kCudfEnableTableScanDefault) && veloxCfg_->get(kCudfEnabled, kCudfEnabledDefault)) { connectorIds_.cudfHiveRegistered = velox::connector::registerConnector( - backend->createCudfHiveConnector(connectorIds_.cudfHive, ioExecutor_.get())); + backend->createCudfHiveConnector(connectorIds_.cudfHive, ioExecutor_.get(), perRuntimeHiveConfig)); GLUTEN_CHECK( connectorIds_.cudfHiveRegistered, "Failed to register scoped cudf hive connector: " + connectorIds_.cudfHive); GLUTEN_CHECK( diff --git a/cpp/velox/tests/RuntimeTest.cc b/cpp/velox/tests/RuntimeTest.cc index df36ceaed1c..153ad4b214b 100644 --- a/cpp/velox/tests/RuntimeTest.cc +++ b/cpp/velox/tests/RuntimeTest.cc @@ -19,8 +19,12 @@ #include #include "compute/VeloxBackend.h" +#include "config/VeloxConfig.h" #include "memory.pb.h" #include "threads/ThreadInitializer.h" +#include "utils/ConfigExtractor.h" +#include "velox/connectors/hive/HiveConfig.h" +#include "velox/connectors/hive/storage_adapters/s3fs/S3Config.h" namespace gluten { @@ -160,15 +164,88 @@ TEST(TestRuntime, CreateRuntime) { } TEST(TestRuntime, CreateVeloxRuntime) { - VeloxBackend::create(AllocationListener::noop(), {}); + VeloxBackend::create( + AllocationListener::noop(), {{kLoadQuantum, "123456"}, {"spark.hadoop.fs.s3a.path.style.access", "true"}}); auto mm = MemoryManager::create(kVeloxBackendKind, AllocationListener::noop()); auto tm = ThreadManager::create(kVeloxBackendKind, ThreadInitializer::noop()); - auto runtime = Runtime::create(kVeloxBackendKind, mm, tm); + auto runtime = Runtime::create(kVeloxBackendKind, mm, tm, {{"spark.hadoop.fs.s3a.path.style.access", "false"}}); ASSERT_EQ(typeid(*runtime), typeid(VeloxRuntime)); + const auto* veloxRuntime = dynamic_cast(runtime); + const auto hiveConnector = facebook::velox::connector::getConnector(veloxRuntime->connectorIds().hive); + ASSERT_NE(hiveConnector, nullptr); + EXPECT_EQ( + hiveConnector->connectorConfig() + ->get(facebook::velox::connector::hive::HiveConfig::kLoadQuantum) + .value(), + "123456"); +#ifdef ENABLE_S3 + EXPECT_EQ( + hiveConnector->connectorConfig() + ->get(facebook::velox::filesystems::S3Config::baseConfigKey( + facebook::velox::filesystems::S3Config::Keys::kPathStyleAccess)) + .value(), + "false"); +#endif Runtime::release(runtime); ThreadManager::release(tm); } +TEST(TestRuntime, MergeFileSystemConfigsPreservesBackendTuning) { + const auto backendConf = + std::make_shared(std::unordered_map{ + {kLoadQuantum, "123456"}, + {"spark.hadoop.fs.s3a.access.key", "backend-access-key"}, + {"spark.sql.unrelated", "backend-value"}}); + const auto runtimeConf = + std::make_shared(std::unordered_map{ + {"spark.hadoop.fs.s3a.access.key", "runtime-access-key"}, + {"spark.hadoop.fs.azure.account.key.example", "runtime-abfs-key"}, + {"spark.sql.unrelated", "runtime-value"}}); + + const auto merged = mergeFileSystemConfigs(backendConf, runtimeConf); + + EXPECT_EQ(merged->get(kLoadQuantum).value(), "123456"); + EXPECT_EQ(merged->get("spark.hadoop.fs.s3a.access.key").value(), "runtime-access-key"); + EXPECT_EQ(merged->get("spark.hadoop.fs.azure.account.key.example").value(), "runtime-abfs-key"); + EXPECT_EQ(merged->get("spark.sql.unrelated").value(), "backend-value"); +} + +#ifdef ENABLE_ABFS +TEST(TestRuntime, ExtractsRuntimeAbfsConfig) { + const auto backendConf = + std::make_shared(std::unordered_map{}); + const auto runtimeConf = std::make_shared( + std::unordered_map{{"spark.hadoop.fs.azure.account.key.example", "runtime-abfs-key"}}); + + const auto hiveConfig = + createHiveConnectorConfig(mergeFileSystemConfigs(backendConf, runtimeConf), FileSystemType::kAll); + + EXPECT_EQ(hiveConfig->get("fs.azure.account.key.example").value(), "runtime-abfs-key"); +} +#endif + +#ifdef ENABLE_GCS +TEST(TestRuntime, ExtractsValidRuntimeGcsConfig) { + const auto backendConf = + std::make_shared(std::unordered_map{}); + const auto runtimeConf = + std::make_shared(std::unordered_map{ + {"spark.hadoop.fs.gs.storage.root.url", "https://storage.example.test"}, + {"spark.hadoop.fs.gs.auth.type", "SERVICE_ACCOUNT_JSON_KEYFILE"}, + {"spark.hadoop.fs.gs.auth.service.account.json.keyfile", "/tmp/gcs-key.json"}}); + + const auto hiveConfig = + createHiveConnectorConfig(mergeFileSystemConfigs(backendConf, runtimeConf), FileSystemType::kAll); + + EXPECT_EQ( + hiveConfig->get(facebook::velox::connector::hive::HiveConfig::kGcsEndpoint).value(), + "https://storage.example.test"); + EXPECT_EQ( + hiveConfig->get(facebook::velox::connector::hive::HiveConfig::kGcsCredentialsPath).value(), + "/tmp/gcs-key.json"); +} +#endif + TEST(TestRuntime, GetResultIterator) { DummyMemoryManager mm(kDummyBackendKind); DummyThreadManager tm(kDummyBackendKind); diff --git a/cpp/velox/utils/ConfigExtractor.cc b/cpp/velox/utils/ConfigExtractor.cc index 7ee2deae8a8..27a6b797713 100644 --- a/cpp/velox/utils/ConfigExtractor.cc +++ b/cpp/velox/utils/ConfigExtractor.cc @@ -246,6 +246,21 @@ std::string parquetSessionProperty(std::string_view key) { } // namespace +std::shared_ptr mergeFileSystemConfigs( + const std::shared_ptr& backendConf, + const std::shared_ptr& runtimeConf) { + constexpr std::string_view kSparkHadoopFsPrefix = "spark.hadoop.fs."; + + auto merged = backendConf->rawConfigs(); + for (const auto& [key, value] : runtimeConf->rawConfigs()) { + if (key.starts_with(kSparkHadoopFsPrefix)) { + merged.insert_or_assign(key, value); + } + } + + return std::make_shared(std::move(merged)); +} + std::shared_ptr createHiveConnectorSessionConfig( const std::shared_ptr& conf) { // The configs below are used at session level. diff --git a/cpp/velox/utils/ConfigExtractor.h b/cpp/velox/utils/ConfigExtractor.h index 77b38c2f295..e94f160c873 100644 --- a/cpp/velox/utils/ConfigExtractor.h +++ b/cpp/velox/utils/ConfigExtractor.h @@ -21,6 +21,7 @@ #include #include +#include #include #include "velox/common/config/Config.h" @@ -29,6 +30,10 @@ namespace gluten { enum class FileSystemType : uint8_t { kHdfs, kS3, kAbfs, kGcs, kAll }; +std::shared_ptr mergeFileSystemConfigs( + const std::shared_ptr& backendConf, + const std::shared_ptr& runtimeConf); + /// Create hive connector session config. std::shared_ptr createHiveConnectorSessionConfig( const std::shared_ptr& conf); 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..8e223bc476e 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 @@ -20,8 +20,31 @@ import org.apache.spark.task.TaskResources import java.util +import scala.collection.JavaConverters._ + object Runtimes { + private val SparkHadoopFsPrefix = "spark.hadoop.fs." + + private[gluten] def resourceId( + backendName: String, + name: String, + extraConf: util.Map[String, String]): String = { + // This id surfaces in TaskResources exception messages, so filesystem credential values (which + // may appear in extraConf) are replaced with an empty placeholder. Phase-1 assumes a single + // credential set per session, so distinct fs values need not yield distinct runtimes. The + // length prefixes keep the encoding unambiguous when keys or values contain delimiters. + val encoded = extraConf.asScala.toSeq + .sortBy { case (key, _) => key } + .map { + case (key, value) => + val safeValue = if (key.startsWith(SparkHadoopFsPrefix)) "" else value + s"${key.length}:$key=${safeValue.length}:$safeValue" + } + .mkString(",") + s"$backendName:$name:$encoded" + } + def contextInstance( backendName: String, name: String, @@ -30,7 +53,7 @@ object Runtimes { throw new IllegalStateException("This method must be called in a Spark task.") } TaskResources.addResourceIfNotRegistered( - s"$backendName:$name:$extraConf", + resourceId(backendName, name, extraConf), () => Runtime(backendName, name, extraConf)) } diff --git a/gluten-core/src/main/scala/org/apache/gluten/config/HadoopConfContributor.scala b/gluten-core/src/main/scala/org/apache/gluten/config/HadoopConfContributor.scala new file mode 100644 index 00000000000..5886f85a2a3 --- /dev/null +++ b/gluten-core/src/main/scala/org/apache/gluten/config/HadoopConfContributor.scala @@ -0,0 +1,21 @@ +/* + * 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.config + +trait HadoopConfContributor { + def interestedPrefixes(): Set[String] = Set.empty +} 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..d8e6c088d0d 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 @@ -72,6 +72,29 @@ trait IteratorApi { wsContext: WholeStageTransformContext = null ): Iterator[ColumnarBatch] + 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] = { + genFirstStageIterator( + inputPartition, + context, + pipelineTime, + updateInputMetrics, + updateNativeMetrics, + partitionIndex, + inputIterators, + enableCudf, + wsContext) + } + /** * Generate Iterator[ColumnarBatch] for final stage. ("Final" means it depends on other SCAN * inputs, maybe it was a mistake to use the word "final") @@ -88,5 +111,31 @@ trait IteratorApi { materializeInput: Boolean = false, enableCudf: Boolean = false, supportsValueStreamDynamicFilter: Boolean = true): Iterator[ColumnarBatch] + + 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] = { + genFinalStageIterator( + context, + inputIterators, + sparkConf, + rootNode, + pipelineTime, + updateNativeMetrics, + partitionIndex, + materializeInput, + enableCudf, + supportsValueStreamDynamicFilter + ) + } // scalastyle:on argcount } 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..610c3901215 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 @@ -16,7 +16,7 @@ */ package org.apache.gluten.execution -import org.apache.gluten.backendsapi.BackendsApiManager +import org.apache.gluten.backendsapi.{BackendsApiManager, IteratorApi} import org.apache.gluten.config.GlutenConfig import org.apache.gluten.extension.ApplyStageInputStatsRule import org.apache.gluten.metrics.{GlutenTimeMetric, IMetrics} @@ -61,7 +61,8 @@ class GlutenWholeStageColumnarRDD( updateInputMetrics: InputMetricsWrapper => Unit, updateNativeMetrics: IMetrics => Unit, enableCudf: Boolean = false, - wsContext: WholeStageTransformContext = null) + wsContext: WholeStageTransformContext = null, + val fsConf: Map[String, String] = Map.empty) extends RDD[ColumnarBatch](sc, rdds.getDependencies) { override def compute(split: Partition, context: TaskContext): Iterator[ColumnarBatch] = { @@ -75,20 +76,34 @@ class GlutenWholeStageColumnarRDD( } val (inputPartition, inputColumnarRDDPartitions) = castNativePartition(split) val inputIterators = rdds.getIterators(inputColumnarRDDPartitions, context) - BackendsApiManager.getIteratorApiInstance.genFirstStageIterator( + createFirstStageIterator( + BackendsApiManager.getIteratorApiInstance, inputPartition, context, - pipelineTime, - updateInputMetrics, - updateNativeMetrics, - split.index, inputIterators, - enableCudf, - wsContext - ) + split.index) } } + private[gluten] def createFirstStageIterator( + iteratorApi: IteratorApi, + inputPartition: BaseGlutenPartition, + context: TaskContext, + inputIterators: Seq[Iterator[ColumnarBatch]], + partitionIndex: Int): Iterator[ColumnarBatch] = { + iteratorApi.genFirstStageIterator( + inputPartition, + context, + pipelineTime, + updateInputMetrics, + updateNativeMetrics, + partitionIndex, + inputIterators, + enableCudf, + wsContext, + fsConf) + } + private def castNativePartition(split: Partition): (BaseGlutenPartition, Seq[Partition]) = { split match { case FirstZippedPartitionsPartition(_, g: BaseGlutenPartition, p) => (g, p) diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/HadoopConfCollector.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/HadoopConfCollector.scala new file mode 100644 index 00000000000..e2363d37cbf --- /dev/null +++ b/gluten-substrait/src/main/scala/org/apache/gluten/execution/HadoopConfCollector.scala @@ -0,0 +1,124 @@ +/* + * 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.gluten.component.Component +import org.apache.gluten.config.HadoopConfContributor + +import org.apache.spark.sql.SparkSession + +import org.apache.hadoop.conf.Configuration + +import java.util.WeakHashMap + +import scala.collection.JavaConverters._ + +object HadoopConfCollector { + private val SparkHadoopPrefix = "spark.hadoop." + private val sessionCache = new WeakHashMap[SparkSession, Map[String, String]]() + + def collect(session: SparkSession): Map[String, String] = { + val prefixes = interestedPrefixes() + if (prefixes.isEmpty) { + Map.empty + } else { + sessionCache.synchronized { + Option(sessionCache.get(session)).getOrElse { + val collected = collectFromSessionProvider(() => session, prefixes) + sessionCache.put(session, collected) + collected + } + } + } + } + + private[gluten] def collectFromSessionProvider( + sessionProvider: () => SparkSession, + prefixes: Set[String]): Map[String, String] = { + if (prefixes.isEmpty) { + Map.empty + } else { + val session = sessionProvider() + val conf = new Configuration(session.sessionState.newHadoopConf()) + collectCopied(conf, prefixes) + } + } + + private[gluten] def collect( + conf: Configuration, + prefixes: Set[String]): Map[String, String] = { + if (prefixes.isEmpty) { + Map.empty + } else { + collectCopied(new Configuration(conf), prefixes) + } + } + + private def interestedPrefixes(): Set[String] = { + Component + .sorted() + .collect { case contributor: HadoopConfContributor => contributor.interestedPrefixes() } + .flatten + .toSet + } + + private def collectCopied( + conf: Configuration, + prefixes: Set[String]): Map[String, String] = { + val entries = conf + .iterator() + .asScala + .map(_.getKey) + .filter(key => matchesPrefix(key, prefixes) && isUserSource(conf, key)) + .map { + key => + val value = + try { + conf.get(key) + } catch { + case _: IllegalStateException => conf.getRaw(key) + } + key -> value + } + .toSeq + // Bare Hadoop keys and their "spark.hadoop." forms normalize to the same key. Emit the bare + // entries last so toMap gives them deterministic precedence regardless of iterator order. + val (prefixedEntries, bareEntries) = entries.partition(_._1.startsWith("spark.")) + (prefixedEntries ++ bareEntries).map { + case (key, value) => normalize(key) -> value + }.toMap + } + + private def matchesPrefix(key: String, prefixes: Set[String]): Boolean = { + prefixes.exists { + prefix => key.startsWith(prefix) || key.startsWith(SparkHadoopPrefix + prefix) + } + } + + private def isUserSource(conf: Configuration, key: String): Boolean = { + val sources = conf.getPropertySources(key) + sources == null || sources.isEmpty || sources.exists(source => !isDefaultSource(source)) + } + + private def isDefaultSource(source: String): Boolean = { + source != null && source.contains("-default.xml") + } + + private def normalize(key: String): String = { + if (key.startsWith("spark.")) key else SparkHadoopPrefix + key + } +} 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..5fc47f55092 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 @@ -322,7 +322,8 @@ case class WholeStageTransformer(child: SparkPlan, materializeInput: Boolean = f leafTransformers: Seq[LeafTransformSupport], wsCtx: WholeStageTransformContext, inputRDDs: ColumnarInputRDDsWrapper, - pipelineTime: SQLMetric): RDD[ColumnarBatch] = { + pipelineTime: SQLMetric, + fsConf: Map[String, String]): RDD[ColumnarBatch] = { // If these are two leaf transformers, they must have same partitions, // otherwise, exchange will be inserted. We should combine the two leaf @@ -380,7 +381,8 @@ case class WholeStageTransformer(child: SparkPlan, materializeInput: Boolean = f wsCtx.substraitContext.registeredAggregationParams ), wsCtx.enableCudf, - wsCtx + wsCtx, + fsConf ) val allInputPartitions = leafTransformers.map(_.getPartitions) @@ -407,10 +409,11 @@ case class WholeStageTransformer(child: SparkPlan, materializeInput: Boolean = f GlutenConfig.get.substraitPlanLogLevel, s"$nodeName generating the substrait plan took: $t ms.")) val inputRDDs = new ColumnarInputRDDsWrapper(columnarInputRDDs) + val fsConf = HadoopConfCollector.collect(session) val leafTransformers = findAllLeafTransformers() if (leafTransformers.nonEmpty) { - generateWholeStageRDD(leafTransformers, wsCtx, inputRDDs, pipelineTime) + generateWholeStageRDD(leafTransformers, wsCtx, inputRDDs, pipelineTime, fsConf) } else { /** @@ -436,7 +439,8 @@ case class WholeStageTransformer(child: SparkPlan, materializeInput: Boolean = f wsCtx.substraitContext.registeredAggregationParams ), materializeInput, - inputRDDs.getPartitionLength + inputRDDs.getPartitionLength, + fsConf ) } } diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/WholeStageZippedPartitionsRDD.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/WholeStageZippedPartitionsRDD.scala index 73068241f7d..f7f41bc0df9 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/execution/WholeStageZippedPartitionsRDD.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/execution/WholeStageZippedPartitionsRDD.scala @@ -16,7 +16,7 @@ */ package org.apache.gluten.execution -import org.apache.gluten.backendsapi.BackendsApiManager +import org.apache.gluten.backendsapi.{BackendsApiManager, IteratorApi} import org.apache.gluten.config.GlutenConfig import org.apache.gluten.extension.ApplyStageInputStatsRule import org.apache.gluten.metrics.{GlutenTimeMetric, IMetrics} @@ -39,7 +39,8 @@ class WholeStageZippedPartitionsRDD( pipelineTime: SQLMetric, updateNativeMetrics: IMetrics => Unit, materializeInput: Boolean, - partitionLength: Int = 0) + partitionLength: Int = 0, + val fsConf: Map[String, String] = Map.empty) extends RDD[ColumnarBatch](sc, rdds.getDependencies) { override def compute(split: Partition, context: TaskContext): Iterator[ColumnarBatch] = { @@ -53,22 +54,34 @@ class WholeStageZippedPartitionsRDD( } val partitions = split.asInstanceOf[ZippedPartitionsPartition].inputColumnarRDDPartitions val inputIterators: Seq[Iterator[ColumnarBatch]] = rdds.getIterators(partitions, context) - BackendsApiManager.getIteratorApiInstance - .genFinalStageIterator( - context, - inputIterators, - sparkConf, - resCtx.root, - pipelineTime, - updateNativeMetrics, - split.index, - materializeInput, - resCtx.enableCudf, - resCtx.supportsValueStreamDynamicFilter - ) + createFinalStageIterator( + BackendsApiManager.getIteratorApiInstance, + context, + inputIterators, + split.index) } } + private[gluten] def createFinalStageIterator( + iteratorApi: IteratorApi, + context: TaskContext, + inputIterators: Seq[Iterator[ColumnarBatch]], + partitionIndex: Int): Iterator[ColumnarBatch] = { + iteratorApi.genFinalStageIterator( + context, + inputIterators, + sparkConf, + resCtx.root, + pipelineTime, + updateNativeMetrics, + partitionIndex, + materializeInput, + resCtx.enableCudf, + resCtx.supportsValueStreamDynamicFilter, + fsConf + ) + } + override def getPartitions: Array[Partition] = { Array.tabulate[Partition](rdds.getPartitionLength) { i => new ZippedPartitionsPartition(i, rdds.getPartitions(i)) diff --git a/gluten-ut/test/src/test/scala/org/apache/gluten/execution/HadoopConfCollectorSuite.scala b/gluten-ut/test/src/test/scala/org/apache/gluten/execution/HadoopConfCollectorSuite.scala new file mode 100644 index 00000000000..224fe487004 --- /dev/null +++ b/gluten-ut/test/src/test/scala/org/apache/gluten/execution/HadoopConfCollectorSuite.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.execution + +import org.apache.spark.SparkFunSuite + +import org.apache.hadoop.conf.Configuration + +class HadoopConfCollectorSuite extends SparkFunSuite { + private val fsPrefixes = Set("fs.") + + test("collect short-circuits without resolving the session when prefixes are empty") { + val collected = HadoopConfCollector.collectFromSessionProvider( + () => throw new IllegalStateException("session provider must not be evaluated"), + Set.empty) + + assert(collected.isEmpty) + } + + test("collect keeps user sources and drops built-in default XML sources") { + val conf = new Configuration(false) + conf.set( + "fs.s3a.default.option", + "built-in-value", + "jar:file:/opt/vendor/cloud.jar!/vendor-cloud-default.xml") + conf.set("fs.s3a.site.option", "site-value", "file:/etc/hadoop/core-site.xml") + conf.set("fs.s3a.programmatic.option", "programmatic-value") + + val collected = HadoopConfCollector.collect(conf, fsPrefixes) + + assert(!collected.contains("spark.hadoop.fs.s3a.default.option")) + assert(collected("spark.hadoop.fs.s3a.site.option") == "site-value") + assert(collected("spark.hadoop.fs.s3a.programmatic.option") == "programmatic-value") + } + + test("collect resolves variables and falls back to raw values for substitution cycles") { + val conf = new Configuration(false) + conf.set("credential.alias", "resolved-value") + conf.set("fs.s3a.resolved.option", "${credential.alias}") + conf.set("fs.s3a.cyclic.option", "${fs.s3a.cyclic.option}") + + val collected = HadoopConfCollector.collect(conf, fsPrefixes) + + assert(collected("spark.hadoop.fs.s3a.resolved.option") == "resolved-value") + assert(collected("spark.hadoop.fs.s3a.cyclic.option") == "${fs.s3a.cyclic.option}") + } + + test("collect normalizes bare keys and gives them precedence over spark.hadoop duplicates") { + val conf = new Configuration(false) + conf.set("fs.s3a.bare.option", "bare-value") + conf.set("spark.hadoop.fs.azure.prefixed.option", "prefixed-value") + // Bare and spark.hadoop forms of the same key must resolve to the bare value. + conf.set("fs.s3a.conflict.option", "bare-wins") + conf.set("spark.hadoop.fs.s3a.conflict.option", "prefixed-loses") + + val collected = HadoopConfCollector.collect(conf, fsPrefixes) + + assert(collected("spark.hadoop.fs.s3a.bare.option") == "bare-value") + assert(collected("spark.hadoop.fs.azure.prefixed.option") == "prefixed-value") + assert(!collected.contains("spark.hadoop.spark.hadoop.fs.azure.prefixed.option")) + assert(collected("spark.hadoop.fs.s3a.conflict.option") == "bare-wins") + } + + test("collect excludes non-filesystem keys and keeps unknown filesystem schemes") { + val conf = new Configuration(false) + conf.set("dfs.blocksize", "1024") + conf.set("fs.oss.endpoint", "oss-endpoint") + + val collected = HadoopConfCollector.collect(conf, fsPrefixes) + + assert(!collected.contains("spark.hadoop.dfs.blocksize")) + assert(collected("spark.hadoop.fs.oss.endpoint") == "oss-endpoint") + } +} diff --git a/gluten-ut/test/src/test/scala/org/apache/gluten/execution/HadoopConfTransportSuite.scala b/gluten-ut/test/src/test/scala/org/apache/gluten/execution/HadoopConfTransportSuite.scala new file mode 100644 index 00000000000..8bceb313617 --- /dev/null +++ b/gluten-ut/test/src/test/scala/org/apache/gluten/execution/HadoopConfTransportSuite.scala @@ -0,0 +1,168 @@ +/* + * 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.gluten.backendsapi.IteratorApi +import org.apache.gluten.metrics.IMetrics +import org.apache.gluten.substrait.plan.{PlanBuilder, PlanNode} +import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat +import org.apache.gluten.substrait.rel.SplitInfo + +import org.apache.spark.{Partition, SparkConf, SparkEnv, SparkFunSuite, TaskContext} +import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics} +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.utils.SparkInputMetricsUtil.InputMetricsWrapper +import org.apache.spark.sql.vectorized.ColumnarBatch + +import scala.reflect.ClassTag + +class HadoopConfTransportSuite extends SparkFunSuite with SharedSparkSession { + + test("whole-stage RDD fsConf survives Spark closure serialization") { + // fsConf must be a serializable field so filesystem credentials reach the executors. + val fsConf = Map( + "spark.hadoop.fs.s3a.access.key" -> "access-key", + "spark.hadoop.fs.azure.account.key.example" -> "account-key") + + assert(roundTrip(newFirstStageRDD(fsConf)).fsConf == fsConf) + assert(roundTrip(newFinalStageRDD(fsConf)).fsConf == fsConf) + } + + test("IteratorApi fsConf overloads delegate to legacy implementations by default") { + // Backends that only implement the pre-fsConf overloads (e.g. ClickHouse) must keep working. + val legacyApi = new LegacyOnlyIteratorApi + val iteratorApi: IteratorApi = legacyApi + val fsConf = Map("spark.hadoop.fs.s3a.access.key" -> "access-key") + + iteratorApi.genFirstStageIterator( + GlutenPartition(0, Array.emptyByteArray), + null, + newPipelineMetric(), + _ => (), + _ => (), + 0, + Seq.empty, + false, + WholeStageTransformContext(PlanBuilder.empty()), + fsConf + ) + iteratorApi.genFinalStageIterator( + null, + Seq.empty, + spark.sparkContext.getConf, + PlanBuilder.empty(), + newPipelineMetric(), + _ => (), + 0, + false, + false, + true, + fsConf) + + assert(legacyApi.firstStageCalled) + assert(legacyApi.finalStageCalled) + } + + private def newFirstStageRDD(fsConf: Map[String, String]): GlutenWholeStageColumnarRDD = { + new GlutenWholeStageColumnarRDD( + spark.sparkContext, + Seq.empty, + new ColumnarInputRDDsWrapper(Seq.empty), + newPipelineMetric(), + _ => (), + _ => (), + false, + null, + fsConf) + } + + private def newFinalStageRDD(fsConf: Map[String, String]): WholeStageZippedPartitionsRDD = { + new WholeStageZippedPartitionsRDD( + spark.sparkContext, + new ColumnarInputRDDsWrapper(Seq.empty), + spark.sparkContext.getConf, + WholeStageTransformContext(PlanBuilder.empty()), + newPipelineMetric(), + _ => (), + false, + 0, + fsConf + ) + } + + private def newPipelineMetric(): SQLMetric = { + SQLMetrics.createTimingMetric(spark.sparkContext, "pipeline time") + } + + private def roundTrip[T: ClassTag](value: T): T = { + val serializer = SparkEnv.get.closureSerializer.newInstance() + serializer.deserialize[T](serializer.serialize(value)) + } + + private class LegacyOnlyIteratorApi extends IteratorApi { + var firstStageCalled: Boolean = false + var finalStageCalled: Boolean = false + + override def genSplitInfo( + partitionIndex: Int, + partition: Seq[Partition], + partitionSchema: StructType, + dataSchema: StructType, + fileFormat: ReadFileFormat, + metadataColumnNames: Seq[String], + properties: Map[String, String]): SplitInfo = { + throw new UnsupportedOperationException() + } + + override def genPartitions( + wsCtx: WholeStageTransformContext, + splitInfos: Seq[Seq[SplitInfo]], + leaves: Seq[LeafTransformSupport]): Seq[BaseGlutenPartition] = Seq.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): Iterator[ColumnarBatch] = { + firstStageCalled = true + Iterator.empty + } + + // scalastyle:off argcount + 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): Iterator[ColumnarBatch] = { + finalStageCalled = true + Iterator.empty + } + // scalastyle:on argcount + } +}