[incubator-streampipes] 01/02: [hotfix] Add initial builder for internal data lake queries

[ri...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

riemer pushed a commit to branch STREAMPIPES-545
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git

commit b0245c21dbe55955f2c042c4a92b7785f8264a7c
Author: Dominik Riemer <do...@gmail.com>
AuthorDate: Fri Jul 22 10:57:53 2022 +0200

    [hotfix] Add initial builder for internal data lake queries
---
 .../dataexplorer/sdk/DataLakeQueryBuilder.java     | 146 +++++++++++++++++++++
 .../dataexplorer/sdk/DataLakeQueryConstants.java   |  29 ++++
 .../v4/params/WhereStatementParams.java            |   2 +
 .../dataexplorer/v4/query/DataExplorerQueryV4.java |  13 ++
 4 files changed, 190 insertions(+)

diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/sdk/DataLakeQueryBuilder.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/sdk/DataLakeQueryBuilder.java
new file mode 100644
index 000000000..4c890843e
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/sdk/DataLakeQueryBuilder.java
@@ -0,0 +1,146 @@
+/*
+ * 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.streampipes.dataexplorer.sdk;
+
+import org.apache.streampipes.config.backend.BackendConfig;
+import org.apache.streampipes.dataexplorer.v4.params.ColumnFunction;
+import org.influxdb.dto.Query;
+import org.influxdb.querybuilder.SelectionQueryImpl;
+import org.influxdb.querybuilder.clauses.*;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.influxdb.querybuilder.BuiltQuery.QueryBuilder.select;
+
+public class DataLakeQueryBuilder {
+
+  private final String measurementId;
+  private final SelectionQueryImpl selectionQuery;
+  private final List<Clause> whereClauses;
+  private final List<Clause> groupByClauses;
+
+  public static DataLakeQueryBuilder create(String measurementId) {
+    return new DataLakeQueryBuilder(measurementId);
+  }
+
+  private DataLakeQueryBuilder(String measurementId) {
+    this.measurementId = measurementId;
+    this.selectionQuery = select();
+    this.whereClauses = new ArrayList<>();
+    this.groupByClauses = new ArrayList<>();
+  }
+
+  public DataLakeQueryBuilder withSimpleColumn(String columnName) {
+    this.selectionQuery.column(columnName);
+
+    return this;
+  }
+
+  public DataLakeQueryBuilder withAggregatedColumn(String columnName,
+                                                   ColumnFunction columnFunction,
+                                                   String targetName) {
+    String transformedColumnName = transform(columnName);
+    String transformedTargetName = transform(targetName);
+    if (columnFunction == ColumnFunction.COUNT) {
+      this.selectionQuery.count(transformedColumnName).as(transformedTargetName);
+    } else if (columnFunction == ColumnFunction.MEAN) {
+      this.selectionQuery.mean(transformedColumnName).as(transformedTargetName);
+    } else if (columnFunction == ColumnFunction.MIN) {
+      this.selectionQuery.min(transformedColumnName).as(transformedTargetName);
+    } else if (columnFunction == ColumnFunction.MAX) {
+      this.selectionQuery.max(transformedColumnName).as(transformedTargetName);
+    }
+
+    // TODO implement all column functions
+
+    return this;
+  }
+
+  public DataLakeQueryBuilder withStartTime(long startTime) {
+    this.whereClauses.add(new SimpleClause("time", ">=", startTime * 1000000));
+    return this;
+  }
+
+
+  public DataLakeQueryBuilder withEndTime(long endTime) {
+    this.whereClauses.add(new SimpleClause("time", "<=", endTime * 1000000));
+    return this;
+  }
+
+  public DataLakeQueryBuilder withTimeBoundary(long startTime,
+                                               long endTime) {
+    this.withStartTime(startTime);
+    this.withEndTime(endTime);
+
+    return this;
+  }
+
+  public DataLakeQueryBuilder withFilter(String field,
+                                         String operator,
+                                         Object value) {
+    this.whereClauses.add(new SimpleClause(field, operator, value));
+    return this;
+  }
+
+  public DataLakeQueryBuilder withExclusiveFilter(String field,
+                                                  String operator,
+                                                  List<?> values) {
+    List<ConjunctionClause> or = new ArrayList<>();
+    values.forEach(value -> {
+      or.add(new OrConjunction(new SimpleClause(transform(field), operator, value)));
+    });
+
+    NestedClause nestedClause = new NestedClause(or);
+    this.whereClauses.add(nestedClause);
+
+    return this;
+  }
+
+  public DataLakeQueryBuilder withGroupByTime(String timeInterval) {
+
+    this.groupByClauses.add(new RawTextClause("time(" + timeInterval + ")"));
+
+    return this;
+  }
+
+  private String transform(String column) {
+    return column.toLowerCase();
+  }
+
+  public Query build() {
+    var selectQuery = this.selectionQuery.from(BackendConfig.INSTANCE.getInfluxDatabaseName(), measurementId);
+    this.whereClauses.forEach(selectQuery::where);
+    this.groupByClauses.forEach(selectQuery::groupBy);
+
+    return selectQuery;
+  }
+
+  public static void main(String[] args) {
+    Query query = DataLakeQueryBuilder.create("abc")
+      .withSimpleColumn("test")
+      .withTimeBoundary(1, 2)
+      .withExclusiveFilter("text", "!=", Arrays.asList("a", "b"))
+      .withGroupByTime("1h")
+      .build();
+    System.out.println(query.getCommand());
+  }
+
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/sdk/DataLakeQueryConstants.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/sdk/DataLakeQueryConstants.java
new file mode 100644
index 000000000..e37389bce
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/sdk/DataLakeQueryConstants.java
@@ -0,0 +1,29 @@
+/*
+ * 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.streampipes.dataexplorer.sdk;
+
+public class DataLakeQueryConstants {
+
+  public static final String GE = ">=";
+  public static final String LE = "<=";
+  public static final String LT = "<";
+  public static final String GT = ">";
+  public static final String EQ = "=";
+  public static final String NEQ = "!=";
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/WhereStatementParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/WhereStatementParams.java
index e3dbd8497..a0af420b1 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/WhereStatementParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/WhereStatementParams.java
@@ -103,6 +103,8 @@ public class WhereStatementParams extends QueryParamsV4 {
   private String returnCondition(String inputCondition) {
     if (NumberUtils.isCreatable(inputCondition) || Boolean.parseBoolean(inputCondition)) {
       return inputCondition;
+    } else if (inputCondition.equals("\"\"")) {
+      return inputCondition;
     } else {
       return "'" + inputCondition + "'";
     }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/DataExplorerQueryV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/DataExplorerQueryV4.java
index c1a512169..973a6ae1b 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/DataExplorerQueryV4.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/DataExplorerQueryV4.java
@@ -44,6 +44,10 @@ public class DataExplorerQueryV4 {
 
     protected int maximumAmountOfEvents;
 
+    public DataExplorerQueryV4() {
+
+    }
+
     public DataExplorerQueryV4(Map<String, QueryParamsV4> params) {
         this.params = params;
         this.maximumAmountOfEvents = -1;
@@ -84,6 +88,15 @@ public class DataExplorerQueryV4 {
         return dataResult;
     }
 
+    public SpQueryResult executeQuery(Query query) {
+        InfluxDB influxDB = DataExplorerUtils.getInfluxDBClient();
+        QueryResult result = influxDB.query(query);
+        SpQueryResult dataResult = postQuery(result);
+        influxDB.close();
+
+        return dataResult;
+    }
+
     private double getAmountOfResults(QueryResult countQueryResult) {
         if (countQueryResult.getResults().get(0).getSeries() != null &&
                 countQueryResult.getResults().get(0).getSeries().get(0).getValues() != null) {