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 2024/03/20 21:05:53 UTC
(streampipes) 01/01: feat: Improve Influx query handling
This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch return-timestamp-influxdb
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit ebd7324fa8345fdb01a885a8344a72b88e02ef49
Author: Dominik Riemer <do...@gmail.com>
AuthorDate: Wed Mar 20 22:02:07 2024 +0100
feat: Improve Influx query handling
---
.../influx/DataExplorerInfluxQueryExecutor.java | 3 +-
.../param/ProvidedRestQueryParams.java | 4 ++
.../query/DataExplorerQueryExecutor.java | 1 +
.../query/StreamedQueryResultProvider.java | 19 +++-----
.../query/writer/item/ItemGenerator.java | 10 +++--
.../streampipes/dataexplorer/utils/TimeParser.java | 50 ----------------------
6 files changed, 20 insertions(+), 67 deletions(-)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java
index 01194fc259..3d97a5110a 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java
@@ -33,6 +33,7 @@ import org.influxdb.dto.QueryResult;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.TimeUnit;
public class DataExplorerInfluxQueryExecutor extends DataExplorerQueryExecutor<Query, QueryResult> {
@@ -96,7 +97,7 @@ public class DataExplorerInfluxQueryExecutor extends DataExplorerQueryExecutor<Q
@Override
protected QueryResult executeQuery(Query query) {
try (final InfluxDB influxDB = InfluxClientProvider.getInfluxDBClient()) {
- return influxDB.query(query);
+ return influxDB.query(query, TimeUnit.MILLISECONDS);
}
}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParams.java
index 308bfba3c4..d703eb934b 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParams.java
@@ -65,6 +65,10 @@ public class ProvidedRestQueryParams {
this.providedParams.put(key, value);
}
+ public void update(String key, long value) {
+ update(key, String.valueOf(value));
+ }
+
public void update(String key, Integer value) {
update(key, String.valueOf(value));
}
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 c28c949780..b5d187fa8a 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
@@ -94,6 +94,7 @@ public abstract class DataExplorerQueryExecutor<X, W> {
public SpQueryResult executeQuery(X query,
boolean ignoreMissingValues) {
+ LOG.info("Data Lake Query " + asQueryString(query));
if (LOG.isDebugEnabled()) {
LOG.debug("Data Lake Query " + asQueryString(query));
}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/StreamedQueryResultProvider.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/StreamedQueryResultProvider.java
index ee356e9368..3921200e84 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/StreamedQueryResultProvider.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/StreamedQueryResultProvider.java
@@ -32,11 +32,10 @@ import java.util.List;
import java.util.Optional;
import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_LIMIT;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_PAGE;
public class StreamedQueryResultProvider extends QueryResultProvider {
- private static final int MAX_RESULTS_PER_QUERY = 500000;
+ private static final int MAX_RESULTS_PER_QUERY = 200000;
private static final String TIME_FIELD = "time";
private final OutputFormat format;
@@ -61,36 +60,32 @@ public class StreamedQueryResultProvider extends QueryResultProvider {
var measurement = findByMeasurementName(queryParams.getMeasurementId()).get();
SpQueryResult dataResult;
- int page = 0;
- if (queryParams.has(QP_PAGE)) {
- page = queryParams.getAsInt(QP_PAGE);
- }
boolean isFirstDataItem = true;
configuredWriter.beforeFirstItem(outputStream);
do {
- queryParams.update(SupportedRestQueryParams.QP_PAGE, String.valueOf(page));
dataResult = getData();
+ long lastTimestamp = 0;
if (dataResult.getTotal() > 0) {
changeTimestampHeader(measurement, dataResult);
var columns = dataResult.getHeaders();
for (List<Object> row : dataResult.getAllDataSeries().get(0).getRows()) {
configuredWriter.writeItem(outputStream, row, columns, isFirstDataItem);
isFirstDataItem = false;
+ lastTimestamp = ((Double) row.get(0)).longValue();
}
}
- page++;
- } while (queryNextPage(dataResult.getTotal(), usesLimit, limit, page));
+ queryParams.update(SupportedRestQueryParams.QP_START_DATE, lastTimestamp + 1);
+ } while (queryNextPage(dataResult.getTotal(), usesLimit, limit));
configuredWriter.afterLastItem(outputStream);
}
private boolean queryNextPage(int lastResultsCount,
boolean usesLimit,
- int limit,
- int lastPage) {
+ int limit) {
if (usesLimit) {
- return !(limit <= (lastPage) * MAX_RESULTS_PER_QUERY);
+ return lastResultsCount >= limit;
} else {
return lastResultsCount > 0;
}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/ItemGenerator.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/ItemGenerator.java
index 7b3aa447fe..31e9eb61d5 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/ItemGenerator.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/ItemGenerator.java
@@ -19,8 +19,6 @@
package org.apache.streampipes.dataexplorer.query.writer.item;
-import org.apache.streampipes.dataexplorer.utils.TimeParser;
-
import java.util.List;
import java.util.StringJoiner;
@@ -39,9 +37,9 @@ public abstract class ItemGenerator {
StringJoiner joiner = new StringJoiner(separator);
for (int i = 0; i < row.size(); i++) {
- var value = row.get(i);
+ Object value = row.get(i);
if (i == 0) {
- value = TimeParser.parseTime(value.toString());
+ value = getTimestampValue((Double) row.get(i));
}
joiner.add(makeItemString(columns.get(i), value));
}
@@ -54,4 +52,8 @@ public abstract class ItemGenerator {
protected abstract String finalizeItem(String item);
+ private long getTimestampValue(Double value) {
+ return value.longValue();
+ }
+
}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/utils/TimeParser.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/utils/TimeParser.java
deleted file mode 100644
index 135a5d89df..0000000000
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/utils/TimeParser.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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.utils;
-
-import java.time.Instant;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.ZonedDateTime;
-import java.time.format.DateTimeFormatter;
-import java.time.format.DateTimeFormatterBuilder;
-import java.time.temporal.ChronoField;
-import java.time.temporal.ChronoUnit;
-import java.time.temporal.TemporalAccessor;
-
-public class TimeParser {
-
- private static final DateTimeFormatter formatter = new DateTimeFormatterBuilder()
- .appendPattern(
- "uuuu[-MM[-dd]]['T'HH[:mm[:ss[.SSSSSSSSS][.SSSSSSSS][.SSSSSSS]"
- + "[.SSSSSS][.SSSSS][.SSSS][.SSS][.SS][.S]]]][XXX]")
- .parseDefaulting(ChronoField.NANO_OF_SECOND, 0)
- .parseDefaulting(ChronoField.OFFSET_SECONDS, 0)
- .toFormatter();
-
- public static Long parseTime(String v) {
- TemporalAccessor temporalAccessor = formatter.parseBest(v,
- ZonedDateTime::from,
- LocalDateTime::from,
- LocalDate::from);
-
- Instant instant = Instant.from(temporalAccessor);
- return Instant.EPOCH.until(instant, ChronoUnit.MILLIS);
- }
-}