Support setting mv properties in core engine

This commit implements ALTER MATERIALIZED VIEW ... SET PROPERTIES... in
the core engine. The implementation supports the use of a newly
introduced SQL keyword "DEFAULT" on the right hand side of a property
assignment. The support of the DEFAULT keyword is implemented by
modifying certain classes related to properties in general (i.e. they
are not specific to materialized view). Consequently, several other SQL
statements also gain the support of the DEFAULT keyword "for free". They
are:

ALTER TABLE... ADD COLUMN...
ANALYZE
CREATE MATERIALIZED VIEW
CREATE SCHEMA
CREATE TABLE (both column properties and table properties)
CREATE TABLE AS

Note, however, that ALTER TABLE ... SET PROPERTIES ... is not in the
list above. Support for DEFAULT in ALTER TABLE...SET PROPERTIES...
requires extensive changes and will therefore be implemented in a
subsequent commit.

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

@@ -45,7 +45,6 @@
 import static io.trino.spi.StandardErrorCode.TABLE_NOT_FOUND;
 import static io.trino.spi.StandardErrorCode.TYPE_NOT_FOUND;
 import static io.trino.spi.connector.ConnectorCapabilities.NOT_NULL_COLUMN_CONSTRAINT;
-import static io.trino.sql.NodeUtils.mapFromProperties;
 import static io.trino.sql.ParameterUtils.parameterExtractor;
 import static io.trino.sql.analyzer.SemanticExceptions.semanticException;
 import static io.trino.sql.analyzer.TypeSignatureTranslator.toTypeSignature;
@@ -117,17 +116,14 @@ public ListenableFuture<Void> execute(
         if (!element.isNullable() && !plannerContext.getMetadata().getConnectorCapabilities(session, catalogName).contains(NOT_NULL_COLUMN_CONSTRAINT)) {
             throw semanticException(NOT_SUPPORTED, element, "Catalog '%s' does not support NOT NULL for column '%s'", catalogName.getCatalogName(), element.getName());
         }
-
-        Map<String, Expression> sqlProperties = mapFromProperties(element.getProperties());
         Map<String, Object> columnProperties = columnPropertyManager.getProperties(
                 catalogName,
                 tableName.getCatalogName(),
-                sqlProperties,
+                element.getProperties(),
                 session,
                 plannerContext,
                 accessControl,
-                parameterExtractor(statement, parameters),
-                true);
+                parameterExtractor(statement, parameters));
 
         ColumnMetadata column = ColumnMetadata.builder()
                 .setName(element.getName().getValue())

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

@@ -14,6 +14,7 @@
 package io.trino.execution;
 
 import com.google.common.util.concurrent.ListenableFuture;
+import io.trino.FeaturesConfig;
 import io.trino.Session;
 import io.trino.connector.CatalogName;
 import io.trino.execution.warnings.WarningCollector;
@@ -41,7 +42,6 @@
 import static com.google.common.util.concurrent.Futures.immediateVoidFuture;
 import static io.trino.metadata.MetadataUtil.createQualifiedObjectName;
 import static io.trino.metadata.MetadataUtil.getRequiredCatalogHandle;
-import static io.trino.sql.NodeUtils.mapFromProperties;
 import static io.trino.sql.ParameterUtils.parameterExtractor;
 import static io.trino.sql.SqlFormatterUtil.getFormattedSql;
 import static java.util.Objects.requireNonNull;
@@ -54,20 +54,23 @@ public class CreateMaterializedViewTask
     private final SqlParser sqlParser;
     private final AnalyzerFactory analyzerFactory;
     private final MaterializedViewPropertyManager materializedViewPropertyManager;
+    private final boolean disableSetPropertiesSecurityCheckForCreateDdl;
 
     @Inject
     public CreateMaterializedViewTask(
             PlannerContext plannerContext,
             AccessControl accessControl,
             SqlParser sqlParser,
             AnalyzerFactory analyzerFactory,
-            MaterializedViewPropertyManager materializedViewPropertyManager)
+            MaterializedViewPropertyManager materializedViewPropertyManager,
+            FeaturesConfig featuresConfig)
     {
         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.materializedViewPropertyManager = requireNonNull(materializedViewPropertyManager, "materializedViewPropertyManager is null");
+        this.disableSetPropertiesSecurityCheckForCreateDdl = featuresConfig.isDisableSetPropertiesSecurityCheckForCreateDdl();
     }
 
     @Override
@@ -99,16 +102,14 @@ public ListenableFuture<Void> execute(
 
         CatalogName catalogName = getRequiredCatalogHandle(plannerContext.getMetadata(), session, statement, name.getCatalogName());
 
-        Map<String, Expression> sqlProperties = mapFromProperties(statement.getProperties());
         Map<String, Object> properties = materializedViewPropertyManager.getProperties(
                 catalogName,
                 name.getCatalogName(),
-                sqlProperties,
+                statement.getProperties(),
                 session,
                 plannerContext,
                 accessControl,
-                parameterLookup,
-                true);
+                parameterLookup);
 
         MaterializedViewDefinition definition = new MaterializedViewDefinition(
                 sql,
@@ -120,6 +121,12 @@ public ListenableFuture<Void> execute(
                 Optional.empty(),
                 properties);
 
+        if (!disableSetPropertiesSecurityCheckForCreateDdl) {
+            accessControl.checkCanCreateMaterializedView(session.toSecurityContext(), name, properties);
+        }
+        else {
+            accessControl.checkCanCreateMaterializedView(session.toSecurityContext(), name);
+        }
         plannerContext.getMetadata().createMaterializedView(session, name, definition, statement.isReplace(), statement.isNotExists());
 
         stateMachine.setOutput(analysis.getTarget());

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

@@ -42,7 +42,6 @@
 import static io.trino.metadata.MetadataUtil.getRequiredCatalogHandle;
 import static io.trino.spi.StandardErrorCode.ALREADY_EXISTS;
 import static io.trino.spi.StandardErrorCode.SCHEMA_ALREADY_EXISTS;
-import static io.trino.sql.NodeUtils.mapFromProperties;
 import static io.trino.sql.ParameterUtils.parameterExtractor;
 import static io.trino.sql.analyzer.SemanticExceptions.semanticException;
 import static java.util.Objects.requireNonNull;
@@ -105,12 +104,11 @@ static ListenableFuture<Void> internalExecute(
         Map<String, Object> properties = schemaPropertyManager.getProperties(
                 catalogName,
                 schema.getCatalogName(),
-                mapFromProperties(statement.getProperties()),
+                statement.getProperties(),
                 session,
                 plannerContext,
                 accessControl,
-                parameterExtractor(statement, parameters),
-                true);
+                parameterExtractor(statement, parameters));
 
         TrinoPrincipal principal = getCreatePrincipal(statement, session, plannerContext.getMetadata(), catalogName.getCatalogName());
         try {

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

@@ -73,7 +73,6 @@
 import static io.trino.spi.StandardErrorCode.TABLE_NOT_FOUND;
 import static io.trino.spi.StandardErrorCode.TYPE_NOT_FOUND;
 import static io.trino.spi.connector.ConnectorCapabilities.NOT_NULL_COLUMN_CONSTRAINT;
-import static io.trino.sql.NodeUtils.mapFromProperties;
 import static io.trino.sql.ParameterUtils.parameterExtractor;
 import static io.trino.sql.analyzer.SemanticExceptions.semanticException;
 import static io.trino.sql.analyzer.TypeSignatureTranslator.toTypeSignature;
@@ -163,17 +162,14 @@ ListenableFuture<Void> internalExecute(CreateTable statement, Session session, L
                 if (!column.isNullable() && !plannerContext.getMetadata().getConnectorCapabilities(session, catalogName).contains(NOT_NULL_COLUMN_CONSTRAINT)) {
                     throw semanticException(NOT_SUPPORTED, column, "Catalog '%s' does not support non-null column for column name '%s'", catalogName.getCatalogName(), column.getName());
                 }
-
-                Map<String, Expression> sqlProperties = mapFromProperties(column.getProperties());
                 Map<String, Object> columnProperties = columnPropertyManager.getProperties(
                         catalogName,
                         tableName.getCatalogName(),
-                        sqlProperties,
+                        column.getProperties(),
                         session,
                         plannerContext,
                         accessControl,
-                        parameterLookup,
-                        true);
+                        parameterLookup);
 
                 columns.put(name, ColumnMetadata.builder()
                         .setName(name)
@@ -247,17 +243,14 @@ else if (element instanceof LikeClause) {
                 throw new TrinoException(GENERIC_INTERNAL_ERROR, "Invalid TableElement: " + element.getClass().getName());
             }
         }
-
-        Map<String, Expression> sqlProperties = mapFromProperties(statement.getProperties());
         Map<String, Object> properties = tablePropertyManager.getProperties(
                 catalogName,
                 tableName.getCatalogName(),
-                sqlProperties,
+                statement.getProperties(),
                 session,
                 plannerContext,
                 accessControl,
-                parameterLookup,
-                true);
+                parameterLookup);
 
         if (!disableSetPropertiesSecurityCheckForCreateDdl && !properties.isEmpty()) {
             accessControl.checkCanCreateTable(session.toSecurityContext(), tableName, properties);
@@ -266,7 +259,10 @@ else if (element instanceof LikeClause) {
             accessControl.checkCanCreateTable(session.toSecurityContext(), tableName);
         }
 
-        Map<String, Object> finalProperties = combineProperties(sqlProperties.keySet(), properties, inheritedProperties);
+        Set<String> specifiedPropertyKeys = statement.getProperties().stream()
+                .map(property -> property.getName().getValue())
+                .collect(toImmutableSet());
+        Map<String, Object> finalProperties = combineProperties(specifiedPropertyKeys, properties, inheritedProperties);
 
         ConnectorTableMetadata tableMetadata = new ConnectorTableMetadata(tableName.asSchemaTableName(), ImmutableList.copyOf(columns.values()), finalProperties, statement.getComment());
         try {

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

@@ -16,9 +16,11 @@
 import com.google.common.util.concurrent.ListenableFuture;
 import io.trino.Session;
 import io.trino.execution.warnings.WarningCollector;
+import io.trino.metadata.Properties;
 import io.trino.metadata.QualifiedObjectName;
 import io.trino.metadata.TableHandle;
 import io.trino.metadata.TablePropertyManager;
+import io.trino.metadata.MaterializedViewPropertyManager;
 import io.trino.security.AccessControl;
 import io.trino.sql.PlannerContext;
 import io.trino.sql.tree.Expression;
@@ -35,9 +37,12 @@
 import static io.trino.metadata.MetadataUtil.getRequiredCatalogHandle;
 import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
 import static io.trino.spi.StandardErrorCode.TABLE_NOT_FOUND;
-import static io.trino.sql.NodeUtils.mapFromProperties;
+import static io.trino.sql.NodeUtils.throwOnDefaultProperty;
 import static io.trino.sql.ParameterUtils.parameterExtractor;
 import static io.trino.sql.analyzer.SemanticExceptions.semanticException;
+import static io.trino.sql.tree.SetProperties.Type.MATERIALIZED_VIEW;
+import static io.trino.sql.tree.SetProperties.Type.TABLE;
+import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 
 public class SetPropertiesTask
@@ -46,13 +51,15 @@ public class SetPropertiesTask
     private final PlannerContext plannerContext;
     private final AccessControl accessControl;
     private final TablePropertyManager tablePropertyManager;
+    private final MaterializedViewPropertyManager materializedViewPropertyManager;
 
     @Inject
-    public SetPropertiesTask(PlannerContext plannerContext, AccessControl accessControl, TablePropertyManager tablePropertyManager)
+    public SetPropertiesTask(PlannerContext plannerContext, AccessControl accessControl, TablePropertyManager tablePropertyManager, MaterializedViewPropertyManager materializedViewPropertyManager)
     {
         this.plannerContext = requireNonNull(plannerContext, "plannerContext is null");
         this.accessControl = requireNonNull(accessControl, "accessControl is null");
         this.tablePropertyManager = requireNonNull(tablePropertyManager, "tablePropertyManager is null");
+        this.materializedViewPropertyManager = requireNonNull(materializedViewPropertyManager, "materializedViewPropertyManager is null");
     }
 
     @Override
@@ -69,21 +76,30 @@ public ListenableFuture<Void> execute(
             WarningCollector warningCollector)
     {
         Session session = stateMachine.getSession();
-        QualifiedObjectName tableName = createQualifiedObjectName(session, statement, statement.getName());
-
-        Map<String, Expression> sqlProperties = mapFromProperties(statement.getProperties());
-
-        if (statement.getType() == SetProperties.Type.TABLE) {
-            Map<String, Object> properties = tablePropertyManager.getProperties(
-                    getRequiredCatalogHandle(plannerContext.getMetadata(), session, statement, tableName.getCatalogName()),
-                    tableName.getCatalogName(),
-                    sqlProperties,
+        QualifiedObjectName objectName = createQualifiedObjectName(session, statement, statement.getName());
+
+        if (statement.getType() == TABLE) {
+            throwOnDefaultProperty(statement.getProperties(), "ALTER TABLE... SET PROPERTIES...");
+            Map<String, Object> properties = tablePropertyManager.getOnlySpecifiedProperties(
+                    getRequiredCatalogHandle(plannerContext.getMetadata(), session, statement, objectName.getCatalogName()),
+                    objectName.getCatalogName(),
+                    statement.getProperties(),
+                    session,
+                    plannerContext,
+                    accessControl,
+                    parameterExtractor(statement, parameters)).getNonNullProperties();
+            setTableProperties(statement, objectName, session, properties);
+        }
+        else if (statement.getType() == MATERIALIZED_VIEW) {
+            Properties properties = materializedViewPropertyManager.getOnlySpecifiedProperties(
+                    getRequiredCatalogHandle(plannerContext.getMetadata(), session, statement, objectName.getCatalogName()),
+                    objectName.getCatalogName(),
+                    statement.getProperties(),
                     session,
                     plannerContext,
                     accessControl,
-                    parameterExtractor(statement, parameters),
-                    false); // skip setting of default properties since they should not be stored explicitly
-            setTableProperties(statement, tableName, session, properties);
+                    parameterExtractor(statement, parameters));
+            setMaterializedViewProperties(statement, objectName, session, properties);
         }
         else {
             throw semanticException(NOT_SUPPORTED, statement, "Unsupported target type: %s", statement.getType());
@@ -111,4 +127,24 @@ private void setTableProperties(SetProperties statement, QualifiedObjectName tab
 
         plannerContext.getMetadata().setTableProperties(session, tableHandle.get(), properties);
     }
+
+    private void setMaterializedViewProperties(
+            SetProperties statement,
+            QualifiedObjectName materializedViewName,
+            Session session,
+            Properties properties)
+    {
+        if (plannerContext.getMetadata().getMaterializedView(session, materializedViewName).isEmpty()) {
+            String exceptionMessage = format("Materialized View '%s' does not exist", materializedViewName);
+            if (plannerContext.getMetadata().getView(session, materializedViewName).isPresent()) {
+                exceptionMessage += ", but a view with that name exists.";
+            }
+            else if (plannerContext.getMetadata().getTableHandle(session, materializedViewName).isPresent()) {
+                exceptionMessage += ", but a table with that name exists. Did you mean ALTER TABLE " + materializedViewName + " SET PROPERTIES ...?";
+            }
+            throw semanticException(TABLE_NOT_FOUND, statement, exceptionMessage);
+        }
+        accessControl.checkCanSetMaterializedViewProperties(session.toSecurityContext(), materializedViewName, properties.getNonNullProperties(), properties.getNullPropertyNames());
+        plannerContext.getMetadata().setMaterializedViewProperties(session, materializedViewName, properties.getNonNullProperties(), properties.getNullPropertyNames());
+    }
 }

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

@@ -22,6 +22,7 @@
 import io.trino.sql.tree.Expression;
 import io.trino.sql.tree.NodeRef;
 import io.trino.sql.tree.Parameter;
+import io.trino.sql.tree.Property;
 
 import java.util.List;
 import java.util.Map;
@@ -44,25 +45,57 @@ public void removeProperties(CatalogName catalogName)
         doRemoveProperties(catalogName);
     }
 
+    /**
+     * Evaluate {@code properties}. The returned {@code Map<String, Object>} contains a supported property iff its value is
+     * (implicitly or explictly) set to a non-{@code null} value. Thus,
+     * <ul>
+     *     <li>If a property does not appear in the result, then its value is {@code null}</li>
+     *     <li>If a {@code Property} in {@code properties} is set to DEFAULT, then it will appear in the result iff the default value is not
+     *     {@code null}.</li>
+     *     <li>If a property does not appear in {@code properties} at all, then it will appear in the result iff its default value is not
+     *     {@code null}. (Thus, specifying a property to have the DEFAULT value is the same as not specifying it at all.)</li>
+     * </ul>
+     */
     public Map<String, Object> getProperties(
             CatalogName catalog,
             String catalogNameForDiagnostics,
-            Map<String, Expression> sqlPropertyValues,
+            Iterable<Property> properties,
             Session session,
             PlannerContext plannerContext,
             AccessControl accessControl,
-            Map<NodeRef<Parameter>, Expression> parameters,
-            boolean setDefaultProperties)
+            Map<NodeRef<Parameter>, Expression> parameters)
     {
         return doGetProperties(
                 catalog,
                 catalogNameForDiagnostics,
-                sqlPropertyValues,
+                properties,
                 session,
                 plannerContext,
                 accessControl,
-                parameters,
-                setDefaultProperties);
+                parameters);
+    }
+
+    /**
+     * Evaluate {@code properties}. Unlike {@link #getProperties}, a property appears in the returned result iff it is specified in
+     * {@code properties}.
+     */
+    public Properties getOnlySpecifiedProperties(
+            CatalogName catalog,
+            String catalogNameForDiagnostics,
+            Iterable<Property> properties,
+            Session session,
+            PlannerContext plannerContext,
+            AccessControl accessControl,
+            Map<NodeRef<Parameter>, Expression> parameters)
+    {
+        return doGetOnlySpecifiedProperties(
+                catalog,
+                catalogNameForDiagnostics,
+                properties,
+                session,
+                plannerContext,
+                accessControl,
+                parameters);
     }
 
     public Map<CatalogName, Map<String, PropertyMetadata<?>>> getAllProperties()