Repository: apache/streampipes
Pull request #3056 (2024-07-19 20:59:43)

fix(#3055): Fix widget rendering in grouping mode
<!--
  ~ 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.
  ~
  -->

  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/community/get-involved/
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar.
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->


See #3055 

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): no

PR introduces (a) deprecation(s): no



Patch:
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/WhereClauseParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/WhereClauseParams.java
index 6e1f633622..3381983268 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/WhereClauseParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/WhereClauseParams.java
@@ -98,7 +98,7 @@ public class WhereClauseParams implements IQueryStatement {
   }
 
   private Object returnCondition(String inputCondition) {
-    if (NumberUtils.isCreatable(inputCondition)) {
+    if (NumberUtils.isParsable(inputCondition)) {
       return Double.parseDouble(inputCondition);
     } else if (isBoolean(inputCondition)) {
       return Boolean.parseBoolean(inputCondition);
diff --git a/streampipes-extensions/streampipes-processors-enricher-jvm/pom.xml b/streampipes-extensions/streampipes-processors-enricher-jvm/pom.xml
index 696ce89520..bc17948896 100644
--- a/streampipes-extensions/streampipes-processors-enricher-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-processors-enricher-jvm/pom.xml
@@ -63,6 +63,12 @@
             <artifactId>junit-jupiter-params</artifactId>
             <scope>test</scope>
         </dependency>
+      <dependency>
+        <groupId>org.apache.streampipes</groupId>
+        <artifactId>streampipes-test-utils-executors</artifactId>
+        <version>0.97.0-SNAPSHOT</version>
+        <scope>test</scope>
+      </dependency>
     </dependencies>
 
     <build>
diff --git a/streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/sizemeasure/SizeMeasureProcessor.java b/streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/sizemeasure/SizeMeasureProcessor.java
index deea211707..c9d2cdc15f 100644
--- a/streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/sizemeasure/SizeMeasureProcessor.java
+++ b/streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/sizemeasure/SizeMeasureProcessor.java
@@ -48,6 +48,10 @@ public class SizeMeasureProcessor extends StreamPipesDataProcessor {
   static final String KILOBYTE_SIZE = "KILOBYTE";
   static final String MEGABYTE_SIZE = "MEGABYTE";
 
+  static final String BYTES_OPTION = "Bytes";
+  static final String KILO_BYTES_OPTION = "Kilobytes (1024 Bytes)";
+  static final String MEGA_BYTES_OPTION = "Megabytes (1024 Kilobytes)";
+
   static final String EVENT_SIZE = "eventSize";
 
   private String sizeUnit;
@@ -55,29 +59,37 @@ public class SizeMeasureProcessor extends StreamPipesDataProcessor {
   @Override
   public DataProcessorDescription declareModel() {
     return ProcessingElementBuilder.create("org.apache.streampipes.processors.enricher.jvm.sizemeasure")
-        .category(DataProcessorType.STRUCTURE_ANALYTICS)
-        .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
-        .withLocales(Locales.EN)
-        .requiredStream(StreamRequirementsBuilder
-            .create()
-            .requiredProperty(EpRequirements.anyProperty())
-            .build())
-        .requiredSingleValueSelection(Labels.withId(SIZE_UNIT),
-            Options.from(new Tuple2<>("Bytes", BYTE_SIZE),
-                new Tuple2<>("Kilobytes (1024 Bytes)", KILOBYTE_SIZE),
-                new Tuple2<>("Megabytes (1024 Kilobytes)", MEGABYTE_SIZE)))
-        .outputStrategy(OutputStrategies.append(EpProperties.doubleEp(
-            Labels.withId(EVENT_SIZE),
-            EVENT_SIZE,
-            "http://schema.org/contentSize")))
-        .build();
+                                   .category(DataProcessorType.STRUCTURE_ANALYTICS)
+                                   .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
+                                   .withLocales(Locales.EN)
+                                   .requiredStream(StreamRequirementsBuilder
+                                                       .create()
+                                                       .requiredProperty(EpRequirements.anyProperty())
+                                                       .build())
+                                   .requiredSingleValueSelection(
+                                       Labels.withId(SIZE_UNIT),
+                                       Options.from(
+                                           new Tuple2<>(BYTES_OPTION, BYTE_SIZE),
+                                           new Tuple2<>(KILO_BYTES_OPTION, KILOBYTE_SIZE),
+                                           new Tuple2<>(MEGA_BYTES_OPTION, MEGABYTE_SIZE)
+                                       )
+                                   )
+                                   .outputStrategy(OutputStrategies.append(EpProperties.doubleEp(
+                                       Labels.withId(EVENT_SIZE),
+                                       EVENT_SIZE,
+                                       "http://schema.org/contentSize"
+                                   )))
+                                   .build();
   }
 
   @Override
-  public void onInvocation(ProcessorParams parameters,
-                           SpOutputCollector spOutputCollector,
-                           EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
-    this.sizeUnit = parameters.extractor().selectedSingleValueInternalName(SIZE_UNIT, String.class);
+  public void onInvocation(
+      ProcessorParams parameters,
+      SpOutputCollector spOutputCollector,
+      EventProcessorRuntimeContext runtimeContext
+  ) throws SpRuntimeException {
+    this.sizeUnit = parameters.extractor()
+                              .selectedSingleValueInternalName(SIZE_UNIT, String.class);
   }
 
   @Override
diff --git a/ui/src/app/data-explorer/components/widgets/table/table-widget.component.html b/ui/src/app/data-explorer/components/widgets/table/table-widget.component.html
index 0b6e90b72c..8588640560 100644
--- a/ui/src/app/data-explorer/components/widgets/table/table-widget.component.html
+++ b/ui/src/app/data-explorer/components/widgets/table/table-widget.component.html
@@ -49,7 +49,7 @@
             (matSortChange)="sortData($event)"
             class="table-widget"
         >
-            <ng-container [cdkColumnDef]="'time'">
+            <ng-container [matColumnDef]="'time'">
                 <div>
                     <th
                         mat-header-cell
@@ -81,7 +81,7 @@
                     let element of dataExplorerWidget.visualizationConfig
                         .selectedColumns
                 "
-                [cdkColumnDef]="element.fullDbName"
+                [matColumnDef]="element.fullDbName"
             >
                 <div *ngIf="!(element.fullDbName === 'time')">
                     <th
@@ -112,6 +112,32 @@
                     </td>
                 </div>
             </ng-container>
+            <ng-container
+                *ngFor="let element of groupByColumnNames"
+                [matColumnDef]="element"
+            >
+                <th
+                    mat-header-cell
+                    mat-sort-header
+                    *matHeaderCellDef
+                    [ngStyle]="{
+                        background:
+                            dataExplorerWidget.baseAppearanceConfig
+                                .backgroundColor,
+                        color: dataExplorerWidget.baseAppearanceConfig.textColor
+                    }"
+                >
+                    <label class="column-header">{{ element }}</label>
+                </th>
+                <td
+                    mat-cell
+                    *matCellDef="let row"
+                    [attr.data-cy]="'data-explorer-table-row-' + element"
+                    style="text-align: left"
+                >
+                    {{ row[element] }}
+                </td>
+            </ng-container>
 
             <tr
                 mat-header-row
diff --git a/ui/src/app/data-explorer/components/widgets/table/table-widget.component.ts b/ui/src/app/data-explorer/components/widgets/table/table-widget.component.ts
index 8cc42a7154..a156b11b14 100644
--- a/ui/src/app/data-explorer/components/widgets/table/table-widget.component.ts
+++ b/ui/src/app/data-explorer/components/widgets/table/table-widget.component.ts
@@ -41,31 +41,56 @@ export class TableWidgetComponent
 
     dataSource = new MatTableDataSource();
     columnNames: string[];
+    groupByColumnNames: string[];
 
     ngOnInit(): void {
         super.ngOnInit();
         this.dataSource.sort = this.sort;
         this.dataSource.paginator = this.paginator;
+        this.regenerateColumnNames();
+    }
+
+    regenerateColumnNames(): void {
+        this.groupByColumnNames = this.makeGroupByColumns();
         this.columnNames = ['time'].concat(
             this.dataExplorerWidget.visualizationConfig.selectedColumns.map(
                 c => c.fullDbName,
             ),
+            ...this.groupByColumnNames,
         );
     }
 
+    makeGroupByColumns(): string[] {
+        return this.dataExplorerWidget.dataConfig.sourceConfigs.flatMap(sc => {
+            return sc.queryConfig.groupBy
+                .filter(g => g.selected)
+                .map(g => g.runtimeName);
+        });
+    }
+
     transformData(spQueryResult: SpQueryResult) {
         return spQueryResult.allDataSeries.flatMap(series =>
             series.rows.map(row =>
-                this.createTableObject(spQueryResult.headers, row),
+                this.createTableObject(spQueryResult.headers, row, series.tags),
             ),
         );
     }
 
-    createTableObject(keys, values) {
-        return keys.reduce((object, key, index) => {
+    createTableObject(
+        keys: string[],
+        values: any[],
+        tags: Record<string, string>,
+    ) {
+        const row = keys.reduce((object, key, index) => {
             object[key] = values[index];
             return object;
         }, {});
+        if (tags !== null) {
+            Object.keys(tags).forEach(key => {
+                row[key] = tags[key];
+            });
+        }
+        return row;
     }
 
     ngOnDestroy(): void {
@@ -116,11 +141,7 @@ export class TableWidgetComponent
     }
 
     onDataReceived(spQueryResults: SpQueryResult[]) {
-        this.columnNames = ['time'].concat(
-            this.dataExplorerWidget.visualizationConfig.selectedColumns.map(
-                c => c.fullDbName,
-            ),
-        );
+        this.regenerateColumnNames();
         const transformedData = spQueryResults
             .map(spQueryResult => this.transformData(spQueryResult))
             .flat();
@@ -148,10 +169,6 @@ export class TableWidgetComponent
     refreshColumns(): void {
         this.dataSource.filter =
             this.dataExplorerWidget.visualizationConfig.searchValue;
-        this.columnNames = ['time'].concat(
-            this.dataExplorerWidget.visualizationConfig.selectedColumns.map(
-                c => c.fullDbName,
-            ),
-        );
+        this.regenerateColumnNames();
     }
 }
diff --git a/ui/src/app/data-explorer/echarts-renderer/echarts-dataset-generator.service.ts b/ui/src/app/data-explorer/echarts-renderer/echarts-dataset-generator.service.ts
index 22debf01ee..ba64bb6a1b 100644
--- a/ui/src/app/data-explorer/echarts-renderer/echarts-dataset-generator.service.ts
+++ b/ui/src/app/data-explorer/echarts-renderer/echarts-dataset-generator.service.ts
@@ -19,6 +19,7 @@
 import { Injectable } from '@angular/core';
 import {
     DataExplorerWidgetModel,
+    DataSeries,
     SpQueryResult,
 } from '@streampipes/platform-services';
 import {
@@ -82,9 +83,15 @@ export class EchartsDatasetGeneratorService<T extends DataExplorerWidgetModel> {
             index: number,
         ) => DataTransformOption[],
     ): PreparedDataset {
-        const data = [
+        const headers = this.makeHeaders(
             queryResult.headers,
-            ...queryResult.allDataSeries.flatMap(series => series.rows),
+            queryResult.allDataSeries,
+        );
+        const data = [
+            headers,
+            ...queryResult.allDataSeries.flatMap(series => {
+                return this.makeRows(series.rows, series.tags);
+            }),
         ];
         const initialTransforms = getInitialTransforms(widgetConfig, index).map(
             transform => ({
@@ -112,7 +119,7 @@ export class EchartsDatasetGeneratorService<T extends DataExplorerWidgetModel> {
             sourceIndex: index,
             tagValues: tags,
             groupedDatasets: groupTransforms,
-            rawDataset: { source: data, dimensions: queryResult.headers },
+            rawDataset: { source: data, dimensions: headers },
             initialTransformDatasets: initialTransforms,
             meta: {
                 preparedDataStartIndex,
@@ -121,6 +128,27 @@ export class EchartsDatasetGeneratorService<T extends DataExplorerWidgetModel> {
         };
     }
 
+    private makeRows(rows: any[][], tags: Record<string, any>): any[][] {
+        const tagValues = tags !== null ? Object.values(tags) : [];
+        return rows.map(row => {
+            return [...row, ...tagValues];
+        });
+    }
+
+    private makeHeaders(
+        headers: string[],
+        allDataSeries: DataSeries[],
+    ): string[] {
+        let tagHeaders = [];
+        if (allDataSeries.length > 0) {
+            const tags = allDataSeries[0].tags;
+            if (tags !== null) {
+                tagHeaders = Object.keys(tags);
+            }
+        }
+        return [...headers, ...tagHeaders];
+    }
+
     increaseDatasetIndex(
         preparedDataset: PreparedDataset,
         currentDatasetIndex: number,


Test patch:
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java
index 52b5552d07..4a9a69dc55 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java
+++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java
@@ -50,6 +50,14 @@ public class WhereStatementParamsTest {
     assertWhereCondition(result, expected);
   }
 
+  @Test
+  public void filterString2() {
+    WhereClauseParams result = WhereClauseParams.from("[fieldName;=;3312476503F]");
+    FilterCondition expected = new FilterCondition("fieldName", "=", "3312476503F");
+
+    assertWhereCondition(result, expected);
+  }
+
   private void assertWhereCondition(WhereClauseParams result, FilterCondition expected) {
     assertEquals(1, result.getWhereConditions().size());
     FilterCondition resultingFilterCondition = result.getWhereConditions().get(0);
diff --git a/streampipes-extensions/streampipes-processors-enricher-jvm/src/test/java/org/apache/streampipes/processors/enricher/jvm/processor/sizemeasure/TestSizeMeasureProcessor.java b/streampipes-extensions/streampipes-processors-enricher-jvm/src/test/java/org/apache/streampipes/processors/enricher/jvm/processor/sizemeasure/TestSizeMeasureProcessor.java
index eb168a7e56..7f2c182d21 100644
--- a/streampipes-extensions/streampipes-processors-enricher-jvm/src/test/java/org/apache/streampipes/processors/enricher/jvm/processor/sizemeasure/TestSizeMeasureProcessor.java
+++ b/streampipes-extensions/streampipes-processors-enricher-jvm/src/test/java/org/apache/streampipes/processors/enricher/jvm/processor/sizemeasure/TestSizeMeasureProcessor.java
@@ -19,99 +19,75 @@
 package org.apache.streampipes.processors.enricher.jvm.processor.sizemeasure;
 
 
-//@RunWith(Parameterized.class)
+import org.apache.streampipes.test.executors.PrefixStrategy;
+import org.apache.streampipes.test.executors.ProcessingElementTestExecutor;
+import org.apache.streampipes.test.executors.TestConfiguration;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.apache.streampipes.processors.enricher.jvm.processor.sizemeasure.SizeMeasureProcessor.BYTES_OPTION;
+import static org.apache.streampipes.processors.enricher.jvm.processor.sizemeasure.SizeMeasureProcessor.EVENT_SIZE;
+import static org.apache.streampipes.processors.enricher.jvm.processor.sizemeasure.SizeMeasureProcessor.KILO_BYTES_OPTION;
+import static org.apache.streampipes.processors.enricher.jvm.processor.sizemeasure.SizeMeasureProcessor.SIZE_UNIT;
+
 public class TestSizeMeasureProcessor {
-//
-//  private static final Logger LOG = LoggerFactory.getLogger(TestSizeMeasureProcessor.class);
-//
-//  @org.junit.runners.Parameterized.Parameters
-//  public static Iterable<Object[]> data() {
-//
-//    /*
-//    249 bytes is the base size of an object when first tested.
-//    Allowing up to 250 bytes comparison error in case of inconsistent base sizes across platforms
-//     */
-//    return Arrays.asList(new Object[][]{
-//            { BYTE_SIZE, 10240 - 249, 10240.0, 250.0},
-//            { KILOBYTE_SIZE, 10240 - 249, 10.0, 0.025},
-//            { MEGABYTE_SIZE, (1024 * 1024) - 249, 1.0, 0.0025}
-//    });
-//  }
-//
-//  @Parameterized.Parameter
-//  public String sizeUnit;
-//
-//  @Parameterized.Parameter(1)
-//  public int numOfBytes;
-//
-//  @Parameterized.Parameter(2)
-//  public double expectedSize;
-//
-//  @Parameterized.Parameter(3)
-//  public double allowableError;
-//
-//  @Test
-//  public void testSizeMeasureProcessor() {
-//    SizeMeasureProcessor processor = new SizeMeasureProcessor();
-//    DataProcessorDescription originalGraph = processor.declareModel();
-//    originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
-//
-//    DataProcessorInvocation graph =
-//            InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
-//
-//
-//    ProcessorParams params = new ProcessorParams(graph);
-//    params.extractor().getStaticPropertyByName(SIZE_UNIT, OneOfStaticProperty.class).getOptions()
-//            .stream().filter(ot -> ot.getInternalName().equals(sizeUnit)).findFirst()
-//            .get().setSelected(true);
-//
-//
-//    SpOutputCollector eventCollector = new StoreEventCollector();
-//
-//    processor.onInvocation(params, eventCollector, null);
-//
-//    LOG.info("Sending event with numOfBytes "
-//        + numOfBytes + ", sizeUnit: " + sizeUnit);
-//    double size = sendEvents(processor, eventCollector);
-//
-//    LOG.info("Expected size is {}", expectedSize);
-//    LOG.info("Actual size is {}", size);
-//    LOG.info("Allowable error is {}", allowableError);
-//    assertEquals(expectedSize, size, allowableError);
-//  }
-//
-//  private Double sendEvents(SizeMeasureProcessor processor, SpOutputCollector spOut) {
-//    double size = 0.0;
-//    Event event = makeEvent(numOfBytes);
-//
-//    processor.onEvent(event, spOut);
-//    try {
-//      Thread.sleep(100);
-//    } catch (InterruptedException e) {
-//      e.printStackTrace();
-//    }
-//    try {
-//      size = event.getFieldBySelector(SizeMeasureProcessor.EVENT_SIZE)
-//              .getAsPrimitive()
-//              .getAsDouble();
-//    } catch (IllegalArgumentException e) {
-//      e.printStackTrace();
-//    }
-//    return size;
-//  }
-//  private Event makeEvent(int numOfBytes) {
-//    Map<String, Object> map = new HashMap<>();
-//    map.put("Test", new ObjectWithSize(numOfBytes));
-//    return EventFactory.fromMap(map,
-//            new SourceInfo("test" + "-topic", "s0"),
-//            new SchemaInfo(null, new ArrayList<>()));
-//  }
-//
-//  public static class ObjectWithSize implements Serializable {
-//    private final byte[] data;
-//
-//    public ObjectWithSize(int numOfBytes) {
-//      this.data = new byte[numOfBytes];
-//    }
-//  }
+  private static final String KEY = "key";
+  private static final String VALUE = "value";
+
+  private static Map<String, Object> inputEvent = Map.of(KEY, VALUE);
+
+  private SizeMeasureProcessor processor;
+
+  @BeforeEach
+  public void setup() {
+    processor = new SizeMeasureProcessor();
+  }
+
+
+  static Stream<Arguments> arguments() {
+    return Stream.of(
+        Arguments.of(
+            BYTES_OPTION,
+            inputEvent,
+            96.0
+        ),
+        Arguments.of(
+            KILO_BYTES_OPTION,
+            inputEvent,
+            0.09375
+        )
+    );
+  }
+
+  @ParameterizedTest
+  @MethodSource("arguments")
+  public void testStringToState(
+      String sizeUnit,
+      Map<String, Object> intpuEvent,
+      double expectedventSize
+  ) {
+
+    var inputEvents = List.of(intpuEvent);
+    List<Map<String, Object>> outputEvents = List.of(Map.of(
+        KEY, VALUE,
+        EVENT_SIZE, expectedventSize
+    ));
+
+    var configuration = TestConfiguration
+        .builder()
+        .config(SIZE_UNIT, sizeUnit)
+        .prefixStrategy(PrefixStrategy.SAME_PREFIX)
+        .build();
+
+    var testExecutor = new ProcessingElementTestExecutor(processor, configuration);
+
+    testExecutor.run(inputEvents, outputEvents);
+  }
 }
\ No newline at end of file