andygrove opened a new pull request, #6860: URL: https://github.com/apache/datafusion-comet/pull/6860
## Which issue does this PR close? Part of #6855. Stacked on #6858; only the last commit is new. ## Rationale for this change #6858 found that `CometColumnarToRowExec` regenerated its projection's source in every partition when it ran outside whole-stage codegen, which a schema with more fields than `spark.sql.codegen.maxFields` always does. Comet's in-memory cache reader, `CachedBatchRowIterator` from #5859, has the same cost. Spark reads a cache through it once per partition whenever it does not read the cache as batches, which is always the case for such a schema, and every call generated the reader's source again; Spark's compiled class cache only saved the compile. With the cache on by default since #5634, any Spark operator reading a wide or deeply nested cached relation pays it in every partition. ## What changes are included in this PR? - `GeneratedClassCache` holds the executor-wide cache that #6858 added inside `CometUnsafeProjection`: the most recently used 100 classes, keyed by a `ColumnLayout` of column types, nullability and `spark.sql.codegen.methodSplitThreshold`. `CometUnsafeProjection` now uses it. - `CachedBatchRowIterator` generates its reader class once per layout. The class takes its batches as a reference, so each reader gets its own batches in its own copy of the references. A class is shared only when the batches are its one reference, which is always the case today. The `spark.sql.codegen.hugeMethodLimit` check runs on every call against the size recorded with the class, so a class generated under one limit is still checked under another. - The fallback for a reader with a method over that limit builds its projection through `CometUnsafeProjection`, so it is not generated in every partition either. ## How are these changes tested? New tests in `CachedBatchRowIteratorSuite` check that a second reader for a layout generates no class and still reads only its own batches, and that a class generated under a 1-byte huge-method limit falls back there but is used without being generated again once the limit allows it. `CometUnsafeProjectionSuite` covers the refactored projection cache. `CachedBatchRowIteratorSuite`, `CometUnsafeProjectionSuite`, `CometInMemoryCacheSuite`, `CometInMemoryCachePruningSuite` and `CometExecSuite` pass on Spark 4.1, `CachedBatchRowIteratorSuite` also on Spark 3.5, and the change compiles with strict warnings. A Spark operator (a `noop` write with Comet off) reading a cached table in Comet's format, with `local[1]` on a Ryzen 9 7950X3D, JDK 17 and Spark 4.1. Times are the best of at least five iterations. The flat tables hold 200,000 rows of nullable bigint columns; the nested one is the 1,000-row fuzz-generated schema from `CometShuffleBenchmark` (100 columns, max depth 6). Before is #6858; the benchmark harness is not part of this PR. | Cached table | Partitions | Before | After | |---|---|---|---| | 200 columns | 4 | 483 ms | 454 ms | | 200 columns | 200 | 1,343 ms | 796 ms | | 1,500 columns | 4 | 4,992 ms | 4,716 ms | | 1,500 columns | 200 | 9,938 ms | 5,802 ms | | fuzz nested, max depth 6 | 4 | 626 ms | 306 ms | | fuzz nested, max depth 6 | 201 | 16,971 ms | 1,390 ms | The savings match the time each call spent generating the reader's source: called directly, after the first call compiled the class, the old reader took about 5 ms per call for 200 columns and 18 ms for 1,500 (Spark's compiled class cache hit every time), against under 0.5 ms now. For the nested schema it was about 78 ms per partition. -- 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]
