This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 585e63fa69 feat: provide basic support for IoTDB as time series
storage (#2835)
585e63fa69 is described below
commit 585e63fa69c61edbcd737b9c8d40db9e23974bd5
Author: Tim <[email protected]>
AuthorDate: Wed May 15 08:58:56 2024 +0200
feat: provide basic support for IoTDB as time series storage (#2835)
* feat: provide basic support for IotDB as time series storage
* dep: align iotdb versions
* update iotdb source
* fix import order
* improve dependency management
* align module name
* remove singletons
* improve writing of values
---
pom.xml | 8 +-
.../influx/DataExplorerManagerInflux.java | 4 +-
.../pom.xml | 29 +++-
.../iotdb/DataExplorerManagerIotDb.java | 33 ++--
.../iotdb/IotDbMeasurementRecord.java | 2 +-
.../iotdb/IotDbPropertyConverter.java | 2 +-
.../dataexplorer}/iotdb/IotDbSessionProvider.java | 2 +-
.../dataexplorer/iotdb/TimeSeriesStorageIotDb.java | 175 +++++++++++++++++++++
.../DataLakeMeasurementSanitizerIotDb.java | 54 +++++++
.../iotdb/sanitize}/IotDbNameSanitizer.java | 2 +-
.../iotdb/sanitize}/IotDbReservedKeywords.java | 2 +-
.../iotdb/sanitize}/MeasureNameSanitizerIotDb.java | 2 +-
.../iotdb/IotDbPropertyConverterTest.java | 4 +-
.../DataLakeMeasurementSanitizerIotDbTest.java | 75 +++++++++
.../iotdb/sanitize}/IotDbNameSanitizerTest.java | 2 +-
.../sanitize/MeasureNameSanitizerIotDbTest.java | 4 +-
streampipes-data-explorer-management/pom.xml | 5 +
.../management/DataExplorerDispatcher.java | 5 +-
.../management/SupportedDataExplorerStorages.java | 1 +
.../streampipes-sinks-databases-jvm/pom.xml | 15 +-
.../sinks/databases/jvm/iotdb/IotDbSink.java | 7 +-
21 files changed, 382 insertions(+), 51 deletions(-)
diff --git a/pom.xml b/pom.xml
index 72bd0ae237..f631832ea6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -67,6 +67,7 @@
<httpcore.version>4.4.9</httpcore.version>
<influxdb.version>2.24</influxdb.version>
<inlong.version>1.12.0</inlong.version>
+ <iotdb.version>1.3.0</iotdb.version>
<jackson.version>2.17.0</jackson.version>
<jackson.databind.version>2.17.0</jackson.databind.version>
<jakarta-annotation.version>3.0.0</jakarta-annotation.version>
@@ -389,6 +390,11 @@
<artifactId>httpcore</artifactId>
<version>${httpcore.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-session</artifactId>
+ <version>${iotdb.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
@@ -825,6 +831,7 @@
<module>streampipes-data-explorer-api</module>
<module>streampipes-data-explorer-export</module>
<module>streampipes-data-explorer-influx</module>
+ <module>streampipes-data-explorer-iotdb</module>
<module>streampipes-data-explorer-management</module>
<module>streampipes-data-export</module>
<module>streampipes-dataformat</module>
@@ -868,7 +875,6 @@
<module>streampipes-storage-management</module>
<module>streampipes-test-utils</module>
<module>streampipes-test-utils-executors</module>
- <module>streampipes-ts-store-iotdb</module>
<module>streampipes-user-management</module>
<module>streampipes-vocabulary</module>
<module>streampipes-wrapper</module>
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 e79cb8d2bc..1c25d5fbc5 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
@@ -34,9 +34,7 @@ import
org.apache.streampipes.storage.management.StorageDispatcher;
import java.util.List;
-public enum DataExplorerManagerInflux implements IDataExplorerManager {
-
- INSTANCE;
+public class DataExplorerManagerInflux implements IDataExplorerManager {
@Override
public IDataLakeMeasurementCounter getMeasurementCounter(
diff --git a/streampipes-ts-store-iotdb/pom.xml
b/streampipes-data-explorer-iotdb/pom.xml
similarity index 74%
rename from streampipes-ts-store-iotdb/pom.xml
rename to streampipes-data-explorer-iotdb/pom.xml
index ff294b445d..763c932a36 100644
--- a/streampipes-ts-store-iotdb/pom.xml
+++ b/streampipes-data-explorer-iotdb/pom.xml
@@ -24,7 +24,7 @@
<version>0.97.0-SNAPSHOT</version>
</parent>
- <artifactId>streampipes-ts-store-iotdb</artifactId>
+ <artifactId>streampipes-data-explorer-iotdb</artifactId>
<properties>
<maven.compiler.source>17</maven.compiler.source>
@@ -44,6 +44,11 @@
<artifactId>streampipes-commons</artifactId>
<version>0.97.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-data-explorer-api</artifactId>
+ <version>0.97.0-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>org.apache.streampipes</groupId>
<artifactId>streampipes-data-explorer</artifactId>
@@ -64,7 +69,16 @@
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>iotdb-session</artifactId>
- <version>1.3.0</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.xerial.snappy</groupId>
+ <artifactId>snappy-java</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.github.luben</groupId>
+ <artifactId>zstd-jni</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
@@ -72,11 +86,22 @@
</dependency>
<!--Test dependencies -->
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-test-utils</artifactId>
+ <version>0.97.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
diff --git
a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
similarity index 54%
copy from
streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java
copy to
streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
index e79cb8d2bc..5457ec166d 100644
---
a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
@@ -16,40 +16,27 @@
*
*/
-package org.apache.streampipes.dataexplorer.influx;
+package org.apache.streampipes.dataexplorer.iotdb;
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.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;
-import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider;
-import
org.apache.streampipes.dataexplorer.influx.sanitize.DataLakeMeasurementSanitizerInflux;
+import org.apache.streampipes.dataexplorer.api.*;
+import
org.apache.streampipes.dataexplorer.iotdb.sanitize.DataLakeMeasurementSanitizerIotDb;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
import org.apache.streampipes.storage.management.StorageDispatcher;
import java.util.List;
-public enum DataExplorerManagerInflux implements IDataExplorerManager {
-
- INSTANCE;
+public class DataExplorerManagerIotDb implements IDataExplorerManager {
@Override
- public IDataLakeMeasurementCounter getMeasurementCounter(
- List<DataLakeMeasure> allMeasurements,
- List<String> measurementsToCount) {
- return new DataLakeMeasurementCounterInflux(allMeasurements,
measurementsToCount);
+ public IDataLakeMeasurementCounter
getMeasurementCounter(List<DataLakeMeasure> allMeasurements, List<String>
measurementsToCount) {
+ return null;
}
@Override
- public IDataExplorerQueryManagement getQueryManagement(
- IDataExplorerSchemaManagement dataExplorerSchemaManagement
- ) {
- return new DataExplorerQueryManagementInflux(dataExplorerSchemaManagement);
+ public IDataExplorerQueryManagement
getQueryManagement(IDataExplorerSchemaManagement dataExplorerSchemaManagement) {
+ return null;
}
@Override
@@ -61,11 +48,11 @@ public enum DataExplorerManagerInflux implements
IDataExplorerManager {
@Override
public ITimeSeriesStorage getTimeseriesStorage(DataLakeMeasure measure) {
- return new TimeSeriesStorageInflux(measure, Environments.getEnvironment(),
new InfluxClientProvider());
+ return new TimeSeriesStorageIotDb(measure, new IotDbPropertyConverter(),
new IotDbSessionProvider());
}
@Override
public IDataLakeMeasurementSanitizer
getMeasurementSanitizer(IStreamPipesClient client, DataLakeMeasure measure) {
- return new DataLakeMeasurementSanitizerInflux(client, measure);
+ return new DataLakeMeasurementSanitizerIotDb(client, measure);
}
}
diff --git
a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbMeasurementRecord.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbMeasurementRecord.java
similarity index 96%
rename from
streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbMeasurementRecord.java
rename to
streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbMeasurementRecord.java
index 63cb18f4b4..24591da3bd 100644
---
a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbMeasurementRecord.java
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbMeasurementRecord.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
diff --git
a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbPropertyConverter.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverter.java
similarity index 98%
rename from
streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbPropertyConverter.java
rename to
streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverter.java
index c0793ba9eb..9772c39c44 100644
---
a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbPropertyConverter.java
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverter.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
diff --git
a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbSessionProvider.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbSessionProvider.java
similarity index 97%
rename from
streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbSessionProvider.java
rename to
streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbSessionProvider.java
index 4a46ede043..6ff4bc38ca 100644
---
a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbSessionProvider.java
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbSessionProvider.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb;
import org.apache.iotdb.session.pool.SessionPool;
import org.apache.streampipes.commons.environment.Environment;
diff --git
a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/TimeSeriesStorageIotDb.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/TimeSeriesStorageIotDb.java
new file mode 100644
index 0000000000..c233580d1f
--- /dev/null
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/TimeSeriesStorageIotDb.java
@@ -0,0 +1,175 @@
+/*
+ * 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.iotdb;
+
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.pool.SessionPool;
+import org.apache.streampipes.commons.environment.Environments;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataexplorer.TimeSeriesStorage;
+import org.apache.streampipes.dataexplorer.iotdb.sanitize.IotDbNameSanitizer;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class TimeSeriesStorageIotDb extends TimeSeriesStorage {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TimeSeriesStorageIotDb.class);
+
+ private final IotDbPropertyConverter propertyConverter;
+ private final SessionPool sessionPool;
+
+ public TimeSeriesStorageIotDb(DataLakeMeasure measure,
+ IotDbPropertyConverter propertyConverter,
+ IotDbSessionProvider iotDbSessionProvider) {
+ super(measure);
+ this.propertyConverter = propertyConverter;
+ this.sessionPool =
iotDbSessionProvider.getSessionPool(Environments.getEnvironment());
+ }
+
+ @Override
+ protected void storeSanitizedRuntimeNames() {
+ measure.getEventSchema()
+ .getEventProperties()
+ .forEach(ep -> sanitizedRuntimeNames.put(
+ ep.getRuntimeName(),
+ new
IotDbNameSanitizer().renameReservedKeywords(ep.getRuntimeName())
+ ));
+ }
+
+ @Override
+ protected void sanitizeRuntimeNamesInEvent(Event event) {
+ event.getRaw()
+ .keySet()
+ .forEach(runtimeName -> {
+ // timestamp field does not need to be renamed since it is not
written as measurement
+ // and taken care of separately
+ if (!runtimeName.equals(measure.getTimestampFieldName())) {
+ event.renameFieldByRuntimeName(
+ runtimeName,
+ new
IotDbNameSanitizer().renameReservedKeywords(runtimeName)
+ );
+ }
+ }
+ );
+ }
+
+ @Override
+ protected void writeToTimeSeriesStorage(Event event) throws
SpRuntimeException {
+ if (event == null) {
+ LOG.warn("Input event is null - skipping event");
+ return;
+ }
+
+ var timestampValue = event.getFieldBySelector(measure.getTimestampField())
+ .getAsPrimitive()
+ .getAsLong();
+
+ if (timestampValue == null) {
+ LOG.warn("Timestamp of input event is null - skipping event");
+ return;
+ }
+
+ if (event.getRaw().size() <= 1) {
+ LOG.warn("Event only consists of timestamp and does not include any
measurement - skipping event");
+ return;
+ }
+
+ var iotDbRecords = extractMeasurementRecords(event);
+
+ if (iotDbRecords.isEmpty()) {
+ return;
+ }
+
+ insertIntoIotDb(timestampValue, iotDbRecords);
+ }
+
+ @Override
+ public void close() throws SpRuntimeException {
+ this.sessionPool.close();
+ }
+
+ /**
+ * Extracts all relevant information for the IotDb from the incoming event.
+ *
+ * @param event The event from which measurement records are extracted.
+ * @return An ArrayList of IotDbMeasurementRecord objects containing the
extracted measurement records.
+ */
+ private ArrayList<IotDbMeasurementRecord> extractMeasurementRecords(Event
event) {
+ var iotDbRecords = new ArrayList<IotDbMeasurementRecord>();
+ allEventProperties.forEach(ep -> {
+ try {
+ // timestamp is already known and is not part of the measurement thus
we ignore it here
+ if (!ep.getRuntimeName().equals(measure.getTimestampFieldName())) {
+
+ try {
+ var eventField =
event.getFieldByRuntimeName(ep.getRuntimeName()).getAsPrimitive();
+
+ if (ep instanceof EventPropertyPrimitive) {
+ iotDbRecords.add(propertyConverter.convertPrimitiveProperty(
+ (EventPropertyPrimitive) ep,
+
event.getFieldByRuntimeName(ep.getRuntimeName()).getAsPrimitive(),
+ sanitizedRuntimeNames.get(ep.getRuntimeName())
+ ));
+ } else {
+ iotDbRecords.add(propertyConverter.convertNonPrimitiveProperty(
+ ep,
+ sanitizedRuntimeNames.get(ep.getRuntimeName())
+ ));
+ }
+ } catch (SpRuntimeException e) {
+ LOG.debug("Value and metadata for event field '{}' could not be
extracted - "
+ + "property will not be written to storage",
+ ep.getRuntimeName()
+ );
+ }
+ }
+ } catch (SpRuntimeException e) {
+ LOG.error("Event could not be converted to the IotDB representation:
{}", e.getMessage());
+ }
+ });
+ return iotDbRecords;
+ }
+
+ /**
+ * Inserts measurement records into an IoTDB database as an aligned time
series of one device.
+ *
+ * @param timestampValue The timestamp value associated with the measurement
records.
+ * @param iotDbRecords An ArrayList of IotDbMeasurementRecord objects
containing the measurement records to be inserted.
+ */
+ private void insertIntoIotDb(long timestampValue,
ArrayList<IotDbMeasurementRecord> iotDbRecords) {
+ try {
+ sessionPool.insertAlignedRecordsOfOneDevice(
+ "root.streampipes.%s".formatted(measure.getMeasureName()),
+ List.of(timestampValue),
+
List.of(iotDbRecords.stream().map(IotDbMeasurementRecord::measurementName).toList()),
+
List.of(iotDbRecords.stream().map(IotDbMeasurementRecord::dataType).toList()),
+
List.of(iotDbRecords.stream().map(IotDbMeasurementRecord::value).toList())
+ );
+ } catch (IoTDBConnectionException | StatementExecutionException e) {
+ LOG.error("Failed to write event to IoTDB - {}", e.getMessage());
+ }
+ }
+}
diff --git
a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/DataLakeMeasurementSanitizerIotDb.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/DataLakeMeasurementSanitizerIotDb.java
new file mode 100644
index 0000000000..bd6637be27
--- /dev/null
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/DataLakeMeasurementSanitizerIotDb.java
@@ -0,0 +1,54 @@
+/*
+ * 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.iotdb.sanitize;
+
+import org.apache.streampipes.client.api.IStreamPipesClient;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataexplorer.DataLakeMeasurementSanitizer;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+
+/**
+ * Ensures proper sanitization of data lake measurements with respect to
Apache IoTDB specifics.
+ * <p>
+ * This includes both sanitizing the measurement name according to IoTDB's
path specifications,
+ * because the measurement name is used as part of the path in IoTDB,
+ * and sanitizing the runtime names of the event properties so that they do
not match any reserved keywords.
+ *
+ * @see <a
href="https://iotdb.apache.org/UserGuide/latest/Basic-Concept/Data-Model-and-Terminology.html#path";>IotDB
Path Spec</a>
+ */
+public class DataLakeMeasurementSanitizerIotDb extends
DataLakeMeasurementSanitizer {
+ public DataLakeMeasurementSanitizerIotDb(IStreamPipesClient client,
DataLakeMeasure measure) {
+ super(client, measure);
+ }
+
+ @Override
+ protected void cleanDataLakeMeasure() throws SpRuntimeException {
+ measure.setMeasureName(new
MeasureNameSanitizerIotDb().sanitize(measure.getMeasureName()));
+
+ measure.getEventSchema()
+ .getEventProperties()
+ .forEach(eventProperty -> eventProperty.setRuntimeName(
+ new
IotDbNameSanitizer().renameReservedKeywords(eventProperty.getRuntimeName()))
+ );
+ }
+
+ protected DataLakeMeasure getMeasure() {
+ return measure;
+ }
+}
diff --git
a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbNameSanitizer.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbNameSanitizer.java
similarity index 96%
rename from
streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbNameSanitizer.java
rename to
streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbNameSanitizer.java
index 8d66978dc9..0f9c4cfe36 100644
---
a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbNameSanitizer.java
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbNameSanitizer.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb.sanitize;
public class IotDbNameSanitizer {
diff --git
a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbReservedKeywords.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbReservedKeywords.java
similarity index 98%
rename from
streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbReservedKeywords.java
rename to
streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbReservedKeywords.java
index 5813cffce1..11c2530b2a 100644
---
a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbReservedKeywords.java
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbReservedKeywords.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb.sanitize;
import java.util.Arrays;
import java.util.List;
diff --git
a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/MeasureNameSanitizerIotDb.java
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/MeasureNameSanitizerIotDb.java
similarity index 97%
rename from
streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/MeasureNameSanitizerIotDb.java
rename to
streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/MeasureNameSanitizerIotDb.java
index bc30a4ed2d..0cd93a45f9 100644
---
a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/MeasureNameSanitizerIotDb.java
+++
b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/MeasureNameSanitizerIotDb.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb.sanitize;
/**
* Ensures that measurement names comply with IoTDB path specifications.
diff --git
a/streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/IotDbPropertyConverterTest.java
b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverterTest.java
similarity index 96%
rename from
streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/IotDbPropertyConverterTest.java
rename to
streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverterTest.java
index 29d0aab2ae..5b7085784e 100644
---
a/streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/IotDbPropertyConverterTest.java
+++
b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverterTest.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
@@ -81,7 +81,7 @@ public class IotDbPropertyConverterTest {
@Test
public void convertPrimitivePropertyNumber() {
- var property = new EventPropertyPrimitive(SO.NUMBER.toString(), "test",
null, null);
+ var property = new EventPropertyPrimitive(SO.NUMBER, "test", null, null);
var field = new PrimitiveField("test", "test", 5.24);
var result = new
IotDbPropertyConverter().convertPrimitiveProperty(property, field,
"sanitizedTest");
diff --git
a/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/DataLakeMeasurementSanitizerIotDbTest.java
b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/DataLakeMeasurementSanitizerIotDbTest.java
new file mode 100644
index 0000000000..12e491637d
--- /dev/null
+++
b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/DataLakeMeasurementSanitizerIotDbTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.iotdb.sanitize;
+
+import org.apache.streampipes.client.api.IStreamPipesClient;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.test.generator.EventPropertyPrimitiveTestBuilder;
+import org.apache.streampipes.test.generator.EventSchemaTestBuilder;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+public class DataLakeMeasurementSanitizerIotDbTest {
+
+ private IStreamPipesClient clientMock;
+
+ @BeforeEach
+ public void setUp() {
+ this.clientMock = mock(IStreamPipesClient.class);
+ }
+
+ @Test
+ public void cleanDataLakeMeasure() {
+ var eventSchema = EventSchemaTestBuilder.create()
+ .withEventProperty(
+ EventPropertyPrimitiveTestBuilder
+ .create()
+ .withRuntimeName("timestamp")
+ .build())
+ .withEventProperty(
+ EventPropertyPrimitiveTestBuilder
+ .create()
+ .withRuntimeName("all")
+ .build())
+ .withEventProperty(
+ EventPropertyPrimitiveTestBuilder
+ .create()
+ .withRuntimeName("pressure")
+ .build())
+ .build();
+ var measure = new DataLakeMeasure(
+ "invalid.Measure",
+ "s0::%s".formatted("timestamp"),
+ eventSchema
+ );
+ var sanitizer = new DataLakeMeasurementSanitizerIotDb(clientMock, measure);
+
+ sanitizer.cleanDataLakeMeasure();
+ var result = sanitizer.getMeasure();
+
+ assertEquals("invalid_Measure", result.getMeasureName());
+ assertEquals(3, result.getEventSchema().getEventProperties().size());
+ assertEquals("timestamp_",
result.getEventSchema().getEventProperties().get(0).getRuntimeName());
+ assertEquals("all_",
result.getEventSchema().getEventProperties().get(1).getRuntimeName());
+ assertEquals("pressure",
result.getEventSchema().getEventProperties().get(2).getRuntimeName());
+ }
+}
diff --git
a/streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/IotDbNameSanitizerTest.java
b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbNameSanitizerTest.java
similarity index 95%
rename from
streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/IotDbNameSanitizerTest.java
rename to
streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbNameSanitizerTest.java
index 16cbf865b2..5cc0a4ccdc 100644
---
a/streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/IotDbNameSanitizerTest.java
+++
b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbNameSanitizerTest.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb.sanitize;
import org.junit.jupiter.api.Test;
diff --git
a/streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/sanitize/MeasureNameSanitizerIotDbTest.java
b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/MeasureNameSanitizerIotDbTest.java
similarity index 91%
rename from
streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/sanitize/MeasureNameSanitizerIotDbTest.java
rename to
streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/MeasureNameSanitizerIotDbTest.java
index 6a11f4dee6..886e924391 100644
---
a/streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/sanitize/MeasureNameSanitizerIotDbTest.java
+++
b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/MeasureNameSanitizerIotDbTest.java
@@ -16,9 +16,8 @@
*
*/
-package org.apache.streampipes.ts.store.iotdb.sanitize;
+package org.apache.streampipes.dataexplorer.iotdb.sanitize;
-import org.apache.streampipes.ts.store.iotdb.MeasureNameSanitizerIotDb;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -30,6 +29,7 @@ public class MeasureNameSanitizerIotDbTest {
var sanitizer = new MeasureNameSanitizerIotDb();
assertEquals("myMeasure", sanitizer.sanitize("myMeasure"));
+ assertEquals("my_Measure", sanitizer.sanitize("my.Measure"));
assertEquals("我的措施", sanitizer.sanitize("我的措施"));
assertEquals("my_Measure", sanitizer.sanitize("my-Measure"));
assertEquals("my_Measure_", sanitizer.sanitize("my-Measure?"));
diff --git a/streampipes-data-explorer-management/pom.xml
b/streampipes-data-explorer-management/pom.xml
index 9e1d6a9600..e82a032919 100644
--- a/streampipes-data-explorer-management/pom.xml
+++ b/streampipes-data-explorer-management/pom.xml
@@ -47,6 +47,11 @@
<artifactId>streampipes-data-explorer-influx</artifactId>
<version>0.97.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-data-explorer-iotdb</artifactId>
+ <version>0.97.0-SNAPSHOT</version>
+ </dependency>
</dependencies>
<properties>
diff --git
a/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/DataExplorerDispatcher.java
b/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/DataExplorerDispatcher.java
index 96ba5b70ae..8614c15b57 100644
---
a/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/DataExplorerDispatcher.java
+++
b/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/DataExplorerDispatcher.java
@@ -21,6 +21,7 @@ package org.apache.streampipes.dataexplorer.management;
import org.apache.streampipes.commons.environment.Environments;
import org.apache.streampipes.dataexplorer.api.IDataExplorerManager;
import org.apache.streampipes.dataexplorer.influx.DataExplorerManagerInflux;
+import org.apache.streampipes.dataexplorer.iotdb.DataExplorerManagerIotDb;
public class DataExplorerDispatcher {
@@ -31,8 +32,8 @@ public class DataExplorerDispatcher {
return switch (Environments.getEnvironment()
.getTsStorage()
.getValueOrDefault()) {
- case SupportedDataExplorerStorages.INFLUX_DB ->
DataExplorerManagerInflux.INSTANCE;
- default -> DataExplorerManagerInflux.INSTANCE;
+ case SupportedDataExplorerStorages.IOT_DB -> new
DataExplorerManagerIotDb();
+ default -> new DataExplorerManagerInflux();
};
}
}
diff --git
a/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/SupportedDataExplorerStorages.java
b/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/SupportedDataExplorerStorages.java
index 70af654be8..1ee48e2acc 100644
---
a/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/SupportedDataExplorerStorages.java
+++
b/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/SupportedDataExplorerStorages.java
@@ -28,4 +28,5 @@ package org.apache.streampipes.dataexplorer.management;
*/
public class SupportedDataExplorerStorages {
public static final String INFLUX_DB = "influxdb";
+ public static final String IOT_DB = "iotdb";
}
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
b/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
index b4f2449ac4..e948bf1e28 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
@@ -52,13 +52,18 @@
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>iotdb-session</artifactId>
- <!-- 0.13.3 is the only recommended production ready version for
edge side (standalone) deployment -->
- <!-- Should be worked for IoTDB version >= 0.13.0. -->
- <version>0.13.3</version>
<exclusions>
<exclusion>
- <groupId>ch.qos.logback</groupId>
- <artifactId>logback-classic</artifactId>
+ <groupId>org.xerial.snappy</groupId>
+ <artifactId>snappy-java</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.antlr</groupId>
+ <artifactId>antlr4-runtime</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.github.luben</groupId>
+ <artifactId>zstd-jni</artifactId>
</exclusion>
</exclusions>
</dependency>
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbSink.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbSink.java
index d326376c1a..44dd0dda0e 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbSink.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbSink.java
@@ -21,6 +21,7 @@ package org.apache.streampipes.sinks.databases.jvm.iotdb;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import
org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
import org.apache.streampipes.model.DataSinkType;
+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.runtime.field.AbstractField;
@@ -30,7 +31,6 @@ import
org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
import org.apache.streampipes.sdk.helpers.EpRequirements;
import org.apache.streampipes.sdk.helpers.Labels;
import org.apache.streampipes.sdk.helpers.Locales;
-import org.apache.streampipes.sdk.utils.Assets;
import org.apache.streampipes.wrapper.params.compat.SinkParams;
import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
@@ -38,7 +38,6 @@ import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.pool.SessionPool;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.utils.Binary;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -77,7 +76,7 @@ public class IotDbSink extends StreamPipesDataSink {
return DataSinkBuilder
.create("org.apache.streampipes.sinks.databases.jvm.iotdb", 0)
.withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON).
+ .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON).
category(DataSinkType.DATABASE)
.requiredStream(
StreamRequirementsBuilder.create()
@@ -169,7 +168,7 @@ public class IotDbSink extends StreamPipesDataSink {
values.add(value);
} else {
types.add(TSDataType.TEXT);
- values.add(Binary.valueOf(value.toString()));
+ values.add(value);
}
}
