Use icebergScanExecutor for reading manifests in IcebergMetadata

Reduces bottleneck on the default iceberg worker pool

Author: raunaqmorarka

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java

@@ -2029,6 +2029,7 @@ private void finishOptimize(ConnectorSession session, IcebergTableExecuteHandle
         // Set dataSequenceNumber to avoid contention between OPTIMIZE and concurrent writing of equality deletes
         rewriteFiles.dataSequenceNumber(snapshot.sequenceNumber());
         rewriteFiles.validateFromSnapshot(snapshot.snapshotId());
+        rewriteFiles.scanManifestsWith(icebergScanExecutor);
         commitUpdateAndTransaction(rewriteFiles, session, transaction, "optimize");
 
         // TODO (https://github.com/trinodb/trino/issues/15439) this may not exactly be the snapshot we committed, if there is another writer
@@ -2129,11 +2130,14 @@ private void executeOptimizeManifests(ConnectorSession session, IcebergTableExec
 
         beginTransaction(icebergTable);
         RewriteManifests rewriteManifests = transaction.rewriteManifests();
-        rewriteManifests.clusterBy(file -> {
-            // Use the first partition field as the clustering key
-            StructLike partition = file.partition();
-            return partition.size() > 1 ? Optional.ofNullable(partition.get(0, Object.class)) : partition;
-        }).commit();
+        rewriteManifests
+                .clusterBy(file -> {
+                    // Use the first partition field as the clustering key
+                    StructLike partition = file.partition();
+                    return partition.size() > 1 ? Optional.ofNullable(partition.get(0, Object.class)) : partition;
+                })
+                .scanManifestsWith(icebergScanExecutor)
+                .commit();
         commitTransaction(transaction, "optimize manifests");
         transaction = null;
     }
@@ -2198,6 +2202,7 @@ private void executeExpireSnapshots(ConnectorSession session, IcebergTableExecut
             table.expireSnapshots()
                     .expireOlderThan(expireTimestampMillis)
                     .deleteWith(deleteFunction)
+                    .planWith(icebergScanExecutor)
                     .commit();
 
             fileSystem.deleteFiles(pathsToDelete);
@@ -3154,6 +3159,7 @@ private void finishWrite(ConnectorSession session, IcebergTableHandle table, Col
         // Ensure a row that is updated by this commit was not deleted by a separate commit
         rowDelta.validateDeletedFiles();
         rowDelta.validateNoConflictingDeleteFiles();
+        rowDelta.scanManifestsWith(icebergScanExecutor);
 
         ImmutableSet.Builder<String> writtenFiles = ImmutableSet.builder();
         ImmutableSet.Builder<String> referencedDataFiles = ImmutableSet.builder();