[HUDI-4025] Add Presto query node to validate presto integration

Author: codope

hudi-integ-test/pom.xml

@@ -360,6 +360,11 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>com.facebook.presto</groupId>
+      <artifactId>presto-jdbc</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>org.awaitility</groupId>
       <artifactId>awaitility</artifactId>

hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/HoodieTestSuiteJob.java

@@ -62,6 +62,7 @@
 import java.util.Map;
 
 import static org.apache.hudi.common.table.HoodieTableConfig.ARCHIVELOG_FOLDER;
+import static org.apache.hudi.common.util.StringUtils.EMPTY_STRING;
 
 /**
  * This is the entry point for running a Hudi Test Suite. Although this class has similarities with {@link HoodieDeltaStreamer} this class does not extend it since do not want to create a dependency
@@ -310,5 +311,16 @@ public static class HoodieTestSuiteConfig extends HoodieDeltaStreamer.Config {
 
     @Parameter(names = {"--use-hudi-data-to-generate-updates"}, description = "Use data from hudi to generate updates for new batches ")
     public Boolean useHudiToGenerateUpdates = false;
+
+    @Parameter(names = {"--presto-jdbc-url"}, description = "Presto JDBC URL in the format jdbc:presto://<host>:<port>/<catalog>/<schema>  "
+        + "e.g. URL to connect to Presto running on localhost port 8080 with the catalog `hive` and the schema `sales`: "
+        + "jdbc:presto://localhost:8080/hive/sales")
+    public String prestoJdbcUrl = EMPTY_STRING;
+
+    @Parameter(names = {"--presto-jdbc-username"}, description = "Username to use for authentication")
+    public String prestoUsername = "test";
+
+    @Parameter(names = {"--presto-jdbc-password"}, description = "Password corresponding to the username to use for authentication")
+    public String prestoPassword;
   }
 }

hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/configuration/DeltaConfig.java

@@ -43,7 +43,7 @@ public class DeltaConfig implements Serializable {
   private final SerializableConfiguration configuration;
 
   public DeltaConfig(DeltaOutputMode deltaOutputMode, DeltaInputType deltaInputType,
-                     SerializableConfiguration configuration) {
+      SerializableConfiguration configuration) {
     this.deltaOutputMode = deltaOutputMode;
     this.deltaInputType = deltaInputType;
     this.configuration = configuration;
@@ -74,6 +74,8 @@ public static class Config {
     public static final String CHILDREN = "children";
     public static final String HIVE_QUERIES = "hive_queries";
     public static final String HIVE_PROPERTIES = "hive_props";
+    public static final String PRESTO_QUERIES = "presto_queries";
+    public static final String PRESTO_PROPERTIES = "presto_props";
     private static String NUM_RECORDS_INSERT = "num_records_insert";
     private static String NUM_RECORDS_UPSERT = "num_records_upsert";
     private static String NUM_RECORDS_DELETE = "num_records_delete";
@@ -278,7 +280,7 @@ public Option<String> getPreCombineField() {
 
     public Option<String> getPartitionField() {
       return !configsMap.containsKey(PARTITION_FIELD) ? Option.empty()
-              : Option.of(configsMap.get(PARTITION_FIELD).toString());
+          : Option.of(configsMap.get(PARTITION_FIELD).toString());
     }
 
     public String getMergeCondition() {
@@ -314,7 +316,7 @@ public Map<String, Object> getOtherConfigs() {
 
     public List<Pair<String, Integer>> getHiveQueries() {
       try {
-        return (List<Pair<String, Integer>>) this.configsMap.getOrDefault("hive_queries", new ArrayList<>());
+        return (List<Pair<String, Integer>>) this.configsMap.getOrDefault(HIVE_QUERIES, new ArrayList<>());
       } catch (Exception e) {
         throw new RuntimeException("unable to get hive queries from configs");
       }
@@ -328,6 +330,18 @@ public List<String> getHiveProperties() {
       return (List<String>) this.configsMap.getOrDefault(HIVE_PROPERTIES, new ArrayList<>());
     }
 
+    public List<String> getPrestoProperties() {
+      return (List<String>) this.configsMap.getOrDefault(PRESTO_PROPERTIES, new ArrayList<>());
+    }
+
+    public List<Pair<String, Integer>> getPrestoQueries() {
+      try {
+        return (List<Pair<String, Integer>>) this.configsMap.getOrDefault(PRESTO_QUERIES, new ArrayList<>());
+      } catch (Exception e) {
+        throw new RuntimeException("unable to get presto queries from configs");
+      }
+    }
+
     @Override
     public String toString() {
       try {
@@ -444,6 +458,16 @@ public Builder withHiveProperties(List<String> hiveProperties) {
         return this;
       }
 
+      public Builder withPrestoProperties(List<String> prestoProperties) {
+        this.configsMap.put(PRESTO_PROPERTIES, prestoProperties);
+        return this;
+      }
+
+      public Builder withPrestoQueryAndResults(List<Pair<String, Integer>> prestoQueries) {
+        this.configsMap.put(PRESTO_QUERIES, prestoQueries);
+        return this;
+      }
+
       public Builder withConfigsMap(Map<String, Object> configsMap) {
         this.configsMap = configsMap;
         return this;

hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/dag/DagUtils.java

@@ -53,7 +53,11 @@
 import java.util.stream.Collectors;
 
 import static org.apache.hudi.integ.testsuite.configuration.DeltaConfig.Config.CONFIG_NAME;
+import static org.apache.hudi.integ.testsuite.configuration.DeltaConfig.Config.HIVE_PROPERTIES;
+import static org.apache.hudi.integ.testsuite.configuration.DeltaConfig.Config.HIVE_QUERIES;
 import static org.apache.hudi.integ.testsuite.configuration.DeltaConfig.Config.NO_DEPENDENCY_VALUE;
+import static org.apache.hudi.integ.testsuite.configuration.DeltaConfig.Config.PRESTO_PROPERTIES;
+import static org.apache.hudi.integ.testsuite.configuration.DeltaConfig.Config.PRESTO_QUERIES;
 
 /**
  * Utility class to SerDe workflow dag.
@@ -172,7 +176,8 @@ private static DagNode convertJsonToDagNode(JsonNode node, String type, String n
       DeltaConfig.Config config = DeltaConfig.Config.newBuilder().withConfigsMap(convertJsonNodeToMap(node))
           .withName(name).build();
       return (DagNode) ReflectionUtils.loadClass(generateFQN(type), config);
-    } catch (ClassNotFoundException e) {
+    }
+    catch (ClassNotFoundException e) {
       throw new RuntimeException(e);
     }
   }
@@ -192,11 +197,17 @@ private static Map<String, Object> convertJsonNodeToMap(JsonNode node) {
     while (itr.hasNext()) {
       Entry<String, JsonNode> entry = itr.next();
       switch (entry.getKey()) {
-        case DeltaConfig.Config.HIVE_QUERIES:
-          configsMap.put(DeltaConfig.Config.HIVE_QUERIES, getHiveQueries(entry));
+        case HIVE_QUERIES:
+          configsMap.put(HIVE_QUERIES, getQueries(entry));
+          break;
+        case HIVE_PROPERTIES:
+          configsMap.put(HIVE_PROPERTIES, getQuerySessionProperties(entry));
           break;
-        case DeltaConfig.Config.HIVE_PROPERTIES:
-          configsMap.put(DeltaConfig.Config.HIVE_PROPERTIES, getProperties(entry));
+        case PRESTO_QUERIES:
+          configsMap.put(PRESTO_QUERIES, getQueries(entry));
+          break;
+        case PRESTO_PROPERTIES:
+          configsMap.put(PRESTO_PROPERTIES, getQuerySessionProperties(entry));
           break;
         default:
           configsMap.put(entry.getKey(), getValue(entry.getValue()));
@@ -206,25 +217,27 @@ private static Map<String, Object> convertJsonNodeToMap(JsonNode node) {
     return configsMap;
   }
 
-  private static List<Pair<String, Integer>> getHiveQueries(Entry<String, JsonNode> entry) {
+  private static List<Pair<String, Integer>> getQueries(Entry<String, JsonNode> entry) {
     List<Pair<String, Integer>> queries = new ArrayList<>();
     try {
       List<JsonNode> flattened = new ArrayList<>();
       flattened.add(entry.getValue());
-      queries = (List<Pair<String, Integer>>)getHiveQueryMapper().readValue(flattened.toString(), List.class);
-    } catch (Exception e) {
+      queries = (List<Pair<String, Integer>>) getQueryMapper().readValue(flattened.toString(), List.class);
+    }
+    catch (Exception e) {
       e.printStackTrace();
     }
     return queries;
   }
 
-  private static List<String> getProperties(Entry<String, JsonNode> entry) {
+  private static List<String> getQuerySessionProperties(Entry<String, JsonNode> entry) {
     List<String> properties = new ArrayList<>();
     try {
       List<JsonNode> flattened = new ArrayList<>();
       flattened.add(entry.getValue());
-      properties = (List<String>)getHivePropertyMapper().readValue(flattened.toString(), List.class);
-    } catch (Exception e) {
+      properties = (List<String>) getQueryEnginePropertyMapper().readValue(flattened.toString(), List.class);
+    }
+    catch (Exception e) {
       e.printStackTrace();
     }
     return properties;
@@ -233,15 +246,20 @@ private static List<String> getProperties(Entry<String, JsonNode> entry) {
   private static Object getValue(JsonNode node) {
     if (node.isInt()) {
       return node.asInt();
-    } else if (node.isLong()) {
+    }
+    else if (node.isLong()) {
       return node.asLong();
-    } else if (node.isShort()) {
+    }
+    else if (node.isShort()) {
       return node.asInt();
-    } else if (node.isBoolean()) {
+    }
+    else if (node.isBoolean()) {
       return node.asBoolean();
-    } else if (node.isDouble()) {
+    }
+    else if (node.isDouble()) {
       return node.asDouble();
-    } else if (node.isFloat()) {
+    }
+    else if (node.isFloat()) {
       return node.asDouble();
     }
     return node.textValue();
@@ -254,13 +272,20 @@ private static JsonNode createJsonNode(DagNode node, String type) throws IOExcep
     while (itr.hasNext()) {
       Entry<String, JsonNode> entry = itr.next();
       switch (entry.getKey()) {
-        case DeltaConfig.Config.HIVE_QUERIES:
-          ((ObjectNode) configNode).put(DeltaConfig.Config.HIVE_QUERIES,
-              MAPPER.readTree(getHiveQueryMapper().writeValueAsString(node.getConfig().getHiveQueries())));
+        case HIVE_QUERIES:
+          ((ObjectNode) configNode).put(HIVE_QUERIES,
+              MAPPER.readTree(getQueryMapper().writeValueAsString(node.getConfig().getHiveQueries())));
+          break;
+        case HIVE_PROPERTIES:
+          ((ObjectNode) configNode).put(HIVE_PROPERTIES,
+              MAPPER.readTree(getQueryEnginePropertyMapper().writeValueAsString(node.getConfig().getHiveProperties())));
+        case PRESTO_QUERIES:
+          ((ObjectNode) configNode).put(PRESTO_QUERIES,
+              MAPPER.readTree(getQueryMapper().writeValueAsString(node.getConfig().getHiveQueries())));
           break;
-        case DeltaConfig.Config.HIVE_PROPERTIES:
-          ((ObjectNode) configNode).put(DeltaConfig.Config.HIVE_PROPERTIES,
-              MAPPER.readTree(getHivePropertyMapper().writeValueAsString(node.getConfig().getHiveProperties())));
+        case PRESTO_PROPERTIES:
+          ((ObjectNode) configNode).put(PRESTO_PROPERTIES,
+              MAPPER.readTree(getQueryEnginePropertyMapper().writeValueAsString(node.getConfig().getHiveProperties())));
           break;
         default:
           break;
@@ -293,21 +318,22 @@ public static String toString(InputStream inputStream) throws IOException {
     return result.toString("utf-8");
   }
 
-  private static ObjectMapper getHiveQueryMapper() {
+  private static ObjectMapper getQueryMapper() {
     SimpleModule module = new SimpleModule();
     ObjectMapper queryMapper = new ObjectMapper();
-    module.addSerializer(List.class, new HiveQuerySerializer());
-    module.addDeserializer(List.class, new HiveQueryDeserializer());
+    module.addSerializer(List.class, new QuerySerializer());
+    module.addDeserializer(List.class, new QueryDeserializer());
     queryMapper.registerModule(module);
     return queryMapper;
   }
 
-  private static final class HiveQuerySerializer extends JsonSerializer<List> {
+  private static final class QuerySerializer extends JsonSerializer<List> {
     Integer index = 0;
+
     @Override
     public void serialize(List pairs, JsonGenerator gen, SerializerProvider serializers) throws IOException {
       gen.writeStartObject();
-      for (Pair pair : (List<Pair>)pairs) {
+      for (Pair pair : (List<Pair>) pairs) {
         gen.writeStringField("query" + index, pair.getLeft().toString());
         gen.writeNumberField("result" + index, Integer.parseInt(pair.getRight().toString()));
         index++;
@@ -316,7 +342,7 @@ public void serialize(List pairs, JsonGenerator gen, SerializerProvider serializ
     }
   }
 
-  private static final class HiveQueryDeserializer extends JsonDeserializer<List> {
+  private static final class QueryDeserializer extends JsonDeserializer<List> {
     @Override
     public List deserialize(JsonParser parser, DeserializationContext context) throws IOException {
       List<Pair<String, Integer>> pairs = new ArrayList<>();
@@ -334,7 +360,8 @@ public List deserialize(JsonParser parser, DeserializationContext context) throw
 
           if (fieldName.contains("query")) {
             query = parser.getValueAsString();
-          } else if (fieldName.contains("result")) {
+          }
+          else if (fieldName.contains("result")) {
             result = parser.getValueAsInt();
             pairs.add(Pair.of(query, result));
           }
@@ -344,29 +371,30 @@ public List deserialize(JsonParser parser, DeserializationContext context) throw
     }
   }
 
-  private static ObjectMapper getHivePropertyMapper() {
+  private static ObjectMapper getQueryEnginePropertyMapper() {
     SimpleModule module = new SimpleModule();
     ObjectMapper propMapper = new ObjectMapper();
-    module.addSerializer(List.class, new HivePropertySerializer());
-    module.addDeserializer(List.class, new HivePropertyDeserializer());
+    module.addSerializer(List.class, new QueryEnginePropertySerializer());
+    module.addDeserializer(List.class, new QueryEnginePropertyDeserializer());
     propMapper.registerModule(module);
     return propMapper;
   }
 
-  private static final class HivePropertySerializer extends JsonSerializer<List> {
+  private static final class QueryEnginePropertySerializer extends JsonSerializer<List> {
     Integer index = 0;
+
     @Override
     public void serialize(List props, JsonGenerator gen, SerializerProvider serializers) throws IOException {
       gen.writeStartObject();
-      for (String prop : (List<String>)props) {
+      for (String prop : (List<String>) props) {
         gen.writeStringField("prop" + index, prop);
         index++;
       }
       gen.writeEndObject();
     }
   }
 
-  private static final class HivePropertyDeserializer extends JsonDeserializer<List> {
+  private static final class QueryEnginePropertyDeserializer extends JsonDeserializer<List> {
     @Override
     public List deserialize(JsonParser parser, DeserializationContext context) throws IOException {
       List<String> props = new ArrayList<>();

hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/dag/nodes/BaseQueryNode.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hudi.integ.testsuite.dag.nodes;
+
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+
+public abstract class BaseQueryNode extends DagNode<Boolean> {
+
+  public void setSessionProperties(List<String> properties, Statement stmt) throws SQLException {
+    for (String prop : properties) {
+      executeStatement(prop, stmt);
+    }
+  }
+
+  public void executeAndValidateQueries(List<Pair<String, Integer>> queriesWithResult, Statement stmt) throws SQLException {
+    for (Pair<String, Integer> queryAndResult : queriesWithResult) {
+      log.info("Running {}", queryAndResult.getLeft());
+      ResultSet res = stmt.executeQuery(queryAndResult.getLeft());
+      if (!res.next()) {
+        log.info("res.next() was False - typically this means the query returned no rows.");
+        assert 0 == queryAndResult.getRight();
+      }
+      else {
+        Integer result = res.getInt(1);
+        if (!queryAndResult.getRight().equals(result)) {
+          throw new AssertionError(
+              "QUERY: " + queryAndResult.getLeft()
+                  + " | EXPECTED RESULT = " + queryAndResult.getRight()
+                  + " | ACTUAL RESULT = " + result
+          );
+        }
+      }
+      log.info("Successfully validated query!");
+    }
+  }
+
+  private void executeStatement(String query, Statement stmt) throws SQLException {
+    log.info("Executing statement {}", stmt.toString());
+    stmt.execute(query);
+  }
+}