Copilot commented on code in PR #11288:
URL: https://github.com/apache/gravitino/pull/11288#discussion_r3317880510
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java:
##########
@@ -421,6 +436,49 @@ public boolean dropNamespace(String[] namespace, boolean
cascade)
}
}
+ /**
+ * 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 e) {
+ throw new NoSuchTableException(ident);
+ }
+ Table sparkTable;
+ try {
+ sparkTable = loadSparkTable(ident);
+ } catch (RuntimeException e) {
+ LOG.warn("Failed to load Spark-side table for view {}: {}", ident,
e.getMessage());
+ throw new NoSuchTableException(ident);
+ }
+ return createSparkView(ident, gravitinoView, sparkTable);
+ }
+
Review Comment:
`loadViewAsTable` catches any `RuntimeException` from `loadSparkTable` and
converts it to `NoSuchTableException`, which can incorrectly mask real failures
(e.g., metastore/network/config errors) as "table not found". Consider only
translating the specific "not found" condition (e.g., when the wrapped cause is
Spark's `NoSuchTableException`) and rethrowing other runtime errors so callers
get actionable failures.
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java:
##########
@@ -312,11 +319,19 @@ public boolean tableExists(Identifier ident) {
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) {
+ LOG.debug("Catalog does not support views for {}, treating viewExists as
false", ident);
+ return false;
+ }
Review Comment:
`tableExists` falls through to `viewExists` but doesn't handle
`ForbiddenException`. If a user lacks view-load privileges,
`ViewCatalog#viewExists` can propagate `ForbiddenException` (runtime) and make
Spark metadata checks fail; this should mirror the table path by catching
`ForbiddenException` and returning false (or otherwise mapping to Spark's
expected behavior).
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/SparkHiveView.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.Iterator;
+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.Batch;
+import org.apache.spark.sql.connector.read.InputPartition;
+import org.apache.spark.sql.connector.read.PartitionReader;
+import org.apache.spark.sql.connector.read.PartitionReaderFactory;
+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><b>Note:</b> View results are materialized on the driver via {@code
executeCollect()}. This is
+ * intentional to avoid executor-side SparkSession access, but limits
scalability to views whose
+ * results fit in driver memory.
+ */
+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() {
+ List<StructField> fields =
+ Arrays.stream(gravitinoView.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 and returns
the resulting rows.
+ * This is used for views in V2 catalogs where Kyuubi's file-based scan
cannot expand the view.
+ */
+ private static class ViewScanBuilder implements ScanBuilder, Scan, Batch,
PartitionReaderFactory {
+
+ 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 Batch toBatch() {
+ return this;
+ }
+
+ @Override
+ public InputPartition[] planInputPartitions() {
+ try {
+ InternalRow[] rows =
spark.sql(viewSql).queryExecution().executedPlan().executeCollect();
+ return new InputPartition[] {new ViewPartition(rows)};
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format("Failed to execute view SQL [%s]: %s", viewSql,
e.getMessage()), e);
+ }
Review Comment:
Catching a blanket `Exception` here is overly broad and conflicts with the
repo guideline to avoid generic `catch (Exception e)`. It also risks wrapping
non-recoverable errors and makes troubleshooting harder; prefer catching the
specific Spark exceptions you expect from `executeCollect` (e.g.,
`AnalysisException`/`SparkException`) and let unexpected failures propagate.
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java:
##########
@@ -421,6 +436,49 @@ public boolean dropNamespace(String[] namespace, boolean
cascade)
}
}
+ /**
+ * 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 e) {
+ throw new NoSuchTableException(ident);
+ }
+ Table sparkTable;
+ try {
+ sparkTable = loadSparkTable(ident);
+ } catch (RuntimeException e) {
+ LOG.warn("Failed to load Spark-side table for view {}: {}", ident,
e.getMessage());
+ throw new NoSuchTableException(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: " + ident.namespace()[0]);
Review Comment:
Default `createSparkView` builds the error message with
`ident.namespace()[0]`, but Spark `Identifier` can have an empty namespace (and
this class already supports that via `getDatabase(ident)`). This can throw
`ArrayIndexOutOfBoundsException` and obscure the intended
`UnsupportedOperationException`; use `getDatabase(ident)` or guard against
empty namespaces when constructing the message.
##########
spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java:
##########
@@ -517,4 +526,160 @@ void testCreateTableWithTimestamp() {
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 {
+ // Spark V2 named catalogs do not support CREATE VIEW via SQL; this
documents the limitation.
+ 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().toUpperCase().contains("VIEW"),
+ "Expected error about CREATE VIEW, got: " + ex.getMessage());
Review Comment:
The assertion uses `ex.getMessage().toUpperCase()` without an explicit
locale, which can behave unexpectedly under non-English default locales (e.g.,
Turkish) and make this test flaky. Use `toUpperCase(Locale.ROOT)` (and add the
import) to ensure deterministic behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
