HyukjinKwon commented on code in PR #59320:
URL: https://github.com/apache/spark/pull/59320#discussion_r4271696331
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/RuntimeConfig.scala:
##########
@@ -46,8 +46,8 @@ class RuntimeConfig private[sql](val sqlConf: SQLConf = new
SQLConf) extends sql
// Connect config RPC writes through this method too, so both front ends
behave the same way.
//
// Scoped to the prefix so an ordinary configuration write does not take
the monitor. Check
- // and write under the monitor that guards the session configurations --
`settings` is a
- // `Collections.synchronizedMap`, so every `put` locks the wrapper this
block locks -- because
+ // and write under the monitor that guards the session configurations --
every method of
Review Comment:
**Nit (P3):** This now says every method of `settings` locks the map itself,
but `ModificationCountingMap`'s `keySet`, `values`, and `entrySet` return
unsynchronized views. Its class doc and `ModificationCountingMapSuite` ("every
method except the views locks the map itself") both carve them out. The block
itself is still correct, because `put` goes through `mutate`, which
synchronizes on the map. The old wording was accurate because
`Collections.synchronizedMap` views lock the same mutex. The new wording could
lead someone to call, say, `sqlConf.settings.keySet.contains(k)` without
holding the monitor and race with a concurrent `SET`. Could it be narrowed to
something like "every method of `settings` except its views locks the map
itself, so every `put` locks the map this block locks"?
**Verification:**
- **Inspection:** The revised comment's locking claim matches
ModificationCountingMap: non-view methods (including put) synchronize on the
map itself, and keySet/values/entrySet are excluded from that claim.
##########
sql/catalyst/benchmarks/RetainedResolutionConfigsBenchmark-results.txt:
##########
@@ -0,0 +1,382 @@
+================================================================================================
+view: 32 settings, 25% retained, warm
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.15+6-Ubuntu-0ubuntu120.04 on Linux
5.4.0-1163-aws-fips
Review Comment:
Thanks, verified: all three result files now come from the "Run benchmarks"
workflow runner (AMD EPYC 7763, Linux 6.17.0-1022-azure) for JDK 17, 21, and
25, consistent with the other catalyst benchmark results.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4258683487","thread_id":"inline:4258683487","verdict_sha256":"ae877c56c9e26aa21afebfcdd9e33ffa742e923b49b144b08d69c09e8df9705d"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -326,15 +326,30 @@ object Analyzer {
newConf: SQLConf,
existingConf: SQLConf,
createSparkVersion: String = ""): Unit = {
- existingConf.getAllConfs.foreach { case (key, value) =>
- if (key.startsWith("spark.sql.catalog.") ||
SQLConf.isSessionBindingPolicy(key)) {
- newConf.settings.put(key, value)
+ // The retained configs are memoized on `existingConf` unless
+ // `ANALYZER_MEMOIZE_RETAINED_RESOLUTION_CONFIGS` is off, which recomputes
them on every build.
+ if
(existingConf.getConf(SQLConf.ANALYZER_MEMOIZE_RETAINED_RESOLUTION_CONFIGS)) {
+ newConf.settings.putAll(existingConf.retainedResolutionConfigs)
+ } else {
+ existingConf.getAllConfs.foreach { case (key, value) =>
+ if (bodyFollowsCaller(key)) {
+ newConf.settings.put(key, value)
+ }
}
}
trySetAnsiValue(newConf, createSparkVersion)
}
+ /**
+ * True if a view or SQL function body takes the caller's value for `key`
rather than the
Review Comment:
Thanks, verified: the narrowed doc now says only the caller's explicitly set
keys replace saved values, and only on paths that go through
`retainResolutionConfigsForAnalysis`.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4258683496","thread_id":"inline:4258683496","verdict_sha256":"ae877c56c9e26aa21afebfcdd9e33ffa742e923b49b144b08d69c09e8df9705d"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2893,6 +2935,19 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val ANALYZER_MEMOIZE_RETAINED_RESOLUTION_CONFIGS =
+ buildConf("spark.sql.analyzer.memoizeRetainedResolutionConfigs.enabled")
+ .internal()
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
Review Comment:
Thanks, verified: the flag is now `NOT_APPLICABLE`, and the on/off
equivalence tests still expect an explicitly set value to follow into body
confs.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4258683499","thread_id":"inline:4258683499","verdict_sha256":"ae877c56c9e26aa21afebfcdd9e33ffa742e923b49b144b08d69c09e8df9705d"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/ModificationCountingMap.scala:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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 java.util
+import java.util.function.{BiConsumer, BiFunction, Function => JFunction}
+
+/**
+ * A thread-safe map of [[SQLConf]] settings that counts the mutating calls on
it, so values
+ * derived from its entries can be reused while [[modificationCount]] is
unchanged.
+ *
+ * Like the `java.util.Collections.synchronizedMap` it replaces, every method
of the map
+ * synchronizes on the map itself, so a caller iterating the map must hold its
monitor. Unlike it,
+ * the key, value, and entry views are read-only, so every mutation goes
through a method that
+ * advances the count, and unsynchronized, so a caller must hold the monitor
while using them.
+ */
+private[sql] final class ModificationCountingMap extends util.Map[String,
String] {
+
+ private val underlying = new util.HashMap[String, String]
+
+ private val readOnlyUnderlying = util.Collections.unmodifiableMap[String,
String](underlying)
+
+ @volatile private var mutations = 0L
+
+ /**
+ * Number of mutating calls so far; a call may advance it without changing
the entries. Values
+ * derived from the entries stay valid while it is unchanged; use
+ * [[snapshotWithModificationCount]] to pair a count with the entries it
describes.
+ */
+ def modificationCount: Long = mutations
+
+ /** A copy of the entries and the [[modificationCount]] they correspond to.
*/
+ def snapshotWithModificationCount: (Long, util.Map[String, String]) =
synchronized {
+ (mutations, new util.HashMap[String, String](underlying))
+ }
+
+ private def mutate[T](f: => T): T = synchronized {
Review Comment:
Thanks, verified: the new test blocks every mutator,
`snapshotWithModificationCount`, and each read on the map's own monitor, so
moving any of them to a private lock or removing the lock would fail it.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4258683512","thread_id":"inline:4258683512","verdict_sha256":"ae877c56c9e26aa21afebfcdd9e33ffa742e923b49b144b08d69c09e8df9705d"}
-->
--
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]
