Add ALTER TABLE SET PROPERTIES statement

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

@@ -114,7 +114,8 @@ public ListenableFuture<Void> execute(
                 session,
                 metadata,
                 accessControl,
-                parameterExtractor(statement, parameters));
+                parameterExtractor(statement, parameters),
+                true);
 
         ColumnMetadata column = ColumnMetadata.builder()
                 .setName(element.getName().getValue())

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

@@ -110,7 +110,8 @@ public ListenableFuture<Void> execute(
                 session,
                 metadata,
                 accessControl,
-                parameterLookup);
+                parameterLookup,
+                true);
 
         ConnectorMaterializedViewDefinition definition = new ConnectorMaterializedViewDefinition(
                 sql,

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

@@ -96,7 +96,8 @@ static ListenableFuture<Void> internalExecute(CreateSchema statement, Metadata m
                 session,
                 metadata,
                 accessControl,
-                parameterExtractor(statement, parameters));
+                parameterExtractor(statement, parameters),
+                true);
 
         TrinoPrincipal principal = getCreatePrincipal(statement, session, metadata, catalogName.getCatalogName());
         try {

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

@@ -33,6 +33,7 @@
 import io.trino.spi.security.AccessDeniedException;
 import io.trino.spi.type.Type;
 import io.trino.spi.type.TypeNotFoundException;
+import io.trino.sql.analyzer.FeaturesConfig;
 import io.trino.sql.analyzer.Output;
 import io.trino.sql.analyzer.OutputColumn;
 import io.trino.sql.tree.ColumnDefinition;
@@ -44,6 +45,8 @@
 import io.trino.sql.tree.TableElement;
 import io.trino.transaction.TransactionManager;
 
+import javax.inject.Inject;
+
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -81,6 +84,14 @@
 public class CreateTableTask
         implements DataDefinitionTask<CreateTable>
 {
+    private final boolean disableSetPropertiesSecurityCheckForCreateDdl;
+
+    @Inject
+    public CreateTableTask(FeaturesConfig featuresConfig)
+    {
+        this.disableSetPropertiesSecurityCheckForCreateDdl = featuresConfig.isDisableSetPropertiesSecurityCheckForCreateDdl();
+    }
+
     @Override
     public String getName()
     {
@@ -155,7 +166,8 @@ ListenableFuture<Void> internalExecute(CreateTable statement, Metadata metadata,
                         session,
                         metadata,
                         accessControl,
-                        parameterLookup);
+                        parameterLookup,
+                        true);
 
                 columns.put(name, ColumnMetadata.builder()
                         .setName(name)
@@ -230,8 +242,6 @@ else if (element instanceof LikeClause) {
             }
         }
 
-        accessControl.checkCanCreateTable(session.toSecurityContext(), tableName);
-
         Map<String, Expression> sqlProperties = mapFromProperties(statement.getProperties());
         Map<String, Object> properties = metadata.getTablePropertyManager().getProperties(
                 catalogName,
@@ -240,7 +250,15 @@ else if (element instanceof LikeClause) {
                 session,
                 metadata,
                 accessControl,
-                parameterLookup);
+                parameterLookup,
+                true);
+
+        if (!disableSetPropertiesSecurityCheckForCreateDdl && !properties.isEmpty()) {
+            accessControl.checkCanCreateTable(session.toSecurityContext(), tableName, properties);
+        }
+        else {
+            accessControl.checkCanCreateTable(session.toSecurityContext(), tableName);
+        }
 
         Map<String, Object> finalProperties = combineProperties(sqlProperties.keySet(), properties, inheritedProperties);
 

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

@@ -0,0 +1,102 @@
+/*
+ * 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.execution;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import io.trino.Session;
+import io.trino.execution.warnings.WarningCollector;
+import io.trino.metadata.Metadata;
+import io.trino.metadata.QualifiedObjectName;
+import io.trino.metadata.TableHandle;
+import io.trino.security.AccessControl;
+import io.trino.sql.tree.Expression;
+import io.trino.sql.tree.SetProperties;
+import io.trino.transaction.TransactionManager;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+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.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.ParameterUtils.parameterExtractor;
+import static io.trino.sql.analyzer.SemanticExceptions.semanticException;
+
+public class SetPropertiesTask
+        implements DataDefinitionTask<SetProperties>
+{
+    @Override
+    public String getName()
+    {
+        return "SET PROPERTIES";
+    }
+
+    @Override
+    public ListenableFuture<Void> execute(
+            SetProperties statement,
+            TransactionManager transactionManager,
+            Metadata metadata,
+            AccessControl accessControl,
+            QueryStateMachine stateMachine,
+            List<Expression> parameters,
+            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 = metadata.getTablePropertyManager().getProperties(
+                    getRequiredCatalogHandle(metadata, session, statement, tableName.getCatalogName()),
+                    tableName.getCatalogName(),
+                    sqlProperties,
+                    session,
+                    metadata,
+                    accessControl,
+                    parameterExtractor(statement, parameters),
+                    false); // skip setting of default properties since they should not be stored explicitly
+            setTableProperties(statement, tableName, metadata, accessControl, session, properties);
+        }
+        else {
+            throw semanticException(NOT_SUPPORTED, statement, "Unsupported target type: %s", statement.getType());
+        }
+
+        return immediateVoidFuture();
+    }
+
+    private void setTableProperties(SetProperties statement, QualifiedObjectName tableName, Metadata metadata, AccessControl accessControl, Session session, Map<String, Object> properties)
+    {
+        if (metadata.getMaterializedView(session, tableName).isPresent()) {
+            throw semanticException(NOT_SUPPORTED, statement, "Cannot set properties to a materialized view in ALTER TABLE");
+        }
+
+        if (metadata.getView(session, tableName).isPresent()) {
+            throw semanticException(NOT_SUPPORTED, statement, "Cannot set properties to a view in ALTER TABLE");
+        }
+
+        Optional<TableHandle> tableHandle = metadata.getTableHandle(session, tableName);
+        if (tableHandle.isEmpty()) {
+            throw semanticException(TABLE_NOT_FOUND, statement, "Table does not exist: %s", tableName);
+        }
+
+        accessControl.checkCanSetTableProperties(session.toSecurityContext(), tableName, properties);
+
+        metadata.setTableProperties(session, tableHandle.get(), properties);
+    }
+}

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

@@ -77,7 +77,8 @@ public final Map<String, Object> getProperties(
             Session session,
             Metadata metadata,
             AccessControl accessControl,
-            Map<NodeRef<Parameter>, Expression> parameters)
+            Map<NodeRef<Parameter>, Expression> parameters,
+            boolean setDefaultProperties)
     {
         Map<String, PropertyMetadata<?>> supportedProperties = connectorProperties.get(catalogName);
         if (supportedProperties == null) {
@@ -134,6 +135,9 @@ public final Map<String, Object> getProperties(
         }
         Map<String, Object> userSpecifiedProperties = properties.build();
 
+        if (!setDefaultProperties) {
+            return properties.build();
+        }
         // Fill in the remaining properties with non-null defaults
         for (PropertyMetadata<?> propertyMetadata : supportedProperties.values()) {
             if (!userSpecifiedProperties.containsKey(propertyMetadata.getName())) {

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

@@ -200,6 +200,11 @@ public interface Metadata
      */
     void renameTable(Session session, TableHandle tableHandle, QualifiedObjectName newTableName);
 
+    /**
+     * Set properties to the specified table.
+     */
+    void setTableProperties(Session session, TableHandle tableHandle, Map<String, Object> properties);
+
     /**
      * Comments to the specified table.
      */

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

@@ -773,6 +773,14 @@ public void renameTable(Session session, TableHandle tableHandle, QualifiedObjec
         metadata.renameTable(session.toConnectorSession(catalog), tableHandle.getConnectorHandle(), newTableName.asSchemaTableName());
     }
 
+    @Override
+    public void setTableProperties(Session session, TableHandle tableHandle, Map<String, Object> properties)
+    {
+        CatalogName catalogName = tableHandle.getCatalogName();
+        ConnectorMetadata metadata = getMetadataForWrite(session, catalogName);
+        metadata.setTableProperties(session.toConnectorSession(catalogName), tableHandle.getConnectorHandle(), properties);
+    }
+
     @Override
     public void setTableComment(Session session, TableHandle tableHandle, Optional<String> comment)
     {

core/trino-main/src/main/java/io/trino/security/AccessControl.java

@@ -27,6 +27,7 @@
 
 import java.security.Principal;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
@@ -163,9 +164,18 @@ public interface AccessControl
      * Check if identity is allowed to create the specified table.
      *
      * @throws AccessDeniedException if not allowed
+     * @deprecated use {@link #checkCanCreateTable(SecurityContext context, QualifiedObjectName tableName, Map properties)}
      */
+    @Deprecated
     void checkCanCreateTable(SecurityContext context, QualifiedObjectName tableName);
 
+    /**
+     * Check if identity is allowed to create the specified table with properties.
+     *
+     * @throws AccessDeniedException if not allowed
+     */
+    void checkCanCreateTable(SecurityContext context, QualifiedObjectName tableName, Map<String, Object> properties);
+
     /**
      * Check if identity is allowed to drop the specified table.
      *
@@ -180,6 +190,13 @@ public interface AccessControl
      */
     void checkCanRenameTable(SecurityContext context, QualifiedObjectName tableName, QualifiedObjectName newTableName);
 
+    /**
+     * Check if identity is allowed to set properties to the specified table.
+     *
+     * @throws AccessDeniedException if not allowed
+     */
+    void checkCanSetTableProperties(SecurityContext context, QualifiedObjectName tableName, Map<String, Object> properties);
+
     /**
      * Check if identity is allowed to comment the specified table.
      *

core/trino-main/src/main/java/io/trino/security/AccessControlManager.java

@@ -426,6 +426,19 @@ public void checkCanCreateTable(SecurityContext securityContext, QualifiedObject
         catalogAuthorizationCheck(tableName.getCatalogName(), securityContext, (control, context) -> control.checkCanCreateTable(context, tableName.asSchemaTableName()));
     }
 
+    @Override
+    public void checkCanCreateTable(SecurityContext securityContext, QualifiedObjectName tableName, Map<String, Object> properties)
+    {
+        requireNonNull(securityContext, "securityContext is null");
+        requireNonNull(tableName, "tableName is null");
+
+        checkCanAccessCatalog(securityContext, tableName.getCatalogName());
+
+        systemAuthorizationCheck(control -> control.checkCanCreateTable(securityContext.toSystemSecurityContext(), tableName.asCatalogSchemaTableName(), properties));
+
+        catalogAuthorizationCheck(tableName.getCatalogName(), securityContext, (control, context) -> control.checkCanCreateTable(context, tableName.asSchemaTableName(), properties));
+    }
+
     @Override
     public void checkCanDropTable(SecurityContext securityContext, QualifiedObjectName tableName)
     {
@@ -453,6 +466,20 @@ public void checkCanRenameTable(SecurityContext securityContext, QualifiedObject
         catalogAuthorizationCheck(tableName.getCatalogName(), securityContext, (control, context) -> control.checkCanRenameTable(context, tableName.asSchemaTableName(), newTableName.asSchemaTableName()));
     }
 
+    @Override
+    public void checkCanSetTableProperties(SecurityContext securityContext, QualifiedObjectName tableName, Map<String, Object> properties)
+    {
+        requireNonNull(securityContext, "securityContext is null");
+        requireNonNull(tableName, "tableName is null");
+        requireNonNull(properties, "properties is null");
+
+        checkCanAccessCatalog(securityContext, tableName.getCatalogName());
+
+        systemAuthorizationCheck(control -> control.checkCanSetTableProperties(securityContext.toSystemSecurityContext(), tableName.asCatalogSchemaTableName(), properties));
+
+        catalogAuthorizationCheck(tableName.getCatalogName(), securityContext, (control, context) -> control.checkCanSetTableProperties(context, tableName.asSchemaTableName(), properties));
+    }
+
     @Override
     public void checkCanSetTableComment(SecurityContext securityContext, QualifiedObjectName tableName)
     {

core/trino-main/src/main/java/io/trino/security/AllowAllAccessControl.java

@@ -22,6 +22,7 @@
 import io.trino.spi.security.TrinoPrincipal;
 
 import java.security.Principal;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
@@ -121,6 +122,11 @@ public void checkCanCreateTable(SecurityContext context, QualifiedObjectName tab
     {
     }
 
+    @Override
+    public void checkCanCreateTable(SecurityContext context, QualifiedObjectName tableName, Map<String, Object> properties)
+    {
+    }
+
     @Override
     public void checkCanDropTable(SecurityContext context, QualifiedObjectName tableName)
     {
@@ -131,6 +137,11 @@ public void checkCanRenameTable(SecurityContext context, QualifiedObjectName tab
     {
     }
 
+    @Override
+    public void checkCanSetTableProperties(SecurityContext context, QualifiedObjectName tableName, Map<String, Object> properties)
+    {
+    }
+
     @Override
     public void checkCanSetTableComment(SecurityContext context, QualifiedObjectName tableName)
     {