xiangfu0 commented on code in PR #19825:
URL: https://github.com/apache/pinot/pull/19825#discussion_r4244083976
##########
pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java:
##########
@@ -108,6 +109,29 @@ public boolean upsert(Record record) {
return upsert(new Key(keyValues), record);
}
+ /// Upserts an already-extracted result into a table with exactly one
aggregation function.
+ /// The table may retain the key and result; callers must not modify them
after this call.
+ /// A null result follows the aggregation function's existing merge
semantics.
+ public boolean upsertSingleAggregationResult(Key key, @Nullable Object
result) {
Review Comment:
[pre-existing, surfaced by this PR's new entry point] The new
`upsertSingleAggregationResult` routes into `addOrUpdateRecord` for
`DeterministicConcurrentIndexedTable`, whose `_lookupMap` is a
`ConcurrentSkipListMap`. Unlike `ConcurrentHashMap`, `CSLM.compute` invokes the
remapping function without any per-key lock and commits via CAS on the value
reference. Since `updateRecord`/`updateSingleAggregationRecord` mutate the
existing `Record`'s values array in place and return the same reference, the
CAS trivially succeeds for every racer: two combine threads merging the same
group key can both read `values[i]` (e.g. count 5), both write 6, and one
segment's contribution is silently lost. Trigger:
`accurateGroupByWithoutOrderBy` query option with a multi-threaded combine. Not
introduced by this PR (the old `upsert(Key, Record)` path has the same race),
but worth fixing since this table's whole purpose is accuracy; the in-place
merge needs a lock or a copy-on-write record for CSLM.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java:
##########
@@ -108,6 +109,29 @@ public boolean upsert(Record record) {
return upsert(new Key(keyValues), record);
}
+ /// Upserts an already-extracted result into a table with exactly one
aggregation function.
+ /// The table may retain the key and result; callers must not modify them
after this call.
+ /// A null result follows the aggregation function's existing merge
semantics.
+ public boolean upsertSingleAggregationResult(Key key, @Nullable Object
result) {
+ assert _aggregationFunctions.length == 1;
Review Comment:
The single-aggregation precondition is enforced only by `assert` (here and
in the `ConcurrentIndexedTable` override), which vanishes under the production
default `-da`. If a future caller invokes this on a table with N>1 aggregation
functions, `createSingleAggregationRecord` silently builds a record of length
`_numKeyColumns + 1` that is shorter than the table's schema: a later merge via
`updateRecord` throws `ArrayIndexOutOfBoundsException` far from the cause, or
serialization reads a missing column (silently wrong results). A cheap hard
check fails fast at the call boundary:
```suggestion
Preconditions.checkState(_aggregationFunctions.length == 1,
"upsertSingleAggregationResult requires exactly one aggregation
function");
```
##########
pinot-core/src/main/java/org/apache/pinot/core/data/table/ConcurrentIndexedTable.java:
##########
@@ -48,6 +50,26 @@ public boolean upsert(Key key, Record record) {
return true;
}
+ @Override
+ public boolean upsertSingleAggregationResult(Key key, @Nullable Object
result) {
+ assert _aggregationFunctions.length == 1;
+ if (_hasOrderBy) {
+ return super.upsertSingleAggregationResult(key, result);
+ }
+ BiFunction<Key, Record, Record> mergeResult = (k, record) -> record == null
+ ? createSingleAggregationRecord(k, result)
+ : updateSingleAggregationRecord(record, result);
+ if (_noMoreNewRecords.get()) {
Review Comment:
This block re-implements `upsertWithoutOrderBy`'s admission control (the
`_noMoreNewRecords` pre-check, the `computeIfPresent` vs `compute` split, and
the `size() >= _resultSize` flag transition) as a parallel copy. Any future fix
to the cap logic (changing `>=`, counting dropped groups for a trim metric,
fixing the size-check race) now has to be made twice, or single-agg and
multi-agg queries silently admit different groups. Consider a shared helper
taking the `BiFunction` (e.g. `upsertWithoutOrderBy(Key, BiFunction<Key,
Record, Record>)`) that both paths call. Related: a subclass overriding
`upsertWithoutOrderBy` to customize unordered admission (the pattern
`UnboundedConcurrentIndexedTable` establishes for ordered upserts) would be
silently bypassed on the single-agg path.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java:
##########
@@ -108,6 +109,29 @@ public boolean upsert(Record record) {
return upsert(new Key(keyValues), record);
}
+ /// Upserts an already-extracted result into a table with exactly one
aggregation function.
+ /// The table may retain the key and result; callers must not modify them
after this call.
+ /// A null result follows the aggregation function's existing merge
semantics.
+ public boolean upsertSingleAggregationResult(Key key, @Nullable Object
result) {
+ assert _aggregationFunctions.length == 1;
+ return upsert(key, createSingleAggregationRecord(key, result));
+ }
+
+ protected Record createSingleAggregationRecord(Key key, @Nullable Object
result) {
+ Object[] values = Arrays.copyOf(key.getValues(), _numKeyColumns + 1);
+ values[_numKeyColumns] = result;
+ return new Record(values);
+ }
+
+ protected Record updateSingleAggregationRecord(Record record, @Nullable
Object result) {
Review Comment:
`updateSingleAggregationRecord` duplicates `updateRecord`'s merge dispatch
(`_hasFinalInput ? mergeFinalResult : merge`, same `_numKeyColumns` indexing,
same `Comparable` casts) specialized to one column. If the merge contract
changes (null handling, a new input mode, layout change), both methods must
change together or the single-agg and multi-agg paths merge differently.
Extracting a shared per-column helper, e.g. `Object mergeValue(int funcIdx,
Object existing, Object input)`, used by both removes the drift risk.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java:
##########
@@ -108,6 +109,29 @@ public boolean upsert(Record record) {
return upsert(new Key(keyValues), record);
}
+ /// Upserts an already-extracted result into a table with exactly one
aggregation function.
+ /// The table may retain the key and result; callers must not modify them
after this call.
+ /// A null result follows the aggregation function's existing merge
semantics.
+ public boolean upsertSingleAggregationResult(Key key, @Nullable Object
result) {
+ assert _aggregationFunctions.length == 1;
+ return upsert(key, createSingleAggregationRecord(key, result));
Review Comment:
The deferred-allocation benefit only reaches
`UnboundedConcurrentIndexedTable`. For a no-order-by combine,
`GroupByUtils.getTrimDisabledIndexedTable` can also return `SimpleIndexedTable`
(numThreads == 1) and `DeterministicConcurrentIndexedTable`, which fall into
this base implementation that still does `Arrays.copyOf` + `new Record` per
group key per segment -- exactly the allocation the PR removes, now plus an
extra call. `HashMap` and `ConcurrentSkipListMap` both support `compute`, so
the lazy create/merge could live in base-class helpers (mirroring
`addOrUpdateRecord`/`updateExistingRecord`) and cover all three table types.
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java:
##########
@@ -77,6 +78,8 @@ public GroupByCombineOperator(List<Operator> operators,
QueryContext queryContex
assert _queryContext.getGroupByExpressions() != null;
_numKeyColumns = _queryContext.getNumGroupByKeyColumns();
_numColumns = _numKeyColumns + _numAggregationFunctions;
+ _useSingleAggregationResult = _numAggregationFunctions == 1
Review Comment:
This caller-side gate duplicates dispatch the table already owns:
`ConcurrentIndexedTable.upsertSingleAggregationResult` falls back to the
generic path when `_hasOrderBy`, and the base implementation is
behavior-identical to the eager loop for any single-aggregation query. As a
result the `_hasOrderBy` fallback branch in the override is production-dead
(reachable only from the new test), and "when is the scalar path valid" is now
encoded in two places that can drift (e.g. a second call site such as the
data-table reducer, or a trim-semantics change in the table). Consider either
reducing this to `_numAggregationFunctions == 1` and letting the table
dispatch, or documenting why the caller must exclude order-by/grouping sets.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
