bvarghese1 commented on code in PR #29373:
URL: https://github.com/apache/flink/pull/29373#discussion_r4209060208
##########
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunctionTest.java:
##########
@@ -96,6 +96,58 @@ void testInsertOnlyRecordsWithCustomSortKey() throws
Exception {
validateRows(actualRows, expectedRows);
}
+ @Test
+ void testInsertWithDuplicateSortKeyAndLastValueAgg() throws Exception {
+ KeyedProcessOperator<RowData, RowData, RowData> operator =
+ new KeyedProcessOperator<>(
+ new NonTimeRowsUnboundedPrecedingFunction<RowData>(
+ 0L,
+ lastValueAggsHandleFunction,
+ GENERATED_ROW_VALUE_EQUALISER,
+ GENERATED_SORT_KEY_EQUALISER,
+ GENERATED_SORT_KEY_COMPARATOR_ASC,
+ lastValueAccTypes,
+ inputFieldTypes,
+ SORT_KEY_TYPES,
+ SORT_KEY_SELECTOR) {});
+
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
+ createTestHarness(operator);
+ testHarness.open();
+
+ testHarness.processElement(insertRecord("key1", 1L, 100L));
+ testHarness.processElement(insertRecord("key1", 2L, 200L));
+ testHarness.processElement(insertRecord("key1", 5L, 500L));
+ testHarness.processElement(insertRecord("key1", 6L, 600L));
+ testHarness.processElement(insertRecord("key1", 4L, 400L));
+ testHarness.processElement(insertRecord("key1", 5L, 503L));
+ testHarness.processElement(updateBeforeRecord("key1", 5L, 500L));
Review Comment:
Good catch, I was able to reproduce it. Fixed it.
Parameterized the tests so that the tests use both the `heap` and
`EmbeddedRocksDBStateBackend` state backends.
Note: early out tests are skipped on RocksDB for now. It can be updated once
[FLINK-40735](https://issues.apache.org/jira/browse/FLINK-40735) is merged.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunction.java:
##########
@@ -273,21 +273,25 @@ protected void processRemainingElements(
// to comply with the sql rows syntax
for (int j = 0; j < ids.size(); j++) {
RowData value = valueMapState.get(ids.get(j));
+ RowData prevAcc =
accMapState.get(GenericRowData.of(ids.get(j)));
aggFuncs.accumulate(value);
- RowData accData =
accMapState.get(GenericRowData.of(ids.get(j)));
+ RowData newAcc = aggFuncs.getAccumulators();
// Logic to early out
// TODO: Move comparison to function i.e. canEarlyOut(prev,
curr)
- if (aggFuncs.getValue().equals(accData)) {
+ if (newAcc.equals(prevAcc)) {
Review Comment:
Makes sense. FLINK-40735 can fix this.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunction.java:
##########
@@ -402,19 +406,27 @@ private RowData getPreviousAccumulator(
*/
private void reAccumulateIdsAndEmitUpdates(
List<Long> ids, int removeIndex, Collector<RowData> out) throws
Exception {
+ RowData baseAcc = aggFuncs.getAccumulators();
for (int j = removeIndex; j < ids.size(); j++) {
RowData value = valueMapState.get(ids.get(j));
if (j == removeIndex) {
- collectDelete(out, value,
accMapState.get(GenericRowData.of(ids.get(j))));
+ RowData deletedAcc =
accMapState.get(GenericRowData.of(ids.get(j)));
+ collectDelete(out, value,
setAccumulatorAndGetValue(deletedAcc));
+ aggFuncs.setAccumulators(baseAcc);
} else {
+ RowData prevAcc =
accMapState.get(GenericRowData.of(ids.get(j)));
aggFuncs.accumulate(value);
+ RowData newAcc = aggFuncs.getAccumulators();
// Logic to early out
- if
(aggFuncs.getValue().equals(accMapState.get(GenericRowData.of(ids.get(j))))) {
+ if (newAcc.equals(prevAcc)) {
break;
Review Comment:
Fixed. `reAccumulateIdsAndEmitUpdates` now returns whether it exited early,
and `removeFromSortedList` skips `processRemainingElements` in that case.
--
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]
