This is an automated email from the ASF dual-hosted git repository. bossenti pushed a commit to branch abstract-data-explorer-from-influx in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit f108c2a2e04d583b45d60469ccf380c35df1989e Author: bossenti <[email protected]> AuthorDate: Tue Apr 30 11:22:08 2024 +0200 refactor: abstract query execution from Influx --- .../influx/DataExplorerInfluxQueryExecutor.java | 19 ++--------- .../dataexplorer}/QueryResultProvider.java | 22 ++++++++---- .../dataexplorer}/StreamedQueryResultProvider.java | 8 +++-- .../query/DataExplorerQueryExecutor.java | 39 ++++++++-------------- 4 files changed, 38 insertions(+), 50 deletions(-) diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java index a500a4bd11..b24d2e4a88 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java +++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java @@ -19,7 +19,6 @@ package org.apache.streampipes.dataexplorer.influx; import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider; -import org.apache.streampipes.dataexplorer.influx.migrate.DeleteDataQuery; import org.apache.streampipes.dataexplorer.param.DeleteQueryParams; import org.apache.streampipes.dataexplorer.param.SelectQueryParams; import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder; @@ -37,6 +36,7 @@ import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; @@ -45,18 +45,6 @@ import static org.apache.streampipes.commons.environment.Environments.getEnviron public class DataExplorerInfluxQueryExecutor extends DataExplorerQueryExecutor<Query, QueryResult> { - public DataExplorerInfluxQueryExecutor() { - super(); - } - - public DataExplorerInfluxQueryExecutor(String forId) { - super(forId); - } - - public DataExplorerInfluxQueryExecutor(int maximumAmountOfEvents) { - super(maximumAmountOfEvents); - } - protected DataSeries convertResult(QueryResult.Series series, boolean ignoreMissingValues) { List<String> columns = series.getColumns(); @@ -79,6 +67,7 @@ public class DataExplorerInfluxQueryExecutor extends DataExplorerQueryExecutor<Q } protected SpQueryResult postQuery(QueryResult queryResult, + Optional<String> forIdOpt, boolean ignoreMissingValues) throws RuntimeException { SpQueryResult result = new SpQueryResult(); AtomicLong lastTimestamp = new AtomicLong(); @@ -96,9 +85,7 @@ public class DataExplorerInfluxQueryExecutor extends DataExplorerQueryExecutor<Q result.setLastTimestamp(lastTimestamp.get()); } - if (this.appendId) { - result.setForId(this.forId); - } + forIdOpt.ifPresent(result::setForId); return result; } diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/QueryResultProvider.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/QueryResultProvider.java similarity index 67% rename from streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/QueryResultProvider.java rename to streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/QueryResultProvider.java index 85a23b188f..cd77b33d07 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/QueryResultProvider.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/QueryResultProvider.java @@ -16,44 +16,54 @@ * */ -package org.apache.streampipes.dataexplorer.influx.migrate; +package org.apache.streampipes.dataexplorer; + +import org.apache.streampipes.dataexplorer.api.IDataExplorerQueryManagement; -import org.apache.streampipes.dataexplorer.influx.DataExplorerInfluxQueryExecutor; import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParamConverter; import org.apache.streampipes.dataexplorer.param.SelectQueryParams; +import org.apache.streampipes.dataexplorer.query.DataExplorerQueryExecutor; import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams; import org.apache.streampipes.model.datalake.param.SupportedRestQueryParams; import org.apache.streampipes.model.datalake.SpQueryResult; +import java.util.Optional; + public class QueryResultProvider { public static final String FOR_ID_KEY = "forId"; protected final boolean ignoreMissingData; + protected final IDataExplorerQueryManagement dataExplorerQueryManagement; + protected final DataExplorerQueryExecutor<?, ?> queryExecutor; protected ProvidedRestQueryParams queryParams; public QueryResultProvider(ProvidedRestQueryParams queryParams, + IDataExplorerQueryManagement dataExplorerQueryManagement, + DataExplorerQueryExecutor<?, ?> queryExecutor, boolean ignoreMissingData) { this.queryParams = queryParams; this.ignoreMissingData = ignoreMissingData; + this.dataExplorerQueryManagement =dataExplorerQueryManagement; + this.queryExecutor = queryExecutor; } public SpQueryResult getData() { if (queryParams.has(SupportedRestQueryParams.QP_AUTO_AGGREGATE)) { - queryParams = new AutoAggregationHandler(queryParams).makeAutoAggregationQueryParams(); + queryParams = new AutoAggregationHandler(queryParams, dataExplorerQueryManagement).makeAutoAggregationQueryParams(); } SelectQueryParams qp = ProvidedRestQueryParamConverter.getSelectQueryParams(queryParams); if (queryParams.getProvidedParams().containsKey(SupportedRestQueryParams.QP_MAXIMUM_AMOUNT_OF_EVENTS)) { int maximumAmountOfEvents = Integer.parseInt(queryParams.getProvidedParams().get(SupportedRestQueryParams.QP_MAXIMUM_AMOUNT_OF_EVENTS)); - return new DataExplorerInfluxQueryExecutor(maximumAmountOfEvents).executeQuery(qp, ignoreMissingData); + return queryExecutor.executeQuery(qp, maximumAmountOfEvents, Optional.empty(), ignoreMissingData); } if (queryParams.getProvidedParams().containsKey(FOR_ID_KEY)) { String forWidgetId = queryParams.getProvidedParams().get(FOR_ID_KEY); - return new DataExplorerInfluxQueryExecutor(forWidgetId).executeQuery(qp, ignoreMissingData); + return queryExecutor.executeQuery(qp, -1, Optional.of(forWidgetId), ignoreMissingData); } else { - return new DataExplorerInfluxQueryExecutor().executeQuery(qp, ignoreMissingData); + return queryExecutor.executeQuery(qp, -1, Optional.empty(), ignoreMissingData); } } } diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/StreamedQueryResultProvider.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/StreamedQueryResultProvider.java similarity index 90% rename from streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/StreamedQueryResultProvider.java rename to streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/StreamedQueryResultProvider.java index 52ed5d7881..d69eb72508 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/StreamedQueryResultProvider.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/StreamedQueryResultProvider.java @@ -16,8 +16,10 @@ * */ -package org.apache.streampipes.dataexplorer.influx.migrate; +package org.apache.streampipes.dataexplorer; +import org.apache.streampipes.dataexplorer.api.IDataExplorerQueryManagement; +import org.apache.streampipes.dataexplorer.query.DataExplorerQueryExecutor; import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams; import org.apache.streampipes.model.datalake.param.SupportedRestQueryParams; import org.apache.streampipes.dataexplorer.export.ConfiguredOutputWriter; @@ -40,8 +42,10 @@ public class StreamedQueryResultProvider extends QueryResultProvider { public StreamedQueryResultProvider(ProvidedRestQueryParams params, OutputFormat format, + IDataExplorerQueryManagement dataExplorerQueryManagement, + DataExplorerQueryExecutor<?, ?> queryExecutor, boolean ignoreMissingValues) { - super(params, ignoreMissingValues); + super(params, dataExplorerQueryManagement, queryExecutor, ignoreMissingValues); this.format = format; } diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java index d56856257f..347070a804 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java @@ -29,54 +29,39 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Map; +import java.util.Optional; public abstract class DataExplorerQueryExecutor<X, W> { private static final Logger LOG = LoggerFactory.getLogger(DataExplorerQueryExecutor.class); - protected int maximumAmountOfEvents; - - protected boolean appendId = false; - protected String forId; - - public DataExplorerQueryExecutor() { - this.maximumAmountOfEvents = -1; - } - - public DataExplorerQueryExecutor(String forId) { - this(); - this.appendId = true; - this.forId = forId; - } - - public DataExplorerQueryExecutor(int maximumAmountOfEvents) { - this(); - this.maximumAmountOfEvents = maximumAmountOfEvents; - } /** * Execute the data explorer query and return the result or a warning message * in case the maximum amount of events to return is defined */ public SpQueryResult executeQuery(SelectQueryParams params, + int maximumAmountOfEvents, + Optional<String> forIdOpt, boolean ignoreMissingValues) throws RuntimeException { X query = makeSelectQuery(params); - var result = executeQuery(query, ignoreMissingValues); - if (this.maximumAmountOfEvents != -1) { - return validateAndReturnQueryResult(result, params.getLimit()); + var result = executeQuery(query, forIdOpt, ignoreMissingValues); + if (maximumAmountOfEvents != -1) { + return validateAndReturnQueryResult(result, params.getLimit(), maximumAmountOfEvents); } else { return result; } } private SpQueryResult validateAndReturnQueryResult(SpQueryResult queryResult, - int limit) { + int limit, + int maximumAmountOfEvents) { var amountOfResults = queryResult.getAllDataSeries() .stream() .mapToInt(DataSeries::getTotal) .sum(); var amountOfQueryResults = limit == Integer.MIN_VALUE ? amountOfResults : Math.min(amountOfResults, limit); - if (amountOfQueryResults > this.maximumAmountOfEvents) { + if (amountOfQueryResults > maximumAmountOfEvents) { return makeTooMuchDataResult(amountOfQueryResults); } else { return queryResult; @@ -91,10 +76,11 @@ public abstract class DataExplorerQueryExecutor<X, W> { } public SpQueryResult executeQuery(DeleteQueryParams params) { - return executeQuery(makeDeleteQuery(params), true); + return executeQuery(makeDeleteQuery(params), Optional.empty(), true); } public SpQueryResult executeQuery(X query, + Optional<String> forIdOpt, boolean ignoreMissingValues) { if (LOG.isDebugEnabled()) { LOG.debug("Data Lake Query {}", asQueryString(query)); @@ -105,10 +91,11 @@ public abstract class DataExplorerQueryExecutor<X, W> { LOG.debug("Data Lake Query Result: {}", result.toString()); } - return postQuery(result, ignoreMissingValues); + return postQuery(result, forIdOpt, ignoreMissingValues); } protected abstract SpQueryResult postQuery(W queryResult, + Optional<String> forIdOpt, boolean ignoreMissingValues); public abstract W executeQuery(X query);
