Add test for async data cache API on node

Author: agrawalreetika

presto-docs/src/main/sphinx/presto_cpp/features.rst

@@ -53,6 +53,30 @@ Other HTTP endpoints include:
 
 The request/response flow of Presto C++ is identical to Java workers. The tasks or new splits are registered via `TaskUpdateRequest`. Resource utilization and query progress are sent to the coordinator via task endpoints.
 
+* GET: /v1/operation/server/clearCache?type=memory: It clears the memory cache on worker node. Here is an example:
+
+  .. sourcecode:: http
+
+   curl -X GET "http://localhost:7777/v1/operation/server/clearCache?type=memory"
+
+   Cleared memory cache
+
+* GET: /v1/operation/server/clearCache?type=ssd: It clears the ssd cache on worker node. Here is an example:
+
+  .. sourcecode:: http
+
+   curl -X GET "http://localhost:7777/v1/operation/server/clearCache?type=ssd"
+
+   Cleared ssd cache
+
+* GET: /v1/operation/server/writeSsd: It writes data from memory cache to the ssd cache on worker node. Here is an example:
+
+  .. sourcecode:: http
+
+   curl -X GET "http://localhost:7777/v1/operation/server/writeSsd"
+
+   Succeeded write ssd cache
+
 Remote Function Execution
 -------------------------
 

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

@@ -34,7 +34,7 @@ public static void main(String[] args)
         javaQueryRunner.close();
 
         // Launch distributed runner.
-        DistributedQueryRunner queryRunner = (DistributedQueryRunner) PrestoNativeQueryRunnerUtils.createQueryRunner(false, false);
+        DistributedQueryRunner queryRunner = (DistributedQueryRunner) PrestoNativeQueryRunnerUtils.createQueryRunner(false, false, false, false);
         Thread.sleep(10);
         Logger log = Logger.get(DistributedQueryRunner.class);
         log.info("======== SERVER STARTED ========");

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

@@ -0,0 +1,72 @@
+/*
+ * 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.nativeworker;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public final class NativeApiEndpointUtils
+{
+    private NativeApiEndpointUtils() {}
+    public static Map<String, Long> fetchScalarLongMetrics(String serverAndPort, String apiEndpoint, String requestMethod) throws Exception
+    {
+        String apiUrl = serverAndPort + apiEndpoint;
+        URL url = new URL(apiUrl);
+        HttpURLConnection connection = (HttpURLConnection) url.openConnection();
+        connection.setRequestMethod(requestMethod);
+        connection.setConnectTimeout(5000);
+        connection.setReadTimeout(5000);
+
+        Map<String, Long> metrics = new HashMap<>();
+        Pattern metricPattern = Pattern.compile("^(\\w+)(?:\\{.*?\\})?\\s+(\\d+)$");
+
+        try (BufferedReader reader = new BufferedReader(new InputStreamReader(connection.getInputStream()))) {
+            String line;
+            while ((line = reader.readLine()) != null) {
+                Matcher matcher = metricPattern.matcher(line);
+                if (matcher.matches()) {
+                    String metricName = matcher.group(1);
+                    long metricValue = Long.parseLong(matcher.group(2));
+                    metrics.put(metricName, metricValue);
+                }
+            }
+        }
+        finally {
+            connection.disconnect();
+        }
+
+        return metrics;
+    }
+
+    public static int sendWorkerRequest(String serverAndPort, String endPoint)
+    {
+        try {
+            URL url = new URL(serverAndPort + endPoint);
+            HttpURLConnection connection = (HttpURLConnection) url.openConnection();
+            connection.setRequestMethod("GET");
+            connection.setRequestProperty("Accept", "application/json");
+            return connection.getResponseCode();
+        }
+        catch (Exception e) {
+            e.printStackTrace();
+            return 500;
+        }
+    }
+}

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

@@ -90,7 +90,7 @@ public class PrestoNativeQueryRunnerUtils
 
     private PrestoNativeQueryRunnerUtils() {}
 
-    public static QueryRunner createQueryRunner(boolean addStorageFormatToPath, boolean isCoordinatorSidecarEnabled)
+    public static QueryRunner createQueryRunner(boolean addStorageFormatToPath, boolean isCoordinatorSidecarEnabled, boolean enableRuntimeMetricsCollection, boolean enableSsdCache)
             throws Exception
     {
         int cacheMaxSize = 4096; // 4GB size cache
@@ -102,7 +102,9 @@ public static QueryRunner createQueryRunner(boolean addStorageFormatToPath, bool
                 cacheMaxSize,
                 DEFAULT_STORAGE_FORMAT,
                 addStorageFormatToPath,
-                isCoordinatorSidecarEnabled);
+                isCoordinatorSidecarEnabled,
+                enableRuntimeMetricsCollection,
+                enableSsdCache);
     }
 
     public static QueryRunner createQueryRunner(
@@ -112,7 +114,9 @@ public static QueryRunner createQueryRunner(
             int cacheMaxSize,
             String storageFormat,
             boolean addStorageFormatToPath,
-            boolean isCoordinatorSidecarEnabled)
+            boolean isCoordinatorSidecarEnabled,
+            boolean enableRuntimeMetricsCollection,
+            boolean enableSsdCache)
             throws Exception
     {
         QueryRunner defaultQueryRunner = createJavaQueryRunner(dataDirectory, storageFormat, addStorageFormatToPath);
@@ -123,7 +127,8 @@ public static QueryRunner createQueryRunner(
 
         defaultQueryRunner.close();
 
-        return createNativeQueryRunner(dataDirectory.get().toString(), prestoServerPath.get(), workerCount, cacheMaxSize, true, Optional.empty(), storageFormat, addStorageFormatToPath, false, isCoordinatorSidecarEnabled, false);
+        return createNativeQueryRunner(dataDirectory.get().toString(), prestoServerPath.get(), workerCount, cacheMaxSize, true, Optional.empty(),
+                storageFormat, addStorageFormatToPath, false, isCoordinatorSidecarEnabled, false, enableRuntimeMetricsCollection, enableSsdCache);
     }
 
     public static QueryRunner createJavaQueryRunner()
@@ -310,7 +315,7 @@ public static QueryRunner createNativeIcebergQueryRunner(
                 .setCreateTpchTables(false)
                 .setAddJmxPlugin(false)
                 .setNodeCount(OptionalInt.of(workerCount.orElse(4)))
-                .setExternalWorkerLauncher(getExternalWorkerLauncher("iceberg", prestoServerPath, cacheMaxSize, remoteFunctionServerUds, false, false))
+                .setExternalWorkerLauncher(getExternalWorkerLauncher("iceberg", prestoServerPath, cacheMaxSize, remoteFunctionServerUds, false, false, false, false))
                 .setAddStorageFormatToPath(addStorageFormatToPath)
                 .setDataDirectory(dataDirectory)
                 .setTpcdsProperties(getNativeWorkerTpcdsProperties())
@@ -328,7 +333,9 @@ public static QueryRunner createNativeQueryRunner(
             boolean addStorageFormatToPath,
             Boolean failOnNestedLoopJoin,
             boolean isCoordinatorSidecarEnabled,
-            boolean singleNodeExecutionEnabled)
+            boolean singleNodeExecutionEnabled,
+            boolean enableRuntimeMetricsCollection,
+            boolean enableSsdCache)
             throws Exception
     {
         // The property "hive.allow-drop-table" needs to be set to true because security is always "legacy" in NativeQueryRunner.
@@ -358,7 +365,8 @@ public static QueryRunner createNativeQueryRunner(
                 hiveProperties,
                 workerCount,
                 Optional.of(Paths.get(addStorageFormatToPath ? dataDirectory + "/" + storageFormat : dataDirectory)),
-                getExternalWorkerLauncher("hive", prestoServerPath, cacheMaxSize, remoteFunctionServerUds, failOnNestedLoopJoin, isCoordinatorSidecarEnabled),
+                getExternalWorkerLauncher("hive", prestoServerPath, cacheMaxSize, remoteFunctionServerUds, failOnNestedLoopJoin,
+                        isCoordinatorSidecarEnabled, enableRuntimeMetricsCollection, enableSsdCache),
                 getNativeWorkerTpcdsProperties());
     }
 
@@ -402,14 +410,14 @@ public static QueryRunner createNativeCteQueryRunner(boolean useThrift, String s
                 hiveProperties,
                 workerCount,
                 Optional.of(Paths.get(addStorageFormatToPath ? dataDirectory + "/" + storageFormat : dataDirectory)),
-                getExternalWorkerLauncher("hive", prestoServerPath, cacheMaxSize, Optional.empty(), false, false),
+                getExternalWorkerLauncher("hive", prestoServerPath, cacheMaxSize, Optional.empty(), false, false, false, false),
                 getNativeWorkerTpcdsProperties());
     }
 
     public static QueryRunner createNativeQueryRunner(String remoteFunctionServerUds)
             throws Exception
     {
-        return createNativeQueryRunner(false, DEFAULT_STORAGE_FORMAT, Optional.ofNullable(remoteFunctionServerUds), false, false, false);
+        return createNativeQueryRunner(false, DEFAULT_STORAGE_FORMAT, Optional.ofNullable(remoteFunctionServerUds), false, false, false, false, false);
     }
 
     public static QueryRunner createNativeQueryRunner(boolean useThrift)
@@ -421,13 +429,13 @@ public static QueryRunner createNativeQueryRunner(boolean useThrift)
     public static QueryRunner createNativeQueryRunner(boolean useThrift, boolean failOnNestedLoopJoin)
             throws Exception
     {
-        return createNativeQueryRunner(useThrift, DEFAULT_STORAGE_FORMAT, Optional.empty(), failOnNestedLoopJoin, false, false);
+        return createNativeQueryRunner(useThrift, DEFAULT_STORAGE_FORMAT, Optional.empty(), failOnNestedLoopJoin, false, false, false, false);
     }
 
     public static QueryRunner createNativeQueryRunner(boolean useThrift, String storageFormat)
             throws Exception
     {
-        return createNativeQueryRunner(useThrift, storageFormat, Optional.empty(), false, false, false);
+        return createNativeQueryRunner(useThrift, storageFormat, Optional.empty(), false, false, false, false, false);
     }
 
     public static QueryRunner createNativeQueryRunner(
@@ -436,7 +444,9 @@ public static QueryRunner createNativeQueryRunner(
             Optional<String> remoteFunctionServerUds,
             Boolean failOnNestedLoopJoin,
             boolean isCoordinatorSidecarEnabled,
-            boolean singleNodeExecutionEnabled)
+            boolean singleNodeExecutionEnabled,
+            boolean enableRuntimeMetricsCollection,
+            boolean enableSSDCache)
             throws Exception
     {
         int cacheMaxSize = 0;
@@ -452,7 +462,9 @@ public static QueryRunner createNativeQueryRunner(
                 true,
                 failOnNestedLoopJoin,
                 isCoordinatorSidecarEnabled,
-                singleNodeExecutionEnabled);
+                singleNodeExecutionEnabled,
+                enableRuntimeMetricsCollection,
+                enableSSDCache);
     }
 
     // Start the remote function server. Return the UDS path used to communicate with it.
@@ -499,7 +511,15 @@ public static NativeQueryRunnerParameters getNativeQueryRunnerParameters()
         return new NativeQueryRunnerParameters(prestoServerPath, dataDirectory, workerCount);
     }
 
-    public static Optional<BiFunction<Integer, URI, Process>> getExternalWorkerLauncher(String catalogName, String prestoServerPath, int cacheMaxSize, Optional<String> remoteFunctionServerUds, Boolean failOnNestedLoopJoin, boolean isCoordinatorSidecarEnabled)
+    public static Optional<BiFunction<Integer, URI, Process>> getExternalWorkerLauncher(
+            String catalogName,
+            String prestoServerPath,
+            int cacheMaxSize,
+            Optional<String> remoteFunctionServerUds,
+            Boolean failOnNestedLoopJoin,
+            boolean isCoordinatorSidecarEnabled,
+            boolean enableRuntimeMetricsCollection,
+            boolean enableSsdCache)
     {
         return
                 Optional.of((workerIndex, discoveryUri) -> {
@@ -521,6 +541,19 @@ public static Optional<BiFunction<Integer, URI, Process>> getExternalWorkerLaunc
                                     "presto.default-namespace=native.default%n", configProperties);
                         }
 
+                        if (enableRuntimeMetricsCollection) {
+                            configProperties = format("%s%n" +
+                                    "runtime-metrics-collection-enabled=true%n", configProperties);
+                        }
+
+                        if (enableSsdCache) {
+                            Path ssdCacheDir = Paths.get(tempDirectoryPath + "/velox-ssd-cache");
+                            Files.createDirectories(ssdCacheDir);
+                            configProperties = format("%s%n" +
+                                    "async-cache-ssd-gb=1%n" +
+                                    "async-cache-ssd-path=%s/%n", configProperties, ssdCacheDir);
+                        }
+
                         if (remoteFunctionServerUds.isPresent()) {
                             String jsonSignaturesPath = Resources.getResource(REMOTE_FUNCTION_JSON_SIGNATURES).getFile();
                             configProperties = format("%s%n" +
@@ -564,7 +597,7 @@ public static Optional<BiFunction<Integer, URI, Process>> getExternalWorkerLaunc
                                 format("connector.name=tpch%n").getBytes());
 
                         // Disable stack trace capturing as some queries (using TRY) generate a lot of exceptions.
-                        return new ProcessBuilder(prestoServerPath, "--logtostderr=1", "--v=1")
+                        return new ProcessBuilder(prestoServerPath, "--logtostderr=1", "--v=1", "--velox_ssd_odirect=false")
                                 .directory(tempDirectoryPath.toFile())
                                 .redirectErrorStream(true)
                                 .redirectOutput(ProcessBuilder.Redirect.to(tempDirectoryPath.resolve("worker." + workerIndex + ".out").toFile()))