diff --git a/pom.xml b/pom.xml index 50a7217da719f..6fc4262077799 100644 --- a/pom.xml +++ b/pom.xml @@ -158,7 +158,7 @@ flexible messaging model and an intuitive client API. 2.10.5 2.5.0 5.1.0 - 3.8.11.2 + 3.36.0.3 8.0.11 42.3.3 0.3.2 diff --git a/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/BaseJdbcAutoSchemaSink.java b/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/BaseJdbcAutoSchemaSink.java index e5efb15362a21..17cfe4cd97b26 100644 --- a/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/BaseJdbcAutoSchemaSink.java +++ b/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/BaseJdbcAutoSchemaSink.java @@ -21,8 +21,8 @@ import com.fasterxml.jackson.databind.JsonNode; import com.google.common.annotations.VisibleForTesting; -import com.google.common.collect.Lists; import java.sql.PreparedStatement; +import java.util.ArrayList; import java.util.HashMap; import java.util.Iterator; import java.util.List; @@ -44,37 +44,28 @@ public abstract class BaseJdbcAutoSchemaSink extends JdbcAbstractSink { @Override - public void bindValue(PreparedStatement statement, - Record message, String action) throws Exception { - final GenericObject record = message.getValue(); - Function recordValueGetter; - if (message.getSchema() != null && message.getSchema() instanceof KeyValueSchema) { - KeyValueSchema keyValueSchema = (KeyValueSchema) message.getSchema(); - - final org.apache.pulsar.client.api.Schema keySchema = keyValueSchema.getKeySchema(); - final org.apache.pulsar.client.api.Schema valueSchema = keyValueSchema.getValueSchema(); - KeyValue keyValue = - (KeyValue) record.getNativeObject(); - - final GenericObject key = keyValue.getKey(); - final GenericObject value = keyValue.getValue(); - - Map data = new HashMap<>(); - fillKeyValueSchemaData(keySchema, key, data); - fillKeyValueSchemaData(valueSchema, value, data); - recordValueGetter = (k) -> data.get(k); - } else { - recordValueGetter = (key) -> ((GenericRecord) record).getField(key); - } + public String generateUpsertQueryStatement() { + throw new IllegalStateException("UPSERT not supported"); + } - List columns = Lists.newArrayList(); - if (action == null || action.equals(INSERT)) { - columns = tableDefinition.getColumns(); - } else if (action.equals(DELETE)){ - columns.addAll(tableDefinition.getKeyColumns()); - } else if (action.equals(UPDATE)){ - columns.addAll(tableDefinition.getNonKeyColumns()); - columns.addAll(tableDefinition.getKeyColumns()); + @Override + public void bindValue(PreparedStatement statement, Mutation mutation) throws Exception { + final List columns = new ArrayList<>(); + switch (mutation.getType()) { + case INSERT: + columns.addAll(tableDefinition.getColumns()); + break; + case UPSERT: + columns.addAll(tableDefinition.getColumns()); + columns.addAll(tableDefinition.getNonKeyColumns()); + break; + case UPDATE: + columns.addAll(tableDefinition.getNonKeyColumns()); + columns.addAll(tableDefinition.getKeyColumns()); + break; + case DELETE: + columns.addAll(tableDefinition.getKeyColumns()); + break; } int index = 1; @@ -82,10 +73,10 @@ public void bindValue(PreparedStatement statement, String colName = columnId.getName(); int colType = columnId.getType(); if (log.isDebugEnabled()) { - log.debug("colName: {} colType: {}", colName, colType); + log.debug("getting value for column: {} type: {}", colName, colType); } try { - Object obj = recordValueGetter.apply(colName); + Object obj = mutation.getValues().apply(colName); if (obj != null) { setColumnValue(statement, index++, obj); } else { @@ -105,6 +96,66 @@ public void bindValue(PreparedStatement statement, } } + @Override + public Mutation createMutation(Record message) { + final GenericObject record = message.getValue(); + Function recordValueGetter; + MutationType mutationType = null; + if (message.getSchema() != null && message.getSchema() instanceof KeyValueSchema) { + KeyValueSchema keyValueSchema = (KeyValueSchema) message.getSchema(); + + final org.apache.pulsar.client.api.Schema keySchema = keyValueSchema.getKeySchema(); + final org.apache.pulsar.client.api.Schema valueSchema = keyValueSchema.getValueSchema(); + KeyValue keyValue = + (KeyValue) record.getNativeObject(); + + final GenericObject key = keyValue.getKey(); + final GenericObject value = keyValue.getValue(); + + boolean isDelete = false; + if (value == null) { + switch (jdbcSinkConfig.getNullValueAction()) { + case DELETE: + isDelete = true; + break; + case FAIL: + throw new IllegalArgumentException("Got record with value NULL with nullValueAction=FAIL"); + default: + break; + } + } + Map data = new HashMap<>(); + fillKeyValueSchemaData(keySchema, key, data); + if (isDelete) { + mutationType = MutationType.DELETE; + } else { + fillKeyValueSchemaData(valueSchema, value, data); + } + recordValueGetter = (k) -> data.get(k); + } else { + recordValueGetter = (key) -> ((GenericRecord) record).getField(key); + } + String action = message.getProperties().get(ACTION_PROPERTY); + if (action != null) { + mutationType = MutationType.valueOf(action); + } else if (mutationType == null) { + switch (jdbcSinkConfig.getInsertMode()) { + case INSERT: + mutationType = MutationType.INSERT; + break; + case UPSERT: + mutationType = MutationType.UPSERT; + break; + case UPDATE: + mutationType = MutationType.UPDATE; + break; + default: + throw new IllegalArgumentException("Unknown insert mode: " + jdbcSinkConfig.getInsertMode()); + } + } + return new Mutation(mutationType, recordValueGetter); + } + private static void setColumnNull(PreparedStatement statement, int index, int type) throws Exception { if (log.isDebugEnabled()) { log.debug("Setting column value to null, statement: {}, index: {}", statement.toString(), index); @@ -163,6 +214,9 @@ private static Object getValueFromJsonNode(final JsonNode fn) { private static void fillKeyValueSchemaData(org.apache.pulsar.client.api.Schema schema, GenericObject record, Map data) { + if (record == null) { + return; + } switch (schema.getSchemaInfo().getType()) { case JSON: final JsonNode jsonNode = (JsonNode) record.getNativeObject(); @@ -190,6 +244,9 @@ private static void fillKeyValueSchemaData(org.apache.pulsar.client.api.Schema implements Sink { // ----- Runtime fields - private JdbcSinkConfig jdbcSinkConfig; + protected JdbcSinkConfig jdbcSinkConfig; @Getter private Connection connection; private String jdbcUrl; @@ -52,13 +55,11 @@ public abstract class JdbcAbstractSink implements Sink { private JdbcUtils.TableId tableId; private PreparedStatement insertStatement; private PreparedStatement updateStatement; + private PreparedStatement upsertStatement; private PreparedStatement deleteStatement; - protected static final String ACTION = "ACTION"; - protected static final String INSERT = "INSERT"; - protected static final String UPDATE = "UPDATE"; - protected static final String DELETE = "DELETE"; + protected static final String ACTION_PROPERTY = "ACTION"; protected JdbcUtils.TableDefinition tableDefinition; @@ -122,12 +123,15 @@ private void initStatement() throws Exception { } tableDefinition = JdbcUtils.getTableDefinition(connection, tableId, keyList, nonKeyList); - insertStatement = JdbcUtils.buildInsertStatement(connection, JdbcUtils.buildInsertSql(tableDefinition)); + insertStatement = JdbcUtils.buildInsertStatement(connection, generateInsertQueryStatement()); + if (jdbcSinkConfig.getInsertMode() == JdbcSinkConfig.InsertMode.UPSERT) { + upsertStatement = JdbcUtils.buildInsertStatement(connection, generateUpsertQueryStatement()); + } if (!nonKeyList.isEmpty()) { - updateStatement = JdbcUtils.buildUpdateStatement(connection, JdbcUtils.buildUpdateSql(tableDefinition)); + updateStatement = JdbcUtils.buildUpdateStatement(connection, generateUpdateQueryStatement()); } if (!keyList.isEmpty()) { - deleteStatement = JdbcUtils.buildDeleteStatement(connection, JdbcUtils.buildDeleteSql(tableDefinition)); + deleteStatement = JdbcUtils.buildDeleteStatement(connection, generateDeleteQueryStatement()); } } @@ -136,6 +140,18 @@ public void close() throws Exception { if (connection != null && !connection.getAutoCommit()) { connection.commit(); } + if (insertStatement != null) { + insertStatement.close(); + } + if (updateStatement != null) { + updateStatement.close(); + } + if (upsertStatement != null) { + upsertStatement.close(); + } + if (deleteStatement != null) { + deleteStatement.close(); + } if (flushExecutor != null) { flushExecutor.shutdown(); flushExecutor = null; @@ -159,10 +175,40 @@ public void write(Record record) throws Exception { } } + public String generateInsertQueryStatement() { + return JdbcUtils.buildInsertSql(tableDefinition); + } + + public String generateUpdateQueryStatement() { + return JdbcUtils.buildUpdateSql(tableDefinition); + } + + public abstract String generateUpsertQueryStatement(); + + public String generateDeleteQueryStatement() { + return JdbcUtils.buildDeleteSql(tableDefinition); + } + // bind value with a PreparedStetement public abstract void bindValue( PreparedStatement statement, - Record message, String action) throws Exception; + Mutation mutation) throws Exception; + + public abstract Mutation createMutation(Record message); + + @Data + @AllArgsConstructor + protected static class Mutation { + private MutationType type; + private Function values; + } + protected enum MutationType { + INSERT, + UPDATE, + UPSERT, + DELETE + } + private void flush() { // if not in flushing state, do flush, else return; @@ -187,42 +233,44 @@ private void flush() { try { // bind each record value for (Record record : swapList) { - String action = record.getProperties().get(ACTION); - if (action == null) { - action = INSERT; - } - switch (action) { + final Mutation mutation = createMutation(record); + switch (mutation.getType()) { case DELETE: - bindValue(deleteStatement, record, action); + bindValue(deleteStatement, mutation); count += 1; deleteStatement.execute(); break; case UPDATE: - bindValue(updateStatement, record, action); + bindValue(updateStatement, mutation); count += 1; updateStatement.execute(); break; case INSERT: - bindValue(insertStatement, record, action); + bindValue(insertStatement, mutation); count += 1; insertStatement.execute(); break; + case UPSERT: + bindValue(upsertStatement, mutation); + count += 1; + upsertStatement.execute(); + break; default: String msg = String.format( "Unsupported action %s, can be one of %s, or not set which indicate %s", - action, Arrays.asList(INSERT, UPDATE, DELETE), INSERT); + mutation.getType(), Arrays.toString(MutationType.values()), MutationType.INSERT); throw new IllegalArgumentException(msg); } } connection.commit(); swapList.forEach(Record::ack); } catch (Exception e) { - log.error("Got exception ", e); + log.error("Got exception ", e.getMessage(), e); swapList.forEach(Record::fail); } if (swapList.size() != count) { - log.error("Update count {} not match total number of records {}", count, swapList.size()); + log.error("Update count {} not match total number of records {}", count, swapList.size()); } // finish flush diff --git a/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcSinkConfig.java b/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcSinkConfig.java index acd35047b0106..89e48980858ce 100644 --- a/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcSinkConfig.java +++ b/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcSinkConfig.java @@ -86,6 +86,35 @@ public class JdbcSinkConfig implements Serializable { ) private int batchSize = 200; + @FieldDoc( + required = false, + defaultValue = "INSERT", + help = "If it is configured as UPSERT, the sink will use upsert semantics rather than " + + "plain INSERT/UPDATE statements. Upsert semantics refer to atomically adding a new row or " + + "updating the existing row if there is a primary key constraint violation, " + + "which provides idempotence." + ) + private InsertMode insertMode = InsertMode.INSERT; + + @FieldDoc( + required = false, + defaultValue = "FAIL", + help = "How to handle records with null values, possible options are DELETE or FAIL." + ) + private NullValueAction nullValueAction = NullValueAction.FAIL; + + public enum InsertMode { + INSERT, + UPSERT, + UPDATE; + } + + public enum NullValueAction { + FAIL, + DELETE + } + + public static JdbcSinkConfig load(String yamlFile) throws IOException { ObjectMapper mapper = new ObjectMapper(new YAMLFactory()); return mapper.readValue(new File(yamlFile), JdbcSinkConfig.class); diff --git a/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcUtils.java b/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcUtils.java index 96a39016c0b09..637de0be96fb0 100644 --- a/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcUtils.java +++ b/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcUtils.java @@ -198,6 +198,13 @@ public static String buildUpdateSql(TableDefinition table) { builder.append("UPDATE "); builder.append(table.tableId.getTableName()); builder.append(" SET "); + StringJoiner setJoiner = buildUpdateSqlSetPart(table); + builder.append(setJoiner); + builder.append(combationWhere(table.keyColumns)); + return builder.toString(); + } + + public static StringJoiner buildUpdateSqlSetPart(TableDefinition table) { StringJoiner setJoiner = new StringJoiner(","); table.nonKeyColumns.forEach((columnId) ->{ @@ -205,9 +212,7 @@ public static String buildUpdateSql(TableDefinition table) { equals.add(columnId.getName()).add("? "); setJoiner.add(equals.toString()); }); - builder.append(setJoiner.toString()); - builder.append(combationWhere(table.keyColumns)); - return builder.toString(); + return setJoiner; } public static PreparedStatement buildUpdateStatement(Connection connection, String updateSQL) throws SQLException { diff --git a/pulsar-io/jdbc/core/src/test/java/org/apache/pulsar/io/jdbc/BaseJdbcAutoSchemaSinkTest.java b/pulsar-io/jdbc/core/src/test/java/org/apache/pulsar/io/jdbc/BaseJdbcAutoSchemaSinkTest.java index de31466466d95..459468facf206 100644 --- a/pulsar-io/jdbc/core/src/test/java/org/apache/pulsar/io/jdbc/BaseJdbcAutoSchemaSinkTest.java +++ b/pulsar-io/jdbc/core/src/test/java/org/apache/pulsar/io/jdbc/BaseJdbcAutoSchemaSinkTest.java @@ -93,20 +93,20 @@ public void testConvertAvroUnion() { @Test(expectedExceptions = UnsupportedOperationException.class, expectedExceptionsMessageRegExp = "Unsupported avro schema type.*") public void testNotSupportedAvroTypesBytes() { - BaseJdbcAutoSchemaSink.convertAvroField(null, createFieldAndGetSchema((builder) -> + BaseJdbcAutoSchemaSink.convertAvroField(new Object(), createFieldAndGetSchema((builder) -> builder.name("field").type().bytesType().noDefault())); } @Test(expectedExceptions = UnsupportedOperationException.class, expectedExceptionsMessageRegExp = "Unsupported avro schema type.*") public void testNotSupportedAvroTypesFixed() { - BaseJdbcAutoSchemaSink.convertAvroField(null, createFieldAndGetSchema((builder) -> + BaseJdbcAutoSchemaSink.convertAvroField(new Object(), createFieldAndGetSchema((builder) -> builder.name("field").type().fixed("fix").size(16).noDefault())); } @Test(expectedExceptions = UnsupportedOperationException.class, expectedExceptionsMessageRegExp = "Unsupported avro schema type.*") public void testNotSupportedAvroTypesRecord() { - BaseJdbcAutoSchemaSink.convertAvroField(null, createFieldAndGetSchema((builder) -> + BaseJdbcAutoSchemaSink.convertAvroField(new Object(), createFieldAndGetSchema((builder) -> builder.name("field").type() .record("myrecord").fields() .name("f1").type().intType().noDefault() @@ -116,7 +116,7 @@ public void testNotSupportedAvroTypesRecord() { @Test(expectedExceptions = UnsupportedOperationException.class, expectedExceptionsMessageRegExp = "Unsupported avro schema type.*") public void testNotSupportedAvroTypesMap() { - BaseJdbcAutoSchemaSink.convertAvroField(null, createFieldAndGetSchema((builder) -> + BaseJdbcAutoSchemaSink.convertAvroField(new Object(), createFieldAndGetSchema((builder) -> builder.name("field").type().map().values().stringType().noDefault())); } @@ -124,11 +124,19 @@ public void testNotSupportedAvroTypesMap() { @Test(expectedExceptions = UnsupportedOperationException.class, expectedExceptionsMessageRegExp = "Unsupported avro schema type.*") public void testNotSupportedAvroTypesArray() { - BaseJdbcAutoSchemaSink.convertAvroField(null, createFieldAndGetSchema((builder) -> + BaseJdbcAutoSchemaSink.convertAvroField(new Object(), createFieldAndGetSchema((builder) -> builder.name("field").type().array().items().stringType().noDefault())); } + @Test + public void testConvertAvroNullValue() { + Object converted = BaseJdbcAutoSchemaSink.convertAvroField(null, createFieldAndGetSchema((builder) -> + builder.name("field").type().stringType().noDefault())); + Assert.assertNull(converted); + } + + private Schema createFieldAndGetSchema(Function, SchemaBuilder.FieldAssembler> consumer) { final SchemaBuilder.FieldAssembler record = SchemaBuilder.record("record") diff --git a/pulsar-io/jdbc/mariadb/src/main/java/org/apache/pulsar/io/jdbc/MariadbJdbcAutoSchemaSink.java b/pulsar-io/jdbc/mariadb/src/main/java/org/apache/pulsar/io/jdbc/MariadbJdbcAutoSchemaSink.java index 3fc45bb3f7b7a..5c232db1aa06a 100644 --- a/pulsar-io/jdbc/mariadb/src/main/java/org/apache/pulsar/io/jdbc/MariadbJdbcAutoSchemaSink.java +++ b/pulsar-io/jdbc/mariadb/src/main/java/org/apache/pulsar/io/jdbc/MariadbJdbcAutoSchemaSink.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.io.jdbc; +import java.util.stream.Collectors; import org.apache.pulsar.io.core.annotations.Connector; import org.apache.pulsar.io.core.annotations.IOType; @@ -29,4 +30,12 @@ ) public class MariadbJdbcAutoSchemaSink extends BaseJdbcAutoSchemaSink { + @Override + public String generateUpsertQueryStatement() { + final String keys = tableDefinition.getKeyColumns().stream().map(JdbcUtils.ColumnId::getName) + .collect(Collectors.joining(",")); + return JdbcUtils.buildInsertSql(tableDefinition) + + "ON DUPLICATE KEY UPDATE " + JdbcUtils.buildUpdateSqlSetPart(tableDefinition); + } + } diff --git a/pulsar-io/jdbc/postgres/src/main/java/org/apache/pulsar/io/jdbc/PostgresJdbcAutoSchemaSink.java b/pulsar-io/jdbc/postgres/src/main/java/org/apache/pulsar/io/jdbc/PostgresJdbcAutoSchemaSink.java index 9ed122eb3bbc0..d29e3e3aaca47 100644 --- a/pulsar-io/jdbc/postgres/src/main/java/org/apache/pulsar/io/jdbc/PostgresJdbcAutoSchemaSink.java +++ b/pulsar-io/jdbc/postgres/src/main/java/org/apache/pulsar/io/jdbc/PostgresJdbcAutoSchemaSink.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.io.jdbc; +import java.util.stream.Collectors; import org.apache.pulsar.io.core.annotations.Connector; import org.apache.pulsar.io.core.annotations.IOType; @@ -29,4 +30,12 @@ ) public class PostgresJdbcAutoSchemaSink extends BaseJdbcAutoSchemaSink { + @Override + public String generateUpsertQueryStatement() { + final String keys = tableDefinition.getKeyColumns().stream().map(JdbcUtils.ColumnId::getName) + .collect(Collectors.joining(",")); + return JdbcUtils.buildInsertSql(tableDefinition) + + " ON CONFLICT(" + keys + ") " + + "DO UPDATE SET " + JdbcUtils.buildUpdateSqlSetPart(tableDefinition); + } } diff --git a/pulsar-io/jdbc/sqlite/src/main/java/org/apache/pulsar/io/jdbc/SqliteJdbcAutoSchemaSink.java b/pulsar-io/jdbc/sqlite/src/main/java/org/apache/pulsar/io/jdbc/SqliteJdbcAutoSchemaSink.java index 8cfd3f0962ed0..76655397003ca 100644 --- a/pulsar-io/jdbc/sqlite/src/main/java/org/apache/pulsar/io/jdbc/SqliteJdbcAutoSchemaSink.java +++ b/pulsar-io/jdbc/sqlite/src/main/java/org/apache/pulsar/io/jdbc/SqliteJdbcAutoSchemaSink.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.io.jdbc; +import java.util.stream.Collectors; import org.apache.pulsar.io.core.annotations.Connector; import org.apache.pulsar.io.core.annotations.IOType; @@ -29,4 +30,12 @@ ) public class SqliteJdbcAutoSchemaSink extends BaseJdbcAutoSchemaSink { + @Override + public String generateUpsertQueryStatement() { + final String keys = tableDefinition.getKeyColumns().stream().map(JdbcUtils.ColumnId::getName) + .collect(Collectors.joining(",")); + return JdbcUtils.buildInsertSql(tableDefinition) + + " ON CONFLICT(" + keys + ") " + + "DO UPDATE SET " + JdbcUtils.buildUpdateSqlSetPart(tableDefinition); + } } diff --git a/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/JdbcUtilsTest.java b/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/JdbcUtilsTest.java index b3ea296bf6f28..deb2fb88aa800 100644 --- a/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/JdbcUtilsTest.java +++ b/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/JdbcUtilsTest.java @@ -69,53 +69,54 @@ public void TestGetTableId() throws Exception { "PRIMARY KEY (firstName, lastName));" ); - Connection connection = sqliteUtils.getConnection(); + try (Connection connection = sqliteUtils.getConnection(true);) { - // Test getTableId - log.info("verify getTableId"); - TableId id = JdbcUtils.getTableId(connection, tableName); - Assert.assertEquals(id.getTableName(), tableName); + // Test getTableId + log.info("verify getTableId"); + TableId id = JdbcUtils.getTableId(connection, tableName); + Assert.assertEquals(id.getTableName(), tableName); - // Test get getTableDefinition - log.info("verify getTableDefinition"); - List keyList = Lists.newArrayList(); - keyList.add("firstName"); - keyList.add("lastName"); - List nonKeyList = Lists.newArrayList(); - nonKeyList.add("age"); - nonKeyList.add("long"); - TableDefinition table = JdbcUtils.getTableDefinition(connection, id, keyList, nonKeyList); - Assert.assertEquals(table.getColumns().get(0).getName(), "firstName"); - Assert.assertEquals(table.getColumns().get(0).getTypeName(), "TEXT"); - Assert.assertEquals(table.getColumns().get(2).getName(), "age"); - Assert.assertEquals(table.getColumns().get(2).getTypeName(), "INTEGER"); - Assert.assertEquals(table.getColumns().get(7).getName(), "float"); - Assert.assertEquals(table.getColumns().get(7).getTypeName(), "NUMERIC"); - Assert.assertEquals(table.getKeyColumns().get(0).getName(), "firstName"); - Assert.assertEquals(table.getKeyColumns().get(0).getTypeName(), "TEXT"); - Assert.assertEquals(table.getKeyColumns().get(1).getName(), "lastName"); - Assert.assertEquals(table.getKeyColumns().get(1).getTypeName(), "TEXT"); - Assert.assertEquals(table.getNonKeyColumns().get(0).getName(), "age"); - Assert.assertEquals(table.getNonKeyColumns().get(0).getTypeName(), "INTEGER"); - Assert.assertEquals(table.getNonKeyColumns().get(1).getName(), "long"); - Assert.assertEquals(table.getNonKeyColumns().get(1).getTypeName(), "INTEGER"); - // Test get getTableDefinition - log.info("verify buildInsertSql"); - String expctedInsertStatement = "INSERT INTO " + tableName + - "(firstName,lastName,age,bool,byte,short,long,float,double,bytes)" + - " VALUES(?,?,?,?,?,?,?,?,?,?)"; - String insertStatement = JdbcUtils.buildInsertSql(table); - Assert.assertEquals(insertStatement, expctedInsertStatement); - log.info("verify buildUpdateSql"); - String expectedUpdateStatement = "UPDATE " + tableName + - " SET age=? ,long=? WHERE firstName=? AND lastName=?"; - String updateStatement = JdbcUtils.buildUpdateSql(table); - Assert.assertEquals(updateStatement, expectedUpdateStatement); - log.info("verify buildDeleteSql"); - String expectedDeleteStatement = "DELETE FROM " + tableName + - " WHERE firstName=? AND lastName=?"; - String deleteStatement = JdbcUtils.buildDeleteSql(table); - Assert.assertEquals(deleteStatement, expectedDeleteStatement); + // Test get getTableDefinition + log.info("verify getTableDefinition"); + List keyList = Lists.newArrayList(); + keyList.add("firstName"); + keyList.add("lastName"); + List nonKeyList = Lists.newArrayList(); + nonKeyList.add("age"); + nonKeyList.add("long"); + TableDefinition table = JdbcUtils.getTableDefinition(connection, id, keyList, nonKeyList); + Assert.assertEquals(table.getColumns().get(0).getName(), "firstName"); + Assert.assertEquals(table.getColumns().get(0).getTypeName(), "TEXT"); + Assert.assertEquals(table.getColumns().get(2).getName(), "age"); + Assert.assertEquals(table.getColumns().get(2).getTypeName(), "INTEGER"); + Assert.assertEquals(table.getColumns().get(7).getName(), "float"); + Assert.assertEquals(table.getColumns().get(7).getTypeName(), "NUMERIC"); + Assert.assertEquals(table.getKeyColumns().get(0).getName(), "firstName"); + Assert.assertEquals(table.getKeyColumns().get(0).getTypeName(), "TEXT"); + Assert.assertEquals(table.getKeyColumns().get(1).getName(), "lastName"); + Assert.assertEquals(table.getKeyColumns().get(1).getTypeName(), "TEXT"); + Assert.assertEquals(table.getNonKeyColumns().get(0).getName(), "age"); + Assert.assertEquals(table.getNonKeyColumns().get(0).getTypeName(), "INTEGER"); + Assert.assertEquals(table.getNonKeyColumns().get(1).getName(), "long"); + Assert.assertEquals(table.getNonKeyColumns().get(1).getTypeName(), "INTEGER"); + // Test get getTableDefinition + log.info("verify buildInsertSql"); + String expctedInsertStatement = "INSERT INTO " + tableName + + "(firstName,lastName,age,bool,byte,short,long,float,double,bytes)" + + " VALUES(?,?,?,?,?,?,?,?,?,?)"; + String insertStatement = JdbcUtils.buildInsertSql(table); + Assert.assertEquals(insertStatement, expctedInsertStatement); + log.info("verify buildUpdateSql"); + String expectedUpdateStatement = "UPDATE " + tableName + + " SET age=? ,long=? WHERE firstName=? AND lastName=?"; + String updateStatement = JdbcUtils.buildUpdateSql(table); + Assert.assertEquals(updateStatement, expectedUpdateStatement); + log.info("verify buildDeleteSql"); + String expectedDeleteStatement = "DELETE FROM " + tableName + + " WHERE firstName=? AND lastName=?"; + String deleteStatement = JdbcUtils.buildDeleteSql(table); + Assert.assertEquals(deleteStatement, expectedDeleteStatement); + } } } diff --git a/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java b/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java index 1ce3bc5b35358..cb1ca84424931 100644 --- a/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java +++ b/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java @@ -19,8 +19,20 @@ package org.apache.pulsar.io.jdbc; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Maps; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import lombok.AllArgsConstructor; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.apache.avro.util.Utf8; @@ -47,22 +59,12 @@ import org.testng.annotations.BeforeMethod; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; - -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.TimeUnit; - -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - /** * Jdbc Sink test */ @Slf4j public class SqliteJdbcSinkTest { - private final SqliteUtils sqliteUtils = new SqliteUtils(getClass().getSimpleName()); + private final SqliteUtils sqliteUtils = new SqliteUtils(UUID.randomUUID().toString()); private BaseJdbcAutoSchemaSink jdbcSink; private final String tableName = "TestOpenAndWriteSink"; @@ -409,14 +411,87 @@ public void TestDeleteAction() throws Exception { Assert.assertEquals(sqliteUtils.select(deleteQuerySql, (resultSet) -> {}), 0); } - @DataProvider(name = "schemaType") + private static class MockKeyValueGenericRecord implements Record { + + public MockKeyValueGenericRecord(Schema> keyValueSchema) { + this.keyValueSchema = keyValueSchema; + } + + int ackCount; + int failCount; + AtomicReference keyValueHolder = new AtomicReference<>(); + Schema> keyValueSchema; + + private final GenericObject genericObject = new GenericObject() { + @Override + public SchemaType getSchemaType() { + return SchemaType.KEY_VALUE; + } + + @Override + public Object getNativeObject() { + return keyValueHolder.get(); + } + }; + @Override + public Optional getTopicName() { + return Optional.of("topic"); + } + + @Override + public org.apache.pulsar.client.api.Schema getSchema() { + return keyValueSchema; + } + + @Override + public GenericObject getValue() { + return genericObject; + } + + public void setKeyValue(KeyValue kv) { + keyValueHolder.set(kv); + } + + @Override + public void ack() { + ackCount++; + } + + @Override + public void fail() { + failCount++; + } + } + + @Data + @AllArgsConstructor + static class InsertModeTestConfig { + SchemaType schemaType; + JdbcSinkConfig.InsertMode insertMode; + } + + @DataProvider(name = "insertModes") public Object[] schemaType() { - return new Object[]{SchemaType.JSON, SchemaType.AVRO}; + List schemas = Arrays.asList(SchemaType.JSON, SchemaType.AVRO); + JdbcSinkConfig.InsertMode[] insertModes = JdbcSinkConfig.InsertMode.values(); + Object[] result = new Object[schemas.size() * insertModes.length]; + int i = 0; + for (SchemaType schema : schemas) { + for (JdbcSinkConfig.InsertMode insertMode : insertModes) { + result[i++] = new InsertModeTestConfig(schema, insertMode); + } + } + return result; } - @Test(dataProvider = "schemaType") - public void testKeyValueSchema(SchemaType schemaType) throws Exception { + @Test(dataProvider = "insertModes") + public void testInsertMode(InsertModeTestConfig config) throws Exception { + jdbcSink.close(); + final JdbcSinkConfig.InsertMode insertMode = config.getInsertMode(); + final SchemaType schemaType = config.getSchemaType(); + + final String tableName = "kvtable_insertmode"; RecordSchemaBuilder keySchemaBuilder = org.apache.pulsar.client.api.schema.SchemaBuilder.record("key"); keySchemaBuilder.field("key").type(SchemaType.STRING).optional().defaultValue(null); GenericSchema keySchema = Schema.generic(keySchemaBuilder.build(schemaType)); @@ -427,6 +502,7 @@ public void testKeyValueSchema(SchemaType schemaType) throws Exception { RecordSchemaBuilder valueSchemaBuilder = org.apache.pulsar.client.api.schema.SchemaBuilder.record("value"); valueSchemaBuilder.field("string").type(SchemaType.STRING).optional().defaultValue(null); valueSchemaBuilder.field("stringutf8").type(SchemaType.STRING).optional().defaultValue(null); + valueSchemaBuilder.field("nulltext").type(SchemaType.STRING).optional().defaultValue(null); valueSchemaBuilder.field("int").type(SchemaType.INT32).optional().defaultValue(null); valueSchemaBuilder.field("bool").type(SchemaType.BOOLEAN).optional().defaultValue(null); valueSchemaBuilder.field("double").type(SchemaType.DOUBLE).optional().defaultValue(null); @@ -434,7 +510,9 @@ public void testKeyValueSchema(SchemaType schemaType) throws Exception { valueSchemaBuilder.field("long").type(SchemaType.INT64).optional().defaultValue(null); GenericSchema valueSchema = Schema.generic(valueSchemaBuilder.build(schemaType)); - GenericRecord valueGenericRecord = valueSchema.newRecordBuilder() + Schema> keyValueSchema = Schema.KeyValue(keySchema, valueSchema, KeyValueEncodingType.INLINE); + MockKeyValueGenericRecord genericObjectRecord = new MockKeyValueGenericRecord(keyValueSchema); + genericObjectRecord.setKeyValue(new KeyValue<>(keyGenericRecord, valueSchema.newRecordBuilder() .set("string", "thestring") .set("stringutf8", schemaType == SchemaType.AVRO ? new Utf8("thestringutf8"): "thestringutf8") .set("int", Integer.MAX_VALUE) @@ -442,40 +520,10 @@ public void testKeyValueSchema(SchemaType schemaType) throws Exception { .set("double", Double.MAX_VALUE) .set("float", Float.MAX_VALUE) .set("long", Long.MIN_VALUE) - .build(); - - Schema> keyValueSchema = Schema.KeyValue(keySchema, valueSchema, KeyValueEncodingType.INLINE); - KeyValue keyValue = new KeyValue<>(keyGenericRecord, valueGenericRecord); - GenericObject genericObject = new GenericObject() { - @Override - public SchemaType getSchemaType() { - return SchemaType.KEY_VALUE; - } - - @Override - public Object getNativeObject() { - return keyValue; - } - }; - Record genericObjectRecord = new Record<>() { - @Override - public Optional getTopicName() { - return Optional.of("topic"); - } - - @Override - public org.apache.pulsar.client.api.Schema getSchema() { - return keyValueSchema; - } + .build())); - @Override - public GenericObject getValue() { - return genericObject; - } - }; - jdbcSink.close(); sqliteUtils.createTable( - "CREATE TABLE kvtable (" + + "CREATE TABLE " + tableName + " (" + " key TEXT," + " int INTEGER," + " string TEXT," + @@ -491,18 +539,19 @@ public GenericObject getValue() { Map conf = Maps.newHashMap(); conf.put("jdbcUrl", jdbcUrl); - conf.put("tableName", "kvtable"); + conf.put("tableName", tableName); conf.put("key", "key"); conf.put("nonKey", "long,int,double,float,bool,nulltext,string,stringutf8"); // change batchSize to 1, to flush on each write. conf.put("batchSize", 1); + conf.put("insertMode", insertMode.toString()); try (SqliteJdbcAutoSchemaSink kvSchemaJdbcSink = new SqliteJdbcAutoSchemaSink();) { kvSchemaJdbcSink.open(conf, null); kvSchemaJdbcSink.write(genericObjectRecord); Awaitility.await().untilAsserted(() -> { final int count = sqliteUtils.select("select int,string,stringutf8,bool,double,float," + - "long,nulltext from kvtable where key='mykey'", (resultSet) -> { + "long,nulltext from " + tableName + " where key='mykey'", (resultSet) -> { int index = 1; Assert.assertEquals(resultSet.getInt(index++), Integer.MAX_VALUE); Assert.assertEquals(resultSet.getString(index++), "thestring"); @@ -513,9 +562,175 @@ public GenericObject getValue() { Assert.assertEquals(resultSet.getLong(index++), Long.MIN_VALUE); Assert.assertNull(resultSet.getString(index++)); }); - Assert.assertEquals(count, 1); + if (insertMode == JdbcSinkConfig.InsertMode.INSERT + || insertMode == JdbcSinkConfig.InsertMode.UPSERT) { + Assert.assertEquals(count, 1); + } else { + Assert.assertEquals(count, 0); + } + }); + } + + if (insertMode == JdbcSinkConfig.InsertMode.UPDATE) { + conf.put("insertMode", JdbcSinkConfig.InsertMode.INSERT); + try (SqliteJdbcAutoSchemaSink kvSchemaJdbcSink = new SqliteJdbcAutoSchemaSink();) { + kvSchemaJdbcSink.open(conf, null); + kvSchemaJdbcSink.write(genericObjectRecord); + Awaitility.await().untilAsserted(() -> { + final int count = sqliteUtils.select("select key from " + tableName + " where key='mykey'", (resultSet) -> { + }); + Assert.assertEquals(count, 1); + }); + } + } + conf.put("insertMode", insertMode.toString()); + try (SqliteJdbcAutoSchemaSink kvSchemaJdbcSink = new SqliteJdbcAutoSchemaSink();) { + kvSchemaJdbcSink.open(conf, null); + genericObjectRecord = new MockKeyValueGenericRecord(keyValueSchema); + genericObjectRecord.setKeyValue(new KeyValue<>(keyGenericRecord, valueSchema.newRecordBuilder() + .set("string", "thestring_updated") + .set("stringutf8", schemaType == SchemaType.AVRO ? + new Utf8("thestringutf8_updated"): "thestringutf8_updated") + .set("int", Integer.MIN_VALUE) + .set("bool", false) + .set("double", Double.MIN_VALUE) + .set("float", Float.MIN_VALUE) + .set("long", Long.MAX_VALUE) + .set("nulltext", "nomore-null") + .build())); + kvSchemaJdbcSink.write(genericObjectRecord); + + Awaitility.await().untilAsserted(() -> { + final int count = sqliteUtils.select("select int,string,stringutf8,bool,double,float," + + "long,nulltext from " + tableName + " where key='mykey'", (resultSet) -> { + int index = 1; + if (insertMode == JdbcSinkConfig.InsertMode.INSERT) { + Assert.assertEquals(resultSet.getInt(index++), Integer.MAX_VALUE); + Assert.assertEquals(resultSet.getString(index++), "thestring"); + Assert.assertEquals(resultSet.getString(index++), "thestringutf8"); + Assert.assertEquals(resultSet.getBoolean(index++), true); + Assert.assertEquals(resultSet.getDouble(index++), Double.MAX_VALUE); + Assert.assertEquals(resultSet.getFloat(index++), Float.MAX_VALUE); + Assert.assertEquals(resultSet.getLong(index++), Long.MIN_VALUE); + Assert.assertNull(resultSet.getString(index++)); + } else { + Assert.assertEquals(resultSet.getInt(index++), Integer.MIN_VALUE); + Assert.assertEquals(resultSet.getString(index++), "thestring_updated"); + Assert.assertEquals(resultSet.getString(index++), "thestringutf8_updated"); + Assert.assertEquals(resultSet.getBoolean(index++), false); + Assert.assertEquals(resultSet.getDouble(index++), Double.MIN_VALUE); + Assert.assertEquals(resultSet.getFloat(index++), Float.MIN_VALUE); + Assert.assertEquals(resultSet.getLong(index++), Long.MAX_VALUE); + Assert.assertEquals(resultSet.getString(index++), "nomore-null"); + } + }); + if (insertMode == JdbcSinkConfig.InsertMode.INSERT) { + Assert.assertEquals(count, 1); + } else if (insertMode == JdbcSinkConfig.InsertMode.UPSERT) { + Assert.assertEquals(count, 1); + } else { + Assert.assertEquals(count, 1); + } }); } } + @Data + @AllArgsConstructor + static class NullValueActionTestConfig { + SchemaType schemaType; + JdbcSinkConfig.NullValueAction nullValueAction; + } + @DataProvider(name = "nullValueActions") + public Object[] nullValueActions() { + List schemas = Arrays.asList(SchemaType.JSON, SchemaType.AVRO); + JdbcSinkConfig.NullValueAction[] nullValueActions = JdbcSinkConfig.NullValueAction.values(); + Object[] result = new Object[schemas.size() * nullValueActions.length]; + int i = 0; + for (SchemaType schema : schemas) { + for (JdbcSinkConfig.NullValueAction nullValueAction : nullValueActions) { + result[i++] = new NullValueActionTestConfig(schema, nullValueAction); + } + } + return result; + } + + @Test(dataProvider = "nullValueActions") + public void testNullValueAction(NullValueActionTestConfig config) throws Exception { + jdbcSink.close(); + final SchemaType schemaType = config.getSchemaType(); + final JdbcSinkConfig.NullValueAction nullValueAction = config.getNullValueAction(); + final String tableName = "kvtable_nullvalue"; + + RecordSchemaBuilder keySchemaBuilder = org.apache.pulsar.client.api.schema.SchemaBuilder.record("key"); + keySchemaBuilder.field("key").type(SchemaType.STRING).optional().defaultValue(null); + GenericSchema keySchema = Schema.generic(keySchemaBuilder.build(schemaType)); + GenericRecord keyGenericRecord = keySchema.newRecordBuilder() + .set("key", "mykey") + .build(); + + RecordSchemaBuilder valueSchemaBuilder = org.apache.pulsar.client.api.schema.SchemaBuilder.record("value"); + valueSchemaBuilder.field("string").type(SchemaType.STRING).optional().defaultValue(null); + GenericSchema valueSchema = Schema.generic(valueSchemaBuilder.build(schemaType)); + + Schema> keyValueSchema = Schema.KeyValue(keySchema, valueSchema, KeyValueEncodingType.INLINE); + MockKeyValueGenericRecord genericObjectRecord = new MockKeyValueGenericRecord(keyValueSchema); + + sqliteUtils.createTable("CREATE TABLE " + tableName + " (" + + " key TEXT," + + " string TEXT," + + "PRIMARY KEY (key));" + ); + String jdbcUrl = sqliteUtils.sqliteUri(); + + Map conf = Maps.newHashMap(); + conf.put("jdbcUrl", jdbcUrl); + conf.put("tableName", tableName); + conf.put("key", "key"); + conf.put("nonKey", "string"); + conf.put("batchSize", 3); + conf.put("insertMode", JdbcSinkConfig.InsertMode.UPSERT.toString()); + conf.put("nullValueAction", nullValueAction.toString()); + try (SqliteJdbcAutoSchemaSink kvSchemaJdbcSink = new SqliteJdbcAutoSchemaSink();) { + kvSchemaJdbcSink.open(conf, null); + genericObjectRecord.setKeyValue(new KeyValue<>(keyGenericRecord, valueSchema.newRecordBuilder() + .set("string", "thestring") + .build())); + kvSchemaJdbcSink.write(genericObjectRecord); + + genericObjectRecord = new MockKeyValueGenericRecord(keyValueSchema); + genericObjectRecord.setKeyValue(new KeyValue<>(keySchema.newRecordBuilder() + .set("key", "mykey2") + .build(), valueSchema.newRecordBuilder() + .set("string", "thestring") + .build())); + kvSchemaJdbcSink.write(genericObjectRecord); + + genericObjectRecord = new MockKeyValueGenericRecord(keyValueSchema); + genericObjectRecord.setKeyValue(new KeyValue<>(keyGenericRecord, null)); + kvSchemaJdbcSink.write(genericObjectRecord); + + Awaitility.await().untilAsserted(() -> { + final int count = sqliteUtils.select("select key,string from " + tableName, (resultSet) -> { + int index = 1; + final String key = resultSet.getString(index++); + final String value = resultSet.getString(index++); + if (key.equals("mykey2")) { + Assert.assertEquals(value, "thestring"); + } else { + throw new IllegalStateException(); + } + }); + if (nullValueAction == JdbcSinkConfig.NullValueAction.DELETE) { + Assert.assertEquals(count, 1); + } else if (nullValueAction == JdbcSinkConfig.NullValueAction.FAIL) { + Assert.assertEquals(count, 0); + } else { + throw new IllegalStateException(); + } + }); + + } + } + } diff --git a/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteUtils.java b/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteUtils.java index bb855d64bf17d..f8cdf76f7d35d 100644 --- a/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteUtils.java +++ b/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteUtils.java @@ -49,9 +49,9 @@ public interface ResultSetReadCallback { private final Path dbPath; - private Connection connection; - - public Connection getConnection() { + public Connection getConnection(boolean autocommit) throws SQLException { + Connection connection = DriverManager.getConnection(sqliteUri()); + connection.setAutoCommit(autocommit); return connection; } @@ -65,12 +65,9 @@ public String sqliteUri() { public void setUp() throws SQLException, IOException { Files.deleteIfExists(dbPath); - connection = DriverManager.getConnection(sqliteUri()); - connection.setAutoCommit(false); } public void tearDown() throws SQLException, IOException { - connection.close(); Files.deleteIfExists(dbPath); } @@ -91,7 +88,8 @@ public void deleteTable(final String table) throws SQLException { public int select(final String query, final ResultSetReadCallback callback) throws SQLException { int count = 0; - try (Statement stmt = connection.createStatement()) { + try (Connection connection = getConnection(true); + Statement stmt = connection.createStatement()) { try (ResultSet rs = stmt.executeQuery(query)) { while (rs.next()) { callback.read(rs); @@ -103,9 +101,9 @@ public int select(final String query, final ResultSetReadCallback callback) thro } public void execute(String sql) throws SQLException { - try (Statement stmt = connection.createStatement()) { + try (Connection connection = getConnection(true); + Statement stmt = connection.createStatement()) { stmt.executeUpdate(sql); - connection.commit(); } } diff --git a/site2/docs/io-jdbc-sink.md b/site2/docs/io-jdbc-sink.md index 87b06c544c40a..abc028132c37d 100644 --- a/site2/docs/io-jdbc-sink.md +++ b/site2/docs/io-jdbc-sink.md @@ -8,6 +8,7 @@ The JDBC sink connectors allow pulling messages from Pulsar topics and persists the messages to ClickHouse, MariaDB, PostgreSQL, and SQLite. > Currently, INSERT, DELETE and UPDATE operations are supported. +> SQLite, MariaDB and PostgreSQL also support UPSERT operations and idempotent writes. ## Configuration @@ -25,6 +26,8 @@ The configuration of all JDBC sink connectors has the following properties. | `key` | String | false | " " (empty string) | A comma-separated list containing the fields used in `where` condition of updating and deleting events. | | `timeoutMs` | int | false | 500 | The JDBC operation timeout in milliseconds. | | `batchSize` | int | false | 200 | The batch size of updates made to the database. | +| `insertMode` | enum( INSERT,UPSERT,UPDATE) | false | INSERT | If it is configured as UPSERT, the sink uses upsert semantics rather than plain INSERT/UPDATE statements. Upsert semantics refer to atomically adding a new row or updating the existing row if there is a primary key constraint violation, which provides idempotence. | +| `nullValueAction` | enum(FAIL, DELETE) | false | FAIL | How to handle records with NULL values. Possible options are `DELETE` or `FAIL`. | ### Example for ClickHouse