Add access control to drop_extended_stats and vacuum in Delta Lake

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/procedure/DropExtendedStatsProcedure.java

@@ -17,6 +17,7 @@
 import io.trino.plugin.deltalake.DeltaLakeMetadataFactory;
 import io.trino.plugin.deltalake.statistics.DeltaLakeStatisticsAccess;
 import io.trino.spi.TrinoException;
+import io.trino.spi.connector.ConnectorAccessControl;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.procedure.Procedure;
@@ -42,6 +43,7 @@ public class DropExtendedStatsProcedure
             DropExtendedStatsProcedure.class,
             "dropStats",
             ConnectorSession.class,
+            ConnectorAccessControl.class,
             // Schema name and table name
             String.class,
             String.class);
@@ -68,7 +70,7 @@ public Procedure get()
                 PROCEDURE_METHOD.bindTo(this));
     }
 
-    public void dropStats(ConnectorSession session, String schema, String table)
+    public void dropStats(ConnectorSession session, ConnectorAccessControl accessControl, String schema, String table)
     {
         checkProcedureArgument(schema != null, "schema_name cannot be null");
         checkProcedureArgument(table != null, "table_name cannot be null");
@@ -78,6 +80,7 @@ public void dropStats(ConnectorSession session, String schema, String table)
         if (metadata.getTableHandle(session, name) == null) {
             throw new TrinoException(INVALID_PROCEDURE_ARGUMENT, format("Table '%s' does not exist", name));
         }
+        accessControl.checkCanInsertIntoTable(null, name);
         statsAccess.deleteDeltaLakeStatistics(session, metadata.getMetastore().getTableLocation(name, session));
     }
 }

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/procedure/VacuumProcedure.java

@@ -30,6 +30,7 @@
 import io.trino.plugin.hive.HdfsEnvironment;
 import io.trino.spi.TrinoException;
 import io.trino.spi.classloader.ThreadContextClassLoader;
+import io.trino.spi.connector.ConnectorAccessControl;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.procedure.Procedure;
@@ -73,6 +74,7 @@ public class VacuumProcedure
             VacuumProcedure.class,
             "vacuum",
             ConnectorSession.class,
+            ConnectorAccessControl.class,
             String.class,
             String.class,
             String.class);
@@ -111,12 +113,13 @@ public Procedure get()
 
     public void vacuum(
             ConnectorSession session,
+            ConnectorAccessControl accessControl,
             String schema,
             String table,
             String retention)
     {
         try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(getClass().getClassLoader())) {
-            doVacuum(session, schema, table, retention);
+            doVacuum(session, accessControl, schema, table, retention);
         }
         catch (TrinoException e) {
             throw e;
@@ -129,6 +132,7 @@ public void vacuum(
 
     private void doVacuum(
             ConnectorSession session,
+            ConnectorAccessControl accessControl,
             String schema,
             String table,
             String retention)
@@ -159,6 +163,9 @@ private void doVacuum(
         DeltaLakeTableHandle handle = metadata.getTableHandle(session, tableName);
         checkProcedureArgument(handle != null, "Table '%s' does not exist", tableName);
 
+        accessControl.checkCanInsertIntoTable(null, tableName);
+        accessControl.checkCanDeleteFromTable(null, tableName);
+
         TableSnapshot tableSnapshot = transactionLogAccess.loadSnapshot(tableName, new Path(handle.getLocation()), session);
         Path tableLocation = tableSnapshot.getTableLocation();
         Path transactionLogDir = getTransactionLogDir(tableLocation);

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/BaseDeltaLakeConnectorSmokeTest.java

@@ -50,6 +50,9 @@
 import static io.trino.SystemSessionProperties.JOIN_DISTRIBUTION_TYPE;
 import static io.trino.plugin.deltalake.DeltaLakeQueryRunner.DELTA_CATALOG;
 import static io.trino.plugin.deltalake.transactionlog.TransactionLogUtil.TRANSACTION_LOG_DIRECTORY;
+import static io.trino.testing.TestingAccessControlManager.TestingPrivilegeType.DELETE_TABLE;
+import static io.trino.testing.TestingAccessControlManager.TestingPrivilegeType.INSERT_TABLE;
+import static io.trino.testing.TestingAccessControlManager.privilege;
 import static io.trino.testing.TestingSession.testSessionBuilder;
 import static io.trino.testing.assertions.Assert.assertEquals;
 import static io.trino.testing.assertions.Assert.assertEventually;
@@ -1257,6 +1260,25 @@ public void testVacuumParameterValidation()
         assertUpdate("DROP TABLE " + tableName);
     }
 
+    @Test
+    public void testVacuumAccessControl()
+    {
+        String tableName = "test_deny_vacuum_" + randomTableSuffix();
+        assertUpdate("CREATE TABLE " + tableName + " WITH (location = '" + getLocationForTable(bucketName, tableName) + "') " +
+                "AS SELECT * FROM orders", "SELECT count(*) FROM orders");
+
+        assertAccessDenied(
+                "CALL system.vacuum(schema_name => CURRENT_SCHEMA, table_name => '" + tableName + "', retention => '30d')",
+                "Cannot insert into table .*",
+                privilege(tableName, INSERT_TABLE));
+        assertAccessDenied(
+                "CALL system.vacuum(schema_name => CURRENT_SCHEMA, table_name => '" + tableName + "', retention => '30d')",
+                "Cannot delete from table .*",
+                privilege(tableName, DELETE_TABLE));
+
+        assertUpdate("DROP TABLE " + tableName);
+    }
+
     @Test
     public void testOptimize()
     {

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeAnalyze.java

@@ -32,6 +32,8 @@
 import static com.google.common.collect.MoreCollectors.onlyElement;
 import static io.trino.plugin.deltalake.DeltaLakeDockerizedMinioDataLake.createDockerizedMinioDataLakeForDeltaLake;
 import static io.trino.plugin.deltalake.DeltaLakeQueryRunner.DELTA_CATALOG;
+import static io.trino.testing.TestingAccessControlManager.TestingPrivilegeType.INSERT_TABLE;
+import static io.trino.testing.TestingAccessControlManager.privilege;
 import static io.trino.testing.sql.TestTable.randomTableSuffix;
 import static java.lang.String.format;
 import static java.time.ZoneOffset.UTC;
@@ -481,6 +483,22 @@ public void testDropMissingStats()
         }
     }
 
+    @Test
+    public void testDropStatsAccessControl()
+    {
+        String path = "test_deny_drop_stats_" + randomTableSuffix();
+
+        try (TestTable table = new TestTable(
+                getQueryRunner()::execute,
+                "test_deny_drop_stats",
+                format("WITH (location = '%s') AS SELECT * FROM tpch.sf1.nation", getLocationForTable(path)))) {
+            assertAccessDenied(
+                    format("CALL %s.system.drop_extended_stats('%s', '%s')", DELTA_CATALOG, SCHEMA, table.getName()),
+                    "Cannot insert into table .*",
+                    privilege(table.getName(), INSERT_TABLE));
+        }
+    }
+
     private void runAnalyzeVerifySplitCount(String tableName, long expectedSplitCount)
     {
         ResultWithQueryId<MaterializedResult> analyzeResult = getDistributedQueryRunner().executeWithQueryId(getSession(), "ANALYZE " + tableName);