This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 0528fd0ccc [Cherry-pick to branch-1.3] [#11287] feat(spark-connector):
Add view support for Hive catalog (#11288) (#11410)
0528fd0ccc is described below
commit 0528fd0ccc0a072f3c13ecefe76368cd2ec832dd
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Jun 4 11:26:54 2026 +0800
[Cherry-pick to branch-1.3] [#11287] feat(spark-connector): Add view
support for Hive catalog (#11288) (#11410)
**Cherry-pick Information:**
- Original commit: 64e8faf4ac8dcdc83b29ab9afbe996f35bf9fdca
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
Co-authored-by: Yuhui <[email protected]>
---
.../apache/gravitino/catalog/hive/HiveView.java | 6 +-
.../catalog/hive/HiveViewCatalogOperations.java | 90 +++++++---
.../catalog/hive/TestHiveCatalogOperations.java | 37 ++--
.../spark/connector/catalog/BaseCatalog.java | 81 +++++++--
.../spark/connector/hive/GravitinoHiveCatalog.java | 13 ++
.../spark/connector/hive/SparkHiveView.java | 192 +++++++++++++++++++++
.../connector/integration/test/SparkEnvIT.java | 5 +
.../integration/test/hive/SparkHiveCatalogIT.java | 165 ++++++++++++++++++
8 files changed, 532 insertions(+), 57 deletions(-)
diff --git
a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveView.java
b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveView.java
index bae828e7f6..a2aeee7a0b 100644
---
a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveView.java
+++
b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveView.java
@@ -52,7 +52,8 @@ import org.apache.gravitino.rel.View;
@ToString
public class HiveView implements View {
- private static final String SPARK_VERSION_KEY = "spark.sql.create.version";
+ static final String SPARK_VERSION_KEY = "spark.sql.create.version";
+ static final String FLINK_PROPERTY_PREFIX = "flink.";
private static final String TRINO_VIEW_MARKER_KEY = "presto_view";
private static final String TRINO_VIEW_PREFIX = "/* Presto View:";
@@ -124,7 +125,8 @@ public class HiveView implements View {
if (parameters != null && parameters.containsKey(SPARK_VERSION_KEY)) {
return Dialects.SPARK;
}
- if (parameters != null && parameters.keySet().stream().anyMatch(k ->
k.startsWith("flink."))) {
+ if (parameters != null
+ && parameters.keySet().stream().anyMatch(k ->
k.startsWith(FLINK_PROPERTY_PREFIX))) {
return Dialects.FLINK;
}
return Dialects.HIVE;
diff --git
a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java
b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java
index 1200391bbc..f3d078ff2e 100644
---
a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java
+++
b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java
@@ -27,6 +27,7 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import java.time.Instant;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.function.Predicate;
import java.util.function.Supplier;
@@ -112,12 +113,13 @@ class HiveViewCatalogOperations implements ViewCatalog {
if (!schemaExistsChecker.test(schemaIdent)) {
throw new NoSuchSchemaException("Schema %s does not exist", schemaIdent);
}
+ Map<String, String> safeProperties = properties == null ?
ImmutableMap.of() : properties;
SQLRepresentation sqlRepresentation =
- validateSQLRepresentation(representations, defaultCatalog,
defaultSchema, ident);
+ validateSQLRepresentation(
+ representations, defaultCatalog, defaultSchema, safeProperties,
ident);
try {
- Map<String, String> params =
- Maps.newHashMap(properties == null ? ImmutableMap.of() : properties);
+ Map<String, String> params = Maps.newHashMap(safeProperties);
params.put(TABLE_TYPE, TableType.VIRTUAL_VIEW.name());
String viewOriginalText = toHmsViewOriginalText(sqlRepresentation,
ident);
@@ -212,6 +214,7 @@ class HiveViewCatalogOperations implements ViewCatalog {
replace.getRepresentations(),
replace.getDefaultCatalog(),
replace.getDefaultSchema(),
+ updatedProperties,
ident);
updatedColumns = copyColumns(replace.getColumns());
updatedComment = replace.getComment();
@@ -363,14 +366,17 @@ class HiveViewCatalogOperations implements ViewCatalog {
Maps.newHashMap(properties != null ? properties : ImmutableMap.of());
String representationSql = viewOriginalText;
String detectedDialect = HiveView.detectDialect(representationSql, params);
- if (!Dialects.HIVE.equalsIgnoreCase(detectedDialect)
- && !Dialects.FLINK.equalsIgnoreCase(detectedDialect)) {
- // TODO(design-docs/gravitino-logical-view-management.md): support
loading trino/spark HMS
- // views.
- throw new UnsupportedOperationException(
- String.format(
- "Hive catalog currently supports only '%s' and '%s' view
dialects, but found '%s' for view %s",
- Dialects.HIVE, Dialects.FLINK, detectedDialect, ident));
+ switch (detectedDialect.toLowerCase(Locale.ROOT)) {
+ case Dialects.HIVE:
+ case Dialects.FLINK:
+ case Dialects.SPARK:
+ break;
+ default:
+ // TODO(design-docs/gravitino-logical-view-management.md): support
loading trino HMS views.
+ throw new UnsupportedOperationException(
+ String.format(
+ "Hive catalog currently supports only '%s', '%s' and '%s' view
dialects, but found '%s' for view %s",
+ Dialects.HIVE, Dialects.FLINK, Dialects.SPARK,
detectedDialect, ident));
}
SQLRepresentation rep =
@@ -389,10 +395,18 @@ class HiveViewCatalogOperations implements ViewCatalog {
.build();
}
+ /**
+ * Validates that {@code representations} contains exactly one {@link
SQLRepresentation} with a
+ * supported dialect, and that dialect-specific constraints are satisfied.
+ *
+ * @param properties view properties used to verify dialect marker keys are
present (e.g. {@code
+ * spark.sql.create.version} for Spark, a {@code flink.*} key for Flink)
+ */
private SQLRepresentation validateSQLRepresentation(
Representation[] representations,
String defaultCatalog,
String defaultSchema,
+ Map<String, String> properties,
NameIdentifier ident) {
int representationCount = representations == null ? 0 :
representations.length;
Representation firstRepresentation =
@@ -406,8 +420,17 @@ class HiveViewCatalogOperations implements ViewCatalog {
firstRepresentation == null ? "null" :
firstRepresentation.getClass().getSimpleName());
SQLRepresentation selected = (SQLRepresentation) firstRepresentation;
- switch (selected.dialect().toLowerCase(java.util.Locale.ROOT)) {
+ switch (selected.dialect().toLowerCase(Locale.ROOT)) {
case Dialects.HIVE:
+ Preconditions.checkArgument(
+ defaultCatalog == null && defaultSchema == null,
+ "Dialect '%s' does not support non-null
defaultCatalog/defaultSchema, but got "
+ + "defaultCatalog=%s, defaultSchema=%s for view %s",
+ selected.dialect(),
+ defaultCatalog,
+ defaultSchema,
+ ident);
+ return selected;
case Dialects.FLINK:
Preconditions.checkArgument(
defaultCatalog == null && defaultSchema == null,
@@ -417,28 +440,45 @@ class HiveViewCatalogOperations implements ViewCatalog {
defaultCatalog,
defaultSchema,
ident);
+ Preconditions.checkArgument(
+ properties.keySet().stream()
+ .anyMatch(k -> k.startsWith(HiveView.FLINK_PROPERTY_PREFIX)),
+ "Flink dialect view '%s' requires at least one property with
prefix '%s' to be set; "
+ + "without it the view silently round-trips as Hive dialect on
reload",
+ ident,
+ HiveView.FLINK_PROPERTY_PREFIX);
+ return selected;
+ case Dialects.SPARK:
+ Preconditions.checkArgument(
+ properties.containsKey(HiveView.SPARK_VERSION_KEY),
+ "Spark dialect view '%s' requires property '%s' to be set; "
+ + "without it the view silently round-trips as Hive dialect on
reload",
+ ident,
+ HiveView.SPARK_VERSION_KEY);
return selected;
default:
- // TODO(design-docs/gravitino-logical-view-management.md): support
creating trino/spark HMS
- // views.
+ // TODO(design-docs/gravitino-logical-view-management.md): support
creating trino HMS views.
throw new UnsupportedOperationException(
String.format(
- "Hive catalog currently supports only '%s' and '%s' view
dialects, but got '%s' for view %s",
- Dialects.HIVE, Dialects.FLINK, selected.dialect(), ident));
+ "Hive catalog currently supports only '%s', '%s' and '%s' view
dialects, but got '%s' for view %s",
+ Dialects.HIVE, Dialects.FLINK, Dialects.SPARK,
selected.dialect(), ident));
}
}
private String toHmsViewOriginalText(SQLRepresentation representation,
NameIdentifier ident) {
- if (!Dialects.HIVE.equalsIgnoreCase(representation.dialect())
- && !Dialects.FLINK.equalsIgnoreCase(representation.dialect())) {
- // TODO(design-docs/gravitino-logical-view-management.md): support
serializing trino/spark HMS
- // view definitions.
- throw new UnsupportedOperationException(
- String.format(
- "Hive catalog currently supports only '%s' and '%s' view
dialects, but got '%s' for view %s",
- Dialects.HIVE, Dialects.FLINK, representation.dialect(), ident));
+ switch (representation.dialect().toLowerCase(Locale.ROOT)) {
+ case Dialects.HIVE:
+ case Dialects.FLINK:
+ case Dialects.SPARK:
+ return representation.sql();
+ default:
+ // TODO(design-docs/gravitino-logical-view-management.md): support
serializing trino HMS
+ // view definitions.
+ throw new UnsupportedOperationException(
+ String.format(
+ "Hive catalog currently supports only '%s', '%s' and '%s' view
dialects, but got '%s' for view %s",
+ Dialects.HIVE, Dialects.FLINK, Dialects.SPARK,
representation.dialect(), ident));
}
- return representation.sql();
}
private String extractRenameTargetName(String originalName, ViewChange[]
changes) {
diff --git
a/catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/TestHiveCatalogOperations.java
b/catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/TestHiveCatalogOperations.java
index a8ab3142d9..fbc1e8c54b 100644
---
a/catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/TestHiveCatalogOperations.java
+++
b/catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/TestHiveCatalogOperations.java
@@ -373,7 +373,7 @@ class TestHiveCatalogOperations {
}
@Test
- void testCreateViewRejectsSparkDialect() throws Exception {
+ void testCreateViewAcceptsSparkDialect() throws Exception {
HiveCatalogOperations op = new HiveCatalogOperations();
op.initialize(Maps.newHashMap(), null, HIVE_PROPERTIES_METADATA);
@@ -381,6 +381,7 @@ class TestHiveCatalogOperations {
HiveClient hiveClient = mock(HiveClient.class);
HiveSchema schema =
HiveSchema.builder().withCatalogName("hive").withName("db").build();
when(hiveClient.getDatabase(anyString(), anyString())).thenReturn(schema);
+ doNothing().when(hiveClient).createTable(any());
when(clientPool.run(any()))
.thenAnswer(
invocation -> {
@@ -389,21 +390,25 @@ class TestHiveCatalogOperations {
});
op.clientPool = clientPool;
- UnsupportedOperationException exception =
- Assertions.assertThrows(
- UnsupportedOperationException.class,
- () ->
- op.createView(
- NameIdentifier.of("db", "v_spark"),
- null,
- new Column[0],
- new SQLRepresentation[] {
-
SQLRepresentation.builder().withDialect("spark").withSql("SELECT 1").build()
- },
- null,
- null,
- Maps.newHashMap()));
- Assertions.assertTrue(exception.getMessage().contains("supports only"));
+ // Callers must supply spark.sql.create.version so HMS detectDialect
identifies the view as
+ // Spark
+ View view =
+ op.createView(
+ NameIdentifier.of("db", "v_spark"),
+ null,
+ new Column[0],
+ new SQLRepresentation[] {
+ SQLRepresentation.builder().withDialect("spark").withSql("SELECT
1").build()
+ },
+ null,
+ null,
+ Maps.newHashMap(ImmutableMap.of("spark.sql.create.version",
"3.5.3")));
+
+ Assertions.assertNotNull(view);
+ Assertions.assertEquals(1, view.representations().length);
+ SQLRepresentation rep = (SQLRepresentation) view.representations()[0];
+ Assertions.assertEquals("spark", rep.dialect());
+ Assertions.assertEquals("SELECT 1", rep.sql());
}
@Test
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
index 24a75bd4a9..2f26beae32 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
@@ -34,12 +34,14 @@ import org.apache.gravitino.SchemaChange;
import org.apache.gravitino.authorization.Privilege;
import org.apache.gravitino.exceptions.ForbiddenException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
+import org.apache.gravitino.exceptions.NoSuchViewException;
import org.apache.gravitino.exceptions.NonEmptySchemaException;
import org.apache.gravitino.exceptions.SchemaAlreadyExistsException;
import org.apache.gravitino.function.Function;
import org.apache.gravitino.function.FunctionDefinition;
import org.apache.gravitino.function.FunctionImpl;
import org.apache.gravitino.function.JavaImpl;
+import org.apache.gravitino.rel.View;
import org.apache.gravitino.spark.connector.ConnectorConstants;
import org.apache.gravitino.spark.connector.PropertiesConverter;
import org.apache.gravitino.spark.connector.SparkTableChangeConverter;
@@ -241,21 +243,22 @@ public abstract class BaseCatalog implements
TableCatalog, SupportsNamespaces, F
@Override
public Table loadTable(Identifier ident) throws NoSuchTableException {
+ org.apache.gravitino.rel.Table gravitinoTable;
try {
- org.apache.gravitino.rel.Table gravitinoTable =
loadGravitinoTable(ident);
- org.apache.spark.sql.connector.catalog.Table sparkTable =
loadSparkTable(ident);
- // Will create a catalog specific table
- return createSparkTable(
- ident,
- gravitinoTable,
- sparkTable,
- sparkCatalog,
- propertiesConverter,
- sparkTransformConverter,
- sparkTypeConverter);
- } catch (org.apache.gravitino.exceptions.NoSuchTableException e) {
- throw new NoSuchTableException(ident);
+ gravitinoTable = loadGravitinoTable(ident);
+ } catch (NoSuchTableException e) {
+ // Not a table in Gravitino; try as a view.
+ return loadViewAsTable(ident);
}
+ Table sparkTable = loadSparkTable(ident);
+ return createSparkTable(
+ ident,
+ gravitinoTable,
+ sparkTable,
+ sparkCatalog,
+ propertiesConverter,
+ sparkTransformConverter,
+ sparkTypeConverter);
}
@Override
@@ -312,11 +315,22 @@ public abstract class BaseCatalog implements
TableCatalog, SupportsNamespaces, F
loadGravitinoTable(ident);
return true;
} catch (NoSuchTableException e) {
- return false;
+ // fall through to view check
} catch (ForbiddenException e) {
// User lacks LOAD_TABLE privilege, return false to allow CREATE TABLE
IF NOT EXISTS
return false;
}
+ try {
+ return gravitinoCatalogClient
+ .asViewCatalog()
+ .viewExists(NameIdentifier.of(getDatabase(ident), ident.name()));
+ } catch (UnsupportedOperationException e) {
+ return false;
+ } catch (ForbiddenException e) {
+ // Same rationale as the table check above: lacking LOAD_VIEW privilege
is treated as
+ // non-existence so that CREATE TABLE IF NOT EXISTS can proceed.
+ return false;
+ }
}
@Override
@@ -421,6 +435,45 @@ public abstract class BaseCatalog implements TableCatalog,
SupportsNamespaces, F
}
}
+ /**
+ * Loads a Gravitino view as a Spark Table when {@link
#loadTable(Identifier)} fails because the
+ * object is a view. Subclasses can override {@link #createSparkView} to
enable view support.
+ *
+ * @param ident the identifier to load
+ * @return Spark Table wrapping the view
+ * @throws NoSuchTableException if no view exists or views are not supported
by this catalog
+ */
+ protected Table loadViewAsTable(Identifier ident) throws
NoSuchTableException {
+ View gravitinoView;
+ try {
+ gravitinoView =
+ gravitinoCatalogClient
+ .asViewCatalog()
+ .loadView(NameIdentifier.of(getDatabase(ident), ident.name()));
+ } catch (NoSuchViewException | UnsupportedOperationException |
ForbiddenException e) {
+ throw new NoSuchTableException(ident);
+ }
+ // Call sparkCatalog.loadTable directly to surface NoSuchTableException as
checked,
+ // avoiding the RuntimeException wrapping in loadSparkTable.
+ Table sparkTable = sparkCatalog.loadTable(ident);
+ return createSparkView(ident, gravitinoView, sparkTable);
+ }
+
+ /**
+ * Creates a catalog-specific Spark Table wrapping a Gravitino view.
Subclasses that support views
+ * must override this method.
+ *
+ * @param ident the identifier
+ * @param gravitinoView the Gravitino view metadata
+ * @param sparkTable the underlying Spark table for IO (view expansion)
+ * @return a Spark Table representing the view
+ * @throws UnsupportedOperationException if views are not supported by this
catalog
+ */
+ protected Table createSparkView(Identifier ident, View gravitinoView, Table
sparkTable) {
+ throw new UnsupportedOperationException(
+ "View not supported by catalog: " + getDatabase(ident) + "." +
ident.name());
+ }
+
protected org.apache.gravitino.rel.Table loadGravitinoTable(Identifier ident)
throws NoSuchTableException {
try {
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 b1df7ad4b4..dd7851828c 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
@@ -21,6 +21,7 @@ package org.apache.gravitino.spark.connector.hive;
import java.util.Map;
import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.View;
import org.apache.gravitino.spark.connector.PropertiesConverter;
import org.apache.gravitino.spark.connector.SparkTransformConverter;
import org.apache.gravitino.spark.connector.SparkTypeConverter;
@@ -77,4 +78,16 @@ public class GravitinoHiveCatalog extends BaseCatalog {
protected SparkTypeConverter getSparkTypeConverter() {
return new SparkHiveTypeConverter();
}
+
+ @Override
+ protected org.apache.spark.sql.connector.catalog.Table createSparkView(
+ Identifier ident,
+ View gravitinoView,
+ org.apache.spark.sql.connector.catalog.Table sparkTable) {
+ return new SparkHiveView(
+ gravitinoView,
+ (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
new file mode 100644
index 0000000000..de93070535
--- /dev/null
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/SparkHiveView.java
@@ -0,0 +1,192 @@
+/*
+ * 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 com.google.common.base.Preconditions;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.gravitino.rel.Dialects;
+import org.apache.gravitino.rel.SQLRepresentation;
+import org.apache.gravitino.rel.View;
+import org.apache.gravitino.spark.connector.ConnectorConstants;
+import org.apache.gravitino.spark.connector.SparkTypeConverter;
+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.connector.read.LocalScan;
+import org.apache.spark.sql.connector.read.Scan;
+import org.apache.spark.sql.connector.read.ScanBuilder;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.MetadataBuilder;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Wraps a Hive view stored in HMS with Gravitino view metadata. Overrides
newScanBuilder to execute
+ * the view's SQL (spark dialect preferred, hive as fallback) instead of using
Kyuubi's file-based
+ * scan, which returns empty results for VIRTUAL_VIEW entries.
+ *
+ * <p>Uses {@link LocalScan} to materialize view results on the driver. This
is a known limitation:
+ * the proper fix (expanding the view SQL into the query plan at analysis
time) requires Spark's V2
+ * catalog framework to route view operations to named catalogs, which Spark
does not yet support —
+ * {@code ResolveSessionCatalog} unconditionally throws for any non-session
catalog view. Until
+ * Spark adds that support, driver-side materialization via {@code
executeCollect()} is the only
+ * viable approach. <b>This implementation is suitable only for small/bounded
views.</b>
+ */
+public class SparkHiveView extends HiveTable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(SparkHiveView.class);
+
+ private final View gravitinoView;
+ private final SparkTypeConverter sparkTypeConverter;
+
+ /**
+ * Creates a SparkHiveView that wraps a Gravitino view with Hive HMS backing.
+ *
+ * @param gravitinoView the Gravitino view metadata
+ * @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,
+ HiveTable hiveTable,
+ HiveTableCatalog hiveTableCatalog,
+ SparkTypeConverter sparkTypeConverter) {
+ super(SparkSession.active(), hiveTable.catalogTable(), hiveTableCatalog);
+ this.gravitinoView = gravitinoView;
+ this.sparkTypeConverter = sparkTypeConverter;
+ }
+
+ @Override
+ public String name() {
+ return gravitinoView.name();
+ }
+
+ @Override
+ @SuppressWarnings("deprecation")
+ public StructType schema() {
+ org.apache.gravitino.rel.Column[] columns = gravitinoView.columns();
+ if (columns.length == 0) {
+ // View API allows an empty column array when the output schema is
unknown;
+ // fall back to the HMS-derived schema from the parent HiveTable.
+ return super.schema();
+ }
+ List<StructField> fields =
+ Arrays.stream(columns)
+ .map(
+ column -> {
+ String comment = column.comment();
+ Metadata metadata =
+ comment != null
+ ? new MetadataBuilder()
+ .putString(ConnectorConstants.COMMENT, comment)
+ .build()
+ : Metadata.empty();
+ return StructField.apply(
+ column.name(),
+ sparkTypeConverter.toSparkType(column.dataType()),
+ column.nullable(),
+ metadata);
+ })
+ .collect(Collectors.toList());
+ return DataTypes.createStructType(fields);
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return gravitinoView.properties();
+ }
+
+ @Override
+ public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) {
+ SparkSession spark = SparkSession.active();
+ Preconditions.checkState(
+ spark != null && !spark.sparkContext().isStopped(),
+ "No active SparkSession available to execute view '%s'",
+ gravitinoView.name());
+ String viewSql =
+ gravitinoView
+ .sqlFor(Dialects.SPARK)
+ .map(SQLRepresentation::sql)
+ .orElseGet(
+ () -> {
+ String hiveSql =
+ gravitinoView
+ .sqlFor(Dialects.HIVE)
+ .map(SQLRepresentation::sql)
+ .orElseThrow(
+ () ->
+ new UnsupportedOperationException(
+ "No SQL representation for view: " +
gravitinoView.name()));
+ LOG.warn(
+ "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;
+ });
+ return new ViewScanBuilder(spark, viewSql, schema());
+ }
+
+ /**
+ * A ScanBuilder that executes the view's SQL via SparkSession using {@link
LocalScan}. Spark
+ * treats LocalScan results as driver-local data and never serializes rows
to executors.
+ */
+ private static class ViewScanBuilder implements ScanBuilder, Scan, LocalScan
{
+
+ private final SparkSession spark;
+ private final String viewSql;
+ private final StructType schema;
+
+ ViewScanBuilder(SparkSession spark, String viewSql, StructType schema) {
+ this.spark = spark;
+ this.viewSql = viewSql;
+ this.schema = schema;
+ }
+
+ @Override
+ public Scan build() {
+ return this;
+ }
+
+ @Override
+ public StructType readSchema() {
+ return schema;
+ }
+
+ @Override
+ 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) {
+ throw new RuntimeException(
+ String.format("Failed to execute view SQL [%s]: %s", viewSql,
e.getMessage()), e);
+ }
+ }
+ }
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java
index d9c4e8a247..a9da4aaa8f 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java
@@ -84,6 +84,11 @@ public abstract class SparkEnvIT extends SparkUtilIT {
return true;
}
+ /** Returns the Gravitino {@link Catalog} for the catalog under test. */
+ protected Catalog getGravitinoCatalog() {
+ return client.loadMetalake(metalakeName).loadCatalog(getCatalogName());
+ }
+
@Override
protected SparkSession getSparkSession() {
Assertions.assertNotNull(sparkSession);
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 c0ae42727d..e5393b525c 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
@@ -22,8 +22,15 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.SQLRepresentation;
+import org.apache.gravitino.rel.ViewCatalog;
+import org.apache.gravitino.rel.types.Types;
import org.apache.gravitino.spark.connector.GravitinoSparkConfig;
import org.apache.gravitino.spark.connector.hive.HivePropertiesConstants;
import org.apache.gravitino.spark.connector.integration.test.SparkCommonIT;
@@ -31,6 +38,7 @@ import
org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfo
import
org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfo.SparkColumnInfo;
import
org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfoChecker;
import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.apache.spark.sql.types.DataTypes;
import org.junit.jupiter.api.Assertions;
@@ -528,4 +536,161 @@ public abstract class SparkHiveCatalogIT extends
SparkCommonIT {
SparkColumnInfo.of("ts", DataTypes.TimestampType));
checkTableColumns(tableName, expectedSparkInfo, tableInfo);
}
+
+ @Test
+ void testCreateViewViaSql() {
+ String tableName = "cv_base_table";
+ String viewName = "cv_test_view";
+ String schemaName = getDefaultDatabase();
+ dropTableIfExists(tableName);
+ createSimpleTable(tableName);
+ try {
+ // CREATE VIEW via SQL is not supported for V2 named catalogs. Spark's
+ // ResolveSessionCatalog unconditionally throws for any non-session
catalog
+ // (see the "case CreateView(ResolvedIdentifier(catalog, _), ...)"
branch), regardless of
+ // whether the catalog implements ViewCatalog. Views must be created
through the
+ // Gravitino API (ViewCatalog.createView) and are readable via SELECT.
+ AnalysisException ex =
+ Assertions.assertThrows(
+ AnalysisException.class,
+ () ->
+ sql(
+ String.format(
+ "CREATE VIEW %s.%s.%s AS SELECT * FROM %s.%s.%s",
+ getCatalogName(),
+ schemaName,
+ viewName,
+ getCatalogName(),
+ schemaName,
+ tableName)));
+ Assertions.assertTrue(
+ ex.getMessage().contains("does not support views"),
+ "Expected error about view support, got: " + ex.getMessage());
+ } finally {
+ dropTableIfExists(tableName);
+ }
+ }
+
+ @Test
+ void testSelectHiveView() {
+ String tableName = "view_base_table";
+ String viewName = "test_hive_view";
+ String schemaName = getDefaultDatabase();
+
+ dropTableIfExists(tableName);
+ createSimpleTable(tableName);
+ sql(String.format("INSERT INTO %s VALUES (1, '1', 1),(2, '2', 2),(3, '3',
3)", tableName));
+
+ ViewCatalog viewCatalog = getGravitinoCatalog().asViewCatalog();
+ NameIdentifier viewIdent = NameIdentifier.of(schemaName, viewName);
+ if (viewCatalog.viewExists(viewIdent)) {
+ viewCatalog.dropView(viewIdent);
+ }
+
+ SQLRepresentation sparkRep =
+ SQLRepresentation.builder()
+ .withDialect("spark")
+ .withSql(
+ String.format("SELECT * FROM %s.%s.%s", getCatalogName(),
schemaName, tableName))
+ .build();
+ viewCatalog.createView(
+ viewIdent,
+ "test 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},
+ getCatalogName(),
+ schemaName,
+ ImmutableMap.of("spark.sql.create.version",
getSparkSession().version()));
+
+ try {
+ List<String> data =
+ getQueryData(
+ String.format("SELECT * FROM %s.%s.%s", getCatalogName(),
schemaName, viewName));
+ data.sort(Comparator.naturalOrder());
+ Assertions.assertEquals(3, data.size());
+ Assertions.assertEquals("1,1,1", data.get(0));
+ Assertions.assertEquals("2,2,2", data.get(1));
+ Assertions.assertEquals("3,3,3", data.get(2));
+ } finally {
+ viewCatalog.dropView(viewIdent);
+ dropTableIfExists(tableName);
+ }
+ }
+
+ @Test
+ void testTableExistsForView() {
+ String viewName = "test_view_exists";
+ String schemaName = getDefaultDatabase();
+
+ ViewCatalog viewCatalog = getGravitinoCatalog().asViewCatalog();
+ NameIdentifier viewIdent = NameIdentifier.of(schemaName, viewName);
+ if (viewCatalog.viewExists(viewIdent)) {
+ viewCatalog.dropView(viewIdent);
+ }
+
+ SQLRepresentation sparkRep =
+ SQLRepresentation.builder().withDialect("spark").withSql("SELECT 1 AS
id").build();
+ viewCatalog.createView(
+ viewIdent,
+ null,
+ new Column[] {Column.of("id", Types.IntegerType.get(), null)},
+ new SQLRepresentation[] {sparkRep},
+ null,
+ null,
+ ImmutableMap.of("spark.sql.create.version",
getSparkSession().version()));
+
+ try {
+ Assertions.assertDoesNotThrow(
+ () ->
+ sql(
+ String.format(
+ "DESCRIBE TABLE %s.%s.%s", getCatalogName(), schemaName,
viewName)));
+ } finally {
+ viewCatalog.dropView(viewIdent);
+ }
+ }
+
+ @Test
+ void testShowTablesExcludesViews() {
+ String tableName = "show_tables_base_table";
+ String viewName = "show_tables_view";
+ String schemaName = getDefaultDatabase();
+
+ dropTableIfExists(tableName);
+ createSimpleTable(tableName);
+
+ ViewCatalog viewCatalog = getGravitinoCatalog().asViewCatalog();
+ NameIdentifier viewIdent = NameIdentifier.of(schemaName, viewName);
+ if (viewCatalog.viewExists(viewIdent)) {
+ viewCatalog.dropView(viewIdent);
+ }
+
+ SQLRepresentation sparkRep =
+ SQLRepresentation.builder()
+ .withDialect("spark")
+ .withSql(
+ String.format("SELECT * FROM %s.%s.%s", getCatalogName(),
schemaName, tableName))
+ .build();
+ viewCatalog.createView(
+ viewIdent,
+ null,
+ new Column[0],
+ new SQLRepresentation[] {sparkRep},
+ getCatalogName(),
+ schemaName,
+ ImmutableMap.of("spark.sql.create.version",
getSparkSession().version()));
+
+ try {
+ Set<String> tableNames = listTableNames(schemaName);
+ Assertions.assertTrue(tableNames.contains(tableName));
+ Assertions.assertFalse(tableNames.contains(viewName));
+ } finally {
+ viewCatalog.dropView(viewIdent);
+ dropTableIfExists(tableName);
+ }
+ }
}
