andygrove commented on code in PR #7238:
URL: https://github.com/apache/datafusion-comet/pull/7238#discussion_r4431693217
##########
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:
Confirmed: the conf defaults to `false` in 3.4.3 and `true` in 3.5.9. Both
passages now say Spark 3.5, and 3.4 only with
`spark.sql.optimizer.canChangeCachedPlanOutputPartitioning` set to `true`; the
Limitations note adds that 3.4 otherwise turns AQE off for the cached plan, so
it is written from the Arrow batches. Fixed in f94fe7060.
##########
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:
You were right: without the conf the test fails on 3.4 (`false did not equal
true AQE is true`), and my claim that it passed there was wrong. It now sets
`CAN_CHANGE_CACHED_PLAN_OUTPUT_PARTITIONING` to `true`, as `withAQECache` does,
and passes on 3.4 locally. Fixed in f94fe7060.
--
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]
