jackylee-ch commented on code in PR #12549: URL: https://github.com/apache/gluten/pull/12549#discussion_r3953988829
########## gluten-core/src/main/scala/org/apache/gluten/config/NativeConfRegistry.scala: ########## @@ -0,0 +1,200 @@ +/* + * 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.apache.spark.internal.Logging + +import scala.collection.JavaConverters._ + +/** + * A registration of one conf key that should be passed to native side. + * + * @param key + * the conf key. + * @param convert + * normalizes a value for native side through the conf's own converter, i.e. the one chosen by + * `stringConf` / `bytesConf(unit)` / `intConf` / ... plus any `transform`. Both a user-set value + * and a resolved default flow through it, so a size conf reaches native as a number rather than + * as "64k", and a `timeParserPolicy` conf reaches native upper-cased even when the user wrote it + * in lower case. + * @param declaredDefault + * the default to deliver when the user did not set the key, in its already-converted native form, + * or `None` to deliver nothing (native's own fallback takes over). Evaluated per delivery rather + * than snapshotted here, since a default may be dynamic - `spark.sql.session.timeZone` follows + * the JVM default time zone, and `spark.sql.ansi.enabled` follows Spark's own default which + * differs between 3.x and 4.x. Which path applies is determined at the declaration site: + * + * - `createOptional`: `None` - nothing is delivered when unset. + * - `createWithDefault(value)`: the stated value in converted form. + * - `createWithDefaultFunction(f)`: `f`'s current result in converted form. + */ +case class NativeConfEntry( + key: String, + convert: String => String = identity, + declaredDefault: () => Option[String] = () => None) + +/** + * A registry for conf keys that should be passed to native side. + * + * This registry is the single conf-passing mechanism on JVM side, but it is not the API to use it. + * Rather than maintaining hard-coded key lists in common conf-passing code, each module (core, + * backend, connector, etc.) declares its own native confs at their definition, through + * `ConfigRegistry`'s four builders plus `ConfigBuilder.passToNative`: + * + * - `buildConf` / `buildStaticConf` for Gluten's own configurations; + * - `registerConf` / `registerStaticConf` for Spark / Hadoop keys that have no Gluten + * `ConfigEntry`, which declare the native delivery only. + * + * Registration happens automatically on entry creation, hence [[register]] is meant for + * `ConfigBuilder` rather than for conf objects. + * + * 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. + * + * There are two delivery channels, matching the two lifecycle stages of a native backend, and which + * ones a conf lands on follows its mutability rather than any argument: + * + * - backend: delivered once during native backend initialization. See + * `GlutenConfig.getNativeBackendConf`. A static conf (`buildStaticConf` / `registerStaticConf`) + * goes here only, since a snapshot taken at init is its value forever. + * - runtime: delivered each time a native runtime instance is created, e.g. per task pipeline / + * native memory manager. See `GlutenConfig.getNativeSessionConf`. A modifiable conf + * (`buildConf` / `registerConf`) goes here *and* to the backend channel, so native observes the + * current value wherever it reads the key. + * + * When the user did not set a registered key, what is delivered is stated at the declaration site + * (see `ConfigBuilder.passToNative`): `createOptional` delivers nothing and leaves native's own + * fallback in charge; `createWithDefault(value)` delivers the stated value; and + * `createWithDefaultFunction(f)` delivers `f`'s result, re-evaluated on every delivery. Delivery is + * always normalized through the conf's own value converter, so all per-key parsing lives at the + * declaration site rather than in per-key transforms at delivery. + */ + +object NativeConfRegistry extends Logging { + + private val runtimeEntries = + new java.util.concurrent.ConcurrentHashMap[String, NativeConfEntry]().asScala + private val backendEntries = + new java.util.concurrent.ConcurrentHashMap[String, NativeConfEntry]().asScala + + // The backend channel is delivered once, during native backend initialization, so a registration + // arriving afterwards can never reach it: the conf would show up on the runtime channel only, and + // native would keep using its own fallback wherever it reads the key at init. Latched on first + // delivery so a late declaration is reported rather than silently half-applied. + @volatile private var backendConfDelivered = false + + /** + * Register a conf key to be passed to native side. Called by `ConfigBuilder` when an entry + * declaring `passToNative` is created; conf objects declare their native confs through the + * builders instead of calling this directly. + * + * @param isStatic + * whether the conf is static to the native backend, i.e. declared by `buildStaticConf` / + * `registerStaticConf`. A static conf is delivered on the backend channel only; a modifiable + * one on both. + * @param convert + * normalizes a value for native through the conf's own value converter, applied whether the + * value comes from the user or from a resolved default. + * @param declaredDefault + * the default declared by the conf itself, in its already-converted native form, evaluated per + * delivery so a dynamic default stays up to date. `None` for a `createOptional` conf, which + * then falls back to the foreign declaration for a Spark-owned key, or is delivered only when + * set. + */ + private[config] def register( + key: String, + isStatic: Boolean, + convert: String => String = identity, + declaredDefault: => Option[String] = None): Unit = { + val entry = NativeConfEntry(key, convert, () => declaredDefault) + if (!isStatic) { + doRegister(runtimeEntries, entry) + } + doRegisterToBackend(entry) + } + + private def doRegisterToBackend(entry: NativeConfEntry): Unit = { + if (backendConfDelivered) { + // Not fatal: the conf still works on the runtime channel, and failing here would take down a + // query for a conf object that merely loaded late. But native backend init has already + // happened, so declare the gap loudly - the usual cause is a conf object that is not declared + // through `Component.confs()`. + logWarning( + s"Native conf ${entry.key} was declared after native backend conf had already been " + + s"delivered, so it will not reach the backend channel. Declare its conf object through " + + s"Component.confs() so that it is initialized before native backend initialization.") + } + 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!") + } + + /** + * Visible for testing. Production code never asks which channel a key is on - it asks for the + * channel's contents through `selectRuntimeConf` / `selectBackendConf`. `private[gluten]` rather + * than `private[config]` because `ComponentSuite` lives in `org.apache.gluten.component`. + */ + private[gluten] def isRuntimeKey(key: String): Boolean = runtimeEntries.contains(key) + + /** Visible for testing. See [[isRuntimeKey]]. */ + private[gluten] def isBackendKey(key: String): Boolean = backendEntries.contains(key) + + /** + * Select runtime-scoped native confs from the given conf map. A key absent from `conf` is + * delivered with its declared default, if it has one. + */ + 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. A key absent from `conf` is + * delivered with its declared default, if it has one. + * + * Marks the backend channel as delivered, so that a declaration arriving afterwards - which can + * no longer reach native backend initialization - is reported rather than silently applied to the + * runtime channel alone. + */ + def selectBackendConf(conf: scala.collection.Map[String, String]): Map[String, String] = { + backendConfDelivered = true + select(backendEntries, conf) + } Review Comment: Good point, and it was not hypothetical — `NativeConfRegistrySuite` was emitting that warning on nearly every test. Fixed in `60b0ea879`: `resetBackendConfDeliveredForTesting()`, called from the suite's `withRegisteredKeys` helper. Warning count in that suite went from ~10 to 0, with all 11 tests still passing. One deviation from your suggestion: it clears **only** the latch, not the entry maps. A JVM running this suite is also running others whose conf objects registered the real confs during class initialization, and class initialization happens exactly once — clearing the maps would leave the registry permanently short of those confs with no way to rebuild it, and the failure would land in whichever suite happened to run next. A test that declared a key itself already removes it through `unregister`, which is the right granularity. It is `private[gluten]` rather than `private[config]` for the same reason `isRuntimeKey` is: `ComponentSuite` lives in `org.apache.gluten.component`. ########## gluten-core/src/test/scala/org/apache/gluten/config/NativeConfRegistrySuite.scala: ########## @@ -0,0 +1,292 @@ +/* + * 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.apache.spark.network.util.ByteUnit +import org.apache.spark.sql.internal.MapProvider + +import org.scalatest.funsuite.AnyFunSuite + +import java.util.Locale + +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) } + } Review Comment: Same fix as the sibling thread — `60b0ea879`. The reset happens in `withRegisteredKeys`, which is the single funnel every declaration in this suite goes through, so each test declares against a clean latch rather than the suite needing `beforeEach`/`afterEach`. I did not take the "avoid invoking `selectBackendConf` until all declarations are complete" option, because several tests exist precisely to assert what the backend channel contains right after a specific declaration — deferring the selection would remove the thing being tested. On "mask regressions": that was the real cost, and worth naming. The warning is Gluten's only signal that a conf object loaded too late to reach native backend init, so a suite that emits it unconditionally makes a genuine occurrence invisible. It now means what it says. ########## gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala: ########## @@ -39,7 +39,7 @@ class GlutenCoreConfig(conf: SQLConf) extends Logging { def offHeapMemorySize: Long = getConf(COLUMNAR_OFFHEAP_SIZE_IN_BYTES) - def taskOffHeapMemorySize: Long = getConf(COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES) + def taskOffHeapMemorySize: Long = getConf(COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES).getOrElse(0L) Review Comment: Answered in two earlier threads with the evidence, so briefly: base declared this entry `createWithDefaultString("0")` ([b77fdef08 `GlutenCoreConfig.scala:137-145`](https://github.com/apache/incubator-gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala#L137-L145)), so `getConf(entry)` already returned `0L` when unset. `getOrElse(0L)` reproduces that value for value, and the only JVM-side consumer is `CHConfig.scala:175`, `(taskOffHeapMemorySize * 0.9).toLong` — the same arithmetic on the same number as before. The optionality that matters is on the native side, and that is where it is expressed: the entry is `createOptional` so nothing is delivered when unset, and native reads the absence as unbounded (`WholeStageResultIterator.cc:553` falls back to `kMaxMemory`). Base delivered a literal `0` there, which collapses the partial-aggregation limit instead. Changing the JVM accessor to `Option[Long]` would change `CHConfig`'s behaviour rather than preserve it, so it stays. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
