feat: Distributed Procedure Support Part 1/X - core code base changes (#26373)

## Description

This PR is the first part of many PRs to support distributed procedure
into Presto. It is a split of the original entire PR which is located
here: #22659.

The whole work in this PR includes the following parts:

1. Re-factor `ProcedureRegistry`/`Procedure` data structure to support
the creation and register of `DistributedProcedure`. And make sure
`ProcedureRegistry` be available in `presto-analyzer` module and
connectors, so that we can recognize distributed procedures in call
statement during prepare analyze stages.

2. Handle call statement on distributed procedures in preparer stage. In
this stage, we figure out the procedure's type in call statement, and
define a new query type `CALL_DISTRIBUTED_PROCEDURE` for `call
distributed procedure` in `BuiltInPreparedQuery`. In this way, `call
distributed procedure` statement would be handled by
`SqlQueryExecutionFactory`, then be created and handled as a
`SqlQueryExecution`.

3. Analyze and plan the `call distributed procedure` statement based on
the subtype of the distributed procedure. For subtype
`TableDataRewriteDistributedProcedure`, ultimately generate a logical
plan for it as follows:

```
OutputNode <- TableFinishNode <- CallDistributedProcedureNode <- FilterNode <- TableScanNode
```

4. Optimize, segmentation, grouped tag and local plan for the logical
plan generated above. The handle logical for
`CallDistributedProcedureNode` is similar as `TableWriterNode`. Besides,
a new optimizer `RewriteWriterTarget` is added, which is placed after
all optimization rules. It is used to update the `TableHandle` held in
`TableFinishNode` and `CallDistributedProcedureNode` based on the
underlying `TableScanNode` after the entire optimization is completed,
considering the possible filter pushing down.

## Motivation and Context

prestodb/rfcs#12

## Impact

N/A

## Test Plan

- Add test cases in each phase involving the procedure architecture
expansion, including creating and registering for distributed
procedures, preparing for call distributed procedure, analyzing for call
distributed procedure, logical planning and optimizing for call
distributed procedure

## Contributor checklist

- [x] Please make sure your submission complies with our [contributing
guide](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md),
in particular [code
style](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#code-style)
and [commit
standards](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#commit-standards).
- [x] PR description addresses the issue accurately and concisely. If
the change is non-trivial, a GitHub Issue is referenced.
- [ ] Documented new properties (with its default value), SQL syntax,
functions, or other functionality.
- [x] If release notes are required, they follow the [release notes
guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines).
- [x] Adequate tests were added if applicable.
- [x] CI passed.
- [ ] If adding new dependencies, verified they have an [OpenSSF
Scorecard](https://securityscorecards.dev/#the-checks) score of 5.0 or
higher (or obtained explicit TSC approval for lower scores).

## Release Notes

```
== RELEASE NOTES ==

General Changes
 * Upgrade the procedure architecture to support distributed executing procedures

```

Author: hantangwangd

presto-analyzer/src/main/java/com/facebook/presto/sql/analyzer/Analysis.java

@@ -34,6 +34,7 @@
 import com.facebook.presto.spi.function.FunctionKind;
 import com.facebook.presto.spi.function.table.Argument;
 import com.facebook.presto.spi.function.table.ConnectorTableFunctionHandle;
+import com.facebook.presto.spi.procedure.DistributedProcedure;
 import com.facebook.presto.spi.security.AccessControl;
 import com.facebook.presto.spi.security.AccessControlContext;
 import com.facebook.presto.spi.security.AllowAllAccessControl;
@@ -176,6 +177,13 @@ public class Analysis
     private final Multiset<ColumnMaskScopeEntry> columnMaskScopes = HashMultiset.create();
     private final Map<NodeRef<Table>, Map<String, Expression>> columnMasks = new LinkedHashMap<>();
 
+    // for call distributed procedure
+    private Optional<DistributedProcedure.DistributedProcedureType> distributedProcedureType = Optional.empty();
+    private Optional<QualifiedObjectName> procedureName = Optional.empty();
+    private Optional<Object[]> procedureArguments = Optional.empty();
+    private Optional<TableHandle> callTarget = Optional.empty();
+    private Optional<QuerySpecification> targetQuery = Optional.empty();
+
     // for create table
     private Optional<QualifiedObjectName> createTableDestination = Optional.empty();
     private Map<String, Expression> createTableProperties = ImmutableMap.of();
@@ -670,6 +678,46 @@ public Optional<QualifiedObjectName> getCreateTableDestination()
         return createTableDestination;
     }
 
+    public Optional<QualifiedObjectName> getProcedureName()
+    {
+        return procedureName;
+    }
+
+    public void setProcedureName(Optional<QualifiedObjectName> procedureName)
+    {
+        this.procedureName = procedureName;
+    }
+
+    public Optional<DistributedProcedure.DistributedProcedureType> getDistributedProcedureType()
+    {
+        return distributedProcedureType;
+    }
+
+    public void setDistributedProcedureType(Optional<DistributedProcedure.DistributedProcedureType> distributedProcedureType)
+    {
+        this.distributedProcedureType = distributedProcedureType;
+    }
+
+    public Optional<Object[]> getProcedureArguments()
+    {
+        return procedureArguments;
+    }
+
+    public void setProcedureArguments(Optional<Object[]> procedureArguments)
+    {
+        this.procedureArguments = procedureArguments;
+    }
+
+    public Optional<TableHandle> getCallTarget()
+    {
+        return callTarget;
+    }
+
+    public void setCallTarget(TableHandle callTarget)
+    {
+        this.callTarget = Optional.of(callTarget);
+    }
+
     public Optional<TableHandle> getAnalyzeTarget()
     {
         return analyzeTarget;
@@ -1044,6 +1092,16 @@ public Optional<QuerySpecification> getCurrentQuerySpecification()
         return currentQuerySpecification;
     }
 
+    public void setTargetQuery(QuerySpecification targetQuery)
+    {
+        this.targetQuery = Optional.of(targetQuery);
+    }
+
+    public Optional<QuerySpecification> getTargetQuery()
+    {
+        return this.targetQuery;
+    }
+
     public Map<FunctionKind, Set<String>> getInvokedFunctions()
     {
         Map<FunctionKind, Set<String>> functionMap = new HashMap<>();

presto-analyzer/src/main/java/com/facebook/presto/sql/analyzer/BuiltInQueryPreparer.java

@@ -13,19 +13,25 @@
  */
 package com.facebook.presto.sql.analyzer;
 
+import com.facebook.presto.common.QualifiedObjectName;
 import com.facebook.presto.common.analyzer.PreparedQuery;
 import com.facebook.presto.common.resourceGroups.QueryType;
+import com.facebook.presto.spi.ConnectorId;
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.PrestoWarning;
+import com.facebook.presto.spi.SchemaTableName;
 import com.facebook.presto.spi.WarningCollector;
 import com.facebook.presto.spi.analyzer.AnalyzerOptions;
 import com.facebook.presto.spi.analyzer.QueryPreparer;
+import com.facebook.presto.spi.procedure.ProcedureRegistry;
 import com.facebook.presto.sql.analyzer.utils.StatementUtils;
 import com.facebook.presto.sql.parser.SqlParser;
+import com.facebook.presto.sql.tree.Call;
 import com.facebook.presto.sql.tree.Execute;
 import com.facebook.presto.sql.tree.Explain;
 import com.facebook.presto.sql.tree.ExplainType;
 import com.facebook.presto.sql.tree.Expression;
+import com.facebook.presto.sql.tree.QualifiedName;
 import com.facebook.presto.sql.tree.Statement;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -36,13 +42,15 @@
 import java.util.Optional;
 
 import static com.facebook.presto.common.WarningHandlingLevel.AS_ERROR;
+import static com.facebook.presto.common.resourceGroups.QueryType.CALL_DISTRIBUTED_PROCEDURE;
 import static com.facebook.presto.spi.StandardErrorCode.NOT_FOUND;
 import static com.facebook.presto.spi.StandardErrorCode.NOT_SUPPORTED;
 import static com.facebook.presto.spi.StandardErrorCode.WARNING_AS_ERROR;
 import static com.facebook.presto.sql.SqlFormatter.formatSql;
 import static com.facebook.presto.sql.analyzer.ConstantExpressionVerifier.verifyExpressionIsConstant;
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.INVALID_PARAMETER_USAGE;
 import static com.facebook.presto.sql.analyzer.utils.AnalyzerUtil.createParsingOptions;
+import static com.facebook.presto.sql.analyzer.utils.MetadataUtils.createQualifiedObjectName;
 import static com.facebook.presto.sql.analyzer.utils.ParameterExtractor.getParameterCount;
 import static com.facebook.presto.sql.tree.ExplainType.Type.VALIDATE;
 import static java.lang.String.format;
@@ -56,11 +64,15 @@ public class BuiltInQueryPreparer
         implements QueryPreparer
 {
     private final SqlParser sqlParser;
+    private final ProcedureRegistry procedureRegistry;
 
     @Inject
-    public BuiltInQueryPreparer(SqlParser sqlParser)
+    public BuiltInQueryPreparer(
+            SqlParser sqlParser,
+            ProcedureRegistry procedureRegistry)
     {
         this.sqlParser = requireNonNull(sqlParser, "sqlParser is null");
+        this.procedureRegistry = requireNonNull(procedureRegistry, "procedureRegistry is null");
     }
 
     @Override
@@ -87,6 +99,18 @@ public BuiltInPreparedQuery prepareQuery(AnalyzerOptions analyzerOptions, Statem
             statement = sqlParser.createStatement(query, createParsingOptions(analyzerOptions));
         }
 
+        Optional<QualifiedObjectName> distributedProcedureName = Optional.empty();
+        if (statement instanceof Call) {
+            QualifiedName qualifiedName = ((Call) statement).getName();
+            QualifiedObjectName qualifiedObjectName = createQualifiedObjectName(analyzerOptions.getSessionCatalogName(), analyzerOptions.getSessionSchemaName(),
+                    statement, qualifiedName, (catalogName, objectName) -> objectName);
+            if (procedureRegistry.isDistributedProcedure(
+                    new ConnectorId(qualifiedObjectName.getCatalogName()),
+                    new SchemaTableName(qualifiedObjectName.getSchemaName(), qualifiedObjectName.getObjectName()))) {
+                distributedProcedureName = Optional.of(qualifiedObjectName);
+            }
+        }
+
         if (statement instanceof Explain && ((Explain) statement).isAnalyze()) {
             Statement innerStatement = ((Explain) statement).getStatement();
             Optional<QueryType> innerQueryType = StatementUtils.getQueryType(innerStatement.getClass());
@@ -103,7 +127,7 @@ public BuiltInPreparedQuery prepareQuery(AnalyzerOptions analyzerOptions, Statem
         if (analyzerOptions.isLogFormattedQueryEnabled()) {
             formattedQuery = Optional.of(getFormattedQuery(statement, parameters));
         }
-        return new BuiltInPreparedQuery(wrappedStatement, statement, parameters, formattedQuery, prepareSql);
+        return new BuiltInPreparedQuery(wrappedStatement, statement, parameters, formattedQuery, prepareSql, distributedProcedureName);
     }
 
     private static String getFormattedQuery(Statement statement, List<Expression> parameters)
@@ -131,13 +155,19 @@ public static class BuiltInPreparedQuery
         private final Statement statement;
         private final Statement wrappedStatement;
         private final List<Expression> parameters;
+        private final Optional<QualifiedObjectName> distributedProcedureName;
 
-        public BuiltInPreparedQuery(Statement wrappedStatement, Statement statement, List<Expression> parameters, Optional<String> formattedQuery, Optional<String> prepareSql)
+        public BuiltInPreparedQuery(
+                Statement wrappedStatement,
+                Statement statement, List<Expression> parameters,
+                Optional<String> formattedQuery, Optional<String> prepareSql,
+                Optional<QualifiedObjectName> distributedProcedureName)
         {
             super(formattedQuery, prepareSql);
             this.wrappedStatement = requireNonNull(wrappedStatement, "wrappedStatement is null");
             this.statement = requireNonNull(statement, "statement is null");
             this.parameters = ImmutableList.copyOf(requireNonNull(parameters, "parameters is null"));
+            this.distributedProcedureName = requireNonNull(distributedProcedureName, "distributedProcedureName is null");
         }
 
         public Statement getStatement()
@@ -157,9 +187,17 @@ public List<Expression> getParameters()
 
         public Optional<QueryType> getQueryType()
         {
+            if (getDistributedProcedureName().isPresent()) {
+                return Optional.of(CALL_DISTRIBUTED_PROCEDURE);
+            }
             return StatementUtils.getQueryType(statement.getClass());
         }
 
+        public Optional<QualifiedObjectName> getDistributedProcedureName()
+        {
+            return this.distributedProcedureName;
+        }
+
         public boolean isTransactionControlStatement()
         {
             return StatementUtils.isTransactionControlStatement(getStatement());

presto-analyzer/src/main/java/com/facebook/presto/sql/analyzer/SemanticErrorCode.java

@@ -90,6 +90,7 @@ public enum SemanticErrorCode
 
     SAMPLE_PERCENTAGE_OUT_OF_RANGE,
 
+    PROCEDURE_NOT_FOUND,
     INVALID_PROCEDURE_ARGUMENTS,
 
     INVALID_SESSION_PROPERTY,

presto-analyzer/src/main/java/com/facebook/presto/sql/analyzer/utils/MetadataUtils.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed 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.
+ */
+package com.facebook.presto.sql.analyzer.utils;
+
+import com.facebook.presto.common.QualifiedObjectName;
+import com.facebook.presto.spi.PrestoException;
+import com.facebook.presto.sql.analyzer.SemanticException;
+import com.facebook.presto.sql.tree.Identifier;
+import com.facebook.presto.sql.tree.Node;
+import com.facebook.presto.sql.tree.QualifiedName;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.function.BiFunction;
+
+import static com.facebook.presto.spi.StandardErrorCode.SYNTAX_ERROR;
+import static com.facebook.presto.sql.analyzer.SemanticErrorCode.CATALOG_NOT_SPECIFIED;
+import static com.facebook.presto.sql.analyzer.SemanticErrorCode.SCHEMA_NOT_SPECIFIED;
+import static java.lang.String.format;
+import static java.util.Locale.ENGLISH;
+import static java.util.Objects.requireNonNull;
+
+public class MetadataUtils
+{
+    private MetadataUtils()
+    {}
+
+    public static QualifiedObjectName createQualifiedObjectName(Optional<String> sessionCatalogName, Optional<String> sessionSchemaName, Node node, QualifiedName name,
+                                                                BiFunction<String, String, String> normalizer)
+    {
+        requireNonNull(sessionCatalogName, "sessionCatalogName is null");
+        requireNonNull(sessionSchemaName, "sessionSchemaName is null");
+        requireNonNull(name, "name is null");
+        if (name.getParts().size() > 3) {
+            throw new PrestoException(SYNTAX_ERROR, format("Too many dots in table name: %s", name));
+        }
+
+        List<Identifier> parts = Lists.reverse(name.getOriginalParts());
+        String objectName = parts.get(0).getValue();
+        String schemaName = (parts.size() > 1) ? parts.get(1).getValue() : sessionSchemaName.orElseThrow(() ->
+                new SemanticException(SCHEMA_NOT_SPECIFIED, node, "Schema must be specified when session schema is not set"));
+        String catalogName = (parts.size() > 2) ? parts.get(2).getValue() : sessionCatalogName.orElseThrow(() ->
+                new SemanticException(CATALOG_NOT_SPECIFIED, node, "Catalog must be specified when session catalog is not set"));
+
+        catalogName = catalogName.toLowerCase(ENGLISH);
+        schemaName = normalizer.apply(catalogName, schemaName);
+        objectName = normalizer.apply(catalogName, objectName);
+        return new QualifiedObjectName(catalogName, schemaName, objectName);
+    }
+}