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
The following commit(s) were added to refs/heads/STREAMPIPES-349 by this push:
new ef5be85 [STREAMPIPES-349] Refactor data retrieval and download
endpoint to support selection of specified columns
ef5be85 is described below
commit ef5be85325e36cb6bd70d2aeee9b0c401d11622c
Author: Daniel Ebi <[email protected]>
AuthorDate: Tue Jul 6 20:37:15 2021 +0200
[STREAMPIPES-349] Refactor data retrieval and download endpoint to support
selection of specified columns
---
.../streampipes/dataexplorer/DataLakeManagementV4.java | 10 +++++-----
.../v4/params/SelectFromStatementParams.java | 18 +++++++++++++++---
.../v4/query/elements/SelectFromStatement.java | 4 ++--
.../dataexplorer/v4/template/QueryTemplatesV4.java | 15 +++++++++++----
.../dataexplorer/v4/utils/DataLakeManagementUtils.java | 4 ++--
.../org/apache/streampipes/ps/DataLakeResourceV4.java | 6 ++++--
6 files changed, 39 insertions(+), 18 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 e06a8cc..359dd3c 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
@@ -50,12 +50,12 @@ 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 =
DataLakeManagementUtils.getSelectQueryParams(measurementID, startDate, endDate,
page, limit, offset, groupBy, order, aggregationFunction, timeInterval);
+ public DataResult getData(String measurementID, String columns, Long
startDate, Long endDate, Integer page, Integer limit, Integer offset, String
groupBy, String order, String aggregationFunction, String timeInterval) {
+ Map<String, QueryParamsV4> queryParts =
DataLakeManagementUtils.getSelectQueryParams(measurementID, columns, startDate,
endDate, page, limit, offset, groupBy, order, aggregationFunction,
timeInterval);
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 {
+ public void getDataAsStream(String measurementID, String columns, 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;
}
@@ -75,7 +75,7 @@ public class DataLakeManagementV4 {
outputStream.write(toBytes("["));
do {
- dataResult = getData(measurementID, startDate, endDate, i,
limit, null, groupBy, order, aggregationFunction, timeInterval);
+ dataResult = getData(measurementID, columns, startDate,
endDate, i, limit, null, groupBy, order, aggregationFunction, timeInterval);
if (dataResult.getTotal() > 0) {
for (List<Object> row : dataResult.getRows()) {
@@ -122,7 +122,7 @@ public class DataLakeManagementV4 {
boolean isFirstDataObject = true;
do {
- dataResult = getData(measurementID, startDate, endDate, i,
limit, null, groupBy, order, aggregationFunction, timeInterval);
+ dataResult = getData(measurementID, columns, startDate,
endDate, i, limit, null, groupBy, order, aggregationFunction, timeInterval);
//Send first header
if (dataResult.getTotal() > 0) {
if (isFirstDataObject) {
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
index 0385963..fd04b5d 100644
---
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
@@ -22,22 +22,34 @@ import javax.annotation.Nullable;
public class SelectFromStatementParams extends QueryParamsV4 {
+ private final String selectedColumns;
private final String aggregationFunction;
- public static SelectFromStatementParams from(String measurementID,
@Nullable String aggregationFunction) {
- return new SelectFromStatementParams(measurementID,
aggregationFunction);
+ public static SelectFromStatementParams from(String measurementID,
@Nullable String columns, @Nullable String aggregationFunction) {
+ return new SelectFromStatementParams(measurementID, columns,
aggregationFunction);
}
public SelectFromStatementParams(String measurementID) {
super(measurementID);
+ this.selectedColumns = "*";
this.aggregationFunction = null;
}
- public SelectFromStatementParams(String measurementID, String
aggregationFunction) {
+ public SelectFromStatementParams(String measurementID, String columns,
String aggregationFunction) {
super(measurementID);
+
+ if (columns != null) {
+ this.selectedColumns = columns;
+ } else {
+ this.selectedColumns = "*";
+ }
+
this.aggregationFunction = aggregationFunction;
}
+ public String getSelectedColumns() {
+ return selectedColumns;
+ }
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
index efd7654..d438179 100644
---
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
@@ -29,9 +29,9 @@ public class SelectFromStatement extends
QueryElement<SelectFromStatementParams>
@Override
protected String buildStatement(SelectFromStatementParams
selectFromStatementParams) {
if (selectFromStatementParams.getAggregationFunction() == null) {
- return
QueryTemplatesV4.selectWildcardFrom(selectFromStatementParams.getIndex());
+ return
QueryTemplatesV4.selectFrom(selectFromStatementParams.getIndex(),
selectFromStatementParams.getSelectedColumns());
} else {
- return
QueryTemplatesV4.selectAggregationFrom(selectFromStatementParams.getIndex(),
selectFromStatementParams.getAggregationFunction());
+ return
QueryTemplatesV4.selectAggregationFrom(selectFromStatementParams.getIndex(),
selectFromStatementParams.getSelectedColumns(),
selectFromStatementParams.getAggregationFunction());
}
}
}
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
index 23c2789..915f6ae 100644
---
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
@@ -19,12 +19,19 @@ package org.apache.streampipes.dataexplorer.v4.template;
public class QueryTemplatesV4 {
- public static String selectWildcardFrom(String index) {
- return "SELECT * FROM " + index;
+ public static String selectFrom(String index, String columns) {
+ return "SELECT " + columns + " FROM " + index;
}
- public static String selectAggregationFrom(String index, String
aggregationFunction) {
- return "SELECT " + aggregationFunction + "(*) FROM " + index;
+ public static String selectAggregationFrom(String index, String columns,
String aggregationFunction) {
+ String[] cols = columns.split(",");
+ String statement = aggregationFunction + "(" + cols[0] + ")";
+
+ for (int i = 1; i < cols.length; i++) {
+ statement = statement + ", " + aggregationFunction + "(" + cols[i]
+ ")";
+ }
+
+ return "SELECT " + statement + " FROM " + index;
}
public static String deleteFrom(String index) {
diff --git
a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/utils/DataLakeManagementUtils.java
b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/utils/DataLakeManagementUtils.java
index ca5195a..3726253 100644
---
a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/utils/DataLakeManagementUtils.java
+++
b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/utils/DataLakeManagementUtils.java
@@ -35,10 +35,10 @@ public class DataLakeManagementUtils {
public static final String DELETE_FROM = "DELETE";
- public static Map<String, QueryParamsV4> getSelectQueryParams(String
measurementID, Long startDate, Long endDate, Integer page, Integer limit,
Integer offset, String groupBy, String order, String aggregationFunction,
String timeInterval) {
+ public static Map<String, QueryParamsV4> getSelectQueryParams(String
measurementID, String columns, 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_FROM,
SelectFromStatementParams.from(measurementID, aggregationFunction));
+ queryParts.put(SELECT_FROM,
SelectFromStatementParams.from(measurementID, columns, aggregationFunction));
if (startDate != null || endDate != null) {
queryParts.put(WHERE, TimeBoundaryParams.from(measurementID,
startDate, endDate));
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 8692af0..c7d0549 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
@@ -129,6 +129,7 @@ public class DataLakeResourceV4 extends
AbstractRestResource {
@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 = "the columns to
be selected (comma-separated)") @QueryParam("columns") String columns
, @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
@@ -139,7 +140,7 @@ 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) {
- DataResult result = this.dataLakeManagement.getData(measurementID,
startDate, endDate, page, limit, offset, groupBy, order, aggregationFunction,
timeInterval);
+ DataResult result = this.dataLakeManagement.getData(measurementID,
columns, startDate, endDate, page, limit, offset, groupBy, order,
aggregationFunction, timeInterval);
return Response.ok(result).build();
}
@@ -153,6 +154,7 @@ public class DataLakeResourceV4 extends
AbstractRestResource {
@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 = "the columns to
be selected (comma-separated)") @QueryParam("columns") String columns
, @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
@@ -169,7 +171,7 @@ public class DataLakeResourceV4 extends
AbstractRestResource {
}
String outputFormat = format;
- StreamingOutput streamingOutput = output ->
dataLakeManagement.getDataAsStream(measurementID, startDate, endDate, page,
limit, offset, groupBy, order, aggregationFunction, timeInterval, outputFormat,
output);
+ StreamingOutput streamingOutput = output ->
dataLakeManagement.getDataAsStream(measurementID, columns, 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 + "\"")
