You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2019/06/07 10:11:32 UTC
[flink] branch master updated (ef150ee -> a0cf205)
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from ef150ee [hotfix][table-common] Introduce an across-catalog object identifier
new c984383 [FLINK-12254][table] Update field references to new type system
new a46bbd6 [hotfix][table] Rename to ValueLiteralExpession.getOutputDataType
new a0cf205 [hotfix][table] Rename to TypeLiteralExpession.getOutputDataType
The 3 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:
.../expressions/LocalReferenceExpression.java | 20 +++++-----
.../flink/table/operations/QueryOperationTest.java | 5 +--
.../expressions/FieldReferenceExpression.java | 20 +++++-----
.../table/expressions/TypeLiteralExpression.java | 2 +-
.../table/expressions/ValueLiteralExpression.java | 2 +-
.../flink/table/expressions/ExpressionTest.java | 3 +-
.../flink/table/expressions/RexNodeConverter.java | 8 ++--
.../codegen/agg/batch/WindowCodeGenerator.scala | 2 +-
.../stream/StreamExecGroupWindowAggregate.scala | 26 ++++++-------
.../logical/BatchLogicalWindowAggregateRule.scala | 8 ++--
.../logical/LogicalWindowAggregateRuleBase.scala | 4 +-
.../logical/StreamLogicalWindowAggregateRule.scala | 8 ++--
.../plan/rules/logical/WindowPropertiesRule.scala | 11 ++++--
.../StreamExecGroupWindowAggregateRule.scala | 4 +-
.../flink/table/plan/util/AggregateUtil.scala | 15 ++++----
.../flink/table/plan/util/WindowEmitStrategy.scala | 4 +-
.../table/sources/tsextractors/ExistingField.scala | 4 +-
.../plan/metadata/FlinkRelMdHandlerTestBase.scala | 7 +++-
.../table/expressions/ExpressionResolver.java | 6 ++-
.../expressions/lookups/FieldReferenceLookup.java | 2 +-
.../operations/AggregateOperationFactory.java | 44 ++++++++++++----------
.../operations/ProjectionOperationFactory.java | 2 +-
.../flink/table/plan/QueryOperationConverter.java | 3 +-
.../expressions/PlannerExpressionConverter.scala | 13 ++++---
.../table/operations/OperationTreeBuilder.scala | 4 +-
.../LogicalCorrelateToTemporalTableJoinRule.scala | 17 +++++----
.../VerifyNoUnresolvedExpressionsRuleTest.java | 12 +++---
.../api/batch/sql/TemporalTableJoinTest.scala | 8 ++--
.../api/stream/table/TemporalTableJoinTest.scala | 9 +++--
29 files changed, 148 insertions(+), 125 deletions(-)
[flink] 02/03: [hotfix][table] Rename to
ValueLiteralExpession.getOutputDataType
Posted by tw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a46bbd6275a0ae979712d5c9da100b60b4d14867
Author: Timo Walther <tw...@apache.org>
AuthorDate: Thu Jun 6 14:19:28 2019 +0200
[hotfix][table] Rename to ValueLiteralExpession.getOutputDataType
---
.../apache/flink/table/expressions/ValueLiteralExpression.java | 2 +-
.../java/org/apache/flink/table/expressions/RexNodeConverter.java | 4 ++--
.../flink/table/codegen/agg/batch/WindowCodeGenerator.scala | 2 +-
.../scala/org/apache/flink/table/plan/util/AggregateUtil.scala | 4 ++--
.../apache/flink/table/operations/AggregateOperationFactory.java | 8 ++++----
.../apache/flink/table/operations/ProjectionOperationFactory.java | 2 +-
.../flink/table/expressions/PlannerExpressionConverter.scala | 6 +++---
7 files changed, 14 insertions(+), 14 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ValueLiteralExpression.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ValueLiteralExpression.java
index fd2144a..6ec25fa 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ValueLiteralExpression.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ValueLiteralExpression.java
@@ -68,7 +68,7 @@ public final class ValueLiteralExpression implements Expression {
this.dataType = dataType;
}
- public DataType getDataType() {
+ public DataType getOutputDataType() {
return dataType;
}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/expressions/RexNodeConverter.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/expressions/RexNodeConverter.java
index 8b48f39..adb9014 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/expressions/RexNodeConverter.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/expressions/RexNodeConverter.java
@@ -239,7 +239,7 @@ public class RexNodeConverter implements ExpressionVisitor<RexNode> {
* This method makes the planner more lenient for new data types defined for literals.
*/
private static TypeInformation<?> getLiteralTypeInfo(ValueLiteralExpression literal) {
- final LogicalType logicalType = literal.getDataType().getLogicalType();
+ final LogicalType logicalType = literal.getOutputDataType().getLogicalType();
if (hasRoot(logicalType, DECIMAL)) {
if (literal.isNull()) {
@@ -267,7 +267,7 @@ public class RexNodeConverter implements ExpressionVisitor<RexNode> {
}
}
- return fromDataTypeToLegacyInfo(literal.getDataType());
+ return fromDataTypeToLegacyInfo(literal.getOutputDataType());
}
/**
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/WindowCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/WindowCodeGenerator.scala
index 5d870ee..1085141 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/WindowCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/WindowCodeGenerator.scala
@@ -755,7 +755,7 @@ object WindowCodeGenerator {
def isTimeIntervalLiteral(expr: Expression): Boolean = expr match {
case literal: ValueLiteralExpression if
- hasRoot(literal.getDataType.getLogicalType, INTERVAL_DAY_TIME) => true
+ hasRoot(literal.getOutputDataType.getLogicalType, INTERVAL_DAY_TIME) => true
case _ => false
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/AggregateUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/AggregateUtil.scala
index 3435034..6492306 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/AggregateUtil.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/AggregateUtil.scala
@@ -711,11 +711,11 @@ object AggregateUtil extends Enumeration {
}
def hasTimeIntervalType(intervalType: ValueLiteralExpression): Boolean = {
- hasRoot(intervalType.getDataType.getLogicalType, LogicalTypeRoot.INTERVAL_DAY_TIME)
+ hasRoot(intervalType.getOutputDataType.getLogicalType, LogicalTypeRoot.INTERVAL_DAY_TIME)
}
def hasRowIntervalType(intervalType: ValueLiteralExpression): Boolean = {
- hasRoot(intervalType.getDataType.getLogicalType, LogicalTypeRoot.BIGINT)
+ hasRoot(intervalType.getOutputDataType.getLogicalType, LogicalTypeRoot.BIGINT)
}
def toLong(literalExpr: ValueLiteralExpression): JLong =
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/AggregateOperationFactory.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/AggregateOperationFactory.java
index 603ac9d..bdc289b 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/AggregateOperationFactory.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/AggregateOperationFactory.java
@@ -298,7 +298,7 @@ public class AggregateOperationFactory {
"A tumble window expects a size value literal.");
final LogicalType timeFieldType = timeField.getOutputDataType().getLogicalType();
- final LogicalType windowSizeType = windowSize.getDataType().getLogicalType();
+ final LogicalType windowSizeType = windowSize.getOutputDataType().getLogicalType();
if (!hasRoot(windowSizeType, BIGINT) && !hasRoot(windowSizeType, INTERVAL_DAY_TIME)) {
throw new ValidationException(
@@ -323,8 +323,8 @@ public class AggregateOperationFactory {
"A sliding window expects a slide value literal.");
final LogicalType timeFieldType = timeField.getOutputDataType().getLogicalType();
- final LogicalType windowSizeType = windowSize.getDataType().getLogicalType();
- final LogicalType windowSlideType = windowSlide.getDataType().getLogicalType();
+ final LogicalType windowSizeType = windowSize.getOutputDataType().getLogicalType();
+ final LogicalType windowSlideType = windowSlide.getOutputDataType().getLogicalType();
if (!hasRoot(windowSizeType, BIGINT) && !hasRoot(windowSizeType, INTERVAL_DAY_TIME)) {
throw new ValidationException(
@@ -352,7 +352,7 @@ public class AggregateOperationFactory {
window.getGap(),
"A session window expects a gap value literal.");
- final LogicalType windowGapType = windowGap.getDataType().getLogicalType();
+ final LogicalType windowGapType = windowGap.getOutputDataType().getLogicalType();
if (!hasRoot(windowGapType, INTERVAL_DAY_TIME)) {
throw new ValidationException("A session window expects a gap literal of a day-time interval type.");
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/ProjectionOperationFactory.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/ProjectionOperationFactory.java
index c87b609..c5fd686 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/ProjectionOperationFactory.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/ProjectionOperationFactory.java
@@ -215,7 +215,7 @@ public final class ProjectionOperationFactory {
Expression child = call.getChildren().get(0);
ValueLiteralExpression key = (ValueLiteralExpression) call.getChildren().get(1);
- final LogicalType keyType = key.getDataType().getLogicalType();
+ final LogicalType keyType = key.getOutputDataType().getLogicalType();
final String keySuffix;
if (hasRoot(keyType, INTEGER)) {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala
index c89b8f8..147221a 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala
@@ -672,7 +672,7 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
}
override def visitValueLiteral(literal: ValueLiteralExpression): PlannerExpression = {
- if (hasRoot(literal.getDataType.getLogicalType, SYMBOL)) {
+ if (hasRoot(literal.getOutputDataType.getLogicalType, SYMBOL)) {
val plannerSymbol = getSymbol(literal.getValueAs(classOf[TableSymbol]).get())
return SymbolPlannerExpression(plannerSymbol)
}
@@ -691,7 +691,7 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
* This method makes the planner more lenient for new data types defined for literals.
*/
private def getLiteralTypeInfo(literal: ValueLiteralExpression): TypeInformation[_] = {
- val logicalType = literal.getDataType.getLogicalType
+ val logicalType = literal.getOutputDataType.getLogicalType
if (hasRoot(logicalType, DECIMAL)) {
if (literal.isNull) {
@@ -719,7 +719,7 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
}
}
- fromDataTypeToLegacyInfo(literal.getDataType)
+ fromDataTypeToLegacyInfo(literal.getOutputDataType)
}
private def getSymbol(symbol: TableSymbol): PlannerSymbol = symbol match {
[flink] 01/03: [FLINK-12254][table] Update field references to new
type system
Posted by tw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c9843837d529ef6f354abadb688c082f7d117999
Author: Timo Walther <tw...@apache.org>
AuthorDate: Thu Jun 6 20:58:32 2019 +0200
[FLINK-12254][table] Update field references to new type system
This closes #8655.
---
.../expressions/LocalReferenceExpression.java | 20 ++++++------
.../flink/table/operations/QueryOperationTest.java | 5 ++-
.../expressions/FieldReferenceExpression.java | 20 ++++++------
.../flink/table/expressions/ExpressionTest.java | 3 +-
.../stream/StreamExecGroupWindowAggregate.scala | 26 ++++++++--------
.../logical/BatchLogicalWindowAggregateRule.scala | 8 +++--
.../logical/LogicalWindowAggregateRuleBase.scala | 4 ++-
.../logical/StreamLogicalWindowAggregateRule.scala | 8 +++--
.../plan/rules/logical/WindowPropertiesRule.scala | 11 ++++---
.../StreamExecGroupWindowAggregateRule.scala | 4 +--
.../flink/table/plan/util/AggregateUtil.scala | 11 +++----
.../flink/table/plan/util/WindowEmitStrategy.scala | 4 +--
.../table/sources/tsextractors/ExistingField.scala | 4 +--
.../plan/metadata/FlinkRelMdHandlerTestBase.scala | 7 +++--
.../table/expressions/ExpressionResolver.java | 6 +++-
.../expressions/lookups/FieldReferenceLookup.java | 2 +-
.../operations/AggregateOperationFactory.java | 36 ++++++++++++----------
.../flink/table/plan/QueryOperationConverter.java | 3 +-
.../expressions/PlannerExpressionConverter.scala | 4 +--
.../table/operations/OperationTreeBuilder.scala | 4 +--
.../LogicalCorrelateToTemporalTableJoinRule.scala | 17 +++++-----
.../VerifyNoUnresolvedExpressionsRuleTest.java | 12 ++++----
.../api/batch/sql/TemporalTableJoinTest.scala | 8 ++---
.../api/stream/table/TemporalTableJoinTest.scala | 9 +++---
24 files changed, 129 insertions(+), 107 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/LocalReferenceExpression.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/LocalReferenceExpression.java
index d74a785..7eec58c 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/LocalReferenceExpression.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/LocalReferenceExpression.java
@@ -18,8 +18,9 @@
package org.apache.flink.table.expressions;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.table.operations.QueryOperation;
+import org.apache.flink.table.types.DataType;
import org.apache.flink.util.Preconditions;
import java.util.Collections;
@@ -31,23 +32,24 @@ import java.util.Objects;
* That entity does not come from any of the Operations input. It might be for example a group
* window in window aggregation.
*/
+@Internal
public class LocalReferenceExpression implements Expression {
private final String name;
- private final TypeInformation<?> resultType;
+ private final DataType dataType;
- public LocalReferenceExpression(String name, TypeInformation<?> resultType) {
+ public LocalReferenceExpression(String name, DataType dataType) {
this.name = Preconditions.checkNotNull(name);
- this.resultType = Preconditions.checkNotNull(resultType);
+ this.dataType = Preconditions.checkNotNull(dataType);
}
public String getName() {
return name;
}
- public TypeInformation<?> getResultType() {
- return resultType;
+ public DataType getOutputDataType() {
+ return dataType;
}
@Override
@@ -69,13 +71,13 @@ public class LocalReferenceExpression implements Expression {
return false;
}
LocalReferenceExpression that = (LocalReferenceExpression) o;
- return Objects.equals(name, that.name) &&
- Objects.equals(resultType, that.resultType);
+ return name.equals(that.name) &&
+ dataType.equals(that.dataType);
}
@Override
public int hashCode() {
- return Objects.hash(name, resultType);
+ return Objects.hash(name, dataType);
}
@Override
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java
index 9178f5b..5253165 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.operations;
-import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.expressions.BuiltInFunctionDefinitions;
@@ -43,7 +42,7 @@ public class QueryOperationTest {
TableSchema schema = TableSchema.builder().field("a", DataTypes.INT()).build();
ProjectQueryOperation tableOperation = new ProjectQueryOperation(
- Collections.singletonList(new FieldReferenceExpression("a", Types.INT, 0, 0)),
+ Collections.singletonList(new FieldReferenceExpression("a", DataTypes.INT(), 0, 0)),
new CatalogQueryOperation(
Arrays.asList("cat1", "db1", "tab1"),
schema), schema);
@@ -65,7 +64,7 @@ public class QueryOperationTest {
@Test
public void testWindowAggregationSummaryString() {
TableSchema schema = TableSchema.builder().field("a", DataTypes.INT()).build();
- FieldReferenceExpression field = new FieldReferenceExpression("a", Types.INT, 0, 0);
+ FieldReferenceExpression field = new FieldReferenceExpression("a", DataTypes.INT(), 0, 0);
WindowAggregateQueryOperation tableOperation = new WindowAggregateQueryOperation(
Collections.singletonList(field),
Collections.singletonList(new CallExpression(BuiltInFunctionDefinitions.SUM,
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/FieldReferenceExpression.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/FieldReferenceExpression.java
index 3256e20..cbedf40 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/FieldReferenceExpression.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/FieldReferenceExpression.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.expressions;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.types.DataType;
import org.apache.flink.util.Preconditions;
import java.util.Collections;
@@ -39,7 +39,7 @@ public final class FieldReferenceExpression implements Expression {
private final String name;
- private final TypeInformation<?> resultType;
+ private final DataType dataType;
/**
* index of an input the field belongs to.
@@ -54,13 +54,13 @@ public final class FieldReferenceExpression implements Expression {
public FieldReferenceExpression(
String name,
- TypeInformation<?> resultType,
+ DataType dataType,
int inputIndex,
int fieldIndex) {
Preconditions.checkArgument(inputIndex >= 0, "Index of input should be a positive number");
Preconditions.checkArgument(fieldIndex >= 0, "Index of field should be a positive number");
- this.name = Preconditions.checkNotNull(name);
- this.resultType = Preconditions.checkNotNull(resultType);
+ this.name = Preconditions.checkNotNull(name, "Field name must not be null.");
+ this.dataType = Preconditions.checkNotNull(dataType, "Field data type must not be null.");
this.inputIndex = inputIndex;
this.fieldIndex = fieldIndex;
}
@@ -69,8 +69,8 @@ public final class FieldReferenceExpression implements Expression {
return name;
}
- public TypeInformation<?> getResultType() {
- return resultType;
+ public DataType getOutputDataType() {
+ return dataType;
}
public int getInputIndex() {
@@ -100,15 +100,15 @@ public final class FieldReferenceExpression implements Expression {
return false;
}
FieldReferenceExpression that = (FieldReferenceExpression) o;
- return Objects.equals(name, that.name) &&
- Objects.equals(resultType, that.resultType) &&
+ return name.equals(that.name) &&
+ dataType.equals(that.dataType) &&
inputIndex == that.inputIndex &&
fieldIndex == that.fieldIndex;
}
@Override
public int hashCode() {
- return Objects.hash(name, resultType, inputIndex, fieldIndex);
+ return Objects.hash(name, dataType, inputIndex, fieldIndex);
}
@Override
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/expressions/ExpressionTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/expressions/ExpressionTest.java
index 68d93be..fed4ce4 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/expressions/ExpressionTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/expressions/ExpressionTest.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.expressions;
-import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.functions.ScalarFunction;
@@ -151,7 +150,7 @@ public class ExpressionTest {
new CallExpression(
EQUALS,
asList(
- new FieldReferenceExpression("field", Types.INT, 0, 0),
+ new FieldReferenceExpression("field", DataTypes.INT(), 0, 0),
new CallExpression(
new ScalarFunctionDefinition("dummy", DUMMY_FUNCTION),
singletonList(new ValueLiteralExpression(nestedValue, DataTypes.INT()))
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecGroupWindowAggregate.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecGroupWindowAggregate.scala
index 840d99c..d7ea87b 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecGroupWindowAggregate.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecGroupWindowAggregate.scala
@@ -26,7 +26,7 @@ import org.apache.flink.table.dataformat.BaseRow
import org.apache.flink.table.plan.logical._
import org.apache.flink.table.plan.nodes.exec.{ExecNode, StreamExecNode}
import org.apache.flink.table.plan.rules.physical.stream.StreamExecRetractionRules
-import org.apache.flink.table.plan.util.AggregateUtil.{isProctimeIndicatorType, hasRowIntervalType, isRowtimeIndicatorType, hasTimeIntervalType, toDuration, toLong, transformToStreamAggregateInfoList}
+import org.apache.flink.table.plan.util.AggregateUtil.{isProctimeAttribute, hasRowIntervalType, isRowtimeAttribute, hasTimeIntervalType, toDuration, toLong, transformToStreamAggregateInfoList}
import org.apache.flink.table.plan.util.{AggregateInfoList, KeySelectorUtil, RelExplainUtil, WindowEmitStrategy}
import org.apache.flink.table.`type`.TypeConverters.createInternalTypeFromTypeInfo
import org.apache.flink.table.api.window.{CountWindow, TimeWindow}
@@ -78,11 +78,11 @@ class StreamExecGroupWindowAggregate(
override def requireWatermark: Boolean = window match {
case TumblingGroupWindow(_, timeField, size)
- if isRowtimeIndicatorType(timeField.getResultType) && hasTimeIntervalType(size) => true
+ if isRowtimeAttribute(timeField) && hasTimeIntervalType(size) => true
case SlidingGroupWindow(_, timeField, size, _)
- if isRowtimeIndicatorType(timeField.getResultType) && hasTimeIntervalType(size) => true
+ if isRowtimeAttribute(timeField) && hasTimeIntervalType(size) => true
case SessionGroupWindow(_, timeField, _)
- if isRowtimeIndicatorType(timeField.getResultType) => true
+ if isRowtimeAttribute(timeField) => true
case _ => false
}
@@ -176,7 +176,7 @@ class StreamExecGroupWindowAggregate(
aggCalls,
namedProperties)
- val timeIdx = if (isRowtimeIndicatorType(window.timeAttribute.getResultType)) {
+ val timeIdx = if (isRowtimeAttribute(window.timeAttribute)) {
if (inputTimeFieldIndex < 0) {
throw new TableException(
"Group window aggregate must defined on a time attribute, " +
@@ -307,15 +307,15 @@ class StreamExecGroupWindowAggregate(
val newBuilder = window match {
case TumblingGroupWindow(_, timeField, size)
- if isProctimeIndicatorType(timeField.getResultType) && hasTimeIntervalType(size) =>
+ if isProctimeAttribute(timeField) && hasTimeIntervalType(size) =>
builder.tumble(toDuration(size), timeZoneOffset).withProcessingTime()
case TumblingGroupWindow(_, timeField, size)
- if isRowtimeIndicatorType(timeField.getResultType) && hasTimeIntervalType(size) =>
+ if isRowtimeAttribute(timeField) && hasTimeIntervalType(size) =>
builder.tumble(toDuration(size), timeZoneOffset).withEventTime(timeIdx)
case TumblingGroupWindow(_, timeField, size)
- if isProctimeIndicatorType(timeField.getResultType) && hasRowIntervalType(size) =>
+ if isProctimeAttribute(timeField) && hasRowIntervalType(size) =>
builder.countWindow(toLong(size))
case TumblingGroupWindow(_, _, _) =>
@@ -326,17 +326,17 @@ class StreamExecGroupWindowAggregate(
"Event-time grouping windows on row intervals are currently not supported.")
case SlidingGroupWindow(_, timeField, size, slide)
- if isProctimeIndicatorType(timeField.getResultType) && hasTimeIntervalType(size) =>
+ if isProctimeAttribute(timeField) && hasTimeIntervalType(size) =>
builder.sliding(toDuration(size), toDuration(slide), timeZoneOffset)
.withProcessingTime()
case SlidingGroupWindow(_, timeField, size, slide)
- if isRowtimeIndicatorType(timeField.getResultType) && hasTimeIntervalType(size) =>
+ if isRowtimeAttribute(timeField) && hasTimeIntervalType(size) =>
builder.sliding(toDuration(size), toDuration(slide), timeZoneOffset)
.withEventTime(timeIdx)
case SlidingGroupWindow(_, timeField, size, slide)
- if isProctimeIndicatorType(timeField.getResultType) && hasRowIntervalType(size) =>
+ if isProctimeAttribute(timeField) && hasRowIntervalType(size) =>
builder.countWindow(toLong(size), toLong(slide))
case SlidingGroupWindow(_, _, _, _) =>
@@ -347,11 +347,11 @@ class StreamExecGroupWindowAggregate(
"Event-time grouping windows on row intervals are currently not supported.")
case SessionGroupWindow(_, timeField, gap)
- if isProctimeIndicatorType(timeField.getResultType) =>
+ if isProctimeAttribute(timeField) =>
builder.session(toDuration(gap)).withProcessingTime()
case SessionGroupWindow(_, timeField, gap)
- if isRowtimeIndicatorType(timeField.getResultType) =>
+ if isRowtimeAttribute(timeField) =>
builder.session(toDuration(gap)).withEventTime(timeIdx)
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/BatchLogicalWindowAggregateRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/BatchLogicalWindowAggregateRule.scala
index 30057ff..c3fb0f5 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/BatchLogicalWindowAggregateRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/BatchLogicalWindowAggregateRule.scala
@@ -23,10 +23,10 @@ import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.calcite.FlinkTypeFactory.toInternalType
import org.apache.flink.table.expressions.FieldReferenceExpression
import org.apache.flink.table.plan.nodes.calcite.LogicalWindowAggregate
-
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.logical.{LogicalAggregate, LogicalProject}
import org.apache.calcite.rex._
+import org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType
/**
* Planner rule that transforms simple [[LogicalAggregate]] on a [[LogicalProject]]
@@ -59,7 +59,8 @@ class BatchLogicalWindowAggregateRule
FlinkTypeFactory.isTimeIndicatorType(c.getType) =>
new FieldReferenceExpression(
rowType.getFieldList.get(windowExprIdx).getName,
- createExternalTypeInfoFromInternalType(toInternalType(c.getType)),
+ fromLegacyInfoToDataType(
+ createExternalTypeInfoFromInternalType(toInternalType(c.getType))),
0, // only one input, should always be 0
windowExprIdx)
case ref: RexInputRef =>
@@ -68,7 +69,8 @@ class BatchLogicalWindowAggregateRule
val fieldType = rowType.getFieldList.get(ref.getIndex).getType
new FieldReferenceExpression(
fieldName,
- createExternalTypeInfoFromInternalType(toInternalType(fieldType)),
+ fromLegacyInfoToDataType(
+ createExternalTypeInfoFromInternalType(toInternalType(fieldType))),
0, // only one input, should always be 0
ref.getIndex)
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalWindowAggregateRuleBase.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalWindowAggregateRuleBase.scala
index 5cc925f..ddd7677 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalWindowAggregateRuleBase.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalWindowAggregateRuleBase.scala
@@ -36,6 +36,7 @@ import org.apache.flink.table.expressions.{FieldReferenceExpression, WindowRefer
import org.apache.flink.table.functions.sql.FlinkSqlOperatorTable
import org.apache.flink.table.plan.logical.{LogicalWindow, SessionGroupWindow, SlidingGroupWindow, TumblingGroupWindow}
import org.apache.flink.table.plan.nodes.calcite.LogicalWindowAggregate
+import org.apache.flink.table.types.utils.TypeConversions.fromDataTypeToLegacyInfo
import _root_.scala.collection.JavaConversions._
@@ -174,7 +175,8 @@ abstract class LogicalWindowAggregateRuleBase(description: String)
}
val timeField = getTimeFieldReference(windowExpr.getOperands.get(0), windowExprIdx, rowType)
- val resultType = Some(createInternalTypeFromTypeInfo(timeField.getResultType))
+ val resultType =
+ Some(createInternalTypeFromTypeInfo(fromDataTypeToLegacyInfo(timeField.getOutputDataType)))
val windowRef = WindowReference("w$", resultType)
windowExpr.getOperator match {
case FlinkSqlOperatorTable.TUMBLE =>
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/StreamLogicalWindowAggregateRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/StreamLogicalWindowAggregateRule.scala
index de87c40..274b9c5 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/StreamLogicalWindowAggregateRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/StreamLogicalWindowAggregateRule.scala
@@ -24,11 +24,11 @@ import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.calcite.FlinkTypeFactory.toInternalType
import org.apache.flink.table.expressions.FieldReferenceExpression
import org.apache.flink.table.plan.nodes.calcite.LogicalWindowAggregate
-
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.logical.{LogicalAggregate, LogicalProject}
import org.apache.calcite.rex._
import org.apache.calcite.sql.`type`.SqlTypeName
+import org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType
/**
* Planner rule that transforms simple [[LogicalAggregate]] on a [[LogicalProject]]
@@ -70,13 +70,15 @@ class StreamLogicalWindowAggregateRule
FlinkTypeFactory.isTimeIndicatorType(c.getType) =>
new FieldReferenceExpression(
rowType.getFieldList.get(windowExprIdx).getName,
- createExternalTypeInfoFromInternalType(toInternalType(c.getType)),
+ fromLegacyInfoToDataType(
+ createExternalTypeInfoFromInternalType(toInternalType(c.getType))),
0, // only one input, should always be 0
windowExprIdx)
case v: RexInputRef if FlinkTypeFactory.isTimeIndicatorType(v.getType) =>
new FieldReferenceExpression(
rowType.getFieldList.get(v.getIndex).getName,
- createExternalTypeInfoFromInternalType(toInternalType(v.getType)),
+ fromLegacyInfoToDataType(
+ createExternalTypeInfoFromInternalType(toInternalType(v.getType))),
0, // only one input, should always be 0
v.getIndex)
case _ =>
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/WindowPropertiesRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/WindowPropertiesRule.scala
index 398ec32..af6878f 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/WindowPropertiesRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/WindowPropertiesRule.scala
@@ -25,13 +25,14 @@ import org.apache.flink.table.functions.sql.FlinkSqlOperatorTable
import org.apache.flink.table.plan.logical.LogicalWindow
import org.apache.flink.table.plan.nodes.calcite.LogicalWindowAggregate
import org.apache.flink.table.plan.util.AggregateUtil
-
import org.apache.calcite.plan.RelOptRule._
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.logical.{LogicalFilter, LogicalProject}
import org.apache.calcite.rex.{RexCall, RexNode}
import org.apache.calcite.tools.RelBuilder
+import org.apache.flink.table.types.logical.LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks.hasRoot
import scala.collection.JavaConversions._
@@ -145,11 +146,13 @@ object WindowPropertiesRules {
}
private def getWindowType(window: LogicalWindow): Symbol = {
- if (AggregateUtil.isRowtimeIndicatorType(window.timeAttribute.getResultType)) {
+ if (AggregateUtil.isRowtimeAttribute(window.timeAttribute)) {
'streamRowtime
- } else if (AggregateUtil.isProctimeIndicatorType(window.timeAttribute.getResultType)) {
+ } else if (AggregateUtil.isProctimeAttribute(window.timeAttribute)) {
'streamProctime
- } else if (Types.SQL_TIMESTAMP == window.timeAttribute.getResultType) {
+ } else if (hasRoot(
+ window.timeAttribute.getOutputDataType.getLogicalType,
+ TIMESTAMP_WITHOUT_TIME_ZONE)) {
'batchRowtime
} else {
throw new TableException("Unknown window type encountered. Please report this bug.")
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/stream/StreamExecGroupWindowAggregateRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/stream/StreamExecGroupWindowAggregateRule.scala
index d4ee8c0..d100f80 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/stream/StreamExecGroupWindowAggregateRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/stream/StreamExecGroupWindowAggregateRule.scala
@@ -24,7 +24,7 @@ import org.apache.flink.table.plan.`trait`.FlinkRelDistribution
import org.apache.flink.table.plan.nodes.FlinkConventions
import org.apache.flink.table.plan.nodes.logical.FlinkLogicalWindowAggregate
import org.apache.flink.table.plan.nodes.physical.stream.StreamExecGroupWindowAggregate
-import org.apache.flink.table.plan.util.AggregateUtil.{isRowtimeIndicatorType, timeFieldIndex}
+import org.apache.flink.table.plan.util.AggregateUtil.{isRowtimeAttribute, timeFieldIndex}
import org.apache.flink.table.plan.util.WindowEmitStrategy
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
@@ -76,7 +76,7 @@ class StreamExecGroupWindowAggregateRule
val emitStrategy = WindowEmitStrategy(config, agg.getWindow)
val timeField = agg.getWindow.timeAttribute
- val inputTimestampIndex = if (isRowtimeIndicatorType(timeField.getResultType)) {
+ val inputTimestampIndex = if (isRowtimeAttribute(timeField)) {
timeFieldIndex(inputRowType, relBuilderFactory.create(cluster, null), timeField)
} else {
-1
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/AggregateUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/AggregateUtil.scala
index 061c5ab..3435034 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/AggregateUtil.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/AggregateUtil.scala
@@ -47,6 +47,7 @@ import org.apache.flink.table.functions.{AggregateFunction, UserDefinedFunction}
import org.apache.flink.table.plan.`trait`.RelModifiedMonotonicity
import org.apache.flink.table.runtime.bundle.trigger.CountBundleTrigger
import org.apache.flink.table.types.logical.LogicalTypeRoot
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks
import org.apache.flink.table.types.logical.utils.LogicalTypeChecks.hasRoot
import org.apache.flink.table.typeutils._
@@ -701,14 +702,12 @@ object AggregateUtil extends Enumeration {
(propPos._1, propPos._2, propPos._3)
}
- def isRowtimeIndicatorType(fieldType: TypeInformation[_]): Boolean = fieldType match {
- case typeInfo: TimeIndicatorTypeInfo => typeInfo.isEventTime
- case _ => false
+ def isRowtimeAttribute(field: FieldReferenceExpression): Boolean = {
+ LogicalTypeChecks.isRowtimeAttribute(field.getOutputDataType.getLogicalType)
}
- def isProctimeIndicatorType(fieldType: TypeInformation[_]): Boolean = fieldType match {
- case typeInfo: TimeIndicatorTypeInfo => !typeInfo.isEventTime
- case _ => false
+ def isProctimeAttribute(field: FieldReferenceExpression): Boolean = {
+ LogicalTypeChecks.isProctimeAttribute(field.getOutputDataType.getLogicalType)
}
def hasTimeIntervalType(intervalType: ValueLiteralExpression): Boolean = {
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/WindowEmitStrategy.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/WindowEmitStrategy.scala
index f13fa0e..6c88470 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/WindowEmitStrategy.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/WindowEmitStrategy.scala
@@ -20,7 +20,7 @@ package org.apache.flink.table.plan.util
import org.apache.flink.table.api.window.TimeWindow
import org.apache.flink.table.api.{TableConfig, TableException}
import org.apache.flink.table.plan.logical.{LogicalWindow, SessionGroupWindow}
-import org.apache.flink.table.plan.util.AggregateUtil.isRowtimeIndicatorType
+import org.apache.flink.table.plan.util.AggregateUtil.isRowtimeAttribute
import org.apache.flink.table.runtime.window.triggers._
import java.time.Duration
@@ -115,7 +115,7 @@ class WindowEmitStrategy(
object WindowEmitStrategy {
def apply(tableConfig: TableConfig, window: LogicalWindow): WindowEmitStrategy = {
- val isEventTime = isRowtimeIndicatorType(window.timeAttribute.getResultType)
+ val isEventTime = isRowtimeAttribute(window.timeAttribute)
val isSessionWindow = window.isInstanceOf[SessionGroupWindow]
val allowLateness = if (isSessionWindow) {
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sources/tsextractors/ExistingField.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sources/tsextractors/ExistingField.scala
index 365b30b..9d3e52b 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sources/tsextractors/ExistingField.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sources/tsextractors/ExistingField.scala
@@ -21,12 +21,10 @@ package org.apache.flink.table.sources.tsextractors
import java.util
import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.table.`type`.DecimalType
import org.apache.flink.table.api.{Types, ValidationException}
import org.apache.flink.table.descriptors.Rowtime
import org.apache.flink.table.expressions._
import org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType
-import org.apache.flink.table.typeutils.DecimalTypeInfo
import scala.collection.JavaConversions._
@@ -66,7 +64,7 @@ final class ExistingField(val field: String) extends TimestampExtractor {
val fieldReferenceExpr = new FieldReferenceExpression(
fieldAccess.name,
- fieldAccess.resultType,
+ fromLegacyInfoToDataType(fieldAccess.resultType),
0,
fieldAccess.fieldIndex)
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/metadata/FlinkRelMdHandlerTestBase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/metadata/FlinkRelMdHandlerTestBase.scala
index 5f82590..a78aa69 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/metadata/FlinkRelMdHandlerTestBase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/metadata/FlinkRelMdHandlerTestBase.scala
@@ -38,7 +38,8 @@ import org.apache.calcite.sql.fun.{SqlCountAggFunction, SqlStdOperatorTable}
import org.apache.calcite.sql.parser.SqlParserPos
import org.apache.calcite.tools.FrameworkConfig
import org.apache.calcite.util.{DateString, ImmutableBitSet, TimeString, TimestampString}
-import org.apache.flink.table.`type`.{InternalType, InternalTypes, TypeConverters}
+import org.apache.flink.table.`type`.TypeConverters.createExternalTypeInfoFromInternalType
+import org.apache.flink.table.`type`.{InternalType, InternalTypes}
import org.apache.flink.table.api.{TableConfig, TableException}
import org.apache.flink.table.calcite.FlinkRelBuilder.NamedWindowProperty
import org.apache.flink.table.calcite.{FlinkCalciteCatalogReader, FlinkRelBuilder, FlinkTypeFactory}
@@ -59,6 +60,7 @@ import org.apache.flink.table.plan.schema.FlinkRelOptTable
import org.apache.flink.table.plan.util.AggregateUtil.transformToStreamAggregateInfoList
import org.apache.flink.table.plan.util._
import org.apache.flink.table.runtime.rank.{ConstantRankRange, RankType, VariableRankRange}
+import org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType
import org.apache.flink.table.util.CountAggFunction
import org.junit.{Before, BeforeClass}
@@ -951,7 +953,8 @@ class FlinkRelMdHandlerTestBase {
windowRef,
new FieldReferenceExpression(
"rowtime",
- TypeConverters.createExternalTypeInfoFromInternalType(InternalTypes.ROWTIME_INDICATOR),
+ fromLegacyInfoToDataType(
+ createExternalTypeInfoFromInternalType(InternalTypes.ROWTIME_INDICATOR)),
0,
4),
intervalOfMillis(900000)
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 a97d5c4..125a88b 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
@@ -53,6 +53,8 @@ import java.util.stream.Collectors;
import scala.Some;
+import static org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType;
+
/**
* Tries to resolve all unresolved expressions such as {@link UnresolvedReferenceExpression}
* or calls such as {@link BuiltInFunctionDefinitions#OVER}.
@@ -213,7 +215,9 @@ public class ExpressionResolver {
.accept(bridgeConverter)
.resultType();
- localReferences.put(windowName, new LocalReferenceExpression(windowName, windowType));
+ localReferences.put(
+ windowName,
+ new LocalReferenceExpression(windowName, fromLegacyInfoToDataType(windowType)));
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/lookups/FieldReferenceLookup.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/lookups/FieldReferenceLookup.java
index 5c69c06..9ee13e6 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/lookups/FieldReferenceLookup.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/expressions/lookups/FieldReferenceLookup.java
@@ -90,7 +90,7 @@ public class FieldReferenceLookup {
return IntStream.range(0, tableSchema.getFieldCount())
.mapToObj(i -> new FieldReferenceExpression(
tableSchema.getFieldName(i).get(),
- tableSchema.getFieldType(i).get(),
+ tableSchema.getFieldDataType(i).get(),
inputIdx,
i))
.collect(Collectors.toMap(
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/AggregateOperationFactory.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/AggregateOperationFactory.java
index ed4fa4f..603ac9d 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/AggregateOperationFactory.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/AggregateOperationFactory.java
@@ -70,7 +70,10 @@ import static org.apache.flink.table.operations.WindowAggregateQueryOperation.Re
import static org.apache.flink.table.operations.WindowAggregateQueryOperation.ResolvedGroupWindow.WindowType.TUMBLE;
import static org.apache.flink.table.types.logical.LogicalTypeRoot.BIGINT;
import static org.apache.flink.table.types.logical.LogicalTypeRoot.INTERVAL_DAY_TIME;
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE;
import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.hasRoot;
+import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isRowtimeAttribute;
+import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isTimeAttribute;
/**
* Utility class for creating a valid {@link AggregateQueryOperation} or {@link WindowAggregateQueryOperation}.
@@ -257,12 +260,15 @@ public class AggregateOperationFactory {
}
FieldReferenceExpression timeField = (FieldReferenceExpression) timeFieldExpr;
- validateTimeAttributeType(timeField);
+
+ final LogicalType timeFieldType = timeField.getOutputDataType().getLogicalType();
+
+ validateTimeAttributeType(timeFieldType);
+
return timeField;
}
- private void validateTimeAttributeType(FieldReferenceExpression timeField) {
- TypeInformation<?> timeFieldType = timeField.getResultType();
+ private void validateTimeAttributeType(LogicalType timeFieldType) {
if (isStreaming) {
validateStreamTimeAttribute(timeFieldType);
} else {
@@ -270,15 +276,15 @@ public class AggregateOperationFactory {
}
}
- private void validateBatchTimeAttribute(TypeInformation<?> timeFieldType) {
- if (!(timeFieldType instanceof SqlTimeTypeInfo || timeFieldType == LONG_TYPE_INFO)) {
+ private void validateBatchTimeAttribute(LogicalType timeFieldType) {
+ if (!(hasRoot(timeFieldType, TIMESTAMP_WITHOUT_TIME_ZONE) || hasRoot(timeFieldType, BIGINT))) {
throw new ValidationException("A group window expects a time attribute for grouping " +
"in a batch environment.");
}
}
- private void validateStreamTimeAttribute(TypeInformation<?> timeFieldType) {
- if (!(timeFieldType instanceof TimeIndicatorTypeInfo)) {
+ private void validateStreamTimeAttribute(LogicalType timeFieldType) {
+ if (!hasRoot(timeFieldType, TIMESTAMP_WITHOUT_TIME_ZONE) || !isTimeAttribute(timeFieldType)) {
throw new ValidationException("A group window expects a time attribute for grouping " +
"in a stream environment.");
}
@@ -291,6 +297,7 @@ public class AggregateOperationFactory {
ValueLiteralExpression windowSize = getAsValueLiteral(window.getSize(),
"A tumble window expects a size value literal.");
+ final LogicalType timeFieldType = timeField.getOutputDataType().getLogicalType();
final LogicalType windowSizeType = windowSize.getDataType().getLogicalType();
if (!hasRoot(windowSizeType, BIGINT) && !hasRoot(windowSizeType, INTERVAL_DAY_TIME)) {
@@ -298,7 +305,7 @@ public class AggregateOperationFactory {
"Tumbling window expects a size literal of a day-time interval or BIGINT type.");
}
- validateWindowIntervalType(timeField, windowSizeType);
+ validateWindowIntervalType(timeFieldType, windowSizeType);
return ResolvedGroupWindow.tumblingWindow(
windowName,
@@ -315,6 +322,7 @@ public class AggregateOperationFactory {
ValueLiteralExpression windowSlide = getAsValueLiteral(window.getSlide(),
"A sliding window expects a slide value literal.");
+ final LogicalType timeFieldType = timeField.getOutputDataType().getLogicalType();
final LogicalType windowSizeType = windowSize.getDataType().getLogicalType();
final LogicalType windowSlideType = windowSlide.getDataType().getLogicalType();
@@ -327,7 +335,7 @@ public class AggregateOperationFactory {
throw new ValidationException("A sliding window expects the same type of size and slide.");
}
- validateWindowIntervalType(timeField, windowSizeType);
+ validateWindowIntervalType(timeFieldType, windowSizeType);
return ResolvedGroupWindow.slidingWindow(
windowName,
@@ -356,8 +364,9 @@ public class AggregateOperationFactory {
windowGap);
}
- private void validateWindowIntervalType(FieldReferenceExpression timeField, LogicalType intervalType) {
- if (isRowTimeIndicator(timeField) && hasRoot(intervalType, BIGINT)) {
+ private void validateWindowIntervalType(LogicalType timeFieldType, LogicalType intervalType) {
+ if (hasRoot(intervalType, TIMESTAMP_WITHOUT_TIME_ZONE) && isRowtimeAttribute(timeFieldType) &&
+ hasRoot(intervalType, BIGINT)) {
// unsupported row intervals on event-time
throw new ValidationException(
"Event-time grouping windows on row intervals in a stream environment " +
@@ -372,11 +381,6 @@ public class AggregateOperationFactory {
return (ValueLiteralExpression) expression;
}
- private boolean isRowTimeIndicator(FieldReferenceExpression field) {
- return field.getResultType() instanceof TimeIndicatorTypeInfo &&
- ((TimeIndicatorTypeInfo) field.getResultType()).isEventTime();
- }
-
private void validateWindowProperties(List<Expression> windowProperties, ResolvedGroupWindow window) {
if (!windowProperties.isEmpty()) {
if (window.getType() == TUMBLE || window.getType() == SLIDE) {
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/plan/QueryOperationConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/plan/QueryOperationConverter.java
index bfcd468..3c3eff2 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/plan/QueryOperationConverter.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/plan/QueryOperationConverter.java
@@ -95,6 +95,7 @@ import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.AS;
import static org.apache.flink.table.expressions.ExpressionUtils.extractValue;
import static org.apache.flink.table.expressions.ExpressionUtils.isFunctionOfType;
import static org.apache.flink.table.expressions.FunctionDefinition.Type.AGGREGATE_FUNCTION;
+import static org.apache.flink.table.types.utils.TypeConversions.fromDataTypeToLegacyInfo;
/**
* Converter from Flink's specific relational representation: {@link QueryOperation} to Calcite's specific relational
@@ -337,7 +338,7 @@ public class QueryOperationConverter extends QueryOperationDefaultVisitor<RelNod
}
private LogicalWindow toLogicalWindow(ResolvedGroupWindow window) {
- TypeInformation<?> windowType = window.getTimeAttribute().getResultType();
+ TypeInformation<?> windowType = fromDataTypeToLegacyInfo(window.getTimeAttribute().getOutputDataType());
WindowReference windowReference = new WindowReference(window.getAlias(), new Some<>(windowType));
switch (window.getType()) {
case SLIDE:
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala
index 24d4ad5..c89b8f8 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala
@@ -756,7 +756,7 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
override def visitFieldReference(fieldReference: FieldReferenceExpression): PlannerExpression = {
PlannerResolvedFieldReference(
fieldReference.getName,
- fieldReference.getResultType)
+ fromDataTypeToLegacyInfo(fieldReference.getOutputDataType))
}
override def visitUnresolvedReference(fieldReference: UnresolvedReferenceExpression)
@@ -804,7 +804,7 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
private def translateWindowReference(reference: Expression): PlannerExpression = reference match {
case expr : LocalReferenceExpression =>
- WindowReference(expr.getName, Some(expr.getResultType))
+ WindowReference(expr.getName, Some(fromDataTypeToLegacyInfo(expr.getOutputDataType)))
//just because how the datastream is converted to table
case expr: UnresolvedReferenceExpression =>
UnresolvedFieldReference(expr.getName)
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 7e9374f..17e9791 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
@@ -282,7 +282,7 @@ class OperationTreeBuilder(private val tableEnv: TableEnvImpl) {
.withLocalReferences(
new LocalReferenceExpression(
resolvedWindow.getAlias,
- resolvedWindow.getTimeAttribute.getResultType))
+ resolvedWindow.getTimeAttribute.getOutputDataType))
.build
val convertedGroupings = resolverWithWindowReferences.resolve(groupingExpressions)
@@ -322,7 +322,7 @@ class OperationTreeBuilder(private val tableEnv: TableEnvImpl) {
.withLocalReferences(
new LocalReferenceExpression(
resolvedWindow.getAlias,
- resolvedWindow.getTimeAttribute.getResultType))
+ resolvedWindow.getTimeAttribute.getOutputDataType))
.build
val convertedGroupings = resolverWithWindowReferences.resolve(newGroupingExpressions)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToTemporalTableJoinRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToTemporalTableJoinRule.scala
index faba822..72e6290 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToTemporalTableJoinRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalCorrelateToTemporalTableJoinRule.scala
@@ -24,15 +24,16 @@ import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.core.TableFunctionScan
import org.apache.calcite.rel.logical.LogicalCorrelate
import org.apache.calcite.rex._
-import org.apache.flink.table.api.{Types, ValidationException}
+import org.apache.flink.table.api.ValidationException
import org.apache.flink.table.calcite.FlinkRelBuilder
-import org.apache.flink.table.calcite.FlinkTypeFactory.{isProctimeIndicatorType, isTimeIndicatorType}
import org.apache.flink.table.expressions.{FieldReferenceExpression, _}
import org.apache.flink.table.functions.utils.TableSqlFunction
import org.apache.flink.table.functions.{TemporalTableFunction, TemporalTableFunctionImpl}
import org.apache.flink.table.operations.QueryOperation
import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin
import org.apache.flink.table.plan.util.RexDefaultVisitor
+import org.apache.flink.table.types.logical.LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks.{hasRoot, isProctimeAttribute}
import org.apache.flink.util.Preconditions.checkState
class LogicalCorrelateToTemporalTableJoinRule
@@ -46,15 +47,18 @@ class LogicalCorrelateToTemporalTableJoinRule
private def extractNameFromTimeAttribute(timeAttribute: Expression): String = {
timeAttribute match {
case f : FieldReferenceExpression
- if f.getResultType == Types.LONG ||
- f.getResultType == Types.SQL_TIMESTAMP ||
- isTimeIndicatorType(f.getResultType) =>
+ if hasRoot(f.getOutputDataType.getLogicalType, TIMESTAMP_WITHOUT_TIME_ZONE) =>
f.getName
case _ => throw new ValidationException(
s"Invalid timeAttribute [$timeAttribute] in TemporalTableFunction")
}
}
+ private def isProctimeReference(temporalTableFunction: TemporalTableFunctionImpl): Boolean = {
+ val fieldRef = temporalTableFunction.getTimeAttribute.asInstanceOf[FieldReferenceExpression]
+ isProctimeAttribute(fieldRef.getOutputDataType.getLogicalType)
+ }
+
private def extractNameFromPrimaryKeyAttribute(expression: Expression): String = {
expression match {
case f: FieldReferenceExpression =>
@@ -100,8 +104,7 @@ class LogicalCorrelateToTemporalTableJoinRule
extractNameFromPrimaryKeyAttribute(rightTemporalTableFunction.getPrimaryKey))
relBuilder.push(
- if (isProctimeIndicatorType(rightTemporalTableFunction.getTimeAttribute
- .asInstanceOf[FieldReferenceExpression].getResultType)) {
+ if (isProctimeReference(rightTemporalTableFunction)) {
LogicalTemporalTableJoin.createProctime(
rexBuilder,
cluster,
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/expressions/rules/VerifyNoUnresolvedExpressionsRuleTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/expressions/rules/VerifyNoUnresolvedExpressionsRuleTest.java
index 8b043b7..e4f7b59 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/expressions/rules/VerifyNoUnresolvedExpressionsRuleTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/expressions/rules/VerifyNoUnresolvedExpressionsRuleTest.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.expressions.rules;
-import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.FieldReferenceExpression;
@@ -54,7 +54,7 @@ public class VerifyNoUnresolvedExpressionsRuleTest {
public void testUnresolvedReferenceIsCatched() {
List<Expression> expressions = asList(
unresolvedRef("field"),
- new FieldReferenceExpression("resolvedField", Types.INT, 0, 0));
+ new FieldReferenceExpression("resolvedField", DataTypes.INT(), 0, 0));
resolverRule.apply(expressions, resolutionContext);
}
@@ -62,14 +62,14 @@ public class VerifyNoUnresolvedExpressionsRuleTest {
public void testNestedUnresolvedReferenceIsCatched() {
List<Expression> expressions = asList(
call(AS, unresolvedRef("field"), valueLiteral("fieldAlias")),
- new FieldReferenceExpression("resolvedField", Types.INT, 0, 0));
+ new FieldReferenceExpression("resolvedField", DataTypes.INT(), 0, 0));
resolverRule.apply(expressions, resolutionContext);
}
@Test(expected = TableException.class)
public void testFlattenCallIsCatched() {
List<Expression> expressions = singletonList(
- call(FLATTEN, new FieldReferenceExpression("resolvedField", Types.INT, 0, 0))
+ call(FLATTEN, new FieldReferenceExpression("resolvedField", DataTypes.INT(), 0, 0))
);
resolverRule.apply(expressions, resolutionContext);
}
@@ -78,7 +78,7 @@ public class VerifyNoUnresolvedExpressionsRuleTest {
public void testUnresolvedOverWindowIsCatched() {
List<Expression> expressions = singletonList(
call(OVER,
- call(COUNT, new FieldReferenceExpression("resolvedField", Types.INT, 0, 0)),
+ call(COUNT, new FieldReferenceExpression("resolvedField", DataTypes.INT(), 0, 0)),
new UnresolvedReferenceExpression("w")
)
);
@@ -88,7 +88,7 @@ public class VerifyNoUnresolvedExpressionsRuleTest {
@Test(expected = TableException.class)
public void testLookupCallIsCatched() {
List<Expression> expressions = singletonList(
- lookupCall("unresolvedCall", new FieldReferenceExpression("resolvedField", Types.INT, 0, 0))
+ lookupCall("unresolvedCall", new FieldReferenceExpression("resolvedField", DataTypes.INT(), 0, 0))
);
resolverRule.apply(expressions, resolutionContext);
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/sql/TemporalTableJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/sql/TemporalTableJoinTest.scala
index cca733b..bed8a5a 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/sql/TemporalTableJoinTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/sql/TemporalTableJoinTest.scala
@@ -50,7 +50,7 @@ class TemporalTableJoinTest extends TableTestBase {
"o_amount * rate as rate " +
"FROM Orders AS o, " +
"LATERAL TABLE (Rates(o_rowtime)) AS r " +
- "WHERE currency = o_currency";
+ "WHERE currency = o_currency"
util.printSql(sqlQuery)
}
@@ -81,7 +81,7 @@ class TemporalTableJoinTest extends TableTestBase {
"LATERAL TABLE (Rates(o_rowtime)) AS r " +
"WHERE currency = o_currency OR secondary_key = o_secondary_key), " +
"Table3 " +
- "WHERE t3_secondary_key = secondary_key";
+ "WHERE t3_secondary_key = secondary_key"
util.printSql(sqlQuery)
}
@@ -95,7 +95,7 @@ class TemporalTableJoinTest extends TableTestBase {
"o_amount * rate as rate " +
"FROM Orders AS o, " +
"LATERAL TABLE (Rates(TIMESTAMP '2016-06-27 10:10:42.123')) AS r " +
- "WHERE currency = o_currency";
+ "WHERE currency = o_currency"
util.printSql(sqlQuery)
}
@@ -105,7 +105,7 @@ class TemporalTableJoinTest extends TableTestBase {
expectedException.expect(classOf[TableException])
expectedException.expectMessage(startsWith("Cannot generate a valid execution plan"))
- val sqlQuery = "SELECT * FROM LATERAL TABLE (Rates(TIMESTAMP '2016-06-27 10:10:42.123'))";
+ val sqlQuery = "SELECT * FROM LATERAL TABLE (Rates(TIMESTAMP '2016-06-27 10:10:42.123'))"
util.printSql(sqlQuery)
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/TemporalTableJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/TemporalTableJoinTest.scala
index 7665192..131e6e0 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/TemporalTableJoinTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/TemporalTableJoinTest.scala
@@ -22,10 +22,11 @@ import java.sql.Timestamp
import org.apache.flink.api.scala._
import org.apache.flink.table.api.scala._
-import org.apache.flink.table.api.{TableSchema, Types, ValidationException}
+import org.apache.flink.table.api.{DataTypes, TableSchema, ValidationException}
import org.apache.flink.table.expressions.{Expression, FieldReferenceExpression}
import org.apache.flink.table.functions.{TemporalTableFunction, TemporalTableFunctionImpl}
import org.apache.flink.table.plan.logical.rel.LogicalTemporalTableJoin._
+import org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType
import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo.{PROCTIME_INDICATOR, ROWTIME_INDICATOR}
import org.apache.flink.table.utils.TableTestUtil._
import org.apache.flink.table.utils._
@@ -197,14 +198,14 @@ class TemporalTableJoinTest extends TableTestBase {
proctime: Boolean = false): Unit = {
val rates = inputRates.asInstanceOf[TemporalTableFunctionImpl]
assertThat(rates.getPrimaryKey,
- equalTo[Expression](new FieldReferenceExpression("currency", Types.STRING, 0, 0)))
+ equalTo[Expression](new FieldReferenceExpression("currency", DataTypes.STRING(), 0, 0)))
val (timeFieldName, timeFieldType) =
if (proctime) {
- ("proctime", PROCTIME_INDICATOR)
+ ("proctime", fromLegacyInfoToDataType(PROCTIME_INDICATOR))
}
else {
- ("rowtime", ROWTIME_INDICATOR)
+ ("rowtime", fromLegacyInfoToDataType(ROWTIME_INDICATOR))
}
assertThat(rates.getTimeAttribute,
[flink] 03/03: [hotfix][table] Rename to
TypeLiteralExpession.getOutputDataType
Posted by tw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a0cf2054cc4ab9ea2a6cce3f60ec0b2dc8a59b27
Author: Timo Walther <tw...@apache.org>
AuthorDate: Thu Jun 6 14:20:09 2019 +0200
[hotfix][table] Rename to TypeLiteralExpession.getOutputDataType
---
.../org/apache/flink/table/expressions/TypeLiteralExpression.java | 2 +-
.../java/org/apache/flink/table/expressions/RexNodeConverter.java | 4 ++--
.../apache/flink/table/expressions/PlannerExpressionConverter.scala | 3 ++-
3 files changed, 5 insertions(+), 4 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/TypeLiteralExpression.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/TypeLiteralExpression.java
index c60c482..3343bfe 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/TypeLiteralExpression.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/TypeLiteralExpression.java
@@ -42,7 +42,7 @@ public final class TypeLiteralExpression implements Expression {
this.dataType = Preconditions.checkNotNull(dataType, "Data type must not be null.");
}
- public DataType getDataType() {
+ public DataType getOutputDataType() {
return dataType;
}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/expressions/RexNodeConverter.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/expressions/RexNodeConverter.java
index adb9014..1aea244 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/expressions/RexNodeConverter.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/expressions/RexNodeConverter.java
@@ -104,7 +104,7 @@ public class RexNodeConverter implements ExpressionVisitor<RexNode> {
return relBuilder.getRexBuilder().makeAbstractCast(
typeFactory.createTypeFromInternalType(
createInternalTypeFromTypeInfo(
- fromDataTypeToLegacyInfo(type.getDataType())),
+ fromDataTypeToLegacyInfo(type.getOutputDataType())),
child.getType().isNullable()),
child);
} else if (call.getFunctionDefinition().equals(BuiltInFunctionDefinitions.REINTERPRET_CAST)) {
@@ -114,7 +114,7 @@ public class RexNodeConverter implements ExpressionVisitor<RexNode> {
return relBuilder.getRexBuilder().makeReinterpretCast(
typeFactory.createTypeFromInternalType(
createInternalTypeFromTypeInfo(
- fromDataTypeToLegacyInfo(type.getDataType())),
+ fromDataTypeToLegacyInfo(type.getOutputDataType())),
child.getType().isNullable()),
child,
checkOverflow);
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala
index 147221a..2b90dbb 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionConverter.scala
@@ -43,7 +43,8 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
assert(children.size == 2)
return Cast(
children.head.accept(this),
- fromDataTypeToLegacyInfo(children(1).asInstanceOf[TypeLiteralExpression].getDataType))
+ fromDataTypeToLegacyInfo(
+ children(1).asInstanceOf[TypeLiteralExpression].getOutputDataType))
case WINDOW_START =>
assert(children.size == 1)