xiangfu0 commented on code in PR #19825:
URL: https://github.com/apache/pinot/pull/19825#discussion_r4245004337
##########
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:
Serialized deterministic-table admission, updates and eviction under the
same table lock for both input paths. The concurrent COUNT test now covers this
table as well, so mutable in-place merges cannot lose a writer’s contribution.
Fixed in
[d4034a5c37](https://github.com/apache/pinot/commit/d4034a5c37d6969d8a45970b6b54f91bd30f4687).
##########
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:
Replaced the assertions with a hard check at the shared scalar API boundary.
A multi-aggregation table now fails before mutation; the test covers both
SimpleIndexedTable and ConcurrentIndexedTable. Fixed in
[d4034a5c37](https://github.com/apache/pinot/commit/d4034a5c37d6969d8a45970b6b54f91bd30f4687).
##########
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:
Both input paths now use the same unordered admission hook in each table.
The concurrent cap transition and saturated update branch have one
implementation, and subclasses retain dispatch through that hook. Fixed in
[d4034a5c37](https://github.com/apache/pinot/commit/d4034a5c37d6969d8a45970b6b54f91bd30f4687).
##########
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:
Extracted shared per-column merge dispatch for eager and scalar inputs. The
multi-aggregation loops keep their existing outer input-mode branch and pass a
constant mode to the helper. Fixed in
[d4034a5c37](https://github.com/apache/pinot/commit/d4034a5c37d6969d8a45970b6b54f91bd30f4687).
##########
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:
Extended lazy row creation to the shared map helpers, so unordered Simple,
Concurrent and Deterministic tables use it while retaining their own admission
policy. Ordered inputs still follow the existing record path and trim behavior.
Fixed in
[d4034a5c37](https://github.com/apache/pinot/commit/d4034a5c37d6969d8a45970b6b54f91bd30f4687).
##########
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:
Reduced the combine gate to a single aggregation function. The table now
owns ordered fallback and unordered admission; the existing
preconstructed-record branch is unchanged. Ordered trim and grouping-set query
tests validate both paths. Fixed in
[d4034a5c37](https://github.com/apache/pinot/commit/d4034a5c37d6969d8a45970b6b54f91bd30f4687).
--
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]
