You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ri...@apache.org on 2026/08/27 21:11:50 UTC
(streampipes) branch dev updated: fix(#4382): Improve handling of existing tables in PostgreSQL sink (#4712)
This is an automated email from the ASF dual-hosted git repository.
dominikriemer 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 a4ad6953e4 fix(#4382): Improve handling of existing tables in PostgreSQL sink (#4712)
a4ad6953e4 is described below
commit a4ad6953e41b6375f70e10ef7e243b47de41125a
Author: Stefanie Sinner <13...@users.noreply.github.com>
AuthorDate: Thu Aug 27 23:11:44 2026 +0200
fix(#4382): Improve handling of existing tables in PostgreSQL sink (#4712)
Co-authored-by: Sven Oehler <oe...@web.de>
---
.../jvm/DatabaseSinksExtensionModuleExport.java | 3 +-
.../sinks/databases/jvm/jdbcclient/JdbcClient.java | 76 +++++-
.../jvm/jdbcclient/model/StatementHandler.java | 43 ++++
.../jvm/jdbcclient/model/TableDescription.java | 95 ++++++--
.../sinks/databases/jvm/postgresql/PostgreSql.java | 23 +-
.../jvm/postgresql/PostgreSqlParameters.java | 17 +-
.../databases/jvm/postgresql/PostgreSqlSink.java | 18 +-
.../migration/PostgreSqlSinkMigrationV1.java | 101 ++++++++
.../documentation.md | 12 +
.../strings.en | 6 +
.../databases/jvm/jdbcclient/JdbcClientTest.java | 238 +++++++++++++++++++
.../jvm/jdbcclient/model/StatementHandlerTest.java | 137 +++++++++++
.../jvm/jdbcclient/model/TableDescriptionTest.java | 258 +++++++++++++++++++++
.../jvm/postgresql/PostgreSqlParametersTest.java | 87 +++++++
.../jvm/postgresql/PostgreSqlSinkTest.java | 106 +++++++++
.../migration/PostgreSqlSinkMigrationV1Test.java | 138 +++++++++++
16 files changed, 1330 insertions(+), 28 deletions(-)
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/DatabaseSinksExtensionModuleExport.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/DatabaseSinksExtensionModuleExport.java
index 5285759baa..4fd6790f43 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/DatabaseSinksExtensionModuleExport.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/DatabaseSinksExtensionModuleExport.java
@@ -28,6 +28,7 @@ import org.apache.streampipes.sinks.databases.jvm.iotdb.IotDbSink;
import org.apache.streampipes.sinks.databases.jvm.milvus.MilvusSink;
import org.apache.streampipes.sinks.databases.jvm.parquet.ParquetSink;
import org.apache.streampipes.sinks.databases.jvm.postgresql.PostgreSqlSink;
+import org.apache.streampipes.sinks.databases.jvm.postgresql.migration.PostgreSqlSinkMigrationV1;
import org.apache.streampipes.sinks.databases.jvm.qdrant.QdrantSink;
import org.apache.streampipes.sinks.databases.jvm.redis.RedisSink;
import org.apache.streampipes.sinks.databases.jvm.tsfile.TsFileSink;
@@ -57,6 +58,6 @@ public class DatabaseSinksExtensionModuleExport implements IExtensionModuleExpor
@Override
public List<IModelMigrator<?, ?>> migrators() {
- return Collections.emptyList();
+ return List.of(new PostgreSqlSinkMigrationV1());
}
}
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/JdbcClient.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/JdbcClient.java
index b62f020c72..c2975a8ead 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/JdbcClient.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/JdbcClient.java
@@ -50,6 +50,18 @@ public class JdbcClient {
protected StatementHandler statementHandler;
+ /*
+ * Controls whether the sink may create the target table if it does not exist yet.
+ * If false, the sink only writes into a table provided by the user and fails if it is missing.
+ */
+ protected boolean allowNewTableCreation = true;
+
+ /*
+ * Number of events to buffer before one batch insert (one round trip to the database).
+ * The default value is 1 which means it sends one statement per event.
+ */
+ protected int batchSize = 1;
+
/**
* A wrapper class for all supported SQL data types (INT, BIGINT, FLOAT, DOUBLE, VARCHAR(255)).
* If no matching type is found, it is interpreted as a String (VARCHAR(255))
@@ -172,21 +184,41 @@ public class JdbcClient {
// Database should exist by now so we can establish a connection
connection = DriverManager.getConnection(url + databaseName, this.dbDescription.getUsername(),
this.dbDescription.getPassword());
+ prepareConnection();
this.statementHandler.setStatement(connection.createStatement());
- ResultSet rs = connection.getMetaData().getTables(null, null, this.tableDescription.getName(), null);
- if (rs.next()) {
+ ResultSet rs = connection.getMetaData().getTables(
+ null,
+ getTableSchemaPattern(),
+ this.tableDescription.getName(),
+ null);
+ boolean tableAlreadyExists = rs.next();
+ rs.close();
+ if (tableAlreadyExists) {
validateTable();
+ } else if (!this.allowNewTableCreation) {
+ // The user does not allow the sink to create tables, so the table has to be there already
+ throw new SpRuntimeException("Table '" + this.tableDescription.getName()
+ + "' does not exist and the option 'Allow New Table Creation' is disabled. "
+ + "Check that the table name matches an existing table in your database, "
+ + "or enable the option to let the sink create one automatically.");
} else {
createTable();
}
this.tableDescription.setTableExists();
- rs.close();
} catch (SQLException e) {
closeAll();
throw new SpRuntimeException(e.getMessage());
}
}
+ protected void prepareConnection() throws SQLException {
+
+ }
+
+ protected String getTableSchemaPattern() throws SQLException {
+ return null;
+ }
+
/**
* Prepares a statement for the insertion of values or the
*
@@ -194,26 +226,41 @@ public class JdbcClient {
* @throws SpRuntimeException When there was an error in the saving process
*/
protected void save(final Event event) throws SpRuntimeException {
- //TODO: Add batch support (https://stackoverflow.com/questions/3784197/efficient-way-to-do-batch-inserts-with-jdbc)
- checkConnected();
- Map<String, Object> eventMap = event.getRaw();
if (event == null) {
throw new SpRuntimeException("event is null");
}
+ checkConnected();
+ Map<String, Object> eventMap = event.getRaw();
if (!this.tableDescription.tableExists()) {
+ if (!this.allowNewTableCreation) {
+ throw new SpRuntimeException("Table '" + this.tableDescription.getName() + "' is not available.");
+ }
// Creates the table
createTable();
this.tableDescription.setTableExists();
}
try {
checkConnected();
- this.statementHandler.executePreparedStatement(
- this.dbDescription, this.tableDescription,
- connection, eventMap);
+ if (this.batchSize > 1) {
+ // Collects events and sends them together in one batch insert
+ this.statementHandler.addToBatch(
+ this.dbDescription, this.tableDescription,
+ connection, eventMap);
+ if (this.statementHandler.getPendingBatchCount() >= this.batchSize) {
+ this.statementHandler.executeBatch();
+ }
+ } else {
+ // Sends one statement per event
+ this.statementHandler.executePreparedStatement(
+ this.dbDescription, this.tableDescription,
+ connection, eventMap);
+ }
} catch (SQLException e) {
- if (e.getSQLState().substring(0, 2).equals("42")) {
+ boolean tableMissing = e.getSQLState() != null && e.getSQLState().startsWith("42");
+ if (tableMissing && this.allowNewTableCreation && this.batchSize <= 1) {
// If the table does not exists (because it got deleted or something, will cause the error
// code "42") we will try to create a new one. Otherwise we do not handle the exception.
+ // Skipped when the user does not allow table creation, or when events are still waiting to be written.
LOG.warn("Table '" + this.tableDescription.getName() + "' was unexpectedly not found and gets recreated.");
this.tableDescription.setTableMissing();
createTable();
@@ -259,6 +306,15 @@ public class JdbcClient {
*/
protected void closeAll() {
boolean error = false;
+ try {
+ // Write out any events still buffered in the batch before we close the connection
+ if (this.statementHandler != null) {
+ this.statementHandler.executeBatch();
+ }
+ } catch (SQLException e) {
+ error = true;
+ LOG.warn("Exception when flushing the batch: " + e.getMessage());
+ }
try {
if (this.statementHandler.statement != null) {
this.statementHandler.statement.close();
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/StatementHandler.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/StatementHandler.java
index 82e577ff38..5178eb0fbe 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/StatementHandler.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/StatementHandler.java
@@ -33,6 +33,10 @@ public class StatementHandler {
public Statement statement;
public PreparedStatement preparedStatement;
+
+ // Number of events currently buffered in the batch but not yet sent to the database
+ private int pendingBatchCount = 0;
+
/**
* The parameters in the prepared statement {@code ps} together with their index and data type
*/
@@ -136,6 +140,9 @@ public class StatementHandler {
} else {
if (!eventParameterMap.containsKey(newKey)) {
//TODO: start the for loop all over again
+
+ // Write the collected events first, otherwise they are lost when the statement is rebuilt
+ executeBatch();
generatePreparedStatement(dbDescription, tableDescription, connection, event);
}
ParameterInformation p = eventParameterMap.get(newKey);
@@ -165,6 +172,42 @@ public class StatementHandler {
this.preparedStatement.executeUpdate();
}
+ /**
+ * Fills the prepared statement with the given event and adds it to the JDBC batch instead of
+ * executing it immediately. To send all buffered rows to the database in one round trip.
+ *
+ * @param event Data to be saved in the SQL table
+ * @throws SQLException When the statement cannot be filled
+ * @throws SpRuntimeException When the table name is not allowed
+ */
+ public void addToBatch(DbDescription dbDescription, TableDescription tableDescription,
+ Connection connection, final Map<String, Object> event)
+ throws SQLException, SpRuntimeException {
+ if (this.getPreparedStatement() != null) {
+ this.preparedStatement.clearParameters();
+ }
+ fillPreparedStatement(dbDescription, tableDescription, connection, event, "");
+ this.preparedStatement.addBatch();
+ this.pendingBatchCount++;
+ }
+
+ /**
+ * Sends all buffered rows to the database in a single round trip and resets the counter.
+ * Does nothing if there is no statement or nothing is buffered.
+ *
+ * @throws SQLException When the batch cannot be executed.
+ */
+ public void executeBatch() throws SQLException {
+ if (this.preparedStatement != null && this.pendingBatchCount > 0) {
+ this.preparedStatement.executeBatch();
+ this.pendingBatchCount = 0;
+ }
+ }
+
+ public int getPendingBatchCount() {
+ return this.pendingBatchCount;
+ }
+
public PreparedStatement getPreparedStatement() {
return preparedStatement;
}
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/TableDescription.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/TableDescription.java
index 81ac2af47c..f90465a654 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/TableDescription.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/TableDescription.java
@@ -28,6 +28,7 @@ import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -116,23 +117,91 @@ public class TableDescription {
}
public void validateTable() throws SpRuntimeException {
+ List<String> missingCols = new ArrayList<>();
+ List<String> typeMismatches = new ArrayList<>();
+
for (EventProperty property : this.eventSchema.getEventProperties()) {
- DbDataTypes existingType = this.getDataTypesHashMap().get(property.getRuntimeName());
- if (existingType != null) {
- if (property instanceof EventPropertyPrimitive) {
- String expected = ((EventPropertyPrimitive) property).getRuntimeType();
- String actual = DbDataTypeFactory.getDataType(existingType).toString();
- if (!expected.equals(actual)) {
- throw new SpRuntimeException("Column '" + property.getRuntimeName()
- + "' in table '" + this.getName() + "' has type mismatch: expected "
- + expected + " but got " + actual);
- }
+ String columnName = property.getRuntimeName();
+ DbDataTypes existingType = this.getDataTypesHashMap().get(columnName);
+
+ if (existingType == null) {
+ // Column is missing in the table
+ missingCols.add("'" + columnName + "'");
+ } else if (property instanceof EventPropertyPrimitive) {
+ // Check the data type of existing columns
+ String expected = ((EventPropertyPrimitive) property).getRuntimeType();
+ String actual = DbDataTypeFactory.getDataType(existingType).toString();
+ if (!expected.equals(actual)) {
+ // Remember this mismatch and keep checking, so all problems are reported together later
+ typeMismatches.add("'" + columnName + "' (the data stream provides " + extractTypeName(expected)
+ + " but the table column is " + existingType + ")");
}
- } else {
- throw new SpRuntimeException("Column '" + property.getRuntimeName()
- + "' is missing in table '" + this.getName() + "'");
}
}
+ reportMissingColumns(missingCols);
+ reportTypeMismatches(typeMismatches);
+ }
+
+ /**
+ * Shortens a runtime type URI such as "http://www.w3.org/2001/XMLSchema#float" to a name that is
+ * readable for users. Only the part after the '#' is kept, for example "float".
+ *
+ * @param runtimeType the runtime type of the event property.
+ * @return the type name without the namespace, or the unchanged input if it contains no '#'.
+ */
+ private String extractTypeName(String runtimeType) {
+ return runtimeType.substring(runtimeType.lastIndexOf('#') + 1);
+ }
+
+ /**
+ * Reports the columns that the data stream expects but that do not exist in the table, so the
+ * user knows which columns to add. Does nothing if no column is missing.
+ *
+ * @param missingCols the names of the missing columns, may be empty.
+ * @throws SpRuntimeException if at least one column is missing.
+ */
+ private void reportMissingColumns(List<String> missingCols) throws SpRuntimeException {
+ if (missingCols.isEmpty()) {
+ return;
+ }
+ String noun = missingCols.size() == 1 ? "Column " : "Columns ";
+ String verb = missingCols.size() == 1 ? " is" : " are";
+ throw new SpRuntimeException(noun + formatColumnEnumeration(missingCols) + verb
+ + " missing in table '" + this.getName() + "'");
+ }
+
+ /**
+ * Reports the columns whose data type in the table does not match the data stream, so the user
+ * knows which columns have the wrong type. Does nothing if all types match.
+ *
+ * @param typeMismatches the descriptions of the mismatching columns, may be empty.
+ * @throws SpRuntimeException if at least one column has a different data type.
+ */
+ private void reportTypeMismatches(List<String> typeMismatches) throws SpRuntimeException {
+ if (typeMismatches.isEmpty()) {
+ return;
+ }
+ String noun = typeMismatches.size() == 1 ? "column " : "columns ";
+ throw new SpRuntimeException("Type mismatch in table '" + this.getName() + "' for "
+ + noun + formatColumnEnumeration(typeMismatches));
+ }
+
+ /**
+ * Formats a list of column names or column problems as a readable enumeration for an error message
+ * to report every problem of the existing table in a single message.
+ *
+ * @param columns the column names or problem descriptions to enumerate.
+ * @return the entries joined into one phrase, using "and" before the last entry.
+ */
+ private String formatColumnEnumeration(List<String> columns) {
+ if (columns.size() == 1) {
+ return columns.get(0);
+ }
+ if (columns.size() == 2) {
+ return columns.get(0) + " and " + columns.get(1);
+ }
+ return String.join(", ", columns.subList(0, columns.size() - 1))
+ + ", and " + columns.get(columns.size() - 1);
}
public boolean tableExists() {
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSql.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSql.java
index dc4c1d4c75..393fd46e84 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSql.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSql.java
@@ -26,9 +26,12 @@ import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.SupportedDbEn
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.sql.SQLException;
+
public class PostgreSql extends JdbcClient {
private static final Logger LOG = LoggerFactory.getLogger(PostgreSql.class);
+ private static final String DEFAULT_SCHEMA = "public";
private PostgreSqlParameters params;
@@ -36,6 +39,8 @@ public class PostgreSql extends JdbcClient {
throws SpRuntimeException {
this.params = parameters;
+ this.allowNewTableCreation = parameters.allowsNewTableCreation();
+ this.batchSize = parameters.getBatchSize();
// get(0) because it is the only input stream of the sink (and not two)
// See (https://www.postgresql.org/docs/current/sql-syntax-lexical.html#SQL-SYNTAX-IDENTIFIERS)
@@ -46,10 +51,26 @@ public class PostgreSql extends JdbcClient {
SupportedDbEngines.POSTGRESQL);
}
+ @Override
+ protected void prepareConnection() throws SQLException {
+ try (var statement = this.connection.prepareStatement("SELECT current_schema()");
+ var resultSet = statement.executeQuery()) {
+ if (resultSet.next() && resultSet.getString(1) == null) {
+ this.connection.setSchema(DEFAULT_SCHEMA);
+ }
+ }
+ }
+
+ @Override
+ protected String getTableSchemaPattern() throws SQLException {
+ return this.connection.getSchema();
+ }
+
@Override
protected void extractTableInformation() {
- String query = "SELECT * FROM information_schema.columns WHERE table_name = ? ;";
+ String query = "SELECT * FROM information_schema.columns "
+ + "WHERE table_name = ? AND table_schema = current_schema() ;";
String[] queryParameter = new String[]{params.getDbTable()};
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlParameters.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlParameters.java
index d00af3ffd3..66fa19fcf8 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlParameters.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlParameters.java
@@ -23,8 +23,12 @@ import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.JdbcConnectio
public class PostgreSqlParameters extends JdbcConnectionParameters {
+ private final boolean allowNewTableCreation;
+ private final int batchSize;
+
public PostgreSqlParameters(DataSinkInvocation graph, String postgreSqlHost, Integer postgreSqlPort,
- String databaseName, String tableName, String user, String password, Boolean sslEnabled) {
+ String databaseName, String tableName, String user, String password, Boolean sslEnabled,
+ Boolean allowNewTableCreation, Integer batchSize) {
super(
graph,
postgreSqlHost,
@@ -36,5 +40,16 @@ public class PostgreSqlParameters extends JdbcConnectionParameters {
sslEnabled,
"org.postgresql.ssl.NonValidatingFactory",
true);
+
+ this.allowNewTableCreation = allowNewTableCreation == null || allowNewTableCreation;
+ this.batchSize = (batchSize == null || batchSize < 1) ? 1 : batchSize;
+ }
+
+ public boolean allowsNewTableCreation() {
+ return allowNewTableCreation;
+ }
+
+ public int getBatchSize() {
+ return batchSize;
}
}
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlSink.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlSink.java
index 3c321640cb..81c290ec54 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlSink.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlSink.java
@@ -37,6 +37,10 @@ import org.apache.streampipes.sdk.helpers.Tuple2;
public class PostgreSqlSink implements IStreamPipesDataSink {
+ public static final String ID = "org.apache.streampipes.sinks.databases.jvm.postgresql";
+ public static final String ALLOW_NEW_TABLE_CREATION_KEY = "allow_new_table_creation";
+ public static final String BATCH_SIZE_KEY = "batch_size";
+
private static final String DATABASE_HOST_KEY = "db_host";
private static final String DATABASE_PORT_KEY = "db_port";
private static final String DATABASE_NAME_KEY = "db_name";
@@ -54,7 +58,7 @@ public class PostgreSqlSink implements IStreamPipesDataSink {
return DataSinkConfiguration.create(
PostgreSqlSink::new,
DataSinkBuilder
- .create("org.apache.streampipes.sinks.databases.jvm.postgresql", 0)
+ .create(ID, 1)
.withLocales(Locales.EN)
.withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
.category(DataSinkType.DATABASE)
@@ -65,12 +69,14 @@ public class PostgreSqlSink implements IStreamPipesDataSink {
.requiredIntegerParameter(Labels.withId(DATABASE_PORT_KEY), 5432)
.requiredTextParameter(Labels.withId(DATABASE_NAME_KEY))
.requiredTextParameter(Labels.withId(DATABASE_TABLE_KEY))
+ .requiredSlideToggle(Labels.withId(ALLOW_NEW_TABLE_CREATION_KEY), true)
.requiredTextParameter(Labels.withId(DATABASE_USER_KEY))
.requiredSecret(Labels.withId(DATABASE_PASSWORD_KEY))
.requiredSingleValueSelection(Labels.withId(SSL_MODE),
Options.from(
new Tuple2<>("Yes", SSL_ENABLED),
new Tuple2<>("No", SSL_DISABLED)))
+ .requiredIntegerParameter(Labels.withId(BATCH_SIZE_KEY), 1)
.build()
);
}
@@ -87,6 +93,12 @@ public class PostgreSqlSink implements IStreamPipesDataSink {
String user = extractor.singleValueParameter(DATABASE_USER_KEY, String.class);
String password = extractor.secretValue(DATABASE_PASSWORD_KEY);
String sslSelection = extractor.selectedSingleValueInternalName(SSL_MODE, String.class);
+ boolean allowNewTableCreation = extractor.slideToggleValue(ALLOW_NEW_TABLE_CREATION_KEY);
+ Integer batchSize = extractor.singleValueParameter(BATCH_SIZE_KEY, Integer.class);
+ if (batchSize == null || batchSize < 1) {
+ throw new SpRuntimeException("Batch size must be at least 1, but was '" + batchSize
+ + "'. Use 1 to write each event immediately, or a higher value to write events in batches.");
+ }
PostgreSqlParameters params = new PostgreSqlParameters(
parameters.getModel(),
@@ -96,7 +108,9 @@ public class PostgreSqlSink implements IStreamPipesDataSink {
tableName,
user,
password,
- sslSelection.equals(SSL_ENABLED));
+ sslSelection.equals(SSL_ENABLED),
+ allowNewTableCreation,
+ batchSize);
this.postgreSql = new PostgreSql();
postgreSql.onInvocation(params);
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/migration/PostgreSqlSinkMigrationV1.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/migration/PostgreSqlSinkMigrationV1.java
new file mode 100644
index 0000000000..1de05530dd
--- /dev/null
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/migration/PostgreSqlSinkMigrationV1.java
@@ -0,0 +1,101 @@
+/*
+ * 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.databases.jvm.postgresql.migration;
+
+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.FreeTextStaticProperty;
+import org.apache.streampipes.model.staticproperty.SlideToggleStaticProperty;
+import org.apache.streampipes.sdk.helpers.Labels;
+import org.apache.streampipes.sinks.databases.jvm.postgresql.PostgreSqlSink;
+import org.apache.streampipes.vocabulary.XSD;
+
+/**
+ * Migrates the PostgreSQL sink from model version 0 to 1.
+ * <p>
+ * Version 1 adds two configuration fields: a toggle that allows the sink to create the target
+ * table, and a batch size for grouping events into a single insert. Existing pipelines keep
+ * their previous behavior through the default values: table creation allowed and batch size 1.
+ */
+public class PostgreSqlSinkMigrationV1 implements IDataSinkMigrator {
+
+ /*
+ * Position of the toggle in a pipeline stored with model version 0, where the fields are
+ * host, port, database, table, user, password and ssl mode. The sink renders the toggle
+ * right below the table name, so a migrated pipeline has to show it in the same place.
+ */
+ private static final int ALLOW_NEW_TABLE_CREATION_INDEX = 4;
+
+ @Override
+ public ModelMigratorConfig config() {
+ return new ModelMigratorConfig(
+ PostgreSqlSink.ID,
+ SpServiceTagPrefix.DATA_SINK,
+ 0,
+ 1
+ );
+ }
+
+ @Override
+ public MigrationResult<DataSinkInvocation> migrate(DataSinkInvocation element,
+ IDataSinkParameterExtractor extractor)
+ throws RuntimeException {
+ addAllowNewTableCreationToggle(element);
+ addBatchSize(element);
+ return MigrationResult.success(element);
+ }
+
+ private void addAllowNewTableCreationToggle(DataSinkInvocation element) {
+ var label = Labels.from(
+ PostgreSqlSink.ALLOW_NEW_TABLE_CREATION_KEY,
+ "Allow New Table Creation",
+ "Let the sink create the table entered above if it does not exist yet"
+ );
+ var staticProperty = new SlideToggleStaticProperty(
+ label.getInternalId(),
+ label.getLabel(),
+ label.getDescription(),
+ true
+ );
+ staticProperty.setSelected(true);
+
+ element.getStaticProperties().add(ALLOW_NEW_TABLE_CREATION_INDEX, staticProperty);
+ }
+
+ private void addBatchSize(DataSinkInvocation element) {
+ var label = Labels.from(
+ PostgreSqlSink.BATCH_SIZE_KEY,
+ "Batch Size",
+ "How many events should the sink collect before writing them together to the database?"
+ );
+ var staticProperty = new FreeTextStaticProperty(
+ label.getInternalId(),
+ label.getLabel(),
+ label.getDescription()
+ );
+ staticProperty.setRequiredDatatype(XSD.INTEGER);
+ staticProperty.setValue("1");
+
+ element.getStaticProperties().add(staticProperty);
+ }
+}
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.postgresql/documentation.md b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.postgresql/documentation.md
index b219701c4d..079c45cda7 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.postgresql/documentation.md
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.postgresql/documentation.md
@@ -54,6 +54,12 @@ The name of the database where events will be stored
The name of the table where events will be stored (will be created if it does not exist)
+### Allow New Table Creation
+
+When enabled, the sink creates the table entered above if it does not exist yet.
+When disabled, the pipeline does not start and asks you to create the table first.
+Use this if the tables in your database have to be created by you.
+
### Username
The username for the PostgreSQL Server.
@@ -62,6 +68,12 @@ The username for the PostgreSQL Server.
The password for the PostgreSQL Server.
+### Batch Size
+
+The number of events collected before they are written together to the database. Use the value 1 to
+write each event on its own. Higher values are faster at high data rates. Buffered events are flushed
+when the pipeline stops.
+
## Output
(not applicable for data sinks)
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.postgresql/strings.en b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.postgresql/strings.en
index 32994c6759..a797632b02 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.postgresql/strings.en
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/resources/org.apache.streampipes.sinks.databases.jvm.postgresql/strings.en
@@ -30,6 +30,9 @@ db_name.description=The name of the database where events will be stored
db_table.title=Table Name
db_table.description=The name of the table where events will be stored (will be created if it does not exist)
+allow_new_table_creation.title=Allow New Table Creation
+allow_new_table_creation.description=Let the sink create the table entered above if it does not exist yet
+
db_user.title=Username
db_user.description=The username for the PostgreSQL Server
@@ -38,3 +41,6 @@ db_password.description=The password for the PostgreSQL Server
ssl_mode.title=SSL Mode
ssl_mode.description=Should the connection be secured via SSL?
+
+batch_size.title=Batch Size
+batch_size.description=How many events should the sink collect before writing them together to the database?
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/JdbcClientTest.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/JdbcClientTest.java
new file mode 100644
index 0000000000..d1f633a5df
--- /dev/null
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/JdbcClientTest.java
@@ -0,0 +1,238 @@
+/*
+ * 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.databases.jvm.jdbcclient;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.DbDescription;
+import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.StatementHandler;
+import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.SupportedDbEngines;
+import org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.TableDescription;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for the implementation of the {@link JdbcClient} class.
+ */
+class JdbcClientTest {
+
+ private static final String ERROR_CODE_UNDEFINED_TABLE = "42P01";
+ private static final String TABLE_NAME = "sensor_data";
+ private static final Map<String, Object> EVENT = Map.of(
+ "temperature", 21.5
+ );
+
+ private JdbcClient client;
+ private StatementHandler statementHandler;
+ private Statement statement;
+ private Event event;
+
+ @BeforeEach
+ void setUp() {
+ statement = mock(Statement.class);
+ statementHandler = mock(StatementHandler.class);
+ statementHandler.statement = statement;
+
+ client = new JdbcClient();
+ client.connection = mock(Connection.class);
+ client.dbDescription = newDbDescription();
+ client.tableDescription = new TableDescription(TABLE_NAME, new EventSchema());
+ client.statementHandler = statementHandler;
+ event = newEvent();
+ }
+
+ DbDescription newDbDescription() {
+ DbDescription dbDescription = mock(DbDescription.class);
+ when(dbDescription.getAllowedRegEx()).thenReturn("^[a-zA-Z_][a-zA-Z0-9_]*$");
+ when(dbDescription.getEngine()).thenReturn(SupportedDbEngines.POSTGRESQL);
+ when(dbDescription.isColumnNameQuoted()).thenReturn(false);
+ return dbDescription;
+ }
+
+ Event newEvent() {
+ Event event = mock(Event.class);
+ when(event.getRaw()).thenReturn(EVENT);
+ return event;
+ }
+
+ @Test
+ void testSave_eventIsNull_throwsException() {
+ try {
+ client.save(null);
+ fail("No exception on #save");
+ } catch (SpRuntimeException e) {
+ assertEquals("event is null", e.getMessage());
+ }
+ }
+
+ @Test
+ void testSave_connectionIsNull_throwsException() {
+ client.connection = null;
+
+ try {
+ client.save(event);
+ fail("No exception on #save");
+ } catch (SpRuntimeException e) {
+ assertEquals("Connection is not established.", e.getMessage());
+ }
+ }
+
+ @Test
+ void testSave_tableIsMissingAndCreationProhibited_throwsException() throws SQLException {
+ client.allowNewTableCreation = false;
+
+ try {
+ client.save(event);
+ fail("No exception on #save");
+ } catch (SpRuntimeException e) {
+ assertEquals("Table '" + TABLE_NAME + "' is not available.", e.getMessage());
+ }
+ verify(statement, never()).executeUpdate(anyString());
+ }
+
+ @Test
+ void testSave_tableIsMissingAndCreationAllowed_createsTable() throws SQLException {
+ client.allowNewTableCreation = true;
+
+ client.save(event);
+
+ verify(statement, times(1)).executeUpdate(anyString());
+ verify(statementHandler, times(1)).executePreparedStatement(any(), any(), any(), anyMap());
+ }
+
+ @Test
+ void testSave_tableExistsAndCreationProhibited_writesIntoExistingTable() throws SQLException {
+ client.tableDescription.setTableExists();
+ client.allowNewTableCreation = false;
+ client.batchSize = 1;
+
+ client.save(event);
+
+ verify(statement, never()).executeUpdate(anyString());
+ verify(statementHandler, times(1)).executePreparedStatement(any(), any(), any(), anyMap());
+ }
+
+ @Test
+ void testSave_tableIsDeletedDuringRuntime_recreatesTableAndRetries() throws SQLException {
+ client.tableDescription.setTableExists();
+ client.allowNewTableCreation = true;
+ client.batchSize = 1;
+
+ doThrow(new SQLException("write failed", ERROR_CODE_UNDEFINED_TABLE))
+ .doNothing()
+ .when(statementHandler).executePreparedStatement(any(), any(), any(), anyMap());
+
+ client.save(event);
+
+ verify(statement, times(1)).executeUpdate(anyString());
+ verify(statementHandler, times(2)).executePreparedStatement(any(), any(), any(), anyMap());
+ }
+
+ @Test
+ void testSave_tableIsDeletedAndCreationProhibited_throwsException() throws SQLException {
+ client.tableDescription.setTableExists();
+ client.allowNewTableCreation = false;
+ client.batchSize = 1;
+
+ doThrow(new SQLException("write failed", ERROR_CODE_UNDEFINED_TABLE))
+ .doNothing()
+ .when(statementHandler).executePreparedStatement(any(), any(), any(), anyMap());
+
+ try {
+ client.save(event);
+ fail("No exception on #save");
+ } catch (SpRuntimeException e) {
+ assertEquals("write failed", e.getMessage());
+ }
+ verify(statement, never()).executeUpdate(anyString());
+ }
+
+ @Test
+ void testSave_tableIsDeletedWhileCollectingEvents_throwsException() throws SQLException {
+ client.tableDescription.setTableExists();
+ client.allowNewTableCreation = true;
+ client.batchSize = 3;
+
+ doThrow(new SQLException("write failed", ERROR_CODE_UNDEFINED_TABLE))
+ .when(statementHandler).addToBatch(any(), any(), any(), anyMap());
+
+ try {
+ client.save(event);
+ fail("No exception on #save");
+ } catch (SpRuntimeException e) {
+ assertEquals("write failed", e.getMessage());
+ }
+ verify(statement, never()).executeUpdate(anyString());
+ }
+
+ @Test
+ void testSave_batchSizeIsOne_sendsOneStatementPerEvent() throws SQLException {
+ client.tableDescription.setTableExists();
+ client.batchSize = 1;
+
+ client.save(event);
+
+ verify(statementHandler, times(1)).executePreparedStatement(any(), any(), any(), anyMap());
+ verify(statementHandler, never()).addToBatch(any(), any(), any(), anyMap());
+ }
+
+ @Test
+ void testSave_batchIsNotFull_collectsEvent() throws SQLException {
+ client.tableDescription.setTableExists();
+ client.batchSize = 3;
+ when(statementHandler.getPendingBatchCount()).thenReturn(1);
+
+ client.save(event);
+
+ verify(statementHandler, times(1)).addToBatch(any(), any(), any(), anyMap());
+ verify(statementHandler, never()).executeBatch();
+ verify(statementHandler, never()).executePreparedStatement(any(), any(), any(), anyMap());
+ }
+
+ @Test
+ void testSave_batchIsFull_sendsCollectedEvents() throws SQLException {
+ client.tableDescription.setTableExists();
+ client.batchSize = 3;
+ when(statementHandler.getPendingBatchCount()).thenReturn(3);
+
+ client.save(event);
+
+ verify(statementHandler, times(1)).addToBatch(any(), any(), any(), anyMap());
+ verify(statementHandler, times(1)).executeBatch();
+ }
+}
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/StatementHandlerTest.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/StatementHandlerTest.java
new file mode 100644
index 0000000000..6c0a21f089
--- /dev/null
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/StatementHandlerTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.databases.jvm.jdbcclient.model;
+
+import org.apache.streampipes.model.schema.EventSchema;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for the implementation of the {@link StatementHandler} class.
+ */
+class StatementHandlerTest {
+
+ private static final String TABLE_NAME = "sensor_data";
+ private static final Map<String, Object> EVENT_TEMPERATURE = Map.of(
+ "temperature", 21.5
+ );
+ private static final Map<String, Object> EVENT_PRESSURE = Map.of(
+ "pressure", 1013.2
+ );
+
+ private StatementHandler statementHandler;
+ private DbDescription dbDescription;
+ private TableDescription tableDescription;
+ private Connection connection;
+ private PreparedStatement preparedStatement;
+
+ @BeforeEach
+ void setUp() throws SQLException {
+ preparedStatement = mock(PreparedStatement.class);
+ connection = mock(Connection.class);
+ when(connection.prepareStatement(anyString())).thenReturn(preparedStatement);
+
+ dbDescription = mock(DbDescription.class);
+ when(dbDescription.getAllowedRegEx()).thenReturn("^[a-zA-Z_][a-zA-Z0-9_]*$");
+ when(dbDescription.getEngine()).thenReturn(SupportedDbEngines.POSTGRESQL);
+ when(dbDescription.isColumnNameQuoted()).thenReturn(false);
+
+ tableDescription = new TableDescription(TABLE_NAME, new EventSchema());
+ statementHandler = new StatementHandler(null, null);
+ }
+
+ @Test
+ void testAddToBatch_threeEvents_setsPendingBatchCountToThree() throws SQLException {
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+
+ verify(preparedStatement, times(3)).addBatch();
+ assertEquals(3, statementHandler.getPendingBatchCount());
+ }
+
+ @Test
+ void testAddToBatch_threeEvents_buildsOneInsertStatement() throws SQLException {
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+
+ verify(connection, times(1)).prepareStatement(anyString());
+ }
+
+ @Test
+ void testAddToBatch_threeEvents_clearsValuesOfPreviousEvent() throws SQLException {
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+
+ verify(preparedStatement, times(2)).clearParameters();
+ }
+
+ @Test
+ void testAddToBatch_eventsChange_doesNotLoseCollectedEvents() throws SQLException {
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_PRESSURE);
+
+ verify(preparedStatement, times(1)).executeBatch();
+ verify(connection, times(2)).prepareStatement(anyString());
+ assertEquals(1, statementHandler.getPendingBatchCount());
+ }
+
+ @Test
+ void testExecuteBatch_threeEvents_setsPendingBatchCountToZero() throws SQLException {
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+ statementHandler.addToBatch(dbDescription, tableDescription, connection, EVENT_TEMPERATURE);
+
+ statementHandler.executeBatch();
+
+ verify(preparedStatement, times(1)).executeBatch();
+ assertEquals(0, statementHandler.getPendingBatchCount());
+ }
+
+ @Test
+ void testExecuteBatch_noCollectedEvents_doesNotSendAnything() throws SQLException {
+ statementHandler.preparedStatement = preparedStatement;
+ statementHandler.executeBatch();
+ verify(preparedStatement, never()).executeBatch();
+ }
+
+ @Test
+ void testExecuteBatch_noStatement_throwsNoException() {
+ StatementHandler emptyHandler = new StatementHandler(null, null);
+ assertDoesNotThrow(emptyHandler::executeBatch);
+ }
+}
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/TableDescriptionTest.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/TableDescriptionTest.java
new file mode 100644
index 0000000000..9c6bf2438c
--- /dev/null
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/jdbcclient/model/TableDescriptionTest.java
@@ -0,0 +1,258 @@
+/*
+ * 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.databases.jvm.jdbcclient.model;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.vocabulary.XSD;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Tests for the implementation of the {@link TableDescription} class.
+ */
+class TableDescriptionTest {
+
+ private static final String TABLE_NAME = "sensor_data";
+
+ private static final List<EventProperty> SENSOR_DATA_STREAM = List.of(
+ new EventPropertyPrimitive(XSD.LONG.toString(), "timestamp", "", null),
+ new EventPropertyPrimitive(XSD.STRING.toString(), "sensorId", "", null),
+ new EventPropertyPrimitive(XSD.FLOAT.toString(), "mass_flow", "", null),
+ new EventPropertyPrimitive(XSD.FLOAT.toString(), "volume_flow", "", null)
+ );
+
+ /**
+ * Set up a table description for test cases.
+ *
+ * @param name the table name in the database.
+ * @param dataStream the fields the sink receives from the data stream.
+ * @param cols the columns the table contains.
+ * @return the defined table description.
+ */
+ TableDescription newTable(String name, List<EventProperty> dataStream, Map<String, DbDataTypes> cols) {
+ TableDescription table = new TableDescription(name, new EventSchema(new ArrayList<>(dataStream)));
+ table.setDataTypesHashMap(new HashMap<>(cols));
+ return table;
+ }
+
+ @Test
+ void testValidateTable_tableMatchesStream_throwsNoException() {
+ Map<String, DbDataTypes> cols = Map.of(
+ "timestamp", DbDataTypes.BIGINT,
+ "sensorId", DbDataTypes.VAR_CHAR,
+ "mass_flow", DbDataTypes.REAL,
+ "volume_flow", DbDataTypes.REAL
+ );
+ TableDescription table = newTable(TABLE_NAME, SENSOR_DATA_STREAM, cols);
+
+ try {
+ table.validateTable();
+ } catch (SpRuntimeException e) {
+ fail("Exception on #validateTable: " + e.getMessage());
+ }
+ }
+
+ @Test
+ void testValidateTable_oneColMissing_reportsMissingCol() {
+ Map<String, DbDataTypes> cols = Map.of(
+ "timestamp", DbDataTypes.BIGINT,
+ "sensorId", DbDataTypes.VAR_CHAR,
+ "volume_flow", DbDataTypes.REAL
+ );
+ TableDescription table = newTable(TABLE_NAME, SENSOR_DATA_STREAM, cols);
+
+ try {
+ table.validateTable();
+ fail("No exception on #validateTable");
+ } catch (SpRuntimeException e) {
+ String expected = "Column 'mass_flow' is missing in table '" + TABLE_NAME + "'";
+ assertEquals(expected, e.getMessage());
+ }
+ }
+
+ @Test
+ void testValidateTable_twoColsMissing_reportsMissingCols() {
+ Map<String, DbDataTypes> cols = Map.of(
+ "sensorId", DbDataTypes.VAR_CHAR,
+ "volume_flow", DbDataTypes.REAL
+ );
+ TableDescription table = newTable(TABLE_NAME, SENSOR_DATA_STREAM, cols);
+
+ try {
+ table.validateTable();
+ fail("No exception on #validateTable");
+ } catch (SpRuntimeException e) {
+ String expected = "Columns 'timestamp' and 'mass_flow' are missing in table '" + TABLE_NAME + "'";
+ assertEquals(expected, e.getMessage());
+ }
+ }
+
+ @Test
+ void testValidateTable_threeColsMissing_reportsMissingCols() {
+ Map<String, DbDataTypes> cols = Map.of(
+ "volume_flow", DbDataTypes.REAL
+ );
+ TableDescription table = newTable(TABLE_NAME, SENSOR_DATA_STREAM, cols);
+
+ try {
+ table.validateTable();
+ fail("No exception on #validateTable");
+ } catch (SpRuntimeException e) {
+ String expected = "Columns 'timestamp', 'sensorId', and 'mass_flow' are missing in table '" + TABLE_NAME + "'";
+ assertEquals(expected, e.getMessage());
+ }
+ }
+
+ @Test
+ void testValidateTable_allColsMissing_reportsMissingCols() {
+ Map<String, DbDataTypes> cols = Map.of(
+ "order_id", DbDataTypes.BIGINT,
+ "customer", DbDataTypes.VAR_CHAR
+ );
+ TableDescription table = newTable(TABLE_NAME, SENSOR_DATA_STREAM, cols);
+
+ try {
+ table.validateTable();
+ fail("No exception on #validateTable");
+ } catch (SpRuntimeException e) {
+ String expected =
+ "Columns 'timestamp', 'sensorId', 'mass_flow', and 'volume_flow' are missing in table '" + TABLE_NAME + "'";
+ assertEquals(expected, e.getMessage());
+ }
+ }
+
+ @Test
+ void testValidateTable_oneTypeMismatch_reportsExpectedType() {
+ Map<String, DbDataTypes> cols = Map.of(
+ "timestamp", DbDataTypes.BIGINT,
+ "sensorId", DbDataTypes.VAR_CHAR,
+ "mass_flow", DbDataTypes.BIGINT,
+ "volume_flow", DbDataTypes.REAL
+ );
+ TableDescription table = newTable(TABLE_NAME, SENSOR_DATA_STREAM, cols);
+
+ try {
+ table.validateTable();
+ fail("No exception on #validateTable");
+ } catch (SpRuntimeException e) {
+ String expected =
+ "Type mismatch in table '" + TABLE_NAME + "' for column "
+ + "'mass_flow' (the data stream provides float but the table column is BIGINT)";
+ assertEquals(expected, e.getMessage());
+ }
+ }
+
+ @Test
+ void testValidateTable_twoTypeMismatches_reportsExpectedTypes() {
+ Map<String, DbDataTypes> cols = Map.of(
+ "timestamp", DbDataTypes.BIGINT,
+ "sensorId", DbDataTypes.VAR_CHAR,
+ "mass_flow", DbDataTypes.BIGINT,
+ "volume_flow", DbDataTypes.BOOLEAN
+ );
+ TableDescription table = newTable(TABLE_NAME, SENSOR_DATA_STREAM, cols);
+
+ try {
+ table.validateTable();
+ fail("No exception on #validateTable");
+ } catch (SpRuntimeException e) {
+ String expected =
+ "Type mismatch in table '" + TABLE_NAME + "' for columns "
+ + "'mass_flow' (the data stream provides float but the table column is BIGINT) "
+ + "and 'volume_flow' (the data stream provides float but the table column is BOOLEAN)";
+ assertEquals(expected, e.getMessage());
+ }
+ }
+
+ @Test
+ void testValidateTable_threeTypeMismatches_reportsExpectedTypes() {
+ Map<String, DbDataTypes> cols = Map.of(
+ "timestamp", DbDataTypes.BOOLEAN,
+ "sensorId", DbDataTypes.VAR_CHAR,
+ "mass_flow", DbDataTypes.BIGINT,
+ "volume_flow", DbDataTypes.BOOLEAN
+ );
+ TableDescription table = newTable(TABLE_NAME, SENSOR_DATA_STREAM, cols);
+
+ try {
+ table.validateTable();
+ fail("No exception on #validateTable");
+ } catch (SpRuntimeException e) {
+ String expected =
+ "Type mismatch in table '" + TABLE_NAME + "' for columns "
+ + "'timestamp' (the data stream provides long but the table column is BOOLEAN), "
+ + "'mass_flow' (the data stream provides float but the table column is BIGINT), "
+ + "and 'volume_flow' (the data stream provides float but the table column is BOOLEAN)";
+ assertEquals(expected, e.getMessage());
+ }
+ }
+
+ @Test
+ void testValidateTable_noTypeMatch_reportsExpectedTypes() {
+ Map<String, DbDataTypes> cols = Map.of(
+ "timestamp", DbDataTypes.BOOLEAN,
+ "sensorId", DbDataTypes.BIGINT,
+ "mass_flow", DbDataTypes.BIGINT,
+ "volume_flow", DbDataTypes.BOOLEAN
+ );
+ TableDescription table = newTable(TABLE_NAME, SENSOR_DATA_STREAM, cols);
+
+ try {
+ table.validateTable();
+ fail("No exception on #validateTable");
+ } catch (SpRuntimeException e) {
+ String expected =
+ "Type mismatch in table '" + TABLE_NAME + "' for columns "
+ + "'timestamp' (the data stream provides long but the table column is BOOLEAN), "
+ + "'sensorId' (the data stream provides string but the table column is BIGINT), "
+ + "'mass_flow' (the data stream provides float but the table column is BIGINT), "
+ + "and 'volume_flow' (the data stream provides float but the table column is BOOLEAN)";
+ assertEquals(expected, e.getMessage());
+ }
+ }
+
+ @Test
+ void testValidateTable_colMissingAndTypeMismatch_reportsMissingCol() {
+ Map<String, DbDataTypes> cols = Map.of(
+ "timestamp", DbDataTypes.BIGINT,
+ "sensorId", DbDataTypes.VAR_CHAR,
+ "mass_flow", DbDataTypes.BIGINT
+ );
+ TableDescription table = newTable(TABLE_NAME, SENSOR_DATA_STREAM, cols);
+
+ try {
+ table.validateTable();
+ fail("No exception on #validateTable");
+ } catch (SpRuntimeException e) {
+ String expected = "Column 'volume_flow' is missing in table '" + TABLE_NAME + "'";
+ assertEquals(expected, e.getMessage());
+ }
+ }
+}
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlParametersTest.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlParametersTest.java
new file mode 100644
index 0000000000..cb41e8dc12
--- /dev/null
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlParametersTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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.databases.jvm.postgresql;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for the implementation of the {@link PostgreSqlParameters} class.
+ */
+class PostgreSqlParametersTest {
+
+ private static final String HOST = "localhost";
+ private static final String DB_NAME = "streampipes";
+ private static final String TABLE_NAME = "sensor_data";
+ private static final String USER = "user";
+ private static final String PASSWORD = "secret";
+ private static final int PORT = 5432;
+ private static final boolean SSL_ENABLED = false;
+
+ PostgreSqlParameters newParameters(Boolean allowNewTableCreation, Integer batchSize) {
+ return new PostgreSqlParameters(
+ null, HOST, PORT, DB_NAME, TABLE_NAME, USER, PASSWORD, SSL_ENABLED,
+ allowNewTableCreation, batchSize);
+ }
+
+ @Test
+ void testAllowsNewTableCreation_isMissing_returnsTrue() {
+ PostgreSqlParameters param = newParameters(null, 1);
+ assertTrue(param.allowsNewTableCreation());
+ }
+
+ @Test
+ void testAllowsNewTableCreation_isOn_returnsTrue() {
+ PostgreSqlParameters param = newParameters(true, 1);
+ assertTrue(param.allowsNewTableCreation());
+ }
+
+ @Test
+ void testAllowsNewTableCreation_isOff_returnsFalse() {
+ PostgreSqlParameters param = newParameters(false, 1);
+ assertFalse(param.allowsNewTableCreation());
+ }
+
+ @Test
+ void testGetBatchSize_isMissing_returnsOne() {
+ PostgreSqlParameters param = newParameters(true, null);
+ assertEquals(1, param.getBatchSize());
+ }
+
+ @Test
+ void testGetBatchSize_isZero_returnsOne() {
+ PostgreSqlParameters param = newParameters(true, 0);
+ assertEquals(1, param.getBatchSize());
+ }
+
+ @Test
+ void testGetBatchSize_isNegative_returnsOne() {
+ PostgreSqlParameters param = newParameters(true, -5);
+ assertEquals(1, param.getBatchSize());
+ }
+
+ @Test
+ void testGetBatchSize_isPositive_returnsPositive() {
+ PostgreSqlParameters param = newParameters(true, 5);
+ assertEquals(5, param.getBatchSize());
+ }
+}
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlSinkTest.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlSinkTest.java
new file mode 100644
index 0000000000..195f5f8dfa
--- /dev/null
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlSinkTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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.databases.jvm.postgresql;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.extensions.api.extractor.IDataSinkParameterExtractor;
+import org.apache.streampipes.extensions.api.pe.param.IDataSinkParameters;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for the implementation of the {@link PostgreSqlSink} class.
+ */
+class PostgreSqlSinkTest {
+
+ private PostgreSqlSink sink;
+
+ /**
+ * Set up the values a user entered in the sink configuration.
+ * Only the batch size is filled in.
+ *
+ * @param batchSize the batch size the user entered.
+ * @return the parameters the sink receives when a pipeline is started.
+ */
+ IDataSinkParameters newParameters(Integer batchSize) {
+ IDataSinkParameterExtractor extractor = mock(IDataSinkParameterExtractor.class);
+ when(extractor.singleValueParameter(PostgreSqlSink.BATCH_SIZE_KEY, Integer.class)).thenReturn(batchSize);
+
+ IDataSinkParameters parameters = mock(IDataSinkParameters.class);
+ when(parameters.extractor()).thenReturn(extractor);
+ return parameters;
+ }
+
+ @BeforeEach
+ void setUp() {
+ sink = new PostgreSqlSink();
+ }
+
+ @Test
+ void testOnPipelineStarted_batchSizeIsMissing_throwsException() {
+ IDataSinkParameters parameters = newParameters(null);
+
+ try {
+ sink.onPipelineStarted(parameters, null);
+ fail("No exception on #onPipelineStarted");
+ } catch (SpRuntimeException e) {
+ String expected = "Batch size must be at least 1, but was 'null'. "
+ + "Use 1 to write each event immediately, or a higher value to write events in batches.";
+ assertEquals(expected, e.getMessage());
+ }
+ }
+
+ @Test
+ void testOnPipelineStarted_batchSizeIsZero_throwsException() {
+ int batchSize = 0;
+
+ IDataSinkParameters parameters = newParameters(batchSize);
+
+ try {
+ sink.onPipelineStarted(parameters, null);
+ fail("No exception on #onPipelineStarted");
+ } catch (SpRuntimeException e) {
+ String expected = "Batch size must be at least 1, but was '" + batchSize + "'. "
+ + "Use 1 to write each event immediately, or a higher value to write events in batches.";
+ assertEquals(expected, e.getMessage());
+ }
+ }
+
+ @Test
+ void testOnPipelineStarted_batchSizeIsNegative_throwsException() {
+ int batchSize = -5;
+
+ IDataSinkParameters parameters = newParameters(batchSize);
+
+ try {
+ sink.onPipelineStarted(parameters, null);
+ fail("No exception on #onPipelineStarted");
+ } catch (SpRuntimeException e) {
+ String expected = "Batch size must be at least 1, but was '" + batchSize + "'. "
+ + "Use 1 to write each event immediately, or a higher value to write events in batches.";
+ assertEquals(expected, e.getMessage());
+ }
+ }
+}
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/postgresql/migration/PostgreSqlSinkMigrationV1Test.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/postgresql/migration/PostgreSqlSinkMigrationV1Test.java
new file mode 100644
index 0000000000..2d0d829e0d
--- /dev/null
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/postgresql/migration/PostgreSqlSinkMigrationV1Test.java
@@ -0,0 +1,138 @@
+/*
+ * 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.databases.jvm.postgresql.migration;
+
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.migration.ModelMigratorConfig;
+import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
+import org.apache.streampipes.model.staticproperty.SlideToggleStaticProperty;
+import org.apache.streampipes.model.staticproperty.StaticProperty;
+import org.apache.streampipes.sinks.databases.jvm.postgresql.PostgreSqlSink;
+import org.apache.streampipes.vocabulary.XSD;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for the implementation of the {@link PostgreSqlSinkMigrationV1} class.
+ */
+class PostgreSqlSinkMigrationV1Test {
+
+ private static final List<StaticProperty> VERSION_ZERO_FIELDS = List.of(
+ new FreeTextStaticProperty("db_host", "db_host", "", XSD.STRING),
+ new FreeTextStaticProperty("db_port", "db_port", "", XSD.STRING),
+ new FreeTextStaticProperty("db_name", "db_name", "", XSD.STRING),
+ new FreeTextStaticProperty("db_table", "db_table", "", XSD.STRING),
+ new FreeTextStaticProperty("db_user", "db_user", "", XSD.STRING),
+ new FreeTextStaticProperty("db_password", "db_password", "", XSD.STRING),
+ new FreeTextStaticProperty("ssl_mode", "ssl_mode", "", XSD.STRING)
+ );
+
+ private DataSinkInvocation dataSink;
+
+ /**
+ * Set up a PostgreSQL sink as it was saved with version 0 and run the migration on it.
+ */
+ @BeforeEach
+ void setUp() {
+ dataSink = new DataSinkInvocation();
+ dataSink.setStaticProperties(new ArrayList<>(VERSION_ZERO_FIELDS));
+ new PostgreSqlSinkMigrationV1().migrate(dataSink, null);
+ }
+
+ /**
+ * Collect the names of all fields of the migrated data sink, in the order they are shown.
+ *
+ * @return the internal names of its fields.
+ */
+ List<String> getFieldNames() {
+ return dataSink.getStaticProperties().stream()
+ .map(StaticProperty::getInternalName)
+ .toList();
+ }
+
+ /**
+ * Look up a single field of the migrated data sink.
+ *
+ * @param name the internal name of the field.
+ * @return the field the migration added under that name.
+ */
+ StaticProperty findField(String name) {
+ return dataSink.getStaticProperties().stream()
+ .filter(field -> name.equals(field.getInternalName()))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("migration did not add a field named '" + name + "'"));
+ }
+
+ @Test
+ void testConfig_migratesFromZeroToOne() {
+ ModelMigratorConfig config = new PostgreSqlSinkMigrationV1().config();
+ assertEquals(0, config.fromVersion());
+ assertEquals(1, config.toVersion());
+ }
+
+ @Test
+ void testConfig_targetsPostgreSqlSink() {
+ ModelMigratorConfig config = new PostgreSqlSinkMigrationV1().config();
+ assertEquals(PostgreSqlSink.ID, config.targetAppId());
+ assertEquals(SpServiceTagPrefix.DATA_SINK, config.modelType());
+ }
+
+ @Test
+ void testMigrate_keepsExistingFieldsAndAddsNewSinkOptions() {
+ List<String> expected = List.of(
+ "db_host", "db_port", "db_name", "db_table", "allow_new_table_creation",
+ "db_user", "db_password", "ssl_mode", "batch_size"
+ );
+ assertEquals(expected, getFieldNames());
+ }
+
+ @Test
+ void testMigrate_turnsToggleOn() {
+ SlideToggleStaticProperty toggle =
+ (SlideToggleStaticProperty) findField(PostgreSqlSink.ALLOW_NEW_TABLE_CREATION_KEY);
+ assertTrue(toggle.isSelected(), "version 0 sinks created missing tables, so the toggle has to be on");
+ }
+
+ @Test
+ void testMigrate_setsBatchSizeToOne() {
+ FreeTextStaticProperty batchSize = (FreeTextStaticProperty) findField(PostgreSqlSink.BATCH_SIZE_KEY);
+ assertEquals("1", batchSize.getValue());
+ assertEquals(XSD.INTEGER, batchSize.getRequiredDatatype());
+ }
+
+ @Test
+ void testMigrate_labelsNewSinkOptions() {
+ StaticProperty toggle = findField(PostgreSqlSink.ALLOW_NEW_TABLE_CREATION_KEY);
+ StaticProperty batchSize = findField(PostgreSqlSink.BATCH_SIZE_KEY);
+
+ assertFalse(toggle.getLabel().isBlank());
+ assertFalse(toggle.getDescription().isBlank());
+ assertFalse(batchSize.getLabel().isBlank());
+ assertFalse(batchSize.getDescription().isBlank());
+ }
+}