Repository navigation
Conversation
gatherColumnStats visited every row to count nulls, including for binary, array, map and struct columns, which record a null count and no bounds. A CometDecodedVector already holds that count, taken from Arrow when the vector is built. Use it when the vector holds exactly the batch's rows: a column without bounds records it without a pass over its rows, and an all-null column of any type skips its bounds loop. Spark's own vectors are still counted row by row. Its vectorized Parquet reader marks a column missing from a file all-null without counting any nulls, so numNulls is 0 there while every row isNullAt. The statistics benchmark now measures Comet's Arrow vectors as well as Spark's, and binary and nested columns. Part of apache#6834.
Spark's DefaultCachedBatch records the same null counts and no bounds for binary, nested and void columns. nullCountsWithoutBounds now checks that the relation is stored as CometCachedBatch before reading its statistics, so the row write path and native scan legs check the format as the Spark vectorized reader leg already did.
encodeBatches now converts a batch that is not written directly before it gathers the statistics, so every column the collector reads is a CometDecodedVector holding the null count Arrow took when it was built. The row-by-row null count for Spark's vectors and the check that a vector holds exactly the batch's rows go, and so do the test of that check and its helper. The statistics benchmark measures only Arrow vectors, which is all the collector reads now.
The columnar cache write stored each input batch as one cached batch, and nothing bounds how small those batches are. A shuffle read returns at least one batch per map output block, so a relation cached straight from a shuffle over 200 map tasks into 50 partitions held 10000 batches of about 500 rows, each with its own message metadata, compression framing and statistics row. CacheBatchCoalescer now sits in front of the columnar write. An Arrow-backed batch laid out the way the reader reads it is passed through when it holds at least half of spark.sql.inMemoryColumnarStorage.batchSize, so full native batches are still written without a copy. Every other batch is copied through ArrowWriter.writeColumns into batches of batchSize rows, the size the row path cuts at: a small batch is worth the copy, and a batch of Spark column vectors was already copied into Arrow. Row order is kept, and each batch is copied before the input advances, because a native producer releases the batch it returned last when asked for the next. Closes apache#6950.
The cache statistics now come from the Arrow batch that is written, so a vector longer than its batch would count nulls past the batch's last row and could record the batch all null. Require each vector to hold exactly the batch's rows to take the direct path, and convert otherwise.
…s' into perf/cache-stats-wide-types
…o perf/cache-stats-wide-types
2 of 13 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Part of #6834.
This PR is stacked on #6853 and #6988, and holds their commits as well. The last four commits are the change here (
e30bc831b9,8a9394fdaa,f20e6e123e,158e105a71). The rest are #6853's four commits and #6988's one. The diff shrinks to those four once both land.Rationale for this change
ArrowCachedBatchSerializer.gatherColumnStatsrecords a lower and an upper bound for each decimal column of every cached batch. For each row,gatherDecimalStatscalledCometVector.getDecimal, which for a precision past 18 digits builds aBigIntegerand aBigDecimalfrom the 16 bytes in Arrow's data buffer, and then aDecimalaround them. Most of the statistics time of a decimal column was that allocation. #6834 lists it as the next step after the null counts of #6853.What changes are included in this PR?
gatherDecimalStatsreads a plain Arrow decimal column's 128-bit unscaled values straight from its data buffer and compares them without building anything per row. A value is a signed high word over an unsigned low word, little-endian. A narrow decimal is the sign extension of its low word, so one comparison covers every precision. All the values of a column share one scale, so their unscaled order is their numeric order. Only the rows of the two bounds become aDecimal, through the samegetDecimalas before, so the bounds keep the type they had.gatherDecimalStatsByValue.CometInMemoryCacheBenchmarkgets a decimal case in its statistics benchmark: precisions 10, 18 and 38 at scale 2, one row in ten null, and values on both sides of zero.How are these changes tested?
New
ArrowCachedBatchStatsSuitebuilds ArrowDecimalVectors at precisions 5, 10, 18, 19, 20, 30 and 38 with a range of scales, and compares the bounds and null count with aBigDecimalminimum and maximum:Comparing the low word as signed instead of unsigned fails the suite at 19, 20, 30 and 38 digits.
CometInMemoryCacheSuite,CometInMemoryCacheKryoSuiteand the new suite pass on Spark 4.1 (103 tests). The new suite and the Kryo suite also pass on 3.5 and 3.4.Benchmark (
CometInMemoryCacheBenchmark --stats-only, Apple M3 Max,local[1], 5M rows of the three decimal columns above, Arrow vectors), run old, new, old in one session so that drift would show:The long and string cases of the same benchmark did not move (160, 160 and 161 ms), so the machine was steady between the runs.