You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by he...@apache.org on 2022/04/25 07:42:32 UTC
[incubator-inlong] branch master updated: [INLONG-3841][Sort] Add distinct support based time column for transform (#3842)
This is an automated email from the ASF dual-hosted git repository.
healchow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 6c5c6a1c3 [INLONG-3841][Sort] Add distinct support based time column for transform (#3842)
6c5c6a1c3 is described below
commit 6c5c6a1c3123cc9b0e779a96bb6509406cb00cdc
Author: yunqingmoswu <44...@users.noreply.github.com>
AuthorDate: Mon Apr 25 15:42:26 2022 +0800
[INLONG-3841][Sort] Add distinct support based time column for transform (#3842)
* Add PROCCESS_TIME BuiltInField to support transform
* Add distinct support based time column for transform
* Optimization of the method name
---
.../inlong/sort/protocol/BuiltInFieldInfo.java | 1 +
.../org/apache/inlong/sort/protocol/node/Node.java | 4 +-
.../sort/protocol/node/transform/DistinctNode.java | 93 ++++++
.../protocol/node/transform/TransformNode.java | 11 +-
.../protocol/transformation/OrderDirection.java | 32 ++
.../protocol/node/transform/DistinctNodeTest.java | 80 +++++
.../flink/parser/impl/FlinkSqlParser.java | 155 +++++++---
.../flink/parser/result/FlinkSqlParseResult.java | 6 +-
.../flink/parser/DistinctNodeSqlParseTest.java | 342 +++++++++++++++++++++
9 files changed, 671 insertions(+), 53 deletions(-)
diff --git a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/BuiltInFieldInfo.java b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/BuiltInFieldInfo.java
index d3cdc6460..085352df3 100644
--- a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/BuiltInFieldInfo.java
+++ b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/BuiltInFieldInfo.java
@@ -72,6 +72,7 @@ public class BuiltInFieldInfo extends FieldInfo {
public enum BuiltInField {
DATA_TIME,
+ PROCCESS_TIME,
MYSQL_METADATA_DATABASE,
MYSQL_METADATA_TABLE,
MYSQL_METADATA_EVENT_TIME,
diff --git a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java
index ed2d7ee24..674347fab 100644
--- a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java
+++ b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java
@@ -25,6 +25,7 @@ import org.apache.inlong.sort.protocol.FieldInfo;
import org.apache.inlong.sort.protocol.node.extract.KafkaExtractNode;
import org.apache.inlong.sort.protocol.node.extract.MySqlExtractNode;
import org.apache.inlong.sort.protocol.node.load.KafkaLoadNode;
+import org.apache.inlong.sort.protocol.node.transform.DistinctNode;
import org.apache.inlong.sort.protocol.node.transform.TransformNode;
import java.util.List;
@@ -39,7 +40,8 @@ import java.util.TreeMap;
@JsonSubTypes.Type(value = MySqlExtractNode.class, name = "mysqlExtract"),
@JsonSubTypes.Type(value = KafkaExtractNode.class, name = "kafkaExtract"),
@JsonSubTypes.Type(value = TransformNode.class, name = "baseTransform"),
- @JsonSubTypes.Type(value = KafkaLoadNode.class, name = "kafkaLoad")
+ @JsonSubTypes.Type(value = KafkaLoadNode.class, name = "kafkaLoad"),
+ @JsonSubTypes.Type(value = DistinctNode.class, name = "distinct")
})
public interface Node {
diff --git a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/transform/DistinctNode.java b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/transform/DistinctNode.java
new file mode 100644
index 000000000..66acdccdd
--- /dev/null
+++ b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/transform/DistinctNode.java
@@ -0,0 +1,93 @@
+/*
+ * 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.inlong.sort.protocol.node.transform;
+
+import com.google.common.base.Preconditions;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.transformation.FieldRelationShip;
+import org.apache.inlong.sort.protocol.transformation.FilterFunction;
+import org.apache.inlong.sort.protocol.transformation.OrderDirection;
+
+import java.util.List;
+
+/**
+ * TimeWindowDistinctNode class is a distinct node based time window
+ * It implements distinct operation by ROW_NUMBER such as:
+ * SELECT f1,f2,f3,f4,ts,rownum
+ * FROM (
+ * SELECT f1, f2, f3,f4,ts,
+ * ROW_NUMBER() OVER (PARTITION BY f2 ORDER BY ts desc) AS row_num -- desc use the latest one,
+ * FROM distinct_table)
+ * WHERE rownum=1
+ * ————————————————
+ */
+@EqualsAndHashCode(callSuper = true)
+@JsonTypeName("distinct")
+@NoArgsConstructor
+@Data
+public class DistinctNode extends TransformNode {
+
+ private static final long serialVersionUID = 5007120031895569715L;
+ @JsonProperty("distinctFields")
+ private List<FieldInfo> distinctFields;
+ @JsonProperty("orderField")
+ private FieldInfo orderField;
+ @JsonProperty("orderDirection")
+ private OrderDirection orderDirection = OrderDirection.ASC;
+
+ /**
+ * TimeWindowDistinctNode constructor
+ *
+ * @param id node id
+ * @param name node name
+ * @param fields The fields used to describe node schema
+ * @param fieldRelationShips field relationShips ysed to describe node relationships
+ * @param filters The filters used for data filter
+ * @param distinctFields The distinct fields used for partition
+ * @param orderField the order field used for sorting in partition
+ * @param orderDirection The orderDirection used for sorting after partition,
+ * support [ASC|DESC] default ASC
+ */
+ @JsonCreator
+ public DistinctNode(@JsonProperty("id") String id,
+ @JsonProperty("name") String name,
+ @JsonProperty("fields") List<FieldInfo> fields,
+ @JsonProperty("fieldRelationShips") List<FieldRelationShip> fieldRelationShips,
+ @JsonProperty("filters") List<FilterFunction> filters,
+ @JsonProperty("distinctFields") List<FieldInfo> distinctFields,
+ @JsonProperty("orderField") FieldInfo orderField,
+ @JsonProperty("orderDirection") OrderDirection orderDirection) {
+ super(id, name, fields, fieldRelationShips, filters);
+ this.distinctFields = Preconditions.checkNotNull(distinctFields, "distinctFields is null");
+ Preconditions.checkState(!distinctFields.isEmpty(), "distinct fields is empty");
+ this.orderField = Preconditions.checkNotNull(orderField, "orderField is null");
+ this.orderDirection = orderDirection != null ? orderDirection : OrderDirection.ASC;
+ }
+
+ @Override
+ public String genTableName() {
+ return "distinct_" + super.getId();
+ }
+
+}
diff --git a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/transform/TransformNode.java b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/transform/TransformNode.java
index 53e3eaf18..da619dc26 100644
--- a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/transform/TransformNode.java
+++ b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/transform/TransformNode.java
@@ -21,6 +21,7 @@ import com.google.common.base.Preconditions;
import lombok.Data;
import lombok.NoArgsConstructor;
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.JsonInclude;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude.Include;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
@@ -34,13 +35,15 @@ import org.apache.inlong.sort.protocol.transformation.FilterFunction;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
@JsonTypeInfo(
use = JsonTypeInfo.Id.NAME,
include = JsonTypeInfo.As.PROPERTY,
property = "type")
@JsonSubTypes({
- @JsonSubTypes.Type(value = TransformNode.class, name = "baseTransform")
+ @JsonSubTypes.Type(value = TransformNode.class, name = "baseTransform"),
+ @JsonSubTypes.Type(value = DistinctNode.class, name = "distinct")
})
@Data
@NoArgsConstructor
@@ -76,6 +79,12 @@ public class TransformNode implements Node, Serializable {
this.filters = filters;
}
+ @JsonIgnore
+ @Override
+ public Map<String, String> getProperties() {
+ return null;
+ }
+
@Override
public String genTableName() {
return "tansform_" + id;
diff --git a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/OrderDirection.java b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/OrderDirection.java
new file mode 100644
index 000000000..ddfe73385
--- /dev/null
+++ b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/OrderDirection.java
@@ -0,0 +1,32 @@
+/*
+ * 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.inlong.sort.protocol.transformation;
+
+/**
+ * OrderDirection class defines whether the sorting direction is ascending or descending
+ */
+public enum OrderDirection {
+ /**
+ * Asc for order direction
+ */
+ ASC,
+ /**
+ * Desc for order direction
+ */
+ DESC
+}
diff --git a/inlong-sort/sort-common/src/test/java/org/apache/inlong/sort/protocol/node/transform/DistinctNodeTest.java b/inlong-sort/sort-common/src/test/java/org/apache/inlong/sort/protocol/node/transform/DistinctNodeTest.java
new file mode 100644
index 000000000..c12566755
--- /dev/null
+++ b/inlong-sort/sort-common/src/test/java/org/apache/inlong/sort/protocol/node/transform/DistinctNodeTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.inlong.sort.protocol.node.transform;
+
+import org.apache.inlong.sort.formats.common.StringFormatInfo;
+import org.apache.inlong.sort.formats.common.TimestampFormatInfo;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.node.Node;
+import org.apache.inlong.sort.protocol.node.NodeBaseTest;
+import org.apache.inlong.sort.protocol.transformation.FieldRelationShip;
+import org.apache.inlong.sort.protocol.transformation.OrderDirection;
+
+import java.util.Arrays;
+
+/**
+ * Test for {@link DistinctNode}
+ */
+public class DistinctNodeTest extends NodeBaseTest {
+
+ @Override
+ public Node getNode() {
+ return new DistinctNode("1", null,
+ Arrays.asList(new FieldInfo("f1", new StringFormatInfo()),
+ new FieldInfo("f2", new StringFormatInfo()),
+ new FieldInfo("f3", new StringFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo())),
+ Arrays.asList(
+ new FieldRelationShip(new FieldInfo("f1", new StringFormatInfo()),
+ new FieldInfo("f1", new StringFormatInfo())),
+ new FieldRelationShip(new FieldInfo("f2", new StringFormatInfo()),
+ new FieldInfo("f2", new StringFormatInfo())),
+ new FieldRelationShip(new FieldInfo("f3", new StringFormatInfo()),
+ new FieldInfo("f3", new StringFormatInfo())),
+ new FieldRelationShip(new FieldInfo("ts", new StringFormatInfo()),
+ new FieldInfo("ts", new StringFormatInfo()))
+ ),
+ null,
+ Arrays.asList(new FieldInfo("f1", new StringFormatInfo()),
+ new FieldInfo("f2", new StringFormatInfo())),
+ new FieldInfo("ts", new StringFormatInfo()), OrderDirection.ASC);
+ }
+
+ @Override
+ public String getExpectSerializeStr() {
+ return "{\"type\":\"distinct\",\"id\":\"1\",\"fields\":[{\"type\":\"base\",\"name\":\"f1\","
+ + "\"formatInfo\":{\"type\":\"string\"}},{\"type\":\"base\",\"name\":\"f2\","
+ + "\"formatInfo\":{\"type\":\"string\"}},{\"type\":\"base\",\"name\":\"f3\","
+ + "\"formatInfo\":{\"type\":\"string\"}},{\"type\":\"base\",\"name\":\"ts\","
+ + "\"formatInfo\":{\"type\":\"timestamp\",\"format\":\"yyyy-MM-dd HH:mm:ss\"}}],"
+ + "\"fieldRelationShips\":[{\"type\":\"fieldRelationShip\",\"inputField\":{\"type\":\"base\","
+ + "\"name\":\"f1\",\"formatInfo\":{\"type\":\"string\"}},\"outputField\":{\"type\":\"base\","
+ + "\"name\":\"f1\",\"formatInfo\":{\"type\":\"string\"}}},{\"type\":\"fieldRelationShip\","
+ + "\"inputField\":{\"type\":\"base\",\"name\":\"f2\",\"formatInfo\":{\"type\":\"string\"}},"
+ + "\"outputField\":{\"type\":\"base\",\"name\":\"f2\",\"formatInfo\":{\"type\":\"string\"}}},"
+ + "{\"type\":\"fieldRelationShip\",\"inputField\":{\"type\":\"base\",\"name\":\"f3\","
+ + "\"formatInfo\":{\"type\":\"string\"}},\"outputField\":{\"type\":\"base\",\"name\":\"f3\","
+ + "\"formatInfo\":{\"type\":\"string\"}}},{\"type\":\"fieldRelationShip\","
+ + "\"inputField\":{\"type\":\"base\",\"name\":\"ts\",\"formatInfo\":{\"type\":\"string\"}},"
+ + "\"outputField\":{\"type\":\"base\",\"name\":\"ts\",\"formatInfo\":{\"type\":\"string\"}}}],"
+ + "\"distinctFields\":[{\"type\":\"base\",\"name\":\"f1\",\"formatInfo\":{\"type\":\"string\"}},"
+ + "{\"type\":\"base\",\"name\":\"f2\",\"formatInfo\":{\"type\":\"string\"}}],"
+ + "\"orderField\":{\"type\":\"base\",\"name\":\"ts\",\"formatInfo\":{\"type\":\"string\"}},"
+ + "\"orderDirection\":\"ASC\"}";
+ }
+}
diff --git a/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/parser/impl/FlinkSqlParser.java b/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/parser/impl/FlinkSqlParser.java
index 20067520f..0d1087a53 100644
--- a/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/parser/impl/FlinkSqlParser.java
+++ b/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/parser/impl/FlinkSqlParser.java
@@ -28,10 +28,13 @@ import org.apache.inlong.sort.protocol.StreamInfo;
import org.apache.inlong.sort.protocol.node.ExtractNode;
import org.apache.inlong.sort.protocol.node.LoadNode;
import org.apache.inlong.sort.protocol.node.Node;
+import org.apache.inlong.sort.protocol.node.transform.DistinctNode;
import org.apache.inlong.sort.protocol.node.transform.TransformNode;
import org.apache.inlong.sort.protocol.transformation.FieldRelationShip;
import org.apache.inlong.sort.protocol.transformation.FilterFunction;
+import org.apache.inlong.sort.protocol.transformation.relation.JoinRelationShip;
import org.apache.inlong.sort.protocol.transformation.relation.NodeRelationShip;
+import org.apache.inlong.sort.protocol.transformation.relation.UnionNodeRelationShip;
import org.apache.inlong.sort.singletenant.flink.parser.Parser;
import org.apache.inlong.sort.singletenant.flink.parser.result.FlinkSqlParseResult;
import org.slf4j.Logger;
@@ -107,7 +110,7 @@ public class FlinkSqlParser implements Parser {
Preconditions.checkState(!streamInfo.getNodes().isEmpty(), "nodes is empty");
Preconditions.checkNotNull(streamInfo.getRelations(), "relations is null");
Preconditions.checkState(!streamInfo.getRelations().isEmpty(), "relations is empty");
- log.info("start parse stream,streamId:{}", streamInfo.getStreamId());
+ log.info("start parse stream, streamId:{}", streamInfo.getStreamId());
Map<String, Node> nodeMap = new HashMap<>(streamInfo.getNodes().size());
streamInfo.getNodes().forEach(s -> {
Preconditions.checkNotNull(s.getId(), "node id is null");
@@ -127,7 +130,7 @@ public class FlinkSqlParser implements Parser {
private void parseNodeRelation(NodeRelationShip relation, Map<String, Node> nodeMap,
Map<String, NodeRelationShip> relationMap) {
- log.info("start parse node relation,relation:{}", relation);
+ log.info("start parse node relation, relation:{}", relation);
Preconditions.checkNotNull(relation, "relation is null");
Preconditions.checkState(relation.getInputs().size() > 0,
"relation must have at least one input node");
@@ -184,60 +187,124 @@ public class FlinkSqlParser implements Parser {
"load node only support one input node");
LoadNode loadNode = (LoadNode) node;
String insertSql = genLoadNodeInsertSql(loadNode, nodeMap.get(relation.getInputs().get(0)));
- log.info("insert sql => {}, node id:{}", insertSql, node.getId());
+ log.info("node id:{}, insert sql:\n{}", node.getId(), insertSql);
insertSqls.add(insertSql);
hasParsedSet.add(node.getId());
} else if (node instanceof TransformNode) {
TransformNode transformNode = (TransformNode) node;
- Preconditions.checkNotNull(transformNode.getFieldRelationShips(), "field relations is null");
+ Preconditions.checkNotNull(transformNode.getFieldRelationShips(),
+ "field relations is null");
Preconditions.checkState(!transformNode.getFieldRelationShips().isEmpty(),
"field relations is empty");
- Preconditions.checkState(relation.getInputs().size() == 1,
- "simple transform only support one input node");
- Preconditions.checkState(relation.getOutputs().size() == 1,
- "join node only support one output node");
- String selectSql = genSimpleTransformSelectSql(transformNode, relation, nodeMap);
String createSql = genCreateSql(node);
log.info("node id:{}, create table sql:\n{}", node.getId(), createSql);
- Preconditions.checkState(relation.getInputs().size() == 1,
- "load node only support one input node");
- log.info("tansform sql => {}, node id:{}", selectSql, node.getId());
- registerTableSql(node, createSql + " AS " + selectSql);
+ String selectSql;
+ if (relation instanceof JoinRelationShip) {
+ Preconditions.checkState(relation.getInputs().size() > 1,
+ "join must have more than one input nodes");
+ Preconditions.checkState(relation.getOutputs().size() == 1,
+ "join node only support one output node");
+ JoinRelationShip joinRelation = (JoinRelationShip) relation;
+ selectSql = genJoinSelectSql(transformNode, joinRelation, nodeMap);
+ } else if (relation instanceof UnionNodeRelationShip) {
+ Preconditions.checkState(relation.getInputs().size() > 1,
+ "union must have more than one input nodes");
+ Preconditions.checkState(relation.getOutputs().size() == 1,
+ "join node only support one output node");
+ UnionNodeRelationShip unionRelation = (UnionNodeRelationShip) relation;
+ selectSql = genUnionNodeSelectSql(transformNode, unionRelation, nodeMap);
+ } else {
+ Preconditions.checkState(relation.getInputs().size() == 1,
+ "simple transform only support one input node");
+ Preconditions.checkState(relation.getOutputs().size() == 1,
+ "join node only support one output node");
+ selectSql = genSimpleTransformSelectSql(transformNode, relation, nodeMap);
+ }
+ log.info("node id:{}, tansform sql:\n{}", node.getId(), selectSql);
+ registerTableSql(node, createSql + " AS\n" + selectSql);
hasParsedSet.add(node.getId());
}
}
log.info("parse node success, node id:{}", node.getId());
}
+ private String genUnionNodeSelectSql(TransformNode transformNode,
+ UnionNodeRelationShip unionRelation, Map<String, Node> nodeMap) {
+ throw new UnsupportedOperationException("Union is not currently supported");
+ }
+
+ private String genJoinSelectSql(TransformNode node,
+ JoinRelationShip relation, Map<String, Node> nodeMap) {
+ throw new UnsupportedOperationException("Join is not currently supported");
+ }
+
+ private StringBuilder genDistinctFilterSql(List<FieldInfo> fields, StringBuilder sb) {
+ String subSql = sb.toString();
+ sb = new StringBuilder("SELECT ");
+ for (FieldInfo field : fields) {
+ sb.append("\n `").append(field.getName()).append("`,");
+ }
+ sb.deleteCharAt(sb.length() - 1).append("\n FROM (").append(subSql)
+ .append(")\nWHERE row_num = 1");
+ return sb;
+ }
+
+ private void genDistinctSql(DistinctNode distinctNode, StringBuilder sb) {
+ Preconditions.checkNotNull(distinctNode.getDistinctFields(), "distinctField is null");
+ Preconditions.checkState(!distinctNode.getDistinctFields().isEmpty(),
+ "distinctField is empty");
+ Preconditions.checkNotNull(distinctNode.getOrderField(), "orderField is null");
+ sb.append(",\n ROW_NUMBER() OVER (PARTITION BY ");
+ for (FieldInfo distinctField : distinctNode.getDistinctFields()) {
+ sb.append(distinctField.format()).append(",");
+ }
+ sb.deleteCharAt(sb.length() - 1);
+ sb.append(" ORDER BY ").append(distinctNode.getOrderField().format()).append(" ")
+ .append(distinctNode.getOrderDirection().name()).append(") AS row_num");
+ }
+
private String genSimpleTransformSelectSql(TransformNode node,
NodeRelationShip relation, Map<String, Node> nodeMap) {
- Preconditions.checkNotNull(node.getFieldRelationShips(), "field relations is null");
- Preconditions.checkState(!node.getFieldRelationShips().isEmpty(),
- "field relations is empty");
StringBuilder sb = new StringBuilder();
sb.append("SELECT ");
Map<String, FieldRelationShip> fieldRelationMap = new HashMap<>(node.getFieldRelationShips().size());
node.getFieldRelationShips().forEach(s -> {
fieldRelationMap.put(s.getOutputField().getName(), s);
});
- for (FieldInfo field : node.getFields()) {
+ parseFieldRelations(node.getFields(), fieldRelationMap, sb);
+ if (node instanceof DistinctNode) {
+ genDistinctSql((DistinctNode) node, sb);
+ }
+ sb.append("\n FROM `").append(nodeMap.get(relation.getInputs().get(0)).genTableName()).append("` ");
+ parseFilterFields(node.getFilters(), sb);
+ if (node instanceof DistinctNode) {
+ sb = genDistinctFilterSql(node.getFields(), sb);
+ }
+ return sb.toString();
+ }
+
+ private void parseFilterFields(List<FilterFunction> filters, StringBuilder sb) {
+ if (filters != null && !filters.isEmpty()) {
+ sb.append("\n WHERE");
+ for (FilterFunction filter : filters) {
+ sb.append(" ").append(filter.format());
+ }
+ }
+ }
+
+ private void parseFieldRelations(List<FieldInfo> fields,
+ Map<String, FieldRelationShip> fieldRelationMap, StringBuilder sb) {
+ for (FieldInfo field : fields) {
FieldRelationShip fieldRelation = fieldRelationMap.get(field.getName());
if (fieldRelation != null) {
- sb.append(fieldRelation.getInputField().format()).append(" AS ").append(field.format()).append(",");
+ sb.append("\n ").append(fieldRelation.getInputField().format())
+ .append(" AS ").append(field.format()).append(",");
} else {
String targetType = TableFormatUtils.deriveLogicalType(field.getFormatInfo()).asSerializableString();
- sb.append("CAST(NULL as ").append(targetType).append(") AS ").append(field.format()).append(",");
+ sb.append("\n CAST(NULL as ").append(targetType).append(") AS ").append(field.format()).append(",");
}
}
sb.deleteCharAt(sb.length() - 1);
- sb.append(" FROM `").append(nodeMap.get(relation.getInputs().get(0)).genTableName()).append("` ");
- if (node.getFilters() != null && !node.getFilters().isEmpty()) {
- sb.append("\n WHERE");
- for (FilterFunction filter : node.getFilters()) {
- sb.append(" ").append(filter.format());
- }
- }
- return sb.toString();
}
private String genLoadNodeInsertSql(LoadNode loadNode, Node inputNode) {
@@ -246,28 +313,14 @@ public class FlinkSqlParser implements Parser {
"field relations is empty");
StringBuilder sb = new StringBuilder();
sb.append("INSERT INTO `").append(loadNode.genTableName()).append("` ");
- sb.append("SELECT ");
+ sb.append("\n SELECT ");
Map<String, FieldRelationShip> fieldRelationMap = new HashMap<>(loadNode.getFieldRelationShips().size());
loadNode.getFieldRelationShips().forEach(s -> {
fieldRelationMap.put(s.getOutputField().getName(), s);
});
- for (FieldInfo field : loadNode.getFields()) {
- FieldRelationShip fieldRelation = fieldRelationMap.get(field.getName());
- if (fieldRelation != null) {
- sb.append(fieldRelation.getInputField().format()).append(" AS ").append(field.format()).append(",");
- } else {
- String targetType = TableFormatUtils.deriveLogicalType(field.getFormatInfo()).asSerializableString();
- sb.append("CAST(NULL as ").append(targetType).append(") AS ").append(field.format()).append(",");
- }
- }
- sb.deleteCharAt(sb.length() - 1);
- sb.append(" FROM `").append(inputNode.genTableName()).append("`");
- if (loadNode.getFilters() != null && !loadNode.getFilters().isEmpty()) {
- sb.append(" WHERE");
- for (FilterFunction filter : loadNode.getFilters()) {
- sb.append(" ").append(filter.format());
- }
- }
+ parseFieldRelations(loadNode.getFields(), fieldRelationMap, sb);
+ sb.append("\n FROM `").append(inputNode.genTableName()).append("`");
+ parseFilterFields(loadNode.getFilters(), sb);
return sb.toString();
}
@@ -295,7 +348,7 @@ public class FlinkSqlParser implements Parser {
}
private String genCreateTransformSql(Node node) {
- return String.format("CREATE VIEW `%s` (\n%s)",
+ return String.format("CREATE VIEW `%s` (%s)",
node.genTableName(), parseTransformNodeFields(node.getFields()));
}
@@ -317,7 +370,7 @@ public class FlinkSqlParser implements Parser {
private String parseTransformNodeFields(List<FieldInfo> fields) {
StringBuilder sb = new StringBuilder();
for (FieldInfo field : fields) {
- sb.append(" `").append(field.getName()).append("`,");
+ sb.append("\n `").append(field.getName()).append("`,");
}
if (sb.length() > 0) {
sb.delete(sb.lastIndexOf(","), sb.length());
@@ -330,7 +383,13 @@ public class FlinkSqlParser implements Parser {
for (FieldInfo field : fields) {
sb.append(" `").append(field.getName()).append("` ");
if (field instanceof BuiltInFieldInfo) {
- throw new UnsupportedOperationException("Metadata field sync is not currently supported");
+ BuiltInFieldInfo builtInFieldInfo = (BuiltInFieldInfo) field;
+ switch (builtInFieldInfo.getBuiltInField()) {
+ case PROCCESS_TIME:
+ sb.append(" AS PROCTIME()");
+ break;
+ default:
+ }
} else {
sb.append(TableFormatUtils.deriveLogicalType(field.getFormatInfo()).asSerializableString());
}
diff --git a/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/parser/result/FlinkSqlParseResult.java b/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/parser/result/FlinkSqlParseResult.java
index 75cb7e8e5..1d382832d 100644
--- a/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/parser/result/FlinkSqlParseResult.java
+++ b/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/parser/result/FlinkSqlParseResult.java
@@ -58,7 +58,7 @@ public class FlinkSqlParseResult implements ParseResult, Serializable {
@Override
public void execute() throws Exception {
executeCreateTableSqls(createTableSqls);
- executeLoadSqls(loadSqls).wait();
+ executeLoadSqls(loadSqls).await();
}
@Override
@@ -70,7 +70,7 @@ public class FlinkSqlParseResult implements ParseResult, Serializable {
private TableResult executeLoadSqls(List<String> sqls) {
StatementSet st = tableEnv.createStatementSet();
for (String sql : sqls) {
- log.info("execute loadSql => {}", sql);
+ log.info("execute loadSql:\n{}", sql);
st.addInsertSql(sql);
}
return st.execute();
@@ -78,7 +78,7 @@ public class FlinkSqlParseResult implements ParseResult, Serializable {
private void executeCreateTableSqls(List<String> sqls) {
for (String sql : sqls) {
- log.info("execute createSql => {}", sql);
+ log.info("execute createSql:\n{}", sql);
tableEnv.executeSql(sql);
}
}
diff --git a/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/parser/DistinctNodeSqlParseTest.java b/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/parser/DistinctNodeSqlParseTest.java
new file mode 100644
index 000000000..c5f1fc52b
--- /dev/null
+++ b/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/parser/DistinctNodeSqlParseTest.java
@@ -0,0 +1,342 @@
+/*
+ * 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.inlong.sort.singletenant.flink.parser;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.inlong.sort.formats.common.FloatFormatInfo;
+import org.apache.inlong.sort.formats.common.IntFormatInfo;
+import org.apache.inlong.sort.formats.common.LongFormatInfo;
+import org.apache.inlong.sort.formats.common.StringFormatInfo;
+import org.apache.inlong.sort.formats.common.TimestampFormatInfo;
+import org.apache.inlong.sort.protocol.BuiltInFieldInfo;
+import org.apache.inlong.sort.protocol.BuiltInFieldInfo.BuiltInField;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.GroupInfo;
+import org.apache.inlong.sort.protocol.StreamInfo;
+import org.apache.inlong.sort.protocol.enums.ScanStartupMode;
+import org.apache.inlong.sort.protocol.node.Node;
+import org.apache.inlong.sort.protocol.node.extract.KafkaExtractNode;
+import org.apache.inlong.sort.protocol.node.format.JsonFormat;
+import org.apache.inlong.sort.protocol.node.load.KafkaLoadNode;
+import org.apache.inlong.sort.protocol.node.transform.DistinctNode;
+import org.apache.inlong.sort.protocol.transformation.FieldRelationShip;
+import org.apache.inlong.sort.protocol.transformation.OrderDirection;
+import org.apache.inlong.sort.protocol.transformation.StringConstantParam;
+import org.apache.inlong.sort.protocol.transformation.TimeUnitConstantParam;
+import org.apache.inlong.sort.protocol.transformation.TimeUnitConstantParam.TimeUnit;
+import org.apache.inlong.sort.protocol.transformation.WatermarkField;
+import org.apache.inlong.sort.protocol.transformation.relation.NodeRelationShip;
+import org.apache.inlong.sort.singletenant.flink.parser.impl.FlinkSqlParser;
+import org.apache.inlong.sort.singletenant.flink.parser.result.FlinkSqlParseResult;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Test for {@link org.apache.inlong.sort.protocol.node.transform.DistinctNode}
+ */
+public class DistinctNodeSqlParseTest extends AbstractTestBase {
+
+ private KafkaExtractNode buildKafkaExtractNode() {
+ List<FieldInfo> fields = Arrays.asList(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("salary", new FloatFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo()),
+ new BuiltInFieldInfo("proctime", new TimestampFormatInfo(), BuiltInField.PROCCESS_TIME));
+ return new KafkaExtractNode("1", "kafka_input", fields, null,
+ null, "topic_input", "localhost:9092",
+ new JsonFormat(), ScanStartupMode.EARLIEST_OFFSET,
+ null);
+ }
+
+ private KafkaExtractNode buildKafkaExtractNode2() {
+ List<FieldInfo> fields = Arrays.asList(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("salary", new FloatFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo()),
+ new BuiltInFieldInfo("proctime", new TimestampFormatInfo(), BuiltInField.PROCCESS_TIME));
+ WatermarkField wk = new WatermarkField(new FieldInfo("ts", new TimestampFormatInfo()),
+ new StringConstantParam("1"),
+ new TimeUnitConstantParam(TimeUnit.SECOND));
+ return new KafkaExtractNode("1", "kafka_input", fields, wk,
+ null, "topic_input", "localhost:9092",
+ new JsonFormat(), ScanStartupMode.EARLIEST_OFFSET,
+ null);
+ }
+
+ private KafkaExtractNode buildKafkaExtractNode3() {
+ List<FieldInfo> fields = Arrays.asList(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("salary", new FloatFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo()));
+ return new KafkaExtractNode("1", "kafka_input", fields, null,
+ null, "topic_input", "localhost:9092",
+ new JsonFormat(), ScanStartupMode.EARLIEST_OFFSET, null);
+ }
+
+ private KafkaLoadNode buildKafkaLoadNode() {
+ List<FieldInfo> fields = Arrays.asList(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("salary", new FloatFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo()));
+ List<FieldRelationShip> relations = Arrays
+ .asList(new FieldRelationShip(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("id", new LongFormatInfo())),
+ new FieldRelationShip(new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo())),
+ new FieldRelationShip(new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo())),
+ new FieldRelationShip(new FieldInfo("ts", new TimestampFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo()))
+ );
+ return new KafkaLoadNode("3", "kafka_output", fields, relations,
+ null, "topic_output", "localhost:9092",
+ new JsonFormat(), null,
+ null, null);
+ }
+
+ private KafkaLoadNode buildKafkaLoadNode2() {
+ List<FieldInfo> fields = Arrays.asList(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("salary", new FloatFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo()));
+ List<FieldRelationShip> relations = Arrays
+ .asList(new FieldRelationShip(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("id", new LongFormatInfo())),
+ new FieldRelationShip(new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo())),
+ new FieldRelationShip(new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo())),
+ new FieldRelationShip(new FieldInfo("ts", new TimestampFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo()))
+ );
+ return new KafkaLoadNode("3", "kafka_output", fields, relations,
+ null, "topic_output", "localhost:9092",
+ new JsonFormat(), null,
+ null, "id");
+ }
+
+ private KafkaLoadNode buildKafkaLoadNode3() {
+ List<FieldInfo> fields = Arrays.asList(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("salary", new FloatFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo()));
+ List<FieldRelationShip> relations = Arrays
+ .asList(new FieldRelationShip(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("id", new LongFormatInfo())),
+ new FieldRelationShip(new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo())),
+ new FieldRelationShip(new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo())),
+ new FieldRelationShip(new FieldInfo("ts", new TimestampFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo()))
+ );
+ return new KafkaLoadNode("3", "kafka_output", fields, relations,
+ null, "topic_output", "localhost:9092",
+ new JsonFormat(), null,
+ null, "id");
+ }
+
+ private Node buildTransformNode() {
+ return new DistinctNode("2", null,
+ Arrays.asList(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("salary", new FloatFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo())
+ ),
+ Arrays.asList(
+ new FieldRelationShip(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("id", new LongFormatInfo())),
+ new FieldRelationShip(new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo())),
+ new FieldRelationShip(new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo())),
+ new FieldRelationShip(new FieldInfo("ts", new TimestampFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo()))
+ ),
+ null,
+ Collections.singletonList(new FieldInfo("name", new StringFormatInfo())),
+ new FieldInfo("proctime", new TimestampFormatInfo()),
+ OrderDirection.ASC);
+ }
+
+ private Node buildTransformNode2() {
+ return new DistinctNode("2", null,
+ Arrays.asList(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("salary", new FloatFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo())
+ ),
+ Arrays.asList(
+ new FieldRelationShip(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("id", new LongFormatInfo())),
+ new FieldRelationShip(new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo())),
+ new FieldRelationShip(new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo())),
+ new FieldRelationShip(new FieldInfo("ts", new TimestampFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo()))
+ ),
+ null,
+ Collections.singletonList(new FieldInfo("name", new StringFormatInfo())),
+ new FieldInfo("ts", new TimestampFormatInfo()),
+ OrderDirection.ASC);
+ }
+
+ private Node buildTransformNode3() {
+ return new DistinctNode("2", null,
+ Arrays.asList(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("salary", new FloatFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo())
+ ),
+ Arrays.asList(
+ new FieldRelationShip(new FieldInfo("id", new LongFormatInfo()),
+ new FieldInfo("id", new LongFormatInfo())),
+ new FieldRelationShip(new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo())),
+ new FieldRelationShip(new FieldInfo("age", new IntFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo())),
+ new FieldRelationShip(new FieldInfo("ts", new TimestampFormatInfo()),
+ new FieldInfo("ts", new TimestampFormatInfo()))
+ ),
+ null,
+ Collections.singletonList(new FieldInfo("name", new StringFormatInfo())),
+ new FieldInfo("ts", new TimestampFormatInfo()),
+ OrderDirection.ASC);
+ }
+
+ public NodeRelationShip buildNodeRelation(List<Node> inputs, List<Node> outputs) {
+ List<String> inputIds = inputs.stream().map(Node::getId).collect(Collectors.toList());
+ List<String> outputIds = outputs.stream().map(Node::getId).collect(Collectors.toList());
+ return new NodeRelationShip(inputIds, outputIds);
+ }
+
+ /**
+ * Test distinct based process time
+ *
+ * @throws Exception The exception may throws when executing
+ */
+ @Test
+ public void testDistinctBasedProcessTime() throws Exception {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inStreamingMode()
+ .build();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.enableCheckpointing(10000);
+ StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
+ Node inputNode = buildKafkaExtractNode();
+ Node tansformNode = buildTransformNode();
+ Node outputNode = buildKafkaLoadNode();
+ StreamInfo streamInfo = new StreamInfo("1", Arrays.asList(inputNode, tansformNode, outputNode),
+ Arrays.asList(
+ buildNodeRelation(Collections.singletonList(inputNode),
+ Collections.singletonList(tansformNode)),
+ buildNodeRelation(Collections.singletonList(tansformNode),
+ Collections.singletonList(outputNode))
+ ));
+ GroupInfo groupInfo = new GroupInfo("group_id", Collections.singletonList(streamInfo));
+ FlinkSqlParser parser = FlinkSqlParser.getInstance(tableEnv, groupInfo);
+ FlinkSqlParseResult result = parser.parse();
+ Assert.assertTrue(result.tryExecute());
+ }
+
+ /**
+ * Test distinct based a time field
+ *
+ * @throws Exception The exception may throws when executing
+ */
+ @Test
+ public void testDistinctBasedTimeField() throws Exception {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inStreamingMode()
+ .build();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.enableCheckpointing(10000);
+ StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
+ Node inputNode = buildKafkaExtractNode3();
+ Node tansformNode = buildTransformNode3();
+ Node outputNode = buildKafkaLoadNode3();
+ StreamInfo streamInfo = new StreamInfo("1", Arrays.asList(inputNode, tansformNode, outputNode),
+ Arrays.asList(
+ buildNodeRelation(Collections.singletonList(inputNode),
+ Collections.singletonList(tansformNode)),
+ buildNodeRelation(Collections.singletonList(tansformNode),
+ Collections.singletonList(outputNode))
+ ));
+ GroupInfo groupInfo = new GroupInfo("group_id", Collections.singletonList(streamInfo));
+ FlinkSqlParser parser = FlinkSqlParser.getInstance(tableEnv, groupInfo);
+ FlinkSqlParseResult result = parser.parse();
+ Assert.assertTrue(result.tryExecute());
+ }
+
+ /**
+ * Test distinct based event time
+ *
+ * @throws Exception The exception may throws when executing
+ */
+ @Test
+ public void testDistinctBasedEventTime() throws Exception {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inStreamingMode()
+ .build();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.enableCheckpointing(10000);
+ StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
+ Node inputNode = buildKafkaExtractNode2();
+ Node tansformNode = buildTransformNode2();
+ Node outputNode = buildKafkaLoadNode2();
+ StreamInfo streamInfo = new StreamInfo("1", Arrays.asList(inputNode, tansformNode, outputNode),
+ Arrays.asList(
+ buildNodeRelation(Collections.singletonList(inputNode),
+ Collections.singletonList(tansformNode)),
+ buildNodeRelation(Collections.singletonList(tansformNode),
+ Collections.singletonList(outputNode))
+ ));
+ GroupInfo groupInfo = new GroupInfo("group_id", Collections.singletonList(streamInfo));
+ FlinkSqlParser parser = FlinkSqlParser.getInstance(tableEnv, groupInfo);
+ FlinkSqlParseResult result = parser.parse();
+ Assert.assertTrue(result.tryExecute());
+ }
+
+}