flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject [2/5] flink git commit: [FLINK-3754] [tableAPI] Add validation phase to Table API before construction of RelNodes.
Date Tue, 17 May 2016 22:49:27 GMT
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


Mime
View raw message