[native] FusionNext - Infrastructure testing

Author: anandamideShakyan

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

@@ -68,6 +68,7 @@ public class ContainerQueryRunner
     private static final Logger logger = Logger.getLogger(ContainerQueryRunner.class.getName());
     private final GenericContainer<?> coordinator;
     private final List<GenericContainer<?>> workers = new ArrayList<>();
+    private GenericContainer<?> sidecar = null;
     private final int coordinatorPort;
     private final String catalog;
     private final String schema;
@@ -123,12 +124,84 @@ public ContainerQueryRunner(int coordinatorPort, String catalog, String schema,
         }
     }
 
+    public ContainerQueryRunner(int numberOfWorkers, boolean isNativeCluster, boolean isSidecarEnabled, boolean isSidecarDelayed)
+            throws IOException, InterruptedException
+    {
+        this.coordinatorPort = DEFAULT_COORDINATOR_PORT;
+        this.catalog = TPCH_CATALOG;
+        this.schema = TINY_SCHEMA;
+        this.numberOfWorkers = numberOfWorkers;
+        coordinator = createCoordinator(isNativeCluster, isSidecarEnabled);
+        coordinator.start();
+        // Delete the temporary files once the containers are started.
+        ContainerQueryRunnerUtils.deleteDirectory(BASE_DIR + "/testcontainers/coordinator");
+
+        if (isNativeCluster) {
+            for (int i = 0; i < numberOfWorkers; i++) {
+                workers.add(createNativeWorker(7777 + i, "native-worker-" + i, isSidecarEnabled, false));
+            }
+            workers.forEach(GenericContainer::start);
+            for (int i = 0; i < numberOfWorkers; i++) {
+                ContainerQueryRunnerUtils.deleteDirectory(BASE_DIR + "/testcontainers/native-worker-" + i);
+            }
+        }
+        else {
+            for (int i = 0; i < numberOfWorkers; i++) {
+                workers.add(createJavaWorker(7777 + i, "java-worker-" + i));
+            }
+            workers.forEach(GenericContainer::start);
+            for (int i = 0; i < numberOfWorkers; i++) {
+                ContainerQueryRunnerUtils.deleteDirectory(BASE_DIR + "/testcontainers/java-worker-" + i);
+            }
+        }
+
+        if (isSidecarEnabled) {
+            sidecar = createSidecar(7777 + numberOfWorkers, "sidecar");
+            if (isSidecarDelayed) {
+                Thread.sleep(10000);
+            }
+            sidecar.start();
+            ContainerQueryRunnerUtils.deleteDirectory(BASE_DIR + "/testcontainers/sidecar");
+        }
+
+        TimeUnit.SECONDS.sleep(5);
+        // Need some extra time for sidecar to register otherwise it throws sidecar not found error
+        if (isSidecarEnabled && !isSidecarDelayed) {
+            TimeUnit.SECONDS.sleep(60);
+        }
+
+        String dockerHostIp = coordinator.getHost();
+        logger.info("Presto UI is accessible at http://" + dockerHostIp + ":" + coordinator.getMappedPort(coordinatorPort));
+
+        String url = String.format("jdbc:presto://%s:%s/%s/%s?%s",
+                dockerHostIp,
+                coordinator.getMappedPort(coordinatorPort),
+                catalog,
+                schema,
+                "timeZoneId=UTC");
+
+        try {
+            connection = getConnection(url, "test", null);
+        }
+        catch (SQLException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     private GenericContainer<?> createCoordinator()
             throws IOException
+    {
+        return createCoordinator(true, false);
+    }
+    private GenericContainer<?> createCoordinator(boolean isNativeCluster, boolean isSidecarEnabled)
+            throws IOException
     {
         ContainerQueryRunnerUtils.createCoordinatorTpchProperties();
         ContainerQueryRunnerUtils.createCoordinatorTpcdsProperties();
-        ContainerQueryRunnerUtils.createCoordinatorConfigProperties(coordinatorPort);
+        ContainerQueryRunnerUtils.createCoordinatorConfigProperties(coordinatorPort, isNativeCluster, isSidecarEnabled);
+        if (isSidecarEnabled) {
+            ContainerQueryRunnerUtils.createCoordinatorSidecarProperties();
+        }
         ContainerQueryRunnerUtils.createCoordinatorJvmConfig();
         ContainerQueryRunnerUtils.createCoordinatorLogProperties();
         ContainerQueryRunnerUtils.createCoordinatorNodeProperties();
@@ -144,11 +217,40 @@ private GenericContainer<?> createCoordinator()
                 .withStartupTimeout(Duration.ofSeconds(Long.parseLong(CONTAINER_TIMEOUT)));
     }
 
-    private GenericContainer<?> createNativeWorker(int port, String nodeId)
+    private GenericContainer<?> createJavaWorker(int port, String nodeId)
             throws IOException
     {
-        ContainerQueryRunnerUtils.createNativeWorkerConfigProperties(coordinatorPort, nodeId);
+        ContainerQueryRunnerUtils.createJavaWorkerConfigProperties(port, coordinatorPort, nodeId);
         ContainerQueryRunnerUtils.createNativeWorkerTpchProperties(nodeId);
+        ContainerQueryRunnerUtils.createJavaEntryPointScript(nodeId);
+        ContainerQueryRunnerUtils.createNativeWorkerNodeProperties(nodeId);
+        return new GenericContainer<>(PRESTO_COORDINATOR_IMAGE)
+                .withExposedPorts(port)
+                .withNetwork(network)
+                .withNetworkAliases(nodeId)
+                .withCopyFileToContainer(MountableFile.forHostPath(BASE_DIR + "/testcontainers/" + nodeId + "/etc"), "/opt/presto-server/etc")
+                .withCopyFileToContainer(MountableFile.forHostPath(BASE_DIR + "/testcontainers/" + nodeId + "/entrypoint.sh"), "/opt/entrypoint.sh");
+    }
+
+    private GenericContainer<?> createSidecar(int port, String nodeId)
+            throws IOException
+    {
+        return createNativeWorker(port, nodeId, true, true);
+    }
+
+    private GenericContainer<?> createNativeWorker(int port, String nodeId)
+            throws IOException
+    {
+        return createNativeWorker(port, nodeId, false, false);
+    }
+
+    private GenericContainer<?> createNativeWorker(int port, String nodeId, boolean isSidecarEnabled, boolean isSidecarNode)
+            throws IOException
+    {
+        ContainerQueryRunnerUtils.createNativeWorkerConfigProperties(coordinatorPort, nodeId, isSidecarEnabled, isSidecarNode);
+        if (!isSidecarEnabled) {
+            ContainerQueryRunnerUtils.createNativeWorkerTpchProperties(nodeId);
+        }
         ContainerQueryRunnerUtils.createNativeWorkerEntryPointScript(nodeId);
         ContainerQueryRunnerUtils.createNativeWorkerNodeProperties(nodeId);
         ContainerQueryRunnerUtils.createNativeWorkerVeloxProperties(nodeId);
@@ -172,6 +274,9 @@ public void close()
         }
         coordinator.stop();
         workers.forEach(GenericContainer::stop);
+        if(sidecar != null) {
+            sidecar.stop();
+        }
     }
 
     @Override
@@ -312,7 +417,8 @@ public MaterializedResult execute(Session session, String sql)
             return ContainerQueryRunnerUtils.toMaterializedResult(resultSet);
         }
         catch (SQLException e) {
-            throw new RuntimeException("Error executing query: " + sql, e);
+            e.printStackTrace();
+            throw new RuntimeException("Error executing query: " + sql + "\n" + e.getMessage());
         }
     }
 }

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

@@ -13,26 +13,7 @@
  */
 package com.facebook.presto.nativeworker;
 
-import com.facebook.presto.common.type.ArrayType;
-import com.facebook.presto.common.type.BigintType;
-import com.facebook.presto.common.type.BooleanType;
-import com.facebook.presto.common.type.CharType;
-import com.facebook.presto.common.type.DateType;
-import com.facebook.presto.common.type.DecimalType;
-import com.facebook.presto.common.type.DoubleType;
-import com.facebook.presto.common.type.IntegerType;
-import com.facebook.presto.common.type.RealType;
-import com.facebook.presto.common.type.SmallintType;
-import com.facebook.presto.common.type.TimeType;
-import com.facebook.presto.common.type.TimestampType;
-import com.facebook.presto.common.type.TimestampWithTimeZoneType;
-import com.facebook.presto.common.type.TinyintType;
-import com.facebook.presto.common.type.Type;
-import com.facebook.presto.common.type.TypeSignature;
-import com.facebook.presto.common.type.TypeSignatureParameter;
-import com.facebook.presto.common.type.UnknownType;
-import com.facebook.presto.common.type.VarbinaryType;
-import com.facebook.presto.common.type.VarcharType;
+import com.facebook.presto.common.type.*;
 import com.facebook.presto.testing.MaterializedResult;
 import com.facebook.presto.testing.MaterializedRow;
 
@@ -45,11 +26,7 @@
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
+import java.util.*;
 
 public class ContainerQueryRunnerUtils
 {
@@ -82,20 +59,35 @@ public static void createNativeWorkerTpchProperties(String nodeId)
         createPropertiesFile("testcontainers/" + nodeId + "/etc/catalog/tpch.properties", properties);
     }
 
-    public static void createNativeWorkerConfigProperties(int coordinatorPort, String nodeId)
+    public static void createNativeWorkerConfigProperties(
+            int coordinatorPort,
+            String nodeId,
+            boolean isSidecarEnabled,
+            boolean isSidecarNode)
             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");
+
+        if (isSidecarEnabled) {
+            properties.setProperty("presto.default-namespace", "native.default");
+        }
+
+        properties.setProperty("native-sidecar", String.valueOf(isSidecarNode));
         createPropertiesFile("testcontainers/" + nodeId + "/etc/config.properties", properties);
     }
 
     public static void createCoordinatorConfigProperties(int port)
             throws IOException
+    {
+        createCoordinatorConfigProperties(port, true, false);
+    }
+
+    public static void createCoordinatorConfigProperties(int port, boolean isNativeCluster, boolean isSidecarEnabled)
+            throws IOException
     {
         Properties properties = new Properties();
         properties.setProperty("coordinator", "true");
@@ -104,16 +96,58 @@ public static void createCoordinatorConfigProperties(int port)
         properties.setProperty("http-server.http.port", Integer.toString(port));
         properties.setProperty("discovery-server.enabled", "true");
         properties.setProperty("discovery.uri", "http://presto-coordinator:" + port);
+        if (isSidecarEnabled) {
+            properties.setProperty("presto.default-namespace", "native.default");
+            properties.setProperty("coordinator-sidecar-enabled", "true");
+            properties.setProperty("exclude-invalid-worker-session-properties", "true");
+        }
 
-        // Get native worker system properties and add them to the coordinator properties
-        Map<String, String> nativeWorkerProperties = NativeQueryRunnerUtils.getNativeWorkerSystemProperties();
-        for (Map.Entry<String, String> entry : nativeWorkerProperties.entrySet()) {
-            properties.setProperty(entry.getKey(), entry.getValue());
+        if (isNativeCluster) {
+            // Get native worker system properties and add them to the coordinator properties
+            Map<String, String> nativeWorkerProperties = NativeQueryRunnerUtils.getNativeWorkerSystemProperties();
+            for (Map.Entry<String, String> entry : nativeWorkerProperties.entrySet()) {
+                properties.setProperty(entry.getKey(), entry.getValue());
+            }
         }
 
         createPropertiesFile("testcontainers/coordinator/etc/config.properties", properties);
     }
 
+    public static void createJavaWorkerConfigProperties(int port, int coordinatorPort, String nodeId)
+            throws IOException
+    {
+        Properties properties = new Properties();
+        properties.setProperty("coordinator", "false");
+        properties.setProperty("presto.version", "testversion");
+        properties.setProperty("node-scheduler.include-coordinator", "false");
+        properties.setProperty("http-server.http.port", Integer.toString(port));
+        properties.setProperty("discovery.uri", "http://presto-coordinator:" + coordinatorPort);
+        createPropertiesFile("testcontainers/" + nodeId + "/etc/config.properties", properties);
+    }
+
+    public static void createCoordinatorSidecarProperties()
+            throws IOException
+    {
+        Properties properties1 = new Properties();
+        properties1.setProperty("function-namespace-manager.name", "native");
+        properties1.setProperty("function-implementation-type", "CPP");
+        properties1.setProperty("supported-function-languages", "CPP");
+
+        createPropertiesFile("testcontainers/coordinator/etc/function-namespace/native.properties", properties1);
+
+        Properties properties2 = new Properties();
+        properties2.setProperty("session-property-provider.name", "native");
+        createPropertiesFile("testcontainers/coordinator/etc/session-property-provider/native.properties", properties2);
+
+        Properties properties3 = new Properties();
+        properties3.setProperty("type-manager.name", "native");
+        createPropertiesFile("testcontainers/coordinator/etc/type-managers/native.properties", properties3);
+
+        Properties properties4 = new Properties();
+        properties4.setProperty("plan-checker-provider.name", "native");
+        createPropertiesFile("testcontainers/coordinator/etc/plan-checker-providers/native.properties", properties4);
+    }
+
     public static void createCoordinatorJvmConfig()
             throws IOException
 
@@ -170,10 +204,7 @@ public static void createNativeWorkerVeloxProperties(String nodeId)
     public static void createCoordinatorEntryPointScript()
             throws IOException
     {
-        String scriptContent = "#!/bin/sh\n" +
-                "set -e\n" +
-                "$PRESTO_HOME/bin/launcher run\n";
-        createScriptFile("testcontainers/coordinator/entrypoint.sh", scriptContent);
+        createJavaEntryPointScript("coordinator");
     }
 
     public static void createNativeWorkerEntryPointScript(String nodeId)
@@ -185,6 +216,15 @@ public static void createNativeWorkerEntryPointScript(String nodeId)
         createScriptFile("testcontainers/" + nodeId + "/entrypoint.sh", scriptContent);
     }
 
+    public static void createJavaEntryPointScript(String nodeId)
+            throws IOException
+    {
+        String scriptContent = "#!/bin/sh\n" +
+                "set -e\n" +
+                "$PRESTO_HOME/bin/launcher run\n";
+        createScriptFile("testcontainers/" + nodeId + "/entrypoint.sh", scriptContent);
+    }
+
     public static void deleteDirectory(String directoryPath)
     {
         File directory = new File(directoryPath);
@@ -353,6 +393,9 @@ private static Type mapSqlTypeToType(int sqlType, String typeName)
             case java.sql.Types.VARBINARY:
             case java.sql.Types.LONGVARBINARY:
                 return VarbinaryType.VARBINARY;
+            case java.sql.Types.NULL:
+                // This happens in select fail() or similar cases
+                return VarcharType.createUnboundedVarcharType();
             case java.sql.Types.OTHER:
                 // Attempt to map based on type name
                 return mapSqlTypeNameToType(typeName);
@@ -363,6 +406,40 @@ private static Type mapSqlTypeToType(int sqlType, String typeName)
 
     private static Type mapSqlTypeNameToType(String typeName)
     {
+        String upperTypeName = typeName.toUpperCase(Locale.ENGLISH).trim();
+
+        // Handle flat ROW types like ROW(VARCHAR, INTEGER) or ROW(a VARCHAR, b INTEGER)
+        if (upperTypeName.startsWith("ROW(") && upperTypeName.endsWith(")")) {
+            String fieldsPart = upperTypeName.substring(4, upperTypeName.length() - 1).trim();
+
+            // Split fields by comma (no nested ROWs expected)
+            String[] fieldTokens = fieldsPart.split(",");
+
+            List<RowType.Field> rowFields = new ArrayList<>();
+            for (int i = 0; i < fieldTokens.length; i++) {
+                String token = fieldTokens[i].trim();
+
+                // Handle both: "a VARCHAR" and "VARCHAR"
+                String[] parts = token.split("\\s+", 2);
+                String fieldName;
+                String fieldType;
+
+                if (parts.length == 2) {
+                    fieldName = parts[0];
+                    fieldType = parts[1];
+                }
+                else {
+                    fieldName = "field" + i;
+                    fieldType = parts[0];
+                }
+
+                Type innerType = mapSqlTypeNameToType(fieldType);
+                rowFields.add(new RowType.Field(Optional.of(fieldName), innerType));
+            }
+
+            return RowType.from(rowFields);
+        }
+
         switch (typeName.toUpperCase()) {
             case "INT":
             case "INTEGER":