You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/06/18 03:07:02 UTC
[GitHub] [flink] KurtYoung commented on a change in pull request #12680: [FLINK-18119][table-runtime-blink] Retract old records in time range …
KurtYoung commented on a change in pull request #12680:
URL: https://github.com/apache/flink/pull/12680#discussion_r441942432
##########
File path: flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/operators/over/RowTimeRangeBoundedPrecedingFunctionTest.java
##########
@@ -0,0 +1,76 @@
+package org.apache.flink.table.runtime.operators.over;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
+import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.AggsHandleFunction;
+import org.apache.flink.table.runtime.generated.GeneratedAggsHandleFunction;
+import org.apache.flink.table.runtime.util.BinaryRowDataKeySelector;
+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.junit.Test;
+
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for {@link RowTimeRangeBoundedPrecedingFunction}.
+ */
+public class RowTimeRangeBoundedPrecedingFunctionTest {
+
+ private static GeneratedAggsHandleFunction aggsHandleFunction =
+ new GeneratedAggsHandleFunction("Function", "", new Object[0]) {
+ @Override
+ public AggsHandleFunction newInstance(ClassLoader classLoader) {
+ return new SumAggsHandleFunction(1);
+ }
+ };
+
+ private LogicalType[] inputFieldTypes = new LogicalType[]{
+ new VarCharType(VarCharType.MAX_LENGTH),
+ new BigIntType(),
+ new BigIntType()
+ };
+ private LogicalType[] accTypes = new LogicalType[]{ new BigIntType() };
+
+ private BinaryRowDataKeySelector keySelector = new BinaryRowDataKeySelector(new int[]{ 0 }, inputFieldTypes);
+ private TypeInformation<RowData> keyType = keySelector.getProducedType();
+
+ @Test
+ public void testRecordRetraction() throws Exception {
+ RowTimeRangeBoundedPrecedingFunction<RowData> function = new RowTimeRangeBoundedPrecedingFunction<>(0, 0, aggsHandleFunction, accTypes, inputFieldTypes, 2000, 2);
+ KeyedProcessOperator<RowData, RowData, RowData> operator = new KeyedProcessOperator<>(function);
+
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);
+
+ testHarness.open();
+
+ HeapKeyedStateBackend stateBackend = (HeapKeyedStateBackend) operator.getKeyedStateBackend();
Review comment:
use `AbstractKeyedStateBackend` instead
##########
File path: flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/operators/over/RowTimeRangeBoundedPrecedingFunctionTest.java
##########
@@ -0,0 +1,76 @@
+package org.apache.flink.table.runtime.operators.over;
Review comment:
missing header
##########
File path: flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/operators/over/ProcTimeRangeBoundedPrecedingFunctionTest.java
##########
@@ -0,0 +1,76 @@
+package org.apache.flink.table.runtime.operators.over;
Review comment:
missing header
##########
File path: flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/operators/over/ProcTimeRangeBoundedPrecedingFunctionTest.java
##########
@@ -0,0 +1,76 @@
+package org.apache.flink.table.runtime.operators.over;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
+import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
+import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.AggsHandleFunction;
+import org.apache.flink.table.runtime.generated.GeneratedAggsHandleFunction;
+import org.apache.flink.table.runtime.util.BinaryRowDataKeySelector;
+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.junit.Test;
+
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for {@link ProcTimeRangeBoundedPrecedingFunction}.
+ */
+public class ProcTimeRangeBoundedPrecedingFunctionTest {
+
+ private static GeneratedAggsHandleFunction aggsHandleFunction =
+ new GeneratedAggsHandleFunction("Function", "", new Object[0]) {
+ @Override
+ public AggsHandleFunction newInstance(ClassLoader classLoader) {
+ return new SumAggsHandleFunction(1);
+ }
+ };
+
+ private LogicalType[] inputFieldTypes = new LogicalType[]{
+ new VarCharType(VarCharType.MAX_LENGTH),
+ new BigIntType(),
+ };
+ private LogicalType[] accTypes = new LogicalType[]{ new BigIntType() };
+
+ private BinaryRowDataKeySelector keySelector = new BinaryRowDataKeySelector(new int[]{ 0 }, inputFieldTypes);
+ private TypeInformation<RowData> keyType = keySelector.getProducedType();
+
+ @Test
+ public void testRecordRetraction() throws Exception {
+ ProcTimeRangeBoundedPrecedingFunction<RowData> function = new ProcTimeRangeBoundedPrecedingFunction<>(0, 0, aggsHandleFunction, accTypes, inputFieldTypes, 2000);
+ KeyedProcessOperator<RowData, RowData, RowData> operator = new KeyedProcessOperator<>(function);
+
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);
+
+ testHarness.open();
+
+ HeapKeyedStateBackend stateBackend = (HeapKeyedStateBackend) operator.getKeyedStateBackend();
Review comment:
Use `AbstractKeyedStateBackend` here so that it will still work if the type of state backend changes to rocksdb.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org