You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ji...@apache.org on 2019/04/19 17:52:47 UTC
[flink] branch master updated: [FLINK-12029][table] Add column
operations for TableApi This closes #8087.
This is an automated email from the ASF dual-hosted git repository.
jincheng 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 e41b6d4 [FLINK-12029][table] Add column operations for TableApi This closes #8087.
e41b6d4 is described below
commit e41b6d4784801677c60b75e353fcf7866106c287
Author: hequn8128 <ch...@gmail.com>
AuthorDate: Thu Apr 11 12:47:20 2019 +0800
[FLINK-12029][table] Add column operations for TableApi
This closes #8087.
---
docs/dev/table/functions.md | 282 ++++++++++++++++
.../expressions/BuiltInFunctionDefinitions.java | 8 +
.../table/expressions/ExpressionResolver.java | 41 ++-
.../rules/ExpandColumnFunctionsRule.java | 235 +++++++++++++
.../table/expressions/rules/ResolverRules.java | 5 +
.../table/operations/ColumnOperationUtils.java | 11 +-
.../flink/table/api/scala/expressionDsl.scala | 37 ++-
.../org/apache/flink/table/api/tableImpl.scala | 4 +-
.../expressions/PlannerExpressionParserImpl.scala | 15 +-
.../table/operations/OperationTreeBuilder.scala | 7 +-
.../api/batch/table/ColumnFunctionsTest.scala | 66 ++++
.../api/stream/table/ColumnFunctionsTest.scala | 367 +++++++++++++++++++++
.../validation/ColumnFunctionsValidationTest.scala | 97 ++++++
13 files changed, 1155 insertions(+), 20 deletions(-)
diff --git a/docs/dev/table/functions.md b/docs/dev/table/functions.md
index c00f8bd..3179838 100644
--- a/docs/dev/table/functions.md
+++ b/docs/dev/table/functions.md
@@ -5800,3 +5800,285 @@ For Table API, please use `_` for spaces (e.g., `DAY_TO_HOUR`).
| | `SQL_TSI_SECOND ` _(SQL-only)_ |
{% top %}
+
+Column Functions
+---------------------------------------
+
+The column functions are used to select or deselect table columns.
+
+| SYNTAX | DESC |
+| :--------------------- | :-------------------------- |
+| withColumns(...) | select the specified columns |
+| withoutColumns(...) | deselect the columns specified |
+
+The detailed syntax is as follows:
+
+{% highlight text %}
+columnFunction:
+ withColumns(columnExprs)
+ withoutColumns(columnExprs)
+
+columnExprs:
+ columnExpr [, columnExpr]*
+
+columnExpr:
+ columnRef | columnIndex to columnIndex | columnName to columnName
+
+columnRef:
+ columnName(The field name that exists in the table) | columnIndex(a positive integer starting from 1)
+{% endhighlight %}
+
+The usage of the column function is illustrated in the following table. (Suppose we have a table with 5 columns: `(a: Int, b: Long, c: String, d:String, e: String)`):
+
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Api</th>
+ <th class="text-center">Usage</th>
+ <th class="text-center">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>
+ withColumns(*)|*
+ </td>
+ <td>
+{% highlight java %}
+select("withColumns(*)") | select("*") = select("a, b, c, d, e")
+{% endhighlight %}
+ </td>
+ <td>
+ all the columns
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withColumns(m to n)
+ </td>
+ <td>
+{% highlight java %}
+select("withColumns(2 to 4)") = select("b, c, d")
+{% endhighlight %}
+ </td>
+ <td>
+ columns from m to n
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withColumns(m, n, k)
+ </td>
+ <td>
+{% highlight java %}
+select("withColumns(1, 3, e)") = select("a, c, e")
+{% endhighlight %}
+ </td>
+ <td>
+ columns m, n, k
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withColumns(m, n to k)
+ </td>
+ <td>
+{% highlight java %}
+select("withColumns(1, 3 to 5)") = select("a, c, d ,e")
+{% endhighlight %}
+ </td>
+ <td>
+ mixing of the above two representation
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withoutColumns(m to n)
+ </td>
+ <td>
+{% highlight java %}
+select("withoutColumns(2 to 4)") = select("a, e")
+{% endhighlight %}
+ </td>
+ <td>
+ deselect columns from m to n
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withoutColumns(m, n, k)
+ </td>
+ <td>
+{% highlight java %}
+select("withoutColumns(1, 3, 5)") = select("b, d")
+{% endhighlight %}
+ </td>
+ <td>
+ deselect columns m, n, k
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withoutColumns(m, n to k)
+ </td>
+ <td>
+{% highlight java %}
+select("withoutColumns(1, 3 to 5)") = select("b")
+{% endhighlight %}
+ </td>
+ <td>
+ mixing of the above two representation
+ </td>
+ </tr>
+
+ </tbody>
+</table>
+</div>
+
+<div data-lang="scala" markdown="1">
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Api</th>
+ <th class="text-center">Usage</th>
+ <th class="text-center">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>
+ withColumns(*)|*
+ </td>
+ <td>
+{% highlight scala %}
+select(withColumns('*)) | select('*) = select('a, 'b, 'c, 'd, 'e)
+{% endhighlight %}
+ </td>
+ <td>
+ all the columns
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withColumns(m to n)
+ </td>
+ <td>
+{% highlight scala %}
+select(withColumns(2 to 4)) = select('b, 'c, 'd)
+{% endhighlight %}
+ </td>
+ <td>
+ columns from m to n
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withColumns(m, n, k)
+ </td>
+ <td>
+{% highlight scala %}
+select(withColumns(1, 3, 'e)) = select('a, 'c, 'e)
+{% endhighlight %}
+ </td>
+ <td>
+ columns m, n, k
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withColumns(m, n to k)
+ </td>
+ <td>
+{% highlight scala %}
+select(withColumns(1, 3 to 5)) = select('a, 'c, 'd, 'e)
+{% endhighlight %}
+ </td>
+ <td>
+ mixing of the above two representation
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withoutColumns(m to n)
+ </td>
+ <td>
+{% highlight scala %}
+select(withoutColumns(2 to 4)) = select('a, 'e)
+{% endhighlight %}
+ </td>
+ <td>
+ deselect columns from m to n
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withoutColumns(m, n, k)
+ </td>
+ <td>
+{% highlight scala %}
+select(withoutColumns(1, 3, 5)) = select('b, 'd)
+{% endhighlight %}
+ </td>
+ <td>
+ deselect columns m, n, k
+ </td>
+ </tr>
+
+ <tr>
+ <td>
+ withoutColumns(m, n to k)
+ </td>
+ <td>
+{% highlight scala %}
+select(withoutColumns(1, 3 to 5)) = select('b)
+{% endhighlight %}
+ </td>
+ <td>
+ mixing of the above two representation
+ </td>
+ </tr>
+
+ </tbody>
+</table>
+</div>
+</div>
+
+The column functions can be used in all places where column fields are expected, such as `select, groupBy, orderBy, UDFs etc.` e.g.:
+
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+table
+ .groupBy("withColumns(1 to 3)")
+ .select("withColumns(a to b), myUDAgg(myUDF(withColumns(5 to 20)))")
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+table
+ .groupBy(withColumns(1 to 3))
+ .select(withColumns('a to 'b), myUDAgg(myUDF(withColumns(5 to 20))))
+{% endhighlight %}
+</div>
+</div>
+
+<span class="label label-info">Note</span> Column functions are only used in Table API.
+
+{% top %}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/BuiltInFunctionDefinitions.java
index 801fad2..838a7b6 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/BuiltInFunctionDefinitions.java
@@ -319,6 +319,12 @@ public final class BuiltInFunctionDefinitions {
public static final FunctionDefinition CURRENT_ROW =
new FunctionDefinition("currentRow", OTHER_FUNCTION);
+ // columns
+ public static final FunctionDefinition WITH_COLUMNS =
+ new FunctionDefinition("withColumns", OTHER_FUNCTION);
+ public static final FunctionDefinition WITHOUT_COLUMNS =
+ new FunctionDefinition("withoutColumns", OTHER_FUNCTION);
+
// etc
public static final FunctionDefinition IN =
new FunctionDefinition("in", SCALAR_FUNCTION);
@@ -328,6 +334,8 @@ public final class BuiltInFunctionDefinitions {
new FunctionDefinition("as", OTHER_FUNCTION);
public static final FunctionDefinition STREAM_RECORD_TIMESTAMP =
new FunctionDefinition("streamRecordTimestamp", OTHER_FUNCTION);
+ public static final FunctionDefinition RANGE_TO =
+ new FunctionDefinition("rangeTo", OTHER_FUNCTION);
public static final Set<FunctionDefinition> WINDOW_PROPERTIES = new HashSet<>(Arrays.asList(
WINDOW_START, WINDOW_END, PROCTIME, ROWTIME
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
index 7b0958d..afa9360 100644
--- 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
@@ -54,15 +54,13 @@ 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>flatten '*' and column functions 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>
@@ -79,6 +77,7 @@ public class ExpressionResolver {
return Arrays.asList(
ResolverRules.LOOKUP_CALL_BY_NAME,
ResolverRules.FLATTEN_STAR_REFERENCE,
+ ResolverRules.EXPAND_COLUMN_FUNCTIONS,
ResolverRules.OVER_WINDOWS,
ResolverRules.FIELD_RESOLVE,
ResolverRules.FLATTEN_CALL,
@@ -160,7 +159,12 @@ public class ExpressionResolver {
}
final String windowName = ((UnresolvedReferenceExpression) alias).getName();
- PlannerExpression timeField = resolveFieldsInSingleExpression(window.getTimeField()).accept(bridgeConverter);
+ List<Expression> resolvedTimeFieldExpression =
+ prepareExpressions(Collections.singletonList(window.getTimeField()));
+ if (resolvedTimeFieldExpression.size() != 1) {
+ throw new ValidationException("Group Window only supports a single time field column.");
+ }
+ PlannerExpression timeField = resolvedTimeFieldExpression.get(0).accept(bridgeConverter);
//TODO replace with LocalReferenceExpression
WindowReference resolvedAlias = new WindowReference(windowName, new Some<>(timeField.resultType()));
@@ -202,9 +206,10 @@ public class ExpressionResolver {
private void prepareLocalReferencesFromGroupWindows(@Nullable GroupWindow groupWindow) {
if (groupWindow != null) {
String windowName = ((UnresolvedReferenceExpression) groupWindow.getAlias()).getName();
- TypeInformation<?> windowType = resolveFieldsInSingleExpression(groupWindow.getTimeField())
- .accept(bridgeConverter)
- .resultType();
+ TypeInformation<?> windowType =
+ prepareExpressions(Collections.singletonList(groupWindow.getTimeField())).get(0)
+ .accept(bridgeConverter)
+ .resultType();
localReferences.put(windowName, new LocalReferenceExpression(windowName, windowType));
}
@@ -219,6 +224,14 @@ public class ExpressionResolver {
));
}
+ private List<Expression> prepareExpressions(List<Expression> expressions) {
+ return expressions.stream()
+ .flatMap(e -> lookupCall(e).stream())
+ .flatMap(e -> resolveColumnFunctions(e).stream())
+ .map(this::resolveFieldsInSingleExpression)
+ .collect(Collectors.toList());
+ }
+
private Expression resolveFieldsInSingleExpression(Expression expression) {
List<Expression> expressions = ResolverRules.FIELD_RESOLVE.apply(Collections.singletonList(expression),
new ExpressionResolverContext());
@@ -230,6 +243,18 @@ public class ExpressionResolver {
return expressions.get(0);
}
+ private List<Expression> resolveColumnFunctions(Expression expression) {
+ List<Expression> expressions = ResolverRules.EXPAND_COLUMN_FUNCTIONS.apply(Collections.singletonList(expression),
+ new ExpressionResolverContext());
+ return expressions;
+ }
+
+ private List<Expression> lookupCall(Expression expression) {
+ List<Expression> expressions = ResolverRules.LOOKUP_CALL_BY_NAME.apply(Collections.singletonList(expression),
+ new ExpressionResolverContext());
+ return expressions;
+ }
+
private class ExpressionResolverContext implements ResolverRule.ResolutionContext {
@Override
@@ -266,7 +291,7 @@ public class ExpressionResolver {
private LogicalOverWindow resolveOverWindow(OverWindow overWindow) {
return new LogicalOverWindow(
overWindow.getAlias(),
- overWindow.getPartitioning().stream().map(this::resolveFieldsInSingleExpression).collect(toList()),
+ prepareExpressions(overWindow.getPartitioning()),
resolveFieldsInSingleExpression(overWindow.getOrder()),
resolveFieldsInSingleExpression(overWindow.getPreceding()),
overWindow.getFollowing().map(this::resolveFieldsInSingleExpression)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ExpandColumnFunctionsRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ExpandColumnFunctionsRule.java
new file mode 100644
index 0000000..a832102
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/rules/ExpandColumnFunctionsRule.java
@@ -0,0 +1,235 @@
+/*
+ * 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.IntegerTypeInfo;
+import org.apache.flink.table.api.TableException;
+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 org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.util.Preconditions;
+
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.AS;
+import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.RANGE_TO;
+import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.WITHOUT_COLUMNS;
+import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.WITH_COLUMNS;
+
+/**
+ * Replaces column functions with all available {@link org.apache.flink.table.expressions.UnresolvedReferenceExpression}s
+ * from underlying inputs.
+ */
+@Internal
+final class ExpandColumnFunctionsRule implements ResolverRule {
+ @Override
+ public List<Expression> apply(List<Expression> expression, ResolutionContext context) {
+ ColumnFunctionsExpander columnFunctionsExpander =
+ new ColumnFunctionsExpander(
+ context.referenceLookup().getAllInputFields().stream()
+ .map(p -> new UnresolvedReferenceExpression(p.getName()))
+ .collect(Collectors.toList())
+ );
+ return expression.stream()
+ .flatMap(expr -> expr.accept(columnFunctionsExpander).stream())
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Expands column functions to it's real parent's input references.
+ */
+ class ColumnFunctionsExpander extends ApiExpressionDefaultVisitor<List<Expression>> {
+
+ private final List<UnresolvedReferenceExpression> inputFieldReferences;
+ private final ColumnsExpressionExpander columnsExpressionExpander;
+
+ public ColumnFunctionsExpander(List<UnresolvedReferenceExpression> inputFieldReferences) {
+ this.inputFieldReferences = inputFieldReferences;
+ this.columnsExpressionExpander = new ColumnsExpressionExpander(inputFieldReferences);
+ }
+
+ @Override
+ public List<Expression> visitCall(CallExpression call) {
+
+ List<Expression> result;
+
+ String definitionName = call.getFunctionDefinition().getName();
+ if (definitionName.equals(WITH_COLUMNS.getName())) {
+ result = resolveArgsOfColumns(call.getChildren(), false);
+ } else if (definitionName.equals(WITHOUT_COLUMNS.getName())) {
+ result = resolveArgsOfColumns(call.getChildren(), true);
+ } else {
+ List<Expression> args = call.getChildren()
+ .stream()
+ .flatMap(c -> c.accept(this).stream())
+ .collect(Collectors.toList());
+ result = Collections.singletonList(new CallExpression(call.getFunctionDefinition(), args));
+
+ // validate as.
+ if (definitionName.equals(AS.getName())) {
+ for (int i = 1; i < args.size(); ++i) {
+ if (!(args.get(i) instanceof ValueLiteralExpression)) {
+ String errorMessage = String.join(
+ ", ",
+ args.stream().map(e -> e.toString()).collect(Collectors.toList()));
+ throw new ValidationException(String.format("Invalid AS, parameters are: [%s].", errorMessage));
+ }
+ }
+ }
+ }
+
+ return result;
+ }
+
+ @Override
+ protected List<Expression> defaultMethod(Expression expression) {
+ return Collections.singletonList(expression);
+ }
+
+ /**
+ * Expand the columns expression in the input Expression List.
+ */
+ private List<Expression> resolveArgsOfColumns(List<Expression> args, boolean isReverseProjection) {
+
+ List<Expression> finalResult = new LinkedList<>();
+ List<UnresolvedReferenceExpression> result = args.stream()
+ .flatMap(e -> e.accept(this.columnsExpressionExpander).stream())
+ .collect(Collectors.toList());
+
+ if (isReverseProjection) {
+ for (UnresolvedReferenceExpression field: inputFieldReferences) {
+ if (indexOfName(result, field.getName()) == -1) {
+ finalResult.add(field);
+ }
+ }
+ } else {
+ finalResult.addAll(result);
+ }
+
+ return finalResult;
+ }
+ }
+
+ /**
+ * Expands a single column expression to it's real parent's input references.
+ */
+ class ColumnsExpressionExpander extends ApiExpressionDefaultVisitor<List<UnresolvedReferenceExpression>> {
+
+ private final List<UnresolvedReferenceExpression> inputFieldReferences;
+ public ColumnsExpressionExpander(List<UnresolvedReferenceExpression> inputFieldReferences) {
+ this.inputFieldReferences = inputFieldReferences;
+ }
+
+ @Override
+ public List<UnresolvedReferenceExpression> visitValueLiteral(ValueLiteralExpression valueLiteralExpression) {
+ if (valueLiteralExpression.getType() instanceof IntegerTypeInfo) {
+ return Collections.singletonList(inputFieldReferences.get((int) valueLiteralExpression.getValue() - 1));
+ } else {
+ return defaultMethod(valueLiteralExpression);
+ }
+ }
+
+ @Override
+ public List<UnresolvedReferenceExpression> visitUnresolvedReference(
+ UnresolvedReferenceExpression unresolvedReference) {
+ if (unresolvedReference.getName().equals("*")) {
+ return inputFieldReferences;
+ } else {
+ return Collections.singletonList(unresolvedReference);
+ }
+ }
+
+ @Override
+ public List<UnresolvedReferenceExpression> visitCall(CallExpression call) {
+ if (isIndexRangeCall(call)) {
+ int start = (int) ((ValueLiteralExpression) call.getChildren().get(0)).getValue();
+ int end = (int) ((ValueLiteralExpression) call.getChildren().get(1)).getValue();
+ Preconditions.checkArgument(
+ start <= end,
+ String.format("The start:%s of %s() or %s() should not bigger than end:%s.",
+ start, WITH_COLUMNS.getName(), WITHOUT_COLUMNS.getName(), end));
+
+ return inputFieldReferences.subList(start - 1, end);
+
+ } else if (isNameRangeCall(call)) {
+ String startName = ((UnresolvedReferenceExpression) call.getChildren().get(0)).getName();
+ String endName = ((UnresolvedReferenceExpression) call.getChildren().get(1)).getName();
+
+ int start = indexOfName(inputFieldReferences, startName);
+ int end = indexOfName(inputFieldReferences, endName);
+ Preconditions.checkArgument(
+ start <= end,
+ String.format("The start name:%s of %s() or %s() should not behind the end:%s.",
+ startName, WITH_COLUMNS.getName(), WITHOUT_COLUMNS.getName(), endName));
+
+ return inputFieldReferences.subList(start, end + 1);
+ } else {
+ return defaultMethod(call);
+ }
+ }
+
+ @Override
+ protected List<UnresolvedReferenceExpression> defaultMethod(Expression expression) {
+ throw new TableException(
+ String.format("The parameters of %s() or %s() only accept column name or column " +
+ "index, but receive %s.",
+ WITH_COLUMNS.getName(),
+ WITHOUT_COLUMNS.getName(),
+ expression.getClass().getSimpleName()));
+ }
+
+ /**
+ * Whether the expression is a column index range expression, e.g. withColumns(1 ~ 2).
+ */
+ private boolean isIndexRangeCall(CallExpression expression) {
+ return expression.getFunctionDefinition().getName().equals(RANGE_TO.getName()) &&
+ expression.getChildren().get(0) instanceof ValueLiteralExpression &&
+ expression.getChildren().get(1) instanceof ValueLiteralExpression;
+ }
+
+ /**
+ * Whether the expression is a column name range expression, e.g. withColumns(a ~ b).
+ */
+ private boolean isNameRangeCall(CallExpression expression) {
+ return expression.getFunctionDefinition().getName().equals(RANGE_TO.getName()) &&
+ expression.getChildren().get(0) instanceof UnresolvedReferenceExpression &&
+ expression.getChildren().get(1) instanceof UnresolvedReferenceExpression;
+ }
+ }
+
+ /**
+ * Find the index of targetName in the list. Return -1 if not found.
+ */
+ private static int indexOfName(List<UnresolvedReferenceExpression> inputFieldReferences, String targetName) {
+ int i;
+ for (i = 0; i < inputFieldReferences.size(); ++i) {
+ if (inputFieldReferences.get(i).getName().equals(targetName)) {
+ break;
+ }
+ }
+ return i == inputFieldReferences.size() ? -1 : i;
+ }
+}
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
index ee874ac..c8e3daf 100644
--- 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
@@ -57,6 +57,11 @@ public final class ResolverRules {
*/
public static final ResolverRule FLATTEN_STAR_REFERENCE = new StarReferenceFlatteningRule();
+ /**
+ * Resolves column functions to corresponding fields of inputs. See {@link ExpandColumnFunctionsRule} for details.
+ */
+ public static final ResolverRule EXPAND_COLUMN_FUNCTIONS = new ExpandColumnFunctionsRule();
+
private ResolverRules() {
}
}
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
index baa944a..40932e9 100644
--- 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
@@ -23,6 +23,7 @@ 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.FieldReferenceExpression;
import org.apache.flink.table.expressions.UnresolvedReferenceExpression;
import java.util.ArrayList;
@@ -116,8 +117,8 @@ public final class ColumnOperationUtils {
private static class DropColumnsExtractor extends ApiExpressionDefaultVisitor<String> {
@Override
- public String visitUnresolvedReference(UnresolvedReferenceExpression unresolvedReference) {
- return unresolvedReference.getName();
+ public String visitFieldReference(FieldReferenceExpression fieldReference) {
+ return fieldReference.getName();
}
@Override
@@ -130,10 +131,10 @@ public final class ColumnOperationUtils {
@Override
public String visitCall(CallExpression call) {
if (call.getFunctionDefinition() == AS &&
- call.getChildren().get(0) instanceof UnresolvedReferenceExpression) {
- UnresolvedReferenceExpression unresolvedReference = (UnresolvedReferenceExpression) call.getChildren()
+ call.getChildren().get(0) instanceof FieldReferenceExpression) {
+ FieldReferenceExpression resolvedFieldReference = (FieldReferenceExpression) call.getChildren()
.get(0);
- return unresolvedReference.getName();
+ return resolvedFieldReference.getName();
} else {
return defaultMethod(call);
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala
index 61fde56..333a8db 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala
@@ -21,10 +21,10 @@ import java.lang.{Boolean => JBoolean, Byte => JByte, Double => JDouble, Float =
import java.math.{BigDecimal => JBigDecimal}
import java.sql.{Date, Time, Timestamp}
-import org.apache.flink.api.common.typeinfo.{SqlTimeTypeInfo, TypeInformation}
+import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, SqlTimeTypeInfo, TypeInformation}
import org.apache.flink.table.api.{Over, Table, ValidationException}
import org.apache.flink.table.expressions.ApiExpressionUtils._
-import org.apache.flink.table.expressions.BuiltInFunctionDefinitions.{E => FDE, UUID => FDUUID, _}
+import org.apache.flink.table.expressions.BuiltInFunctionDefinitions.{WITH_COLUMNS, RANGE_TO, E => FDE, UUID => FDUUID, _}
import org.apache.flink.table.expressions._
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils.{getAccumulatorTypeOfAggregateFunction, getResultTypeOfAggregateFunction}
import org.apache.flink.table.functions.{AggregateFunction, ScalarFunction, TableFunction}
@@ -185,6 +185,12 @@ trait ImplicitExpressionOperations {
def % (other: Expression): Expression = mod(other)
/**
+ * Indicates the range from left to right, i.e. [left, right], which can be used in columns
+ * selection, e.g.: withColumns(1 to 3).
+ */
+ def to (other: Expression): Expression = call(RANGE_TO, expr, other)
+
+ /**
* Similar to a SQL distinct aggregation clause such as COUNT(DISTINCT a), declares that an
* aggregation function is only applied on distinct input values.
*
@@ -1126,6 +1132,12 @@ trait ImplicitExpressionConversions {
implicit def symbol2FieldExpression(sym: Symbol): Expression =
unresolvedRef(sym.name)
+ implicit def scalaRange2RangeExpression(range: Range.Inclusive): Expression = {
+ val startExpression = new ValueLiteralExpression(range.start, BasicTypeInfo.INT_TYPE_INFO)
+ val endExpression = new ValueLiteralExpression(range.end, BasicTypeInfo.INT_TYPE_INFO)
+ startExpression to endExpression
+ }
+
implicit def byte2Literal(b: Byte): Expression = valueLiteral(b)
implicit def short2Literal(s: Short): Expression = valueLiteral(s)
@@ -1592,4 +1604,25 @@ object ifThenElse {
}
}
+/**
+ * Creates a withColumns expression.
+ */
+object withColumns {
+
+ def apply(head: Expression, tail: Expression*): Expression = {
+ call(WITH_COLUMNS, head +: tail: _*)
+ }
+}
+
+/**
+ * Creates a withoutColumns expression.
+ */
+object withoutColumns {
+
+ def apply(head: Expression, tail: Expression*): Expression = {
+ call(WITHOUT_COLUMNS, head +: tail: _*)
+ }
+}
+
+
// scalastyle:on object.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 4f97bc6..94fb266 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
@@ -456,7 +456,7 @@ class TableImpl(
*/
class GroupedTableImpl(
private[flink] val table: Table,
- private[flink] val groupKey: Seq[Expression])
+ private[flink] val groupKeys: Seq[Expression])
extends GroupedTable {
private val tableImpl = table.asInstanceOf[TableImpl]
@@ -477,7 +477,7 @@ class GroupedTableImpl(
new TableImpl(tableImpl.tableEnv,
tableImpl.operationTreeBuilder.project(extracted.getProjections,
tableImpl.operationTreeBuilder.aggregate(
- groupKey.asJava,
+ groupKeys.asJava,
extracted.getAggregations,
tableImpl.operationTree
)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionParserImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionParserImpl.scala
index 9d587cd..3850c3f 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionParserImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionParserImpl.scala
@@ -133,6 +133,7 @@ object PlannerExpressionParserImpl extends JavaTokenParsers
lazy val TRIM_MODE_LEADING: Keyword = Keyword("LEADING")
lazy val TRIM_MODE_TRAILING: Keyword = Keyword("TRAILING")
lazy val TRIM_MODE_BOTH: Keyword = Keyword("BOTH")
+ lazy val TO: Keyword = Keyword("TO")
def functionIdent: PlannerExpressionParserImpl.Parser[String] = super.ident
@@ -641,7 +642,19 @@ object PlannerExpressionParserImpl extends JavaTokenParsers
case e ~ _ ~ name => call(BuiltInFunctionDefinitions.AS, e, valueLiteral(name.getName))
}
- lazy val expression: PackratParser[Expression] = overConstant | alias |
+ // columns
+
+ lazy val fieldNameRange: PackratParser[Expression] = fieldReference ~ TO ~ fieldReference ^^ {
+ case start ~ _ ~ end => call(BuiltInFunctionDefinitions.RANGE_TO, start, end)
+ }
+
+ lazy val fieldIndexRange: PackratParser[Expression] = numberLiteral ~ TO ~ numberLiteral ^^ {
+ case start ~ _ ~ end => call(BuiltInFunctionDefinitions.RANGE_TO, start, end)
+ }
+
+ lazy val range = fieldNameRange | fieldIndexRange
+
+ lazy val expression: PackratParser[Expression] = range | overConstant | alias |
failure("Invalid expression.")
lazy val expressionList: Parser[List[Expression]] = rep1sep(expression, ",")
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala
index c6fec01..ca70598 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala
@@ -129,8 +129,10 @@ class OperationTreeBuilder(private val tableEnv: TableEnvironment) {
child: TableOperation)
: TableOperation = {
+ val resolver = resolverFor(tableCatalog, functionCatalog, child).build()
val inputFieldNames = child.getTableSchema.getFieldNames.toList.asJava
- val validateAliases = ColumnOperationUtils.renameColumns(inputFieldNames, aliases)
+ val validateAliases =
+ ColumnOperationUtils.renameColumns(inputFieldNames, resolver.resolve(aliases))
project(validateAliases, child)
}
@@ -140,8 +142,9 @@ class OperationTreeBuilder(private val tableEnv: TableEnvironment) {
child: TableOperation)
: TableOperation = {
+ val resolver = resolverFor(tableCatalog, functionCatalog, child).build()
val inputFieldNames = child.getTableSchema.getFieldNames.toList.asJava
- val finalFields = ColumnOperationUtils.dropFields(inputFieldNames, fieldLists)
+ val finalFields = ColumnOperationUtils.dropFields(inputFieldNames, resolver.resolve(fieldLists))
project(finalFields, child)
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/table/ColumnFunctionsTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/table/ColumnFunctionsTest.scala
new file mode 100644
index 0000000..786ee84
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/table/ColumnFunctionsTest.scala
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.api.batch.table
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.Table
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.functions.ScalarFunction
+import org.apache.flink.table.utils.TableTestUtil._
+import org.apache.flink.table.utils.{BatchTableTestUtil, TableTestBase}
+import org.junit.Test
+
+/**
+ * Tests for column functions.
+ */
+class ColumnFunctionsTest extends TableTestBase {
+
+ val util = new BatchTableTestUtil()
+
+ private def verifyAll(tab1: Table, tab2: Table, expected: String): Unit = {
+ util.verifyTable(tab1, expected)
+ this.verifyTableEquals(tab1, tab2)
+ }
+
+ @Test
+ def testOrderBy(): Unit = {
+ val t = util.addTable[(Int, Long, String, Int, Long, String)]('a, 'b, 'c, 'd, 'e, 'f)
+
+ val tab1 = t
+ .orderBy(withColumns(1, 2 to 3))
+
+ val tab2 = t
+ .orderBy("withColumns(1, 2 to 3)")
+
+ val expected =
+ unaryNode(
+ "DataSetSort",
+ batchTableNode(0),
+ term("orderBy", "a ASC", "b ASC", "c ASC")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+}
+
+object TestFunc extends ScalarFunction {
+ def eval(a: Double, b: Long): Double = {
+ a
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/ColumnFunctionsTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/ColumnFunctionsTest.scala
new file mode 100644
index 0000000..bfde639
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/ColumnFunctionsTest.scala
@@ -0,0 +1,367 @@
+/*
+ * 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.api.stream.table
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.{Over, Slide, Table}
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.functions.ScalarFunction
+import org.apache.flink.table.functions.aggfunctions.CountAggFunction
+import org.apache.flink.table.runtime.utils.JavaUserDefinedAggFunctions.{CountDistinct, WeightedAvg}
+import org.apache.flink.table.utils.TableTestUtil.{binaryNode, streamTableNode, term, unaryNode}
+import org.apache.flink.table.utils.{StreamTableTestUtil, TableFunc0, TableTestBase}
+import org.junit.Test
+
+/**
+ * Tests for column functions which includes tests for different column functions.
+ */
+class ColumnFunctionsTest extends TableTestBase {
+
+ val util = new StreamTableTestUtil()
+
+ private def verifyAll(tab1: Table, tab2: Table, expected: String): Unit = {
+ util.verifyTable(tab1, expected)
+ this.verifyTableEquals(tab1, tab2)
+ }
+
+ @Test
+ def testStar(): Unit = {
+
+ val t = util.addTable[(Double, Long)]('double, 'long)
+
+ util.tableEnv.registerFunction("TestFunc", TestFunc)
+ val tab1 = t.select(TestFunc(withColumns('*)))
+ val tab2 = t.select("TestFunc(withColumns(*))")
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "TestFunc$(double, long) AS _c0")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+
+ @Test
+ def testColumnRange(): Unit = {
+ val t = util.addTable[(Int, Long, String, Int, Long, String)]('a, 'b, 'c, 'd, 'e, 'f)
+
+ val tab1 = t.select(withColumns('b to 'c), 'a, withColumns(5 to 6, 'd))
+ val tab2 = t.select("withColumns(b to c), a, withColumns(5 to 6, d)")
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "b", "c", "a", "e", "f", "d")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+
+ @Test
+ def testColumnWithoutRange(): Unit = {
+ val t = util.addTable[(Int, Long, String, Int, Long, String)]('a, 'b, 'c, 'd, 'e, 'f)
+
+ val tab1 = t.select(withColumns(1, 'b, 'c), 'f)
+ val tab2 = t.select("withColumns(1, b, c), f")
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "b", "c", "f")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+
+ @Test
+ def testInverseSelection(): Unit = {
+ val t = util.addTable[(Int, Long, String, Int, Long, String)]('a, 'b, 'c, 'd, 'e, 'f)
+
+ val tab1 = t
+ .select(withoutColumns(1, 'b))
+ .select(withoutColumns(1 to 2))
+
+ val tab2 = t
+ .select("withoutColumns(1, b)")
+ .select("withoutColumns(1 to 2)")
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "e", "f")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+
+ @Test
+ def testColumnFunctionsInUDF(): Unit = {
+ val t = util.addTable[(Int, Long, String, String)]('int, 'long, 'string1, 'string2)
+
+ val tab1 = t.select(concat(withColumns('string1 to 'string2)))
+ val tab2 = t.select("concat(withColumns(string1 to string2))")
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "CONCAT(string1, string2) AS _c0")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+
+ @Test
+ def testJoin(): Unit = {
+ val t1 = util.addTable[(Int, Long, String)]('int1, 'long1, 'string1)
+ val t2 = util.addTable[(Int, Long, String)]('int2, 'long2, 'string2)
+
+ val tab1 = t1.join(t2, withColumns(1) === withColumns(4))
+ val tab2 = t1.join(t2, "withColumns(1) === withColumns(4)")
+
+ val expected =
+ binaryNode(
+ "DataStreamJoin",
+ streamTableNode(0),
+ streamTableNode(1),
+ term("where", "=(int1, int2)"),
+ term("join", "int1", "long1", "string1", "int2", "long2", "string2"),
+ term("joinType", "InnerJoin")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+
+ @Test
+ def testJoinLateral(): Unit = {
+ val t = util.addTable[(Double, Long, String)]('int, 'long, 'string)
+ val func0 = new TableFunc0
+ util.tableEnv.registerFunction("func0", func0)
+
+ val tab1 = t.joinLateral(func0(withColumns('string)))
+ val tab2 = t.joinLateral("func0(withColumns(string))")
+
+ val expected =
+ unaryNode(
+ "DataStreamCorrelate",
+ streamTableNode(0),
+ term("invocation",
+ "org$apache$flink$table$utils$TableFunc0$497a630d2a145bca99673bcd05a53d2b($2)"),
+ term("correlate", "table(TableFunc0(string))"),
+ term("select", "int", "long", "string", "name", "age"),
+ term("rowType",
+ "RecordType(DOUBLE int, BIGINT long, VARCHAR(65536) string, VARCHAR(65536) name," +
+ " INTEGER age)"),
+ term("joinType", "INNER")
+ )
+
+ util.verifyTable(tab1, expected)
+ util.verify2Tables(tab1, tab2)
+ }
+
+ @Test
+ def testFilter(): Unit = {
+ val t = util.addTable[(Int, Long, String, String)]('int, 'long, 'string1, 'string2)
+
+ val tab1 = t.where(concat(withColumns('string1 to 'string2)) === "a")
+ val tab2 = t.where("concat(withColumns(string1 to string2)) = 'a'")
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "int", "long", "string1", "string2"),
+ term("where", "=(CONCAT(string1, string2), 'a')")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+
+ @Test
+ def testGroupBy(): Unit = {
+ val t = util.addTable[(Int, Long, String, Int, Long, String)]('a, 'b, 'c, 'd, 'e, 'f)
+
+ val tab1 = t
+ .groupBy(withColumns(1), 'b)
+ .select('a, 'b, withColumns('c).count)
+
+ val tab2 = t
+ .groupBy("withColumns(1), b")
+ .select("a, b, withColumns(c).count")
+
+ val expected =
+ unaryNode(
+ "DataStreamGroupAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "b", "c")
+ ),
+ term("groupBy", "a", "b"),
+ term("select", "a", "b", "COUNT(c) AS TMP_0")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+
+ @Test
+ def testWindowGroupBy(): Unit = {
+ val t = util.addTable[(Int, Long, String, Int)]('a, 'b.rowtime, 'c, 'd)
+
+ val tab1 = t
+ .window(Slide over 3.milli every 10.milli on withColumns('b) as 'w)
+ .groupBy(withColumns('a, 'b), 'w)
+ .select(withColumns(1 to 2), withColumns('c).count as 'c)
+
+ val tab2 = t
+ .window(Slide.over("3.milli").every("10.milli").on("withColumns(b)").as("w"))
+ .groupBy("withColumns(a, b), w")
+ .select("withColumns(1 to 2), withColumns(c).count as c")
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamGroupWindowAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "CAST(b) AS b", "c")
+ ),
+ term("groupBy", "a", "b"),
+ term("window", "SlidingGroupWindow('w, 'b, 3.millis, 10.millis)"),
+ term("select", "a", "b", "COUNT(c) AS TMP_0")
+ ),
+ term("select", "a", "b", "TMP_0 AS c")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+
+ @Test
+ def testOver(): Unit = {
+ val table = util.addTable[(Long, Int, String)]('a, 'b, 'c, 'proctime.proctime)
+
+ val countFun = new CountAggFunction
+ val weightAvgFun = new WeightedAvg
+ val countDist = new CountDistinct
+
+ util.tableEnv.registerFunction("countFun", countFun)
+ util.tableEnv.registerFunction("weightAvgFun", weightAvgFun)
+ util.tableEnv.registerFunction("countDist", countDist)
+
+ val tab1 = table
+ .window(
+ Over partitionBy withColumns('c) orderBy 'proctime preceding UNBOUNDED_ROW as 'w)
+ .select('c,
+ countFun(withColumns('b)) over 'w as 'mycount,
+ weightAvgFun(withColumns('a to 'b)) over 'w as 'wAvg,
+ countDist('a) over 'w as 'countDist)
+ .select('c, 'mycount, 'wAvg, 'countDist)
+
+ val tab2 = table
+ .window(
+ Over.partitionBy("withColumns(c)")
+ .orderBy("proctime")
+ .preceding("UNBOUNDED_ROW")
+ .as("w"))
+ .select("c, countFun(withColumns(b)) over w as mycount, " +
+ "weightAvgFun(withColumns(a to b)) over w as wAvg, countDist(a) over w as countDist")
+ .select('c, 'mycount, 'wAvg, 'countDist)
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ streamTableNode(0),
+ term("partitionBy", "c"),
+ term("orderBy", "proctime"),
+ term("rows", "BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"),
+ term("select", "a", "b", "c", "proctime", "CountAggFunction(b) AS w0$o0",
+ "WeightedAvg(a, b) AS w0$o1", "CountDistinct(a) AS w0$o2")
+ ),
+ term("select", "c", "w0$o0 AS mycount", "w0$o1 AS wAvg", "w0$o2 AS countDist")
+ )
+ verifyAll(tab1, tab2, expected)
+ }
+
+ @Test
+ def testAddColumns(): Unit = {
+ val t = util.addTable[(Double, Long, String)]('a, 'b, 'c)
+
+ util.tableEnv.registerFunction("TestFunc", TestFunc)
+ val tab1 = t.addColumns(TestFunc(withColumns('a, 'b)) as 'd)
+ val tab2 = t.addColumns("TestFunc(withColumns(a, b)) as d")
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "b", "c", "TestFunc$(a, b) AS d")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+
+ @Test
+ def testRenameColumns(): Unit = {
+ val t = util.addTable[(Double, Long, String)]('a, 'b, 'c)
+
+ val tab1 = t.renameColumns(withColumns('a) as 'd).select("d, b")
+ val tab2 = t.renameColumns("withColumns(a) as d").select('d, 'b)
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a AS d", "b")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+
+ @Test
+ def testDropColumns(): Unit = {
+ val t = util.addTable[(Double, Long, String)]('a, 'b, 'c)
+
+ val tab1 = t.dropColumns(withColumns('a to 'b))
+ val tab2 = t.dropColumns("withColumns(a to b)")
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "c")
+ )
+
+ verifyAll(tab1, tab2, expected)
+ }
+}
+
+object TestFunc extends ScalarFunction {
+ def eval(a: Double, b: Long): Double = {
+ a
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/validation/ColumnFunctionsValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/validation/ColumnFunctionsValidationTest.scala
new file mode 100644
index 0000000..d141dc0
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/validation/ColumnFunctionsValidationTest.scala
@@ -0,0 +1,97 @@
+/*
+ * 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.api.validation
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.{Slide, TableException, ValidationException}
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.expressions.BuiltInFunctionDefinitions
+import org.apache.flink.table.utils.{StreamTableTestUtil, TableTestBase}
+import org.junit.Test
+
+/**
+ * Tests to validate exceptions for column functions. This test can also cover the batch
+ * scenarios.
+ */
+class ColumnFunctionsValidationTest extends TableTestBase {
+
+ val util = new StreamTableTestUtil()
+ val withCol = BuiltInFunctionDefinitions.WITH_COLUMNS.getName
+ val withoutCol = BuiltInFunctionDefinitions.WITHOUT_COLUMNS.getName
+
+ @Test
+ def testIndexRangeInvalid(): Unit = {
+ expectedException.expect(classOf[IllegalArgumentException])
+ expectedException.expectMessage(
+ s"The start:2 of $withCol() or $withoutCol() should not bigger than end:1")
+
+ val t = util.addTable[(Int, Long, String, Int, Long, String)]('a, 'b, 'c, 'd, 'e, 'f)
+ val tab = t.select(withColumns(2 to 1))
+ util.verifyTable(tab, "")
+ }
+
+ @Test
+ def testNameRangeInvalid(): Unit = {
+ expectedException.expect(classOf[IllegalArgumentException])
+ expectedException.expectMessage(
+ s"The start name:b of $withCol() or $withoutCol() should not behind the end:a.")
+
+ val t = util.addTable[(Int, Long, String, Int, Long, String)]('a, 'b, 'c, 'd, 'e, 'f)
+ val tab = t.select(withColumns('b to 'a))
+ util.verifyTable(tab, "")
+ }
+
+ @Test
+ def testInvalidParameters(): Unit = {
+ expectedException.expect(classOf[TableException])
+ expectedException.expectMessage(
+ s"The parameters of $withCol() or $withoutCol() only accept column name or " +
+ "column index, but receive CallExpression.")
+
+ val t = util.addTable[(Int, Long, String, Int, Long, String)]('a, 'b, 'c, 'd, 'e, 'f)
+ val tab = t.select(withColumns(concat('f)))
+ util.verifyTable(tab, "")
+ }
+
+ @Test
+ def testInvalidRenameColumns(): Unit = {
+ expectedException.expect(classOf[ValidationException])
+ expectedException.expectMessage(
+ "Invalid AS, parameters are: [a, b, 'a'].")
+
+ val t = util.addTable[(Int, Long, String, Int, Long, String)]('a, 'b, 'c, 'd, 'e, 'f)
+ val tab = t.renameColumns(withColumns(1 to 2) as 'a) // failed, invalid as
+ util.verifyTable(tab, "")
+ }
+
+ @Test
+ def testInvalidWindowTimeField(): Unit = {
+ expectedException.expect(classOf[ValidationException])
+ expectedException.expectMessage(
+ "Group Window only supports a single time field column.")
+
+ val t = util.addTable[(Int, Long, String, Int)]('a, 'b.rowtime, 'c, 'd)
+ val tab = t
+ // failed, time field only support one column
+ .window(Slide over 3.milli every 10.milli on withColumns('b, 'a) as 'w)
+ .groupBy(withColumns('a, 'b), 'w)
+ .select(withColumns(1 to 2), withColumns('c).count as 'c)
+
+ util.verifyTable(tab, "")
+ }
+}