feat(PartitionedOutput): Partition shared vectors only once - #2487
Open
xin-zhang2 wants to merge 38 commits into
Open
xin-zhang2 wants to merge 38 commits into
xin-zhang2 wants to merge 38 commits into
Conversation
This commit introduces `PartitionedVector` - a low-level execution abstraction that provides an in-place, partition-aware layout of a vector based on per-row partition IDs. 1. **In-place rearrangement**: Rearrange vector data in memory without creating multiple copies 2. **Buffer reuse**: Allow reuse of temporary buffers across multiple partitioning operations 3. **Minimal abstraction**: Similar to `DecodedVector`, focus on efficient execution rather than operator semantics 4. **Thread-unsafe by design**: Optimized for single-threaded execution contexts For more information please see IBM#1703 Alchemy-item: (ID = 1150) Introducing PartitionedVector commit 1/1 - 960f41b Alchemy-item: (ID = 1327) Optimized PartitionedOutput staging hub commit 1/11 - 76dc41a Alchemy-item: (ID = 1596) Optimized PartitionedOutput staging hub commit 1/11 - 2c59ee6 Alchemy-item: (ID = 1682) Optimized PartitionedOutput staging hub commit 1/11 - 23e8d1d Alchemy-item: (ID = 1803) Optimized PartitionedOutput staging hub commit 1/11 - c576556
Signed-off-by: Xin Zhang <xin-zhang2@ibm.com> Alchemy-item: (ID = 1167) Add PartitionedRowVector commit 1/1 - f2af427 Alchemy-item: (ID = 1327) Optimized PartitionedOutput staging hub commit 2/11 - 3853bf6 Alchemy-item: (ID = 1596) Optimized PartitionedOutput staging hub commit 2/11 - 71705c7 Alchemy-item: (ID = 1682) Optimized PartitionedOutput staging hub commit 2/11 - 7df0be4 Alchemy-item: (ID = 1803) Optimized PartitionedOutput staging hub commit 2/11 - b887d69
…dthValuesInPlace Alchemy-item: (ID = 1327) Optimized PartitionedOutput staging hub commit 3/11 - ff2e34b Alchemy-item: (ID = 1596) Optimized PartitionedOutput staging hub commit 3/11 - 3d9e709 Alchemy-item: (ID = 1682) Optimized PartitionedOutput staging hub commit 3/11 - 5719f90 Alchemy-item: (ID = 1803) Optimized PartitionedOutput staging hub commit 3/11 - 4732eb6
Alchemy-item: (ID = 1327) Optimized PartitionedOutput staging hub commit 4/11 - 875c92c Alchemy-item: (ID = 1596) Optimized PartitionedOutput staging hub commit 4/11 - d787419 Alchemy-item: (ID = 1682) Optimized PartitionedOutput staging hub commit 4/11 - fad8064 Alchemy-item: (ID = 1803) Optimized PartitionedOutput staging hub commit 4/11 - 1b947dc
PartitionedFlatVector::partition() and PartitionedRowVector::partition() called mutableRawNulls() unconditionally. mutableRawNulls() allocates a null buffer if one does not exist, causing mayHaveNulls() to return true for every vector after partitioning, even when the original had no nulls. Fix both sites to check rawNulls() first and only call mutableRawNulls() when a null buffer already exists. Add noNullBufferAllocatedForNullFreeFlat and noNullBufferAllocatedForNullFreeRow tests to PartitionedVectorTest to cover this case. # Conflicts: # velox/vector/PartitionedVector.cpp Alchemy-item: (ID = 1327) Optimized PartitionedOutput staging hub commit 5/11 - 281a365 Alchemy-item: (ID = 1596) Optimized PartitionedOutput staging hub commit 5/11 - 652dd0a Alchemy-item: (ID = 1682) Optimized PartitionedOutput staging hub commit 5/11 - 04fbad0 Alchemy-item: (ID = 1803) Optimized PartitionedOutput staging hub commit 5/11 - 4dcce78
Alchemy-item: (ID = 1327) Optimized PartitionedOutput staging hub commit 6/11 - 6519a8f Alchemy-item: (ID = 1596) Optimized PartitionedOutput staging hub commit 6/11 - 28c45bd Alchemy-item: (ID = 1682) Optimized PartitionedOutput staging hub commit 6/11 - c3a52c9 Alchemy-item: (ID = 1803) Optimized PartitionedOutput staging hub commit 6/11 - d57e0e3
Alchemy-item: (ID = 1327) Optimized PartitionedOutput staging hub commit 7/11 - d8f34b4 Alchemy-item: (ID = 1596) Optimized PartitionedOutput staging hub commit 7/11 - 59b321a Alchemy-item: (ID = 1682) Optimized PartitionedOutput staging hub commit 7/11 - 9efe82a Alchemy-item: (ID = 1803) Optimized PartitionedOutput staging hub commit 7/11 - a4fa81e
Alchemy-item: (ID = 1327) Optimized PartitionedOutput staging hub commit 8/11 - 9eafc9d Alchemy-item: (ID = 1596) Optimized PartitionedOutput staging hub commit 8/11 - b27c492 Alchemy-item: (ID = 1682) Optimized PartitionedOutput staging hub commit 8/11 - ff888e7 Alchemy-item: (ID = 1803) Optimized PartitionedOutput staging hub commit 8/11 - 91c0156
This commit introduces PrestoIterativePartitioningSerializer, which buffers RowVectors across multiple append() calls, partitions rows in-place using PartitionedVector, and on flush() serializes each non-empty partition into a Presto wire-format IOBuf. The serializer has no dependency on velox_exec: it returns raw folly::IOBuf objects, leaving SerializedPage creation to the caller. Alchemy-item: (ID = 1327) Optimized PartitionedOutput staging hub commit 9/11 - 6f09ea9 Alchemy-item: (ID = 1596) Optimized PartitionedOutput staging hub commit 9/11 - 48018b3 Alchemy-item: (ID = 1682) Optimized PartitionedOutput staging hub commit 9/11 - 1511c01 Alchemy-item: (ID = 1803) Optimized PartitionedOutput staging hub commit 9/11 - 9e82e6c
This commit introduces OptimizedPartitionedOutput, a PartitionedOutput operator backed by PrestoIterativePartitioningSerializer. Enabled via query config key "optimized_repartitioning" (default off). LocalPlanner selects it over the standard PartitionedOutput when the flag is set. TODO: replicateNullsAndAny is not yet supported and raises a user error.
…geBenchmark - Added normal vs optimized PartitionedOutput comparison by running each exchange case twice with kOptimizedPartitionedOutputEnabled=false/true. - Added per-mode benchmark names: - exchange<Case>_normalPartitionedOutput - exchange<Case>_optimizedPartitionedOutput in ExchangeBenchmark.cpp. - Refactored result printing into shared helpers and fixed output consistency in ExchangeBenchmark.cpp.
…mark Split the local partition exchange benchmark out of ExchangeBenchmark into its own executable and CMake target, while keeping the local benchmark logic and statistics reporting available in a dedicated binary.
…tioningSerializer
…fferManager listeners Pass an OutputBufferManager-backed listener factory into PrestoIterativePartitioningSerializer so the optimized path uses the same listener source as normal PartitionedOutput. Create per-partition listeners during flush, set the checksum bit only when a listener is present, and compute the page checksum only for PrestoOutputStreamListener instances. Also add tests that verify checksum headers are written and that the serialized pages round-trip through the standard deserializer.
…rting - add explicit simple-schema benchmark cases by type and column count - register normal and optimized runs as separate named benchmark cases - make `dictPct` apply per generated vector and recurse into nested types - generate benchmark input vectors directly with optional nulls - replace ad hoc flat input generation with explicit input specs - return `ExchangeRunStats` from benchmark runs and centralize query config - group printed results by dataset with normal vs. optimized stats
The new OptimizedVectorHasher is up to 2-3x faster than VectorHasher.
BatchMaker always allocates a null buffer even when no rows are null. This commit removes it so benchmarks measure the non-nullable path faithfully. Plus some minor format cleanups.
Currently Velox never passes CMAKE_BUILD_TYPE into Folly's own configure
step, while cmake_install only forwards arbitrary caller flags, so Folly
was not built in release mode when Velox is built in release mode. This
commit adds -DCMAKE_BUILD_TYPE="${CMAKE_BUILD_TYPE}" to FOLLY_FLAGS, so
a release Velox dependency setup now builds release Folly on macOS and
Linux.
- Remove benchmarks with vector size 1000000 and only keep vector size 10000. - Add tinyint and smallint benchmarks.
Introduce OptimizedHashPartitionFunction as a faster drop-in replacement for HashPartitionFunction, gated behind a new query config flag optimized_hash_partition_function_enabled (default false). partition() is improved from 50% to over 200x. Add HashPartitionFunctionBase as a common base exposing numPartitions(), and createHashPartitionFunction() factories that select the implementation based on the flag. Thread QueryConfig* through PartitionFunctionSpec::create() and update callsites (LocalPartition, PartitionedOutput, MarkDistinct, RowNumber, Window, SubPartitionedSortWindowBuild, HiveConnector) to construct partition functions via the factory. Register CMake targets for the new test and benchmark binaries.
…ormance
The old in-place partitioning followed cycles of swaps,
values[cursor[targetPartition]++] swapped with values[offset];
targetPartition = partitions[cursor[targetPartition] - 1];
so the next partition id depended on the previous swap's store. The
loop ran at one row per cache-latency cycle and could not be
vectorized.
Replace it with a contiguous scatter,
output[cursor[partitions[i]]++] = input[i];
into a scratch buffer (ctx.tempBuffer), then memcpy the scratch back
into the caller's buffer. The per-row chain through partitions[] is
broken: rows targeting distinct partitions have independent address
computations and can issue in parallel. The scatter loop is unrolled
by 4 and marked __restrict to expose that ILP, and to let the compiler
hoist the partition / cursor loads above the stores instead of
assuming aliasing.
The bool path takes the same shape via scatterBitsInPlace: a
sequential bit scatter into a scratch byte buffer, then memcpy back.
The old per-bit swap had the same data dependency as the value swap.
Benchmarks (release, M1 macOS, noNulls, time/iter, partitionedVectorBenchmark):
case before after speedup
---- ------ ----- -------
BOOLEAN_1Cols_P64 37.18us 11.41us 3.26x
BOOLEAN_10Cols_P64 338.63us 82.52us 4.10x
INTEGER_1Cols_P64 30.30us 9.37us 3.23x
INTEGER_10Cols_P64 267.21us 59.06us 4.52x
BIGINT_1Cols_P64 31.08us 10.26us 3.03x
BIGINT_10Cols_P64 281.92us 70.45us 4.00x
HUGEINT_1Cols_P64 32.61us 12.76us 2.56x
HUGEINT_10Cols_P64 280.05us 97.02us 2.89x
VARCHAR_1Cols_P64 30.46us 12.93us 2.36x
VARCHAR_10Cols_P64 280.05us 98.45us 2.84x
Mixed_1Cols_P64 37.17us 11.45us 3.25x
Mixed_10Cols_P64 281.84us 70.66us 3.99x
Across the full sweep (BOOLEAN/INTEGER/BIGINT/HUGEINT/VARCHAR/Mixed
x {1, 10} cols x {4, 16, 64, 256} partitions), the new path is
1.58x to 5.06x faster than the in-place version, with the largest
wins on narrow types and 10-column inputs where the unrolled scalar
scatter delivers the most parallelism.
The previous commit copied ctx.tempBuffer back into the caller's buffer on every partition() call. That memcpy scales with element width and is the dominant per-call cost for wide fixed-width types — ~40% overhead for HUGEINT/VARCHAR on the 10-column benchmark. BOOLEAN was unaffected because its bit path already memcpys via scatterBitsInPlace. Remove the memcpy by switching ctx.tempBuffer back to a std::swap. Cross-pool ownership — the reason the swap was originally given up — is avoided by allocating ctx.tempBuffer from inputBuffer->pool() instead of the operator's 'pool' argument. The swapped-in buffer then lives in the same pool the caller's FlatVector already belonged to, so the operator's pool sees no foreign reference at teardown. ctx.tempBuffer is reused across columns of the same input (which share a pool); when the input's pool changes between calls, the cached buffer is reset so the next allocation comes from the new pool. The setSize before the swap is also added — without it, FlatVector::slice() can read a stale size from a reused tempBuffer and underflow numValues = size / sizeof(T) to 0 when a narrower column preceded a wider one. Benchmarks (release, M1 macOS, noNulls, time/iter, partitionedVectorBenchmark): 10-column input, 64 partitions: type memcpy swap+inputPool delta BOOLEAN 82.52us 82.35us -0.2% INTEGER 59.06us 54.06us -8.5% BIGINT 70.45us 57.79us -18.0% HUGEINT 97.02us 68.54us -29.4% VARCHAR 98.45us 70.00us -28.9% Mixed 70.66us 60.93us -13.8% 1-column input, 64 partitions: type memcpy swap+inputPool delta INTEGER 9.37us 8.91us -4.9% BIGINT 10.26us 9.03us -12.0% HUGEINT 12.76us 9.95us -22.0% VARCHAR 12.93us 10.23us -20.9% The win tracks the byte count per element: every extra byte per element is an extra byte the previous memcpy had to write. Tests: - velox_vector_test --gtest_filter='*Partition*' (44 passed) - velox_serializer_test_PrestoIterativePartitioningSerializerTest (56 passed) - velox_exec_test_OptimizedPartitionedOutputTest (301 passed)
1) Skip running hash function on all rows if the input is constant vector 2) Skip partition array fill for single-partition vectors OptimizedHashPartitionFunctionBenchmark Before ---------------------------------------------------------------------------- partition_bool_remote_p16_constant_all_null 7.15us 139.87K optimized_partition_bool_remote_p16_constant_al 349.51% 2.05us 488.84K ---------------------------------------------------------------------------- partition_bool_remote_p100_constant_no_null 7.14us 139.97K optimized_partition_bool_remote_p100_constant_n 190.04% 3.76us 265.99K ---------------------------------------------------------------------------- OptimizedHashPartitionFunctionBenchmark After ---------------------------------------------------------------------------- partition_bool_remote_p16_constant_all_null 7.15us 139.82K optimized_partition_bool_remote_p16_constant_al 15456.% 46.28ns 21.61M ---------------------------------------------------------------------------- partition_bool_remote_p100_constant_no_null 7.14us 140.12K optimized_partition_bool_remote_p100_constant_n 14819.% 48.16ns 20.76M ---------------------------------------------------------------------------- Other data types show the same pattern. ExchangeBenchmark After (_c100 means 100% of the vectors are constant vectors) ============================================================================ [...]exec/benchmarks/ExchangeBenchmark.cpp relative time/iter iters/s ============================================================================ 10K_Integer_col1_d0_c100_n0_normal 8.88ms 112.65 10K_Integer_col1_d0_c100_n0_optimized 296.25% 3.00ms 333.74 10K_Integer_col1_d0_c100_n100_normal 9.48ms 105.46 10K_Integer_col1_d0_c100_n100_optimized 553.59% 1.71ms 583.81 ---------------------------------------------------------------------------- 10K_Integer_col4_d0_c100_n0_normal 17.75ms 56.32 10K_Integer_col4_d0_c100_n0_optimized 224.66% 7.90ms 126.54 10K_Integer_col4_d0_c100_n100_normal 16.80ms 59.52 10K_Integer_col4_d0_c100_n100_optimized 318.17% 5.28ms 189.37 ---------------------------------------------------------------------------- 10K_Integer_col16_d0_c100_n0_normal 55.97ms 17.87 10K_Integer_col16_d0_c100_n0_optimized 180.95% 30.93ms 32.33 10K_Integer_col16_d0_c100_n100_normal 45.23ms 22.11 10K_Integer_col16_d0_c100_n100_optimized 474.32% 9.53ms 104.88 ----------------------------------------------------------------------------
…in verification The test computed expected per-partition row assignments with HashPartitionFunction, but OptimizedPartitionedOutput routes rows using OptimizedHashPartitionFunction. The two hash functions produce different partition assignments, so every parameterized case whose verification relied on the partition mapping was failing. Switch the test to HashPartitionFunctionSpec::create() with useOptimizedPartitionFunction=true, matching the operator's own construction path.
Move the HashPartitionFunctionSpec::create() call into a protected helper on OptimizedPartitionedOutputTest so verifyDataIntegrity and duplicateOutputColumns share the same construction path.
…titioningSerializer
`PrestoIterativePartitioningSerializer` can now serialize struct columns, including structs nested inside structs, with nulls at any level. Serializing a struct column does not write into the vector handed to `append()`. Given `ROW(ROW(ROW(INTEGER)))` where the outer struct is null at row 0 and the inner struct is null at row 2, combining the rows the outer struct discards back into the caller's own null bitmap would leave the caller's inner struct reporting two nulls instead of one, so the mask that tells a child which rows survive is built separately: only when a level really drops rows, and in a buffer of its own when both the level and an ancestor drop rows. A struct whose null bitmap is allocated but holds no null does not count as having nulls. Such a bitmap drops no row, so treating it as nullable would reject `ROW(ROW(VARCHAR))` with "Variable-width columns nested under a ROW with nulls are not supported" and push fixed-width leaves off the bulk copy path onto one stream write per value. A struct column can arrive CONSTANT-encoded, which produces a `PartitionedConstantVector`; that is reported as an unsupported encoding rather than relying on a cast that returns null. The ROW block footer reuses the same null-section writer as simple columns instead of repeating the Velox-to-Presto bitmap conversion. Also adds nested-ROW coverage to the benchmarks: the `PartitionedVector` type generator was off by one nesting level, so its one-level case built a plain BIGINT column, and `ExchangeBenchmark` reused a single null window at every level, so no level ever discarded a row its child would otherwise have written. An unused `PartitionedVector` setter and a test that asserted nothing are removed.
…itioningSerializer
xin-zhang2
force-pushed
the
PartitionedOutput-aliased-children
branch
from
September 1, 2026 13:44
97a2681 to
00ffd83
Compare
unidevel
force-pushed
the
optimized_partitionedoutput
branch
from
September 16, 2026 18:39
1912dee to
c4129dd
Compare
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
2 participants
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.
No description provided.