Output metrics about remove_orphan_files execution

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

@@ -121,7 +121,7 @@ Optional<TableExecuteHandle> getTableHandleForExecute(
 
     void finishTableExecute(Session session, TableExecuteHandle handle, Collection<Slice> fragments, List<Object> tableExecuteState);
 
-    void executeTableExecute(Session session, TableExecuteHandle handle);
+    Map<String, Long> executeTableExecute(Session session, TableExecuteHandle handle);
 
     TableProperties getTableProperties(Session session, TableHandle handle);
 

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

@@ -362,11 +362,11 @@ public void finishTableExecute(Session session, TableExecuteHandle tableExecuteH
     }
 
     @Override
-    public void executeTableExecute(Session session, TableExecuteHandle tableExecuteHandle)
+    public Map<String, Long> executeTableExecute(Session session, TableExecuteHandle tableExecuteHandle)
     {
         CatalogHandle catalogHandle = tableExecuteHandle.catalogHandle();
         ConnectorMetadata metadata = getMetadata(session, catalogHandle);
-        metadata.executeTableExecute(session.toConnectorSession(catalogHandle), tableExecuteHandle.connectorHandle());
+        return metadata.executeTableExecute(session.toConnectorSession(catalogHandle), tableExecuteHandle.connectorHandle());
     }
 
     @Override

core/trino-main/src/main/java/io/trino/operator/SimpleTableExecuteOperator.java

@@ -13,20 +13,26 @@
  */
 package io.trino.operator;
 
+import com.google.common.collect.ImmutableList;
+import io.airlift.slice.Slices;
 import io.trino.Session;
 import io.trino.metadata.Metadata;
 import io.trino.metadata.TableExecuteHandle;
 import io.trino.spi.Page;
+import io.trino.spi.PageBuilder;
+import io.trino.spi.block.BlockBuilder;
+import io.trino.spi.type.Type;
 import io.trino.sql.planner.plan.PlanNodeId;
 
+import java.util.List;
+import java.util.Map;
+
 import static com.google.common.base.Preconditions.checkState;
 import static java.util.Objects.requireNonNull;
 
 public class SimpleTableExecuteOperator
         implements Operator
 {
-    private static final Page PAGE = new Page(0);
-
     public static class SimpleTableExecuteOperatorOperatorFactory
             implements OperatorFactory
     {
@@ -35,20 +41,23 @@ public static class SimpleTableExecuteOperatorOperatorFactory
         private final Metadata metadata;
         private final Session session;
         private final TableExecuteHandle executeHandle;
+        private final List<Type> types;
         private boolean closed;
 
         public SimpleTableExecuteOperatorOperatorFactory(
                 int operatorId,
                 PlanNodeId planNodeId,
                 Metadata metadata,
                 Session session,
-                TableExecuteHandle executeHandle)
+                TableExecuteHandle executeHandle,
+                List<Type> types)
         {
             this.operatorId = operatorId;
             this.planNodeId = requireNonNull(planNodeId, "planNodeId is null");
             this.metadata = requireNonNull(metadata, "planNodeId is null");
             this.session = requireNonNull(session, "planNodeId is null");
             this.executeHandle = requireNonNull(executeHandle, "executeHandle is null");
+            this.types = ImmutableList.copyOf(requireNonNull(types, "types is null"));
         }
 
         @Override
@@ -60,7 +69,8 @@ public Operator createOperator(DriverContext driverContext)
                     context,
                     metadata,
                     session,
-                    executeHandle);
+                    executeHandle,
+                    types);
         }
 
         @Override
@@ -77,27 +87,31 @@ public OperatorFactory duplicate()
                     planNodeId,
                     metadata,
                     session,
-                    executeHandle);
+                    executeHandle,
+                    types);
         }
     }
 
     private final OperatorContext operatorContext;
     private final Metadata metadata;
     private final Session session;
     private final TableExecuteHandle executeHandle;
+    private final List<Type> types;
 
     private boolean finished;
 
     public SimpleTableExecuteOperator(
             OperatorContext operatorContext,
             Metadata metadata,
             Session session,
-            TableExecuteHandle executeHandle)
+            TableExecuteHandle executeHandle,
+            List<Type> types)
     {
         this.operatorContext = requireNonNull(operatorContext, "operatorContext is null");
         this.metadata = requireNonNull(metadata, "metadata is null");
         this.session = requireNonNull(session, "session is null");
         this.executeHandle = requireNonNull(executeHandle, "executeHandle is null");
+        this.types = ImmutableList.copyOf(requireNonNull(types, "types is null"));
     }
 
     @Override
@@ -125,9 +139,17 @@ public Page getOutput()
             return null;
         }
 
-        metadata.executeTableExecute(session, executeHandle);
+        Map<String, Long> metrics = metadata.executeTableExecute(session, executeHandle);
         finished = true;
-        return PAGE;
+        PageBuilder pageBuilder = new PageBuilder(types);
+        BlockBuilder metricNameBuilder = pageBuilder.getBlockBuilder(0);
+        BlockBuilder metricValueBuilder = pageBuilder.getBlockBuilder(1);
+        for (Map.Entry<String, Long> entry : metrics.entrySet()) {
+            types.get(0).writeSlice(metricNameBuilder, Slices.utf8Slice(entry.getKey()));
+            types.get(1).writeLong(metricValueBuilder, entry.getValue());
+            pageBuilder.declarePosition();
+        }
+        return pageBuilder.build();
     }
 
     @Override

core/trino-main/src/main/java/io/trino/sql/analyzer/StatementAnalyzer.java

@@ -1323,6 +1323,9 @@ protected Scope visitTableExecute(TableExecute node, Optional<Scope> scope)
             analysis.setUpdateType("ALTER TABLE EXECUTE");
             analysis.setUpdateTarget(executeHandle.catalogHandle().getVersion(), tableName, Optional.of(table), Optional.empty());
 
+            if (!procedureMetadata.getExecutionMode().isReadsData()) {
+                return createAndAssignScope(node, scope, Field.newUnqualified("metric_name", VARCHAR), Field.newUnqualified("metric_value", BIGINT));
+            }
             return createAndAssignScope(node, scope, Field.newUnqualified("rows", BIGINT));
         }
 

core/trino-main/src/main/java/io/trino/sql/planner/LocalExecutionPlanner.java

@@ -3476,7 +3476,8 @@ public PhysicalOperation visitSimpleTableExecuteNode(SimpleTableExecuteNode node
                             node.getId(),
                             metadata,
                             session,
-                            node.getExecuteHandle());
+                            node.getExecuteHandle(),
+                            getSymbolTypes(node.getOutputSymbols()));
 
             return new PhysicalOperation(operatorFactory, makeLayout(node));
         }

core/trino-main/src/main/java/io/trino/sql/planner/LogicalPlanner.java

@@ -966,7 +966,9 @@ private RelationPlan createTableExecutePlan(Analysis analysis, TableExecute stat
         if (!analysis.isTableExecuteReadsData()) {
             SimpleTableExecuteNode node = new SimpleTableExecuteNode(
                     idAllocator.getNextId(),
-                    symbolAllocator.newSymbol("rows", BIGINT),
+                    ImmutableList.of(
+                            symbolAllocator.newSymbol("metricName", VARCHAR),
+                            symbolAllocator.newSymbol("metricValue", BIGINT)),
                     executeHandle);
             return new RelationPlan(node, analysis.getRootScope(), node.getOutputSymbols(), Optional.empty());
         }

core/trino-main/src/main/java/io/trino/sql/planner/optimizations/UnaliasSymbolReferences.java

@@ -687,12 +687,11 @@ public PlanAndMappings visitSimpleTableExecuteNode(SimpleTableExecuteNode node,
         {
             Map<Symbol, Symbol> mapping = new HashMap<>(context.getCorrelationMapping());
             SymbolMapper mapper = symbolMapper(mapping);
-            Symbol newOutput = mapper.map(node.getOutput());
 
             return new PlanAndMappings(
                     new SimpleTableExecuteNode(
                             node.getId(),
-                            newOutput,
+                            mapper.map(node.getOutputSymbols()),
                             node.getExecuteHandle()),
                     mapping);
         }

core/trino-main/src/main/java/io/trino/sql/planner/plan/SimpleTableExecuteNode.java

@@ -27,17 +27,17 @@
 public class SimpleTableExecuteNode
         extends PlanNode
 {
-    private final Symbol output;
+    private final List<Symbol> outputs;
     private final TableExecuteHandle executeHandle;
 
     @JsonCreator
     public SimpleTableExecuteNode(
             @JsonProperty("id") PlanNodeId id,
-            @JsonProperty("output") Symbol output,
+            @JsonProperty("outputs") List<Symbol> outputs,
             @JsonProperty("executeHandle") TableExecuteHandle executeHandle)
     {
         super(id);
-        this.output = requireNonNull(output, "output is null");
+        this.outputs = ImmutableList.copyOf(requireNonNull(outputs, "outputs is null"));
         this.executeHandle = requireNonNull(executeHandle, "executeHandle is null");
     }
 
@@ -48,16 +48,11 @@ public List<PlanNode> getSources()
         return ImmutableList.of();
     }
 
+    @JsonProperty("outputs")
     @Override
     public List<Symbol> getOutputSymbols()
     {
-        return ImmutableList.of(output);
-    }
-
-    @JsonProperty
-    public Symbol getOutput()
-    {
-        return output;
+        return outputs;
     }
 
     @Override

core/trino-main/src/main/java/io/trino/tracing/TracingConnectorMetadata.java

@@ -181,11 +181,11 @@ public void finishTableExecute(ConnectorSession session, ConnectorTableExecuteHa
     }
 
     @Override
-    public void executeTableExecute(ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle)
+    public Map<String, Long> executeTableExecute(ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle)
     {
         Span span = startSpan("executeTableExecute", tableExecuteHandle);
         try (var _ = scopedSpan(span)) {
-            delegate.executeTableExecute(session, tableExecuteHandle);
+            return delegate.executeTableExecute(session, tableExecuteHandle);
         }
     }
 

core/trino-main/src/main/java/io/trino/tracing/TracingMetadata.java

@@ -236,11 +236,11 @@ public void finishTableExecute(Session session, TableExecuteHandle handle, Colle
     }
 
     @Override
-    public void executeTableExecute(Session session, TableExecuteHandle handle)
+    public Map<String, Long> executeTableExecute(Session session, TableExecuteHandle handle)
     {
         Span span = startSpan("executeTableExecute", handle);
         try (var _ = scopedSpan(span)) {
-            delegate.executeTableExecute(session, handle);
+            return delegate.executeTableExecute(session, handle);
         }
     }
 

core/trino-main/src/test/java/io/trino/connector/MockConnector.java

@@ -882,7 +882,10 @@ public Optional<ConnectorTableExecuteHandle> getTableHandleForExecute(ConnectorS
         }
 
         @Override
-        public void executeTableExecute(ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle) {}
+        public Map<String, Long> executeTableExecute(ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle)
+        {
+            return ImmutableMap.of();
+        }
 
         @Override
         public void finishTableExecute(ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle, Collection<Slice> fragments, List<Object> tableExecuteState) {}

core/trino-main/src/test/java/io/trino/metadata/AbstractMockMetadata.java

@@ -170,7 +170,7 @@ public void finishTableExecute(Session session, TableExecuteHandle handle, Colle
     }
 
     @Override
-    public void executeTableExecute(Session session, TableExecuteHandle handle)
+    public Map<String, Long> executeTableExecute(Session session, TableExecuteHandle handle)
     {
         throw new UnsupportedOperationException();
     }

core/trino-spi/pom.xml

@@ -832,6 +832,11 @@
                                     <new>method void io.trino.spi.connector.Connector::shutdown()</new>
                                     <justification>Require connector to implement shutdown to prevent leaks</justification>
                                 </item>
+                                <item>
+                                    <code>java.method.returnTypeChanged</code>
+                                    <old>method void io.trino.spi.connector.ConnectorMetadata::executeTableExecute(io.trino.spi.connector.ConnectorSession, io.trino.spi.connector.ConnectorTableExecuteHandle)</old>
+                                    <new>method java.util.Map&lt;java.lang.String, java.lang.Long&gt; io.trino.spi.connector.ConnectorMetadata::executeTableExecute(io.trino.spi.connector.ConnectorSession, io.trino.spi.connector.ConnectorTableExecuteHandle)</new>
+                                </item>
                             </differences>
                         </revapi.differences>
                     </analysisConfiguration>

core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorMetadata.java

@@ -160,9 +160,10 @@ default void finishTableExecute(ConnectorSession session, ConnectorTableExecuteH
     }
 
     /**
-     * Execute a {@link TableProcedureExecutionMode#coordinatorOnly() coordinator-only} table procedure.
+     * Execute a {@link TableProcedureExecutionMode#coordinatorOnly() coordinator-only} table procedure
+     * and return procedure execution metrics that will be populated in the query output.
      */
-    default void executeTableExecute(ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle)
+    default Map<String, Long> executeTableExecute(ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle)
     {
         throw new TrinoException(GENERIC_INTERNAL_ERROR, "ConnectorMetadata executeTableExecute() is not implemented");
     }

docs/src/main/sphinx/connector/iceberg.md

@@ -907,12 +907,39 @@ time is recommended to keep size of a table's data directory under control.
 ALTER TABLE test_table EXECUTE remove_orphan_files(retention_threshold => '7d');
 ```
 
+```text
+        metric_name         | metric_value
+----------------------------+--------------
+ processed_manifests_count  |            2
+ active_files_count          |           98
+ scanned_files_count        |           97
+ deleted_files_count        |            0
+```
+
 The value for `retention_threshold` must be higher than or equal to
 `iceberg.remove-orphan-files.min-retention` in the catalog otherwise the
 procedure fails with a similar message: `Retention specified (1.00d) is shorter
 than the minimum retention configured in the system (7.00d)`. The default value
 for this property is `7d`.
 
+The output of the query has the following metrics:
+
+:::{list-table} Output
+:widths: 40, 60
+:header-rows: 1
+
+* - Property name
+  - Description
+* - `processed_manifests_count`
+  - The count of manifest files read by remove_orphan_files.
+* - `active_files_count`
+  - The count of files belonging to snapshots that have not been expired.
+* - `scanned_files_count`
+  - The count of files scanned from the file system.
+* - `deleted_files_count`
+  - The count of files deleted by remove_orphan_files.
+:::
+
 (drop-extended-stats)=
 ##### drop_extended_stats
 

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/classloader/ClassLoaderSafeConnectorMetadata.java

@@ -211,10 +211,10 @@ public Optional<ConnectorTableExecuteHandle> getTableHandleForExecute(ConnectorS
     }
 
     @Override
-    public void executeTableExecute(ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle)
+    public Map<String, Long> executeTableExecute(ConnectorSession session, ConnectorTableExecuteHandle tableExecuteHandle)
     {
         try (ThreadContextClassLoader _ = new ThreadContextClassLoader(classLoader)) {
-            delegate.executeTableExecute(session, tableExecuteHandle);
+            return delegate.executeTableExecute(session, tableExecuteHandle);
         }
     }