You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by eb...@apache.org on 2021/06/11 18:43:14 UTC
[incubator-streampipes] branch STREAMPIPES-349 updated (3fb5a97 ->
895508d)
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a change to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git.
discard 3fb5a97 Add DataLakeResourceV4 to openapi.yaml
discard 5915b93 Add DataLakeResourceV4 to StreamPipesResourceConfig
discard f14a05b Add implementation of endpoint for removing entire measurement series and related event property
discard f974f4d Add endpoint definition for dropping entire measurement series from data lake and removing related event property
discard bee08cc Add implementation of endpoint for downloading data from data lake
discard b416e91 Add endpoint definition for downloading data from data lake (as csv- or json-file)
discard c51509e Add implementation of endpoint for getting data from data lake
discard 9d14cf9 Add implementation of endpoint for removing data from a single measurement series
discard 015641e Add Data Explorer Query class
discard f53de50 Add Query Builder class
discard ee2fe10 Add query element "Delete From Statement" and corresponding parameter class
discard 2513b45 Add query element "Offset" and corresponding parameter class
discard 6dbf7c6 Add query element "Item Limitation" and corresponding parameter class
discard aec29d6 Add query element "Grouping By Time" and corresponding parameter class
discard 4dd3411 Add query element "Grouping By Tags" and corresponding parameter class
discard 029334f Add query element "Ordering By Time" and corresponding parameter class
discard c8ca2a6 Add query element "Time Boundary" and corresponding parameter class
discard bf7eb2e Add query element "Select From Statement" and corresponding parameter class
discard 3d11553 Add query template class (includes database specific templates for various query statements)
discard ec066a6 Add abstract query element class and query parameter class
discard 12c2b39 [STREAMPIPES-349] Add first implementation of methods for getting and setting data lake configuration
discard 414b619 [STREAMPIPES-349] Update definition of configuration endpoints and add implementation
discard 7a4f1c3 [STREAMPIPES-349] Add query defining classes for changes at the retention policies
discard 20b5586 [STREAMPIPES-349] Add label definition for data lake
discard 1cbb5da [STREAMPIPES-349] Add data lake configuration and retention policy defining classes
discard 2540b3e [STREAMPIPES-349] Add implementation of endpoint for 'removing all stored measurement series'
discard 20f8838 [STREAMPIPES-349] Rename method as well as list containing all stored measurement series
discard 33c617c [STREAMPIPES-349] Add implementation of endpoint for 'getting all stored measurement series'
discard ded2489 [STREAMPIPES-349] Add skeleton of the revised REST API
omit af1a6ff Merge pull request #41 from anushkrishnav/anushkrishnav
new a3d1389 [STREAMPIPES-349] Add skeleton of the revised REST API
new d4e35f3 [STREAMPIPES-349] Add implementation of endpoint for 'getting all stored measurement series'
new a066b5d [STREAMPIPES-349] Rename method as well as list containing all stored measurement series
new 3e303a8 [STREAMPIPES-349] Add implementation of endpoint for 'removing all stored measurement series'
new 00767ef [STREAMPIPES-349] Add data lake configuration and retention policy defining classes
new b5bc9f8 [STREAMPIPES-349] Add label definition for data lake
new f3ced17 [STREAMPIPES-349] Add query defining classes for changes at the retention policies
new 3c30237 [STREAMPIPES-349] Update definition of configuration endpoints and add implementation
new f7cb766 [STREAMPIPES-349] Add first implementation of methods for getting and setting data lake configuration
new 3c3d74d [STREAMPIPES-349] Add abstract query element class and query parameter class
new de0279a [STREAMPIPES-349] Add query template class (includes database specific templates for various query statements)
new 84f16a6 [STREAMPIPES-349] Add query element 'Select From Statement' and corresponding parameter class
new a79e5eb [STREAMPIPES-349] Add query element 'Time Boundary' and corresponding parameter class
new e7a68d0 Add query element "Ordering By Time" and corresponding parameter class
new b8366e2 [STREAMPIPES-349] Add query element 'Grouping By Tags' and corresponding parameter class
new 02063e7 [STREAMPIPES-349] Add query element 'Grouping By Time' and corresponding parameter class
new 7613792 [STREAMPIPES-349] Add query element 'Item Limitation' and corresponding parameter class
new acfc799 [STREAMPIPES-349] Add query element 'Offset' and corresponding parameter class
new 60a69b0 [STREAMPIPES-349] Add query element 'Delete From Statement' and corresponding parameter class
new ba5c5a8 [STREAMPIPES-349] Add Query Builder class
new 98759e0 [STREAMPIPES-349] Add Data Explorer Query class
new 228b5d1 [STREAMPIPES-349] Add implementation of endpoint for removing data from a single measurement series
new f24f63d [STREAMPIPES-349] Add implementation of endpoint for getting data from data lake
new 6bf1dc1 [STREAMPIPES-349] Add endpoint definition for downloading data from data lake (as csv- or json-file)
new c9ec353 [STREAMPIPES-349] Add implementation of endpoint for downloading data from data lake
new 8b4e07c [STREAMPIPES-349] Add endpoint definition for dropping entire measurement series from data lake and removing related event property
new 307383d [STREAMPIPES-349] Add implementation of endpoint for removing entire measurement series and related event property
new d9e9d79 [STREAMPIPES-349] Add DataLakeResourceV4 to StreamPipesResourceConfig
new 895508d [STREAMPIPES-349] Add DataLakeResourceV4 to openapi.yaml
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (3fb5a97)
\
N -- N -- N refs/heads/STREAMPIPES-349 (895508d)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 29 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
[incubator-streampipes] 06/29: [STREAMPIPES-349] Add label
definition for data lake
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit b5bc9f8f9f898f5165f0b4b011bb6d69b3e48561
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Wed May 5 10:21:43 2021 +0200
[STREAMPIPES-349] Add label definition for data lake
---
.../model/datalake/LabelDefinition.java | 46 ++++++++++++++++++++++
1 file changed, 46 insertions(+)
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/LabelDefinition.java b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/LabelDefinition.java
new file mode 100644
index 0000000..60eb7db
--- /dev/null
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/LabelDefinition.java
@@ -0,0 +1,46 @@
+/*
+ * 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.model.datalake;
+
+public class LabelDefinition {
+
+ private String classLabel;
+ private String labelColumn;
+
+ public LabelDefinition(String classLabel, String labelColumn) {
+ this.classLabel = classLabel;
+ this.labelColumn = labelColumn;
+ }
+
+ public String getClassLabel() {
+ return classLabel;
+ }
+
+ public void setClassLabel(String classLabel) {
+ this.classLabel = classLabel;
+ }
+
+ public String getLabelColumn() {
+ return labelColumn;
+ }
+
+ public void setLabelColumn(String labelColumn) {
+ this.labelColumn = labelColumn;
+ }
+}
[incubator-streampipes] 05/29: [STREAMPIPES-349] Add data lake
configuration and retention policy defining classes
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 00767efb957e7ce0c8128eadfad695ed34812aad
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Wed May 5 10:18:32 2021 +0200
[STREAMPIPES-349] Add data lake configuration and retention policy defining classes
---
.../model/datalake/DataLakeConfiguration.java | 60 +++++++++++++++++++++
.../model/datalake/DataLakeRetentionPolicy.java | 62 ++++++++++++++++++++++
2 files changed, 122 insertions(+)
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeConfiguration.java b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeConfiguration.java
new file mode 100644
index 0000000..48c8875
--- /dev/null
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeConfiguration.java
@@ -0,0 +1,60 @@
+package org.apache.streampipes.model.datalake;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class DataLakeConfiguration {
+
+ private Integer batchSize;
+ private Integer flushDuration;
+
+ private List<DataLakeRetentionPolicy> retentionPolicies;
+
+ public DataLakeConfiguration() {
+ this.batchSize = 2000;
+ this.flushDuration = 500;
+ this.retentionPolicies = new ArrayList<DataLakeRetentionPolicy>();
+ }
+
+ public DataLakeConfiguration(List<DataLakeRetentionPolicy> retentionPolicies) {
+ this.batchSize = 2000;
+ this.flushDuration = 500;
+ this.retentionPolicies = retentionPolicies;
+ }
+
+ public DataLakeConfiguration(Integer batchSize, Integer flushDuration) {
+ this.batchSize = batchSize;
+ this.flushDuration = flushDuration;
+ this.retentionPolicies = new ArrayList<DataLakeRetentionPolicy>();
+ }
+
+ public DataLakeConfiguration(Integer batchSize, Integer flushDuration, List<DataLakeRetentionPolicy> retentionPolicies) {
+ this.batchSize = batchSize;
+ this.flushDuration = flushDuration;
+ this.retentionPolicies = retentionPolicies;
+ }
+
+ public List<DataLakeRetentionPolicy> getRetentionPolicies() {
+ return retentionPolicies;
+ }
+
+ public void setRetentionPolicies(List<DataLakeRetentionPolicy> retentionPolicies) {
+ this.retentionPolicies = retentionPolicies;
+ }
+
+ public Integer getBatchSize() {
+ return batchSize;
+ }
+
+ public void setBatchSize(Integer batchSize) {
+ this.batchSize = batchSize;
+ }
+
+ public Integer getFlushDuration() {
+ return flushDuration;
+ }
+
+ public void setFlushDuration(Integer flushDuration) {
+ this.flushDuration = flushDuration;
+ }
+}
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeRetentionPolicy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeRetentionPolicy.java
new file mode 100644
index 0000000..c069ce1
--- /dev/null
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeRetentionPolicy.java
@@ -0,0 +1,62 @@
+package org.apache.streampipes.model.datalake;
+
+public class DataLakeRetentionPolicy {
+
+ private String name;
+ private String durationLiteral;
+ private boolean isDefault;
+
+ public DataLakeRetentionPolicy(String name, int duration, String durationUnit, boolean isDefault) {
+ this.name = name;
+ this.durationLiteral = duration + durationUnit;
+ this.isDefault = isDefault;
+ }
+
+ public DataLakeRetentionPolicy(String name) {
+ this.name = name;
+ this.isDefault = false;
+ setDurationLiteralToInfinity();
+ }
+
+ public DataLakeRetentionPolicy(String name, String durationLiteral, boolean isDefault) {
+ this.name = name;
+ this.durationLiteral = durationLiteral;
+ this.isDefault = isDefault;
+ }
+
+ public String getDurationLiteral() {
+ return durationLiteral;
+ }
+
+ public void setDurationLiteral(int duration, String durationUnit) {
+ this.durationLiteral = duration + durationUnit;
+ }
+
+ public void setDurationLiteralToInfinity() {
+ this.durationLiteral = "INF";
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public boolean isDefault() {
+ return isDefault;
+ }
+
+ public void setDefault(boolean isDefault) {
+ this.isDefault = isDefault;
+ }
+
+ @Override
+ public String toString() {
+ return "DataLakeRetentionPolicy{" +
+ "name='" + name + '\'' +
+ ", durationLiteral='" + durationLiteral + '\'' +
+ '}';
+ }
+}
[incubator-streampipes] 22/29: [STREAMPIPES-349] Add implementation
of endpoint for removing data from a single measurement series
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 228b5d137dc1f10561cd2501eca0789f33c9aa7b
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 19:57:54 2021 +0200
[STREAMPIPES-349] Add implementation of endpoint for removing data from a single measurement series
---
.../dataexplorer/DataLakeManagementV4.java | 21 +++++++++++++++++++++
.../apache/streampipes/ps/DataLakeResourceV4.java | 12 ++++++------
2 files changed, 27 insertions(+), 6 deletions(-)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
index f1bc058..76a7577 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
@@ -23,12 +23,19 @@ import org.apache.streampipes.dataexplorer.query.DeleteDataQuery;
import org.apache.streampipes.dataexplorer.query.EditRetentionPolicyQuery;
import org.apache.streampipes.dataexplorer.query.ShowRetentionPolicyQuery;
import org.apache.streampipes.dataexplorer.utils.DataExplorerUtils;
+import org.apache.streampipes.dataexplorer.v4.params.DeleteFromStatementParams;
+import org.apache.streampipes.dataexplorer.v4.params.QueryParamsV4;
+import org.apache.streampipes.dataexplorer.v4.params.TimeBoundaryParams;
+import org.apache.streampipes.dataexplorer.v4.query.DataExplorerQueryV4;
import org.apache.streampipes.model.datalake.DataLakeConfiguration;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
import org.apache.streampipes.model.datalake.DataLakeRetentionPolicy;
+import org.apache.streampipes.model.datalake.DataResult;
import org.influxdb.dto.QueryResult;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
public class DataLakeManagementV4 {
@@ -49,6 +56,11 @@ public class DataLakeManagementV4 {
return true;
}
+ public DataResult deleteData(String measurementID, Long startDate, Long endDate) {
+ Map<String, QueryParamsV4> queryParts = getDeleteQueryParams(measurementID, startDate, endDate);
+ return new DataExplorerQueryV4(queryParts).executeQuery();
+ }
+
public DataLakeConfiguration getDataLakeConfiguration() {
List<DataLakeRetentionPolicy> retentionPolicies = getAllExistingRetentionPolicies();
return new DataLakeConfiguration(retentionPolicies);
@@ -91,4 +103,13 @@ public class DataLakeManagementV4 {
*/
return new ShowRetentionPolicyQuery(RetentionPolicyQueryParams.from("", "0s")).executeQuery();
}
+
+ public Map<String, QueryParamsV4> getDeleteQueryParams(String measurementID, Long startDate, Long endDate) {
+ Map<String, QueryParamsV4> queryParts = new HashMap<>();
+ queryParts.put("DELETE", DeleteFromStatementParams.from(measurementID));
+ if (startDate != null || endDate != null) {
+ queryParts.put("WHERE", TimeBoundaryParams.from(measurementID, startDate, endDate));
+ }
+ return queryParts;
+ }
}
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 91807c1..cbef9de 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
@@ -28,6 +28,7 @@ import io.swagger.v3.oas.annotations.responses.ApiResponse;
import org.apache.streampipes.dataexplorer.DataLakeManagementV4;
import org.apache.streampipes.model.datalake.DataLakeConfiguration;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.model.datalake.DataResult;
import org.apache.streampipes.rest.impl.AbstractRestResource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -76,12 +77,11 @@ public class DataLakeResourceV4 extends AbstractRestResource {
@ApiResponse(responseCode = "400", description = "Measurement series with given id not found")})
public Response deleteData(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username
, @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathParam("measurementID") String measurementID
- , @Parameter(in = ParameterIn.QUERY, description = "start date for slicing operation") @QueryParam("startDate") String startDate
- , @Parameter(in = ParameterIn.QUERY, description = "end date for slicing operation") @QueryParam("endDate") String endDate) {
- /**
- * TODO: implementation of method stump
- */
- return null;
+ , @Parameter(in = ParameterIn.QUERY, description = "start date for slicing operation") @QueryParam("startDate") Long startDate
+ , @Parameter(in = ParameterIn.QUERY, description = "end date for slicing operation") @QueryParam("endDate") Long endDate) {
+
+ DataResult result = this.dataLakeManagement.deleteData(measurementID, startDate, endDate);
+ return ok();
}
@GET
[incubator-streampipes] 18/29: [STREAMPIPES-349] Add query element
'Offset' and corresponding parameter class
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit acfc799f015d4e6a8371179ffcb78e954da88dae
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 19:14:55 2021 +0200
[STREAMPIPES-349] Add query element 'Offset' and corresponding parameter class
---
.../dataexplorer/v4/params/OffsetParams.java | 37 ++++++++++++++++++++++
.../dataexplorer/v4/query/elements/Offset.java | 34 ++++++++++++++++++++
2 files changed, 71 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/OffsetParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/OffsetParams.java
new file mode 100644
index 0000000..ac6a1b3
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/OffsetParams.java
@@ -0,0 +1,37 @@
+/*
+ * 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.v4.params;
+
+public class OffsetParams extends QueryParamsV4 {
+
+ private final Integer offset;
+
+ public static OffsetParams from(String measurementID, Integer offset) {
+ return new OffsetParams(measurementID, offset);
+ }
+
+ public OffsetParams(String measurementID, Integer offset) {
+ super(measurementID);
+ this.offset = offset;
+ }
+
+ public Integer getOffset() {
+ return offset;
+ }
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/Offset.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/Offset.java
new file mode 100644
index 0000000..c8c8c0a
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/Offset.java
@@ -0,0 +1,34 @@
+/*
+ * 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.v4.query.elements;
+
+import org.apache.streampipes.dataexplorer.v4.params.OffsetParams;
+import org.apache.streampipes.dataexplorer.v4.template.QueryTemplatesV4;
+
+public class Offset extends QueryElement<OffsetParams> {
+
+ public Offset(OffsetParams offsetParams) {
+ super(offsetParams);
+ }
+
+ @Override
+ protected String buildStatement(OffsetParams offsetParams) {
+ return QueryTemplatesV4.offset(offsetParams.getOffset());
+ }
+}
[incubator-streampipes] 08/29: [STREAMPIPES-349] Update definition
of configuration endpoints and add implementation
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 3c3023713262880d87eafddaeb92e30c368a54c4
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Wed May 5 10:30:09 2021 +0200
[STREAMPIPES-349] Update definition of configuration endpoints and add implementation
---
.../apache/streampipes/ps/DataLakeResourceV4.java | 32 +++++++++-------------
1 file changed, 13 insertions(+), 19 deletions(-)
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 f16dc3a..91807c1 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
@@ -26,6 +26,7 @@ import io.swagger.v3.oas.annotations.media.Content;
import io.swagger.v3.oas.annotations.media.Schema;
import io.swagger.v3.oas.annotations.responses.ApiResponse;
import org.apache.streampipes.dataexplorer.DataLakeManagementV4;
+import org.apache.streampipes.model.datalake.DataLakeConfiguration;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
import org.apache.streampipes.rest.impl.AbstractRestResource;
import org.slf4j.Logger;
@@ -57,17 +58,14 @@ public class DataLakeResourceV4 extends AbstractRestResource {
@POST
- @Path("/measurements/{measurementID}/configuration")
+ @Path("/configuration")
@Consumes(MediaType.APPLICATION_JSON)
- @Operation(summary = "Configure parameters of measurement series with a given id", tags = {"Data Lake"},
+ @Operation(summary = "Configure the parameters of the data lake", tags = {"Data Lake"},
responses = {@ApiResponse(responseCode = "200", description = "Configuration was successful")})
- public Response configureMeasurement(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username
- , @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathParam("measurementID") String measurementID
- , @Parameter(in = ParameterIn.DEFAULT, description = "the configuration parameters") Placeholder body) {
- /**
- * TODO: implementation of method stump
- */
- return null;
+ public Response configureMeasurement(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username,
+ @Parameter(in = ParameterIn.QUERY, description = "should any parameter be reset to its default value?") @DefaultValue("false") @QueryParam("resetToDefault") boolean resetToDefault,
+ @Parameter(in = ParameterIn.DEFAULT, description = "the configuration parameters") DataLakeConfiguration config) {
+ return ok(this.dataLakeManagement.editMeasurementConfiguration(config, resetToDefault));
}
@DELETE
@@ -122,18 +120,14 @@ public class DataLakeResourceV4 extends AbstractRestResource {
}
@GET
- @Path("/measurements/{measurementID}/configuration")
+ @Path("/configuration")
@Produces(MediaType.APPLICATION_JSON)
- @Operation(summary = "Get configuration parameters of a measurement series by a given id", tags = {"Data Lake"},
+ @Operation(summary = "Get the configuration parameters of the data lake", tags = {"Data Lake"},
responses = {
- @ApiResponse(responseCode = "200", description = "configuration parameters", content = @Content(schema = @Schema(implementation = Placeholder.class)))})
- public Response getMeasurementConfiguration(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username
- , @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathParam("measurementID") String measurementID
- , @Parameter(in = ParameterIn.QUERY, description = "the id of a specific configuration parameter") @QueryParam("parameterID") String parameterID) {
- /**
- * TODO: implementation of method stump
- */
- return null;
+ @ApiResponse(responseCode = "200", description = "configuration parameters", content = @Content(schema = @Schema(implementation = DataLakeConfiguration.class)))})
+ public Response getMeasurementConfiguration(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username,
+ @Parameter(in = ParameterIn.QUERY, description = "the id of a specific configuration parameter") @QueryParam("parameterID") String parameterID) {
+ return ok(this.dataLakeManagement.getDataLakeConfiguration());
}
@POST
[incubator-streampipes] 07/29: [STREAMPIPES-349] Add query defining
classes for changes at the retention policies
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit f3ced17db8cdb8e1d882e457f74c832a25163c3f
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Wed May 5 10:24:49 2021 +0200
[STREAMPIPES-349] Add query defining classes for changes at the retention policies
---
.../param/RetentionPolicyQueryParams.java | 20 ++++++++
.../query/EditRetentionPolicyQuery.java | 56 ++++++++++++++++++++++
.../query/ShowRetentionPolicyQuery.java | 38 +++++++++++++++
3 files changed, 114 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/RetentionPolicyQueryParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/RetentionPolicyQueryParams.java
new file mode 100644
index 0000000..e224570
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/RetentionPolicyQueryParams.java
@@ -0,0 +1,20 @@
+package org.apache.streampipes.dataexplorer.param;
+
+public class RetentionPolicyQueryParams extends QueryParams {
+ private final String durationLiteral;
+
+ public static RetentionPolicyQueryParams from(String index, String durationLiteral) {
+ return new RetentionPolicyQueryParams(index, durationLiteral);
+ }
+
+ protected RetentionPolicyQueryParams(String index, String durationLiteral) {
+ super(index);
+ this.durationLiteral = durationLiteral;
+ }
+
+ public String getDurationLiteral() {
+ return durationLiteral;
+ }
+
+
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/EditRetentionPolicyQuery.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/EditRetentionPolicyQuery.java
new file mode 100644
index 0000000..ddaed73
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/EditRetentionPolicyQuery.java
@@ -0,0 +1,56 @@
+package org.apache.streampipes.dataexplorer.query;
+
+import org.apache.streampipes.dataexplorer.param.RetentionPolicyQueryParams;
+import org.influxdb.dto.QueryResult;
+
+public class EditRetentionPolicyQuery extends ParameterizedDataExplorerQuery<RetentionPolicyQueryParams, String> {
+
+ private static final String CREATE_OPERATOR = "CREATE";
+ private static final String ALTER_OPERATOR = "ALTER";
+ private static final String DROP_OPERATOR = "DROP";
+ private static final String RESET_OPERATOR = "DEFAULT";
+
+ private String operationToPerform;
+
+ public EditRetentionPolicyQuery(RetentionPolicyQueryParams queryParams, String operation) {
+ super(queryParams);
+ this.operationToPerform = operation;
+ }
+
+
+ @Override
+ protected void getQuery(DataExplorerQueryBuilder queryBuilder) {
+ if (this.operationToPerform.equals(CREATE_OPERATOR)) {
+ queryBuilder.add(createRetentionPolicyStatement(params.getIndex()));
+ } else if (this.operationToPerform.equals(ALTER_OPERATOR)) {
+ queryBuilder.add(alterRetentionPolicyStatement(params.getIndex()));
+ } else if (this.operationToPerform.equals(DROP_OPERATOR)) {
+ queryBuilder.add(dropRetentionPolicyStatement(params.getIndex()));
+ } else if (this.operationToPerform.equals(RESET_OPERATOR)) {
+ queryBuilder.add(resetRetentionPolicyStatement());
+ }
+
+ }
+
+ @Override
+ protected String postQuery(QueryResult result) throws RuntimeException {
+ return result.toString();
+ }
+
+ private String createRetentionPolicyStatement(String index) {
+ return "CREATE RETENTION POLICY " + index + " ON " + "sp DURATION " + params.getDurationLiteral() + " REPLICATION 1 DEFAULT";
+ }
+
+ private String alterRetentionPolicyStatement(String index) {
+ return "ALTER RETENTION POLICY " + index + " ON " + "sp DURATION " + params.getDurationLiteral() + " REPLICATION 1 DEFAULT";
+ }
+
+ private String dropRetentionPolicyStatement(String index) {
+ return "DROP RETENTION POLICY " + index + " ON " + "sp";
+ }
+
+ private String resetRetentionPolicyStatement() {
+ return "ALTER RETENTION POLICY " + "autogen" + " ON " + "sp DURATION " + "0s" + " REPLICATION 1 DEFAULT";
+ }
+
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/ShowRetentionPolicyQuery.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/ShowRetentionPolicyQuery.java
new file mode 100644
index 0000000..5c4c3d8
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/ShowRetentionPolicyQuery.java
@@ -0,0 +1,38 @@
+package org.apache.streampipes.dataexplorer.query;
+
+import org.apache.streampipes.dataexplorer.param.RetentionPolicyQueryParams;
+import org.apache.streampipes.model.datalake.DataLakeRetentionPolicy;
+import org.influxdb.dto.QueryResult;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ShowRetentionPolicyQuery extends ParameterizedDataExplorerQuery<RetentionPolicyQueryParams, List<DataLakeRetentionPolicy>> {
+
+ public ShowRetentionPolicyQuery(RetentionPolicyQueryParams queryParams) {
+ super(queryParams);
+ }
+
+
+ @Override
+ protected void getQuery(DataExplorerQueryBuilder queryBuilder) {
+ queryBuilder.add(showRetentionPolicyStatement());
+ }
+
+ @Override
+ protected List<DataLakeRetentionPolicy> postQuery(QueryResult result) throws RuntimeException {
+ List<DataLakeRetentionPolicy> policies = new ArrayList<>();
+ for (List<Object> a : result.getResults().get(0).getSeries().get(0).getValues()) {
+ boolean isDefault = false;
+ if (a.get(4).toString().equals("true")) {
+ isDefault = true;
+ }
+ policies.add(new DataLakeRetentionPolicy(a.get(0).toString(), a.get(1).toString(), isDefault));
+ }
+ return policies;
+ }
+
+ private String showRetentionPolicyStatement() {
+ return "SHOW RETENTION POLICIES ON " + "sp";
+ }
+}
[incubator-streampipes] 14/29: Add query element "Ordering By Time"
and corresponding parameter class
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit e7a68d0aae9ad26f0b83a4cba65ee013fbe7b73e
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 19:06:01 2021 +0200
Add query element "Ordering By Time" and corresponding parameter class
---
.../v4/params/OrderingByTimeParams.java | 36 ++++++++++++++++++++++
.../v4/query/elements/OrderingByTime.java | 34 ++++++++++++++++++++
2 files changed, 70 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/OrderingByTimeParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/OrderingByTimeParams.java
new file mode 100644
index 0000000..b7b71e9
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/OrderingByTimeParams.java
@@ -0,0 +1,36 @@
+/*
+ * 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.v4.params;
+
+public class OrderingByTimeParams extends QueryParamsV4 {
+ private final String ordering;
+
+ public static OrderingByTimeParams from(String measurementID, String ordering) {
+ return new OrderingByTimeParams(measurementID, ordering);
+ }
+
+ public OrderingByTimeParams(String measurementID, String ordering) {
+ super(measurementID);
+ this.ordering = ordering;
+ }
+
+ public String getOrdering() {
+ return this.ordering;
+ }
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/OrderingByTime.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/OrderingByTime.java
new file mode 100644
index 0000000..877d7e0
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/OrderingByTime.java
@@ -0,0 +1,34 @@
+/*
+ * 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.v4.query.elements;
+
+import org.apache.streampipes.dataexplorer.v4.params.OrderingByTimeParams;
+import org.apache.streampipes.dataexplorer.v4.template.QueryTemplatesV4;
+
+public class OrderingByTime extends QueryElement<OrderingByTimeParams> {
+
+ public OrderingByTime(OrderingByTimeParams orderingByTimeParams) {
+ super(orderingByTimeParams);
+ }
+
+ @Override
+ protected String buildStatement(OrderingByTimeParams orderingByTimeParams) {
+ return QueryTemplatesV4.orderByTime(orderingByTimeParams.getOrdering());
+ }
+}
[incubator-streampipes] 25/29: [STREAMPIPES-349] Add implementation
of endpoint for downloading data from data lake
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit c9ec353dc3942f38499fd66b7c4bc10431959f56
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 20:07:26 2021 +0200
[STREAMPIPES-349] Add implementation of endpoint for downloading data from data lake
---
.../dataexplorer/DataLakeManagementV4.java | 123 +++++++++++++++++++++
.../apache/streampipes/ps/DataLakeResourceV4.java | 15 ++-
2 files changed, 135 insertions(+), 3 deletions(-)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
index 8e175b8..b2b7192 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
@@ -18,6 +18,7 @@
package org.apache.streampipes.dataexplorer;
+import com.google.gson.Gson;
import org.apache.streampipes.dataexplorer.param.RetentionPolicyQueryParams;
import org.apache.streampipes.dataexplorer.query.DeleteDataQuery;
import org.apache.streampipes.dataexplorer.query.EditRetentionPolicyQuery;
@@ -31,6 +32,10 @@ import org.apache.streampipes.model.datalake.DataLakeRetentionPolicy;
import org.apache.streampipes.model.datalake.DataResult;
import org.influxdb.dto.QueryResult;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -47,6 +52,120 @@ public class DataLakeManagementV4 {
return new DataExplorerQueryV4(queryParts).executeQuery();
}
+ public void getDataAsStream(String measurementID, Long startDate, Long endDate, Integer page, Integer limit, Integer offset, String groupBy, String order, String aggregationFunction, String timeInterval, String format, OutputStream outputStream) throws IOException {
+ if (limit == null) {
+ limit = 500000;
+ }
+ SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
+
+ DataResult dataResult;
+ //JSON
+ if (format.equals("json")) {
+
+ Gson gson = new Gson();
+ int i = 0;
+ if (page != null) {
+ i = page;
+ }
+
+ boolean isFirstDataObject = true;
+
+ outputStream.write(toBytes("["));
+ do {
+ dataResult = getData(measurementID, startDate, endDate, i, limit, null, groupBy, order, aggregationFunction, timeInterval);
+
+ if (dataResult.getTotal() > 0) {
+ for (List<Object> row : dataResult.getRows()) {
+ if (!isFirstDataObject) {
+ outputStream.write(toBytes(","));
+ }
+
+ //produce one json object
+ boolean isFirstElementInRow = true;
+ outputStream.write(toBytes("{"));
+ for (int i1 = 0; i1 < row.size(); i1++) {
+ Object element = row.get(i1);
+ if (!isFirstElementInRow) {
+ outputStream.write(toBytes(","));
+ }
+ isFirstElementInRow = false;
+ if (i1 == 0) {
+ try {
+ element = formatter.parse(element.toString()).getTime();
+ } catch (ParseException e) {
+ element = element.toString();
+ }
+ }
+ //produce json e.g. "name": "Pipes" or "load": 42
+ outputStream.write(toBytes("\"" + dataResult.getHeaders().get(i1) + "\": "
+ + gson.toJson(element)));
+ }
+ outputStream.write(toBytes("}"));
+ isFirstDataObject = false;
+ }
+
+ i++;
+ }
+ } while (dataResult.getTotal() > 0);
+ outputStream.write(toBytes("]"));
+
+ //CSV
+ } else if (format.equals("csv")) {
+ int i = 0;
+ if (page != null) {
+ i = page;
+ }
+
+ boolean isFirstDataObject = true;
+
+ do {
+ dataResult = getData(measurementID, startDate, endDate, i, limit, null, groupBy, order, aggregationFunction, timeInterval);
+ //Send first header
+ if (dataResult.getTotal() > 0) {
+ if (isFirstDataObject) {
+ boolean isFirst = true;
+ for (int i1 = 0; i1 < dataResult.getHeaders().size(); i1++) {
+ if (!isFirst) {
+ outputStream.write(toBytes(";"));
+ }
+ isFirst = false;
+ outputStream.write(toBytes(dataResult.getHeaders().get(i1)));
+ }
+ }
+ outputStream.write(toBytes("\n"));
+ isFirstDataObject = false;
+ }
+
+ if (dataResult.getTotal() > 0) {
+ for (List<Object> row : dataResult.getRows()) {
+ boolean isFirstInRow = true;
+ for (int i1 = 0; i1 < row.size(); i1++) {
+ Object element = row.get(i1);
+ if (!isFirstInRow) {
+ outputStream.write(toBytes(";"));
+ }
+ isFirstInRow = false;
+ if (i1 == 0) {
+ try {
+ element = formatter.parse(element.toString()).getTime();
+ } catch (ParseException e) {
+ element = element.toString();
+ }
+ }
+ if (element == null) {
+ outputStream.write(toBytes(""));
+ } else {
+ outputStream.write(toBytes(element.toString()));
+ }
+ }
+ outputStream.write(toBytes("\n"));
+ }
+ }
+ i++;
+ } while (dataResult.getTotal() > 0);
+ }
+ }
+
public boolean removeAllMeasurements() {
List<DataLakeMeasure> allMeasurements = getAllMeasurements();
@@ -157,4 +276,8 @@ public class DataLakeManagementV4 {
}
return queryParts;
}
+
+ private byte[] toBytes(String value) {
+ return value.getBytes();
+ }
}
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 caf76e8..f2bd49b 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
@@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory;
import javax.ws.rs.*;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
+import javax.ws.rs.core.StreamingOutput;
import java.util.List;
class Placeholder {
@@ -139,9 +140,17 @@ public class DataLakeResourceV4 extends AbstractRestResource {
, @Parameter(in = ParameterIn.QUERY, description = "name of aggregation function used for grouping operation") @QueryParam("aggregationFunction") String aggregationFunction
, @Parameter(in = ParameterIn.QUERY, description = "time interval for aggregation (e.g. 1m - one minute) for grouping operation") @QueryParam("timeInterval") String timeInterval
, @Parameter(in = ParameterIn.QUERY, description = "format specification (csv, json - default is csv) for data download") @QueryParam("format") String format) {
- /**
- * TODO: implementation of method stump
- */
+
+ if (format == null) {
+ format = "csv";
+ }
+ String outputFormat = format;
+
+ StreamingOutput streamingOutput = output -> dataLakeManagement.getDataAsStream(measurementID, startDate, endDate, page, limit, offset, groupBy, order, aggregationFunction, timeInterval, outputFormat, output);
+
+ return Response.ok(streamingOutput, MediaType.APPLICATION_OCTET_STREAM).
+ header("Content-Disposition", "attachment; filename=\"datalake." + outputFormat + "\"")
+ .build();
}
@GET
[incubator-streampipes] 17/29: [STREAMPIPES-349] Add query element
'Item Limitation' and corresponding parameter class
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 7613792d3bdc7a2e79ecdb281469af7875d3479e
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 19:13:26 2021 +0200
[STREAMPIPES-349] Add query element 'Item Limitation' and corresponding parameter class
---
.../v4/params/ItemLimitationParams.java | 37 ++++++++++++++++++++++
.../v4/query/elements/ItemLimitation.java | 34 ++++++++++++++++++++
2 files changed, 71 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/ItemLimitationParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/ItemLimitationParams.java
new file mode 100644
index 0000000..6656ad5
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/ItemLimitationParams.java
@@ -0,0 +1,37 @@
+/*
+ * 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.v4.params;
+
+public class ItemLimitationParams extends QueryParamsV4 {
+
+ private final Integer limit;
+
+ public static ItemLimitationParams from(String measurementID, Integer limit) {
+ return new ItemLimitationParams(measurementID, limit);
+ }
+
+ public ItemLimitationParams(String measurementID, Integer limit) {
+ super(measurementID);
+ this.limit = limit;
+ }
+
+ public Integer getLimit() {
+ return limit;
+ }
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/ItemLimitation.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/ItemLimitation.java
new file mode 100644
index 0000000..b4f42e0
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/ItemLimitation.java
@@ -0,0 +1,34 @@
+/*
+ * 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.v4.query.elements;
+
+import org.apache.streampipes.dataexplorer.v4.params.ItemLimitationParams;
+import org.apache.streampipes.dataexplorer.v4.template.QueryTemplatesV4;
+
+public class ItemLimitation extends QueryElement<ItemLimitationParams> {
+
+ public ItemLimitation(ItemLimitationParams itemLimitationParams) {
+ super(itemLimitationParams);
+ }
+
+ @Override
+ protected String buildStatement(ItemLimitationParams itemLimitationParams) {
+ return QueryTemplatesV4.limitItems(itemLimitationParams.getLimit());
+ }
+}
[incubator-streampipes] 29/29: [STREAMPIPES-349] Add
DataLakeResourceV4 to openapi.yaml
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 895508d5188a51f292882496a034147b6626e506
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 20:17:18 2021 +0200
[STREAMPIPES-349] Add DataLakeResourceV4 to openapi.yaml
---
streampipes-backend/src/main/resources/openapi.yaml | 1 +
1 file changed, 1 insertion(+)
diff --git a/streampipes-backend/src/main/resources/openapi.yaml b/streampipes-backend/src/main/resources/openapi.yaml
index db3b2a6..77a224d 100644
--- a/streampipes-backend/src/main/resources/openapi.yaml
+++ b/streampipes-backend/src/main/resources/openapi.yaml
@@ -2,6 +2,7 @@ scannerClass: io.swagger.v3.oas.integration.GenericOpenApiScanner
resourceClasses:
- org.apache.streampipes.ps.PipelineElementTemplateResource
- org.apache.streampipes.rest.impl.PipelineResource
+ - org.apache.streampipes.ps.DataLakeResourceV4
prettyPrint: true
cacheTTL: 0
openAPI:
[incubator-streampipes] 12/29: [STREAMPIPES-349] Add query element
'Select From Statement' and corresponding parameter class
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 84f16a616c419093752e19583256f292771838f0
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 19:02:30 2021 +0200
[STREAMPIPES-349] Add query element 'Select From Statement' and corresponding parameter class
---
.../v4/params/SelectFromStatementParams.java | 45 ++++++++++++++++++++++
.../v4/query/elements/SelectFromStatement.java | 37 ++++++++++++++++++
2 files changed, 82 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/SelectFromStatementParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/SelectFromStatementParams.java
new file mode 100644
index 0000000..0385963
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/SelectFromStatementParams.java
@@ -0,0 +1,45 @@
+/*
+ * 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.v4.params;
+
+import javax.annotation.Nullable;
+
+public class SelectFromStatementParams extends QueryParamsV4 {
+
+ private final String aggregationFunction;
+
+ public static SelectFromStatementParams from(String measurementID, @Nullable String aggregationFunction) {
+ return new SelectFromStatementParams(measurementID, aggregationFunction);
+ }
+
+ public SelectFromStatementParams(String measurementID) {
+ super(measurementID);
+ this.aggregationFunction = null;
+ }
+
+ public SelectFromStatementParams(String measurementID, String aggregationFunction) {
+ super(measurementID);
+ this.aggregationFunction = aggregationFunction;
+ }
+
+
+ public String getAggregationFunction() {
+ return aggregationFunction;
+ }
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/SelectFromStatement.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/SelectFromStatement.java
new file mode 100644
index 0000000..efd7654
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/SelectFromStatement.java
@@ -0,0 +1,37 @@
+/*
+ * 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.v4.query.elements;
+
+import org.apache.streampipes.dataexplorer.v4.params.SelectFromStatementParams;
+import org.apache.streampipes.dataexplorer.v4.template.QueryTemplatesV4;
+
+public class SelectFromStatement extends QueryElement<SelectFromStatementParams> {
+ public SelectFromStatement(SelectFromStatementParams selectFromStatementParams) {
+ super(selectFromStatementParams);
+ }
+
+ @Override
+ protected String buildStatement(SelectFromStatementParams selectFromStatementParams) {
+ if (selectFromStatementParams.getAggregationFunction() == null) {
+ return QueryTemplatesV4.selectWildcardFrom(selectFromStatementParams.getIndex());
+ } else {
+ return QueryTemplatesV4.selectAggregationFrom(selectFromStatementParams.getIndex(), selectFromStatementParams.getAggregationFunction());
+ }
+ }
+}
[incubator-streampipes] 26/29: [STREAMPIPES-349] Add endpoint
definition for dropping entire measurement series from data lake and
removing related event property
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 8b4e07c225b5a683aa9249920b76973adb3e48f3
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 20:11:40 2021 +0200
[STREAMPIPES-349] Add endpoint definition for dropping entire measurement series from data lake and removing related event property
---
.../java/org/apache/streampipes/ps/DataLakeResourceV4.java | 14 ++++++++++++++
1 file changed, 14 insertions(+)
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 f2bd49b..53b3414 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
@@ -85,6 +85,20 @@ public class DataLakeResourceV4 extends AbstractRestResource {
return ok();
}
+ @DELETE
+ @Path("/measurements/{measurementID}/drop")
+ @Operation(summary = "Drop a single measurement series with given id from Data Lake and remove related event property", tags = {"Data Lake"},
+ responses = {
+ @ApiResponse(responseCode = "200", description = "Measurement series successfully dropped from Data Lake"),
+ @ApiResponse(responseCode = "400", description = "Measurement series with given id or related event property not found")})
+ public Response dropMeasurementSeries(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username
+ , @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathParam("measurementID") String measurementID) {
+
+ /**
+ * TODO: implementation of method stump
+ */
+ }
+
@GET
@Path("/measurements")
@Produces(MediaType.APPLICATION_JSON)
[incubator-streampipes] 19/29: [STREAMPIPES-349] Add query element
'Delete From Statement' and corresponding parameter class
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 60a69b007bf4fd430742fd1e56dd1bdfeba96ff5
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 19:16:10 2021 +0200
[STREAMPIPES-349] Add query element 'Delete From Statement' and corresponding parameter class
---
.../v4/params/DeleteFromStatementParams.java | 31 ++++++++++++++++++++
.../v4/query/elements/DeleteFromStatement.java | 33 ++++++++++++++++++++++
2 files changed, 64 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/DeleteFromStatementParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/DeleteFromStatementParams.java
new file mode 100644
index 0000000..df93347
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/DeleteFromStatementParams.java
@@ -0,0 +1,31 @@
+/*
+ * 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.v4.params;
+
+public class DeleteFromStatementParams extends QueryParamsV4 {
+
+ public static DeleteFromStatementParams from(String measurementID) {
+ return new DeleteFromStatementParams(measurementID);
+ }
+
+ public DeleteFromStatementParams(String measurementID) {
+ super(measurementID);
+ }
+
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/DeleteFromStatement.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/DeleteFromStatement.java
new file mode 100644
index 0000000..cabd8b6
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/DeleteFromStatement.java
@@ -0,0 +1,33 @@
+/*
+ * 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.v4.query.elements;
+
+import org.apache.streampipes.dataexplorer.v4.params.DeleteFromStatementParams;
+import org.apache.streampipes.dataexplorer.v4.template.QueryTemplatesV4;
+
+public class DeleteFromStatement extends QueryElement<DeleteFromStatementParams> {
+ public DeleteFromStatement(DeleteFromStatementParams deleteFromStatementParams) {
+ super(deleteFromStatementParams);
+ }
+
+ @Override
+ protected String buildStatement(DeleteFromStatementParams deleteFromStatementParams) {
+ return QueryTemplatesV4.deleteFrom(deleteFromStatementParams.getIndex());
+ }
+}
[incubator-streampipes] 13/29: [STREAMPIPES-349] Add query element
'Time Boundary' and corresponding parameter class
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit a79e5eb0223fd0b08f1c00955574801889896aa3
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 19:03:56 2021 +0200
[STREAMPIPES-349] Add query element 'Time Boundary' and corresponding parameter class
---
.../dataexplorer/v4/params/TimeBoundaryParams.java | 43 ++++++++++++++++++++++
.../v4/query/elements/TimeBoundary.java | 40 ++++++++++++++++++++
2 files changed, 83 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/TimeBoundaryParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/TimeBoundaryParams.java
new file mode 100644
index 0000000..c893829
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/TimeBoundaryParams.java
@@ -0,0 +1,43 @@
+/*
+ * 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.v4.params;
+
+public class TimeBoundaryParams extends QueryParamsV4 {
+
+ private final Long startDate;
+ private final Long endDate;
+
+ public static TimeBoundaryParams from(String measurementID, Long startDate, Long endDate) {
+ return new TimeBoundaryParams(measurementID, startDate, endDate);
+ }
+
+ protected TimeBoundaryParams(String measurementID, Long startDate, Long endDate) {
+ super(measurementID);
+ this.startDate = startDate;
+ this.endDate = endDate;
+ }
+
+ public Long getStartDate() {
+ return startDate;
+ }
+
+ public Long getEndDate() {
+ return endDate;
+ }
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/TimeBoundary.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/TimeBoundary.java
new file mode 100644
index 0000000..bae5f27
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/TimeBoundary.java
@@ -0,0 +1,40 @@
+/*
+ * 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.v4.query.elements;
+
+import org.apache.streampipes.dataexplorer.v4.params.TimeBoundaryParams;
+import org.apache.streampipes.dataexplorer.v4.template.QueryTemplatesV4;
+
+public class TimeBoundary extends QueryElement<TimeBoundaryParams> {
+
+ public TimeBoundary(TimeBoundaryParams timeBoundaryParams) {
+ super(timeBoundaryParams);
+ }
+
+ @Override
+ protected String buildStatement(TimeBoundaryParams timeBoundaryParams) {
+ if (timeBoundaryParams.getStartDate() == null) {
+ return QueryTemplatesV4.whereTimeRightBound(timeBoundaryParams.getEndDate());
+ } else if (timeBoundaryParams.getEndDate() == null) {
+ return QueryTemplatesV4.whereTimeLeftBound(timeBoundaryParams.getStartDate());
+ } else {
+ return QueryTemplatesV4.whereTimeWithin(timeBoundaryParams.getStartDate(), timeBoundaryParams.getEndDate());
+ }
+ }
+}
[incubator-streampipes] 27/29: [STREAMPIPES-349] Add implementation
of endpoint for removing entire measurement series and related event
property
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 307383deb17db3e6d4e90c440013a984aad52e43
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 20:13:54 2021 +0200
[STREAMPIPES-349] Add implementation of endpoint for removing entire measurement series and related event property
---
.../dataexplorer/DataLakeManagementV4.java | 39 ++++++++++++++++++++++
.../apache/streampipes/ps/DataLakeResourceV4.java | 15 +++++++--
2 files changed, 51 insertions(+), 3 deletions(-)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
index b2b7192..1eed837 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
@@ -19,6 +19,7 @@
package org.apache.streampipes.dataexplorer;
import com.google.gson.Gson;
+import com.google.gson.JsonObject;
import org.apache.streampipes.dataexplorer.param.RetentionPolicyQueryParams;
import org.apache.streampipes.dataexplorer.query.DeleteDataQuery;
import org.apache.streampipes.dataexplorer.query.EditRetentionPolicyQuery;
@@ -30,7 +31,9 @@ import org.apache.streampipes.model.datalake.DataLakeConfiguration;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
import org.apache.streampipes.model.datalake.DataLakeRetentionPolicy;
import org.apache.streampipes.model.datalake.DataResult;
+import org.apache.streampipes.storage.couchdb.utils.Utils;
import org.influxdb.dto.QueryResult;
+import org.lightcouch.CouchDbClient;
import java.io.IOException;
import java.io.OutputStream;
@@ -178,6 +181,21 @@ public class DataLakeManagementV4 {
return true;
}
+ public boolean removeMeasurement(String measurementID) {
+ List<DataLakeMeasure> allMeasurements = getAllMeasurements();
+ for (DataLakeMeasure measure : allMeasurements) {
+ if (measure.getMeasureName().equals(measurementID)) {
+ QueryResult queryResult = new DeleteDataQuery(new DataLakeMeasure(measurementID, null)).executeQuery();
+
+ if (queryResult.hasError() || queryResult.getResults().get(0).getError() != null) {
+ return false;
+ }
+ return true;
+ }
+ }
+ return false;
+ }
+
public DataResult deleteData(String measurementID, Long startDate, Long endDate) {
Map<String, QueryParamsV4> queryParts = getDeleteQueryParams(measurementID, startDate, endDate);
return new DataExplorerQueryV4(queryParts).executeQuery();
@@ -226,6 +244,27 @@ public class DataLakeManagementV4 {
return new ShowRetentionPolicyQuery(RetentionPolicyQueryParams.from("", "0s")).executeQuery();
}
+ public boolean removeEventProperty(String measurementID) {
+ boolean isSuccess = false;
+ CouchDbClient couchDbClient = Utils.getCouchDbDataLakeClient();
+ List<JsonObject> docs = couchDbClient.view("_all_docs").includeDocs(true).query(JsonObject.class);
+
+ for (JsonObject document : docs) {
+ if (document.get("measureName").toString().replace("\"", "").equals(measurementID)) {
+ couchDbClient.remove(document.get("_id").toString().replace("\"", ""), document.get("_rev").toString().replace("\"", ""));
+ isSuccess = true;
+ break;
+ }
+ }
+
+ try {
+ couchDbClient.close();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ return isSuccess;
+ }
+
private Map<String, QueryParamsV4> getQueryParams(String measurementID, Long startDate, Long endDate, Integer page, Integer limit, Integer offset, String groupBy, String order, String aggregationFunction, String timeInterval) {
Map<String, QueryParamsV4> queryParts = new HashMap<>();
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 53b3414..8692af0 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
@@ -94,9 +94,18 @@ public class DataLakeResourceV4 extends AbstractRestResource {
public Response dropMeasurementSeries(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username
, @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathParam("measurementID") String measurementID) {
- /**
- * TODO: implementation of method stump
- */
+ boolean isSuccessDataLake = this.dataLakeManagement.removeMeasurement(measurementID);
+
+ if (isSuccessDataLake) {
+ boolean isSuccessEventProperty = this.dataLakeManagement.removeEventProperty(measurementID);
+ if (isSuccessEventProperty) {
+ return ok();
+ } else {
+ return Response.status(Response.Status.NOT_FOUND).entity("Event property related to measurement series with given id not found.").build();
+ }
+ } else {
+ return Response.status(Response.Status.NOT_FOUND).entity("Measurement series with given id not found.").build();
+ }
}
@GET
[incubator-streampipes] 16/29: [STREAMPIPES-349] Add query element
'Grouping By Time' and corresponding parameter class
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 02063e7292c87bcbea787d76f8503fe555c62175
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 19:09:57 2021 +0200
[STREAMPIPES-349] Add query element 'Grouping By Time' and corresponding parameter class
---
.../v4/params/GroupingByTimeParams.java | 36 ++++++++++++++++++++++
.../v4/query/elements/GroupingByTime.java | 34 ++++++++++++++++++++
2 files changed, 70 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/GroupingByTimeParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/GroupingByTimeParams.java
new file mode 100644
index 0000000..1751bcc
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/GroupingByTimeParams.java
@@ -0,0 +1,36 @@
+/*
+ * 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.v4.params;
+
+public class GroupingByTimeParams extends QueryParamsV4 {
+ private final String timeInterval;
+
+ public static GroupingByTimeParams from(String measurementID, String timeInterval) {
+ return new GroupingByTimeParams(measurementID, timeInterval);
+ }
+
+ public GroupingByTimeParams(String measurementID, String timeInterval) {
+ super(measurementID);
+ this.timeInterval = timeInterval;
+ }
+
+ public String getTimeInterval() {
+ return this.timeInterval;
+ }
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/GroupingByTime.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/GroupingByTime.java
new file mode 100644
index 0000000..dc16b65
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/GroupingByTime.java
@@ -0,0 +1,34 @@
+/*
+ * 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.v4.query.elements;
+
+import org.apache.streampipes.dataexplorer.v4.params.GroupingByTimeParams;
+import org.apache.streampipes.dataexplorer.v4.template.QueryTemplatesV4;
+
+public class GroupingByTime extends QueryElement<GroupingByTimeParams> {
+
+ public GroupingByTime(GroupingByTimeParams groupingByTimeParams) {
+ super(groupingByTimeParams);
+ }
+
+ @Override
+ protected String buildStatement(GroupingByTimeParams groupingByTimeParams) {
+ return QueryTemplatesV4.groupByTime(groupingByTimeParams.getTimeInterval());
+ }
+}
[incubator-streampipes] 20/29: [STREAMPIPES-349] Add Query Builder
class
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit ba5c5a8facacc1d1cbcce2eeaa05543436b12c3c
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 19:19:43 2021 +0200
[STREAMPIPES-349] Add Query Builder class
---
.../dataexplorer/v4/query/QueryBuilder.java | 54 ++++++++++++++++++++++
1 file changed, 54 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/QueryBuilder.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/QueryBuilder.java
new file mode 100644
index 0000000..8606797
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/QueryBuilder.java
@@ -0,0 +1,54 @@
+/*
+ * 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.v4.query;
+
+import org.apache.streampipes.dataexplorer.v4.query.elements.QueryElement;
+import org.influxdb.dto.Query;
+
+import java.util.List;
+import java.util.StringJoiner;
+
+public class QueryBuilder {
+
+ private StringJoiner queryParts;
+ private String databaseName;
+
+ public static QueryBuilder create(String databaseName) {
+ return new QueryBuilder(databaseName);
+ }
+
+ private QueryBuilder(String databaseName) {
+ this.queryParts = new StringJoiner(" ");
+ this.databaseName = databaseName;
+ }
+
+ public Query build(List<QueryElement> queryElements) {
+ for (QueryElement queryPart : queryElements) {
+ this.queryParts.add(queryPart.getStatement());
+ }
+ return toQuery();
+ }
+
+ public Query toQuery() {
+ return new Query(this.queryParts.toString(), this.databaseName);
+ }
+}
+
+
+
[incubator-streampipes] 02/29: [STREAMPIPES-349] Add implementation
of endpoint for 'getting all stored measurement series'
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit d4e35f322fbc20d340f7180cb3428f1c936e8528
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Apr 30 10:06:29 2021 +0200
[STREAMPIPES-349] Add implementation of endpoint for 'getting all stored measurement series'
---
.../dataexplorer/DataLakeManagementV4.java | 32 ++++++++++++++++++++++
.../apache/streampipes/ps/DataLakeResourceV4.java | 24 ++++++++++++----
2 files changed, 51 insertions(+), 5 deletions(-)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
new file mode 100644
index 0000000..07f2b90
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
@@ -0,0 +1,32 @@
+/*
+ * 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;
+
+import org.apache.streampipes.dataexplorer.utils.DataExplorerUtils;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+
+import java.util.List;
+
+public class DataLakeManagementV4 {
+
+ public List<DataLakeMeasure> getAllMeasurementSeries() {
+ List<DataLakeMeasure> dataLakeMeasuresList = DataExplorerUtils.getInfos();
+ return dataLakeMeasuresList;
+ }
+}
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 35ac486..d6f51f6 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
@@ -25,20 +25,36 @@ import io.swagger.v3.oas.annotations.media.ArraySchema;
import io.swagger.v3.oas.annotations.media.Content;
import io.swagger.v3.oas.annotations.media.Schema;
import io.swagger.v3.oas.annotations.responses.ApiResponse;
+import org.apache.streampipes.dataexplorer.DataLakeManagementV4;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.rest.impl.AbstractRestResource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.ws.rs.*;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
+import java.util.List;
+
+class Placeholder {
+}
@Path("v4/users/{username}/datalake")
-public class DataLakeResourceV4 {
+public class DataLakeResourceV4 extends AbstractRestResource {
private static final Logger logger = LoggerFactory.getLogger(DataLakeResourceV4.class);
+ private DataLakeManagementV4 dataLakeManagement;
+
+ public DataLakeResourceV4() {
+ this.dataLakeManagement = new DataLakeManagementV4();
+ }
+
+ public DataLakeResourceV4(DataLakeManagementV4 dataLakeManagement) {
+ this.dataLakeManagement = dataLakeManagement;
+ }
+
@POST
@Path("/measurements/{measurementID}/configuration")
@@ -77,10 +93,8 @@ public class DataLakeResourceV4 {
responses = {
@ApiResponse(responseCode = "200", description = "array of stored measurement series", content = @Content(array = @ArraySchema(schema = @Schema(implementation = DataLakeMeasure.class))))})
public Response getAll(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username) {
- /**
- * TODO: implementation of method stump
- */
- return null;
+ List<DataLakeMeasure> allMeasurementSeries = this.dataLakeManagement.getAllMeasurementSeries();
+ return ok(allMeasurementSeries);
}
@GET
[incubator-streampipes] 01/29: [STREAMPIPES-349] Add skeleton of
the revised REST API
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit a3d138984242b72d5676212b1dde1b7ce0bfc667
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Thu Apr 29 09:55:53 2021 +0200
[STREAMPIPES-349] Add skeleton of the revised REST API
---
.../apache/streampipes/ps/DataLakeResourceV4.java | 154 +++++++++++++++++++++
1 file changed, 154 insertions(+)
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
new file mode 100644
index 0000000..35ac486
--- /dev/null
+++ b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java
@@ -0,0 +1,154 @@
+/*
+ * 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.ps;
+
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.enums.ParameterIn;
+import io.swagger.v3.oas.annotations.media.ArraySchema;
+import io.swagger.v3.oas.annotations.media.Content;
+import io.swagger.v3.oas.annotations.media.Schema;
+import io.swagger.v3.oas.annotations.responses.ApiResponse;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.ws.rs.*;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+
+
+@Path("v4/users/{username}/datalake")
+public class DataLakeResourceV4 {
+
+ private static final Logger logger = LoggerFactory.getLogger(DataLakeResourceV4.class);
+
+
+ @POST
+ @Path("/measurements/{measurementID}/configuration")
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Operation(summary = "Configure parameters of measurement series with a given id", tags = {"Data Lake"},
+ responses = {@ApiResponse(responseCode = "200", description = "Configuration was successful")})
+ public Response configureMeasurement(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username
+ , @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathParam("measurementID") String measurementID
+ , @Parameter(in = ParameterIn.DEFAULT, description = "the configuration parameters") Placeholder body) {
+ /**
+ * TODO: implementation of method stump
+ */
+ return null;
+ }
+
+ @DELETE
+ @Path("/measurements/{measurementID}")
+ @Operation(summary = "Remove data from a single measurement series with given id", tags = {"Data Lake"},
+ responses = {
+ @ApiResponse(responseCode = "200", description = "Data from measurement series successfully removed"),
+ @ApiResponse(responseCode = "400", description = "Measurement series with given id not found")})
+ public Response deleteData(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username
+ , @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathParam("measurementID") String measurementID
+ , @Parameter(in = ParameterIn.QUERY, description = "start date for slicing operation") @QueryParam("startDate") String startDate
+ , @Parameter(in = ParameterIn.QUERY, description = "end date for slicing operation") @QueryParam("endDate") String endDate) {
+ /**
+ * TODO: implementation of method stump
+ */
+ return null;
+ }
+
+ @GET
+ @Path("/measurements")
+ @Produces(MediaType.APPLICATION_JSON)
+ @Operation(summary = "Get a list of all measurement series", tags = {"Data Lake"},
+ responses = {
+ @ApiResponse(responseCode = "200", description = "array of stored measurement series", content = @Content(array = @ArraySchema(schema = @Schema(implementation = DataLakeMeasure.class))))})
+ public Response getAll(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username) {
+ /**
+ * TODO: implementation of method stump
+ */
+ return null;
+ }
+
+ @GET
+ @Path("/measurements/{measurementID}")
+ @Produces(MediaType.APPLICATION_JSON)
+ @Operation(summary = "Get data from a single measurement series by a given id", tags = {"Data Lake"},
+ responses = {
+ @ApiResponse(responseCode = "400", description = "Measurement series with given id and requested query specification not found"),
+ @ApiResponse(responseCode = "200", description = "requested data", content = @Content(schema = @Schema(implementation = Placeholder.class)))})
+ public Response getData(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username
+ , @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathParam("measurementID") String measurementID
+ , @Parameter(in = ParameterIn.QUERY, description = "start date for slicing operation") @QueryParam("startDate") String startDate
+ , @Parameter(in = ParameterIn.QUERY, description = "end date for slicing operation") @QueryParam("endDate") String endDate
+ , @Parameter(in = ParameterIn.QUERY, description = "page number for paging operation") @QueryParam("page") String page
+ , @Parameter(in = ParameterIn.QUERY, description = "items per page limitation for paging operation") @QueryParam("limit") Integer limit
+ , @Parameter(in = ParameterIn.QUERY, description = "offset for paging operation") @QueryParam("offset") Integer offset
+ , @Parameter(in = ParameterIn.QUERY, description = "grouping tags (comma-separated) for grouping operation") @QueryParam("groupBy") String groupBy
+ , @Parameter(in = ParameterIn.QUERY, description = "name of aggregation function used for grouping operation") @QueryParam("aggregationFunction") String aggregationFunction
+ , @Parameter(in = ParameterIn.QUERY, description = "time interval for aggregation (e.g. 1m - one minute) for grouping operation") @QueryParam("timeInterval") String timeInterval
+ , @Parameter(in = ParameterIn.QUERY, description = "format specification (csv, json) for data download") @QueryParam("format") String format) {
+ /**
+ * TODO: implementation of method stump
+ */
+ return null;
+ }
+
+ @GET
+ @Path("/measurements/{measurementID}/configuration")
+ @Produces(MediaType.APPLICATION_JSON)
+ @Operation(summary = "Get configuration parameters of a measurement series by a given id", tags = {"Data Lake"},
+ responses = {
+ @ApiResponse(responseCode = "200", description = "configuration parameters", content = @Content(schema = @Schema(implementation = Placeholder.class)))})
+ public Response getMeasurementConfiguration(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username
+ , @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathParam("measurementID") String measurementID
+ , @Parameter(in = ParameterIn.QUERY, description = "the id of a specific configuration parameter") @QueryParam("parameterID") String parameterID) {
+ /**
+ * TODO: implementation of method stump
+ */
+ return null;
+ }
+
+ @POST
+ @Path("/measurements/{measurementID}/labeling")
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Operation(summary = "Label data points of the measurement series with given id", tags = {"Data Lake"},
+ responses = {
+ @ApiResponse(responseCode = "200", description = "Labeling was successful")})
+ public Response labelData(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username
+ , @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathParam("measurementID") String measurementID
+ , @Parameter(in = ParameterIn.DEFAULT, description = "the label details that should be written into database") Placeholder body
+
+ , @Parameter(in = ParameterIn.QUERY, description = "start date for slicing operation") @QueryParam("startDate") String startDate
+ , @Parameter(in = ParameterIn.QUERY, description = "end date for slicing operation") @QueryParam("endDate") String endDate) {
+ /**
+ * TODO: implementation of method stump
+ */
+ return null;
+ }
+
+ @DELETE
+ @Path("/measurements")
+ @Operation(summary = "Remove all stored measurement series from Data Lake", tags = {"Data Lake"},
+ responses = {
+ @ApiResponse(responseCode = "200", description = "All measurement series successfully removed")})
+ public Response removeAll(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username) {
+ /**
+ * TODO: implementation of method stump
+ */
+ return null;
+ }
+}
[incubator-streampipes] 15/29: [STREAMPIPES-349] Add query element
'Grouping By Tags' and corresponding parameter class
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit b8366e22324f6862ebf7c1cbea061e861a8283ac
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 19:08:43 2021 +0200
[STREAMPIPES-349] Add query element 'Grouping By Tags' and corresponding parameter class
---
.../v4/params/GroupingByTagsParams.java | 42 +++++++++++++++++++++
.../v4/query/elements/GroupingByTags.java | 43 ++++++++++++++++++++++
2 files changed, 85 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/GroupingByTagsParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/GroupingByTagsParams.java
new file mode 100644
index 0000000..bcbb02b
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/GroupingByTagsParams.java
@@ -0,0 +1,42 @@
+/*
+ * 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.v4.params;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class GroupingByTagsParams extends QueryParamsV4 {
+ private final List<String> groupingTags;
+
+ public static GroupingByTagsParams from(String measurementID, String groupingTagsSeparatedByComma) {
+ return new GroupingByTagsParams(measurementID, groupingTagsSeparatedByComma);
+ }
+
+ public GroupingByTagsParams(String measurementID, String groupingTagsSeparatedByComma) {
+ super(measurementID);
+ this.groupingTags = new ArrayList<>();
+ for (String tag : groupingTagsSeparatedByComma.split(",")) {
+ this.groupingTags.add(tag);
+ }
+ }
+
+ public List<String> getGroupingTags() {
+ return groupingTags;
+ }
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/GroupingByTags.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/GroupingByTags.java
new file mode 100644
index 0000000..61b9b9e
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/GroupingByTags.java
@@ -0,0 +1,43 @@
+/*
+ * 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.v4.query.elements;
+
+import org.apache.streampipes.dataexplorer.v4.params.GroupingByTagsParams;
+import org.apache.streampipes.dataexplorer.v4.template.QueryTemplatesV4;
+
+public class GroupingByTags extends QueryElement<GroupingByTagsParams> {
+
+ public GroupingByTags(GroupingByTagsParams groupingByTagsParams) {
+ super(groupingByTagsParams);
+ }
+
+ @Override
+ protected String buildStatement(GroupingByTagsParams groupingByTagsParams) {
+ String tags = "";
+ for (String tag : groupingByTagsParams.getGroupingTags()) {
+ if (tags.equals("")) {
+ tags = tag;
+ } else {
+ tags = tags + ", " + tag;
+ }
+ }
+
+ return QueryTemplatesV4.groupByTags(tags);
+ }
+}
[incubator-streampipes] 10/29: [STREAMPIPES-349] Add abstract query
element class and query parameter class
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 3c3d74d28902a56586a9333f8bfea4db7f37e239
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 18:53:02 2021 +0200
[STREAMPIPES-349] Add abstract query element class and query parameter class
---
.../dataexplorer/v4/params/QueryParamsV4.java | 36 ++++++++++++++++++++++
.../v4/query/elements/QueryElement.java | 33 ++++++++++++++++++++
2 files changed, 69 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/QueryParamsV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/QueryParamsV4.java
new file mode 100644
index 0000000..c5d2699
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/params/QueryParamsV4.java
@@ -0,0 +1,36 @@
+/*
+ * 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.v4.params;
+
+public class QueryParamsV4 {
+
+ private final String index;
+
+ public static QueryParamsV4 from(String index) {
+ return new QueryParamsV4(index);
+ }
+
+ protected QueryParamsV4(String index) {
+ this.index = index;
+ }
+
+ public String getIndex() {
+ return index;
+ }
+
+}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/QueryElement.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/QueryElement.java
new file mode 100644
index 0000000..8fd3f50
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/elements/QueryElement.java
@@ -0,0 +1,33 @@
+/*
+ * 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.v4.query.elements;
+
+public abstract class QueryElement<QueryElementParams> {
+ protected String queryStatement;
+
+ public QueryElement(QueryElementParams params) {
+ this.queryStatement = buildStatement(params);
+ }
+
+ protected abstract String buildStatement(QueryElementParams params);
+
+ public String getStatement() {
+ return queryStatement;
+ }
+}
[incubator-streampipes] 24/29: [STREAMPIPES-349] Add endpoint
definition for downloading data from data lake (as csv- or json-file)
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 6bf1dc10bbdc209b43a05dcabea8618eb006e740
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 20:04:14 2021 +0200
[STREAMPIPES-349] Add endpoint definition for downloading data from data lake (as csv- or json-file)
---
.../apache/streampipes/ps/DataLakeResourceV4.java | 24 ++++++++++++++++++++++
1 file changed, 24 insertions(+)
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 fd6ab92..caf76e8 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
@@ -121,6 +121,30 @@ public class DataLakeResourceV4 extends AbstractRestResource {
}
@GET
+ @Path("/measurements/{measurementID}/download")
+ @Produces(MediaType.APPLICATION_OCTET_STREAM)
+ @Operation(summary = "Download data from a single measurement series by a given id", tags = {"Data Lake"},
+ responses = {
+ @ApiResponse(responseCode = "400", description = "Measurement series with given id and requested query specification not found"),
+ @ApiResponse(responseCode = "200", description = "requested data", content = @Content(schema = @Schema(implementation = DataResult.class)))})
+ public Response downloadData(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username
+ , @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathParam("measurementID") String measurementID
+ , @Parameter(in = ParameterIn.QUERY, description = "start date for slicing operation") @QueryParam("startDate") Long startDate
+ , @Parameter(in = ParameterIn.QUERY, description = "end date for slicing operation") @QueryParam("endDate") Long endDate
+ , @Parameter(in = ParameterIn.QUERY, description = "page number for paging operation") @QueryParam("page") Integer page
+ , @Parameter(in = ParameterIn.QUERY, description = "maximum number of retrieved query results") @QueryParam("limit") Integer limit
+ , @Parameter(in = ParameterIn.QUERY, description = "offset") @QueryParam("offset") Integer offset
+ , @Parameter(in = ParameterIn.QUERY, description = "grouping tags (comma-separated) for grouping operation") @QueryParam("groupBy") String groupBy
+ , @Parameter(in = ParameterIn.QUERY, description = "ordering of retrieved query results (ASC or DESC - default is ASC)") @QueryParam("order") String order
+ , @Parameter(in = ParameterIn.QUERY, description = "name of aggregation function used for grouping operation") @QueryParam("aggregationFunction") String aggregationFunction
+ , @Parameter(in = ParameterIn.QUERY, description = "time interval for aggregation (e.g. 1m - one minute) for grouping operation") @QueryParam("timeInterval") String timeInterval
+ , @Parameter(in = ParameterIn.QUERY, description = "format specification (csv, json - default is csv) for data download") @QueryParam("format") String format) {
+ /**
+ * TODO: implementation of method stump
+ */
+ }
+
+ @GET
@Path("/configuration")
@Produces(MediaType.APPLICATION_JSON)
@Operation(summary = "Get the configuration parameters of the data lake", tags = {"Data Lake"},
[incubator-streampipes] 11/29: [STREAMPIPES-349] Add query template
class (includes database specific templates for various query statements)
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit de0279acf63aa37c0d13a8a6ea746e146a42e140
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 19:00:47 2021 +0200
[STREAMPIPES-349] Add query template class (includes database specific templates for various query statements)
---
.../dataexplorer/v4/template/QueryTemplatesV4.java | 72 ++++++++++++++++++++++
1 file changed, 72 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/template/QueryTemplatesV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/template/QueryTemplatesV4.java
new file mode 100644
index 0000000..23c2789
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/template/QueryTemplatesV4.java
@@ -0,0 +1,72 @@
+/*
+ * 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.v4.template;
+
+public class QueryTemplatesV4 {
+
+ public static String selectWildcardFrom(String index) {
+ return "SELECT * FROM " + index;
+ }
+
+ public static String selectAggregationFrom(String index, String aggregationFunction) {
+ return "SELECT " + aggregationFunction + "(*) FROM " + index;
+ }
+
+ public static String deleteFrom(String index) {
+ return "DELETE FROM " + index;
+ }
+
+
+ public static String whereTimeWithin(long startDate, long endDate) {
+ return "WHERE time > "
+ + startDate * 1000000
+ + " AND time < "
+ + endDate * 1000000;
+ }
+
+ public static String whereTimeLeftBound(long startDate) {
+ return "WHERE time > "
+ + startDate * 1000000;
+ }
+
+ public static String whereTimeRightBound(long endDate) {
+ return "WHERE time < "
+ + endDate * 1000000;
+ }
+
+ public static String groupByTags(String tags) {
+ return "GROUP BY " + tags;
+ }
+
+ public static String groupByTime(String timeInterval) {
+ return "GROUP BY time(" + timeInterval + ")";
+ }
+
+ public static String orderByTime(String ordering) {
+ return "ORDER BY time " + ordering.toUpperCase();
+ }
+
+ public static String limitItems(int limit) {
+ return "LIMIT " + limit;
+ }
+
+ public static String offset(int offset) {
+ return "OFFSET " + offset;
+ }
+
+}
[incubator-streampipes] 09/29: [STREAMPIPES-349] Add first
implementation of methods for getting and setting data lake configuration
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit f7cb7662c320b109c124cdc87d33322dad28706a
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Wed May 5 10:33:09 2021 +0200
[STREAMPIPES-349] Add first implementation of methods for getting and setting data lake configuration
---
.../dataexplorer/DataLakeManagementV4.java | 49 +++++++++++++++++++++-
1 file changed, 48 insertions(+), 1 deletion(-)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
index 2f177f6..f1bc058 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
@@ -18,9 +18,14 @@
package org.apache.streampipes.dataexplorer;
+import org.apache.streampipes.dataexplorer.param.RetentionPolicyQueryParams;
import org.apache.streampipes.dataexplorer.query.DeleteDataQuery;
+import org.apache.streampipes.dataexplorer.query.EditRetentionPolicyQuery;
+import org.apache.streampipes.dataexplorer.query.ShowRetentionPolicyQuery;
import org.apache.streampipes.dataexplorer.utils.DataExplorerUtils;
+import org.apache.streampipes.model.datalake.DataLakeConfiguration;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.model.datalake.DataLakeRetentionPolicy;
import org.influxdb.dto.QueryResult;
import java.util.List;
@@ -43,5 +48,47 @@ public class DataLakeManagementV4 {
}
return true;
}
-}
+
+ public DataLakeConfiguration getDataLakeConfiguration() {
+ List<DataLakeRetentionPolicy> retentionPolicies = getAllExistingRetentionPolicies();
+ return new DataLakeConfiguration(retentionPolicies);
+ }
+
+ public String editMeasurementConfiguration(DataLakeConfiguration config, boolean resetToDefault) {
+
+ List<DataLakeRetentionPolicy> existingRetentionPolicies = getAllExistingRetentionPolicies();
+
+ if (resetToDefault) {
+ if (existingRetentionPolicies.size() > 1) {
+ String drop = new EditRetentionPolicyQuery(RetentionPolicyQueryParams.from("custom", "0s"), "DROP").executeQuery();
+ }
+ String reset = new EditRetentionPolicyQuery(RetentionPolicyQueryParams.from("autogen", "0s"), "DEFAULT").executeQuery();
+ return reset;
+ } else {
+
+ Integer batchSize = config.getBatchSize();
+ Integer flushDuration = config.getFlushDuration();
+
+ /**
+ * TODO:
+ * - Implementation of parameter update for batchSize and flushDuration
+ * - Updating multiple retention policies
+ */
+
+ String operation = "CREATE";
+ if (existingRetentionPolicies.size() > 1) {
+ operation = "ALTER";
+ }
+ String result = new EditRetentionPolicyQuery(RetentionPolicyQueryParams.from("custom", "1d"), operation).executeQuery();
+ return result;
+ }
+ }
+
+ public List<DataLakeRetentionPolicy> getAllExistingRetentionPolicies() {
+ /**
+ * TODO:
+ * - Implementation of parameter return for batchSize and flushDuration
+ */
+ return new ShowRetentionPolicyQuery(RetentionPolicyQueryParams.from("", "0s")).executeQuery();
+ }
}
[incubator-streampipes] 21/29: [STREAMPIPES-349] Add Data Explorer
Query class
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 98759e0b0fe64e13c159a3c89a06159c4ec8b8c7
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 19:46:03 2021 +0200
[STREAMPIPES-349] Add Data Explorer Query class
---
.../dataexplorer/v4/query/DataExplorerQueryV4.java | 121 +++++++++++++++++++++
1 file changed, 121 insertions(+)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/DataExplorerQueryV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/DataExplorerQueryV4.java
new file mode 100644
index 0000000..6b06f21
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/query/DataExplorerQueryV4.java
@@ -0,0 +1,121 @@
+/*
+ * 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.v4.query;
+
+import org.apache.streampipes.config.backend.BackendConfig;
+import org.apache.streampipes.dataexplorer.utils.DataExplorerUtils;
+import org.apache.streampipes.dataexplorer.v4.params.*;
+import org.apache.streampipes.dataexplorer.v4.query.elements.*;
+import org.apache.streampipes.model.datalake.DataResult;
+import org.influxdb.InfluxDB;
+import org.influxdb.dto.Query;
+import org.influxdb.dto.QueryResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class DataExplorerQueryV4 {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DataExplorerQueryV4.class);
+
+
+ protected Map<String, QueryParamsV4> params;
+
+ public DataExplorerQueryV4(Map<String, QueryParamsV4> params) {
+ this.params = params;
+ }
+
+ public DataResult executeQuery() throws RuntimeException {
+ InfluxDB influxDB = DataExplorerUtils.getInfluxDBClient();
+ List<QueryElement> queryElements = getQueryElements();
+
+ QueryBuilder queryBuilder = QueryBuilder.create(BackendConfig.INSTANCE.getInfluxDatabaseName());
+ Query query = queryBuilder.build(queryElements);
+ LOG.debug("Data Lake Query (database:" + query.getDatabase() + "): " + query.getCommand());
+
+ QueryResult result = influxDB.query(query);
+ LOG.debug("Data Lake Query Result: " + result.toString());
+ DataResult dataResult = postQuery(result);
+
+ influxDB.close();
+ return dataResult;
+ }
+
+ protected DataResult convertResult(QueryResult result) {
+ if (result.getResults().get(0).getSeries() != null) {
+ return convertResult(result.getResults().get(0).getSeries().get(0));
+ } else {
+ return new DataResult();
+ }
+ }
+
+ protected DataResult convertResult(QueryResult.Series series) {
+ List<String> columns = series.getColumns();
+ for (int i = 0; i < columns.size(); i++) {
+ String replacedColumnName = columns.get(i).replaceAll("mean_", "");
+ columns.set(i, replacedColumnName);
+ }
+ List values = series.getValues();
+ return new DataResult(values.size(), columns, values);
+ }
+
+ protected DataResult postQuery(QueryResult result) throws RuntimeException {
+ return convertResult(result);
+ }
+
+ protected List<QueryElement> getQueryElements() {
+ List<QueryElement> queryElements = new ArrayList<>();
+
+ if (this.params.containsKey("SELECT")) {
+ queryElements.add(new SelectFromStatement((SelectFromStatementParams) this.params.get("SELECT")));
+ } else {
+ queryElements.add(new DeleteFromStatement((DeleteFromStatementParams) this.params.get("DELETE")));
+ }
+
+ if (this.params.containsKey("WHERE")) {
+ queryElements.add(new TimeBoundary((TimeBoundaryParams) this.params.get("WHERE")));
+ }
+
+ if (this.params.containsKey("GROUPBYTIME")) {
+ queryElements.add(new GroupingByTime((GroupingByTimeParams) this.params.get("GROUPBYTIME")));
+
+ } else if (this.params.containsKey("GROUPBY")) {
+ queryElements.add(new GroupingByTags((GroupingByTagsParams) this.params.get("GROUPBY")));
+ }
+
+ if (this.params.containsKey("DESCENDING")) {
+ queryElements.add(new OrderingByTime((OrderingByTimeParams) this.params.get("DESCENDING")));
+ } else if (this.params.containsKey("SELECT")) {
+ queryElements.add(new OrderingByTime(OrderingByTimeParams.from(this.params.get("SELECT").getIndex(), "ASC")));
+ }
+
+ if (this.params.containsKey("LIMIT")) {
+ queryElements.add(new ItemLimitation((ItemLimitationParams) this.params.get("LIMIT")));
+ }
+
+ if (this.params.containsKey("OFFSET")) {
+ queryElements.add(new Offset((OffsetParams) this.params.get("OFFSET")));
+ }
+
+ return queryElements;
+ }
+}
[incubator-streampipes] 03/29: [STREAMPIPES-349] Rename method as
well as list containing all stored measurement series
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit a066b5d1d2bb9576daea07dd86af2a870899d287
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Apr 30 10:17:52 2021 +0200
[STREAMPIPES-349] Rename method as well as list containing all stored measurement series
---
.../org/apache/streampipes/dataexplorer/DataLakeManagementV4.java | 6 +++---
.../src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java | 4 ++--
2 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
index 07f2b90..06e055c 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
@@ -25,8 +25,8 @@ import java.util.List;
public class DataLakeManagementV4 {
- public List<DataLakeMeasure> getAllMeasurementSeries() {
- List<DataLakeMeasure> dataLakeMeasuresList = DataExplorerUtils.getInfos();
- return dataLakeMeasuresList;
+ public List<DataLakeMeasure> getAllMeasurements() {
+ List<DataLakeMeasure> allMeasurements = DataExplorerUtils.getInfos();
+ return allMeasurements;
}
}
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 d6f51f6..942d32a 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
@@ -93,8 +93,8 @@ public class DataLakeResourceV4 extends AbstractRestResource {
responses = {
@ApiResponse(responseCode = "200", description = "array of stored measurement series", content = @Content(array = @ArraySchema(schema = @Schema(implementation = DataLakeMeasure.class))))})
public Response getAll(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username) {
- List<DataLakeMeasure> allMeasurementSeries = this.dataLakeManagement.getAllMeasurementSeries();
- return ok(allMeasurementSeries);
+ List<DataLakeMeasure> allMeasurements = this.dataLakeManagement.getAllMeasurements();
+ return ok(allMeasurements);
}
@GET
[incubator-streampipes] 28/29: [STREAMPIPES-349] Add
DataLakeResourceV4 to StreamPipesResourceConfig
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit d9e9d7980d448f4d4cc1ef5df8abc071478f3029
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 20:16:46 2021 +0200
[STREAMPIPES-349] Add DataLakeResourceV4 to StreamPipesResourceConfig
---
.../backend/StreamPipesResourceConfig.java | 156 +++++++++++----------
1 file changed, 79 insertions(+), 77 deletions(-)
diff --git a/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesResourceConfig.java b/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesResourceConfig.java
index 2261149..1e66b44 100644
--- a/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesResourceConfig.java
+++ b/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesResourceConfig.java
@@ -20,6 +20,8 @@ package org.apache.streampipes.backend;
import io.swagger.v3.jaxrs2.integration.resources.OpenApiResource;
import org.apache.streampipes.connect.container.master.rest.*;
+import org.apache.streampipes.ps.DataLakeResourceV3;
+import org.apache.streampipes.ps.DataLakeResourceV4;
import org.apache.streampipes.ps.PipelineElementTemplateResource;
import org.apache.streampipes.rest.impl.*;
import org.apache.streampipes.rest.impl.dashboard.Dashboard;
@@ -27,7 +29,6 @@ import org.apache.streampipes.rest.impl.dashboard.DashboardWidget;
import org.apache.streampipes.rest.impl.dashboard.VisualizablePipeline;
import org.apache.streampipes.rest.impl.datalake.DataLakeDashboardResource;
import org.apache.streampipes.rest.impl.datalake.DataLakeNoUserResourceV3;
-import org.apache.streampipes.ps.DataLakeResourceV3;
import org.apache.streampipes.rest.impl.datalake.DataLakeWidgetResource;
import org.apache.streampipes.rest.impl.nouser.FileServingResource;
import org.apache.streampipes.rest.impl.nouser.PipelineElementImportNoUser;
@@ -47,86 +48,87 @@ import javax.ws.rs.ApplicationPath;
@ApplicationPath("/api")
public class StreamPipesResourceConfig extends ResourceConfig {
- public StreamPipesResourceConfig() {
- register(Authentication.class);
- register(AssetDashboard.class);
- register(AutoComplete.class);
- register(CategoryResource.class);
- register(ConsulConfig.class);
- register(ContainerProvidedOptions.class);
- register(Couchdb.class);
- register(DashboardWidget.class);
- register(Dashboard.class);
- register(DataLakeResourceV3.class);
- register(DataLakeNoUserResourceV3.class);
- register(DataStream.class);
- register(Deployment.class);
- register(FileServingResource.class);
- register(InternalPipelineTemplates.class);
- register(LabelResource.class);
- register(MeasurementUnitResource.class);
- register(Notification.class);
- register(OntologyContext.class);
- register(OntologyKnowledge.class);
- register(OntologyMeasurementUnit.class);
- register(OntologyPipelineElement.class);
- register(PipelineCache.class);
- register(PipelineCategory.class);
- register(PipelineElementAsset.class);
- register(PipelineElementCategory.class);
- register(PipelineElementFile.class);
- register(PipelineElementImportNoUser.class);
- register(PipelineElementImport.class);
- register(PipelineElementRuntimeInfo.class);
- register(PipelineMonitoring.class);
- register(PipelineNoUserResource.class);
- register(PipelineTemplate.class);
- register(PipelineResource.class);
- register(RdfEndpoint.class);
- register(SemanticEventConsumer.class);
- register(SemanticEventProcessingAgent.class);
- register(SemanticEventProducer.class);
- register(Setup.class);
- register(User.class);
- register(Version.class);
- register(PipelineElementAsset.class);
- register(DataLakeDashboardResource.class);
- register(DataLakeWidgetResource.class);
- register(DataLakeResourceV3.class);
- register(DataLakeNoUserResourceV3.class);
- register(PipelineElementFile.class);
- register(FileServingResource.class);
- register(DashboardWidget.class);
- register(Dashboard.class);
- register(VirtualSensor.class);
- register(Visualization.class);
- register(VisualizablePipeline.class);
+ public StreamPipesResourceConfig() {
+ register(Authentication.class);
+ register(AssetDashboard.class);
+ register(AutoComplete.class);
+ register(CategoryResource.class);
+ register(ConsulConfig.class);
+ register(ContainerProvidedOptions.class);
+ register(Couchdb.class);
+ register(DashboardWidget.class);
+ register(Dashboard.class);
+ register(DataLakeResourceV3.class);
+ register(DataLakeNoUserResourceV3.class);
+ register(DataStream.class);
+ register(Deployment.class);
+ register(FileServingResource.class);
+ register(InternalPipelineTemplates.class);
+ register(LabelResource.class);
+ register(MeasurementUnitResource.class);
+ register(Notification.class);
+ register(OntologyContext.class);
+ register(OntologyKnowledge.class);
+ register(OntologyMeasurementUnit.class);
+ register(OntologyPipelineElement.class);
+ register(PipelineCache.class);
+ register(PipelineCategory.class);
+ register(PipelineElementAsset.class);
+ register(PipelineElementCategory.class);
+ register(PipelineElementFile.class);
+ register(PipelineElementImportNoUser.class);
+ register(PipelineElementImport.class);
+ register(PipelineElementRuntimeInfo.class);
+ register(PipelineMonitoring.class);
+ register(PipelineNoUserResource.class);
+ register(PipelineTemplate.class);
+ register(PipelineResource.class);
+ register(RdfEndpoint.class);
+ register(SemanticEventConsumer.class);
+ register(SemanticEventProcessingAgent.class);
+ register(SemanticEventProducer.class);
+ register(Setup.class);
+ register(User.class);
+ register(Version.class);
+ register(PipelineElementAsset.class);
+ register(DataLakeDashboardResource.class);
+ register(DataLakeWidgetResource.class);
+ register(DataLakeResourceV3.class);
+ register(DataLakeNoUserResourceV3.class);
+ register(PipelineElementFile.class);
+ register(FileServingResource.class);
+ register(DashboardWidget.class);
+ register(Dashboard.class);
+ register(VirtualSensor.class);
+ register(Visualization.class);
+ register(VisualizablePipeline.class);
+
+ // Serializers
+ register(GsonWithIdProvider.class);
+ register(GsonWithoutIdProvider.class);
+ register(GsonClientModelProvider.class);
+ register(JsonLdProvider.class);
+ register(JacksonSerializationProvider.class);
+ register(MultiPartFeature.class);
- // Serializers
- register(GsonWithIdProvider.class);
- register(GsonWithoutIdProvider.class);
- register(GsonClientModelProvider.class);
- register(JsonLdProvider.class);
- register(JacksonSerializationProvider.class);
- register(MultiPartFeature.class);
-
- // Platform Services
- register(PipelineElementTemplateResource.class);
+ // Platform Services
+ register(PipelineElementTemplateResource.class);
+ register(DataLakeResourceV4.class);
- register(OpenApiResource.class);
+ register(OpenApiResource.class);
- // Connect Master
- register(WelcomePageMaster.class);
- register(AdapterResource.class);
- register(AdapterTemplateResource.class);
- register(DescriptionResource.class);
- register(SourcesResource.class);
- register(GuessResource.class);
+ // Connect Master
+ register(WelcomePageMaster.class);
+ register(AdapterResource.class);
+ register(AdapterTemplateResource.class);
+ register(DescriptionResource.class);
+ register(SourcesResource.class);
+ register(GuessResource.class);
// register(MultiPartFeature.class);
- register(UnitResource.class);
- register(WorkerAdministrationResource.class);
- register(RuntimeResolvableResource.class);
- }
+ register(UnitResource.class);
+ register(WorkerAdministrationResource.class);
+ register(RuntimeResolvableResource.class);
+ }
}
[incubator-streampipes] 04/29: [STREAMPIPES-349] Add implementation
of endpoint for 'removing all stored measurement series'
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 3e303a85ea96dd0b5321e4133ffc06242d39c13f
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Apr 30 10:31:21 2021 +0200
[STREAMPIPES-349] Add implementation of endpoint for 'removing all stored measurement series'
---
.../streampipes/dataexplorer/DataLakeManagementV4.java | 15 +++++++++++++++
.../org/apache/streampipes/ps/DataLakeResourceV4.java | 6 ++----
2 files changed, 17 insertions(+), 4 deletions(-)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
index 06e055c..2f177f6 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
@@ -18,8 +18,10 @@
package org.apache.streampipes.dataexplorer;
+import org.apache.streampipes.dataexplorer.query.DeleteDataQuery;
import org.apache.streampipes.dataexplorer.utils.DataExplorerUtils;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.influxdb.dto.QueryResult;
import java.util.List;
@@ -29,4 +31,17 @@ public class DataLakeManagementV4 {
List<DataLakeMeasure> allMeasurements = DataExplorerUtils.getInfos();
return allMeasurements;
}
+
+ public boolean removeAllMeasurements() {
+ List<DataLakeMeasure> allMeasurements = getAllMeasurements();
+
+ for (DataLakeMeasure measure : allMeasurements) {
+ QueryResult queryResult = new DeleteDataQuery(measure).executeQuery();
+ if (queryResult.hasError() || queryResult.getResults().get(0).getError() != null) {
+ return false;
+ }
+ }
+ return true;
+ }
+}
}
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 942d32a..f16dc3a 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
@@ -160,9 +160,7 @@ public class DataLakeResourceV4 extends AbstractRestResource {
responses = {
@ApiResponse(responseCode = "200", description = "All measurement series successfully removed")})
public Response removeAll(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username) {
- /**
- * TODO: implementation of method stump
- */
- return null;
+ boolean isSuccess = this.dataLakeManagement.removeAllMeasurements();
+ return Response.ok(isSuccess).build();
}
}
[incubator-streampipes] 23/29: [STREAMPIPES-349] Add implementation
of endpoint for getting data from data lake
Posted by eb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ebi pushed a commit to branch STREAMPIPES-349
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit f24f63d82939ee526f88021f52d0b5d1d5f3eb41
Author: Daniel Ebi <eb...@fzi.de>
AuthorDate: Fri Jun 11 20:01:22 2021 +0200
[STREAMPIPES-349] Add implementation of endpoint for getting data from data lake
---
.../dataexplorer/DataLakeManagementV4.java | 51 ++++++++++++++++++++--
.../apache/streampipes/ps/DataLakeResourceV4.java | 25 ++++++-----
2 files changed, 61 insertions(+), 15 deletions(-)
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
index 76a7577..8e175b8 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
@@ -23,9 +23,7 @@ import org.apache.streampipes.dataexplorer.query.DeleteDataQuery;
import org.apache.streampipes.dataexplorer.query.EditRetentionPolicyQuery;
import org.apache.streampipes.dataexplorer.query.ShowRetentionPolicyQuery;
import org.apache.streampipes.dataexplorer.utils.DataExplorerUtils;
-import org.apache.streampipes.dataexplorer.v4.params.DeleteFromStatementParams;
-import org.apache.streampipes.dataexplorer.v4.params.QueryParamsV4;
-import org.apache.streampipes.dataexplorer.v4.params.TimeBoundaryParams;
+import org.apache.streampipes.dataexplorer.v4.params.*;
import org.apache.streampipes.dataexplorer.v4.query.DataExplorerQueryV4;
import org.apache.streampipes.model.datalake.DataLakeConfiguration;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
@@ -44,6 +42,11 @@ public class DataLakeManagementV4 {
return allMeasurements;
}
+ public DataResult getData(String measurementID, Long startDate, Long endDate, Integer page, Integer limit, Integer offset, String groupBy, String order, String aggregationFunction, String timeInterval) {
+ Map<String, QueryParamsV4> queryParts = getQueryParams(measurementID, startDate, endDate, page, limit, offset, groupBy, order, aggregationFunction, timeInterval);
+ return new DataExplorerQueryV4(queryParts).executeQuery();
+ }
+
public boolean removeAllMeasurements() {
List<DataLakeMeasure> allMeasurements = getAllMeasurements();
@@ -104,6 +107,48 @@ public class DataLakeManagementV4 {
return new ShowRetentionPolicyQuery(RetentionPolicyQueryParams.from("", "0s")).executeQuery();
}
+ private Map<String, QueryParamsV4> getQueryParams(String measurementID, Long startDate, Long endDate, Integer page, Integer limit, Integer offset, String groupBy, String order, String aggregationFunction, String timeInterval) {
+ Map<String, QueryParamsV4> queryParts = new HashMap<>();
+
+ queryParts.put("SELECT", SelectFromStatementParams.from(measurementID, aggregationFunction));
+
+ if (startDate != null || endDate != null) {
+ queryParts.put("WHERE", TimeBoundaryParams.from(measurementID, startDate, endDate));
+ }
+
+
+ if (timeInterval != null && aggregationFunction != null) {
+ if (groupBy == null) {
+ queryParts.put("GROUPBYTIME", GroupingByTimeParams.from(measurementID, timeInterval));
+ } else {
+ groupBy = groupBy + ",time(" + timeInterval + ")";
+ }
+ }
+
+ if (groupBy != null) {
+ queryParts.put("GROUPBY", GroupingByTagsParams.from(measurementID, groupBy));
+ }
+
+ if (order != null) {
+ if (order.equals("DESC")) {
+ queryParts.put("DESCENDING", OrderingByTimeParams.from(measurementID, order));
+ }
+ }
+
+ if (limit != null) {
+ queryParts.put("LIMIT", ItemLimitationParams.from(measurementID, limit));
+ }
+
+ if (offset != null) {
+ queryParts.put("OFFSET", OffsetParams.from(measurementID, offset));
+ } else if (limit != null && page != null) {
+ queryParts.put("OFFSET", OffsetParams.from(measurementID, page * limit));
+ }
+
+ return queryParts;
+
+ }
+
public Map<String, QueryParamsV4> getDeleteQueryParams(String measurementID, Long startDate, Long endDate) {
Map<String, QueryParamsV4> queryParts = new HashMap<>();
queryParts.put("DELETE", DeleteFromStatementParams.from(measurementID));
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 cbef9de..fd6ab92 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
@@ -95,28 +95,29 @@ public class DataLakeResourceV4 extends AbstractRestResource {
return ok(allMeasurements);
}
+
@GET
@Path("/measurements/{measurementID}")
@Produces(MediaType.APPLICATION_JSON)
@Operation(summary = "Get data from a single measurement series by a given id", tags = {"Data Lake"},
responses = {
@ApiResponse(responseCode = "400", description = "Measurement series with given id and requested query specification not found"),
- @ApiResponse(responseCode = "200", description = "requested data", content = @Content(schema = @Schema(implementation = Placeholder.class)))})
+ @ApiResponse(responseCode = "200", description = "requested data", content = @Content(schema = @Schema(implementation = DataResult.class)))})
public Response getData(@Parameter(in = ParameterIn.PATH, description = "username", required = true) @PathParam("username") String username
, @Parameter(in = ParameterIn.PATH, description = "the id of the measurement series", required = true) @PathParam("measurementID") String measurementID
- , @Parameter(in = ParameterIn.QUERY, description = "start date for slicing operation") @QueryParam("startDate") String startDate
- , @Parameter(in = ParameterIn.QUERY, description = "end date for slicing operation") @QueryParam("endDate") String endDate
- , @Parameter(in = ParameterIn.QUERY, description = "page number for paging operation") @QueryParam("page") String page
- , @Parameter(in = ParameterIn.QUERY, description = "items per page limitation for paging operation") @QueryParam("limit") Integer limit
- , @Parameter(in = ParameterIn.QUERY, description = "offset for paging operation") @QueryParam("offset") Integer offset
+ , @Parameter(in = ParameterIn.QUERY, description = "start date for slicing operation") @QueryParam("startDate") Long startDate
+ , @Parameter(in = ParameterIn.QUERY, description = "end date for slicing operation") @QueryParam("endDate") Long endDate
+ , @Parameter(in = ParameterIn.QUERY, description = "page number for paging operation") @QueryParam("page") Integer page
+ , @Parameter(in = ParameterIn.QUERY, description = "maximum number of retrieved query results") @QueryParam("limit") Integer limit
+ , @Parameter(in = ParameterIn.QUERY, description = "offset") @QueryParam("offset") Integer offset
, @Parameter(in = ParameterIn.QUERY, description = "grouping tags (comma-separated) for grouping operation") @QueryParam("groupBy") String groupBy
+ , @Parameter(in = ParameterIn.QUERY, description = "ordering of retrieved query results (ASC or DESC - default is ASC)") @QueryParam("order") String order
, @Parameter(in = ParameterIn.QUERY, description = "name of aggregation function used for grouping operation") @QueryParam("aggregationFunction") String aggregationFunction
- , @Parameter(in = ParameterIn.QUERY, description = "time interval for aggregation (e.g. 1m - one minute) for grouping operation") @QueryParam("timeInterval") String timeInterval
- , @Parameter(in = ParameterIn.QUERY, description = "format specification (csv, json) for data download") @QueryParam("format") String format) {
- /**
- * TODO: implementation of method stump
- */
- return null;
+ , @Parameter(in = ParameterIn.QUERY, description = "time interval for aggregation (e.g. 1m - one minute) for grouping operation") @QueryParam("timeInterval") String timeInterval) {
+
+ DataResult result = this.dataLakeManagement.getData(measurementID, startDate, endDate, page, limit, offset, groupBy, order, aggregationFunction, timeInterval);
+
+ return Response.ok(result).build();
}
@GET