feat(#3448): Add excel export feature

streampipes-data-explorer-export/pom.xml

@@ -29,13 +29,50 @@
         This module contains all components and functionalities related to exporting data from the data explorer storage.
     </description>
 
+    <properties>
+        <apache-poi.version>5.4.0</apache-poi.version>
+    </properties>
+
     <dependencies>
         <!-- StreamPipes dependencies -->
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-model</artifactId>
             <version>0.98.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-storage-management</artifactId>
+            <version>0.98.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-pipeline-management</artifactId>
+            <version>0.98.0-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.poi</groupId>
+            <artifactId>poi</artifactId>
+            <version>${apache-poi.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.logging.log4j</groupId>
+                    <artifactId>log4j-api</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.poi</groupId>
+            <artifactId>poi-ooxml</artifactId>
+            <version>${apache-poi.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.logging.log4j</groupId>
+                    <artifactId>log4j-api</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
 
         <!-- Test dependencies -->
         <dependency>
@@ -44,11 +81,4 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
-
-    <properties>
-        <maven.compiler.source>17</maven.compiler.source>
-        <maven.compiler.target>17</maven.compiler.target>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    </properties>
-
-</project>
+</project>

streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredCsvOutputWriter.java

@@ -19,6 +19,7 @@
 package org.apache.streampipes.dataexplorer.export;
 
 import org.apache.streampipes.dataexplorer.export.item.CsvItemGenerator;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
 import org.apache.streampipes.model.datalake.param.SupportedRestQueryParams;
 
@@ -35,10 +36,18 @@ public class ConfiguredCsvOutputWriter extends ConfiguredOutputWriter {
 
   private CsvItemGenerator itemGenerator;
   private String delimiter = COMMA;
+  private DataLakeMeasure schema;
+  private String headerColumnNameStrategy;
 
   @Override
-  public void configure(ProvidedRestQueryParams params,
+  public void configure(DataLakeMeasure schema,
+                        ProvidedRestQueryParams params,
                         boolean ignoreMissingValues) {
+    this.schema = schema;
+    this.headerColumnNameStrategy = params
+        .getProvidedParams()
+        .getOrDefault(SupportedRestQueryParams.QP_HEADER_COLUMN_NAME, "key");
+
     if (params.has(SupportedRestQueryParams.QP_CSV_DELIMITER)) {
       delimiter = params.getAsString(SupportedRestQueryParams.QP_CSV_DELIMITER).equals("comma") ? COMMA : SEMICOLON;
     }
@@ -69,7 +78,7 @@ public void writeItem(OutputStream outputStream,
 
   private String makeHeaderLine(List<String> columns) {
     StringJoiner joiner = new StringJoiner(this.delimiter);
-    columns.forEach(joiner::add);
+    columns.forEach(c -> joiner.add(getHeaderName(schema, c, headerColumnNameStrategy)));
     return joiner + LINE_SEPARATOR;
   }
 }

streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredExcelOutputWriter.java

@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.dataexplorer.export;
+
+import org.apache.streampipes.manager.file.FileManager;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
+import org.apache.streampipes.model.datalake.param.SupportedRestQueryParams;
+import org.apache.streampipes.model.file.FileMetadata;
+import org.apache.streampipes.storage.api.CRUDStorage;
+
+import org.apache.poi.ss.usermodel.Sheet;
+import org.apache.poi.xssf.streaming.SXSSFWorkbook;
+import org.apache.poi.xssf.usermodel.XSSFWorkbook;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.util.List;
+import java.util.Objects;
+
+public class ConfiguredExcelOutputWriter extends ConfiguredOutputWriter {
+
+  private final CRUDStorage<FileMetadata> storage;
+
+  private SXSSFWorkbook wb;
+  private Sheet ws;
+  private boolean useTemplate = false;
+  private int startRow = 0;
+  private String templateId;
+  private DataLakeMeasure schema;
+  private String headerColumnNameStrategy;
+
+  public ConfiguredExcelOutputWriter(CRUDStorage<FileMetadata> fileMetadataStorage) {
+    this.storage = fileMetadataStorage;
+  }
+
+  @Override
+  public void configure(DataLakeMeasure schema,
+                        ProvidedRestQueryParams params,
+                        boolean ignoreMissingValues) {
+    var qp = params.getProvidedParams();
+    this.schema = schema;
+    this.headerColumnNameStrategy = qp.getOrDefault(SupportedRestQueryParams.QP_HEADER_COLUMN_NAME, "key");
+    if (qp.containsKey(SupportedRestQueryParams.QP_XLSX_USE_TEMPLATE)) {
+      this.useTemplate = true;
+      this.startRow = Integer.parseInt(qp.getOrDefault(SupportedRestQueryParams.QP_XLSX_START_ROW, "0"));
+      this.templateId = qp.getOrDefault(SupportedRestQueryParams.QP_XLSX_TEMPLATE_ID, null);
+    }
+  }
+
+  @Override
+  public void beforeFirstItem(OutputStream outputStream) {
+    if (useTemplate && Objects.nonNull(templateId)) {
+      var fileMetadata = storage.getElementById(templateId);
+      if (fileMetadata != null) {
+        var path = new FileManager().getFile(fileMetadata.getFilename()).getAbsoluteFile().toPath();
+        try (InputStream is = Files.newInputStream(path)) {
+          XSSFWorkbook templateWorkbook = new XSSFWorkbook(is);
+          wb = new SXSSFWorkbook(templateWorkbook);
+          ws = wb.getSheetAt(0);
+        } catch (IOException e) {
+          createNewWorkbook();
+        }
+      }
+    } else {
+      createNewWorkbook();
+    }
+  }
+
+  private void createNewWorkbook() {
+    wb = new SXSSFWorkbook();
+    ws = wb.createSheet();
+  }
+
+  @Override
+  public void afterLastItem(OutputStream outputStream) throws IOException {
+    wb.write(outputStream);
+    wb.close();
+  }
+
+  @Override
+  public void writeItem(OutputStream outputStream,
+                        List<Object> row,
+                        List<String> columnNames, boolean firstObject) throws IOException {
+    var excelRow = ws.createRow(startRow);
+    int columnIndex = 0;
+    if (firstObject) {
+      for (String column : columnNames) {
+        var cell = excelRow.createCell(columnIndex);
+        var headerName = getHeaderName(schema, String.valueOf(column), headerColumnNameStrategy);
+        cell.setCellValue(headerName);
+        columnIndex++;
+      }
+      startRow++;
+    }
+    for (Object column : row) {
+      var cell = excelRow.createCell(columnIndex);
+      String entry = ExportUtils.formatValue(column);
+      cell.setCellValue(entry);
+      columnIndex++;
+    }
+    startRow++;
+  }
+}

streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredJsonOutputWriter.java

@@ -19,8 +19,9 @@
 package org.apache.streampipes.dataexplorer.export;
 
 import org.apache.streampipes.dataexplorer.export.item.ItemGenerator;
-import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
 import org.apache.streampipes.dataexplorer.export.item.JsonItemGenerator;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -40,7 +41,8 @@ public ConfiguredJsonOutputWriter() {
   }
 
   @Override
-  public void configure(ProvidedRestQueryParams params,
+  public void configure(DataLakeMeasure schema,
+                        ProvidedRestQueryParams params,
                         boolean ignoreMissingValues) {
     // do nothing
   }

streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredOutputWriter.java

@@ -18,24 +18,58 @@
 
 package org.apache.streampipes.dataexplorer.export;
 
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
+import org.apache.streampipes.model.schema.EventProperty;
 
 import java.io.IOException;
 import java.io.OutputStream;
+import java.text.DecimalFormat;
 import java.util.List;
+import java.util.Objects;
 
 public abstract class ConfiguredOutputWriter {
 
-  public static ConfiguredOutputWriter getConfiguredWriter(OutputFormat format,
+  private final DecimalFormat df = new DecimalFormat("#");
+
+  public static ConfiguredOutputWriter getConfiguredWriter(DataLakeMeasure schema,
+                                                           OutputFormat format,
                                                            ProvidedRestQueryParams params,
                                                            boolean ignoreMissingValues) {
     var writer = format.getWriter();
-    writer.configure(params, ignoreMissingValues);
+    writer.configure(schema, params, ignoreMissingValues);
 
     return writer;
   }
 
-  public abstract void configure(ProvidedRestQueryParams params,
+  protected String getHeaderName(DataLakeMeasure schema,
+                                 String runtimeName,
+                                 String headerColumnNameStrategy) {
+    if (Objects.nonNull(schema) && headerColumnNameStrategy.equals("label")) {
+      return schema
+          .getEventSchema()
+          .getEventProperties()
+          .stream()
+          .filter(ep -> ep.getRuntimeName().equals(runtimeName))
+          .findFirst()
+          .map(ep -> extractLabel(ep, runtimeName))
+          .orElse(runtimeName);
+    } else {
+      return runtimeName;
+    }
+  }
+
+  private String extractLabel(EventProperty ep,
+                              String runtimeName) {
+    if (Objects.nonNull(ep.getLabel())) {
+      return ep.getLabel();
+    } else {
+      return runtimeName;
+    }
+  }
+
+  public abstract void configure(DataLakeMeasure schema,
+                                 ProvidedRestQueryParams params,
                                  boolean ignoreMissingValues);
 
   public abstract void beforeFirstItem(OutputStream outputStream) throws IOException;

streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ExportUtils.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.dataexplorer.export;
+
+import java.math.BigDecimal;
+import java.text.DecimalFormat;
+
+public class ExportUtils {
+
+  private static final DecimalFormat df = new DecimalFormat("#");
+
+  public static String formatValue(Object value) {
+    if (value instanceof Double) {
+      return BigDecimal.valueOf((Double) value).toPlainString();
+    } else {
+      return String.valueOf(value);
+    }
+  }
+}

streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/OutputFormat.java

@@ -18,11 +18,15 @@
 
 package org.apache.streampipes.dataexplorer.export;
 
+import org.apache.streampipes.storage.management.StorageDispatcher;
+
+import java.util.Arrays;
 import java.util.function.Supplier;
 
 public enum OutputFormat {
   JSON(ConfiguredJsonOutputWriter::new),
-  CSV(ConfiguredCsvOutputWriter::new);
+  CSV(ConfiguredCsvOutputWriter::new),
+  XLSX(() -> new ConfiguredExcelOutputWriter(StorageDispatcher.INSTANCE.getNoSqlStore().getFileMetadataStorage()));
 
   private final Supplier<ConfiguredOutputWriter> writerSupplier;
 
@@ -33,4 +37,12 @@ public enum OutputFormat {
   public ConfiguredOutputWriter getWriter() {
     return writerSupplier.get();
   }
+
+  public static OutputFormat fromString(String desiredFormat) {
+    return Arrays.stream(
+            OutputFormat.values())
+        .filter(format -> format.name().equalsIgnoreCase(desiredFormat))
+        .findFirst()
+        .orElseThrow(() -> new IllegalArgumentException(String.format("Could not find format %s", desiredFormat)));
+  }
 }

streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/item/CsvItemGenerator.java

@@ -19,6 +19,8 @@
 
 package org.apache.streampipes.dataexplorer.export.item;
 
+import org.apache.streampipes.dataexplorer.export.ExportUtils;
+
 public class CsvItemGenerator extends ItemGenerator {
 
   public CsvItemGenerator(String delimiter) {
@@ -27,7 +29,7 @@ public CsvItemGenerator(String delimiter) {
 
   @Override
   protected String makeItemString(String key, Object value) {
-    return value != null ? value.toString() : "";
+    return value != null ? ExportUtils.formatValue(value) : "";
   }
 
   @Override