This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 7a76f129672 [FLINK-40883][table] Fix accumulator corruption in
NonTimeRowsUnboundedPrecedingFunction (#29373)
7a76f129672 is described below
commit 7a76f129672193aa170ba18c25090f8586fc54e4
Author: Bonnie Varghese <[email protected]>
AuthorDate: Wed Oct 7 13:08:48 2026 -0700
[FLINK-40883][table] Fix accumulator corruption in
NonTimeRowsUnboundedPrecedingFunction (#29373)
- processRemainingElements and reAccumulateIdsAndEmitUpdates stored
aggFuncs.getValue() into accMapState instead of
aggFuncs.getAccumulators(), and emitted the raw stored accumulator as
the "before" value. For aggregates whose accumulator differs from their
result (e.g. LAST_VALUE with retraction), a recalculation after an
out-of-order insert or a delete wrote a result-shaped row into the
accumulator state, which fails on RocksDB when it is serialized
(ArrayIndexOutOfBoundsException in RowDataSerializer).
- Compare accumulator to accumulator for the early-out check, derive the
before/after values via setAccumulatorAndGetValue(), and store the
accumulator in accMapState, matching
NonTimeRangeUnboundedPrecedingFunction.
- When the early-out fires on the delete path, skip processing of the
following sort keys.
Generated-by: Claude Code (Claude Opus 5.5)
---
.../NonTimeRowsUnboundedPrecedingFunction.java | 43 ++++--
.../over/LastValueAggsHandleFunction.java | 96 ++++++++++++
.../operators/over/NonTimeOverWindowTestBase.java | 69 +++++++--
...NonTimeRangeUnboundedPrecedingFunctionTest.java | 114 ++++++++++----
.../NonTimeRowsUnboundedPrecedingFunctionTest.java | 164 +++++++++++++++++----
.../operators/over/RowTimeOverWindowTestBase.java | 10 ++
6 files changed, 413 insertions(+), 83 deletions(-)
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunction.java
index e5a1ca7e958..6d6f89760cc 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunction.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunction.java
@@ -273,11 +273,12 @@ public class NonTimeRowsUnboundedPrecedingFunction<K>
// 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)) {
// Previous accumulator is the same as the current
accumulator.
// This means all the ids will have no change in the
accumulated value.
// Skip sending downstream updates in such cases to reduce
network traffic
@@ -285,9 +286,12 @@ public class NonTimeRowsUnboundedPrecedingFunction<K>
"Prev accumulator is same as curr accumulator.
Skipping further updates.");
return;
}
- collectUpdateBefore(out, value, accData);
- collectUpdateAfter(out, value, aggFuncs.getValue());
- accMapState.put(GenericRowData.of(ids.get(j)),
aggFuncs.getValue());
+ RowData newValue = aggFuncs.getValue();
+ RowData prevValue = setAccumulatorAndGetValue(prevAcc);
+ aggFuncs.setAccumulators(newAcc);
+ collectUpdateBefore(out, value, prevValue);
+ collectUpdateAfter(out, value, newValue);
+ accMapState.put(GenericRowData.of(ids.get(j)), newAcc);
}
}
}
@@ -335,7 +339,7 @@ public class NonTimeRowsUnboundedPrecedingFunction<K>
aggFuncs.setAccumulators(prevAcc);
// Re-accumulate ids and emit updates for ids
- reAccumulateIdsAndEmitUpdates(ids, removeIndex, out);
+ boolean isEarlyOut = reAccumulateIdsAndEmitUpdates(ids, removeIndex,
out);
// Remove id and update sortedList
Long deletedId = ids.remove(removeIndex);
@@ -346,6 +350,9 @@ public class NonTimeRowsUnboundedPrecedingFunction<K>
accMapState.remove(GenericRowData.of(deletedId));
sortedListState.update(sortedList);
+ if (isEarlyOut) {
+ return;
+ }
processRemainingElements(sortedList, idxOfSortKey,
aggFuncs.getAccumulators(), out);
}
@@ -398,24 +405,34 @@ public class NonTimeRowsUnboundedPrecedingFunction<K>
* @param ids
* @param removeIndex
* @param out
+ * @return true if an id's accumulator was unchanged, so all following ids
are unchanged too
* @throws Exception
*/
- private void reAccumulateIdsAndEmitUpdates(
+ private boolean 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))))) {
- break;
+ if (newAcc.equals(prevAcc)) {
+ return true;
}
- collectUpdateBefore(out, value,
accMapState.get(GenericRowData.of(ids.get(j))));
- collectUpdateAfter(out, value, aggFuncs.getValue());
- accMapState.put(GenericRowData.of(ids.get(j)),
aggFuncs.getValue());
+ RowData newValue = aggFuncs.getValue();
+ RowData prevValue = setAccumulatorAndGetValue(prevAcc);
+ aggFuncs.setAccumulators(newAcc);
+ collectUpdateBefore(out, value, prevValue);
+ collectUpdateAfter(out, value, newValue);
+ accMapState.put(GenericRowData.of(ids.get(j)), newAcc);
}
}
+ return false;
}
}
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/LastValueAggsHandleFunction.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/LastValueAggsHandleFunction.java
new file mode 100644
index 00000000000..2df315d8a70
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/LastValueAggsHandleFunction.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.over;
+
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.dataview.StateDataViewStore;
+import org.apache.flink.table.runtime.generated.AggsHandleFunction;
+
+/**
+ * Test {@link AggsHandleFunction} whose accumulator shape, (lastValue,
count), differs from its
+ * result shape, (lastValue) only - mirroring built-in aggregates like
LAST_VALUE with retraction,
+ * where the accumulator carries bookkeeping fields beyond what getValue()
returns.
+ */
+public class LastValueAggsHandleFunction implements AggsHandleFunction {
+
+ private final int inputIndex;
+ private Long lastValue;
+ private long count;
+
+ public LastValueAggsHandleFunction(int inputIndex) {
+ this.inputIndex = inputIndex;
+ }
+
+ @Override
+ public void open(StateDataViewStore store) throws Exception {}
+
+ @Override
+ public void accumulate(RowData input) throws Exception {
+ lastValue = input.getLong(inputIndex);
+ count++;
+ }
+
+ @Override
+ public void retract(RowData input) throws Exception {
+ count--;
+ }
+
+ @Override
+ public void merge(RowData accumulator) throws Exception {
+ lastValue = accumulator.getLong(0);
+ count += accumulator.getLong(1);
+ }
+
+ @Override
+ public void setAccumulators(RowData accumulator) throws Exception {
+ lastValue = accumulator.getLong(0);
+ count = accumulator.getLong(1);
+ }
+
+ @Override
+ public void resetAccumulators() throws Exception {
+ lastValue = null;
+ count = 0;
+ }
+
+ @Override
+ public RowData getAccumulators() throws Exception {
+ return GenericRowData.of(lastValue, count);
+ }
+
+ @Override
+ public RowData createAccumulators() throws Exception {
+ return GenericRowData.of(0L, 0L);
+ }
+
+ @Override
+ public RowData getValue() throws Exception {
+ return GenericRowData.of(lastValue);
+ }
+
+ @Override
+ public void setWindowSize(int windowSize) {}
+
+ @Override
+ public void cleanup() throws Exception {}
+
+ @Override
+ public void close() throws Exception {}
+}
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeOverWindowTestBase.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeOverWindowTestBase.java
index 2c42980c389..ca557ae3a28 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeOverWindowTestBase.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeOverWindowTestBase.java
@@ -19,6 +19,10 @@
package org.apache.flink.table.runtime.operators.over;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
+import org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackend;
+import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
@@ -30,20 +34,37 @@ import
org.apache.flink.table.runtime.generated.RecordComparator;
import org.apache.flink.table.runtime.generated.RecordEqualiser;
import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import
org.apache.flink.table.runtime.util.StateParameterizedHarnessTestBase.StateBackendMode;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.utils.HandwrittenSelectorUtil;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import
org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.util.Arrays;
+import java.util.Collection;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assumptions.assumeThat;
/** Base class for non-time over window test. */
+@ExtendWith(ParameterizedTestExtension.class)
public abstract class NonTimeOverWindowTestBase extends
RowTimeOverWindowTestBase {
+ @Parameter private StateBackendMode stateBackendMode;
+
+ @Parameters(name = "StateBackend={0}")
+ public static Collection<StateBackendMode> parameters() {
+ return Arrays.asList(StateBackendMode.HEAP, StateBackendMode.ROCKSDB);
+ }
+
private static final int SORT_KEY_IDX = 1;
static final LogicalType[] SORT_KEY_TYPES = new LogicalType[] {new
BigIntType()};
@@ -142,13 +163,11 @@ public abstract class NonTimeOverWindowTestBase extends
RowTimeOverWindowTestBas
public boolean equals(RowData row1, RowData row2) {
if (row1 instanceof BinaryRowData && row2 instanceof
BinaryRowData) {
return row1.equals(row2);
- } else if (row1 instanceof GenericRowData && row2 instanceof
GenericRowData) {
- return row1.getString(0).equals(row2.getString(0))
- && row1.getLong(1) == row2.getLong(1)
- && row1.getLong(2) == row2.getLong(2);
- } else {
- throw new UnsupportedOperationException();
}
+ // Rows read back from RocksDB are BinaryRowData while input rows
are GenericRowData
+ return row1.getString(0).equals(row2.getString(0))
+ && row1.getLong(1) == row2.getLong(1)
+ && row1.getLong(2) == row2.getLong(2);
}
}
@@ -169,6 +188,33 @@ public abstract class NonTimeOverWindowTestBase extends
RowTimeOverWindowTestBas
}
}
+ @Override
+ protected OneInputStreamOperatorTestHarness<RowData, RowData>
createTestHarness(
+ KeyedProcessOperator<RowData, RowData, RowData> operator) throws
Exception {
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
+ super.createTestHarness(operator);
+ switch (stateBackendMode) {
+ case HEAP:
+ testHarness.setStateBackend(new HashMapStateBackend());
+ break;
+ case ROCKSDB:
+ testHarness.setStateBackend(new EmbeddedRocksDBStateBackend());
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown mode: " +
stateBackendMode);
+ }
+ return testHarness;
+ }
+
+ /**
+ * The early-out compares a {@link GenericRowData} accumulator with the
{@link BinaryRowData}
+ * read back from RocksDB, which never matches until FLINK-40735 compares
them with a {@link
+ * RecordEqualiser}.
+ */
+ void assumeEarlyOutSupported() {
+ assumeThat(stateBackendMode).isEqualTo(StateBackendMode.HEAP);
+ }
+
abstract void validateEntry(
AbstractNonTimeUnboundedPrecedingOver<RowData> function, RowData
record, int idOffset)
throws Exception;
@@ -222,12 +268,11 @@ public abstract class NonTimeOverWindowTestBase extends
RowTimeOverWindowTestBas
if (isInsertion) {
// Validate if record was successfully inserted in the
valueMapState
- assertThat(
- function.getRuntimeContext()
- .getMapState(function.valueStateDescriptor)
- .get(Long.MIN_VALUE + idOffset)
- .toString())
- .isEqualTo(record.toString());
+ RowData storedRow =
+ function.getRuntimeContext()
+ .getMapState(function.valueStateDescriptor)
+ .get(Long.MIN_VALUE + idOffset);
+ assertThat(new TestRowValueEqualiser().equals(storedRow,
record)).isTrue();
} else {
// Validate if record was successfully removed from the
valueMapState
assertThat(
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeRangeUnboundedPrecedingFunctionTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeRangeUnboundedPrecedingFunctionTest.java
index 7f7e45a705c..03c748fdf90 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeRangeUnboundedPrecedingFunctionTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeRangeUnboundedPrecedingFunctionTest.java
@@ -24,10 +24,11 @@ import
org.apache.flink.streaming.api.operators.KeyedProcessOperator;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
import org.apache.flink.types.RowKind;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
import java.time.Duration;
import java.util.Arrays;
@@ -56,7 +57,7 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
SORT_KEY_SELECTOR) {};
}
- @Test
+ @TestTemplate
void testInsertOnlyRecordsWithCustomSortKey() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -97,7 +98,61 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
+ void testInsertWithDuplicateSortKeyAndLastValueAgg() throws Exception {
+ KeyedProcessOperator<RowData, RowData, RowData> operator =
+ new KeyedProcessOperator<>(
+ new NonTimeRangeUnboundedPrecedingFunction<>(
+ 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));
+
+ List<RowData> expectedRows =
+ Arrays.asList(
+ outputRecord(RowKind.INSERT, "key1", 1L, 100L, 100L),
+ outputRecord(RowKind.INSERT, "key1", 2L, 200L, 200L),
+ outputRecord(RowKind.INSERT, "key1", 5L, 500L, 500L),
+ outputRecord(RowKind.INSERT, "key1", 6L, 600L, 600L),
+ outputRecord(RowKind.INSERT, "key1", 4L, 400L, 400L),
+ outputRecord(RowKind.UPDATE_BEFORE, "key1", 5L, 500L,
500L),
+ outputRecord(RowKind.UPDATE_AFTER, "key1", 5L, 500L,
500L),
+ outputRecord(RowKind.UPDATE_BEFORE, "key1", 6L, 600L,
600L),
+ outputRecord(RowKind.UPDATE_AFTER, "key1", 6L, 600L,
600L),
+ outputRecord(RowKind.UPDATE_BEFORE, "key1", 5L, 500L,
500L),
+ outputRecord(RowKind.UPDATE_AFTER, "key1", 5L, 500L,
503L),
+ outputRecord(RowKind.INSERT, "key1", 5L, 503L, 503L),
+ outputRecord(RowKind.UPDATE_BEFORE, "key1", 6L, 600L,
600L),
+ outputRecord(RowKind.UPDATE_AFTER, "key1", 6L, 600L,
600L),
+ outputRecord(RowKind.DELETE, "key1", 5L, 500L, 503L),
+ outputRecord(RowKind.UPDATE_BEFORE, "key1", 5L, 503L,
503L),
+ outputRecord(RowKind.UPDATE_AFTER, "key1", 5L, 503L,
503L),
+ outputRecord(RowKind.UPDATE_BEFORE, "key1", 6L, 600L,
600L),
+ outputRecord(RowKind.UPDATE_AFTER, "key1", 6L, 600L,
600L));
+
+ List<RowData> actualRows = testHarness.extractOutputValues();
+
+ validateRows(actualRows, expectedRows);
+ }
+
+ @TestTemplate
void testInsertOnlyRecordsWithCustomSortKeyAndLongSumAgg() throws
Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -146,7 +201,7 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testInsertOnlyRecordsWithDuplicateSortKeys() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -226,7 +281,7 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testRetractingRecordsWithCustomSortKey() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -294,7 +349,7 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testRetractWithFirstDuplicateSortKey() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -344,7 +399,7 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testRetractWithMiddleDuplicateSortKey() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -394,7 +449,7 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testRetractWithLastDuplicateSortKey() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -444,7 +499,7 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testRetractWithDescendingSort() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -500,8 +555,9 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testRetractWithEarlyOut() throws Exception {
+ assumeEarlyOutSupported();
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
getNonTimeRangeUnboundedPrecedingFunction(
@@ -550,7 +606,7 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testInsertAndRetractAllWithStateValidation() throws Exception {
NonTimeRangeUnboundedPrecedingFunction<RowData> function =
getNonTimeRangeUnboundedPrecedingFunction(0L,
GENERATED_SORT_KEY_COMPARATOR_ASC);
@@ -562,31 +618,31 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
testHarness.open();
// put some records
- GenericRowData firstRecord = GenericRowData.of("key1", 1L, 100L);
+ GenericRowData firstRecord =
GenericRowData.of(StringData.fromString("key1"), 1L, 100L);
testHarness.processElement(insertRecord("key1", 1L, 100L));
validateState(function, firstRecord, 0, 1, 0, 1, 0, 1, true);
- GenericRowData secondRecord = GenericRowData.of("key1", 2L, 200L);
+ GenericRowData secondRecord =
GenericRowData.of(StringData.fromString("key1"), 2L, 200L);
testHarness.processElement(insertRecord("key1", 2L, 200L));
validateState(function, secondRecord, 1, 2, 0, 1, 1, 2, true);
- GenericRowData thirdRecord = GenericRowData.of("key1", 2L, 201L);
+ GenericRowData thirdRecord =
GenericRowData.of(StringData.fromString("key1"), 2L, 201L);
testHarness.processElement(insertRecord("key1", 2L, 201L));
validateState(function, thirdRecord, 1, 2, 1, 2, 2, 3, true);
- GenericRowData fourthRecord = GenericRowData.of("key1", 5L, 500L);
+ GenericRowData fourthRecord =
GenericRowData.of(StringData.fromString("key1"), 5L, 500L);
testHarness.processElement(insertRecord("key1", 5L, 500L));
validateState(function, fourthRecord, 2, 3, 0, 1, 3, 4, true);
- GenericRowData fifthRecord = GenericRowData.of("key1", 5L, 502L);
+ GenericRowData fifthRecord =
GenericRowData.of(StringData.fromString("key1"), 5L, 502L);
testHarness.processElement(insertRecord("key1", 5L, 502L));
validateState(function, fifthRecord, 2, 3, 1, 2, 4, 5, true);
- GenericRowData sixthRecord = GenericRowData.of("key1", 5L, 501L);
+ GenericRowData sixthRecord =
GenericRowData.of(StringData.fromString("key1"), 5L, 501L);
testHarness.processElement(insertRecord("key1", 5L, 501L));
validateState(function, sixthRecord, 2, 3, 2, 3, 5, 6, true);
- GenericRowData seventhRecord = GenericRowData.of("key1", 6L, 600L);
+ GenericRowData seventhRecord =
GenericRowData.of(StringData.fromString("key1"), 6L, 600L);
testHarness.processElement(insertRecord("key1", 6L, 600L));
validateState(function, seventhRecord, 3, 4, 0, 1, 6, 7, true);
@@ -617,7 +673,7 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
assertThat(function.getNumOfIdsNotFound().getCount()).isZero();
}
- @Test
+ @TestTemplate
void testInsertWithStateTTLExpiration() throws Exception {
Duration stateTtlTime = Duration.ofMillis(10);
NonTimeRangeUnboundedPrecedingFunction<RowData> function =
@@ -631,15 +687,15 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
testHarness.open();
// put some records
- GenericRowData firstRecord = GenericRowData.of("key1", 1L, 100L);
+ GenericRowData firstRecord =
GenericRowData.of(StringData.fromString("key1"), 1L, 100L);
testHarness.processElement(insertRecord("key1", 1L, 100L));
validateState(function, firstRecord, 0, 1, 0, 1, 0, 1, true);
- GenericRowData secondRecord = GenericRowData.of("key1", 2L, 200L);
+ GenericRowData secondRecord =
GenericRowData.of(StringData.fromString("key1"), 2L, 200L);
testHarness.processElement(insertRecord("key1", 2L, 200L));
validateState(function, secondRecord, 1, 2, 0, 1, 1, 2, true);
- GenericRowData thirdRecord = GenericRowData.of("key1", 2L, 201L);
+ GenericRowData thirdRecord =
GenericRowData.of(StringData.fromString("key1"), 2L, 201L);
testHarness.processElement(insertRecord("key1", 2L, 201L));
validateState(function, thirdRecord, 1, 2, 1, 2, 2, 3, true);
@@ -651,7 +707,7 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
testHarness.setProcessingTime(stateTtlTime.toMillis() + 1);
// After insertion of the following record, there should be only 1
record in state
- GenericRowData fourthRecord = GenericRowData.of("key1", 5L, 500L);
+ GenericRowData fourthRecord =
GenericRowData.of(StringData.fromString("key1"), 5L, 500L);
testHarness.processElement(insertRecord("key1", 5L, 500L));
validateState(function, fourthRecord, 0, 1, 0, 1, 0, 1, true);
@@ -668,7 +724,7 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
assertThat(function.getNumOfIdsNotFound().getCount()).isZero();
}
- @Test
+ @TestTemplate
void testInsertAndRetractWithStateTTLExpiration() throws Exception {
Duration stateTtlTime = Duration.ofMillis(10);
NonTimeRangeUnboundedPrecedingFunction<RowData> function =
@@ -682,23 +738,23 @@ class NonTimeRangeUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBa
testHarness.open();
// put some records
- GenericRowData firstRecord = GenericRowData.of("key1", 1L, 100L);
+ GenericRowData firstRecord =
GenericRowData.of(StringData.fromString("key1"), 1L, 100L);
testHarness.processElement(insertRecord("key1", 1L, 100L));
validateState(function, firstRecord, 0, 1, 0, 1, 0, 1, true);
- GenericRowData secondRecord = GenericRowData.of("key1", 2L, 200L);
+ GenericRowData secondRecord =
GenericRowData.of(StringData.fromString("key1"), 2L, 200L);
testHarness.processElement(insertRecord("key1", 2L, 200L));
validateState(function, secondRecord, 1, 2, 0, 1, 1, 2, true);
- GenericRowData thirdRecord = GenericRowData.of("key1", 2L, 201L);
+ GenericRowData thirdRecord =
GenericRowData.of(StringData.fromString("key1"), 2L, 201L);
testHarness.processElement(insertRecord("key1", 2L, 201L));
validateState(function, thirdRecord, 1, 2, 1, 2, 2, 3, true);
- GenericRowData fourthRecord = GenericRowData.of("key1", 5L, 500L);
+ GenericRowData fourthRecord =
GenericRowData.of(StringData.fromString("key1"), 5L, 500L);
testHarness.processElement(insertRecord("key1", 5L, 500L));
validateState(function, fourthRecord, 2, 3, 0, 1, 3, 4, true);
- GenericRowData fifthRecord = GenericRowData.of("key1", 5L, 502L);
+ GenericRowData fifthRecord =
GenericRowData.of(StringData.fromString("key1"), 5L, 502L);
testHarness.processElement(insertRecord("key1", 5L, 502L));
validateState(function, fifthRecord, 2, 3, 1, 2, 4, 5, true);
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunctionTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunctionTest.java
index 7452169f661..0e687e3b59a 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunctionTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunctionTest.java
@@ -24,15 +24,19 @@ import
org.apache.flink.streaming.api.operators.KeyedProcessOperator;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.runtime.generated.AggsHandleFunction;
+import org.apache.flink.table.runtime.generated.GeneratedAggsHandleFunction;
import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
import org.apache.flink.types.RowKind;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
+import static
org.apache.flink.table.runtime.util.StreamRecordUtils.deleteRecord;
import static
org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord;
import static
org.apache.flink.table.runtime.util.StreamRecordUtils.updateAfterRecord;
import static
org.apache.flink.table.runtime.util.StreamRecordUtils.updateBeforeRecord;
@@ -55,7 +59,7 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
SORT_KEY_SELECTOR) {};
}
- @Test
+ @TestTemplate
void testInsertOnlyRecordsWithCustomSortKey() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -96,7 +100,108 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
+ 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));
+
+ List<RowData> expectedRows =
+ Arrays.asList(
+ outputRecord(RowKind.INSERT, "key1", 1L, 100L, 100L),
+ outputRecord(RowKind.INSERT, "key1", 2L, 200L, 200L),
+ outputRecord(RowKind.INSERT, "key1", 5L, 500L, 500L),
+ outputRecord(RowKind.INSERT, "key1", 6L, 600L, 600L),
+ outputRecord(RowKind.INSERT, "key1", 4L, 400L, 400L),
+ outputRecord(RowKind.UPDATE_BEFORE, "key1", 5L, 500L,
500L),
+ outputRecord(RowKind.UPDATE_AFTER, "key1", 5L, 500L,
500L),
+ outputRecord(RowKind.UPDATE_BEFORE, "key1", 6L, 600L,
600L),
+ outputRecord(RowKind.UPDATE_AFTER, "key1", 6L, 600L,
600L),
+ outputRecord(RowKind.INSERT, "key1", 5L, 503L, 503L),
+ outputRecord(RowKind.UPDATE_BEFORE, "key1", 6L, 600L,
600L),
+ outputRecord(RowKind.UPDATE_AFTER, "key1", 6L, 600L,
600L),
+ outputRecord(RowKind.DELETE, "key1", 5L, 500L, 500L),
+ outputRecord(RowKind.UPDATE_BEFORE, "key1", 5L, 503L,
503L),
+ outputRecord(RowKind.UPDATE_AFTER, "key1", 5L, 503L,
503L),
+ outputRecord(RowKind.UPDATE_BEFORE, "key1", 6L, 600L,
600L),
+ outputRecord(RowKind.UPDATE_AFTER, "key1", 6L, 600L,
600L));
+
+ List<RowData> actualRows = testHarness.extractOutputValues();
+
+ validateRows(actualRows, expectedRows);
+ }
+
+ @TestTemplate
+ void testRetractWithEarlyOutDoesNotRestartFollowingSortKeys() throws
Exception {
+ assumeEarlyOutSupported();
+ // Sum ts rather than the sort key so rows sharing a sort key change
the accumulator
+ // differently
+ GeneratedAggsHandleFunction sumOfTsAggsHandleFunction =
+ new GeneratedAggsHandleFunction("Function", "", new Object[0])
{
+ @Override
+ public AggsHandleFunction newInstance(ClassLoader
classLoader) {
+ return new SumAggsHandleFunction(2);
+ }
+ };
+ KeyedProcessOperator<RowData, RowData, RowData> operator =
+ new KeyedProcessOperator<>(
+ new NonTimeRowsUnboundedPrecedingFunction<RowData>(
+ 0L,
+ sumOfTsAggsHandleFunction,
+ GENERATED_ROW_VALUE_EQUALISER,
+ GENERATED_SORT_KEY_EQUALISER,
+ GENERATED_SORT_KEY_COMPARATOR_ASC,
+ accTypes,
+ inputFieldTypes,
+ SORT_KEY_TYPES,
+ SORT_KEY_SELECTOR) {});
+
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
+ createTestHarness(operator);
+ testHarness.open();
+
+ testHarness.processElement(insertRecord("key1", 1L, 0L));
+ testHarness.processElement(insertRecord("key1", 1L, 5L));
+ testHarness.processElement(insertRecord("key1", 1L, 7L));
+ testHarness.processElement(insertRecord("key1", 2L, 1L));
+ // Removing a zero leaves (1, 5) unchanged, so nothing after it
changes either
+ testHarness.processElement(deleteRecord("key1", 1L, 0L));
+
+ List<RowData> expectedRows =
+ Arrays.asList(
+ outputRecord(RowKind.INSERT, "key1", 1L, 0L, 0L),
+ outputRecord(RowKind.INSERT, "key1", 1L, 5L, 5L),
+ outputRecord(RowKind.INSERT, "key1", 1L, 7L, 12L),
+ outputRecord(RowKind.INSERT, "key1", 2L, 1L, 13L),
+ outputRecord(RowKind.DELETE, "key1", 1L, 0L, 0L));
+
+ List<RowData> actualRows = testHarness.extractOutputValues();
+
+ validateRows(actualRows, expectedRows);
+ }
+
+ @TestTemplate
void testInsertOnlyRecordsWithCustomSortKeyAndLongSumAgg() throws
Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -145,7 +250,7 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testInsertOnlyRecordsWithDuplicateSortKeys() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -213,7 +318,7 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testRetractingRecordsWithCustomSortKey() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -281,7 +386,7 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testRetractWithFirstDuplicateSortKey() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -323,7 +428,7 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testRetractWithMiddleDuplicateSortKey() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -363,7 +468,7 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testRetractWithLastDuplicateSortKey() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -401,7 +506,7 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testRetractWithDescendingSort() throws Exception {
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
@@ -455,8 +560,9 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testRetractWithEarlyOut() throws Exception {
+ assumeEarlyOutSupported();
KeyedProcessOperator<RowData, RowData, RowData> operator =
new KeyedProcessOperator<>(
getNonTimeRowsUnboundedPrecedingFunction(
@@ -497,7 +603,7 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
validateRows(actualRows, expectedRows);
}
- @Test
+ @TestTemplate
void testInsertAndRetractAllWithStateValidation() throws Exception {
NonTimeRowsUnboundedPrecedingFunction<RowData> function =
getNonTimeRowsUnboundedPrecedingFunction(0L,
GENERATED_SORT_KEY_COMPARATOR_ASC);
@@ -509,31 +615,31 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
testHarness.open();
// put some records
- GenericRowData firstRecord = GenericRowData.of("key1", 1L, 100L);
+ GenericRowData firstRecord =
GenericRowData.of(StringData.fromString("key1"), 1L, 100L);
testHarness.processElement(insertRecord("key1", 1L, 100L));
validateState(function, firstRecord, 0, 1, 0, 1, 0, 1, true);
- GenericRowData secondRecord = GenericRowData.of("key1", 2L, 200L);
+ GenericRowData secondRecord =
GenericRowData.of(StringData.fromString("key1"), 2L, 200L);
testHarness.processElement(insertRecord("key1", 2L, 200L));
validateState(function, secondRecord, 1, 2, 0, 1, 1, 2, true);
- GenericRowData thirdRecord = GenericRowData.of("key1", 2L, 201L);
+ GenericRowData thirdRecord =
GenericRowData.of(StringData.fromString("key1"), 2L, 201L);
testHarness.processElement(insertRecord("key1", 2L, 201L));
validateState(function, thirdRecord, 1, 2, 1, 2, 2, 3, true);
- GenericRowData fourthRecord = GenericRowData.of("key1", 5L, 500L);
+ GenericRowData fourthRecord =
GenericRowData.of(StringData.fromString("key1"), 5L, 500L);
testHarness.processElement(insertRecord("key1", 5L, 500L));
validateState(function, fourthRecord, 2, 3, 0, 1, 3, 4, true);
- GenericRowData fifthRecord = GenericRowData.of("key1", 5L, 502L);
+ GenericRowData fifthRecord =
GenericRowData.of(StringData.fromString("key1"), 5L, 502L);
testHarness.processElement(insertRecord("key1", 5L, 502L));
validateState(function, fifthRecord, 2, 3, 1, 2, 4, 5, true);
- GenericRowData sixthRecord = GenericRowData.of("key1", 5L, 501L);
+ GenericRowData sixthRecord =
GenericRowData.of(StringData.fromString("key1"), 5L, 501L);
testHarness.processElement(insertRecord("key1", 5L, 501L));
validateState(function, sixthRecord, 2, 3, 2, 3, 5, 6, true);
- GenericRowData seventhRecord = GenericRowData.of("key1", 6L, 600L);
+ GenericRowData seventhRecord =
GenericRowData.of(StringData.fromString("key1"), 6L, 600L);
testHarness.processElement(insertRecord("key1", 6L, 600L));
validateState(function, seventhRecord, 3, 4, 0, 1, 6, 7, true);
@@ -564,7 +670,7 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
assertThat(function.getNumOfIdsNotFound().getCount()).isZero();
}
- @Test
+ @TestTemplate
void testInsertWithStateTTLExpiration() throws Exception {
Duration stateTtlTime = Duration.ofMillis(10);
NonTimeRowsUnboundedPrecedingFunction<RowData> function =
@@ -578,15 +684,15 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
testHarness.open();
// put some records
- GenericRowData firstRecord = GenericRowData.of("key1", 1L, 100L);
+ GenericRowData firstRecord =
GenericRowData.of(StringData.fromString("key1"), 1L, 100L);
testHarness.processElement(insertRecord("key1", 1L, 100L));
validateState(function, firstRecord, 0, 1, 0, 1, 0, 1, true);
- GenericRowData secondRecord = GenericRowData.of("key1", 2L, 200L);
+ GenericRowData secondRecord =
GenericRowData.of(StringData.fromString("key1"), 2L, 200L);
testHarness.processElement(insertRecord("key1", 2L, 200L));
validateState(function, secondRecord, 1, 2, 0, 1, 1, 2, true);
- GenericRowData thirdRecord = GenericRowData.of("key1", 2L, 201L);
+ GenericRowData thirdRecord =
GenericRowData.of(StringData.fromString("key1"), 2L, 201L);
testHarness.processElement(insertRecord("key1", 2L, 201L));
validateState(function, thirdRecord, 1, 2, 1, 2, 2, 3, true);
@@ -599,7 +705,7 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
// After insertion of the following record, there should be only 1
record in state due to
// state ttl expiration
- GenericRowData fourthRecord = GenericRowData.of("key1", 5L, 500L);
+ GenericRowData fourthRecord =
GenericRowData.of(StringData.fromString("key1"), 5L, 500L);
testHarness.processElement(insertRecord("key1", 5L, 500L));
validateState(function, fourthRecord, 0, 1, 0, 1, 0, 1, true);
@@ -616,7 +722,7 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
assertThat(function.getNumOfIdsNotFound().getCount()).isZero();
}
- @Test
+ @TestTemplate
void testInsertAndRetractWithStateTTLExpiration() throws Exception {
Duration stateTtlTime = Duration.ofMillis(10);
NonTimeRowsUnboundedPrecedingFunction<RowData> function =
@@ -630,23 +736,23 @@ class NonTimeRowsUnboundedPrecedingFunctionTest extends
NonTimeOverWindowTestBas
testHarness.open();
// put some records
- GenericRowData firstRecord = GenericRowData.of("key1", 1L, 100L);
+ GenericRowData firstRecord =
GenericRowData.of(StringData.fromString("key1"), 1L, 100L);
testHarness.processElement(insertRecord("key1", 1L, 100L));
validateState(function, firstRecord, 0, 1, 0, 1, 0, 1, true);
- GenericRowData secondRecord = GenericRowData.of("key1", 2L, 200L);
+ GenericRowData secondRecord =
GenericRowData.of(StringData.fromString("key1"), 2L, 200L);
testHarness.processElement(insertRecord("key1", 2L, 200L));
validateState(function, secondRecord, 1, 2, 0, 1, 1, 2, true);
- GenericRowData thirdRecord = GenericRowData.of("key1", 2L, 201L);
+ GenericRowData thirdRecord =
GenericRowData.of(StringData.fromString("key1"), 2L, 201L);
testHarness.processElement(insertRecord("key1", 2L, 201L));
validateState(function, thirdRecord, 1, 2, 1, 2, 2, 3, true);
- GenericRowData fourthRecord = GenericRowData.of("key1", 5L, 500L);
+ GenericRowData fourthRecord =
GenericRowData.of(StringData.fromString("key1"), 5L, 500L);
testHarness.processElement(insertRecord("key1", 5L, 500L));
validateState(function, fourthRecord, 2, 3, 0, 1, 3, 4, true);
- GenericRowData fifthRecord = GenericRowData.of("key1", 5L, 502L);
+ GenericRowData fifthRecord =
GenericRowData.of(StringData.fromString("key1"), 5L, 502L);
testHarness.processElement(insertRecord("key1", 5L, 502L));
validateState(function, fifthRecord, 2, 3, 1, 2, 4, 5, true);
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/RowTimeOverWindowTestBase.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/RowTimeOverWindowTestBase.java
index 2ea90fa1ad9..48a558f878c 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/RowTimeOverWindowTestBase.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/RowTimeOverWindowTestBase.java
@@ -49,9 +49,19 @@ public class RowTimeOverWindowTestBase {
}
};
+ protected static GeneratedAggsHandleFunction lastValueAggsHandleFunction =
+ new GeneratedAggsHandleFunction("Function", "", new Object[0]) {
+ @Override
+ public AggsHandleFunction newInstance(ClassLoader classLoader)
{
+ return new LastValueAggsHandleFunction(2);
+ }
+ };
+
protected LogicalType[] inputFieldTypes =
new LogicalType[] {VarCharType.STRING_TYPE, new BigIntType(), new
BigIntType()};
protected LogicalType[] accTypes = new LogicalType[] {new BigIntType()};
+ protected LogicalType[] lastValueAccTypes =
+ new LogicalType[] {new BigIntType(), new BigIntType()};
protected RowDataKeySelector keySelector =
HandwrittenSelectorUtil.getRowDataSelector(new int[] {0},
inputFieldTypes);
