andygrove opened a new pull request, #5051: URL: https://github.com/apache/datafusion-comet/pull/5051
## Which issue does this PR close? Closes #2391. ## Rationale for this change This PR continues the work started by @pchintar in #4591, who is no longer able to work on it. All credit for the original implementation goes to them. This branch is based on their branch with latest `apache/main` merged in, and picks up the remaining review feedback and CI failures. Comet currently has limited support for Spark's in-memory cache. When a table is cached and later read, the cached data cannot be consumed directly by Comet operators. Instead, the execution plan falls back to Spark's cache scan path and introduces an additional `CometSparkColumnarToColumnar` conversion before execution can continue in Comet. This extra conversion adds overhead to cached table scans and prevents cached data from remaining on a native Comet execution path. ## What changes are included in this PR? A native cache path for in-memory cached tables behind a new configuration: `spark.comet.exec.inMemoryCache.enabled` When enabled: * Cached data is stored using a Comet-specific cache serializer (compressed Arrow IPC). * Cached data is represented as `CometCachedBatch`. * Cached tables are scanned using `CometInMemoryTableScanExec`. * Cached data can be consumed directly by Comet operators without introducing a `CometSparkColumnarToColumnar` conversion. * Per-batch column statistics (lower bound, upper bound, null count, row count) are written in the format expected by `SimpleMetricsCachedBatchSerializer`, so Spark's `buildFilter` can prune cached batches before they are decoded. When disabled: * Spark's existing cache serializer continues to be used. * Existing cache scan behavior is preserved. ## How are these changes tested? `CometInMemoryCacheSuite` covers: - Comet-native cache scan over `CometCachedBatch` - Fallback behavior when native cache support is disabled - Multi-partition cached tables - Empty cached tables - Projection-only cache reads - Shuffle execution after cached table scans - Stats-based batch pruning - Empty projection scans (`SELECT count(*)`) - Floating-point pruning with NaN values - Fallback read path for Spark `DefaultCachedBatch` `CometInMemoryCacheBenchmark` compares the native cache path against the existing fallback path for repeated cached table scans and selective-filter scans exercising cache pruning. Benchmark results (release build, Apple M3 Ultra, JDK 17, Spark 3.5 profile, 5M-row cached table): ### Repeated full scan (`SELECT sum(id), sum(k), sum(v)`) | Case | Best (ms) | Avg (ms) | Relative | |------|----------:|---------:|---------:| | Comet cache disabled | 180 | 201 | 1.0x | | Comet cache enabled | 121 | 128 | **1.5x** | ### Selective filter (`WHERE id >= 4500000 AND id < 4750000`) | Case | Best (ms) | Avg (ms) | Relative | |------|----------:|---------:|---------:| | Comet cache disabled | 46 | 53 | 1.0x | | Comet cache enabled | 42 | 48 | **1.1x** | Performance follow-ups are tracked in #4781. -- 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]
