You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by et...@apache.org on 2024/02/26 14:51:50 UTC
(iceberg) branch main updated: Spark 3.4: Include catalog name in view errors (#9810)
This is an automated email from the ASF dual-hosted git repository.
etudenhoefner pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 10ec85ffca Spark 3.4: Include catalog name in view errors (#9810)
10ec85ffca is described below
commit 10ec85ffcab6ac12ecac7fc6df9c3e4139f857b7
Author: Eduard Tudenhoefner <et...@gmail.com>
AuthorDate: Mon Feb 26 15:51:43 2024 +0100
Spark 3.4: Include catalog name in view errors (#9810)
---
.../spark/sql/catalyst/analysis/CheckViews.scala | 17 ++++++------
.../catalyst/analysis/RewriteViewCommands.scala | 17 +++++-------
.../apache/iceberg/spark/extensions/TestViews.java | 30 ++++++++++++++--------
.../catalyst/analysis/RewriteViewCommands.scala | 7 ++---
4 files changed, 38 insertions(+), 33 deletions(-)
diff --git a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckViews.scala b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckViews.scala
index 5e9429ac30..4debc4d343 100644
--- a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckViews.scala
+++ b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckViews.scala
@@ -24,7 +24,6 @@ import org.apache.spark.sql.catalyst.plans.logical.AlterViewAs
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.plans.logical.views.CreateIcebergView
import org.apache.spark.sql.catalyst.plans.logical.views.ResolvedV2View
-import org.apache.spark.sql.connector.catalog.Identifier
import org.apache.spark.sql.connector.catalog.ViewCatalog
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.util.SchemaUtils
@@ -33,9 +32,9 @@ object CheckViews extends (LogicalPlan => Unit) {
override def apply(plan: LogicalPlan): Unit = {
plan foreach {
- case CreateIcebergView(ResolvedIdentifier(_: ViewCatalog, ident), _, query, columnAliases, _,
+ case CreateIcebergView(resolvedIdent@ResolvedIdentifier(_: ViewCatalog, _), _, query, columnAliases, _,
_, _, _, _, _, _) =>
- verifyColumnCount(ident, columnAliases, query)
+ verifyColumnCount(resolvedIdent, columnAliases, query)
SchemaUtils.checkColumnNameDuplication(query.schema.fieldNames, SQLConf.get.resolver)
case AlterViewAs(ResolvedV2View(_, _), _, _) =>
@@ -45,16 +44,18 @@ object CheckViews extends (LogicalPlan => Unit) {
}
}
- private def verifyColumnCount(ident: Identifier, columns: Seq[String], query: LogicalPlan): Unit = {
+ private def verifyColumnCount(ident: ResolvedIdentifier, columns: Seq[String], query: LogicalPlan): Unit = {
if (columns.nonEmpty) {
if (columns.length > query.output.length) {
- throw new AnalysisException(String.format("Cannot create view %s, the reason is not enough data columns:\n" +
+ throw new AnalysisException(String.format("Cannot create view %s.%s, the reason is not enough data columns:\n" +
"View columns: %s\n" +
- "Data columns: %s", ident.toString, columns.mkString(", "), query.output.map(c => c.name).mkString(", ")))
+ "Data columns: %s", ident.catalog.name(), ident.identifier, columns.mkString(", "),
+ query.output.map(c => c.name).mkString(", ")))
} else if (columns.length < query.output.length) {
- throw new AnalysisException(String.format("Cannot create view %s, the reason is too many data columns:\n" +
+ throw new AnalysisException(String.format("Cannot create view %s.%s, the reason is too many data columns:\n" +
"View columns: %s\n" +
- "Data columns: %s", ident.toString, columns.mkString(", "), query.output.map(c => c.name).mkString(", ")))
+ "Data columns: %s", ident.catalog.name(), ident.identifier, columns.mkString(", "),
+ query.output.map(c => c.name).mkString(", ")))
}
}
}
diff --git a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteViewCommands.scala b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteViewCommands.scala
index e483579712..ece851a25b 100644
--- a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteViewCommands.scala
+++ b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteViewCommands.scala
@@ -35,10 +35,7 @@ import org.apache.spark.sql.catalyst.plans.logical.views.ShowIcebergViews
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.UNRESOLVED_FUNCTION
import org.apache.spark.sql.connector.catalog.CatalogManager
-import org.apache.spark.sql.connector.catalog.CatalogPlugin
-import org.apache.spark.sql.connector.catalog.Identifier
import org.apache.spark.sql.connector.catalog.LookupCatalog
-import org.apache.spark.sql.connector.catalog.ViewCatalog
import scala.collection.mutable
/**
@@ -58,7 +55,7 @@ case class RewriteViewCommands(spark: SparkSession) extends Rule[LogicalPlan] wi
case CreateView(ResolvedIdent(resolved), userSpecifiedColumns, comment, properties,
Some(queryText), query, allowExisting, replace) =>
val q = CTESubstitution.apply(query)
- verifyTemporaryObjectsDontExist(resolved.identifier, q)
+ verifyTemporaryObjectsDontExist(resolved, q)
CreateIcebergView(child = resolved,
queryText = queryText,
query = q,
@@ -113,22 +110,22 @@ case class RewriteViewCommands(spark: SparkSession) extends Rule[LogicalPlan] wi
* Permanent views are not allowed to reference temp objects
*/
private def verifyTemporaryObjectsDontExist(
- name: Identifier,
+ identifier: ResolvedIdentifier,
child: LogicalPlan): Unit = {
val tempViews = collectTemporaryViews(child)
if (tempViews.nonEmpty) {
- throw invalidRefToTempObject(name, tempViews.map(v => v.quoted).mkString("[", ", ", "]"), "view")
+ throw invalidRefToTempObject(identifier, tempViews.map(v => v.quoted).mkString("[", ", ", "]"), "view")
}
val tempFunctions = collectTemporaryFunctions(child)
if (tempFunctions.nonEmpty) {
- throw invalidRefToTempObject(name, tempFunctions.mkString("[", ", ", "]"), "function")
+ throw invalidRefToTempObject(identifier, tempFunctions.mkString("[", ", ", "]"), "function")
}
}
- private def invalidRefToTempObject(name: Identifier, tempObjectNames: String, tempObjectType: String) = {
- new AnalysisException(String.format("Cannot create view %s that references temporary %s: %s",
- name, tempObjectType, tempObjectNames))
+ private def invalidRefToTempObject(ident: ResolvedIdentifier, tempObjectNames: String, tempObjectType: String) = {
+ new AnalysisException(String.format("Cannot create view %s.%s that references temporary %s: %s",
+ ident.catalog.name(), ident.identifier, tempObjectType, tempObjectNames))
}
/**
diff --git a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java
index 26bcb03ba1..7314b4bb0f 100644
--- a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java
+++ b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java
@@ -981,7 +981,8 @@ public class TestViews extends SparkExtensionsTestBase {
() -> sql("CREATE VIEW %s AS SELECT id FROM %s", viewReferencingTempView, tempView))
.isInstanceOf(AnalysisException.class)
.hasMessageContaining(
- String.format("Cannot create view %s.%s", NAMESPACE, viewReferencingTempView))
+ String.format(
+ "Cannot create view %s.%s.%s", catalogName, NAMESPACE, viewReferencingTempView))
.hasMessageContaining("that references temporary view:")
.hasMessageContaining(tempView);
}
@@ -1004,7 +1005,8 @@ public class TestViews extends SparkExtensionsTestBase {
viewReferencingTempView, globalTempView))
.isInstanceOf(AnalysisException.class)
.hasMessageContaining(
- String.format("Cannot create view %s.%s", NAMESPACE, viewReferencingTempView))
+ String.format(
+ "Cannot create view %s.%s.%s", catalogName, NAMESPACE, viewReferencingTempView))
.hasMessageContaining("that references temporary view:")
.hasMessageContaining(String.format("%s.%s", "global_temp", globalTempView));
}
@@ -1022,7 +1024,8 @@ public class TestViews extends SparkExtensionsTestBase {
assertThatThrownBy(
() -> sql("CREATE VIEW %s AS SELECT %s(id) FROM %s", viewName, functionName, tableName))
.isInstanceOf(AnalysisException.class)
- .hasMessageContaining(String.format("Cannot create view %s.%s", NAMESPACE, viewName))
+ .hasMessageContaining(
+ String.format("Cannot create view %s.%s.%s", catalogName, NAMESPACE, viewName))
.hasMessageContaining("that references temporary function:")
.hasMessageContaining(functionName);
}
@@ -1073,7 +1076,8 @@ public class TestViews extends SparkExtensionsTestBase {
assertThatThrownBy(
() -> sql("CREATE VIEW %s (id, data) AS SELECT id FROM %s", viewName, tableName))
.isInstanceOf(AnalysisException.class)
- .hasMessageContaining(String.format("Cannot create view %s.%s", NAMESPACE, viewName))
+ .hasMessageContaining(
+ String.format("Cannot create view %s.%s.%s", catalogName, NAMESPACE, viewName))
.hasMessageContaining("not enough data columns")
.hasMessageContaining("View columns: id, data")
.hasMessageContaining("Data columns: id");
@@ -1081,7 +1085,8 @@ public class TestViews extends SparkExtensionsTestBase {
assertThatThrownBy(
() -> sql("CREATE VIEW %s (id) AS SELECT id, data FROM %s", viewName, tableName))
.isInstanceOf(AnalysisException.class)
- .hasMessageContaining(String.format("Cannot create view %s.%s", NAMESPACE, viewName))
+ .hasMessageContaining(
+ String.format("Cannot create view %s.%s.%s", catalogName, NAMESPACE, viewName))
.hasMessageContaining("too many data columns")
.hasMessageContaining("View columns: id")
.hasMessageContaining("Data columns: id, data");
@@ -1200,7 +1205,8 @@ public class TestViews extends SparkExtensionsTestBase {
assertThatThrownBy(() -> sql("CREATE VIEW %s AS %s", viewName, sql))
.isInstanceOf(AnalysisException.class)
- .hasMessageContaining(String.format("Cannot create view %s.%s", NAMESPACE, viewName))
+ .hasMessageContaining(
+ String.format("Cannot create view %s.%s.%s", catalogName, NAMESPACE, viewName))
.hasMessageContaining("that references temporary view:")
.hasMessageContaining(tempViewInCTE);
}
@@ -1221,7 +1227,8 @@ public class TestViews extends SparkExtensionsTestBase {
assertThatThrownBy(() -> sql("CREATE VIEW %s AS %s", viewName, sql))
.isInstanceOf(AnalysisException.class)
- .hasMessageContaining(String.format("Cannot create view %s.%s", NAMESPACE, viewName))
+ .hasMessageContaining(
+ String.format("Cannot create view %s.%s.%s", catalogName, NAMESPACE, viewName))
.hasMessageContaining("that references temporary function:")
.hasMessageContaining(functionName);
}
@@ -1249,7 +1256,8 @@ public class TestViews extends SparkExtensionsTestBase {
assertThatThrownBy(() -> sql("CREATE VIEW %s AS %s", viewName, sql))
.isInstanceOf(AnalysisException.class)
- .hasMessageContaining(String.format("Cannot create view %s.%s", NAMESPACE, viewName))
+ .hasMessageContaining(
+ String.format("Cannot create view %s.%s.%s", catalogName, NAMESPACE, viewName))
.hasMessageContaining("that references temporary view:")
.hasMessageContaining(tempView);
}
@@ -1269,7 +1277,8 @@ public class TestViews extends SparkExtensionsTestBase {
assertThatThrownBy(() -> sql("CREATE VIEW %s AS %s", viewName, sql))
.isInstanceOf(AnalysisException.class)
- .hasMessageContaining(String.format("Cannot create view %s.%s", NAMESPACE, viewName))
+ .hasMessageContaining(
+ String.format("Cannot create view %s.%s.%s", catalogName, NAMESPACE, viewName))
.hasMessageContaining("that references temporary view:")
.hasMessageContaining(String.format("%s.%s", "global_temp", globalTempView));
}
@@ -1289,7 +1298,8 @@ public class TestViews extends SparkExtensionsTestBase {
assertThatThrownBy(() -> sql("CREATE VIEW %s AS %s", viewName, sql))
.isInstanceOf(AnalysisException.class)
- .hasMessageContaining(String.format("Cannot create view %s.%s", NAMESPACE, viewName))
+ .hasMessageContaining(
+ String.format("Cannot create view %s.%s.%s", catalogName, NAMESPACE, viewName))
.hasMessageContaining("that references temporary function:")
.hasMessageContaining(functionName);
}
diff --git a/spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteViewCommands.scala b/spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteViewCommands.scala
index 22c0bb3301..ef0ed5c996 100644
--- a/spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteViewCommands.scala
+++ b/spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteViewCommands.scala
@@ -123,12 +123,9 @@ case class RewriteViewCommands(spark: SparkSession) extends Rule[LogicalPlan] wi
}
}
- private def invalidRefToTempObject(
- identifier: ResolvedIdentifier,
- tempObjectNames: String,
- tempObjectType: String) = {
+ private def invalidRefToTempObject(ident: ResolvedIdentifier, tempObjectNames: String, tempObjectType: String) = {
new AnalysisException(String.format("Cannot create view %s.%s that references temporary %s: %s",
- identifier.catalog.name(), identifier.identifier, tempObjectType, tempObjectNames))
+ ident.catalog.name(), ident.identifier, tempObjectType, tempObjectNames))
}
/**