Skip to content

compute: consolidate operator output only while it recovers records - #38674

Open
frankmcsherry wants to merge 1 commit into
MaterializeInc:mainfrom
frankmcsherry:adaptive-consolidation
Open

compute: consolidate operator output only while it recovers records#38674
frankmcsherry wants to merge 1 commit into
MaterializeInc:mainfrom
frankmcsherry:adaptive-consolidation

Conversation

@frankmcsherry

Copy link
Copy Markdown
Contributor

Several render operators emit through differential's ConsolidatingContainerBuilder, which sorts and consolidates every output chunk of about a thousand records before it leaves the operator. That collapses repeated (data, time) pairs before they are exchanged or arranged, a large saving when the data has few distinct keys. When it has many, the sort recovers nothing: a GROUP BY over 1.25M groups spread across 10M rows merges under one percent of each chunk, and the sort was the largest cost of the reduce's key formation and of the accumulator explosion.

AdaptiveConsolidatingContainerBuilder (in mz-timely-util) keeps differential's buffering and measures what each consolidation recovers, as an exponential moving average in permille. While that stays below two percent it passes chunks through unsorted, consolidating one chunk in every thirty-two so a change in the data is noticed within that many chunks. Consumers never relied on the output being consolidated or ordered, so nothing downstream changes.

It replaces the differential builder at four sites: the reduce's key formation, explode_one (the accumulable reduce's accumulator explosion), the MFP flat_map on collections, and the FlatMap operator's Ok output. Error outputs and the join closures keep the differential builder.

Hydration at 10M rows, one worker, median of three, on a local build with the other hydration PRs in flight applied:

shape before after operators
SELECT k % g, count(*), sum(v) ... GROUP BY 1 (1.25M groups) 4.42s 3.81s ExplodeOne 919 to 514ms, CollectionFlatMap 1233 to 968ms, ArrangeAccumulable 1073 to 1097ms
max(v) by k % g with AGGREGATE INPUT GROUP SIZE = 8 4.50s 4.28s CollectionFlatMap 1166 to 916ms
SELECT DISTINCT k, v 3.01s 2.93s
index on a table 2.00s 1.95s

The downstream arrangements pay slightly more for unconsolidated input (ArrangeAccumulable +24ms); the sorts they no longer need paid far more.

Unit tests cover the three regimes: repeated keys stay consolidated, distinct keys stop the sort without losing records, and a probe resumes consolidation when keys start repeating.

@frankmcsherry
frankmcsherry requested a review from a team as a code owner September 5, 2026 00:21
@frankmcsherry
frankmcsherry force-pushed the adaptive-consolidation branch 2 times, most recently from e6373c4 to c2bfe82 Compare September 5, 2026 00:23
The reduce's key formation, explode_one, the MFP flat_map and FlatMap emit
through a builder that sorts and consolidates every output chunk. That
collapses repeated keys before the exchange when there are few distinct
keys, and recovers nothing when there are many. Add a builder that measures
what each consolidation recovers and passes chunks through unsorted while it
stays under two percent, probing one chunk in thirty-two.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant