andygrove commented on code in PR #7082:
URL: https://github.com/apache/datafusion-comet/pull/7082#discussion_r4345446138
##########
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala:
##########
@@ -3483,6 +3484,75 @@ class CometExecSuite extends CometTestBase {
}
}
+ /** The footer metadata cache hits and misses of the native scan that ran
`df`. */
+ private def footerCacheCounts(df: DataFrame): (Long, Long) = {
+ val scan = find(df.queryExecution.executedPlan)(s =>
+ s.isInstanceOf[CometScanExec] || s.isInstanceOf[CometNativeScanExec])
+ assert(scan.isDefined, "Expected to find a Comet scan node")
+ val metrics = scan.get.metrics
+ (metrics("scan_io_metadata_cache_hits").value,
metrics("scan_io_metadata_cache_misses").value)
+ }
+
+ test("Parquet footers are shared by the tasks of a stage") {
+ withTempDir { dir =>
+ val path = new Path(dir.toURI.toString, "one-file").toString
+ spark
+ .range(0, 200000)
+ .selectExpr("id", "cast(id * 7 as string) as s")
+ .coalesce(1)
+ .write
+ .parquet(path)
+
+ // One file split into many tasks, so every task needs the same footer.
+ withSQLConf(
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ SQLConf.FILES_MAX_PARTITION_BYTES.key -> "64k",
+ SQLConf.FILES_OPEN_COST_IN_BYTES.key -> "0") {
+ val query = s"SELECT sum(id), count(s) FROM parquet.`$path`"
+ val tasks = spark.read.parquet(path).rdd.getNumPartitions
+ assert(tasks > 8, s"Expected the file to be split into many tasks, got
$tasks")
+
+ withSQLConf(CometConf.COMET_PARQUET_SHARED_METADATA_CACHE_ENABLED.key
-> "true") {
+ val df = sql(query)
+ df.collect()
+ val (hits, misses) = footerCacheCounts(df)
+ // Tasks that start together can each miss, but not all of them do.
+ assert(hits > 0, s"hits=$hits misses=$misses")
+ assert(misses < tasks, s"hits=$hits misses=$misses tasks=$tasks")
+ }
+
+ withSQLConf(CometConf.COMET_PARQUET_SHARED_METADATA_CACHE_ENABLED.key
-> "false") {
+ val df = sql(query)
+ df.collect()
+ val (hits, misses) = footerCacheCounts(df)
+ assert(hits == 0, s"hits=$hits misses=$misses")
+ assert(misses >= tasks, s"hits=$hits misses=$misses tasks=$tasks")
+ }
+ }
+ }
+ }
+
+ test("Parquet footers are not shared between queries") {
+ withTempDir { dir =>
+ val path = new Path(dir.toURI.toString, "table").toString
+ spark.range(0, 1000).repartition(4).write.parquet(path)
+ withSQLConf(
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_PARQUET_SHARED_METADATA_CACHE_ENABLED.key -> "true") {
+ // The relation keeps its file listing, as a cached table does, so the
second query is
+ // given a file that no longer exists. It must find that out by
opening the file.
+ val table = spark.read.parquet(path)
+ assert(table.count() == 1000)
+ val deleted = new java.io.File(dir, "table")
+ .listFiles()
+ .filter(_.getName.endsWith(".parquet"))
+ .head
+ assert(deleted.delete())
+ intercept[SparkException](table.count())
Review Comment:
Done. The test now asserts that the first read's plan has a
`CometNativeScanExec`. It also checks that the second read fails because of the
missing file: a `FileNotFoundException` in the cause chain. That is the cause
of `FAILED_READ_FILE.FILE_NOT_EXIST` on Spark 4.0+, which
`HiveMetadataCacheSuite` checks for, and of Spark 3.x's file-not-found read
error. It passes on 4.1 and 3.5.
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -184,6 +184,19 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(false)
+ val COMET_PARQUET_SHARED_METADATA_CACHE_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.parquet.metadataCache.sharedWithinStage.enabled")
+ .category(CATEGORY_PARQUET)
+ .doc(
+ "When enabled, the tasks of one Spark stage that run on the same
executor share the " +
+ "native Parquet scan's cache of file footers and page indexes, so a
file split " +
+ "across many tasks has its footer fetched and parsed once per
executor instead of " +
Review Comment:
Added. The accounting-gap section of `memory_management.md` now has a bullet
for native caches that outlive their task. The metadata cache isn't reserved
from the pool. A stage's cache holds up to 50 MiB per running task. The two
most recent caches stay resident after their tasks finish, at 50 MiB each, so
up to 100 MiB can be held with no task running.
##########
native/core/src/execution/jni_api.rs:
##########
@@ -1009,6 +1020,13 @@ fn prepare_datafusion_session_context(
.with_object_store_registry(Arc::new(CometObjectStoreRegistry::default()));
rt_config = rt_config.with_memory_pool(memory_pool);
+ // Let the tasks of one stage on this executor share the Parquet footer
and page index cache
+ // instead of each fetching and parsing the footers of the files it reads.
+ let task_slots = spark_config.get_usize(SPARK_EXECUTOR_CORES, 1) /
task_cpus.max(1);
Review Comment:
You're right, thanks for the reproduction. That count comes from
`numDriverOrExecutorCores`, which is an estimate on a standalone executor
without `spark.executor.cores`, and it shouldn't size memory. Nothing in the
task gets the executor's real core count, so the cache no longer uses one.
Each task now holds a `TaskCache` handle on its stage's cache while its plan
runs. Taking the handle counts the task in, and dropping it counts the task
out. The cache holds 50 MiB for each task that is running, and 50 MiB once none
is. On your two-core executor that gives at most two tasks' worth, 100 MiB,
whatever the host's processor count. A retained cache shrinks to 50 MiB, so the
two that outlive their tasks hold at most 100 MiB together.
`a_stage_cache_holds_one_task_limit_per_running_task` covers the counting and
the shrink. (40beedd74)
This is the same idea as the running-task count in #7096, which the fair
pool divides by. This one counts the tasks of one stage rather than of the
executor, since only those share the cache.
--
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]
