You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2016/05/17 22:49:27 UTC
[2/5] flink git commit: [FLINK-3754] [tableAPI] Add validation phase
to Table API before construction of RelNodes.
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/RexNodeTranslator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/RexNodeTranslator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/RexNodeTranslator.scala
index f946ed9..095cf04 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/RexNodeTranslator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/RexNodeTranslator.scala
@@ -18,57 +18,49 @@
package org.apache.flink.api.table.plan
-import org.apache.calcite.tools.RelBuilder.AggCall
import org.apache.flink.api.table.TableEnvironment
-
import org.apache.flink.api.table.expressions._
object RexNodeTranslator {
/**
- * Extracts all aggregation expressions (zero, one, or more) from an expression, translates
- * these aggregation expressions into Calcite AggCalls, and replaces the original aggregation
- * expressions by field accesses expressions.
+ * Extracts all aggregation expressions (zero, one, or more) from an expression,
+ * and replaces the original aggregation expressions by field accesses expressions.
*/
- def extractAggCalls(
+ def extractAggregations(
exp: Expression,
- tableEnv: TableEnvironment): Pair[Expression, List[AggCall]] = {
-
- val relBuilder = tableEnv.getRelBuilder
+ tableEnv: TableEnvironment): Pair[Expression, List[NamedExpression]] = {
exp match {
case agg: Aggregation =>
val name = tableEnv.createUniqueAttributeName()
- val aggCall = agg.toAggCall(name)(relBuilder)
+ val aggCall = Alias(agg, name)
val fieldExp = new UnresolvedFieldReference(name)
(fieldExp, List(aggCall))
- case n@Naming(agg: Aggregation, name) =>
- val aggCall = agg.toAggCall(name)(relBuilder)
+ case n @ Alias(agg: Aggregation, name) =>
val fieldExp = new UnresolvedFieldReference(name)
- (fieldExp, List(aggCall))
+ (fieldExp, List(n))
case l: LeafExpression =>
(l, Nil)
case u: UnaryExpression =>
- val c = extractAggCalls(u.child, tableEnv)
- (u.makeCopy(List(c._1)), c._2)
+ val c = extractAggregations(u.child, tableEnv)
+ (u.makeCopy(Array(c._1)), c._2)
case b: BinaryExpression =>
- val l = extractAggCalls(b.left, tableEnv)
- val r = extractAggCalls(b.right, tableEnv)
- (b.makeCopy(List(l._1, r._1)), l._2 ::: r._2)
- case e: Eval =>
- val c = extractAggCalls(e.condition, tableEnv)
- val t = extractAggCalls(e.ifTrue, tableEnv)
- val f = extractAggCalls(e.ifFalse, tableEnv)
- (e.makeCopy(List(c._1, t._1, f._1)), c._2 ::: t._2 ::: f._2)
+ val l = extractAggregations(b.left, tableEnv)
+ val r = extractAggregations(b.right, tableEnv)
+ (b.makeCopy(Array(l._1, r._1)), l._2 ::: r._2)
// Scalar functions
- case c@Call(name, args@_*) =>
- val newArgs = args.map(extractAggCalls(_, tableEnv)).toList
- (c.makeCopy(name :: newArgs.map(_._1)), newArgs.flatMap(_._2))
+ case c @ Call(name, args) =>
+ val newArgs = args.map(extractAggregations(_, tableEnv))
+ (c.makeCopy((name :: newArgs.map(_._1) :: Nil).toArray), newArgs.flatMap(_._2).toList)
- case e@AnyRef =>
- throw new IllegalArgumentException(
- s"Expression $e of type ${e.getClass} not supported yet")
+ case e: Expression =>
+ val newArgs = e.productIterator.map {
+ case arg: Expression =>
+ extractAggregations(arg, tableEnv)
+ }
+ (e.makeCopy(newArgs.map(_._1).toArray), newArgs.flatMap(_._2).toList)
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/LogicalNode.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/LogicalNode.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/LogicalNode.scala
new file mode 100644
index 0000000..dae02bd
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/LogicalNode.scala
@@ -0,0 +1,162 @@
+/*
+ * 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.api.table.plan.logical
+
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.tools.RelBuilder
+
+import org.apache.flink.api.table.{TableEnvironment, ValidationException}
+import org.apache.flink.api.table.expressions._
+import org.apache.flink.api.table.trees.TreeNode
+import org.apache.flink.api.table.typeutils.TypeCoercion
+import org.apache.flink.api.table.validate._
+
+/**
+ * LogicalNode is created and validated as we construct query plan using Table API.
+ *
+ * The main validation procedure is separated into two phases:
+ *
+ * Expressions' resolution and transformation ([[resolveExpressions]]):
+ *
+ * - translate [[UnresolvedFieldReference]] into [[ResolvedFieldReference]]
+ * using child operator's output
+ * - translate [[Call]](UnresolvedFunction) into solid Expression
+ * - generate alias names for query output
+ * - ....
+ *
+ * LogicalNode validation ([[validate]]):
+ *
+ * - check no [[UnresolvedFieldReference]] exists any more
+ * - check if all expressions have children of needed type
+ * - check each logical operator have desired input
+ *
+ * Once we pass the validation phase, we can safely convert LogicalNode into Calcite's RelNode.
+ */
+abstract class LogicalNode extends TreeNode[LogicalNode] {
+ def output: Seq[Attribute]
+
+ def resolveExpressions(tableEnv: TableEnvironment): LogicalNode = {
+ // resolve references and function calls
+ val exprResolved = expressionPostOrderTransform {
+ case u @ UnresolvedFieldReference(name) =>
+ resolveReference(name).getOrElse(u)
+ case c @ Call(name, children) if c.childrenValid =>
+ tableEnv.getFunctionCatalog.lookupFunction(name, children)
+ }
+
+ exprResolved.expressionPostOrderTransform {
+ case ips: InputTypeSpec if ips.childrenValid =>
+ var changed: Boolean = false
+ val newChildren = ips.expectedTypes.zip(ips.children).map { case (tpe, child) =>
+ val childType = child.resultType
+ if (childType != tpe && TypeCoercion.canSafelyCast(childType, tpe)) {
+ changed = true
+ Cast(child, tpe)
+ } else {
+ child
+ }
+ }.toArray[AnyRef]
+ if (changed) ips.makeCopy(newChildren) else ips
+ }
+ }
+
+ final def toRelNode(relBuilder: RelBuilder): RelNode = construct(relBuilder).build()
+
+ protected[logical] def construct(relBuilder: RelBuilder): RelBuilder
+
+ def validate(tableEnv: TableEnvironment): LogicalNode = {
+ val resolvedNode = resolveExpressions(tableEnv)
+ resolvedNode.expressionPostOrderTransform {
+ case a: Attribute if !a.valid =>
+ val from = children.flatMap(_.output).map(_.name).mkString(", ")
+ failValidation(s"cannot resolve [${a.name}] given input [$from]")
+
+ case e: Expression if e.validateInput().isFailure =>
+ failValidation(s"Expression $e failed on input check: " +
+ s"${e.validateInput().asInstanceOf[ValidationFailure].message}")
+ }
+ }
+
+ /**
+ * Resolves the given strings to a [[NamedExpression]] using the input from all child
+ * nodes of this LogicalPlan.
+ */
+ def resolveReference(name: String): Option[NamedExpression] = {
+ val childrenOutput = children.flatMap(_.output)
+ val candidates = childrenOutput.filter(_.name.equalsIgnoreCase(name))
+ if (candidates.length > 1) {
+ failValidation(s"Reference $name is ambiguous")
+ } else if (candidates.length == 0) {
+ None
+ } else {
+ Some(candidates.head.withName(name))
+ }
+ }
+
+ /**
+ * Runs [[postOrderTransform]] with `rule` on all expressions present in this logical node.
+ *
+ * @param rule the rule to be applied to every expression in this logical node.
+ */
+ def expressionPostOrderTransform(rule: PartialFunction[Expression, Expression]): LogicalNode = {
+ var changed = false
+
+ def expressionPostOrderTransform(e: Expression): Expression = {
+ val newExpr = e.postOrderTransform(rule)
+ if (newExpr.fastEquals(e)) {
+ e
+ } else {
+ changed = true
+ newExpr
+ }
+ }
+
+ val newArgs = productIterator.map {
+ case e: Expression => expressionPostOrderTransform(e)
+ case Some(e: Expression) => Some(expressionPostOrderTransform(e))
+ case seq: Traversable[_] => seq.map {
+ case e: Expression => expressionPostOrderTransform(e)
+ case other => other
+ }
+ case other: AnyRef => other
+ }.toArray
+
+ if (changed) makeCopy(newArgs) else this
+ }
+
+ protected def failValidation(msg: String): Nothing = {
+ throw new ValidationException(msg)
+ }
+}
+
+abstract class LeafNode extends LogicalNode {
+ override def children: Seq[LogicalNode] = Nil
+}
+
+abstract class UnaryNode extends LogicalNode {
+ def child: LogicalNode
+
+ override def children: Seq[LogicalNode] = child :: Nil
+}
+
+abstract class BinaryNode extends LogicalNode {
+ def left: LogicalNode
+ def right: LogicalNode
+
+ override def children: Seq[LogicalNode] = left :: right :: Nil
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala
new file mode 100644
index 0000000..d347651
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala
@@ -0,0 +1,339 @@
+/*
+ * 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.api.table.plan.logical
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.JoinRelType
+import org.apache.calcite.rel.logical.LogicalProject
+import org.apache.calcite.tools.RelBuilder
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
+import org.apache.flink.api.java.operators.join.JoinType
+import org.apache.flink.api.table._
+import org.apache.flink.api.table.expressions._
+import org.apache.flink.api.table.typeutils.TypeConverter
+
+case class Project(projectList: Seq[NamedExpression], child: LogicalNode) extends UnaryNode {
+ override def output: Seq[Attribute] = projectList.map(_.toAttribute)
+
+ override def resolveExpressions(tableEnv: TableEnvironment): LogicalNode = {
+ val afterResolve = super.resolveExpressions(tableEnv).asInstanceOf[Project]
+ val newProjectList =
+ afterResolve.projectList.zipWithIndex.map { case (e, i) =>
+ e match {
+ case u @ UnresolvedAlias(child) => child match {
+ case ne: NamedExpression => ne
+ case e if !e.valid => u
+ case c @ Cast(ne: NamedExpression, tp) => Alias(c, s"${ne.name}-$tp")
+ case other => Alias(other, s"_c$i")
+ }
+ case _ => throw new IllegalArgumentException
+ }
+ }
+ Project(newProjectList, child)
+ }
+
+ override def validate(tableEnv: TableEnvironment): LogicalNode = {
+ val resolvedProject = super.validate(tableEnv).asInstanceOf[Project]
+
+ def checkUniqueNames(exprs: Seq[Expression]): Unit = {
+ val names: mutable.Set[String] = mutable.Set()
+ exprs.foreach {
+ case n: Alias =>
+ // explicit name
+ if (names.contains(n.name)) {
+ throw new ValidationException(s"Duplicate field name $n.name.")
+ } else {
+ names.add(n.name)
+ }
+ case r: ResolvedFieldReference =>
+ // simple field forwarding
+ if (names.contains(r.name)) {
+ throw new ValidationException(s"Duplicate field name $r.name.")
+ } else {
+ names.add(r.name)
+ }
+ case _ => // Do nothing
+ }
+ }
+ checkUniqueNames(resolvedProject.projectList)
+ resolvedProject
+ }
+
+ override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
+ val allAlias = projectList.forall(_.isInstanceOf[Alias])
+ child.construct(relBuilder)
+ if (allAlias) {
+ // Calcite's RelBuilder does not translate identity projects even if they rename fields.
+ // Add a projection ourselves (will be automatically removed by translation rules).
+ relBuilder.push(
+ LogicalProject.create(relBuilder.peek(),
+ projectList.map(_.toRexNode(relBuilder)).asJava,
+ projectList.map(_.name).asJava))
+ } else {
+ relBuilder.project(projectList.map(_.toRexNode(relBuilder)): _*)
+ }
+ }
+}
+
+case class AliasNode(aliasList: Seq[Expression], child: LogicalNode) extends UnaryNode {
+ override def output: Seq[Attribute] =
+ throw new UnresolvedException("Invalid call to output on AliasNode")
+
+ override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder =
+ throw new UnresolvedException("Invalid call to toRelNode on AliasNode")
+
+ override def resolveExpressions(tableEnv: TableEnvironment): LogicalNode = {
+ if (aliasList.length > child.output.length) {
+ failValidation("Aliasing more fields than we actually have")
+ } else if (!aliasList.forall(_.isInstanceOf[UnresolvedFieldReference])) {
+ failValidation("Alias only accept name expressions as arguments")
+ } else {
+ val names = aliasList.map(_.asInstanceOf[UnresolvedFieldReference].name)
+ val input = child.output
+ Project(
+ names.zip(input).map { case (name, attr) =>
+ Alias(attr, name)} ++ input.drop(names.length), child)
+ }
+ }
+}
+
+case class Distinct(child: LogicalNode) extends UnaryNode {
+ override def output: Seq[Attribute] = child.output
+
+ override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
+ child.construct(relBuilder)
+ relBuilder.distinct()
+ }
+
+ override def validate(tableEnv: TableEnvironment): LogicalNode = {
+ if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
+ throw new TableException(s"Distinct on stream tables is currently not supported.")
+ }
+ this
+ }
+}
+
+case class Sort(order: Seq[Ordering], child: LogicalNode) extends UnaryNode {
+ override def output: Seq[Attribute] = child.output
+
+ override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
+ child.construct(relBuilder)
+ relBuilder.sort(order.map(_.toRexNode(relBuilder)).asJava)
+ }
+
+ override def validate(tableEnv: TableEnvironment): LogicalNode = {
+ if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
+ throw new TableException(s"Distinct on stream tables is currently not supported.")
+ }
+ super.validate(tableEnv)
+ }
+}
+
+case class Filter(condition: Expression, child: LogicalNode) extends UnaryNode {
+ override def output: Seq[Attribute] = child.output
+
+ override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
+ child.construct(relBuilder)
+ relBuilder.filter(condition.toRexNode(relBuilder))
+ }
+
+ override def validate(tableEnv: TableEnvironment): LogicalNode = {
+ val resolvedFilter = super.validate(tableEnv).asInstanceOf[Filter]
+ if (resolvedFilter.condition.resultType != BOOLEAN_TYPE_INFO) {
+ failValidation(s"filter expression ${resolvedFilter.condition} of" +
+ s" ${resolvedFilter.condition.resultType} is not a boolean")
+ }
+ resolvedFilter
+ }
+}
+
+case class Aggregate(
+ groupingExpressions: Seq[Expression],
+ aggregateExpressions: Seq[NamedExpression],
+ child: LogicalNode) extends UnaryNode {
+
+ override def output: Seq[Attribute] = {
+ (groupingExpressions ++ aggregateExpressions) map { agg =>
+ agg match {
+ case ne: NamedExpression => ne.toAttribute
+ case e => Alias(e, e.toString).toAttribute
+ }
+ }
+ }
+
+ override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
+ child.construct(relBuilder)
+ relBuilder.aggregate(
+ relBuilder.groupKey(groupingExpressions.map(_.toRexNode(relBuilder)).asJava),
+ aggregateExpressions.map { e =>
+ e match {
+ case Alias(agg: Aggregation, name) => agg.toAggCall(name)(relBuilder)
+ case _ => throw new RuntimeException("This should never happen.")
+ }
+ }.asJava)
+ }
+
+ override def validate(tableEnv: TableEnvironment): LogicalNode = {
+ if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
+ throw new TableException(s"Aggregate on stream tables is currently not supported.")
+ }
+
+ val resolvedAggregate = super.validate(tableEnv).asInstanceOf[Aggregate]
+ val groupingExprs = resolvedAggregate.groupingExpressions
+ val aggregateExprs = resolvedAggregate.aggregateExpressions
+ aggregateExprs.foreach(validateAggregateExpression)
+ groupingExprs.foreach(validateGroupingExpression)
+
+ def validateAggregateExpression(expr: Expression): Unit = expr match {
+ // check no nested aggregation exists.
+ case aggExpr: Aggregation =>
+ aggExpr.children.foreach { child =>
+ child.preOrderVisit {
+ case agg: Aggregation =>
+ failValidation(
+ "It's not allowed to use an aggregate function as " +
+ "input of another aggregate function")
+ case _ => // OK
+ }
+ }
+ case a: Attribute if !groupingExprs.exists(_.checkEquals(a)) =>
+ failValidation(
+ s"expression '$a' is invalid because it is neither" +
+ " present in group by nor an aggregate function")
+ case e if groupingExprs.exists(_.checkEquals(e)) => // OK
+ case e => e.children.foreach(validateAggregateExpression)
+ }
+
+ def validateGroupingExpression(expr: Expression): Unit = {
+ if (!expr.resultType.isKeyType) {
+ failValidation(
+ s"expression $expr cannot be used as a grouping expression " +
+ "because it's not a valid key type")
+ }
+ }
+ resolvedAggregate
+ }
+}
+
+case class Union(left: LogicalNode, right: LogicalNode) extends BinaryNode {
+ override def output: Seq[Attribute] = left.output
+
+ override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
+ left.construct(relBuilder)
+ right.construct(relBuilder)
+ relBuilder.union(true)
+ }
+
+ override def validate(tableEnv: TableEnvironment): LogicalNode = {
+ val resolvedUnion = super.validate(tableEnv).asInstanceOf[Union]
+ if (left.output.length != right.output.length) {
+ failValidation(s"Union two table of different column sizes:" +
+ s" ${left.output.size} and ${right.output.size}")
+ }
+ val sameSchema = left.output.zip(right.output).forall { case (l, r) =>
+ l.resultType == r.resultType && l.name == r.name }
+ if (!sameSchema) {
+ failValidation(s"Union two table of different schema:" +
+ s" [${left.output.map(a => (a.name, a.resultType)).mkString(", ")}] and" +
+ s" [${right.output.map(a => (a.name, a.resultType)).mkString(", ")}]")
+ }
+ resolvedUnion
+ }
+}
+
+case class Join(
+ left: LogicalNode,
+ right: LogicalNode,
+ joinType: JoinType,
+ condition: Option[Expression]) extends BinaryNode {
+
+ override def output: Seq[Attribute] = {
+ joinType match {
+ case JoinType.INNER => left.output ++ right.output
+ case j => throw new ValidationException(s"Unsupported JoinType: $j")
+ }
+ }
+
+ override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
+ joinType match {
+ case JoinType.INNER =>
+ left.construct(relBuilder)
+ right.construct(relBuilder)
+ relBuilder.join(JoinRelType.INNER,
+ condition.map(_.toRexNode(relBuilder)).getOrElse(relBuilder.literal(true)))
+ case _ =>
+ throw new ValidationException(s"Unsupported JoinType: $joinType")
+ }
+ }
+
+ private def ambiguousName: Set[String] =
+ left.output.map(_.name).toSet.intersect(right.output.map(_.name).toSet)
+
+ override def validate(tableEnv: TableEnvironment): LogicalNode = {
+ if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
+ throw new TableException(s"Join on stream tables is currently not supported.")
+ }
+
+ val resolvedJoin = super.validate(tableEnv).asInstanceOf[Join]
+ if (!resolvedJoin.condition.forall(_.resultType == BOOLEAN_TYPE_INFO)) {
+ failValidation(s"filter expression ${resolvedJoin.condition} is not a boolean")
+ } else if (!ambiguousName.isEmpty) {
+ failValidation(s"join relations with ambiguous names: ${ambiguousName.mkString(", ")}")
+ }
+ resolvedJoin
+ }
+}
+
+case class CatalogNode(
+ tableName: String,
+ rowType: RelDataType) extends LeafNode {
+
+ val output: Seq[Attribute] = rowType.getFieldList.asScala.map { field =>
+ ResolvedFieldReference(
+ field.getName, TypeConverter.sqlTypeToTypeInfo(field.getType.getSqlTypeName))
+ }
+
+ override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
+ relBuilder.scan(tableName)
+ }
+
+ override def validate(tableEnv: TableEnvironment): LogicalNode = this
+}
+
+/**
+ * Wrapper for valid logical plans generated from SQL String.
+ */
+case class LogicalRelNode(
+ relNode: RelNode) extends LeafNode {
+
+ val output: Seq[Attribute] = relNode.getRowType.getFieldList.asScala.map { field =>
+ ResolvedFieldReference(
+ field.getName, TypeConverter.sqlTypeToTypeInfo(field.getType.getSqlTypeName))
+ }
+
+ override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = {
+ relBuilder.push(relNode)
+ }
+
+ override def validate(tableEnv: TableEnvironment): LogicalNode = this
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
index 5356a9d..4f111c9 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
@@ -17,28 +17,18 @@
*/
package org.apache.flink.api.table
+import scala.collection.JavaConverters._
+
import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.`type`.RelDataTypeField
-import org.apache.calcite.rel.core.JoinRelType
-import org.apache.calcite.rel.logical.LogicalProject
-import org.apache.calcite.rex.{RexCall, RexInputRef, RexLiteral, RexNode}
-import org.apache.calcite.sql.SqlKind
-import org.apache.calcite.tools.RelBuilder.{AggCall, GroupKey}
-import org.apache.calcite.util.NlsString
+
import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.table.plan.PlanGenException
-import org.apache.flink.api.table.plan.RexNodeTranslator.extractAggCalls
+import org.apache.flink.api.java.operators.join.JoinType
+import org.apache.flink.api.table.plan.RexNodeTranslator.extractAggregations
import org.apache.flink.api.table.expressions._
+import org.apache.flink.api.table.plan.logical._
import org.apache.flink.api.table.sinks.TableSink
import org.apache.flink.api.table.typeutils.TypeConverter
-import scala.collection.mutable
-import scala.collection.JavaConverters._
-
-case class BaseTable(
- private[flink] val relNode: RelNode,
- private[flink] val tableEnv: TableEnvironment)
-
/**
* A Table is the core component of the Table API.
* Similar to how the batch and streaming APIs have DataSet and DataStream,
@@ -66,18 +56,16 @@ case class BaseTable(
* in a Scala DSL or as an expression String. Please refer to the documentation for the expression
* syntax.
*
- * @param relNode The root node of the relational Calcite [[RelNode]] tree.
* @param tableEnv The [[TableEnvironment]] to which the table is bound.
+ * @param logicalPlan
*/
class Table(
- private[flink] override val relNode: RelNode,
- private[flink] override val tableEnv: TableEnvironment)
- extends BaseTable(relNode, tableEnv)
-{
+ private[flink] val tableEnv: TableEnvironment,
+ private[flink] val logicalPlan: LogicalNode) {
def relBuilder = tableEnv.getRelBuilder
- def getRelNode: RelNode = relNode
+ def getRelNode: RelNode = logicalPlan.toRelNode(relBuilder)
/**
* Performs a selection operation. Similar to an SQL SELECT statement. The field expressions
@@ -90,44 +78,17 @@ class Table(
* }}}
*/
def select(fields: Expression*): Table = {
-
- checkUniqueNames(fields)
-
- relBuilder.push(relNode)
-
- // separate aggregations and selection expressions
- val extractedAggCalls: List[(Expression, List[AggCall])] = fields
- .map(extractAggCalls(_, tableEnv)).toList
-
- // get aggregation calls
- val aggCalls: List[AggCall] = extractedAggCalls.flatMap(_._2)
-
- // apply aggregations
- if (aggCalls.nonEmpty) {
- // aggregation on stream table is not currently supported
- tableEnv match {
- case _: StreamTableEnvironment =>
- throw new TableException("Aggregation on stream tables is currently not supported.")
- case _ =>
- val emptyKey: GroupKey = relBuilder.groupKey()
- relBuilder.aggregate(emptyKey, aggCalls.toIterable.asJava)
- }
- }
-
- // get selection expressions
- val exprs: List[RexNode] = extractedAggCalls.map(_._1.toRexNode(relBuilder))
-
- relBuilder.project(exprs.toIterable.asJava)
- val projected = relBuilder.build()
-
- if(relNode == projected) {
- // Calcite's RelBuilder does not translate identity projects even if they rename fields.
- // Add a projection ourselves (will be automatically removed by translation rules).
- new Table(createRenamingProject(exprs), tableEnv)
+ val projectionOnAggregates = fields.map(extractAggregations(_, tableEnv))
+ val aggregations = projectionOnAggregates.flatMap(_._2)
+ if (aggregations.nonEmpty) {
+ new Table(tableEnv,
+ Project(projectionOnAggregates.map(e => UnresolvedAlias(e._1)),
+ Aggregate(Nil, aggregations, logicalPlan).validate(tableEnv)).validate(tableEnv))
} else {
- new Table(projected, tableEnv)
+ new Table(tableEnv,
+ Project(
+ projectionOnAggregates.map(e => UnresolvedAlias(e._1)), logicalPlan).validate(tableEnv))
}
-
}
/**
@@ -156,30 +117,7 @@ class Table(
* }}}
*/
def as(fields: Expression*): Table = {
-
- val curNames = relNode.getRowType.getFieldNames.asScala
-
- // validate that AS has only field references
- if (! fields.forall( _.isInstanceOf[UnresolvedFieldReference] )) {
- throw new IllegalArgumentException("All expressions must be field references.")
- }
- // validate that we have not more field references than fields
- if ( fields.length > curNames.size) {
- throw new IllegalArgumentException("More field references than fields.")
- }
-
- val curFields = curNames.map(new UnresolvedFieldReference(_))
-
- val renamings = fields.zip(curFields).map {
- case (newName, oldName) => new Naming(oldName, newName.name)
- }
- val remaining = curFields.drop(fields.size)
-
- relBuilder.push(relNode)
-
- val exprs = (renamings ++ remaining).map(_.toRexNode(relBuilder))
-
- new Table(createRenamingProject(exprs), tableEnv)
+ new Table(tableEnv, AliasNode(fields, logicalPlan).validate(tableEnv))
}
/**
@@ -208,11 +146,7 @@ class Table(
* }}}
*/
def filter(predicate: Expression): Table = {
-
- relBuilder.push(relNode)
- relBuilder.filter(predicate.toRexNode(relBuilder))
-
- new Table(relBuilder.build(), tableEnv)
+ new Table(tableEnv, Filter(predicate, logicalPlan).validate(tableEnv))
}
/**
@@ -269,19 +203,10 @@ class Table(
* }}}
*/
def groupBy(fields: Expression*): GroupedTable = {
-
- // group by on stream tables is currently not supported
- tableEnv match {
- case _: StreamTableEnvironment =>
- throw new TableException("Group by on stream tables is currently not supported.")
- case _ => {
- relBuilder.push(relNode)
- val groupExpr = fields.map(_.toRexNode(relBuilder)).toIterable.asJava
- val groupKey = relBuilder.groupKey(groupExpr)
-
- new GroupedTable(relBuilder.build(), tableEnv, groupKey)
- }
+ if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
+ throw new TableException(s"Group by on stream tables is currently not supported.")
}
+ new GroupedTable(this, fields)
}
/**
@@ -309,15 +234,7 @@ class Table(
* }}}
*/
def distinct(): Table = {
- // distinct on stream table is not currently supported
- tableEnv match {
- case _: StreamTableEnvironment =>
- throw new TableException("Distinct on stream tables is currently not supported.")
- case _ =>
- relBuilder.push(relNode)
- relBuilder.distinct()
- new Table(relBuilder.build(), tableEnv)
- }
+ new Table(tableEnv, Distinct(logicalPlan).validate(tableEnv))
}
/**
@@ -334,32 +251,12 @@ class Table(
* }}}
*/
def join(right: Table): Table = {
-
- // join on stream tables is currently not supported
- tableEnv match {
- case _: StreamTableEnvironment =>
- throw new TableException("Join on stream tables is currently not supported.")
- case _ => {
- // check that right table belongs to the same TableEnvironment
- if (right.tableEnv != this.tableEnv) {
- throw new TableException("Only tables from the same TableEnvironment can be joined.")
- }
-
- // check that join inputs do not have overlapping field names
- val leftFields = relNode.getRowType.getFieldNames.asScala.toSet
- val rightFields = right.relNode.getRowType.getFieldNames.asScala.toSet
- if (leftFields.intersect(rightFields).nonEmpty) {
- throw new IllegalArgumentException("Overlapping fields names on join input.")
- }
-
- relBuilder.push(relNode)
- relBuilder.push(right.relNode)
-
- relBuilder.join(JoinRelType.INNER, relBuilder.literal(true))
- val join = relBuilder.build()
- new Table(join, tableEnv)
- }
+ // check that right table belongs to the same TableEnvironment
+ if (right.tableEnv != this.tableEnv) {
+ throw new ValidationException("Only tables from the same TableEnvironment can be joined.")
}
+ new Table(tableEnv,
+ Join(this.logicalPlan, right.logicalPlan, JoinType.INNER, None).validate(tableEnv))
}
/**
@@ -375,32 +272,11 @@ class Table(
* }}}
*/
def unionAll(right: Table): Table = {
-
// check that right table belongs to the same TableEnvironment
if (right.tableEnv != this.tableEnv) {
- throw new TableException("Only tables from the same TableEnvironment can be unioned.")
- }
-
- val leftRowType: List[RelDataTypeField] = relNode.getRowType.getFieldList.asScala.toList
- val rightRowType: List[RelDataTypeField] = right.relNode.getRowType.getFieldList.asScala.toList
-
- if (leftRowType.length != rightRowType.length) {
- throw new IllegalArgumentException("Unioned tables have varying row schema.")
+ throw new ValidationException("Only tables from the same TableEnvironment can be unioned.")
}
- else {
- val zipped: List[(RelDataTypeField, RelDataTypeField)] = leftRowType.zip(rightRowType)
- zipped.foreach { case (x, y) =>
- if (!x.getName.equals(y.getName) || x.getType != y.getType) {
- throw new IllegalArgumentException("Unioned tables have varying row schema.")
- }
- }
- }
-
- relBuilder.push(relNode)
- relBuilder.push(right.relNode)
-
- relBuilder.union(true)
- new Table(relBuilder.build(), tableEnv)
+ new Table(tableEnv, Union(logicalPlan, right.logicalPlan).validate(tableEnv))
}
/**
@@ -414,22 +290,13 @@ class Table(
* }}}
*/
def orderBy(fields: Expression*): Table = {
- relBuilder.push(relNode)
-
- if (! fields.forall {
- case x : UnresolvedFieldReference => true
- case x : Ordering => x.child.isInstanceOf[UnresolvedFieldReference]
- case _ => false
- }) {
- throw new IllegalArgumentException("All expressions must be field references " +
- "or asc/desc expressions.")
+ val order: Seq[Ordering] = fields.map { case e =>
+ e match {
+ case o: Ordering => o
+ case _ => Asc(e)
+ }
}
-
- val exprs = fields.map(_.toRexNode(relBuilder))
-
- relBuilder.sort(exprs.asJava)
- new Table(relBuilder.build(), tableEnv)
-
+ new Table(tableEnv, Sort(order, logicalPlan).validate(tableEnv))
}
/**
@@ -460,7 +327,7 @@ class Table(
def toSink[T](sink: TableSink[T]): Unit = {
// get schema information of table
- val rowType = relNode.getRowType
+ val rowType = getRelNode.getRowType
val fieldNames: Array[String] = rowType.getFieldNames.asScala.toArray
val fieldTypes: Array[TypeInformation[_]] = rowType.getFieldList.asScala
.map(f => TypeConverter.sqlTypeToTypeInfo(f.getType.getSqlTypeName)).toArray
@@ -471,62 +338,14 @@ class Table(
// emit the table to the configured table sink
tableEnv.emitToSink(this, configuredSink)
}
-
- private def createRenamingProject(exprs: Seq[RexNode]): LogicalProject = {
-
- val names = exprs.map{ e =>
- e.getKind match {
- case SqlKind.AS =>
- e.asInstanceOf[RexCall].getOperands.get(1)
- .asInstanceOf[RexLiteral].getValue
- .asInstanceOf[NlsString].getValue
- case SqlKind.INPUT_REF =>
- relNode.getRowType.getFieldNames.get(e.asInstanceOf[RexInputRef].getIndex)
- case _ =>
- throw new PlanGenException("Unexpected expression type encountered.")
- }
-
- }
- LogicalProject.create(relNode, exprs.toList.asJava, names.toList.asJava)
- }
-
- private def checkUniqueNames(exprs: Seq[Expression]): Unit = {
- val names: mutable.Set[String] = mutable.Set()
-
- exprs.foreach {
- case n: Naming =>
- // explicit name
- if (names.contains(n.name)) {
- throw new IllegalArgumentException(s"Duplicate field name $n.name.")
- } else {
- names.add(n.name)
- }
- case u: UnresolvedFieldReference =>
- // simple field forwarding
- if (names.contains(u.name)) {
- throw new IllegalArgumentException(s"Duplicate field name $u.name.")
- } else {
- names.add(u.name)
- }
- case _ => // Do nothing
- }
- }
-
}
/**
* A table that has been grouped on a set of grouping keys.
- *
- * @param relNode The root node of the relational Calcite [[RelNode]] tree.
- * @param tableEnv The [[TableEnvironment]] to which the table is bound.
- * @param groupKey The Calcite [[GroupKey]] of this table.
*/
class GroupedTable(
- private[flink] override val relNode: RelNode,
- private[flink] override val tableEnv: TableEnvironment,
- private[flink] val groupKey: GroupKey) extends BaseTable(relNode, tableEnv) {
-
- def relBuilder = tableEnv.getRelBuilder
+ private[flink] val table: Table,
+ private[flink] val groupKey: Seq[Expression]) {
/**
* Performs a selection operation on a grouped table. Similar to an SQL SELECT statement.
@@ -540,31 +359,19 @@ class GroupedTable(
*/
def select(fields: Expression*): Table = {
- relBuilder.push(relNode)
-
- // separate aggregations and selection expressions
- val extractedAggCalls: List[(Expression, List[AggCall])] = fields
- .map(extractAggCalls(_, tableEnv)).toList
-
- // get aggregation calls
- val aggCalls: List[AggCall] = extractedAggCalls.flatMap(_._2)
+ val projectionOnAggregates = fields.map(extractAggregations(_, table.tableEnv))
+ val aggregations = projectionOnAggregates.flatMap(_._2)
- // apply aggregations
- relBuilder.aggregate(groupKey, aggCalls.toIterable.asJava)
-
- // get selection expressions
- val exprs: List[RexNode] = try {
- extractedAggCalls.map(_._1.toRexNode(relBuilder))
- } catch {
- case iae: IllegalArgumentException =>
- throw new IllegalArgumentException(
- "Only grouping fields and aggregations allowed after groupBy.", iae)
- case e: Exception => throw e
+ val logical = if (aggregations.nonEmpty) {
+ Project(projectionOnAggregates.map(e => UnresolvedAlias(e._1)),
+ Aggregate(groupKey, aggregations, table.logicalPlan).validate(table.tableEnv)
+ )
+ } else {
+ Project(projectionOnAggregates.map(e => UnresolvedAlias(e._1)),
+ Aggregate(groupKey, Nil, table.logicalPlan).validate(table.tableEnv))
}
- relBuilder.project(exprs.toIterable.asJava)
-
- new Table(relBuilder.build(), tableEnv)
+ new Table(table.tableEnv, logical.validate(table.tableEnv))
}
/**
@@ -581,5 +388,4 @@ class GroupedTable(
val fieldExprs = ExpressionParser.parseExpressionList(fields)
select(fieldExprs: _*)
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/trees/TreeNode.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/trees/TreeNode.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/trees/TreeNode.scala
new file mode 100644
index 0000000..63c7013
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/trees/TreeNode.scala
@@ -0,0 +1,114 @@
+/*
+ * 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.api.table.trees
+
+import org.apache.commons.lang.ClassUtils
+
+/**
+ * Generic base class for trees that can be transformed and traversed.
+ */
+abstract class TreeNode[A <: TreeNode[A]] extends Product { self: A =>
+
+ /**
+ * List of child nodes that should be considered when doing transformations. Other values
+ * in the Product will not be transformed, only handed through.
+ */
+ def children: Seq[A]
+
+ /**
+ * Tests for equality by first testing for reference equality.
+ */
+ def fastEquals(other: TreeNode[_]): Boolean = this.eq(other) || this == other
+
+ /**
+ * Do tree transformation in post order.
+ */
+ def postOrderTransform(rule: PartialFunction[A, A]): A = {
+ def childrenTransform(rule: PartialFunction[A, A]): A = {
+ var changed = false
+ val newArgs = productIterator.map {
+ case arg: TreeNode[_] if children.contains(arg) =>
+ val newChild = arg.asInstanceOf[A].postOrderTransform(rule)
+ if (!(newChild fastEquals arg)) {
+ changed = true
+ newChild
+ } else {
+ arg
+ }
+ case args: Traversable[_] => args.map {
+ case arg: TreeNode[_] if children.contains(arg) =>
+ val newChild = arg.asInstanceOf[A].postOrderTransform(rule)
+ if (!(newChild fastEquals arg)) {
+ changed = true
+ newChild
+ } else {
+ arg
+ }
+ case other => other
+ }
+ case nonChild: AnyRef => nonChild
+ case null => null
+ }.toArray
+ if (changed) makeCopy(newArgs) else this
+ }
+
+ val afterChildren = childrenTransform(rule)
+ if (afterChildren fastEquals this) {
+ rule.applyOrElse(this, identity[A])
+ } else {
+ rule.applyOrElse(afterChildren, identity[A])
+ }
+ }
+
+ /**
+ * Runs the given function first on the node and then recursively on all its children.
+ */
+ def preOrderVisit(f: A => Unit): Unit = {
+ f(this)
+ children.foreach(_.preOrderVisit(f))
+ }
+
+ /**
+ * Creates a new copy of this expression with new children. This is used during transformation
+ * if children change.
+ */
+ def makeCopy(newArgs: Array[AnyRef]): A = {
+ val ctors = getClass.getConstructors.filter(_.getParameterTypes.size > 0)
+ if (ctors.isEmpty) {
+ throw new RuntimeException(s"No valid constructor for ${getClass.getSimpleName}")
+ }
+
+ val defaultCtor = ctors.find { ctor =>
+ if (ctor.getParameterTypes.size != newArgs.length) {
+ false
+ } else if (newArgs.contains(null)) {
+ false
+ } else {
+ val argsClasses: Array[Class[_]] = newArgs.map(_.getClass)
+ ClassUtils.isAssignable(argsClasses, ctor.getParameterTypes)
+ }
+ }.getOrElse(ctors.maxBy(_.getParameterTypes.size))
+
+ try {
+ defaultCtor.newInstance(newArgs: _*).asInstanceOf[A]
+ } catch {
+ case e: java.lang.IllegalArgumentException =>
+ throw new IllegalArgumentException(s"Fail to copy treeNode ${getClass.getName}")
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCheckUtils.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCheckUtils.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCheckUtils.scala
new file mode 100644
index 0000000..1da1d2c
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCheckUtils.scala
@@ -0,0 +1,40 @@
+/*
+ * 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.api.table.typeutils
+
+import org.apache.flink.api.common.typeinfo.{NumericTypeInfo, TypeInformation}
+import org.apache.flink.api.table.validate._
+
+object TypeCheckUtils {
+
+ def assertNumericExpr(dataType: TypeInformation[_], caller: String): ExprValidationResult = {
+ if (dataType.isInstanceOf[NumericTypeInfo[_]]) {
+ ValidationSuccess
+ } else {
+ ValidationFailure(s"$caller requires numeric types, get $dataType here")
+ }
+ }
+
+ def assertOrderableExpr(dataType: TypeInformation[_], caller: String): ExprValidationResult = {
+ if (dataType.isSortKeyType) {
+ ValidationSuccess
+ } else {
+ ValidationFailure(s"$caller requires orderable types, get $dataType here")
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCoercion.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCoercion.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCoercion.scala
new file mode 100644
index 0000000..218996d
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeCoercion.scala
@@ -0,0 +1,92 @@
+/*
+ * 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.api.table.typeutils
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
+import org.apache.flink.api.common.typeinfo.{NumericTypeInfo, TypeInformation}
+
+/**
+ * Utilities for type conversions.
+ */
+object TypeCoercion {
+
+ val numericWideningPrecedence: IndexedSeq[TypeInformation[_]] =
+ IndexedSeq(
+ BYTE_TYPE_INFO,
+ SHORT_TYPE_INFO,
+ INT_TYPE_INFO,
+ LONG_TYPE_INFO,
+ FLOAT_TYPE_INFO,
+ DOUBLE_TYPE_INFO)
+
+ def widerTypeOf(tp1: TypeInformation[_], tp2: TypeInformation[_]): Option[TypeInformation[_]] = {
+ (tp1, tp2) match {
+ case (tp1, tp2) if tp1 == tp2 => Some(tp1)
+
+ case (_, STRING_TYPE_INFO) => Some(STRING_TYPE_INFO)
+ case (STRING_TYPE_INFO, _) => Some(STRING_TYPE_INFO)
+
+ case tuple if tuple.productIterator.forall(numericWideningPrecedence.contains) =>
+ val higherIndex = numericWideningPrecedence.lastIndexWhere(t => t == tp1 || t == tp2)
+ Some(numericWideningPrecedence(higherIndex))
+
+ case _ => None
+ }
+ }
+
+ /**
+ * Test if we can do cast safely without lose of information.
+ */
+ def canSafelyCast(from: TypeInformation[_], to: TypeInformation[_]): Boolean = (from, to) match {
+ case (_, STRING_TYPE_INFO) => true
+
+ case tuple if tuple.productIterator.forall(numericWideningPrecedence.contains) =>
+ if (numericWideningPrecedence.indexOf(from) < numericWideningPrecedence.indexOf(to)) {
+ true
+ } else {
+ false
+ }
+
+ case _ => false
+ }
+
+ /**
+ * All the supported cast types in flink-table.
+ * Note: This may lose information during the cast.
+ */
+ def canCast(from: TypeInformation[_], to: TypeInformation[_]): Boolean = (from, to) match {
+ case (from, to) if from == to => true
+
+ case (_, STRING_TYPE_INFO) => true
+
+ case (_, DATE_TYPE_INFO) => false // Date type not supported yet.
+ case (_, VOID_TYPE_INFO) => false // Void type not supported
+ case (_, CHAR_TYPE_INFO) => false // Character type not supported.
+
+ case (STRING_TYPE_INFO, _: NumericTypeInfo[_]) => true
+ case (STRING_TYPE_INFO, BOOLEAN_TYPE_INFO) => true
+
+ case (BOOLEAN_TYPE_INFO, _: NumericTypeInfo[_]) => true
+ case (_: NumericTypeInfo[_], BOOLEAN_TYPE_INFO) => true
+
+ case (_: NumericTypeInfo[_], _: NumericTypeInfo[_]) => true
+
+ case _ => false
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/ExprValidationResult.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/ExprValidationResult.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/ExprValidationResult.scala
new file mode 100644
index 0000000..8571051
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/ExprValidationResult.scala
@@ -0,0 +1,41 @@
+/*
+ * 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.api.table.validate
+
+/**
+ * Represents the result of `Expression.validateInput`.
+ */
+sealed trait ExprValidationResult {
+ def isFailure: Boolean = !isSuccess
+ def isSuccess: Boolean
+}
+
+/**
+ * Represents the successful result of `Expression.checkInputDataTypes`.
+ */
+object ValidationSuccess extends ExprValidationResult {
+ val isSuccess: Boolean = true
+}
+
+/**
+ * Represents the failing result of `Expression.checkInputDataTypes`,
+ * with a error message to show the reason of failure.
+ */
+case class ValidationFailure(message: String) extends ExprValidationResult {
+ val isSuccess: Boolean = false
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala
new file mode 100644
index 0000000..726917e
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala
@@ -0,0 +1,124 @@
+/*
+ * 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.api.table.validate
+
+import scala.collection.mutable
+import scala.util.{Failure, Success, Try}
+
+import org.apache.flink.api.table.expressions._
+import org.apache.flink.api.table.ValidationException
+
+/**
+ * A catalog for looking up user defined functions, used during validation phase.
+ */
+class FunctionCatalog {
+
+ private val functionBuilders = mutable.HashMap.empty[String, Class[_]]
+
+ def registerFunction(name: String, builder: Class[_]): Unit =
+ functionBuilders.put(name.toLowerCase, builder)
+
+ /**
+ * Lookup and create an expression if we find a match.
+ */
+ def lookupFunction(name: String, children: Seq[Expression]): Expression = {
+ val funcClass = functionBuilders.get(name.toLowerCase).getOrElse {
+ throw new ValidationException(s"undefined function $name")
+ }
+ withChildren(funcClass, children)
+ }
+
+ /**
+ * Instantiate a function using the provided `children`.
+ */
+ private def withChildren(func: Class[_], children: Seq[Expression]): Expression = {
+ // Try to find a constructor accepts `Seq[Expression]`
+ Try(func.getDeclaredConstructor(classOf[Seq[_]])) match {
+ case Success(seqCtor) =>
+ Try(seqCtor.newInstance(children).asInstanceOf[Expression]) match {
+ case Success(expr) => expr
+ case Failure(e) => throw new ValidationException(e.getMessage)
+ }
+ case Failure(e) =>
+ val childrenClass = Seq.fill(children.length)(classOf[Expression])
+ // Try to find a constructor matching the exact number of children
+ Try(func.getDeclaredConstructor(childrenClass: _*)) match {
+ case Success(ctor) =>
+ Try(ctor.newInstance(children: _*).asInstanceOf[Expression]) match {
+ case Success(expr) => expr
+ case Failure(e) => throw new ValidationException(e.getMessage)
+ }
+ case Failure(e) =>
+ throw new ValidationException(s"Invalid number of arguments for function $func")
+ }
+ }
+ }
+
+ /**
+ * Drop a function and return if the function existed.
+ */
+ def dropFunction(name: String): Boolean =
+ functionBuilders.remove(name.toLowerCase).isDefined
+
+ /**
+ * Drop all registered functions.
+ */
+ def clear(): Unit = functionBuilders.clear()
+}
+
+object FunctionCatalog {
+
+ val buildInFunctions: Map[String, Class[_]] = Map(
+ // aggregate functions
+ "avg" -> classOf[Avg],
+ "count" -> classOf[Count],
+ "max" -> classOf[Max],
+ "min" -> classOf[Min],
+ "sum" -> classOf[Sum],
+
+ // string functions
+ "charLength" -> classOf[CharLength],
+ "initCap" -> classOf[InitCap],
+ "like" -> classOf[Like],
+ "lowerCase" -> classOf[Lower],
+ "similar" -> classOf[Similar],
+ "subString" -> classOf[SubString],
+ "trim" -> classOf[Trim],
+ "upperCase" -> classOf[Upper],
+
+ // math functions
+ "abs" -> classOf[Abs],
+ "ceil" -> classOf[Ceil],
+ "exp" -> classOf[Exp],
+ "floor" -> classOf[Floor],
+ "log10" -> classOf[Log10],
+ "ln" -> classOf[Ln],
+ "power" -> classOf[Power],
+ "mod" -> classOf[Mod]
+ )
+
+ /**
+ * Create a new function catalog with build-in functions.
+ */
+ def withBuildIns: FunctionCatalog = {
+ val catalog = new FunctionCatalog()
+ buildInFunctions.foreach { case (n, c) => catalog.registerFunction(n, c) }
+ catalog
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/TableEnvironmentITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/TableEnvironmentITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/TableEnvironmentITCase.java
index 0a2cf57..8fdb2da 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/TableEnvironmentITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/TableEnvironmentITCase.java
@@ -24,10 +24,7 @@ import org.apache.flink.api.java.table.BatchTableEnvironment;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.scala.batch.utils.TableProgramsTestBase;
-import org.apache.flink.api.table.Row;
-import org.apache.flink.api.table.Table;
-import org.apache.flink.api.table.TableEnvironment;
-import org.apache.flink.api.table.TableException;
+import org.apache.flink.api.table.*;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -99,7 +96,7 @@ public class TableEnvironmentITCase extends TableProgramsTestBase {
tableEnv.registerDataSet("MyTable", ds2);
}
- @Test(expected = TableException.class)
+ @Test(expected = ValidationException.class)
public void testScanUnregisteredTable() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env, config());
@@ -127,7 +124,7 @@ public class TableEnvironmentITCase extends TableProgramsTestBase {
compareResultAsText(results, expected);
}
- @Test(expected = TableException.class)
+ @Test(expected = ValidationException.class)
public void testIllegalName() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env, config());
@@ -138,7 +135,7 @@ public class TableEnvironmentITCase extends TableProgramsTestBase {
tableEnv.registerTable("_DataSetTable_42", t);
}
- @Test(expected = TableException.class)
+ @Test(expected = ValidationException.class)
public void testRegisterTableFromOtherEnv() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv1 = TableEnvironment.getTableEnvironment(env, config());
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/AggregationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/AggregationsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/AggregationsITCase.java
index d4af08d..ddf5884 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/AggregationsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/AggregationsITCase.java
@@ -27,7 +27,7 @@ import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.api.table.TableEnvironment;
-import org.apache.flink.api.table.plan.PlanGenException;
+import org.apache.flink.api.table.ValidationException;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
@@ -59,7 +59,7 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
compareResultAsText(results, expected);
}
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = ValidationException.class)
public void testAggregationOnNonExistingField() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -141,7 +141,7 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
compareResultAsText(results, expected);
}
- @Test(expected = PlanGenException.class)
+ @Test(expected = ValidationException.class)
public void testNonWorkingDataTypes() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -161,7 +161,7 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
compareResultAsText(results, expected);
}
- @Test(expected = UnsupportedOperationException.class)
+ @Test(expected = ValidationException.class)
public void testNoNestedAggregation() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java
index 0246904..83c7cfd 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/ExpressionsITCase.java
@@ -30,6 +30,8 @@ import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.table.TableEnvironment;
import org.apache.flink.api.table.codegen.CodeGenException;
import static org.junit.Assert.fail;
+
+import org.apache.flink.api.table.ValidationException;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -154,7 +156,7 @@ public class ExpressionsITCase extends TableProgramsTestBase {
compareResultAsText(results, expected);
}
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = ValidationException.class)
public void testEvalInvalidTypes() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env, config());
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/FilterITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/FilterITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/FilterITCase.java
index 2848347..7a2bedf 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/FilterITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/FilterITCase.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.table.BatchTableEnvironment;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.table.TableEnvironment;
+import org.apache.flink.api.table.ValidationException;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -152,7 +153,7 @@ public class FilterITCase extends TableProgramsTestBase {
compareResultAsText(results, expected);
}
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = ValidationException.class)
public void testFilterInvalidField() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env, config());
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/GroupedAggregationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/GroupedAggregationsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/GroupedAggregationsITCase.java
index 70dd793..1906040 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/GroupedAggregationsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/GroupedAggregationsITCase.java
@@ -25,6 +25,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.table.BatchTableEnvironment;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.table.TableEnvironment;
+import org.apache.flink.api.table.ValidationException;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
@@ -40,7 +41,7 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = ValidationException.class)
public void testGroupingOnNonExistentField() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -54,7 +55,7 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase {
.select("a.avg");
}
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = ValidationException.class)
public void testGroupingInvalidSelection() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/JoinITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/JoinITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/JoinITCase.java
index 4c40596..e6db3b0 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/JoinITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/JoinITCase.java
@@ -26,7 +26,7 @@ import org.apache.flink.api.java.table.BatchTableEnvironment;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.table.TableEnvironment;
-import org.apache.flink.api.table.TableException;
+import org.apache.flink.api.table.ValidationException;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
@@ -121,7 +121,7 @@ public class JoinITCase extends MultipleProgramsTestBase {
compareResultAsText(results, expected);
}
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = ValidationException.class)
public void testJoinNonExistingKey() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -136,7 +136,7 @@ public class JoinITCase extends MultipleProgramsTestBase {
in1.join(in2).where("foo === e").select("c, g");
}
- @Test(expected = TableException.class)
+ @Test(expected = ValidationException.class)
public void testJoinWithNonMatchingKeyTypes() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -154,7 +154,7 @@ public class JoinITCase extends MultipleProgramsTestBase {
tableEnv.toDataSet(result, Row.class).collect();
}
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = ValidationException.class)
public void testJoinWithAmbiguousFields() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -189,7 +189,7 @@ public class JoinITCase extends MultipleProgramsTestBase {
compareResultAsText(results, expected);
}
- @Test(expected = TableException.class)
+ @Test(expected = ValidationException.class)
public void testJoinTablesFromDifferentEnvs() {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tEnv1 = TableEnvironment.getTableEnvironment(env);
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/SelectITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/SelectITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/SelectITCase.java
index 7c9478a..e48914c 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/SelectITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/SelectITCase.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.table.BatchTableEnvironment;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.table.TableEnvironment;
+import org.apache.flink.api.table.ValidationException;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -103,7 +104,7 @@ public class SelectITCase extends TableProgramsTestBase {
compareResultAsText(results, expected);
}
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = ValidationException.class)
public void testSelectInvalidField() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env, config());
@@ -115,7 +116,7 @@ public class SelectITCase extends TableProgramsTestBase {
.select("a + 1, foo + 2");
}
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = ValidationException.class)
public void testSelectAmbiguousFieldNames() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env, config());
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/StringExpressionsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/StringExpressionsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/StringExpressionsITCase.java
index e55bd22..db5eac9 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/StringExpressionsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/StringExpressionsITCase.java
@@ -21,10 +21,10 @@ package org.apache.flink.api.java.batch.table;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.table.TableEnvironment;
+import org.apache.flink.api.table.ValidationException;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
import org.apache.flink.api.table.Row;
import org.apache.flink.api.table.Table;
-import org.apache.flink.api.table.codegen.CodeGenException;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.table.BatchTableEnvironment;
@@ -64,6 +64,26 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
}
@Test
+ public void testSubstringWithByteStart() throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
+
+ DataSet<Tuple2<String, Byte>> ds = env.fromElements(
+ new Tuple2<>("AAAA", (byte) 2),
+ new Tuple2<>("BBBB", (byte) 1));
+
+ Table in = tableEnv.fromDataSet(ds, "a, b");
+
+ Table result = in
+ .select("a.substring(1, b)");
+
+ DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = resultSet.collect();
+ String expected = "AA\nB";
+ compareResultAsText(results, expected);
+ }
+
+ @Test
public void testSubstringWithMaxEnd() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -83,7 +103,7 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
compareResultAsText(results, expected);
}
- @Test(expected = CodeGenException.class)
+ @Test(expected = ValidationException.class)
public void testNonWorkingSubstring1() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -102,7 +122,7 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
resultSet.collect();
}
- @Test(expected = CodeGenException.class)
+ @Test(expected = ValidationException.class)
public void testNonWorkingSubstring2() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -121,7 +141,7 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
resultSet.collect();
}
- @Test(expected = CodeGenException.class)
+ @Test(expected = ValidationException.class)
public void testGeneratedCodeForStringComparison() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -132,7 +152,7 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
DataSet<Row> resultSet = tableEnv.toDataSet(res, Row.class);
}
- @Test(expected = CodeGenException.class)
+ @Test(expected = ValidationException.class)
public void testGeneratedCodeForIntegerEqualsComparison() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -143,7 +163,7 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
DataSet<Row> resultSet = tableEnv.toDataSet(res, Row.class);
}
- @Test(expected = CodeGenException.class)
+ @Test(expected = ValidationException.class)
public void testGeneratedCodeForIntegerGreaterComparison() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/UnionITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/UnionITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/UnionITCase.java
index a7805f8..853cd7f 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/UnionITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/batch/table/UnionITCase.java
@@ -26,7 +26,7 @@ import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.table.Row;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.table.TableEnvironment;
-import org.apache.flink.api.table.TableException;
+import org.apache.flink.api.table.ValidationException;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
@@ -80,7 +80,7 @@ public class UnionITCase extends MultipleProgramsTestBase {
compareResultAsText(results, expected);
}
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = ValidationException.class)
public void testUnionIncompatibleNumberOfFields() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -95,7 +95,7 @@ public class UnionITCase extends MultipleProgramsTestBase {
in1.unionAll(in2);
}
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = ValidationException.class)
public void testUnionIncompatibleFieldsName() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -110,7 +110,7 @@ public class UnionITCase extends MultipleProgramsTestBase {
in1.unionAll(in2);
}
- @Test(expected = IllegalArgumentException.class)
+ @Test(expected = ValidationException.class)
public void testUnionIncompatibleFieldTypes() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
@@ -168,7 +168,7 @@ public class UnionITCase extends MultipleProgramsTestBase {
compareResultAsText(results, expected);
}
- @Test(expected = TableException.class)
+ @Test(expected = ValidationException.class)
public void testUnionTablesFromDifferentEnvs() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tEnv1 = TableEnvironment.getTableEnvironment(env);
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/TableEnvironmentITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/TableEnvironmentITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/TableEnvironmentITCase.scala
index 80a491b..c33e1ef 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/TableEnvironmentITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/TableEnvironmentITCase.scala
@@ -23,7 +23,7 @@ import org.apache.flink.api.scala.batch.utils.TableProgramsTestBase
import org.apache.flink.api.scala.batch.utils.TableProgramsTestBase.TableConfigMode
import org.apache.flink.api.scala.table._
import org.apache.flink.api.scala.util.CollectionDataSets
-import org.apache.flink.api.table.{Row, TableEnvironment, TableException}
+import org.apache.flink.api.table.{Row, TableEnvironment, TableException, ValidationException}
import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
import org.apache.flink.test.util.TestBaseUtils
import org.junit._
@@ -89,7 +89,7 @@ class TableEnvironmentITCase(
tEnv.registerDataSet("MyTable", ds2)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testScanUnregisteredTable(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val tEnv = TableEnvironment.getTableEnvironment(env, config)
@@ -130,7 +130,7 @@ class TableEnvironmentITCase(
tEnv.registerDataSet("MyTable", t2)
}
- @Test(expected = classOf[TableException])
+ @Test(expected = classOf[ValidationException])
def testRegisterTableFromOtherEnv(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val tEnv1 = TableEnvironment.getTableEnvironment(env, config)
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/AggregationsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/AggregationsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/AggregationsITCase.scala
index c34edbc..6c413e5 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/AggregationsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/AggregationsITCase.scala
@@ -21,8 +21,7 @@ package org.apache.flink.api.scala.batch.table
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.table._
import org.apache.flink.api.scala.util.CollectionDataSets
-import org.apache.flink.api.table.plan.PlanGenException
-import org.apache.flink.api.table.{Row, TableEnvironment}
+import org.apache.flink.api.table.{Row, TableEnvironment, ValidationException}
import org.apache.flink.examples.scala.WordCountTable.{WC => MyWC}
import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
@@ -49,7 +48,7 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test(expected = classOf[IllegalArgumentException])
+ @Test(expected = classOf[ValidationException])
def testAggregationOnNonExistingField(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
@@ -137,7 +136,7 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
TestBaseUtils.compareResultAsText(result.asJava, expected)
}
- @Test(expected = classOf[PlanGenException])
+ @Test(expected = classOf[ValidationException])
def testNonWorkingAggregationDataTypes(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
@@ -150,7 +149,7 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
t.collect()
}
- @Test(expected = classOf[UnsupportedOperationException])
+ @Test(expected = classOf[ValidationException])
def testNoNestedAggregations(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/ExpressionsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/ExpressionsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/ExpressionsITCase.scala
index fdbe3c9..9a0a035 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/ExpressionsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/ExpressionsITCase.scala
@@ -27,7 +27,7 @@ import org.apache.flink.api.scala.batch.utils.TableProgramsTestBase.TableConfigM
import org.apache.flink.api.scala.table._
import org.apache.flink.api.table.codegen.CodeGenException
import org.apache.flink.api.table.expressions.{Literal, Null}
-import org.apache.flink.api.table.{Row, TableEnvironment}
+import org.apache.flink.api.table.{Row, TableEnvironment, ValidationException}
import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
import org.apache.flink.test.util.TestBaseUtils
import org.junit.Assert._
@@ -144,7 +144,7 @@ class ExpressionsITCase(
TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test(expected = classOf[IllegalArgumentException])
+ @Test(expected = classOf[ValidationException])
def testEvalInvalidTypes(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val tEnv = TableEnvironment.getTableEnvironment(env, config)
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/FilterITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/FilterITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/FilterITCase.scala
index 5deff9e..ee0356f 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/FilterITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/FilterITCase.scala
@@ -24,7 +24,7 @@ import org.apache.flink.api.scala.batch.utils.TableProgramsTestBase.TableConfigM
import org.apache.flink.api.scala.table._
import org.apache.flink.api.scala.util.CollectionDataSets
import org.apache.flink.api.table.expressions.Literal
-import org.apache.flink.api.table.{Row, TableEnvironment}
+import org.apache.flink.api.table.{Row, TableEnvironment, ValidationException}
import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
import org.apache.flink.test.util.TestBaseUtils
import org.junit._
@@ -174,7 +174,7 @@ class FilterITCase(
TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test(expected = classOf[IllegalArgumentException])
+ @Test(expected = classOf[ValidationException])
def testFilterInvalidFieldName(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val tEnv = TableEnvironment.getTableEnvironment(env, config)
http://git-wip-us.apache.org/repos/asf/flink/blob/f0d543f8/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/GroupedAggregationsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/GroupedAggregationsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/GroupedAggregationsITCase.scala
index cbd42b2..8889b37 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/GroupedAggregationsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/GroupedAggregationsITCase.scala
@@ -21,7 +21,7 @@ package org.apache.flink.api.scala.batch.table
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.table._
import org.apache.flink.api.scala.util.CollectionDataSets
-import org.apache.flink.api.table.{Row, TableEnvironment}
+import org.apache.flink.api.table.{Row, TableEnvironment, ValidationException}
import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
import org.junit._
@@ -33,7 +33,7 @@ import scala.collection.JavaConverters._
@RunWith(classOf[Parameterized])
class GroupedAggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
- @Test(expected = classOf[IllegalArgumentException])
+ @Test(expected = classOf[ValidationException])
def testGroupingOnNonExistentField(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
@@ -45,7 +45,7 @@ class GroupedAggregationsITCase(mode: TestExecutionMode) extends MultipleProgram
.select('a.avg)
}
- @Test(expected = classOf[IllegalArgumentException])
+ @Test(expected = classOf[ValidationException])
def testGroupingInvalidSelection(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment