diff --git a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala index 2a3f34483ec..9b9b5f51837 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala @@ -533,6 +533,8 @@ object VeloxConfig extends ConfigRegistry { val COLUMNAR_VELOX_FILE_HANDLE_CACHE_ENABLED = buildStaticConf("spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled") + .passToNative() + .passDefault() .doc( "Disables caching if false. File handle cache should be disabled " + "if files are mutable, i.e. file content may change while file path stays the same.") @@ -586,24 +588,47 @@ object VeloxConfig extends ConfigRegistry { val CACHE_PREFETCH_MINPCT = buildStaticConf("spark.gluten.sql.columnar.backend.velox.cachePrefetchMinPct") + .passToNative() .doc("Set prefetch cache min pct for velox file scan") .intConf .createWithDefault(0) val AWS_SDK_LOG_LEVEL = - buildConf("spark.gluten.velox.awsSdkLogLevel") + buildStaticConf("spark.gluten.velox.awsSdkLogLevel") .internal() + .passToNative() + .passDefault() .doc("Log granularity of AWS C++ SDK in velox.") .stringConf .createWithDefault("FATAL") val AWS_S3_RETRY_MODE = - buildConf("spark.gluten.velox.fs.s3a.retry.mode") + buildStaticConf("spark.gluten.velox.fs.s3a.retry.mode") .internal() + .passToNative() + .passDefault() .doc("Retry mode for AWS s3 connection error: legacy, standard and adaptive.") .stringConf .createWithDefault("legacy") + val S3_USE_PROXY_FROM_ENV = + buildStaticConf("spark.gluten.velox.s3UseProxyFromEnv") + .internal() + .passToNative() + .passDefault() + .doc("Whether to use proxy from environment variables for S3 C++ client.") + .booleanConf + .createWithDefault(false) + + val S3_PAYLOAD_SIGNING_POLICY = + buildStaticConf("spark.gluten.velox.s3PayloadSigningPolicy") + .internal() + .passToNative() + .passDefault() + .doc("The S3 payload signing policy: Always, RequestDependent and Never.") + .stringConf + .createWithDefault("Never") + val S3_UPLOAD_PART_ASYNC = buildStaticConf("spark.gluten.velox.s3UploadPartAsync") .doc("If true, S3 multipart upload parts are uploaded asynchronously.") @@ -790,7 +815,8 @@ object VeloxConfig extends ConfigRegistry { .createWithDefault(50) val CUDF_ENABLE_TABLE_SCAN = - buildStaticConf("spark.gluten.sql.columnar.backend.velox.cudf.enableTableScan") + buildConf("spark.gluten.sql.columnar.backend.velox.cudf.enableTableScan") + .passToNative() .doc("Enable cudf table scan") .booleanConf .createWithDefault(false) diff --git a/docs/developers/NativeConfPassing.md b/docs/developers/NativeConfPassing.md new file mode 100644 index 00000000000..44f677d3767 --- /dev/null +++ b/docs/developers/NativeConfPassing.md @@ -0,0 +1,191 @@ +--- +layout: page +title: Passing Configurations from JVM to Native +nav_order: 19 +parent: Developer Overview +--- + +# Passing Configurations from JVM to Native + +This document describes the design of Gluten's JVM-to-native configuration-passing mechanism, +built around `NativeConfRegistry` and the `ConfigBuilder.passToNative()` declaration API. + +## Background and motivation + +Native code (Velox / ClickHouse backends) consumes Spark, Hadoop and Gluten configurations. +Historically, the JVM side decided which keys to deliver via hard-coded string lists inside +`GlutenConfig.getNativeSessionConf` / `getNativeBackendConf`: + +- A `nativeKeys` set of 40+ raw string keys, including backend-specific keys living in common + code (e.g. Velox-only S3 keys in gluten-substrait). +- Two "configs having default values" `Seq`s duplicating each conf's key and default. +- Ad-hoc per-key special cases (byte-unit conversion, upper-casing, etc.) inline in the + selection methods. + +This was hard to maintain: adding a native conf required editing central lists far away from +the conf's definition, defaults were duplicated and could drift, and backend-specific keys +leaked into common code. + +## Core model + +### Two delivery channels, two scopes + +There are exactly two points where the JVM delivers confs to native, matching the two +lifecycle stages of a native backend: + +| `NativeScope` | Delivery point | JVM entry | Native receiver | +|---|---|---|---| +| `BACKEND` | Once, during native backend initialization | `GlutenConfig.getNativeBackendConf` | e.g. Velox `VeloxBackend::init` (`backendConf_`), CH `BackendInitializerUtil` | +| `RUNTIME` | Each time a native runtime instance is created (per task pipeline / native memory manager) | `GlutenConfig.getNativeSessionConf` | e.g. Velox `VeloxRuntime` (`confMap_` / `veloxCfg_`) | + +`NativeScope.ALL` means both. + +### Staticness determines the scope + +Gluten follows Spark's static/non-static SQL conf semantics on the JVM side, and derives the +native scope from it: + +- A **static** conf (`buildStaticConf`) cannot be modified after startup. Passing it once at + backend initialization is therefore lossless: the snapshot taken at init time is the value, + forever. Hence **static confs map to `BACKEND`**. +- A **non-static** conf (`buildConf`) may be changed per session/query. It must be re-delivered + each time a native runtime is created so native always observes the current session value. + Hence **dynamic confs map to `RUNTIME`**. + +Note that this describes the *delivery* semantics, not usability: a `RUNTIME` conf can still be +set at startup (spark-defaults / `SparkConf`) and the session inherits that value; the point is +that its native delivery happens at runtime creation, picking up whatever the session currently +holds. + +Conversely, a conf whose declared staticness disagrees with where native actually consumes it +indicates the declaration is wrong and should be fixed: + +1. Declared static, but native reads it per runtime → it should be non-static. +2. Declared non-static, but native reads it only at backend init and later changes can never + take effect → it should be static. +3. Consumed at both backend init and per runtime → allowed, but must be dual-registered and its + split-brain semantics documented (see "Dual-scope confs" below). + +## Declaration API + +### Gluten confs: builder markers + +For configurations defined through Gluten's `ConfigBuilder`, chain the markers at the conf +definition: + +```scala +val COLUMNAR_MAX_BATCH_SIZE = + buildConf("spark.gluten.sql.columnar.maxBatchSize") // dynamic -> RUNTIME + .passToNative() + .intConf + .createWithDefault(4096) + +val COLUMNAR_VELOX_FILE_HANDLE_CACHE_ENABLED = + buildStaticConf("spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled") + .passToNative() // static -> BACKEND + .passDefault() // native relies on the key being always present + .booleanConf + .createWithDefault(false) +``` + +- `passToNative()`: registers the conf to `NativeConfRegistry` on entry creation. The scope is + derived from staticness as described above; there is no scope argument. +- `passDefault()`: additionally passes the conf's own default value (in parsed form, e.g. a + "64MB" bytes conf is passed as "67108864") even when the user did not set the conf. Use it + when native relies on the key being always present. It requires `passToNative()` and a + defined default value; both constraints are checked eagerly at entry creation. + +### Non-Gluten keys: explicit registration + +Spark SQL / Spark core / Hadoop keys that have no Gluten `ConfigEntry` are registered +explicitly, from the conf object of the module that needs them: + +```scala +NativeConfRegistry.register(SQLConf.ANSI_ENABLED.key, RUNTIME, Some(defaultString)) +NativeConfRegistry.register("spark.sql.orc.compression.codec", BACKEND, Some("snappy")) +NativeConfRegistry.register( + SPARK_SHUFFLE_FILE_BUFFER, + RUNTIME, + transform = Some(v => (JavaUtils.byteStringAs(v, ByteUnit.KiB) * 1024).toString)) +``` + +`register(key, scope, defaultToPass, transform)`: + +- `defaultToPass`: when defined, the value is delivered even if the user did not set the key + (same role as `passDefault()` for builder confs). +- `transform`: normalization applied to user-set values before delivery, e.g. size-string to + bytes, upper-casing. Not applied to `defaultToPass` (which is already in final form). + +Runtime and backend scopes are tracked separately, so the same key may carry different +semantics per scope (e.g. plain filter in one scope, always-passed with default in the other). +Duplicate registration within one scope fails fast. + +### Modularity + +Registrations live in the conf object of the owning module (`GlutenCoreConfig`, +`GlutenConfig`, `VeloxConfig`, ...) and take effect when that object is loaded. A backend's or +connector's registrations therefore never leak into another deployment: Velox-only keys are +declared in backends-velox and simply do not exist when running the ClickHouse backend. + +## Selection at delivery time + +- `getNativeSessionConf` = `NativeConfRegistry.selectRuntimeConf(sessionConf)` + prefix rules + (`spark.gluten.sql.columnar.backend.` for non-static keys, + `spark.gluten.`) + UGI tokens. +- `getNativeBackendConf` = `NativeConfRegistry.selectBackendConf(conf)` + prefix rules + (`spark.gluten.sql.columnar.backend.`, `spark.hadoop.fs.s3a.` / `fs.azure.` / + `fs.gs.`, `spark.gluten.`). + +Prefix rules are pattern-based and intentionally kept: they cover open-ended key families +(e.g. arbitrary `fs.s3a` client options) that cannot be enumerated. The registry covers the +keys native depends on individually — especially those needing defaults or transforms. + +## Dual-scope confs (BACKEND + RUNTIME) + +A small set of dynamic confs is also consumed during native backend initialization. They keep +the builder-derived `RUNTIME` registration and add an explicit `BACKEND` registration. The +backend-init delivery is a one-time snapshot of the startup value; later session changes only +affect the runtime channel. Each conf's split semantics: + +| Conf | Backend-init consumption (frozen at startup) | Runtime consumption (follows session) | +|---|---|---| +| `spark.gluten.sql.debug` | keeps user glog levels | per-task input/plan debug dumps | +| `spark.gluten.sql.columnar.cudf` | one-time GPU environment initialization | per-query CPU/GPU offload decision. Enabling in-session without startup enablement will not initialize the GPU | +| `spark.gluten.memory.task.offHeap.size.in.bytes` | CH external sort/aggregation thresholds | Velox per-task spill memory limit | +| `spark.sql.legacy.statisticalAggregate`, `spark.sql.decimalOperations.allowPrecisionLoss`, `spark.sql.legacy.timeParserPolicy` | expression/aggregate behavior fixed into reused backend structures | per-query expression evaluation | +| `spark.sql.session.timeZone`, `spark.redaction.regex` | registered `ALL` | registered `ALL` | +| `spark.hadoop.fs.s3a.*` connection confs (ssl, path-style, retry attempts, connection maximum, ...) | reused HiveConnector construction (defaults passed when unset) | per-query file system access | + +Two confs are BACKEND-only on the native side but still must stay non-static on the JVM side, +solely because `GlutenAutoAdjustStageResourceProfile` mutates them on `SQLConf` at runtime +when a new resource profile is applied: + +- `spark.gluten.memory.offHeap.size.in.bytes` +- `spark.gluten.numTaskSlotsPerExecutor` (native warns and falls back to 1 when the key is + missing, hence its default is always passed) + +If the resource-profile adjustment ever moves to a dedicated passing mechanism, these two can +be tightened to static confs. + +Backend note: the ClickHouse backend calls `getNativeBackendConf` both at `initNative` and per +kernel pipeline, so for CH the BACKEND scope effectively refreshes per query as well; the +Velox backend consumes BACKEND scope strictly once at init. The scope contract above is +defined by the stricter (Velox) behavior. + +## What was removed + +- `nativeKeys` hard-coded string set (including Velox keys in common code, now declared in + `VeloxConfig`). +- The two default-value `Seq`s and all per-key special cases in `getNativeSessionConf` / + `getNativeBackendConf`. +- `BackendSettingsApi.extraNativeSessionConfKeys` / `extraNativeBackendConfKeys`. +- `GlutenConfigUtil.mapByteConfValue` (superseded by per-entry `transform`). + +The selected key/value results of both methods are intended to be byte-for-byte identical to +the previous implementation. + +## Testing + +`org.apache.gluten.config.NativeConfRegistrySuite` covers scope derivation from staticness, +`passDefault` constraints, per-scope mixed semantics, duplicate-registration rejection, +`defaultToPass` and `transform` behavior. diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md index 965b03802e5..da99344efe8 100644 --- a/docs/velox-configuration.md +++ b/docs/velox-configuration.md @@ -22,7 +22,7 @@ nav_order: 16 | spark.gluten.sql.columnar.backend.velox.cudf.allowCpuFallback | ⚓ Static | true | Allow cuDF to fall back to CPU execution for unsupported operators. | | spark.gluten.sql.columnar.backend.velox.cudf.batchSize | 🔄 Dynamic | 2147483647 | Cudf input batch size after shuffle reader | | spark.gluten.sql.columnar.backend.velox.cudf.concurrentGpuTasks | ⚓ Static | 1 | The number of concurrent GPU tasks to run. | -| spark.gluten.sql.columnar.backend.velox.cudf.enableTableScan | ⚓ Static | false | Enable cudf table scan | +| spark.gluten.sql.columnar.backend.velox.cudf.enableTableScan | 🔄 Dynamic | false | Enable cudf table scan | | spark.gluten.sql.columnar.backend.velox.cudf.enableValidation | ⚓ Static | true | Heuristics you can apply to validate a cuDF/GPU plan and only offload when the entire stage can be fully and profitably executed on GPU | | spark.gluten.sql.columnar.backend.velox.cudf.memoryPercent | ⚓ Static | 50 | The initial percent of GPU memory to allocate for memory resource for one thread. | | spark.gluten.sql.columnar.backend.velox.cudf.memoryResource | ⚓ Static | async | GPU RMM memory resource. | diff --git a/gluten-core/src/main/scala/org/apache/gluten/config/ConfigBuilder.scala b/gluten-core/src/main/scala/org/apache/gluten/config/ConfigBuilder.scala index 2c71f46bad1..e84a837da96 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/config/ConfigBuilder.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/config/ConfigBuilder.scala @@ -36,6 +36,9 @@ private[gluten] case class ConfigBuilder(key: String) { private[config] var _experimental = false private[config] var _alternatives = List.empty[String] private[config] var _onCreate: Option[ConfigEntry[_] => Unit] = None + private[config] var _isStatic = false + private[config] var _passToNative = false + private[config] var _passDefault = false def doc(s: String): ConfigBuilder = { _doc = s @@ -78,6 +81,57 @@ private[gluten] case class ConfigBuilder(key: String) { this } + /** Marks this config as a static (non-session-mutable) conf. Set by `buildStaticConf`. */ + private[config] def markStatic(): ConfigBuilder = { + _isStatic = true + this + } + + /** + * Marks this config to be passed to native side when it is set by user. The config is registered + * to [[NativeConfRegistry]] on entry creation. + * + * The native scope is derived from the conf's staticness: static confs (`buildStaticConf`) are + * passed once during native backend initialization; dynamic confs (`buildConf`) are passed each + * time a native runtime instance is created. For confs that need a different or an additional + * scope, call `NativeConfRegistry.register` directly instead. + */ + def passToNative(): ConfigBuilder = { + _passToNative = true + this + } + + /** + * Marks that this config's default value should also be passed to native side even when the conf + * is not set by user. The default is passed in parsed form, e.g. a "64MB" bytes conf is passed as + * "67108864". Use this when native side relies on the key being always present. Must be used + * together with [[passToNative]], and the entry must define a default value. + */ + def passDefault(): ConfigBuilder = { + _passDefault = true + this + } + + private[config] def registerToNative(entry: ConfigEntry[_]): Unit = { + require( + !_passDefault || _passToNative, + s"Config $key: passDefault() must be used together with passToNative()") + if (!_passToNative) { + return + } + val scope = if (_isStatic) NativeScope.BACKEND else NativeScope.RUNTIME + val defaultToPass = if (_passDefault) { + // Use the parsed default value rather than the raw default string, so that e.g. + // a "64MB" bytes conf is passed to native as "67108864". + val d = entry.defaultValue.map(_.toString) + require(d.isDefined, s"Config $key is marked with passDefault() but has no default value") + d + } else { + None + } + NativeConfRegistry.register(key, scope, defaultToPass) + } + def intConf: TypedConfigBuilder[Int] = { new TypedConfigBuilder(this, toNumber(_, _.toInt, key, "int")) } @@ -118,6 +172,7 @@ private[gluten] case class ConfigBuilder(key: String) { _alternatives, fallback) _onCreate.foreach(_(entry)) + registerToNative(entry) entry } } @@ -207,6 +262,7 @@ private[gluten] class TypedConfigBuilder[T]( converter, stringConverter) parent._onCreate.foreach(_(entry)) + parent.registerToNative(entry) entry } @@ -229,6 +285,7 @@ private[gluten] class TypedConfigBuilder[T]( transformedDefault ) parent._onCreate.foreach(_(entry)) + parent.registerToNative(entry) entry } } @@ -247,6 +304,7 @@ private[gluten] class TypedConfigBuilder[T]( default ) parent._onCreate.foreach(_(entry)) + parent.registerToNative(entry) entry } } diff --git a/gluten-core/src/main/scala/org/apache/gluten/config/ConfigRegistry.scala b/gluten-core/src/main/scala/org/apache/gluten/config/ConfigRegistry.scala index 4ebc29de9e6..5fbc0c1ddf4 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/config/ConfigRegistry.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/config/ConfigRegistry.scala @@ -73,12 +73,14 @@ trait ConfigRegistry { } protected def buildStaticConf(key: String): ConfigBuilder = { - ConfigBuilder(key).onCreate { - entry => - register(entry) - registerToSQLConf(entry, isStatic = true) - ConfigRegistry.registerToAllEntries(entry) - } + ConfigBuilder(key) + .markStatic() + .onCreate { + entry => + register(entry) + registerToSQLConf(entry, isStatic = true) + ConfigRegistry.registerToAllEntries(entry) + } } def get: GlutenCoreConfig diff --git a/gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala b/gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala index eb08396772f..a838f9bbe8a 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala @@ -78,6 +78,9 @@ object GlutenCoreConfig extends ConfigRegistry { val SPARK_ONHEAP_SIZE_KEY = "spark.executor.memory" + // Spark keys without Gluten ConfigEntry that should be passed to native backend. + NativeConfRegistry.register(SPARK_OFFHEAP_ENABLED_KEY, NativeScope.BACKEND) + val GLUTEN_ENABLED = buildConf("spark.gluten.enabled") .doc( @@ -117,8 +120,9 @@ object GlutenCoreConfig extends ConfigRegistry { .createWithDefault(false) val COLUMNAR_OVERHEAD_SIZE_IN_BYTES = - buildConf("spark.gluten.memoryOverhead.size.in.bytes") + buildStaticConf("spark.gluten.memoryOverhead.size.in.bytes") .internal() + .passToNative() .doc( "Must provide default value since non-execution operations " + "(e.g. org.apache.spark.sql.Dataset#summary) doesn't propagate configurations using " + @@ -139,6 +143,7 @@ object GlutenCoreConfig extends ConfigRegistry { val COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES = buildConf("spark.gluten.memory.task.offHeap.size.in.bytes") .internal() + .passToNative() .doc( "Must provide default value since non-execution operations " + "(e.g. org.apache.spark.sql.Dataset#summary) doesn't propagate configurations using " + @@ -208,4 +213,25 @@ object GlutenCoreConfig extends ConfigRegistry { .doubleConf .checkValue(v => v >= 0 && v <= 1, "offheap sizing memory fraction must between [0, 1]") .createWithDefault(0.6) + // Special-case native registrations. The confs below cannot be static even though native + // consumes them during backend initialization: they are initialized on SparkConf at driver + // start, but GlutenAutoAdjustStageResourceProfile mutates them on SQLConf at runtime when a + // new resource profile is applied, so they must stay session-modifiable on JVM side. + // - COLUMNAR_OFFHEAP_SIZE_IN_BYTES: native consumption is backend-init only (ClickHouse + // derives spill thresholds from it), hence a BACKEND-only registration without a builder + // marker. + // - NUM_TASK_SLOTS_PER_EXECUTOR: native consumption is backend-init only (Velox io/spill + // thread sizing) and Velox relies on the key being always present, hence a BACKEND-only + // registration passing the default when unset. + // - COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES: consumed by both sides of native - backend init + // (ClickHouse external sort/aggregation settings) and each task's runtime (Velox spill + // memory limit), so it keeps the builder-derived RUNTIME registration (`passToNative` at + // its definition) plus the BACKEND one here. + NativeConfRegistry.register(COLUMNAR_OFFHEAP_SIZE_IN_BYTES.key, NativeScope.BACKEND) + NativeConfRegistry.register(COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key, NativeScope.BACKEND) + NativeConfRegistry.register( + NUM_TASK_SLOTS_PER_EXECUTOR.key, + NativeScope.BACKEND, + Some(NUM_TASK_SLOTS_PER_EXECUTOR.defaultValueString)) + } diff --git a/gluten-core/src/main/scala/org/apache/gluten/config/NativeConfRegistry.scala b/gluten-core/src/main/scala/org/apache/gluten/config/NativeConfRegistry.scala new file mode 100644 index 00000000000..2ccc3738408 --- /dev/null +++ b/gluten-core/src/main/scala/org/apache/gluten/config/NativeConfRegistry.scala @@ -0,0 +1,158 @@ +/* + * 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 + +import scala.collection.JavaConverters._ + +/** + * The native scope where a conf is passed to, i.e. when the conf is delivered from JVM to native + * side. + * + * - RUNTIME: the conf is dynamic. It is passed to native side each time a native runtime instance + * is created, e.g. per task pipeline / native memory manager. See + * `GlutenConfig.getNativeSessionConf`. + * - BACKEND: the conf is static to native backend. It is passed to native side once during native + * backend initialization. See `GlutenConfig.getNativeBackendConf`. + * - ALL: the conf is passed in both of the above cases. + */ +object NativeScope extends Enumeration { + type NativeScope = Value + val RUNTIME, BACKEND, ALL = Value +} + +/** + * A registration of one conf key that should be passed to native side. + * + * @param key + * the conf key. + * @param defaultToPass + * if it yields a defined value, the default value is passed to native side even when the conf is + * not explicitly set by user. Use this when native side relies on the key being always present. + * If it yields None, the conf is passed only when it's set by user. Re-evaluated on each + * selection rather than snapshotted at registration, since some defaults are dynamic, e.g. + * `spark.sql.session.timeZone` defaults to the current JVM default time zone. + * @param transform + * if defined, applied to the user-set value before passing to native. Useful for value + * normalization, e.g. converting size strings like "64k" to byte numbers, or upper-casing. Not + * applied to `defaultToPass`. + */ +case class NativeConfEntry( + key: String, + defaultToPass: () => Option[String] = () => None, + transform: Option[String => String] = None) + +/** + * A registry for conf keys that should be passed to native side. + * + * This registry is the single conf-passing declaration mechanism on JVM side. Rather than + * maintaining hard-coded key lists in common conf-passing code, each module (core, backend, + * connector, etc.) declares its own native confs on demand: + * + * - For Gluten config entries defined via `ConfigBuilder`, chain `.passToNative()` (and + * optionally `.passDefault()`) in the builder. The native scope is derived from the conf's + * staticness: static confs go to BACKEND, dynamic confs go to RUNTIME. Registration happens + * automatically on entry creation. + * - For non-Gluten keys (e.g. `spark.sql.*`, `spark.hadoop.fs.s3a.*` and other Spark/Hadoop + * native keys that have no Gluten `ConfigEntry`), call `NativeConfRegistry.register` + * explicitly, typically from the owning conf object's initializer so the registration ships + * with the module that needs it. + * + * Registrations are naturally modular: a backend's or connector's registrations only take effect + * when its conf object is loaded, so e.g. Velox-only keys never leak into a ClickHouse deployment. + * + * Runtime and backend scopes are tracked separately, so the same key can be registered with + * different semantics per scope, e.g. filter-only in runtime scope while always-passed with a + * default value in backend scope. + */ +object NativeConfRegistry { + import NativeScope._ + + private val runtimeEntries = + new java.util.concurrent.ConcurrentHashMap[String, NativeConfEntry]().asScala + private val backendEntries = + new java.util.concurrent.ConcurrentHashMap[String, NativeConfEntry]().asScala + + /** + * Register a conf key to be passed to native side. + * + * @param defaultToPass + * if it yields a defined value, the default value is passed even when the conf is not set by + * user. Evaluated lazily on each selection so dynamic defaults stay up to date. + * @param transform + * if defined, applied to the user-set value before passing. + */ + def register( + key: String, + scope: NativeScope, + defaultToPass: => Option[String] = None, + transform: Option[String => String] = None): Unit = { + val entry = NativeConfEntry(key, () => defaultToPass, transform) + scope match { + case RUNTIME => doRegister(runtimeEntries, entry) + case BACKEND => doRegister(backendEntries, entry) + case ALL => + doRegister(runtimeEntries, entry) + doRegister(backendEntries, entry) + } + } + + private def doRegister( + entries: scala.collection.concurrent.Map[String, NativeConfEntry], + entry: NativeConfEntry): Unit = { + val existing = entries.putIfAbsent(entry.key, entry) + require(existing.isEmpty, s"Native conf ${entry.key} already registered!") + } + + def isRuntimeKey(key: String): Boolean = runtimeEntries.contains(key) + + def isBackendKey(key: String): Boolean = backendEntries.contains(key) + + /** + * Select runtime-scoped native confs from the given conf map. Entries with `defaultToPass` + * defined are always included, using the default value when not set in `conf`. + */ + def selectRuntimeConf(conf: scala.collection.Map[String, String]): Map[String, String] = { + select(runtimeEntries, conf) + } + + /** + * Select backend(static)-scoped native confs from the given conf map. Entries with + * `defaultToPass` defined are always included, using the default value when not set in `conf`. + */ + def selectBackendConf(conf: scala.collection.Map[String, String]): Map[String, String] = { + select(backendEntries, conf) + } + + private def select( + entries: scala.collection.concurrent.Map[String, NativeConfEntry], + conf: scala.collection.Map[String, String]): Map[String, String] = { + entries.values.flatMap { + entry => + val value = conf.get(entry.key) match { + case Some(v) => Some(entry.transform.map(_(v)).getOrElse(v)) + case None => entry.defaultToPass() + } + value.map(v => entry.key -> v) + }.toMap + } + + /** Visible for testing. */ + private[config] def unregister(key: String): Unit = { + runtimeEntries.remove(key) + backendEntries.remove(key) + } +} diff --git a/gluten-core/src/main/scala/org/apache/spark/sql/internal/GlutenConfigUtil.scala b/gluten-core/src/main/scala/org/apache/spark/sql/internal/GlutenConfigUtil.scala index 99ead64d893..bb61ef0217e 100644 --- a/gluten-core/src/main/scala/org/apache/spark/sql/internal/GlutenConfigUtil.scala +++ b/gluten-core/src/main/scala/org/apache/spark/sql/internal/GlutenConfigUtil.scala @@ -18,8 +18,6 @@ package org.apache.spark.sql.internal import org.apache.gluten.config._ -import org.apache.spark.network.util.{ByteUnit, JavaUtils} - object GlutenConfigUtil { private def getConfString( configProvider: GlutenConfigProvider, @@ -43,9 +41,4 @@ object GlutenConfigUtil { val parsedConf = glutenConf.map { case (k, v) => (k, getConfString(provider, k, v)) } parsedConf ++ otherConf } - - def mapByteConfValue(conf: Map[String, String], key: String, unit: ByteUnit)( - f: Long => Unit): Unit = { - conf.get(key).foreach(v => f(JavaUtils.byteStringAs(v, unit))) - } } diff --git a/gluten-core/src/test/scala/org/apache/gluten/config/NativeConfRegistrySuite.scala b/gluten-core/src/test/scala/org/apache/gluten/config/NativeConfRegistrySuite.scala new file mode 100644 index 00000000000..d1e10baed9b --- /dev/null +++ b/gluten-core/src/test/scala/org/apache/gluten/config/NativeConfRegistrySuite.scala @@ -0,0 +1,199 @@ +/* + * 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 + +import org.scalatest.funsuite.AnyFunSuite + +class NativeConfRegistrySuite extends AnyFunSuite { + + private def withRegisteredKeys(keys: String*)(f: => Unit): Unit = { + try f + finally keys.foreach(NativeConfRegistry.unregister) + } + + // Other suites in the same JVM may have loaded conf objects (e.g. GlutenCoreConfig) whose + // initializers register real native confs. Restrict select results to the keys under test so + // assertions are not affected by those global registrations. + private def selectRuntime( + conf: Map[String, String], + keys: String*): Map[String, String] = { + NativeConfRegistry.selectRuntimeConf(conf).filter { case (k, _) => keys.contains(k) } + } + + private def selectBackend( + conf: Map[String, String], + keys: String*): Map[String, String] = { + NativeConfRegistry.selectBackendConf(conf).filter { case (k, _) => keys.contains(k) } + } + + test("ConfigBuilder.passToNative derives the native scope from conf staticness") { + val runtimeKey = "spark.gluten.test.native.runtime.conf" + val backendKey = "spark.gluten.test.native.backend.conf" + val jvmOnlyKey = "spark.gluten.test.jvmOnly.conf" + withRegisteredKeys(runtimeKey, backendKey) { + // Dynamic conf -> RUNTIME scope. + ConfigBuilder(runtimeKey) + .passToNative() + .booleanConf + .createWithDefault(true) + // Static conf (as built by `buildStaticConf`) -> BACKEND scope. + ConfigBuilder(backendKey) + .markStatic() + .passToNative() + .intConf + .createWithDefault(1) + ConfigBuilder(jvmOnlyKey).booleanConf.createWithDefault(false) + + assert(NativeConfRegistry.isRuntimeKey(runtimeKey)) + assert(!NativeConfRegistry.isBackendKey(runtimeKey)) + + assert(!NativeConfRegistry.isRuntimeKey(backendKey)) + assert(NativeConfRegistry.isBackendKey(backendKey)) + + assert(!NativeConfRegistry.isRuntimeKey(jvmOnlyKey)) + assert(!NativeConfRegistry.isBackendKey(jvmOnlyKey)) + } + } + + test("ConfigBuilder.passDefault registers the entry default value") { + val key = "spark.gluten.test.native.withEntryDefault.conf" + val bytesKey = "spark.gluten.test.native.withBytesDefault.conf" + withRegisteredKeys(key, bytesKey) { + ConfigBuilder(key) + .passToNative() + .passDefault() + .booleanConf + .createWithDefault(true) + // Bytes conf default is passed as the parsed value in bytes, not the raw string. + ConfigBuilder(bytesKey) + .passToNative() + .passDefault() + .bytesConf(org.apache.spark.network.util.ByteUnit.BYTE) + .createWithDefaultString("1KB") + + val selected = selectRuntime(Map.empty[String, String], key, bytesKey) + assert(selected === Map(key -> "true", bytesKey -> "1024")) + } + } + + test("passDefault on optional conf without default value is rejected") { + val key = "spark.gluten.test.native.optionalNoDefault.conf" + withRegisteredKeys(key) { + assertThrows[IllegalArgumentException] { + ConfigBuilder(key) + .passToNative() + .passDefault() + .stringConf + .createOptional + } + } + } + + test("passDefault without passToNative is rejected") { + val key = "spark.gluten.test.native.defaultWithoutPass.conf" + withRegisteredKeys(key) { + assertThrows[IllegalArgumentException] { + ConfigBuilder(key).passDefault().booleanConf.createWithDefault(true) + } + } + } + + test("same key can be registered with different semantics per scope") { + val key = "spark.gluten.test.native.mixedScopes.conf" + withRegisteredKeys(key) { + NativeConfRegistry.register(key, NativeScope.RUNTIME) + NativeConfRegistry.register(key, NativeScope.BACKEND, Some("false")) + + // Runtime: passed only when set. + assert(selectRuntime(Map.empty[String, String], key).isEmpty) + assert(selectRuntime(Map(key -> "true"), key) === Map(key -> "true")) + // Backend: always passed with default. + assert(selectBackend(Map.empty[String, String], key) === Map(key -> "false")) + } + } + + test("duplicate registration in the same scope is not allowed") { + val key = "spark.gluten.test.native.duplicated.conf" + withRegisteredKeys(key) { + NativeConfRegistry.register(key, NativeScope.RUNTIME) + assertThrows[IllegalArgumentException] { + NativeConfRegistry.register(key, NativeScope.RUNTIME) + } + // Different scope is fine. + NativeConfRegistry.register(key, NativeScope.BACKEND) + } + } + + test("selectRuntimeConf/selectBackendConf pick up user-set confs by scope") { + val runtimeKey = "spark.gluten.test.select.runtime.conf" + val backendKey = "spark.gluten.test.select.backend.conf" + val allKey = "spark.gluten.test.select.all.conf" + withRegisteredKeys(runtimeKey, backendKey, allKey) { + NativeConfRegistry.register(runtimeKey, NativeScope.RUNTIME) + NativeConfRegistry.register(backendKey, NativeScope.BACKEND) + NativeConfRegistry.register(allKey, NativeScope.ALL) + + val unregisteredKey = "spark.gluten.test.select.unregistered.conf" + val conf = Map( + runtimeKey -> "v1", + backendKey -> "v2", + allKey -> "v3", + unregisteredKey -> "v4") + + val testKeys = Seq(runtimeKey, backendKey, allKey, unregisteredKey) + assert( + selectRuntime(conf, testKeys: _*) === Map(runtimeKey -> "v1", allKey -> "v3")) + assert( + selectBackend(conf, testKeys: _*) === Map(backendKey -> "v2", allKey -> "v3")) + } + } + + test("confs not set by user are skipped unless defaultToPass is defined") { + val noDefaultKey = "spark.gluten.test.select.noDefault.conf" + val withDefaultKey = "spark.gluten.test.select.withDefault.conf" + withRegisteredKeys(noDefaultKey, withDefaultKey) { + NativeConfRegistry.register(noDefaultKey, NativeScope.RUNTIME) + NativeConfRegistry.register( + withDefaultKey, + NativeScope.RUNTIME, + defaultToPass = Some("defaultValue")) + + // Neither conf is set by user. + val selected = selectRuntime(Map.empty[String, String], noDefaultKey, withDefaultKey) + assert(selected === Map(withDefaultKey -> "defaultValue")) + + // User-set value takes precedence over defaultToPass. + val selected2 = + selectRuntime(Map(withDefaultKey -> "userValue"), noDefaultKey, withDefaultKey) + assert(selected2 === Map(withDefaultKey -> "userValue")) + } + } + + test("transform is applied to user-set values but not to defaultToPass") { + val key = "spark.gluten.test.select.transform.conf" + withRegisteredKeys(key) { + NativeConfRegistry.register( + key, + NativeScope.RUNTIME, + defaultToPass = Some("raw-default"), + transform = Some(_.toUpperCase(java.util.Locale.ROOT))) + + assert(selectRuntime(Map(key -> "legacy"), key) === Map(key -> "LEGACY")) + assert(selectRuntime(Map.empty[String, String], key) === Map(key -> "raw-default")) + } + } +} diff --git a/gluten-core/src/test/scala/org/apache/spark/sql/internal/GlutenConfigUtilSuite.scala b/gluten-core/src/test/scala/org/apache/spark/sql/internal/GlutenConfigUtilSuite.scala deleted file mode 100644 index f3ddabcd477..00000000000 --- a/gluten-core/src/test/scala/org/apache/spark/sql/internal/GlutenConfigUtilSuite.scala +++ /dev/null @@ -1,35 +0,0 @@ -/* - * 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.internal - -import org.apache.spark.network.util.ByteUnit - -import org.scalatest.funsuite.AnyFunSuite - -class GlutenConfigUtilSuite extends AnyFunSuite { - - test("mapByteConfValue should return correct value") { - val conf = Map( - "spark.unsafe.sorter.spill.reader.buffer.size" -> "2m" - ) - - GlutenConfigUtil.mapByteConfValue( - conf, - "spark.unsafe.sorter.spill.reader.buffer.size", - ByteUnit.BYTE)(v => assert(2097152L.equals(v))) - } -} diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala index 85b146e1956..75c94be4d4c 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala @@ -168,16 +168,4 @@ trait BackendSettingsApi { /** Whether the backend supports columnar shuffle with empty schema. */ def supportEmptySchemaColumnarShuffle(): Boolean = true - - /** - * Backend-specific non-prefixed session configs that should be forwarded to native runtime / - * memory manager creation. - */ - def extraNativeSessionConfKeys(): Set[String] = Set.empty - - /** - * Backend-specific non-prefixed backend initialization configs that should be forwarded during - * native backend initialization. - */ - def extraNativeBackendConfKeys(): Set[String] = Set.empty } diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala index 93fb3888e99..69400fc4223 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala @@ -16,11 +16,10 @@ */ package org.apache.gluten.config -import org.apache.gluten.backendsapi.BackendsApiManager import org.apache.gluten.shuffle.SupportsColumnarShuffle import org.apache.spark.network.util.{ByteUnit, JavaUtils} -import org.apache.spark.sql.internal.{GlutenConfigUtil, SQLConf} +import org.apache.spark.sql.internal.SQLConf import org.apache.hadoop.security.UserGroupInformation @@ -480,109 +479,133 @@ object GlutenConfig extends ConfigRegistry { def prefixOf(backendName: String): String = s"spark.gluten.sql.columnar.backend.$backendName" def prefixSessionOf(backendName: String): String = s"spark.gluten.$backendName" - private lazy val nativeKeys = Set( - DEBUG_ENABLED.key, - BENCHMARK_SAVE_DIR.key, - GlutenCoreConfig.COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key, - COLUMNAR_MAX_BATCH_SIZE.key, - SHUFFLE_WRITER_BUFFER_SIZE.key, - COLUMNAR_CUDF_ENABLED.key, - SQLConf.LEGACY_SIZE_OF_NULL.key, - SQLConf.LEGACY_STATISTICAL_AGGREGATE.key, - SQLConf.JSON_GENERATOR_IGNORE_NULL_FIELDS.key, - SQLConf.RUNTIME_BLOOM_FILTER_EXPECTED_NUM_ITEMS.key, - SQLConf.RUNTIME_BLOOM_FILTER_NUM_BITS.key, - SQLConf.RUNTIME_BLOOM_FILTER_MAX_NUM_BITS.key, - SQLConf.RUNTIME_BLOOM_FILTER_MAX_NUM_ITEMS.key, - "spark.io.compression.codec", - "spark.sql.decimalOperations.allowPrecisionLoss", - // s3 config - SPARK_S3_ACCESS_KEY, - SPARK_S3_SECRET_KEY, - SPARK_S3_ENDPOINT, - SPARK_S3_CONNECTION_SSL_ENABLED, - SPARK_S3_PATH_STYLE_ACCESS, - SPARK_S3_USE_INSTANCE_CREDENTIALS, - SPARK_S3_IAM, - SPARK_S3_IAM_SESSION_NAME, - SPARK_S3_RETRY_MAX_ATTEMPTS, - SPARK_S3_CONNECTION_MAXIMUM, - SPARK_S3_ENDPOINT_REGION, - SPARK_S3_AWS_IMDS_ENABLED, - "spark.gluten.velox.fs.s3a.retry.mode", - "spark.gluten.velox.awsSdkLogLevel", - "spark.gluten.velox.s3UseProxyFromEnv", - "spark.gluten.velox.s3PayloadSigningPolicy", - "spark.gluten.velox.s3LogLocation", - // gcs config - SPARK_GCS_STORAGE_ROOT_URL, - SPARK_GCS_AUTH_TYPE, - SPARK_GCS_AUTH_SERVICE_ACCOUNT_JSON_KEYFILE, - SPARK_REDACTION_REGEX, - "spark.gluten.sql.columnar.backend.velox.queryTraceEnabled", - "spark.gluten.sql.columnar.backend.velox.queryTraceDir", - "spark.gluten.sql.columnar.backend.velox.queryTraceNodeIds", - "spark.gluten.sql.columnar.backend.velox.queryTraceMaxBytes", - "spark.gluten.sql.columnar.backend.velox.queryTraceTaskRegExp", - "spark.gluten.sql.columnar.backend.velox.opTraceDirectoryCreateConfig", - "spark.gluten.sql.columnar.backend.velox.enableUserExceptionStacktrace", - "spark.gluten.sql.columnar.backend.velox.enableSystemExceptionStacktrace", - "spark.gluten.sql.columnar.backend.velox.memoryUseHugePages", - "spark.gluten.sql.columnar.backend.velox.cachePrefetchMinPct", - "spark.gluten.sql.columnar.backend.velox.memoryPoolCapacityTransferAcrossTasks", - "spark.gluten.sql.columnar.backend.velox.preferredBatchBytes", - "spark.gluten.sql.columnar.backend.velox.cudf.enableTableScan", - "spark.gluten.sql.columnar.backend.velox.columnarBatchSerializerCompression" - ) - - private def backendSettings(backendName: String) = { - // Only one backend is loaded in a running Gluten session. Use backend settings hooks to avoid - // hard-coding backend-specific configs in common code. - BackendsApiManager.getSettings - } + // Registrations of non-Gluten configurations (Spark SQL / Spark core / Hadoop keys that have + // no Gluten ConfigEntry) to be passed to native side. Gluten's own configurations declare + // native passing via `ConfigBuilder.passToNative` at their definitions instead. + private def registerNativeConfs(): Unit = { + import NativeScope._ - /** Get dynamic configs. */ - def getNativeSessionConf(backendName: String, conf: Map[String, String]): Map[String, String] = { - val settings = backendSettings(backendName) - val nativeConfMap = mutable.Map[String, String](conf.filter { - case (key, _) => - nativeKeys.contains(key) || settings.extraNativeSessionConfKeys().contains(key) - }.toSeq: _*) + // Reference GlutenCoreConfig to force its object initialization, so that its own + // `passToNative` registrations are in place before native confs are selected. + require(GlutenCoreConfig.SPARK_OFFHEAP_ENABLED_KEY.nonEmpty) + // Dynamic Spark SQL confs consumed by native. Passed when set by user. Seq( - (SQLConf.CASE_SENSITIVE.key, SQLConf.CASE_SENSITIVE.defaultValueString), - (SQLConf.IGNORE_MISSING_FILES.key, SQLConf.IGNORE_MISSING_FILES.defaultValueString), - ( - SQLConf.LEGACY_STATISTICAL_AGGREGATE.key, - SQLConf.LEGACY_STATISTICAL_AGGREGATE.defaultValueString), - ( - COLUMNAR_MEMORY_BACKTRACE_ALLOCATION.key, - COLUMNAR_MEMORY_BACKTRACE_ALLOCATION.defaultValueString), - ( - GLUTEN_COLUMNAR_TO_ROW_MEM_THRESHOLD.key, - GLUTEN_COLUMNAR_TO_ROW_MEM_THRESHOLD.defaultValue.get.toString), - (SPARK_SHUFFLE_SPILL_COMPRESS, SPARK_SHUFFLE_SPILL_COMPRESS_DEFAULT.toString), - (SQLConf.MAP_KEY_DEDUP_POLICY.key, SQLConf.MAP_KEY_DEDUP_POLICY.defaultValueString), - (SQLConf.SESSION_LOCAL_TIMEZONE.key, SQLConf.SESSION_LOCAL_TIMEZONE.defaultValueString), - (SQLConf.ANSI_ENABLED.key, SQLConf.ANSI_ENABLED.defaultValueString) - ).foreach { case (k, defaultValue) => nativeConfMap.put(k, conf.getOrElse(k, defaultValue)) } + SQLConf.LEGACY_SIZE_OF_NULL.key, + SQLConf.JSON_GENERATOR_IGNORE_NULL_FIELDS.key, + SQLConf.RUNTIME_BLOOM_FILTER_EXPECTED_NUM_ITEMS.key, + SQLConf.RUNTIME_BLOOM_FILTER_NUM_BITS.key, + SQLConf.RUNTIME_BLOOM_FILTER_MAX_NUM_BITS.key, + SQLConf.RUNTIME_BLOOM_FILTER_MAX_NUM_ITEMS.key, + "spark.io.compression.codec" + ).foreach(NativeConfRegistry.register(_, RUNTIME)) + + // Dynamic Spark SQL confs that native relies on being always present. The Spark default + // value is passed when not set by user. + Seq( + SQLConf.CASE_SENSITIVE, + SQLConf.IGNORE_MISSING_FILES, + SQLConf.MAP_KEY_DEDUP_POLICY, + SQLConf.ANSI_ENABLED + ).foreach(e => NativeConfRegistry.register(e.key, RUNTIME, Some(e.defaultValueString))) + NativeConfRegistry.register( + SQLConf.SESSION_LOCAL_TIMEZONE.key, + ALL, + Some(SQLConf.SESSION_LOCAL_TIMEZONE.defaultValueString)) + // Also passed to native backend(static) conf as a plain filter. + NativeConfRegistry.register( + SQLConf.LEGACY_STATISTICAL_AGGREGATE.key, + RUNTIME, + Some(SQLConf.LEGACY_STATISTICAL_AGGREGATE.defaultValueString)) + NativeConfRegistry.register(SQLConf.LEGACY_STATISTICAL_AGGREGATE.key, BACKEND) + NativeConfRegistry.register(SQLConf.DECIMAL_OPERATIONS_ALLOW_PREC_LOSS.key, RUNTIME) + NativeConfRegistry.register( + SQLConf.DECIMAL_OPERATIONS_ALLOW_PREC_LOSS.key, + BACKEND, + Some(SQLConf.DECIMAL_OPERATIONS_ALLOW_PREC_LOSS.defaultValueString)) + // Native side expects the upper-cased value in session scope. + NativeConfRegistry.register( + SQLConf.LEGACY_TIME_PARSER_POLICY.key, + RUNTIME, + transform = Some(_.toUpperCase(Locale.ROOT))) + NativeConfRegistry.register(SQLConf.LEGACY_TIME_PARSER_POLICY.key, BACKEND) + + // Spark core confs. Size strings (e.g. "64k") are normalized to numbers in bytes for native. + NativeConfRegistry.register( + SPARK_SHUFFLE_SPILL_COMPRESS, + RUNTIME, + Some(SPARK_SHUFFLE_SPILL_COMPRESS_DEFAULT.toString)) + NativeConfRegistry.register(SPARK_REDACTION_REGEX, ALL) + Seq(SPARK_UNSAFE_SORTER_SPILL_READER_BUFFER_SIZE, SPARK_SHUFFLE_SPILL_DISK_WRITE_BUFFER_SIZE) + .foreach( + NativeConfRegistry.register( + _, + RUNTIME, + transform = Some(v => JavaUtils.byteStringAs(v, ByteUnit.BYTE).toString))) + NativeConfRegistry.register( + SPARK_SHUFFLE_FILE_BUFFER, + RUNTIME, + transform = Some(v => (JavaUtils.byteStringAs(v, ByteUnit.KiB) * 1024).toString)) + + // Object store (S3/GCS) confs. Passed to session conf when set by user. In backend(static) + // scope they are covered by the `spark.hadoop.fs.s3a.`/`spark.hadoop.fs.gs.` prefix rules, + // except the ones below that native relies on being always present. + Seq( + SPARK_S3_ACCESS_KEY, + SPARK_S3_SECRET_KEY, + SPARK_S3_ENDPOINT, + SPARK_S3_IAM, + SPARK_S3_IAM_SESSION_NAME, + SPARK_S3_ENDPOINT_REGION, + SPARK_S3_AWS_IMDS_ENABLED, + SPARK_GCS_STORAGE_ROOT_URL, + SPARK_GCS_AUTH_TYPE, + SPARK_GCS_AUTH_SERVICE_ACCOUNT_JSON_KEYFILE + ).foreach(NativeConfRegistry.register(_, RUNTIME)) + Seq( + (SPARK_S3_CONNECTION_SSL_ENABLED, "false"), + (SPARK_S3_PATH_STYLE_ACCESS, "true"), + (SPARK_S3_USE_INSTANCE_CREDENTIALS, "false"), + (SPARK_S3_RETRY_MAX_ATTEMPTS, "20"), + (SPARK_S3_CONNECTION_MAXIMUM, "15") + ).foreach { + case (key, default) => + NativeConfRegistry.register(key, RUNTIME) + NativeConfRegistry.register(key, BACKEND, Some(default)) + } + // Shuffle codec confs (COLUMNAR_SHUFFLE_CODEC(_BACKEND)) are optional entries with no + // default; native backend init relies on the keys being always present, so pass an empty + // string when unset. Key literals are used since the entry vals are defined after this + // method runs during object initialization. + NativeConfRegistry.register("spark.gluten.sql.columnar.shuffle.codec", BACKEND, Some("")) + NativeConfRegistry.register( + "spark.gluten.sql.columnar.shuffle.codecBackend", + BACKEND, + Some("")) + + // Datasource confs consumed by native backend(static) initialization. + NativeConfRegistry.register(SPARK_SQL_PARQUET_COMPRESSION_CODEC, BACKEND) + NativeConfRegistry.register(SQLConf.PARQUET_WRITE_LEGACY_FORMAT.key, BACKEND) + NativeConfRegistry.register("spark.sql.orc.compression.codec", BACKEND, Some("snappy")) + + // Hadoop client confs consumed by native backend(static) initialization. Seq( - (SPARK_UNSAFE_SORTER_SPILL_READER_BUFFER_SIZE, ByteUnit.BYTE, (v: Long) => v.toString), - (SPARK_SHUFFLE_SPILL_DISK_WRITE_BUFFER_SIZE, ByteUnit.BYTE, (v: Long) => v.toString), - (SPARK_SHUFFLE_FILE_BUFFER, ByteUnit.KiB, (v: Long) => (v * 1024).toString) - ) - .foreach { - case (k, unit, f) => - GlutenConfigUtil.mapByteConfValue(conf, k, unit)(v => nativeConfMap.put(k, f(v))) - } + "spark.hadoop.input.connect.timeout", + "spark.hadoop.input.read.timeout", + "spark.hadoop.input.write.timeout" + ).foreach(NativeConfRegistry.register(_, BACKEND, Some("180000"))) + NativeConfRegistry.register("spark.hadoop.dfs.client.log.severity", BACKEND, Some("INFO")) + } + registerNativeConfs() - conf - .get(SQLConf.LEGACY_TIME_PARSER_POLICY.key) - .foreach( - v => - nativeConfMap - .put(SQLConf.LEGACY_TIME_PARSER_POLICY.key, v.toUpperCase(Locale.ROOT))) + /** Get dynamic configs. */ + def getNativeSessionConf(backendName: String, conf: Map[String, String]): Map[String, String] = { + val nativeConfMap = mutable.Map[String, String]() + + // Confs declared to be passed to native session, either by `ConfigBuilder.passToNative` at + // conf definitions or by `NativeConfRegistry.register`. + nativeConfMap ++= NativeConfRegistry.selectRuntimeConf(conf) val confPrefixSession = prefixSessionOf(backendName) val confPrefix = prefixOf(backendName) @@ -633,54 +656,9 @@ object GlutenConfig extends ConfigRegistry { val nativeConfMap = mutable.HashMap.empty[String, String] - // some configs having default values - Seq( - (SPARK_S3_CONNECTION_SSL_ENABLED, "false"), - (SPARK_S3_PATH_STYLE_ACCESS, "true"), - (SPARK_S3_USE_INSTANCE_CREDENTIALS, "false"), - (SPARK_S3_RETRY_MAX_ATTEMPTS, "20"), - (SPARK_S3_CONNECTION_MAXIMUM, "15"), - ("spark.gluten.velox.fs.s3a.retry.mode", "legacy"), - ( - GlutenCoreConfig.NUM_TASK_SLOTS_PER_EXECUTOR.key, - GlutenCoreConfig.NUM_TASK_SLOTS_PER_EXECUTOR.defaultValueString), - (COLUMNAR_SHUFFLE_CODEC.key, ""), - (COLUMNAR_SHUFFLE_CODEC_BACKEND.key, ""), - (DEBUG_CUDF.key, DEBUG_CUDF.defaultValueString), - ("spark.hadoop.input.connect.timeout", "180000"), - ("spark.hadoop.input.read.timeout", "180000"), - ("spark.hadoop.input.write.timeout", "180000"), - ("spark.hadoop.dfs.client.log.severity", "INFO"), - ("spark.sql.orc.compression.codec", "snappy"), - ("spark.sql.decimalOperations.allowPrecisionLoss", "true"), - ("spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled", "false"), - ("spark.gluten.velox.awsSdkLogLevel", "FATAL"), - ("spark.gluten.velox.s3UseProxyFromEnv", "false"), - ("spark.gluten.velox.s3PayloadSigningPolicy", "Never"), - (SQLConf.SESSION_LOCAL_TIMEZONE.key, SQLConf.SESSION_LOCAL_TIMEZONE.defaultValueString) - ).foreach { case (k, defaultValue) => nativeConfMap.put(k, conf.getOrElse(k, defaultValue)) } - - val settings = backendSettings(backendName) - val keys = Set( - DEBUG_ENABLED.key, - // datasource config - SPARK_SQL_PARQUET_COMPRESSION_CODEC, - SQLConf.PARQUET_WRITE_LEGACY_FORMAT.key, - // datasource config end - GlutenCoreConfig.COLUMNAR_OVERHEAD_SIZE_IN_BYTES.key, - GlutenCoreConfig.COLUMNAR_OFFHEAP_SIZE_IN_BYTES.key, - GlutenCoreConfig.COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key, - GlutenCoreConfig.SPARK_OFFHEAP_ENABLED_KEY, - SQLConf.DECIMAL_OPERATIONS_ALLOW_PREC_LOSS.key, - SPARK_REDACTION_REGEX, - SQLConf.LEGACY_TIME_PARSER_POLICY.key, - SQLConf.LEGACY_STATISTICAL_AGGREGATE.key, - COLUMNAR_CUDF_ENABLED.key - ) - - nativeConfMap ++= conf.filter { - case (k, _) => keys.contains(k) || settings.extraNativeBackendConfKeys().contains(k) - } + // Confs declared to be passed to native backend, either by `ConfigBuilder.passToNative` at + // conf definitions or by `NativeConfRegistry.register`. + nativeConfMap ++= NativeConfRegistry.selectBackendConf(conf) val confPrefix = prefixOf(backendName) val s3Prefix = HADOOP_PREFIX + S3A_PREFIX @@ -1129,7 +1107,8 @@ object GlutenConfig extends ConfigRegistry { .createOptional val COLUMNAR_SHUFFLE_CODEC_BACKEND = - buildConf("spark.gluten.sql.columnar.shuffle.codecBackend").stringConf + buildConf("spark.gluten.sql.columnar.shuffle.codecBackend") + .stringConf .transform(_.toLowerCase(Locale.ROOT)) .createOptional @@ -1175,18 +1154,24 @@ object GlutenConfig extends ConfigRegistry { .createWithDefault(false) val COLUMNAR_MAX_BATCH_SIZE = - buildConf("spark.gluten.sql.columnar.maxBatchSize").intConf + buildConf("spark.gluten.sql.columnar.maxBatchSize") + .passToNative() + .intConf .checkValue(_ > 0, s"must be positive.") .createWithDefault(4096) val GLUTEN_COLUMNAR_TO_ROW_MEM_THRESHOLD = buildConf("spark.gluten.sql.columnarToRowMemoryThreshold") + .passToNative() + .passDefault() .bytesConf(ByteUnit.BYTE) .createWithDefaultString("64MB") // if not set, use COLUMNAR_MAX_BATCH_SIZE instead val SHUFFLE_WRITER_BUFFER_SIZE = - buildConf("spark.gluten.shuffleWriter.bufferSize").intConf + buildConf("spark.gluten.shuffleWriter.bufferSize") + .passToNative() + .intConf .checkValue(_ > 0, s"must be positive.") .createOptional @@ -1262,6 +1247,8 @@ object GlutenConfig extends ConfigRegistry { val COLUMNAR_MEMORY_BACKTRACE_ALLOCATION = buildConf("spark.gluten.memory.backtrace.allocation") .internal() + .passToNative() + .passDefault() .doc("Print backtrace information for large memory allocations. This helps debugging when " + "Spark OOM happens due to large acquire requests.") .booleanConf @@ -1332,6 +1319,7 @@ object GlutenConfig extends ConfigRegistry { val DEBUG_ENABLED = buildConf("spark.gluten.sql.debug") .internal() + .passToNative() .booleanConf .createWithDefault(false) @@ -1360,6 +1348,8 @@ object GlutenConfig extends ConfigRegistry { val DEBUG_CUDF = buildStaticConf("spark.gluten.sql.debug.cudf") .internal() + .passToNative() + .passDefault() .booleanConf .createWithDefault(false) @@ -1390,6 +1380,7 @@ object GlutenConfig extends ConfigRegistry { val BENCHMARK_SAVE_DIR = buildConf("spark.gluten.saveDir") .internal() + .passToNative() .stringConf .createWithDefault("") @@ -1690,6 +1681,7 @@ object GlutenConfig extends ConfigRegistry { val COLUMNAR_CUDF_ENABLED = buildConf("spark.gluten.sql.columnar.cudf") .experimental() + .passToNative() .doc("Enable or disable cudf support. This is an experimental feature.") .booleanConf .createWithDefault(false) @@ -1709,4 +1701,15 @@ object GlutenConfig extends ConfigRegistry { "total size of small files is below this threshold.") .doubleConf .createWithDefault(0.5) + // Confs below are consumed by both native backend initialization and native session, so + // they keep the builder-derived RUNTIME registration and add an explicit BACKEND one here. + // - DEBUG_ENABLED: backend init uses it to keep user-provided glog levels; each native + // runtime also reads it to dump task input/plan. The value read at backend init is + // frozen for logging; session-level changes only affect per-query debug dumps. + // - COLUMNAR_CUDF_ENABLED: backend init uses it to initialize the GPU environment (a + // one-time decision); each native runtime reads it to decide CPU/GPU offload. Enabling + // it in a session without having it enabled at startup will not initialize the GPU. + NativeConfRegistry.register(DEBUG_ENABLED.key, NativeScope.BACKEND) + NativeConfRegistry.register(COLUMNAR_CUDF_ENABLED.key, NativeScope.BACKEND) + }