feat(#4188): Add support for nested filters in data explorer

streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataLakeQueryBuilder.java

@@ -23,6 +23,7 @@
 import org.apache.streampipes.model.datalake.AggregationFunction;
 import org.apache.streampipes.model.datalake.DataLakeQueryOrdering;
 import org.apache.streampipes.model.datalake.FilterCondition;
+import org.apache.streampipes.model.datalake.FilterExpressionGroup;
 
 import java.util.List;
 
@@ -65,6 +66,8 @@ IDataLakeQueryBuilder<T> withInclusiveFilter(String field,
 
   IDataLakeQueryBuilder<T> withInclusiveFilter(List<FilterCondition> filterConditions);
 
+  IDataLakeQueryBuilder<T> withFilterExpression(FilterExpressionGroup filterExpression);
+
   IDataLakeQueryBuilder<T> withGroupByTime(String timeInterval);
 
   IDataLakeQueryBuilder<T> withGroupByTime(String timeInterval,

streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java

@@ -24,6 +24,10 @@
 import org.apache.streampipes.model.datalake.AggregationFunction;
 import org.apache.streampipes.model.datalake.DataLakeQueryOrdering;
 import org.apache.streampipes.model.datalake.FilterCondition;
+import org.apache.streampipes.model.datalake.FilterExpressionCondition;
+import org.apache.streampipes.model.datalake.FilterExpressionGroup;
+import org.apache.streampipes.model.datalake.FilterExpressionLogicalOperator;
+import org.apache.streampipes.model.datalake.FilterExpressionNode;
 
 import org.influxdb.dto.Query;
 import org.influxdb.querybuilder.Ordering;
@@ -177,6 +181,17 @@ public IDataLakeQueryBuilder<Query> withInclusiveFilter(List<FilterCondition> fi
     return this;
   }
 
+  @Override
+  public IDataLakeQueryBuilder<Query> withFilterExpression(FilterExpressionGroup filterExpression) {
+    if (Objects.nonNull(filterExpression)
+        && Objects.nonNull(filterExpression.children())
+        && !filterExpression.children().isEmpty()) {
+      this.whereClauses.add(toNestedClause(filterExpression));
+    }
+
+    return this;
+  }
+
   @Override
   public DataLakeInfluxQueryBuilder withGroupByTime(String timeInterval) {
 
@@ -276,6 +291,32 @@ private void addNestedWhereClause(List<ConjunctionClause> clauses) {
     this.whereClauses.add(nestedClause);
   }
 
+  private NestedClause toNestedClause(FilterExpressionGroup group) {
+    List<ConjunctionClause> conjunctionClauses = new ArrayList<>();
+    var conjunction = Objects.nonNull(group.operator()) ? group.operator() : FilterExpressionLogicalOperator.AND;
+    group.children().forEach(node -> conjunctionClauses.add(makeConjunction(node, conjunction)));
+    return new NestedClause(conjunctionClauses);
+  }
+
+  private ConjunctionClause makeConjunction(FilterExpressionNode node, FilterExpressionLogicalOperator conjunction) {
+    Clause clause = toClause(node);
+    return conjunction == FilterExpressionLogicalOperator.OR
+        ? new OrConjunction(clause)
+        : new AndConjunction(clause);
+  }
+
+  private Clause toClause(FilterExpressionNode node) {
+    if (node instanceof FilterExpressionCondition condition) {
+      return new SimpleClause(condition.field(), condition.operator(), condition.condition());
+    }
+
+    if (node instanceof FilterExpressionGroup group) {
+      return toNestedClause(group);
+    }
+
+    throw new IllegalArgumentException("Unsupported filter expression node: " + node);
+  }
+
   private String escapeIndex(String index) {
     return "\"" + index + "\"";
   }

streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/SelectQueryParamsTest.java

@@ -206,4 +206,87 @@ public void testGroupByTags() {
         + " time > 1000000) GROUP BY sensorId,sensorId2;", query);
   }
 
+  @Test
+  public void testFilterExpressionOr() {
+    var params = ProvidedQueryParameterBuilder.create("abc")
+        .withStartDate(1)
+        .withEndDate(2)
+        .withSimpleColumns(Arrays.asList("p1", "p2"))
+        .withFilterExpression(
+            "{\"type\":\"group\",\"operator\":\"OR\",\"children\":["
+                + "{\"type\":\"condition\",\"field\":\"p1\",\"operator\":\"=\",\"condition\":1},"
+                + "{\"type\":\"condition\",\"field\":\"p2\",\"operator\":\"=\",\"condition\":2}"
+                + "]}")
+        .build();
+
+    SelectQueryParams qp = ProvidedRestQueryParamConverter.getSelectQueryParams(params);
+
+    String query = qp.toQuery(DataLakeInfluxQueryBuilder.create("abc")).getCommand();
+
+    assertEquals("SELECT p1,p2 FROM \"abc\" WHERE (time < 2000000 AND time > 1000000) "
+        + "AND (p1 = 1 OR p2 = 2);", query);
+  }
+
+  @Test
+  public void testFilterExpressionTakesPrecedenceOverLegacyFilter() {
+    var params = ProvidedQueryParameterBuilder.create("abc")
+        .withStartDate(1)
+        .withEndDate(2)
+        .withSimpleColumns(Arrays.asList("p1", "p2"))
+        .withFilter("[p1;=;1]")
+        .withFilterExpression(
+            "{\"type\":\"group\",\"operator\":\"OR\",\"children\":["
+                + "{\"type\":\"condition\",\"field\":\"p1\",\"operator\":\"=\",\"condition\":1},"
+                + "{\"type\":\"condition\",\"field\":\"p2\",\"operator\":\"=\",\"condition\":2}"
+                + "]}")
+        .build();
+
+    SelectQueryParams qp = ProvidedRestQueryParamConverter.getSelectQueryParams(params);
+
+    String query = qp.toQuery(DataLakeInfluxQueryBuilder.create("abc")).getCommand();
+
+    assertEquals("SELECT p1,p2 FROM \"abc\" WHERE (time < 2000000 AND time > 1000000) "
+        + "AND (p1 = 1 OR p2 = 2);", query);
+  }
+
+  @Test
+  public void testFilterExpressionStringBooleanIsParsedAsBoolean() {
+    var params = ProvidedQueryParameterBuilder.create("abc")
+        .withStartDate(1)
+        .withEndDate(2)
+        .withSimpleColumns(Arrays.asList("p1", "p2"))
+        .withFilterExpression(
+            "{\"type\":\"group\",\"operator\":\"AND\",\"children\":["
+                + "{\"type\":\"condition\",\"field\":\"p1\",\"operator\":\"=\",\"condition\":\"true\"}"
+                + "]}")
+        .build();
+
+    SelectQueryParams qp = ProvidedRestQueryParamConverter.getSelectQueryParams(params);
+
+    String query = qp.toQuery(DataLakeInfluxQueryBuilder.create("abc")).getCommand();
+
+    assertEquals("SELECT p1,p2 FROM \"abc\" WHERE (time < 2000000 AND time > 1000000) "
+        + "AND (p1 = true);", query);
+  }
+
+  @Test
+  public void testFilterExpressionStringNumberIsParsedAsNumber() {
+    var params = ProvidedQueryParameterBuilder.create("abc")
+        .withStartDate(1)
+        .withEndDate(2)
+        .withSimpleColumns(Arrays.asList("p1", "p2"))
+        .withFilterExpression(
+            "{\"type\":\"group\",\"operator\":\"AND\",\"children\":["
+                + "{\"type\":\"condition\",\"field\":\"p1\",\"operator\":\"=\",\"condition\":\"1\"}"
+                + "]}")
+        .build();
+
+    SelectQueryParams qp = ProvidedRestQueryParamConverter.getSelectQueryParams(params);
+
+    String query = qp.toQuery(DataLakeInfluxQueryBuilder.create("abc")).getCommand();
+
+    assertEquals("SELECT p1,p2 FROM \"abc\" WHERE (time < 2000000 AND time > 1000000) "
+        + "AND (p1 = 1.0);", query);
+  }
+
 }

streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/utils/ProvidedQueryParameterBuilder.java

@@ -74,6 +74,12 @@ public ProvidedQueryParameterBuilder withFilter(String filter) {
     return this;
   }
 
+  public ProvidedQueryParameterBuilder withFilterExpression(String filterExpression) {
+    this.queryParams.put(SupportedRestQueryParams.QP_FILTER_EXPRESSION, filterExpression);
+
+    return this;
+  }
+
   public ProvidedQueryParameterBuilder withPage(int page) {
     this.queryParams.put(SupportedRestQueryParams.QP_PAGE, String.valueOf(page));
 

streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java

@@ -26,9 +26,13 @@
 import org.apache.streampipes.dataexplorer.param.model.OrderByClauseParams;
 import org.apache.streampipes.dataexplorer.param.model.SelectClauseParams;
 import org.apache.streampipes.dataexplorer.param.model.WhereClauseParams;
+import org.apache.streampipes.model.datalake.FilterExpressionGroup;
 import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
 import org.apache.streampipes.model.datalake.param.SupportedRestQueryParams;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -40,6 +44,7 @@ public class ProvidedRestQueryParamConverter {
   public static final String BRACKET_CLOSE = "\\]";
 
   public static final String ORDER_DESCENDING = "DESC";
+  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
   public static SelectQueryParams getSelectQueryParams(ProvidedRestQueryParams params) {
     SelectQueryParams queryParameters = new SelectQueryParams(params.getMeasurementId());
@@ -54,15 +59,20 @@ public static SelectQueryParams getSelectQueryParams(ProvidedRestQueryParams par
           params.getAsString(SupportedRestQueryParams.QP_AGGREGATION_FUNCTION)));
     }
 
-    String filterConditions = params.getAsString(SupportedRestQueryParams.QP_FILTER);
+    String filterExpression = params.getAsString(SupportedRestQueryParams.QP_FILTER_EXPRESSION);
+    FilterExpressionGroup parsedFilterExpression = parseFilterExpression(filterExpression);
+    String filterConditions = parsedFilterExpression == null
+        ? params.getAsString(SupportedRestQueryParams.QP_FILTER)
+        : null;
 
     if (hasTimeParams(params)) {
       queryParameters.withWhereParams(WhereClauseParams.from(
           params.getAsLong(SupportedRestQueryParams.QP_START_DATE),
           params.getAsLong(SupportedRestQueryParams.QP_END_DATE),
-          filterConditions));
-    } else if (filterConditions != null) {
-      queryParameters.withWhereParams(WhereClauseParams.from(filterConditions));
+          filterConditions,
+          parsedFilterExpression));
+    } else if (filterConditions != null || parsedFilterExpression != null) {
+      queryParameters.withWhereParams(WhereClauseParams.from(filterConditions, parsedFilterExpression));
     }
 
     if (params.has(SupportedRestQueryParams.QP_TIME_INTERVAL)) {
@@ -133,4 +143,16 @@ public static String[] buildSingleCondition(String queryPart) {
         .replaceAll(BRACKET_CLOSE, "")
         .split(";");
   }
+
+  private static FilterExpressionGroup parseFilterExpression(String filterExpression) {
+    if (filterExpression == null || filterExpression.isBlank()) {
+      return null;
+    }
+
+    try {
+      return OBJECT_MAPPER.readValue(filterExpression, FilterExpressionGroup.class);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Invalid filter expression provided", e);
+    }
+  }
 }

streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/WhereClauseParams.java

@@ -21,6 +21,9 @@
 import org.apache.streampipes.dataexplorer.api.IQueryStatement;
 import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParamConverter;
 import org.apache.streampipes.model.datalake.FilterCondition;
+import org.apache.streampipes.model.datalake.FilterExpressionCondition;
+import org.apache.streampipes.model.datalake.FilterExpressionGroup;
+import org.apache.streampipes.model.datalake.FilterExpressionNode;
 
 import org.apache.commons.lang3.math.NumberUtils;
 
@@ -33,13 +36,17 @@ public class WhereClauseParams implements IQueryStatement {
   private static final String LT = "<";
 
   private final List<FilterCondition> filterConditions;
+  private final FilterExpressionGroup filterExpression;
 
   private WhereClauseParams(
       Long startTime,
       Long endTime,
-      String whereConditions
+      String whereConditions,
+      FilterExpressionGroup filterExpression
   ) {
-    this(startTime, endTime);
+    this.filterConditions = new ArrayList<>();
+    this.filterExpression = normalizeFilterExpression(filterExpression);
+    this.buildTimeConditions(startTime, endTime);
     if (whereConditions != null) {
       buildConditions(whereConditions);
     }
@@ -50,11 +57,13 @@ private WhereClauseParams(
       Long endTime
   ) {
     this.filterConditions = new ArrayList<>();
+    this.filterExpression = null;
     this.buildTimeConditions(startTime, endTime);
   }
 
-  private WhereClauseParams(String whereConditions) {
+  private WhereClauseParams(String whereConditions, FilterExpressionGroup filterExpression) {
     this.filterConditions = new ArrayList<>();
+    this.filterExpression = normalizeFilterExpression(filterExpression);
     if (whereConditions != null) {
       buildConditions(whereConditions);
     }
@@ -68,15 +77,31 @@ public static WhereClauseParams from(
   }
 
   public static WhereClauseParams from(String whereConditions) {
-    return new WhereClauseParams(whereConditions);
+    return new WhereClauseParams(whereConditions, null);
+  }
+
+  public static WhereClauseParams from(
+      String whereConditions,
+      FilterExpressionGroup filterExpression
+  ) {
+    return new WhereClauseParams(whereConditions, filterExpression);
   }
 
   public static WhereClauseParams from(
       Long startTime,
       Long endTime,
       String whereConditions
   ) {
-    return new WhereClauseParams(startTime, endTime, whereConditions);
+    return new WhereClauseParams(startTime, endTime, whereConditions, null);
+  }
+
+  public static WhereClauseParams from(
+      Long startTime,
+      Long endTime,
+      String whereConditions,
+      FilterExpressionGroup filterExpression
+  ) {
+    return new WhereClauseParams(startTime, endTime, whereConditions, filterExpression);
   }
 
   private void buildTimeConditions(
@@ -119,6 +144,35 @@ private Object returnCondition(String inputCondition) {
     }
   }
 
+  private FilterExpressionGroup normalizeFilterExpression(FilterExpressionGroup input) {
+    if (input == null) {
+      return null;
+    }
+
+    var children = input.children().stream()
+        .map(this::normalizeFilterExpressionNode)
+        .toList();
+
+    return new FilterExpressionGroup(input.operator(), children);
+  }
+
+  private FilterExpressionNode normalizeFilterExpressionNode(FilterExpressionNode node) {
+    if (node instanceof FilterExpressionGroup group) {
+      return normalizeFilterExpression(group);
+    }
+
+    if (node instanceof FilterExpressionCondition condition) {
+      Object normalizedCondition = condition.condition();
+      if (normalizedCondition instanceof String conditionAsString) {
+        normalizedCondition = returnCondition(conditionAsString);
+      }
+
+      return new FilterExpressionCondition(condition.field(), condition.operator(), normalizedCondition);
+    }
+
+    return node;
+  }
+
   private boolean isQuotedString(String input) {
     if (input.startsWith("\"") && input.endsWith("\"")) {
       String content = removeQuotes(input);
@@ -141,6 +195,12 @@ public List<FilterCondition> getWhereConditions() {
 
   @Override
   public void buildStatement(IDataLakeQueryBuilder<?> builder) {
-    builder.withInclusiveFilter(filterConditions);
+    if (!filterConditions.isEmpty()) {
+      builder.withInclusiveFilter(filterConditions);
+    }
+
+    if (filterExpression != null) {
+      builder.withFilterExpression(filterExpression);
+    }
   }
 }

streampipes-model/src/main/java/org/apache/streampipes/model/datalake/FilterExpressionCondition.java

@@ -0,0 +1,21 @@
+/*
+ * 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.model.datalake;
+
+public record FilterExpressionCondition(String field, String operator, Object condition) implements FilterExpressionNode {
+}