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/09 12:05:12 UTC
[flink] branch master updated: [FLINK-11884][table] Implement
expression resolution on top of new 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 413e3fe [FLINK-11884][table] Implement expression resolution on top of new Expressions
413e3fe is described below
commit 413e3feb786df9b0df951eae7e942ac45ba3d22a
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Thu Mar 7 15:54:22 2019 +0100
[FLINK-11884][table] Implement expression resolution on top of new
Expressions
---
.../java/org/apache/flink/table/api/Table.java | 6 +-
.../expressions/ApiExpressionDefaultVisitor.java | 5 +
.../table/expressions/ApiExpressionUtils.java | 43 +++
.../table/expressions/ApiExpressionVisitor.java | 4 +
...pression.java => LocalReferenceExpression.java} | 28 +-
.../expressions/UnresolvedReferenceExpression.java | 5 +-
.../flink/table/expressions/ExpressionVisitor.java | 1 -
.../table/codegen/agg/batch/AggCodeGenHelper.scala | 3 +-
.../codegen/agg/batch/HashAggCodeGenHelper.scala | 3 +-
.../table/expressions/ExpressionResolver.java | 317 ++++++++++++++++
.../expressions/lookups/FieldReferenceLookup.java | 110 ++++++
.../expressions/lookups/TableReferenceLookup.java} | 35 +-
.../table/expressions/rules/FlattenCallRule.java | 90 +++++
.../expressions/rules/LookupCallByNameRule.java} | 35 +-
.../expressions/rules/OverWindowResolverRule.java | 101 +++++
.../expressions/rules/ReferenceResolverRule.java | 85 +++++
.../rules/ResolveCallByArgumentsRule.java | 116 ++++++
.../table/expressions/rules/ResolverRule.java | 79 ++++
.../table/expressions/rules/ResolverRules.java | 62 ++++
.../expressions/rules/RuleExpressionVisitor.java} | 29 +-
.../rules/StarReferenceFlatteningRule.java | 65 ++++
.../table/operations/ColumnOperationUtils.java | 151 ++++++++
.../table/operations/ProjectionOperationUtils.java | 166 +++++++++
.../flink/table/operations/SortOperationUtils.java | 71 ++++
.../apache/flink/table/api/TableEnvironment.scala | 4 +-
.../org/apache/flink/table/api/tableImpl.scala | 42 +--
.../expressions/PlannerExpressionConverter.scala | 87 +++--
.../apache/flink/table/expressions/composite.scala | 4 +-
.../flink/table/expressions/fieldExpression.scala | 2 -
.../table/operations/OperationTreeBuilder.scala | 399 ++++++++++++++++++++
.../flink/table/plan/OperationTreeBuilder.scala | 413 ---------------------
.../flink/table/plan/ProjectionTranslator.scala | 155 --------
.../flink/table/plan/logical/LogicalNode.scala | 110 +-----
.../flink/table/plan/logical/groupWindows.scala | 15 +-
.../flink/table/plan/logical/operators.scala | 144 ++-----
.../flink/table/util/JavaScalaConversionUtil.scala | 6 +
.../StreamTableEnvironmentValidationTest.scala | 10 +-
.../table/validation/CalcValidationTest.scala | 20 +-
.../validation/OverWindowValidationTest.scala | 2 +-
39 files changed, 2077 insertions(+), 946 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java
index 0fafc87..5d62202 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java
@@ -973,8 +973,7 @@ public interface Table {
Table renameColumns(Expression... fields);
/**
- * Drops existing columns. The field expressions
- * should be field reference expressions, and only existing fields can be dropped.
+ * Drops existing columns. The field expressions should be field reference expressions.
*
* <p>Example:
*
@@ -987,8 +986,7 @@ public interface Table {
Table dropColumns(String fields);
/**
- * Drops existing columns. The field expressions
- * should be field reference expressions, and only existing fields can be dropped.
+ * Drops existing columns. The field expressions should be field reference expressions.
*
* <p>Scala Example:
* <pre>
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionDefaultVisitor.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionDefaultVisitor.java
index 978ed14..23e1438 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionDefaultVisitor.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionDefaultVisitor.java
@@ -52,6 +52,11 @@ public abstract class ApiExpressionDefaultVisitor<T> extends ApiExpressionVisito
}
@Override
+ public T visitLocalReference(LocalReferenceExpression localReference) {
+ return defaultMethod(localReference);
+ }
+
+ @Override
public T visitTypeLiteral(TypeLiteralExpression typeLiteral) {
return defaultMethod(typeLiteral);
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionUtils.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionUtils.java
index 30dc45d..c2c535b 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionUtils.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionUtils.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.expressions;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.typeutils.RowIntervalTypeInfo;
@@ -35,6 +36,7 @@ import java.util.Optional;
import java.util.function.Supplier;
import java.util.stream.Collectors;
+import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.AS;
import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.CAST;
import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.TIMES;
import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.WINDOW_PROPERTIES;
@@ -46,6 +48,8 @@ import static org.apache.flink.table.expressions.FunctionDefinition.Type.AGGREGA
@Internal
public final class ApiExpressionUtils {
+ private static final ExtractNameVisitor extractNameVisitor = new ExtractNameVisitor();
+
public static final long MILLIS_PER_SECOND = 1000L;
public static final long MILLIS_PER_MINUTE = 60000L;
@@ -338,4 +342,43 @@ public final class ApiExpressionUtils {
return Collections.emptyList();
}
}
+
+ public static List<Optional<String>> extractNames(List<Expression> expressions) {
+ return expressions.stream().map(ApiExpressionUtils::extractName).collect(Collectors.toList());
+ }
+
+ public static Optional<String> extractName(Expression expression) {
+ return expression.accept(extractNameVisitor);
+ }
+
+ private static class ExtractNameVisitor extends ApiExpressionDefaultVisitor<Optional<String>> {
+ @Override
+ public Optional<String> visitCall(CallExpression call) {
+ if (call.getFunctionDefinition().equals(AS)) {
+ return extractValue(call.getChildren().get(1), Types.STRING);
+ } else {
+ return Optional.empty();
+ }
+ }
+
+ @Override
+ public Optional<String> visitLocalReference(LocalReferenceExpression localReference) {
+ return Optional.of(localReference.getName());
+ }
+
+ @Override
+ public Optional<String> visitTableReference(TableReferenceExpression tableReference) {
+ return Optional.of(tableReference.getName());
+ }
+
+ @Override
+ public Optional<String> visitFieldReference(FieldReferenceExpression fieldReference) {
+ return Optional.of(fieldReference.getName());
+ }
+
+ @Override
+ protected Optional<String> defaultMethod(Expression expression) {
+ return Optional.empty();
+ }
+ }
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionVisitor.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionVisitor.java
index f228c1a..f5ab190 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionVisitor.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionVisitor.java
@@ -28,6 +28,8 @@ public abstract class ApiExpressionVisitor<R> implements ExpressionVisitor<R> {
public abstract R visitTableReference(TableReferenceExpression tableReference);
+ public abstract R visitLocalReference(LocalReferenceExpression localReference);
+
public abstract R visitLookupCall(LookupCallExpression lookupCall);
public abstract R visitUnresolvedReference(UnresolvedReferenceExpression unresolvedReference);
@@ -35,6 +37,8 @@ public abstract class ApiExpressionVisitor<R> implements ExpressionVisitor<R> {
public final R visit(Expression other) {
if (other instanceof TableReferenceExpression) {
return visitTableReference((TableReferenceExpression) other);
+ } else if (other instanceof LocalReferenceExpression) {
+ return visitLocalReference((LocalReferenceExpression) other);
} else if (other instanceof LookupCallExpression) {
return visitLookupCall((LookupCallExpression) other);
} else if (other instanceof UnresolvedReferenceExpression) {
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/UnresolvedReferenceExpression.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/LocalReferenceExpression.java
similarity index 65%
copy from flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/UnresolvedReferenceExpression.java
copy to flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/LocalReferenceExpression.java
index e454ea2..03f7dd3 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/UnresolvedReferenceExpression.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/LocalReferenceExpression.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.expressions;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Preconditions;
import java.util.Collections;
@@ -26,24 +26,29 @@ import java.util.List;
import java.util.Objects;
/**
- * An unresolved reference to a field or table.
- *
- * <p>This is a purely API facing expression that will be resolved into
- * {@link FieldReferenceExpression} or {@link TableReferenceExpression}.
+ * Reference to entity local to a certain {@link org.apache.flink.table.operations.TableOperation}.
+ * That entity does not come from any of the Operations input. It might be for example a group
+ * window in window aggregation.
*/
-@PublicEvolving
-public final class UnresolvedReferenceExpression implements Expression {
+public class LocalReferenceExpression implements Expression {
private final String name;
- public UnresolvedReferenceExpression(String name) {
+ private final TypeInformation<?> resultType;
+
+ public LocalReferenceExpression(String name, TypeInformation<?> resultType) {
this.name = Preconditions.checkNotNull(name);
+ this.resultType = Preconditions.checkNotNull(resultType);
}
public String getName() {
return name;
}
+ public TypeInformation<?> getResultType() {
+ return resultType;
+ }
+
@Override
public List<Expression> getChildren() {
return Collections.emptyList();
@@ -62,13 +67,14 @@ public final class UnresolvedReferenceExpression implements Expression {
if (o == null || getClass() != o.getClass()) {
return false;
}
- UnresolvedReferenceExpression that = (UnresolvedReferenceExpression) o;
- return Objects.equals(name, that.name);
+ LocalReferenceExpression that = (LocalReferenceExpression) o;
+ return Objects.equals(name, that.name) &&
+ Objects.equals(resultType, that.resultType);
}
@Override
public int hashCode() {
- return Objects.hash(name);
+ return Objects.hash(name, resultType);
}
@Override
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/UnresolvedReferenceExpression.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/UnresolvedReferenceExpression.java
index e454ea2..d9cf34a 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/UnresolvedReferenceExpression.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/UnresolvedReferenceExpression.java
@@ -26,10 +26,11 @@ import java.util.List;
import java.util.Objects;
/**
- * An unresolved reference to a field or table.
+ * An unresolved reference to a field, table, or local reference.
*
* <p>This is a purely API facing expression that will be resolved into
- * {@link FieldReferenceExpression} or {@link TableReferenceExpression}.
+ * {@link FieldReferenceExpression}, {@link LocalReferenceExpression},
+ * or {@link TableReferenceExpression}.
*/
@PublicEvolving
public final class UnresolvedReferenceExpression implements Expression {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java
index 55d0d64..869de04 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java
@@ -38,5 +38,4 @@ public interface ExpressionVisitor<R> {
R visitTypeLiteral(TypeLiteralExpression typeLiteral);
R visit(Expression other);
-
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/AggCodeGenHelper.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/AggCodeGenHelper.scala
index 683973f..8c4848c 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/AggCodeGenHelper.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/AggCodeGenHelper.scala
@@ -28,12 +28,11 @@ import org.apache.flink.table.codegen.CodeGenUtils.{boxedTypeTermForExternalType
import org.apache.flink.table.codegen.OperatorCodeGenerator.STREAM_RECORD
import org.apache.flink.table.codegen.{CodeGenUtils, CodeGeneratorContext, ExprCodeGenerator, GenerateUtils, GeneratedExpression, OperatorCodeGenerator}
import org.apache.flink.table.dataformat.{BaseRow, GenericRow}
-import org.apache.flink.table.expressions.{CallExpression, Expression, ExpressionVisitor, FieldReferenceExpression, ResolvedAggInputReference, ResolvedAggLocalReference, RexNodeConverter, SymbolExpression, TypeLiteralExpression, UnresolvedReferenceExpression, ValueLiteralExpression}
+import org.apache.flink.table.expressions.{CallExpression, Expression, ExpressionVisitor, FieldReferenceExpression, LocalReferenceExpression, ResolvedAggInputReference, ResolvedAggLocalReference, RexNodeConverter, SymbolExpression, TypeLiteralExpression, UnresolvedReferenceExpression, ValueLiteralExpression}
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils.{getAccumulatorTypeOfAggregateFunction, getAggUserDefinedInputTypes, getResultTypeOfAggregateFunction}
import org.apache.flink.table.functions.{AggregateFunction, DeclarativeAggregateFunction, UserDefinedFunction}
import org.apache.flink.table.generated.{GeneratedAggsHandleFunction, GeneratedOperator}
import org.apache.flink.table.runtime.context.ExecutionContextImpl
-
import org.apache.calcite.rel.core.AggregateCall
import org.apache.calcite.rex.RexNode
import org.apache.calcite.tools.RelBuilder
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/HashAggCodeGenHelper.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/HashAggCodeGenHelper.scala
index 59e3413..9d9c7e8 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/HashAggCodeGenHelper.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/HashAggCodeGenHelper.scala
@@ -27,13 +27,12 @@ import org.apache.flink.table.codegen.CodeGenUtils.{binaryRowFieldSetAccess, bin
import org.apache.flink.table.codegen.agg.batch.AggCodeGenHelper.buildAggregateArgsMapping
import org.apache.flink.table.codegen.{CodeGenUtils, CodeGeneratorContext, ExprCodeGenerator, GenerateUtils, GeneratedExpression, OperatorCodeGenerator, SortCodeGenerator}
import org.apache.flink.table.dataformat.{BaseRow, BinaryRow, GenericRow, JoinedRow}
-import org.apache.flink.table.expressions.{CallExpression, Expression, ExpressionVisitor, FieldReferenceExpression, ResolvedAggInputReference, RexNodeConverter, SymbolExpression, TypeLiteralExpression, UnresolvedReferenceExpression, ValueLiteralExpression}
+import org.apache.flink.table.expressions.{CallExpression, Expression, ExpressionVisitor, FieldReferenceExpression, LocalReferenceExpression, ResolvedAggInputReference, RexNodeConverter, SymbolExpression, TypeLiteralExpression, UnresolvedReferenceExpression, ValueLiteralExpression}
import org.apache.flink.table.functions.{AggregateFunction, DeclarativeAggregateFunction, UserDefinedFunction}
import org.apache.flink.table.generated.{NormalizedKeyComputer, RecordComparator}
import org.apache.flink.table.runtime.aggregate.{BytesHashMap, BytesHashMapSpillMemorySegmentPool}
import org.apache.flink.table.runtime.sort.BufferedKVExternalSorter
import org.apache.flink.table.typeutils.BinaryRowSerializer
-
import org.apache.calcite.rel.core.AggregateCall
import org.apache.calcite.tools.RelBuilder
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/ExpressionResolver.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/ExpressionResolver.java
new file mode 100644
index 0000000..7b0958d
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/ExpressionResolver.java
@@ -0,0 +1,317 @@
+/*
+ * 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.expressions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.api.GroupWindow;
+import org.apache.flink.table.api.OverWindow;
+import org.apache.flink.table.api.SessionWithGapOnTimeWithAlias;
+import org.apache.flink.table.api.SlideWithSizeAndSlideOnTimeWithAlias;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TumbleWithSizeOnTimeWithAlias;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.expressions.catalog.FunctionDefinitionCatalog;
+import org.apache.flink.table.expressions.lookups.FieldReferenceLookup;
+import org.apache.flink.table.expressions.lookups.TableReferenceLookup;
+import org.apache.flink.table.expressions.rules.ResolverRule;
+import org.apache.flink.table.expressions.rules.ResolverRules;
+import org.apache.flink.table.operations.TableOperation;
+import org.apache.flink.table.plan.logical.LogicalOverWindow;
+import org.apache.flink.table.plan.logical.LogicalWindow;
+import org.apache.flink.table.plan.logical.SessionGroupWindow;
+import org.apache.flink.table.plan.logical.SlidingGroupWindow;
+import org.apache.flink.table.plan.logical.TumblingGroupWindow;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import scala.Some;
+
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Tries to resolve all unresolved expressions such as {@link UnresolvedReferenceExpression}
+ * or calls such as {@link BuiltInFunctionDefinitions#OVER}.
+ *
+ * <p>The default set of rules ({@link ExpressionResolver#getResolverRules()}) will resolve following references:
+ * <ul>
+ * <li>flatten '*' to all fields of underlying inputs</li>
+ * <li>join over aggregates with corresponding over windows into a single resolved call</li>
+ * <li>resolve remaining unresolved references to fields, tables or local references</li>
+ * <li>replace call to {@link BuiltInFunctionDefinitions#FLATTEN}</li>
+ * <li>performs call arguments types validation and inserts additional casts if possible</li>
+ * </ul>
+ */
+@Internal
+public class ExpressionResolver {
+
+ /**
+ * List of rules that will be applied during expression resolution.
+ */
+ public static List<ResolverRule> getResolverRules() {
+ return Arrays.asList(
+ ResolverRules.LOOKUP_CALL_BY_NAME,
+ ResolverRules.FLATTEN_STAR_REFERENCE,
+ ResolverRules.OVER_WINDOWS,
+ ResolverRules.FIELD_RESOLVE,
+ ResolverRules.FLATTEN_CALL,
+ ResolverRules.RESOLVE_CALL_BY_ARGUMENTS);
+ }
+
+ private final PlannerExpressionConverter bridgeConverter = PlannerExpressionConverter.INSTANCE();
+
+ private final FieldReferenceLookup fieldLookup;
+
+ private final TableReferenceLookup tableLookup;
+
+ private final FunctionDefinitionCatalog functionLookup;
+
+ //TODO change to LocalReferenceLookup, once we don't need to resolve fields to create LocalReferenceExpression
+ private final Map<String, LocalReferenceExpression> localReferences = new HashMap<>();
+
+ private final Map<Expression, LogicalOverWindow> overWindows;
+
+ private final Function<List<Expression>, List<Expression>> resolveFunction;
+
+ private ExpressionResolver(
+ TableReferenceLookup tableLookup,
+ FunctionDefinitionCatalog functionCatalog,
+ FieldReferenceLookup fieldLookup,
+ List<OverWindow> overWindows,
+ @Nullable GroupWindow groupWindow,
+ List<ResolverRule> rules) {
+ this.tableLookup = Preconditions.checkNotNull(tableLookup);
+ this.fieldLookup = Preconditions.checkNotNull(fieldLookup);
+ this.functionLookup = Preconditions.checkNotNull(functionCatalog);
+ this.resolveFunction = concatenateRules(rules);
+
+ this.overWindows = prepareOverWindows(overWindows);
+ prepareLocalReferencesFromGroupWindows(groupWindow);
+ }
+
+ /**
+ * Creates a builder for {@link ExpressionResolver}. One can add additional properties to the resolver
+ * like e.g. {@link GroupWindow} or {@link OverWindow}. You can also add additional {@link ResolverRule}.
+ *
+ * @param tableCatalog a way to lookup a table reference by name
+ * @param functionDefinitionCatalog a way to lookup call by name
+ * @param inputs inputs to use for field resolution
+ * @return builder for resolver
+ */
+ public static ExpressionResolverBuilder resolverFor(
+ TableReferenceLookup tableCatalog,
+ FunctionDefinitionCatalog functionDefinitionCatalog,
+ TableOperation... inputs) {
+ return new ExpressionResolverBuilder(inputs, tableCatalog, functionDefinitionCatalog);
+ }
+
+ /**
+ * Resolves given expressions with configured set of rules. All expressions of an operation should be
+ * given at once as some rules might assume the order of expressions.
+ *
+ * <p>After this method is applied the returned expressions should be ready to be converted to planner specific
+ * expressions.
+ *
+ * @param expressions list of expressions to resolve.
+ * @return resolved list of expression
+ */
+ public List<Expression> resolve(List<Expression> expressions) {
+ return resolveFunction.apply(expressions);
+ }
+
+ /**
+ * Converts an API class to a logical window for planning with expressions already resolved.
+ *
+ * @param window window to resolve
+ * @return logical window with expressions resolved
+ */
+ public LogicalWindow resolveGroupWindow(GroupWindow window) {
+ Expression alias = window.getAlias();
+
+ if (!(alias instanceof UnresolvedReferenceExpression)) {
+ throw new ValidationException("Alias of group window should be an UnresolvedFieldReference");
+ }
+
+ final String windowName = ((UnresolvedReferenceExpression) alias).getName();
+ PlannerExpression timeField = resolveFieldsInSingleExpression(window.getTimeField()).accept(bridgeConverter);
+
+ //TODO replace with LocalReferenceExpression
+ WindowReference resolvedAlias = new WindowReference(windowName, new Some<>(timeField.resultType()));
+
+ if (window instanceof TumbleWithSizeOnTimeWithAlias) {
+ TumbleWithSizeOnTimeWithAlias tw = (TumbleWithSizeOnTimeWithAlias) window;
+ return new TumblingGroupWindow(
+ resolvedAlias,
+ timeField,
+ resolveFieldsInSingleExpression(tw.getSize()).accept(bridgeConverter));
+ } else if (window instanceof SlideWithSizeAndSlideOnTimeWithAlias) {
+ SlideWithSizeAndSlideOnTimeWithAlias sw = (SlideWithSizeAndSlideOnTimeWithAlias) window;
+ return new SlidingGroupWindow(
+ resolvedAlias,
+ timeField,
+ resolveFieldsInSingleExpression(sw.getSize()).accept(bridgeConverter),
+ resolveFieldsInSingleExpression(sw.getSlide()).accept(bridgeConverter));
+ } else if (window instanceof SessionWithGapOnTimeWithAlias) {
+ SessionWithGapOnTimeWithAlias sw = (SessionWithGapOnTimeWithAlias) window;
+ return new SessionGroupWindow(
+ resolvedAlias,
+ timeField,
+ resolveFieldsInSingleExpression(sw.getGap()).accept(bridgeConverter));
+ } else {
+ throw new TableException("Unknown window type");
+ }
+ }
+
+ private Function<List<Expression>, List<Expression>> concatenateRules(List<ResolverRule> rules) {
+ return rules.stream()
+ .reduce(
+ Function.identity(),
+ (function, resolverRule) -> function.andThen(exprs -> resolverRule.apply(exprs,
+ new ExpressionResolverContext())),
+ Function::andThen
+ );
+ }
+
+ private void prepareLocalReferencesFromGroupWindows(@Nullable GroupWindow groupWindow) {
+ if (groupWindow != null) {
+ String windowName = ((UnresolvedReferenceExpression) groupWindow.getAlias()).getName();
+ TypeInformation<?> windowType = resolveFieldsInSingleExpression(groupWindow.getTimeField())
+ .accept(bridgeConverter)
+ .resultType();
+
+ localReferences.put(windowName, new LocalReferenceExpression(windowName, windowType));
+ }
+ }
+
+ private Map<Expression, LogicalOverWindow> prepareOverWindows(List<OverWindow> overWindows) {
+ return overWindows.stream()
+ .map(this::resolveOverWindow)
+ .collect(Collectors.toMap(
+ LogicalOverWindow::alias,
+ Function.identity()
+ ));
+ }
+
+ private Expression resolveFieldsInSingleExpression(Expression expression) {
+ List<Expression> expressions = ResolverRules.FIELD_RESOLVE.apply(Collections.singletonList(expression),
+ new ExpressionResolverContext());
+
+ if (expressions.size() != 1) {
+ throw new TableException("Expected a single expression as a result. Got: " + expressions);
+ }
+
+ return expressions.get(0);
+ }
+
+ private class ExpressionResolverContext implements ResolverRule.ResolutionContext {
+
+ @Override
+ public FieldReferenceLookup referenceLookup() {
+ return fieldLookup;
+ }
+
+ @Override
+ public TableReferenceLookup tableLookup() {
+ return tableLookup;
+ }
+
+ @Override
+ public FunctionDefinitionCatalog functionDefinitionLookup() {
+ return functionLookup;
+ }
+
+ @Override
+ public Optional<LocalReferenceExpression> getLocalReference(String alias) {
+ return Optional.ofNullable(localReferences.get(alias));
+ }
+
+ @Override
+ public Optional<LogicalOverWindow> getOverWindow(Expression alias) {
+ return Optional.ofNullable(overWindows.get(alias));
+ }
+
+ @Override
+ public PlannerExpression bridge(Expression expression) {
+ return expression.accept(bridgeConverter);
+ }
+ }
+
+ private LogicalOverWindow resolveOverWindow(OverWindow overWindow) {
+ return new LogicalOverWindow(
+ overWindow.getAlias(),
+ overWindow.getPartitioning().stream().map(this::resolveFieldsInSingleExpression).collect(toList()),
+ resolveFieldsInSingleExpression(overWindow.getOrder()),
+ resolveFieldsInSingleExpression(overWindow.getPreceding()),
+ overWindow.getFollowing().map(this::resolveFieldsInSingleExpression)
+ );
+ }
+
+ /**
+ * Builder for creating {@link ExpressionResolver}.
+ */
+ public static class ExpressionResolverBuilder {
+
+ private final List<TableOperation> tableOperations;
+ private final TableReferenceLookup tableCatalog;
+ private final FunctionDefinitionCatalog functionCatalog;
+ private List<OverWindow> logicalOverWindows = new ArrayList<>();
+ private GroupWindow groupWindow = null;
+ private List<ResolverRule> rules = new ArrayList<>(getResolverRules());
+
+ private ExpressionResolverBuilder(
+ TableOperation[] tableOperations,
+ TableReferenceLookup tableCatalog,
+ FunctionDefinitionCatalog functionCatalog) {
+ this.tableOperations = Arrays.asList(tableOperations);
+ this.tableCatalog = tableCatalog;
+ this.functionCatalog = functionCatalog;
+ }
+
+ public ExpressionResolverBuilder withOverWindows(List<OverWindow> windows) {
+ this.logicalOverWindows = Preconditions.checkNotNull(windows);
+ return this;
+ }
+
+ public ExpressionResolverBuilder withGroupWindow(GroupWindow window) {
+ this.groupWindow = Preconditions.checkNotNull(window);
+ return this;
+ }
+
+ public ExpressionResolver build() {
+ return new ExpressionResolver(
+ tableCatalog,
+ functionCatalog,
+ new FieldReferenceLookup(tableOperations),
+ logicalOverWindows,
+ groupWindow,
+ rules);
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/lookups/FieldReferenceLookup.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/lookups/FieldReferenceLookup.java
new file mode 100644
index 0000000..5cd6410
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/lookups/FieldReferenceLookup.java
@@ -0,0 +1,110 @@
+/*
+ * 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.expressions.lookups;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.operations.TableOperation;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Provides a way to look up field reference by the name of the field.
+ */
+@Internal
+public class FieldReferenceLookup {
+
+ private final List<Map<String, FieldReferenceExpression>> fieldReferences;
+
+ public FieldReferenceLookup(List<TableOperation> tableOperations) {
+ fieldReferences = prepareFieldReferences(tableOperations);
+ }
+
+ /**
+ * Tries to resolve {@link FieldReferenceExpression} using given name in underlying inputs.
+ *
+ * @param name name of field to look for
+ * @return resolved field reference or empty if could not find field with given name.
+ * @throws org.apache.flink.table.api.ValidationException if the name is ambiguous.
+ */
+ public Optional<FieldReferenceExpression> lookupField(String name) {
+ List<FieldReferenceExpression> matchingFields = fieldReferences.stream()
+ .map(input -> input.get(name))
+ .filter(Objects::nonNull)
+ .collect(toList());
+
+ if (matchingFields.size() == 1) {
+ return Optional.of(matchingFields.get(0));
+ } else if (matchingFields.size() == 0) {
+ return Optional.empty();
+ } else {
+ throw failAmbigousColumn(name);
+ }
+ }
+
+ /**
+ * Gives all fields of underlying inputs in order of those inputs and order of fields within input.
+ *
+ * @return concatenated list of fields of all inputs.
+ */
+ public List<FieldReferenceExpression> getAllInputFields() {
+ return fieldReferences.stream().flatMap(input -> input.values().stream()).collect(toList());
+ }
+
+ private static List<Map<String, FieldReferenceExpression>> prepareFieldReferences(
+ List<TableOperation> tableOperations) {
+ return IntStream.range(0, tableOperations.size())
+ .mapToObj(idx -> prepareFieldsInInput(tableOperations.get(idx), idx))
+ .collect(Collectors.toList());
+ }
+
+ private static Map<String, FieldReferenceExpression> prepareFieldsInInput(TableOperation input, int inputIdx) {
+ TableSchema tableSchema = input.getTableSchema();
+ return IntStream.range(0, tableSchema.getFieldCount())
+ .mapToObj(i -> new FieldReferenceExpression(
+ tableSchema.getFieldName(i).get(),
+ tableSchema.getFieldType(i).get(),
+ inputIdx,
+ i))
+ .collect(Collectors.toMap(
+ FieldReferenceExpression::getName,
+ Function.identity(),
+ (fieldRef1, fieldRef2) -> {
+ throw failAmbigousColumn(fieldRef1.getName());
+ },
+ // we need to maintain order of fields within input for resolving e.g. '*' reference
+ LinkedHashMap::new
+ ));
+ }
+
+ private static ValidationException failAmbigousColumn(String name) {
+ return new ValidationException("Ambigous column name: " + name);
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/lookups/TableReferenceLookup.java
similarity index 58%
copy from flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java
copy to flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/lookups/TableReferenceLookup.java
index 55d0d64..0d6112c 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/lookups/TableReferenceLookup.java
@@ -16,27 +16,24 @@
* limitations under the License.
*/
-package org.apache.flink.table.expressions;
+package org.apache.flink.table.expressions.lookups;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.expressions.TableReferenceExpression;
+
+import java.util.Optional;
/**
- * The visitor definition of {@link Expression}. An expression visitor transforms an
- * expression to instances of {@code R}.
+ * Provides a way to look up table reference by the name of the table.
*/
-@PublicEvolving
-public interface ExpressionVisitor<R> {
-
- R visitCall(CallExpression call);
-
- R visitSymbol(SymbolExpression symbolExpression);
-
- R visitValueLiteral(ValueLiteralExpression valueLiteralExpression);
-
- R visitFieldReference(FieldReferenceExpression fieldReference);
-
- R visitTypeLiteral(TypeLiteralExpression typeLiteral);
-
- R visit(Expression other);
-
+@Internal
+public interface TableReferenceLookup {
+
+ /**
+ * Tries to resolve given name to {@link TableReferenceExpression}.
+ *
+ * @param name name of table to look for
+ * @return resolved field reference or empty if could not find table with given name.
+ */
+ Optional<TableReferenceExpression> lookupTable(String name);
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/FlattenCallRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/FlattenCallRule.java
new file mode 100644
index 0000000..5a4020f
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/FlattenCallRule.java
@@ -0,0 +1,90 @@
+/*
+ * 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.expressions.rules;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.table.expressions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.PlannerExpression;
+
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.singletonList;
+import static org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral;
+import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.GET;
+
+/**
+ * Replaces {@link BuiltInFunctionDefinitions#FLATTEN} with calls to {@link BuiltInFunctionDefinitions#GET} for all
+ * fields of underlying field of complex type.
+ */
+@Internal
+final class FlattenCallRule implements ResolverRule {
+
+ @Override
+ public List<Expression> apply(List<Expression> expression, ResolutionContext context) {
+ return expression.stream()
+ .flatMap(expr -> expr.accept(new FlatteningCallVisitor(context)).stream())
+ .collect(Collectors.toList());
+ }
+
+ private class FlatteningCallVisitor extends RuleExpressionVisitor<List<Expression>> {
+
+ FlatteningCallVisitor(ResolutionContext context) {
+ super(context);
+ }
+
+ @Override
+ public List<Expression> visitCall(CallExpression call) {
+ if (call.getFunctionDefinition() == BuiltInFunctionDefinitions.FLATTEN) {
+ return executeFlatten(call);
+ }
+
+ return singletonList(call);
+ }
+
+ private List<Expression> executeFlatten(CallExpression call) {
+ Expression arg = call.getChildren().get(0);
+ PlannerExpression plannerExpression = resolutionContext.bridge(arg);
+ plannerExpression.validateInput();
+ TypeInformation<?> resultType = plannerExpression.resultType();
+ if (resultType instanceof CompositeType) {
+ return flattenCompositeType(arg, (CompositeType<?>) resultType);
+ } else {
+ return singletonList(arg);
+ }
+ }
+
+ private List<Expression> flattenCompositeType(Expression arg, CompositeType<?> resultType) {
+ return IntStream.range(0, resultType.getArity())
+ .mapToObj(idx -> new CallExpression(GET, asList(arg, valueLiteral(resultType.getFieldNames()[idx]))))
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ protected List<Expression> defaultMethod(Expression expression) {
+ return singletonList(expression);
+ }
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/LookupCallByNameRule.java
similarity index 50%
copy from flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java
copy to flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/LookupCallByNameRule.java
index 55d0d64..4f76f08 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/LookupCallByNameRule.java
@@ -16,27 +16,24 @@
* limitations under the License.
*/
-package org.apache.flink.table.expressions;
+package org.apache.flink.table.expressions.rules;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.LookupCallResolver;
+
+import java.util.List;
+import java.util.stream.Collectors;
/**
- * The visitor definition of {@link Expression}. An expression visitor transforms an
- * expression to instances of {@code R}.
+ * Resolves {@link org.apache.flink.table.expressions.LookupCallExpression} to
+ * a corresponding {@link org.apache.flink.table.expressions.FunctionDefinition}.
*/
-@PublicEvolving
-public interface ExpressionVisitor<R> {
-
- R visitCall(CallExpression call);
-
- R visitSymbol(SymbolExpression symbolExpression);
-
- R visitValueLiteral(ValueLiteralExpression valueLiteralExpression);
-
- R visitFieldReference(FieldReferenceExpression fieldReference);
-
- R visitTypeLiteral(TypeLiteralExpression typeLiteral);
-
- R visit(Expression other);
-
+@Internal
+final class LookupCallByNameRule implements ResolverRule {
+ @Override
+ public List<Expression> apply(List<Expression> expression, ResolutionContext context) {
+ LookupCallResolver lookupCallResolver = new LookupCallResolver(context.functionDefinitionLookup());
+ return expression.stream().map(expr -> expr.accept(lookupCallResolver)).collect(Collectors.toList());
+ }
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/OverWindowResolverRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/OverWindowResolverRule.java
new file mode 100644
index 0000000..c3603d7
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/OverWindowResolverRule.java
@@ -0,0 +1,101 @@
+/*
+ * 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.expressions.rules;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.expressions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.PlannerExpression;
+import org.apache.flink.table.plan.logical.LogicalOverWindow;
+import org.apache.flink.table.typeutils.RowIntervalTypeInfo;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.emptyList;
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Joins call to {@link BuiltInFunctionDefinitions#OVER} with corresponding over window
+ * and creates a fully resolved over aggregation.
+ */
+@Internal
+final class OverWindowResolverRule implements ResolverRule {
+
+ @Override
+ public List<Expression> apply(List<Expression> expression, ResolutionContext context) {
+ return expression.stream()
+ .map(expr -> expr.accept(new ExpressionResolverVisitor(context)))
+ .collect(Collectors.toList());
+ }
+
+ private class ExpressionResolverVisitor extends RuleExpressionVisitor<Expression> {
+
+ ExpressionResolverVisitor(ResolutionContext context) {
+ super(context);
+ }
+
+ @Override
+ public Expression visitCall(CallExpression call) {
+
+ if (call.getFunctionDefinition() == BuiltInFunctionDefinitions.OVER) {
+ List<Expression> children = call.getChildren();
+ Expression alias = children.get(1);
+
+ LogicalOverWindow referenceWindow = resolutionContext.getOverWindow(alias)
+ .orElseThrow(() -> new ValidationException("Could not resolve over call."));
+
+ Expression following = calculateOverWindowFollowing(referenceWindow);
+ List<Expression> newArgs = new ArrayList<>(asList(
+ children.get(0),
+ referenceWindow.orderBy(),
+ referenceWindow.preceding(),
+ following));
+
+ newArgs.addAll(referenceWindow.partitionBy());
+ return new CallExpression(call.getFunctionDefinition(), newArgs);
+ } else {
+ return new CallExpression(
+ call.getFunctionDefinition(),
+ call.getChildren().stream().map(expr -> expr.accept(this)).collect(toList()));
+ }
+ }
+
+ private Expression calculateOverWindowFollowing(LogicalOverWindow referenceWindow) {
+ return referenceWindow.following().orElseGet(() -> {
+ PlannerExpression preceding = resolutionContext.bridge(referenceWindow.preceding());
+ if (preceding.resultType() instanceof RowIntervalTypeInfo) {
+ return new CallExpression(BuiltInFunctionDefinitions.CURRENT_ROW, emptyList());
+ } else {
+ return new CallExpression(BuiltInFunctionDefinitions.CURRENT_RANGE, emptyList());
+ }
+ }
+ );
+ }
+
+ @Override
+ protected Expression defaultMethod(Expression expression) {
+ return expression;
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ReferenceResolverRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ReferenceResolverRule.java
new file mode 100644
index 0000000..378a8c4
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ReferenceResolverRule.java
@@ -0,0 +1,85 @@
+/*
+ * 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.expressions.rules;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.UnresolvedReferenceExpression;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static java.lang.String.format;
+
+/**
+ * Resolves {@link UnresolvedReferenceExpression} to either
+ * {@link org.apache.flink.table.expressions.FieldReferenceExpression},
+ * {@link org.apache.flink.table.expressions.TableReferenceExpression}, or
+ * {@link org.apache.flink.table.expressions.LocalReferenceExpression} in this order.
+ */
+@Internal
+final class ReferenceResolverRule implements ResolverRule {
+
+ @Override
+ public List<Expression> apply(List<Expression> expression, ResolutionContext context) {
+ return expression.stream()
+ .map(expr -> expr.accept(new ExpressionResolverVisitor(context)))
+ .collect(Collectors.toList());
+ }
+
+ private class ExpressionResolverVisitor extends RuleExpressionVisitor<Expression> {
+
+ ExpressionResolverVisitor(ResolutionContext resolutionContext) {
+ super(resolutionContext);
+ }
+
+ @Override
+ public Expression visitCall(CallExpression call) {
+ List<Expression> resolvedArgs = call.getChildren()
+ .stream()
+ .map(expr -> expr.accept(this))
+ .collect(Collectors.toList());
+
+ return new CallExpression(call.getFunctionDefinition(), resolvedArgs);
+ }
+
+ @Override
+ public Expression visitUnresolvedReference(UnresolvedReferenceExpression unresolvedReference) {
+ return resolutionContext.referenceLookup().lookupField(unresolvedReference.getName())
+ .map(expr -> (Expression) expr)
+ .orElseGet(() ->
+ resolutionContext.tableLookup().lookupTable(unresolvedReference.getName())
+ .map(expr -> (Expression) expr)
+ .orElseGet(() -> resolutionContext.getLocalReference(unresolvedReference.getName())
+ .orElseThrow(() -> failForField(unresolvedReference)
+ )));
+ }
+
+ private ValidationException failForField(UnresolvedReferenceExpression fieldReference) {
+ return new ValidationException(format("Cannot resolve field [%s]", fieldReference.getName()));
+ }
+
+ @Override
+ protected Expression defaultMethod(Expression expression) {
+ return expression;
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ResolveCallByArgumentsRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ResolveCallByArgumentsRule.java
new file mode 100644
index 0000000..af3b4e3
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ResolveCallByArgumentsRule.java
@@ -0,0 +1,116 @@
+/*
+ * 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.expressions.rules;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.expressions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.InputTypeSpec;
+import org.apache.flink.table.expressions.PlannerExpression;
+import org.apache.flink.table.typeutils.TypeCoercion;
+
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static java.util.Arrays.asList;
+import static org.apache.flink.table.expressions.ApiExpressionUtils.typeLiteral;
+import static org.apache.flink.table.util.JavaScalaConversionUtil.toJava;
+
+/**
+ * It checks if a {@link CallExpression} can work with given arguments.
+ * If the call expects different types of arguments, but the given arguments
+ * have types that can be casted, a {@link BuiltInFunctionDefinitions#CAST}
+ * expression is inserted.
+ */
+@Internal
+final class ResolveCallByArgumentsRule implements ResolverRule {
+
+ @Override
+ public List<Expression> apply(List<Expression> expression, ResolutionContext context) {
+ return expression.stream()
+ .map(expr -> expr.accept(new CallArgumentsCastingVisitor(context)))
+ .collect(Collectors.toList());
+ }
+
+ private class CallArgumentsCastingVisitor extends RuleExpressionVisitor<Expression> {
+
+ CallArgumentsCastingVisitor(ResolutionContext context) {
+ super(context);
+ }
+
+ @Override
+ public Expression visitCall(CallExpression call) {
+ PlannerExpression plannerCall = resolutionContext.bridge(call);
+ if (plannerCall instanceof InputTypeSpec) {
+ List<TypeInformation<?>> expectedTypes = toJava(((InputTypeSpec) plannerCall).expectedTypes());
+ return performTypeResolution(call, expectedTypes);
+ } else {
+ return validateArguments(call, plannerCall);
+ }
+ }
+
+ private Expression performTypeResolution(CallExpression call, List<TypeInformation<?>> expectedTypes) {
+ List<PlannerExpression> args = call.getChildren()
+ .stream()
+ .map(resolutionContext::bridge)
+ .collect(Collectors.toList());
+
+ List<Expression> newArgs = IntStream.range(0, args.size())
+ .mapToObj(idx -> castIfNeeded(args.get(idx), expectedTypes.get(idx)))
+ .collect(Collectors.toList());
+
+ return new CallExpression(call.getFunctionDefinition(), newArgs);
+ }
+
+ private Expression validateArguments(CallExpression call, PlannerExpression plannerCall) {
+ if (!plannerCall.valid()) {
+ throw new ValidationException(String.format("Invalid arguments [%s] for call: %s",
+ call.getChildren(),
+ call));
+ }
+
+ return call;
+ }
+
+ private Expression castIfNeeded(PlannerExpression childExpression, TypeInformation<?> expectedType) {
+ TypeInformation<?> actualType = childExpression.resultType();
+ if (actualType.equals(expectedType)) {
+ return childExpression;
+ } else if (TypeCoercion.canSafelyCast(actualType, expectedType)) {
+ return new CallExpression(
+ BuiltInFunctionDefinitions.CAST,
+ asList(childExpression, typeLiteral(expectedType))
+ );
+ } else {
+ throw new ValidationException(String.format("Incompatible type of argument: %s Expected: %s",
+ childExpression,
+ expectedType));
+ }
+ }
+
+ @Override
+ protected Expression defaultMethod(Expression expression) {
+ return expression;
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ResolverRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ResolverRule.java
new file mode 100644
index 0000000..119ae63
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ResolverRule.java
@@ -0,0 +1,79 @@
+/*
+ * 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.expressions.rules;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.LocalReferenceExpression;
+import org.apache.flink.table.expressions.PlannerExpression;
+import org.apache.flink.table.expressions.catalog.FunctionDefinitionCatalog;
+import org.apache.flink.table.expressions.lookups.FieldReferenceLookup;
+import org.apache.flink.table.expressions.lookups.TableReferenceLookup;
+import org.apache.flink.table.plan.logical.LogicalOverWindow;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Rule that can be applied during resolution of {@link Expression}. Rules are applied to a collection of expressions
+ * at once, e.g. all expressions in a projection. One must consider order in which rules are applied. Some rules
+ * might e.g. require that references to fields have to be already resolved.
+ */
+@Internal
+public interface ResolverRule {
+
+ List<Expression> apply(List<Expression> expression, ResolutionContext context);
+
+ /**
+ * Contextual information that can be used during application of the rule. E.g. one can access fields in inputs by
+ * name etc.
+ */
+ interface ResolutionContext {
+
+ /**
+ * Access to available {@link org.apache.flink.table.expressions.FieldReferenceExpression} in inputs.
+ */
+ FieldReferenceLookup referenceLookup();
+
+ /**
+ * Access to available {@link org.apache.flink.table.expressions.TableReferenceExpression}.
+ */
+ TableReferenceLookup tableLookup();
+
+ /**
+ * Access to available {@link org.apache.flink.table.expressions.FunctionDefinition}s.
+ */
+ FunctionDefinitionCatalog functionDefinitionLookup();
+
+ /**
+ * Access to available local references.
+ */
+ Optional<LocalReferenceExpression> getLocalReference(String alias);
+
+ /**
+ * Lookup for over windows.
+ */
+ Optional<LogicalOverWindow> getOverWindow(Expression alias);
+
+ /**
+ * Temporary way to convert expression to PlannerExpression to evaluate type.
+ */
+ PlannerExpression bridge(Expression expression);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ResolverRules.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ResolverRules.java
new file mode 100644
index 0000000..ee874ac
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ResolverRules.java
@@ -0,0 +1,62 @@
+/*
+ * 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.expressions.rules;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.expressions.UnresolvedReferenceExpression;
+
+/**
+ * Contains instances of {@link ResolverRule}.
+ */
+@Internal
+public final class ResolverRules {
+
+ /**
+ * Rule that resolves flatten call. See {@link FlattenCallRule} for details.
+ */
+ public static final ResolverRule FLATTEN_CALL = new FlattenCallRule();
+
+ /**
+ * Resolves {@link UnresolvedReferenceExpression}. See {@link ReferenceResolverRule} for details.
+ */
+ public static final ResolverRule FIELD_RESOLVE = new ReferenceResolverRule();
+
+ /**
+ * Resolves call based on argument types. See {@link ResolveCallByArgumentsRule} for details.
+ */
+ public static final ResolverRule RESOLVE_CALL_BY_ARGUMENTS = new ResolveCallByArgumentsRule();
+
+ /**
+ * Looks up unresolved call by name. See {@link LookupCallByNameRule} for details.
+ */
+ public static final ResolverRule LOOKUP_CALL_BY_NAME = new LookupCallByNameRule();
+
+ /**
+ * Concatenates over aggregations with corresponding over window. See {@link OverWindowResolverRule} for details.
+ */
+ public static final ResolverRule OVER_WINDOWS = new OverWindowResolverRule();
+
+ /**
+ * Resolves '*' expressions to corresponding fields of inputs. See {@link StarReferenceFlatteningRule} for details.
+ */
+ public static final ResolverRule FLATTEN_STAR_REFERENCE = new StarReferenceFlatteningRule();
+
+ private ResolverRules() {
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/RuleExpressionVisitor.java
similarity index 57%
copy from flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java
copy to flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/RuleExpressionVisitor.java
index 55d0d64..59a2338 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ExpressionVisitor.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/RuleExpressionVisitor.java
@@ -16,27 +16,22 @@
* limitations under the License.
*/
-package org.apache.flink.table.expressions;
+package org.apache.flink.table.expressions.rules;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.expressions.ApiExpressionDefaultVisitor;
+import org.apache.flink.table.expressions.rules.ResolverRule.ResolutionContext;
/**
- * The visitor definition of {@link Expression}. An expression visitor transforms an
- * expression to instances of {@code R}.
+ * Utility class for {@link ResolverRule} specific visitor that unifies access to
+ * {@link ResolutionContext}.
*/
-@PublicEvolving
-public interface ExpressionVisitor<R> {
+@Internal
+abstract class RuleExpressionVisitor<T> extends ApiExpressionDefaultVisitor<T> {
- R visitCall(CallExpression call);
-
- R visitSymbol(SymbolExpression symbolExpression);
-
- R visitValueLiteral(ValueLiteralExpression valueLiteralExpression);
-
- R visitFieldReference(FieldReferenceExpression fieldReference);
-
- R visitTypeLiteral(TypeLiteralExpression typeLiteral);
-
- R visit(Expression other);
+ protected final ResolutionContext resolutionContext;
+ RuleExpressionVisitor(ResolutionContext resolutionContext) {
+ this.resolutionContext = resolutionContext;
+ }
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/StarReferenceFlatteningRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/StarReferenceFlatteningRule.java
new file mode 100644
index 0000000..86ff7c6
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/StarReferenceFlatteningRule.java
@@ -0,0 +1,65 @@
+/*
+ * 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.expressions.rules;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.UnresolvedReferenceExpression;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static java.util.Collections.singletonList;
+
+/**
+ * Replaces '*' with all available {@link org.apache.flink.table.expressions.FieldReferenceExpression}s
+ * from underlying inputs.
+ */
+@Internal
+final class StarReferenceFlatteningRule implements ResolverRule {
+
+ @Override
+ public List<Expression> apply(List<Expression> expression, ResolutionContext context) {
+ return expression.stream()
+ .flatMap(expr -> expr.accept(new FieldFlatteningVisitor(context)).stream())
+ .collect(Collectors.toList());
+ }
+
+ private static class FieldFlatteningVisitor extends RuleExpressionVisitor<List<Expression>> {
+
+ FieldFlatteningVisitor(ResolutionContext resolutionContext) {
+ super(resolutionContext);
+ }
+
+ @Override
+ public List<Expression> visitUnresolvedReference(UnresolvedReferenceExpression unresolvedReference) {
+ if (unresolvedReference.getName().equals("*")) {
+ return new ArrayList<>(resolutionContext.referenceLookup().getAllInputFields());
+ } else {
+ return singletonList(unresolvedReference);
+ }
+ }
+
+ @Override
+ protected List<Expression> defaultMethod(Expression expression) {
+ return singletonList(expression);
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/ColumnOperationUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/ColumnOperationUtils.java
new file mode 100644
index 0000000..9439702
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/ColumnOperationUtils.java
@@ -0,0 +1,151 @@
+/*
+ * 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.UnresolvedReferenceExpression;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static java.lang.String.format;
+import static org.apache.flink.table.expressions.ApiExpressionUtils.extractName;
+import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.AS;
+
+/**
+ * Utility class for creating projection expressions from column operation.
+ */
+@Internal
+public final class ColumnOperationUtils {
+
+ private static final DropColumnsExtractor dropColumnsExtractor = new DropColumnsExtractor();
+ private static final RenameColumnExtractor renameColumnExtractor = new RenameColumnExtractor();
+
+ /**
+ * Creates a projection list that renames existing columns to new names.
+ *
+ * <p><b>NOTE:</b> Resulting expression are still unresolved.
+ *
+ * @param inputFields names of current columns
+ * @param newAliases new aliases for current columns
+ * @return projection expressions
+ */
+ public static List<Expression> renameColumns(List<String> inputFields, List<Expression> newAliases) {
+ LinkedHashMap<String, Expression> finalFields = new LinkedHashMap<>();
+
+ inputFields.forEach(field -> finalFields.put(field, new UnresolvedReferenceExpression(field)));
+ newAliases.forEach(expr -> {
+ String name = expr.accept(renameColumnExtractor);
+ finalFields.put(name, expr);
+ });
+
+ return new ArrayList<>(finalFields.values());
+ }
+
+ /**
+ * Creates a projection list that adds new or replaces existing (if a column with corresponding name already exists)
+ * columns.
+ *
+ * <p><b>NOTE:</b> Resulting expression are still unresolved.
+ *
+ * @param inputFields names of current columns
+ * @param newExpressions new columns to add
+ * @return projection expressions
+ */
+ public static List<Expression> addOrReplaceColumns(List<String> inputFields, List<Expression> newExpressions) {
+ LinkedHashMap<String, Expression> finalFields = new LinkedHashMap<>();
+
+ inputFields.forEach(field -> finalFields.put(field, new UnresolvedReferenceExpression(field)));
+ newExpressions.forEach(expr -> {
+ String name = extractName(expr).orElse(expr.toString());
+ finalFields.put(name, expr);
+ });
+
+ return new ArrayList<>(finalFields.values());
+ }
+
+ /**
+ * Creates a projection list that removes given columns.
+ *
+ * <p><b>NOTE:</b> Resulting expression are still unresolved.
+ *
+ * @param inputFields names of current columns
+ * @param dropExpressions columns to remove
+ * @return projection expressions
+ */
+ public static List<Expression> dropFields(List<String> inputFields, List<Expression> dropExpressions) {
+ Set<String> columnsToDrop = dropExpressions.stream()
+ .map(expr -> expr.accept(dropColumnsExtractor))
+ .collect(Collectors.toSet());
+
+ columnsToDrop.forEach(c -> {
+ if (!inputFields.contains(c)) {
+ throw new ValidationException(format("Field %s does not exist in source table", c));
+ }
+ });
+
+ return inputFields.stream()
+ .filter(oldName -> !columnsToDrop.contains(oldName))
+ .map(UnresolvedReferenceExpression::new)
+ .collect(Collectors.toList());
+ }
+
+ private static class DropColumnsExtractor extends ApiExpressionDefaultVisitor<String> {
+
+ @Override
+ public String visitUnresolvedReference(UnresolvedReferenceExpression unresolvedReference) {
+ return unresolvedReference.getName();
+ }
+
+ @Override
+ protected String defaultMethod(Expression expression) {
+ throw new ValidationException("Unexpected drop column expression: " + expression);
+ }
+ }
+
+ private static class RenameColumnExtractor extends ApiExpressionDefaultVisitor<String> {
+ @Override
+ public String visitCall(CallExpression call) {
+ if (call.getFunctionDefinition() == AS &&
+ call.getChildren().get(0) instanceof UnresolvedReferenceExpression) {
+ UnresolvedReferenceExpression unresolvedReference = (UnresolvedReferenceExpression) call.getChildren()
+ .get(0);
+ return unresolvedReference.getName();
+ } else {
+ return defaultMethod(call);
+ }
+ }
+
+ @Override
+ protected String defaultMethod(Expression expression) {
+ throw new ValidationException(format("Unexpected field expression type [%s]. " +
+ "Renaming must add an alias to the original field, e.g., a as a1.", expression));
+ }
+ }
+
+ private ColumnOperationUtils() {
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/ProjectionOperationUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/ProjectionOperationUtils.java
new file mode 100644
index 0000000..7d32559
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/ProjectionOperationUtils.java
@@ -0,0 +1,166 @@
+/*
+ * 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.api.common.typeinfo.Types;
+import org.apache.flink.table.expressions.ApiExpressionDefaultVisitor;
+import org.apache.flink.table.expressions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.FunctionDefinition;
+import org.apache.flink.table.expressions.LocalReferenceExpression;
+import org.apache.flink.table.expressions.TableReferenceExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral;
+import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.AS;
+import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.CAST;
+import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.GET;
+
+/**
+ * Utility class for projection specific expressions transformations.
+ */
+@Internal
+public final class ProjectionOperationUtils {
+
+ private final ExtractNameVisitor extractNameVisitor = new ExtractNameVisitor();
+ private final NamingVisitor namingVisitor = new NamingVisitor();
+ private int currentFieldIndex = 0;
+
+ /**
+ * Ensures that all expressions have a derivable name. There is a few categories and naming
+ * rules that apply:
+ * <ul>
+ * <li>{@link FieldReferenceExpression}, {@link TableReferenceExpression},
+ * {@link LocalReferenceExpression} and {@link BuiltInFunctionDefinitions#AS}
+ * are already named}</li>
+ * <li>{@link BuiltInFunctionDefinitions#CAST} use the name of underlying expression
+ * appended with the name of the type</li>
+ * <li>{@link BuiltInFunctionDefinitions#GET} uses pattern
+ * <i>[underlying name][$fieldName]{1, }</i></li>
+ * <li>if none of the above apply a name <i>[_c(idx)]</i> is used where idx is
+ * the index within given expressions</li>
+ * </ul>
+ */
+ public List<Expression> nameExpressions(List<Expression> expression) {
+ return IntStream.range(0, expression.size())
+ .mapToObj(idx -> {
+ currentFieldIndex = idx;
+ return expression.get(idx).accept(namingVisitor);
+ })
+ .collect(Collectors.toList());
+ }
+
+ private class NamingVisitor extends ApiExpressionDefaultVisitor<Expression> {
+
+ @Override
+ public Expression visitCall(CallExpression call) {
+ FunctionDefinition functionDefinition = call.getFunctionDefinition();
+ final Optional<String> rename;
+ if (functionDefinition.equals(CAST)) {
+ rename = nameForCast(call);
+ } else if (functionDefinition.equals(GET)) {
+ rename = nameForGet(call);
+ } else if (functionDefinition.equals(AS)) {
+ rename = Optional.empty();
+ } else {
+ rename = Optional.of(getUniqueName());
+ }
+
+ return rename.map(name -> new CallExpression(AS, Arrays.asList(call, valueLiteral(name)))).orElse(call);
+ }
+
+ private Optional<String> nameForGet(CallExpression call) {
+ return Optional.of(call.accept(extractNameVisitor).orElseGet(ProjectionOperationUtils.this::getUniqueName));
+ }
+
+ private Optional<String> nameForCast(CallExpression call) {
+ Optional<String> innerName = call.getChildren().get(0).accept(extractNameVisitor);
+ Expression type = call.getChildren().get(1);
+ return Optional.of(innerName.map(n -> String.format("%s-%s", n, type))
+ .orElseGet(ProjectionOperationUtils.this::getUniqueName));
+ }
+
+ @Override
+ public Expression visitValueLiteral(ValueLiteralExpression valueLiteralExpression) {
+ return new CallExpression(AS, Arrays.asList(valueLiteralExpression, valueLiteral(getUniqueName())));
+ }
+
+ @Override
+ protected Expression defaultMethod(Expression expression) {
+ return expression;
+ }
+ }
+
+ private class ExtractNameVisitor extends ApiExpressionDefaultVisitor<Optional<String>> {
+
+ @Override
+ public Optional<String> visitCall(CallExpression call) {
+ if (call.getFunctionDefinition().equals(GET)) {
+ return extractNameFromGet(call);
+ } else {
+ return Optional.empty();
+ }
+ }
+
+ private Optional<String> extractNameFromGet(CallExpression call) {
+ Expression child = call.getChildren().get(0);
+ ValueLiteralExpression key = (ValueLiteralExpression) call.getChildren().get(1);
+ final String keySuffix;
+ if (key.getType().equals(Types.INT)) {
+ keySuffix = "$_" + key.getValue();
+ } else {
+ keySuffix = "$" + key.getValue();
+ }
+ return child.accept(this).map(p -> p + keySuffix);
+ }
+
+ @Override
+ public Optional<String> visitLocalReference(LocalReferenceExpression localReference) {
+ return Optional.of(localReference.getName());
+ }
+
+ @Override
+ public Optional<String> visitTableReference(TableReferenceExpression tableReference) {
+ return Optional.of(tableReference.getName());
+ }
+
+ @Override
+ public Optional<String> visitFieldReference(FieldReferenceExpression fieldReference) {
+ return Optional.of(fieldReference.getName());
+ }
+
+ @Override
+ protected Optional<String> defaultMethod(Expression expression) {
+ return Optional.empty();
+ }
+ }
+
+ private String getUniqueName() {
+ return "_c" + currentFieldIndex++;
+ }
+}
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
new file mode 100644
index 0000000..418541f
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/SortOperationUtils.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.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/api/TableEnvironment.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
index 7fe36ae..855ce46 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
@@ -53,7 +53,7 @@ import org.apache.flink.table.descriptors.{ConnectorDescriptor, TableDescriptor}
import org.apache.flink.table.expressions._
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils._
import org.apache.flink.table.functions.{AggregateFunction, ScalarFunction, TableFunction}
-import org.apache.flink.table.plan.OperationTreeBuilder
+import org.apache.flink.table.operations.OperationTreeBuilder
import org.apache.flink.table.plan.cost.DataSetCostFactory
import org.apache.flink.table.plan.logical.{CatalogNode, LogicalRelNode}
import org.apache.flink.table.plan.nodes.FlinkConventions
@@ -1093,7 +1093,7 @@ abstract class TableEnvironment(val config: TableConfig) {
case p: PojoTypeInfo[A] =>
exprs flatMap {
- case (UnresolvedFieldReference(name: String)) =>
+ case UnresolvedFieldReference(name: String) =>
referenceByName(name, p).map((_, name))
case Alias(UnresolvedFieldReference(origName), name: String, _) =>
referenceByName(origName, p).map((_, name))
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/tableImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/tableImpl.scala
index 3d67b44..fa6ccc3 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/tableImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/tableImpl.scala
@@ -25,9 +25,7 @@ import org.apache.flink.api.java.operators.join.JoinType
import org.apache.flink.table.expressions.ApiExpressionUtils.{extractAggregationsAndProperties, extractFieldReferences, replaceAggregationsAndProperties}
import org.apache.flink.table.expressions._
import org.apache.flink.table.functions.{TemporalTableFunction, TemporalTableFunctionImpl}
-import org.apache.flink.table.operations.TableOperation
-import org.apache.flink.table.plan.OperationTreeBuilder
-import org.apache.flink.table.plan.ProjectionTranslator.getLeadingNonAliasExpr
+import org.apache.flink.table.operations.{OperationTreeBuilder, TableOperation}
import org.apache.flink.table.plan.logical._
import org.apache.flink.table.util.JavaScalaConversionUtil.toJava
@@ -125,15 +123,13 @@ class TableImpl(
timeAttribute: Expression,
primaryKey: Expression)
: TemporalTableFunction = {
- val temporalTable = operationTreeBuilder.createTemporalTable(
- timeAttribute,
- primaryKey,
- operationTree)
+ val resolvedTimeAttribute = operationTreeBuilder.resolveExpression(timeAttribute, operationTree)
+ val resolvedPrimaryKey = operationTreeBuilder.resolveExpression(primaryKey, operationTree)
TemporalTableFunctionImpl.create(
operationTree,
- temporalTable.timeAttribute,
- temporalTable.primaryKey)
+ resolvedTimeAttribute,
+ resolvedPrimaryKey)
}
override def as(fields: String): Table = {
@@ -157,8 +153,7 @@ class TableImpl(
}
private def filterInternal(predicate: Expression): Table = {
- val resolvedCallPredicate = predicate.accept(callResolver)
- new TableImpl(tableEnv, operationTreeBuilder.filter(resolvedCallPredicate, operationTree))
+ new TableImpl(tableEnv, operationTreeBuilder.filter(predicate, operationTree))
}
override def where(predicate: String): Table = {
@@ -299,10 +294,10 @@ class TableImpl(
}
wrap(operationTreeBuilder.joinLateral(
operationTree,
- callExpr.accept(callResolver),
+ callExpr,
joinType,
- toJava(joinPredicate.map(_.accept(callResolver))
- )))
+ toJava(joinPredicate)
+ ))
}
override def minus(right: Table): Table = {
@@ -377,7 +372,7 @@ class TableImpl(
}
private def orderByInternal(fields: Seq[Expression]): Table = {
- wrap(operationTreeBuilder.sort(fields.map(_.accept(callResolver)).asJava, operationTree))
+ wrap(operationTreeBuilder.sort(fields.asJava, operationTree))
}
override def offset(offset: Int): Table = {
@@ -417,7 +412,7 @@ class TableImpl(
}
override def addColumns(fields: String): Table = {
- addColumns(ExpressionParser.parseExpressionList(fields): _*);
+ addColumns(ExpressionParser.parseExpressionList(fields): _*)
}
override def addColumns(fields: Expression*): Table = {
@@ -441,7 +436,7 @@ class TableImpl(
val aggNames = extracted.getAggregations
if(aggNames.nonEmpty){
- throw new TableException(
+ throw new ValidationException(
s"The added field expression cannot be an aggregation, found [${aggNames.head}].")
}
@@ -470,15 +465,7 @@ class TableImpl(
}
override def map(mapFunction: Expression): Table = {
- val resolvedMapFunction = mapFunction.accept(callResolver)
- getLeadingNonAliasExpr(resolvedMapFunction) match {
- case callExpr: CallExpression if callExpr.getFunctionDefinition.getType ==
- FunctionDefinition.Type.SCALAR_FUNCTION =>
- case _ =>
- throw new ValidationException("Only ScalarFunction can be used in the map operator.")
- }
-
- wrap(operationTreeBuilder.map(resolvedMapFunction, operationTree))
+ wrap(operationTreeBuilder.map(mapFunction, operationTree))
}
/**
@@ -653,12 +640,11 @@ class OverWindowedTableImpl(
fields: Seq[Expression],
logicalOverWindows: Seq[OverWindow])
: Table = {
- val expressionsWithResolvedCalls = fields.map(_.accept(tableImpl.callResolver))
new TableImpl(
tableImpl.tableEnv,
tableImpl.operationTreeBuilder
- .project(expressionsWithResolvedCalls.asJava,
+ .project(fields.asJava,
tableImpl.operationTree,
logicalOverWindows.asJava)
)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala
index 7470477..1a76c56 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala
@@ -31,22 +31,49 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
override def visitCall(call: CallExpression): PlannerExpression = {
val func = call.getFunctionDefinition
+ val children = call.getChildren.asScala
- // special case: casting requires a type literal
- if (func.equals(CAST)) {
- assert(call.getChildren.size() == 2)
- return Cast(
- call.getChildren.get(0).accept(this),
- call.getChildren.get(1).asInstanceOf[TypeLiteralExpression].getType)
+ // special case: requires individual handling of child expressions
+ func match {
+ case CAST =>
+ assert(children.size == 2)
+ return Cast(
+ children.head.accept(this),
+ children(1).asInstanceOf[TypeLiteralExpression].getType)
+
+ case WINDOW_START =>
+ assert(children.size == 1)
+ val windowReference = translateWindowReference(children.head)
+ return WindowStart(windowReference)
+
+ case WINDOW_END =>
+ assert(children.size == 1)
+ val windowReference = translateWindowReference(children.head)
+ return WindowEnd(windowReference)
+
+ case PROCTIME =>
+ assert(children.size == 1)
+ val windowReference = translateWindowReference(children.head)
+ return ProctimeAttribute(windowReference)
+
+ case ROWTIME =>
+ assert(children.size == 1)
+ val windowReference = translateWindowReference(children.head)
+ return RowtimeAttribute(windowReference)
+
+ case _ =>
}
- val args = call.getChildren.asScala.map(_.accept(this))
+ val args = children.map(_.accept(this))
func match {
case sfd: ScalarFunctionDefinition =>
- PlannerScalarFunctionCall(
+ val call = PlannerScalarFunctionCall(
sfd.getScalarFunction,
args)
+ //it configures underlying state
+ call.validateInput()
+ call
case tfd: TableFunctionDefinition =>
PlannerTableFunctionCall(
@@ -79,7 +106,10 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
case GET =>
assert(args.size == 2)
- GetCompositeField(args.head, getValue(args.last))
+ val expr = GetCompositeField(args.head, getValue(args.last))
+ //it configures underlying state
+ expr.validateInput()
+ expr
case AND =>
assert(args.size == 2)
@@ -569,14 +599,6 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
case ROW =>
RowConstructor(args)
- case WINDOW_START =>
- assert(args.size == 1)
- WindowStart(args.head)
-
- case WINDOW_END =>
- assert(args.size == 1)
- WindowEnd(args.head)
-
case ORDER_ASC =>
assert(args.size == 1)
Asc(args.head)
@@ -613,19 +635,14 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
assert(args.size == 2)
Sha2(args.head, args.last)
- case PROCTIME =>
- assert(args.size == 1)
- ProctimeAttribute(args.head)
-
- case ROWTIME =>
- assert(args.size == 1)
- RowtimeAttribute(args.head)
-
case OVER =>
- assert(args.size == 2)
- UnresolvedOverCall(
+ assert(args.size >= 4)
+ OverCall(
args.head,
- args.last
+ args.slice(4, args.size),
+ args(1),
+ args(2),
+ args(3)
)
case UNBOUNDED_RANGE =>
@@ -715,6 +732,10 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
)
}
+ override def visitLocalReference(localReference: LocalReferenceExpression): PlannerExpression =
+ throw new TableException(
+ "Local reference should be handled individually by a call: " + localReference)
+
override def visitLookupCall(lookupCall: LookupCallExpression): PlannerExpression =
throw new TableException("Unsupported function call: " + lookupCall)
@@ -737,6 +758,16 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
throw new ValidationException("Invalid number of arguments for function.")
}
}
+
+ private def translateWindowReference(reference: Expression): PlannerExpression = reference match {
+ case expr : LocalReferenceExpression =>
+ WindowReference(expr.getName, Some(expr.getResultType))
+ //just because how the datastream is converted to table
+ case expr: UnresolvedReferenceExpression =>
+ UnresolvedFieldReference(expr.getName)
+ case _ =>
+ throw new ValidationException(s"Expected LocalReferenceExpression. Got: $reference")
+ }
}
object PlannerExpressionConverter {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/composite.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/composite.scala
index d23a281..3277540 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/composite.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/composite.scala
@@ -100,7 +100,9 @@ case class GetCompositeField(child: PlannerExpression, key: Any) extends UnaryEx
} else {
None
}
- case c: ResolvedFieldReference => Some(s"${c.name}$$$key")
+ case c: ResolvedFieldReference =>
+ val keySuffix = if (key.isInstanceOf[Int]) s"_$key" else key
+ Some(s"${c.name}$$$keySuffix")
case _ => None
}
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
index 767ff32..edd74a3 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
@@ -99,8 +99,6 @@ case class Alias(child: PlannerExpression, name: String, extraNames: Seq[String]
override private[flink] def validateInput(): ValidationResult = {
if (name == "*") {
ValidationFailure("Alias can not accept '*' as name.")
- } else if (extraNames.nonEmpty) {
- ValidationFailure("Invalid call to Alias with multiple names.")
} else {
ValidationSuccess
}
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
new file mode 100644
index 0000000..20d62d1
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala
@@ -0,0 +1,399 @@
+/*
+ * 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 java.util.{Collections, Optional, List => JList, Map => JMap}
+
+import org.apache.flink.api.java.operators.join.JoinType
+import org.apache.flink.table.api._
+import org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral
+import org.apache.flink.table.expressions.BuiltInFunctionDefinitions.AS
+import org.apache.flink.table.expressions.ExpressionResolver.resolverFor
+import org.apache.flink.table.expressions.FunctionDefinition.Type.SCALAR_FUNCTION
+import org.apache.flink.table.expressions._
+import org.apache.flink.table.expressions.catalog.FunctionDefinitionCatalog
+import org.apache.flink.table.expressions.lookups.TableReferenceLookup
+import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils
+import org.apache.flink.table.plan.logical.{Minus => LMinus, _}
+import org.apache.flink.table.util.JavaScalaConversionUtil
+import org.apache.flink.table.util.JavaScalaConversionUtil.toScala
+import org.apache.flink.util.Preconditions
+
+import _root_.scala.collection.JavaConverters._
+
+/**
+ * Builder for [[[Operation]] tree.
+ */
+class OperationTreeBuilder(private val tableEnv: TableEnvironment) {
+
+ private val expressionBridge: ExpressionBridge[PlannerExpression] = tableEnv.expressionBridge
+ private val projectionOperationUtils = new ProjectionOperationUtils
+ private val functionCatalog: FunctionDefinitionCatalog = tableEnv.functionCatalog
+
+ private val noWindowPropertyChecker = new NoWindowPropertyChecker(
+ "Window start and end properties are not available for Over windows.")
+
+ private def bridgeExpression(expression: Expression): PlannerExpression = {
+ val expr = expressionBridge.bridge(expression)
+ if (!expr.valid) {
+ throw new ValidationException(s"Could not validate expression: $expression")
+ }
+ expr
+ }
+
+ private val tableCatalog = new TableReferenceLookup {
+ override def lookupTable(name: String): Optional[TableReferenceExpression] =
+ JavaScalaConversionUtil
+ .toJava(tableEnv.scanInternal(Array(name))
+ .map(op => new TableReferenceExpression(name, op.getTableOperation)))
+ }
+
+ def project(
+ projectList: JList[Expression],
+ child: TableOperation,
+ explicitAlias: Boolean = false)
+ : TableOperation = {
+
+ val childNode = child.asInstanceOf[LogicalNode]
+
+ projectInternal(projectList, childNode, explicitAlias, Collections.emptyList())
+ }
+
+ def project(
+ projectList: JList[Expression],
+ child: TableOperation,
+ overWindows: JList[OverWindow])
+ : TableOperation = {
+
+ Preconditions.checkArgument(!overWindows.isEmpty)
+
+ val childNode = child.asInstanceOf[LogicalNode]
+
+ projectList.asScala.map(_.accept(noWindowPropertyChecker))
+
+ projectInternal(projectList,
+ childNode,
+ explicitAlias = true,
+ overWindows)
+ }
+
+ private def projectInternal(
+ projectList: JList[Expression],
+ child: LogicalNode,
+ explicitAlias: Boolean,
+ overWindows: JList[OverWindow])
+ : LogicalNode = {
+
+ val resolver = resolverFor(tableCatalog, functionCatalog, child).withOverWindows(overWindows)
+ .build
+ val projections = resolver.resolve(projectList)
+ val namedProjections = projectionOperationUtils.nameExpressions(projections)
+
+ Project(namedProjections.asScala.map(bridgeExpression), child, explicitAlias)
+ .validate(tableEnv)
+ }
+
+ /**
+ * Adds additional columns. Existing fields will be replaced if replaceIfExist is true.
+ */
+ def addColumns(
+ replaceIfExist: Boolean,
+ fieldLists: JList[Expression],
+ child: TableOperation)
+ : TableOperation = {
+ val newColumns = if (replaceIfExist) {
+ val fieldNames = child.getTableSchema.getFieldNames.toList.asJava
+ ColumnOperationUtils.addOrReplaceColumns(fieldNames, fieldLists)
+ } else {
+ (new UnresolvedReferenceExpression("*") +: fieldLists.asScala).asJava
+ }
+ project(newColumns, child)
+ }
+
+ def renameColumns(
+ aliases: JList[Expression],
+ child: TableOperation)
+ : TableOperation = {
+
+ val inputFieldNames = child.getTableSchema.getFieldNames.toList.asJava
+ val validateAliases = ColumnOperationUtils.renameColumns(inputFieldNames, aliases)
+
+ project(validateAliases, child)
+ }
+
+ def dropColumns(
+ fieldLists: JList[Expression],
+ child: TableOperation)
+ : TableOperation = {
+
+ val inputFieldNames = child.getTableSchema.getFieldNames.toList.asJava
+ val finalFields = ColumnOperationUtils.dropFields(inputFieldNames, fieldLists)
+
+ project(finalFields, child)
+ }
+
+ def aggregate(
+ groupingExpressions: JList[Expression],
+ namedAggregates: JMap[Expression, String],
+ child: TableOperation)
+ : TableOperation = {
+
+ val childNode = child.asInstanceOf[LogicalNode]
+ val resolver = resolverFor(tableCatalog, functionCatalog, childNode).build
+
+ val convertedGroupings = resolveExpressions(groupingExpressions, resolver)
+ val convertedAggregates = resolveNamedExpressions(namedAggregates, resolver)
+
+ Aggregate(convertedGroupings, convertedAggregates, childNode).validate(tableEnv)
+ }
+
+ private def resolveExpressions(
+ expressions: JList[Expression],
+ resolver: ExpressionResolver)
+ : Seq[PlannerExpression] = {
+ resolver.resolve(expressions).asScala.map(bridgeExpression)
+ }
+
+ private def resolveNamedExpressions(
+ namedExpressions: JMap[Expression, String],
+ resolver: ExpressionResolver)
+ : Seq[PlannerExpression] = {
+
+ val renamedExpressions = namedExpressions.asScala
+ .map(expr => new CallExpression(AS, List(expr._1, valueLiteral(expr._2)).asJava)
+ .asInstanceOf[Expression])
+ .toList.asJava
+
+ resolver.resolve(renamedExpressions).asScala.map(bridgeExpression)
+ }
+
+ def windowAggregate(
+ groupingExpressions: JList[Expression],
+ window: GroupWindow,
+ namedProperties: JMap[Expression, String],
+ namedAggregates: JMap[Expression, String],
+ child: TableOperation)
+ : TableOperation = {
+
+ val childNode = child.asInstanceOf[LogicalNode]
+
+ val resolver = resolverFor(tableCatalog, functionCatalog, childNode).withGroupWindow(window)
+ .build
+
+ val convertedGroupings = resolveExpressions(groupingExpressions, resolver)
+
+ val convertedAggregates = resolveNamedExpressions(namedAggregates, resolver)
+
+ val convertedProperties = resolveNamedExpressions(namedProperties, resolver)
+
+ WindowAggregate(
+ convertedGroupings,
+ resolver.resolveGroupWindow(window),
+ convertedProperties,
+ convertedAggregates,
+ childNode)
+ .validate(tableEnv)
+ }
+
+ def join(
+ left: TableOperation,
+ right: TableOperation,
+ joinType: JoinType,
+ condition: Optional[Expression],
+ correlated: Boolean)
+ : TableOperation = {
+
+ val leftNode = left.asInstanceOf[LogicalNode]
+ val rightNode = right.asInstanceOf[LogicalNode]
+
+ val resolver = resolverFor(tableCatalog, functionCatalog, leftNode, rightNode).build()
+
+ val resolvedCondition = toScala(condition).map(c => resolver.resolve(List(c).asJava)) match {
+ case Some(resolvedExprs) if resolvedExprs.size != 1 =>
+ throw new ValidationException(s"Invalid join condition $condition")
+ case Some(resolvedExprs) =>
+ Some(resolvedExprs.get(0))
+ case None => None
+ }
+
+ val plannerExpression = resolvedCondition.map(bridgeExpression)
+
+ Join(leftNode, rightNode, joinType, plannerExpression, correlated).validate(tableEnv)
+ }
+
+ def joinLateral(
+ left: TableOperation,
+ tableFunction: Expression,
+ joinType: JoinType,
+ condition: Optional[Expression])
+ : TableOperation = {
+
+ val leftNode = left.asInstanceOf[LogicalNode]
+
+ val resolver = resolverFor(tableCatalog, functionCatalog, leftNode).build()
+ val resolvedFunction = resolveSingleExpression(tableFunction, resolver)
+
+ val temporalTable = UserDefinedFunctionUtils.createLogicalFunctionCall(
+ expressionBridge.bridge(resolvedFunction),
+ leftNode).validate(tableEnv)
+
+ join(left, temporalTable, joinType, condition, correlated = true)
+ }
+
+ def resolveExpression(expression: Expression, tableOperation: TableOperation*)
+ : PlannerExpression = {
+ val resolver = resolverFor(tableCatalog, functionCatalog, tableOperation: _*).build()
+
+ resolveSingleExpression(expression, resolver)
+ }
+
+ private def resolveSingleExpression(
+ expression: Expression,
+ resolver: ExpressionResolver)
+ : PlannerExpression = {
+ val resolvedTimeAttributes = resolver.resolve(List(expression).asJava)
+ if (resolvedTimeAttributes.size() != 1) {
+ throw new ValidationException("Expected single expression")
+ } else {
+ expressionBridge.bridge(resolvedTimeAttributes.get(0))
+ }
+ }
+
+ def sort(
+ 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)
+
+ Sort(order, childNode).validate(tableEnv)
+ }
+
+ def limitWithOffset(offset: Int, child: TableOperation): TableOperation = {
+ limit(offset, -1, child)
+ }
+
+ def limitWithFetch(fetch: Int, child: TableOperation): TableOperation = {
+ applyFetch(fetch, child)
+ }
+
+ private def applyFetch(fetch: Int, child: TableOperation): TableOperation = {
+ child match {
+ case Limit(o, -1, c) =>
+ // replace LIMIT without FETCH by LIMIT with FETCH
+ limit(o, fetch, c)
+ case Limit(_, _, _) =>
+ throw new ValidationException("FETCH is already defined.")
+ case _ =>
+ limit(0, fetch, child)
+ }
+ }
+
+ def limit(offset: Int, fetch: Int, child: TableOperation): TableOperation = {
+ Limit(offset, fetch, child.asInstanceOf[LogicalNode]).validate(tableEnv)
+ }
+
+ def alias(
+ fields: JList[Expression],
+ child: TableOperation)
+ : TableOperation = {
+
+ val childNode = child.asInstanceOf[LogicalNode]
+ val convertedFields = fields.asScala.map(expressionBridge.bridge)
+
+ AliasNode(convertedFields, childNode).validate(tableEnv)
+ }
+
+ def filter(
+ condition: Expression,
+ child: TableOperation)
+ : TableOperation = {
+
+ val childNode = child.asInstanceOf[LogicalNode]
+ val resolver = resolverFor(tableCatalog, functionCatalog, childNode).build()
+ val convertedFields = expressionBridge.bridge(resolveSingleExpression(condition, resolver))
+
+ Filter(convertedFields, childNode).validate(tableEnv)
+ }
+
+ def distinct(
+ child: TableOperation)
+ : TableOperation = {
+ Distinct(child.asInstanceOf[LogicalNode]).validate(tableEnv)
+ }
+
+ def minus(
+ left: TableOperation,
+ right: TableOperation,
+ all: Boolean)
+ : TableOperation = {
+ LMinus(left.asInstanceOf[LogicalNode], right.asInstanceOf[LogicalNode], all).validate(tableEnv)
+ }
+
+ def intersect(
+ left: TableOperation,
+ right: TableOperation,
+ all: Boolean)
+ : TableOperation = {
+ Intersect(left.asInstanceOf[LogicalNode], right.asInstanceOf[LogicalNode], all)
+ .validate(tableEnv)
+ }
+
+ def union(
+ left: TableOperation,
+ right: TableOperation,
+ all: Boolean)
+ : TableOperation = {
+ Union(left.asInstanceOf[LogicalNode], right.asInstanceOf[LogicalNode], all).validate(tableEnv)
+ }
+
+ def map(mapFunction: Expression, child: TableOperation): TableOperation = {
+
+ if (!isScalarFunction(mapFunction)) {
+ throw new ValidationException("Only ScalarFunction can be used in the map operator.")
+ }
+
+ val expandedFields = new CallExpression(BuiltInFunctionDefinitions.FLATTEN,
+ List(mapFunction).asJava)
+ project(Collections.singletonList(expandedFields), child)
+ }
+
+ private def isScalarFunction(mapFunction: Expression) = {
+ mapFunction.isInstanceOf[CallExpression] &&
+ mapFunction.asInstanceOf[CallExpression].getFunctionDefinition.getType == SCALAR_FUNCTION
+ }
+
+ class NoWindowPropertyChecker(val exceptionMessage: String)
+ extends ApiExpressionDefaultVisitor[Void] {
+ override def visitCall(call: CallExpression): Void = {
+ val functionDefinition = call.getFunctionDefinition
+ if (BuiltInFunctionDefinitions.WINDOW_PROPERTIES
+ .contains(functionDefinition)) {
+ throw new ValidationException(exceptionMessage)
+ }
+ call.getChildren.asScala.foreach(expr => expr.accept(this))
+ null
+ }
+
+ override protected def defaultMethod(expression: Expression): Void = null
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/OperationTreeBuilder.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/OperationTreeBuilder.scala
deleted file mode 100644
index 32ca8c3..0000000
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/OperationTreeBuilder.scala
+++ /dev/null
@@ -1,413 +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.plan
-
-import java.util.{Optional, List => JList, Map => JMap}
-
-import org.apache.flink.api.java.operators.join.JoinType
-import org.apache.flink.table.api._
-import org.apache.flink.table.expressions.{Alias, Asc, Expression, ExpressionBridge, Flattening, Ordering, PlannerExpression, UnresolvedAlias, UnresolvedFieldReference, WindowProperty}
-import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils
-import org.apache.flink.table.operations.TableOperation
-import org.apache.flink.table.plan.ProjectionTranslator.{expandProjectList, flattenExpression, resolveOverWindows}
-import org.apache.flink.table.plan.logical._
-import org.apache.flink.table.util.JavaScalaConversionUtil.toScala
-
-import _root_.scala.collection.JavaConverters._
-
-/**
- * Builder for [[[Operation]] tree.
- */
-class OperationTreeBuilder(private val tableEnv: TableEnvironment) {
-
- private val expressionBridge: ExpressionBridge[PlannerExpression] = tableEnv.expressionBridge
-
- def project(
- projectList: JList[Expression],
- child: TableOperation,
- explicitAlias: Boolean = false)
- : TableOperation = {
-
- val childNode = child.asInstanceOf[LogicalNode]
-
- val convertedProjectList = projectList.asScala
- .map(expressionBridge.bridge)
- .flatMap(expr => flattenExpression(expr, childNode, tableEnv))
- .map(UnresolvedAlias).toList
-
- Project(convertedProjectList, childNode, explicitAlias).validate(tableEnv)
- }
-
- /**
- * Adds additional columns. Existing fields will be replaced if replaceIfExist is true.
- */
- def addColumns(
- replaceIfExist: Boolean,
- fieldLists: Seq[Expression],
- child: TableOperation)
- : TableOperation = {
-
- val childNode = child.asInstanceOf[LogicalNode]
- val childFields = childNode.output
- .map(a => UnresolvedFieldReference(a.name).asInstanceOf[PlannerExpression])
-
- val addFields = fieldLists.map(expressionBridge.bridge)
-
- if (replaceIfExist) {
- val finalFields = childFields.toBuffer
-
- // replace field if exist.
- addFields.foreach {
- case e@Alias(_, name, _) =>
- val index = finalFields.indexWhere(
- p => p match {
- case u: UnresolvedFieldReference => u.name.equals(name)
- case a: Alias => a.name.equals(name)
- case _ => false
- })
- if (index >= 0) {
- finalFields(index) = e
- } else {
- finalFields.append(e)
- }
- case e =>
- finalFields.append(e)
- }
-
- val convertedProjectList = finalFields.map(UnresolvedAlias).toList
- Project(convertedProjectList, childNode).validate(tableEnv)
- } else {
- val convertedProjectList = (childFields ++ addFields)
- .flatMap(expr => flattenExpression(expr, childNode, tableEnv))
- .map(UnresolvedAlias)
- Project(convertedProjectList, childNode).validate(tableEnv)
- }
- }
-
- def renameColumns(
- fieldLists: Seq[Expression],
- child: TableOperation)
- : TableOperation = {
-
- val childNode = child.asInstanceOf[LogicalNode]
- val finalFields = childNode.output
- .map(a => UnresolvedFieldReference(a.name).asInstanceOf[PlannerExpression]).toArray
-
- val renameFields = fieldLists.map(tableEnv.expressionBridge.bridge)
-
- // Rename existing fields
- renameFields.foreach {
- case e@Alias(child: UnresolvedFieldReference, _, _) =>
- val index = finalFields.indexWhere(
- p => p match {
- case u: UnresolvedFieldReference => u.name.equals(child.name)
- case _ => false
- })
- if (index >= 0) {
- finalFields(index) = e
- } else {
- throw new TableException(s"Rename field [${child.name}] does not exist in source table.")
- }
- case e =>
- throw new TableException(
- s"Unexpected field expression type [$e]. " +
- s"Renaming must add an alias to the original field, e.g., a as a1.")
- }
-
- Project(finalFields.map(UnresolvedAlias), childNode).validate(tableEnv)
- }
-
- def dropColumns(
- fieldLists: Seq[Expression],
- child: TableOperation)
- : TableOperation = {
-
- val childNode = child.asInstanceOf[LogicalNode]
- val finalFields = childNode.output.map(a => UnresolvedFieldReference(a.name)).toBuffer
- val dropFields = fieldLists.map(tableEnv.expressionBridge.bridge).distinct
-
- // Remove the fields which should be deleted in the final list
- dropFields.foreach {
- case UnresolvedFieldReference(name) =>
- val index = finalFields.indexWhere(
- p => p match {
- case u: UnresolvedFieldReference => u.name.equalsIgnoreCase(name)
- case _ => false
- })
-
- if (index >= 0) {
- finalFields.remove(index)
- } else {
- throw new TableException(s"Drop field [$name] does not exist in source table.")
- }
- case e =>
- throw new TableException(s"Unexpected field expression type [$e].")
- }
- Project(finalFields.map(UnresolvedAlias), childNode).validate(tableEnv)
- }
-
- def aggregate(
- groupingExpressions: JList[Expression],
- namedAggregates: JMap[Expression, String],
- child: TableOperation)
- : TableOperation = {
-
- val childNode = child.asInstanceOf[LogicalNode]
-
- val convertedGroupings = groupingExpressions.asScala
- .map(expressionBridge.bridge)
-
- val convertedAggregates = namedAggregates.asScala
- .map(a => Alias(expressionBridge.bridge(a._1), a._2)).toSeq
-
- Aggregate(convertedGroupings, convertedAggregates, childNode).validate(tableEnv)
- }
-
- def windowAggregate(
- groupingExpressions: JList[Expression],
- window: GroupWindow,
- namedProperties: JMap[Expression, String],
- namedAggregates: JMap[Expression, String],
- child: TableOperation)
- : TableOperation = {
-
- val childNode = child.asInstanceOf[LogicalNode]
-
- val convertedGroupings = groupingExpressions.asScala
- .map(expressionBridge.bridge)
-
- val convertedAggregates = namedAggregates.asScala
- .map(a => Alias(expressionBridge.bridge(a._1), a._2)).toSeq
-
- val convertedProperties = namedProperties.asScala
- .map(a => Alias(expressionBridge.bridge(a._1), a._2)).toSeq
-
- WindowAggregate(
- convertedGroupings,
- createLogicalWindow(window),
- convertedProperties,
- convertedAggregates,
- childNode)
- .validate(tableEnv)
- }
-
- def project(
- projectList: JList[Expression],
- child: TableOperation,
- overWindows: JList[OverWindow])
- : TableOperation = {
-
- val childNode = child.asInstanceOf[LogicalNode]
-
- val expandedFields = expandProjectList(
- projectList.asScala.map(expressionBridge.bridge),
- childNode,
- tableEnv)
-
- if (expandedFields.exists(_.isInstanceOf[WindowProperty])){
- throw new ValidationException(
- "Window start and end properties are not available for Over windows.")
- }
-
- val expandedOverFields = resolveOverWindows(
- expandedFields,
- overWindows.asScala.map(createLogicalWindow))
- .map(UnresolvedAlias)
-
- Project(expandedOverFields, childNode, explicitAlias = true).validate(tableEnv)
- }
-
- def join(
- left: TableOperation,
- right: TableOperation,
- joinType: JoinType,
- condition: Optional[Expression],
- correlated: Boolean)
- : TableOperation = {
- Join(
- left.asInstanceOf[LogicalNode],
- right.asInstanceOf[LogicalNode],
- joinType,
- toScala(condition).map(expressionBridge.bridge),
- correlated).validate(tableEnv)
- }
-
- def joinLateral(
- left: TableOperation,
- tableFunction: Expression,
- joinType: JoinType,
- condition: Optional[Expression])
- : TableOperation = {
-
- val leftNode = left.asInstanceOf[LogicalNode]
-
- val temporalTable = UserDefinedFunctionUtils.createLogicalFunctionCall(
- expressionBridge.bridge(tableFunction),
- leftNode).validate(tableEnv)
-
- join(left, temporalTable, joinType, condition, correlated = true)
- }
-
- def createTemporalTable(
- timeAttribute: Expression,
- primaryKey: Expression,
- underlyingOperation: TableOperation)
- : TemporalTable = {
- val underlyingOperationNode = underlyingOperation.asInstanceOf[LogicalNode]
- TemporalTable(
- expressionBridge.bridge(timeAttribute),
- expressionBridge.bridge(primaryKey),
- underlyingOperationNode)
- .validate(tableEnv)
- .asInstanceOf[TemporalTable]
- }
-
- def sort(
- fields: JList[Expression],
- child: TableOperation)
- : TableOperation = {
- val childNode = child.asInstanceOf[LogicalNode]
-
- val order: Seq[Ordering] = fields.asScala.map(expressionBridge.bridge).map {
- case o: Ordering => o
- case e => Asc(e)
- }
-
- Sort(order, childNode).validate(tableEnv)
- }
-
- def limitWithOffset(offset: Int, child: TableOperation): TableOperation = {
- limit(offset, -1, child)
- }
-
- def limitWithFetch(fetch: Int, child: TableOperation): TableOperation = {
- applyFetch(fetch, child)
- }
-
- private def applyFetch(fetch: Int, child: TableOperation): TableOperation = {
- child match {
- case Limit(o, -1, c) =>
- // replace LIMIT without FETCH by LIMIT with FETCH
- limit(o, fetch, c)
- case Limit(_, _, _) =>
- throw new ValidationException("FETCH is already defined.")
- case _ =>
- limit(0, fetch, child)
- }
- }
-
- def limit(offset: Int, fetch: Int, child: TableOperation): TableOperation = {
- Limit(offset, fetch, child.asInstanceOf[LogicalNode]).validate(tableEnv)
- }
-
- def alias(
- fields: JList[Expression],
- child: TableOperation)
- : TableOperation = {
-
- val childNode = child.asInstanceOf[LogicalNode]
- val convertedFields = fields.asScala.map(expressionBridge.bridge)
-
- AliasNode(convertedFields, childNode).validate(tableEnv)
- }
-
- def filter(
- condition: Expression,
- child: TableOperation)
- : TableOperation = {
-
- val childNode = child.asInstanceOf[LogicalNode]
- val convertedFields = expressionBridge.bridge(condition)
-
- Filter(convertedFields, childNode).validate(tableEnv)
- }
-
- def distinct(
- child: TableOperation)
- : TableOperation = {
- Distinct(child.asInstanceOf[LogicalNode]).validate(tableEnv)
- }
-
- def minus(
- left: TableOperation,
- right: TableOperation,
- all: Boolean)
- : TableOperation = {
- Minus(left.asInstanceOf[LogicalNode], right.asInstanceOf[LogicalNode], all).validate(tableEnv)
- }
-
- def intersect(
- left: TableOperation,
- right: TableOperation,
- all: Boolean)
- : TableOperation = {
- Intersect(left.asInstanceOf[LogicalNode], right.asInstanceOf[LogicalNode], all)
- .validate(tableEnv)
- }
-
- def union(
- left: TableOperation,
- right: TableOperation,
- all: Boolean)
- : TableOperation = {
- Union(left.asInstanceOf[LogicalNode], right.asInstanceOf[LogicalNode], all).validate(tableEnv)
- }
-
- def map(mapFunction: Expression, child: TableOperation): TableOperation = {
- val childNode = child.asInstanceOf[LogicalNode]
- val expandedFields = expandProjectList(
- Seq(mapFunction).map(e => Flattening(expressionBridge.bridge(e))), childNode, tableEnv)
- Project(expandedFields.map(UnresolvedAlias), childNode).validate(tableEnv)
- }
-
- /**
- * Converts an API class to a logical window for planning.
- */
- private def createLogicalWindow(overWindow: OverWindow): LogicalOverWindow = {
- LogicalOverWindow(
- expressionBridge.bridge(overWindow.getAlias),
- overWindow.getPartitioning.asScala.map(expressionBridge.bridge),
- expressionBridge.bridge(overWindow.getOrder),
- expressionBridge.bridge(overWindow.getPreceding),
- toScala(overWindow.getFollowing).map(expressionBridge.bridge)
- )
- }
-
- /**
- * Converts an API class to a logical window for planning.
- */
- private def createLogicalWindow(window: GroupWindow): LogicalWindow = window match {
- case tw: TumbleWithSizeOnTimeWithAlias =>
- TumblingGroupWindow(
- expressionBridge.bridge(tw.getAlias),
- expressionBridge.bridge(tw.getTimeField),
- expressionBridge.bridge(tw.getSize))
- case sw: SlideWithSizeAndSlideOnTimeWithAlias =>
- SlidingGroupWindow(
- expressionBridge.bridge(sw.getAlias),
- expressionBridge.bridge(sw.getTimeField),
- expressionBridge.bridge(sw.getSize),
- expressionBridge.bridge(sw.getSlide))
- case sw: SessionWithGapOnTimeWithAlias =>
- SessionGroupWindow(
- expressionBridge.bridge(sw.getAlias),
- expressionBridge.bridge(sw.getTimeField),
- expressionBridge.bridge(sw.getGap))
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/ProjectionTranslator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/ProjectionTranslator.scala
deleted file mode 100644
index aacd7fd..0000000
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/ProjectionTranslator.scala
+++ /dev/null
@@ -1,155 +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.plan
-
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.table.api.TableEnvironment
-import org.apache.flink.table.expressions._
-import org.apache.flink.table.plan.logical.{LogicalNode, LogicalOverWindow, Project}
-import org.apache.flink.table.typeutils.RowIntervalTypeInfo
-
-object ProjectionTranslator {
-
- /**
- * Expands an UnresolvedFieldReference("*") to parent's full project list.
- */
- def expandProjectList(
- exprs: Seq[PlannerExpression],
- parent: LogicalNode,
- tableEnv: TableEnvironment)
- : Seq[PlannerExpression] = {
- exprs.flatMap(expr => flattenExpression(expr, parent, tableEnv))
- }
-
- def flattenExpression(
- expr: PlannerExpression,
- parent: LogicalNode,
- tableEnv: TableEnvironment)
- : Seq[PlannerExpression] = {
- expr match {
- case n: UnresolvedFieldReference if n.name == "*" =>
- parent.output.map(a => UnresolvedFieldReference(a.name))
-
- case Flattening(unresolved) =>
- // simulate a simple project to resolve fields using current parent
- val project = Project(Seq(UnresolvedAlias(unresolved)), parent).validate(tableEnv)
- val resolvedExpr = project
- .output
- .headOption
- .getOrElse(throw new RuntimeException("Could not find resolved composite."))
- resolvedExpr.validateInput()
- resolvedExpr.resultType match {
- case ct: CompositeType[_] =>
- (0 until ct.getArity).map { idx =>
- GetCompositeField(unresolved, ct.getFieldNames()(idx))
- }
- case _ =>
- Seq(unresolved)
- }
-
- case e: Expression => Seq(e)
- }
- }
-
- /**
- * Find and replace UnresolvedOverCall with OverCall
- *
- * @param exprs the expressions to check
- * @param overWindows windows to use to resolve expressions
- * @return an expression with correct resolved OverCall
- */
- def resolveOverWindows(
- exprs: Seq[PlannerExpression],
- overWindows: Seq[LogicalOverWindow])
- : Seq[PlannerExpression] = {
-
- exprs.map(e => replaceOverCall(e, overWindows))
- }
-
- private def replaceOverCall(
- expr: PlannerExpression,
- overWindows: Seq[LogicalOverWindow])
- : PlannerExpression = {
-
- expr match {
- case u: UnresolvedOverCall =>
- overWindows.find(_.alias.equals(u.alias)) match {
- case Some(overWindow) =>
- OverCall(
- u.agg,
- overWindow.partitionBy,
- overWindow.orderBy,
- overWindow.preceding,
- overWindow.following.getOrElse {
- // set following to CURRENT_ROW / CURRENT_RANGE if not defined
- if (overWindow.preceding.resultType.isInstanceOf[RowIntervalTypeInfo]) {
- CurrentRow()
- } else {
- CurrentRange()
- }
- })
-
- case None => u
- }
-
- case u: UnaryExpression =>
- val c = replaceOverCall(u.child, overWindows)
- u.makeCopy(Array(c))
-
- case b: BinaryExpression =>
- val l = replaceOverCall(b.left, overWindows)
- val r = replaceOverCall(b.right, overWindows)
- b.makeCopy(Array(l, r))
-
- // Scala functions
- case sfc @ PlannerScalarFunctionCall(clazz, args: Seq[PlannerExpression]) =>
- val newArgs: Seq[PlannerExpression] =
- args.map(
- (exp: PlannerExpression) =>
- replaceOverCall(exp, overWindows))
- sfc.makeCopy(Array(clazz, newArgs))
-
- // Array constructor
- case c @ ArrayConstructor(args) =>
- val newArgs =
- c.elements
- .map((exp: PlannerExpression) => replaceOverCall(exp, overWindows))
- c.makeCopy(Array(newArgs))
-
- // Other expressions
- case e: PlannerExpression => e
- }
- }
-
-
- /**
- * Extracts the leading non-alias expression.
- *
- * @param expr the expression to extract
- * @return the top non-alias expression
- */
- def getLeadingNonAliasExpr(expr: Expression): Expression = {
- expr match {
- case callExpr: CallExpression if callExpr.getFunctionDefinition.equals(
- BuiltInFunctionDefinitions.AS) =>
- getLeadingNonAliasExpr(callExpr.getChildren.get(0))
- case _ => expr
- }
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/LogicalNode.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/LogicalNode.scala
index 50518f6..a6fef54 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/LogicalNode.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/LogicalNode.scala
@@ -19,25 +19,16 @@ package org.apache.flink.table.plan.logical
import org.apache.calcite.rel.RelNode
import org.apache.calcite.tools.RelBuilder
-import org.apache.flink.table.plan.TreeNode
import org.apache.flink.table.api.{TableEnvironment, TableSchema, ValidationException}
import org.apache.flink.table.expressions._
import org.apache.flink.table.operations.TableOperation
-import org.apache.flink.table.typeutils.TypeCoercion
-import org.apache.flink.table.validate._
+import org.apache.flink.table.plan.TreeNode
/**
* LogicalNode is created and validated as we construct query plan using Table API.
*
* The main validation procedure is separated into two phases:
*
- * Expressions' resolution and transformation ([[resolveExpressions]]):
- *
- * - translate [[UnresolvedFieldReference]] into [[ResolvedFieldReference]]
- * using child operator's output
- * - generate alias names for query output
- * - ....
- *
* LogicalNode validation ([[validate]]):
*
* - check no [[UnresolvedFieldReference]] exists any more
@@ -54,108 +45,11 @@ abstract class LogicalNode extends TreeNode[LogicalNode] with TableOperation {
new TableSchema(attributes.map(_.name).toArray, attributes.map(_.resultType).toArray)
}
- def resolveExpressions(tableEnv: TableEnvironment): LogicalNode = {
- // resolve references and function calls
- val exprResolved = expressionPostOrderTransform {
- case u @ UnresolvedFieldReference(name) =>
- // try resolve a field
- resolveReference(tableEnv, name).getOrElse(u)
- }
-
- exprResolved.expressionPostOrderTransform {
- case ips: InputTypeSpec if ips.childrenValid =>
- var changed: Boolean = false
- val newChildren = ips.expectedTypes.zip(ips.children).map { case (tpe, child) =>
- val childType = child.resultType
- if (childType != tpe && TypeCoercion.canSafelyCast(childType, tpe)) {
- changed = true
- Cast(child, tpe)
- } else {
- child
- }
- }.toArray[AnyRef]
- if (changed) ips.makeCopy(newChildren) else ips
- }
- }
-
final def toRelNode(relBuilder: RelBuilder): RelNode = construct(relBuilder).build()
protected[logical] def construct(relBuilder: RelBuilder): RelBuilder
- def validate(tableEnv: TableEnvironment): LogicalNode = {
- val resolvedNode = resolveExpressions(tableEnv)
- resolvedNode.expressionPostOrderTransform {
- case a: Attribute if !a.valid =>
- val from = children.flatMap(_.output).map(_.name).mkString(", ")
- // give helpful error message for null literals
- if (a.name == "null") {
- failValidation(s"Cannot resolve field [${a.name}] given input [$from]. If you want to " +
- s"express a null literal, use 'nullOf(TYPE)' for typed null expressions. " +
- s"For example: nullOf(INT)")
- } else {
- failValidation(s"Cannot resolve field [${a.name}] given input [$from].")
- }
-
- case e: PlannerExpression if e.validateInput().isFailure =>
- failValidation(s"Expression $e failed on input check: " +
- s"${e.validateInput().asInstanceOf[ValidationFailure].message}")
- }
- }
-
- /**
- * Resolves the given strings to a [[NamedExpression]] using the input from all child
- * nodes of this LogicalPlan.
- */
- def resolveReference(tableEnv: TableEnvironment, name: String): Option[NamedExpression] = {
- // try to resolve a field
- val childrenOutput = children.flatMap(_.output)
- val fieldCandidates = childrenOutput.filter(_.name.equalsIgnoreCase(name))
- if (fieldCandidates.length > 1) {
- failValidation(s"Reference $name is ambiguous.")
- } else if (fieldCandidates.nonEmpty) {
- return Some(fieldCandidates.head.withName(name))
- }
-
- // try to resolve a table
- tableEnv.scanInternal(Array(name)) match {
- case Some(table) => Some(TableReference(name, table.getTableOperation))
- case None => None
- }
- }
-
- /**
- * Runs [[postOrderTransform]] with `rule` on all expressions present in this logical node.
- *
- * @param rule the rule to be applied to every expression in this logical node.
- */
- def expressionPostOrderTransform(
- rule: PartialFunction[PlannerExpression, PlannerExpression])
- : LogicalNode = {
- var changed = false
-
- def expressionPostOrderTransform(e: PlannerExpression): PlannerExpression = {
- val newExpr = e.postOrderTransform(rule)
- if (newExpr.fastEquals(e)) {
- e
- } else {
- changed = true
- newExpr
- }
- }
-
- val newArgs = productIterator.map {
- case e: PlannerExpression => expressionPostOrderTransform(e)
- case Some(e: PlannerExpression) => Some(expressionPostOrderTransform(e))
- case seq: Traversable[_] => seq.map {
- case e: PlannerExpression => expressionPostOrderTransform(e)
- case other => other
- }
- case r: Resolvable[_] => r.resolveExpressions(e => expressionPostOrderTransform(e))
- case other: AnyRef => other
- }.toArray
-
- if (changed) makeCopy(newArgs) else this
- }
+ def validate(tableEnv: TableEnvironment): LogicalNode = this
protected def failValidation(msg: String): Nothing = {
throw new ValidationException(msg)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/groupWindows.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/groupWindows.scala
index 6afffa0..d3ad5ad 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/groupWindows.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/groupWindows.scala
@@ -18,10 +18,13 @@
package org.apache.flink.table.plan.logical
+import java.util.Optional
+import java.util.{List => JList}
+
import org.apache.flink.table.api.{BatchTableEnvironment, StreamTableEnvironment, TableEnvironment}
import org.apache.flink.table.expressions.PlannerExpressionUtils.{isRowCountLiteral, isRowtimeAttribute, isTimeAttribute, isTimeIntervalLiteral}
import org.apache.flink.table.expressions._
-import org.apache.flink.table.typeutils.TypeCheckUtils.{isTimePoint, isLong}
+import org.apache.flink.table.typeutils.TypeCheckUtils.{isLong, isTimePoint}
import org.apache.flink.table.validate.{ValidationFailure, ValidationResult, ValidationSuccess}
// ------------------------------------------------------------------------------------------------
@@ -29,11 +32,11 @@ import org.apache.flink.table.validate.{ValidationFailure, ValidationResult, Val
// ------------------------------------------------------------------------------------------------
case class LogicalOverWindow(
- alias: PlannerExpression,
- partitionBy: Seq[PlannerExpression],
- orderBy: PlannerExpression,
- preceding: PlannerExpression,
- following: Option[PlannerExpression])
+ alias: Expression,
+ partitionBy: JList[Expression],
+ orderBy: Expression,
+ preceding: Expression,
+ following: Optional[Expression])
// ------------------------------------------------------------------------------------------------
// Tumbling group windows
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 2836c08..f12b596 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
@@ -43,34 +43,14 @@ import scala.collection.JavaConverters._
import scala.collection.mutable
case class Project(
- projectList: Seq[NamedExpression],
+ projectList: Seq[PlannerExpression],
child: LogicalNode,
explicitAlias: Boolean = false)
extends UnaryNode {
- override def output: Seq[Attribute] = projectList.map(_.toAttribute)
-
- override def resolveExpressions(tableEnv: TableEnvironment): LogicalNode = {
- val afterResolve = super.resolveExpressions(tableEnv).asInstanceOf[Project]
- val newProjectList =
- afterResolve.projectList.zipWithIndex.map { case (e, i) =>
- e match {
- case u @ UnresolvedAlias(c) => c match {
- case ne: NamedExpression => ne
- case expr if !expr.valid => u
- case c @ Cast(ne: NamedExpression, tp) => Alias(c, s"${ne.name}-$tp")
- case gcf: GetCompositeField => Alias(gcf, gcf.aliasName().getOrElse(s"_c$i"))
- case other => Alias(other, s"_c$i")
- }
- case _ =>
- throw new RuntimeException("This should never be called and probably points to a bug.")
- }
- }
- Project(newProjectList, child, explicitAlias)
- }
+ override def output: Seq[Attribute] = projectList.map(_.asInstanceOf[NamedExpression].toAttribute)
override def validate(tableEnv: TableEnvironment): LogicalNode = {
- val resolvedProject = super.validate(tableEnv).asInstanceOf[Project]
val names: mutable.Set[String] = mutable.Set()
def checkName(name: String): Unit = {
@@ -81,7 +61,7 @@ case class Project(
}
}
- resolvedProject.projectList.foreach {
+ projectList.foreach {
case n: Alias =>
// explicit name
checkName(n.name)
@@ -90,12 +70,13 @@ case class Project(
checkName(r.name)
case _ => // Do nothing
}
- resolvedProject
+ this
}
override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
child.construct(relBuilder)
+ val projectNames = projectList.map(_.asInstanceOf[NamedExpression].name).asJava
val exprs = if (explicitAlias) {
projectList
} else {
@@ -108,7 +89,7 @@ case class Project(
relBuilder.project(
exprs.map(_.toRexNode(relBuilder)).asJava,
- projectList.map(_.name).asJava,
+ projectNames,
true)
}
}
@@ -120,7 +101,7 @@ case class AliasNode(aliasList: Seq[PlannerExpression], child: LogicalNode) exte
override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder =
throw UnresolvedException("Invalid call to toRelNode on AliasNode")
- override def resolveExpressions(tableEnv: TableEnvironment): LogicalNode = {
+ override def validate(tableEnv: TableEnvironment): LogicalNode = {
if (aliasList.length > child.output.length) {
failValidation("Aliasing more fields than we actually have")
} else if (!aliasList.forall(_.isInstanceOf[UnresolvedFieldReference])) {
@@ -148,7 +129,7 @@ case class Distinct(child: LogicalNode) extends UnaryNode {
}
}
-case class Sort(order: Seq[Ordering], child: LogicalNode) extends UnaryNode {
+case class Sort(order: Seq[PlannerExpression], child: LogicalNode) extends UnaryNode {
override def output: Seq[Attribute] = child.output
override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
@@ -160,7 +141,7 @@ case class Sort(order: Seq[Ordering], child: LogicalNode) extends UnaryNode {
if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
failValidation(s"Sort on stream tables is currently not supported.")
}
- super.validate(tableEnv)
+ this
}
}
@@ -195,18 +176,17 @@ case class Filter(condition: PlannerExpression, child: LogicalNode) extends Unar
}
override def validate(tableEnv: TableEnvironment): LogicalNode = {
- val resolvedFilter = super.validate(tableEnv).asInstanceOf[Filter]
- if (resolvedFilter.condition.resultType != BOOLEAN_TYPE_INFO) {
+ if (condition.resultType != BOOLEAN_TYPE_INFO) {
failValidation(s"Filter operator requires a boolean expression as input," +
- s" but ${resolvedFilter.condition} is of type ${resolvedFilter.condition.resultType}")
+ s" but $condition is of type ${condition.resultType}")
}
- resolvedFilter
+ this
}
}
case class Aggregate(
groupingExpressions: Seq[PlannerExpression],
- aggregateExpressions: Seq[NamedExpression],
+ aggregateExpressions: Seq[PlannerExpression],
child: LogicalNode) extends UnaryNode {
override def output: Seq[Attribute] = {
@@ -228,9 +208,8 @@ case class Aggregate(
override def validate(tableEnv: TableEnvironment): LogicalNode = {
implicit val relBuilder: RelBuilder = tableEnv.getRelBuilder
- val resolvedAggregate = super.validate(tableEnv).asInstanceOf[Aggregate]
- val groupingExprs = resolvedAggregate.groupingExpressions
- val aggregateExprs = resolvedAggregate.aggregateExpressions
+ val groupingExprs = groupingExpressions
+ val aggregateExprs = aggregateExpressions
aggregateExprs.foreach(validateAggregateExpression)
groupingExprs.foreach(validateGroupingExpression)
@@ -273,7 +252,7 @@ case class Aggregate(
"because it's not a valid key type which must be hashable and comparable")
}
}
- resolvedAggregate
+ this
}
}
@@ -416,8 +395,7 @@ case class Join(
}
}
- override def resolveExpressions(tableEnv: TableEnvironment): LogicalNode = {
- val node = super.resolveExpressions(tableEnv).asInstanceOf[Join]
+ def resolveCondition(): Option[PlannerExpression] = {
val partialFunction: PartialFunction[PlannerExpression, PlannerExpression] = {
case field: ResolvedFieldReference => JoinFieldReference(
field.name,
@@ -425,8 +403,7 @@ case class Join(
left,
right)
}
- val resolvedCondition = node.condition.map(_.postOrderTransform(partialFunction))
- Join(node.left, node.right, node.joinType, resolvedCondition, correlated)
+ condition.map(_.postOrderTransform(partialFunction))
}
override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
@@ -455,16 +432,16 @@ case class Join(
left.output.map(_.name).toSet.intersect(right.output.map(_.name).toSet)
override def validate(tableEnv: TableEnvironment): LogicalNode = {
- val resolvedJoin = super.validate(tableEnv).asInstanceOf[Join]
- if (!resolvedJoin.condition.forall(_.resultType == BOOLEAN_TYPE_INFO)) {
+ if (!condition.forall(_.resultType == BOOLEAN_TYPE_INFO)) {
failValidation(s"Filter operator requires a boolean expression as input, " +
- s"but ${resolvedJoin.condition} is of type ${resolvedJoin.joinType}")
+ s"but $condition is of type $joinType")
} else if (ambiguousName.nonEmpty) {
failValidation(s"join relations with ambiguous names: ${ambiguousName.mkString(", ")}")
}
- resolvedJoin.condition.foreach(testJoinCondition)
- resolvedJoin
+ val resolvedCondition = resolveCondition()
+ resolvedCondition.foreach(testJoinCondition)
+ Join(left, right, joinType, resolvedCondition, correlated)
}
private def testJoinCondition(expression: PlannerExpression): Unit = {
@@ -555,8 +532,8 @@ case class LogicalRelNode(
case class WindowAggregate(
groupingExpressions: Seq[PlannerExpression],
window: LogicalWindow,
- propertyExpressions: Seq[NamedExpression],
- aggregateExpressions: Seq[NamedExpression],
+ propertyExpressions: Seq[PlannerExpression],
+ aggregateExpressions: Seq[PlannerExpression],
child: LogicalNode)
extends UnaryNode {
@@ -567,43 +544,6 @@ case class WindowAggregate(
}
}
- // resolve references of this operator's parameters
- override def resolveReference(
- tableEnv: TableEnvironment,
- name: String)
- : Option[NamedExpression] = {
-
- def resolveAlias(alias: String) = {
- // check if reference can already be resolved by input fields
- val found = super.resolveReference(tableEnv, name)
- if (found.isDefined) {
- failValidation(s"Reference $name is ambiguous.")
- } else {
- // resolve type of window reference
- val resolvedType = window.timeAttribute match {
- case UnresolvedFieldReference(n) =>
- super.resolveReference(tableEnv, n) match {
- case Some(ResolvedFieldReference(_, tpe)) => Some(tpe)
- case _ => None
- }
- case _ => None
- }
- // let validation phase throw an error if type could not be resolved
- Some(WindowReference(name, resolvedType))
- }
- }
-
- window.aliasAttribute match {
- // resolve reference to this window's name
- case UnresolvedFieldReference(alias) if name == alias =>
- resolveAlias(alias)
-
- case _ =>
- // resolve references as usual
- super.resolveReference(tableEnv, name)
- }
- }
-
override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
val flinkRelBuilder = relBuilder.asInstanceOf[FlinkRelBuilder]
child.construct(flinkRelBuilder)
@@ -622,9 +562,8 @@ case class WindowAggregate(
override def validate(tableEnv: TableEnvironment): LogicalNode = {
implicit val relBuilder: RelBuilder = tableEnv.getRelBuilder
- val resolvedWindowAggregate = super.validate(tableEnv).asInstanceOf[WindowAggregate]
- val groupingExprs = resolvedWindowAggregate.groupingExpressions
- val aggregateExprs = resolvedWindowAggregate.aggregateExpressions
+ val groupingExprs = groupingExpressions
+ val aggregateExprs = aggregateExpressions
aggregateExprs.foreach(validateAggregateExpression)
groupingExprs.foreach(validateGroupingExpression)
@@ -636,7 +575,7 @@ case class WindowAggregate(
case aggExpr: DistinctAgg =>
validateAggregateExpression(aggExpr.child)
// check no nested aggregation exists.
- case aggExpr: Aggregation =>
+ case aggExpr: Aggregation =>PlannerExpressionConverter
aggExpr.children.foreach { child =>
child.preOrderVisit {
case agg: Aggregation =>
@@ -663,7 +602,7 @@ case class WindowAggregate(
}
// validate window
- resolvedWindowAggregate.window.validate(tableEnv) match {
+ window.validate(tableEnv) match {
case ValidationFailure(msg) =>
failValidation(s"$window is invalid: $msg")
case ValidationSuccess => // ok
@@ -671,7 +610,7 @@ case class WindowAggregate(
// validate property
if (propertyExpressions.nonEmpty) {
- resolvedWindowAggregate.window match {
+ window match {
case TumblingGroupWindow(_, _, size) if isRowCountLiteral(size) =>
failValidation("Window start and Window end cannot be selected " +
"for a row-count Tumbling window.")
@@ -684,21 +623,7 @@ case class WindowAggregate(
}
}
- resolvedWindowAggregate
- }
-}
-
-case class TemporalTable(
- timeAttribute: PlannerExpression,
- primaryKey: PlannerExpression,
- child: LogicalNode)
- extends UnaryNode {
-
- override def output: Seq[Attribute] = child.output
-
- override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
- throw new UnsupportedOperationException(
- "This should never be called. This node is supposed to be used only for validation")
+ this
}
}
@@ -736,13 +661,12 @@ case class LogicalTableFunctionCall(
}
override def validate(tableEnv: TableEnvironment): LogicalNode = {
- val node = super.validate(tableEnv).asInstanceOf[LogicalTableFunctionCall]
// check if not Scala object
checkNotSingleton(tableFunction.getClass)
// check if class could be instantiated
checkForInstantiation(tableFunction.getClass)
// look for a signature that matches the input types
- val signature = node.parameters.map(_.resultType)
+ val signature = parameters.map(_.resultType)
val foundMethod = getUserDefinedMethod(tableFunction, "eval", typeInfoToClass(signature))
if (foundMethod.isEmpty) {
failValidation(
@@ -750,9 +674,9 @@ case class LogicalTableFunctionCall(
s"Actual: ${signatureToString(signature)} \n" +
s"Expected: ${signaturesToString(tableFunction, "eval")}")
} else {
- node.evalMethod = foundMethod.get
+ evalMethod = foundMethod.get
}
- node
+ this
}
override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/util/JavaScalaConversionUtil.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/util/JavaScalaConversionUtil.scala
index 6dc79ed..a7fd210 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/util/JavaScalaConversionUtil.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/util/JavaScalaConversionUtil.scala
@@ -22,6 +22,8 @@ import java.util.Optional
import java.util.function.{BiConsumer, Consumer, Function}
import org.apache.flink.api.java.tuple.{Tuple2 => JTuple2}
+import java.util.{List => JList}
+import scala.collection.JavaConverters._
/**
* Utilities for interoperability between Scala and Java classes.
@@ -58,4 +60,8 @@ object JavaScalaConversionUtil {
def toJava[T0, T1](tuple: (T0, T1)): JTuple2[T0, T1] = {
new JTuple2[T0, T1](tuple._1, tuple._2)
}
+
+ def toJava[T](seq: Seq[T]): JList[T] = {
+ seq.asJava
+ }
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentValidationTest.scala
index f1b7bcd..a6be848 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentValidationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentValidationTest.scala
@@ -22,7 +22,7 @@ import java.math.BigDecimal
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
-import org.apache.flink.table.api.TableException
+import org.apache.flink.table.api.{TableException, ValidationException}
import org.apache.flink.table.api.scala._
import org.apache.flink.table.runtime.stream.TimeAttributesITCase.TimestampWithEqualWatermark
import org.apache.flink.table.utils.TableTestBase
@@ -41,21 +41,21 @@ class StreamTableEnvironmentValidationTest extends TableTestBase {
util.addTable[(Long, Int, String, Int, Long)]('a.rowtime as 'b, 'b, 'c, 'd, 'e)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testInvalidRowtimeAttributesByPosition(): Unit = {
val util = streamTestUtil()
// table definition makes no sense
util.addTable[(Long, Int, String, Int, Long)]('a.rowtime.rowtime, 'b, 'c, 'd, 'e)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testInvalidProctimeAttributesByPosition(): Unit = {
val util = streamTestUtil()
// table definition makes no sense
util.addTable[(Long, Int, String, Int, Long)]('a.proctime.proctime, 'b, 'c, 'd, 'e)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testInvalidTimeAttributesByPosition(): Unit = {
val util = streamTestUtil()
// table definition makes no sense
@@ -160,7 +160,7 @@ class StreamTableEnvironmentValidationTest extends TableTestBase {
util.addTable[(Long, Int, String, Int, Long)]('x, '_1)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testInvalidAliasWithProctimeAttribute(): Unit = {
val util = streamTestUtil()
// alias in proctime not allowed
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/validation/CalcValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/validation/CalcValidationTest.scala
index ab2eed0..53ba73c 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/validation/CalcValidationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/validation/CalcValidationTest.scala
@@ -48,67 +48,67 @@ class CalcValidationTest extends TableTestBase {
.select('w.end.rowtime, 'int.count as 'int) // no rowtime on non-window reference
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testAddColumnsWithAgg(): Unit = {
val util = streamTestUtil()
val tab = util.addTable[(Int, Long, String)]("Table3",'a, 'b, 'c)
tab.addColumns('a.sum)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testAddOrReplaceColumnsWithAgg(): Unit = {
val util = streamTestUtil()
val tab = util.addTable[(Int, Long, String)]("Table3",'a, 'b, 'c)
tab.addOrReplaceColumns('a.sum)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testRenameColumnsWithAgg(): Unit = {
val util = streamTestUtil()
val tab = util.addTable[(Int, Long, String)]("Table3",'a, 'b, 'c)
tab.renameColumns('a.sum)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testRenameColumnsWithoutAlias(): Unit = {
val util = streamTestUtil()
val tab = util.addTable[(Int, Long, String)]("Table3",'a, 'b, 'c)
tab.renameColumns('a)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testRenameColumnsWithFunctallCall(): Unit = {
val util = streamTestUtil()
val tab = util.addTable[(Int, Long, String)]("Table3",'a, 'b, 'c)
tab.renameColumns('a + 1 as 'a2)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testRenameColumnsNotExist(): Unit = {
val util = streamTestUtil()
val tab = util.addTable[(Int, Long, String)]("Table3",'a, 'b, 'c)
tab.renameColumns('e as 'e2)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testDropColumnsWithAgg(): Unit = {
val util = streamTestUtil()
val tab = util.addTable[(Int, Long, String)]("Table3",'a, 'b, 'c)
tab.dropColumns('a.sum)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testDropColumnsNotExist(): Unit = {
val util = streamTestUtil()
val tab = util.addTable[(Int, Long, String)]("Table3",'a, 'b, 'c)
tab.dropColumns('e)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testDropColumnsWithValueLiteral(): Unit = {
val util = streamTestUtil()
val tab = util.addTable[(Int, Long, String)]("Table3",'a, 'b, 'c)
- tab.dropColumns("'last'")
+ tab.dropColumns("'a'")
}
@Test(expected = classOf[ValidationException])
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/validation/OverWindowValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/validation/OverWindowValidationTest.scala
index aa74dcc..0798e4e 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/validation/OverWindowValidationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/validation/OverWindowValidationTest.scala
@@ -148,6 +148,6 @@ class OverWindowValidationTest extends TableTestBase {
table
.window(Over orderBy 'rowtime preceding 1.minutes as 'w)
- .select('c, 'a.count over 'w, 'w.start, 'w.end)
+ .select('c, 'a.count over 'w, 'w.start + 1, 'w.end)
}
}