Update changes

Author: Joe-Abraham

presto-native-execution/src/test/java/com/facebook/presto/nativeworker/ContainerQueryRunner.java

@@ -37,7 +37,6 @@
 
 import java.io.IOException;
 import java.sql.Connection;
-import java.sql.DriverManager;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
@@ -51,6 +50,7 @@
 import java.util.logging.Logger;
 
 import static com.facebook.presto.testing.TestingSession.testSessionBuilder;
+import static java.sql.DriverManager.getConnection;
 
 public class ContainerQueryRunner
         implements QueryRunner
@@ -62,32 +62,31 @@ public class ContainerQueryRunner
     private static final String CLUSTER_SHUTDOWN_TIMEOUT = System.getProperty("clusterShutDownTimeout", "10");
     private static final String BASE_DIR = System.getProperty("user.dir");
     private static final int DEFAULT_COORDINATOR_PORT = 8080;
-    private static final int DEFAULT_FUNCTION_SERVER_PORT = 1122;
     private static final String TPCH_CATALOG = "tpch";
     private static final String TINY_SCHEMA = "tiny";
     private static final int DEFAULT_NUMBER_OF_WORKERS = 4;
     private static final Logger logger = Logger.getLogger(ContainerQueryRunner.class.getName());
     private final GenericContainer<?> coordinator;
     private final List<GenericContainer<?>> workers = new ArrayList<>();
     private final int coordinatorPort;
-    private final int functionServerPort;
     private final String catalog;
     private final String schema;
-    private Statement statement;
+    private final int numberOfWorkers;
+    private Connection connection;
 
     public ContainerQueryRunner()
             throws InterruptedException, IOException
     {
-        this(DEFAULT_COORDINATOR_PORT, DEFAULT_FUNCTION_SERVER_PORT, TPCH_CATALOG, TINY_SCHEMA, DEFAULT_NUMBER_OF_WORKERS);
+        this(DEFAULT_COORDINATOR_PORT, TPCH_CATALOG, TINY_SCHEMA, DEFAULT_NUMBER_OF_WORKERS);
     }
 
-    public ContainerQueryRunner(int coordinatorPort, int functionServerPort, String catalog, String schema, int numberOfWorkers)
+    public ContainerQueryRunner(int coordinatorPort, String catalog, String schema, int numberOfWorkers)
             throws InterruptedException, IOException
     {
         this.coordinatorPort = coordinatorPort;
-        this.functionServerPort = functionServerPort;
         this.catalog = catalog;
         this.schema = schema;
+        this.numberOfWorkers = numberOfWorkers;
 
         // The container details can be added as properties in VM options for testing in IntelliJ.
         coordinator = createCoordinator();
@@ -109,9 +108,7 @@ public ContainerQueryRunner(int coordinatorPort, int functionServerPort, String
                 "timeZoneId=UTC");
 
         try {
-            Connection connection = DriverManager.getConnection(url, "test", null);
-            statement = connection.createStatement();
-            statement.execute("set session remote_functions_enabled=true");
+            connection = getConnection(url, "test", null);
         }
         catch (SQLException e) {
             throw new RuntimeException(e);
@@ -133,14 +130,11 @@ private GenericContainer<?> createCoordinator()
         ContainerQueryRunnerUtils.createCoordinatorLogProperties();
         ContainerQueryRunnerUtils.createCoordinatorNodeProperties();
         ContainerQueryRunnerUtils.createCoordinatorEntryPointScript();
-        ContainerQueryRunnerUtils.createFunctionNamespaceRemoteProperties(functionServerPort);
-        ContainerQueryRunnerUtils.createFunctionServerConfigProperties(functionServerPort);
 
         return new GenericContainer<>(PRESTO_COORDINATOR_IMAGE)
                 .withExposedPorts(coordinatorPort)
                 .withNetwork(network).withNetworkAliases("presto-coordinator")
                 .withFileSystemBind(BASE_DIR + "/testcontainers/coordinator/etc", "/opt/presto-server/etc", BindMode.READ_WRITE)
-                .withFileSystemBind(BASE_DIR + "/testcontainers/coordinator/etc/function-server", "/opt/function-server/etc", BindMode.READ_ONLY)
                 .withFileSystemBind(BASE_DIR + "/testcontainers/coordinator/entrypoint.sh", "/opt/entrypoint.sh", BindMode.READ_ONLY)
                 .waitingFor(Wait.forLogMessage(".*======== SERVER STARTED ========.*", 1))
                 .withStartupTimeout(Duration.ofSeconds(Long.parseLong(CONTAINER_TIMEOUT)));
@@ -149,7 +143,7 @@ private GenericContainer<?> createCoordinator()
     private GenericContainer<?> createNativeWorker(int port, String nodeId)
             throws IOException
     {
-        ContainerQueryRunnerUtils.createNativeWorkerConfigProperties(coordinatorPort, functionServerPort, nodeId);
+        ContainerQueryRunnerUtils.createNativeWorkerConfigProperties(coordinatorPort, nodeId);
         ContainerQueryRunnerUtils.createNativeWorkerTpchProperties(nodeId);
         ContainerQueryRunnerUtils.createNativeWorkerEntryPointScript(nodeId);
         ContainerQueryRunnerUtils.createNativeWorkerNodeProperties(nodeId);
@@ -302,12 +296,12 @@ public Session getDefaultSession()
     public MaterializedResult execute(Session session, String sql)
     {
         try {
+            Statement statement = connection.createStatement();
             ResultSet resultSet = statement.executeQuery(sql);
-            return ContainerQueryRunnerUtils
-                    .toMaterializedResult(resultSet);
+            return ContainerQueryRunnerUtils.toMaterializedResult(resultSet);
         }
         catch (SQLException e) {
-            throw new RuntimeException(e);
+            throw new RuntimeException("Error executing query: " + sql, e);
         }
     }
 }

presto-native-execution/src/test/java/com/facebook/presto/nativeworker/ContainerQueryRunnerUtils.java

@@ -71,7 +71,19 @@ public static void createNativeWorkerTpchProperties(String nodeId)
         createPropertiesFile("testcontainers/" + nodeId + "/etc/catalog/tpch.properties", properties);
     }
 
-    public static void createNativeWorkerConfigProperties(int coordinatorPort, int functionServerPort, String nodeId)
+    public static void createNativeWorkerConfigProperties(int coordinatorPort, String nodeId)
+            throws IOException
+    {
+        Properties properties = new Properties();
+        properties.setProperty("presto.version", "testversion");
+        properties.setProperty("http-server.http.port", "7777");
+        properties.setProperty("discovery.uri", "http://presto-coordinator:" + coordinatorPort);
+        properties.setProperty("system-memory-gb", "2");
+        properties.setProperty("native.sidecar", "false");
+        createPropertiesFile("testcontainers/" + nodeId + "/etc/config.properties", properties);
+    }
+
+    public static void createNativeWorkerConfigPropertiesWithFunctionServer(int coordinatorPort, int functionServerPort, String nodeId)
             throws IOException
     {
         Properties properties = new Properties();
@@ -106,7 +118,24 @@ public static void createCoordinatorConfigProperties(int port)
         createPropertiesFile("testcontainers/coordinator/etc/config.properties", properties);
     }
 
-    public static void createFunctionNamespaceRemoteProperties(int functionServerPort)
+    public static void createFunctionNamespaceRemoteProperties()
+            throws IOException
+    {
+        Properties properties = new Properties();
+        properties.setProperty("function-namespace-manager.name", "rest");
+        properties.setProperty("supported-function-languages", "Java");
+        properties.setProperty("function-implementation-type", "REST");
+
+        String directoryPath = "testcontainers/function-namespace";
+        File directory = new File(directoryPath);
+        if (!directory.exists()) {
+            directory.mkdirs();
+        }
+
+        createPropertiesFile("testcontainers/coordinator/etc/function-namespace/remote.properties", properties);
+    }
+
+    public static void createFunctionNamespaceRemotePropertiesWithFunctionServer(int functionServerPort)
             throws IOException
     {
         Properties properties = new Properties();
@@ -241,10 +270,6 @@ public static void createPropertiesFile(String filePath, Properties properties)
             parentDir.mkdirs();
         }
 
-        if (file.exists()) {
-            throw new IOException("File exists: " + filePath);
-        }
-
         try (OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(file), StandardCharsets.UTF_8)) {
             for (String key : properties.stringPropertyNames()) {
                 writer.write(key + "=" + properties.getProperty(key) + "\n");
@@ -261,10 +286,6 @@ public static void createScriptFile(String filePath, String scriptContent)
             parentDir.mkdirs();
         }
 
-        if (file.exists()) {
-            throw new IOException("File exists: " + filePath);
-        }
-
         try (OutputStream output = new FileOutputStream(file);
                 OutputStreamWriter writer = new OutputStreamWriter(output, StandardCharsets.UTF_8)) {
             writer.write(scriptContent);