From ef39b068408c671cfa9a6a3e3907f163b32b4581 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Wed, 23 Jul 2025 16:00:47 +0200 Subject: [PATCH 01/28] add loading indicator --- ui/app/manager/src/pages/page-export.ts | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index f56f896366..10e6b719c8 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -16,6 +16,7 @@ import {isAxiosError} from "@openremote/rest"; import { showSnackbar } from "@openremote/or-mwc-components/or-mwc-snackbar"; import { showDialog } from "@openremote/or-mwc-components/or-mwc-dialog"; const tableStyle = require("@material/data-table/dist/mdc.data-table.css"); +import { when } from "lit/directives/when.js"; export function pageExportProvider(store: Store): PageProvider { return { @@ -200,6 +201,9 @@ export class PageExport extends Page { @property({type: Number}) private latestTimestamp: number = moment().valueOf(); + @property() + protected _loading: boolean = false; + private config?: OrExportConfig; private realm: string; private isClearExportBtnDisabled: boolean = true; @@ -240,11 +244,17 @@ export class PageExport extends Page { ${i18next.t("dataExport")} +

${i18next.t("dataSelection")}

+ ${when(this._loading, () => html` +
+ +
+ `)} ${headers.map(header => html` @@ -268,7 +278,8 @@ export class PageExport extends Page {
- +
+
@@ -290,7 +301,7 @@ export class PageExport extends Page { attributeName: attrRef.name, }); }); - + this._loading = true; Promise.all(dataPointInfoPromises).then(datapointPeriod => { const assetInfoPromises = datapointPeriod.map(result => { @@ -301,6 +312,7 @@ export class PageExport extends Page { const allAssets = assetInfos.map(attr => attr.data); const allDatapoints = datapointPeriod.map(datapoints => datapoints.data); + if (allDatapoints.length > 0) { this.tableRows = allDatapoints.map(dataInfo => { const relevantAssetInfo = allAssets.find(asset => asset.id === dataInfo.assetId); @@ -315,6 +327,7 @@ export class PageExport extends Page { this.renderTableRows(); this.isClearExportBtnDisabled = false; this.isExportBtnDisabled = false; + this._loading = false; } else { this.isClearExportBtnDisabled = true; this.isExportBtnDisabled = true; @@ -375,6 +388,7 @@ export class PageExport extends Page { } protected export = () => { + this._loading = true; manager.rest.api.AssetDatapointResource.getDatapointExport({ attributeRefs: JSON.stringify(this.tableRows.map(attr => ({id: attr.assetId, name: attr.attributeName}))), fromTimestamp: this.oldestTimestamp, @@ -397,8 +411,15 @@ export class PageExport extends Page { if(ex.response?.status === 413) { showSnackbar(undefined, "exportTooLargeError"); } + else if(ex.message.includes("timeout")) { + showSnackbar(undefined, "noAttributeDataTimeout"); + } + else { + showSnackbar(undefined, "errorOccurred"); + } } }); + this._loading = false; } protected async loadConfig() { From 8340c440a117a6c68f70e6c4faa47a487dec70c4 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Wed, 23 Jul 2025 16:40:04 +0200 Subject: [PATCH 02/28] add ui for selecting different csv format --- ui/app/manager/src/pages/page-export.ts | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index 10e6b719c8..b962554990 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -179,6 +179,9 @@ export class PageExport extends Page { margin-bottom: 2em } + .format-wrapper { + margin-bottom: 2em + } .export-btn-wrapper { display: flex; justify-content: space-between; @@ -204,6 +207,11 @@ export class PageExport extends Page { @property() protected _loading: boolean = false; + protected _exportFormats: String[] = ["Default", "Column per attribute"]; + + @property() + private selectedFormat: String = "Default"; + private config?: OrExportConfig; private realm: string; private isClearExportBtnDisabled: boolean = true; @@ -275,6 +283,10 @@ export class PageExport extends Page {
+ +
+
+
From 89f89fef14ca03bb50ef819129d05025c17d33f8 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Wed, 23 Jul 2025 17:12:41 +0200 Subject: [PATCH 03/28] add extra formatting option in api and datapoint service --- .../datapoint/AssetDatapointResourceImpl.java | 4 +- .../datapoint/AssetDatapointService.java | 67 ++++++++++++++++--- .../datapoint/AssetDatapointResource.java | 3 +- 3 files changed, 60 insertions(+), 14 deletions(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java index 05c655a180..2a7933c8a0 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java @@ -174,7 +174,7 @@ public DatapointPeriod getDatapointPeriod(RequestParams requestParams, String as } @Override - public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefsString, long fromTimestamp, long toTimestamp) { + public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefsString, long fromTimestamp, long toTimestamp, String format) { try { AttributeRef[] attributeRefs = JSON.readValue(attributeRefsString, AttributeRef[].class); @@ -201,7 +201,7 @@ public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefs DATA_EXPORT_LOG.info("User '" + getUsername() + "' started data export for " + attributeRefsString + " from " + fromTimestamp + " to " + toTimestamp); - ScheduledFuture exportFuture = assetDatapointService.exportDatapoints(attributeRefs, fromTimestamp, toTimestamp); + ScheduledFuture exportFuture = assetDatapointService.exportDatapoints(attributeRefs, fromTimestamp, toTimestamp, format); asyncResponse.register((ConnectionCallback) disconnected -> exportFuture.cancel(true)); diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index 0b8fcf384a..415109b040 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -37,6 +37,8 @@ import java.util.logging.Level; import java.util.logging.Logger; import java.util.stream.Collectors; +import java.util.Set; +import java.util.HashSet; import static java.time.temporal.ChronoUnit.DAYS; import static java.util.stream.Collectors.groupingBy; @@ -251,13 +253,14 @@ protected String buildWhereClause(List>> attributes, b */ public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, long fromTimestamp, - long toTimestamp) { + long toTimestamp, + String format) { try { String query = getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp); // Verify the query is 'legal' and can be executed if(canQueryDatapoints(query, null, datapointExportLimit)) { - return doExportDatapoints(attributeRefs, fromTimestamp, toTimestamp); + return doExportDatapoints(attributeRefs, fromTimestamp, toTimestamp, format); } throw new RuntimeException("Could not export datapoints."); @@ -270,25 +273,67 @@ public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, long fromTimestamp, - long toTimestamp) { + long toTimestamp, + String format) { return scheduledExecutorService.schedule(() -> { String fileName = UniqueIdentifierGenerator.generateId() + ".csv"; - StringBuilder sb = new StringBuilder("copy (") - .append(getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp)) - .append(") to '/storage/") - .append(EXPORT_STORAGE_DIR_NAME) - .append("/") - .append(fileName) - .append("' delimiter ',' CSV HEADER;"); + if (format.equals("Column per attribute")) { + StringBuilder sb = new StringBuilder(String.format( + "copy (select * from crosstab( " + + "'select ad.timestamp, a.name || '' \\: '' || ad.attribute_name as header, ad.value " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) " + + "order by ad.timestamp, header', " + + "'select distinct a.name || '' \\: '' || ad.attribute_name as header " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "order by header') " + + "as ct(timestamp timestamp, %s) " + + ") to '/storage/" + EXPORT_STORAGE_DIR_NAME + "/" + fileName + "' delimiter ',' CSV HEADER;", + fromTimestamp / 1000, toTimestamp / 1000, getAttributeColumns(attributeRefs) + )); + persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); + } else { + StringBuilder sb = new StringBuilder("copy (") + .append(getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp)) + .append(") to '/storage/") + .append(EXPORT_STORAGE_DIR_NAME) + .append("/") + .append(fileName) + .append("' delimiter ',' CSV HEADER;"); + persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); + } + - persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); // The same path must resolve in both the postgresql container and the manager container return exportPath.resolve(fileName).toFile(); }, 0, TimeUnit.MILLISECONDS); } + /** + * Function for building CSV attribute headers when format will make + * a separate column per attribute. + */ + private String getAttributeColumns(AttributeRef[] attributeRefs) { + // Create a set to keep track of unique headers + Set headers = new HashSet<>(); + + // Build unique headers in the format "AssetName: AttributeName" + Arrays.stream(attributeRefs).forEach(attr -> { + String assetName = assetStorageService.findNames(attr.getId()).toString().replaceAll("(^\\[|\\]$)", ""); + String attributeName = attr.getName(); + headers.add(assetName + ": " + attributeName); + }); + + // Return as a single string with the assembled header. + return headers.stream() + .map(header -> "\"" + header + "\" text") + .collect(Collectors.joining(", ")); + } + /** * Function for building an SQL query that SELECTs the content for the data export. * Currently, it includes timestamp, asset name, attribute name, and the data point value. diff --git a/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java b/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java index d2a182dc16..e34de3a684 100644 --- a/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java +++ b/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java @@ -69,6 +69,7 @@ DatapointPeriod getDatapointPeriod(@BeanParam RequestParams requestParams, void getDatapointExport(@Suspended AsyncResponse asyncResponse, @QueryParam("attributeRefs") String attributeRefsString, @QueryParam("fromTimestamp") long fromTimestamp, - @QueryParam("toTimestamp") long toTimestamp); + @QueryParam("toTimestamp") long toTimestamp, + @QueryParam("format") String format); } From e8e18ec3dada05162de9fdabfbdaa7e844a29df4 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Thu, 24 Jul 2025 10:29:21 +0200 Subject: [PATCH 04/28] add tanblefunc migration, fix typo --- .../manager/setup/database/V20250724_01__addTableFunc.sql | 1 + ui/app/manager/src/pages/page-export.ts | 7 ++++--- 2 files changed, 5 insertions(+), 3 deletions(-) create mode 100644 manager/src/main/resources/org/openremote/manager/setup/database/V20250724_01__addTableFunc.sql diff --git a/manager/src/main/resources/org/openremote/manager/setup/database/V20250724_01__addTableFunc.sql b/manager/src/main/resources/org/openremote/manager/setup/database/V20250724_01__addTableFunc.sql new file mode 100644 index 0000000000..c24323fa87 --- /dev/null +++ b/manager/src/main/resources/org/openremote/manager/setup/database/V20250724_01__addTableFunc.sql @@ -0,0 +1 @@ +CREATE EXTENSION IF NOT EXISTS tablefunc; diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index b962554990..c2941e86de 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -207,10 +207,10 @@ export class PageExport extends Page { @property() protected _loading: boolean = false; - protected _exportFormats: String[] = ["Default", "Column per attribute"]; + protected _exportFormats: string[] = ["Default", "Column per attribute"]; @property() - private selectedFormat: String = "Default"; + private selectedFormat: string = "Default"; private config?: OrExportConfig; private realm: string; @@ -404,7 +404,8 @@ export class PageExport extends Page { manager.rest.api.AssetDatapointResource.getDatapointExport({ attributeRefs: JSON.stringify(this.tableRows.map(attr => ({id: attr.assetId, name: attr.attributeName}))), fromTimestamp: this.oldestTimestamp, - toTimestamp: this.latestTimestamp + toTimestamp: this.latestTimestamp, + format: this.selectedFormat }, { responseType: "blob", From 12c5ddac2f3a0c9e57a8909c263bac407c397edf Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Thu, 24 Jul 2025 11:45:20 +0200 Subject: [PATCH 05/28] add translations, make format optional queryparam --- .../manager/datapoint/AssetDatapointResourceImpl.java | 2 +- .../openremote/manager/datapoint/AssetDatapointService.java | 2 +- .../openremote/model/datapoint/AssetDatapointResource.java | 2 +- ui/app/manager/src/pages/page-export.ts | 6 +++--- ui/app/shared/locales/en/or.json | 1 + ui/app/shared/locales/nl/or.json | 1 + 6 files changed, 8 insertions(+), 6 deletions(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java index 2a7933c8a0..0e762be393 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java @@ -199,7 +199,7 @@ public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefs ); } - DATA_EXPORT_LOG.info("User '" + getUsername() + "' started data export for " + attributeRefsString + " from " + fromTimestamp + " to " + toTimestamp); + DATA_EXPORT_LOG.info("User '" + getUsername() + "' started data export for " + attributeRefsString + " from " + fromTimestamp + " to " + toTimestamp + " in format " + format); ScheduledFuture exportFuture = assetDatapointService.exportDatapoints(attributeRefs, fromTimestamp, toTimestamp, format); diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index 415109b040..938bb636b2 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -278,7 +278,7 @@ protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, return scheduledExecutorService.schedule(() -> { String fileName = UniqueIdentifierGenerator.generateId() + ".csv"; - if (format.equals("Column per attribute")) { + if (format.equals("Column per attribute CSV")) { StringBuilder sb = new StringBuilder(String.format( "copy (select * from crosstab( " + "'select ad.timestamp, a.name || '' \\: '' || ad.attribute_name as header, ad.value " + diff --git a/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java b/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java index e34de3a684..ee45c70866 100644 --- a/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java +++ b/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java @@ -70,6 +70,6 @@ void getDatapointExport(@Suspended AsyncResponse asyncResponse, @QueryParam("attributeRefs") String attributeRefsString, @QueryParam("fromTimestamp") long fromTimestamp, @QueryParam("toTimestamp") long toTimestamp, - @QueryParam("format") String format); + @QueryParam("format") @DefaultValue("Default CSV") String format); } diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index c2941e86de..38e45d9842 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -207,10 +207,10 @@ export class PageExport extends Page { @property() protected _loading: boolean = false; - protected _exportFormats: string[] = ["Default", "Column per attribute"]; + protected _exportFormats: string[] = ["Default CSV", "Column per attribute CSV"]; @property() - private selectedFormat: string = "Default"; + private selectedFormat: string = "Default CSV"; private config?: OrExportConfig; private realm: string; @@ -286,7 +286,7 @@ export class PageExport extends Page {
- +
diff --git a/ui/app/shared/locales/en/or.json b/ui/app/shared/locales/en/or.json index 6abf68f5d7..62e7d1171a 100644 --- a/ui/app/shared/locales/en/or.json +++ b/ui/app/shared/locales/en/or.json @@ -607,6 +607,7 @@ "writePermissions": "Write permissions", "valueType": "Value type", "dataExport": "Data export", + "exportFormat": "Export format", "dataSelection": "Data selection", "exportFrom": "Export from", "assetName": "Asset name", diff --git a/ui/app/shared/locales/nl/or.json b/ui/app/shared/locales/nl/or.json index be64bd3df4..8553238cdd 100644 --- a/ui/app/shared/locales/nl/or.json +++ b/ui/app/shared/locales/nl/or.json @@ -606,6 +606,7 @@ "writePermissions": "Schrijfrechten", "valueType": "Waarde type", "dataExport": "Data export", + "exportFormat": "Export format", "dataSelection": "Data selectie", "exportFrom": "Exporteer vanaf", "assetName": "Asset naam", From 88c6f2fabd78709c846f89a42e519769ff1c67ef Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Thu, 24 Jul 2025 12:37:20 +0200 Subject: [PATCH 06/28] improve style --- ui/app/manager/src/pages/page-export.ts | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index 38e45d9842..b1e6861c94 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -176,10 +176,10 @@ export class PageExport extends Page { } .timerange-wrapper { - margin-bottom: 2em + margin-bottom: 1em } - - .format-wrapper { + .options { + width: fit-content; margin-bottom: 2em } .export-btn-wrapper { @@ -280,13 +280,14 @@ export class PageExport extends Page {
${header}
-
- - - -
-
- +
+
+ + +
+
+ +
From 44059906a4049d853d01898dfa66a6c8e49f6ab8 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Thu, 24 Jul 2025 15:06:56 +0200 Subject: [PATCH 07/28] add aggregate option --- .../datapoint/AssetDatapointService.java | 25 ++++++++++++++++++- ui/app/manager/src/pages/page-export.ts | 2 +- 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index 938bb636b2..e2bb67e239 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -295,7 +295,30 @@ protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, fromTimestamp / 1000, toTimestamp / 1000, getAttributeColumns(attributeRefs) )); persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); - } else { + } else if (format.equals("Column per attribute, 30 second average CSV")) { + String bucketInterval = "30 seconds"; // Bucket interval + StringBuilder sb = new StringBuilder(String.format( + "copy (select * from crosstab( " + + "'select public.time_bucket(''%s'', ad.timestamp) as bucket_timestamp, " + + "a.name || '' \\: '' || ad.attribute_name as header, " + + "round(avg(ad.value::numeric), 3) as value " + // Using avg() with rounding to 3 decimal places + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) " + + "group by bucket_timestamp, header " + + "order by bucket_timestamp, header', " + + "'select distinct a.name || '' \\: '' || ad.attribute_name as header " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "order by header') " + + "as ct(timestamp timestamp, %s) " + + ") to '/storage/" + EXPORT_STORAGE_DIR_NAME + "/" + fileName + "' delimiter ',' CSV HEADER;", + bucketInterval, fromTimestamp / 1000, toTimestamp / 1000, getAttributeColumns(attributeRefs) + )); + + persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); + } + else { StringBuilder sb = new StringBuilder("copy (") .append(getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp)) .append(") to '/storage/") diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index b1e6861c94..58ce4326d5 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -207,7 +207,7 @@ export class PageExport extends Page { @property() protected _loading: boolean = false; - protected _exportFormats: string[] = ["Default CSV", "Column per attribute CSV"]; + protected _exportFormats: string[] = ["Default CSV", "Column per attribute CSV", "Column per attribute, 30 second average CSV"]; @property() private selectedFormat: string = "Default CSV"; From 288b331feb9b34e6ff46e294877b52e1a3584843 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Thu, 24 Jul 2025 15:46:42 +0200 Subject: [PATCH 08/28] fix potential resource leak using try-with-resources --- .../datapoint/AssetDatapointResourceImpl.java | 16 ++++++++-------- .../manager/datapoint/AssetDatapointService.java | 7 +++---- ui/app/manager/src/pages/page-export.ts | 2 +- 3 files changed, 12 insertions(+), 13 deletions(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java index 0e762be393..054011fd7e 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java @@ -210,14 +210,14 @@ public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefs try { exportFile = exportFuture.get(); - ZipOutputStream zipOut = new ZipOutputStream(response.getOutputStream()); - FileInputStream fin = new FileInputStream(exportFile); - ZipEntry zipEntry = new ZipEntry(exportFile.getName()); - zipOut.putNextEntry(zipEntry); - IOUtils.copy(fin, zipOut); - zipOut.closeEntry(); - zipOut.close(); - fin.close(); + try (FileInputStream fin = new FileInputStream(exportFile); + ZipOutputStream zipOut = new ZipOutputStream(response.getOutputStream())) { + + ZipEntry zipEntry = new ZipEntry(exportFile.getName()); + zipOut.putNextEntry(zipEntry); + IOUtils.copy(fin, zipOut); + zipOut.closeEntry(); + } response.setContentType("application/zip"); response.setHeader(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"dataexport.zip\""); diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index e2bb67e239..04f4651f9b 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -278,7 +278,7 @@ protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, return scheduledExecutorService.schedule(() -> { String fileName = UniqueIdentifierGenerator.generateId() + ".csv"; - if (format.equals("Column per attribute CSV")) { + if ("Column per attribute CSV".equals(format)) { StringBuilder sb = new StringBuilder(String.format( "copy (select * from crosstab( " + "'select ad.timestamp, a.name || '' \\: '' || ad.attribute_name as header, ad.value " + @@ -295,8 +295,7 @@ protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, fromTimestamp / 1000, toTimestamp / 1000, getAttributeColumns(attributeRefs) )); persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); - } else if (format.equals("Column per attribute, 30 second average CSV")) { - String bucketInterval = "30 seconds"; // Bucket interval + } else if ("Column per attribute CSV - 1 minute averages".equals(format)) { StringBuilder sb = new StringBuilder(String.format( "copy (select * from crosstab( " + "'select public.time_bucket(''%s'', ad.timestamp) as bucket_timestamp, " + @@ -313,7 +312,7 @@ protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, "order by header') " + "as ct(timestamp timestamp, %s) " + ") to '/storage/" + EXPORT_STORAGE_DIR_NAME + "/" + fileName + "' delimiter ',' CSV HEADER;", - bucketInterval, fromTimestamp / 1000, toTimestamp / 1000, getAttributeColumns(attributeRefs) + "1 minute", fromTimestamp / 1000, toTimestamp / 1000, getAttributeColumns(attributeRefs) )); persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index 58ce4326d5..723f778955 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -207,7 +207,7 @@ export class PageExport extends Page { @property() protected _loading: boolean = false; - protected _exportFormats: string[] = ["Default CSV", "Column per attribute CSV", "Column per attribute, 30 second average CSV"]; + protected _exportFormats: string[] = ["Default CSV", "Column per attribute CSV", "Column per attribute CSV - 1 minute averages"]; @property() private selectedFormat: string = "Default CSV"; From 6e428fad01629bdd09c494778584bd1d2ac2f455 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Thu, 24 Jul 2025 17:03:00 +0200 Subject: [PATCH 09/28] convert queryparameter to a number in backend and keyed object in frontend --- .../manager/datapoint/AssetDatapointResourceImpl.java | 6 +++++- .../manager/datapoint/AssetDatapointService.java | 8 ++++---- .../model/datapoint/AssetDatapointResource.java | 2 +- ui/app/manager/src/pages/page-export.ts | 6 +++--- 4 files changed, 13 insertions(+), 9 deletions(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java index 054011fd7e..c9fbe35e97 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java @@ -174,7 +174,7 @@ public DatapointPeriod getDatapointPeriod(RequestParams requestParams, String as } @Override - public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefsString, long fromTimestamp, long toTimestamp, String format) { + public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefsString, long fromTimestamp, long toTimestamp, int format) { try { AttributeRef[] attributeRefs = JSON.readValue(attributeRefsString, AttributeRef[].class); @@ -189,6 +189,10 @@ public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefs throw new WebApplicationException(Response.Status.NOT_FOUND); } + if (format > 3 || format < 1){ + throw new WebApplicationException(Response.Status.BAD_REQUEST); + } + if (!isRealmActiveAndAccessible(asset.getRealm())) { DATA_EXPORT_LOG.info("Forbidden access for user '" + getUsername() + "': " + asset); throw new WebApplicationException(Response.Status.FORBIDDEN); diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index 04f4651f9b..3d199ed8e9 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -254,7 +254,7 @@ protected String buildWhereClause(List>> attributes, b public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, long fromTimestamp, long toTimestamp, - String format) { + int format) { try { String query = getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp); @@ -274,11 +274,11 @@ public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, long fromTimestamp, long toTimestamp, - String format) { + int format) { return scheduledExecutorService.schedule(() -> { String fileName = UniqueIdentifierGenerator.generateId() + ".csv"; - if ("Column per attribute CSV".equals(format)) { + if (format==2) { StringBuilder sb = new StringBuilder(String.format( "copy (select * from crosstab( " + "'select ad.timestamp, a.name || '' \\: '' || ad.attribute_name as header, ad.value " + @@ -295,7 +295,7 @@ protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, fromTimestamp / 1000, toTimestamp / 1000, getAttributeColumns(attributeRefs) )); persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); - } else if ("Column per attribute CSV - 1 minute averages".equals(format)) { + } else if (format==3) { StringBuilder sb = new StringBuilder(String.format( "copy (select * from crosstab( " + "'select public.time_bucket(''%s'', ad.timestamp) as bucket_timestamp, " + diff --git a/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java b/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java index ee45c70866..ef0c371d8e 100644 --- a/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java +++ b/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java @@ -70,6 +70,6 @@ void getDatapointExport(@Suspended AsyncResponse asyncResponse, @QueryParam("attributeRefs") String attributeRefsString, @QueryParam("fromTimestamp") long fromTimestamp, @QueryParam("toTimestamp") long toTimestamp, - @QueryParam("format") @DefaultValue("Default CSV") String format); + @QueryParam("format") @DefaultValue("1") int format); } diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index 723f778955..3ecc6cdc0d 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -207,10 +207,10 @@ export class PageExport extends Page { @property() protected _loading: boolean = false; - protected _exportFormats: string[] = ["Default CSV", "Column per attribute CSV", "Column per attribute CSV - 1 minute averages"]; + protected _exportFormats: {[key: number]: string} = {1: 'Default .csv', 2: "Column per attribute .csv", 3: "Column per attributes .csv with 1 minute averages"}; @property() - private selectedFormat: string = "Default CSV"; + private selectedFormat: number = 1; private config?: OrExportConfig; private realm: string; @@ -286,7 +286,7 @@ export class PageExport extends Page {
- +
From 1a9870b087c38ec0d85e61e9cf4d9b4c2b05d3dc Mon Sep 17 00:00:00 2001 From: Wouter Voorberg <119453250+Hackerberg43@users.noreply.github.com> Date: Fri, 25 Jul 2025 22:04:22 +0200 Subject: [PATCH 10/28] Update manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../org/openremote/manager/datapoint/AssetDatapointService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index 3d199ed8e9..d6f4302a44 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -278,7 +278,7 @@ protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, return scheduledExecutorService.schedule(() -> { String fileName = UniqueIdentifierGenerator.generateId() + ".csv"; - if (format==2) { + if (format == 2) { StringBuilder sb = new StringBuilder(String.format( "copy (select * from crosstab( " + "'select ad.timestamp, a.name || '' \\: '' || ad.attribute_name as header, ad.value " + From 7bc414ca8b73d0bce3ea7ac319f34500e54c15f3 Mon Sep 17 00:00:00 2001 From: Wouter Voorberg <119453250+Hackerberg43@users.noreply.github.com> Date: Fri, 25 Jul 2025 22:04:46 +0200 Subject: [PATCH 11/28] Update manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../org/openremote/manager/datapoint/AssetDatapointService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index d6f4302a44..3ebc6dc5f7 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -295,7 +295,7 @@ protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, fromTimestamp / 1000, toTimestamp / 1000, getAttributeColumns(attributeRefs) )); persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); - } else if (format==3) { + } else if (format == 3) { StringBuilder sb = new StringBuilder(String.format( "copy (select * from crosstab( " + "'select public.time_bucket(''%s'', ad.timestamp) as bucket_timestamp, " + From 6b3ad70fb3763748bfd168b48723eb82cc3a3173 Mon Sep 17 00:00:00 2001 From: Wouter Voorberg <119453250+Hackerberg43@users.noreply.github.com> Date: Fri, 25 Jul 2025 22:04:59 +0200 Subject: [PATCH 12/28] Update manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../manager/datapoint/AssetDatapointResourceImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java index c9fbe35e97..a1b92d0de8 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java @@ -189,7 +189,7 @@ public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefs throw new WebApplicationException(Response.Status.NOT_FOUND); } - if (format > 3 || format < 1){ + if (format > 3 || format < 1) { throw new WebApplicationException(Response.Status.BAD_REQUEST); } From 041abe3b916093ad9ef2cbf7d4b5e8d36ec653c7 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Fri, 17 Oct 2025 14:39:10 +0200 Subject: [PATCH 13/28] update migration script, change export format names, fix html typo --- ...4_01__addTableFunc.sql => V20251017_01__addTableFunc.sql} | 0 ui/app/manager/src/pages/page-export.ts | 5 +---- 2 files changed, 1 insertion(+), 4 deletions(-) rename manager/src/main/resources/org/openremote/manager/setup/database/{V20250724_01__addTableFunc.sql => V20251017_01__addTableFunc.sql} (100%) diff --git a/manager/src/main/resources/org/openremote/manager/setup/database/V20250724_01__addTableFunc.sql b/manager/src/main/resources/org/openremote/manager/setup/database/V20251017_01__addTableFunc.sql similarity index 100% rename from manager/src/main/resources/org/openremote/manager/setup/database/V20250724_01__addTableFunc.sql rename to manager/src/main/resources/org/openremote/manager/setup/database/V20251017_01__addTableFunc.sql diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index c864277df2..e271639e7c 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -207,7 +207,7 @@ export class PageExport extends Page { @property() protected _loading: boolean = false; - protected _exportFormats: {[key: number]: string} = {1: 'Default .csv', 2: "Column per attribute .csv", 3: "Column per attributes .csv with 1 minute averages"}; + protected _exportFormats: {[key: number]: string} = {1: 'CSV: Default', 2: "CSV: Multicolumn", 3: "CSV: Multicolumn | 1min averages"}; @property() private selectedFormat: number = 1; @@ -290,9 +290,6 @@ export class PageExport extends Page {
- -
-
From 9d719b8e0dcce823136504dffe2604c278f2281d Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Fri, 17 Oct 2025 15:20:47 +0200 Subject: [PATCH 14/28] add overload for backwards compatibility - fixes failing test -added tests for the new formats --- .../datapoint/AssetDatapointService.java | 11 ++++ .../assets/AssetDatapointExportTest.groovy | 61 ++++++++++++++++++- 2 files changed, 70 insertions(+), 2 deletions(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index 3ebc6dc5f7..ea206acaa7 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -247,6 +247,17 @@ protected String buildWhereClause(List>> attributes, b return " and (dp.assetId, dp.attributeName) " + (negate ? "not " : "") + "in (" + whereStr + ")"; } + /** + * Exports datapoints as CSV using SQL; the export path used in the SQL query must also be mapped into the manager + * container so it can be accessed by this process. + * Backwards compatible overload with default format. + */ + public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, + long fromTimestamp, + long toTimestamp) { + return exportDatapoints(attributeRefs, fromTimestamp, toTimestamp, 1); + } + /** * Exports datapoints as CSV using SQL; the export path used in the SQL query must also be mapped into the manager * container so it can be accessed by this process. diff --git a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy index 9a047c240f..0175ceb411 100644 --- a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy +++ b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy @@ -66,7 +66,7 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer ] ) - then: "the CSV export should return a file" + then: "the default CSV export should return a file" def csvExport1Future = assetDatapointService.exportDatapoints( [new AttributeRef(asset.id, attributeName)] as AttributeRef[], dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), @@ -87,11 +87,68 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer /* ------------------------- */ + when: "exporting with format 2 (crosstab)" + def csvExport2Future = assetDatapointService.exportDatapoints( + [new AttributeRef(asset.id, attributeName)] as AttributeRef[], + dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), + dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), + 2 + ) + assert csvExport2Future != null + def csvExport2 = csvExport2Future.get() + assert csvExport2 != null + + then: "the crosstab CSV export should contain all 5 data points in columnar format" + def csvExport2Lines = csvExport2.readLines() + assert csvExport2Lines.size() == 6 // header + 5 data rows + assert csvExport2Lines[0].contains("timestamp") + assert csvExport2Lines[0].contains(assetName + ": " + attributeName) + + /* ------------------------- */ + + when: "exporting with format 3 (time-bucketed)" + def csvExport3Future = assetDatapointService.exportDatapoints( + [new AttributeRef(asset.id, attributeName)] as AttributeRef[], + dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), + dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), + 3 + ) + assert csvExport3Future != null + def csvExport3 = csvExport3Future.get() + assert csvExport3 != null + + then: "the time-bucketed CSV export should contain aggregated data with correct values" + def csvExport3Lines = csvExport3.readLines() + assert csvExport3Lines.size() >= 2 // header + at least 1 aggregated row + assert csvExport3Lines[0].contains("timestamp") + assert csvExport3Lines[0].contains(assetName + ": " + attributeName) + + and: "the aggregated values should be correct averages within 1-minute buckets" + // Each datapoint is 5 minutes apart, so each should be in its own 1-minute bucket + // Therefore we should have 5 rows of data (one per bucket) + assert csvExport3Lines.size() == 6 // header + 5 buckets + + // Extract the values from the CSV (skip header, get last column which is the brightness value) + def values = csvExport3Lines[1..5].collect { line -> + def parts = line.split(',') + parts.size() > 1 ? parts[1].trim() : null + }.findAll { it != null && it != '' } + + // Should have 5 values, one per bucket, matching our input: 10, 20, 30, 40, 50 + assert values.size() == 5 + assert values.contains("10.000") || values.contains("10") + assert values.contains("20.000") || values.contains("20") + assert values.contains("30.000") || values.contains("30") + assert values.contains("40.000") || values.contains("40") + assert values.contains("50.000") || values.contains("50") + + /* ------------------------- */ + when: "the limit of data export is lowered to 4" assetDatapointService.datapointExportLimit = 4 and: "we try to export the same amount of data points" - def csvExport2Future = assetDatapointService.exportDatapoints( + def csvExport4Future = assetDatapointService.exportDatapoints( [new AttributeRef(asset.id, attributeName)] as AttributeRef[], dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() From 92465091f7acd80df65bea8eae0ef83a148511bc Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Fri, 17 Oct 2025 17:14:39 +0200 Subject: [PATCH 15/28] review comment - fix for non numeric data --- .../datapoint/AssetDatapointService.java | 30 +++++++++++++++---- ui/app/manager/src/pages/page-export.ts | 6 ++-- 2 files changed, 28 insertions(+), 8 deletions(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index ea206acaa7..5b1cd58a8c 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -290,40 +290,48 @@ protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, return scheduledExecutorService.schedule(() -> { String fileName = UniqueIdentifierGenerator.generateId() + ".csv"; if (format == 2) { + String attributeFilter = getAttributeFilter(attributeRefs); StringBuilder sb = new StringBuilder(String.format( "copy (select * from crosstab( " + "'select ad.timestamp, a.name || '' \\: '' || ad.attribute_name as header, ad.value " + "from asset_datapoint ad " + "join asset a on ad.entity_id = a.id " + - "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) and (%s) " + "order by ad.timestamp, header', " + "'select distinct a.name || '' \\: '' || ad.attribute_name as header " + "from asset_datapoint ad " + "join asset a on ad.entity_id = a.id " + + "where %s " + "order by header') " + "as ct(timestamp timestamp, %s) " + ") to '/storage/" + EXPORT_STORAGE_DIR_NAME + "/" + fileName + "' delimiter ',' CSV HEADER;", - fromTimestamp / 1000, toTimestamp / 1000, getAttributeColumns(attributeRefs) + fromTimestamp / 1000, toTimestamp / 1000, attributeFilter, attributeFilter, getAttributeColumns(attributeRefs) )); persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); } else if (format == 3) { + String attributeFilter = getAttributeFilter(attributeRefs); StringBuilder sb = new StringBuilder(String.format( "copy (select * from crosstab( " + "'select public.time_bucket(''%s'', ad.timestamp) as bucket_timestamp, " + "a.name || '' \\: '' || ad.attribute_name as header, " + - "round(avg(ad.value::numeric), 3) as value " + // Using avg() with rounding to 3 decimal places + "CASE " + + " WHEN jsonb_typeof((array_agg(ad.value))[1]) = ''number'' THEN " + + " round(avg((ad.value#>>''{}'')::numeric) FILTER (WHERE jsonb_typeof(ad.value) = ''number''), 3)::text " + + " ELSE (array_agg(ad.value ORDER BY ad.timestamp DESC) FILTER (WHERE jsonb_typeof(ad.value) != ''number''))[1]#>>''{}''" + + "END as value " + "from asset_datapoint ad " + "join asset a on ad.entity_id = a.id " + - "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) and (%s) " + "group by bucket_timestamp, header " + "order by bucket_timestamp, header', " + "'select distinct a.name || '' \\: '' || ad.attribute_name as header " + "from asset_datapoint ad " + "join asset a on ad.entity_id = a.id " + + "where %s " + "order by header') " + "as ct(timestamp timestamp, %s) " + ") to '/storage/" + EXPORT_STORAGE_DIR_NAME + "/" + fileName + "' delimiter ',' CSV HEADER;", - "1 minute", fromTimestamp / 1000, toTimestamp / 1000, getAttributeColumns(attributeRefs) + "1 minute", fromTimestamp / 1000, toTimestamp / 1000, attributeFilter, attributeFilter, getAttributeColumns(attributeRefs) )); persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); @@ -346,6 +354,18 @@ protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, }, 0, TimeUnit.MILLISECONDS); } + /** + * Function for building a WHERE clause filter for the requested attributes. + * Used in crosstab queries to ensure only the requested attributes are included. + * Note: Uses double single quotes for SQL string escaping within crosstab queries. + */ + private String getAttributeFilter(AttributeRef[] attributeRefs) { + return Arrays.stream(attributeRefs) + .map(attr -> String.format("(ad.entity_id = ''%s'' and ad.attribute_name = ''%s'')", + attr.getId(), attr.getName())) + .collect(Collectors.joining(" or ")); + } + /** * Function for building CSV attribute headers when format will make * a separate column per attribute. diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index e271639e7c..ec78f7e9b6 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -207,7 +207,7 @@ export class PageExport extends Page { @property() protected _loading: boolean = false; - protected _exportFormats: {[key: number]: string} = {1: 'CSV: Default', 2: "CSV: Multicolumn", 3: "CSV: Multicolumn | 1min averages"}; + protected _exportFormats: {[key: number]: string} = {1: 'CSVDefault', 2: "CSVCrosstab", 3: "CSVCrosstab | 1min interval"}; @property() private selectedFormat: number = 1; @@ -418,7 +418,7 @@ export class PageExport extends Page { link.setAttribute("download", "dataexport.zip"); document.body.appendChild(link); link.click(); - + this._loading = false; }).catch(ex => { if(isAxiosError(ex)) { if(ex.response?.status === 413) { @@ -431,8 +431,8 @@ export class PageExport extends Page { showSnackbar(undefined, "errorOccurred"); } } + this._loading = false; }); - this._loading = false; } protected async loadConfig() { From 332d3bb7f7c39d583abdc1afbae03b63a152c288 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Thu, 27 Nov 2025 11:59:11 +0100 Subject: [PATCH 16/28] fix enum review comments --- .../datapoint/AssetDatapointResourceImpl.java | 11 ++++---- .../datapoint/AssetDatapointService.java | 11 ++++---- .../datapoint/AssetDatapointResource.java | 2 +- .../datapoint/DatapointExportFormat.java | 27 +++++++++++++++++++ .../assets/AssetDatapointExportTest.groovy | 9 ++++--- ui/app/manager/src/pages/page-export.ts | 12 ++++++--- ui/app/shared/locales/en/or.json | 3 +++ ui/app/shared/locales/nl/or.json | 3 +++ 8 files changed, 59 insertions(+), 19 deletions(-) create mode 100644 model/src/main/java/org/openremote/model/datapoint/DatapointExportFormat.java diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java index a1b92d0de8..a8f619851c 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java @@ -39,6 +39,7 @@ import org.openremote.model.attribute.AttributeRef; import org.openremote.model.datapoint.AssetDatapointResource; import org.openremote.model.datapoint.DatapointPeriod; +import org.openremote.model.datapoint.DatapointExportFormat; import org.openremote.model.datapoint.DatapointQueryTooLargeException; import org.openremote.model.datapoint.ValueDatapoint; import org.openremote.model.datapoint.query.AssetDatapointQuery; @@ -174,8 +175,12 @@ public DatapointPeriod getDatapointPeriod(RequestParams requestParams, String as } @Override - public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefsString, long fromTimestamp, long toTimestamp, int format) { + public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefsString, long fromTimestamp, long toTimestamp, DatapointExportFormat format) { try { + if (format == null) { + throw new WebApplicationException(Response.Status.BAD_REQUEST); + } + AttributeRef[] attributeRefs = JSON.readValue(attributeRefsString, AttributeRef[].class); for (AttributeRef attributeRef : attributeRefs) { @@ -189,10 +194,6 @@ public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefs throw new WebApplicationException(Response.Status.NOT_FOUND); } - if (format > 3 || format < 1) { - throw new WebApplicationException(Response.Status.BAD_REQUEST); - } - if (!isRealmActiveAndAccessible(asset.getRealm())) { DATA_EXPORT_LOG.info("Forbidden access for user '" + getUsername() + "': " + asset); throw new WebApplicationException(Response.Status.FORBIDDEN); diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index 72c9c898ca..c462eed875 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -3,6 +3,7 @@ import org.openremote.agent.protocol.ProtocolDatapointService; import org.openremote.container.timer.TimerService; import org.openremote.manager.asset.OutdatedAttributeEvent; +import org.openremote.model.datapoint.DatapointExportFormat; import org.openremote.model.datapoint.DatapointQueryTooLargeException; import org.openremote.model.util.UniqueIdentifierGenerator; import org.openremote.manager.asset.AssetProcessingException; @@ -255,7 +256,7 @@ protected String buildWhereClause(List>> attributes, b public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, long fromTimestamp, long toTimestamp) { - return exportDatapoints(attributeRefs, fromTimestamp, toTimestamp, 1); + return exportDatapoints(attributeRefs, fromTimestamp, toTimestamp, DatapointExportFormat.CSV); } /** @@ -265,7 +266,7 @@ public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, long fromTimestamp, long toTimestamp, - int format) { + DatapointExportFormat format) { try { String query = getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp); @@ -285,11 +286,11 @@ public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, long fromTimestamp, long toTimestamp, - int format) { + DatapointExportFormat format) { return scheduledExecutorService.schedule(() -> { String fileName = UniqueIdentifierGenerator.generateId() + ".csv"; - if (format == 2) { + if (format == DatapointExportFormat.CSV_CROSSTAB) { String attributeFilter = getAttributeFilter(attributeRefs); StringBuilder sb = new StringBuilder(String.format( "copy (select * from crosstab( " + @@ -308,7 +309,7 @@ protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, fromTimestamp / 1000, toTimestamp / 1000, attributeFilter, attributeFilter, getAttributeColumns(attributeRefs) )); persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); - } else if (format == 3) { + } else if (format == DatapointExportFormat.CSV_CROSSTAB_MINUTE) { String attributeFilter = getAttributeFilter(attributeRefs); StringBuilder sb = new StringBuilder(String.format( "copy (select * from crosstab( " + diff --git a/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java b/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java index ef0c371d8e..d98e919a28 100644 --- a/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java +++ b/model/src/main/java/org/openremote/model/datapoint/AssetDatapointResource.java @@ -70,6 +70,6 @@ void getDatapointExport(@Suspended AsyncResponse asyncResponse, @QueryParam("attributeRefs") String attributeRefsString, @QueryParam("fromTimestamp") long fromTimestamp, @QueryParam("toTimestamp") long toTimestamp, - @QueryParam("format") @DefaultValue("1") int format); + @QueryParam("format") @DefaultValue("CSV") DatapointExportFormat format); } diff --git a/model/src/main/java/org/openremote/model/datapoint/DatapointExportFormat.java b/model/src/main/java/org/openremote/model/datapoint/DatapointExportFormat.java new file mode 100644 index 0000000000..4ae4e72479 --- /dev/null +++ b/model/src/main/java/org/openremote/model/datapoint/DatapointExportFormat.java @@ -0,0 +1,27 @@ +/* + * Copyright 2025, OpenRemote Inc. + * + * See the CONTRIBUTORS.txt file in the distribution for a + * full listing of individual contributors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package org.openremote.model.datapoint; + +public enum DatapointExportFormat { + + CSV, + CSV_CROSSTAB, + CSV_CROSSTAB_MINUTE +} diff --git a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy index 0175ceb411..d7af45a893 100644 --- a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy +++ b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy @@ -5,6 +5,7 @@ import org.openremote.manager.datapoint.AssetDatapointService import org.openremote.manager.setup.SetupService import org.openremote.model.asset.impl.LightAsset import org.openremote.model.attribute.AttributeRef +import org.openremote.model.datapoint.DatapointExportFormat import org.openremote.model.datapoint.DatapointQueryTooLargeException import org.openremote.model.datapoint.ValueDatapoint import org.openremote.model.query.AssetQuery @@ -87,12 +88,12 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer /* ------------------------- */ - when: "exporting with format 2 (crosstab)" + when: "exporting with format CSV_CROSSTAB" def csvExport2Future = assetDatapointService.exportDatapoints( [new AttributeRef(asset.id, attributeName)] as AttributeRef[], dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), - 2 + DatapointExportFormat.CSV_CROSSTAB ) assert csvExport2Future != null def csvExport2 = csvExport2Future.get() @@ -106,12 +107,12 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer /* ------------------------- */ - when: "exporting with format 3 (time-bucketed)" + when: "exporting with format CSV_CROSSTAB_MINUTE" def csvExport3Future = assetDatapointService.exportDatapoints( [new AttributeRef(asset.id, attributeName)] as AttributeRef[], dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), - 3 + DatapointExportFormat.CSV_CROSSTAB_MINUTE ) assert csvExport3Future != null def csvExport3 = csvExport3Future.get() diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index ec78f7e9b6..0bb2293ee5 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -8,7 +8,7 @@ import {i18next} from "@openremote/or-translate"; import manager, { DefaultColor3, Util } from "@openremote/core"; import { InputType, OrInputChangedEvent } from "@openremote/or-mwc-components/or-mwc-input"; import {OrAssetAttributePickerPickedEvent, OrAssetAttributePicker} from "@openremote/or-attribute-picker"; -import { AttributeRef } from "@openremote/model"; +import { AttributeRef, DatapointExportFormat } from "@openremote/model"; import moment from "moment"; import { buttonStyle } from "@openremote/or-rules"; import {createSelector} from "reselect"; @@ -207,10 +207,14 @@ export class PageExport extends Page { @property() protected _loading: boolean = false; - protected _exportFormats: {[key: number]: string} = {1: 'CSVDefault', 2: "CSVCrosstab", 3: "CSVCrosstab | 1min interval"}; + protected _exportFormats: Record = { + [DatapointExportFormat.CSV]: i18next.t('exportFormatCSV'), + [DatapointExportFormat.CSV_CROSSTAB]: i18next.t('exportFormatCSVCrosstab'), + [DatapointExportFormat.CSV_CROSSTAB_MINUTE]: i18next.t('exportFormatCSVCrosstabMinute') + }; @property() - private selectedFormat: number = 1; + private selectedFormat: DatapointExportFormat = DatapointExportFormat.CSV; private config?: OrExportConfig; private realm: string; @@ -286,7 +290,7 @@ export class PageExport extends Page {
- +
diff --git a/ui/app/shared/locales/en/or.json b/ui/app/shared/locales/en/or.json index cbdca952f8..eb13fd0c1a 100644 --- a/ui/app/shared/locales/en/or.json +++ b/ui/app/shared/locales/en/or.json @@ -637,6 +637,9 @@ "valueType": "Value type", "dataExport": "Data export", "exportFormat": "Export format", + "exportFormatCSV": "CSV", + "exportFormatCSVCrosstab": "CSV crosstab", + "exportFormatCSVCrosstabMinute": "CSV crosstab (1 min interval)", "dataSelection": "Data selection", "exportFrom": "Export from", "assetName": "Asset name", diff --git a/ui/app/shared/locales/nl/or.json b/ui/app/shared/locales/nl/or.json index 7231af5987..96c1a6d712 100644 --- a/ui/app/shared/locales/nl/or.json +++ b/ui/app/shared/locales/nl/or.json @@ -636,6 +636,9 @@ "valueType": "Waarde type", "dataExport": "Data export", "exportFormat": "Export format", + "exportFormatCSV": "CSV", + "exportFormatCSVCrosstab": "CSV kruistabel", + "exportFormatCSVCrosstabMinute": "CSV kruistabel (1 min interval)", "dataSelection": "Data selectie", "exportFrom": "Exporteer vanaf", "assetName": "Asset naam", From 496355b09531498a37ca087b6bbf343dde55e3f0 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Mon, 1 Dec 2025 15:11:33 +0100 Subject: [PATCH 17/28] add export frontend test --- ui/app/manager/test/export.test.ts | 112 +++++++++++++++++++++++++++++ 1 file changed, 112 insertions(+) create mode 100644 ui/app/manager/test/export.test.ts diff --git a/ui/app/manager/test/export.test.ts b/ui/app/manager/test/export.test.ts new file mode 100644 index 0000000000..9f5c9c0451 --- /dev/null +++ b/ui/app/manager/test/export.test.ts @@ -0,0 +1,112 @@ +import { expect } from "@openremote/test"; +import { test, userStatePath } from "./fixtures/manager"; +import { preparedAssetsForRules } from "./fixtures/data/assets"; +import { AttributeEvent } from "@openremote/model"; + +test.use({ storageState: userStatePath }); + +test.beforeEach(async ({ manager }) => { + await manager.setup("smartcity", { assets: exportAssets }); +}); + +test.afterEach(async ({ manager }) => { + await manager.cleanUp(); +}); + +// Reuse assets from rules test which have storeDataPoints: true +const exportAssets = preparedAssetsForRules; +const exportAttributeName = "energyLevel"; + +/** + * Generate attribute events with timestamps for creating historical datapoints. + * @param assetId The asset ID to create events for + * @param attributeName The attribute name to create events for + * @param count Number of datapoints to create + * @param intervalMinutes Interval between datapoints in minutes + */ +function generateDatapointEvents(assetId: string, attributeName: string, count: number = 5, intervalMinutes: number = 5): AttributeEvent[] { + const now = Date.now(); + const events: AttributeEvent[] = []; + + for (let i = 0; i < count; i++) { + events.push({ + eventType: "attribute", + ref: { + id: assetId, + name: attributeName + }, + value: 20 + i * 2, + timestamp: now - (count - 1 - i) * intervalMinutes * 60 * 1000 + }); + } + + return events; +} + +/** + * Add an attribute to the export table using the attribute picker dialog. + * @param page The page to find "add asset attribute" on + * @param assetName The name of the asset to select in the asset tree + * @param attributeName The name of the attribute to select + */ +async function addAttributeViaPicker(page: import("@playwright/test").Page, assetName: string, attributeName: string) { + await page.locator("a.button:has-text('Add asset attribute')").click(); + const dialog = page.locator("or-asset-attribute-picker"); + await dialog.locator(".mdc-dialog--open").waitFor(); + + const assetTree = dialog.locator("or-asset-tree"); + await assetTree.locator("li.asset-list-element").first().waitFor(); + await assetTree.locator(`li.asset-list-element:has-text("${assetName}")`).click(); + + const attributeList = dialog.locator("#attribute-selector"); + await attributeList.waitFor(); + const displayName = attributeName.replace(/([a-z])([A-Z])/g, "$1 $2").replace(/^./, c => c.toUpperCase()); + const attributeCheckbox = attributeList.locator("li.mdc-list-item").filter({ hasText: displayName }); + await attributeCheckbox.click(); + + await dialog.locator("#add-btn").click(); + await dialog.locator(".mdc-dialog--open").waitFor({ state: "detached" }); +} + +/** + * @given An asset with datapoints exists + * @and Logged into the "smartcity" realm + * @and Navigated to the "Export" page + * @when Selecting an attribute with datapoints + * @and Clicking export + * @then The API call to the file export download should return ok + */ +test("Export datapoints successfully triggers download", async ({ page, manager }) => { + const asset = manager.assets[0]; + const events = generateDatapointEvents(asset.id!, exportAttributeName, 5); + await manager.api.AssetResource.writeAttributeEvents(events); + + await manager.goToRealmStartPage("smartcity"); + await page.waitForLoadState("networkidle"); + await manager.navigateToMenuItem("export"); + await page.locator("text=Data export").waitFor({ state: "visible" }); + + await addAttributeViaPicker(page, asset.name!, exportAttributeName); + await page.locator("table tbody tr").first().waitFor({ state: "visible", timeout: 15000 }); + + const oldestDatapoint = await page.locator("table tbody tr").first().locator("td").nth(2).textContent(); + const latestDatapoint = await page.locator("table tbody tr").first().locator("td").nth(3).textContent(); + const toDatetimeLocal = (dateStr: string) => { + const date = new Date(dateStr); + return date.toISOString().slice(0, 16); + }; + await page.locator("or-mwc-input[label='Export from']").locator("input").fill(toDatetimeLocal(oldestDatapoint!)); + await page.locator("or-mwc-input[label='To']").locator("input").fill(toDatetimeLocal(latestDatapoint!)); + + await page.locator("or-mwc-input[label='Export format']").click(); + await page.locator("li.mdc-list-item:has-text('CSV')").first().click(); + + const exportResponsePromise = page.waitForResponse( + response => response.url().includes("/asset/datapoint/export") + ); + await page.getByRole("button", { name: "Export", exact: true }).click(); + + const response = await exportResponsePromise; + expect(response.ok()).toBe(true); + await expect(page.locator("or-mwc-snackbar .mdc-snackbar--open")).not.toBeAttached({ timeout: 2000 }); +}); From e693666b1293b53d342a65e30a5f9fef2d7adae1 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Mon, 1 Dec 2025 15:26:57 +0100 Subject: [PATCH 18/28] bypass camelCaseToSentenceCase --- ui/app/manager/src/pages/page-export.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ui/app/manager/src/pages/page-export.ts b/ui/app/manager/src/pages/page-export.ts index 0bb2293ee5..92e7612d3e 100644 --- a/ui/app/manager/src/pages/page-export.ts +++ b/ui/app/manager/src/pages/page-export.ts @@ -290,7 +290,7 @@ export class PageExport extends Page {
- +
From 08fb8c2c7cecb99caba3fcedb4eb25b3fc35ad4f Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Mon, 1 Dec 2025 15:34:04 +0100 Subject: [PATCH 19/28] adjust migration script date --- ...251017_01__addTableFunc.sql => V20251201_01__addTableFunc.sql} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename manager/src/main/resources/org/openremote/manager/setup/database/{V20251017_01__addTableFunc.sql => V20251201_01__addTableFunc.sql} (100%) diff --git a/manager/src/main/resources/org/openremote/manager/setup/database/V20251017_01__addTableFunc.sql b/manager/src/main/resources/org/openremote/manager/setup/database/V20251201_01__addTableFunc.sql similarity index 100% rename from manager/src/main/resources/org/openremote/manager/setup/database/V20251017_01__addTableFunc.sql rename to manager/src/main/resources/org/openremote/manager/setup/database/V20251201_01__addTableFunc.sql From ec1f7059e2377df9e97dfe20e993712029eaed56 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Wed, 3 Dec 2025 15:26:45 +0100 Subject: [PATCH 20/28] read http return status --- ui/app/manager/test/export.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ui/app/manager/test/export.test.ts b/ui/app/manager/test/export.test.ts index 9f5c9c0451..1f6ac6df2f 100644 --- a/ui/app/manager/test/export.test.ts +++ b/ui/app/manager/test/export.test.ts @@ -107,6 +107,6 @@ test("Export datapoints successfully triggers download", async ({ page, manager await page.getByRole("button", { name: "Export", exact: true }).click(); const response = await exportResponsePromise; - expect(response.ok()).toBe(true); + expect(response.status).toBe(200); await expect(page.locator("or-mwc-snackbar .mdc-snackbar--open")).not.toBeAttached({ timeout: 2000 }); }); From 6e9bad29a71a83f3f06c103cbe2407faa1cae941 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Wed, 3 Dec 2025 16:13:16 +0100 Subject: [PATCH 21/28] oops --- ui/app/manager/test/export.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ui/app/manager/test/export.test.ts b/ui/app/manager/test/export.test.ts index 1f6ac6df2f..4fb88e7224 100644 --- a/ui/app/manager/test/export.test.ts +++ b/ui/app/manager/test/export.test.ts @@ -107,6 +107,6 @@ test("Export datapoints successfully triggers download", async ({ page, manager await page.getByRole("button", { name: "Export", exact: true }).click(); const response = await exportResponsePromise; - expect(response.status).toBe(200); + expect(response.status()).toBe(200); await expect(page.locator("or-mwc-snackbar .mdc-snackbar--open")).not.toBeAttached({ timeout: 2000 }); }); From d184d265286040b958e943c4ff77844ad8813a05 Mon Sep 17 00:00:00 2001 From: Koen <44117179+Ekhorn@users.noreply.github.com> Date: Wed, 24 Dec 2025 17:14:49 +0100 Subject: [PATCH 22/28] Use stdout for datapoint exports (#9) * Use stdout for datapoint exports * Remove OR_DATA_POINTS_EXPORT_LIMIT --- .../datapoint/AssetDatapointResourceImpl.java | 63 +++-- .../datapoint/AssetDatapointService.java | 232 ++++++++---------- profile/deploy.yml | 3 - .../assets/AssetDatapointExportTest.groovy | 42 +--- 4 files changed, 131 insertions(+), 209 deletions(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java index a8f619851c..5ae592a65a 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java @@ -46,11 +46,10 @@ import org.openremote.model.http.RequestParams; import org.openremote.model.security.ClientRole; import org.openremote.model.syslog.SyslogCategory; +import org.openremote.model.util.UniqueIdentifierGenerator; import org.openremote.model.value.MetaItemType; -import java.io.File; -import java.io.FileInputStream; -import java.util.concurrent.ScheduledFuture; +import java.io.*; import java.util.logging.Level; import java.util.logging.Logger; import java.util.zip.ZipEntry; @@ -206,47 +205,43 @@ public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefs DATA_EXPORT_LOG.info("User '" + getUsername() + "' started data export for " + attributeRefsString + " from " + fromTimestamp + " to " + toTimestamp + " in format " + format); - ScheduledFuture exportFuture = assetDatapointService.exportDatapoints(attributeRefs, fromTimestamp, toTimestamp, format); + PipedInputStream pipedInputStream = assetDatapointService.exportDatapoints(attributeRefs, fromTimestamp, toTimestamp, format); - asyncResponse.register((ConnectionCallback) disconnected -> exportFuture.cancel(true)); - - File exportFile = null; - - try { - exportFile = exportFuture.get(); - - try (FileInputStream fin = new FileInputStream(exportFile); - ZipOutputStream zipOut = new ZipOutputStream(response.getOutputStream())) { - - ZipEntry zipEntry = new ZipEntry(exportFile.getName()); - zipOut.putNextEntry(zipEntry); - IOUtils.copy(fin, zipOut); - zipOut.closeEntry(); + asyncResponse.register((ConnectionCallback) disconnected -> { + try { + pipedInputStream.close(); + } catch (IOException e) { + DATA_EXPORT_LOG.log(Level.SEVERE, "Could not close input stream: ", e); + throw new RuntimeException(e); } + }); - response.setContentType("application/zip"); - response.setHeader(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"dataexport.zip\""); + try (InputStream fin = pipedInputStream; + ZipOutputStream zipOut = new ZipOutputStream(response.getOutputStream())) { - asyncResponse.resume( - response - ); - } catch (Exception ex) { - exportFuture.cancel(true); + String fileName = UniqueIdentifierGenerator.generateId() + ".csv"; + ZipEntry zipEntry = new ZipEntry(fileName); + zipOut.putNextEntry(zipEntry); + IOUtils.copy(fin, zipOut); + zipOut.closeEntry(); + } catch (IOException ex) { asyncResponse.resume(new WebApplicationException(Response.Status.INTERNAL_SERVER_ERROR)); - DATA_EXPORT_LOG.log(Level.SEVERE, "Exception in ScheduledFuture: ", ex); - } finally { - if (exportFile != null && exportFile.exists()) { - try { - exportFile.delete(); - } catch (Exception e) { - DATA_EXPORT_LOG.log(Level.SEVERE, "Failed to delete temporary export file: " + exportFile.getPath(), e); - } - } + DATA_EXPORT_LOG.log(Level.SEVERE, "Zip exception: ", ex); } + + response.setContentType("application/zip"); + response.setHeader(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"dataexport.zip\""); + + asyncResponse.resume( + response + ); } catch (JsonProcessingException ex) { asyncResponse.resume(new BadRequestException(ex)); } catch (DatapointQueryTooLargeException dqex) { asyncResponse.resume(new WebApplicationException(dqex, Response.Status.REQUEST_ENTITY_TOO_LARGE)); + } catch (IOException ex) { + asyncResponse.resume(new WebApplicationException(Response.Status.INTERNAL_SERVER_ERROR)); + DATA_EXPORT_LOG.log(Level.SEVERE, "Failed to create piped output stream: ", ex); } } } diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index c462eed875..4003bf7ad2 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -1,11 +1,10 @@ package org.openremote.manager.datapoint; +import org.hibernate.Session; import org.openremote.agent.protocol.ProtocolDatapointService; import org.openremote.container.timer.TimerService; import org.openremote.manager.asset.OutdatedAttributeEvent; import org.openremote.model.datapoint.DatapointExportFormat; -import org.openremote.model.datapoint.DatapointQueryTooLargeException; -import org.openremote.model.util.UniqueIdentifierGenerator; import org.openremote.manager.asset.AssetProcessingException; import org.openremote.manager.asset.AssetStorageService; import org.openremote.manager.event.ClientEventService; @@ -21,19 +20,19 @@ import org.openremote.model.query.AssetQuery; import org.openremote.model.query.filter.AttributePredicate; import org.openremote.model.query.filter.NameValuePredicate; +import org.openremote.model.syslog.SyslogCategory; import org.openremote.model.util.Pair; import org.openremote.model.value.MetaHolder; import org.openremote.model.value.MetaItemType; +import org.postgresql.PGConnection; +import org.postgresql.copy.CopyManager; -import java.io.File; -import java.nio.file.Files; -import java.nio.file.Path; +import java.io.*; import java.sql.Date; import java.time.*; import java.util.Arrays; import java.util.List; import java.util.Map; -import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.logging.Level; import java.util.logging.Logger; @@ -44,6 +43,7 @@ import static java.time.temporal.ChronoUnit.DAYS; import static java.util.stream.Collectors.groupingBy; import static java.util.stream.Collectors.toList; +import static org.openremote.model.syslog.SyslogCategory.DATA; import static org.openremote.model.util.MapAccess.getInteger; import static org.openremote.model.value.MetaItemType.STORE_DATA_POINTS; @@ -57,13 +57,10 @@ public class AssetDatapointService extends AbstractDatapointService - file.isFile() - && file.getName().endsWith("csv") - && file.lastModified() < timerService.getCurrentTimeMillis() - oneDayMillis - ); - - if (obsoleteExports != null) { - Arrays.stream(obsoleteExports) - .forEach(file -> { - boolean success = false; - try { - success = file.delete(); - } catch (SecurityException e) { - LOG.log(Level.WARNING, "Cannot access the export file to delete it", e); - } - if (!success) { - LOG.log(Level.WARNING, "Failed to delete obsolete export '" + file.getName() + "'"); - } - }); - } - } catch (Exception e) { - LOG.log(Level.WARNING, "Failed to purge old exports", e); - } } protected String buildWhereClause(List>> attributes, boolean negate) { @@ -253,9 +206,9 @@ protected String buildWhereClause(List>> attributes, b * container so it can be accessed by this process. * Backwards compatible overload with default format. */ - public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, + public PipedInputStream exportDatapoints(AttributeRef[] attributeRefs, long fromTimestamp, - long toTimestamp) { + long toTimestamp) throws IOException { return exportDatapoints(attributeRefs, fromTimestamp, toTimestamp, DatapointExportFormat.CSV); } @@ -263,96 +216,103 @@ public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, * Exports datapoints as CSV using SQL; the export path used in the SQL query must also be mapped into the manager * container so it can be accessed by this process. */ - public ScheduledFuture exportDatapoints(AttributeRef[] attributeRefs, + public PipedInputStream exportDatapoints(AttributeRef[] attributeRefs, long fromTimestamp, long toTimestamp, - DatapointExportFormat format) { - try { - String query = getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp); - - // Verify the query is 'legal' and can be executed - if(canQueryDatapoints(query, null, datapointExportLimit)) { - return doExportDatapoints(attributeRefs, fromTimestamp, toTimestamp, format); - } - throw new RuntimeException("Could not export datapoints."); + DatapointExportFormat format) throws IOException { + String query = getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp); - } catch (DatapointQueryTooLargeException dqex) { - String msg = "Could not export data points. It exceeds the data limit of " + datapointExportLimit + " data points."; - getLogger().log(Level.WARNING, msg, dqex); - throw dqex; + // Verify the query is 'legal' and can be executed + if (canQueryDatapoints(query, null, 0)) { + return doExportDatapoints(attributeRefs, fromTimestamp, toTimestamp, format); } + throw new RuntimeException("Could not export datapoints."); } - protected ScheduledFuture doExportDatapoints(AttributeRef[] attributeRefs, + protected PipedInputStream doExportDatapoints(AttributeRef[] attributeRefs, long fromTimestamp, long toTimestamp, - DatapointExportFormat format) { - - return scheduledExecutorService.schedule(() -> { - String fileName = UniqueIdentifierGenerator.generateId() + ".csv"; - if (format == DatapointExportFormat.CSV_CROSSTAB) { - String attributeFilter = getAttributeFilter(attributeRefs); - StringBuilder sb = new StringBuilder(String.format( - "copy (select * from crosstab( " + - "'select ad.timestamp, a.name || '' \\: '' || ad.attribute_name as header, ad.value " + - "from asset_datapoint ad " + - "join asset a on ad.entity_id = a.id " + - "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) and (%s) " + - "order by ad.timestamp, header', " + - "'select distinct a.name || '' \\: '' || ad.attribute_name as header " + - "from asset_datapoint ad " + - "join asset a on ad.entity_id = a.id " + - "where %s " + - "order by header') " + - "as ct(timestamp timestamp, %s) " + - ") to '/storage/" + EXPORT_STORAGE_DIR_NAME + "/" + fileName + "' delimiter ',' CSV HEADER;", - fromTimestamp / 1000, toTimestamp / 1000, attributeFilter, attributeFilter, getAttributeColumns(attributeRefs) - )); - persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); - } else if (format == DatapointExportFormat.CSV_CROSSTAB_MINUTE) { - String attributeFilter = getAttributeFilter(attributeRefs); - StringBuilder sb = new StringBuilder(String.format( - "copy (select * from crosstab( " + - "'select public.time_bucket(''%s'', ad.timestamp) as bucket_timestamp, " + - "a.name || '' \\: '' || ad.attribute_name as header, " + - "CASE " + - " WHEN jsonb_typeof((array_agg(ad.value))[1]) = ''number'' THEN " + - " round(avg((ad.value#>>''{}'')::numeric) FILTER (WHERE jsonb_typeof(ad.value) = ''number''), 3)::text " + - " ELSE (array_agg(ad.value ORDER BY ad.timestamp DESC) FILTER (WHERE jsonb_typeof(ad.value) != ''number''))[1]#>>''{}''" + - "END as value " + - "from asset_datapoint ad " + - "join asset a on ad.entity_id = a.id " + - "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) and (%s) " + - "group by bucket_timestamp, header " + - "order by bucket_timestamp, header', " + - "'select distinct a.name || '' \\: '' || ad.attribute_name as header " + - "from asset_datapoint ad " + - "join asset a on ad.entity_id = a.id " + - "where %s " + - "order by header') " + - "as ct(timestamp timestamp, %s) " + - ") to '/storage/" + EXPORT_STORAGE_DIR_NAME + "/" + fileName + "' delimiter ',' CSV HEADER;", - "1 minute", fromTimestamp / 1000, toTimestamp / 1000, attributeFilter, attributeFilter, getAttributeColumns(attributeRefs) - )); - - persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); - } - else { - StringBuilder sb = new StringBuilder("copy (") - .append(getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp)) - .append(") to '/storage/") - .append(EXPORT_STORAGE_DIR_NAME) - .append("/") - .append(fileName) - .append("' delimiter ',' CSV HEADER;"); - persistenceService.doTransaction(em -> em.createNativeQuery(sb.toString()).executeUpdate()); - } - + DatapointExportFormat format) throws IOException { + // Increase buffer size (default is only 1 KB) + PipedInputStream in = new PipedInputStream(1024 * 1024 * 4); // 4 MB + PipedOutputStream out = new PipedOutputStream(in); + + StringBuilder sb; + + final String TO_STDOUT_WITH_CSV_FORMAT = ") TO STDOUT WITH (FORMAT CSV, HEADER, DELIMITER ',');"; + + if (format == DatapointExportFormat.CSV_CROSSTAB) { + String attributeFilter = getAttributeFilter(attributeRefs); + sb = new StringBuilder(String.format( + "copy (select * from crosstab( " + + "'select ad.timestamp, a.name || '' \\: '' || ad.attribute_name as header, ad.value " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) and (%s) " + + "order by ad.timestamp, header', " + + "'select distinct a.name || '' \\: '' || ad.attribute_name as header " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "where %s " + + "order by header') " + + "as ct(timestamp timestamp, %s) " + + TO_STDOUT_WITH_CSV_FORMAT, + fromTimestamp / 1000, toTimestamp / 1000, attributeFilter, attributeFilter, getAttributeColumns(attributeRefs) + )); + } else if (format == DatapointExportFormat.CSV_CROSSTAB_MINUTE) { + String attributeFilter = getAttributeFilter(attributeRefs); + sb = new StringBuilder(String.format( + "copy (select * from crosstab( " + + "'select public.time_bucket(''%s'', ad.timestamp) as bucket_timestamp, " + + "a.name || '' \\: '' || ad.attribute_name as header, " + + "CASE " + + " WHEN jsonb_typeof((array_agg(ad.value))[1]) = ''number'' THEN " + + " round(avg((ad.value#>>''{}'')::numeric) FILTER (WHERE jsonb_typeof(ad.value) = ''number''), 3)::text " + + " ELSE (array_agg(ad.value ORDER BY ad.timestamp DESC) FILTER (WHERE jsonb_typeof(ad.value) != ''number''))[1]#>>''{}''" + + "END as value " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) and (%s) " + + "group by bucket_timestamp, header " + + "order by bucket_timestamp, header', " + + "'select distinct a.name || '' \\: '' || ad.attribute_name as header " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "where %s " + + "order by header') " + + "as ct(timestamp timestamp, %s) " + + TO_STDOUT_WITH_CSV_FORMAT, + "1 minute", fromTimestamp / 1000, toTimestamp / 1000, attributeFilter, attributeFilter, getAttributeColumns(attributeRefs) + )); + } else { + sb = new StringBuilder("copy (") + .append(getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp)) + .append(TO_STDOUT_WITH_CSV_FORMAT); + } + scheduledExecutorService.schedule(() -> persistenceService.doTransaction(em -> { + Session session = em.unwrap(Session.class); + session.doWork(connection -> { + PGConnection pgConnection = connection.unwrap(PGConnection.class); + CopyManager copyManager = pgConnection.getCopyAPI(); + try { + copyManager.copyOut(sb.toString(), out); + out.flush(); + out.close(); + } catch (IOException e) { + // Either a database connection- or output stream failure + DATA_EXPORT_LOG.log(Level.SEVERE, "Datapoint export failed", e); + try { + out.close(); + in.close(); + } catch (IOException ignored) { + DATA_EXPORT_LOG.log(Level.SEVERE, "Failed to close piped input stream", e); + } + } + }); + }), 0, TimeUnit.MILLISECONDS); - // The same path must resolve in both the postgresql container and the manager container - return exportPath.resolve(fileName).toFile(); - }, 0, TimeUnit.MILLISECONDS); + return in; } /** diff --git a/profile/deploy.yml b/profile/deploy.yml index 6385fe8e1f..7f74a5714f 100644 --- a/profile/deploy.yml +++ b/profile/deploy.yml @@ -345,9 +345,6 @@ services: # Defaults to 100.000. When set to 0, it disables the limit. # OR_DATA_POINTS_QUERY_LIMIT: 100000 - # Configure the limit of data points that can be exported to CSV. Defaults to 1 million data points. - # OR_DATA_POINTS_EXPORT_LIMIT: 10000000 - # App id for the API of OpenWeather: https://openweathermap.org # OR_OPEN_WEATHER_API_APP_ID diff --git a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy index d7af45a893..224900a925 100644 --- a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy +++ b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy @@ -30,7 +30,6 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer def keycloakTestSetup = container.getService(SetupService.class).getTaskOfType(KeycloakTestSetup.class) def assetStorageService = container.getService(AssetStorageService.class) def assetDatapointService = container.getService(AssetDatapointService.class) - assetDatapointService.datapointExportLimit = 1000 when: "requesting the first light asset in City realm" def asset = assetStorageService.find( @@ -68,17 +67,14 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer ) then: "the default CSV export should return a file" - def csvExport1Future = assetDatapointService.exportDatapoints( + def inputStream1 = assetDatapointService.exportDatapoints( [new AttributeRef(asset.id, attributeName)] as AttributeRef[], dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() ) - assert csvExport1Future != null - def csvExport1 = csvExport1Future.get() - assert csvExport1 != null and: "the CSV export should contain all 5 data points" - def csvExport1Lines = csvExport1.readLines() + def csvExport1Lines = inputStream1.readLines() assert csvExport1Lines.size() == 6 assert csvExport1Lines[1].endsWith("10.0") assert csvExport1Lines[2].endsWith("20.0") @@ -89,18 +85,15 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer /* ------------------------- */ when: "exporting with format CSV_CROSSTAB" - def csvExport2Future = assetDatapointService.exportDatapoints( + def inputStream2 = assetDatapointService.exportDatapoints( [new AttributeRef(asset.id, attributeName)] as AttributeRef[], dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), DatapointExportFormat.CSV_CROSSTAB ) - assert csvExport2Future != null - def csvExport2 = csvExport2Future.get() - assert csvExport2 != null then: "the crosstab CSV export should contain all 5 data points in columnar format" - def csvExport2Lines = csvExport2.readLines() + def csvExport2Lines = inputStream2.readLines() assert csvExport2Lines.size() == 6 // header + 5 data rows assert csvExport2Lines[0].contains("timestamp") assert csvExport2Lines[0].contains(assetName + ": " + attributeName) @@ -108,18 +101,15 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer /* ------------------------- */ when: "exporting with format CSV_CROSSTAB_MINUTE" - def csvExport3Future = assetDatapointService.exportDatapoints( + def inputStream3 = assetDatapointService.exportDatapoints( [new AttributeRef(asset.id, attributeName)] as AttributeRef[], dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), DatapointExportFormat.CSV_CROSSTAB_MINUTE ) - assert csvExport3Future != null - def csvExport3 = csvExport3Future.get() - assert csvExport3 != null then: "the time-bucketed CSV export should contain aggregated data with correct values" - def csvExport3Lines = csvExport3.readLines() + def csvExport3Lines = inputStream3.readLines() assert csvExport3Lines.size() >= 2 // header + at least 1 aggregated row assert csvExport3Lines[0].contains("timestamp") assert csvExport3Lines[0].contains(assetName + ": " + attributeName) @@ -142,25 +132,5 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer assert values.contains("30.000") || values.contains("30") assert values.contains("40.000") || values.contains("40") assert values.contains("50.000") || values.contains("50") - - /* ------------------------- */ - - when: "the limit of data export is lowered to 4" - assetDatapointService.datapointExportLimit = 4 - - and: "we try to export the same amount of data points" - def csvExport4Future = assetDatapointService.exportDatapoints( - [new AttributeRef(asset.id, attributeName)] as AttributeRef[], - dateTime.minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), - dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() - ) - - then: "the CSV export should throw an exception" - thrown(DatapointQueryTooLargeException) - - /* ------------------------- */ - - cleanup: "Remove the limit on datapoint querying" - assetDatapointService.datapointExportLimit = assetDatapointService.OR_DATA_POINTS_EXPORT_LIMIT_DEFAULT; } } From e9c34dfb73858a5a2b14a2f26f6170d17047edc3 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Wed, 14 Jan 2026 14:00:14 +0100 Subject: [PATCH 23/28] revert dataexportlimit removal --- .../datapoint/AssetDatapointService.java | 29 +++++++++++++++---- profile/deploy.yml | 3 ++ .../assets/AssetDatapointExportTest.groovy | 21 ++++++++++++++ 3 files changed, 48 insertions(+), 5 deletions(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index 4003bf7ad2..f3990ebe22 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -5,6 +5,7 @@ import org.openremote.container.timer.TimerService; import org.openremote.manager.asset.OutdatedAttributeEvent; import org.openremote.model.datapoint.DatapointExportFormat; +import org.openremote.model.datapoint.DatapointQueryTooLargeException; import org.openremote.manager.asset.AssetProcessingException; import org.openremote.manager.asset.AssetStorageService; import org.openremote.manager.event.ClientEventService; @@ -57,10 +58,13 @@ public class AssetDatapointService extends AbstractDatapointService Date: Thu, 15 Jan 2026 16:44:54 +0100 Subject: [PATCH 24/28] multiple improvements -fix crosstab exporting attributes with no data, -improve stream error handling -cleanup -add 400 exception when no attributerefs, because this causes the crosstab db query to fail multiple improvements -fix crosstab exporting attributes with no data, -improve stream error handling -cleanup -add 400 exception when no attributerefs, because this causes the crosstab db query to fail --- .../datapoint/AssetDatapointResourceImpl.java | 4 + .../datapoint/AssetDatapointService.java | 198 +++++++++--------- 2 files changed, 108 insertions(+), 94 deletions(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java index 5ae592a65a..21d5787cc6 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointResourceImpl.java @@ -182,6 +182,10 @@ public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefs AttributeRef[] attributeRefs = JSON.readValue(attributeRefsString, AttributeRef[].class); + if (attributeRefs == null || attributeRefs.length == 0) { + throw new WebApplicationException(Response.Status.BAD_REQUEST); + } + for (AttributeRef attributeRef : attributeRefs) { if (isRestrictedUser() && !assetStorageService.isUserAsset(getUserId(), attributeRef.getId())) { throw new WebApplicationException(Response.Status.FORBIDDEN); diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index f3990ebe22..8dbccb4cbf 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -39,7 +39,7 @@ import java.util.logging.Logger; import java.util.stream.Collectors; import java.util.Set; -import java.util.HashSet; +import java.util.TreeSet; import static java.time.temporal.ChronoUnit.DAYS; import static java.util.stream.Collectors.groupingBy; @@ -233,7 +233,7 @@ public PipedInputStream exportDatapoints(AttributeRef[] attributeRefs, long toTimestamp, DatapointExportFormat format) throws IOException { try { - String query = getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp); + String query = getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp, null); // Verify the query is 'legal' and can be executed if(canQueryDatapoints(query, null, datapointExportLimit)) { @@ -256,125 +256,135 @@ protected PipedInputStream doExportDatapoints(AttributeRef[] attributeRefs, PipedInputStream in = new PipedInputStream(1024 * 1024 * 4); // 4 MB PipedOutputStream out = new PipedOutputStream(in); - StringBuilder sb; - final String TO_STDOUT_WITH_CSV_FORMAT = ") TO STDOUT WITH (FORMAT CSV, HEADER, DELIMITER ',');"; - - if (format == DatapointExportFormat.CSV_CROSSTAB) { - String attributeFilter = getAttributeFilter(attributeRefs); - sb = new StringBuilder(String.format( - "copy (select * from crosstab( " + - "'select ad.timestamp, a.name || '' \\: '' || ad.attribute_name as header, ad.value " + - "from asset_datapoint ad " + - "join asset a on ad.entity_id = a.id " + - "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) and (%s) " + - "order by ad.timestamp, header', " + - "'select distinct a.name || '' \\: '' || ad.attribute_name as header " + - "from asset_datapoint ad " + - "join asset a on ad.entity_id = a.id " + - "where %s " + - "order by header') " + - "as ct(timestamp timestamp, %s) " + - TO_STDOUT_WITH_CSV_FORMAT, - fromTimestamp / 1000, toTimestamp / 1000, attributeFilter, attributeFilter, getAttributeColumns(attributeRefs) - )); - } else if (format == DatapointExportFormat.CSV_CROSSTAB_MINUTE) { - String attributeFilter = getAttributeFilter(attributeRefs); - sb = new StringBuilder(String.format( - "copy (select * from crosstab( " + - "'select public.time_bucket(''%s'', ad.timestamp) as bucket_timestamp, " + - "a.name || '' \\: '' || ad.attribute_name as header, " + - "CASE " + - " WHEN jsonb_typeof((array_agg(ad.value))[1]) = ''number'' THEN " + - " round(avg((ad.value#>>''{}'')::numeric) FILTER (WHERE jsonb_typeof(ad.value) = ''number''), 3)::text " + - " ELSE (array_agg(ad.value ORDER BY ad.timestamp DESC) FILTER (WHERE jsonb_typeof(ad.value) != ''number''))[1]#>>''{}''" + - "END as value " + - "from asset_datapoint ad " + - "join asset a on ad.entity_id = a.id " + - "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) and (%s) " + - "group by bucket_timestamp, header " + - "order by bucket_timestamp, header', " + - "'select distinct a.name || '' \\: '' || ad.attribute_name as header " + - "from asset_datapoint ad " + - "join asset a on ad.entity_id = a.id " + - "where %s " + - "order by header') " + - "as ct(timestamp timestamp, %s) " + - TO_STDOUT_WITH_CSV_FORMAT, - "1 minute", fromTimestamp / 1000, toTimestamp / 1000, attributeFilter, attributeFilter, getAttributeColumns(attributeRefs) - )); + String query; + + if (format == DatapointExportFormat.CSV_CROSSTAB || format == DatapointExportFormat.CSV_CROSSTAB_MINUTE) { + Set headers = getAttributeHeaders(attributeRefs); + + // Category query using VALUES clause with dollar-quoting to avoid escaping issues + String categoryValues = headers.stream() + .map(header -> "('" + header.replace("'", "''") + "')") + .collect(Collectors.joining(", ")); + String categoryQuery = "SELECT header FROM (VALUES " + categoryValues + ") AS t(header)"; + + // Column definitions for crosstab result + String attributeColumns = headers.stream() + .map(header -> "\"" + header + "\" text") + .collect(Collectors.joining(", ")); + + query = String.format( + "copy (select * from crosstab('%s', $cat$%s$cat$) as ct(timestamp timestamp, %s)%s", + getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp, format), + categoryQuery, attributeColumns, TO_STDOUT_WITH_CSV_FORMAT); } else { - sb = new StringBuilder("copy (") - .append(getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp)) - .append(TO_STDOUT_WITH_CSV_FORMAT); + query = "copy (" + getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp, format) + TO_STDOUT_WITH_CSV_FORMAT; } - scheduledExecutorService.schedule(() -> persistenceService.doTransaction(em -> { - Session session = em.unwrap(Session.class); - session.doWork(connection -> { - PGConnection pgConnection = connection.unwrap(PGConnection.class); - CopyManager copyManager = pgConnection.getCopyAPI(); + // Execute query asynchronously + scheduledExecutorService.schedule(() -> { + boolean success = false; + try { + persistenceService.doTransaction(em -> { + Session session = em.unwrap(Session.class); + session.doWork(connection -> { + try { + PGConnection pgConnection = connection.unwrap(PGConnection.class); + CopyManager copyManager = pgConnection.getCopyAPI(); + copyManager.copyOut(query, out); + out.flush(); + } catch (IOException e) { + throw new RuntimeException("Failed to export datapoints", e); + } + }); + }); + success = true; + } catch (Exception e) { + DATA_EXPORT_LOG.log(Level.SEVERE, "Datapoint export failed", e); + } finally { try { - copyManager.copyOut(sb.toString(), out); - out.flush(); out.close(); } catch (IOException e) { - // Either a database connection- or output stream failure - DATA_EXPORT_LOG.log(Level.SEVERE, "Datapoint export failed", e); + DATA_EXPORT_LOG.log(Level.WARNING, "Failed to close output stream", e); + } + if (!success) { try { - out.close(); in.close(); } catch (IOException ignored) { DATA_EXPORT_LOG.log(Level.SEVERE, "Failed to close piped input stream", e); } } - }); - }), 0, TimeUnit.MILLISECONDS); + } + }, 0, TimeUnit.MILLISECONDS); return in; } - - /** - * Function for building a WHERE clause filter for the requested attributes. - * Used in crosstab queries to ensure only the requested attributes are included. - * Note: Uses double single quotes for SQL string escaping within crosstab queries. - */ - private String getAttributeFilter(AttributeRef[] attributeRefs) { - return Arrays.stream(attributeRefs) - .map(attr -> String.format("(ad.entity_id = ''%s'' and ad.attribute_name = ''%s'')", - attr.getId(), attr.getName())) - .collect(Collectors.joining(" or ")); - } - + /** * Function for building CSV attribute headers when format will make * a separate column per attribute. + * Uses TreeSet for alphabetical ordering required by crosstab queries. */ - private String getAttributeColumns(AttributeRef[] attributeRefs) { - // Create a set to keep track of unique headers - Set headers = new HashSet<>(); - - // Build unique headers in the format "AssetName: AttributeName" + private Set getAttributeHeaders(AttributeRef[] attributeRefs) { + Set headers = new TreeSet<>(); Arrays.stream(attributeRefs).forEach(attr -> { String assetName = assetStorageService.findNames(attr.getId()).toString().replaceAll("(^\\[|\\]$)", ""); - String attributeName = attr.getName(); - headers.add(assetName + ": " + attributeName); + headers.add(assetName + " : " + attr.getName()); }); - - // Return as a single string with the assembled header. - return headers.stream() - .map(header -> "\"" + header + "\" text") - .collect(Collectors.joining(", ")); + return headers; } /** * Function for building an SQL query that SELECTs the content for the data export. - * Currently, it includes timestamp, asset name, attribute name, and the data point value. - * Be aware: this SQL query does NOT contain any CSV-related statements. + * For crosstab formats, uses escaped quotes ('') as these are embedded in a crosstab string argument. */ - protected String getSelectExportQuery(AttributeRef[] attributeRefs, long fromTimestamp, long toTimestamp) { - return new StringBuilder(String.format("select ad.timestamp, a.name, ad.attribute_name, value from asset_datapoint ad, asset a where ad.entity_id = a.id and ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) and (", fromTimestamp / 1000, toTimestamp / 1000)) - .append(Arrays.stream(attributeRefs).map(attributeRef -> String.format("(ad.entity_id = '%s' and ad.attribute_name = '%s')", attributeRef.getId(), attributeRef.getName())).collect(Collectors.joining(" or "))) - .append(")").toString(); + protected String getSelectExportQuery(AttributeRef[] attributeRefs, long fromTimestamp, long toTimestamp, DatapointExportFormat format) { + long fromSeconds = fromTimestamp / 1000; + long toSeconds = toTimestamp / 1000; + + if (format == null) { + format = DatapointExportFormat.CSV; + } + + if (format == DatapointExportFormat.CSV_CROSSTAB_MINUTE) { + String attributeFilter = Arrays.stream(attributeRefs) + .map(attr -> String.format("(ad.entity_id = ''%s'' and ad.attribute_name = ''%s'')", attr.getId(), attr.getName())) + .collect(Collectors.joining(" or ")); + return String.format( + "select public.time_bucket(''1 minute'', ad.timestamp) as bucket_timestamp, " + + "a.name || '' : '' || ad.attribute_name as header, " + + "CASE " + + " WHEN jsonb_typeof((array_agg(ad.value))[1]) = ''number'' THEN " + + " round(avg((ad.value#>>''{}'')::numeric) FILTER (WHERE jsonb_typeof(ad.value) = ''number''), 3)::text " + + " ELSE (array_agg(ad.value ORDER BY ad.timestamp DESC) FILTER (WHERE jsonb_typeof(ad.value) != ''number''))[1]#>>''{}''" + + "END as value " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) and (%s) " + + "group by bucket_timestamp, header " + + "order by bucket_timestamp, header", + fromSeconds, toSeconds, attributeFilter); + } else if (format == DatapointExportFormat.CSV_CROSSTAB) { + String attributeFilter = Arrays.stream(attributeRefs) + .map(attr -> String.format("(ad.entity_id = ''%s'' and ad.attribute_name = ''%s'')", attr.getId(), attr.getName())) + .collect(Collectors.joining(" or ")); + return String.format( + "select ad.timestamp, a.name || '' : '' || ad.attribute_name as header, ad.value " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) and (%s) " + + "order by ad.timestamp, header", + fromSeconds, toSeconds, attributeFilter); + } else { + // Plain CSV format - no quote escaping needed + String attributeFilter = Arrays.stream(attributeRefs) + .map(attr -> String.format("(ad.entity_id = '%s' and ad.attribute_name = '%s')", attr.getId(), attr.getName())) + .collect(Collectors.joining(" or ")); + return String.format( + "select ad.timestamp, a.name, ad.attribute_name, value " + + "from asset_datapoint ad, asset a " + + "where ad.entity_id = a.id and ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) and (%s)", + fromSeconds, toSeconds, attributeFilter); + } } } From 9173845f5486925b9f96810889e6b7eaf2ba742a Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Thu, 15 Jan 2026 16:57:31 +0100 Subject: [PATCH 25/28] oops --- .../org/openremote/manager/datapoint/AssetDatapointService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index 8dbccb4cbf..013945158d 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -310,7 +310,7 @@ protected PipedInputStream doExportDatapoints(AttributeRef[] attributeRefs, if (!success) { try { in.close(); - } catch (IOException ignored) { + } catch (IOException e) { DATA_EXPORT_LOG.log(Level.SEVERE, "Failed to close piped input stream", e); } } From e81ba728fa55a8258933a9c5dff094319fc13eb2 Mon Sep 17 00:00:00 2001 From: Hackerberg43 Date: Thu, 15 Jan 2026 17:19:04 +0100 Subject: [PATCH 26/28] adjust test to account for header now in sql query format --- .../openremote/test/assets/AssetDatapointExportTest.groovy | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy index b8fe0e7637..25266657ec 100644 --- a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy +++ b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy @@ -97,7 +97,7 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer def csvExport2Lines = inputStream2.readLines() assert csvExport2Lines.size() == 6 // header + 5 data rows assert csvExport2Lines[0].contains("timestamp") - assert csvExport2Lines[0].contains(assetName + ": " + attributeName) + assert csvExport2Lines[0].contains(assetName + " : " + attributeName) /* ------------------------- */ @@ -113,7 +113,7 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer def csvExport3Lines = inputStream3.readLines() assert csvExport3Lines.size() >= 2 // header + at least 1 aggregated row assert csvExport3Lines[0].contains("timestamp") - assert csvExport3Lines[0].contains(assetName + ": " + attributeName) + assert csvExport3Lines[0].contains(assetName + " : " + attributeName) and: "the aggregated values should be correct averages within 1-minute buckets" // Each datapoint is 5 minutes apart, so each should be in its own 1-minute bucket From 93f314ef49a38b5a51c76af01433e501c34f900f Mon Sep 17 00:00:00 2001 From: Wouter Voorberg <119453250+Hackerberg43@users.noreply.github.com> Date: Wed, 18 Feb 2026 11:57:33 +0100 Subject: [PATCH 27/28] Update manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java Co-authored-by: Koen <44117179+Ekhorn@users.noreply.github.com> --- .../org/openremote/manager/datapoint/AssetDatapointService.java | 1 - 1 file changed, 1 deletion(-) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index 013945158d..5d53b0e206 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -62,7 +62,6 @@ public class AssetDatapointService extends AbstractDatapointService Date: Fri, 20 Feb 2026 16:32:03 +0100 Subject: [PATCH 28/28] adapt to sql injection fix --- .../datapoint/AssetDatapointService.java | 184 ++++++++++-------- ...unc.sql => V20260120_01__addTableFunc.sql} | 0 .../assets/AssetDatapointExportTest.groovy | 3 +- 3 files changed, 105 insertions(+), 82 deletions(-) rename manager/src/main/resources/org/openremote/manager/setup/database/{V20251201_01__addTableFunc.sql => V20260120_01__addTableFunc.sql} (100%) diff --git a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java index 6e76c37661..21572721b8 100644 --- a/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java +++ b/manager/src/main/java/org/openremote/manager/datapoint/AssetDatapointService.java @@ -239,7 +239,7 @@ public PipedInputStream exportDatapoints(AttributeRef[] attributeRefs, // Verify the query is 'legal' and can be executed if(canQueryDatapoints(exportQuery.query, exportQuery.parameters, datapointExportLimit)) { - return doExportDatapoints(attributeRefs, fromTimestamp, toTimestamp); + return doExportDatapoints(attributeRefs, fromTimestamp, toTimestamp, format); } throw new RuntimeException("Could not export datapoints."); @@ -258,69 +258,44 @@ protected PipedInputStream doExportDatapoints(AttributeRef[] attributeRefs, PipedInputStream in = new PipedInputStream(1024 * 1024 * 4); // 4 MB PipedOutputStream out = new PipedOutputStream(in); - final String TO_STDOUT_WITH_CSV_FORMAT = ") TO STDOUT WITH (FORMAT CSV, HEADER, DELIMITER ',');"; - String query; - - return scheduledExecutorService.schedule(() -> { - String fileName = UniqueIdentifierGenerator.generateId() + ".csv"; - String tempTableName = "tmp_export_attributes"; - String copySql = buildCopyExportSql(tempTableName, fileName, fromTimestamp, toTimestamp); - - persistenceService.doTransaction(em -> em.unwrap(Session.class).doWork(connection -> { - // It is not possible to use parametrized prepared statements with the copy statement, - // so we're going through a temporary table for safely binding the parameters - try (Statement statement = connection.createStatement()) { - statement.execute( - "create temp table " + tempTableName + " (entity_id text, attribute_name text) on commit drop" - ); - } - - try (PreparedStatement insertStatement = connection.prepareStatement( - "insert into " + tempTableName + " (entity_id, attribute_name) values (?, ?)" - )) { - for (AttributeRef attributeRef : attributeRefs) { - insertStatement.setString(1, validateAssetId(attributeRef.getId())); - insertStatement.setString(2, attributeRef.getName()); - insertStatement.addBatch(); - } - insertStatement.executeBatch(); - } - - try (Statement statement = connection.createStatement()) { - statement.execute(copySql); - } - })); - - // Column definitions for crosstab result - String attributeColumns = headers.stream() - .map(header -> "\"" + header + "\" text") - .collect(Collectors.joining(", ")); - - query = String.format( - "copy (select * from crosstab('%s', $cat$%s$cat$) as ct(timestamp timestamp, %s)%s", - getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp, format), - categoryQuery, attributeColumns, TO_STDOUT_WITH_CSV_FORMAT); - } else { - query = "copy (" + getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp, format) + TO_STDOUT_WITH_CSV_FORMAT; - } - // Execute query asynchronously scheduledExecutorService.schedule(() -> { boolean success = false; try { - persistenceService.doTransaction(em -> { - Session session = em.unwrap(Session.class); - session.doWork(connection -> { - try { - PGConnection pgConnection = connection.unwrap(PGConnection.class); - CopyManager copyManager = pgConnection.getCopyAPI(); - copyManager.copyOut(query, out); - out.flush(); - } catch (IOException e) { - throw new RuntimeException("Failed to export datapoints", e); + persistenceService.doTransaction(em -> em.unwrap(Session.class).doWork(connection -> { + String tempTableName = "tmp_export_attributes"; + + // Create temp table and insert attribute refs safely via PreparedStatement + // to prevent SQL injection (instead of string-interpolating IDs into the query) + try (Statement statement = connection.createStatement()) { + statement.execute( + "create temp table " + tempTableName + " (entity_id text, attribute_name text) on commit drop" + ); + } + + try (PreparedStatement insertStatement = connection.prepareStatement( + "insert into " + tempTableName + " (entity_id, attribute_name) values (?, ?)" + )) { + for (AttributeRef attributeRef : attributeRefs) { + insertStatement.setString(1, validateAssetId(attributeRef.getId())); + insertStatement.setString(2, attributeRef.getName()); + insertStatement.addBatch(); } - }); - }); + insertStatement.executeBatch(); + } + + // Build the COPY ... TO STDOUT query based on format + String copyQuery = buildCopyToStdoutQuery(tempTableName, fromTimestamp, toTimestamp, format, attributeRefs); + + try { + PGConnection pgConnection = connection.unwrap(PGConnection.class); + CopyManager copyManager = pgConnection.getCopyAPI(); + copyManager.copyOut(copyQuery, out); + out.flush(); + } catch (IOException e) { + throw new RuntimeException("Failed to export datapoints", e); + } + })); success = true; } catch (Exception e) { DATA_EXPORT_LOG.log(Level.SEVERE, "Datapoint export failed", e); @@ -343,6 +318,76 @@ protected PipedInputStream doExportDatapoints(AttributeRef[] attributeRefs, return in; } + /** + * Builds a COPY ... TO STDOUT query for streaming CSV data based on the export format. + * Uses a temporary table for safe attribute filtering (SQL injection prevention). + */ + private String buildCopyToStdoutQuery(String tempTableName, long fromTimestamp, long toTimestamp, + DatapointExportFormat format, AttributeRef[] attributeRefs) { + long fromSeconds = fromTimestamp / 1000; + long toSeconds = toTimestamp / 1000; + final String TO_STDOUT_WITH_CSV_FORMAT = ") TO STDOUT WITH (FORMAT CSV, HEADER, DELIMITER ',');"; + + if (format == DatapointExportFormat.CSV_CROSSTAB || format == DatapointExportFormat.CSV_CROSSTAB_MINUTE) { + Set headers = getAttributeHeaders(attributeRefs); + + // Category query using VALUES clause with dollar-quoting to avoid escaping issues + String categoryValues = headers.stream() + .map(header -> "('" + header.replace("'", "''") + "')") + .collect(Collectors.joining(", ")); + String categoryQuery = "SELECT header FROM (VALUES " + categoryValues + ") AS t(header)"; + + // Column definitions for crosstab result + String attributeColumns = headers.stream() + .map(header -> "\"" + header + "\" text") + .collect(Collectors.joining(", ")); + + // Inner SELECT query for crosstab (uses escaped quotes since it's inside a string argument) + String innerQuery; + if (format == DatapointExportFormat.CSV_CROSSTAB_MINUTE) { + innerQuery = String.format( + "select public.time_bucket(''1 minute'', ad.timestamp) as bucket_timestamp, " + + "a.name || '' : '' || ad.attribute_name as header, " + + "CASE " + + " WHEN jsonb_typeof((array_agg(ad.value))[1]) = ''number'' THEN " + + " round(avg((ad.value#>>''{}'')::numeric) FILTER (WHERE jsonb_typeof(ad.value) = ''number''), 3)::text " + + " ELSE (array_agg(ad.value ORDER BY ad.timestamp DESC) FILTER (WHERE jsonb_typeof(ad.value) != ''number''))[1]#>>''{}''" + + "END as value " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "join %s t on ad.entity_id = t.entity_id and ad.attribute_name = t.attribute_name " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) " + + "group by bucket_timestamp, header " + + "order by bucket_timestamp, header", + tempTableName, fromSeconds, toSeconds); + } else { + innerQuery = String.format( + "select ad.timestamp, a.name || '' : '' || ad.attribute_name as header, ad.value " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "join %s t on ad.entity_id = t.entity_id and ad.attribute_name = t.attribute_name " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) " + + "order by ad.timestamp, header", + tempTableName, fromSeconds, toSeconds); + } + + return String.format( + "copy (select * from crosstab('%s', $cat$%s$cat$) as ct(timestamp timestamp, %s)%s", + innerQuery, categoryQuery, attributeColumns, TO_STDOUT_WITH_CSV_FORMAT); + } else { + // Plain CSV format using temp table join + String innerQuery = String.format( + "select ad.timestamp, a.name, ad.attribute_name, value " + + "from asset_datapoint ad " + + "join asset a on ad.entity_id = a.id " + + "join %s t on ad.entity_id = t.entity_id and ad.attribute_name = t.attribute_name " + + "where ad.timestamp >= to_timestamp(%d) and ad.timestamp <= to_timestamp(%d) " + + "order by ad.timestamp desc", + tempTableName, fromSeconds, toSeconds); + return "copy (" + innerQuery + TO_STDOUT_WITH_CSV_FORMAT; + } + } + /** * Function for building CSV attribute headers when format will make * a separate column per attribute. @@ -389,27 +434,6 @@ protected ExportQuery getSelectExportQuery(AttributeRef[] attributeRefs, long fr protected record ExportQuery(String query, Map parameters) { } - private static String buildCopyExportSql(String tempTableName, String fileName, long fromTimestamp, long toTimestamp) { - long fromSeconds = fromTimestamp / 1000; - long toSeconds = toTimestamp / 1000; - - StringBuilder sb = new StringBuilder("copy (") - .append("select ad.timestamp, a.name, ad.attribute_name, value ") - .append("from asset_datapoint ad ") - .append("join asset a on ad.entity_id = a.id ") - .append("join ").append(tempTableName).append(" t on ad.entity_id = t.entity_id and ad.attribute_name = t.attribute_name ") - .append("where ad.timestamp >= to_timestamp(").append(fromSeconds).append(") ") - .append("and ad.timestamp <= to_timestamp(").append(toSeconds).append(") ") - .append("order by ad.timestamp desc") - .append(") to '/storage/") - .append(EXPORT_STORAGE_DIR_NAME) - .append("/") - .append(fileName) - .append("' delimiter ',' CSV HEADER;"); - - return sb.toString(); - } - private static String validateAssetId(String assetId) { if (assetId == null || assetId.length() != 22) { throw new IllegalArgumentException("Invalid asset id"); diff --git a/manager/src/main/resources/org/openremote/manager/setup/database/V20251201_01__addTableFunc.sql b/manager/src/main/resources/org/openremote/manager/setup/database/V20260120_01__addTableFunc.sql similarity index 100% rename from manager/src/main/resources/org/openremote/manager/setup/database/V20251201_01__addTableFunc.sql rename to manager/src/main/resources/org/openremote/manager/setup/database/V20260120_01__addTableFunc.sql diff --git a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy index 1776cc25c1..046b172c8a 100644 --- a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy +++ b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy @@ -187,14 +187,13 @@ class AssetDatapointExportTest extends Specification implements ManagerContainer when: "exporting with a malicious attribute name" def injectedAttributeName = "x')) UNION ((SELECT now()::timestamp, table_name::text, 'injected'::text, NULL::jsonb FROM information_schema.tables WHERE table_schema='public" - def csvExportFuture = assetDatapointService.exportDatapoints( + def csvExport = assetDatapointService.exportDatapoints( [new AttributeRef(asset.id, injectedAttributeName)] as AttributeRef[], LocalDateTime.now().minusMinutes(30).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() ) then: "the CSV export should not contain injected rows" - def csvExport = csvExportFuture.get() assert csvExport != null def csvExportLines = csvExport.readLines() assert csvExportLines.size() == 1