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 de44bd9c43 feat: IoTDB delete capabilities (#2870)
de44bd9c43 is described below
commit de44bd9c43a7b17178e29338eb3927e785525297
Author: Tim <[email protected]>
AuthorDate: Fri May 17 08:36:48 2024 +0200
feat: IoTDB delete capabilities (#2870)
* feat: add delete capabilities
* style: add missing import
* style: add whitespace
* fix: delete query verification
---
.../api/IDataExplorerQueryManagement.java | 2 +-
.../influx/DataExplorerInfluxQueryExecutor.java | 2 +-
.../influx/DataExplorerQueryManagementInflux.java | 9 ++-
.../iotdb/DataExplorerManagerIotDb.java | 13 +++-
.../iotdb/DataExplorerQueryManagementIotDb.java | 85 ++++++++++++++++++++++
.../dataexplorer/param/DeleteQueryParams.java | 2 +-
.../param/ProvidedRestQueryParamConverter.java | 4 +-
.../apache/streampipes/ps/DataLakeResourceV4.java | 11 ++-
8 files changed, 115 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..0129fd4beb 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
@@ -25,6 +25,7 @@ import
org.apache.streampipes.dataexplorer.QueryResultProvider;
import org.apache.streampipes.dataexplorer.StreamedQueryResultProvider;
import org.apache.streampipes.dataexplorer.param.DeleteQueryParams;
import
org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParamConverter;
+import org.apache.streampipes.model.datalake.SpQueryStatus;
import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
import org.apache.streampipes.model.datalake.SpQueryResult;
@@ -92,10 +93,12 @@ 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)
+ .getSpQueryStatus()
+ .equals(SpQueryStatus.OK);
}
@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 e29b570f4a..97ef58e99e 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..dfc88efea8
--- /dev/null
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerQueryManagementIotDb.java
@@ -0,0 +1,85 @@
+/*
+ * 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.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..d4bda50882 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")
