Native built in functions demo 1 (prestodb#25661)

Summary:
This PR includes changes for Built in functions for different use cases
- Register SQL invoked function in built in namespace when they come from plugins (prestodb#25597)
- Register Native functions in built in namespace when they come from sidecar (prestodb/rfcs#41)

Since these use cases have a lot of overlap, we can introduce an abstract class in order to reduce duplicate logic.

## Description

## Motivation and Context

## Impact

Pull Request resolved: prestodb#25661

Test Plan:
<!---Please fill in how you tested your change-->

## Contributor checklist

- [ ] 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).
- [ ] 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.
- [ ] If release notes are required, they follow the [release notes guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines).
- [ ] Adequate tests were added if applicable.
- [ ] CI passed.

## Release Notes
Please follow [release notes guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines) and fill in the release notes below.

```
== RELEASE NOTES ==

General Changes
* ...
* ...

Hive Connector Changes
* ...
* ...
```

If release note is NOT required, use:

```
== NO RELEASE NOTE ==
```

Rollback Plan:

Differential Revision: D79301760

Pulled By: kevintang2022

Author: kevintang2022

presto-main-base/pom.xml

@@ -91,6 +91,11 @@
             <artifactId>bootstrap</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>com.facebook.airlift</groupId>
+            <artifactId>http-client</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>io.airlift</groupId>
             <artifactId>aircompressor</artifactId>

presto-main-base/src/main/java/com/facebook/presto/metadata/BuiltInFunctionHandle.java

@@ -15,6 +15,7 @@
 
 import com.facebook.presto.common.CatalogSchemaName;
 import com.facebook.presto.common.type.TypeSignature;
+import com.facebook.presto.spi.function.BuiltInKind;
 import com.facebook.presto.spi.function.FunctionHandle;
 import com.facebook.presto.spi.function.FunctionKind;
 import com.facebook.presto.spi.function.Signature;
@@ -30,12 +31,16 @@ public class BuiltInFunctionHandle
         implements FunctionHandle
 {
     private final Signature signature;
+    private final BuiltInKind builtInKind;
 
     @JsonCreator
-    public BuiltInFunctionHandle(@JsonProperty("signature") Signature signature)
+    public BuiltInFunctionHandle(
+            @JsonProperty("signature") Signature signature,
+            @JsonProperty("builtInKind") BuiltInKind builtInKind)
     {
         this.signature = requireNonNull(signature, "signature is null");
         checkArgument(signature.getTypeVariableConstraints().isEmpty(), "%s has unbound type parameters", signature);
+        this.builtInKind = builtInKind;
     }
 
     @JsonProperty
@@ -62,6 +67,13 @@ public List<TypeSignature> getArgumentTypes()
         return signature.getArgumentTypes();
     }
 
+    @JsonProperty
+    @Override
+    public BuiltInKind getBuiltInKind()
+    {
+        return builtInKind;
+    }
+
     @Override
     public CatalogSchemaName getCatalogSchemaName()
     {
@@ -78,13 +90,14 @@ public boolean equals(Object o)
             return false;
         }
         BuiltInFunctionHandle that = (BuiltInFunctionHandle) o;
-        return Objects.equals(signature, that.signature);
+        return Objects.equals(signature, that.signature)
+                && Objects.equals(builtInKind, that.builtInKind);
     }
 
     @Override
     public int hashCode()
     {
-        return Objects.hash(signature);
+        return Objects.hash(signature, builtInKind);
     }
 
     @Override

presto-main-base/src/main/java/com/facebook/presto/metadata/BuiltInNativeFunctionNamespaceManager.java

@@ -0,0 +1,112 @@
+/*
+ * 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.metadata;
+
+import com.facebook.airlift.log.Logger;
+import com.facebook.presto.spi.PrestoException;
+import com.facebook.presto.spi.function.BuiltInKind;
+import com.facebook.presto.spi.function.FunctionHandle;
+import com.facebook.presto.spi.function.FunctionMetadata;
+import com.facebook.presto.spi.function.Parameter;
+import com.facebook.presto.spi.function.Signature;
+import com.facebook.presto.spi.function.SqlFunction;
+import com.facebook.presto.spi.function.SqlInvokedFunction;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+
+import javax.inject.Provider;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static com.facebook.presto.spi.function.FunctionImplementationType.CPP;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Throwables.throwIfInstanceOf;
+import static com.google.common.collect.ImmutableList.toImmutableList;
+import static java.util.Objects.requireNonNull;
+
+public class BuiltInNativeFunctionNamespaceManager
+        extends BuiltInSpecialFunctionNamespaceManager
+{
+    private static final Logger log = Logger.get(BuiltInNativeFunctionNamespaceManager.class);
+
+    private final Provider<NativeFunctionRegistryTool> nativeFunctionRegistryToolProvider;
+
+    public BuiltInNativeFunctionNamespaceManager(FunctionAndTypeManager functionAndTypeManager, Provider<NativeFunctionRegistryTool> nativeFunctionRegistryToolProvider)
+    {
+        super(functionAndTypeManager);
+        this.nativeFunctionRegistryToolProvider = requireNonNull(nativeFunctionRegistryToolProvider, "nativeFunctionRegistryToolProvider is null");
+    }
+
+    public synchronized void registerNativeFunctions()
+    {
+        // only register functions once
+        if (!this.functions.list().isEmpty()) {
+            return;
+        }
+
+        List<SqlFunction> allNativeFunctions = nativeFunctionRegistryToolProvider
+                .get()
+                .getNativeFunctionSignatureMap()
+                .getUDFSignatureMap()
+                .entrySet()
+                .stream()
+                .flatMap(entry -> entry.getValue().stream()
+                        .map(metaInfo -> NativeSidecarFunctionUtil.createSqlInvokedFunction(entry.getKey(), metaInfo)))
+                .collect(Collectors.toList());
+        this.functions = new FunctionMap(this.functions, allNativeFunctions);
+    }
+
+    @Override
+    public FunctionHandle getFunctionHandle(Signature signature)
+    {
+        return new BuiltInFunctionHandle(signature, BuiltInKind.NATIVE);
+    }
+
+    @Override
+    public FunctionMetadata getFunctionMetadata(FunctionHandle functionHandle)
+    {
+        checkArgument(functionHandle instanceof BuiltInFunctionHandle, "Expect BuiltInFunctionHandle");
+        Signature signature = ((BuiltInFunctionHandle) functionHandle).getSignature();
+        SpecializedFunctionKey functionKey;
+        try {
+            functionKey = specializedFunctionKeyCache.getUnchecked(signature);
+        }
+        catch (UncheckedExecutionException e) {
+            throwIfInstanceOf(e.getCause(), PrestoException.class);
+            throw e;
+        }
+        SqlFunction function = functionKey.getFunction();
+        checkArgument(function instanceof SqlInvokedFunction, "BuiltInPluginFunctionNamespaceManager only support SqlInvokedFunctions");
+        SqlInvokedFunction sqlFunction = (SqlInvokedFunction) function;
+        List<String> argumentNames = sqlFunction.getParameters().stream().map(Parameter::getName).collect(toImmutableList());
+        return new FunctionMetadata(
+                signature.getName(),
+                signature.getArgumentTypes(),
+                argumentNames,
+                signature.getReturnType(),
+                signature.getKind(),
+                sqlFunction.getRoutineCharacteristics().getLanguage(),
+                CPP,
+                function.isDeterministic(),
+                function.isCalledOnNullInput(),
+                sqlFunction.getVersion(),
+                sqlFunction.getComplexTypeFunctionDescriptor());
+    }
+
+    @Override
+    protected synchronized void checkForNamingConflicts(Collection<? extends SqlFunction> functions)
+    {
+    }
+}

presto-main-base/src/main/java/com/facebook/presto/metadata/BuiltInPluginFunctionNamespaceManager.java

@@ -0,0 +1,94 @@
+/*
+ * 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.metadata;
+
+import com.facebook.presto.spi.PrestoException;
+import com.facebook.presto.spi.function.BuiltInKind;
+import com.facebook.presto.spi.function.FunctionHandle;
+import com.facebook.presto.spi.function.FunctionMetadata;
+import com.facebook.presto.spi.function.Parameter;
+import com.facebook.presto.spi.function.Signature;
+import com.facebook.presto.spi.function.SqlFunction;
+import com.facebook.presto.spi.function.SqlInvokedFunction;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+
+import java.util.Collection;
+import java.util.List;
+
+import static com.facebook.presto.spi.function.FunctionImplementationType.SQL;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Throwables.throwIfInstanceOf;
+import static com.google.common.collect.ImmutableList.toImmutableList;
+
+public class BuiltInPluginFunctionNamespaceManager
+        extends BuiltInSpecialFunctionNamespaceManager
+{
+    public BuiltInPluginFunctionNamespaceManager(FunctionAndTypeManager functionAndTypeManager)
+    {
+        super(functionAndTypeManager);
+    }
+
+    public synchronized void registerPluginFunctions(List<? extends SqlFunction> functions)
+    {
+        checkForNamingConflicts(functions);
+        this.functions = new FunctionMap(this.functions, functions);
+    }
+
+    @Override
+    public FunctionMetadata getFunctionMetadata(FunctionHandle functionHandle)
+    {
+        checkArgument(functionHandle instanceof BuiltInFunctionHandle, "Expect BuiltInFunctionHandle");
+        Signature signature = ((BuiltInFunctionHandle) functionHandle).getSignature();
+        SpecializedFunctionKey functionKey;
+        try {
+            functionKey = specializedFunctionKeyCache.getUnchecked(signature);
+        }
+        catch (UncheckedExecutionException e) {
+            throwIfInstanceOf(e.getCause(), PrestoException.class);
+            throw e;
+        }
+        SqlFunction function = functionKey.getFunction();
+        checkArgument(function instanceof SqlInvokedFunction, "BuiltInPluginFunctionNamespaceManager only support SqlInvokedFunctions");
+        SqlInvokedFunction sqlFunction = (SqlInvokedFunction) function;
+        List<String> argumentNames = sqlFunction.getParameters().stream().map(Parameter::getName).collect(toImmutableList());
+        return new FunctionMetadata(
+                signature.getName(),
+                signature.getArgumentTypes(),
+                argumentNames,
+                signature.getReturnType(),
+                signature.getKind(),
+                sqlFunction.getRoutineCharacteristics().getLanguage(),
+                SQL,
+                function.isDeterministic(),
+                function.isCalledOnNullInput(),
+                sqlFunction.getVersion(),
+                sqlFunction.getComplexTypeFunctionDescriptor());
+    }
+
+    @Override
+    public FunctionHandle getFunctionHandle(Signature signature)
+    {
+        return new BuiltInFunctionHandle(signature, BuiltInKind.PLUGIN);
+    }
+
+    @Override
+    protected synchronized void checkForNamingConflicts(Collection<? extends SqlFunction> functions)
+    {
+        for (SqlFunction function : functions) {
+            for (SqlFunction existingFunction : this.functions.list()) {
+                checkArgument(!function.getSignature().equals(existingFunction.getSignature()), "Function already registered: %s", function.getSignature());
+            }
+        }
+    }
+}