[GLUTEN-10113][VL] Pass hadoop fs related session level configurations to native#12483
[GLUTEN-10113][VL] Pass hadoop fs related session level configurations to native#12483zhouyuan wants to merge 1 commit into
Conversation
|
Run Gluten Clickhouse CI on x86 |
c997243 to
bf6e0dc
Compare
|
Run Gluten Clickhouse CI on x86 |
bf6e0dc to
ef67195
Compare
|
Run Gluten Clickhouse CI on x86 |
ef67195 to
5bffe1c
Compare
|
Run Gluten Clickhouse CI on x86 |
Signed-off-by: Yuan <yuanzhou@apache.org> Revert "[VL] allow to pass hadoop options to native backend" This reverts commit 1a96976. fix Signed-off-by: Yuan <yuanzhou@apache.org> fix format Signed-off-by: Yuan <yuanzhou@apache.org> fix Signed-off-by: Yuan <yuanzhou@apache.org>
5bffe1c to
17216aa
Compare
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
This PR aims to propagate Hadoop filesystem-related Spark session/runtime configurations (notably fs.azure.*, fs.s3a.*, fs.gs.*) from the Spark driver/executors into the native backend (Velox) by threading an FS config map through GlutenWholeStageColumnarRDD into the iterator/JNI creation path, with additional safeguards intended to reduce accidental credential exposure via toString/logging.
Changes:
- Collect FS-related configs in
WholeStageTransformer(Hadoop conf + SQLConf + scan reader options) and pass them intoGlutenWholeStageColumnarRDD. - Extend iterator API to accept an
fsConfmap and include it in VeloxNativePlanEvaluator.create(..., extraConf). - Introduce redaction-focused wrappers/keys (
FsCredentialConfand a hashed runtime cache key) to reduce accidental credential leakage.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| gluten-substrait/src/main/scala/org/apache/gluten/execution/WholeStageTransformer.scala | Collect FS-related configs and pass them into the whole-stage RDD. |
| gluten-substrait/src/main/scala/org/apache/gluten/execution/GlutenWholeStageColumnarRDD.scala | Carry FS config on the RDD and attempt to redact credentials in toString. |
| gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala | Expose DSv1 reader options to support collecting per-scan FS options. |
| gluten-substrait/src/main/scala/org/apache/gluten/execution/BatchScanExecTransformer.scala | Add readerOptions override (currently empty) for DSv2 scan transformers. |
| gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala | Add readerOptions API to scan transformers for propagating per-scan options. |
| gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/IteratorApi.scala | Extend iterator API to accept an FS config map. |
| gluten-arrow/src/main/scala/org/apache/gluten/runtime/Runtimes.scala | Change runtime cache key generation to avoid embedding raw config strings. |
| backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala | Pass FS config into native evaluator creation via extraConf. |
| backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHIteratorApi.scala | Update iterator API signature to accept FS config (unused for CH). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| // 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) | ||
| } |
| // 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) | ||
| } |
| postDriverMetrics() | ||
| } | ||
|
|
||
| override def readerOptions: Map[String, String] = Map.empty |
| 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)" | ||
| } | ||
| } | ||
| } |
What changes are proposed in this pull request?
This patch adds the runtime hadoop related configurations to native.
How was this patch tested?
Was this patch authored or co-authored using generative AI tooling?
Related issue: #10113