Use stdout for datapoint exports

Ekhorn · Ekhorn · commit 4e2addf5177f · 2025-12-24T13:21:24.000+01:00
diff --git 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,19 +205,23 @@ 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<File> 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;
+            asyncResponse.register((ConnectionCallback) disconnected -> {
+                try {
+                    pipedInputStream.close();
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
+                }
+            });
 
             try {
-                exportFile = exportFuture.get();
 
-                try (FileInputStream fin = new FileInputStream(exportFile);
+                try (InputStream fin = pipedInputStream;
                      ZipOutputStream zipOut = new ZipOutputStream(response.getOutputStream())) {
 
-                    ZipEntry zipEntry = new ZipEntry(exportFile.getName());
+                    String fileName = UniqueIdentifierGenerator.generateId() + ".csv";
+                    ZipEntry zipEntry = new ZipEntry(fileName);
                     zipOut.putNextEntry(zipEntry);
                     IOUtils.copy(fin, zipOut);
                     zipOut.closeEntry();
@@ -231,22 +234,16 @@ public void getDatapointExport(AsyncResponse asyncResponse, String attributeRefs
                     response
                 );
             } catch (Exception ex) {
-                exportFuture.cancel(true);
+                pipedInputStream.close();
                 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);
-                    }
-                }
             }
         } catch (JsonProcessingException ex) {
             asyncResponse.resume(new BadRequestException(ex));
         } catch (DatapointQueryTooLargeException dqex) {
             asyncResponse.resume(new WebApplicationException(dqex, Response.Status.REQUEST_ENTITY_TOO_LARGE));
+        } catch (IOException e) {
+            asyncResponse.resume(new WebApplicationException(Response.Status.INTERNAL_SERVER_ERROR));
         }
     }
 }
diff --git 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,11 @@
 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;
@@ -24,16 +24,17 @@
 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.io.*;
 import java.nio.file.Files;
 import java.nio.file.Path;
 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;
@@ -253,25 +254,25 @@ protected String buildWhereClause(List<Pair<String, Attribute<?>>> attributes, b
      * container so it can be accessed by this process.
      * Backwards compatible overload with default format.
      */
-    public ScheduledFuture<File> exportDatapoints(AttributeRef[] attributeRefs,
+    public PipedInputStream exportDatapoints(AttributeRef[] attributeRefs,
                                                   long fromTimestamp,
-                                                  long toTimestamp) {
+                                                  long toTimestamp) throws IOException {
         return exportDatapoints(attributeRefs, fromTimestamp, toTimestamp, DatapointExportFormat.CSV);
     }
 
     /**
      * 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<File> exportDatapoints(AttributeRef[] attributeRefs,
+    public PipedInputStream exportDatapoints(AttributeRef[] attributeRefs,
                                                   long fromTimestamp,
                                                   long toTimestamp,
-                                                  DatapointExportFormat format) {
+                                                  DatapointExportFormat format) throws IOException {
         try {
             String query = getSelectExportQuery(attributeRefs, fromTimestamp, toTimestamp);
 
             // Verify the query is 'legal' and can be executed
-            if(canQueryDatapoints(query, null, datapointExportLimit)) {
+            if (canQueryDatapoints(query, null, datapointExportLimit)) {
                 return doExportDatapoints(attributeRefs, fromTimestamp, toTimestamp, format);
             }
             throw new RuntimeException("Could not export datapoints.");
@@ -283,76 +284,89 @@ public ScheduledFuture<File> exportDatapoints(AttributeRef[] attributeRefs,
         }
     }
 
-    protected ScheduledFuture<File> 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 database connection or output stream failure
+                    getLogger().log(Level.WARNING, "Datapoint export failed", e);
+                    try {
+                        in.close();
+                    } catch (IOException ignored) {
+                        getLogger().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/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointExportTest.groovy
