You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ri...@apache.org on 2021/08/22 16:53:51 UTC
[incubator-streampipes] 01/02: [STREAMPIPES-410] Improve
performance of auto-aggregation
This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 656e67e5d0f669ec31ba0b04e69ed21be2c2b498
Author: Dominik Riemer <ri...@fzi.de>
AuthorDate: Sun Aug 22 18:48:58 2021 +0200
[STREAMPIPES-410] Improve performance of auto-aggregation
---
.../dataexplorer/v4/AutoAggregationHandler.java | 56 ++++++++++++++--------
.../v4/params/SelectFromStatementParams.java | 9 ++--
.../v4/query/elements/SelectFromStatement.java | 2 +-
.../dataexplorer/v4/template/QueryTemplatesV4.java | 26 +++++-----
.../v4/utils/DataLakeManagementUtils.java | 2 +-
5 files changed, 59 insertions(+), 36 deletions(-)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/AutoAggregationHandler.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/AutoAggregationHandler.java
index 4a2ecb6..6a69dae 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/AutoAggregationHandler.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/AutoAggregationHandler.java
@@ -48,21 +48,26 @@ public class AutoAggregationHandler {
public ProvidedQueryParams makeAutoAggregationQueryParams() throws IllegalArgumentException {
checkAllArgumentsPresent();
- Integer count = getCount();
+ try {
+ DataResult newest = getSingleRecord(Order.DESC);
+ DataResult oldest = getSingleRecord(Order.ASC);
+ String sampleField = getSampleField(newest);
+ Integer count = getCount(sampleField);
if (count <= MAX_RETURN_LIMIT) {
LOG.debug("Auto-Aggregation disabled as {} results <= max return limit {}", count, MAX_RETURN_LIMIT);
return disableAutoAgg(this.queryParams);
} else {
LOG.debug("Performing auto-aggregation");
- try {
- int aggValue = getAggregationValue();
+
+ int aggValue = getAggregationValue(newest, oldest);
LOG.debug("Setting auto-aggregation value to {} ms", aggValue);
queryParams.update(QP_TIME_INTERVAL, aggValue + "ms");
- } catch (ParseException e) {
- e.printStackTrace();
- }
return disableAutoAgg(queryParams);
}
+ } catch (ParseException e) {
+ e.printStackTrace();
+ }
+ return null;
}
private void checkAllArgumentsPresent() throws IllegalArgumentException {
@@ -76,11 +81,12 @@ public class AutoAggregationHandler {
return params;
}
- public Integer getCount() {
+ public Integer getCount(String fieldName) {
ProvidedQueryParams countParams = disableAutoAgg(new ProvidedQueryParams(queryParams));
countParams.remove(QP_TIME_INTERVAL);
countParams.remove(QP_AGGREGATION_FUNCTION);
countParams.update(QP_COUNT_ONLY, true);
+ countParams.update(QP_COLUMNS, fieldName);
DataResult result = new DataLakeManagementV4().getData(countParams);
return result.getTotal() > 0 ? ((Double) result.getRows().get(0).get(1)).intValue() : 0;
@@ -90,29 +96,41 @@ public class AutoAggregationHandler {
return dataLakeManagement.getData(params);
}
- private int getAggregationValue() throws ParseException {
- long timerange = getNewestTimestamp() - getOldestTimestamp();
+ private int getAggregationValue(DataResult newest, DataResult oldest) throws ParseException {
+ long timerange = extractTimestamp(newest) - extractTimestamp(oldest);
double v = timerange / MAX_RETURN_LIMIT;
return Double.valueOf(v).intValue();
}
- private long getNewestTimestamp() throws ParseException {
- return getTimestamp(Order.DESC);
- }
-
- private long getOldestTimestamp() throws ParseException {
- return getTimestamp(Order.ASC);
- }
+// private long getNewestTimestamp(DataResult result) throws ParseException {
+// return extractTimestamp(result);
+// }
+//
+// private long getOldestTimestamp(DataResult result) throws ParseException {
+// return extractTimestamp(result);
+// }
- private long getTimestamp(Order order) throws ParseException {
+ private DataResult getSingleRecord(Order order) throws ParseException {
ProvidedQueryParams singleEvent = disableAutoAgg(new ProvidedQueryParams(queryParams));
singleEvent.remove(QP_AGGREGATION_FUNCTION);
singleEvent.update(QP_LIMIT, 1);
singleEvent.update(QP_ORDER, order.toValue());
- DataResult result = fireQuery(singleEvent);
- int timestampIndex = result.getHeaders().indexOf(TIMESTAMP_FIELD);
+ return fireQuery(singleEvent);
+
+ }
+
+ private String getSampleField(DataResult result) {
+ for (String column : result.getHeaders()) {
+ if (!column.equals(TIMESTAMP_FIELD)) {
+ return column;
+ }
+ }
+ throw new IllegalArgumentException("No columns present");
+ }
+ private long extractTimestamp(DataResult result) throws ParseException {
+ int timestampIndex = result.getHeaders().indexOf(TIMESTAMP_FIELD);
return tryParseDate(result.getRows().get(0).get(timestampIndex).toString()).getTime();
}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/SelectFromStatementParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/SelectFromStatementParams.java
index f702d15..6790ece 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/SelectFromStatementParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/SelectFromStatementParams.java
@@ -22,8 +22,8 @@ import javax.annotation.Nullable;
public class SelectFromStatementParams extends QueryParamsV4 {
- private final String selectedColumns;
- private final String aggregationFunction;
+ private String selectedColumns;
+ private String aggregationFunction;
private boolean countOnly = false;
public static SelectFromStatementParams from(String measurementID,
@@ -33,8 +33,9 @@ public class SelectFromStatementParams extends QueryParamsV4 {
}
public static SelectFromStatementParams from(String measurementId,
+ String columns,
boolean countOnly) {
- return new SelectFromStatementParams(measurementId, countOnly);
+ return new SelectFromStatementParams(measurementId, columns, countOnly);
}
public SelectFromStatementParams(String measurementID) {
@@ -44,8 +45,10 @@ public class SelectFromStatementParams extends QueryParamsV4 {
}
public SelectFromStatementParams(String measurementId,
+ String columns,
boolean countOnly) {
this(measurementId);
+ this.selectedColumns = columns;
this.countOnly = countOnly;
}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/SelectFromStatement.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/SelectFromStatement.java
index 15bc05d..abbe5a5 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/SelectFromStatement.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/SelectFromStatement.java
@@ -29,7 +29,7 @@ public class SelectFromStatement extends QueryElement<SelectFromStatementParams>
@Override
protected String buildStatement(SelectFromStatementParams selectFromStatementParams) {
if (selectFromStatementParams.isCountOnly()) {
- return QueryTemplatesV4.selectCountFrom(selectFromStatementParams.getIndex());
+ return QueryTemplatesV4.selectCountFrom(selectFromStatementParams.getIndex(), selectFromStatementParams.getSelectedColumns());
} else if (selectFromStatementParams.getAggregationFunction() == null) {
return QueryTemplatesV4.selectFrom(selectFromStatementParams.getIndex(), selectFromStatementParams.getSelectedColumns());
} else {
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/template/QueryTemplatesV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/template/QueryTemplatesV4.java
index 9761c0b..5840736 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/template/QueryTemplatesV4.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/template/QueryTemplatesV4.java
@@ -17,30 +17,32 @@
*/
package org.apache.streampipes.dataexplorer.v4.template;
+import java.util.StringJoiner;
+
public class QueryTemplatesV4 {
public static String selectFrom(String index, String columns) {
return "SELECT " + columns + " FROM " + index;
}
- public static String selectCountFrom(String index) {
- return "SELECT COUNT(*) FROM " + index;
+ public static String selectCountFrom(String index, String selectedColumns) {
+ return selectAggregationFrom(index, selectedColumns, "COUNT");
}
public static String selectAggregationFrom(String index, String columns, String aggregationFunction) {
String[] cols = columns.split(",");
- StringBuilder statement = new StringBuilder(aggregationFunction + "(" + cols[0] + ")");
-
- for (int i = 1; i < cols.length; i++) {
- statement
- .append(", ")
- .append(aggregationFunction)
- .append("(")
- .append(cols[i])
- .append(")");
+ StringJoiner joiner = new StringJoiner(", ");
+ //StringBuilder statement = new StringBuilder(aggregationFunction + "(" + cols[0] + ")");
+
+ for (int i = 0; i < cols.length; i++) {
+ String builder = aggregationFunction +
+ "(" +
+ cols[i] +
+ ")" + " AS " + cols[i];
+ joiner.add(builder);
}
- return "SELECT " + statement + " FROM " + index;
+ return "SELECT " + joiner + " FROM " + index;
}
public static String deleteFrom(String index) {
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/utils/DataLakeManagementUtils.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/utils/DataLakeManagementUtils.java
index c8807e8..f084e49 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/utils/DataLakeManagementUtils.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/utils/DataLakeManagementUtils.java
@@ -45,7 +45,7 @@ public class DataLakeManagementUtils {
String measurementId = params.getMeasurementId();
if (params.has(QP_COUNT_ONLY) && params.getAsBoolean(QP_COUNT_ONLY)) {
- queryParts.put(SELECT_FROM, SelectFromStatementParams.from(measurementId, true));
+ queryParts.put(SELECT_FROM, SelectFromStatementParams.from(measurementId, params.getAsString(QP_COLUMNS), true));
} else {
queryParts.put(SELECT_FROM, SelectFromStatementParams.from(measurementId, params.getAsString(QP_COLUMNS), params.getAsString(QP_AGGREGATION_FUNCTION)));
}