You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2019/04/08 13:28:07 UTC
[flink] branch master updated: [hotfix][table] Replaced Table in
TableReference with TableOperation
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 646349c [hotfix][table] Replaced Table in TableReference with TableOperation
646349c is described below
commit 646349c4f47c4bab881234da6cdfc396c5e325ba
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Wed Mar 27 08:19:09 2019 +0100
[hotfix][table] Replaced Table in TableReference with TableOperation
---
.../flink/table/expressions/ApiExpressionUtils.java | 2 +-
.../table/expressions/TableReferenceExpression.java | 17 +++++++++--------
.../table/expressions/PlannerExpressionConverter.scala | 2 +-
.../flink/table/expressions/fieldExpression.scala | 5 ++++-
.../org/apache/flink/table/expressions/subquery.scala | 13 ++++++++-----
.../apache/flink/table/plan/logical/LogicalNode.scala | 2 +-
6 files changed, 24 insertions(+), 17 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionUtils.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionUtils.java
index 232b421..30dc45d 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionUtils.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionUtils.java
@@ -83,7 +83,7 @@ public final class ApiExpressionUtils {
}
public static TableReferenceExpression tableRef(String name, Table table) {
- return new TableReferenceExpression(name, table);
+ return new TableReferenceExpression(name, table.getTableOperation());
}
public static LookupCallExpression lookupCall(String name, Expression... args) {
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/TableReferenceExpression.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/TableReferenceExpression.java
index 7a53621..d547454 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/TableReferenceExpression.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/TableReferenceExpression.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.expressions;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.table.api.Table;
+import org.apache.flink.table.operations.TableOperation;
import org.apache.flink.util.Preconditions;
import java.util.Collections;
@@ -35,19 +35,19 @@ import java.util.Objects;
public final class TableReferenceExpression implements Expression {
private final String name;
- private final Table table;
+ private final TableOperation tableOperation;
- public TableReferenceExpression(String name, Table table) {
+ public TableReferenceExpression(String name, TableOperation tableOperation) {
this.name = Preconditions.checkNotNull(name);
- this.table = Preconditions.checkNotNull(table);
+ this.tableOperation = Preconditions.checkNotNull(tableOperation);
}
public String getName() {
return name;
}
- public Table getTable() {
- return table;
+ public TableOperation getTableOperation() {
+ return tableOperation;
}
@Override
@@ -69,12 +69,13 @@ public final class TableReferenceExpression implements Expression {
return false;
}
TableReferenceExpression that = (TableReferenceExpression) o;
- return Objects.equals(name, that.name) && Objects.equals(table, that.table);
+ return Objects.equals(name, that.name) &&
+ Objects.equals(tableOperation, that.tableOperation);
}
@Override
public int hashCode() {
- return Objects.hash(name, table);
+ return Objects.hash(name, tableOperation);
}
@Override
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 e63cb05..7470477 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
@@ -711,7 +711,7 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
override def visitTableReference(tableRef: TableReferenceExpression): PlannerExpression = {
TableReference(
tableRef.asInstanceOf[TableReferenceExpression].getName,
- tableRef.asInstanceOf[TableReferenceExpression].getTable
+ tableRef.asInstanceOf[TableReferenceExpression].getTableOperation
)
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
index 7425578..767ff32 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala
@@ -24,6 +24,7 @@ import org.apache.flink.table.api._
import org.apache.flink.table.calcite.FlinkRelBuilder.NamedWindowProperty
import org.apache.flink.table.calcite.FlinkTypeFactory._
import org.apache.flink.table.functions.sql.StreamRecordTimestampSqlFunction
+import org.apache.flink.table.operations.TableOperation
import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
import org.apache.flink.table.validate.{ValidationFailure, ValidationResult, ValidationSuccess}
@@ -139,7 +140,9 @@ case class WindowReference(name: String, tpe: Option[TypeInformation[_]] = None)
override def toString: String = s"'$name"
}
-case class TableReference(name: String, table: Table) extends LeafExpression with NamedExpression {
+case class TableReference(name: String, tableOperation: TableOperation)
+ extends LeafExpression
+ with NamedExpression {
override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode =
throw new UnsupportedOperationException(s"Table reference '$name' can not be used solely.")
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/subquery.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/subquery.scala
index bbdffea..654ae01 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/subquery.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/subquery.scala
@@ -24,7 +24,8 @@ import org.apache.calcite.sql.fun.SqlStdOperatorTable
import org.apache.calcite.tools.RelBuilder
import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.table.api.TableImpl
+import org.apache.flink.table.operations.TableOperation
+import org.apache.flink.table.plan.logical.LogicalNode
import org.apache.flink.table.typeutils.TypeCheckUtils._
import org.apache.flink.table.validate.{ValidationFailure, ValidationResult, ValidationSuccess}
@@ -39,8 +40,10 @@ case class In(expression: PlannerExpression, elements: Seq[PlannerExpression])
// check if this is a sub-query expression or an element list
elements.head match {
- case TableReference(_, table: TableImpl) =>
- RexSubQuery.in(table.getRelNode, ImmutableList.of(expression.toRexNode))
+ case TableReference(_, tableOperation: TableOperation) =>
+ RexSubQuery.in(
+ tableOperation.asInstanceOf[LogicalNode].toRelNode(relBuilder),
+ ImmutableList.of(expression.toRexNode))
case _ =>
relBuilder.call(SqlStdOperatorTable.IN, children.map(_.toRexNode): _*)
@@ -51,11 +54,11 @@ case class In(expression: PlannerExpression, elements: Seq[PlannerExpression])
// check if this is a sub-query expression or an element list
elements.head match {
- case TableReference(name, table: TableImpl) =>
+ case TableReference(name, tableOperation: TableOperation) =>
if (elements.length != 1) {
return ValidationFailure("IN operator supports only one table reference.")
}
- val tableSchema = table.operationTree.getTableSchema
+ val tableSchema = tableOperation.getTableSchema
if (tableSchema.getFieldCount > 1) {
return ValidationFailure(
s"The sub-query table '$name' must not have more than one column.")
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/LogicalNode.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/LogicalNode.scala
index 137a247..50518f6 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/LogicalNode.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/logical/LogicalNode.scala
@@ -118,7 +118,7 @@ abstract class LogicalNode extends TreeNode[LogicalNode] with TableOperation {
// try to resolve a table
tableEnv.scanInternal(Array(name)) match {
- case Some(table) => Some(TableReference(name, table))
+ case Some(table) => Some(TableReference(name, table.getTableOperation))
case None => None
}
}