jackylee-ch commented on code in PR #12549:
URL: https://github.com/apache/gluten/pull/12549#discussion_r3949765399
##########
gluten-core/src/main/scala/org/apache/gluten/config/ConfigBuilder.scala:
##########
@@ -77,6 +80,131 @@ 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 as owned by Spark / Hadoop rather than by Gluten. Set
by `registerConf` /
+ * `registerStaticConf`.
+ *
+ * A foreign config is not registered as a Gluten config entry and not
registered to `SQLConf`
+ * (Spark or Hadoop already did that); the builder is only used to declare
how the key is
+ * delivered to native side.
+ */
+ private[config] def markForeign(): ConfigBuilder = {
+ _isForeign = true
+ this
+ }
+
+ /**
+ * Marks this config to be passed to native side. A value set by the user is
always delivered;
+ * what happens when it is not set is stated by the terminal method, which
is the whole of the
+ * rule:
+ *
+ * - `createOptional`: nothing is delivered, leaving native's own fallback
in charge. This is
+ * the common case for a foreign key, since native usually declares the
same fallback Spark /
+ * Hadoop does, or branches on the key being absent at all.
+ * - `createWithDefault(value)`: the stated value is delivered. For a
Gluten config this is its
+ * own default; for a foreign one it says Gluten deliberately departs
from what both Spark /
+ * Hadoop and native would apply, e.g. `fs.s3a.path.style.access` where
native falls back to
+ * `false` and Gluten wants `true`.
+ * - `createWithDefaultFunction(f)`: `f` is evaluated at each delivery.
Use it when the default
+ * cannot be a literal - because it follows JVM or session state
(`spark.sql.session.timeZone`
+ * follows the JVM default time zone) or has changed across Spark
versions
+ * (`spark.sql.ansi.enabled` flipped its default in 4.0). For a foreign
key, read it back
+ * through the owner's own accessor rather than restating it: a restated
default is exactly
+ * what drifts.
+ *
+ * The config is registered to [[NativeConfRegistry]] on entry creation.
+ *
+ * Which delivery channels it lands on follows the conf's mutability, so
there is no argument:
+ * - `buildConf` / `registerConf`: modifiable at any time and usable at
any time. Delivered both
+ * during native backend initialization and on each native runtime
creation, so native
+ * observes the current value wherever it reads the key.
+ * - `buildStaticConf` / `registerStaticConf`: set while the native
backend is initialized and
+ * not modifiable afterwards. Delivered once during native backend
initialization.
+ */
+ def passToNative(): ConfigBuilder = {
+ _passToNative = true
+ this
+ }
+
+ /**
+ * Normalizes a value for native side through the conf's own value
converter, i.e. the one chosen
+ * by `stringConf` / `bytesConf(unit)` / `intConf` / ... plus any
`transform`. A conf therefore
+ * states how its value is parsed exactly once, at its declaration, and both
a user-set value and
+ * a resolved default go through it.
+ *
+ * This is what makes a size conf reach native as a number rather than as
"64k": a
+ * `bytesConf(ByteUnit.KiB)` yields the KiB count the foreign entry would
yield, and a
+ * `bytesConf(ByteUnit.BYTE)` the byte count. A foreign conf declares the
same converter Spark /
+ * Hadoop declares, so JVM and native agree on the value's meaning and
native applies whatever
+ * unit conversion it needs on top - `spark.shuffle.file.buffer` is KiB on
both sides, and native
+ * multiplies by 1024.
+ *
+ * Falls back to the raw string when the entry has no usable converter (e.g.
a fallback entry),
+ * since delivering the value unchanged is always better than dropping it.
+ */
+ private def convertForNative(entry: ConfigEntry[_], raw: String): String = {
+ try {
+ entry.valueConverter(raw) match {
+ // An `OptionalConfigEntry` wraps its converter's result in `Option`.
+ case o: Option[_] => o.map(_.toString).getOrElse(raw)
+ case null => raw
+ case v => v.toString
+ }
+ } catch {
+ // A value Spark or Hadoop would reject is not this mechanism's business
to validate: Spark or
+ // Hadoop raises on it at its own read site, with its own message.
Deliver it unchanged rather
+ // than failing conf selection, which runs per task.
+ case _: IllegalArgumentException => raw
+ }
+ }
+
+ private[config] def registerToNative(entry: ConfigEntry[_]): Unit = {
+ require(
+ !_isForeign || _passToNative,
+ s"Config $key: a config declared by registerConf() /
registerStaticConf() must be marked " +
+ s"with passToNative(), otherwise declaring it has no effect"
+ )
+ if (!_passToNative) {
+ return
+ }
+ // The channel follows the conf's mutability. A modifiable conf is
delivered on both channels so
+ // native observes the current value wherever it reads the key; a static
conf is set while the
+ // native backend is initialized and not modifiable afterwards, so
delivering it once there is
+ // lossless.
+ NativeConfRegistry.register(
+ key,
+ _isStatic,
+ convert = convertForNative(entry, _),
+ declaredDefault = declaredDefault(entry))
+ }
+
+ /**
+ * The default delivered to native for a key the user did not set, or `None`
to deliver nothing
+ * and leave native's own fallback in charge. Read per delivery rather than
snapshotted, so a
+ * `createWithDefaultFunction` default that follows JVM or session state
keeps delivering its
+ * current value.
+ *
+ * Which of the two the caller gets is stated by the terminal method - see
[[passToNative]].
+ */
+ private def declaredDefault(entry: ConfigEntry[_]): Option[String] = entry
match {
+ // A fallback entry reports the *target* conf's default as its own, and
the target is delivered
+ // under its own key. Delivering it here would also contradict the user:
with only the target
+ // conf set, this key would carry the target's default rather than the
value the user chose.
+ case _: ConfigEntryFallback[_] | _: ConfigEntryForeignFallback[_] => None
+ // `createWithDefault(value)` / `createWithDefaultFunction(f)`. Reading
the parsed default
+ // rather than the raw default string means a "64MB" bytes conf reaches
native as "67108864";
+ // for the function form, reading it here is what re-evaluates `f` on
every delivery.
+ case e if e.defaultValue.isDefined => e.defaultValue.map(_.toString)
+ // `createOptional`: nothing is delivered when the key is not set.
+ case _ => None
Review Comment:
Fixed in `01c3fd7ac`. `declaredDefault` no longer has a guard at all:
```scala
private def declaredDefault(entry: ConfigEntry[_]): Option[String] = entry
match {
case _: ConfigEntryFallback[_] | _: ConfigEntryForeignFallback[_] => None
case e => e.defaultValue.map(_.toString)
}
```
`createOptional` lands in the last case and yields `None` on its own, since
an `OptionalConfigEntry` declares no default value — so the `isDefined` guard
was not carrying any weight, and dropping it is what makes the read happen
exactly once.
##########
cpp/core/jni/JniWrapper.cc:
##########
@@ -845,6 +844,22 @@
Java_org_apache_gluten_vectorized_LocalPartitionWriterJniWrapper_createPartition
auto dataFile = jStringToCString(env, dataFileJstr);
auto localDirs = splitPaths(jStringToCString(env, localDirsJstr));
+ // `spark.shuffle.file.buffer` is declared with `bytesConf(ByteUnit.KiB)` on
the JVM side, matching
+ // Spark's own declaration, so the delivered value is a KiB count. Convert
it to bytes here, which
+ // is the unit every reader of `shuffleFileBufferSize` uses.
+ auto shuffleFileBufferSize = kDefaultShuffleFileBufferSize;
+ auto& conf = ctx->getConfMap();
+ if (auto it = conf.find(kShuffleFileBufferSize); it != conf.end()) {
+ try {
+ shuffleFileBufferSize = std::stoll(it->second) * 1024;
Review Comment:
Fixed in `01c3fd7ac`, range-checked before the multiplication rather than
after:
```cpp
auto kib = std::stoll(it->second);
GLUTEN_CHECK(
kib > 0 && kib <= std::numeric_limits<int64_t>::max() / 1024,
"out of range for a KiB count: " + it->second);
shuffleFileBufferSize = kib * 1024;
```
`GLUTEN_CHECK` throws, so it lands in the same `catch` as the parse failure
and falls back to `kDefaultShuffleFileBufferSize` with a warning — nothing
propagates out of the JNI method.
The lower bound is deliberate too: this is an allocation size, and a
non-positive value is not merely odd. `Spill::openForRead` takes it as
unsigned, so a negative becomes an enormous prefetch size, and `0` divides by
zero in `MmapFileStream`.
--
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]
