Copilot commented on code in PR #11288: URL: https://github.com/apache/gravitino/pull/11288#discussion_r3329875257
########## 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 (RuntimeException e) { Review Comment: `planInputPartitions()` executes the view and embeds the collected `InternalRow[]` inside an `InputPartition`. In Spark, `InputPartition` instances are serialized and shipped to executors; `InternalRow` implementations are not guaranteed to be safely serializable across Spark versions/serializers, and this also re-sends data from driver -> executors after already collecting it from the cluster. This can fail in distributed mode and/or be prohibitively expensive for non-trivial views. Consider using a driver-only scan API (e.g., Spark's local scan interfaces if available for your supported Spark versions) or serializing rows into a stable transport form (bytes) and reconstructing on the executor side. ########## 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( Review Comment: `schema()` builds the Spark schema exclusively from `gravitinoView.columns()`. The View API allows an empty column array when the output schema is unknown; in that case this returns an empty `StructType`, which will cause Spark to treat the view as having 0 columns (and can mismatch the rows produced by `viewSql`). Consider falling back to the HMS-derived schema from the parent `HiveTable` when the Gravitino view schema is empty/unknown. ########## catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java: ########## @@ -119,6 +119,9 @@ public View createView( Map<String, String> params = Maps.newHashMap(properties == null ? ImmutableMap.of() : properties); params.put(TABLE_TYPE, TableType.VIRTUAL_VIEW.name()); + if (Dialects.SPARK.equalsIgnoreCase(sqlRepresentation.dialect())) { + params.putIfAbsent(HiveView.SPARK_VERSION_KEY, "gravitino"); + } Review Comment: For Spark-dialect views, `defaultCatalog`/`defaultSchema` are accepted by `createView(...)` but are currently not persisted anywhere. This is silent data loss: a subsequent `loadView()` cannot return the defaults, and consumers that rely on them to resolve unqualified identifiers will behave incorrectly. Consider persisting them as HMS table parameters (with Gravitino-scoped keys) so they round-trip through HMS. ########## catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java: ########## @@ -363,14 +366,12 @@ private HiveView toHiveView( 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. + if (Dialects.TRINO.equalsIgnoreCase(detectedDialect)) { Review Comment: `toHiveView(...)` currently ignores any persisted `defaultCatalog`/`defaultSchema` values (and `HiveView` instances always return null for these fields). If you persist defaults in HMS parameters (e.g., via Gravitino-scoped keys), load-time should restore them onto the `HiveView` and remove the internal keys from `properties()` to avoid leaking implementation details. -- 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]
