Replace default group-by combine with non-blocking per-thread operator (CONCURRENT fallback)

[xiangfu0]

Summary

Replace the default group-by combine operator with a non-blocking implementation that uses per-thread SimpleIndexedTables instead of a shared ConcurrentHashMap, eliminating cross-thread contention.

Motivation

The existing GroupByCombineOperator uses a single ConcurrentIndexedTable backed by ConcurrentHashMap shared across all worker threads. Under high concurrency this creates significant lock contention — benchmarks show performance actually degrades as threads increase (e.g., 4→8 threads makes queries slower, not faster). The non-blocking approach gives each thread its own SimpleIndexedTable, merging them at the end, which scales linearly with thread count.

How the non-blocking combine works

Old approach (CONCURRENT):

Thread 1 ──┐
Thread 2 ──┤──► shared ConcurrentHashMap (contention on every upsert)
Thread 3 ──┤
Thread 4 ──┘

New approach (default, NON-BLOCKING):

Thread 1 ──► local HashMap₁ ──┐
Thread 2 ──► local HashMap₂ ──┤──► lock-free pairwise merge
Thread 3 ──► local HashMap₃ ──┤
Thread 4 ──► local HashMap₄ ──┘

Each worker thread creates its own SimpleIndexedTable (backed by plain HashMap, zero synchronization) and processes segments independently. When a thread finishes all its segments, it merges its local table with others via a lock-free exchange protocol:

Thread A finishes → sets _indexedTable = tableA           (no merge, just a swap)
Thread B finishes → takes tableA, merges tableB into it   (1 merge)
Thread C finishes → sets _indexedTable = tableC           (no merge)
Thread D finishes → takes tableC, merges tableD into it   (1 merge)
Thread B done     → takes tableD+C, merges into A+B       (1 merge)

The merge cost is small relative to processing because:

• With 8 threads and 32 segments, each thread processes ~4 segments. The merge adds ~7 × K_unique upserts on top of 32 × K_per_segment upserts — a small fraction of total work
• Merge upserts use uncontended HashMap.compute(), which is ~3-5x faster per call than ConcurrentHashMap.compute()
• With ORDER BY + LIMIT (the common case), trim keeps each local table at ~trimSize entries (e.g., 5000), so the merge is trivially small

Phase	Old CONCURRENT	New NON-BLOCKING (default)
Segment processing	Contended ConcurrentHashMap.compute()	Uncontended HashMap.compute()
Merge	N/A (already merged)	Lock-free pairwise merge
Memory	1 shared table	N_threads local tables (trimmed)
Scaling with threads	Worse (more contention)	Better (zero contention)

Approaches evaluated

We benchmarked several optimization strategies before settling on the non-blocking approach:

1. Partitioned group-by combine (PartitionedGroupByCombineOperator)
Hash-partitions group keys across independent tables with per-partition locks and parallel tree-reduction merge. Showed ~2x speedup at moderate cardinality (10K-100K records/segment) but hit severe GC pressure at high cardinality (1M+) due to N_threads × N_partitions SimpleIndexedTable instances. Removed in favor of the simpler non-blocking approach which scales better across all cardinalities.

2. fastutil Object2ObjectOpenHashMap replacing HashMap
Benchmarked open-addressing hash map vs JDK HashMap for the IndexedTable upsert hot path. Results (single-thread, GROUP BY d1, d2, d1×d2 cardinality up to 8M):

Records	HashMap (ms)	fastutil (ms)	Winner
10K	0.34	0.67	HashMap 2.0x faster
100K	13.5	16.4	HashMap 1.2x faster
500K	99.9	153.4	HashMap 1.5x faster

HashMap wins because: (a) JDK's compute() is heavily optimized, (b) chaining with tree bins degrades more gracefully than linear probing at high load, (c) fastutil's advantage is with primitive types (avoids boxing), not Object keys.

3. Cached Key.hashCode()
Pre-computed Arrays.hashCode() at Key construction instead of recomputing per lookup. No measurable benefit — within error margins at all sizes. The JDK likely already caches or the hash computation is not the bottleneck vs map probing/resizing.

4. Non-blocking per-thread combine (chosen approach)
Each worker thread builds its own SimpleIndexedTable (plain HashMap, no synchronization). After processing, threads merge via a lock-free exchange protocol. This eliminates all contention during the hot processing path.

Records/Segment	CONCURRENT (ms)	NON-BLOCKING (ms)	Speedup
10K	158	87	1.8x
100K	1,676	690	2.4x
500K	8,996	3,095	2.9x
1M	19,358	6,223	3.1x

(JMH, 32 segments, 8 threads, GROUP BY d1, d2 ORDER BY SUM(m1) DESC LIMIT 500, d1 cardinality=4096, d2 cardinality=2048)

The speedup increases with data volume (1.8x→3.1x) because contention on the shared ConcurrentHashMap grows with the number of keys being merged.

Key changes

• NonblockingGroupByCombineOperator: new default group-by combine operator that uses per-thread SimpleIndexedTables with lock-free merge at the end
• absorbTrimStats() on IndexedTable: propagates resize/trim statistics when merging tables so metrics remain accurate
• CombinePlanNode: routes all group-by queries (except safe-trim sorted path) to NonblockingGroupByCombineOperator by default
• Instance-level config: servers can set pinot.server.query.executor.group.by.algorithm=CONCURRENT to revert the default fleet-wide without modifying queries
• Per-query override: SET groupByAlgorithm='CONCURRENT' reverts to the legacy ConcurrentHashMap-based operator for a single query
• The sorted combine path (SortedGroupByCombineOperator / SequentialSortedGroupByCombineOperator) for safe-trim with small limits is preserved

Algorithm names

groupByAlgorithm value	Operator	Description
(unset, default)	NonblockingGroupByCombineOperator	Per-thread HashMap + lock-free merge
NON-BLOCKING	NonblockingGroupByCombineOperator	Same as default (explicit)
CONCURRENT	GroupByCombineOperator	Legacy shared ConcurrentHashMap (fallback)

Validation

./mvnw -pl pinot-core -am -Dtest=InterSegmentNonBlockingGroupBySingleValueQueriesTest,InterSegmentGroupBySingleValueQueriesTest,CombinePlanNodeTest,IndexedTableTest,GroupByTrimTest,ExplainPlanQueriesTest -Dsurefire.failIfNoSpecifiedTests=false test

[codecov-commenter]

Codecov Report

❌ Patch coverage is 72.69737% with 83 lines in your changes missing coverage. Please review.
✅ Project coverage is 57.09%. Comparing base (fae8080) to head (6d08e07).

Files with missing lines	Patch %	Lines
...tor/combine/PartitionedGroupByCombineOperator.java	69.17%	27 Missing and 18 partials ⚠️
.../core/operator/combine/GroupByCombineOperator.java	73.01%	13 Missing and 4 partials ⚠️
...tor/combine/NonblockingGroupByCombineOperator.java	71.42%	6 Missing and 4 partials ⚠️
.../java/org/apache/pinot/core/util/GroupByUtils.java	68.00%	2 Missing and 6 partials ⚠️
...pinot/core/plan/maker/InstancePlanMakerImplV2.java	71.42%	2 Missing ⚠️
...org/apache/pinot/core/data/table/IndexedTable.java	87.50%	0 Missing and 1 partial ⚠️

❗ There is a different number of reports uploaded between BASE (fae8080) and HEAD (6d08e07). Click for more details.

HEAD has 16 uploads less than BASE

Flag	BASE (fae8080)	HEAD (6d08e07)
java-21	5	1
unittests	2	1
temurin	5	1
unittests2	1	0
integration	3	0
integration1	1	0
integration2	1	0
custom-integration1	1	0

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #14698      +/-   ##
============================================
- Coverage     65.37%   57.09%   -8.29%     
+ Complexity     1405        1    -1404     
============================================
  Files          3423     2628     -795     
  Lines        215972   156564   -59408     
  Branches      34186    25586    -8600     
============================================
- Hits         141199    89387   -51812     
+ Misses        63399    59515    -3884     
+ Partials      11374     7662    -3712     

Flag	Coverage Δ
custom-integration1	?
integration	?
integration1	?
integration2	?
java-21	57.09% <72.69%> (-8.29%)	⬇️
temurin	57.09% <72.69%> (-8.29%)	⬇️
unittests	57.09% <72.69%> (-8.29%)	⬇️
unittests1	57.09% <72.69%> (+<0.01%)	⬆️
unittests2	?

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[ankitsultana]

@xiangfu0 : are you folks also planning to run some benchmarks? And any other ideas you are already trying out?

[wirybeaver]

@xiangfu0 Glad to see the master is working on the partitioned groupBy algorithm. Do we plan to support disk spilled mode? #12080

[xiangfu0]

Updated with commit fc14886. The partitioned combine path now uses thread-local per-partition tables plus parallel final reduction, which removes the 8-thread initialization race and improves the partitioned path on the same local synthetic workload from 106.341 ms to 77.933 ms at 4 threads (~27% faster). On 8 threads the previous implementation reproduced the NPE; the current code completes at 59.218 ms, and in the current 3-way comparison the partitioned path is 54.288 ms vs 68.945 ms for non-blocking and 150.688 ms for default.

[Copilot]

Pull request overview

Copilot reviewed 15 out of 15 changed files in this pull request and generated 1 comment.

[Copilot]

Pull request overview

Copilot reviewed 16 out of 16 changed files in this pull request and generated 2 comments.

[Copilot]

Pull request overview

Copilot reviewed 9 out of 9 changed files in this pull request and generated 5 comments.