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 2023/11/17 09:01:31 UTC
(flink) branch master updated (e656e17352f -> 1986cfe1cb3)
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from e656e17352f [hotfix][build] Remove unnecessary notes element in pom file (#23734)
new 7c38b12388c [hotfix] Implement type inference for STREAM_RECORD and THROW_EXCEPTION
new 1986cfe1cb3 [FLINK-33410] Implement type inference for OVER function
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../pyflink/table/tests/test_expression.py | 8 +-
flink-python/pyflink/table/tests/test_window.py | 3 +-
.../org/apache/flink/table/api/Expressions.java | 11 +-
.../table/api/OverWindowPartitionedOrdered.java | 5 +-
.../resolver/rules/OverWindowResolverRule.java | 35 ++--
.../table/api/ImplicitExpressionConversions.scala | 9 +-
...JsonExistsOnError.java => OverWindowRange.java} | 12 +-
.../functions/BuiltInFunctionDefinitions.java | 34 +---
.../inference/strategies/OverTypeStrategy.java | 199 +++++++++++++++++++
.../strategies/SpecificInputTypeStrategies.java | 3 +
.../inference/strategies/OverTypeStrategyTest.java | 218 +++++++++++++++++++++
.../expressions/converter/OverConvertRule.java | 109 ++++++-----
.../functions/InternalFunctionDefinitions.java | 11 +-
.../expressions/PlannerExpressionConverter.scala | 45 +----
.../flink/table/planner/expressions/call.scala | 125 ------------
.../table/planner/expressions/overOffsets.scala | 53 -----
16 files changed, 538 insertions(+), 342 deletions(-)
copy flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/{JsonExistsOnError.java => OverWindowRange.java} (83%)
create mode 100644 flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategy.java
create mode 100644 flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategyTest.java
delete mode 100644 flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/overOffsets.scala
(flink) 02/02: [FLINK-33410] Implement type inference for OVER function
Posted by dw...@apache.org.
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
commit 1986cfe1cb3606fc00fd13735f7ba901b806f56f
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Tue Nov 14 17:43:42 2023 +0100
[FLINK-33410] Implement type inference for OVER function
---
.../pyflink/table/tests/test_expression.py | 8 +-
flink-python/pyflink/table/tests/test_window.py | 3 +-
.../org/apache/flink/table/api/Expressions.java | 11 +-
.../table/api/OverWindowPartitionedOrdered.java | 5 +-
.../resolver/rules/OverWindowResolverRule.java | 35 ++--
.../table/api/ImplicitExpressionConversions.scala | 9 +-
.../apache/flink/table/api/OverWindowRange.java | 32 +++
.../functions/BuiltInFunctionDefinitions.java | 31 +--
.../inference/strategies/OverTypeStrategy.java | 199 +++++++++++++++++++
.../strategies/SpecificInputTypeStrategies.java | 3 +
.../inference/strategies/OverTypeStrategyTest.java | 218 +++++++++++++++++++++
.../expressions/converter/OverConvertRule.java | 109 ++++++-----
.../expressions/PlannerExpressionConverter.scala | 35 +---
.../flink/table/planner/expressions/call.scala | 106 ----------
.../table/planner/expressions/overOffsets.scala | 53 -----
15 files changed, 553 insertions(+), 304 deletions(-)
diff --git a/flink-python/pyflink/table/tests/test_expression.py b/flink-python/pyflink/table/tests/test_expression.py
index 069f88cd67f..589d0894961 100644
--- a/flink-python/pyflink/table/tests/test_expression.py
+++ b/flink-python/pyflink/table/tests/test_expression.py
@@ -247,10 +247,10 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase):
from pyflink.table.expressions import UNBOUNDED_ROW, UNBOUNDED_RANGE, CURRENT_ROW, \
CURRENT_RANGE
- self.assertEqual('unboundedRow()', str(UNBOUNDED_ROW))
- self.assertEqual('unboundedRange()', str(UNBOUNDED_RANGE))
- self.assertEqual('currentRow()', str(CURRENT_ROW))
- self.assertEqual('currentRange()', str(CURRENT_RANGE))
+ self.assertEqual('UNBOUNDED_ROW', str(UNBOUNDED_ROW))
+ self.assertEqual('UNBOUNDED_RANGE', str(UNBOUNDED_RANGE))
+ self.assertEqual('CURRENT_ROW', str(CURRENT_ROW))
+ self.assertEqual('CURRENT_RANGE', str(CURRENT_RANGE))
self.assertEqual('currentDatabase()', str(current_database()))
self.assertEqual('currentDate()', str(current_date()))
diff --git a/flink-python/pyflink/table/tests/test_window.py b/flink-python/pyflink/table/tests/test_window.py
index 6fd376da3ec..f5440e55982 100644
--- a/flink-python/pyflink/table/tests/test_window.py
+++ b/flink-python/pyflink/table/tests/test_window.py
@@ -37,7 +37,8 @@ class StreamTableWindowTests(PyFlinkStreamTableTestCase):
.alias("w"))
self.assertRaisesRegex(
- Py4JJavaError, "Ordering must be defined on a time attribute",
+ Py4JJavaError,
+ "Second argument in OVER window must be a TIME ATTRIBUTE, but is: BIGINT",
result.select, expr.col("b").sum.over(expr.col("w")))
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
index 213e6b111ed..fbf4b930c60 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
@@ -212,30 +212,27 @@ public final class Expressions {
* Use this constant for a time interval. Unbounded over windows start with the first row of a
* partition.
*/
- public static final ApiExpression UNBOUNDED_ROW =
- apiCall(BuiltInFunctionDefinitions.UNBOUNDED_ROW);
+ public static final ApiExpression UNBOUNDED_ROW = lit(OverWindowRange.UNBOUNDED_ROW);
/**
* Offset constant to be used in the {@code preceding} clause of unbounded {@link Over} windows.
* Use this constant for a row-count interval. Unbounded over windows start with the first row
* of a partition.
*/
- public static final ApiExpression UNBOUNDED_RANGE =
- apiCall(BuiltInFunctionDefinitions.UNBOUNDED_RANGE);
+ public static final ApiExpression UNBOUNDED_RANGE = lit(OverWindowRange.UNBOUNDED_RANGE);
/**
* Offset constant to be used in the {@code following} clause of {@link Over} windows. Use this
* for setting the upper bound of the window to the current row.
*/
- public static final ApiExpression CURRENT_ROW = apiCall(BuiltInFunctionDefinitions.CURRENT_ROW);
+ public static final ApiExpression CURRENT_ROW = lit(OverWindowRange.CURRENT_ROW);
/**
* Offset constant to be used in the {@code following} clause of {@link Over} windows. Use this
* for setting the upper bound of the window to the sort key of the current row, i.e., all rows
* with the same sort key as the current row are included in the window.
*/
- public static final ApiExpression CURRENT_RANGE =
- apiCall(BuiltInFunctionDefinitions.CURRENT_RANGE);
+ public static final ApiExpression CURRENT_RANGE = lit(OverWindowRange.CURRENT_RANGE);
/**
* Returns the current SQL date in local time zone, the return type of this expression is {@link
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrdered.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrdered.java
index 15f5dd456ed..c4ed5657cd5 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrdered.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrdered.java
@@ -20,13 +20,12 @@ package org.apache.flink.table.api;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.table.expressions.Expression;
-import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import java.util.List;
import java.util.Optional;
-import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedCall;
import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedRef;
+import static org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral;
/** Partially defined over window with (optional) partitioning and order. */
@PublicEvolving
@@ -71,7 +70,7 @@ public final class OverWindowPartitionedOrdered {
alias,
partitionBy,
orderBy,
- unresolvedCall(BuiltInFunctionDefinitions.UNBOUNDED_RANGE),
+ valueLiteral(OverWindowRange.UNBOUNDED_RANGE),
Optional.empty());
}
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/OverWindowResolverRule.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/OverWindowResolverRule.java
index 4d7ec3fdd0f..b6cf186ca68 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/OverWindowResolverRule.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/OverWindowResolverRule.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.expressions.resolver.rules;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.OverWindowRange;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.UnresolvedCallExpression;
@@ -26,7 +27,6 @@ import org.apache.flink.table.expressions.ValueLiteralExpression;
import org.apache.flink.table.expressions.resolver.LocalOverWindow;
import org.apache.flink.table.expressions.utils.ApiExpressionDefaultVisitor;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
-import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.types.logical.LogicalType;
import java.util.ArrayList;
@@ -35,6 +35,7 @@ import java.util.stream.Collectors;
import static java.util.Arrays.asList;
import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedCall;
+import static org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral;
import static org.apache.flink.table.types.logical.LogicalTypeRoot.BIGINT;
import static org.apache.flink.table.types.logical.LogicalTypeRoot.INTERVAL_DAY_TIME;
@@ -105,9 +106,9 @@ final class OverWindowResolverRule implements ResolverRule {
.getPreceding()
.accept(OVER_WINDOW_KIND_EXTRACTOR);
if (kind == WindowKind.ROW) {
- return unresolvedCall(BuiltInFunctionDefinitions.CURRENT_ROW);
+ return valueLiteral(OverWindowRange.CURRENT_ROW);
} else {
- return unresolvedCall(BuiltInFunctionDefinitions.CURRENT_RANGE);
+ return valueLiteral(OverWindowRange.CURRENT_RANGE);
}
});
}
@@ -133,18 +134,24 @@ final class OverWindowResolverRule implements ResolverRule {
} else if (literalType.is(INTERVAL_DAY_TIME)) {
return WindowKind.RANGE;
}
- return defaultMethod(valueLiteral);
- }
- @Override
- public WindowKind visit(UnresolvedCallExpression unresolvedCall) {
- final FunctionDefinition definition = unresolvedCall.getFunctionDefinition();
- if (definition == BuiltInFunctionDefinitions.UNBOUNDED_ROW) {
- return WindowKind.ROW;
- } else if (definition == BuiltInFunctionDefinitions.UNBOUNDED_RANGE) {
- return WindowKind.RANGE;
- }
- return defaultMethod(unresolvedCall);
+ return valueLiteral
+ .getValueAs(OverWindowRange.class)
+ .map(
+ v -> {
+ switch (v) {
+ case CURRENT_ROW:
+ case UNBOUNDED_ROW:
+ return WindowKind.ROW;
+ case CURRENT_RANGE:
+ case UNBOUNDED_RANGE:
+ return WindowKind.RANGE;
+ default:
+ throw new IllegalArgumentException(
+ "Unexpected window range: " + v);
+ }
+ })
+ .orElseGet(() -> defaultMethod(valueLiteral));
}
@Override
diff --git a/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala b/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
index 543d3014cb7..f3044c6d83c 100644
--- a/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
+++ b/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
@@ -58,28 +58,27 @@ trait ImplicitExpressionConversions {
* Offset constant to be used in the `preceding` clause of unbounded [[Over]] windows. Use this
* constant for a time interval. Unbounded over windows start with the first row of a partition.
*/
- implicit val UNBOUNDED_ROW: Expression = unresolvedCall(BuiltInFunctionDefinitions.UNBOUNDED_ROW)
+ implicit val UNBOUNDED_ROW: Expression = lit(OverWindowRange.UNBOUNDED_ROW)
/**
* Offset constant to be used in the `preceding` clause of unbounded [[Over]] windows. Use this
* constant for a row-count interval. Unbounded over windows start with the first row of a
* partition.
*/
- implicit val UNBOUNDED_RANGE: Expression =
- unresolvedCall(BuiltInFunctionDefinitions.UNBOUNDED_RANGE)
+ implicit val UNBOUNDED_RANGE: Expression = lit(OverWindowRange.UNBOUNDED_RANGE)
/**
* Offset constant to be used in the `following` clause of [[Over]] windows. Use this for setting
* the upper bound of the window to the current row.
*/
- implicit val CURRENT_ROW: Expression = unresolvedCall(BuiltInFunctionDefinitions.CURRENT_ROW)
+ implicit val CURRENT_ROW: Expression = lit(OverWindowRange.CURRENT_ROW)
/**
* Offset constant to be used in the `following` clause of [[Over]] windows. Use this for setting
* the upper bound of the window to the sort key of the current row, i.e., all rows with the same
* sort key as the current row are included in the window.
*/
- implicit val CURRENT_RANGE: Expression = unresolvedCall(BuiltInFunctionDefinitions.CURRENT_RANGE)
+ implicit val CURRENT_RANGE: Expression = lit(OverWindowRange.CURRENT_RANGE)
// ----------------------------------------------------------------------------------------------
// Implicit conversions
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/OverWindowRange.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/OverWindowRange.java
new file mode 100644
index 00000000000..dd3e4e3388c
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/OverWindowRange.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.api;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.expressions.TableSymbol;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+
+/** Range specification for {@link BuiltInFunctionDefinitions#OVER} window. */
+@PublicEvolving
+public enum OverWindowRange implements TableSymbol {
+ CURRENT_ROW,
+ CURRENT_RANGE,
+ UNBOUNDED_ROW,
+ UNBOUNDED_RANGE
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index c7461864131..cfaf37fab71 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -2061,35 +2061,8 @@ public final class BuiltInFunctionDefinitions {
BuiltInFunctionDefinition.newBuilder()
.name("over")
.kind(OTHER)
- .outputTypeStrategy(TypeStrategies.MISSING)
- .build();
-
- public static final BuiltInFunctionDefinition UNBOUNDED_RANGE =
- BuiltInFunctionDefinition.newBuilder()
- .name("unboundedRange")
- .kind(OTHER)
- .outputTypeStrategy(TypeStrategies.MISSING)
- .build();
-
- public static final BuiltInFunctionDefinition UNBOUNDED_ROW =
- BuiltInFunctionDefinition.newBuilder()
- .name("unboundedRow")
- .kind(OTHER)
- .outputTypeStrategy(TypeStrategies.MISSING)
- .build();
-
- public static final BuiltInFunctionDefinition CURRENT_RANGE =
- BuiltInFunctionDefinition.newBuilder()
- .name("currentRange")
- .kind(OTHER)
- .outputTypeStrategy(TypeStrategies.MISSING)
- .build();
-
- public static final BuiltInFunctionDefinition CURRENT_ROW =
- BuiltInFunctionDefinition.newBuilder()
- .name("currentRow")
- .kind(OTHER)
- .outputTypeStrategy(TypeStrategies.MISSING)
+ .inputTypeStrategy(SpecificInputTypeStrategies.OVER)
+ .outputTypeStrategy(TypeStrategies.argument(0))
.build();
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategy.java
new file mode 100644
index 00000000000..3fd35498759
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategy.java
@@ -0,0 +1,199 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.OverWindowRange;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.ConstantArgumentCount;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+
+import java.math.BigDecimal;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+
+/** Specific {@link InputTypeStrategy} for {@link BuiltInFunctionDefinitions#OVER}. */
+@Internal
+public class OverTypeStrategy implements InputTypeStrategy {
+ @Override
+ public ArgumentCount getArgumentCount() {
+ return ConstantArgumentCount.from(4);
+ }
+
+ @Override
+ public Optional<List<DataType>> inferInputTypes(
+ CallContext callContext, boolean throwOnFailure) {
+ final List<DataType> argumentDataTypes = callContext.getArgumentDataTypes();
+ final DataType timeAttribute = argumentDataTypes.get(1);
+
+ if (!LogicalTypeChecks.isTimeAttribute(timeAttribute.getLogicalType())) {
+ return callContext.fail(
+ throwOnFailure,
+ "Second argument in OVER window must be a TIME ATTRIBUTE, but is: "
+ + timeAttribute);
+ }
+
+ if (!callContext.isArgumentLiteral(2)) {
+ return callContext.fail(
+ throwOnFailure, "Preceding must be a row interval or time interval literal.");
+ }
+
+ if (!callContext.isArgumentLiteral(3)) {
+ return callContext.fail(
+ throwOnFailure, "Following must be a row interval or time interval literal.");
+ }
+
+ final DataType preceding = argumentDataTypes.get(2);
+ final DataType following = argumentDataTypes.get(3);
+
+ final LogicalTypeRoot precedingTypeRoot = preceding.getLogicalType().getTypeRoot();
+ final Optional<String> precedingErr =
+ validateFollowingPreceding(
+ precedingTypeRoot, FollowingPreceding.PRECEDING, callContext);
+
+ if (precedingErr.isPresent()) {
+ return callContext.fail(throwOnFailure, precedingErr.get());
+ }
+
+ final LogicalTypeRoot followingTypeRoot = following.getLogicalType().getTypeRoot();
+ final Optional<String> followingErr =
+ validateFollowingPreceding(
+ followingTypeRoot, FollowingPreceding.FOLLOWING, callContext);
+
+ if (followingErr.isPresent()) {
+ return callContext.fail(throwOnFailure, followingErr.get());
+ }
+
+ final boolean isPrecedingRowWindow =
+ isRowWindow(callContext, precedingTypeRoot, FollowingPreceding.PRECEDING);
+ final boolean isFollowingRowWindow =
+ isRowWindow(callContext, followingTypeRoot, FollowingPreceding.FOLLOWING);
+
+ if (isPrecedingRowWindow && !isFollowingRowWindow
+ || !isPrecedingRowWindow && isFollowingRowWindow) {
+ return callContext.fail(
+ throwOnFailure, "Preceding and following must be of same interval type.");
+ }
+
+ return Optional.of(argumentDataTypes);
+ }
+
+ private static boolean isRowWindow(
+ CallContext callContext,
+ LogicalTypeRoot typeRoot,
+ FollowingPreceding followingPreceding) {
+ switch (typeRoot) {
+ case BIGINT:
+ return true;
+ case SYMBOL:
+ final OverWindowRange windowRange =
+ callContext
+ .getArgumentValue(followingPreceding.pos, OverWindowRange.class)
+ .get();
+ return windowRange == OverWindowRange.UNBOUNDED_ROW
+ || windowRange == OverWindowRange.CURRENT_ROW;
+ default:
+ return false;
+ }
+ }
+
+ private enum FollowingPreceding {
+ PRECEDING("Preceding", 2),
+ FOLLOWING("Following", 3);
+
+ private final int pos;
+ private final String name;
+
+ FollowingPreceding(String name, int pos) {
+ this.name = name;
+ this.pos = pos;
+ }
+ }
+
+ private Optional<String> validateFollowingPreceding(
+ LogicalTypeRoot typeRoot,
+ FollowingPreceding followingPreceding,
+ CallContext callContext) {
+ switch (typeRoot) {
+ case BIGINT:
+ if (callContext.getArgumentValue(followingPreceding.pos, Long.class).get() <= 0) {
+ return Optional.of(
+ String.format(
+ "%s row interval must be larger than 0.",
+ followingPreceding.name));
+ }
+ break;
+ case INTERVAL_DAY_TIME:
+ if (callContext
+ .getArgumentValue(followingPreceding.pos, BigDecimal.class)
+ .get()
+ .compareTo(BigDecimal.ZERO)
+ < 0) {
+ return Optional.of(
+ String.format(
+ "%s time interval must be equal or larger than 0.",
+ followingPreceding.name));
+ }
+ break;
+ case SYMBOL:
+ final Optional<OverWindowRange> precedingSymbol =
+ callContext.getArgumentValue(followingPreceding.pos, OverWindowRange.class);
+ if (!precedingSymbol.isPresent()) {
+ return Optional.of(
+ String.format(
+ "%s must be a row interval or time interval literal.",
+ followingPreceding.name));
+ }
+ break;
+ default:
+ return Optional.of(
+ String.format(
+ "%s must be a row interval or time interval literal.",
+ followingPreceding.name));
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
+ return Arrays.asList(
+ Signature.of(
+ Signature.Argument.ofGroup("ANY"),
+ Signature.Argument.ofGroup("TIME ATTRIBUTE"),
+ Signature.Argument.ofGroup(
+ "INTERVAL LITERAL | CURRENT_RANGE | UNBOUNDED_RANGE"),
+ Signature.Argument.ofGroup(
+ "INTERVAL LITERAL | CURRENT_RANGE | UNBOUNDED_RANGE"),
+ Signature.Argument.ofGroupVarying("ANY")),
+ Signature.of(
+ Signature.Argument.ofGroup("ANY"),
+ Signature.Argument.ofGroup("TIME ATTRIBUTE"),
+ Signature.Argument.ofGroup("BIGINT | CURRENT_ROW | UNBOUNDED_ROW"),
+ Signature.Argument.ofGroup("BIGINT | CURRENT_ROW | UNBOUNDED_ROW"),
+ Signature.Argument.ofGroupVarying("ANY")));
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
index 20688cca2e0..3bb5e2f0a5b 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
@@ -60,6 +60,9 @@ public final class SpecificInputTypeStrategies {
public static final InputTypeStrategy CURRENT_WATERMARK =
new CurrentWatermarkInputTypeStrategy();
+ /** See {@link OverTypeStrategy}. */
+ public static final InputTypeStrategy OVER = new OverTypeStrategy();
+
/** See {@link WindowTimeIndictorInputTypeStrategy}. */
public static InputTypeStrategy windowTimeIndicator(TimestampKind timestampKind) {
return new WindowTimeIndictorInputTypeStrategy(timestampKind);
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategyTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategyTest.java
new file mode 100644
index 00000000000..efc80e61ff8
--- /dev/null
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategyTest.java
@@ -0,0 +1,218 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.OverWindowRange;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.InputTypeStrategiesTestBase;
+import org.apache.flink.table.types.logical.SymbolType;
+import org.apache.flink.table.types.logical.TimestampKind;
+import org.apache.flink.table.types.logical.TimestampType;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.math.BigDecimal;
+import java.util.stream.Stream;
+
+/** Tests for {@link OverTypeStrategy}. */
+class OverTypeStrategyTest extends InputTypeStrategiesTestBase {
+
+ private static final DataType TIME_ATTRIBUTE_TYPE =
+ TypeConversions.fromLogicalToDataType(
+ new TimestampType(false, TimestampKind.ROWTIME, 3));
+
+ private static final DataType SYMBOL_TYPE =
+ TypeConversions.fromLogicalToDataType(new SymbolType<>());
+
+ @Override
+ protected Stream<TestSpec> testData() {
+ return Stream.of(
+ TestSpec.forStrategy("Row window literals", SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(), TIME_ATTRIBUTE_TYPE, SYMBOL_TYPE, SYMBOL_TYPE)
+ .calledWithLiteralAt(2, OverWindowRange.UNBOUNDED_ROW)
+ .calledWithLiteralAt(3, OverWindowRange.UNBOUNDED_ROW)
+ .expectArgumentTypes(
+ DataTypes.INT(), TIME_ATTRIBUTE_TYPE, SYMBOL_TYPE, SYMBOL_TYPE),
+ TestSpec.forStrategy("Range window literals", SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(), TIME_ATTRIBUTE_TYPE, SYMBOL_TYPE, SYMBOL_TYPE)
+ .calledWithLiteralAt(2, OverWindowRange.CURRENT_RANGE)
+ .calledWithLiteralAt(3, OverWindowRange.UNBOUNDED_RANGE)
+ .expectArgumentTypes(
+ DataTypes.INT(), TIME_ATTRIBUTE_TYPE, SYMBOL_TYPE, SYMBOL_TYPE),
+ TestSpec.forStrategy(
+ "Row window literal with row number following",
+ SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ SYMBOL_TYPE,
+ DataTypes.BIGINT())
+ .calledWithLiteralAt(2, OverWindowRange.CURRENT_ROW)
+ .calledWithLiteralAt(3, 10L)
+ .expectArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ SYMBOL_TYPE,
+ DataTypes.BIGINT()),
+ TestSpec.forStrategy(
+ "Row window literal with row number preceding",
+ SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ DataTypes.BIGINT(),
+ SYMBOL_TYPE)
+ .calledWithLiteralAt(2, 10L)
+ .calledWithLiteralAt(3, OverWindowRange.CURRENT_ROW)
+ .expectArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ DataTypes.BIGINT(),
+ SYMBOL_TYPE),
+ TestSpec.forStrategy(
+ "Row window literal with row number preceding and following",
+ SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ DataTypes.BIGINT(),
+ DataTypes.BIGINT())
+ .calledWithLiteralAt(2, 10L)
+ .calledWithLiteralAt(3, 10L)
+ .expectArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ DataTypes.BIGINT(),
+ DataTypes.BIGINT()),
+ TestSpec.forStrategy(
+ "Range window literal with time following",
+ SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ SYMBOL_TYPE,
+ DataTypes.INTERVAL(DataTypes.SECOND()))
+ .calledWithLiteralAt(2, OverWindowRange.CURRENT_RANGE)
+ .calledWithLiteralAt(3, BigDecimal.valueOf(10L))
+ .expectArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ SYMBOL_TYPE,
+ DataTypes.INTERVAL(DataTypes.SECOND())),
+ TestSpec.forStrategy(
+ "Range window literal with time preceding",
+ SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ DataTypes.INTERVAL(DataTypes.SECOND()),
+ SYMBOL_TYPE)
+ .calledWithLiteralAt(2, BigDecimal.valueOf(10L))
+ .calledWithLiteralAt(3, OverWindowRange.CURRENT_RANGE)
+ .expectArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ DataTypes.INTERVAL(DataTypes.SECOND()),
+ SYMBOL_TYPE),
+ TestSpec.forStrategy(
+ "Range window literal with time preceding and following",
+ SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ DataTypes.INTERVAL(DataTypes.SECOND()),
+ DataTypes.INTERVAL(DataTypes.SECOND()))
+ .calledWithLiteralAt(2, BigDecimal.valueOf(10L))
+ .calledWithLiteralAt(3, BigDecimal.valueOf(10L))
+ .expectArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ DataTypes.INTERVAL(DataTypes.SECOND()),
+ DataTypes.INTERVAL(DataTypes.SECOND())),
+ TestSpec.forStrategy(
+ "Different window kind literals", SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(), TIME_ATTRIBUTE_TYPE, SYMBOL_TYPE, SYMBOL_TYPE)
+ .calledWithLiteralAt(2, OverWindowRange.CURRENT_ROW)
+ .calledWithLiteralAt(3, OverWindowRange.UNBOUNDED_RANGE)
+ .expectErrorMessage(
+ "Preceding and following must be of same interval type."),
+ TestSpec.forStrategy("Non literal preceding", SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ DataTypes.BIGINT(),
+ SYMBOL_TYPE)
+ .calledWithLiteralAt(3, OverWindowRange.UNBOUNDED_ROW)
+ .expectErrorMessage(
+ "Preceding must be a row interval or time interval literal."),
+ TestSpec.forStrategy("Non literal following", SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ SYMBOL_TYPE,
+ DataTypes.BIGINT())
+ .calledWithLiteralAt(2, OverWindowRange.UNBOUNDED_ROW)
+ .expectErrorMessage(
+ "Following must be a row interval or time interval literal."),
+ TestSpec.forStrategy("Negative preceding", SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ DataTypes.BIGINT(),
+ SYMBOL_TYPE)
+ .calledWithLiteralAt(2, -2L)
+ .calledWithLiteralAt(3, OverWindowRange.UNBOUNDED_ROW)
+ .expectErrorMessage("Preceding row interval must be larger than 0."),
+ TestSpec.forStrategy("Negative following", SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ SYMBOL_TYPE,
+ DataTypes.BIGINT())
+ .calledWithLiteralAt(2, OverWindowRange.UNBOUNDED_ROW)
+ .calledWithLiteralAt(3, -2L)
+ .expectErrorMessage("Following row interval must be larger than 0."),
+ TestSpec.forStrategy("Negative time preceding", SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ DataTypes.INTERVAL(DataTypes.SECOND()),
+ SYMBOL_TYPE)
+ .calledWithLiteralAt(2, BigDecimal.valueOf(-10L))
+ .calledWithLiteralAt(3, OverWindowRange.UNBOUNDED_RANGE)
+ .expectErrorMessage(
+ "Preceding time interval must be equal or larger than 0."),
+ TestSpec.forStrategy("Zero time following", SpecificInputTypeStrategies.OVER)
+ .calledWithArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ SYMBOL_TYPE,
+ DataTypes.INTERVAL(DataTypes.SECOND()))
+ .calledWithLiteralAt(2, OverWindowRange.UNBOUNDED_RANGE)
+ .calledWithLiteralAt(3, BigDecimal.valueOf(0))
+ .expectArgumentTypes(
+ DataTypes.INT(),
+ TIME_ATTRIBUTE_TYPE,
+ SYMBOL_TYPE,
+ DataTypes.INTERVAL(DataTypes.SECOND())));
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java
index 72a50297bcb..46dbe94310e 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.expressions.converter;
+import org.apache.flink.table.api.OverWindowRange;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.expressions.CallExpression;
import org.apache.flink.table.expressions.Expression;
@@ -111,9 +112,7 @@ public class OverConvertRule implements CallExpressionConvertRule {
.collect(Collectors.toList());
// assemble bounds
Expression preceding = children.get(2);
- boolean isPhysical =
- fromDataTypeToLogicalType(((ResolvedExpression) preceding).getOutputDataType())
- .is(LogicalTypeRoot.BIGINT);
+ boolean isRows = isRows((ValueLiteralExpression) preceding);
Expression following = children.get(3);
RexWindowBound lowerBound = createBound(context, preceding, SqlKind.PRECEDING);
RexWindowBound upperBound = createBound(context, following, SqlKind.FOLLOWING);
@@ -130,7 +129,7 @@ public class OverConvertRule implements CallExpressionConvertRule {
orderKey,
lowerBound,
upperBound,
- isPhysical,
+ isRows,
true,
false,
isDistinct,
@@ -139,6 +138,16 @@ public class OverConvertRule implements CallExpressionConvertRule {
return Optional.empty();
}
+ private static boolean isRows(ValueLiteralExpression preceding) {
+ return preceding
+ .getValueAs(OverWindowRange.class)
+ .map(r -> r == OverWindowRange.CURRENT_ROW || r == OverWindowRange.UNBOUNDED_ROW)
+ .orElseGet(
+ () ->
+ fromDataTypeToLogicalType(preceding.getOutputDataType())
+ .is(LogicalTypeRoot.BIGINT));
+ }
+
private RexNode createCollation(
RexNode node,
RelFieldCollation.Direction direction,
@@ -182,57 +191,59 @@ public class OverConvertRule implements CallExpressionConvertRule {
}
private RexWindowBound createBound(ConvertContext context, Expression bound, SqlKind sqlKind) {
- if (bound instanceof CallExpression) {
- CallExpression callExpr = (CallExpression) bound;
- FunctionDefinition func = callExpr.getFunctionDefinition();
- if (BuiltInFunctionDefinitions.UNBOUNDED_ROW.equals(func)
- || BuiltInFunctionDefinitions.UNBOUNDED_RANGE.equals(func)) {
+ if (bound instanceof ValueLiteralExpression) {
+ final ValueLiteralExpression literal = (ValueLiteralExpression) bound;
+ return literal.getValueAs(OverWindowRange.class)
+ .map(r -> createSymbolBound(r, sqlKind))
+ .orElseGet(
+ () ->
+ createLiteralBound(
+ context, (ValueLiteralExpression) bound, sqlKind));
+ } else {
+ throw new TableException("Unexpected expression: " + bound);
+ }
+ }
+
+ private static RexWindowBound createLiteralBound(
+ ConvertContext context, ValueLiteralExpression bound, SqlKind sqlKind) {
+ RelDataType returnType =
+ context.getTypeFactory()
+ .createFieldTypeFromLogicalType(new DecimalType(true, 19, 0));
+ SqlOperator sqlOperator =
+ new SqlPostfixOperator(
+ sqlKind.name(), sqlKind, 2, new OrdinalReturnTypeInference(0), null, null);
+ SqlNode[] operands = new SqlNode[] {SqlLiteral.createExactNumeric("1", SqlParserPos.ZERO)};
+ SqlNode node = new SqlBasicCall(sqlOperator, operands, SqlParserPos.ZERO);
+
+ RexNode literalRexNode =
+ bound.getValueAs(BigDecimal.class)
+ .map(v -> context.getRelBuilder().literal(v))
+ .orElse(context.getRelBuilder().literal(extractValue(bound, Object.class)));
+
+ List<RexNode> expressions = new ArrayList<>();
+ expressions.add(literalRexNode);
+ RexNode rexNode =
+ context.getRelBuilder()
+ .getRexBuilder()
+ .makeCall(returnType, sqlOperator, expressions);
+ return RexWindowBounds.create(node, rexNode);
+ }
+
+ private static RexWindowBound createSymbolBound(OverWindowRange bound, SqlKind sqlKind) {
+ switch (bound) {
+ case CURRENT_ROW:
+ case CURRENT_RANGE:
+ SqlNode currentRow = SqlWindow.createCurrentRow(SqlParserPos.ZERO);
+ return RexWindowBounds.create(currentRow, null);
+ case UNBOUNDED_ROW:
+ case UNBOUNDED_RANGE:
SqlNode unbounded =
sqlKind.equals(SqlKind.PRECEDING)
? SqlWindow.createUnboundedPreceding(SqlParserPos.ZERO)
: SqlWindow.createUnboundedFollowing(SqlParserPos.ZERO);
return RexWindowBounds.create(unbounded, null);
- } else if (BuiltInFunctionDefinitions.CURRENT_ROW.equals(func)
- || BuiltInFunctionDefinitions.CURRENT_RANGE.equals(func)) {
- SqlNode currentRow = SqlWindow.createCurrentRow(SqlParserPos.ZERO);
- return RexWindowBounds.create(currentRow, null);
- } else {
+ default:
throw new IllegalArgumentException("Unexpected expression: " + bound);
- }
- } else if (bound instanceof ValueLiteralExpression) {
- RelDataType returnType =
- context.getTypeFactory()
- .createFieldTypeFromLogicalType(new DecimalType(true, 19, 0));
- SqlOperator sqlOperator =
- new SqlPostfixOperator(
- sqlKind.name(),
- sqlKind,
- 2,
- new OrdinalReturnTypeInference(0),
- null,
- null);
- SqlNode[] operands =
- new SqlNode[] {SqlLiteral.createExactNumeric("1", SqlParserPos.ZERO)};
- SqlNode node = new SqlBasicCall(sqlOperator, operands, SqlParserPos.ZERO);
-
- ValueLiteralExpression literalExpr = (ValueLiteralExpression) bound;
- RexNode literalRexNode =
- literalExpr
- .getValueAs(BigDecimal.class)
- .map(v -> context.getRelBuilder().literal(v))
- .orElse(
- context.getRelBuilder()
- .literal(extractValue(literalExpr, Object.class)));
-
- List<RexNode> expressions = new ArrayList<>();
- expressions.add(literalRexNode);
- RexNode rexNode =
- context.getRelBuilder()
- .getRexBuilder()
- .makeCall(returnType, sqlOperator, expressions);
- return RexWindowBounds.create(node, rexNode);
- } else {
- throw new TableException("Unexpected expression: " + bound);
}
}
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala
index 2cbfcd563e9..a9b5b52c3f9 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala
@@ -22,7 +22,6 @@ import org.apache.flink.table.api.{TableException, ValidationException}
import org.apache.flink.table.expressions._
import org.apache.flink.table.functions._
import org.apache.flink.table.functions.BuiltInFunctionDefinitions._
-import org.apache.flink.table.planner.functions.InternalFunctionDefinitions.THROW_EXCEPTION
import org.apache.flink.table.runtime.types.TypeInfoDataTypeConverter.fromDataTypeToTypeInfo
import org.apache.flink.table.types.logical.LogicalTypeRoot.{CHAR, DECIMAL, SYMBOL}
import org.apache.flink.table.types.logical.utils.LogicalTypeChecks.{hasLength, hasPrecision, hasScale}
@@ -108,38 +107,8 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
tafd.getAccumulatorTypeInfo,
args)
- case fd: FunctionDefinition =>
- fd match {
-
- case OVER =>
- assert(args.size >= 4)
- OverCall(
- args.head,
- args.slice(4, args.size),
- args(1),
- args(2),
- args(3)
- )
-
- case UNBOUNDED_RANGE =>
- assert(args.isEmpty)
- UnboundedRange()
-
- case UNBOUNDED_ROW =>
- assert(args.isEmpty)
- UnboundedRow()
-
- case CURRENT_RANGE =>
- assert(args.isEmpty)
- CurrentRange()
-
- case CURRENT_ROW =>
- assert(args.isEmpty)
- CurrentRow()
-
- case _ =>
- unknownFunctionHandler()
- }
+ case _: FunctionDefinition =>
+ unknownFunctionHandler()
}
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/call.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/call.scala
index cb2b9e79cad..a6401076759 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/call.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/call.scala
@@ -60,112 +60,6 @@ case class UnresolvedOverCall(agg: PlannerExpression, alias: PlannerExpression)
override private[flink] def children = Seq()
}
-/**
- * Over expression for Calcite over transform.
- *
- * @param agg
- * over-agg expression
- * @param partitionBy
- * The fields by which the over window is partitioned
- * @param orderBy
- * The field by which the over window is sorted
- * @param preceding
- * The lower bound of the window
- * @param following
- * The upper bound of the window
- */
-case class OverCall(
- agg: PlannerExpression,
- partitionBy: Seq[PlannerExpression],
- orderBy: PlannerExpression,
- preceding: PlannerExpression,
- following: PlannerExpression)
- extends PlannerExpression {
-
- override def toString: String = s"$agg OVER (" +
- s"PARTITION BY (${partitionBy.mkString(", ")}) " +
- s"ORDER BY $orderBy " +
- s"PRECEDING $preceding " +
- s"FOLLOWING $following)"
-
- override private[flink] def children: Seq[PlannerExpression] =
- Seq(agg) ++ Seq(orderBy) ++ partitionBy ++ Seq(preceding) ++ Seq(following)
-
- override private[flink] def resultType = agg.resultType
-
- override private[flink] def validateInput(): ValidationResult = {
-
- // check that agg expression is aggregation
- agg match {
- case _: Aggregation =>
- ValidationSuccess
- case _ =>
- return ValidationFailure(s"OVER can only be applied on an aggregation.")
- }
-
- // check partitionBy expression keys are resolved field reference
- partitionBy.foreach {
- case r: PlannerResolvedFieldReference if r.resultType.isKeyType =>
- ValidationSuccess
- case r: PlannerResolvedFieldReference =>
- return ValidationFailure(
- s"Invalid PartitionBy expression: $r. " +
- s"Expression must return key type.")
- case r =>
- return ValidationFailure(
- s"Invalid PartitionBy expression: $r. " +
- s"Expression must be a resolved field reference.")
- }
-
- // check preceding is valid
- preceding match {
- case _: CurrentRow | _: CurrentRange | _: UnboundedRow | _: UnboundedRange =>
- ValidationSuccess
- case Literal(v: Long, BasicTypeInfo.LONG_TYPE_INFO) if v > 0 =>
- ValidationSuccess
- case Literal(_, BasicTypeInfo.LONG_TYPE_INFO) =>
- return ValidationFailure("Preceding row interval must be larger than 0.")
- case Literal(v: Long, _: TimeIntervalTypeInfo[_]) if v >= 0 =>
- ValidationSuccess
- case Literal(_, _: TimeIntervalTypeInfo[_]) =>
- return ValidationFailure("Preceding time interval must be equal or larger than 0.")
- case Literal(_, _) =>
- return ValidationFailure("Preceding must be a row interval or time interval literal.")
- }
-
- // check following is valid
- following match {
- case _: CurrentRow | _: CurrentRange | _: UnboundedRow | _: UnboundedRange =>
- ValidationSuccess
- case Literal(v: Long, BasicTypeInfo.LONG_TYPE_INFO) if v > 0 =>
- ValidationSuccess
- case Literal(_, BasicTypeInfo.LONG_TYPE_INFO) =>
- return ValidationFailure("Following row interval must be larger than 0.")
- case Literal(v: Long, _: TimeIntervalTypeInfo[_]) if v >= 0 =>
- ValidationSuccess
- case Literal(_, _: TimeIntervalTypeInfo[_]) =>
- return ValidationFailure("Following time interval must be equal or larger than 0.")
- case Literal(_, _) =>
- return ValidationFailure("Following must be a row interval or time interval literal.")
- }
-
- // check that preceding and following are of same type
- (preceding, following) match {
- case (p: PlannerExpression, f: PlannerExpression) if p.resultType == f.resultType =>
- ValidationSuccess
- case _ =>
- return ValidationFailure("Preceding and following must be of same interval type.")
- }
-
- // check time field
- if (!PlannerExpressionUtils.isTimeAttribute(orderBy)) {
- return ValidationFailure("Ordering must be defined on a time attribute.")
- }
-
- ValidationSuccess
- }
-}
-
/**
* Expression for calling a user-defined scalar functions.
*
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/overOffsets.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/overOffsets.scala
deleted file mode 100644
index 545f467dfab..00000000000
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/overOffsets.scala
+++ /dev/null
@@ -1,53 +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.planner.expressions
-
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo
-import org.apache.flink.table.typeutils.TimeIntervalTypeInfo
-
-case class CurrentRow() extends PlannerExpression {
- override private[flink] def resultType = BasicTypeInfo.LONG_TYPE_INFO
-
- override private[flink] def children = Seq()
-
- override def toString = "CURRENT ROW"
-}
-
-case class CurrentRange() extends PlannerExpression {
- override private[flink] def resultType = TimeIntervalTypeInfo.INTERVAL_MILLIS
-
- override private[flink] def children = Seq()
-
- override def toString = "CURRENT RANGE"
-}
-
-case class UnboundedRow() extends PlannerExpression {
- override private[flink] def resultType = BasicTypeInfo.LONG_TYPE_INFO
-
- override private[flink] def children = Seq()
-
- override def toString = "UNBOUNDED ROW"
-}
-
-case class UnboundedRange() extends PlannerExpression {
- override private[flink] def resultType = TimeIntervalTypeInfo.INTERVAL_MILLIS
-
- override private[flink] def children = Seq()
-
- override def toString = "UNBOUNDED RANGE"
-}
(flink) 01/02: [hotfix] Implement type inference for STREAM_RECORD and THROW_EXCEPTION
Posted by dw...@apache.org.
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
commit 7c38b12388cf61fe27c05fe5b20ce4f9b8ca382b
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Tue Nov 14 17:43:17 2023 +0100
[hotfix] Implement type inference for STREAM_RECORD and THROW_EXCEPTION
---
.../table/functions/BuiltInFunctionDefinitions.java | 3 ++-
.../functions/InternalFunctionDefinitions.java | 11 +++++++++--
.../expressions/PlannerExpressionConverter.scala | 10 ----------
.../apache/flink/table/planner/expressions/call.scala | 19 -------------------
4 files changed, 11 insertions(+), 32 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 4b208552d73..c7461864131 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -2306,7 +2306,8 @@ public final class BuiltInFunctionDefinitions {
BuiltInFunctionDefinition.newBuilder()
.name("streamRecordTimestamp")
.kind(OTHER)
- .outputTypeStrategy(TypeStrategies.MISSING)
+ .inputTypeStrategy(NO_ARGS)
+ .outputTypeStrategy(explicit(DataTypes.BIGINT()))
.build();
public static final BuiltInFunctionDefinition RANGE_TO =
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/InternalFunctionDefinitions.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/InternalFunctionDefinitions.java
index 158f8464df6..6f838383e82 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/InternalFunctionDefinitions.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/InternalFunctionDefinitions.java
@@ -18,10 +18,14 @@
package org.apache.flink.table.planner.functions;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.functions.BuiltInFunctionDefinition;
-import org.apache.flink.table.types.inference.TypeStrategies;
+import org.apache.flink.table.types.inference.InputTypeStrategies;
import static org.apache.flink.table.functions.FunctionKind.SCALAR;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.explicit;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.sequence;
+import static org.apache.flink.table.types.inference.TypeStrategies.argument;
/** Dictionary of function definitions for all internal used functions. */
public class InternalFunctionDefinitions {
@@ -30,6 +34,9 @@ public class InternalFunctionDefinitions {
new BuiltInFunctionDefinition.Builder()
.name("throwException")
.kind(SCALAR)
- .outputTypeStrategy(TypeStrategies.MISSING)
+ .inputTypeStrategy(
+ sequence(
+ explicit(DataTypes.STRING()), InputTypeStrategies.TYPE_LITERAL))
+ .outputTypeStrategy(argument(1))
.build();
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala
index 5f3953bfdd7..2cbfcd563e9 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala
@@ -79,12 +79,6 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
val windowReference = translateWindowReference(children.head)
return WindowEnd(windowReference)
- case THROW_EXCEPTION =>
- assert(children.size == 2)
- return ThrowException(
- children.head.accept(this),
- fromDataTypeToTypeInfo(children(1).asInstanceOf[TypeLiteralExpression].getOutputDataType))
-
case _ =>
}
@@ -143,10 +137,6 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
assert(args.isEmpty)
CurrentRow()
- case STREAM_RECORD_TIMESTAMP =>
- assert(args.isEmpty)
- StreamRecordTimestamp()
-
case _ =>
unknownFunctionHandler()
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/call.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/call.scala
index c69df8351fa..cb2b9e79cad 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/call.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/call.scala
@@ -242,22 +242,3 @@ case class PlannerTableFunctionCall(
override def toString =
s"${tableFunction.getClass.getCanonicalName}(${parameters.mkString(", ")})"
}
-
-case class ThrowException(msg: PlannerExpression, tp: TypeInformation[_]) extends UnaryExpression {
-
- override private[flink] def resultType: TypeInformation[_] = tp
-
- override private[flink] def child: PlannerExpression = msg
-
- override private[flink] def validateInput(): ValidationResult = {
- if (child.resultType == Types.STRING) {
- ValidationSuccess
- } else {
- ValidationFailure(
- s"ThrowException operator requires String input, " +
- s"but $child is of type ${child.resultType}")
- }
- }
-
- override def toString: String = s"ThrowException($msg)"
-}