[spark] branch branch-3.1 updated: [SPARK-32680][SQL] Don't Preprocess V2 CTAS with Unresolved Query

[we...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

wenchen pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-3.1 by this push:
     new 88a3c39  [SPARK-32680][SQL] Don't Preprocess V2 CTAS with Unresolved Query
88a3c39 is described below

commit 88a3c39c16f1f3a4d693d8593b7bf949dd3b376a
Author: Linhong Liu <li...@databricks.com>
AuthorDate: Mon Dec 7 13:25:43 2020 +0000

    [SPARK-32680][SQL] Don't Preprocess V2 CTAS with Unresolved Query
    
    ### What changes were proposed in this pull request?
    The analyzer rule `PreprocessTableCreation` will preprocess table creation related logical plan. But for
    CTAS, if the sub-query can't be resolved, preprocess it will cause "Invalid call to toAttribute on unresolved
    object" (instead of a user-friendly error msg: "table or view not found").
    This PR fixes this wrongly preprocess for CTAS using V2 catalog.
    
    ### Why are the changes needed?
    bug fix
    
    ### Does this PR introduce _any_ user-facing change?
    The error message for CTAS with a non-exists table changed from:
    `UnresolvedException: Invalid call to toAttribute on unresolved object, tree: xxx` to
    `AnalysisException: Table or view not found: xxx`
    
    ### How was this patch tested?
    added test
    
    Closes #30637 from linhongliu-db/fix-ctas.
    
    Authored-by: Linhong Liu <li...@databricks.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
    (cherry picked from commit d730b6bdaa92f2ca19cc8852ac58035e28d47a4f)
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../scala/org/apache/spark/sql/execution/datasources/rules.scala | 2 +-
 .../src/test/scala/org/apache/spark/sql/DataFrameSuite.scala     | 9 +++++++++
 2 files changed, 10 insertions(+), 1 deletion(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
index 2cc7825..b9866e4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
@@ -239,7 +239,7 @@ case class PreprocessTableCreation(sparkSession: SparkSession) extends Rule[Logi
         c.copy(tableDesc = normalizedTable.copy(schema = reorderedSchema))
       }
 
-    case create: V2CreateTablePlan =>
+    case create: V2CreateTablePlan if create.childrenResolved =>
       val schema = create.tableSchema
       val partitioning = create.partitioning
       val identifier = create.tableName
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
index d34dcb4..a45bf12 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
@@ -36,6 +36,7 @@ import org.apache.spark.sql.catalyst.expressions.Uuid
 import org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation
 import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, OneRowRelation}
 import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.connector.FakeV2Provider
 import org.apache.spark.sql.execution.{FilterExec, QueryExecution, WholeStageCodegenExec}
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
 import org.apache.spark.sql.execution.aggregate.HashAggregateExec
@@ -2451,6 +2452,14 @@ class DataFrameSuite extends QueryTest
     assert(e.getMessage.contains("Table or view not found:"))
   }
 
+  test("SPARK-32680: Don't analyze CTAS with unresolved query") {
+    val v2Source = classOf[FakeV2Provider].getName
+    val e = intercept[AnalysisException] {
+      sql(s"CREATE TABLE t USING $v2Source AS SELECT * from nonexist")
+    }
+    assert(e.getMessage.contains("Table or view not found:"))
+  }
+
   test("CalendarInterval reflection support") {
     val df = Seq((1, new CalendarInterval(1, 2, 3))).toDF("a", "b")
     checkAnswer(df.selectExpr("b"), Row(new CalendarInterval(1, 2, 3)))


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org