Repository: apache/streampipes
Pull request #3231 (2024-09-16 11:38:20)

feat: Manually override dimensions in data lake sink
<!--
  ~ 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.
-->

The data lake sink, which stores data to the underlying InfluxDB, uses the `PropertyScope` of the input event to determine which fields are stored as dimension/tag. Users should be able to manually override these settings when using the data lake sink.

The PR adds an additional option to the data lake sink which resolves at runtime and allows users to mark boolean, integer and string fields as dimension.

### 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/installer/k8s/README.md b/installer/k8s/README.md
index 598e034ce9..48afd5a59e 100644
--- a/installer/k8s/README.md
+++ b/installer/k8s/README.md
@@ -78,8 +78,7 @@ ui-b94bd9766-rm6zb                             2/2     Running   0          3m27
 For **minikube users**:
 > **NOTE**: If you're running Docker Desktop or Minikube with a local k8s cluster, the above step to use your host IP
 > might not work. Luckily, you can port-forward a service port to your localhost using the following command to be able to
-> access the UI either via `http://localhost` or `http://<HOST_IP>` (you require sudo to run this command in order to bind
-> to a privileged port).
+> access the UI either via `http://localhost:8088` or `http://<HOST_IP>:8088` (If you want to use privileged ports such as 80, you need to run this command with sudo to bind to the privileged port).
 
 ```bash
 kubectl port-forward svc/ui --address=0.0.0.0 8088:8088
diff --git a/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerManager.java b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerManager.java
index ef1b68a0ca..586a77706c 100644
--- a/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerManager.java
+++ b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerManager.java
@@ -34,15 +34,19 @@ public interface IDataExplorerManager {
    * @return An instance of {@link IDataLakeMeasurementCounter} configured to count the sizes of the specified measurements.
    */
   IDataLakeMeasurementCounter getMeasurementCounter(
-    List<DataLakeMeasure> allMeasurements,
-    List<String> measurementsToCount
+      List<DataLakeMeasure> allMeasurements,
+      List<String> measurementsToCount
   );
 
   IDataExplorerQueryManagement getQueryManagement(IDataExplorerSchemaManagement dataExplorerSchemaManagement);
 
   IDataExplorerSchemaManagement getSchemaManagement();
 
-  ITimeSeriesStorage getTimeseriesStorage(DataLakeMeasure measure);
+  default ITimeSeriesStorage getTimeseriesStorage(DataLakeMeasure measure) {
+    return getTimeseriesStorage(measure, false);
+  }
+
+  ITimeSeriesStorage getTimeseriesStorage(DataLakeMeasure measure, boolean ignoreDuplicates);
 
   IDataLakeMeasurementSanitizer getMeasurementSanitizer(IStreamPipesClient client, DataLakeMeasure measure);
 }
diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java
index 1c25d5fbc5..869f314b1d 100644
--- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java
@@ -21,9 +21,9 @@ package org.apache.streampipes.dataexplorer.influx;
 import org.apache.streampipes.client.api.IStreamPipesClient;
 import org.apache.streampipes.commons.environment.Environments;
 import org.apache.streampipes.dataexplorer.DataExplorerSchemaManagement;
+import org.apache.streampipes.dataexplorer.api.IDataExplorerManager;
 import org.apache.streampipes.dataexplorer.api.IDataExplorerQueryManagement;
 import org.apache.streampipes.dataexplorer.api.IDataExplorerSchemaManagement;
-import org.apache.streampipes.dataexplorer.api.IDataExplorerManager;
 import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementCounter;
 import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementSanitizer;
 import org.apache.streampipes.dataexplorer.api.ITimeSeriesStorage;
@@ -38,28 +38,33 @@ public class DataExplorerManagerInflux implements IDataExplorerManager {
 
   @Override
   public IDataLakeMeasurementCounter getMeasurementCounter(
-    List<DataLakeMeasure> allMeasurements,
-    List<String> measurementsToCount) {
+      List<DataLakeMeasure> allMeasurements,
+      List<String> measurementsToCount) {
     return new DataLakeMeasurementCounterInflux(allMeasurements, measurementsToCount);
   }
 
   @Override
   public IDataExplorerQueryManagement getQueryManagement(
-    IDataExplorerSchemaManagement dataExplorerSchemaManagement
-    ) {
+      IDataExplorerSchemaManagement dataExplorerSchemaManagement
+  ) {
     return new DataExplorerQueryManagementInflux(dataExplorerSchemaManagement);
   }
 
   @Override
   public IDataExplorerSchemaManagement getSchemaManagement() {
     return new DataExplorerSchemaManagement(StorageDispatcher.INSTANCE
-                                                             .getNoSqlStore()
-                                                             .getDataLakeStorage());
+        .getNoSqlStore()
+        .getDataLakeStorage());
   }
 
   @Override
-  public ITimeSeriesStorage getTimeseriesStorage(DataLakeMeasure measure) {
-    return new TimeSeriesStorageInflux(measure, Environments.getEnvironment(), new InfluxClientProvider());
+  public ITimeSeriesStorage getTimeseriesStorage(DataLakeMeasure measure, boolean ignoreDuplicates) {
+    return new TimeSeriesStorageInflux(
+        measure,
+        ignoreDuplicates,
+        Environments.getEnvironment(),
+        new InfluxClientProvider()
+    );
   }
 
   @Override
diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/PropertyDuplicateFilter.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/PropertyDuplicateFilter.java
new file mode 100644
index 0000000000..fe70b15eae
--- /dev/null
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/PropertyDuplicateFilter.java
@@ -0,0 +1,45 @@
+/*
+ * 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.influx;
+
+import org.apache.streampipes.model.runtime.field.PrimitiveField;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+public class PropertyDuplicateFilter {
+
+  private final boolean ignoreDuplicates;
+  private final Map<String, Object> lastValues;
+
+  public PropertyDuplicateFilter(boolean ignoreDuplicates) {
+    this.ignoreDuplicates = ignoreDuplicates;
+    this.lastValues = new HashMap<>();
+  }
+
+  public boolean shouldIgnoreField(String sanitizedRuntimeName,
+                                   PrimitiveField eventPropertyPrimitiveField) {
+    var rawValue = eventPropertyPrimitiveField.getRawValue();
+    boolean shouldIgnore = ignoreDuplicates && lastValues.containsKey(sanitizedRuntimeName)
+        && Objects.equals(lastValues.get(sanitizedRuntimeName), rawValue);
+    lastValues.put(sanitizedRuntimeName, rawValue);
+    return shouldIgnore;
+  }
+}
diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/PropertyHandler.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/PropertyHandler.java
index 74d03ba3e2..a19c7d9ed8 100644
--- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/PropertyHandler.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/PropertyHandler.java
@@ -34,10 +34,13 @@ public class PropertyHandler {
 
   private static final Logger LOG = LoggerFactory.getLogger(PropertyHandler.class);
 
-  private RawFieldSerializer rawFieldSerializer;
+  private final RawFieldSerializer rawFieldSerializer;
+  private final PropertyDuplicateFilter duplicateFilter;
 
-  public PropertyHandler() {
+
+  public PropertyHandler(PropertyDuplicateFilter duplicateFilter) {
     rawFieldSerializer = new RawFieldSerializer();
+    this.duplicateFilter = duplicateFilter;
   }
 
   /**
@@ -109,30 +112,32 @@ public class PropertyHandler {
       String sanitizedRuntimeName,
       PrimitiveField eventPropertyPrimitiveField
   ) {
-    // Store property according to property type
-    if (XSD.INTEGER.toString()
-                   .equals(runtimeType)) {
-      handleIntegerProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
-    } else if (XSD.LONG.toString()
-                       .equals(runtimeType)) {
-      handleLongProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
-    } else if (XSD.FLOAT.toString()
-                        .equals(runtimeType)) {
-      handleFloatProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
-    } else if (XSD.DOUBLE.toString()
-                         .equals(runtimeType)) {
-      handleDoubleProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
-    } else if (XSD.BOOLEAN.toString()
-                          .equals(runtimeType)) {
-      handleBooleanProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
-    } else if (SO.NUMBER.equals(runtimeType)) {
-      handleDoubleProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
-    } else {
-      handleStringProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
+
+    if (!duplicateFilter.shouldIgnoreField(sanitizedRuntimeName, eventPropertyPrimitiveField)) {
+      // Store property according to property type
+      if (XSD.INTEGER.toString()
+          .equals(runtimeType)) {
+        handleIntegerProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
+      } else if (XSD.LONG.toString()
+          .equals(runtimeType)) {
+        handleLongProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
+      } else if (XSD.FLOAT.toString()
+          .equals(runtimeType)) {
+        handleFloatProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
+      } else if (XSD.DOUBLE.toString()
+          .equals(runtimeType)) {
+        handleDoubleProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
+      } else if (XSD.BOOLEAN.toString()
+          .equals(runtimeType)) {
+        handleBooleanProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
+      } else if (SO.NUMBER.equals(runtimeType)) {
+        handleDoubleProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
+      } else {
+        handleStringProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
+      }
     }
   }
 
-
   private void handleStringProperty(
       Point.Builder p,
       String sanitizedRuntimeName,
diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java
index fa35cbe9b9..799808541e 100644
--- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java
@@ -38,15 +38,23 @@ public class TimeSeriesStorageInflux extends TimeSeriesStorage {
 
   private final PropertyHandler propertyHandler;
 
+  public TimeSeriesStorageInflux(
+      DataLakeMeasure measure,
+      Environment environment,
+      InfluxClientProvider influxClientProvider
+  ) throws SpRuntimeException {
+    this(measure, false, environment, influxClientProvider);
+  }
 
   public TimeSeriesStorageInflux(
       DataLakeMeasure measure,
+      boolean ignoreDuplicates,
       Environment environment,
       InfluxClientProvider influxClientProvider
   ) throws SpRuntimeException {
     super(measure);
     this.influxDb = influxClientProvider.getSetUpInfluxDBClient(environment);
-    propertyHandler = new PropertyHandler();
+    propertyHandler = new PropertyHandler(new PropertyDuplicateFilter(ignoreDuplicates));
   }
 
   protected void writeToTimeSeriesStorage(Event event) throws SpRuntimeException {
diff --git a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
index 97ef58e99e..612c2dee6c 100644
--- a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
+++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
@@ -56,7 +56,7 @@ public class DataExplorerManagerIotDb implements IDataExplorerManager {
   }
 
   @Override
-  public ITimeSeriesStorage getTimeseriesStorage(DataLakeMeasure measure) {
+  public ITimeSeriesStorage getTimeseriesStorage(DataLakeMeasure measure, boolean ignoreDuplicates) {
     return new TimeSeriesStorageIotDb(measure, new IotDbPropertyConverter(), new IotDbSessionProvider());
   }
 
diff --git a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/extractor/IParameterExtractor.java b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/extractor/IParameterExtractor.java
index 396f0f3961..b9bd660f08 100644
--- a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/extractor/IParameterExtractor.java
+++ b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/extractor/IParameterExtractor.java
@@ -102,4 +102,6 @@ public interface IParameterExtractor {
   List<String> getEventPropertiesSelectorByScope(PropertyScope scope);
 
   List<EventProperty> getEventPropertiesByScope(PropertyScope scope);
+
+  List<EventProperty> getInputEventProperties(int streamIndex);
 }
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/InternalSinksExtensionModuleExports.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/InternalSinksExtensionModuleExports.java
index 8dba55f9aa..72bcbcb7d1 100644
--- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/InternalSinksExtensionModuleExports.java
+++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/InternalSinksExtensionModuleExports.java
@@ -24,6 +24,7 @@ import org.apache.streampipes.extensions.api.migration.IModelMigrator;
 import org.apache.streampipes.extensions.api.pe.IStreamPipesPipelineElement;
 import org.apache.streampipes.sinks.internal.jvm.datalake.DataLakeSink;
 import org.apache.streampipes.sinks.internal.jvm.datalake.migrations.DataLakeSinkMigrationV1;
+import org.apache.streampipes.sinks.internal.jvm.datalake.migrations.DataLakeSinkMigrationV2;
 import org.apache.streampipes.sinks.internal.jvm.notification.InternalStreamPipesNotificationSink;
 
 import java.util.Collections;
@@ -45,6 +46,8 @@ public class InternalSinksExtensionModuleExports implements IExtensionModuleExpo
 
   @Override
   public List<IModelMigrator<?, ?>> migrators() {
-    return List.of(new DataLakeSinkMigrationV1());
+    return List.of(
+        new DataLakeSinkMigrationV1(),
+        new DataLakeSinkMigrationV2());
   }
 }
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeDimensionProvider.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeDimensionProvider.java
new file mode 100644
index 0000000000..05c02730ee
--- /dev/null
+++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeDimensionProvider.java
@@ -0,0 +1,70 @@
+/*
+ * 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.sinks.internal.jvm.datalake;
+
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.model.staticproperty.Option;
+import org.apache.streampipes.model.staticproperty.RuntimeResolvableAnyStaticProperty;
+import org.apache.streampipes.sdk.utils.Datatypes;
+import org.apache.streampipes.vocabulary.SO;
+
+import java.net.URI;
+import java.util.List;
+
+public class DataLakeDimensionProvider {
+
+  public void applyOptions(List<EventProperty> inputFields,
+                           RuntimeResolvableAnyStaticProperty staticProperty) {
+    var primitiveFields = getPrimitiveFields(inputFields);
+    primitiveFields
+        .forEach(field -> addFieldIfNotExists(field, staticProperty.getOptions()));
+    staticProperty.getOptions().removeIf(o -> !existsInFields(o, primitiveFields));
+  }
+
+  private List<EventPropertyPrimitive> getPrimitiveFields(List<EventProperty> inputFields) {
+    return inputFields
+        .stream()
+        .filter(field -> field instanceof EventPropertyPrimitive)
+        .filter(field -> satisfiesFilter((EventPropertyPrimitive) field))
+        .map(field -> (EventPropertyPrimitive) field)
+        .toList();
+  }
+
+  private boolean satisfiesFilter(EventPropertyPrimitive field) {
+    return !field.getRuntimeType().equals(Datatypes.Float.toString())
+        &&  !(field.getDomainProperties().stream().map(URI::toString).toList().contains(SO.DATE_TIME));
+  }
+
+  private void addFieldIfNotExists(EventPropertyPrimitive field,
+                                   List<Option> options) {
+    if (options.stream().noneMatch(o -> o.getName().equals(field.getRuntimeName()))) {
+      options.add(new Option(
+          field.getRuntimeName(),
+          PropertyScope.valueOf(field.getPropertyScope()) == PropertyScope.DIMENSION_PROPERTY)
+      );
+    }
+  }
+
+  private boolean existsInFields(Option o,
+                                 List<EventPropertyPrimitive> primitiveFields) {
+    return primitiveFields.stream().anyMatch(field -> field.getRuntimeName().equals(o.getName()));
+  }
+}
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java
index ef88bbdcdc..5cdeae32a0 100644
--- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java
+++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java
@@ -22,14 +22,19 @@ import org.apache.streampipes.commons.environment.Environments;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.dataexplorer.TimeSeriesStore;
 import org.apache.streampipes.dataexplorer.management.DataExplorerDispatcher;
+import org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
+import org.apache.streampipes.extensions.api.runtime.SupportsRuntimeConfig;
 import org.apache.streampipes.model.DataSinkType;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.datalake.DataLakeMeasureSchemaUpdateStrategy;
 import org.apache.streampipes.model.extensions.ExtensionAssetType;
 import org.apache.streampipes.model.graph.DataSinkDescription;
 import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.schema.EventSchema;
 import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.model.staticproperty.RuntimeResolvableAnyStaticProperty;
+import org.apache.streampipes.model.staticproperty.StaticProperty;
 import org.apache.streampipes.sdk.builder.DataSinkBuilder;
 import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
 import org.apache.streampipes.sdk.helpers.EpRequirements;
@@ -39,15 +44,21 @@ import org.apache.streampipes.sdk.helpers.Options;
 import org.apache.streampipes.wrapper.params.compat.SinkParams;
 import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
 
-public class DataLakeSink extends StreamPipesDataSink {
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DataLakeSink extends StreamPipesDataSink implements SupportsRuntimeConfig {
 
   private static final String DATABASE_MEASUREMENT_KEY = "db_measurement";
   private static final String TIMESTAMP_MAPPING_KEY = "timestamp_mapping";
   public static final String SCHEMA_UPDATE_KEY = "schema_update";
+  public static final String DIMENSIONS_KEY = "dimensions_selection";
+  public static final String IGNORE_DUPLICATES_KEY = "ignore_duplicates";
 
   public static final String SCHEMA_UPDATE_OPTION = "Update schema";
 
   public static final String EXTEND_EXISTING_SCHEMA_OPTION = "Extend existing schema";
+  private static final Logger LOG = LoggerFactory.getLogger(DataLakeSink.class);
 
   private TimeSeriesStore timeSeriesStore;
 
@@ -55,25 +66,26 @@ public class DataLakeSink extends StreamPipesDataSink {
   @Override
   public DataSinkDescription declareModel() {
     return DataSinkBuilder
-      .create("org.apache.streampipes.sinks.internal.jvm.datalake", 1)
-      .withLocales(Locales.EN)
-      .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
-      .category(DataSinkType.INTERNAL)
-      .requiredStream(StreamRequirementsBuilder
-                        .create()
-                        .requiredPropertyWithUnaryMapping(
-                          EpRequirements.timestampReq(),
-                          Labels.withId(TIMESTAMP_MAPPING_KEY),
-                          PropertyScope.NONE
-                        )
-                        .build())
-      .requiredTextParameter(Labels.withId(DATABASE_MEASUREMENT_KEY))
-      .requiredSingleValueSelection(
-        Labels.withId(SCHEMA_UPDATE_KEY),
-        Options.from(SCHEMA_UPDATE_OPTION, EXTEND_EXISTING_SCHEMA_OPTION)
-      )
-
-      .build();
+        .create("org.apache.streampipes.sinks.internal.jvm.datalake", 2)
+        .withLocales(Locales.EN)
+        .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
+        .category(DataSinkType.INTERNAL)
+        .requiredStream(StreamRequirementsBuilder
+            .create()
+            .requiredPropertyWithUnaryMapping(
+                EpRequirements.timestampReq(),
+                Labels.withId(TIMESTAMP_MAPPING_KEY),
+                PropertyScope.NONE
+            )
+            .build())
+        .requiredTextParameter(Labels.withId(DATABASE_MEASUREMENT_KEY))
+        .requiredSingleValueSelection(
+            Labels.withId(SCHEMA_UPDATE_KEY),
+            Options.from(SCHEMA_UPDATE_OPTION, EXTEND_EXISTING_SCHEMA_OPTION)
+        )
+        .requiredMultiValueSelectionFromContainer(Labels.withId(DIMENSIONS_KEY))
+        .requiredSlideToggle(Labels.withId(IGNORE_DUPLICATES_KEY), false)
+        .build();
   }
 
   @Override
@@ -81,9 +93,21 @@ public class DataLakeSink extends StreamPipesDataSink {
     var extractor = parameters.extractor();
     var timestampField = extractor.mappingPropertyValue(TIMESTAMP_MAPPING_KEY);
     var measureName = extractor.singleValueParameter(DATABASE_MEASUREMENT_KEY, String.class);
-    var eventSchema = parameters.getInputSchemaInfos()
-                                .get(0)
-                                .getEventSchema();
+    var dimensions = extractor.selectedMultiValues(DIMENSIONS_KEY, String.class);
+    var ignoreDuplicates = extractor.slideToggleValue(IGNORE_DUPLICATES_KEY);
+    var eventSchema = new EventSchema(parameters.getInputSchemaInfos()
+        .get(0)
+        .getEventSchema()
+        .getEventProperties()
+        .stream()
+        .peek(ep -> {
+          if (dimensions.contains(ep.getRuntimeName())) {
+            LOG.info("Using {} as dimension", ep.getRuntimeName());
+            ep.setPropertyScope(PropertyScope.DIMENSION_PROPERTY.name());
+          }
+        })
+        .toList()
+    );
 
 
     var measure = new DataLakeMeasure(measureName, timestampField, eventSchema);
@@ -101,7 +125,7 @@ public class DataLakeSink extends StreamPipesDataSink {
         .sanitizeAndRegister();
 
     this.timeSeriesStore = new TimeSeriesStore(
-        new DataExplorerDispatcher().getDataExplorerManager().getTimeseriesStorage(measure),
+        new DataExplorerDispatcher().getDataExplorerManager().getTimeseriesStorage(measure, ignoreDuplicates),
         measure,
         Environments.getEnvironment(),
         true
@@ -118,4 +142,13 @@ public class DataLakeSink extends StreamPipesDataSink {
   public void onDetach() throws SpRuntimeException {
     this.timeSeriesStore.close();
   }
+
+  @Override
+  public StaticProperty resolveConfiguration(String staticPropertyInternalName,
+                                             IStaticPropertyExtractor extractor) {
+    var staticProperty = extractor.getStaticPropertyByName(DIMENSIONS_KEY, RuntimeResolvableAnyStaticProperty.class);
+    var inputFields = extractor.getInputEventProperties(0);
+    new DataLakeDimensionProvider().applyOptions(inputFields, staticProperty);
+    return staticProperty;
+  }
 }
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/migrations/DataLakeSinkMigrationV2.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/migrations/DataLakeSinkMigrationV2.java
new file mode 100644
index 0000000000..5a81ba4cd7
--- /dev/null
+++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/migrations/DataLakeSinkMigrationV2.java
@@ -0,0 +1,83 @@
+/*
+ * 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.sinks.internal.jvm.datalake.migrations;
+
+import org.apache.streampipes.extensions.api.extractor.IDataSinkParameterExtractor;
+import org.apache.streampipes.extensions.api.migration.IDataSinkMigrator;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.migration.MigrationResult;
+import org.apache.streampipes.model.migration.ModelMigratorConfig;
+import org.apache.streampipes.model.staticproperty.RuntimeResolvableAnyStaticProperty;
+import org.apache.streampipes.model.staticproperty.SlideToggleStaticProperty;
+import org.apache.streampipes.sdk.helpers.Labels;
+import org.apache.streampipes.sinks.internal.jvm.datalake.DataLakeDimensionProvider;
+import org.apache.streampipes.sinks.internal.jvm.datalake.DataLakeSink;
+
+public class DataLakeSinkMigrationV2 implements IDataSinkMigrator {
+  @Override
+  public ModelMigratorConfig config() {
+    return new ModelMigratorConfig(
+        "org.apache.streampipes.sinks.internal.jvm.datalake",
+        SpServiceTagPrefix.DATA_SINK,
+        1,
+        2
+    );
+  }
+
+  @Override
+  public MigrationResult<DataSinkInvocation> migrate(DataSinkInvocation element,
+                                                     IDataSinkParameterExtractor extractor) throws RuntimeException {
+    addDimensionSelection(element);
+    addDuplicateToggle(element);
+    return MigrationResult.success(element);
+  }
+
+  private void addDimensionSelection(DataSinkInvocation element) {
+    var label = Labels.from(
+        DataLakeSink.DIMENSIONS_KEY,
+        "Dimensions",
+        "Selected fields will be stored as dimensions."
+    );
+    var staticProperty = new RuntimeResolvableAnyStaticProperty(
+        label.getInternalId(),
+        label.getLabel(),
+        label.getDescription()
+    );
+    var inputFields = element.getInputStreams().get(0).getEventSchema().getEventProperties();
+    new DataLakeDimensionProvider().applyOptions(inputFields, staticProperty);
+
+    element.getStaticProperties().add(staticProperty);
+  }
+
+  private void addDuplicateToggle(DataSinkInvocation element) {
+    var label = Labels.from(
+        DataLakeSink.IGNORE_DUPLICATES_KEY,
+        "Ignore duplicates",
+        "Fields having the same value than the previous event are not stored."
+    );
+    var staticProperty = new SlideToggleStaticProperty(
+        label.getInternalId(),
+        label.getLabel(),
+        label.getDescription(),
+        false);
+
+    element.getStaticProperties().add(staticProperty);
+  }
+}
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/documentation.md b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/documentation.md
index 2d58b7e3d9..b9dde8e06d 100644
--- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/documentation.md
+++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/documentation.md
@@ -41,6 +41,16 @@ This sink requires an event that provides a timestamp value (a field that is mar
 
 ## Configuration
 
+### Dimensions
+
+The fields which will be stored as dimensional values in the time series storage. Dimensions are typically identifiers 
+such as the ID of a sensor.
+Dimensions support grouping in the data explorer, but will be converted to a text-based field and provide less advanced 
+filtering capabilities.
+
+Be careful when modifying dimensions of existing pipelines! This might have impact on how you are able to view data in 
+the data explorer due to schema incompatibilities.
+
 ### Identifier
 
 The name of the measurement (table) where the events are stored.
@@ -60,3 +70,22 @@ The Schema Update Options dictate the behavior when encountering a measurement (
 - **Description:** Keeps old event fields in the event schema.
 - **Strategy:** This follows an append-only strategy, allowing continued work with historic data.
 - **Consideration:** Old properties may exist for which no new data is generated.
+
+
+### Dimensions
+
+Select fields which will be marked as dimensions. Dimensions reflect tags in the underlying time-series database. 
+Dimensions support grouping operations and can be used for fields with a limited set of values, e.g., boolean flags or 
+fields representing IDs. Dimensions are not a good choice for fields with a high number of different values since they 
+slow down database queries.
+
+By default, all fields which are marked as dimensions in the metadata are chosen and can be manually overridden 
+with this configuration.
+
+Data types which can be marked as dimensional values are booleans, integer, and strings.
+
+### Ignore Duplicates
+
+Before writing an event to the time series storage, a duplicate check is performed. By activating this option, only 
+fields having a different value than the previous event are stored. 
+This setting only affects measurement fields, not dimensions.
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/strings.en b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/strings.en
index 0b604bdfd9..d1fcc39c7e 100644
--- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/strings.en
+++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/strings.en
@@ -26,3 +26,9 @@ timestamp_mapping.description=The value which contains a timestamp
 
 schema_update.title=Schema Update
 schema_update.description=Update existing schemas with the new one or extend the existing schema with new properties
+
+dimensions_selection.title=Dimensions
+dimensions_selection.description=Selected fields will be stored as dimensions.
+
+ignore_duplicates.title=Ignore duplicates
+ignore_duplicates.description=Fields having the same value than the previous event are not stored. This only affects measurements, not tags.
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/dashboard/DashboardItem.java b/streampipes-model/src/main/java/org/apache/streampipes/model/dashboard/DashboardItem.java
index c7962b0411..81759642cb 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/dashboard/DashboardItem.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/dashboard/DashboardItem.java
@@ -18,6 +18,7 @@
 package org.apache.streampipes.model.dashboard;
 
 import java.util.List;
+import java.util.Map;
 
 public class DashboardItem {
 
@@ -31,6 +32,7 @@ public class DashboardItem {
   private Integer rows;
   private Integer x;
   private Integer y;
+  private Map<String, Object> timeSettings;
 
   public DashboardItem() {
 
@@ -99,4 +101,12 @@ public class DashboardItem {
   public void setY(Integer y) {
     this.y = y;
   }
+
+  public Map<String, Object> getTimeSettings() {
+    return timeSettings;
+  }
+
+  public void setTimeSettings(Map<String, Object> timeSettings) {
+    this.timeSettings = timeSettings;
+  }
 }
diff --git a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/extractor/AbstractParameterExtractor.java b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/extractor/AbstractParameterExtractor.java
index b7230510f3..bb365ec3c3 100644
--- a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/extractor/AbstractParameterExtractor.java
+++ b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/extractor/AbstractParameterExtractor.java
@@ -666,4 +666,13 @@ public abstract class AbstractParameterExtractor<T extends InvocableStreamPipesE
                                                        .equals(scope.name()))
         .collect(Collectors.toList());
   }
+
+  @Override
+  public List<EventProperty> getInputEventProperties(int streamIndex) {
+    if (!sepaElement.getInputStreams().isEmpty()) {
+      return sepaElement.getInputStreams().get(streamIndex).getEventSchema().getEventProperties();
+    } else {
+      return List.of();
+    }
+  }
 }
diff --git a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/function/StreamPipesFunctionHandler.java b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/function/StreamPipesFunctionHandler.java
index 7b02dc4db4..a36d788a4d 100644
--- a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/function/StreamPipesFunctionHandler.java
+++ b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/function/StreamPipesFunctionHandler.java
@@ -96,7 +96,7 @@ public enum StreamPipesFunctionHandler {
     this.runningInstances.forEach((key, value) -> {
       value.discardRuntime();
     });
-    new Thread(new FunctionDeregistrationHandler(getFunctionDefinitions())).start();
+    new FunctionDeregistrationHandler(getFunctionDefinitions()).run();
   }
 
   private List<FunctionDefinition> getFunctionDefinitions() {
diff --git a/ui/.lintstagedrc.json b/ui/.lintstagedrc.json
index 059fea6ac0..ca7497c491 100644
--- a/ui/.lintstagedrc.json
+++ b/ui/.lintstagedrc.json
@@ -1,4 +1,4 @@
 {
-  "src/**/*.ts": "prettier --write",
-  "src/**/*.ts": ["ng-lint-staged lint --fix"]
+  "*.*": "prettier --check",
+  "src/**/*.ts": ["ng-lint-staged lint --"]
 }
diff --git a/ui/projects/streampipes/platform-services/src/lib/model/dashboard/dashboard.model.ts b/ui/projects/streampipes/platform-services/src/lib/model/dashboard/dashboard.model.ts
index 8a65f934ed..532a19b2d4 100644
--- a/ui/projects/streampipes/platform-services/src/lib/model/dashboard/dashboard.model.ts
+++ b/ui/projects/streampipes/platform-services/src/lib/model/dashboard/dashboard.model.ts
@@ -25,6 +25,7 @@ export interface DashboardConfig extends GridsterConfig {}
 export interface ClientDashboardItem extends GridsterItem {
     widgetId: string;
     widgetType: string;
+    timeSettings?: TimeSettings;
     id: string;
 }
 
diff --git a/ui/projects/streampipes/platform-services/src/lib/model/datalake/DateRange.ts b/ui/projects/streampipes/platform-services/src/lib/model/datalake/DateRange.ts
index 6c305f6dd4..9f3cff7d29 100644
--- a/ui/projects/streampipes/platform-services/src/lib/model/datalake/DateRange.ts
+++ b/ui/projects/streampipes/platform-services/src/lib/model/datalake/DateRange.ts
@@ -24,6 +24,11 @@ export interface TimeSettings {
     timeSelectionId?: TimeSelectionId;
 }
 
+export interface WidgetTimeSettings {
+    timeSettings: TimeSettings;
+    widgetIndex?: number;
+}
+
 export interface TimeString {
     startDate: string;
     startTime: string;
diff --git a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
index 7c9650ec20..398a67c5f7 100644
--- a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
+++ b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
@@ -20,7 +20,7 @@
 /* tslint:disable */
 /* eslint-disable */
 // @ts-nocheck
-// Generated using typescript-generator version 3.2.1263 on 2024-09-10 17:11:03.
+// Generated using typescript-generator version 3.2.1263 on 2024-09-18 15:50:05.
 
 export class NamedStreamPipesEntity implements Storable {
     '@class':
@@ -970,6 +970,7 @@ export class DashboardItem {
     name: string;
     rows: number;
     settings: string[];
+    timeSettings: { [index: string]: any };
     x: number;
     y: number;
 
@@ -989,6 +990,9 @@ export class DashboardItem {
         instance.settings = __getCopyArrayFn(__identity<string>())(
             data.settings,
         );
+        instance.timeSettings = __getCopyObjectFn(__identity<any>())(
+            data.timeSettings,
+        );
         instance.x = data.x;
         instance.y = data.y;
         return instance;
diff --git a/ui/src/app/core-ui/static-properties/static-slide-toggle/static-slide-toggle.component.html b/ui/src/app/core-ui/static-properties/static-slide-toggle/static-slide-toggle.component.html
index 0863899fa6..e3a4e5c69c 100644
--- a/ui/src/app/core-ui/static-properties/static-slide-toggle/static-slide-toggle.component.html
+++ b/ui/src/app/core-ui/static-properties/static-slide-toggle/static-slide-toggle.component.html
@@ -22,7 +22,6 @@
             fxFlex
             formControlName="{{ fieldName }}"
             color="accent"
-            required
             [attr.data-cy]="fieldName"
         >
             {{ staticProperty.description }}
diff --git a/ui/src/app/core-ui/static-properties/static-slide-toggle/static-slide-toggle.component.ts b/ui/src/app/core-ui/static-properties/static-slide-toggle/static-slide-toggle.component.ts
index 64e17a6005..e74274ae7f 100644
--- a/ui/src/app/core-ui/static-properties/static-slide-toggle/static-slide-toggle.component.ts
+++ b/ui/src/app/core-ui/static-properties/static-slide-toggle/static-slide-toggle.component.ts
@@ -33,6 +33,7 @@ export class StaticSlideToggleComponent
     ngOnInit(): void {
         this.addValidator(this.staticProperty.selected, Validators.required);
         this.enableValidators();
+        this.emitUpdate();
     }
 
     emitUpdate() {
diff --git a/ui/src/app/data-explorer/components/time-selector/time-selector-menu/custom-time-range-selection/custom-time-range-selection.component.html b/ui/src/app/data-explorer/components/time-selector/time-selector-menu/custom-time-range-selection/custom-time-range-selection.component.html
index 9e277f0ca6..1564d9b5dc 100644
--- a/ui/src/app/data-explorer/components/time-selector/time-selector-menu/custom-time-range-selection/custom-time-range-selection.component.html
+++ b/ui/src/app/data-explorer/components/time-selector/time-selector-menu/custom-time-range-selection/custom-time-range-selection.component.html
@@ -74,6 +74,7 @@
         </div>
     </div>
     <div fxLayout="row" fxLayoutAlign="end center" class="mt-10">
+        <ng-content> </ng-content>
         <button
             mat-raised-button
             color="accent"
diff --git a/ui/src/app/data-explorer/components/time-selector/time-selector-menu/time-selector-menu.component.html b/ui/src/app/data-explorer/components/time-selector/time-selector-menu/time-selector-menu.component.html
index 70ded63676..e53550aa43 100644
--- a/ui/src/app/data-explorer/components/time-selector/time-selector-menu/time-selector-menu.component.html
+++ b/ui/src/app/data-explorer/components/time-selector/time-selector-menu/time-selector-menu.component.html
@@ -46,6 +46,7 @@
                     (timeSettingsEmitter)="timeSettingsEmitter.emit($event)"
                     class="w-100"
                 >
+                    <ng-content> </ng-content>
                 </sp-custom-time-range-selection>
             </div>
         </div>
diff --git a/ui/src/app/data-explorer/components/widget/data-explorer-dashboard-widget.component.html b/ui/src/app/data-explorer/components/widget/data-explorer-dashboard-widget.component.html
index fa9617a0ed..ea476a9846 100644
--- a/ui/src/app/data-explorer/components/widget/data-explorer-dashboard-widget.component.html
+++ b/ui/src/app/data-explorer/components/widget/data-explorer-dashboard-widget.component.html
@@ -111,6 +111,34 @@
                 >
                     <mat-icon>get_app</mat-icon>
                 </button>
+                <button
+                    mat-icon-button
+                    [matMenuTriggerFor]="optMenu"
+                    *ngIf="!globalTimeEnabled"
+                    aria-label="Options"
+                    data-cy="options-data-explorer"
+                    #menuTrigger="matMenuTrigger"
+                >
+                    <mat-icon
+                        [color]="timeSettingsModified ? 'primary' : 'default'"
+                        >alarm_clock</mat-icon
+                    >
+                </button>
+                <mat-menu #optMenu="matMenu" class="large-menu">
+                    <sp-time-selector-menu
+                        [timeSettings]="clonedTimeSettings"
+                        (timeSettingsEmitter)="modifyWidgetTimeSettings($event)"
+                        class="w-100"
+                    >
+                        <button
+                            mat-raised-button
+                            class="mat-basic"
+                            (click)="resetWidgetTimeSettings()"
+                        >
+                            Reset
+                        </button>
+                    </sp-time-selector-menu>
+                </mat-menu>
                 <button
                     mat-icon-button
                     (click)="removeWidget()"
diff --git a/ui/src/app/data-explorer/components/widget/data-explorer-dashboard-widget.component.ts b/ui/src/app/data-explorer/components/widget/data-explorer-dashboard-widget.component.ts
index 9e3e962de2..21e5170c0e 100644
--- a/ui/src/app/data-explorer/components/widget/data-explorer-dashboard-widget.component.ts
+++ b/ui/src/app/data-explorer/components/widget/data-explorer-dashboard-widget.component.ts
@@ -22,9 +22,11 @@ import {
     ComponentRef,
     EventEmitter,
     Input,
+    OnChanges,
     OnDestroy,
     OnInit,
     Output,
+    SimpleChanges,
     ViewChild,
 } from '@angular/core';
 import { GridsterItemComponent } from 'angular-gridster2';
@@ -45,13 +47,18 @@ import { UserPrivilege } from '../../../_enums/user-privilege.enum';
 import { CurrentUserService } from '@streampipes/shared-ui';
 import { BaseWidgetData } from '../../models/dataview-dashboard.model';
 import { DataExplorerDashboardService } from '../../services/data-explorer-dashboard.service';
+import { TimeSelectionService } from '../../services/time-selection.service';
+import { MatMenuTrigger } from '@angular/material/menu';
 
 @Component({
     selector: 'sp-data-explorer-dashboard-widget',
     templateUrl: './data-explorer-dashboard-widget.component.html',
     styleUrls: ['./data-explorer-dashboard-widget.component.scss'],
 })
-export class DataExplorerDashboardWidgetComponent implements OnInit, OnDestroy {
+export class DataExplorerDashboardWidgetComponent
+    implements OnInit, OnDestroy, OnChanges
+{
+    @ViewChild('menuTrigger') menu: MatMenuTrigger;
     @Input()
     dashboardItem: DashboardItem;
 
@@ -97,6 +104,9 @@ export class DataExplorerDashboardWidgetComponent implements OnInit, OnDestroy {
     timerActive = false;
     loadingTime = 0;
 
+    clonedTimeSettings: TimeSettings;
+    timeSettingsModified: boolean = false;
+
     hasDataExplorerWritePrivileges = false;
 
     authSubscription: Subscription;
@@ -116,8 +126,16 @@ export class DataExplorerDashboardWidgetComponent implements OnInit, OnDestroy {
         private widgetTypeService: WidgetTypeService,
         private authService: AuthService,
         private currentUserService: CurrentUserService,
+        private timeSelectionService: TimeSelectionService,
     ) {}
 
+    ngOnChanges(changes: SimpleChanges): void {
+        if (changes.widgetIndex && this.componentRef?.instance) {
+            this.componentRef.instance.widgetIndex =
+                changes.widgetIndex.currentValue;
+        }
+    }
+
     ngOnInit(): void {
         this.authSubscription = this.currentUserService.user$.subscribe(
             user => {
@@ -139,6 +157,14 @@ export class DataExplorerDashboardWidgetComponent implements OnInit, OnDestroy {
                 },
             );
         this.chooseWidget(this.configuredWidget.widgetType);
+        this.clonedTimeSettings = {
+            startTime: this.timeSettings.startTime,
+            endTime: this.timeSettings.endTime,
+            timeSelectionId: this.timeSettings.timeSelectionId,
+        };
+        if (this.dashboardItem.timeSettings !== undefined) {
+            this.timeSettingsModified = true;
+        }
     }
 
     ngOnDestroy() {
@@ -176,6 +202,7 @@ export class DataExplorerDashboardWidgetComponent implements OnInit, OnDestroy {
         this.componentRef.instance.dataExplorerWidget = this.configuredWidget;
         this.componentRef.instance.previewMode = this.previewMode;
         this.componentRef.instance.gridMode = this.gridMode;
+        this.componentRef.instance.widgetIndex = this.widgetIndex;
         const removeSub =
             this.componentRef.instance.removeWidgetCallback.subscribe(ev =>
                 this.removeWidget(),
@@ -196,9 +223,13 @@ export class DataExplorerDashboardWidgetComponent implements OnInit, OnDestroy {
     }
 
     getTimeSettings(): TimeSettings {
-        return this.globalTimeEnabled
-            ? this.timeSettings
-            : (this.configuredWidget.timeSettings as TimeSettings);
+        if (this.globalTimeEnabled) {
+            return this.timeSettings;
+        } else if (this.dashboardItem.timeSettings !== undefined) {
+            return this.dashboardItem.timeSettings as TimeSettings;
+        } else {
+            return this.configuredWidget.timeSettings as TimeSettings;
+        }
     }
 
     removeWidget() {
@@ -233,4 +264,26 @@ export class DataExplorerDashboardWidgetComponent implements OnInit, OnDestroy {
             this.configuredWidget,
         );
     }
+
+    modifyWidgetTimeSettings(timeSettings: TimeSettings): void {
+        this.dashboardItem.timeSettings = timeSettings;
+        this.timeSelectionService.notify(timeSettings, this.widgetIndex);
+        this.menu.closeMenu();
+        this.timeSettingsModified = true;
+    }
+
+    resetWidgetTimeSettings(): void {
+        this.dashboardItem.timeSettings = undefined;
+        this.clonedTimeSettings = {
+            startTime: this.timeSettings.startTime,
+            endTime: this.timeSettings.endTime,
+            timeSelectionId: this.timeSettings.timeSelectionId,
+        };
+        this.timeSelectionService.notify(
+            this.getTimeSettings(),
+            this.widgetIndex,
+        );
+        this.menu.closeMenu();
+        this.timeSettingsModified = false;
+    }
 }
diff --git a/ui/src/app/data-explorer/components/widgets/base/base-data-explorer-widget.directive.ts b/ui/src/app/data-explorer/components/widgets/base/base-data-explorer-widget.directive.ts
index b8e86ffdb9..5df627e9e7 100644
--- a/ui/src/app/data-explorer/components/widgets/base/base-data-explorer-widget.directive.ts
+++ b/ui/src/app/data-explorer/components/widgets/base/base-data-explorer-widget.directive.ts
@@ -83,6 +83,9 @@ export abstract class BaseDataExplorerWidgetDirective<
     @Input() dataViewDashboardItem: DashboardItem;
     @Input() dataExplorerWidget: T;
 
+    @Input()
+    widgetIndex: number;
+
     @HostBinding('class') className = 'h-100';
 
     public selectedProperties: string[];
@@ -193,16 +196,21 @@ export abstract class BaseDataExplorerWidgetDirective<
         }
         this.timeSelectionSub =
             this.timeSelectionService.timeSelectionChangeSubject.subscribe(
-                ts => {
-                    if (ts) {
-                        this.timeSettings = ts;
-                    } else {
-                        this.timeSelectionService.updateTimeSettings(
-                            this.timeSettings,
-                            new Date(),
-                        );
+                widgetTimeSettings => {
+                    if (
+                        widgetTimeSettings.widgetIndex === undefined ||
+                        widgetTimeSettings.widgetIndex === this.widgetIndex
+                    ) {
+                        if (widgetTimeSettings.timeSettings) {
+                            this.timeSettings = widgetTimeSettings.timeSettings;
+                        } else {
+                            this.timeSelectionService.updateTimeSettings(
+                                this.timeSettings,
+                                new Date(),
+                            );
+                        }
+                        this.updateData();
                     }
-                    this.updateData();
                 },
             );
         this.updateData();
diff --git a/ui/src/app/data-explorer/dialogs/edit-dashboard/data-explorer-edit-dashboard-dialog.component.html b/ui/src/app/data-explorer/dialogs/edit-dashboard/data-explorer-edit-dashboard-dialog.component.html
index de73ccaeff..98b4312145 100644
--- a/ui/src/app/data-explorer/dialogs/edit-dashboard/data-explorer-edit-dashboard-dialog.component.html
+++ b/ui/src/app/data-explorer/dialogs/edit-dashboard/data-explorer-edit-dashboard-dialog.component.html
@@ -67,7 +67,8 @@
                         [(ngModel)]="
                             dashboard.dashboardGeneralSettings.globalTimeEnabled
                         "
-                        >Use global time instead of widget time settings
+                        >Use global time settings instead of widget time
+                        settings
                     </mat-checkbox>
                 </div>
                 <!--<mat-checkbox [(ngModel)]="dashboard.displayHeader">Show name and description in dashboard</mat-checkbox>-->
diff --git a/ui/src/app/data-explorer/models/dataview-dashboard.model.ts b/ui/src/app/data-explorer/models/dataview-dashboard.model.ts
index 670c73a29d..d3597f69db 100644
--- a/ui/src/app/data-explorer/models/dataview-dashboard.model.ts
+++ b/ui/src/app/data-explorer/models/dataview-dashboard.model.ts
@@ -54,6 +54,7 @@ export interface BaseWidgetData<T extends DataExplorerWidgetModel> {
     dataExplorerWidget: T;
     previewMode: boolean;
     gridMode: boolean;
+    widgetIndex?: number;
 
     cleanupSubscriptions(): void;
 }
diff --git a/ui/src/app/data-explorer/services/time-selection.service.ts b/ui/src/app/data-explorer/services/time-selection.service.ts
index 11a9515cfc..5c55e7756a 100644
--- a/ui/src/app/data-explorer/services/time-selection.service.ts
+++ b/ui/src/app/data-explorer/services/time-selection.service.ts
@@ -23,6 +23,7 @@ import {
     QuickTimeSelection,
     TimeSelectionId,
     TimeSettings,
+    WidgetTimeSettings,
 } from '@streampipes/platform-services';
 import {
     startOfDay,
@@ -178,10 +179,11 @@ export class TimeSelectionService {
         }
     }
 
-    public timeSelectionChangeSubject: Subject<TimeSettings | undefined> =
-        new Subject<TimeSettings | undefined>();
+    public timeSelectionChangeSubject: Subject<WidgetTimeSettings | undefined> =
+        new Subject<WidgetTimeSettings | undefined>();
 
-    public notify(timeSettings?: TimeSettings): void {
-        this.timeSelectionChangeSubject.next(timeSettings);
+    public notify(timeSettings?: TimeSettings, widgetIndex?: number): void {
+        const widgetTimeSettings = { timeSettings, widgetIndex };
+        this.timeSelectionChangeSubject.next(widgetTimeSettings);
     }
 }


Test patch:
diff --git a/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/PropertyDuplicateFilterTest.java b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/PropertyDuplicateFilterTest.java
new file mode 100644
index 0000000000..0b9acb8a62
--- /dev/null
+++ b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/PropertyDuplicateFilterTest.java
@@ -0,0 +1,63 @@
+/*
+ * 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.influx;
+
+import org.apache.streampipes.model.runtime.field.PrimitiveField;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class PropertyDuplicateFilterTest {
+
+  @Test
+  public void testDuplicate() {
+    var filter = new PropertyDuplicateFilter(true);
+    var field = new PrimitiveField(null, null, 1);
+    var result = filter.shouldIgnoreField("test", field);
+
+    assertFalse(result);
+
+    field = new PrimitiveField(null, null, 2);
+    result = filter.shouldIgnoreField("test", field);
+    assertFalse(result);
+
+    field = new PrimitiveField(null, null, 2);
+    result = filter.shouldIgnoreField("test1", field);
+    assertFalse(result);
+
+    field = new PrimitiveField(null, null, 2);
+    result = filter.shouldIgnoreField("test", field);
+    assertTrue(result);
+  }
+
+  @Test
+  public void testDuplicateDeactivated() {
+    var filter = new PropertyDuplicateFilter(false);
+    var field = new PrimitiveField(null, null, 1);
+    var result = filter.shouldIgnoreField("test", field);
+
+    assertFalse(result);
+
+    field = new PrimitiveField(null, null, 1);
+    result = filter.shouldIgnoreField("test", field);
+    assertFalse(result);
+  }
+}
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/test/java/org/apache/streampipes/sinks/internal/jvm/datalake/migrations/DataLakeSinkMigrationV2Test.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/test/java/org/apache/streampipes/sinks/internal/jvm/datalake/migrations/DataLakeSinkMigrationV2Test.java
new file mode 100644
index 0000000000..3a3e2e272f
--- /dev/null
+++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/test/java/org/apache/streampipes/sinks/internal/jvm/datalake/migrations/DataLakeSinkMigrationV2Test.java
@@ -0,0 +1,95 @@
+/*
+ * 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.sinks.internal.jvm.datalake.migrations;
+
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.migration.MigrationResult;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.model.staticproperty.RuntimeResolvableAnyStaticProperty;
+import org.apache.streampipes.sdk.builder.PrimitivePropertyBuilder;
+import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
+import org.apache.streampipes.sdk.utils.Datatypes;
+import org.apache.streampipes.sinks.internal.jvm.datalake.DataLakeSink;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.mockito.Mockito.mock;
+
+public class DataLakeSinkMigrationV2Test {
+
+  @Test
+  public void migrate() {
+    var dataLakeSinkMigrationV2 = new DataLakeSinkMigrationV2();
+
+    var stream = new SpDataStream();
+    stream.setEventSchema(makeSchema());
+    var extractor = mock(DataSinkParameterExtractor.class);
+    var invocation = new DataSinkInvocation();
+    invocation.setStaticProperties(new ArrayList<>());
+    invocation.setInputStreams(List.of(stream));
+
+    var actual = dataLakeSinkMigrationV2.migrate(invocation, extractor);
+
+    Assertions.assertTrue(actual.success());
+    Assertions.assertEquals(actual.element()
+        .getStaticProperties()
+        .size(), 2);
+    var dimensionConfig = getAnyStaticProperty(actual);
+    Assertions.assertEquals(dimensionConfig.getInternalName(), DataLakeSink.DIMENSIONS_KEY);
+    Assertions.assertEquals(dimensionConfig.getOptions().size(), 3);
+    Assertions.assertTrue(dimensionConfig.getOptions().get(0).isSelected());
+    Assertions.assertFalse(dimensionConfig.getOptions().get(1).isSelected());
+    Assertions.assertFalse(dimensionConfig.getOptions().get(2).isSelected());
+  }
+
+  private static RuntimeResolvableAnyStaticProperty getAnyStaticProperty(MigrationResult<DataSinkInvocation> actual) {
+    return (RuntimeResolvableAnyStaticProperty) actual.element()
+        .getStaticProperties()
+        .get(0);
+  }
+
+  private static EventSchema makeSchema() {
+    return new EventSchema(
+        List.of(
+            PrimitivePropertyBuilder
+                .create(Datatypes.String, "a")
+                .scope(PropertyScope.DIMENSION_PROPERTY)
+                .build(),
+            PrimitivePropertyBuilder
+                .create(Datatypes.Float, "b")
+                .scope(PropertyScope.MEASUREMENT_PROPERTY)
+                .build(),
+            PrimitivePropertyBuilder
+                .create(Datatypes.Integer, "c")
+                .scope(PropertyScope.MEASUREMENT_PROPERTY)
+                .build(),
+            PrimitivePropertyBuilder
+                .create(Datatypes.Boolean, "d")
+                .scope(PropertyScope.MEASUREMENT_PROPERTY)
+                .build()
+        )
+    );
+  }
+}