Support multiple catalogs for Presto spark native execution (prestodb#25943)

## Description
This PR implements native catalog properties support for Presto Spark,
enabling proper configuration and management of catalog properties for
native execution processes.

**Key Changes:**

* **Added `NativeExecutionCatalogProperties` class**: A new class that
holds catalog properties for native execution processes, where each
catalog generates a separate `<catalog-name>.properties` file.
    
* **Enhanced `WorkerProperty` and `PrestoSparkWorkerProperty`**:
Extended the worker property classes to support catalog properties
configuration and proper property file generation.
    
* **Updated Native Execution Module**: Modified `NativeExecutionModule`
and `NativeExecutionProcess` to integrate catalog properties into the
native execution workflow.
    
* **Improved Configuration Integration**: Updated `PrestoSparkModule`,
`PrestoSparkServiceFactory` to properly wire the new catalog properties
functionality.
    
* **Enhanced Test Coverage**: Added and updated tests in
`TestNativeExecutionSystemConfig`, `TestNativeExecutionProcess`, and
other test classes to ensure proper catalog properties handling.

## Motivation and Context
This change is required to support proper catalog configuration in
Presto Spark's native execution mode. Previously, catalog properties
were not properly managed for native execution processes, which limited
the ability to configure connectors effectively in native mode.

## Test Plan
*   **Unit Tests**: Added test `NativeExecutionCatalogProperties`
* **Integration Tests**: Updated `TestNativeExecutionProcess` and
`TestPrestoSparkHttpClient` to verify catalog properties integration

## 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.
- [x] Adequate tests were added if applicable.
- [x] CI passed.

## Release Notes
```
== NO RELEASE NOTE ==
```

Author: pradeepvaka

presto-native-execution/src/test/java/com/facebook/presto/spark/PrestoSparkNativeQueryRunnerUtils.java

@@ -18,7 +18,6 @@
 import com.facebook.presto.hive.metastore.ExtendedHiveMetastore;
 import com.facebook.presto.nativeworker.PrestoNativeQueryRunnerUtils;
 import com.facebook.presto.spark.execution.nativeprocess.NativeExecutionModule;
-import com.facebook.presto.spark.execution.property.NativeExecutionConnectorConfig;
 import com.facebook.presto.spi.security.PrincipalType;
 import com.facebook.presto.testing.QueryRunner;
 import com.google.common.collect.ImmutableList;
@@ -125,7 +124,8 @@ public static PrestoSparkQueryRunner createTpchRunner()
         return createRunner(
                 "tpchstandard",
                 new NativeExecutionModule(
-                        Optional.of(new NativeExecutionConnectorConfig().setConnectorName("tpch"))));
+                        Optional.of(
+                                ImmutableMap.of("hive", ImmutableMap.of("connector.name", "tpch")))));
     }
 
     public static PrestoSparkQueryRunner createRunner(String defaultCatalog, Optional<Path> baseDir, Map<String, String> additionalConfigProperties, Map<String, String> additionalSparkProperties, ImmutableList<Module> nativeModules)

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

@@ -120,7 +120,6 @@
 import com.facebook.presto.spark.execution.PrestoSparkBroadcastTableCacheManager;
 import com.facebook.presto.spark.execution.PrestoSparkExecutionExceptionFactory;
 import com.facebook.presto.spark.execution.http.BatchTaskUpdateRequest;
-import com.facebook.presto.spark.execution.property.NativeExecutionConnectorConfig;
 import com.facebook.presto.spark.execution.property.NativeExecutionNodeConfig;
 import com.facebook.presto.spark.execution.shuffle.PrestoSparkLocalShuffleReadInfo;
 import com.facebook.presto.spark.execution.shuffle.PrestoSparkLocalShuffleWriteInfo;
@@ -284,7 +283,6 @@ protected void setup(Binder binder)
         configBinder(binder).bindConfig(PrestoSparkConfig.class);
         configBinder(binder).bindConfig(TracingConfig.class);
         configBinder(binder).bindConfig(NativeExecutionNodeConfig.class);
-        configBinder(binder).bindConfig(NativeExecutionConnectorConfig.class);
         configBinder(binder).bindConfig(PlanCheckerProviderManagerConfig.class);
         configBinder(binder).bindConfig(SecurityConfig.class);
 

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

@@ -27,6 +27,7 @@
 import com.google.inject.Module;
 
 import java.util.List;
+import java.util.Optional;
 
 import static com.facebook.presto.spark.classloader_interface.PrestoSparkConfiguration.METADATA_STORAGE_TYPE_LOCAL;
 import static com.google.common.base.Preconditions.checkArgument;
@@ -70,7 +71,7 @@ protected List<Module> getAdditionalModules(PrestoSparkConfiguration configurati
         return ImmutableList.of(
                 new PrestoSparkLocalMetadataStorageModule(),
                 // TODO: Need to let NativeExecutionModule addition be controlled by configuration as well.
-                new NativeExecutionModule());
+                new NativeExecutionModule(Optional.of(configuration.getCatalogProperties())));
     }
 
     protected SqlParserOptions getSqlParserOptions()

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/nativeprocess/NativeExecutionModule.java

@@ -13,9 +13,7 @@
  */
 package com.facebook.presto.spark.execution.nativeprocess;
 
-import com.facebook.presto.spark.execution.property.NativeExecutionConnectorConfig;
-import com.facebook.presto.spark.execution.property.NativeExecutionNodeConfig;
-import com.facebook.presto.spark.execution.property.NativeExecutionSystemConfig;
+import com.facebook.presto.spark.execution.property.NativeExecutionCatalogProperties;
 import com.facebook.presto.spark.execution.property.PrestoSparkWorkerProperty;
 import com.facebook.presto.spark.execution.property.WorkerProperty;
 import com.facebook.presto.spark.execution.shuffle.PrestoSparkLocalShuffleInfoTranslator;
@@ -32,6 +30,7 @@
 import com.google.inject.TypeLiteral;
 import okhttp3.OkHttpClient;
 
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
@@ -40,20 +39,20 @@
 public class NativeExecutionModule
         implements Module
 {
-    private Optional<NativeExecutionConnectorConfig> connectorConfig;
+    private Optional<Map<String, Map<String, String>>> catalogProperties;
 
     // For use by production system where the configurations can only be tuned via configurations.
     public NativeExecutionModule()
     {
-        this.connectorConfig = Optional.empty();
+        this.catalogProperties = Optional.empty();
     }
 
     // In the future, we would make more bindings injected into NativeExecutionModule
     // to be able to test various configuration parameters
     @VisibleForTesting
-    public NativeExecutionModule(Optional<NativeExecutionConnectorConfig> connectorConfig)
+    public NativeExecutionModule(Optional<Map<String, Map<String, String>>> catalogProperties)
     {
-        this.connectorConfig = connectorConfig;
+        this.catalogProperties = catalogProperties;
     }
 
     @Override
@@ -74,17 +73,14 @@ protected void bindShuffle(Binder binder)
 
     protected void bindWorkerProperties(Binder binder)
     {
+        // Bind NativeExecutionCatalogProperties - this is not bound elsewhere
+        binder.bind(NativeExecutionCatalogProperties.class).toInstance(
+                new NativeExecutionCatalogProperties(catalogProperties.orElse(ImmutableMap.of())));
+
+        // Bind worker property classes
         newOptionalBinder(binder, new TypeLiteral<WorkerProperty<?, ?, ?>>() {
         }).setDefault().to(PrestoSparkWorkerProperty.class).in(Scopes.SINGLETON);
-        if (connectorConfig.isPresent()) {
-            binder.bind(PrestoSparkWorkerProperty.class).toInstance(
-                    new PrestoSparkWorkerProperty(connectorConfig.get(),
-                            new NativeExecutionNodeConfig(), new NativeExecutionSystemConfig(
-                            ImmutableMap.of())));
-        }
-        else {
-            binder.bind(PrestoSparkWorkerProperty.class).in(Scopes.SINGLETON);
-        }
+        binder.bind(PrestoSparkWorkerProperty.class).in(Scopes.SINGLETON);
     }
 
     protected void bindHttpClient(Binder binder)

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/nativeprocess/NativeExecutionProcess.java

@@ -343,8 +343,7 @@ private void populateConfigurationFiles(String configBasePath)
         workerProperty.populateAllProperties(
                 Paths.get(configBasePath, WORKER_CONFIG_FILE),
                 Paths.get(configBasePath, WORKER_NODE_CONFIG_FILE),
-                Paths.get(configBasePath, format("%s%s.properties", WORKER_CONNECTOR_CONFIG_FILE,
-                    getNativeExecutionCatalogName(session))));
+                Paths.get(configBasePath, WORKER_CONNECTOR_CONFIG_FILE));  // Directory path for catalogs
     }
 
     private void doGetServerInfo(SettableFuture<ServerInfo> future)

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/property/NativeExecutionCatalogProperties.java

@@ -0,0 +1,37 @@
+/*
+ * 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.spark.execution.property;
+
+import java.util.Map;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * This class holds catalog properties for native execution process.
+ * Each catalog will generate a separate <catalog-name>.properties file.
+ */
+public class NativeExecutionCatalogProperties
+{
+    private final Map<String, Map<String, String>> catalogProperties;
+
+    public NativeExecutionCatalogProperties(Map<String, Map<String, String>> catalogProperties)
+    {
+        this.catalogProperties = requireNonNull(catalogProperties, "catalogProperties is null");
+    }
+
+    public Map<String, Map<String, String>> getAllCatalogProperties()
+    {
+        return catalogProperties;
+    }
+}

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/property/NativeExecutionConnectorConfig.java

@@ -19,14 +19,14 @@
  * A utility class that helps with properties and its materialization.
  */
 public class PrestoSparkWorkerProperty
-        extends WorkerProperty<NativeExecutionConnectorConfig, NativeExecutionNodeConfig, NativeExecutionSystemConfig>
+        extends WorkerProperty<NativeExecutionCatalogProperties, NativeExecutionNodeConfig, NativeExecutionSystemConfig>
 {
     @Inject
     public PrestoSparkWorkerProperty(
-            NativeExecutionConnectorConfig connectorConfig,
+            NativeExecutionCatalogProperties catalogProperties,
             NativeExecutionNodeConfig nodeConfig,
             NativeExecutionSystemConfig systemConfig)
     {
-        super(connectorConfig, nodeConfig, systemConfig);
+        super(catalogProperties, nodeConfig, systemConfig);
     }
 }

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/property/PrestoSparkWorkerProperty.java

@@ -64,25 +64,25 @@
  * }
  * }
  */
-public class WorkerProperty<T1 extends NativeExecutionConnectorConfig, T2 extends NativeExecutionNodeConfig, T3 extends NativeExecutionSystemConfig>
+public class WorkerProperty<T1 extends NativeExecutionCatalogProperties, T2 extends NativeExecutionNodeConfig, T3 extends NativeExecutionSystemConfig>
 {
-    private final T1 connectorConfig;
+    private final T1 catalogProperties;
     private final T2 nodeConfig;
     private final T3 systemConfig;
 
     public WorkerProperty(
-            T1 connectorConfig,
+            T1 catalogProperties,
             T2 nodeConfig,
             T3 systemConfig)
     {
         this.systemConfig = requireNonNull(systemConfig, "systemConfig is null");
         this.nodeConfig = requireNonNull(nodeConfig, "nodeConfig is null");
-        this.connectorConfig = requireNonNull(connectorConfig, "connectorConfig is null");
+        this.catalogProperties = requireNonNull(catalogProperties, "catalogProperties is null");
     }
 
-    public T1 getConnectorConfig()
+    public T1 getCatalogProperties()
     {
-        return connectorConfig;
+        return catalogProperties;
     }
 
     public T2 getNodeConfig()
@@ -95,12 +95,12 @@ public T3 getSystemConfig()
         return systemConfig;
     }
 
-    public void populateAllProperties(Path systemConfigPath, Path nodeConfigPath, Path connectorConfigPath)
+    public void populateAllProperties(Path systemConfigPath, Path nodeConfigPath, Path catalogConfigsPath)
             throws IOException
     {
         populateProperty(systemConfig.getAllProperties(), systemConfigPath);
         populateProperty(nodeConfig.getAllProperties(), nodeConfigPath);
-        populateProperty(connectorConfig.getAllProperties(), connectorConfigPath);
+        populateCatalogProperties(catalogProperties.getAllCatalogProperties(), catalogConfigsPath);
     }
 
     private void populateProperty(Map<String, String> properties, Path path)
@@ -122,4 +122,20 @@ private void populateProperty(Map<String, String> properties, Path path)
             throw e;
         }
     }
+
+    private void populateCatalogProperties(Map<String, Map<String, String>> catalogProperties, Path path)
+            throws IOException
+    {
+        File catalogDir = path.toFile();
+        if (!catalogDir.exists()) {
+            catalogDir.mkdirs();
+        }
+
+        for (Map.Entry<String, Map<String, String>> catalogEntry : catalogProperties.entrySet()) {
+            String catalogName = catalogEntry.getKey();
+            Map<String, String> properties = catalogEntry.getValue();
+            Path catalogConfigPath = path.resolve(catalogName + ".properties");
+            populateProperty(properties, catalogConfigPath);
+        }
+    }
 }

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/property/WorkerProperty.java

@@ -22,7 +22,7 @@
 import com.facebook.presto.spark.execution.http.TestPrestoSparkHttpClient;
 import com.facebook.presto.spark.execution.nativeprocess.NativeExecutionProcess;
 import com.facebook.presto.spark.execution.nativeprocess.NativeExecutionProcessFactory;
-import com.facebook.presto.spark.execution.property.NativeExecutionConnectorConfig;
+import com.facebook.presto.spark.execution.property.NativeExecutionCatalogProperties;
 import com.facebook.presto.spark.execution.property.NativeExecutionNodeConfig;
 import com.facebook.presto.spark.execution.property.NativeExecutionSystemConfig;
 import com.facebook.presto.spark.execution.property.PrestoSparkWorkerProperty;
@@ -88,7 +88,7 @@ private NativeExecutionProcessFactory createNativeExecutionProcessFactory()
         TaskId taskId = new TaskId("testid", 0, 0, 0, 0);
         ScheduledExecutorService errorScheduler = newScheduledThreadPool(4);
         PrestoSparkWorkerProperty workerProperty = new PrestoSparkWorkerProperty(
-                new NativeExecutionConnectorConfig(),
+                new NativeExecutionCatalogProperties(ImmutableMap.of()),
                 new NativeExecutionNodeConfig(),
                 new NativeExecutionSystemConfig(ImmutableMap.of()));
         NativeExecutionProcessFactory factory = new NativeExecutionProcessFactory(