Add support for creating view with properties in engine

core/trino-main/src/main/java/io/trino/connector/CatalogServiceProviderModule.java

@@ -29,6 +29,7 @@
 import io.trino.metadata.SessionPropertyManager;
 import io.trino.metadata.TableProceduresPropertyManager;
 import io.trino.metadata.TablePropertyManager;
+import io.trino.metadata.ViewPropertyManager;
 import io.trino.security.AccessControlManager;
 import io.trino.spi.connector.ConnectorAccessControl;
 import io.trino.spi.connector.ConnectorIndexProvider;
@@ -134,6 +135,13 @@ public static TablePropertyManager createTablePropertyManager(ConnectorServicesP
         return new TablePropertyManager(new ConnectorCatalogServiceProvider<>("table properties", connectorServicesProvider, ConnectorServices::getTableProperties));
     }
 
+    @Provides
+    @Singleton
+    public static ViewPropertyManager createViewPropertyManager(ConnectorServicesProvider connectorServicesProvider)
+    {
+        return new ViewPropertyManager(new ConnectorCatalogServiceProvider<>("view properties", connectorServicesProvider, ConnectorServices::getViewProperties));
+    }
+
     @Provides
     @Singleton
     public static MaterializedViewPropertyManager createMaterializedViewPropertyManager(ConnectorServicesProvider connectorServicesProvider)

core/trino-main/src/main/java/io/trino/connector/ConnectorServices.java

@@ -86,6 +86,7 @@ public class ConnectorServices
     private final List<EventListener> eventListeners;
     private final Map<String, PropertyMetadata<?>> sessionProperties;
     private final Map<String, PropertyMetadata<?>> tableProperties;
+    private final Map<String, PropertyMetadata<?>> viewProperties;
     private final Map<String, PropertyMetadata<?>> materializedViewProperties;
     private final Map<String, PropertyMetadata<?>> schemaProperties;
     private final Map<String, PropertyMetadata<?>> columnProperties;
@@ -194,6 +195,10 @@ public ConnectorServices(Tracer tracer, CatalogHandle catalogHandle, Connector c
         requireNonNull(tableProperties, format("Connector '%s' returned a null table properties set", catalogHandle));
         this.tableProperties = Maps.uniqueIndex(tableProperties, PropertyMetadata::getName);
 
+        List<PropertyMetadata<?>> viewProperties = connector.getViewProperties();
+        requireNonNull(viewProperties, format("Connector '%s' returned a null view properties set", catalogHandle));
+        this.viewProperties = Maps.uniqueIndex(viewProperties, PropertyMetadata::getName);
+
         List<PropertyMetadata<?>> materializedViewProperties = connector.getMaterializedViewProperties();
         requireNonNull(materializedViewProperties, format("Connector '%s' returned a null materialized view properties set", catalogHandle));
         this.materializedViewProperties = Maps.uniqueIndex(materializedViewProperties, PropertyMetadata::getName);
@@ -306,6 +311,11 @@ public Map<String, PropertyMetadata<?>> getTableProperties()
         return tableProperties;
     }
 
+    public Map<String, PropertyMetadata<?>> getViewProperties()
+    {
+        return viewProperties;
+    }
+
     public Map<String, PropertyMetadata<?>> getMaterializedViewProperties()
     {
         return materializedViewProperties;

core/trino-main/src/main/java/io/trino/execution/CreateViewTask.java

@@ -22,23 +22,28 @@
 import io.trino.metadata.QualifiedObjectName;
 import io.trino.metadata.ViewColumn;
 import io.trino.metadata.ViewDefinition;
+import io.trino.metadata.ViewPropertyManager;
 import io.trino.security.AccessControl;
+import io.trino.spi.connector.CatalogHandle;
 import io.trino.spi.security.Identity;
 import io.trino.sql.PlannerContext;
 import io.trino.sql.analyzer.Analysis;
 import io.trino.sql.analyzer.AnalyzerFactory;
 import io.trino.sql.parser.SqlParser;
 import io.trino.sql.tree.CreateView;
 import io.trino.sql.tree.Expression;
+import io.trino.sql.tree.NodeRef;
+import io.trino.sql.tree.Parameter;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.util.concurrent.Futures.immediateVoidFuture;
 import static io.trino.execution.ParameterExtractor.bindParameters;
 import static io.trino.metadata.MetadataUtil.createQualifiedObjectName;
-import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
+import static io.trino.metadata.MetadataUtil.getRequiredCatalogHandle;
 import static io.trino.spi.StandardErrorCode.TABLE_ALREADY_EXISTS;
 import static io.trino.sql.SqlFormatterUtil.getFormattedSql;
 import static io.trino.sql.analyzer.SemanticExceptions.semanticException;
@@ -52,14 +57,21 @@ public class CreateViewTask
     private final AccessControl accessControl;
     private final SqlParser sqlParser;
     private final AnalyzerFactory analyzerFactory;
+    private final ViewPropertyManager viewPropertyManager;
 
     @Inject
-    public CreateViewTask(PlannerContext plannerContext, AccessControl accessControl, SqlParser sqlParser, AnalyzerFactory analyzerFactory)
+    public CreateViewTask(
+            PlannerContext plannerContext,
+            AccessControl accessControl,
+            SqlParser sqlParser,
+            AnalyzerFactory analyzerFactory,
+            ViewPropertyManager viewPropertyManager)
     {
         this.plannerContext = requireNonNull(plannerContext, "plannerContext is null");
         this.accessControl = requireNonNull(accessControl, "accessControl is null");
         this.sqlParser = requireNonNull(sqlParser, "sqlParser is null");
         this.analyzerFactory = requireNonNull(analyzerFactory, "analyzerFactory is null");
+        this.viewPropertyManager = requireNonNull(viewPropertyManager, "viewPropertyManager is null");
     }
 
     @Override
@@ -76,9 +88,8 @@ public ListenableFuture<Void> execute(
             WarningCollector warningCollector)
     {
         Metadata metadata = plannerContext.getMetadata();
-        if (!statement.getProperties().isEmpty()) {
-            throw semanticException(NOT_SUPPORTED, statement, "Creating views with properties is not supported");
-        }
+        Map<NodeRef<Parameter>, Expression> parameterLookup = bindParameters(statement, parameters);
+
         Session session = stateMachine.getSession();
         QualifiedObjectName name = createQualifiedObjectName(session, statement, statement.getName());
 
@@ -98,7 +109,7 @@ else if (metadata.getTableHandle(session, name).isPresent()) {
 
         String sql = getFormattedSql(statement.getQuery(), sqlParser);
 
-        Analysis analysis = analyzerFactory.createAnalyzer(session, parameters, bindParameters(statement, parameters), stateMachine.getWarningCollector(), stateMachine.getPlanOptimizersStatsCollector())
+        Analysis analysis = analyzerFactory.createAnalyzer(session, parameters, parameterLookup, stateMachine.getWarningCollector(), stateMachine.getPlanOptimizersStatsCollector())
                 .analyze(statement);
 
         List<ViewColumn> columns = analysis.getOutputDescriptor(statement.getQuery())
@@ -112,6 +123,19 @@ else if (metadata.getTableHandle(session, name).isPresent()) {
             owner = Optional.empty();
         }
 
+        String catalogName = name.getCatalogName();
+        CatalogHandle catalogHandle = getRequiredCatalogHandle(metadata, session, statement, catalogName);
+
+        Map<String, Object> properties = viewPropertyManager.getProperties(
+                name.getCatalogName(),
+                catalogHandle,
+                statement.getProperties(),
+                session,
+                plannerContext,
+                accessControl,
+                parameterLookup,
+                true);
+
         ViewDefinition definition = new ViewDefinition(
                 sql,
                 session.getCatalog(),
@@ -124,7 +148,7 @@ else if (metadata.getTableHandle(session, name).isPresent()) {
                         .filter(element -> !element.getCatalogName().equals(GlobalSystemConnector.NAME))
                         .collect(toImmutableList()));
 
-        metadata.createView(session, name, definition, statement.isReplace());
+        metadata.createView(session, name, definition, properties, statement.isReplace());
 
         stateMachine.setOutput(analysis.getTarget());
         stateMachine.setReferencedTables(analysis.getReferencedTables());

core/trino-main/src/main/java/io/trino/metadata/Metadata.java

@@ -492,6 +492,11 @@ default boolean isView(Session session, QualifiedObjectName viewName)
      */
     Optional<ViewDefinition> getView(Session session, QualifiedObjectName viewName);
 
+    /**
+     * Returns the view definition for the specified view name.
+     */
+    Map<String, Object> getViewProperties(Session session, QualifiedObjectName viewName);
+
     /**
      * Gets the schema properties for the specified schema.
      */
@@ -505,7 +510,7 @@ default boolean isView(Session session, QualifiedObjectName viewName)
     /**
      * Creates the specified view with the specified view definition.
      */
-    void createView(Session session, QualifiedObjectName viewName, ViewDefinition definition, boolean replace);
+    void createView(Session session, QualifiedObjectName viewName, ViewDefinition definition, Map<String, Object> properties, boolean replace);
 
     /**
      * Rename the specified view.

core/trino-main/src/main/java/io/trino/metadata/MetadataManager.java

@@ -1469,6 +1469,23 @@ public Optional<ViewDefinition> getView(Session session, QualifiedObjectName vie
         return Optional.of(createViewDefinition(viewName, connectorView.get(), Optional.of(runAsIdentity)));
     }
 
+    @Override
+    public Map<String, Object> getViewProperties(Session session, QualifiedObjectName viewName)
+    {
+        Optional<CatalogMetadata> catalog = getOptionalCatalogMetadata(session, viewName.getCatalogName());
+        if (catalog.isPresent()) {
+            CatalogMetadata catalogMetadata = catalog.get();
+            CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, viewName);
+            ConnectorMetadata metadata = catalogMetadata.getMetadataFor(session, catalogHandle);
+
+            ConnectorSession connectorSession = session.toConnectorSession(catalogHandle);
+            return ImmutableMap.copyOf(metadata.getViewProperties(
+                    connectorSession,
+                    viewName.asSchemaTableName()));
+        }
+        return ImmutableMap.of();
+    }
+
     private static ViewDefinition createViewDefinition(QualifiedObjectName viewName, ConnectorViewDefinition view, Optional<Identity> runAsIdentity)
     {
         if (view.isRunAsInvoker() && runAsIdentity.isPresent()) {
@@ -1509,13 +1526,13 @@ private Optional<ConnectorViewDefinition> getViewInternal(Session session, Quali
     }
 
     @Override
-    public void createView(Session session, QualifiedObjectName viewName, ViewDefinition definition, boolean replace)
+    public void createView(Session session, QualifiedObjectName viewName, ViewDefinition definition, Map<String, Object> viewProperties, boolean replace)
     {
         CatalogMetadata catalogMetadata = getCatalogMetadataForWrite(session, viewName.getCatalogName());
         CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle();
         ConnectorMetadata metadata = catalogMetadata.getMetadata(session);
 
-        metadata.createView(session.toConnectorSession(catalogHandle), viewName.asSchemaTableName(), definition.toConnectorViewDefinition(), replace);
+        metadata.createView(session.toConnectorSession(catalogHandle), viewName.asSchemaTableName(), definition.toConnectorViewDefinition(), viewProperties, replace);
         if (catalogMetadata.getSecurityManagement() == SYSTEM) {
             systemSecurityMetadata.tableCreated(session, viewName.asCatalogSchemaTableName());
         }

core/trino-main/src/main/java/io/trino/metadata/ViewPropertyManager.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed 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 io.trino.metadata;
+
+import io.trino.connector.CatalogServiceProvider;
+import io.trino.spi.session.PropertyMetadata;
+
+import java.util.Map;
+
+import static io.trino.spi.StandardErrorCode.INVALID_VIEW_PROPERTY;
+
+public class ViewPropertyManager
+        extends AbstractCatalogPropertyManager
+{
+    public ViewPropertyManager(CatalogServiceProvider<Map<String, PropertyMetadata<?>>> connectorProperties)
+    {
+        super("view", INVALID_VIEW_PROPERTY, connectorProperties);
+    }
+}

core/trino-main/src/main/java/io/trino/sql/rewrite/ShowQueriesRewrite.java

@@ -39,6 +39,7 @@
 import io.trino.metadata.TableHandle;
 import io.trino.metadata.TablePropertyManager;
 import io.trino.metadata.ViewDefinition;
+import io.trino.metadata.ViewPropertyManager;
 import io.trino.security.AccessControl;
 import io.trino.spi.StandardErrorCode;
 import io.trino.spi.TrinoException;
@@ -130,6 +131,7 @@
 import static io.trino.spi.StandardErrorCode.INVALID_SCHEMA_PROPERTY;
 import static io.trino.spi.StandardErrorCode.INVALID_TABLE_PROPERTY;
 import static io.trino.spi.StandardErrorCode.INVALID_VIEW;
+import static io.trino.spi.StandardErrorCode.INVALID_VIEW_PROPERTY;
 import static io.trino.spi.StandardErrorCode.MISSING_CATALOG_NAME;
 import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
 import static io.trino.spi.StandardErrorCode.SCHEMA_NOT_FOUND;
@@ -180,6 +182,7 @@ public final class ShowQueriesRewrite
     private final SchemaPropertyManager schemaPropertyManager;
     private final ColumnPropertyManager columnPropertyManager;
     private final TablePropertyManager tablePropertyManager;
+    private final ViewPropertyManager viewPropertyManager;
     private final MaterializedViewPropertyManager materializedViewPropertyManager;
 
     @Inject
@@ -191,6 +194,7 @@ public ShowQueriesRewrite(
             SchemaPropertyManager schemaPropertyManager,
             ColumnPropertyManager columnPropertyManager,
             TablePropertyManager tablePropertyManager,
+            ViewPropertyManager viewPropertyManager,
             MaterializedViewPropertyManager materializedViewPropertyManager)
     {
         this.metadata = requireNonNull(metadata, "metadata is null");
@@ -200,6 +204,7 @@ public ShowQueriesRewrite(
         this.schemaPropertyManager = requireNonNull(schemaPropertyManager, "schemaPropertyManager is null");
         this.columnPropertyManager = requireNonNull(columnPropertyManager, "columnPropertyManager is null");
         this.tablePropertyManager = requireNonNull(tablePropertyManager, "tablePropertyManager is null");
+        this.viewPropertyManager = requireNonNull(viewPropertyManager, "viewPropertyManager is null");
         this.materializedViewPropertyManager = requireNonNull(materializedViewPropertyManager, "materializedViewPropertyManager is null");
     }
 
@@ -221,6 +226,7 @@ public Statement rewrite(
                 schemaPropertyManager,
                 columnPropertyManager,
                 tablePropertyManager,
+                viewPropertyManager,
                 materializedViewPropertyManager);
         return (Statement) visitor.process(node, null);
     }
@@ -236,6 +242,7 @@ private static class Visitor
         private final SchemaPropertyManager schemaPropertyManager;
         private final ColumnPropertyManager columnPropertyManager;
         private final TablePropertyManager tablePropertyManager;
+        private final ViewPropertyManager viewPropertyManager;
         private final MaterializedViewPropertyManager materializedViewPropertyManager;
 
         public Visitor(
@@ -247,6 +254,7 @@ public Visitor(
                 SchemaPropertyManager schemaPropertyManager,
                 ColumnPropertyManager columnPropertyManager,
                 TablePropertyManager tablePropertyManager,
+                ViewPropertyManager viewPropertyManager,
                 MaterializedViewPropertyManager materializedViewPropertyManager)
         {
             this.metadata = requireNonNull(metadata, "metadata is null");
@@ -257,6 +265,7 @@ public Visitor(
             this.schemaPropertyManager = requireNonNull(schemaPropertyManager, "schemaPropertyManager is null");
             this.columnPropertyManager = requireNonNull(columnPropertyManager, "columnPropertyManager is null");
             this.tablePropertyManager = requireNonNull(tablePropertyManager, "tablePropertyManager is null");
+            this.viewPropertyManager = requireNonNull(viewPropertyManager, "viewPropertyManager is null");
             this.materializedViewPropertyManager = requireNonNull(materializedViewPropertyManager, "materializedViewPropertyManager is null");
         }
 
@@ -646,14 +655,18 @@ protected Node visitShowCreate(ShowCreate node, Void context)
 
                 accessControl.checkCanShowCreateTable(session.toSecurityContext(), new QualifiedObjectName(catalogName.getValue(), schemaName.getValue(), tableName.getValue()));
 
+                Map<String, Object> properties = metadata.getViewProperties(session, objectName);
+                CatalogHandle catalogHandle = getRequiredCatalogHandle(metadata, session, node, catalogName.getValue());
+                Collection<PropertyMetadata<?>> allViewProperties = viewPropertyManager.getAllProperties(catalogHandle);
+                List<Property> propertyNodes = buildProperties(objectName, Optional.empty(), INVALID_VIEW_PROPERTY, properties, allViewProperties);
                 CreateView.Security security = viewDefinition.get().isRunAsInvoker() ? INVOKER : DEFINER;
                 String sql = formatSql(new CreateView(
                         QualifiedName.of(ImmutableList.of(catalogName, schemaName, tableName)),
                         query,
                         false,
                         viewDefinition.get().getComment(),
                         Optional.of(security),
-                        ImmutableList.of()))
+                        propertyNodes))
                         .trim();
                 return singleValueQuery("Create View", sql);
             }