Summary

Adds an opt-in non-blocking server-side GROUP BY combine operator. Each worker accumulates into an exclusively owned
SimpleIndexedTable, then the workers combine their tables through an atomic tournament that merges the smaller table
into the larger one.

The existing CONCURRENT implementation remains the server default and rollout gate. The non-blocking path is used
only for general GROUP BY queries with ORDER BY after an operator enables it in server configuration. Sorted
safe-trim operators still take precedence, and GROUP BY queries without ORDER BY use the shared concurrent table.

The experimental partitioned implementation was removed after review found correctness and peak-memory risks that
were not justified by its additional complexity.

Motivation

GroupByCombineOperator sends every worker's upserts through one shared ConcurrentIndexedTable. High-cardinality
queries spend substantial time contending on that table. The non-blocking implementation moves the hot accumulation
loop to ordinary per-worker HashMaps and performs a bounded reduction after segment processing.

Design

• Build one thread-confined SimpleIndexedTable per active combine worker.
• Exchange completed tables through an AtomicReference without a central merge lock.
• Merge smaller tables into larger tables to reduce repeated hash-map growth.
• Transfer existing Key and Record ownership while destructively draining source maps, reducing allocation and
  peak retained memory during reduction.
• Stop partially drained tournament merges after a worker failure and discard both exclusively owned tables.
• Avoid capturing map-update lambdas in the simple-table hot path.
• Reuse TableResizer candidate storage instead of allocating one wrapper per examined record.
• Preserve cancellation checks and aggregate trim/resize statistics across table merges.
• Preserve the legacy unordered-table insertion behavior used by downstream reducers.
• Keep CONCURRENT as the default because per-worker tables can require more peak memory.
• Prevent a query option from bypassing the server rollout gate.
• Route no-ORDER BY GROUP BY queries to CONCURRENT; worker-local LIMIT admission can otherwise discard partial
  aggregates before the global merge.

Rollout and usage

Enable the algorithm on a server after validating representative query cardinality, execution-thread count, and
memory headroom:

pinot.server.query.executor.group.by.algorithm=NON-BLOCKING

With that server gate enabled, a query can explicitly select the optimized path:

SET groupByAlgorithm = 'NON-BLOCKING';
SELECT d1, d2, SUM(m1), MAX(m2)
FROM testTable
GROUP BY d1, d2
ORDER BY SUM(m1) DESC
LIMIT 500;

A query can fall back to the shared implementation without changing server configuration:

SET groupByAlgorithm = 'CONCURRENT';
SELECT d1, d2, SUM(m1), MAX(m2)
FROM testTable
GROUP BY d1, d2
ORDER BY SUM(m1) DESC
LIMIT 500;

If the server remains configured as CONCURRENT, a query cannot opt itself into NON-BLOCKING.

Performance

JMH on JDK 25, 8 combine threads, 4 forks, and 3 measured iterations per fork. The harness explicitly rejects unknown
algorithm names and was invoked with -p '_algorithm=CONCURRENT,NON-BLOCKING'. The benchmark query is:

SELECT d1, d2, SUM(m1), MAX(m2)
FROM testTable
GROUP BY d1, d2
ORDER BY SUM(m1) DESC
LIMIT 500

Normalized allocation:

Validation

Targeted correctness coverage includes legacy and non-blocking inter-segment queries, trim behavior, cancellation,
rollout/default selection, a forced four-worker CAS tournament, deterministic partial-merge failure, no-ORDER BY
partial aggregates, and the gapfill regressions found by the JDK 25 CI shard.

TZ=UTC ./mvnw -pl pinot-core \
  -Dtest=GapfillQueriesTest,GapfillQueriesScalabilityTest,NonblockingGroupByCombineOperatorTest,CombinePlanNodeTest,IndexedTableTest,DeterministicIndexedTableTest,TableResizerTest,SortedGroupByCombineOperatorsTest,InstancePlanMakerImplV2Test,InterSegmentGroupBySingleValueQueriesTest,InterSegmentNonBlockingGroupBySingleValueQueriesTest \
  test

Result on Temurin 25: 247 tests passed.

Required module checks also pass:

./mvnw spotless:apply -pl pinot-common,pinot-core,pinot-spi,pinot-perf
./mvnw checkstyle:check -pl pinot-common,pinot-core,pinot-spi,pinot-perf
./mvnw license:format -pl pinot-common,pinot-core,pinot-spi,pinot-perf
./mvnw license:check -pl pinot-common,pinot-core,pinot-spi,pinot-perf

CI

All 11/11 required GitHub Actions checks passed on exact head bb9a6b212a2f63b607272a69a78c8f7ade5be7ad, including both unit shards, both integration shards, quickstart, linter, binary compatibility, and all four compatibility regressions.