You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by go...@apache.org on 2021/03/19 02:42:48 UTC
[flink] branch master updated: [FLINK-21805][table-planner-blink]
Support json serialization/deserialization for StreamExecRank,
StreamExecSortLimit and StreamExecLimit
This is an automated email from the ASF dual-hosted git repository.
godfrey 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 ca80fc9 [FLINK-21805][table-planner-blink] Support json serialization/deserialization for StreamExecRank, StreamExecSortLimit and StreamExecLimit
ca80fc9 is described below
commit ca80fc9e63c923893efd3d61b816bb731d5400b1
Author: 龙三 <we...@alibaba-inc.com>
AuthorDate: Tue Mar 16 15:00:44 2021 +0800
[FLINK-21805][table-planner-blink] Support json serialization/deserialization for StreamExecRank, StreamExecSortLimit and StreamExecLimit
This closes #15231
---
.../plan/nodes/exec/spec/PartitionSpec.java | 11 +-
.../planner/plan/nodes/exec/spec/SortSpec.java | 31 ++-
.../plan/nodes/exec/stream/StreamExecLimit.java | 39 +++-
.../plan/nodes/exec/stream/StreamExecRank.java | 70 +++++-
.../nodes/exec/stream/StreamExecSortLimit.java | 40 +++-
.../planner/plan/utils/RankProcessStrategy.java | 39 +++-
.../factories/TestValuesRuntimeFunctions.java | 2 +
.../nodes/exec/serde/PartitionSpecSerdeTest.java | 48 ++++
.../exec/serde/RankProcessStrategySerdeTest.java | 50 ++++
.../plan/nodes/exec/serde/RankRangeSerdeTest.java | 50 ++++
.../plan/nodes/exec/serde/RankTypeSerdeTest.java} | 35 ++-
.../plan/nodes/exec/serde/SortSpecSerdeTest.java | 54 +++++
.../nodes/exec/stream/JsonSerdeCoverageTest.java | 3 -
.../plan/nodes/exec/stream/LimitJsonPlanTest.java | 66 ++++++
.../plan/nodes/exec/stream/RankJsonPlanTest.java | 71 ++++++
.../nodes/exec/stream/SortLimitJsonPlanTest.java | 66 ++++++
.../nodes/exec/stream/TableSourceJsonPlanTest.java | 6 -
.../stream/jsonplan/LimitJsonPlanITCase.java | 50 ++++
.../stream/jsonplan/RankJsonPlanITCase.java | 53 +++++
.../stream/jsonplan/SortLimitJsonPlanITCase.java | 50 ++++
.../LimitJsonPlanTest_jsonplan/testLimit.out | 204 +++++++++++++++++
.../stream/RankJsonPlanTest_jsonplan/testRank.out | 254 +++++++++++++++++++++
.../testSortLimit.out | 213 +++++++++++++++++
.../testLimitPushDown.out | 164 +++++++++++++
.../runtime/operators/rank/ConstantRankRange.java | 26 ++-
.../rank/ConstantRankRangeWithoutEnd.java | 12 +-
.../table/runtime/operators/rank/RankRange.java | 10 +-
.../runtime/operators/rank/VariableRankRange.java | 18 +-
28 files changed, 1677 insertions(+), 58 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/PartitionSpec.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/PartitionSpec.java
index c7eeea5..53f1a37 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/PartitionSpec.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/PartitionSpec.java
@@ -18,6 +18,10 @@
package org.apache.flink.table.planner.plan.nodes.exec.spec;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
import java.util.Arrays;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -25,17 +29,22 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/** {@link PartitionSpec} describes how data is partitioned in Rank. */
public class PartitionSpec {
+ public static final String FIELD_NAME_FIELDS = "fields";
+
/** PartitionSpec makes all data in one partition. */
public static final PartitionSpec ALL_IN_ONE = new PartitionSpec(new int[0]);
/** 0-based index of field used in partitioning. */
+ @JsonProperty(FIELD_NAME_FIELDS)
private final int[] fields;
- public PartitionSpec(int[] fields) {
+ @JsonCreator
+ public PartitionSpec(@JsonProperty(FIELD_NAME_FIELDS) int[] fields) {
this.fields = checkNotNull(fields);
}
/** Gets field index of all fields in input. */
+ @JsonIgnore
public int[] getFieldIndices() {
return fields;
}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/SortSpec.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/SortSpec.java
index 3a3492b..55696da 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/SortSpec.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/SortSpec.java
@@ -22,6 +22,10 @@ import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
@@ -34,12 +38,16 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/** {@link SortSpec} describes how the data will be sorted. */
public class SortSpec {
+ public static final String FIELD_NAME_FIELDS = "fields";
+
/** SortSpec does not require sort. */
public static final SortSpec ANY = new SortSpec(new SortFieldSpec[0]);
+ @JsonProperty(FIELD_NAME_FIELDS)
private final SortFieldSpec[] fieldSpecs;
- public SortSpec(SortFieldSpec[] fieldSpecs) {
+ @JsonCreator
+ public SortSpec(@JsonProperty(FIELD_NAME_FIELDS) SortFieldSpec[] fieldSpecs) {
this.fieldSpecs = checkNotNull(fieldSpecs);
}
@@ -52,11 +60,13 @@ public class SortSpec {
}
/** Gets field index of all fields in input. */
+ @JsonIgnore
public int[] getFieldIndices() {
return Arrays.stream(fieldSpecs).mapToInt(SortFieldSpec::getFieldIndex).toArray();
}
/** Gets flags of all fields for whether to sort in ascending order or not. */
+ @JsonIgnore
public boolean[] getAscendingOrders() {
boolean[] orders = new boolean[fieldSpecs.length];
IntStream.range(0, fieldSpecs.length)
@@ -65,6 +75,7 @@ public class SortSpec {
}
/** Gets flags of all fields for whether to put null at last or not. */
+ @JsonIgnore
public boolean[] getNullsIsLast() {
boolean[] nullIsLasts = new boolean[fieldSpecs.length];
IntStream.range(0, fieldSpecs.length)
@@ -73,6 +84,7 @@ public class SortSpec {
}
/** Gets all {@link SortFieldSpec} in the SortSpec. */
+ @JsonIgnore
public SortFieldSpec[] getFieldSpecs() {
return fieldSpecs;
}
@@ -83,6 +95,7 @@ public class SortSpec {
}
/** Gets num of field in the spec. */
+ @JsonIgnore
public int getFieldSize() {
return fieldSpecs.length;
}
@@ -116,27 +129,41 @@ public class SortSpec {
/** Sort info for a Field. */
public static class SortFieldSpec {
+ public static final String FIELD_NAME_INDEX = "index";
+ public static final String FIELD_NAME_IS_ASCENDING = "isAscending";
+ public static final String FIELD_NAME_NULL_IS_LAST = "nullIsLast";
+
/** 0-based index of field being sorted. */
+ @JsonProperty(FIELD_NAME_INDEX)
private final int fieldIndex;
/** in ascending order or not. */
+ @JsonProperty(FIELD_NAME_IS_ASCENDING)
private final boolean isAscendingOrder;
/** put null at last or not. */
+ @JsonProperty(FIELD_NAME_NULL_IS_LAST)
private final boolean nullIsLast;
- public SortFieldSpec(int fieldIndex, boolean isAscendingOrder, boolean nullIsLast) {
+ @JsonCreator
+ public SortFieldSpec(
+ @JsonProperty(FIELD_NAME_INDEX) int fieldIndex,
+ @JsonProperty(FIELD_NAME_IS_ASCENDING) boolean isAscendingOrder,
+ @JsonProperty(FIELD_NAME_NULL_IS_LAST) boolean nullIsLast) {
this.fieldIndex = fieldIndex;
this.isAscendingOrder = isAscendingOrder;
this.nullIsLast = nullIsLast;
}
+ @JsonIgnore
public int getFieldIndex() {
return fieldIndex;
}
+ @JsonIgnore
public boolean getIsAscendingOrder() {
return isAscendingOrder;
}
+ @JsonIgnore
public boolean getNullIsLast() {
return nullIsLast;
}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLimit.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLimit.java
index 958b577..b1d7210 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLimit.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLimit.java
@@ -31,10 +31,19 @@ import org.apache.flink.table.runtime.operators.rank.ConstantRankRange;
import org.apache.flink.table.runtime.operators.rank.RankType;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Collections;
+import java.util.List;
+
/** Stream {@link ExecNode} for Limit. */
+@JsonIgnoreProperties(ignoreUnknown = true)
public class StreamExecLimit extends StreamExecRank {
- private final long limitEnd;
+ @JsonIgnore private final long limitEnd;
public StreamExecLimit(
long limitStart,
@@ -44,18 +53,38 @@ public class StreamExecLimit extends StreamExecRank {
InputProperty inputProperty,
RowType outputType,
String description) {
+ this(
+ new ConstantRankRange(limitStart + 1, limitEnd),
+ getRankStrategy(needRetraction),
+ generateUpdateBefore,
+ getNewNodeId(),
+ Collections.singletonList(inputProperty),
+ outputType,
+ description);
+ }
+
+ @JsonCreator
+ public StreamExecLimit(
+ @JsonProperty(FIELD_NAME_RANK_RANG) ConstantRankRange rankRange,
+ @JsonProperty(FIELD_NAME_RANK_STRATEGY) RankProcessStrategy rankStrategy,
+ @JsonProperty(FIELD_NAME_GENERATE_UPDATE_BEFORE) boolean generateUpdateBefore,
+ @JsonProperty(FIELD_NAME_ID) int id,
+ @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
+ @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
+ @JsonProperty(FIELD_NAME_DESCRIPTION) String description) {
super(
RankType.ROW_NUMBER,
PartitionSpec.ALL_IN_ONE,
SortSpec.ANY,
- new ConstantRankRange(limitStart + 1, limitEnd),
- getRankStrategy(needRetraction),
+ rankRange,
+ rankStrategy,
false, // outputRankNumber
generateUpdateBefore,
- inputProperty,
+ id,
+ inputProperties,
outputType,
description);
- this.limitEnd = limitEnd;
+ this.limitEnd = rankRange.getRankEnd();
}
private static RankProcessStrategy getRankStrategy(boolean needRetraction) {
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecRank.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecRank.java
index 7804cc9..b616042 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecRank.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecRank.java
@@ -53,10 +53,19 @@ import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
import java.util.Collections;
+import java.util.List;
import java.util.stream.IntStream;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
/** Stream {@link ExecNode} for Rank. */
+@JsonIgnoreProperties(ignoreUnknown = true)
public class StreamExecRank extends ExecNodeBase<RowData>
implements StreamExecNode<RowData>, SingleTransformationTranslator<RowData> {
@@ -70,12 +79,33 @@ public class StreamExecRank extends ExecNodeBase<RowData>
"TopN operator has a cache which caches partial state contents to reduce"
+ " state access. Cache size is the number of records in each TopN task.");
+ public static final String FIELD_NAME_RANK_TYPE = "rankType";
+ public static final String FIELD_NAME_PARTITION_SPEC = "partitionSpec";
+ public static final String FIELD_NAME_SORT_SPEC = "sortSpec";
+ public static final String FIELD_NAME_RANK_RANG = "rankRange";
+ public static final String FIELD_NAME_RANK_STRATEGY = "rankStrategy";
+ public static final String FIELD_NAME_GENERATE_UPDATE_BEFORE = "generateUpdateBefore";
+ public static final String FIELD_NAME_OUTPUT_RANK_NUMBER = "outputRowNumber";
+
+ @JsonProperty(FIELD_NAME_RANK_TYPE)
private final RankType rankType;
+
+ @JsonProperty(FIELD_NAME_PARTITION_SPEC)
private final PartitionSpec partitionSpec;
+
+ @JsonProperty(FIELD_NAME_SORT_SPEC)
private final SortSpec sortSpec;
+
+ @JsonProperty(FIELD_NAME_RANK_RANG)
private final RankRange rankRange;
+
+ @JsonProperty(FIELD_NAME_RANK_STRATEGY)
private final RankProcessStrategy rankStrategy;
+
+ @JsonProperty(FIELD_NAME_OUTPUT_RANK_NUMBER)
private final boolean outputRankNumber;
+
+ @JsonProperty(FIELD_NAME_GENERATE_UPDATE_BEFORE)
private final boolean generateUpdateBefore;
public StreamExecRank(
@@ -89,12 +119,40 @@ public class StreamExecRank extends ExecNodeBase<RowData>
InputProperty inputProperty,
RowType outputType,
String description) {
- super(Collections.singletonList(inputProperty), outputType, description);
- this.rankType = rankType;
- this.rankRange = rankRange;
- this.rankStrategy = rankStrategy;
- this.sortSpec = sortSpec;
- this.partitionSpec = partitionSpec;
+ this(
+ rankType,
+ partitionSpec,
+ sortSpec,
+ rankRange,
+ rankStrategy,
+ outputRankNumber,
+ generateUpdateBefore,
+ getNewNodeId(),
+ Collections.singletonList(inputProperty),
+ outputType,
+ description);
+ }
+
+ @JsonCreator
+ public StreamExecRank(
+ @JsonProperty(FIELD_NAME_RANK_TYPE) RankType rankType,
+ @JsonProperty(FIELD_NAME_PARTITION_SPEC) PartitionSpec partitionSpec,
+ @JsonProperty(FIELD_NAME_SORT_SPEC) SortSpec sortSpec,
+ @JsonProperty(FIELD_NAME_RANK_RANG) RankRange rankRange,
+ @JsonProperty(FIELD_NAME_RANK_STRATEGY) RankProcessStrategy rankStrategy,
+ @JsonProperty(FIELD_NAME_OUTPUT_RANK_NUMBER) boolean outputRankNumber,
+ @JsonProperty(FIELD_NAME_GENERATE_UPDATE_BEFORE) boolean generateUpdateBefore,
+ @JsonProperty(FIELD_NAME_ID) int id,
+ @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
+ @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
+ @JsonProperty(FIELD_NAME_DESCRIPTION) String description) {
+ super(id, inputProperties, outputType, description);
+ checkArgument(inputProperties.size() == 1);
+ this.rankType = checkNotNull(rankType);
+ this.rankRange = checkNotNull(rankRange);
+ this.rankStrategy = checkNotNull(rankStrategy);
+ this.sortSpec = checkNotNull(sortSpec);
+ this.partitionSpec = checkNotNull(partitionSpec);
this.outputRankNumber = outputRankNumber;
this.generateUpdateBefore = generateUpdateBefore;
}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSortLimit.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSortLimit.java
index 74791af..ed27223 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSortLimit.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSortLimit.java
@@ -30,10 +30,19 @@ import org.apache.flink.table.runtime.operators.rank.ConstantRankRange;
import org.apache.flink.table.runtime.operators.rank.RankType;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Collections;
+import java.util.List;
+
/** {@link StreamExecNode} for Sort with limit. */
+@JsonIgnoreProperties(ignoreUnknown = true)
public class StreamExecSortLimit extends StreamExecRank {
- private final long limitEnd;
+ @JsonIgnore private final long limitEnd;
public StreamExecSortLimit(
SortSpec sortSpec,
@@ -44,18 +53,41 @@ public class StreamExecSortLimit extends StreamExecRank {
InputProperty inputProperty,
RowType outputType,
String description) {
+ this(
+ sortSpec,
+ new ConstantRankRange(limitStart + 1, limitEnd),
+ rankStrategy,
+ generateUpdateBefore,
+ getNewNodeId(),
+ Collections.singletonList(inputProperty),
+ outputType,
+ description);
+ }
+
+ @JsonCreator
+ public StreamExecSortLimit(
+ @JsonProperty(FIELD_NAME_SORT_SPEC) SortSpec sortSpec,
+ @JsonProperty(FIELD_NAME_RANK_RANG) ConstantRankRange rankRange,
+ @JsonProperty(FIELD_NAME_RANK_STRATEGY) RankProcessStrategy rankStrategy,
+ @JsonProperty(FIELD_NAME_GENERATE_UPDATE_BEFORE) boolean generateUpdateBefore,
+ @JsonProperty(FIELD_NAME_ID) int id,
+ @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
+ @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
+ @JsonProperty(FIELD_NAME_DESCRIPTION) String description) {
+
super(
RankType.ROW_NUMBER,
PartitionSpec.ALL_IN_ONE,
sortSpec,
- new ConstantRankRange(limitStart + 1, limitEnd),
+ rankRange,
rankStrategy,
false, // outputRankNumber
generateUpdateBefore,
- inputProperty,
+ id,
+ inputProperties,
outputType,
description);
- this.limitEnd = limitEnd;
+ this.limitEnd = rankRange.getRankEnd();
}
@Override
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/utils/RankProcessStrategy.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/utils/RankProcessStrategy.java
index 079c767..074b45d 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/utils/RankProcessStrategy.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/utils/RankProcessStrategy.java
@@ -23,6 +23,14 @@ import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalR
import org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram;
import org.apache.flink.table.planner.plan.trait.RelModifiedMonotonicity;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonSubTypes;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
+
import org.apache.calcite.rel.RelCollation;
import org.apache.calcite.rel.RelFieldCollation;
import org.apache.calcite.rel.RelNode;
@@ -37,6 +45,13 @@ import java.util.List;
import java.util.Set;
/** Base class of Strategy to choose different rank process function. */
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
+@JsonSubTypes({
+ @JsonSubTypes.Type(value = RankProcessStrategy.UndefinedStrategy.class),
+ @JsonSubTypes.Type(value = RankProcessStrategy.AppendFastStrategy.class),
+ @JsonSubTypes.Type(value = RankProcessStrategy.RetractStrategy.class),
+ @JsonSubTypes.Type(value = RankProcessStrategy.UpdateFastStrategy.class)
+})
public interface RankProcessStrategy {
UndefinedStrategy UNDEFINED_STRATEGY = new UndefinedStrategy();
@@ -49,8 +64,11 @@ public interface RankProcessStrategy {
* A placeholder strategy which will be inferred after {@link
* FlinkChangelogModeInferenceProgram}.
*/
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ @JsonTypeName("Undefined")
class UndefinedStrategy implements RankProcessStrategy {
- private UndefinedStrategy() {}
+ @JsonCreator
+ public UndefinedStrategy() {}
@Override
public String toString() {
@@ -59,8 +77,11 @@ public interface RankProcessStrategy {
}
/** A strategy which only works when input only contains insertion changes. */
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ @JsonTypeName("AppendFast")
class AppendFastStrategy implements RankProcessStrategy {
- private AppendFastStrategy() {}
+ @JsonCreator
+ public AppendFastStrategy() {}
@Override
public String toString() {
@@ -69,8 +90,11 @@ public interface RankProcessStrategy {
}
/** A strategy which works when input contains update or deletion changes. */
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ @JsonTypeName("Retract")
class RetractStrategy implements RankProcessStrategy {
- private RetractStrategy() {}
+ @JsonCreator
+ public RetractStrategy() {}
@Override
public String toString() {
@@ -82,14 +106,21 @@ public interface RankProcessStrategy {
* A strategy which only works when input shouldn't contains deletion changes and input should
* have the given {@link #primaryKeys} and should be monotonic on the order by field.
*/
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ @JsonTypeName("UpdateFast")
class UpdateFastStrategy implements RankProcessStrategy {
+ public static final String FIELD_NAME_PRIMARY_KEYS = "primaryKeys";
+
+ @JsonProperty(FIELD_NAME_PRIMARY_KEYS)
private final int[] primaryKeys;
- public UpdateFastStrategy(int[] primaryKeys) {
+ @JsonCreator
+ public UpdateFastStrategy(@JsonProperty(FIELD_NAME_PRIMARY_KEYS) int[] primaryKeys) {
this.primaryKeys = primaryKeys;
}
+ @JsonIgnore
public int[] getPrimaryKeys() {
return primaryKeys;
}
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java
index 3796e7c..82b4bdd 100644
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java
@@ -507,8 +507,10 @@ final class TestValuesRuntimeFunctions {
assert row != null;
localRawResult.add(kind.shortString() + "(" + row.toString() + ")");
if (kind == RowKind.INSERT || kind == RowKind.UPDATE_AFTER) {
+ row.setKind(RowKind.INSERT);
localRetractResult.add(row.toString());
} else {
+ row.setKind(RowKind.INSERT);
boolean contains = localRetractResult.remove(row.toString());
if (!contains) {
throw new RuntimeException(
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/PartitionSpecSerdeTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/PartitionSpecSerdeTest.java
new file mode 100644
index 0000000..3c37a14
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/PartitionSpecSerdeTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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.planner.plan.nodes.exec.serde;
+
+import org.apache.flink.table.planner.plan.nodes.exec.spec.PartitionSpec;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/** Test PartitionSpec json ser/de. */
+public class PartitionSpecSerdeTest {
+
+ @Test
+ public void testPartitionSpec() throws JsonProcessingException {
+ PartitionSpec spec = new PartitionSpec(new int[] {1, 2, 3});
+ ObjectMapper mapper = new ObjectMapper();
+ assertEquals(spec, mapper.readValue(mapper.writeValueAsString(spec), PartitionSpec.class));
+ }
+
+ @Test
+ public void testAllInOne() throws JsonProcessingException {
+ ObjectMapper mapper = new ObjectMapper();
+ assertEquals(
+ PartitionSpec.ALL_IN_ONE,
+ mapper.readValue(
+ mapper.writeValueAsString(PartitionSpec.ALL_IN_ONE), PartitionSpec.class));
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RankProcessStrategySerdeTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RankProcessStrategySerdeTest.java
new file mode 100644
index 0000000..46085fb
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RankProcessStrategySerdeTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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.planner.plan.nodes.exec.serde;
+
+import org.apache.flink.table.planner.plan.utils.RankProcessStrategy;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/** Test RankProcessStrategy json ser/de. */
+public class RankProcessStrategySerdeTest {
+
+ @Test
+ public void testRankRange() throws JsonProcessingException {
+ ObjectMapper mapper = new ObjectMapper();
+ RankProcessStrategy[] strategies =
+ new RankProcessStrategy[] {
+ RankProcessStrategy.UNDEFINED_STRATEGY,
+ RankProcessStrategy.APPEND_FAST_STRATEGY,
+ RankProcessStrategy.RETRACT_STRATEGY,
+ new RankProcessStrategy.UpdateFastStrategy(new int[] {1, 2})
+ };
+ for (RankProcessStrategy strategy : strategies) {
+ RankProcessStrategy result =
+ mapper.readValue(
+ mapper.writeValueAsString(strategy), RankProcessStrategy.class);
+ assertEquals(strategy.toString(), result.toString());
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RankRangeSerdeTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RankRangeSerdeTest.java
new file mode 100644
index 0000000..1d9125a
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RankRangeSerdeTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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.planner.plan.nodes.exec.serde;
+
+import org.apache.flink.table.runtime.operators.rank.ConstantRankRange;
+import org.apache.flink.table.runtime.operators.rank.ConstantRankRangeWithoutEnd;
+import org.apache.flink.table.runtime.operators.rank.RankRange;
+import org.apache.flink.table.runtime.operators.rank.VariableRankRange;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/** Test RankRange json ser/de. */
+public class RankRangeSerdeTest {
+
+ @Test
+ public void testRankRange() throws JsonProcessingException {
+ ObjectMapper mapper = new ObjectMapper();
+ RankRange[] ranges =
+ new RankRange[] {
+ new ConstantRankRange(1, 2),
+ new VariableRankRange(3),
+ new ConstantRankRangeWithoutEnd(4)
+ };
+ for (RankRange range : ranges) {
+ RankRange result = mapper.readValue(mapper.writeValueAsString(range), RankRange.class);
+ assertEquals(range.toString(), result.toString());
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/VariableRankRange.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RankTypeSerdeTest.java
similarity index 51%
copy from flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/VariableRankRange.java
copy to flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RankTypeSerdeTest.java
index cce9b04..2ff6584 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/VariableRankRange.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RankTypeSerdeTest.java
@@ -16,31 +16,26 @@
* limitations under the License.
*/
-package org.apache.flink.table.runtime.operators.rank;
+package org.apache.flink.table.planner.plan.nodes.exec.serde;
-import java.util.List;
+import org.apache.flink.table.runtime.operators.rank.RankType;
-/** changing rank limit depends on input. */
-public class VariableRankRange implements RankRange {
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
- private static final long serialVersionUID = 5579785886506433955L;
- private int rankEndIndex;
+import org.junit.Test;
- public VariableRankRange(int rankEndIndex) {
- this.rankEndIndex = rankEndIndex;
- }
-
- public int getRankEndIndex() {
- return rankEndIndex;
- }
+import static org.junit.Assert.assertEquals;
- @Override
- public String toString(List<String> inputFieldNames) {
- return "rankEnd=" + inputFieldNames.get(rankEndIndex);
- }
+/** Test RankType json ser/de. */
+public class RankTypeSerdeTest {
- @Override
- public String toString() {
- return "rankEnd=$" + rankEndIndex;
+ @Test
+ public void testRankType() throws JsonProcessingException {
+ ObjectMapper mapper = new ObjectMapper();
+ for (RankType type : RankType.values()) {
+ RankType result = mapper.readValue(mapper.writeValueAsString(type), RankType.class);
+ assertEquals(type, result);
+ }
}
}
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/SortSpecSerdeTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/SortSpecSerdeTest.java
new file mode 100644
index 0000000..8dd3ca6
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/SortSpecSerdeTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.planner.plan.nodes.exec.serde;
+
+import org.apache.flink.table.planner.plan.nodes.exec.spec.SortSpec;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/** Test SortSpec json ser/de. */
+public class SortSpecSerdeTest {
+
+ @Test
+ public void testSortSpec() throws JsonProcessingException {
+ SortSpec sortSpec =
+ SortSpec.builder()
+ .addField(1, true, true)
+ .addField(2, true, false)
+ .addField(3, false, true)
+ .addField(4, false, false)
+ .build();
+ ObjectMapper mapper = new ObjectMapper();
+ assertEquals(
+ sortSpec, mapper.readValue(mapper.writeValueAsString(sortSpec), SortSpec.class));
+ }
+
+ @Test
+ public void testAny() throws JsonProcessingException {
+ ObjectMapper mapper = new ObjectMapper();
+ assertEquals(
+ SortSpec.ANY,
+ mapper.readValue(mapper.writeValueAsString(SortSpec.ANY), SortSpec.class));
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/JsonSerdeCoverageTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/JsonSerdeCoverageTest.java
index 6806e17..f24452f 100644
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/JsonSerdeCoverageTest.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/JsonSerdeCoverageTest.java
@@ -58,10 +58,7 @@ public class JsonSerdeCoverageTest {
"StreamExecPythonOverAggregate",
"StreamExecCorrelate",
"StreamExecPythonCorrelate",
- "StreamExecRank",
"StreamExecPythonCalc",
- "StreamExecLimit",
- "StreamExecSortLimit",
"StreamExecTemporalSort",
"StreamExecSort",
"StreamExecExpand",
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LimitJsonPlanTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LimitJsonPlanTest.java
new file mode 100644
index 0000000..062edca
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LimitJsonPlanTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.Before;
+import org.junit.Test;
+
+/** Test json serialization for sort limit. */
+public class LimitJsonPlanTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+ private TableEnvironment tEnv;
+
+ @Before
+ public void setup() {
+ util = streamTestUtil(TableConfig.getDefault());
+ tEnv = util.getTableEnv();
+
+ String srcTableDdl =
+ "CREATE TABLE MyTable (\n"
+ + " a bigint,\n"
+ + " b int not null,\n"
+ + " c varchar,\n"
+ + " d timestamp(3)\n"
+ + ") with (\n"
+ + " 'connector' = 'values',\n"
+ + " 'bounded' = 'false')";
+ tEnv.executeSql(srcTableDdl);
+ }
+
+ @Test
+ public void testLimit() {
+ String sinkTableDdl =
+ "CREATE TABLE MySink (\n"
+ + " a bigint,\n"
+ + " b bigint\n"
+ + ") with (\n"
+ + " 'connector' = 'values',\n"
+ + " 'sink-insert-only' = 'false',\n"
+ + " 'table-sink-class' = 'DEFAULT')";
+ tEnv.executeSql(sinkTableDdl);
+ String sql = "insert into MySink SELECT a, a from MyTable limit 10";
+ util.verifyJsonPlan(sql);
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest.java
new file mode 100644
index 0000000..ab3e141
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.Before;
+import org.junit.Test;
+
+/** Test json serialization for rank. */
+public class RankJsonPlanTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+ private TableEnvironment tEnv;
+
+ @Before
+ public void setup() {
+ util = streamTestUtil(TableConfig.getDefault());
+ tEnv = util.getTableEnv();
+
+ String srcTableDdl =
+ "CREATE TABLE MyTable (\n"
+ + " a bigint,\n"
+ + " b int not null,\n"
+ + " c varchar,\n"
+ + " d timestamp(3)\n"
+ + ") with (\n"
+ + " 'connector' = 'values',\n"
+ + " 'bounded' = 'false')";
+ tEnv.executeSql(srcTableDdl);
+ }
+
+ @Test
+ public void testRank() {
+ String sinkTableDdl =
+ "CREATE TABLE MySink (\n"
+ + " a bigint,\n"
+ + " b bigint\n"
+ + ") with (\n"
+ + " 'connector' = 'values',\n"
+ + " 'sink-insert-only' = 'false',\n"
+ + " 'table-sink-class' = 'DEFAULT')";
+ tEnv.executeSql(sinkTableDdl);
+ String sql =
+ "insert into MySink SELECT a, row_num\n"
+ + "FROM (\n"
+ + " SELECT a, ROW_NUMBER() OVER (PARTITION BY b ORDER BY a) as row_num\n"
+ + " FROM MyTable)\n"
+ + "WHERE row_num <= a";
+ util.verifyJsonPlan(sql);
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/SortLimitJsonPlanTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/SortLimitJsonPlanTest.java
new file mode 100644
index 0000000..089f4d6
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/SortLimitJsonPlanTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.Before;
+import org.junit.Test;
+
+/** Test json serialization for sort limit. */
+public class SortLimitJsonPlanTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+ private TableEnvironment tEnv;
+
+ @Before
+ public void setup() {
+ util = streamTestUtil(TableConfig.getDefault());
+ tEnv = util.getTableEnv();
+
+ String srcTableDdl =
+ "CREATE TABLE MyTable (\n"
+ + " a bigint,\n"
+ + " b int not null,\n"
+ + " c varchar,\n"
+ + " d timestamp(3)\n"
+ + ") with (\n"
+ + " 'connector' = 'values',\n"
+ + " 'bounded' = 'false')";
+ tEnv.executeSql(srcTableDdl);
+ }
+
+ @Test
+ public void testSortLimit() {
+ String sinkTableDdl =
+ "CREATE TABLE MySink (\n"
+ + " a bigint,\n"
+ + " b bigint\n"
+ + ") with (\n"
+ + " 'connector' = 'values',\n"
+ + " 'sink-insert-only' = 'false',\n"
+ + " 'table-sink-class' = 'DEFAULT')";
+ tEnv.executeSql(sinkTableDdl);
+ String sql = "insert into MySink SELECT a, a from MyTable order by b limit 10";
+ util.verifyJsonPlan(sql);
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSourceJsonPlanTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSourceJsonPlanTest.java
index c142bb6..96e1c09 100644
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSourceJsonPlanTest.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSourceJsonPlanTest.java
@@ -20,7 +20,6 @@ package org.apache.flink.table.planner.plan.nodes.exec.stream;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.table.api.TableException;
import org.apache.flink.table.planner.utils.StreamTableTestUtil;
import org.apache.flink.table.planner.utils.TableTestBase;
@@ -118,11 +117,6 @@ public class TableSourceJsonPlanTest extends TableTestBase {
@Test
public void testLimitPushDown() {
- exception.expect(TableException.class);
- // currently, there is a StreamExecLimit in the plan, once StreamExecLimit does support
- // json serialization/deserialization, the following exception checking should be removed.
- exception.expectMessage(
- "StreamExecLimit does not implement @JsonCreator annotation on constructor");
util.verifyJsonPlan("insert into MySink select * from MyTable limit 3");
}
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/LimitJsonPlanITCase.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/LimitJsonPlanITCase.java
new file mode 100644
index 0000000..66e37b8
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/LimitJsonPlanITCase.java
@@ -0,0 +1,50 @@
+/*
+ * 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.planner.runtime.stream.jsonplan;
+
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.TestData;
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
+import org.apache.flink.table.planner.utils.JsonPlanTestBase;
+
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+/** Test for limit JsonPlan ser/de. */
+public class LimitJsonPlanITCase extends JsonPlanTestBase {
+ @Test
+ public void testLimit() throws ExecutionException, InterruptedException, IOException {
+ createTestValuesSourceTable(
+ "MyTable",
+ JavaScalaConversionUtil.toJava(TestData.data1()),
+ "a int",
+ "b varchar",
+ "c int");
+ createTestNonInsertOnlyValuesSinkTable("`result`", "a int", "b varchar", "c bigint");
+ String sql = "insert into `result` select * from MyTable limit 3";
+ executeSqlWithJsonPlanVerified(sql).await();
+
+ List<String> expected = Arrays.asList("+I[2, a, 6]", "+I[4, b, 8]", "+I[6, c, 10]");
+ assertResult(expected, TestValuesTableFactory.getResults("result"));
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/RankJsonPlanITCase.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/RankJsonPlanITCase.java
new file mode 100644
index 0000000..7e76bc2
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/RankJsonPlanITCase.java
@@ -0,0 +1,53 @@
+/*
+ * 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.planner.runtime.stream.jsonplan;
+
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.TestData;
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
+import org.apache.flink.table.planner.utils.JsonPlanTestBase;
+
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+/** Test for Rank JsonPlan ser/de. */
+public class RankJsonPlanITCase extends JsonPlanTestBase {
+ @Test
+ public void testRank() throws ExecutionException, InterruptedException, IOException {
+ createTestValuesSourceTable(
+ "MyTable",
+ JavaScalaConversionUtil.toJava(TestData.data1()),
+ "a int",
+ "b varchar",
+ "c int");
+ createTestNonInsertOnlyValuesSinkTable("`result`", "a int", "b varchar", "c bigint");
+ String sql =
+ "insert into `result` select * from "
+ + "(select a, b, row_number() over(partition by b order by c) as c from MyTable)"
+ + " where c = 1";
+ executeSqlWithJsonPlanVerified(sql).await();
+
+ List<String> expected = Arrays.asList("+I[1, a, 1]", "+I[3, b, 1]", "+I[5, c, 1]");
+ assertResult(expected, TestValuesTableFactory.getResults("result"));
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/SortLimitJsonPlanITCase.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/SortLimitJsonPlanITCase.java
new file mode 100644
index 0000000..438814c
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/SortLimitJsonPlanITCase.java
@@ -0,0 +1,50 @@
+/*
+ * 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.planner.runtime.stream.jsonplan;
+
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.TestData;
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
+import org.apache.flink.table.planner.utils.JsonPlanTestBase;
+
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+/** Test for sort limit JsonPlan ser/de. */
+public class SortLimitJsonPlanITCase extends JsonPlanTestBase {
+ @Test
+ public void testSortLimit() throws ExecutionException, InterruptedException, IOException {
+ createTestValuesSourceTable(
+ "MyTable",
+ JavaScalaConversionUtil.toJava(TestData.data1()),
+ "a int",
+ "b varchar",
+ "c int");
+ createTestNonInsertOnlyValuesSinkTable("`result`", "a int", "b varchar", "c bigint");
+ String sql = "insert into `result` select * from MyTable order by a limit 3";
+ executeSqlWithJsonPlanVerified(sql).await();
+
+ List<String> expected = Arrays.asList("+I[1, a, 5]", "+I[2, a, 6]", "+I[3, b, 7]");
+ assertResult(expected, TestValuesTableFactory.getResults("result"));
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LimitJsonPlanTest_jsonplan/testLimit.out b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LimitJsonPlanTest_jsonplan/testLimit.out
new file mode 100644
index 0000000..d0d8d5b
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/LimitJsonPlanTest_jsonplan/testLimit.out
@@ -0,0 +1,204 @@
+{
+ "flinkVersion" : "",
+ "nodes" : [ {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecTableSourceScan",
+ "scanTableSource" : {
+ "identifier" : {
+ "catalogName" : "default_catalog",
+ "databaseName" : "default_database",
+ "tableName" : "MyTable"
+ },
+ "catalogTable" : {
+ "schema.3.data-type" : "TIMESTAMP(3)",
+ "schema.2.data-type" : "VARCHAR(2147483647)",
+ "schema.3.name" : "d",
+ "connector" : "values",
+ "schema.0.data-type" : "BIGINT",
+ "schema.2.name" : "c",
+ "schema.1.name" : "b",
+ "bounded" : "false",
+ "schema.0.name" : "a",
+ "schema.1.data-type" : "INT NOT NULL"
+ },
+ "sourceAbilitySpecs" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ] ],
+ "producedType" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "a" : "BIGINT"
+ } ]
+ }
+ }, {
+ "type" : "LimitPushDown",
+ "limit" : 10
+ } ]
+ },
+ "id" : 15,
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ } ]
+ },
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a], limit=[10]]], fields=[a])",
+ "inputProperties" : [ ]
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecExchange",
+ "id" : 16,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ } ]
+ },
+ "description" : "Exchange(distribution=[single])"
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecLimit",
+ "rankRange" : {
+ "type" : "Constant",
+ "start" : 1,
+ "end" : 10
+ },
+ "rankStrategy" : {
+ "type" : "AppendFast"
+ },
+ "generateUpdateBefore" : false,
+ "id" : 17,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ } ]
+ },
+ "description" : "Limit(offset=[0], fetch=[10])",
+ "rankType" : "ROW_NUMBER",
+ "partitionSpec" : {
+ "fields" : [ ]
+ },
+ "sortSpec" : {
+ "fields" : [ ]
+ },
+ "outputRowNumber" : false
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : true
+ }
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : true
+ }
+ } ],
+ "condition" : null,
+ "id" : 18,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "a0" : "BIGINT"
+ } ]
+ },
+ "description" : "Calc(select=[a, a AS a0])"
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink",
+ "dynamicTableSink" : {
+ "identifier" : {
+ "catalogName" : "default_catalog",
+ "databaseName" : "default_database",
+ "tableName" : "MySink"
+ },
+ "catalogTable" : {
+ "sink-insert-only" : "false",
+ "table-sink-class" : "DEFAULT",
+ "connector" : "values",
+ "schema.0.data-type" : "BIGINT",
+ "schema.1.name" : "b",
+ "schema.0.name" : "a",
+ "schema.1.data-type" : "BIGINT"
+ }
+ },
+ "inputChangelogMode" : [ "INSERT" ],
+ "id" : 19,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "a0" : "BIGINT"
+ } ]
+ },
+ "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, a0])"
+ } ],
+ "edges" : [ {
+ "source" : 15,
+ "target" : 16,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 16,
+ "target" : 17,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 17,
+ "target" : 18,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 18,
+ "target" : 19,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out
new file mode 100644
index 0000000..9f83a94
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out
@@ -0,0 +1,254 @@
+{
+ "flinkVersion" : "",
+ "nodes" : [ {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecTableSourceScan",
+ "scanTableSource" : {
+ "identifier" : {
+ "catalogName" : "default_catalog",
+ "databaseName" : "default_database",
+ "tableName" : "MyTable"
+ },
+ "catalogTable" : {
+ "schema.3.data-type" : "TIMESTAMP(3)",
+ "schema.2.data-type" : "VARCHAR(2147483647)",
+ "schema.3.name" : "d",
+ "connector" : "values",
+ "schema.0.data-type" : "BIGINT",
+ "schema.2.name" : "c",
+ "schema.1.name" : "b",
+ "bounded" : "false",
+ "schema.0.name" : "a",
+ "schema.1.data-type" : "INT NOT NULL"
+ }
+ },
+ "id" : 9,
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "b" : "INT NOT NULL"
+ }, {
+ "c" : "VARCHAR(2147483647)"
+ }, {
+ "d" : {
+ "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
+ "nullable" : true,
+ "precision" : 3,
+ "kind" : "REGULAR"
+ }
+ } ]
+ },
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])",
+ "inputProperties" : [ ]
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : true
+ }
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : {
+ "typeName" : "INTEGER",
+ "nullable" : false
+ }
+ } ],
+ "condition" : null,
+ "id" : 10,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "b" : "INT NOT NULL"
+ } ]
+ },
+ "description" : "Calc(select=[a, b])"
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecExchange",
+ "id" : 11,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 1 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "b" : "INT NOT NULL"
+ } ]
+ },
+ "description" : "Exchange(distribution=[hash[b]])"
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecRank",
+ "rankType" : "ROW_NUMBER",
+ "partitionSpec" : {
+ "fields" : [ 1 ]
+ },
+ "sortSpec" : {
+ "fields" : [ {
+ "index" : 0,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "rankRange" : {
+ "type" : "Variable",
+ "endIndex" : 0
+ },
+ "rankStrategy" : {
+ "type" : "AppendFast"
+ },
+ "outputRowNumber" : true,
+ "generateUpdateBefore" : true,
+ "id" : 12,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "b" : "INT NOT NULL"
+ }, {
+ "w0$o0" : "BIGINT NOT NULL"
+ } ]
+ },
+ "description" : "Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER], rankRange=[rankEnd=a], partitionBy=[b], orderBy=[a ASC], select=[a, b, w0$o0])"
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : true
+ }
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : false
+ }
+ } ],
+ "condition" : null,
+ "id" : 13,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "w0$o0" : "BIGINT NOT NULL"
+ } ]
+ },
+ "description" : "Calc(select=[a, w0$o0])"
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink",
+ "dynamicTableSink" : {
+ "identifier" : {
+ "catalogName" : "default_catalog",
+ "databaseName" : "default_database",
+ "tableName" : "MySink"
+ },
+ "catalogTable" : {
+ "sink-insert-only" : "false",
+ "table-sink-class" : "DEFAULT",
+ "connector" : "values",
+ "schema.0.data-type" : "BIGINT",
+ "schema.1.name" : "b",
+ "schema.0.name" : "a",
+ "schema.1.data-type" : "BIGINT"
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE" ],
+ "id" : 14,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "w0$o0" : "BIGINT NOT NULL"
+ } ]
+ },
+ "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, w0$o0])"
+ } ],
+ "edges" : [ {
+ "source" : 9,
+ "target" : 10,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 10,
+ "target" : 11,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 11,
+ "target" : 12,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 12,
+ "target" : 13,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 13,
+ "target" : 14,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/SortLimitJsonPlanTest_jsonplan/testSortLimit.out b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/SortLimitJsonPlanTest_jsonplan/testSortLimit.out
new file mode 100644
index 0000000..3090fc4
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/SortLimitJsonPlanTest_jsonplan/testSortLimit.out
@@ -0,0 +1,213 @@
+{
+ "flinkVersion" : "",
+ "nodes" : [ {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecTableSourceScan",
+ "scanTableSource" : {
+ "identifier" : {
+ "catalogName" : "default_catalog",
+ "databaseName" : "default_database",
+ "tableName" : "MyTable"
+ },
+ "catalogTable" : {
+ "schema.3.data-type" : "TIMESTAMP(3)",
+ "schema.2.data-type" : "VARCHAR(2147483647)",
+ "schema.3.name" : "d",
+ "connector" : "values",
+ "schema.0.data-type" : "BIGINT",
+ "schema.2.name" : "c",
+ "schema.1.name" : "b",
+ "bounded" : "false",
+ "schema.0.name" : "a",
+ "schema.1.data-type" : "INT NOT NULL"
+ },
+ "sourceAbilitySpecs" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 1 ] ],
+ "producedType" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "b" : "INT NOT NULL"
+ } ]
+ }
+ } ]
+ },
+ "id" : 24,
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "b" : "INT NOT NULL"
+ } ]
+ },
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b]]], fields=[a, b])",
+ "inputProperties" : [ ]
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecExchange",
+ "id" : 25,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "b" : "INT NOT NULL"
+ } ]
+ },
+ "description" : "Exchange(distribution=[single])"
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSortLimit",
+ "sortSpec" : {
+ "fields" : [ {
+ "index" : 1,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "rankRange" : {
+ "type" : "Constant",
+ "start" : 1,
+ "end" : 10
+ },
+ "rankStrategy" : {
+ "type" : "AppendFast"
+ },
+ "generateUpdateBefore" : true,
+ "id" : 26,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "b" : "INT NOT NULL"
+ } ]
+ },
+ "description" : "SortLimit(orderBy=[b ASC], offset=[0], fetch=[10], strategy=[AppendFastStrategy])",
+ "rankType" : "ROW_NUMBER",
+ "partitionSpec" : {
+ "fields" : [ ]
+ },
+ "outputRowNumber" : false
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : true
+ }
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : true
+ }
+ } ],
+ "condition" : null,
+ "id" : 27,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "a1" : "BIGINT"
+ } ]
+ },
+ "description" : "Calc(select=[a, a AS a1])"
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink",
+ "dynamicTableSink" : {
+ "identifier" : {
+ "catalogName" : "default_catalog",
+ "databaseName" : "default_database",
+ "tableName" : "MySink"
+ },
+ "catalogTable" : {
+ "sink-insert-only" : "false",
+ "table-sink-class" : "DEFAULT",
+ "connector" : "values",
+ "schema.0.data-type" : "BIGINT",
+ "schema.1.name" : "b",
+ "schema.0.name" : "a",
+ "schema.1.data-type" : "BIGINT"
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE" ],
+ "id" : 28,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "a1" : "BIGINT"
+ } ]
+ },
+ "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, a1])"
+ } ],
+ "edges" : [ {
+ "source" : 24,
+ "target" : 25,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 25,
+ "target" : 26,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 26,
+ "target" : 27,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 27,
+ "target" : 28,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSourceJsonPlanTest_jsonplan/testLimitPushDown.out b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSourceJsonPlanTest_jsonplan/testLimitPushDown.out
new file mode 100644
index 0000000..d53d7c9
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSourceJsonPlanTest_jsonplan/testLimitPushDown.out
@@ -0,0 +1,164 @@
+{
+ "flinkVersion" : "",
+ "nodes" : [ {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecTableSourceScan",
+ "scanTableSource" : {
+ "identifier" : {
+ "catalogName" : "default_catalog",
+ "databaseName" : "default_database",
+ "tableName" : "MyTable"
+ },
+ "catalogTable" : {
+ "schema.2.data-type" : "VARCHAR(2147483647)",
+ "connector" : "values",
+ "schema.0.data-type" : "BIGINT",
+ "schema.2.name" : "c",
+ "schema.1.name" : "b",
+ "bounded" : "false",
+ "schema.0.name" : "a",
+ "schema.1.data-type" : "INT"
+ },
+ "sourceAbilitySpecs" : [ {
+ "type" : "LimitPushDown",
+ "limit" : 3
+ } ]
+ },
+ "id" : 47,
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "b" : "INT"
+ }, {
+ "c" : "VARCHAR(2147483647)"
+ } ]
+ },
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable, limit=[3]]], fields=[a, b, c])",
+ "inputProperties" : [ ]
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecExchange",
+ "id" : 48,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "b" : "INT"
+ }, {
+ "c" : "VARCHAR(2147483647)"
+ } ]
+ },
+ "description" : "Exchange(distribution=[single])"
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecLimit",
+ "rankRange" : {
+ "type" : "Constant",
+ "start" : 1,
+ "end" : 3
+ },
+ "rankStrategy" : {
+ "type" : "AppendFast"
+ },
+ "generateUpdateBefore" : false,
+ "id" : 49,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "b" : "INT"
+ }, {
+ "c" : "VARCHAR(2147483647)"
+ } ]
+ },
+ "description" : "Limit(offset=[0], fetch=[3])",
+ "rankType" : "ROW_NUMBER",
+ "partitionSpec" : {
+ "fields" : [ ]
+ },
+ "sortSpec" : {
+ "fields" : [ ]
+ },
+ "outputRowNumber" : false
+ }, {
+ "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink",
+ "dynamicTableSink" : {
+ "identifier" : {
+ "catalogName" : "default_catalog",
+ "databaseName" : "default_database",
+ "tableName" : "MySink"
+ },
+ "catalogTable" : {
+ "table-sink-class" : "DEFAULT",
+ "schema.2.data-type" : "VARCHAR(2147483647)",
+ "connector" : "values",
+ "schema.0.data-type" : "BIGINT",
+ "schema.2.name" : "c",
+ "schema.1.name" : "b",
+ "schema.0.name" : "a",
+ "schema.1.data-type" : "INT"
+ }
+ },
+ "inputChangelogMode" : [ "INSERT" ],
+ "id" : 50,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "nullable" : true,
+ "fields" : [ {
+ "a" : "BIGINT"
+ }, {
+ "b" : "INT"
+ }, {
+ "c" : "VARCHAR(2147483647)"
+ } ]
+ },
+ "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, b, c])"
+ } ],
+ "edges" : [ {
+ "source" : 47,
+ "target" : 48,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 48,
+ "target" : 49,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 49,
+ "target" : 50,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/ConstantRankRange.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/ConstantRankRange.java
index 60f656c..63ac931 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/ConstantRankRange.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/ConstantRankRange.java
@@ -18,24 +18,44 @@
package org.apache.flink.table.runtime.operators.rank;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
+
import java.util.List;
/** rankStart and rankEnd are inclusive, rankStart always start from one. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonTypeName("Constant")
public class ConstantRankRange implements RankRange {
+ public static final String FIELD_NAME_START = "start";
+ public static final String FIELD_NAME_END = "end";
+
private static final long serialVersionUID = 9062345289888078376L;
- private long rankStart;
- private long rankEnd;
- public ConstantRankRange(long rankStart, long rankEnd) {
+ @JsonProperty(FIELD_NAME_START)
+ private final long rankStart;
+
+ @JsonProperty(FIELD_NAME_END)
+ private final long rankEnd;
+
+ @JsonCreator
+ public ConstantRankRange(
+ @JsonProperty(FIELD_NAME_START) long rankStart,
+ @JsonProperty(FIELD_NAME_END) long rankEnd) {
this.rankStart = rankStart;
this.rankEnd = rankEnd;
}
+ @JsonIgnore
public long getRankStart() {
return rankStart;
}
+ @JsonIgnore
public long getRankEnd() {
return rankEnd;
}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/ConstantRankRangeWithoutEnd.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/ConstantRankRangeWithoutEnd.java
index fadc9a0..6c33b21 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/ConstantRankRangeWithoutEnd.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/ConstantRankRangeWithoutEnd.java
@@ -18,16 +18,26 @@
package org.apache.flink.table.runtime.operators.rank;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
+
import java.util.List;
/** ConstantRankRangeWithoutEnd is a RankRange which not specify RankEnd. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonTypeName("ConstantWithoutEnd")
public class ConstantRankRangeWithoutEnd implements RankRange {
private static final long serialVersionUID = -1944057111062598696L;
+ @JsonProperty(ConstantRankRange.FIELD_NAME_START)
private final long rankStart;
- public ConstantRankRangeWithoutEnd(long rankStart) {
+ @JsonCreator
+ public ConstantRankRangeWithoutEnd(
+ @JsonProperty(ConstantRankRange.FIELD_NAME_START) long rankStart) {
this.rankStart = rankStart;
}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/RankRange.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/RankRange.java
index 1ecb736..6026e01 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/RankRange.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/RankRange.java
@@ -18,6 +18,9 @@
package org.apache.flink.table.runtime.operators.rank;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonSubTypes;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeInfo;
+
import java.io.Serializable;
import java.util.List;
@@ -25,7 +28,12 @@ import java.util.List;
* RankRange for Rank, including following 3 types : ConstantRankRange, ConstantRankRangeWithoutEnd,
* VariableRankRange.
*/
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
+@JsonSubTypes({
+ @JsonSubTypes.Type(value = ConstantRankRange.class),
+ @JsonSubTypes.Type(value = ConstantRankRangeWithoutEnd.class),
+ @JsonSubTypes.Type(value = VariableRankRange.class)
+})
public interface RankRange extends Serializable {
-
String toString(List<String> inputFieldNames);
}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/VariableRankRange.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/VariableRankRange.java
index cce9b04..e00789c 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/VariableRankRange.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/rank/VariableRankRange.java
@@ -18,18 +18,32 @@
package org.apache.flink.table.runtime.operators.rank;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
+
import java.util.List;
/** changing rank limit depends on input. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonTypeName("Variable")
public class VariableRankRange implements RankRange {
+ public static final String FIELD_NAME_END_INDEX = "endIndex";
+
private static final long serialVersionUID = 5579785886506433955L;
- private int rankEndIndex;
- public VariableRankRange(int rankEndIndex) {
+ @JsonProperty(FIELD_NAME_END_INDEX)
+ private final int rankEndIndex;
+
+ @JsonCreator
+ public VariableRankRange(@JsonProperty(FIELD_NAME_END_INDEX) int rankEndIndex) {
this.rankEndIndex = rankEndIndex;
}
+ @JsonIgnore
public int getRankEndIndex() {
return rankEndIndex;
}