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 2020/05/26 14:10:08 UTC
[flink] 02/02: [FLINK-17936][table] Introduce new type inference
for AS
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1b9e1eb34eb05ca1dacd2913f09ee3e98a2bd011
Author: Timo Walther <tw...@apache.org>
AuthorDate: Mon May 25 14:02:01 2020 +0200
[FLINK-17936][table] Introduce new type inference for AS
Introduces the last missing pieces required to start porting the
other expressions for a consistent API behavior.
This closes #12331.
---
.../functions/BuiltInFunctionDefinitions.java | 13 ++++++-
.../expressions/PlannerExpressionConverter.scala | 8 ----
.../planner/expressions/fieldExpression.scala | 43 ----------------------
.../table/validation/CalcValidationTest.scala | 8 ----
.../table/validation/CalcValidationTest.scala | 8 ----
5 files changed, 12 insertions(+), 68 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 d3523a8..66fa84a 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
@@ -23,6 +23,7 @@ import org.apache.flink.table.api.TableException;
import org.apache.flink.table.types.inference.ConstantArgumentCount;
import org.apache.flink.table.types.inference.InputTypeStrategies;
import org.apache.flink.table.types.inference.TypeStrategies;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
import org.apache.flink.util.Preconditions;
import java.lang.reflect.Field;
@@ -35,6 +36,11 @@ import java.util.Set;
import static org.apache.flink.table.functions.FunctionKind.AGGREGATE;
import static org.apache.flink.table.functions.FunctionKind.OTHER;
import static org.apache.flink.table.functions.FunctionKind.SCALAR;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.OUTPUT_IF_NULL;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.and;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.logical;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.or;
+import static org.apache.flink.table.types.inference.InputTypeStrategies.varyingSequence;
/**
* Dictionary of function definitions for all built-in functions.
@@ -886,7 +892,12 @@ public final class BuiltInFunctionDefinitions {
new BuiltInFunctionDefinition.Builder()
.name("as")
.kind(OTHER)
- .outputTypeStrategy(TypeStrategies.MISSING)
+ .inputTypeStrategy(
+ varyingSequence(
+ or(OUTPUT_IF_NULL, InputTypeStrategies.ANY),
+ and(InputTypeStrategies.LITERAL, logical(LogicalTypeFamily.CHARACTER_STRING)),
+ and(InputTypeStrategies.LITERAL, logical(LogicalTypeFamily.CHARACTER_STRING))))
+ .outputTypeStrategy(TypeStrategies.argument(0))
.build();
public static final BuiltInFunctionDefinition STREAM_RECORD_TIMESTAMP =
new BuiltInFunctionDefinition.Builder()
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala
index e338055..3f6ced9 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala
@@ -149,14 +149,6 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
case fd: FunctionDefinition =>
fd match {
- case AS =>
- assert(args.size >= 2)
- val name = getValue[String](args(1))
- val extraNames = args
- .drop(2)
- .map(e => getValue[String](e))
- Alias(args.head, name, extraNames)
-
case FLATTEN =>
assert(args.size == 1)
Flattening(args.head)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/fieldExpression.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/fieldExpression.scala
index 1a0f5b3..fd86f67 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/fieldExpression.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/fieldExpression.scala
@@ -80,49 +80,6 @@ case class PlannerResolvedFieldReference(
}
}
-case class Alias(child: PlannerExpression, name: String, extraNames: Seq[String] = Seq())
- extends UnaryExpression with NamedExpression {
-
- override def toString = s"$child as '$name"
-
- override private[flink] def resultType: TypeInformation[_] = child.resultType
-
- override private[flink] def makeCopy(anyRefs: Array[AnyRef]): this.type = {
- val child: PlannerExpression = anyRefs.head.asInstanceOf[PlannerExpression]
- copy(child, name, extraNames).asInstanceOf[this.type]
- }
-
- override private[flink] def toAttribute: Attribute = {
- if (valid) {
- PlannerResolvedFieldReference(name, child.resultType)
- } else {
- UnresolvedFieldReference(name)
- }
- }
-
- override private[flink] def validateInput(): ValidationResult = {
- if (name == "*") {
- ValidationFailure("Alias can not accept '*' as name.")
- } else {
- ValidationSuccess
- }
- }
-}
-
-case class UnresolvedAlias(child: PlannerExpression) extends UnaryExpression with NamedExpression {
-
- override private[flink] def name: String =
- throw new UnresolvedException("Invalid call to name on UnresolvedAlias")
-
- override private[flink] def toAttribute: Attribute =
- throw new UnresolvedException("Invalid call to toAttribute on UnresolvedAlias")
-
- override private[flink] def resultType: TypeInformation[_] =
- throw new UnresolvedException("Invalid call to resultType on UnresolvedAlias")
-
- override private[flink] lazy val valid = false
-}
-
case class WindowReference(name: String, tpe: Option[TypeInformation[_]] = None) extends Attribute {
override private[flink] def resultType: TypeInformation[_] =
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/table/validation/CalcValidationTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/table/validation/CalcValidationTest.scala
index b07fbce..14b2a19 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/table/validation/CalcValidationTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/table/validation/CalcValidationTest.scala
@@ -85,14 +85,6 @@ class CalcValidationTest extends TableTestBase {
}
try {
- util.addTableSource[(Int, Long, String)]("Table2")
- .select('_1 as '*, '_2 as 'b, '_1 as 'c)
- fail("ValidationException expected")
- } catch {
- case _: ValidationException => //ignore
- }
-
- try {
util.addTableSource[(Int, Long, String)]("Table3").as("*", "b", "c")
fail("ValidationException expected")
} catch {
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/table/validation/CalcValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/table/validation/CalcValidationTest.scala
index b168aa9..1be3689 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/table/validation/CalcValidationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/batch/table/validation/CalcValidationTest.scala
@@ -103,14 +103,6 @@ class CalcValidationTest extends TableTestBase {
}
try {
- util.addTable[(Int, Long, String)]("Table2")
- .select('_1 as '*, '_2 as 'b, '_1 as 'c)
- fail("ValidationException expected")
- } catch {
- case _: ValidationException => //ignore
- }
-
- try {
util.addTable[(Int, Long, String)]("Table3").as("*", "b", "c")
fail("ValidationException expected")
} catch {