HIVE-29578: Iceberg: add support for logical views

common/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java

@@ -445,6 +445,7 @@ public enum ErrorMsg {
   @Deprecated // kept for backwards reference
   REPLACE_VIEW_WITH_MATERIALIZED(10400, "Attempt to replace view {0} with materialized view", true),
   REPLACE_MATERIALIZED_WITH_VIEW(10401, "Attempt to replace materialized view {0} with view", true),
+  VIEW_STORAGE_HANDLER_UNSUPPORTED(10448, "Storage handler {0} doesn't support external logical views", true),
   UPDATE_DELETE_VIEW(10402, "You cannot update or delete records in a view"),
   MATERIALIZED_VIEW_DEF_EMPTY(10403, "Query for the materialized view rebuild could not be retrieved"),
   MERGE_PREDIACTE_REQUIRED(10404, "MERGE statement with both UPDATE and DELETE clauses " +

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HMSTablePropertyHelper.java

@@ -161,6 +161,7 @@ public static void updateHmsTableForIcebergView(
         HiveOperationsBase.ICEBERG_VIEW_TYPE_VALUE.toUpperCase(Locale.ENGLISH),
         metadata.schema(),
         maxHiveTablePropertySize);
+    parameters.put(hive_metastoreConstants.META_TABLE_STORAGE, HIVE_ICEBERG_STORAGE_HANDLER);
     tbl.setParameters(parameters);
   }
 

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveOperationsBase.java

@@ -44,7 +44,7 @@
 import org.slf4j.LoggerFactory;
 
 /** All the HMS operations like table,view,materialized_view should implement this. */
-interface HiveOperationsBase {
+public interface HiveOperationsBase {
 
   Logger LOG = LoggerFactory.getLogger(HiveOperationsBase.class);
   // The max size is based on HMS backend database. For Hive versions below 2.3, the max table

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveViewOperations.java

@@ -302,7 +302,7 @@ private Table newHMSView(ViewMetadata metadata) {
         tableType().name());
   }
 
-  private String sqlFor(ViewMetadata metadata) {
+  public static String sqlFor(ViewMetadata metadata) {
     SQLViewRepresentation closest = null;
     for (ViewRepresentation representation : metadata.currentVersion().representations()) {
       if (representation instanceof SQLViewRepresentation) {

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/IcebergViewSupport.java

@@ -0,0 +1,143 @@
+/*
+ * 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.iceberg.hive;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.catalog.ViewCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.view.ViewBuilder;
+
+/**
+ * Commits a native Iceberg view through the configured default Iceberg catalog (HiveCatalog or REST
+ * catalog, etc.) when {@code Catalog} also implements {@link ViewCatalog}.
+ */
+public final class IcebergViewSupport {
+
+  private IcebergViewSupport() {
+  }
+
+  /**
+   * Loads the native Iceberg logical view definition and applies SQL, schema, and Iceberg params to {@code hmsTable}
+   */
+  public static void enrichHmsTableFromIcebergView(
+      org.apache.hadoop.hive.metastore.api.Table hmsTable, Configuration conf) {
+    TableIdentifier identifier = TableIdentifier.of(hmsTable.getDbName(), hmsTable.getTableName());
+    String catalogName = IcebergCatalogProperties.getCatalogName(conf);
+    Map<String, String> catalogProps = IcebergCatalogProperties.getCatalogProperties(conf, catalogName);
+    Catalog catalog = CatalogUtil.buildIcebergCatalog(catalogName, catalogProps, conf);
+    runWithCatalog(catalog, () -> loadAndApplyView(hmsTable, conf, catalog, catalogName, identifier));
+  }
+
+  private static void loadAndApplyView(
+      org.apache.hadoop.hive.metastore.api.Table hmsTable,
+      Configuration conf,
+      Catalog catalog,
+      String catalogName,
+      TableIdentifier identifier) {
+    ViewCatalog viewCatalog = asViewCatalog(catalog, catalogName);
+    MetastoreUtil.applyIcebergViewToHmsTable(hmsTable, viewCatalog.loadView(identifier), conf);
+  }
+
+  public static void createOrReplaceView(
+      Configuration conf,
+      String databaseName,
+      String viewName,
+      List<FieldSchema> fieldSchemas,
+      String viewSql,
+      Map<String, String> tblProperties,
+      String comment) {
+
+    TableIdentifier identifier = TableIdentifier.of(databaseName, viewName);
+    String catalogName = IcebergCatalogProperties.getCatalogName(conf);
+    Map<String, String> catalogProps = IcebergCatalogProperties.getCatalogProperties(conf, catalogName);
+    Catalog catalog = CatalogUtil.buildIcebergCatalog(catalogName, catalogProps, conf);
+    runWithCatalog(
+        catalog,
+        () -> commitView(catalog, catalogName, identifier, fieldSchemas, viewSql, tblProperties, comment));
+  }
+
+  /**
+   * Runs {@code action} with {@code catalog}, then closes it when the catalog implements
+   * {@link Closeable} (e.g. REST catalog clients).
+   */
+  private static void runWithCatalog(Catalog catalog, Runnable action) {
+    if (catalog instanceof Closeable closeable) {
+      try (Closeable ignored = closeable) {
+        action.run();
+      } catch (IOException e) {
+        throw new UncheckedIOException("Failed to close Iceberg catalog", e);
+      }
+    } else {
+      action.run();
+    }
+  }
+
+  private static void commitView(
+      Catalog catalog,
+      String catalogName,
+      TableIdentifier identifier,
+      List<FieldSchema> fieldSchemas,
+      String viewSql,
+      Map<String, String> tblProperties,
+      String comment) {
+    ViewCatalog viewCatalog = asViewCatalog(catalog, catalogName);
+
+    ViewBuilder builder =
+        viewCatalog
+            .buildView(identifier)
+            .withSchema(HiveSchemaUtil.convert(fieldSchemas, Collections.emptyMap(), true))
+            .withDefaultNamespace(Namespace.of(identifier.namespace().level(0)))
+            .withQuery("hive", viewSql);
+
+    if (StringUtils.isNotBlank(comment)) {
+      builder = builder.withProperty("comment", comment);
+    }
+
+    Map<String, String> tblProps =
+        tblProperties == null ? Maps.newHashMap() : Maps.newHashMap(tblProperties);
+
+    builder.withProperties(tblProps);
+
+    builder.createOrReplace();
+  }
+
+  private static ViewCatalog asViewCatalog(Catalog catalog, String catalogName) {
+    if (catalog instanceof ViewCatalog viewCatalog) {
+      return viewCatalog;
+    }
+    throw new UnsupportedOperationException(
+        String.format(
+                "Iceberg catalog '%s' does not implement ViewCatalog.",
+                catalogName) +
+            " Iceberg views require a catalog that implements ViewCatalog (e.g. HiveCatalog or REST).");
+  }
+}

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/MetastoreUtil.java

@@ -36,8 +36,10 @@
 import org.apache.hadoop.hive.metastore.api.SkewedInfo;
 import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
 import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
 import org.apache.hadoop.hive.serde.serdeConstants;
+import org.apache.iceberg.BaseMetastoreTableOperations;
 import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.Schema;
@@ -46,6 +48,11 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.view.BaseView;
+import org.apache.iceberg.view.SQLViewRepresentation;
+import org.apache.iceberg.view.View;
+import org.apache.iceberg.view.ViewMetadata;
 import org.apache.thrift.TException;
 
 public class MetastoreUtil {
@@ -134,20 +141,101 @@ public static Table toHiveTable(org.apache.iceberg.Table table, Configuration co
     result.setDbName(tableName.getDb());
     result.setTableName(tableName.getTable());
     result.setTableType(TableType.EXTERNAL_TABLE.toString());
-    result.setPartitionKeys(getPartitionKeys(table, table.spec().specId()));
+
+    // TODO: Revert after HIVE-29633 is fixed
+    // result.setPartitionKeys(getPartitionKeys(table, table.spec().specId()));
+    result.setPartitionKeys(Lists.newArrayList());
+
     TableMetadata metadata = ((BaseTable) table).operations().current();
     long maxHiveTablePropertySize = conf.getLong(HiveOperationsBase.HIVE_TABLE_PROPERTY_MAX_SIZE,
         HiveOperationsBase.HIVE_TABLE_PROPERTY_MAX_SIZE_DEFAULT);
     HMSTablePropertyHelper.updateHmsTableForIcebergTable(metadata.metadataFileLocation(), result, metadata,
         null, true, maxHiveTablePropertySize, null);
     String catalogType = IcebergCatalogProperties.getCatalogType(conf);
     if (!StringUtils.isEmpty(catalogType) && !IcebergCatalogProperties.NO_CATALOG_TYPE.equals(catalogType)) {
-      result.getParameters().put(CatalogUtil.ICEBERG_CATALOG_TYPE, IcebergCatalogProperties.getCatalogType(conf));
+      result.getParameters().put(CatalogUtil.ICEBERG_CATALOG_TYPE, catalogType);
     }
     result.setSd(getHiveStorageDescriptor(table));
     return result;
   }
 
+  /**
+   * Builds a minimal HMS {@link Table} shell for Iceberg view (identity, view type,
+   * and Iceberg storage-handler markers only). The storage handler {@code postGetTable} hook enriches
+   * this object via {@link IcebergViewSupport#enrichHmsTableFromIcebergView} (view SQL,
+   * schema, and Iceberg parameters).
+   */
+  public static Table buildMinimalHMSView(String catName, String dbName, String tableName) {
+    Table result = new Table();
+    result.setCatName(catName);
+    result.setDbName(dbName);
+    result.setTableName(tableName);
+    result.setTableType(TableType.VIRTUAL_VIEW.toString());
+
+    Map<String, String> parameters = Maps.newHashMap();
+    parameters.put(
+        BaseMetastoreTableOperations.TABLE_TYPE_PROP, HiveOperationsBase.ICEBERG_VIEW_TYPE_VALUE);
+    parameters.put(
+        hive_metastoreConstants.META_TABLE_STORAGE, HMSTablePropertyHelper.HIVE_ICEBERG_STORAGE_HANDLER);
+    result.setParameters(parameters);
+    return result;
+  }
+
+  /**
+   * Applies Iceberg view metadata (SQL, schema, params) onto an existing HMS {@link Table}.
+   */
+  public static void applyIcebergViewToHmsTable(Table hmsTable, View view, Configuration conf) {
+    ViewMetadata metadata = ((BaseView) view).operations().current();
+    String sqlText = viewSqlText(view, metadata);
+
+    boolean hiveEngineEnabled = false;
+    hmsTable.setSd(HiveOperationsBase.storageDescriptor(metadata.schema(), metadata.location(), hiveEngineEnabled));
+    StorageDescriptor sd = hmsTable.getSd();
+
+    if (sd.getBucketCols() == null) {
+      sd.setBucketCols(Lists.newArrayList());
+    }
+
+    if (sd.getSortCols() == null) {
+      sd.setSortCols(Lists.newArrayList());
+    }
+
+    long maxHiveTablePropertySize =
+        conf.getLong(
+            HiveOperationsBase.HIVE_TABLE_PROPERTY_MAX_SIZE,
+            HiveOperationsBase.HIVE_TABLE_PROPERTY_MAX_SIZE_DEFAULT);
+    HMSTablePropertyHelper.updateHmsTableForIcebergView(
+        metadata.metadataFileLocation(),
+        hmsTable,
+        metadata,
+        Collections.emptySet(),
+        maxHiveTablePropertySize,
+        null);
+
+    hmsTable.setCreateTime((int) (metadata.version(1).timestampMillis() / 1000));
+    hmsTable.setLastAccessTime((int) (metadata.currentVersion().timestampMillis() / 1000));
+    hmsTable.setOwner(
+        PropertyUtil.propertyAsString(
+            metadata.properties(), HiveCatalog.HMS_TABLE_OWNER, HiveHadoopUtil.currentUser()));
+
+    // In-memory overlay for compile/describe: authoritative SQL comes from Iceberg metadata.
+    hmsTable.setViewOriginalText(sqlText);
+    hmsTable.setViewExpandedText(sqlText);
+
+    String catalogType = IcebergCatalogProperties.getCatalogType(conf);
+    if (!StringUtils.isEmpty(catalogType) && !IcebergCatalogProperties.NO_CATALOG_TYPE.equals(catalogType)) {
+      hmsTable.getParameters().put(CatalogUtil.ICEBERG_CATALOG_TYPE, IcebergCatalogProperties.getCatalogType(conf));
+    }
+  }
+
+  private static String viewSqlText(View view, ViewMetadata metadata) {
+    SQLViewRepresentation hiveRepr = view.sqlFor("hive");
+    if (hiveRepr != null) {
+      return hiveRepr.sql();
+    }
+    return HiveViewOperations.sqlFor(metadata);
+  }
+
   private static StorageDescriptor getHiveStorageDescriptor(org.apache.iceberg.Table table) {
     var result = new StorageDescriptor();
     result.setCols(HiveSchemaUtil.convert(table.schema()));