comphead commented on code in PR #7238:
URL: https://github.com/apache/datafusion-comet/pull/7238#discussion_r4414769808
##########
spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala:
##########
@@ -2758,6 +2758,47 @@ class CometInMemoryCacheSuite extends CometTestBase {
assert(zstdSize == plainSize, s"the relation size should not depend on the
codec: $sizes")
}
+ test("Comet in-memory cache is built from rows under AQE before Spark 4.0") {
+ // Spark builds a cache from the batches of its plan only when the plan
reports that it
+ // supports columnar output. A cached AdaptiveSparkPlanExec does not
before Spark 4.0, so there
+ // a relation cached under AQE goes native plan, columnar-to-row,
UnsafeRow and back to Arrow
+ // before it is compressed. Without AQE, or from 4.0 on, it is written
from the Arrow batches.
+ // The user guide's build timings are from Spark 4.1 and say so.
+ Seq(true, false).foreach { adaptive =>
+ withConversions(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString,
+ SQLConf.SHUFFLE_PARTITIONS.key -> "3",
+ CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
+ spark.catalog.clearCache()
+ try {
+ spark
+ .range(0, 100, 1, 2)
+ .selectExpr("id % 7 AS k", "id AS v")
+ .groupBy("k")
+ .agg(sum("v").as("s"))
+ .createOrReplaceTempView("rows_build_cache")
+ spark.catalog.cacheTable("rows_build_cache")
+ assert(spark.table("rows_build_cache").count() == 7)
+ val cachedPlan = spark.sharedState.cacheManager
+ .lookupCachedData(spark.table("rows_build_cache"))
+ .get
+ .cachedRepresentation
+ .cacheBuilder
+ .cachedPlan
+ assert(
+ cachedPlan.isInstanceOf[AdaptiveSparkPlanExec] == adaptive,
Review Comment:
For the same reason, I expect this to fail on Spark 3.4 with `adaptive =
true`, since the test does not set
`spark.sql.optimizer.canChangeCachedPlanOutputPartitioning` and the cached plan
comes back non-adaptive. `withAQECache` and the planning test set that conf to
`true` for exactly this. The description says it passes on 3.4, and I could not
find what makes it pass there. I have not run it. Could you recheck with
`run-spark-3.4-tests` or `run-all-spark-profiles`, and set the conf here if
needed?
##########
docs/source/user-guide/latest/in-memory-cache.md:
##########
@@ -301,6 +302,14 @@ operator reads a relation cached in Comet's format depends
on the scan below it:
Whenever Spark's scan reads Comet's format, which also happens for a relation
whose cached plan
records `Dataset.observe` metrics, Comet records a fallback reason on the scan.
+On **Spark 3.4 and 3.5**, a relation cached under AQE
(`spark.sql.adaptive.enabled`, on by default) is
Review Comment:
On Spark 3.4 with default settings the cache is not built under AQE.
`CacheManager.getOrCloneSessionWithConfigsOff` turns
`spark.sql.adaptive.enabled` off for the cached plan unless
`spark.sql.optimizer.canChangeCachedPlanOutputPartitioning` is true, and that
conf defaults to false in 3.4.3 and true in 3.5.9. So on 3.4 the cached plan is
not an `AdaptiveSparkPlanExec`, and the cache is written from Arrow batches.
The suite already says this in the comment on the "does not build the cached
RDD while planning" test. Could this passage and the one in the build-cost
section say 3.5, and 3.4 only with that conf set?
--
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]
