Repository: apache/streampipes
Pull request #2766 (2024-04-23 14:19:56)

refactor: replace `StorageManager` by `StorageDispatcher`
<!--
  ~ 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/getinvolved.html
    - 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.
-->

### 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): yes

PR introduces (a) deprecation(s): no

#### Breaking Changes

- removes `StorageManager`. Pleaase use `StorageDispatcher` instead.



Patch:
diff --git a/.github/workflows/pr-validation.yml b/.github/workflows/pr-validation.yml
index 350e9ef559..0e0002698d 100644
--- a/.github/workflows/pr-validation.yml
+++ b/.github/workflows/pr-validation.yml
@@ -142,7 +142,9 @@ jobs:
       fail-fast: false
       matrix:
         python: [3.8, 3.9, "3.10", "3.11"]
-        os: [ubuntu-latest, macos-latest, windows-latest]
+        # temporarily stick to macos13 until Python 3.8 and 3.9 are available for macos-latest again:
+        # https://github.com/actions/setup-python/issues/850#issuecomment-2072657610
+        os: [ubuntu-latest, macos-13, windows-latest]
     defaults:
       run:
         shell: bash
diff --git a/pom.xml b/pom.xml
index 7a38fc7449..db897491a9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -52,7 +52,7 @@
         <commons-io.version>2.16.1</commons-io.version>
         <commons-lang3.version>3.14.0</commons-lang3.version>
         <commons-pool2.version>2.12.0</commons-pool2.version>
-        <commons-text.version>1.11.0</commons-text.version>
+        <commons-text.version>1.12.0</commons-text.version>
         <ditto-client.version>1.0.0</ditto-client.version>
         <eclipse.milo.version>0.6.9</eclipse.milo.version>
         <file-management.version>3.1.0</file-management.version>
@@ -134,7 +134,7 @@
         <!-- Maven settings & dependencies -->
         <apache-rat-plugin.version>0.13</apache-rat-plugin.version>
         <checksum-maven-plugin.version>1.11</checksum-maven-plugin.version>
-        <checkstyle.version>10.12.1</checkstyle.version>
+        <checkstyle.version>10.15.0</checkstyle.version>
         <extra-enforcer-rules.version>1.8.0</extra-enforcer-rules.version>
         <formatter-maven-plugin.version>2.23.0</formatter-maven-plugin.version>
         <maven.compiler.source>17</maven.compiler.source>
diff --git a/streampipes-client-python/poetry.lock b/streampipes-client-python/poetry.lock
index 4b5fa170cc..ecda35ffcd 100644
--- a/streampipes-client-python/poetry.lock
+++ b/streampipes-client-python/poetry.lock
@@ -1670,38 +1670,38 @@ files = [
 
 [[package]]
 name = "mypy"
-version = "1.9.0"
+version = "1.10.0"
 description = "Optional static typing for Python"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "mypy-1.9.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:f8a67616990062232ee4c3952f41c779afac41405806042a8126fe96e098419f"},
-    {file = "mypy-1.9.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:d357423fa57a489e8c47b7c85dfb96698caba13d66e086b412298a1a0ea3b0ed"},
-    {file = "mypy-1.9.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:49c87c15aed320de9b438ae7b00c1ac91cd393c1b854c2ce538e2a72d55df150"},
-    {file = "mypy-1.9.0-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:48533cdd345c3c2e5ef48ba3b0d3880b257b423e7995dada04248725c6f77374"},
-    {file = "mypy-1.9.0-cp310-cp310-win_amd64.whl", hash = "sha256:4d3dbd346cfec7cb98e6cbb6e0f3c23618af826316188d587d1c1bc34f0ede03"},
-    {file = "mypy-1.9.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:653265f9a2784db65bfca694d1edd23093ce49740b2244cde583aeb134c008f3"},
-    {file = "mypy-1.9.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:3a3c007ff3ee90f69cf0a15cbcdf0995749569b86b6d2f327af01fd1b8aee9dc"},
-    {file = "mypy-1.9.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2418488264eb41f69cc64a69a745fad4a8f86649af4b1041a4c64ee61fc61129"},
-    {file = "mypy-1.9.0-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:68edad3dc7d70f2f17ae4c6c1b9471a56138ca22722487eebacfd1eb5321d612"},
-    {file = "mypy-1.9.0-cp311-cp311-win_amd64.whl", hash = "sha256:85ca5fcc24f0b4aeedc1d02f93707bccc04733f21d41c88334c5482219b1ccb3"},
-    {file = "mypy-1.9.0-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:aceb1db093b04db5cd390821464504111b8ec3e351eb85afd1433490163d60cd"},
-    {file = "mypy-1.9.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:0235391f1c6f6ce487b23b9dbd1327b4ec33bb93934aa986efe8a9563d9349e6"},
-    {file = "mypy-1.9.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d4d5ddc13421ba3e2e082a6c2d74c2ddb3979c39b582dacd53dd5d9431237185"},
-    {file = "mypy-1.9.0-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:190da1ee69b427d7efa8aa0d5e5ccd67a4fb04038c380237a0d96829cb157913"},
-    {file = "mypy-1.9.0-cp312-cp312-win_amd64.whl", hash = "sha256:fe28657de3bfec596bbeef01cb219833ad9d38dd5393fc649f4b366840baefe6"},
-    {file = "mypy-1.9.0-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:e54396d70be04b34f31d2edf3362c1edd023246c82f1730bbf8768c28db5361b"},
-    {file = "mypy-1.9.0-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:5e6061f44f2313b94f920e91b204ec600982961e07a17e0f6cd83371cb23f5c2"},
-    {file = "mypy-1.9.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:81a10926e5473c5fc3da8abb04119a1f5811a236dc3a38d92015cb1e6ba4cb9e"},
-    {file = "mypy-1.9.0-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:b685154e22e4e9199fc95f298661deea28aaede5ae16ccc8cbb1045e716b3e04"},
-    {file = "mypy-1.9.0-cp38-cp38-win_amd64.whl", hash = "sha256:5d741d3fc7c4da608764073089e5f58ef6352bedc223ff58f2f038c2c4698a89"},
-    {file = "mypy-1.9.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:587ce887f75dd9700252a3abbc9c97bbe165a4a630597845c61279cf32dfbf02"},
-    {file = "mypy-1.9.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:f88566144752999351725ac623471661c9d1cd8caa0134ff98cceeea181789f4"},
-    {file = "mypy-1.9.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:61758fabd58ce4b0720ae1e2fea5cfd4431591d6d590b197775329264f86311d"},
-    {file = "mypy-1.9.0-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:e49499be624dead83927e70c756970a0bc8240e9f769389cdf5714b0784ca6bf"},
-    {file = "mypy-1.9.0-cp39-cp39-win_amd64.whl", hash = "sha256:571741dc4194b4f82d344b15e8837e8c5fcc462d66d076748142327626a1b6e9"},
-    {file = "mypy-1.9.0-py3-none-any.whl", hash = "sha256:a260627a570559181a9ea5de61ac6297aa5af202f06fd7ab093ce74e7181e43e"},
-    {file = "mypy-1.9.0.tar.gz", hash = "sha256:3cc5da0127e6a478cddd906068496a97a7618a21ce9b54bde5bf7e539c7af974"},
+    {file = "mypy-1.10.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:da1cbf08fb3b851ab3b9523a884c232774008267b1f83371ace57f412fe308c2"},
+    {file = "mypy-1.10.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:12b6bfc1b1a66095ab413160a6e520e1dc076a28f3e22f7fb25ba3b000b4ef99"},
+    {file = "mypy-1.10.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:9e36fb078cce9904c7989b9693e41cb9711e0600139ce3970c6ef814b6ebc2b2"},
+    {file = "mypy-1.10.0-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:2b0695d605ddcd3eb2f736cd8b4e388288c21e7de85001e9f85df9187f2b50f9"},
+    {file = "mypy-1.10.0-cp310-cp310-win_amd64.whl", hash = "sha256:cd777b780312ddb135bceb9bc8722a73ec95e042f911cc279e2ec3c667076051"},
+    {file = "mypy-1.10.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:3be66771aa5c97602f382230165b856c231d1277c511c9a8dd058be4784472e1"},
+    {file = "mypy-1.10.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:8b2cbaca148d0754a54d44121b5825ae71868c7592a53b7292eeb0f3fdae95ee"},
+    {file = "mypy-1.10.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:1ec404a7cbe9fc0e92cb0e67f55ce0c025014e26d33e54d9e506a0f2d07fe5de"},
+    {file = "mypy-1.10.0-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:e22e1527dc3d4aa94311d246b59e47f6455b8729f4968765ac1eacf9a4760bc7"},
+    {file = "mypy-1.10.0-cp311-cp311-win_amd64.whl", hash = "sha256:a87dbfa85971e8d59c9cc1fcf534efe664d8949e4c0b6b44e8ca548e746a8d53"},
+    {file = "mypy-1.10.0-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:a781f6ad4bab20eef8b65174a57e5203f4be627b46291f4589879bf4e257b97b"},
+    {file = "mypy-1.10.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:b808e12113505b97d9023b0b5e0c0705a90571c6feefc6f215c1df9381256e30"},
+    {file = "mypy-1.10.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:8f55583b12156c399dce2df7d16f8a5095291354f1e839c252ec6c0611e86e2e"},
+    {file = "mypy-1.10.0-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:4cf18f9d0efa1b16478c4c129eabec36148032575391095f73cae2e722fcf9d5"},
+    {file = "mypy-1.10.0-cp312-cp312-win_amd64.whl", hash = "sha256:bc6ac273b23c6b82da3bb25f4136c4fd42665f17f2cd850771cb600bdd2ebeda"},
+    {file = "mypy-1.10.0-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:9fd50226364cd2737351c79807775136b0abe084433b55b2e29181a4c3c878c0"},
+    {file = "mypy-1.10.0-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:f90cff89eea89273727d8783fef5d4a934be2fdca11b47def50cf5d311aff727"},
+    {file = "mypy-1.10.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:fcfc70599efde5c67862a07a1aaf50e55bce629ace26bb19dc17cece5dd31ca4"},
+    {file = "mypy-1.10.0-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:075cbf81f3e134eadaf247de187bd604748171d6b79736fa9b6c9685b4083061"},
+    {file = "mypy-1.10.0-cp38-cp38-win_amd64.whl", hash = "sha256:3f298531bca95ff615b6e9f2fc0333aae27fa48052903a0ac90215021cdcfa4f"},
+    {file = "mypy-1.10.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:fa7ef5244615a2523b56c034becde4e9e3f9b034854c93639adb667ec9ec2976"},
+    {file = "mypy-1.10.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:3236a4c8f535a0631f85f5fcdffba71c7feeef76a6002fcba7c1a8e57c8be1ec"},
+    {file = "mypy-1.10.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:4a2b5cdbb5dd35aa08ea9114436e0d79aceb2f38e32c21684dcf8e24e1e92821"},
+    {file = "mypy-1.10.0-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:92f93b21c0fe73dc00abf91022234c79d793318b8a96faac147cd579c1671746"},
+    {file = "mypy-1.10.0-cp39-cp39-win_amd64.whl", hash = "sha256:28d0e038361b45f099cc086d9dd99c15ff14d0188f44ac883010e172ce86c38a"},
+    {file = "mypy-1.10.0-py3-none-any.whl", hash = "sha256:f8c083976eb530019175aabadb60921e73b4f45736760826aa1689dda8208aee"},
+    {file = "mypy-1.10.0.tar.gz", hash = "sha256:3d087fcbec056c4ee34974da493a826ce316947485cef3901f511848e687c131"},
 ]
 
 [package.dependencies]
@@ -3826,4 +3826,4 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p
 [metadata]
 lock-version = "2.0"
 python-versions = ">3.8.1,<3.12"
-content-hash = "abad1d1443750976a1a067c0fecbf0042c66fa9d4d2754a0b980975e6a15e110"
+content-hash = "73f0e6358363576f9f83703189f190e2a88cadfa19438aad6af1a9c4ba34e2d4"
diff --git a/streampipes-client-python/pyproject.toml b/streampipes-client-python/pyproject.toml
index 01469bfbad..874aa59b5b 100644
--- a/streampipes-client-python/pyproject.toml
+++ b/streampipes-client-python/pyproject.toml
@@ -71,7 +71,7 @@ blacken-docs = "1.16.0"
 flake8 = "6.1.0"
 interrogate = { version = "1.7.0", extras = ["png"] }
 isort = "5.13.0"
-mypy = "1.9.0"
+mypy = "1.10.0"
 ruff = "0.4.0"
 pre-commit = "3.5.0"
 pytest = "8.1.1"
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerQueryManagement.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerQueryManagement.java
index 07c5058b6c..426262f1cb 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerQueryManagement.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerQueryManagement.java
@@ -18,33 +18,22 @@
 
 package org.apache.streampipes.dataexplorer;
 
-import org.apache.streampipes.commons.environment.Environment;
-import org.apache.streampipes.commons.environment.Environments;
 import org.apache.streampipes.dataexplorer.api.IDataExplorerQueryManagement;
 import org.apache.streampipes.dataexplorer.api.IDataExplorerSchemaManagement;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
 import org.apache.streampipes.dataexplorer.influx.DataExplorerInfluxQueryExecutor;
 import org.apache.streampipes.dataexplorer.param.DeleteQueryParams;
 import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParamConverter;
 import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
-import org.apache.streampipes.dataexplorer.query.DeleteDataQuery;
 import org.apache.streampipes.dataexplorer.query.QueryResultProvider;
 import org.apache.streampipes.dataexplorer.query.StreamedQueryResultProvider;
 import org.apache.streampipes.dataexplorer.query.writer.OutputFormat;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.datalake.SpQueryResult;
 
-import org.influxdb.InfluxDB;
-import org.influxdb.dto.Query;
-import org.influxdb.dto.QueryResult;
-
 import java.io.IOException;
 import java.io.OutputStream;
-import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 public class DataExplorerQueryManagement implements IDataExplorerQueryManagement {
 
@@ -72,10 +61,11 @@ public class DataExplorerQueryManagement implements IDataExplorerQueryManagement
   @Override
   public boolean deleteAllData() {
     List<DataLakeMeasure> allMeasurements = getAllMeasurements();
+    var queryExecutor = new DataExplorerInfluxQueryExecutor();
 
     for (DataLakeMeasure measure : allMeasurements) {
-      QueryResult queryResult = new DeleteDataQuery(measure).executeQuery();
-      if (queryResult.hasError() || queryResult.getResults().get(0).getError() != null) {
+      boolean success = queryExecutor.deleteData(measure);
+      if (!success) {
         return false;
       }
     }
@@ -85,14 +75,13 @@ public class DataExplorerQueryManagement implements IDataExplorerQueryManagement
   @Override
   public boolean deleteData(String measurementID) {
     List<DataLakeMeasure> allMeasurements = getAllMeasurements();
-    for (DataLakeMeasure measure : allMeasurements) {
-      if (measure.getMeasureName().equals(measurementID)) {
-        QueryResult queryResult = new DeleteDataQuery(new DataLakeMeasure(measurementID, null)).executeQuery();
 
-        return !queryResult.hasError();
-      }
-    }
-    return false;
+    var measureToDeleteOpt = allMeasurements.stream()
+                                            .filter(measure -> measure.getMeasureName().equals(measurementID))
+                                            .findFirst();
+
+    return measureToDeleteOpt.filter(measure -> new DataExplorerInfluxQueryExecutor().deleteData(measure))
+                             .isPresent();
   }
 
   @Override
@@ -105,38 +94,10 @@ public class DataExplorerQueryManagement implements IDataExplorerQueryManagement
   @Override
   public Map<String, Object> getTagValues(String measurementId,
                                           String fields) {
-    InfluxDB influxDB = InfluxClientProvider.getInfluxDBClient();
-    String databaseName = getEnvironment().getTsStorageBucket().getValueOrDefault();
-    Map<String, Object> tags = new HashMap<>();
-    if (fields != null && !(fields.isEmpty())) {
-      List<String> fieldList = Arrays.asList(fields.split(","));
-      fieldList.forEach(f -> {
-        String q =
-            "SHOW TAG VALUES ON \"" + databaseName + "\" FROM \"" + measurementId
-                + "\" WITH KEY = \"" + f + "\"";
-        Query query = new Query(q);
-        QueryResult queryResult = influxDB.query(query);
-        queryResult.getResults().forEach(res -> {
-          res.getSeries().forEach(series -> {
-            if (!series.getValues().isEmpty()) {
-              String field = series.getValues().get(0).get(0).toString();
-              List<String> values =
-                  series.getValues().stream().map(v -> v.get(1).toString()).collect(Collectors.toList());
-              tags.put(field, values);
-            }
-          });
-        });
-      });
-    }
-
-    return tags;
+    return new DataExplorerInfluxQueryExecutor().getTagValues(measurementId, fields);
   }
 
   private List<DataLakeMeasure> getAllMeasurements() {
     return this.dataExplorerSchemaManagement.getAllMeasurements();
   }
-
-  private Environment getEnvironment() {
-    return Environments.getEnvironment();
-  }
 }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java
index dbb8a9a0b2..822d5c9f23 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java
@@ -18,12 +18,13 @@
 
 package org.apache.streampipes.dataexplorer.influx;
 
-import org.apache.streampipes.commons.environment.Environments;
 import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
 import org.apache.streampipes.dataexplorer.param.DeleteQueryParams;
 import org.apache.streampipes.dataexplorer.param.SelectQueryParams;
 import org.apache.streampipes.dataexplorer.query.DataExplorerQueryExecutor;
+import org.apache.streampipes.dataexplorer.query.DeleteDataQuery;
 import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.datalake.DataSeries;
 import org.apache.streampipes.model.datalake.SpQueryResult;
 
@@ -32,9 +33,15 @@ import org.influxdb.dto.Query;
 import org.influxdb.dto.QueryResult;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+import static org.apache.streampipes.commons.environment.Environments.getEnvironment;
 
 public class DataExplorerInfluxQueryExecutor extends DataExplorerQueryExecutor<Query, QueryResult> {
 
@@ -114,12 +121,12 @@ public class DataExplorerInfluxQueryExecutor extends DataExplorerQueryExecutor<Q
 
   @Override
   protected Query makeDeleteQuery(DeleteQueryParams params) {
-    String query = "DELETE FROM \"" + params.getMeasurementId() + "\"";
-    if (params.isTimeRestricted()) {
+    String query = "DELETE FROM \"" + params.measurementId() + "\"";
+    if (params.timeRestricted()) {
       query += "WHERE time > "
-          + params.getStartTime() * 1000000
+          + params.startTime() * 1000000
           + " AND time < "
-          + params.getEndTime() * 1000000;
+          + params.endTime() * 1000000;
     }
     return new Query(query, getDatabaseName());
   }
@@ -142,6 +149,43 @@ public class DataExplorerInfluxQueryExecutor extends DataExplorerQueryExecutor<Q
   }
 
   private String getDatabaseName() {
-    return Environments.getEnvironment().getTsStorageBucket().getValueOrDefault();
+    return getEnvironment().getTsStorageBucket().getValueOrDefault();
+  }
+
+  public Map<String, Object> getTagValues(String measurementId, String fields) {
+    try (final InfluxDB influxDB = InfluxClientProvider.getInfluxDBClient()) {
+      Map<String, Object> tags = new HashMap<>();
+      if (fields != null && !(fields.isEmpty())) {
+        List<String> fieldList = Arrays.asList(fields.split(","));
+        fieldList.forEach(f -> {
+          String q =
+              "SHOW TAG VALUES ON \"" + getDatabaseName() + "\" FROM \"" + measurementId
+              + "\" WITH KEY = \"" + f + "\"";
+          Query query = new Query(q);
+          QueryResult queryResult = influxDB.query(query);
+          queryResult.getResults().forEach(res -> {
+            res.getSeries().forEach(series -> {
+              if (!series.getValues().isEmpty()) {
+                String field = series.getValues().get(0).get(0).toString();
+                List<String> values =
+                    series.getValues().stream().map(v -> v.get(1).toString()).collect(Collectors.toList());
+                tags.put(field, values);
+              }
+            });
+          });
+        });
+      }
+
+      return tags;
+    }
+  }
+
+  public boolean deleteData(DataLakeMeasure measure) {
+    QueryResult queryResult = new DeleteDataQuery(measure).executeQuery();
+
+    return !queryResult.hasError() && (queryResult.getResults() == null || queryResult.getResults()
+                                                                                      .get(0)
+                                                                                      .getError() == null);
+
   }
 }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
index 8c968ca885..abed48684e 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
@@ -177,13 +177,6 @@ public class DataLakeInfluxQueryBuilder implements IDataLakeQueryBuilder<Query>
     return this;
   }
 
-  @Override
-  public DataLakeInfluxQueryBuilder withFilter(NestedClause clause) {
-    this.whereClauses.add(clause);
-
-    return this;
-  }
-
   @Override
   public DataLakeInfluxQueryBuilder withGroupByTime(String timeInterval) {
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/DeleteQueryParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/DeleteQueryParams.java
index 8bdda43de4..33d3634903 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/DeleteQueryParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/DeleteQueryParams.java
@@ -18,43 +18,11 @@
 
 package org.apache.streampipes.dataexplorer.param;
 
-public class DeleteQueryParams {
 
-  private final String measurementId;
+public record DeleteQueryParams(String measurementId, Long startTime, Long endTime, boolean timeRestricted) {
 
-  private long startTime;
-
-  private long endTime;
-
-  private boolean timeRestricted;
-
-  public DeleteQueryParams(String measurementId) {
-    this.measurementId = measurementId;
-    this.timeRestricted = false;
-  }
-
-  public DeleteQueryParams(String measurementId,
-                           Long startTime,
-                           Long endTime) {
-    this(measurementId);
-    this.startTime = startTime;
-    this.endTime = endTime;
-    this.timeRestricted = true;
-  }
-
-  public String getMeasurementId() {
-    return measurementId;
-  }
-
-  public long getStartTime() {
-    return startTime;
-  }
-
-  public long getEndTime() {
-    return endTime;
-  }
-
-  public boolean isTimeRestricted() {
-    return timeRestricted;
+  public DeleteQueryParams(String measurementId, Long startTime, Long endTime) {
+    this(measurementId, startTime != null ? startTime : 0, endTime != null ? endTime : 99999999999999L, true);
   }
 }
+
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java
index 9b6f341dd7..186a5eb74c 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java
@@ -111,13 +111,9 @@ public class ProvidedRestQueryParamConverter {
   }
 
   public static DeleteQueryParams getDeleteQueryParams(String measurementId,
-                                                                    Long startTime,
-                                                                    Long endTime) {
-    if (startTime != null || endTime != null) {
-      return new DeleteQueryParams(measurementId, startTime, endTime);
-    } else {
-      return new DeleteQueryParams(measurementId);
-    }
+                                                       Long startTime,
+                                                       Long endTime) {
+    return new DeleteQueryParams(measurementId, startTime, endTime);
   }
 
   private static boolean hasTimeParams(ProvidedRestQueryParams params) {
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/SelectQueryParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/SelectQueryParams.java
index be5392d8c3..4131d9bf22 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/SelectQueryParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/SelectQueryParams.java
@@ -95,7 +95,7 @@ public class SelectQueryParams {
 
   public int getLimit() {
     if (Objects.nonNull(limitParams)) {
-      return limitParams.getLimit();
+      return limitParams.limit();
     } else {
       return Integer.MIN_VALUE;
     }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/LimitClauseParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/LimitClauseParams.java
index bdff3d070d..f1ecef0e77 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/LimitClauseParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/LimitClauseParams.java
@@ -21,22 +21,12 @@ package org.apache.streampipes.dataexplorer.param.model;
 import org.apache.streampipes.dataexplorer.api.IQueryStatement;
 import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
 
-public class LimitClauseParams implements IQueryStatement {
-
-  private final Integer limit;
-
-  public LimitClauseParams(Integer limit) {
-    this.limit = limit;
-  }
+public record LimitClauseParams(Integer limit) implements IQueryStatement {
 
   public static LimitClauseParams from(Integer limit) {
     return new LimitClauseParams(limit);
   }
 
-  public Integer getLimit() {
-    return limit;
-  }
-
   @Override
   public void buildStatement(IDataLakeQueryBuilder<?> builder) {
     builder.withLimit(limit);
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/OffsetClauseParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/OffsetClauseParams.java
index b8dbc94aea..6419b7a73e 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/OffsetClauseParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/OffsetClauseParams.java
@@ -21,22 +21,12 @@ package org.apache.streampipes.dataexplorer.param.model;
 import org.apache.streampipes.dataexplorer.api.IQueryStatement;
 import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
 
-public class OffsetClauseParams implements IQueryStatement {
-
-  private final Integer offset;
-
-  public OffsetClauseParams(Integer offset) {
-    this.offset = offset;
-  }
+public record OffsetClauseParams(Integer offset) implements IQueryStatement {
 
   public static OffsetClauseParams from(Integer offset) {
     return new OffsetClauseParams(offset);
   }
 
-  public Integer getOffset() {
-    return offset;
-  }
-
   @Override
   public void buildStatement(IDataLakeQueryBuilder<?> builder) {
     builder.withOffset(offset);
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/SelectClauseParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/SelectClauseParams.java
index fe7f8d0f65..02db7227f2 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/SelectClauseParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/SelectClauseParams.java
@@ -32,10 +32,6 @@ public class SelectClauseParams implements IQueryStatement {
   private List<SelectColumn> selectedColumnsCountOnly;
   private boolean selectWildcard = false;
 
-  public SelectClauseParams() {
-    this.selectWildcard = true;
-  }
-
   public SelectClauseParams(String columns,
                             boolean countOnly) {
     this.selectWildcard = false;
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java
index 205046303a..f5e12bb247 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java
@@ -18,7 +18,6 @@
 
 package org.apache.streampipes.dataexplorer.query;
 
-import org.apache.streampipes.dataexplorer.influx.DataExplorerInfluxQueryExecutor;
 import org.apache.streampipes.dataexplorer.param.DeleteQueryParams;
 import org.apache.streampipes.dataexplorer.param.SelectQueryParams;
 import org.apache.streampipes.model.datalake.DataSeries;
@@ -30,7 +29,7 @@ import org.slf4j.LoggerFactory;
 
 public abstract class DataExplorerQueryExecutor<X, W> {
 
-  private static final Logger LOG = LoggerFactory.getLogger(DataExplorerInfluxQueryExecutor.class);
+  private static final Logger LOG = LoggerFactory.getLogger(DataExplorerQueryExecutor.class);
   protected int maximumAmountOfEvents;
 
   protected boolean appendId = false;
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredJsonOutputWriter.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredJsonOutputWriter.java
index e1311c9e47..e14e866e05 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredJsonOutputWriter.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredJsonOutputWriter.java
@@ -22,7 +22,7 @@ import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
 import org.apache.streampipes.dataexplorer.query.writer.item.ItemGenerator;
 import org.apache.streampipes.dataexplorer.query.writer.item.JsonItemGenerator;
 
-import com.google.gson.Gson;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
 import java.io.IOException;
 import java.io.OutputStream;
@@ -36,8 +36,7 @@ public class ConfiguredJsonOutputWriter extends ConfiguredOutputWriter {
   private final ItemGenerator itemGenerator;
 
   public ConfiguredJsonOutputWriter() {
-    Gson gson = new Gson();
-    this.itemGenerator = new JsonItemGenerator(gson);
+    this.itemGenerator = new JsonItemGenerator(new ObjectMapper());
   }
 
   @Override
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemGenerator.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemGenerator.java
index e135bb8cdf..03d37197f0 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemGenerator.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemGenerator.java
@@ -19,28 +19,35 @@
 
 package org.apache.streampipes.dataexplorer.query.writer.item;
 
-import com.google.gson.Gson;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class JsonItemGenerator extends ItemGenerator {
 
+  private static final Logger LOG = LoggerFactory.getLogger(JsonItemGenerator.class);
+
   private static final String BEGIN_OBJECT = "{";
   private static final String END_OBJECT = "}";
 
-  private final Gson gson;
+  private final ObjectMapper objectMapper;
 
-  public JsonItemGenerator(Gson gson) {
+  public JsonItemGenerator(ObjectMapper objectMapper) {
     super(COMMA_SEPARATOR);
-    this.gson = gson;
+    this.objectMapper = objectMapper;
   }
 
   @Override
   protected String makeItemString(String key,
                                   Object value) {
-    var stringValue = value != null ? gson.toJson(value) : null;
-    return "\""
-        + key
-        + "\": "
-        + stringValue;
+    String valueJsonString = null;
+    try {
+      valueJsonString = value != null ? this.objectMapper.writeValueAsString(value) : null;
+    } catch (JsonProcessingException e) {
+      LOG.error("Error while converting value to JSON string: {}", e.getMessage());
+    }
+    return "\"%s\": %s".formatted(key, valueJsonString);
   }
 
   @Override
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/IDataLakeQueryBuilder.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/IDataLakeQueryBuilder.java
index 065c185621..3298a8ed4d 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/IDataLakeQueryBuilder.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/IDataLakeQueryBuilder.java
@@ -20,8 +20,6 @@ package org.apache.streampipes.dataexplorer.querybuilder;
 
 import org.apache.streampipes.dataexplorer.param.model.AggregationFunction;
 
-import org.influxdb.querybuilder.clauses.NestedClause;
-
 import java.util.List;
 
 public interface IDataLakeQueryBuilder<T> {
@@ -63,8 +61,6 @@ public interface IDataLakeQueryBuilder<T> {
 
   IDataLakeQueryBuilder<T> withInclusiveFilter(List<FilterCondition> filterConditions);
 
-  IDataLakeQueryBuilder<T> withFilter(NestedClause clause);
-
   IDataLakeQueryBuilder<T> withGroupByTime(String timeInterval);
 
   IDataLakeQueryBuilder<T> withGroupByTime(String timeInterval,
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/recommender/ElementRecommender.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/recommender/ElementRecommender.java
index 92edbee27c..6cb0911b27 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/recommender/ElementRecommender.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/recommender/ElementRecommender.java
@@ -36,7 +36,6 @@ import org.apache.streampipes.resource.management.SpResourceManager;
 import org.apache.streampipes.storage.api.INoSqlStorage;
 import org.apache.streampipes.storage.api.IPipelineElementDescriptionStorage;
 import org.apache.streampipes.storage.management.StorageDispatcher;
-import org.apache.streampipes.storage.management.StorageManager;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -185,7 +184,7 @@ public class ElementRecommender {
   }
 
   private IPipelineElementDescriptionStorage getTripleStore() {
-    return StorageManager.INSTANCE.getPipelineElementStorage();
+    return StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineElementDescriptionStorage();
   }
 
   private INoSqlStorage getNoSqlStorage() {
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineGenerator.java
index 19cec5d904..134e37a1d2 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineGenerator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineGenerator.java
@@ -27,7 +27,7 @@ import org.apache.streampipes.model.pipeline.Pipeline;
 import org.apache.streampipes.model.pipeline.PipelineModification;
 import org.apache.streampipes.model.template.BoundPipelineElement;
 import org.apache.streampipes.model.template.PipelineTemplateDescription;
-import org.apache.streampipes.storage.management.StorageManager;
+import org.apache.streampipes.storage.management.StorageDispatcher;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -126,10 +126,9 @@ public class PipelineGenerator {
   }
 
   private SpDataStream getStream(String datasetId) {
-    return StorageManager
-        .INSTANCE
-        .getPipelineElementStorage()
-        .getEventStreamById(datasetId);
+    return StorageDispatcher.INSTANCE.getNoSqlStore()
+                                     .getPipelineElementDescriptionStorage()
+                                     .getEventStreamById(datasetId);
   }
 
 
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineTemplateGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineTemplateGenerator.java
index 235e1b7e18..4531fc88f2 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineTemplateGenerator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineTemplateGenerator.java
@@ -29,7 +29,7 @@ import org.apache.streampipes.model.graph.DataSinkDescription;
 import org.apache.streampipes.model.graph.DataSinkInvocation;
 import org.apache.streampipes.model.template.PipelineTemplateDescription;
 import org.apache.streampipes.storage.api.IPipelineElementDescriptionStorage;
-import org.apache.streampipes.storage.management.StorageManager;
+import org.apache.streampipes.storage.management.StorageDispatcher;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -131,8 +131,7 @@ public class PipelineTemplateGenerator {
   }
 
   protected IPipelineElementDescriptionStorage getStorage() {
-    return StorageManager
-        .INSTANCE
-        .getPipelineElementStorage();
+    return StorageDispatcher.INSTANCE.getNoSqlStore()
+                                     .getPipelineElementDescriptionStorage();
   }
 }
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/ElementVerifier.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/ElementVerifier.java
index 2537057698..797a40bf40 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/ElementVerifier.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/ElementVerifier.java
@@ -36,30 +36,29 @@ import org.apache.streampipes.model.message.SuccessMessage;
 import org.apache.streampipes.resource.management.SpResourceManager;
 import org.apache.streampipes.serializers.json.JacksonSerializer;
 import org.apache.streampipes.storage.api.IPipelineElementDescriptionStorage;
-import org.apache.streampipes.storage.management.StorageManager;
+import org.apache.streampipes.storage.management.StorageDispatcher;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.logging.Logger;
 
 public abstract class ElementVerifier<T extends NamedStreamPipesEntity> {
 
-  protected static final Logger LOGGER = Logger.getAnonymousLogger();
-
   private String graphData;
   private Class<T> elementClass;
-  private boolean shouldTransform;
+  private final boolean shouldTransform;
 
   protected T elementDescription;
 
   protected List<VerificationResult> validationResults;
   protected List<Verifier> validators;
 
-  protected IPipelineElementDescriptionStorage storageApi =
-      StorageManager.INSTANCE.getPipelineElementStorage();
+  protected IPipelineElementDescriptionStorage storageApi = StorageDispatcher
+      .INSTANCE
+      .getNoSqlStore()
+      .getPipelineElementDescriptionStorage();
 
   public ElementVerifier(String graphData, Class<T> elementClass) {
     this.elementClass = elementClass;
@@ -119,7 +118,7 @@ public abstract class ElementVerifier<T extends NamedStreamPipesEntity> {
 
   }
 
-  public Message verifyAndUpdate() throws SepaParseException {
+  public Message verifyAndUpdate() {
     try {
       this.elementDescription = transform();
     } catch (JsonProcessingException e) {
@@ -178,15 +177,19 @@ public abstract class ElementVerifier<T extends NamedStreamPipesEntity> {
   }
 
   private boolean isVerifiedSuccessfully() {
-    return validationResults.stream().noneMatch(validator -> (validator instanceof VerificationError));
+    return validationResults.stream()
+                            .noneMatch(validator -> (validator instanceof VerificationError));
   }
 
   protected T transform() throws JsonProcessingException {
-    return JacksonSerializer.getObjectMapper().readValue(graphData, elementClass);
+    return JacksonSerializer.getObjectMapper()
+                            .readValue(graphData, elementClass);
   }
 
-  private void createAndStorePermission(String principalSid,
-                                        boolean publicElement) {
+  private void createAndStorePermission(
+      String principalSid,
+      boolean publicElement
+  ) {
     Permission permission = makePermission(
         this.elementDescription.getElementId(),
         this.elementDescription.getClass(),
@@ -196,17 +199,20 @@ public abstract class ElementVerifier<T extends NamedStreamPipesEntity> {
     storeNewObjectPermission(permission);
   }
 
-  protected Permission makePermission(String objectInstanceId,
-                                      Class<?> objectInstanceClass,
-                                      String principalSid,
-                                      boolean publicElement) {
+  protected Permission makePermission(
+      String objectInstanceId,
+      Class<?> objectInstanceClass,
+      String principalSid,
+      boolean publicElement
+  ) {
     return PermissionBuilder
-        .create(objectInstanceId, objectInstanceClass, principalSid)
-        .publicElement(publicElement)
-        .build();
+      .create(objectInstanceId, objectInstanceClass, principalSid)
+      .publicElement(publicElement)
+      .build();
   }
 
   protected void storeNewObjectPermission(Permission permission) {
-    new SpResourceManager().managePermissions().create(permission);
+    new SpResourceManager().managePermissions()
+                           .create(permission);
   }
 }
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementCategory.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementCategory.java
index 1b73e6fe93..9e1957e4a8 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementCategory.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementCategory.java
@@ -24,7 +24,7 @@ import org.apache.streampipes.model.DataSinkType;
 import org.apache.streampipes.model.SpDataStream;
 import org.apache.streampipes.model.client.Category;
 import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource;
-import org.apache.streampipes.storage.management.StorageManager;
+import org.apache.streampipes.storage.management.StorageDispatcher;
 
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
@@ -41,7 +41,9 @@ public class PipelineElementCategory extends AbstractAuthGuardedRestResource {
 
   @GetMapping(path = "/ep", produces = MediaType.APPLICATION_JSON_VALUE)
   public ResponseEntity<List<Category>> getEps() {
-    return ok(makeCategories(StorageManager.INSTANCE.getPipelineElementStorage().getAllDataStreams()));
+    return ok(makeCategories(StorageDispatcher.INSTANCE.getNoSqlStore()
+                                                       .getPipelineElementDescriptionStorage()
+                                                       .getAllDataStreams()));
   }
 
   @GetMapping(path = "/epa", produces = MediaType.APPLICATION_JSON_VALUE)
diff --git a/streampipes-storage-management/src/main/java/org/apache/streampipes/storage/management/StorageManager.java b/streampipes-storage-management/src/main/java/org/apache/streampipes/storage/management/StorageManager.java
deleted file mode 100644
index 01e9a58117..0000000000
--- a/streampipes-storage-management/src/main/java/org/apache/streampipes/storage/management/StorageManager.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.storage.management;
-
-import org.apache.streampipes.storage.api.IPipelineElementDescriptionStorage;
-import org.apache.streampipes.storage.couchdb.CouchDbStorageManager;
-
-public enum StorageManager {
-
-  INSTANCE;
-
-  public IPipelineElementDescriptionStorage getPipelineElementStorage() {
-    return CouchDbStorageManager.INSTANCE.getPipelineElementDescriptionStorage();
-  }
-
-
-}


Test patch:
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemGenerator.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemGenerator.java
index 2b3dcb58b1..786570bc83 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemGenerator.java
+++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemGenerator.java
@@ -18,7 +18,7 @@
 
 package org.apache.streampipes.dataexplorer.query.writer.item;
 
-import com.google.gson.Gson;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -29,7 +29,7 @@ public class TestJsonItemGenerator extends TestItemGenerator {
 
   @Test
   public void testJsonWriter() {
-    var writer = new JsonItemGenerator(new Gson());
+    var writer = new JsonItemGenerator(new ObjectMapper());
 
     String result = writer.createItem(row, columns);