fix: Presto spark add X509 certificates to SessionContext (#27183)

## Description
<!---Describe your changes in detail-->
The X509 certificates are available from the Session of the query, so it
can also be passed to the PrestoSparkSession, PrestoSparkRunnerContext,
and PrestoSparkSessionContext so that they can be stored as a field.

## Motivation and Context
<!---Why is this change required? What problem does it solve?-->
<!---If it fixes an open issue, please link to the issue here.-->
Compared to HttpRequestSessionContext, PrestoSparkSessionContext does
not have the certificates, and it always returns an empty list because
it uses the default implementation of the SessionContext interface.


## Impact
<!---Describe any public API or user-facing feature change or any
performance impact-->
Adds extra field to:
PrestoSparkSessionContext, PrestoSparkRunnerContext, PrestoSparkSession

## Test Plan
<!---Please fill in how you tested your change-->
No impact on logic, and the field was accessible in the SessionContext
when running PrestoSpark

## 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.

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


Differential Revision: D93777694

Author: kevintang2022

presto-spark-base/src/main/java/com/facebook/presto/spark/PrestoSparkSessionContext.java

@@ -29,6 +29,8 @@
 import com.google.common.collect.ImmutableSet;
 import jakarta.annotation.Nullable;
 
+import java.security.cert.X509Certificate;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -77,7 +79,8 @@ public static PrestoSparkSessionContext createFromSessionInfo(
                         extraCredentials.build(),
                         extraTokenAuthenticators.build(),
                         Optional.empty(),
-                        Optional.empty()),
+                        Optional.empty(),
+                        prestoSparkSession.getCertificates()),
                 prestoSparkSession.getCatalog().orElse(null),
                 prestoSparkSession.getSchema().orElse(null),
                 prestoSparkSession.getSource().orElse(null),
@@ -128,6 +131,12 @@ public Identity getIdentity()
         return identity;
     }
 
+    @Override
+    public List<X509Certificate> getCertificates()
+    {
+        return identity.getCertificates();
+    }
+
     @Nullable
     @Override
     public String getCatalog()

presto-spark-base/src/test/java/com/facebook/presto/spark/PrestoSparkQueryRunner.java

@@ -339,15 +339,15 @@ public PrestoSparkQueryRunner(
                 // Sql-Standard Access Control Checker
                 // needs us to specify our role
                 .setIdentity(
-                    new Identity(
-                        "hive",
-                        Optional.empty(),
-                        ImmutableMap.of(defaultCatalog,
-                            new SelectedRole(Type.ROLE, Optional.of("admin"))),
-                        ImmutableMap.of(),
-                        ImmutableMap.of(),
-                        Optional.empty(),
-                        Optional.empty()))
+                        new Identity(
+                                "hive",
+                                Optional.empty(),
+                                ImmutableMap.of(defaultCatalog,
+                                        new SelectedRole(Type.ROLE, Optional.of("admin"))),
+                                ImmutableMap.of(),
+                                ImmutableMap.of(),
+                                Optional.empty(),
+                                Optional.empty()))
                 .build();
 
         transactionManager = injector.getInstance(TransactionManager.class);
@@ -659,6 +659,7 @@ private static PrestoSparkSession createSessionInfo(Session session)
                 session.getIdentity().getUser(),
                 session.getIdentity().getPrincipal(),
                 session.getIdentity().getExtraCredentials(),
+                session.getIdentity().getCertificates(),
                 session.getCatalog(),
                 session.getSchema(),
                 session.getSource(),

presto-spark-classloader-interface/src/main/java/com/facebook/presto/spark/classloader_interface/PrestoSparkSession.java

@@ -14,12 +14,16 @@
 package com.facebook.presto.spark.classloader_interface;
 
 import java.security.Principal;
+import java.security.cert.X509Certificate;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
+import static java.util.Collections.unmodifiableList;
 import static java.util.Collections.unmodifiableMap;
 import static java.util.Collections.unmodifiableSet;
 import static java.util.Objects.requireNonNull;
@@ -33,6 +37,7 @@ public class PrestoSparkSession
     private final String user;
     private final Optional<Principal> principal;
     private final Map<String, String> extraCredentials;
+    private final List<X509Certificate> certificates;
     private final Optional<String> catalog;
     private final Optional<String> schema;
     private final Optional<String> source;
@@ -49,6 +54,7 @@ public PrestoSparkSession(
             String user,
             Optional<Principal> principal,
             Map<String, String> extraCredentials,
+            List<X509Certificate> certificates,
             Optional<String> catalog,
             Optional<String> schema,
             Optional<String> source,
@@ -65,6 +71,7 @@ public PrestoSparkSession(
         this.user = requireNonNull(user, "user is null");
         this.principal = requireNonNull(principal, "principal is null");
         this.extraCredentials = unmodifiableMap(new HashMap<>(requireNonNull(extraCredentials, "extraCredentials is null")));
+        this.certificates = unmodifiableList(new ArrayList<>(requireNonNull(certificates, "certificates is null")));
         this.catalog = requireNonNull(catalog, "catalog is null");
         this.schema = requireNonNull(schema, "schema is null");
         this.source = requireNonNull(source, "source is null");
@@ -94,6 +101,11 @@ public Map<String, String> getExtraCredentials()
         return extraCredentials;
     }
 
+    public List<X509Certificate> getCertificates()
+    {
+        return certificates;
+    }
+
     public Optional<String> getCatalog()
     {
         return catalog;

presto-spark-launcher/src/main/java/com/facebook/presto/spark/launcher/PrestoSparkLauncherCommand.java

@@ -14,6 +14,7 @@
 package com.facebook.presto.spark.launcher;
 
 import com.facebook.presto.spark.classloader_interface.PrestoSparkConfInitializer;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import io.airlift.airline.Command;
@@ -67,14 +68,15 @@ public void run()
                 Optional.empty(),
                 clientOptions.sessionPropertyConfig == null ? Optional.empty() : Optional.of(
                         loadProperties(checkFile(new File(clientOptions.sessionPropertyConfig)))),
-                    Optional.empty(),
+                Optional.empty(),
                 Optional.empty());
 
         try (PrestoSparkRunner runner = new PrestoSparkRunner(distribution)) {
             runner.run(
                     "test",
                     Optional.empty(),
                     ImmutableMap.of(),
+                    ImmutableList.of(),
                     clientOptions.catalog,
                     clientOptions.schema,
                     Optional.empty(),

presto-spark-launcher/src/main/java/com/facebook/presto/spark/launcher/PrestoSparkRunner.java

@@ -35,6 +35,7 @@
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.security.Principal;
+import java.security.cert.X509Certificate;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -84,6 +85,7 @@ public void run(
             String user,
             Optional<Principal> principal,
             Map<String, String> extraCredentials,
+            List<X509Certificate> certificates,
             String catalog,
             String schema,
             Optional<String> source,
@@ -106,6 +108,7 @@ public void run(
                 user,
                 principal,
                 extraCredentials,
+                certificates,
                 catalog,
                 schema,
                 source,
@@ -154,6 +157,7 @@ private void execute(IPrestoSparkQueryExecutionFactory queryExecutionFactory, Pr
                 prestoSparkRunnerContext.getUser(),
                 prestoSparkRunnerContext.getPrincipal(),
                 prestoSparkRunnerContext.getExtraCredentials(),
+                prestoSparkRunnerContext.getCertificates(),
                 Optional.ofNullable(prestoSparkRunnerContext.getCatalog()),
                 Optional.ofNullable(prestoSparkRunnerContext.getSchema()),
                 prestoSparkRunnerContext.getSource(),

presto-spark-launcher/src/main/java/com/facebook/presto/spark/launcher/PrestoSparkRunnerContext.java

@@ -16,6 +16,7 @@
 import com.facebook.presto.spark.classloader_interface.ExecutionStrategy;
 
 import java.security.Principal;
+import java.security.cert.X509Certificate;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -28,6 +29,7 @@ public class PrestoSparkRunnerContext
     private final String user;
     private final Optional<Principal> principal;
     private final Map<String, String> extraCredentials;
+    private final List<X509Certificate> certificates;
     private final String catalog;
     private final String schema;
     private final Optional<String> source;
@@ -50,6 +52,7 @@ public PrestoSparkRunnerContext(
             String user,
             Optional<Principal> principal,
             Map<String, String> extraCredentials,
+            List<X509Certificate> certificates,
             String catalog,
             String schema,
             Optional<String> source,
@@ -71,6 +74,7 @@ public PrestoSparkRunnerContext(
         this.user = user;
         this.principal = principal;
         this.extraCredentials = extraCredentials;
+        this.certificates = certificates;
         this.catalog = catalog;
         this.schema = schema;
         this.source = source;
@@ -105,6 +109,11 @@ public Map<String, String> getExtraCredentials()
         return extraCredentials;
     }
 
+    public List<X509Certificate> getCertificates()
+    {
+        return certificates;
+    }
+
     public String getCatalog()
     {
         return catalog;
@@ -195,6 +204,7 @@ public static class Builder
         private String user;
         private Optional<Principal> principal;
         private Map<String, String> extraCredentials;
+        private List<X509Certificate> certificates;
         private String catalog;
         private String schema;
         private Optional<String> source;
@@ -218,6 +228,7 @@ public Builder(PrestoSparkRunnerContext prestoSparkRunnerContext)
             this.user = prestoSparkRunnerContext.getUser();
             this.principal = prestoSparkRunnerContext.getPrincipal();
             this.extraCredentials = prestoSparkRunnerContext.getExtraCredentials();
+            this.certificates = prestoSparkRunnerContext.getCertificates();
             this.catalog = prestoSparkRunnerContext.getCatalog();
             this.schema = prestoSparkRunnerContext.getSchema();
             this.source = prestoSparkRunnerContext.getSource();
@@ -249,6 +260,7 @@ public PrestoSparkRunnerContext build()
                     user,
                     principal,
                     extraCredentials,
+                    certificates,
                     catalog,
                     schema,
                     source,