felipepessoto commented on code in PR #12967:
URL: https://github.com/apache/gluten/pull/12967#discussion_r4077456648
##########
backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/stats/GlutenDeltaJobStatsTracker.scala:
##########
@@ -86,6 +86,22 @@ private[stats] class GlutenDeltaJobStatsTracker(val
delegate: DeltaJobStatistics
}
object GlutenDeltaJobStatsTracker extends Logging {
+ private val statsPlanObserverLock = new Object
+ @volatile private var statsPlanObserver: Option[(Path, SparkPlan) => Unit] =
None
+
+ /** Observes task-local statistics plans in local-mode tests; callbacks run
on task threads. */
+ private[delta] def withStatsPlanObserver[T](observer: (Path, SparkPlan) =>
Unit)(f: => T): T =
+ statsPlanObserverLock.synchronized {
+ require(Utils.isTesting, "Statistics plan observation is only available
in tests")
+ require(statsPlanObserver.isEmpty, "A statistics plan observer is
already registered")
+ statsPlanObserver = Some(observer)
+ try {
+ f
+ } finally {
+ statsPlanObserver = None
+ }
+ }
Review Comment:
The current implementation serializes observer registration scopes with a
lock held across the write, publishes the callback through @volatile , and
clears it in finally . The caller filters observations by its unique table
path and stores matching plans in a ConcurrentLinkedQueue , so unrelated
writes do not contaminate its assertions. A DynamicVariable would not
reliably propagate into Spark’s existing task threads. I’d retain this scoped
local-mode hook unless we need concurrent independent observers.
--
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]
