[FLINK-6884] [table] Improve confused exception information.

This closes apache#4100

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

@@ -718,16 +718,20 @@ abstract class TableEnvironment(val config: TableConfig) {
     // validate that at least the field types of physical and logical type match
     // we do that here to make sure that plan translation was correct
     if (schema.physicalTypeInfo != inputTypeInfo) {
-      throw TableException("The field types of physical and logical row types do not match." +
-        "This is a bug and should not happen. Please file an issue.")
+      throw TableException(
+        s"The field types of physical and logical row types do not match. " +
+        s"Physical type is [${schema.physicalTypeInfo}], Logical type is [${inputTypeInfo}]. " +
+        s"This is a bug and should not happen. Please file an issue.")
     }
 
     val fieldTypes = schema.physicalFieldTypeInfo
     val fieldNames = schema.physicalFieldNames
 
     // validate requested type
     if (requestedTypeInfo.getArity != fieldTypes.length) {
-      throw new TableException("Arity of result does not match requested type.")
+      throw new TableException(
+        s"Arity[${fieldTypes.length}] of result[${fieldTypes}] does not match " +
+        s"the number[${requestedTypeInfo.getArity}] of requested type[${requestedTypeInfo}].")
     }
 
     requestedTypeInfo match {
@@ -761,7 +765,7 @@ abstract class TableEnvironment(val config: TableConfig) {
       case at: AtomicType[_] =>
         if (fieldTypes.size != 1) {
           throw new TableException(s"Requested result type is an atomic type but " +
-            s"result has more or less than a single field.")
+            s"result[$fieldTypes] has more or less than a single field.")
         }
         val fieldTypeInfo = fieldTypes.head
         if (fieldTypeInfo != at) {
@@ -805,10 +809,6 @@ abstract class TableEnvironment(val config: TableConfig) {
   */
 object TableEnvironment {
 
-  // default names that can be used in in TableSources etc.
-  val DEFAULT_ROWTIME_ATTRIBUTE = "rowtime"
-  val DEFAULT_PROCTIME_ATTRIBUTE = "proctime"
-
   /**
     * Returns a [[JavaBatchTableEnv]] for a Java [[JavaBatchExecEnv]].
     *

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala

@@ -937,31 +937,40 @@ class CodeGenerator(
     : GeneratedExpression = {
     // initial type check
     if (returnType.getArity != fieldExprs.length) {
-      throw new CodeGenException("Arity of result type does not match number of expressions.")
+      throw new CodeGenException(
+        s"Arity[${returnType.getArity}] of result type[$returnType] does not match " +
+        s"number[${fieldExprs.length}] of expressions[$fieldExprs].")
     }
     if (resultFieldNames.length != fieldExprs.length) {
-      throw new CodeGenException("Arity of result field names does not match number of " +
-        "expressions.")
+      throw new CodeGenException(
+        s"Arity[${resultFieldNames.length}] of result field names[$resultFieldNames] does not " +
+        s"match number[${fieldExprs.length}] of expressions[$fieldExprs].")
     }
     // type check
     returnType match {
       case pt: PojoTypeInfo[_] =>
         fieldExprs.zipWithIndex foreach {
           case (fieldExpr, i) if fieldExpr.resultType != pt.getTypeAt(resultFieldNames(i)) =>
-            throw new CodeGenException("Incompatible types of expression and result type.")
+            throw new CodeGenException(
+              s"Incompatible types of expression and result type. Expression[$fieldExpr] type is " +
+              s"[${fieldExpr.resultType}], result type is [${pt.getTypeAt(resultFieldNames(i))}]")
 
           case _ => // ok
         }
 
       case ct: CompositeType[_] =>
         fieldExprs.zipWithIndex foreach {
           case (fieldExpr, i) if fieldExpr.resultType != ct.getTypeAt(i) =>
-            throw new CodeGenException("Incompatible types of expression and result type.")
+            throw new CodeGenException(
+              s"Incompatible types of expression and result type. Expression[$fieldExpr] type is " +
+              s"[${fieldExpr.resultType}], result type is [${ct.getTypeAt(i)}]")
           case _ => // ok
         }
 
       case at: AtomicType[_] if at != fieldExprs.head.resultType =>
-        throw new CodeGenException("Incompatible types of expression and result type.")
+        throw new CodeGenException(
+          s"Incompatible types of expression and result type. Expression[${fieldExprs.head}] " +
+          s"type is [${fieldExprs.head.resultType}], result type is [$at]")
 
       case _ => // ok
     }