dwsmith1983 commented on code in PR #5615:
URL: https://github.com/apache/datafusion-comet/pull/5615#discussion_r3921078525
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -95,6 +95,56 @@ private[comet] trait PlanDataInjector {
* Registry and utilities for injecting per-partition planning data into
operator trees.
*/
private[comet] object PlanDataInjector extends Logging {
+ import java.nio.ByteBuffer
+ import java.util.{LinkedHashMap, Map => JMap}
+
+ private[comet] final val maxCachedBasePlans = 16
+
+ // Every task of a stage deserializes its own byte-identical copy of the
base plan, so
+ // without a cache an executor re-parses the same operator tree once per
task. Parsed
+ // Operators are immutable, so one instance is safely shared across
concurrent tasks.
+ // Keyed by content (ByteBuffer hashes/compares the bytes) since the arrays
are distinct.
+ //
+ // Entries are whole parsed plan trees, so the entry count is what bounds
executor memory:
+ // at most 16 recent stages' plans stay live, LRU-evicted as stages turn
over. A stage
+ // rerun that misses after eviction simply re-parses.
+ private val basePlanCache = java.util.Collections.synchronizedMap(
+ new LinkedHashMap[ByteBuffer, Operator](4, 0.75f, true) {
+ override def removeEldestEntry(eldest: JMap.Entry[ByteBuffer,
Operator]): Boolean = {
+ size() > maxCachedBasePlans
+ }
+ })
+
+ /**
+ * Look up `key`, computing and inserting the value on a miss. The
computation runs outside any
+ * lock so unrelated misses never serialize behind each other; when two
threads race the same
+ * cold key, the first insert wins and the loser adopts it, keeping the
cached value
+ * reference-shared (which the sourceKey memo's identity fast path relies
on).
+ */
+ private[comet] def cachedOrCompute[K, V](cache: JMap[K, V], key: K)(compute:
=> V): V = {
+ val cached = cache.get(key)
+ if (cached != null) {
+ cached
+ } else {
+ val computed = compute
+ cache.synchronized {
+ val winner = cache.get(key)
+ if (winner != null) {
+ winner
+ } else {
+ cache.put(key, computed)
+ computed
+ }
+ }
+ }
+ }
+
+ /**
+ * Parse a stage's base plan bytes, sharing the parsed tree across the
executor's tasks. Falls
+ * back to a plain parse on eviction, so a stage rerun is always correct.
+ */
+ def parseBasePlan(bytes: Array[Byte]): Operator =
+ cachedOrCompute(basePlanCache,
ByteBuffer.wrap(bytes))(Operator.parseFrom(bytes))
Review Comment:
Done in 36fa94d3a. The base plan cache now keys on a PlanKey that stores its
hash, computed once per task before the monitor; equals is identity then
Arrays.equals. Driver transport was not practical for this one since the plan
bytes are the task binary itself, so this is your measured middle option. The
other two caches are gone entirely, see the main comment.
--
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]
