This is an automated email from the ASF dual-hosted git repository. bossenti pushed a commit to branch iot-delete-capabilities in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 4912ea1583285eab2cb4709c2413bf4c9e633043 Author: bossenti <[email protected]> AuthorDate: Wed May 15 13:38:46 2024 +0200 feat: add delete capabilities --- .../api/IDataExplorerQueryManagement.java | 2 +- .../influx/DataExplorerInfluxQueryExecutor.java | 2 +- .../influx/DataExplorerQueryManagementInflux.java | 6 +- .../iotdb/DataExplorerManagerIotDb.java | 13 ++++- .../iotdb/DataExplorerQueryManagementIotDb.java | 67 ++++++++++++++++++++++ .../dataexplorer/param/DeleteQueryParams.java | 2 +- .../param/ProvidedRestQueryParamConverter.java | 4 +- .../apache/streampipes/ps/DataLakeResourceV4.java | 11 +++- 8 files changed, 94 insertions(+), 13 deletions(-) diff --git a/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerQueryManagement.java b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerQueryManagement.java index 1e3b500925..2f5f3eb6c3 100644 --- a/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerQueryManagement.java +++ b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerQueryManagement.java @@ -39,7 +39,7 @@ public interface IDataExplorerQueryManagement { boolean deleteData(String measurementID); - SpQueryResult deleteData(String measurementID, Long startDate, Long endDate); + boolean deleteData(String measurementName, Long startDate, Long endDate); boolean deleteAllData(); 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 b24d2e4a88..4bc17c2515 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 @@ -108,7 +108,7 @@ public class DataExplorerInfluxQueryExecutor extends DataExplorerQueryExecutor<Q @Override protected Query makeDeleteQuery(DeleteQueryParams params) { - String query = "DELETE FROM \"" + params.measurementId() + "\""; + String query = "DELETE FROM \"" + params.measurementName() + "\""; if (params.timeRestricted()) { query += "WHERE time > " + params.startTime() * 1000000 diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerQueryManagementInflux.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerQueryManagementInflux.java index 44b9374b80..a98df579a5 100644 --- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerQueryManagementInflux.java +++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerQueryManagementInflux.java @@ -92,10 +92,10 @@ public class DataExplorerQueryManagementInflux implements IDataExplorerQueryMana } @Override - public SpQueryResult deleteData(String measurementID, Long startDate, Long endDate) { + public boolean deleteData(String measurementName, Long startDate, Long endDate) { DeleteQueryParams params = - ProvidedRestQueryParamConverter.getDeleteQueryParams(measurementID, startDate, endDate); - return new DataExplorerInfluxQueryExecutor().executeQuery(params); + ProvidedRestQueryParamConverter.getDeleteQueryParams(measurementName, startDate, endDate); + return !new DataExplorerInfluxQueryExecutor().executeQuery(params).getHeaders().isEmpty(); } @Override 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..1e743ec799 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 @@ -19,8 +19,14 @@ package org.apache.streampipes.dataexplorer.iotdb; import org.apache.streampipes.client.api.IStreamPipesClient; +import org.apache.streampipes.commons.environment.Environments; import org.apache.streampipes.dataexplorer.DataExplorerSchemaManagement; -import org.apache.streampipes.dataexplorer.api.*; +import org.apache.streampipes.dataexplorer.api.IDataExplorerManager; +import org.apache.streampipes.dataexplorer.api.IDataExplorerQueryManagement; +import org.apache.streampipes.dataexplorer.api.IDataExplorerSchemaManagement; +import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementCounter; +import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementSanitizer; +import org.apache.streampipes.dataexplorer.api.ITimeSeriesStorage; import org.apache.streampipes.dataexplorer.iotdb.sanitize.DataLakeMeasurementSanitizerIotDb; import org.apache.streampipes.model.datalake.DataLakeMeasure; import org.apache.streampipes.storage.management.StorageDispatcher; @@ -36,7 +42,10 @@ public class DataExplorerManagerIotDb implements IDataExplorerManager { @Override public IDataExplorerQueryManagement getQueryManagement(IDataExplorerSchemaManagement dataExplorerSchemaManagement) { - return null; + return new DataExplorerQueryManagementIotDb( + dataExplorerSchemaManagement, + new DataExplorerIotDbQueryExecutor(new IotDbSessionProvider().getSessionPool(Environments.getEnvironment())) + ); } @Override diff --git a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerQueryManagementIotDb.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerQueryManagementIotDb.java new file mode 100644 index 0000000000..67314f3da1 --- /dev/null +++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerQueryManagementIotDb.java @@ -0,0 +1,67 @@ +package org.apache.streampipes.dataexplorer.iotdb; + +import org.apache.streampipes.dataexplorer.api.IDataExplorerQueryManagement; +import org.apache.streampipes.dataexplorer.api.IDataExplorerSchemaManagement; +import org.apache.streampipes.dataexplorer.export.OutputFormat; +import org.apache.streampipes.model.datalake.SpQueryResult; +import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.io.OutputStream; +import java.util.Map; + +public class DataExplorerQueryManagementIotDb implements IDataExplorerQueryManagement { + + private static final Logger LOG = LoggerFactory.getLogger(DataExplorerIotDbQueryExecutor.class); + + private final DataExplorerIotDbQueryExecutor queryExecutor; + private final IDataExplorerSchemaManagement dataExplorerSchemaManagement; + + public DataExplorerQueryManagementIotDb(IDataExplorerSchemaManagement dataExplorerSchemaManagement, + DataExplorerIotDbQueryExecutor queryExecutor) { + this.dataExplorerSchemaManagement = dataExplorerSchemaManagement; + this.queryExecutor = queryExecutor; + } + + @Override + public SpQueryResult getData(ProvidedRestQueryParams queryParams, boolean ignoreMissingData) throws IllegalArgumentException { + return null; + } + + @Override + public void getDataAsStream(ProvidedRestQueryParams params, OutputFormat format, boolean ignoreMissingValues, OutputStream outputStream) throws IOException { + + } + + @Override + public boolean deleteData(String measurementID) { + var allMeasurements = this.dataExplorerSchemaManagement.getAllMeasurements(); + + var measureToDeleteOpt = allMeasurements.stream() + .filter(measure -> measure.getMeasureName().equals(measurementID)) + .findFirst(); + return measureToDeleteOpt.filter(queryExecutor::deleteData).isPresent(); + } + + @Override + public boolean deleteData(String measurementName, Long startDate, Long endDate) { + var queryString = "DELETE FROM root.streampipes.%s.* WHERE time > %s AND time < %s".formatted(measurementName, startDate, endDate); + return queryExecutor.executeNonQueryStatement(queryString); + } + + @Override + public boolean deleteAllData() { + var allMeasurements = this.dataExplorerSchemaManagement.getAllMeasurements(); + + return allMeasurements.stream() + .allMatch(queryExecutor::deleteData); // Check if all results are true else return false + } + + @Override + public Map<String, Object> getTagValues(String measurementId, String fields) { + LOG.error("Retrieval of tag values is not supported with IoTDB as storage"); + return Map.of(); + } +} diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/DeleteQueryParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/DeleteQueryParams.java index 33d3634903..1fa6948c38 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/DeleteQueryParams.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/DeleteQueryParams.java @@ -19,7 +19,7 @@ package org.apache.streampipes.dataexplorer.param; -public record DeleteQueryParams(String measurementId, Long startTime, Long endTime, boolean timeRestricted) { +public record DeleteQueryParams(String measurementName, Long startTime, Long endTime, boolean timeRestricted) { public DeleteQueryParams(String measurementId, Long startTime, Long endTime) { this(measurementId, startTime != null ? startTime : 0, endTime != null ? endTime : 99999999999999L, true); diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java index dd658d0575..5628b5f992 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java @@ -105,10 +105,10 @@ public class ProvidedRestQueryParamConverter { return queryParameters; } - public static DeleteQueryParams getDeleteQueryParams(String measurementId, + public static DeleteQueryParams getDeleteQueryParams(String measurementName, Long startTime, Long endTime) { - return new DeleteQueryParams(measurementId, startTime, endTime); + return new DeleteQueryParams(measurementName, startTime, endTime); } private static boolean hasTimeParams(ProvidedRestQueryParams params) { diff --git a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java index 2bc5cf4a79..a1bd9d7705 100644 --- a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java +++ b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java @@ -104,7 +104,7 @@ public class DataLakeResourceV4 extends AbstractRestResource { responses = { @ApiResponse(responseCode = "200", description = "Data from measurement series successfully removed"), @ApiResponse(responseCode = "400", description = "Measurement series with given id not found")}) - public ResponseEntity<Void> deleteData( + public ResponseEntity<?> deleteData( @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathVariable("measurementID") String measurementID , @Parameter(in = ParameterIn.QUERY, description = "start date for slicing operation") @@ -112,8 +112,13 @@ public class DataLakeResourceV4 extends AbstractRestResource { , @Parameter(in = ParameterIn.QUERY, description = "end date for slicing operation") @RequestParam(value = "endDate", required = false) Long endDate) { - SpQueryResult result = this.dataExplorerQueryManagement.deleteData(measurementID, startDate, endDate); - return ok(); + if(this.dataExplorerQueryManagement.deleteData(measurementID, startDate, endDate)){ + return ok(); + } else { + return ResponseEntity + .status(HttpStatus.NOT_FOUND) + .body("Given measurement could not be found"); + } } @DeleteMapping(path = "/measurements/{measurementID}/drop")
