xiangfu0 commented on code in PR #19825:
URL: https://github.com/apache/pinot/pull/19825#discussion_r4251772883
##########
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:
Removed the containsKey precheck and restored unconditional
computeIfPresent, so updates wait for an in-flight insertion. Added a
controlled regression for both eager and scalar updates: both cases failed
before the fix with count 1 instead of 2, and pass afterward; all 52 focused
table tests and the required style/license checks passed. Fixed in
[97acbecfd9](https://github.com/apache/pinot/commit/97acbecfd9ed0e4f7fef270412aaeb60b40bb42c).
##########
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
Review Comment:
Removed this precheck in
[97acbecfd9](https://github.com/apache/pinot/commit/97acbecfd9ed0e4f7fef270412aaeb60b40bb42c)
after the pending-insertion race identified by Jackie. computeIfPresent must
run unconditionally to wait for a reservation node; it already ignores
genuinely absent keys. The eager/scalar regression verifies both contributions
are retained.
--
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]
