Jackie-Jiang commented on code in PR #19825:
URL: https://github.com/apache/pinot/pull/19825#discussion_r4251214878
##########
pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java:
##########
@@ -108,14 +109,59 @@ 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.
+ /// Final inputs must be Comparable, as required by the aggregation
function's final-result merge.
+ /// Returns true even when new groups are rejected; existing groups can
still accept updates.
+ public boolean upsertSingleAggregationResult(Key key, @Nullable Object
result) {
+ Preconditions.checkState(_aggregationFunctions.length == 1,
+ "upsertSingleAggregationResult requires exactly one aggregation
function");
+ Preconditions.checkArgument(!_hasFinalInput || result == null || result
instanceof Comparable,
+ "Final aggregation inputs must be Comparable");
+ if (_hasOrderBy) {
+ return upsert(key, createSingleAggregationRecord(key, result));
+ }
+ upsertWithoutOrderBy(key, null, result);
+ return true;
+ }
+
+ /// Applies the table's unordered admission policy to either an eager record
or a single aggregation result.
+ /// A null record selects the single-result input, which may itself be null.
+ protected abstract void upsertWithoutOrderBy(Key key, @Nullable Record
record, @Nullable Object 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) {
+ Object[] values = record.getValues();
+ values[_numKeyColumns] = mergeValue(0, values[_numKeyColumns], result,
_hasFinalInput);
+ return record;
+ }
+
/// Adds a record with new key or updates a record with existing key.
protected void addOrUpdateRecord(Key key, Record newRecord) {
- _lookupMap.compute(key, (k, v) -> v == null ? newRecord : updateRecord(v,
newRecord));
+ addOrUpdateRecord(key, newRecord, null);
+ }
+
+ protected void addOrUpdateRecord(Key key, @Nullable Record newRecord,
@Nullable Object result) {
+ _lookupMap.compute(key, (k, existing) -> {
+ if (existing == null) {
+ return newRecord != null ? newRecord :
createSingleAggregationRecord(k, result);
+ }
+ return newRecord != null ? updateRecord(existing, newRecord) :
updateSingleAggregationRecord(existing, result);
+ });
}
/// Updates a record with existing key. Record with new key will be ignored.
- protected void updateExistingRecord(Key key, Record newRecord) {
- _lookupMap.computeIfPresent(key, (k, v) -> updateRecord(v, newRecord));
+ protected void updateExistingRecord(Key key, @Nullable Record newRecord,
@Nullable Object result) {
+ if (_lookupMap.containsKey(key)) {
Review Comment:
[P1] Preserve atomic updates during group-limit transitions
The new `containsKey` guard can silently drop an aggregation contribution
while a group's initial insertion is still in progress:
1. T1 starts inserting `K → 1` through `compute`, installing a reservation
node before publishing the record.
2. T2 inserts a different-bucket key, reaches `_resultSize`, and sets
`_noMoreNewRecords` (with enough initial capacity to avoid a resize).
3. T3 merges another `K → 1` through this saturated path. `containsKey(K)`
returns false while T1's insertion is pending, so T3's contribution is
discarded.
4. T1 publishes K, leaving its aggregate at 1 instead of 2.
`ConcurrentHashMap.containsKey` uses a nonblocking lookup that sees no value
for the reservation node. The previous unconditional `computeIfPresent` waits
for the pending insertion and retries against the published record, preserving
the contribution.
Please remove the `containsKey` precheck; `computeIfPresent` already ignores
absent keys. A concurrency test covering insertion during the limit transition
would catch this—the current saturated-update test prepopulates its hot key
before starting the writers.
--
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]
