DeltaLake support optimzie output metrics in result

Author: Ray Cheng

docs/src/main/sphinx/connector/delta-lake.md

@@ -714,6 +714,13 @@ EXECUTE <alter-table-execute>`.
 
 ```{include} optimize.fragment
 ```
+```text
+        metric_name         | metric_value
+----------------------------+--------------
+ rewritten_data_files_count |            2
+ removed_delete_files_count |            0
+ added_data_files_count     |            1
+```
 
 Use a `WHERE` clause with [metadata columns](delta-lake-special-columns) to filter
 which files are optimized.

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

@@ -2943,13 +2943,12 @@ public Map<String, Long> finishTableExecute(ConnectorSession session, ConnectorT
         DeltaLakeTableExecuteHandle executeHandle = (DeltaLakeTableExecuteHandle) tableExecuteHandle;
         switch (executeHandle.procedureId()) {
             case OPTIMIZE:
-                finishOptimize(session, executeHandle, fragments, splitSourceInfo);
-                return ImmutableMap.of();
+                return finishOptimize(session, executeHandle, fragments, splitSourceInfo);
         }
         throw new IllegalArgumentException("Unknown procedure '" + executeHandle.procedureId() + "'");
     }
 
-    private void finishOptimize(ConnectorSession session, DeltaLakeTableExecuteHandle executeHandle, Collection<Slice> fragments, List<Object> splitSourceInfo)
+    private Map<String, Long> finishOptimize(ConnectorSession session, DeltaLakeTableExecuteHandle executeHandle, Collection<Slice> fragments, List<Object> splitSourceInfo)
     {
         DeltaTableOptimizeHandle optimizeHandle = (DeltaTableOptimizeHandle) executeHandle.procedureHandle();
         String tableLocation = executeHandle.tableLocation();
@@ -3002,6 +3001,20 @@ private void finishOptimize(ConnectorSession session, DeltaLakeTableExecuteHandl
             }
             throw new TrinoException(DELTA_LAKE_BAD_WRITE, "Failed to write Delta Lake transaction log entry", e);
         }
+        // No files should be removed (this is VACUUM's job)
+        return new OptimizeResult(scannedDataFiles.size(), 0, dataFileInfos.size()).toMap();
+    }
+
+    private record OptimizeResult(long rewrittenDataFiles, long removedDeleteFiles, long addedDataFiles)
+    {
+        Map<String, Long> toMap()
+        {
+            return ImmutableMap.<String, Long>builder()
+                    .put("rewritten_data_files_count", rewrittenDataFiles)
+                    .put("removed_delete_files_count", removedDeleteFiles)
+                    .put("added_data_files_count", addedDataFiles)
+                    .buildOrThrow();
+        }
     }
 
     private long commitOptimizeOperation(

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

@@ -392,10 +392,10 @@ public void testOptimizeRewritesTable()
 
             for (int i = 0; i < 3; i++) {
                 Set<String> initialFiles = getActiveFiles(tableName);
-                computeActual("ALTER TABLE " + tableName + " EXECUTE OPTIMIZE");
+                computeActual(withSingleWriterPerTask(getSession()), "ALTER TABLE " + tableName + " EXECUTE OPTIMIZE");
                 Set<String> filesAfterOptimize = getActiveFiles(tableName);
                 assertThat(filesAfterOptimize)
-                        .hasSizeBetween(1, workerCount)
+                        .hasSize(1)
                         .containsExactlyElementsOf(initialFiles);
             }
 
@@ -1949,18 +1949,25 @@ public void testOptimize()
                     // Verify we have sufficiently many test rows with respect to worker count.
                     .hasSizeGreaterThan(workerCount);
 
-            computeActual("ALTER TABLE " + tableName + " EXECUTE OPTIMIZE");
+            computeActual(withSingleWriterPerTask(getSession()),"ALTER TABLE " + tableName + " EXECUTE OPTIMIZE");
             assertThat(query("SELECT sum(key), listagg(value, ' ') WITHIN GROUP (ORDER BY key) FROM " + tableName))
                     .matches("VALUES (BIGINT '65', VARCHAR 'eleven zwölf trzynaście quatorze пʼятнадцять')");
             Set<String> updatedFiles = getActiveFiles(tableName);
             assertThat(updatedFiles)
-                    .hasSizeBetween(1, workerCount)
+                    .hasSize(1)
                     .doesNotContainAnyElementsOf(initialFiles);
             // No files should be removed (this is VACUUM's job)
             assertThat(getAllDataFilesFromTableDirectory(tableName)).isEqualTo(union(initialFiles, updatedFiles));
 
             // optimize with low retention threshold, nothing should change
-            computeActual("ALTER TABLE " + tableName + " EXECUTE OPTIMIZE (file_size_threshold => '33B')");
+            MaterializedResult lowThresholdResult = computeActual(withSingleWriterPerTask(getSession()), "ALTER TABLE " + tableName + " EXECUTE OPTIMIZE (file_size_threshold => '33B')");
+            Map<String, Long> lowThresholdMetrics = lowThresholdResult.getMaterializedRows().stream()
+                    .collect(ImmutableMap.toImmutableMap(
+                            row -> (String) row.getField(0),
+                            row -> (Long) row.getField(1)));
+            assertThat(lowThresholdMetrics.get("rewritten_data_files_count")).isEqualTo(0L);
+            assertThat(lowThresholdMetrics.get("added_data_files_count")).isEqualTo(0L);
+            assertThat(lowThresholdMetrics.get("removed_delete_files_count")).isEqualTo(0L);
             assertThat(query("SELECT sum(key), listagg(value, ' ') WITHIN GROUP (ORDER BY key) FROM " + tableName))
                     .matches("VALUES (BIGINT '65', VARCHAR 'eleven zwölf trzynaście quatorze пʼятнадцять')");
             assertThat(getActiveFiles(tableName)).isEqualTo(updatedFiles);
@@ -2005,14 +2012,22 @@ public void testOptimizeWithPartitionedTable()
             Set<String> initialFiles = getActiveFiles(tableName);
             assertThat(initialFiles).hasSize(9);
 
-            computeActual("ALTER TABLE " + tableName + " EXECUTE OPTIMIZE");
+            MaterializedResult optimizeResult = computeActual(withSingleWriterPerTask(getSession()), "ALTER TABLE " + tableName + " EXECUTE OPTIMIZE");
+            // Verify optimize metrics
+            Map<String, Long> metrics = optimizeResult.getMaterializedRows().stream()
+                    .collect(ImmutableMap.toImmutableMap(
+                            row -> (String) row.getField(0),
+                            row -> (Long) row.getField(1)));
+            assertThat(metrics.get("rewritten_data_files_count")).isEqualTo(3L);
+            assertThat(metrics.get("added_data_files_count")).isEqualTo(1L);
+            assertThat(metrics.get("removed_delete_files_count")).isEqualTo(0L);
 
             assertThat(query("SELECT sum(key), listagg(value, ' ') WITHIN GROUP (ORDER BY value) FROM " + tableName))
                     .matches("VALUES (BIGINT '508', VARCHAR 'ONE Three four one one one tHrEe three two')");
 
             Set<String> updatedFiles = getActiveFiles(tableName);
             assertThat(updatedFiles)
-                    .hasSizeBetween(7, initialFiles.size());
+                    .hasSize(7);
             assertThat(getAllDataFilesFromTableDirectory(tableName)).isEqualTo(union(initialFiles, updatedFiles));
         }
         finally {
@@ -2046,7 +2061,15 @@ public void testOptimizeWithEnforcedRepartitioning()
             Set<String> initialFiles = getActiveFiles(tableName, currentSession);
             assertThat(initialFiles).hasSize(10);
 
-            computeActual(currentSession, "ALTER TABLE " + tableName + " EXECUTE OPTIMIZE");
+            MaterializedResult optimizeResult = computeActual(withSingleWriterPerTask(getSession()), "ALTER TABLE " + tableName + " EXECUTE OPTIMIZE");
+
+            Map<String, Long> metrics = optimizeResult.getMaterializedRows().stream()
+                    .collect(ImmutableMap.toImmutableMap(
+                            row -> (String) row.getField(0),
+                            row -> (Long) row.getField(1)));
+            assertThat(metrics.get("rewritten_data_files_count")).isEqualTo(9L);
+            assertThat(metrics.get("added_data_files_count")).isEqualTo(2L);
+            assertThat(metrics.get("removed_delete_files_count")).isEqualTo(0L);
 
             assertThat(query(currentSession, "SELECT sum(key), listagg(value, ' ') WITHIN GROUP (ORDER BY value) FROM " + tableName))
                     .matches("VALUES (BIGINT '55', VARCHAR 'one one one one one one one three two two')");
@@ -2144,7 +2167,7 @@ public void testOptimizeUsingForcedPartitioning()
         Set<String> initialFiles = getActiveFiles(tableName);
         assertThat(initialFiles).hasSize(10);
 
-        computeActual("ALTER TABLE " + tableName + " EXECUTE OPTIMIZE");
+        computeActual(withSingleWriterPerTask(getSession()),"ALTER TABLE " + tableName + " EXECUTE OPTIMIZE");
 
         assertThat(query("SELECT " +
                 "sum(value1), " +
@@ -2878,6 +2901,13 @@ private Set<String> getActiveFiles(String tableName, Session session)
                 .collect(toImmutableSet());
     }
 
+    private Session withSingleWriterPerTask(Session session)
+    {
+        return Session.builder(session)
+                .setSystemProperty("task_min_writer_count", "1")
+                .build();
+    }
+
     private Set<String> getAllDataFilesFromTableDirectory(String tableName)
     {
         return getTableFiles(tableName).stream()