andygrove opened a new pull request, #6853: URL: https://github.com/apache/datafusion-comet/pull/6853
## Which issue does this PR close? Part of #6834: the first step of its statistics-gathering item. ## Rationale for this change `ArrowCachedBatchSerializer.gatherColumnStats` visits every row of every column to count nulls, including binary, array, map and struct columns, which record a null count and no bounds. When Comet operators or the row write path build the cache, its columns are Comet's Arrow vectors, and each one already holds that count: `CometDecodedVector` takes it from Arrow when the vector is built. The pass over the rows only recomputes it. ## What changes are included in this PR? - When a column is a `CometDecodedVector` holding exactly the batch's rows, `gatherColumnStats` takes its null count from the vector. A column without bounds records that count without visiting its rows, and a column of any type whose rows are all null skips its bounds loop. - Spark's own vectors are still counted row by row. Spark's vectorized Parquet reader marks a column missing from a file all-null without counting any nulls, so that vector's `numNulls` is 0 while every row `isNullAt`. Taking the count from it would record no nulls for a binary or nested column, and an `IS NULL` filter over the cache would then prune every batch read from that file. - The statistics cases in `CometInMemoryCacheBenchmark` now measure Comet's Arrow vectors as well as Spark's, plus binary and nested columns. `CometInMemoryCacheBenchmark --stats-only`, 5M row visits, best of three interleaved runs of each side (Apple M3 Max, JDK 17, Spark 4.1): | Case | `main` | This PR | | ----------------------------------------------------------------- | -----: | ------: | | 3 bigint and 3 string columns, Spark vectors | 128 ms | 126 ms | | The same columns, Arrow vectors | 152 ms | 155 ms | | Struct, array and binary columns, 1 row in 10 null, Arrow vectors | 27 ms | 0 ms | The flat columns are unchanged, as expected. Their loops still visit every row for the bounds, and the null check is a small part of each visit. I measured where that time goes on `main`, with three columns of one type at a time, Spark vectors then Arrow vectors: bigint 12 and 8 ms, double 12 and 16 ms, `decimal(10,2)` 44 and 47 ms, string 122 and 152 ms, and `decimal(38,2)` 283 and 324 ms, where every row builds a `BigInteger` and a `BigDecimal`. Wide decimals and strings are the next steps for this item. ## How are these changes tested? New `CometInMemoryCacheSuite` tests: - "statistics count nulls in columns that record no bounds": binary, array, map and struct columns, and a bigint column with no values, cached from the row write path (with a void column too), from Comet's native scan and from Spark's vectorized reader. - "statistics count the nulls of a column missing from a file": the missing-column case above, through Spark's vectorized reader. It checks both the recorded count and the `IS NULL` answer. - "statistics count nulls only within a batch's rows": a batch that reports fewer rows than its vectors hold, whose count must not come from the vectors. Each test fails against a mutant of the code it guards. Taking the count from every vector's `numNulls` makes the `IS NULL` query return 0 rows instead of 400. The other mutants drop the row-count check, or zero the count for columns without bounds or for all-null columns. The native-scan case catches the zeroed count by itself. `CometInMemoryCacheSuite`, `CometInMemoryCachePruningSuite` and `CometInMemoryCacheKryoSuite` pass locally on Spark 3.4, 3.5, 4.0 and 4.1. -- 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]
