[SPARK-16672][SQL] SQLBuilder should not raise exceptions on EXISTS queries

## What changes were proposed in this pull request?

Currently, `SQLBuilder` raises `empty.reduceLeft` exceptions on *unoptimized* `EXISTS` queries. We had better prevent this.
```scala
scala> sql("CREATE TABLE t1(a int)")
scala> val df = sql("select * from t1 b where exists (select * from t1 a)")
scala> new org.apache.spark.sql.catalyst.SQLBuilder(df).toSQL
java.lang.UnsupportedOperationException: empty.reduceLeft
```

## How was this patch tested?

Pass the Jenkins tests with a new test suite.

Author: Dongjoon Hyun <dongjoon@apache.org>

Closes #14307 from dongjoon-hyun/SPARK-16672.

3 files changed, 21 insertions, 2 deletions

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/SQLBuilder.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/SQLBuilder.scala
index a8cc72f2e7..9a02e3c8f3 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/SQLBuilder.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/SQLBuilder.scala
@@ -512,8 +512,13 @@ class SQLBuilder(logicalPlan: LogicalPlan) extends Logging {
           ScalarSubquery(rewrite, Seq.empty, exprId)
 
         case PredicateSubquery(query, conditions, false, exprId) =>
-          val plan = Project(Seq(Alias(Literal(1), "1")()),
-            Filter(conditions.reduce(And), addSubqueryIfNeeded(query)))
+          val subquery = addSubqueryIfNeeded(query)
+          val plan = if (conditions.isEmpty) {
+            subquery
+          } else {
+            Project(Seq(Alias(Literal(1), "1")()),
+              Filter(conditions.reduce(And), subquery))
+          }
           Exists(plan, exprId)
 
         case PredicateSubquery(query, conditions, true, exprId) =>
diff --git a/sql/hive/src/test/resources/sqlgen/predicate_subquery.sql b/sql/hive/src/test/resources/sqlgen/predicate_subquery.sql
new file mode 100644
index 0000000000..2e06b4fe77
--- /dev/null
+++ b/sql/hive/src/test/resources/sqlgen/predicate_subquery.sql
@@ -0,0 +1,4 @@
+-- This file is automatically generated by LogicalPlanToSQLSuite.
+select * from t1 b where exists (select * from t1 a)
+--------------------------------------------------------------------------------
+SELECT `gen_attr` AS `a` FROM (SELECT `gen_attr` FROM (SELECT `a` AS `gen_attr` FROM `default`.`t1`) AS gen_subquery_0 WHERE EXISTS(SELECT `gen_attr` AS `a` FROM ((SELECT `gen_attr` FROM (SELECT `a` AS `gen_attr` FROM `default`.`t1`) AS gen_subquery_0) AS gen_subquery_1) AS gen_subquery_1)) AS b
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/catalyst/LogicalPlanToSQLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/catalyst/LogicalPlanToSQLSuite.scala
index 1f5078dc9c..ebece38773 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/catalyst/LogicalPlanToSQLSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/catalyst/LogicalPlanToSQLSuite.scala
@@ -25,6 +25,7 @@ import scala.util.control.NonFatal
 import org.apache.spark.sql.Column
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.functions._
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SQLTestUtils
 
 /**
@@ -927,6 +928,15 @@ class LogicalPlanToSQLSuite extends SQLBuilderTest with SQLTestUtils {
     }
   }
 
+  test("predicate subquery") {
+    withTable("t1") {
+      withSQLConf(SQLConf.CROSS_JOINS_ENABLED.key -> "true") {
+        sql("CREATE TABLE t1(a int)")
+        checkSQL("select * from t1 b where exists (select * from t1 a)", "predicate_subquery")
+      }
+    }
+  }
+
   test("SPARK-14933 - select orc table") {
     withTable("orc_t") {
       sql("create table orc_t stored as orc as select 1 as c1, 'abc' as c2")