This is an automated email from the ASF dual-hosted git repository.
diqiu50 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 6904c69be5 [#14293] fix(spark-connector): Resolve table names in a
Hive view against the view's namespace (#14297)
6904c69be5 is described below
commit 6904c69be5506a5e9587b2ec9aaa55e9d0e99866
Author: YangJie <[email protected]>
AuthorDate: Mon Nov 2 22:09:27 2026 -0500
[#14293] fix(spark-connector): Resolve table names in a Hive view against
the view's namespace (#14297)
### What changes were proposed in this pull request?
- `SparkHiveView` parses the view SQL and qualifies its table names
before running it, with the new `ViewRelationQualifier`. As in Spark's
own view resolution, a one-part name gets the catalog and namespace, and
a name whose first part is not a registered catalog gets the catalog.
CTE names are substituted first, and names inside subqueries and time
travel clauses (`VERSION AS OF`, `TIMESTAMP AS OF (subquery)`) are
qualified too.
- The namespace is the one Spark recorded in
`view.catalogAndNamespace.*` when it created the view, with
`spark_catalog` mapped to the Gravitino catalog the view was loaded
from. Without a recorded namespace, it is the view's own database.
- When the view has no Spark SQL, the Hive fallback uses the view's
expanded text, in which Hive qualifies the table names.
- The session's current catalog and database are not changed.
### Why are the changes needed?
The view SQL ran in the session's current catalog and database, so a
view read the wrong table, or failed with `TABLE_OR_VIEW_NOT_FOUND`,
whenever the session was not in the view's database.
Fix: #14293
### Does this PR introduce _any_ user-facing change?
Views return the rows of the tables they were defined on, regardless of
the session's current database. Function names, names written with
`IDENTIFIER(...)` and the SQL configs Spark records with a view are
still taken from the session, as before.
### How was this patch tested?
- New `TestViewRelationQualifier` covers one- and two-part names, names
that start with a catalog, subqueries, CTE names, `VERSION AS OF` and a
`TIMESTAMP AS OF` subquery, and nested `WITH` under the legacy CTE
precedence policy.
-
`SparkHiveCatalogIT.testSelectViewWithUnqualifiedSqlResolvesInViewNamespace`
switches the session to `spark_catalog.default` and reads three views
whose SQL has no database name, with a same-named table in another
database: one without a recorded namespace (its SQL has a CTE and a
subquery), one with a recorded Spark namespace, and one created by `hive
-e`.
---
.../spark/connector/hive/GravitinoHiveCatalog.java | 2 +
.../spark/connector/hive/SparkHiveView.java | 66 ++++++++++-
.../connector/hive/ViewRelationQualifier.java | 126 +++++++++++++++++++++
.../connector/hive/TestViewRelationQualifier.java | 126 +++++++++++++++++++++
.../integration/test/hive/SparkHiveCatalogIT.java | 105 +++++++++++++++++
5 files changed, 420 insertions(+), 5 deletions(-)
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/GravitinoHiveCatalog.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/GravitinoHiveCatalog.java
index 4179a79da2..e0275730e3 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/GravitinoHiveCatalog.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/GravitinoHiveCatalog.java
@@ -120,6 +120,8 @@ public class GravitinoHiveCatalog extends BaseCatalog {
org.apache.spark.sql.connector.catalog.Table sparkTable) {
return new SparkHiveView(
gravitinoView,
+ name(),
+ getDatabase(ident),
(HiveTable) sparkTable,
(HiveTableCatalog) sparkCatalog,
getSparkTypeConverter());
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/SparkHiveView.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/SparkHiveView.java
index de93070535..39d0eb5123 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/SparkHiveView.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/SparkHiveView.java
@@ -20,10 +20,12 @@
package org.apache.gravitino.spark.connector.hive;
import com.google.common.base.Preconditions;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.rel.Dialects;
import org.apache.gravitino.rel.SQLRepresentation;
import org.apache.gravitino.rel.View;
@@ -33,9 +35,13 @@ import org.apache.kyuubi.spark.connector.hive.HiveTable;
import org.apache.kyuubi.spark.connector.hive.HiveTableCatalog;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.parser.ParseException;
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
+import org.apache.spark.sql.connector.catalog.CatalogManager;
import org.apache.spark.sql.connector.read.LocalScan;
import org.apache.spark.sql.connector.read.Scan;
import org.apache.spark.sql.connector.read.ScanBuilder;
+import org.apache.spark.sql.execution.CommandExecutionMode;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.MetadataBuilder;
@@ -44,6 +50,7 @@ import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import scala.collection.JavaConverters;
/**
* Wraps a Hive view stored in HMS with Gravitino view metadata. Overrides
newScanBuilder to execute
@@ -63,22 +70,30 @@ public class SparkHiveView extends HiveTable {
private final View gravitinoView;
private final SparkTypeConverter sparkTypeConverter;
+ private final String catalogName;
+ private final String schemaName;
/**
* Creates a SparkHiveView that wraps a Gravitino view with Hive HMS backing.
*
* @param gravitinoView the Gravitino view metadata
+ * @param catalogName the Spark catalog name the view was loaded from
+ * @param schemaName the view's schema (database) name
* @param hiveTable the backing Kyuubi HiveTable from HMS
* @param hiveTableCatalog the Kyuubi HiveTableCatalog for HMS access
* @param sparkTypeConverter converter for Gravitino-to-Spark type mapping
*/
public SparkHiveView(
View gravitinoView,
+ String catalogName,
+ String schemaName,
HiveTable hiveTable,
HiveTableCatalog hiveTableCatalog,
SparkTypeConverter sparkTypeConverter) {
super(SparkSession.active(), hiveTable.catalogTable(), hiveTableCatalog);
this.gravitinoView = gravitinoView;
+ this.catalogName = catalogName;
+ this.schemaName = schemaName;
this.sparkTypeConverter = sparkTypeConverter;
}
@@ -147,9 +162,36 @@ public class SparkHiveView extends HiveTable {
"View '{}' has no Spark SQL representation; falling back
to Hive SQL. "
+ "Results may differ if the SQL uses Hive-specific
syntax.",
gravitinoView.name());
- return hiveSql;
+ // For a view created by Hive, the expanded text has its
table names qualified.
+ String expandedText = catalogTable().viewText().getOrElse(()
-> "");
+ return StringUtils.isNotBlank(expandedText) ? expandedText :
hiveSql;
});
- return new ViewScanBuilder(spark, viewSql, schema());
+ return new ViewScanBuilder(spark, viewSql, schema(),
relationQualifier(spark));
+ }
+
+ /**
+ * Builds the qualifier for the namespace the view SQL resolves in: the
catalog and namespace
+ * Spark recorded when it created the view, or else the view's own catalog
and schema. A recorded
+ * {@code spark_catalog} is mapped to the catalog the view was loaded from.
This assumes that the
+ * session catalog of the Spark session that created the view used the same
Hive metastore as this
+ * catalog.
+ */
+ @SuppressWarnings("deprecation") // JavaConverters is deprecated on Scala
2.13 but needed on 2.12
+ private ViewRelationQualifier relationQualifier(SparkSession spark) {
+ CatalogManager catalogManager = spark.sessionState().catalogManager();
+ List<String> recorded =
JavaConverters.seqAsJavaList(catalogTable().viewCatalogAndNamespace());
+ String viewCatalog = catalogName;
+ List<String> viewNamespace = new ArrayList<>();
+ if (recorded.isEmpty()) {
+ viewNamespace.add(schemaName);
+ } else {
+ if
(!CatalogManager.SESSION_CATALOG_NAME().equalsIgnoreCase(recorded.get(0))) {
+ viewCatalog = recorded.get(0);
+ }
+ viewNamespace.addAll(recorded.subList(1, recorded.size()));
+ }
+ return new ViewRelationQualifier(
+ viewCatalog, viewNamespace, catalogManager::isCatalogRegistered);
}
/**
@@ -161,11 +203,17 @@ public class SparkHiveView extends HiveTable {
private final SparkSession spark;
private final String viewSql;
private final StructType schema;
+ private final ViewRelationQualifier relationQualifier;
- ViewScanBuilder(SparkSession spark, String viewSql, StructType schema) {
+ ViewScanBuilder(
+ SparkSession spark,
+ String viewSql,
+ StructType schema,
+ ViewRelationQualifier relationQualifier) {
this.spark = spark;
this.viewSql = viewSql;
this.schema = schema;
+ this.relationQualifier = relationQualifier;
}
@Override
@@ -182,8 +230,16 @@ public class SparkHiveView extends HiveTable {
public InternalRow[] rows() {
// executeCollect() pulls all rows to driver memory — see class-level
Javadoc for why.
try {
- return
spark.sql(viewSql).queryExecution().executedPlan().executeCollect();
- } catch (RuntimeException e) {
+ // Qualify the table names instead of switching the session's current
namespace, so the
+ // session and other queries running on it are not affected.
+ LogicalPlan plan =
+
relationQualifier.qualify(spark.sessionState().sqlParser().parsePlan(viewSql));
+ return spark
+ .sessionState()
+ .executePlan(plan, CommandExecutionMode.ALL())
+ .executedPlan()
+ .executeCollect();
+ } catch (ParseException | RuntimeException e) {
throw new RuntimeException(
String.format("Failed to execute view SQL [%s]: %s", viewSql,
e.getMessage()), e);
}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/ViewRelationQualifier.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/ViewRelationQualifier.java
new file mode 100644
index 0000000000..c10aa40823
--- /dev/null
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/ViewRelationQualifier.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.spark.connector.hive;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Predicate;
+import org.apache.spark.sql.catalyst.analysis.CTESubstitution;
+import org.apache.spark.sql.catalyst.analysis.RelationTimeTravel;
+import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation;
+import org.apache.spark.sql.catalyst.expressions.Expression;
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression;
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
+import scala.collection.JavaConverters;
+
+/**
+ * Qualifies the table names in a parsed view query with the namespace the
view resolves in, the way
+ * Spark expands table names inside a view: a one-part name gets the catalog
and the namespace, and
+ * a name whose first part is not a registered catalog gets the catalog.
Function names and names
+ * built with {@code IDENTIFIER(...)} are not qualified.
+ */
+@SuppressWarnings("deprecation") // JavaConverters is deprecated on Scala 2.13
but required on 2.12
+final class ViewRelationQualifier {
+
+ // The analyzer applies CTESubstitution in a batch of at most 100 iterations
by default.
+ private static final int MAX_CTE_SUBSTITUTIONS = 100;
+
+ private final String catalog;
+ private final List<String> namespace;
+ private final Predicate<String> isCatalog;
+
+ /**
+ * @param catalog the catalog that unqualified names resolve in
+ * @param namespace the namespace that one-part names resolve in
+ * @param isCatalog tests whether a name is a catalog registered in the
session
+ */
+ ViewRelationQualifier(String catalog, List<String> namespace,
Predicate<String> isCatalog) {
+ this.catalog = catalog;
+ this.namespace = namespace;
+ this.isCatalog = isCatalog;
+ }
+
+ /**
+ * Returns the plan with its table names qualified. CTE references are
substituted first, so a CTE
+ * name is not taken for a table.
+ *
+ * @param plan the parsed, unresolved view query
+ * @return the plan with qualified table names
+ */
+ LogicalPlan qualify(LogicalPlan plan) {
+ return qualifyPlan(substituteCtes(plan));
+ }
+
+ // The legacy CTE precedence policy substitutes one level of nested WITH per
run, so run it until
+ // nothing changes, as the analyzer does.
+ private static LogicalPlan substituteCtes(LogicalPlan plan) {
+ LogicalPlan current = plan;
+ for (int i = 0; i < MAX_CTE_SUBSTITUTIONS; i++) {
+ LogicalPlan next = CTESubstitution.apply(current);
+ if (next.fastEquals(current)) {
+ break;
+ }
+ current = next;
+ }
+ return current;
+ }
+
+ private LogicalPlan qualifyPlan(LogicalPlan plan) {
+ if (plan instanceof UnresolvedRelation) {
+ return qualifyRelation((UnresolvedRelation) plan);
+ }
+ if (plan instanceof RelationTimeTravel) {
+ // The relation is a field, not a child, of this leaf node.
+ RelationTimeTravel timeTravel = (RelationTimeTravel) plan;
+ return new RelationTimeTravel(
+ qualifyPlan(timeTravel.relation()), timeTravel.timestamp(),
timeTravel.version())
+ .mapExpressions(this::qualifyExpression);
+ }
+ return
plan.mapChildren(this::qualifyPlan).mapExpressions(this::qualifyExpression);
+ }
+
+ private Expression qualifyExpression(Expression expression) {
+ Expression qualified = expression;
+ if (expression instanceof SubqueryExpression) {
+ // Spark analyzes a subquery on its own, which substitutes the CTEs it
defines.
+ SubqueryExpression subquery = (SubqueryExpression) expression;
+ qualified = subquery.withNewPlan(qualify(subquery.plan()));
+ }
+ return qualified.mapChildren(this::qualifyExpression);
+ }
+
+ private LogicalPlan qualifyRelation(UnresolvedRelation relation) {
+ List<String> parts =
JavaConverters.seqAsJavaList(relation.multipartIdentifier());
+ List<String> qualified = new ArrayList<>();
+ if (parts.size() == 1) {
+ qualified.add(catalog);
+ qualified.addAll(namespace);
+ } else if (isCatalog.test(parts.get(0))) {
+ return relation;
+ } else {
+ qualified.add(catalog);
+ }
+ qualified.addAll(parts);
+ return new UnresolvedRelation(
+ JavaConverters.asScalaBuffer(qualified).toList(),
+ relation.options(),
+ relation.isStreaming());
+ }
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/hive/TestViewRelationQualifier.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/hive/TestViewRelationQualifier.java
new file mode 100644
index 0000000000..9c6a5c8c8d
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/hive/TestViewRelationQualifier.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.spark.connector.hive;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.spark.sql.catalyst.analysis.RelationTimeTravel;
+import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation;
+import org.apache.spark.sql.catalyst.expressions.Expression;
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression;
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser;
+import org.apache.spark.sql.catalyst.parser.ParseException;
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
+import org.apache.spark.sql.internal.SQLConf;
+import org.junit.jupiter.api.Test;
+import scala.collection.JavaConverters;
+
+@SuppressWarnings("deprecation") // JavaConverters is deprecated on Scala 2.13
but required on 2.12
+public class TestViewRelationQualifier {
+
+ private static final ViewRelationQualifier QUALIFIER =
+ new ViewRelationQualifier("hive", Arrays.asList("db1"), name ->
name.equals("other"));
+
+ @Test
+ void testQualifiesOneAndTwoPartNames() throws ParseException {
+ assertEquals(
+ Arrays.asList("hive.db1.t", "hive.db2.u"),
+ relations("SELECT * FROM t JOIN db2.u ON t.id = u.id"));
+ }
+
+ @Test
+ void testKeepsNamesThatStartWithACatalog() throws ParseException {
+ assertEquals(Arrays.asList("other.db.t"), relations("SELECT * FROM
other.db.t"));
+ }
+
+ @Test
+ void testQualifiesNamesInSubqueries() throws ParseException {
+ assertEquals(
+ Arrays.asList("hive.db1.a", "hive.db1.b"),
+ relations("SELECT * FROM a WHERE id IN (SELECT id FROM b)"));
+ }
+
+ @Test
+ void testDoesNotQualifyCteReferences() throws ParseException {
+ // The CTE named t shadows the table t; only the table inside the CTE body
is qualified.
+ assertEquals(
+ Arrays.asList("hive.db1.base"),
+ relations("WITH t AS (SELECT * FROM base) SELECT * FROM t"));
+ }
+
+ @Test
+ void testQualifiesTimeTravelRelations() throws ParseException {
+ assertEquals(Arrays.asList("hive.db1.t"), relations("SELECT * FROM t
VERSION AS OF 1"));
+ assertEquals(
+ Arrays.asList("hive.db1.t", "hive.db1.u"),
+ relations("SELECT * FROM t TIMESTAMP AS OF (SELECT max(ts) FROM u)"));
+ }
+
+ @Test
+ void testQualifiesNestedCtesUnderLegacyPolicy() throws ParseException {
+ SQLConf conf = SQLConf.get();
+ String key = SQLConf.LEGACY_CTE_PRECEDENCE_POLICY().key();
+ String previous = conf.getConfString(key, null);
+ conf.setConfString(key, "LEGACY");
+ try {
+ assertEquals(
+ Arrays.asList("hive.db1.t"),
+ relations("WITH a AS (WITH b AS (SELECT * FROM t) SELECT * FROM b)
SELECT * FROM a"));
+ assertEquals(
+ Arrays.asList("hive.db1.a", "hive.db1.b"),
+ relations("SELECT * FROM a WHERE id IN (WITH c AS (SELECT id FROM b)
SELECT id FROM c)"));
+ } finally {
+ if (previous == null) {
+ conf.unsetConf(key);
+ } else {
+ conf.setConfString(key, previous);
+ }
+ }
+ }
+
+ private static List<String> relations(String sql) throws ParseException {
+ List<String> names = new ArrayList<>();
+ collect(QUALIFIER.qualify(new CatalystSqlParser().parsePlan(sql)), names);
+ return names;
+ }
+
+ private static void collect(LogicalPlan plan, List<String> names) {
+ if (plan instanceof RelationTimeTravel) {
+ collect(((RelationTimeTravel) plan).relation(), names);
+ }
+ if (plan instanceof UnresolvedRelation) {
+ List<String> parts =
+ JavaConverters.seqAsJavaList(((UnresolvedRelation)
plan).multipartIdentifier());
+ names.add(String.join(".", parts));
+ }
+ JavaConverters.seqAsJavaList(plan.children()).forEach(child ->
collect(child, names));
+ JavaConverters.seqAsJavaList(plan.expressions()).forEach(e -> collect(e,
names));
+ }
+
+ private static void collect(Expression expression, List<String> names) {
+ if (expression instanceof SubqueryExpression) {
+ collect(((SubqueryExpression) expression).plan(), names);
+ }
+ JavaConverters.seqAsJavaList(expression.children()).forEach(child ->
collect(child, names));
+ }
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java
index bca934c64e..83cdcfd6fc 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java
@@ -24,12 +24,14 @@ import com.google.common.collect.Maps;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.integration.test.container.ContainerSuite;
import org.apache.gravitino.rel.Column;
import org.apache.gravitino.rel.SQLRepresentation;
import org.apache.gravitino.rel.ViewCatalog;
@@ -59,6 +61,7 @@ import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import org.testcontainers.containers.Container;
import scala.collection.JavaConverters;
@Tag("gravitino-docker-test")
@@ -634,6 +637,108 @@ public abstract class SparkHiveCatalogIT extends
SparkCommonIT {
}
}
+ @Test
+ void testSelectViewWithUnqualifiedSqlResolvesInViewNamespace() {
+ String tableName = "view_base_table_unq";
+ String schemaName = getDefaultDatabase();
+ String otherDb = "db_view_unq_other";
+ String ownView = "test_hive_view_unq_own";
+ String recordedView = "test_hive_view_unq_recorded";
+
+ // The real table is in the default database; a same-named table in
otherDb holds other rows.
+ dropTableIfExists(tableName);
+ createSimpleTable(tableName);
+ sql(String.format("INSERT INTO %s VALUES (1, '1', 1)", tableName));
+ createDatabaseIfNotExists(otherDb, getProvider());
+ String shadowTable = String.join(".", otherDb, tableName);
+ dropTableIfExists(shadowTable);
+ createSimpleTable(shadowTable);
+ sql(String.format("INSERT INTO %s VALUES (9, 'shadow', 9)", shadowTable));
+
+ ViewCatalog viewCatalog = getGravitinoCatalog().asViewCatalog();
+ NameIdentifier ownIdent = NameIdentifier.of(schemaName, ownView);
+ NameIdentifier recordedIdent = NameIdentifier.of(otherDb, recordedView);
+ // No recorded namespace: the SQL resolves in the view's own database. The
CTE and the
+ // subquery check that their table names are qualified too.
+ createUnqualifiedView(
+ viewCatalog,
+ ownIdent,
+ String.format(
+ "WITH x AS (SELECT * FROM %1$s) SELECT * FROM x WHERE id IN
(SELECT id FROM %1$s)",
+ tableName),
+ ImmutableMap.of());
+ // Like a Spark view created by "USE default_db; CREATE VIEW otherDb.v AS
SELECT * FROM t".
+ createUnqualifiedView(
+ viewCatalog,
+ recordedIdent,
+ String.format("SELECT * FROM %s", tableName),
+ ImmutableMap.of(
+ "view.catalogAndNamespace.numParts", "2",
+ "view.catalogAndNamespace.part.0", "spark_catalog",
+ "view.catalogAndNamespace.part.1", schemaName));
+ // A view created by Hive itself, whose expanded text qualifies the table
with default_db.
+ String hiveView = "test_hive_view_unq_hive";
+ NameIdentifier hiveIdent = NameIdentifier.of(otherDb, hiveView);
+ Container.ExecResult created =
+ ContainerSuite.getInstance()
+ .getHiveContainer()
+ .executeInContainer(
+ "hive",
+ "-e",
+ String.format(
+ "USE %s; CREATE VIEW %s.%s AS SELECT * FROM %s",
+ schemaName, otherDb, hiveView, tableName));
+ Assertions.assertEquals(0, created.getExitCode(), created.getStderr());
+
+ try {
+ // The session is in another catalog, so unqualified names in the
session resolve elsewhere.
+ sql("USE spark_catalog.default");
+ for (NameIdentifier view : new NameIdentifier[] {ownIdent,
recordedIdent, hiveIdent}) {
+ List<String> data =
+ getQueryData(
+ String.format(
+ "SELECT * FROM %s.%s.%s",
+ getCatalogName(), view.namespace().level(0), view.name()));
+ Assertions.assertEquals(Arrays.asList("1,1,1"), data, view.toString());
+ }
+ Assertions.assertEquals("spark_catalog",
getSparkSession().catalog().currentCatalog());
+ } finally {
+ sql("USE " + String.join(".", getCatalogName(), getDefaultDatabase()));
+ viewCatalog.dropView(ownIdent);
+ viewCatalog.dropView(recordedIdent);
+ viewCatalog.dropView(hiveIdent);
+ dropTableIfExists(shadowTable);
+ dropDatabaseIfExists(otherDb);
+ dropTableIfExists(tableName);
+ }
+ }
+
+ private void createUnqualifiedView(
+ ViewCatalog viewCatalog,
+ NameIdentifier viewIdent,
+ String viewSql,
+ Map<String, String> extraProperties) {
+ if (viewCatalog.viewExists(viewIdent)) {
+ viewCatalog.dropView(viewIdent);
+ }
+ SQLRepresentation sparkRep =
+
SQLRepresentation.builder().withDialect("spark").withSql(viewSql).build();
+ Map<String, String> properties = new HashMap<>(extraProperties);
+ properties.put("spark.sql.create.version", getSparkSession().version());
+ viewCatalog.createView(
+ viewIdent,
+ "unqualified view",
+ new Column[] {
+ Column.of("id", Types.IntegerType.get(), null),
+ Column.of("name", Types.StringType.get(), null),
+ Column.of("age", Types.IntegerType.get(), null)
+ },
+ new SQLRepresentation[] {sparkRep},
+ null,
+ null,
+ properties);
+ }
+
@Test
void testTableExistsForView() {
String viewName = "test_view_exists";
