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 #7235, which is stacked on #6853 and #6988, and holds all of their commits. The last two commits are the change here (
011718272e,0f01c13981). The rest are #7235's four commits, #6853's four and #6988's one. The diff shrinks to those two once the others land.Rationale for this change
ArrowCachedBatchSerializer.gatherColumnStatsrecords a lower and an upper bound for each string column of every cached batch. For each row,gatherStringStatscalledgetUTF8String, which allocates aUTF8String, and compared it through a boxedOrdering[Any]. That left strings as the most expensive column type in the statistics loop after the decimals fixed in #7235, and #6834 lists them as the next step.What changes are included in this PR?
StringType,gatherStringStatscompares the bytes in the vector's data buffer and builds noUTF8Stringper row. The comparison is that ofUTF8String.compareTo: eight bytes at a time as big-endian unsigned words, then byte by byte as unsigned values, and a shorter string sorts before a longer one it prefixes. A row is compared with the lower bound first, and with the upper bound only when it is not below it. Only the rows of the two bounds becomeUTF8Strings, as owned copies.gatherStringStatsByValue, and so keep comparing through the orderingboundsOrderingsresolved for the column. The default collation is recognised bydataType == StringType, which a collatedStringTypeis not equal to, so no version shim is needed.How are these changes tested?
Four new tests in
ArrowCachedBatchStatsSuitecompare the bounds with the minimum and maximum of the strings' UTF-8 bytes under an unsigned byte-wise order:Each of these fails if the comparison is changed: treating bytes as signed, comparing the words in native byte order, or dropping the tie-break on length.
CometInMemoryCacheSuite,CometInMemoryCacheKryoSuiteandArrowCachedBatchStatsSuitepass on Spark 4.1 (107 tests), and the stats and Kryo suites pass on 3.5 and 3.4. Scalafix and the strict-warnings build pass.Benchmark (
CometInMemoryCacheBenchmark --stats-only, Apple M3 Max,local[1], 5M rows, Arrow vectors), run before, after, before in one session so that drift would show:The decimal case did not move, so the machine was steady between the runs. The string columns' share of the first case is not measured separately.