[#10280] Support Iceberg snapshot maintenance procedures via Gravitino Trino

trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/lakehouse-iceberg/00013_snapshot_maintenance.sql

@@ -0,0 +1,63 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements.  See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership.  The ASF licenses this file
+-- to you 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.
+
+-- Test Iceberg snapshot maintenance procedures via Gravitino connector
+
+CREATE SCHEMA IF NOT EXISTS gt_snapshot_test;
+
+CREATE TABLE gt_snapshot_test.maintenance_table (
+    id int,
+    name varchar
+);
+
+-- Insert data to create snapshots
+INSERT INTO gt_snapshot_test.maintenance_table VALUES (1, 'alice');
+
+INSERT INTO gt_snapshot_test.maintenance_table VALUES (2, 'bob');
+
+INSERT INTO gt_snapshot_test.maintenance_table VALUES (3, 'charlie');
+
+-- Verify we have multiple snapshots
+SELECT count(*) >= 3 FROM "gt_snapshot_test"."maintenance_table$snapshots";
+
+-- Test expire_snapshots procedure (expire snapshots older than a far future timestamp keeps all)
+ALTER TABLE gt_snapshot_test.maintenance_table EXECUTE expire_snapshots(retention_threshold => '0s');
+
+-- Verify table data is still intact after expire_snapshots
+SELECT count(*) FROM gt_snapshot_test.maintenance_table;
+
+-- Test remove_orphan_files procedure
+ALTER TABLE gt_snapshot_test.maintenance_table EXECUTE remove_orphan_files(retention_threshold => '0s');
+
+-- Verify table data is still intact after remove_orphan_files
+SELECT count(*) FROM gt_snapshot_test.maintenance_table;
+
+-- Insert more data to create small files for optimize (rewrite_data_files)
+INSERT INTO gt_snapshot_test.maintenance_table VALUES (4, 'dave');
+
+INSERT INTO gt_snapshot_test.maintenance_table VALUES (5, 'eve');
+
+-- Test optimize (rewrite_data_files) table procedure
+ALTER TABLE gt_snapshot_test.maintenance_table EXECUTE optimize;
+
+-- Verify all data is still accessible after optimize
+SELECT count(*) FROM gt_snapshot_test.maintenance_table;
+
+-- Cleanup
+DROP TABLE gt_snapshot_test.maintenance_table;
+
+DROP SCHEMA gt_snapshot_test;

trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/lakehouse-iceberg/00013_snapshot_maintenance.txt

@@ -0,0 +1,31 @@
+CREATE SCHEMA
+
+CREATE TABLE
+
+INSERT: 1 row
+
+INSERT: 1 row
+
+INSERT: 1 row
+
+"true"
+
+EXECUTE expire_snapshots
+
+"3"
+
+EXECUTE remove_orphan_files
+
+"3"
+
+INSERT: 1 row
+
+INSERT: 1 row
+
+EXECUTE optimize
+
+"5"
+
+DROP TABLE
+
+DROP SCHEMA

trino-connector/trino-connector-435-439/src/main/java/org/apache/gravitino/trino/connector/GravitinoMetadata435.java

@@ -24,11 +24,13 @@
 import io.trino.spi.connector.ConnectorMergeTableHandle;
 import io.trino.spi.connector.ConnectorOutputMetadata;
 import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.ConnectorTableExecuteHandle;
 import io.trino.spi.connector.ConnectorTableHandle;
 import io.trino.spi.connector.RetryMode;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.statistics.ComputedStatistics;
 import java.util.Collection;
+import java.util.Map;
 import java.util.Optional;
 import org.apache.gravitino.trino.connector.catalog.CatalogConnectorMetadata;
 import org.apache.gravitino.trino.connector.catalog.CatalogConnectorMetadataAdapter;
@@ -50,6 +52,23 @@ public void addColumn(
     catalogConnectorMetadata.addColumn(getTableName(tableHandle), gravitinoColumn);
   }
 
+  @Override
+  public Optional<ConnectorTableExecuteHandle> getTableHandleForExecute(
+      ConnectorSession session,
+      ConnectorTableHandle tableHandle,
+      String procedureName,
+      Map<String, Object> executeProperties,
+      RetryMode retryMode) {
+    return internalMetadata.getTableHandleForExecute(
+        session, GravitinoHandle.unWrap(tableHandle), procedureName, executeProperties, retryMode);
+  }
+
+  @Override
+  public void executeTableExecute(
+      ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle) {
+    internalMetadata.executeTableExecute(session, tableExecuteHandle);
+  }
+
   @Override
   public Optional<ConnectorOutputMetadata> finishInsert(
       ConnectorSession session,

trino-connector/trino-connector-440-445/src/main/java/org/apache/gravitino/trino/connector/GravitinoMetadata440.java

@@ -24,12 +24,14 @@
 import io.trino.spi.connector.ConnectorMergeTableHandle;
 import io.trino.spi.connector.ConnectorOutputMetadata;
 import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.ConnectorTableExecuteHandle;
 import io.trino.spi.connector.ConnectorTableHandle;
 import io.trino.spi.connector.RetryMode;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.statistics.ComputedStatistics;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import org.apache.gravitino.trino.connector.catalog.CatalogConnectorMetadata;
@@ -52,6 +54,23 @@ public void addColumn(
     catalogConnectorMetadata.addColumn(getTableName(tableHandle), gravitinoColumn);
   }
 
+  @Override
+  public Optional<ConnectorTableExecuteHandle> getTableHandleForExecute(
+      ConnectorSession session,
+      ConnectorTableHandle tableHandle,
+      String procedureName,
+      Map<String, Object> executeProperties,
+      RetryMode retryMode) {
+    return internalMetadata.getTableHandleForExecute(
+        session, GravitinoHandle.unWrap(tableHandle), procedureName, executeProperties, retryMode);
+  }
+
+  @Override
+  public void executeTableExecute(
+      ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle) {
+    internalMetadata.executeTableExecute(session, tableExecuteHandle);
+  }
+
   @Override
   public Optional<ConnectorOutputMetadata> finishInsert(
       ConnectorSession session,

trino-connector/trino-connector-446-451/src/main/java/org/apache/gravitino/trino/connector/GravitinoMetadata446.java

@@ -24,12 +24,14 @@
 import io.trino.spi.connector.ConnectorMergeTableHandle;
 import io.trino.spi.connector.ConnectorOutputMetadata;
 import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.ConnectorTableExecuteHandle;
 import io.trino.spi.connector.ConnectorTableHandle;
 import io.trino.spi.connector.RetryMode;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.statistics.ComputedStatistics;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import org.apache.gravitino.trino.connector.catalog.CatalogConnectorMetadata;
@@ -52,6 +54,23 @@ public void addColumn(
     catalogConnectorMetadata.addColumn(getTableName(tableHandle), gravitinoColumn);
   }
 
+  @Override
+  public Optional<ConnectorTableExecuteHandle> getTableHandleForExecute(
+      ConnectorSession session,
+      ConnectorTableHandle tableHandle,
+      String procedureName,
+      Map<String, Object> executeProperties,
+      RetryMode retryMode) {
+    return internalMetadata.getTableHandleForExecute(
+        session, GravitinoHandle.unWrap(tableHandle), procedureName, executeProperties, retryMode);
+  }
+
+  @Override
+  public void executeTableExecute(
+      ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle) {
+    internalMetadata.executeTableExecute(session, tableExecuteHandle);
+  }
+
   @Override
   public Optional<ConnectorOutputMetadata> finishInsert(
       ConnectorSession session,

trino-connector/trino-connector-452-468/src/main/java/org/apache/gravitino/trino/connector/GravitinoMetadata452.java

@@ -24,12 +24,14 @@
 import io.trino.spi.connector.ConnectorMergeTableHandle;
 import io.trino.spi.connector.ConnectorOutputMetadata;
 import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.ConnectorTableExecuteHandle;
 import io.trino.spi.connector.ConnectorTableHandle;
 import io.trino.spi.connector.RetryMode;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.statistics.ComputedStatistics;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import org.apache.gravitino.trino.connector.catalog.CatalogConnectorMetadata;
@@ -67,6 +69,23 @@ public Optional<ConnectorOutputMetadata> finishInsert(
         computedStatistics);
   }
 
+  @Override
+  public Optional<ConnectorTableExecuteHandle> getTableHandleForExecute(
+      ConnectorSession session,
+      ConnectorTableHandle tableHandle,
+      String procedureName,
+      Map<String, Object> executeProperties,
+      RetryMode retryMode) {
+    return internalMetadata.getTableHandleForExecute(
+        session, GravitinoHandle.unWrap(tableHandle), procedureName, executeProperties, retryMode);
+  }
+
+  @Override
+  public void executeTableExecute(
+      ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle) {
+    internalMetadata.executeTableExecute(session, tableExecuteHandle);
+  }
+
   @SuppressWarnings("deprecation")
   @Override
   public ConnectorMergeTableHandle beginMerge(

trino-connector/trino-connector-469-472/src/main/java/org/apache/gravitino/trino/connector/GravitinoMetadata469.java

@@ -22,10 +22,12 @@
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
 import io.trino.spi.connector.ColumnPosition;
+import io.trino.spi.connector.ConnectorAccessControl;
 import io.trino.spi.connector.ConnectorInsertTableHandle;
 import io.trino.spi.connector.ConnectorMergeTableHandle;
 import io.trino.spi.connector.ConnectorOutputMetadata;
 import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.ConnectorTableExecuteHandle;
 import io.trino.spi.connector.ConnectorTableHandle;
 import io.trino.spi.connector.RetryMode;
 import io.trino.spi.connector.SchemaTableName;
@@ -73,6 +75,29 @@ public Optional<ConnectorOutputMetadata> finishInsert(
         computedStatistics);
   }
 
+  @Override
+  public Optional<ConnectorTableExecuteHandle> getTableHandleForExecute(
+      ConnectorSession session,
+      ConnectorAccessControl accessControl,
+      ConnectorTableHandle tableHandle,
+      String procedureName,
+      Map<String, Object> executeProperties,
+      RetryMode retryMode) {
+    return internalMetadata.getTableHandleForExecute(
+        session,
+        accessControl,
+        GravitinoHandle.unWrap(tableHandle),
+        procedureName,
+        executeProperties,
+        retryMode);
+  }
+
+  @Override
+  public void executeTableExecute(
+      ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle) {
+    internalMetadata.executeTableExecute(session, tableExecuteHandle);
+  }
+
   @Override
   public ConnectorMergeTableHandle beginMerge(
       ConnectorSession session,

trino-connector/trino-connector-473-478/src/main/java/org/apache/gravitino/trino/connector/GravitinoMetadata478.java

@@ -22,10 +22,12 @@
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
 import io.trino.spi.connector.ColumnPosition;
+import io.trino.spi.connector.ConnectorAccessControl;
 import io.trino.spi.connector.ConnectorInsertTableHandle;
 import io.trino.spi.connector.ConnectorMergeTableHandle;
 import io.trino.spi.connector.ConnectorOutputMetadata;
 import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.ConnectorTableExecuteHandle;
 import io.trino.spi.connector.ConnectorTableHandle;
 import io.trino.spi.connector.RetryMode;
 import io.trino.spi.connector.SchemaTableName;
@@ -66,6 +68,29 @@ public void addColumn(
     catalogConnectorMetadata.addColumn(getTableName(tableHandle), gravitinoColumn, columnPosition);
   }
 
+  @Override
+  public Optional<ConnectorTableExecuteHandle> getTableHandleForExecute(
+      ConnectorSession session,
+      ConnectorAccessControl accessControl,
+      ConnectorTableHandle tableHandle,
+      String procedureName,
+      Map<String, Object> executeProperties,
+      RetryMode retryMode) {
+    return internalMetadata.getTableHandleForExecute(
+        session,
+        accessControl,
+        GravitinoHandle.unWrap(tableHandle),
+        procedureName,
+        executeProperties,
+        retryMode);
+  }
+
+  @Override
+  public Map<String, Long> executeTableExecute(
+      ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle) {
+    return internalMetadata.executeTableExecute(session, tableExecuteHandle);
+  }
+
   @Override
   public Optional<ConnectorOutputMetadata> finishInsert(
       ConnectorSession session,

trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnector.java

@@ -33,6 +33,8 @@
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.ConnectorSplitManager;
 import io.trino.spi.connector.ConnectorTransactionHandle;
+import io.trino.spi.connector.TableProcedureMetadata;
+import io.trino.spi.procedure.Procedure;
 import io.trino.spi.session.PropertyMetadata;
 import io.trino.spi.transaction.IsolationLevel;
 import java.util.List;
@@ -101,6 +103,18 @@ protected GravitinoMetadata createGravitinoMetadata(
     throw new TrinoException(NOT_SUPPORTED, "Should be overridden in subclass");
   }
 
+  @Override
+  public Set<Procedure> getProcedures() {
+    Connector internalConnector = catalogConnectorContext.getInternalConnector();
+    return internalConnector.getProcedures();
+  }
+
+  @Override
+  public Set<TableProcedureMetadata> getTableProcedures() {
+    Connector internalConnector = catalogConnectorContext.getInternalConnector();
+    return internalConnector.getTableProcedures();
+  }
+
   @Override
   public List<PropertyMetadata<?>> getTableProperties() {
     return catalogConnectorContext.getTableProperties();