Add View support to the Iceberg Glue Catalog

plugin/trino-hive/src/main/java/io/trino/plugin/hive/ViewReaderUtil.java

@@ -166,6 +166,11 @@ public static class PrestoViewReader
     {
         @Override
         public ConnectorViewDefinition decodeViewData(String viewData, Table table, CatalogName catalogName)
+        {
+            return decodeViewData(viewData);
+        }
+
+        public static ConnectorViewDefinition decodeViewData(String viewData)
         {
             checkCondition(viewData.startsWith(VIEW_PREFIX), HIVE_INVALID_VIEW_DATA, "View data missing prefix: %s", viewData);
             checkCondition(viewData.endsWith(VIEW_SUFFIX), HIVE_INVALID_VIEW_DATA, "View data missing suffix: %s", viewData);

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/AbstractTrinoCatalog.java

@@ -13,10 +13,15 @@
  */
 package io.trino.plugin.iceberg.catalog;
 
+import com.google.common.collect.ImmutableMap;
 import io.trino.plugin.hive.HdfsEnvironment;
+import io.trino.plugin.hive.HiveMetadata;
+import io.trino.plugin.hive.HiveViewNotSupportedException;
+import io.trino.plugin.hive.ViewReaderUtil;
 import io.trino.plugin.iceberg.ColumnIdentity;
 import io.trino.spi.TrinoException;
 import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.ConnectorViewDefinition;
 import io.trino.spi.connector.SchemaTableName;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -31,8 +36,13 @@
 import java.util.Map;
 import java.util.Optional;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static io.trino.plugin.hive.HiveMetadata.TABLE_COMMENT;
+import static io.trino.plugin.hive.ViewReaderUtil.PRESTO_VIEW_FLAG;
+import static io.trino.plugin.hive.ViewReaderUtil.isHiveOrPrestoView;
+import static io.trino.plugin.hive.ViewReaderUtil.isPrestoView;
 import static io.trino.plugin.iceberg.IcebergErrorCode.ICEBERG_FILESYSTEM_ERROR;
+import static io.trino.spi.StandardErrorCode.TABLE_NOT_FOUND;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 import static java.util.UUID.randomUUID;
@@ -42,14 +52,25 @@
 public abstract class AbstractTrinoCatalog
         implements TrinoCatalog
 {
+    // Be compatible with views defined by the Hive connector, which can be useful under certain conditions.
+    protected static final String TRINO_CREATED_BY = HiveMetadata.TRINO_CREATED_BY;
+    protected static final String TRINO_CREATED_BY_VALUE = "Trino Iceberg connector";
+    protected static final String PRESTO_VIEW_COMMENT = HiveMetadata.PRESTO_VIEW_COMMENT;
+    protected static final String PRESTO_VERSION_NAME = HiveMetadata.PRESTO_VERSION_NAME;
+    protected static final String PRESTO_QUERY_ID_NAME = HiveMetadata.PRESTO_QUERY_ID_NAME;
+    protected static final String PRESTO_VIEW_EXPANDED_TEXT_MARKER = HiveMetadata.PRESTO_VIEW_EXPANDED_TEXT_MARKER;
+
     protected final IcebergTableOperationsProvider tableOperationsProvider;
+    private final String trinoVersion;
     private final boolean useUniqueTableLocation;
 
     protected AbstractTrinoCatalog(
             IcebergTableOperationsProvider tableOperationsProvider,
+            String trinoVersion,
             boolean useUniqueTableLocation)
     {
         this.tableOperationsProvider = requireNonNull(tableOperationsProvider, "tableOperationsProvider is null");
+        this.trinoVersion = requireNonNull(trinoVersion, "trinoVersion is null");
         this.useUniqueTableLocation = useUniqueTableLocation;
     }
 
@@ -72,6 +93,26 @@ public void updateColumnComment(ConnectorSession session, SchemaTableName schema
         icebergTable.updateSchema().updateColumnDoc(columnIdentity.getName(), comment.orElse(null)).commit();
     }
 
+    @Override
+    public Map<SchemaTableName, ConnectorViewDefinition> getViews(ConnectorSession session, Optional<String> namespace)
+    {
+        ImmutableMap.Builder<SchemaTableName, ConnectorViewDefinition> views = ImmutableMap.builder();
+        for (SchemaTableName name : listViews(session, namespace)) {
+            try {
+                getView(session, name).ifPresent(view -> views.put(name, view));
+            }
+            catch (TrinoException e) {
+                if (e.getErrorCode().equals(TABLE_NOT_FOUND.toErrorCode())) {
+                    // Ignore view that was dropped during query execution (race condition)
+                }
+                else {
+                    throw e;
+                }
+            }
+        }
+        return views.buildOrThrow();
+    }
+
     protected Transaction newCreateTableTransaction(
             ConnectorSession session,
             SchemaTableName schemaTableName,
@@ -115,4 +156,54 @@ protected void deleteTableDirectory(
             throw new TrinoException(ICEBERG_FILESYSTEM_ERROR, format("Failed to delete directory %s of the table %s", tableLocation, schemaTableName), e);
         }
     }
+
+    protected Optional<ConnectorViewDefinition> getView(
+            SchemaTableName viewName,
+            Optional<String> viewOriginalText,
+            String tableType,
+            Map<String, String> tableParameters,
+            Optional<String> tableOwner)
+    {
+        if (!isView(tableType, tableParameters)) {
+            // Filter out Tables and Materialized Views
+            return Optional.empty();
+        }
+
+        if (!isPrestoView(tableParameters)) {
+            // Hive views are not compatible
+            throw new HiveViewNotSupportedException(viewName);
+        }
+
+        checkArgument(viewOriginalText.isPresent(), "viewOriginalText must be present");
+        ConnectorViewDefinition definition = ViewReaderUtil.PrestoViewReader.decodeViewData(viewOriginalText.get());
+        // use owner from table metadata if it exists
+        if (tableOwner.isPresent() && !definition.isRunAsInvoker()) {
+            definition = new ConnectorViewDefinition(
+                    definition.getOriginalSql(),
+                    definition.getCatalog(),
+                    definition.getSchema(),
+                    definition.getColumns(),
+                    definition.getComment(),
+                    tableOwner,
+                    false);
+        }
+        return Optional.of(definition);
+    }
+
+    private static boolean isView(String tableType, Map<String, String> tableParameters)
+
+    {
+        return isHiveOrPrestoView(tableType) && PRESTO_VIEW_COMMENT.equals(tableParameters.get(TABLE_COMMENT));
+    }
+
+    protected Map<String, String> createViewProperties(ConnectorSession session)
+    {
+        return ImmutableMap.<String, String>builder()
+                .put(PRESTO_VIEW_FLAG, "true") // Ensures compatibility with views created by the Hive connector
+                .put(TRINO_CREATED_BY, TRINO_CREATED_BY_VALUE)
+                .put(PRESTO_VERSION_NAME, trinoVersion)
+                .put(PRESTO_QUERY_ID_NAME, session.getQueryId())
+                .put(TABLE_COMMENT, PRESTO_VIEW_COMMENT)
+                .buildOrThrow();
+    }
 }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/TrinoCatalog.java

@@ -101,7 +101,7 @@ Transaction newCreateTableTransaction(
 
     Map<SchemaTableName, ConnectorViewDefinition> getViews(ConnectorSession session, Optional<String> namespace);
 
-    Optional<ConnectorViewDefinition> getView(ConnectorSession session, SchemaTableName viewIdentifier);
+    Optional<ConnectorViewDefinition> getView(ConnectorSession session, SchemaTableName viewName);
 
     List<SchemaTableName> listMaterializedViews(ConnectorSession session, Optional<String> namespace);
 

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/GlueIcebergUtil.java

@@ -15,10 +15,14 @@
 
 import com.amazonaws.services.glue.model.TableInput;
 
+import javax.annotation.Nullable;
+
 import java.util.Map;
 import java.util.Optional;
 
+import static io.trino.plugin.hive.HiveMetadata.PRESTO_VIEW_EXPANDED_TEXT_MARKER;
 import static org.apache.hadoop.hive.metastore.TableType.EXTERNAL_TABLE;
+import static org.apache.hadoop.hive.metastore.TableType.VIRTUAL_VIEW;
 
 public final class GlueIcebergUtil
 {
@@ -33,4 +37,15 @@ public static TableInput getTableInput(String tableName, Optional<String> owner,
                 // Iceberg does not distinguish managed and external tables, all tables are treated the same and marked as EXTERNAL
                 .withTableType(EXTERNAL_TABLE.name());
     }
+
+    public static TableInput getViewTableInput(String viewName, String viewOriginalText, @Nullable String owner, Map<String, String> parameters)
+    {
+        return new TableInput()
+                .withName(viewName)
+                .withTableType(VIRTUAL_VIEW.name())
+                .withViewOriginalText(viewOriginalText)
+                .withViewExpandedText(PRESTO_VIEW_EXPANDED_TEXT_MARKER)
+                .withOwner(owner)
+                .withParameters(parameters);
+    }
 }