This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 583c61b322 feat: implement measurement counting for IoTDB (#2867)
583c61b322 is described below
commit 583c61b32279ea83788106ad2ca82cc37c334073
Author: Tim <[email protected]>
AuthorDate: Thu May 16 08:52:01 2024 +0200
feat: implement measurement counting for IoTDB (#2867)
* feat: implement measurement counting for IoTDB
* style: fix style violations
---
.../influx/DataLakeMeasurementCounterInflux.java | 82 +++----------
.../iotdb/DataExplorerIotDbQueryExecutor.java | 96 +++++++++++++++
.../iotdb/DataExplorerManagerIotDb.java | 2 +-
.../iotdb/DataLakeMeasurementCounterIotDb.java | 73 ++++++++++++
.../dataexplorer/iotdb/IotDbPropertyConverter.java | 2 +-
.../query/DataLakeMeasurementCounter.java | 130 +++++++++++++++++++++
6 files changed, 317 insertions(+), 68 deletions(-)
diff --git
a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java
b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java
index 4f516f52c1..db11222324 100644
---
a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java
+++
b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java
@@ -18,86 +18,36 @@
package org.apache.streampipes.dataexplorer.influx;
-import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementCounter;
+import org.apache.streampipes.dataexplorer.query.DataLakeMeasurementCounter;
import org.apache.streampipes.model.datalake.AggregationFunction;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
-import org.apache.streampipes.model.datalake.SpQueryResult;
-import org.apache.streampipes.model.schema.EventProperty;
-import org.apache.streampipes.model.schema.PropertyScope;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.stream.Collectors;
-public class DataLakeMeasurementCounterInflux implements
IDataLakeMeasurementCounter {
-
- private final List<DataLakeMeasure> allMeasurements;
- private final List<String> measurementNames;
+public class DataLakeMeasurementCounterInflux extends
DataLakeMeasurementCounter {
private static final String COUNT_FIELD = "count";
public DataLakeMeasurementCounterInflux(List<DataLakeMeasure>
allMeasurements,
List<String> measurementNames) {
- this.allMeasurements = allMeasurements;
- this.measurementNames = measurementNames;
+ super(allMeasurements, measurementNames);
}
- public Map<String, Integer> countMeasurementSizes() {
- Map<String, CompletableFuture<Integer>> futures = measurementNames.stream()
- .distinct()
- .map(this::getMeasure)
- .collect(Collectors.toMap(DataLakeMeasure::getMeasureName, m ->
CompletableFuture.supplyAsync(() -> {
- var firstColumn = getFirstColumn(m);
- var builder = DataLakeInfluxQueryBuilder
-
.create(m.getMeasureName()).withEndTime(System.currentTimeMillis())
- .withAggregatedColumn(firstColumn, AggregationFunction.COUNT);
- var queryResult = new
DataExplorerInfluxQueryExecutor().executeQuery(builder.build(),
Optional.empty(), true);
- if (queryResult.getTotal() > 0) {
- var headers = queryResult.getHeaders();
- return extractResult(queryResult, headers);
- } else {
- return 0;
- }
- })));
-
- Map<String, Integer> result = new HashMap<>();
- futures.entrySet().forEach((entry -> {
- try {
- result.put(entry.getKey(), entry.getValue().get());
- } catch (InterruptedException | ExecutionException e) {
- result.put(entry.getKey(), 0);
+ @Override
+ protected CompletableFuture<Integer>
createQueryAsAsyncFuture(DataLakeMeasure measure) {
+ return CompletableFuture.supplyAsync(() -> {
+ var firstColumn = getFirstMeasurementProperty(measure);
+ var builder = DataLakeInfluxQueryBuilder
+
.create(measure.getMeasureName()).withEndTime(System.currentTimeMillis())
+ .withAggregatedColumn(firstColumn, AggregationFunction.COUNT);
+ var queryResult = new
DataExplorerInfluxQueryExecutor().executeQuery(builder.build(),
Optional.empty(), true);
+ if (queryResult.getTotal() > 0) {
+ return extractResult(queryResult, COUNT_FIELD);
+ } else {
+ return 0;
}
- }));
-
- return result;
- }
-
- private Integer extractResult(SpQueryResult queryResult,
- List<String> headers) {
- return ((Double) (
-
queryResult.getAllDataSeries().get(0).getRows().get(0).get(headers.indexOf(COUNT_FIELD)))
- ).intValue();
- }
-
- private DataLakeMeasure getMeasure(String measureName) {
- return allMeasurements
- .stream()
- .filter(m -> m.getMeasureName().equals(measureName))
- .findFirst()
- .orElse(null);
- }
-
- private String getFirstColumn(DataLakeMeasure measure) {
- return measure.getEventSchema().getEventProperties()
- .stream()
- .filter(ep -> ep.getPropertyScope() != null
- &&
ep.getPropertyScope().equals(PropertyScope.MEASUREMENT_PROPERTY.name()))
- .map(EventProperty::getRuntimeName)
- .findFirst()
- .orElse(null);
+ });
}
}
diff --git
a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerIotDbQueryExecutor.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerIotDbQueryExecutor.java
new file mode 100644
index 0000000000..9622aebd48
--- /dev/null
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerIotDbQueryExecutor.java
@@ -0,0 +1,96 @@
+/*
+ * 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.iotdb;
+
+import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.pool.SessionPool;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataexplorer.param.DeleteQueryParams;
+import org.apache.streampipes.dataexplorer.param.SelectQueryParams;
+import org.apache.streampipes.dataexplorer.query.DataExplorerQueryExecutor;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.model.datalake.SpQueryResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Optional;
+
+public class DataExplorerIotDbQueryExecutor extends
DataExplorerQueryExecutor<String, SessionDataSetWrapper> {
+
+ private final SessionPool sessionPool;
+
+ public DataExplorerIotDbQueryExecutor(SessionPool sessionPool){
+ this.sessionPool = sessionPool;
+ }
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DataExplorerIotDbQueryExecutor.class);
+ @Override
+ protected SpQueryResult postQuery(SessionDataSetWrapper queryResult,
Optional<String> forIdOpt, boolean ignoreMissingValues) {
+ return null;
+ }
+
+ @Override
+ public SessionDataSetWrapper executeQuery(String query) {
+ try (var result = sessionPool.executeQueryStatement(query)) {
+ return result;
+ } catch (IoTDBConnectionException | StatementExecutionException e) {
+ throw new SpRuntimeException(e);
+ }
+ }
+
+ public boolean executeNonQueryStatement(String statement) {
+ try {
+ sessionPool.executeNonQueryStatement(statement);
+ } catch (StatementExecutionException | IoTDBConnectionException e) {
+ LOG.error("Error while executing non-query statement '{}': {}",
statement, e.getMessage());
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ protected String asQueryString(String query) {
+ return "";
+ }
+
+ @Override
+ protected String makeDeleteQuery(DeleteQueryParams params) {
+ LOG.error("This method is not required for IoTDB.");
+ return "";
+ }
+
+ @Override
+ protected String makeSelectQuery(SelectQueryParams params) {
+ return "";
+ }
+
+ @Override
+ public Map<String, Object> getTagValues(String measurementId, String fields)
{
+ return Map.of();
+ }
+
+ @Override
+ public boolean deleteData(DataLakeMeasure measure) {
+ var deleteTimeSeriesQuery = "DELETE timeseries
root.streampipes.%s.*".formatted(measure.getMeasureName());
+ return executeNonQueryStatement(deleteTimeSeriesQuery);
+ }
+}
\ No newline at end of file
diff --git
a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
index 5457ec166d..e29b570f4a 100644
---
a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
@@ -31,7 +31,7 @@ public class DataExplorerManagerIotDb implements
IDataExplorerManager {
@Override
public IDataLakeMeasurementCounter
getMeasurementCounter(List<DataLakeMeasure> allMeasurements, List<String>
measurementsToCount) {
- return null;
+ return new DataLakeMeasurementCounterIotDb(allMeasurements,
measurementsToCount);
}
@Override
diff --git
a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataLakeMeasurementCounterIotDb.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataLakeMeasurementCounterIotDb.java
new file mode 100644
index 0000000000..95ed59d6ac
--- /dev/null
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataLakeMeasurementCounterIotDb.java
@@ -0,0 +1,73 @@
+/*
+ * 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.iotdb;
+
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.streampipes.commons.environment.Environments;
+import org.apache.streampipes.dataexplorer.query.DataLakeMeasurementCounter;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+public class DataLakeMeasurementCounterIotDb extends
DataLakeMeasurementCounter {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DataLakeMeasurementCounterIotDb.class);
+
+ public DataLakeMeasurementCounterIotDb(List<DataLakeMeasure> allMeasurements,
+ List<String> measurementNames) {
+ super(allMeasurements, measurementNames);
+ }
+
+ /**
+ * Creates a CompletableFuture to execute a count query on a DataLakeMeasure
asynchronously.
+ *
+ * @param measure The DataLakeMeasure object representing the measure to
query.
+ * @return A CompletableFuture<Integer> representing the count query result
as a future.
+ */
+ @Override
+ protected CompletableFuture<Integer>
createQueryAsAsyncFuture(DataLakeMeasure measure) {
+ var sessionPool = new
IotDbSessionProvider().getSessionPool(Environments.getEnvironment());
+ return CompletableFuture.supplyAsync(() -> {
+
+ // We want to apply the count query to only one property of the
measurement, as this is sufficient and
+ // significantly reduces the complexity of the query compared to
counting all available properties.
+ // So we can just take the first measurement
+ var propertyName = getFirstMeasurementProperty(measure);
+
+ try (var result = new
DataExplorerIotDbQueryExecutor(sessionPool).executeQuery(
+ "Select count(%s) from root.streampipes.%s".formatted(propertyName,
measure.getMeasureName())
+ )) {
+ var resultRecord = result.next();
+ if (resultRecord == null) {
+ LOG.error("Result of count query does not contain any row - empty
result");
+ return 0;
+ }
+ return (int) resultRecord.getFields().get(0).getLongV();
+ } catch (IoTDBConnectionException | StatementExecutionException e) {
+ LOG.error("Error during count query execution: {}", e.getMessage());
+ return 0;
+ }
+ });
+ }
+}
diff --git
a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverter.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverter.java
index 9772c39c44..4bb46a2cfd 100644
---
a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverter.java
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverter.java
@@ -32,7 +32,7 @@ import org.apache.streampipes.vocabulary.XSD;
public class IotDbPropertyConverter {
/**
- * Converts a {@link EventPropertyPrimitive }into an {@link
IotDbMeasurementRecord}.
+ * Converts a {@link EventPropertyPrimitive } into an {@link
IotDbMeasurementRecord}.
*
* @param eventPropertyPrimitive The primitive event property to be converted
* @param primitiveField The primitive field containing the property
value
diff --git
a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java
b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java
new file mode 100644
index 0000000000..0c1026a627
--- /dev/null
+++
b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java
@@ -0,0 +1,130 @@
+/*
+ * 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.query;
+
+import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementCounter;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.model.datalake.SpQueryResult;
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.PropertyScope;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+public abstract class DataLakeMeasurementCounter implements
IDataLakeMeasurementCounter {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DataLakeMeasurementCounter.class);
+
+ protected final List<DataLakeMeasure> allMeasurements;
+ protected final List<String> measurementNames;
+
+ public DataLakeMeasurementCounter(List<DataLakeMeasure> allMeasurements,
+ List<String> measurementNames) {
+ this.allMeasurements = allMeasurements;
+ this.measurementNames = measurementNames;
+ }
+
+ @Override
+ public Map<String, Integer> countMeasurementSizes() {
+
+ // create async futures so that count queries can be executed parallel
+ Map<String, CompletableFuture<Integer>> countQueriesFutures =
measurementNames.stream()
+ .map(this::getMeasure)
+ .collect(Collectors.toMap(
+ DataLakeMeasure::getMeasureName,
+ this::createQueryAsAsyncFuture)
+ );
+
+ return getQueryResults(countQueriesFutures);
+ }
+
+ /**
+ * Retrieves the {@link DataLakeMeasure} with the specified measure name
from the collection of all measurements.
+ *
+ * @param measureName The name of the measure to retrieve.
+ * @return The DataLakeMeasure corresponding to the provided measure name,
or null if no such measure is found.
+ */
+ private DataLakeMeasure getMeasure(String measureName) {
+ return allMeasurements
+ .stream()
+ .filter(m -> m.getMeasureName().equals(measureName))
+ .findFirst()
+ .orElse(null);
+ }
+
+ /**
+ * Retrieves the results of asynchronous count queries from the given
CompletableFuture map.
+ *
+ * @param queryFutures A Map containing the futures of
+ * asynchronous count queries mapped by their respective
keys.
+ * @return A Map representing the results of the queries, where each key
corresponds to
+ * a measure name and the value is the count result.
+ */
+ private Map<String, Integer> getQueryResults(Map<String,
CompletableFuture<Integer>> queryFutures) {
+ Map<String, Integer> resultPerMeasure = new HashMap<>();
+ queryFutures.forEach((key, value) -> {
+ try {
+ resultPerMeasure.put(key, value.get());
+ } catch (InterruptedException | ExecutionException e) {
+ LOG.error("Async execution of count query failed: {}", e.getMessage());
+ resultPerMeasure.put(key, 0);
+ }
+ });
+
+ return resultPerMeasure;
+ }
+
+ private Integer getQueryResult(Map<String, CompletableFuture<Integer>>
queryFuture) {
+ return null;
+ }
+
+ /**
+ * Retrieves the runtime name of the first measurement property from the
event schema of the given DataLakeMeasure.
+ *
+ * @param measure The {@link DataLakeMeasure} from which to retrieve the
first measurement property.
+ * @return The runtime name of the first measurement property, or null if no
such property is found.
+ */
+ protected String getFirstMeasurementProperty(DataLakeMeasure measure) {
+ return measure.getEventSchema().getEventProperties()
+ .stream()
+ .filter(ep -> ep.getPropertyScope() != null
+ &&
ep.getPropertyScope().equals(PropertyScope.MEASUREMENT_PROPERTY.name()))
+ .map(EventProperty::getRuntimeName)
+ .findFirst()
+ .orElse(null);
+ }
+
+ protected Integer extractResult(SpQueryResult queryResult, String fieldName)
{
+ return ((Double) (
+
queryResult.getAllDataSeries().get(0).getRows().get(0).get(queryResult.getHeaders().indexOf(fieldName)))
+ ).intValue();
+ }
+
+ /**
+ * Create the count query for the given DataLakeMeasure and return it as a
{@link CompletableFuture}.
+ */
+ protected abstract CompletableFuture<Integer>
createQueryAsAsyncFuture(DataLakeMeasure measure);
+}
