You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2019/04/15 11:14:35 UTC
[flink] branch master updated: [FLINK-11884][table] Ported sort &
limit validation on top of Expressions
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz 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 2c503c2 [FLINK-11884][table] Ported sort & limit validation on top of Expressions
2c503c2 is described below
commit 2c503c2488f786ae684fa956b23d4e2f8e1797de
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Mon Apr 1 15:17:25 2019 +0200
[FLINK-11884][table] Ported sort & limit validation on top of Expressions
---
.../expressions/BuiltInFunctionDefinitions.java | 2 +
.../table/operations/SortOperationFactory.java | 118 +++++++++++++++++++++
.../flink/table/operations/SortOperationUtils.java | 71 -------------
.../table/operations/OperationTreeBuilder.scala | 12 +--
.../flink/table/plan/logical/operators.scala | 23 +---
5 files changed, 128 insertions(+), 98 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/BuiltInFunctionDefinitions.java
index db6fcd3..801fad2 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/BuiltInFunctionDefinitions.java
@@ -333,6 +333,8 @@ public final class BuiltInFunctionDefinitions {
WINDOW_START, WINDOW_END, PROCTIME, ROWTIME
));
+ public static final List<FunctionDefinition> ORDERING = Arrays.asList(ORDER_ASC, ORDER_DESC);
+
public static List<FunctionDefinition> getDefinitions() {
final Field[] fields = BuiltInFunctionDefinitions.class.getFields();
final List<FunctionDefinition> list = new ArrayList<>(fields.length);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/SortOperationFactory.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/SortOperationFactory.java
new file mode 100644
index 0000000..8540f7b
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/SortOperationFactory.java
@@ -0,0 +1,118 @@
+/*
+ * 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.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.expressions.ApiExpressionDefaultVisitor;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.ExpressionBridge;
+import org.apache.flink.table.expressions.PlannerExpression;
+import org.apache.flink.table.plan.logical.Limit;
+import org.apache.flink.table.plan.logical.LogicalNode;
+import org.apache.flink.table.plan.logical.Sort;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static java.util.Collections.singletonList;
+import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.ORDERING;
+import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.ORDER_ASC;
+
+/**
+ * Utility class for creating a valid {@link Sort} operation.
+ */
+@Internal
+public class SortOperationFactory {
+
+ private final boolean isStreaming;
+ private final ExpressionBridge<PlannerExpression> expressionBridge;
+ private final OrderWrapper orderWrapper = new OrderWrapper();
+
+ public SortOperationFactory(
+ ExpressionBridge<PlannerExpression> expressionBridge,
+ boolean isStreaming) {
+ this.expressionBridge = expressionBridge;
+ this.isStreaming = isStreaming;
+ }
+
+ /**
+ * Creates a valid {@link Sort} operation.
+ *
+ * <p><b>NOTE:</b> if the collation is not explicitly specified for any expression, it is wrapped in a
+ * default ascending order
+ *
+ * @param orders expressions describing order,
+ * @param child relational expression on top of which to apply the sort operation
+ * @return valid sort operation
+ */
+ public Sort createSort(List<Expression> orders, TableOperation child) {
+ failIfStreaming();
+
+ List<PlannerExpression> convertedOrders = orders.stream()
+ .map(f -> f.accept(orderWrapper))
+ .map(expressionBridge::bridge)
+ .collect(Collectors.toList());
+ return new Sort(convertedOrders, (LogicalNode) child);
+ }
+
+ /**
+ * Creates a valid {@link Limit} operation.
+ *
+ * @param offset offset to start from
+ * @param fetch number of records to fetch
+ * @return valid limit operation
+ */
+ public Limit createLimit(int offset, int fetch, TableOperation child) {
+ failIfStreaming();
+
+ if (!(child instanceof Sort)) {
+ throw new ValidationException("A limit operation must be preceded by a sort operation.");
+ }
+ if (offset < 0) {
+ throw new ValidationException("Offset should be greater than or equal to zero.");
+ }
+
+ return new Limit(offset, fetch, (LogicalNode) child);
+ }
+
+ private void failIfStreaming() {
+ if (isStreaming) {
+ throw new ValidationException("A limit operation on unbounded tables is currently not supported.");
+ }
+ }
+
+ private class OrderWrapper extends ApiExpressionDefaultVisitor<Expression> {
+
+ @Override
+ public Expression visitCall(CallExpression call) {
+ if (ORDERING.contains(call.getFunctionDefinition())) {
+ return call;
+ } else {
+ return defaultMethod(call);
+ }
+ }
+
+ @Override
+ protected Expression defaultMethod(Expression expression) {
+ return new CallExpression(ORDER_ASC, singletonList(expression));
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/SortOperationUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/SortOperationUtils.java
deleted file mode 100644
index 418541f..0000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/SortOperationUtils.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * 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.operations;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.expressions.ApiExpressionDefaultVisitor;
-import org.apache.flink.table.expressions.CallExpression;
-import org.apache.flink.table.expressions.Expression;
-import org.apache.flink.table.expressions.FunctionDefinition;
-
-import java.util.Arrays;
-import java.util.List;
-import java.util.stream.Collectors;
-
-import static java.util.Collections.singletonList;
-import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.ORDER_ASC;
-import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.ORDER_DESC;
-
-/**
- * Utility class for sort specific expressions transformations.
- */
-@Internal
-public final class SortOperationUtils {
-
- private static final List<FunctionDefinition> ORDERING = Arrays.asList(ORDER_ASC, ORDER_DESC);
-
- /**
- * Makes sure that all expressions are wrapped in ordering expression. If the expression is either
- * {@link org.apache.flink.table.expressions.BuiltInFunctionDefinitions#ORDER_ASC} or
- * {@link org.apache.flink.table.expressions.BuiltInFunctionDefinitions#ORDER_DESC} it does nothing, otherwise
- * it inserts {@link org.apache.flink.table.expressions.BuiltInFunctionDefinitions#ORDER_ASC}.
- */
- public static List<Expression> wrapInOrder(List<Expression> expression) {
- return expression.stream()
- .map(expr -> expr.accept(new OrderWrapper()))
- .collect(Collectors.toList());
- }
-
- private static class OrderWrapper extends ApiExpressionDefaultVisitor<Expression> {
-
- @Override
- public Expression visitCall(CallExpression call) {
- if (ORDERING.contains(call.getFunctionDefinition())) {
- return call;
- } else {
- return defaultMethod(call);
- }
- }
-
- @Override
- protected Expression defaultMethod(Expression expression) {
- return new CallExpression(ORDER_ASC, singletonList(expression));
- }
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala
index 7321cb8..f3289f4 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala
@@ -44,7 +44,9 @@ class OperationTreeBuilder(private val tableEnv: TableEnvironment) {
private val expressionBridge: ExpressionBridge[PlannerExpression] = tableEnv.expressionBridge
private val functionCatalog: FunctionDefinitionCatalog = tableEnv.functionCatalog
+ private val isStreaming = tableEnv.isInstanceOf[StreamTableEnvironment]
private val projectionOperationFactory = new ProjectionOperationFactory(expressionBridge)
+ private val sortOperationFactory = new SortOperationFactory(expressionBridge, isStreaming)
private val noWindowPropertyChecker = new NoWindowPropertyChecker(
"Window start and end properties are not available for Over windows.")
@@ -260,13 +262,11 @@ class OperationTreeBuilder(private val tableEnv: TableEnvironment) {
fields: JList[Expression],
child: TableOperation)
: TableOperation = {
- val childNode = child.asInstanceOf[LogicalNode]
- val resolver = resolverFor(tableCatalog, functionCatalog, childNode).build()
- val resolvedExpressions = resolver.resolve(fields)
- val order = SortOperationUtils.wrapInOrder(resolvedExpressions).asScala.map(bridgeExpression)
+ val resolver = resolverFor(tableCatalog, functionCatalog, child).build()
+ val resolvedFields = resolver.resolve(fields)
- Sort(order, childNode).validate(tableEnv)
+ sortOperationFactory.createSort(resolvedFields, child)
}
def limitWithOffset(offset: Int, child: TableOperation): TableOperation = {
@@ -290,7 +290,7 @@ class OperationTreeBuilder(private val tableEnv: TableEnvironment) {
}
def limit(offset: Int, fetch: Int, child: TableOperation): TableOperation = {
- Limit(offset, fetch, child.asInstanceOf[LogicalNode]).validate(tableEnv)
+ sortOperationFactory.createLimit(offset, fetch, child)
}
def alias(
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/operators.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/operators.scala
index 0ee5ca5..8d1c067 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/operators.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/operators.scala
@@ -81,19 +81,12 @@ case class Distinct(child: LogicalNode) extends UnaryNode {
}
}
-case class Sort(order: Seq[PlannerExpression], child: LogicalNode) extends UnaryNode {
+case class Sort(order: JList[PlannerExpression], child: LogicalNode) extends UnaryNode {
override def output: Seq[Attribute] = child.output
override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
child.construct(relBuilder)
- relBuilder.sort(order.map(_.toRexNode(relBuilder)).asJava)
- }
-
- override def validate(tableEnv: TableEnvironment): LogicalNode = {
- if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
- failValidation(s"Sort on stream tables is currently not supported.")
- }
- this
+ relBuilder.sort(order.asScala.map(_.toRexNode(relBuilder)).asJava)
}
}
@@ -105,18 +98,6 @@ case class Limit(offset: Int, fetch: Int = -1, child: LogicalNode) extends Unary
relBuilder.limit(offset, fetch)
}
- override def validate(tableEnv: TableEnvironment): LogicalNode = {
- if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
- failValidation(s"Limit on stream tables is currently not supported.")
- }
- if (!child.isInstanceOf[Sort]) {
- failValidation(s"Limit operator must be preceded by an OrderBy operator.")
- }
- if (offset < 0) {
- failValidation(s"Offset should be greater than or equal to zero.")
- }
- super.validate(tableEnv)
- }
}
case class Filter(condition: PlannerExpression, child: LogicalNode) extends UnaryNode {