Merge branch 'FLINK-4086'

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/Expression.scala

@@ -29,14 +29,14 @@ abstract class Expression extends TreeNode[Expression] {
     * Returns the [[TypeInformation]] for evaluating this expression.
     * It is sometimes not available until the expression is valid.
     */
-  def resultType: TypeInformation[_]
+  private[flink] def resultType: TypeInformation[_]
 
   /**
     * One pass validation of the expression tree in post order.
     */
-  lazy val valid: Boolean = childrenValid && validateInput().isSuccess
+  private[flink] lazy val valid: Boolean = childrenValid && validateInput().isSuccess
 
-  def childrenValid: Boolean = children.forall(_.valid)
+  private[flink] def childrenValid: Boolean = children.forall(_.valid)
 
   /**
     * Check input data types, inputs number or other properties specified by this expression.
@@ -44,17 +44,17 @@ abstract class Expression extends TreeNode[Expression] {
     * or `ValidationFailure` with supplement message explaining the error.
     * Note: we should only call this method until `childrenValid == true`
     */
-  def validateInput(): ExprValidationResult = ValidationSuccess
+  private[flink] def validateInput(): ExprValidationResult = ValidationSuccess
 
   /**
     * Convert Expression to its counterpart in Calcite, i.e. RexNode
     */
-  def toRexNode(implicit relBuilder: RelBuilder): RexNode =
+  private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode =
     throw new UnsupportedOperationException(
       s"${this.getClass.getName} cannot be transformed to RexNode"
     )
 
-  def checkEquals(other: Expression): Boolean = {
+  private[flink] def checkEquals(other: Expression): Boolean = {
     if (this.getClass != other.getClass) {
       false
     } else {
@@ -73,16 +73,16 @@ abstract class Expression extends TreeNode[Expression] {
 }
 
 abstract class BinaryExpression extends Expression {
-  def left: Expression
-  def right: Expression
-  def children = Seq(left, right)
+  private[flink] def left: Expression
+  private[flink] def right: Expression
+  private[flink] def children = Seq(left, right)
 }
 
 abstract class UnaryExpression extends Expression {
-  def child: Expression
-  def children = Seq(child)
+  private[flink] def child: Expression
+  private[flink] def children = Seq(child)
 }
 
 abstract class LeafExpression extends Expression {
-  val children = Nil
+  private[flink] val children = Nil
 }

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/InputTypeSpec.scala

@@ -36,9 +36,9 @@ trait InputTypeSpec extends Expression {
     *   def expectedTypes: Seq[TypeInformation[_]] = DOUBLE_TYPE_INFO :: DOUBLE_TYPE_INFO :: Nil
     * }}}
     */
-  def expectedTypes: Seq[TypeInformation[_]]
+  private[flink] def expectedTypes: Seq[TypeInformation[_]]
 
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     val typeMismatches = mutable.ArrayBuffer.empty[String]
     children.zip(expectedTypes).zipWithIndex.foreach { case ((e, tpe), i) =>
       if (e.resultType != tpe) {

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/aggregations.scala

@@ -29,69 +29,73 @@ abstract sealed class Aggregation extends UnaryExpression {
 
   override def toString = s"Aggregate($child)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode =
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode =
     throw new UnsupportedOperationException("Aggregate cannot be transformed to RexNode")
 
   /**
     * Convert Aggregate to its counterpart in Calcite, i.e. AggCall
     */
-  def toAggCall(name: String)(implicit relBuilder: RelBuilder): AggCall
+  private[flink] def toAggCall(name: String)(implicit relBuilder: RelBuilder): AggCall
 }
 
 case class Sum(child: Expression) extends Aggregation {
   override def toString = s"sum($child)"
 
-  override def toAggCall(name: String)(implicit relBuilder: RelBuilder): AggCall = {
+  override private[flink] def toAggCall(name: String)(implicit relBuilder: RelBuilder): AggCall = {
     relBuilder.aggregateCall(SqlStdOperatorTable.SUM, false, null, name, child.toRexNode)
   }
 
-  override def resultType = child.resultType
+  override private[flink] def resultType = child.resultType
 
-  override def validateInput = TypeCheckUtils.assertNumericExpr(child.resultType, "sum")
+  override private[flink] def validateInput =
+    TypeCheckUtils.assertNumericExpr(child.resultType, "sum")
 }
 
 case class Min(child: Expression) extends Aggregation {
   override def toString = s"min($child)"
 
-  override def toAggCall(name: String)(implicit relBuilder: RelBuilder): AggCall = {
+  override private[flink] def toAggCall(name: String)(implicit relBuilder: RelBuilder): AggCall = {
     relBuilder.aggregateCall(SqlStdOperatorTable.MIN, false, null, name, child.toRexNode)
   }
 
-  override def resultType = child.resultType
+  override private[flink] def resultType = child.resultType
 
-  override def validateInput = TypeCheckUtils.assertOrderableExpr(child.resultType, "min")
+  override private[flink] def validateInput =
+    TypeCheckUtils.assertOrderableExpr(child.resultType, "min")
 }
 
 case class Max(child: Expression) extends Aggregation {
   override def toString = s"max($child)"
 
-  override def toAggCall(name: String)(implicit relBuilder: RelBuilder): AggCall = {
+  override private[flink] def toAggCall(name: String)(implicit relBuilder: RelBuilder): AggCall = {
     relBuilder.aggregateCall(SqlStdOperatorTable.MAX, false, null, name, child.toRexNode)
   }
 
-  override def resultType = child.resultType
+  override private[flink] def resultType = child.resultType
 
-  override def validateInput = TypeCheckUtils.assertOrderableExpr(child.resultType, "max")
+  override private[flink] def validateInput =
+    TypeCheckUtils.assertOrderableExpr(child.resultType, "max")
 }
 
 case class Count(child: Expression) extends Aggregation {
   override def toString = s"count($child)"
 
-  override def toAggCall(name: String)(implicit relBuilder: RelBuilder): AggCall = {
+  override private[flink] def toAggCall(name: String)(implicit relBuilder: RelBuilder): AggCall = {
     relBuilder.aggregateCall(SqlStdOperatorTable.COUNT, false, null, name, child.toRexNode)
   }
 
-  override def resultType = BasicTypeInfo.LONG_TYPE_INFO
+  override private[flink] def resultType = BasicTypeInfo.LONG_TYPE_INFO
 }
 
 case class Avg(child: Expression) extends Aggregation {
   override def toString = s"avg($child)"
 
-  override def toAggCall(name: String)(implicit relBuilder: RelBuilder): AggCall = {
+  override private[flink] def toAggCall(name: String)(implicit relBuilder: RelBuilder): AggCall = {
     relBuilder.aggregateCall(SqlStdOperatorTable.AVG, false, null, name, child.toRexNode)
   }
 
-  override def resultType = child.resultType
+  override private[flink] def resultType = child.resultType
 
-  override def validateInput = TypeCheckUtils.assertNumericExpr(child.resultType, "avg")
+  override private[flink] def validateInput =
+    TypeCheckUtils.assertNumericExpr(child.resultType, "avg")
 }

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/arithmetic.scala

@@ -29,13 +29,13 @@ import org.apache.flink.api.table.validate._
 import scala.collection.JavaConversions._
 
 abstract class BinaryArithmetic extends BinaryExpression {
-  def sqlOperator: SqlOperator
+  private[flink] def sqlOperator: SqlOperator
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(sqlOperator, children.map(_.toRexNode))
   }
 
-  override def resultType: TypeInformation[_] =
+  override private[flink] def resultType: TypeInformation[_] =
     TypeCoercion.widerTypeOf(left.resultType, right.resultType) match {
       case Some(t) => t
       case None =>
@@ -43,7 +43,7 @@ abstract class BinaryArithmetic extends BinaryExpression {
     }
 
   // TODO: tighten this rule once we implemented type coercion rules during validation
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (!isNumeric(left.resultType) || !isNumeric(right.resultType)) {
       ValidationFailure(s"$this requires both operands Numeric, got " +
         s"${left.resultType} and ${right.resultType}")
@@ -56,9 +56,9 @@ abstract class BinaryArithmetic extends BinaryExpression {
 case class Plus(left: Expression, right: Expression) extends BinaryArithmetic {
   override def toString = s"($left + $right)"
 
-  val sqlOperator = SqlStdOperatorTable.PLUS
+  private[flink] val sqlOperator = SqlStdOperatorTable.PLUS
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     if(isString(left.resultType)) {
       val castedRight = Cast(right, BasicTypeInfo.STRING_TYPE_INFO)
       relBuilder.call(SqlStdOperatorTable.PLUS, left.toRexNode, castedRight.toRexNode)
@@ -73,7 +73,7 @@ case class Plus(left: Expression, right: Expression) extends BinaryArithmetic {
   }
 
   // TODO: tighten this rule once we implemented type coercion rules during validation
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (isString(left.resultType) || isString(right.resultType)) {
       ValidationSuccess
     } else if (!isNumeric(left.resultType) || !isNumeric(right.resultType)) {
@@ -88,36 +88,36 @@ case class Plus(left: Expression, right: Expression) extends BinaryArithmetic {
 case class UnaryMinus(child: Expression) extends UnaryExpression {
   override def toString = s"-($child)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.UNARY_MINUS, child.toRexNode)
   }
 
-  override def resultType = child.resultType
+  override private[flink] def resultType = child.resultType
 
-  override def validateInput(): ExprValidationResult =
+  override private[flink] def validateInput(): ExprValidationResult =
     TypeCheckUtils.assertNumericExpr(child.resultType, "unary minus")
 }
 
 case class Minus(left: Expression, right: Expression) extends BinaryArithmetic {
   override def toString = s"($left - $right)"
 
-  val sqlOperator = SqlStdOperatorTable.MINUS
+  private[flink] val sqlOperator = SqlStdOperatorTable.MINUS
 }
 
 case class Div(left: Expression, right: Expression) extends BinaryArithmetic {
   override def toString = s"($left / $right)"
 
-  val sqlOperator = SqlStdOperatorTable.DIVIDE
+  private[flink] val sqlOperator = SqlStdOperatorTable.DIVIDE
 }
 
 case class Mul(left: Expression, right: Expression) extends BinaryArithmetic {
   override def toString = s"($left * $right)"
 
-  val sqlOperator = SqlStdOperatorTable.MULTIPLY
+  private[flink] val sqlOperator = SqlStdOperatorTable.MULTIPLY
 }
 
 case class Mod(left: Expression, right: Expression) extends BinaryArithmetic {
   override def toString = s"($left % $right)"
 
-  val sqlOperator = SqlStdOperatorTable.MOD
+  private[flink] val sqlOperator = SqlStdOperatorTable.MOD
 }

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/call.scala

@@ -29,17 +29,17 @@ import org.apache.flink.api.table.validate.{ExprValidationResult, ValidationFail
   */
 case class Call(functionName: String, args: Seq[Expression]) extends Expression {
 
-  override def children: Seq[Expression] = args
+  override private[flink] def children: Seq[Expression] = args
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     throw new UnresolvedException(s"trying to convert UnresolvedFunction $functionName to RexNode")
   }
 
   override def toString = s"\\$functionName(${args.mkString(", ")})"
 
-  override def resultType =
+  override private[flink] def resultType =
     throw new UnresolvedException(s"calling resultType on UnresolvedFunction $functionName")
 
-  override def validateInput(): ExprValidationResult =
+  override private[flink] def validateInput(): ExprValidationResult =
     ValidationFailure(s"Unresolved function call: $functionName")
 }

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/cast.scala

@@ -28,16 +28,16 @@ case class Cast(child: Expression, resultType: TypeInformation[_]) extends Unary
 
   override def toString = s"$child.cast($resultType)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.cast(child.toRexNode, TypeConverter.typeInfoToSqlType(resultType))
   }
 
-  override def makeCopy(anyRefs: Array[AnyRef]): this.type = {
+  override private[flink] def makeCopy(anyRefs: Array[AnyRef]): this.type = {
     val child: Expression = anyRefs.head.asInstanceOf[Expression]
     copy(child, resultType).asInstanceOf[this.type]
   }
 
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (TypeCoercion.canCast(child.resultType, resultType)) {
       ValidationSuccess
     } else {

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/comparison.scala

@@ -28,93 +28,96 @@ import org.apache.flink.api.table.validate._
 import scala.collection.JavaConversions._
 
 abstract class BinaryComparison extends BinaryExpression {
-  def sqlOperator: SqlOperator
+  private[flink] def sqlOperator: SqlOperator
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(sqlOperator, children.map(_.toRexNode))
   }
 
-  override def resultType = BOOLEAN_TYPE_INFO
+  override private[flink] def resultType = BOOLEAN_TYPE_INFO
 
   // TODO: tighten this rule once we implemented type coercion rules during validation
-  override def validateInput(): ExprValidationResult = (left.resultType, right.resultType) match {
-    case (lType, rType) if isNumeric(lType) && isNumeric(rType) => ValidationSuccess
-    case (lType, rType) if isComparable(lType) && lType == rType => ValidationSuccess
-    case (lType, rType) =>
-      ValidationFailure(
-        s"Comparison is only supported for numeric types and " +
-          s"comparable types of same type, got $lType and $rType")
-  }
+  override private[flink] def validateInput(): ExprValidationResult =
+    (left.resultType, right.resultType) match {
+      case (lType, rType) if isNumeric(lType) && isNumeric(rType) => ValidationSuccess
+      case (lType, rType) if isComparable(lType) && lType == rType => ValidationSuccess
+      case (lType, rType) =>
+        ValidationFailure(
+          s"Comparison is only supported for numeric types and " +
+            s"comparable types of same type, got $lType and $rType")
+    }
 }
 
 case class EqualTo(left: Expression, right: Expression) extends BinaryComparison {
   override def toString = s"$left === $right"
 
-  val sqlOperator: SqlOperator = SqlStdOperatorTable.EQUALS
+  private[flink] val sqlOperator: SqlOperator = SqlStdOperatorTable.EQUALS
 
-  override def validateInput(): ExprValidationResult = (left.resultType, right.resultType) match {
-    case (lType, rType) if isNumeric(lType) && isNumeric(rType) => ValidationSuccess
-    // TODO widen this rule once we support custom objects as types (FLINK-3916)
-    case (lType, rType) if lType == rType => ValidationSuccess
-    case (lType, rType) =>
-      ValidationFailure(s"Equality predicate on incompatible types: $lType and $rType")
-  }
+  override private[flink] def validateInput(): ExprValidationResult =
+    (left.resultType, right.resultType) match {
+      case (lType, rType) if isNumeric(lType) && isNumeric(rType) => ValidationSuccess
+      // TODO widen this rule once we support custom objects as types (FLINK-3916)
+      case (lType, rType) if lType == rType => ValidationSuccess
+      case (lType, rType) =>
+        ValidationFailure(s"Equality predicate on incompatible types: $lType and $rType")
+    }
 }
 
 case class NotEqualTo(left: Expression, right: Expression) extends BinaryComparison {
   override def toString = s"$left !== $right"
 
-  val sqlOperator: SqlOperator = SqlStdOperatorTable.NOT_EQUALS
+  private[flink] val sqlOperator: SqlOperator = SqlStdOperatorTable.NOT_EQUALS
 
-  override def validateInput(): ExprValidationResult = (left.resultType, right.resultType) match {
-    case (lType, rType) if isNumeric(lType) && isNumeric(rType) => ValidationSuccess
-    // TODO widen this rule once we support custom objects as types (FLINK-3916)
-    case (lType, rType) if lType == rType => ValidationSuccess
-    case (lType, rType) =>
-      ValidationFailure(s"Inequality predicate on incompatible types: $lType and $rType")
-  }
+  override private[flink] def validateInput(): ExprValidationResult =
+    (left.resultType, right.resultType) match {
+      case (lType, rType) if isNumeric(lType) && isNumeric(rType) => ValidationSuccess
+      // TODO widen this rule once we support custom objects as types (FLINK-3916)
+      case (lType, rType) if lType == rType => ValidationSuccess
+      case (lType, rType) =>
+        ValidationFailure(s"Inequality predicate on incompatible types: $lType and $rType")
+    }
 }
 
 case class GreaterThan(left: Expression, right: Expression) extends BinaryComparison {
   override def toString = s"$left > $right"
 
-  val sqlOperator: SqlOperator = SqlStdOperatorTable.GREATER_THAN
+  private[flink] val sqlOperator: SqlOperator = SqlStdOperatorTable.GREATER_THAN
 }
 
 case class GreaterThanOrEqual(left: Expression, right: Expression) extends BinaryComparison {
   override def toString = s"$left >= $right"
 
-  val sqlOperator: SqlOperator = SqlStdOperatorTable.GREATER_THAN_OR_EQUAL
+  private[flink] val sqlOperator: SqlOperator = SqlStdOperatorTable.GREATER_THAN_OR_EQUAL
 }
 
 case class LessThan(left: Expression, right: Expression) extends BinaryComparison {
   override def toString = s"$left < $right"
 
-  val sqlOperator: SqlOperator = SqlStdOperatorTable.LESS_THAN
+  private[flink] val sqlOperator: SqlOperator = SqlStdOperatorTable.LESS_THAN
 }
 
 case class LessThanOrEqual(left: Expression, right: Expression) extends BinaryComparison {
   override def toString = s"$left <= $right"
 
-  val sqlOperator: SqlOperator = SqlStdOperatorTable.LESS_THAN_OR_EQUAL
+  private[flink] val sqlOperator: SqlOperator = SqlStdOperatorTable.LESS_THAN_OR_EQUAL
 }
 
 case class IsNull(child: Expression) extends UnaryExpression {
   override def toString = s"($child).isNull"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.isNull(child.toRexNode)
   }
 
-  override def resultType = BOOLEAN_TYPE_INFO
+  override private[flink] def resultType = BOOLEAN_TYPE_INFO
 }
 
 case class IsNotNull(child: Expression) extends UnaryExpression {
   override def toString = s"($child).isNotNull"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.isNotNull(child.toRexNode)
   }
 
-  override def resultType = BOOLEAN_TYPE_INFO
+  override private[flink] def resultType = BOOLEAN_TYPE_INFO
 }

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/fieldExpression.scala

@@ -25,26 +25,27 @@ import org.apache.flink.api.table.UnresolvedException
 import org.apache.flink.api.table.validate.{ExprValidationResult, ValidationFailure}
 
 trait NamedExpression extends Expression {
-  def name: String
-  def toAttribute: Attribute
+  private[flink] def name: String
+  private[flink] def toAttribute: Attribute
 }
 
 abstract class Attribute extends LeafExpression with NamedExpression {
-  override def toAttribute: Attribute = this
+  override private[flink] def toAttribute: Attribute = this
 
-  def withName(newName: String): Attribute
+  private[flink] def withName(newName: String): Attribute
 }
 
 case class UnresolvedFieldReference(name: String) extends Attribute {
 
   override def toString = "\"" + name
 
-  override def withName(newName: String): Attribute = UnresolvedFieldReference(newName)
+  override private[flink] def withName(newName: String): Attribute =
+    UnresolvedFieldReference(newName)
 
-  override def resultType: TypeInformation[_] =
+  override private[flink] def resultType: TypeInformation[_] =
     throw new UnresolvedException(s"calling resultType on ${this.getClass}")
 
-  override def validateInput(): ExprValidationResult =
+  override private[flink] def validateInput(): ExprValidationResult =
     ValidationFailure(s"Unresolved reference $name")
 }
 
@@ -54,11 +55,11 @@ case class ResolvedFieldReference(
 
   override def toString = s"'$name"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.field(name)
   }
 
-  override def withName(newName: String): Attribute = {
+  override private[flink] def withName(newName: String): Attribute = {
     if (newName == name) {
       this
     } else {
@@ -72,18 +73,18 @@ case class Alias(child: Expression, name: String)
 
   override def toString = s"$child as '$name"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.alias(child.toRexNode, name)
   }
 
-  override def resultType: TypeInformation[_] = child.resultType
+  override private[flink] def resultType: TypeInformation[_] = child.resultType
 
-  override def makeCopy(anyRefs: Array[AnyRef]): this.type = {
+  override private[flink] def makeCopy(anyRefs: Array[AnyRef]): this.type = {
     val child: Expression = anyRefs.head.asInstanceOf[Expression]
     copy(child, name).asInstanceOf[this.type]
   }
 
-  override def toAttribute: Attribute = {
+  override private[flink] def toAttribute: Attribute = {
     if (valid) {
       ResolvedFieldReference(name, child.resultType)
     } else {
@@ -94,14 +95,14 @@ case class Alias(child: Expression, name: String)
 
 case class UnresolvedAlias(child: Expression) extends UnaryExpression with NamedExpression {
 
-  override def name: String =
+  override private[flink] def name: String =
     throw new UnresolvedException("Invalid call to name on UnresolvedAlias")
 
-  override def toAttribute: Attribute =
+  override private[flink] def toAttribute: Attribute =
     throw new UnresolvedException("Invalid call to toAttribute on UnresolvedAlias")
 
-  override def resultType: TypeInformation[_] =
+  override private[flink] def resultType: TypeInformation[_] =
     throw new UnresolvedException("Invalid call to resultType on UnresolvedAlias")
 
-  override lazy val valid = false
+  override private[flink] lazy val valid = false
 }

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/literals.scala

@@ -27,7 +27,7 @@ import org.apache.flink.api.common.typeinfo.{SqlTimeTypeInfo, BasicTypeInfo, Typ
 import org.apache.flink.api.table.typeutils.TypeConverter
 
 object Literal {
-  def apply(l: Any): Literal = l match {
+  private[flink] def apply(l: Any): Literal = l match {
     case i: Int => Literal(i, BasicTypeInfo.INT_TYPE_INFO)
     case s: Short => Literal(s, BasicTypeInfo.SHORT_TYPE_INFO)
     case b: Byte => Literal(b, BasicTypeInfo.BYTE_TYPE_INFO)
@@ -48,7 +48,7 @@ object Literal {
 case class Literal(value: Any, resultType: TypeInformation[_]) extends LeafExpression {
   override def toString = s"$value"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     resultType match {
       case BasicTypeInfo.BIG_DEC_TYPE_INFO =>
         val bigDecValue = value.asInstanceOf[java.math.BigDecimal]
@@ -80,7 +80,7 @@ case class Literal(value: Any, resultType: TypeInformation[_]) extends LeafExpre
 case class Null(resultType: TypeInformation[_]) extends LeafExpression {
   override def toString = s"null"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.getRexBuilder.makeNullLiteral(TypeConverter.typeInfoToSqlType(resultType))
   }
 }

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/logic.scala

@@ -25,9 +25,9 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo
 import org.apache.flink.api.table.validate._
 
 abstract class BinaryPredicate extends BinaryExpression {
-  override def resultType = BasicTypeInfo.BOOLEAN_TYPE_INFO
+  override private[flink] def resultType = BasicTypeInfo.BOOLEAN_TYPE_INFO
 
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (left.resultType == BasicTypeInfo.BOOLEAN_TYPE_INFO &&
         right.resultType == BasicTypeInfo.BOOLEAN_TYPE_INFO) {
       ValidationSuccess
@@ -42,13 +42,13 @@ case class Not(child: Expression) extends UnaryExpression {
 
   override def toString = s"!($child)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.not(child.toRexNode)
   }
 
-  override def resultType = BasicTypeInfo.BOOLEAN_TYPE_INFO
+  override private[flink] def resultType = BasicTypeInfo.BOOLEAN_TYPE_INFO
 
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (child.resultType == BasicTypeInfo.BOOLEAN_TYPE_INFO) {
       ValidationSuccess
     } else {
@@ -62,7 +62,7 @@ case class And(left: Expression, right: Expression) extends BinaryPredicate {
 
   override def toString = s"$left && $right"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.and(left.toRexNode, right.toRexNode)
   }
 }
@@ -71,7 +71,7 @@ case class Or(left: Expression, right: Expression) extends BinaryPredicate {
 
   override def toString = s"$left || $right"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.or(left.toRexNode, right.toRexNode)
   }
 }
@@ -81,20 +81,20 @@ case class If(
     ifTrue: Expression,
     ifFalse: Expression)
   extends Expression {
-  def children = Seq(condition, ifTrue, ifFalse)
+  private[flink] def children = Seq(condition, ifTrue, ifFalse)
 
-  override def resultType = ifTrue.resultType
+  override private[flink] def resultType = ifTrue.resultType
 
   override def toString = s"($condition)? $ifTrue : $ifFalse"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     val c = condition.toRexNode
     val t = ifTrue.toRexNode
     val f = ifFalse.toRexNode
     relBuilder.call(SqlStdOperatorTable.CASE, c, t, f)
   }
 
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (condition.resultType == BasicTypeInfo.BOOLEAN_TYPE_INFO &&
         ifTrue.resultType == ifFalse.resultType) {
       ValidationSuccess

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/mathExpressions.scala

@@ -27,89 +27,90 @@ import org.apache.flink.api.table.typeutils.TypeCheckUtils
 import org.apache.flink.api.table.validate._
 
 case class Abs(child: Expression) extends UnaryExpression {
-  override def resultType: TypeInformation[_] = child.resultType
+  override private[flink] def resultType: TypeInformation[_] = child.resultType
 
-  override def validateInput(): ExprValidationResult =
+  override private[flink] def validateInput(): ExprValidationResult =
     TypeCheckUtils.assertNumericExpr(child.resultType, "Abs")
 
-  override def toString(): String = s"abs($child)"
+  override def toString: String = s"abs($child)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.ABS, child.toRexNode)
   }
 }
 
 case class Ceil(child: Expression) extends UnaryExpression {
-  override def resultType: TypeInformation[_] = LONG_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = LONG_TYPE_INFO
 
-  override def validateInput(): ExprValidationResult =
+  override private[flink] def validateInput(): ExprValidationResult =
     TypeCheckUtils.assertNumericExpr(child.resultType, "Ceil")
 
-  override def toString(): String = s"ceil($child)"
+  override def toString: String = s"ceil($child)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.CEIL, child.toRexNode)
   }
 }
 
 case class Exp(child: Expression) extends UnaryExpression with InputTypeSpec {
-  override def resultType: TypeInformation[_] = DOUBLE_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = DOUBLE_TYPE_INFO
 
-  override def expectedTypes: Seq[TypeInformation[_]] = DOUBLE_TYPE_INFO :: Nil
+  override private[flink] def expectedTypes: Seq[TypeInformation[_]] = DOUBLE_TYPE_INFO :: Nil
 
-  override def toString(): String = s"exp($child)"
+  override def toString: String = s"exp($child)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.EXP, child.toRexNode)
   }
 }
 
 
 case class Floor(child: Expression) extends UnaryExpression {
-  override def resultType: TypeInformation[_] = LONG_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = LONG_TYPE_INFO
 
-  override def validateInput(): ExprValidationResult =
+  override private[flink] def validateInput(): ExprValidationResult =
     TypeCheckUtils.assertNumericExpr(child.resultType, "Floor")
 
-  override def toString(): String = s"floor($child)"
+  override def toString: String = s"floor($child)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.FLOOR, child.toRexNode)
   }
 }
 
 case class Log10(child: Expression) extends UnaryExpression with InputTypeSpec {
-  override def resultType: TypeInformation[_] = DOUBLE_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = DOUBLE_TYPE_INFO
 
-  override def expectedTypes: Seq[TypeInformation[_]] = DOUBLE_TYPE_INFO :: Nil
+  override private[flink] def expectedTypes: Seq[TypeInformation[_]] = DOUBLE_TYPE_INFO :: Nil
 
-  override def toString(): String = s"log10($child)"
+  override def toString: String = s"log10($child)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.LOG10, child.toRexNode)
   }
 }
 
 case class Ln(child: Expression) extends UnaryExpression with InputTypeSpec {
-  override def resultType: TypeInformation[_] = DOUBLE_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = DOUBLE_TYPE_INFO
 
-  override def expectedTypes: Seq[TypeInformation[_]] = DOUBLE_TYPE_INFO :: Nil
+  override private[flink] def expectedTypes: Seq[TypeInformation[_]] = DOUBLE_TYPE_INFO :: Nil
 
-  override def toString(): String = s"ln($child)"
+  override def toString: String = s"ln($child)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.LN, child.toRexNode)
   }
 }
 
 case class Power(left: Expression, right: Expression) extends BinaryExpression with InputTypeSpec {
-  override def resultType: TypeInformation[_] = DOUBLE_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = DOUBLE_TYPE_INFO
 
-  override def expectedTypes: Seq[TypeInformation[_]] = DOUBLE_TYPE_INFO :: DOUBLE_TYPE_INFO :: Nil
+  override private[flink] def expectedTypes: Seq[TypeInformation[_]] =
+    DOUBLE_TYPE_INFO :: DOUBLE_TYPE_INFO :: Nil
 
-  override def toString(): String = s"pow($left, $right)"
+  override def toString: String = s"pow($left, $right)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.POWER, left.toRexNode, right.toRexNode)
   }
 }

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/ordering.scala

@@ -24,7 +24,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.table.validate._
 
 abstract class Ordering extends UnaryExpression {
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (!child.isInstanceOf[NamedExpression]) {
       ValidationFailure(s"Sort should only based on field reference")
     } else {
@@ -36,19 +36,19 @@ abstract class Ordering extends UnaryExpression {
 case class Asc(child: Expression) extends Ordering {
   override def toString: String = s"($child).asc"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     child.toRexNode
   }
 
-  override def resultType: TypeInformation[_] = child.resultType
+  override private[flink] def resultType: TypeInformation[_] = child.resultType
 }
 
 case class Desc(child: Expression) extends Ordering {
   override def toString: String = s"($child).desc"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.desc(child.toRexNode)
   }
 
-  override def resultType: TypeInformation[_] = child.resultType
+  override private[flink] def resultType: TypeInformation[_] = child.resultType
 }

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/stringExpressions.scala

@@ -31,9 +31,9 @@ import org.apache.flink.api.table.validate._
   * Returns the length of this `str`.
   */
 case class CharLength(child: Expression) extends UnaryExpression {
-  override def resultType: TypeInformation[_] = INT_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = INT_TYPE_INFO
 
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (child.resultType == STRING_TYPE_INFO) {
       ValidationSuccess
     } else {
@@ -41,9 +41,9 @@ case class CharLength(child: Expression) extends UnaryExpression {
     }
   }
 
-  override def toString(): String = s"($child).charLength()"
+  override def toString: String = s"($child).charLength()"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.CHAR_LENGTH, child.toRexNode)
   }
 }
@@ -53,9 +53,9 @@ case class CharLength(child: Expression) extends UnaryExpression {
   * All other letters are in lowercase. Words are delimited by white space.
   */
 case class InitCap(child: Expression) extends UnaryExpression {
-  override def resultType: TypeInformation[_] = STRING_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = STRING_TYPE_INFO
 
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (child.resultType == STRING_TYPE_INFO) {
       ValidationSuccess
     } else {
@@ -63,9 +63,9 @@ case class InitCap(child: Expression) extends UnaryExpression {
     }
   }
 
-  override def toString(): String = s"($child).initCap()"
+  override def toString: String = s"($child).initCap()"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.INITCAP, child.toRexNode)
   }
 }
@@ -74,12 +74,12 @@ case class InitCap(child: Expression) extends UnaryExpression {
   * Returns true if `str` matches `pattern`.
   */
 case class Like(str: Expression, pattern: Expression) extends BinaryExpression {
-  def left: Expression = str
-  def right: Expression = pattern
+  private[flink] def left: Expression = str
+  private[flink] def right: Expression = pattern
 
-  override def resultType: TypeInformation[_] = BOOLEAN_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = BOOLEAN_TYPE_INFO
 
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (str.resultType == STRING_TYPE_INFO && pattern.resultType == STRING_TYPE_INFO) {
       ValidationSuccess
     } else {
@@ -88,9 +88,9 @@ case class Like(str: Expression, pattern: Expression) extends BinaryExpression {
     }
   }
 
-  override def toString(): String = s"($str).like($pattern)"
+  override def toString: String = s"($str).like($pattern)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.LIKE, children.map(_.toRexNode))
   }
 }
@@ -99,9 +99,9 @@ case class Like(str: Expression, pattern: Expression) extends BinaryExpression {
   * Returns str with all characters changed to lowercase.
   */
 case class Lower(child: Expression) extends UnaryExpression {
-  override def resultType: TypeInformation[_] = STRING_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = STRING_TYPE_INFO
 
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (child.resultType == STRING_TYPE_INFO) {
       ValidationSuccess
     } else {
@@ -109,9 +109,9 @@ case class Lower(child: Expression) extends UnaryExpression {
     }
   }
 
-  override def toString(): String = s"($child).toLowerCase()"
+  override def toString: String = s"($child).toLowerCase()"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.LOWER, child.toRexNode)
   }
 }
@@ -120,12 +120,12 @@ case class Lower(child: Expression) extends UnaryExpression {
   * Returns true if `str` is similar to `pattern`.
   */
 case class Similar(str: Expression, pattern: Expression) extends BinaryExpression {
-  def left: Expression = str
-  def right: Expression = pattern
+  private[flink] def left: Expression = str
+  private[flink] def right: Expression = pattern
 
-  override def resultType: TypeInformation[_] = BOOLEAN_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = BOOLEAN_TYPE_INFO
 
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (str.resultType == STRING_TYPE_INFO && pattern.resultType == STRING_TYPE_INFO) {
       ValidationSuccess
     } else {
@@ -134,9 +134,9 @@ case class Similar(str: Expression, pattern: Expression) extends BinaryExpressio
     }
   }
 
-  override def toString(): String = s"($str).similarTo($pattern)"
+  override def toString: String = s"($str).similarTo($pattern)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.SIMILAR_TO, children.map(_.toRexNode))
   }
 }
@@ -151,16 +151,16 @@ case class SubString(
 
   def this(str: Expression, begin: Expression) = this(str, begin, CharLength(str))
 
-  override def children: Seq[Expression] = str :: begin :: end :: Nil
+  override private[flink] def children: Seq[Expression] = str :: begin :: end :: Nil
 
-  override def resultType: TypeInformation[_] = STRING_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = STRING_TYPE_INFO
 
-  override def expectedTypes: Seq[TypeInformation[_]] =
+  override private[flink] def expectedTypes: Seq[TypeInformation[_]] =
     Seq(STRING_TYPE_INFO, INT_TYPE_INFO, INT_TYPE_INFO)
 
-  override def toString(): String = s"$str.subString($begin, $end)"
+  override def toString: String = s"$str.subString($begin, $end)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.SUBSTRING, children.map(_.toRexNode))
   }
 }
@@ -174,16 +174,16 @@ case class Trim(
     trimString: Expression,
     str: Expression) extends Expression with InputTypeSpec {
 
-  override def children: Seq[Expression] = trimFlag :: trimString :: str :: Nil
+  override private[flink] def children: Seq[Expression] = trimFlag :: trimString :: str :: Nil
 
-  override def resultType: TypeInformation[_] = STRING_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = STRING_TYPE_INFO
 
-  override def expectedTypes: Seq[TypeInformation[_]] =
+  override private[flink] def expectedTypes: Seq[TypeInformation[_]] =
     Seq(INT_TYPE_INFO, STRING_TYPE_INFO, STRING_TYPE_INFO)
 
-  override def toString(): String = s"trim($trimFlag, $trimString, $str)"
+  override def toString: String = s"trim($trimFlag, $trimString, $str)"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.TRIM, children.map(_.toRexNode))
   }
 }
@@ -202,9 +202,9 @@ object TrimConstants {
   * Returns str with all characters changed to uppercase.
   */
 case class Upper(child: Expression) extends UnaryExpression {
-  override def resultType: TypeInformation[_] = STRING_TYPE_INFO
+  override private[flink] def resultType: TypeInformation[_] = STRING_TYPE_INFO
 
-  override def validateInput(): ExprValidationResult = {
+  override private[flink] def validateInput(): ExprValidationResult = {
     if (child.resultType == STRING_TYPE_INFO) {
       ValidationSuccess
     } else {
@@ -212,9 +212,9 @@ case class Upper(child: Expression) extends UnaryExpression {
     }
   }
 
-  override def toString(): String = s"($child).toUpperCase()"
+  override def toString: String = s"($child).toUpperCase()"
 
-  override def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
     relBuilder.call(SqlStdOperatorTable.UPPER, child.toRexNode)
   }
 }

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/trees/TreeNode.scala

@@ -28,17 +28,17 @@ 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]
+  private[flink] def children: Seq[A]
 
   /**
    * Tests for equality by first testing for reference equality.
    */
-  def fastEquals(other: TreeNode[_]): Boolean = this.eq(other) || this == other
+  private[flink] def fastEquals(other: TreeNode[_]): Boolean = this.eq(other) || this == other
 
   /**
     * Do tree transformation in post order.
     */
-  def postOrderTransform(rule: PartialFunction[A, A]): A = {
+  private[flink] def postOrderTransform(rule: PartialFunction[A, A]): A = {
     def childrenTransform(rule: PartialFunction[A, A]): A = {
       var changed = false
       val newArgs = productIterator.map {
@@ -78,7 +78,7 @@ abstract class TreeNode[A <: TreeNode[A]] extends Product { self: A =>
   /**
     * Runs the given function first on the node and then recursively on all its children.
     */
-  def preOrderVisit(f: A => Unit): Unit = {
+  private[flink] def preOrderVisit(f: A => Unit): Unit = {
     f(this)
     children.foreach(_.preOrderVisit(f))
   }
@@ -87,7 +87,7 @@ abstract class TreeNode[A <: TreeNode[A]] extends Product { self: A =>
    * Creates a new copy of this expression with new children. This is used during transformation
    * if children change.
    */
-  def makeCopy(newArgs: Array[AnyRef]): A = {
+  private[flink] 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}")