This is an automated email from the ASF dual-hosted git repository. bossenti pushed a commit to branch iotdb-measurement-counting in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit ec59b43e795ec142415a9a55d14baa75977b78f0 Author: bossenti <[email protected]> AuthorDate: Wed May 15 10:58:26 2024 +0200 feat: implement measurement counting for IoTDB --- .../influx/DataLakeMeasurementCounterInflux.java | 82 +++--------- .../iotdb/DataExplorerIotDbQueryExecutor.java | 96 ++++++++++++++ .../iotdb/DataExplorerManagerIotDb.java | 2 +- .../iotdb/DataLakeMeasurementCounterIotDb.java | 73 +++++++++++ .../dataexplorer/iotdb/IotDbPropertyConverter.java | 2 +- .../query/DataLakeMeasurementCounter.java | 139 +++++++++++++++++++++ 6 files changed, 326 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..6891b09e63 --- /dev/null +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java @@ -0,0 +1,139 @@ +/* + * 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); + } + + @Override + public int countMeasurementSize(String measurementName) { + var measure = getMeasure(measurementName); + var query = createQueryAsAsyncFuture(measure); + try { + return query.get(); + } catch (InterruptedException | ExecutionException e) { + LOG.error("Could not count measurement size - query failed: {}", e.getMessage()); + return 0; + } + } + + /** + * 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<String, CompletableFuture<Integer>> containing the futures of asynchronous count queries mapped by their respective keys. + * @return A Map<String, Integer> 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<Integer>} + */ + protected abstract CompletableFuture<Integer> createQueryAsAsyncFuture(DataLakeMeasure measure); +}
