Copilot commented on code in PR #12549:
URL: https://github.com/apache/gluten/pull/12549#discussion_r3949398949
##########
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:
`std::stoll(it->second) * 1024` can overflow (multiplication happens after
parsing succeeds), potentially resulting in a negative/garbled buffer size that
then propagates into `LocalPartitionWriterOptions`. Consider guarding the
multiplication (e.g., bounds-check before multiplying) and falling back to
`kDefaultShuffleFileBufferSize` on overflow.
##########
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:
`COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES` is now declared `createOptional`
(absence is meaningful), but this accessor reintroduces a sentinel by mapping
missing to `0L`. That can blur the distinction between “unset” vs “explicitly
set to 0” for JVM-side consumers and makes it harder to preserve the intended
semantics. Prefer returning `Option[Long]` here (or otherwise exposing an
accessor that preserves ‘unset’) so callers can handle absence explicitly.
##########
gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala:
##########
@@ -489,110 +490,155 @@ 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",
- "spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing",
- // 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"
- )
+ // Declarations of non-Gluten configurations (Spark SQL / Spark core /
Hadoop keys that have no
+ // Gluten ConfigEntry) to be passed to native side. `registerConf` /
`registerStaticConf` declare
+ // only the native delivery: the key stays owned by Spark / Hadoop, so
nothing is registered as a
+ // Gluten config entry or to SQLConf. Gluten's own configurations declare
native passing via
+ // `ConfigBuilder.passToNative` at their definitions instead.
+ private def registerNativeConfs(): Unit = {
+ // Force GlutenCoreConfig's object initialization, so that its own
`passToNative`
+ // registrations are in place before native confs are selected.
+ GlutenCoreConfig.ensureRegistered()
+
+ // Spark SQL confs read by native. All of these rely on native's own
fallback matching Spark's
+ // default, so nothing is delivered when the key is unset - see
`ConfigBuilder.passToNative`.
+ // `spark.sql.legacy.sizeOfNull` is `passToNative()` for documentation
purposes only: it is
+ // never read from the conf map, since the value is baked as a substrait
literal at plan
+ // conversion (see `ExpressionConverter`).
+
registerConf(SQLConf.LEGACY_SIZE_OF_NULL.key).stringConf.passToNative().createOptional
+ // Read by `ConfigExtractor` as a bool with its own fallback of `true`,
matching Spark's
+ // default. A string literal because not every supported Spark version has
the entry.
+ registerConf("spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing")
+ .booleanConf
+ .passToNative()
+ .createOptional
+ registerConf(SQLConf.JSON_GENERATOR_IGNORE_NULL_FIELDS.key)
+ .stringConf
+ .passToNative()
+ .createOptional
+ registerConf(SQLConf.RUNTIME_BLOOM_FILTER_EXPECTED_NUM_ITEMS.key)
+ .stringConf
+ .passToNative()
+ .createOptional
+ registerConf(SQLConf.RUNTIME_BLOOM_FILTER_NUM_BITS.key)
+ .stringConf
+ .passToNative()
+ .createOptional
+ registerConf(SQLConf.RUNTIME_BLOOM_FILTER_MAX_NUM_BITS.key)
+ .stringConf
+ .passToNative()
+ .createOptional
+ registerConf(SQLConf.RUNTIME_BLOOM_FILTER_MAX_NUM_ITEMS.key)
+ .stringConf
+ .passToNative()
+ .createOptional
+
registerConf(SPARK_IO_COMPRESSION_CODEC).stringConf.passToNative().createOptional
+ // Velox compares the value against upper-cased literals; ClickHouse
lower-cases it itself.
+ // Declaring `transform(toUpperCase)` mirrors Spark's own entry which also
upper-cases.
+ registerConf(SQLConf.LEGACY_TIME_PARSER_POLICY.key)
+ .stringConf
+ .transform(_.toUpperCase(Locale.ROOT))
+ .passToNative()
+ .createOptional
+
registerConf(SQLConf.CASE_SENSITIVE.key).stringConf.passToNative().createOptional
+
registerConf(SQLConf.IGNORE_MISSING_FILES.key).stringConf.passToNative().createOptional
+ registerConf(SQLConf.LEGACY_STATISTICAL_AGGREGATE.key)
+ .stringConf
+ .passToNative()
+ .createOptional
+ registerConf(SQLConf.DECIMAL_OPERATIONS_ALLOW_PREC_LOSS.key)
+ .stringConf
+ .passToNative()
+ .createOptional
Review Comment:
These Spark-owned keys are boolean confs in Spark SQLConf, but they’re
registered here as `stringConf`. That bypasses Spark-like normalization/parsing
(and can allow non-canonical values to reach native). Use the matching typed
builders (`booleanConf` for boolean keys, etc.) so the declared converter
mirrors the owner’s semantics and consistently normalizes the delivered values.
##########
gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala:
##########
Review Comment:
The comment states this entry has “No `passToNative`”, but the builder now
calls `.passToNative()`. Update the comment to reflect the new behavior (or
remove `.passToNative()` if the intent is truly JVM-only). As written, the code
and documentation contradict each other.
##########
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:
`case e if e.defaultValue.isDefined => e.defaultValue.map(...)` can evaluate
a dynamic default twice because `defaultValue` is invoked once for `isDefined`
and again for `map`. For `createWithDefaultFunction`, this can produce
inconsistent results (and unnecessary work) within a single selection. Evaluate
`defaultValue` once (e.g., assign to a local val and `map` it) or drop the
guard and rely on a single `entry.defaultValue.map(_.toString)` after excluding
fallback entries.
--
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]
