nullability and remove rand

Author: petermaxlee

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveInlineTables.scala

@@ -17,8 +17,10 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
+import scala.util.control.NonFatal
+
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{Cast, InterpretedProjection, Unevaluable}
+import org.apache.spark.sql.catalyst.expressions.Cast
 import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.types.{StructField, StructType}
@@ -34,26 +36,6 @@ object ResolveInlineTables extends Rule[LogicalPlan] {
       convert(table)
   }
 
-  /**
-   * Validates that all inline table data are valid expressions that can be evaluated.
-   *
-   * This is package visible for unit testing.
-   */
-  private[analysis] def validateInputEvaluable(table: UnresolvedInlineTable): Unit = {
-    table.rows.foreach { row =>
-      row.foreach { e =>
-        // We want to support foldable expressions and nondeterministic expressions (e.g. rand)
-        // that are evaluable.
-        // Note that there are expressions that are evaluable but not actually valid for inline
-        // tables. One example is AttributeReference. However, since UnresolvedInlineTable is a
-        // leaf node, the analyzer would not resolve UnresolvedAttribute into AttributeReference.
-        if (!e.resolved || e.isInstanceOf[Unevaluable]) {
-          e.failAnalysis(s"cannot evaluate expression ${e.sql} in inline table definition")
-        }
-      }
-    }
-  }
-
   /**
    * Validates the input data dimension:
    * 1. All rows have the same cardinality.
@@ -72,6 +54,23 @@ object ResolveInlineTables extends Rule[LogicalPlan] {
     }
   }
 
+  /**
+   * Validates that all inline table data are valid expressions that can be evaluated
+   * (in this they must be foldable).
+   *
+   * This is package visible for unit testing.
+   */
+  private[analysis] def validateInputEvaluable(table: UnresolvedInlineTable): Unit = {
+    table.rows.foreach { row =>
+      row.foreach { e =>
+        // Note that nondeterministic expressions are not supported since they are not foldable.
+        if (!e.resolved || !e.foldable) {
+          e.failAnalysis(s"cannot evaluate expression ${e.sql} in inline table definition")
+        }
+      }
+    }
+  }
+
   /**
    * Convert a valid (with right shape and foldable inputs) [[UnresolvedInlineTable]]
    * into a [[LocalRelation]].
@@ -81,24 +80,33 @@ object ResolveInlineTables extends Rule[LogicalPlan] {
    * This is package visible for unit testing.
    */
   private[analysis] def convert(table: UnresolvedInlineTable): LocalRelation = {
-    // For each column, traverse all the values and find a common data type.
-    val targetTypes = table.rows.transpose.zip(table.names).map { case (column, name) =>
+    // For each column, traverse all the values and find a common data type and nullability.
+    val fields = table.rows.transpose.zip(table.names).map { case (column, name) =>
       val inputTypes = column.map(_.dataType)
-      TypeCoercion.findWiderTypeWithoutStringPromotion(inputTypes).getOrElse {
+      val tpe = TypeCoercion.findWiderTypeWithoutStringPromotion(inputTypes).getOrElse {
         table.failAnalysis(s"incompatible types found in column $name for inline table")
       }
+      StructField(name, tpe, nullable = column.exists(_.nullable))
     }
-    assert(targetTypes.size == table.names.size)
+    val attributes = StructType(fields).toAttributes
+    assert(fields.size == table.names.size)
 
     val newRows: Seq[InternalRow] = table.rows.map { row =>
-      new InterpretedProjection(row.zipWithIndex.map { case (e, ci) =>
-        val targetType = targetTypes(ci)
-        if (e.dataType.sameType(targetType)) e else Cast(e, targetType)
-      }).apply(null)
+      InternalRow.fromSeq(row.zipWithIndex.map { case (e, ci) =>
+        val targetType = fields(ci).dataType
+        try {
+          if (e.dataType.sameType(targetType)) {
+            e.eval()
+          } else {
+            Cast(e, targetType).eval()
+          }
+        } catch {
+          case NonFatal(ex) =>
+            table.failAnalysis(s"failed to evaluate expression ${e.sql}: ${ex.getMessage}")
+        }
+      })
     }
 
-    val attributes = StructType(targetTypes.zip(table.names)
-      .map { case (typ, name) => StructField(name, typ) }).toAttributes
     LocalRelation(attributes, newRows)
   }
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveInlineTablesSuite.scala

@@ -23,7 +23,7 @@ import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.expressions.{Literal, Rand}
 import org.apache.spark.sql.catalyst.expressions.aggregate.Count
 import org.apache.spark.sql.catalyst.plans.PlanTest
-import org.apache.spark.sql.types.LongType
+import org.apache.spark.sql.types.{LongType, NullType}
 
 /**
  * Unit tests for [[ResolveInlineTables]]. Note that there are also test cases defined in
@@ -38,20 +38,22 @@ class ResolveInlineTablesSuite extends PlanTest with BeforeAndAfter {
     ResolveInlineTables.validateInputEvaluable(
       UnresolvedInlineTable(Seq("c1", "c2"), Seq(Seq(lit(1)))))
 
-    // nondeterministic (rand) should be fine
-    ResolveInlineTables.validateInputEvaluable(
-      UnresolvedInlineTable(Seq("c1", "c2"), Seq(Seq(Rand(1)))))
+    // nondeterministic (rand) should not work
+    intercept[AnalysisException] {
+      ResolveInlineTables.validateInputEvaluable(
+        UnresolvedInlineTable(Seq("c1"), Seq(Seq(Rand(1)))))
+    }
 
     // aggregate should not work
     intercept[AnalysisException] {
       ResolveInlineTables.validateInputEvaluable(
-        UnresolvedInlineTable(Seq("c1", "c2"), Seq(Seq(Count(lit(1))))))
+        UnresolvedInlineTable(Seq("c1"), Seq(Seq(Count(lit(1))))))
     }
 
     // unresolved attribute should not work
     intercept[AnalysisException] {
       ResolveInlineTables.validateInputEvaluable(
-        UnresolvedInlineTable(Seq("c1", "c2"), Seq(Seq(UnresolvedAttribute("A")))))
+        UnresolvedInlineTable(Seq("c1"), Seq(Seq(UnresolvedAttribute("A")))))
     }
   }
 
@@ -86,4 +88,14 @@ class ResolveInlineTablesSuite extends PlanTest with BeforeAndAfter {
     assert(converted.data(0).getLong(0) == 1L)
     assert(converted.data(1).getLong(0) == 2L)
   }
+
+  test("nullability inference in convert") {
+    val table1 = UnresolvedInlineTable(Seq("c1"), Seq(Seq(lit(1)), Seq(lit(2L))))
+    val converted1 = ResolveInlineTables.convert(table1)
+    assert(!converted1.schema.fields(0).nullable)
+
+    val table2 = UnresolvedInlineTable(Seq("c1"), Seq(Seq(lit(1)), Seq(Literal(null, NullType))))
+    val converted2 = ResolveInlineTables.convert(table2)
+    assert(converted2.schema.fields(0).nullable)
+  }
 }

sql/core/src/test/resources/sql-tests/results/inline-table.sql.out

@@ -94,10 +94,10 @@ two	3.0
 -- !query 10
 select * from values ("one", rand(5)), ("two", 3.0D) as data(a, b)
 -- !query 10 schema
-struct<a:string,b:double>
+struct<>
 -- !query 10 output
-one	0.087440518337355
-two	3.0
+org.apache.spark.sql.AnalysisException
+cannot evaluate expression rand(5) in inline table definition; line 1 pos 29
 
 
 -- !query 11