viirya commented on code in PR #7082:
URL: https://github.com/apache/datafusion-comet/pull/7082#discussion_r4339809671
##########
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:
Agreed. Could `memory_management.md` get the same note? Its "Who allocates
what" table and the accounting-gap section don't cover native allocations that
outlive the task. As far as I know, the two retained caches are the first ones
that do.
##########
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 " +
+ "once per task. The cache is never shared between stages or stage
attempts, so a " +
+ "query always reads the footers of its files afresh. When disabled,
each task " +
+ "keeps its own cache.")
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
I'd like to understand how often this hits outside the benchmark setup. 470
tasks over a 474 MB file is roughly a 1 MB `maxPartitionBytes`. At the default
128 MB that file is 4 splits.
Spark sorts splits by length with a stable sort, so the splits of one file
land in adjacent partitions. `TaskSchedulerImpl.resourceOfferSingleTaskSet`
hands out one task per shuffled offer per round. So those adjacent tasks start
on different executors at about the same time. Tasks that do land on the same
executor together each miss, since there is no single-flight. So I'd expect
hits mainly when splits per file far exceed the number of executors.
Could you share `scan_io_metadata_cache_hits` and
`scan_io_metadata_cache_misses` at the default split size with more than one
executor? If the win is limited to small splits or very large files, would it
make sense to ship this off by default, like `prefetchNextFileMetadata` in
#7094? That would also avoid changing the per-task headroom comphead raised for
every scan.
##########
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:
Could this test assert that the plan has a `CometNativeScanExec`, like the
restart suite does? If the read falls back to Spark, the test passes without
exercising the cache. `intercept[SparkException]` also accepts any failure.
Checking that the cause is the missing-file error would tie it to the case
`HiveMetadataCacheSuite` cares about.
##########
native/core/src/execution/jni_api.rs:
##########
@@ -1009,6 +1011,14 @@ 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.
+ if let Some(cache) = shared_metadata_cache_for_task(spark_config) {
+ rt_config = rt_config.with_cache_manager(
+
CacheManagerConfig::default().with_file_metadata_cache(Some(cache)),
Review Comment:
One thing to watch if the limit gets scaled, as comphead suggested on
`shared_metadata_cache.rs`. `CacheManager::try_new` in DataFusion 55.2 calls
`file_metadata_cache.update_cache_limit(config.metadata_cache_limit)` on the
cache it is given. This passes `CacheManagerConfig::default()`, so every task
that builds its `RuntimeEnv` resets the shared cache to 50 MiB. Today that does
nothing because both values are 50 MiB. But a larger limit set only in
`DefaultCache::new` would be undone by the next task. So it would need to go
through `with_metadata_cache_limit` here as well.
--
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]
