[SPARK-24536] Validate that an evaluated limit clause cannot be null

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

@@ -62,6 +62,9 @@ trait CheckAnalysis extends PredicateHelper {
 
   private def checkLimitClause(limitExpr: Expression): Unit = {
     limitExpr match {
+      case e if e.nullable => failAnalysis(
+        "The limit expression must not be nullable, but got " +
+          limitExpr.sql)
       case e if !e.foldable => failAnalysis(
         "The limit expression must evaluate to a constant value, but got " +
           limitExpr.sql)

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

@@ -399,6 +399,12 @@ class AnalysisErrorSuite extends AnalysisTest {
     "Generators are not supported outside the SELECT clause, but got: Sort" :: Nil
   )
 
+  errorTest(
+    "limit clause must not be nullable",
+    testRelation.limit(Literal(null, IntegerType)),
+    "The limit expression must not be nullable, but got " :: Nil
+  )
+
   errorTest(
     "num_rows in limit clause must be equal to or greater than 0",
     listRelation.limit(-1),

sql/core/src/test/resources/sql-tests/inputs/limit.sql

@@ -13,6 +13,9 @@ SELECT * FROM testdata LIMIT CAST(1 AS int);
 SELECT * FROM testdata LIMIT -1;
 SELECT * FROM testData TABLESAMPLE (-1 ROWS);
 
+-- limit may not be nullable
+SELECT * FROM testdata LIMIT CAST(NULL AS INT);
+
 -- limit must be foldable
 SELECT * FROM testdata LIMIT key > 3;
 

sql/core/src/test/resources/sql-tests/results/limit.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 12
+-- Number of queries: 13
 
 
 -- !query 0
@@ -66,44 +66,53 @@ The limit expression must be equal to or greater than 0, but got -1;
 
 
 -- !query 7
-SELECT * FROM testdata LIMIT key > 3
+SELECT * FROM testdata LIMIT CAST(NULL AS INT)
 -- !query 7 schema
 struct<>
 -- !query 7 output
 org.apache.spark.sql.AnalysisException
-The limit expression must evaluate to a constant value, but got (testdata.`key` > 3);
+The limit expression must not be nullable, but got CAST(NULL AS INT);
 
 
 -- !query 8
-SELECT * FROM testdata LIMIT true
+SELECT * FROM testdata LIMIT key > 3
 -- !query 8 schema
 struct<>
 -- !query 8 output
 org.apache.spark.sql.AnalysisException
-The limit expression must be integer type, but got boolean;
+The limit expression must evaluate to a constant value, but got (testdata.`key` > 3);
 
 
 -- !query 9
-SELECT * FROM testdata LIMIT 'a'
+SELECT * FROM testdata LIMIT true
 -- !query 9 schema
 struct<>
 -- !query 9 output
 org.apache.spark.sql.AnalysisException
-The limit expression must be integer type, but got string;
+The limit expression must be integer type, but got boolean;
 
 
 -- !query 10
-SELECT * FROM (SELECT * FROM range(10) LIMIT 5) WHERE id > 3
+SELECT * FROM testdata LIMIT 'a'
 -- !query 10 schema
-struct<id:bigint>
+struct<>
 -- !query 10 output
-4
+org.apache.spark.sql.AnalysisException
+The limit expression must be integer type, but got string;
 
 
 -- !query 11
-SELECT * FROM testdata WHERE key < 3 LIMIT ALL
+SELECT * FROM (SELECT * FROM range(10) LIMIT 5) WHERE id > 3
 -- !query 11 schema
-struct<key:int,value:string>
+struct<id:bigint>
 -- !query 11 output
+4
+
+
+-- !query 12
+SELECT * FROM testdata WHERE key < 3 LIMIT ALL
+-- !query 12 schema
+struct<key:int,value:string>
+-- !query 12 output
 1	1
 2	2