[Address Comment]Refactor Procedure and DistributedProcedure into abstract classes

And introduce a new class `LocalProcedure` to represent the former
coordinator-only procedures

Author: hantangwangd

presto-analyzer/src/test/java/com/facebook/presto/sql/analyzer/TestBuiltInQueryPreparer.java

@@ -18,7 +18,9 @@
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.WarningCollector;
 import com.facebook.presto.spi.analyzer.AnalyzerOptions;
+import com.facebook.presto.spi.procedure.LocalProcedure;
 import com.facebook.presto.spi.procedure.Procedure;
+import com.facebook.presto.spi.procedure.Procedure.Argument;
 import com.facebook.presto.spi.procedure.ProcedureRegistry;
 import com.facebook.presto.spi.procedure.TableDataRewriteDistributedProcedure;
 import com.facebook.presto.sql.analyzer.BuiltInQueryPreparer.BuiltInPreparedQuery;
@@ -61,12 +63,12 @@ public class TestBuiltInQueryPreparer
     public void setup()
     {
         procedureRegistry = new TestProcedureRegistry();
-        List<Procedure.Argument> arguments = new ArrayList<>();
-        arguments.add(new Procedure.Argument(SCHEMA, VARCHAR));
-        arguments.add(new Procedure.Argument(TABLE_NAME, VARCHAR));
+        List<Argument> arguments = new ArrayList<>();
+        arguments.add(new Argument(SCHEMA, VARCHAR));
+        arguments.add(new Argument(TABLE_NAME, VARCHAR));
 
         List<Procedure> procedures = new ArrayList<>();
-        procedures.add(new Procedure("system", "fun", arguments));
+        procedures.add(new LocalProcedure("system", "fun", arguments));
         procedures.add(new TableDataRewriteDistributedProcedure("system", "distributed_fun",
                 arguments,
                 (session, transactionContext, procedureHandle, fragments) -> null,

presto-hive-metastore/src/main/java/com/facebook/presto/hive/metastore/InvalidateMetastoreCacheProcedure.java

@@ -17,6 +17,7 @@
 import com.facebook.presto.spi.ConnectorSession;
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.classloader.ThreadContextClassLoader;
+import com.facebook.presto.spi.procedure.LocalProcedure;
 import com.facebook.presto.spi.procedure.Procedure;
 import com.facebook.presto.spi.procedure.Procedure.Argument;
 import com.google.common.collect.ImmutableList;
@@ -65,7 +66,7 @@ public InvalidateMetastoreCacheProcedure(ExtendedHiveMetastore extendedHiveMetas
     @Override
     public Procedure get()
     {
-        return new Procedure(
+        return new LocalProcedure(
                 "system",
                 "invalidate_metastore_cache",
                 ImmutableList.of(

presto-hive/src/main/java/com/facebook/presto/hive/CreateEmptyPartitionProcedure.java

@@ -22,6 +22,7 @@
 import com.facebook.presto.spi.ConnectorSession;
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.classloader.ThreadContextClassLoader;
+import com.facebook.presto.spi.procedure.LocalProcedure;
 import com.facebook.presto.spi.procedure.Procedure;
 import com.facebook.presto.spi.procedure.Procedure.Argument;
 import com.google.common.collect.ImmutableList;
@@ -83,7 +84,7 @@ public CreateEmptyPartitionProcedure(
     @Override
     public Procedure get()
     {
-        return new Procedure(
+        return new LocalProcedure(
                 "system",
                 "create_empty_partition",
                 ImmutableList.of(

presto-hive/src/main/java/com/facebook/presto/hive/DirectoryListCacheInvalidationProcedure.java

@@ -16,6 +16,7 @@
 import com.facebook.presto.spi.ConnectorSession;
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.classloader.ThreadContextClassLoader;
+import com.facebook.presto.spi.procedure.LocalProcedure;
 import com.facebook.presto.spi.procedure.Procedure;
 import com.facebook.presto.spi.procedure.Procedure.Argument;
 import com.google.common.collect.ImmutableList;
@@ -52,7 +53,7 @@ public DirectoryListCacheInvalidationProcedure(DirectoryLister directoryLister)
     @Override
     public Procedure get()
     {
-        return new Procedure(
+        return new LocalProcedure(
                 "system",
                 "invalidate_directory_list_cache",
                 ImmutableList.of(

presto-hive/src/main/java/com/facebook/presto/hive/SyncPartitionMetadataProcedure.java

@@ -24,6 +24,7 @@
 import com.facebook.presto.spi.SchemaTableName;
 import com.facebook.presto.spi.TableNotFoundException;
 import com.facebook.presto.spi.classloader.ThreadContextClassLoader;
+import com.facebook.presto.spi.procedure.LocalProcedure;
 import com.facebook.presto.spi.procedure.Procedure;
 import com.facebook.presto.spi.procedure.Procedure.Argument;
 import com.google.common.collect.ImmutableList;
@@ -95,7 +96,7 @@ public SyncPartitionMetadataProcedure(
     @Override
     public Procedure get()
     {
-        return new Procedure(
+        return new LocalProcedure(
                 "system",
                 "sync_partition_metadata",
                 ImmutableList.of(

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergAbstractMetadata.java

@@ -1080,7 +1080,7 @@ public ConnectorDistributedProcedureHandle beginCallDistributedProcedure(
         procedureContext = Optional.of((IcebergProcedureContext) ((DistributedProcedure) procedure).createContext());
         procedureContext.get().setTable(icebergTable);
         procedureContext.get().setTransaction(transaction);
-        return ((DistributedProcedure) procedure).getBeginCallDistributedProcedure().begin(session, procedureContext.get(), tableLayoutHandle, arguments);
+        return ((DistributedProcedure) procedure).begin(session, procedureContext.get(), tableLayoutHandle, arguments);
     }
 
     @Override
@@ -1093,7 +1093,7 @@ public void finishCallDistributedProcedure(ConnectorSession session, ConnectorDi
                         procedureName.getObjectName()));
         verify(procedure instanceof DistributedProcedure, "procedure must be DistributedProcedure");
         verify(procedureContext.isPresent(), "procedure context must be present");
-        ((DistributedProcedure) procedure).getFinishCallDistributedProcedure().finish(procedureContext.get(), procedureHandle, fragments);
+        ((DistributedProcedure) procedure).finish(procedureContext.get(), procedureHandle, fragments);
         transaction.commitTransaction();
         procedureContext.get().destroy();
     }

presto-iceberg/src/main/java/com/facebook/presto/iceberg/RewriteDataFilesProcedure.java

@@ -21,7 +21,8 @@
 import com.facebook.presto.spi.ConnectorSplitSource;
 import com.facebook.presto.spi.FixedSplitSource;
 import com.facebook.presto.spi.classloader.ThreadContextClassLoader;
-import com.facebook.presto.spi.procedure.Procedure;
+import com.facebook.presto.spi.procedure.DistributedProcedure;
+import com.facebook.presto.spi.procedure.Procedure.Argument;
 import com.facebook.presto.spi.procedure.TableDataRewriteDistributedProcedure;
 import com.google.common.base.VerifyException;
 import com.google.common.collect.ImmutableList;
@@ -63,7 +64,7 @@
 import static java.util.Objects.requireNonNull;
 
 public class RewriteDataFilesProcedure
-        implements Provider<Procedure>
+        implements Provider<DistributedProcedure>
 {
     TypeManager typeManager;
     JsonCodec<CommitTaskData> commitTaskCodec;
@@ -78,16 +79,16 @@ public RewriteDataFilesProcedure(
     }
 
     @Override
-    public Procedure get()
+    public DistributedProcedure get()
     {
         return new TableDataRewriteDistributedProcedure(
                 "system",
                 "rewrite_data_files",
                 ImmutableList.of(
-                        new Procedure.Argument(SCHEMA, VARCHAR),
-                        new Procedure.Argument(TABLE_NAME, VARCHAR),
-                        new Procedure.Argument("filter", VARCHAR, false, "TRUE"),
-                        new Procedure.Argument("options", "map(varchar, varchar)", false, null)),
+                        new Argument(SCHEMA, VARCHAR),
+                        new Argument(TABLE_NAME, VARCHAR),
+                        new Argument("filter", VARCHAR, false, "TRUE"),
+                        new Argument("options", "map(varchar, varchar)", false, null)),
                 (session, procedureContext, tableLayoutHandle, arguments) -> beginCallDistributedProcedure(session, (IcebergProcedureContext) procedureContext, (IcebergTableLayoutHandle) tableLayoutHandle, arguments),
                 ((procedureContext, tableHandle, fragments) -> finishCallDistributedProcedure((IcebergProcedureContext) procedureContext, tableHandle, fragments)),
                 IcebergProcedureContext::new);

presto-iceberg/src/main/java/com/facebook/presto/iceberg/procedure/ExpireSnapshotsProcedure.java

@@ -20,7 +20,9 @@
 import com.facebook.presto.spi.ConnectorSession;
 import com.facebook.presto.spi.SchemaTableName;
 import com.facebook.presto.spi.classloader.ThreadContextClassLoader;
+import com.facebook.presto.spi.procedure.LocalProcedure;
 import com.facebook.presto.spi.procedure.Procedure;
+import com.facebook.presto.spi.procedure.Procedure.Argument;
 import com.google.common.collect.ImmutableList;
 import jakarta.inject.Inject;
 import org.apache.iceberg.ExpireSnapshots;
@@ -61,15 +63,15 @@ public ExpireSnapshotsProcedure(IcebergMetadataFactory metadataFactory)
     @Override
     public Procedure get()
     {
-        return new Procedure(
+        return new LocalProcedure(
                 "system",
                 "expire_snapshots",
                 ImmutableList.of(
-                        new Procedure.Argument("schema", VARCHAR),
-                        new Procedure.Argument("table_name", VARCHAR),
-                        new Procedure.Argument("older_than", TIMESTAMP, false, null),
-                        new Procedure.Argument("retain_last", INTEGER, false, null),
-                        new Procedure.Argument("snapshot_ids", "array(bigint)", false, null)),
+                        new Argument("schema", VARCHAR),
+                        new Argument("table_name", VARCHAR),
+                        new Argument("older_than", TIMESTAMP, false, null),
+                        new Argument("retain_last", INTEGER, false, null),
+                        new Argument("snapshot_ids", "array(bigint)", false, null)),
                 EXPIRE_SNAPSHOTS.bindTo(this));
     }
 

presto-iceberg/src/main/java/com/facebook/presto/iceberg/procedure/FastForwardBranchProcedure.java

@@ -17,7 +17,9 @@
 import com.facebook.presto.spi.ConnectorSession;
 import com.facebook.presto.spi.SchemaTableName;
 import com.facebook.presto.spi.connector.ConnectorMetadata;
+import com.facebook.presto.spi.procedure.LocalProcedure;
 import com.facebook.presto.spi.procedure.Procedure;
+import com.facebook.presto.spi.procedure.Procedure.Argument;
 import com.google.common.collect.ImmutableList;
 import jakarta.inject.Inject;
 import org.apache.iceberg.Table;
@@ -54,14 +56,14 @@ public FastForwardBranchProcedure(IcebergMetadataFactory metadataFactory)
     @Override
     public Procedure get()
     {
-        return new Procedure(
+        return new LocalProcedure(
                 "system",
                 "fast_forward",
                 ImmutableList.of(
-                        new Procedure.Argument("schema", VARCHAR),
-                        new Procedure.Argument("table_name", VARCHAR),
-                        new Procedure.Argument("branch", VARCHAR),
-                        new Procedure.Argument("to", VARCHAR)),
+                        new Argument("schema", VARCHAR),
+                        new Argument("table_name", VARCHAR),
+                        new Argument("branch", VARCHAR),
+                        new Argument("to", VARCHAR)),
                 FAST_FORWARD.bindTo(this));
     }
 

presto-iceberg/src/main/java/com/facebook/presto/iceberg/procedure/ManifestFileCacheInvalidationProcedure.java

@@ -15,6 +15,7 @@
 
 import com.facebook.presto.iceberg.ManifestFileCache;
 import com.facebook.presto.spi.classloader.ThreadContextClassLoader;
+import com.facebook.presto.spi.procedure.LocalProcedure;
 import com.facebook.presto.spi.procedure.Procedure;
 import com.google.common.collect.ImmutableList;
 import jakarta.inject.Inject;
@@ -44,7 +45,7 @@ public ManifestFileCacheInvalidationProcedure(ManifestFileCache manifestFileCach
     @Override
     public Procedure get()
     {
-        return new Procedure(
+        return new LocalProcedure(
                 "system",
                 "invalidate_manifest_file_cache",
                 ImmutableList.of(),