This is an automated email from the ASF dual-hosted git repository.
mthomsen pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 405c393 NIFI-8031: Add UPSERT capability for MySQL in
PutDatabaseRecord
405c393 is described below
commit 405c393cb59407aec89280934e353db7252b4b20
Author: Matthew Burgess <[email protected]>
AuthorDate: Thu Nov 19 15:52:47 2020 -0500
NIFI-8031: Add UPSERT capability for MySQL in PutDatabaseRecord
This closes #4678
Signed-off-by: Mike Thomsen <[email protected]>
---
.../processors/standard/PutDatabaseRecord.java | 12 ++-
.../processors/standard/db/DatabaseAdapter.java | 10 ++
.../standard/db/impl/MySQLDatabaseAdapter.java | 54 ++++++++++
.../standard/db/impl/TestMySQLDatabaseAdapter.java | 109 +++++++++++++++++++++
4 files changed, 184 insertions(+), 1 deletion(-)
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java
index d6384d2..0a27ebb 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutDatabaseRecord.java
@@ -450,7 +450,7 @@ public class PutDatabaseRecord extends
AbstractSessionFactoryProcessor {
}, (fc, inputFlowFile, r, e) -> {
- getLogger().warn("Failed to process {} due to {}", new
Object[]{inputFlowFile, e}, e);
+ getLogger().error("Failed to process {} due to {}", new
Object[]{inputFlowFile, e}, e);
// Check if there was a BatchUpdateException or if multiple SQL
statements were being executed and one failed
final String statementTypeProperty =
context.getProperty(STATEMENT_TYPE).getValue();
@@ -752,6 +752,11 @@ public class PutDatabaseRecord extends
AbstractSessionFactoryProcessor {
if (DELETE_TYPE.equalsIgnoreCase(statementType)) {
ps.setObject(i * 2 + 1, currentValue, sqlType);
ps.setObject(i * 2 + 2, currentValue, sqlType);
+ } else if
(UPSERT_TYPE.equalsIgnoreCase(statementType)) {
+ final int timesToAddObjects =
databaseAdapter.getTimesToAddColumnObjectsForUpsert();
+ for (int j = 0; j < timesToAddObjects; j++) {
+ ps.setObject(i + (fieldIndexes.size() * j)
+ 1, currentValue, sqlType);
+ }
} else {
ps.setObject(i + 1, currentValue, sqlType);
}
@@ -766,6 +771,11 @@ public class PutDatabaseRecord extends
AbstractSessionFactoryProcessor {
if (DELETE_TYPE.equalsIgnoreCase(statementType)) {
ps.setObject(i * 2 + 1, currentValue, sqlType);
ps.setObject(i * 2 + 2, currentValue, sqlType);
+ } else if
(UPSERT_TYPE.equalsIgnoreCase(statementType)) {
+ final int timesToAddObjects =
databaseAdapter.getTimesToAddColumnObjectsForUpsert();
+ for (int j = 0; j < timesToAddObjects; j++) {
+ ps.setObject(i + (fieldIndexes.size() * j)
+ 1, currentValue, sqlType);
+ }
} else {
ps.setObject(i + 1, currentValue, sqlType);
}
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/DatabaseAdapter.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/DatabaseAdapter.java
index 40de0b8..1fba3f6 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/DatabaseAdapter.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/DatabaseAdapter.java
@@ -68,6 +68,16 @@ public interface DatabaseAdapter {
}
/**
+ * Tells How many times the column values need to be inserted into the
prepared statement. Some DBs (such as MySQL) need the values specified twice in
the statement,
+ * some need only to specify them once.
+ *
+ * @return An integer corresponding to the number of times to insert
column values into the prepared statement for UPSERT, or -1 if upsert is not
supported.
+ */
+ default int getTimesToAddColumnObjectsForUpsert() {
+ return supportsUpsert() ? 1 : -1;
+ }
+
+ /**
* Returns an SQL UPSERT statement - i.e. UPDATE record or INSERT if id
doesn't exist.
* <br /><br />
* There is no standard way of doing this so not all adapters support it -
use together with {@link #supportsUpsert()}!
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/impl/MySQLDatabaseAdapter.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/impl/MySQLDatabaseAdapter.java
index fdbc205..cba7768 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/impl/MySQLDatabaseAdapter.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/impl/MySQLDatabaseAdapter.java
@@ -16,6 +16,14 @@
*/
package org.apache.nifi.processors.standard.db.impl;
+import com.google.common.base.Preconditions;
+import org.apache.nifi.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
+
/**
* A generic database adapter that generates MySQL compatible SQL.
*/
@@ -35,4 +43,50 @@ public class MySQLDatabaseAdapter extends
GenericDatabaseAdapter {
// Removes double quotes and back-ticks.
return identifier == null ? null : identifier.replaceAll("[\"`]", "");
}
+
+ @Override
+ public boolean supportsUpsert() {
+ return true;
+ }
+
+ /**
+ * Tells How many times the column values need to be inserted into the
prepared statement. Some DBs (such as MySQL) need the values specified twice in
the statement,
+ * some need only to specify them once.
+ *
+ * @return An integer corresponding to the number of times to insert
column values into the prepared statement for UPSERT, or -1 if upsert is not
supported.
+ */
+ @Override
+ public int getTimesToAddColumnObjectsForUpsert() {
+ return 2;
+ }
+
+ @Override
+ public String getUpsertStatement(String table, List<String> columnNames,
Collection<String> uniqueKeyColumnNames) {
+ Preconditions.checkArgument(!StringUtils.isEmpty(table), "Table name
cannot be null or blank");
+ Preconditions.checkArgument(columnNames != null &&
!columnNames.isEmpty(), "Column names cannot be null or empty");
+ Preconditions.checkArgument(uniqueKeyColumnNames != null &&
!uniqueKeyColumnNames.isEmpty(), "Key column names cannot be null or empty");
+
+ String columns = columnNames.stream()
+ .collect(Collectors.joining(", "));
+
+ String parameterizedInsertValues = columnNames.stream()
+ .map(__ -> "?")
+ .collect(Collectors.joining(", "));
+
+ List<String> updateValues = new ArrayList<>();
+ for (int i = 0; i < columnNames.size(); i++) {
+ updateValues.add(columnNames.get(i) + " = ?");
+ }
+ String parameterizedUpdateValues = String.join(", ", updateValues);
+
+ StringBuilder statementStringBuilder = new StringBuilder("INSERT INTO
")
+ .append(table)
+ .append("(").append(columns).append(")")
+ .append(" VALUES ")
+ .append("(").append(parameterizedInsertValues).append(")")
+ .append(" ON DUPLICATE KEY UPDATE ")
+ .append(parameterizedUpdateValues);
+
+ return statementStringBuilder.toString();
+ }
}
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/db/impl/TestMySQLDatabaseAdapter.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/db/impl/TestMySQLDatabaseAdapter.java
new file mode 100644
index 0000000..a5f6eb7
--- /dev/null
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/db/impl/TestMySQLDatabaseAdapter.java
@@ -0,0 +1,109 @@
+/*
+ * 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.nifi.processors.standard.db.impl;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class TestMySQLDatabaseAdapter {
+
+ private MySQLDatabaseAdapter testSubject;
+
+ @Before
+ public void setUp() throws Exception {
+ testSubject = new MySQLDatabaseAdapter();
+ }
+
+ @Test
+ public void testSupportsUpsert() throws Exception {
+ assertTrue(testSubject.getClass().getSimpleName() + " should support
upsert", testSubject.supportsUpsert());
+ }
+
+ @Test
+ public void testGetUpsertStatementWithNullTableName() throws Exception {
+ testGetUpsertStatement(null, Collections.singletonList("notEmpty"),
Collections.singletonList("notEmpty"), new IllegalArgumentException("Table name
cannot be null or blank"));
+ }
+
+ @Test
+ public void testGetUpsertStatementWithBlankTableName() throws Exception {
+ testGetUpsertStatement("", Collections.singletonList("notEmpty"),
Collections.singletonList("notEmpty"), new IllegalArgumentException("Table name
cannot be null or blank"));
+ }
+
+ @Test
+ public void testGetUpsertStatementWithNullColumnNames() throws Exception {
+ testGetUpsertStatement("notEmpty", null,
Collections.singletonList("notEmpty"), new IllegalArgumentException("Column
names cannot be null or empty"));
+ }
+
+ @Test
+ public void testGetUpsertStatementWithEmptyColumnNames() throws Exception {
+ testGetUpsertStatement("notEmpty", Collections.emptyList(),
Collections.singletonList("notEmpty"), new IllegalArgumentException("Column
names cannot be null or empty"));
+ }
+
+ @Test
+ public void testGetUpsertStatementWithNullKeyColumnNames() throws
Exception {
+ testGetUpsertStatement("notEmpty",
Collections.singletonList("notEmpty"), null, new IllegalArgumentException("Key
column names cannot be null or empty"));
+ }
+
+ @Test
+ public void testGetUpsertStatementWithEmptyKeyColumnNames() throws
Exception {
+ testGetUpsertStatement("notEmpty",
Collections.singletonList("notEmpty"), Collections.emptyList(), new
IllegalArgumentException("Key column names cannot be null or empty"));
+ }
+
+ @Test
+ public void testGetUpsertStatement() {
+ // GIVEN
+ String tableName = "table";
+ List<String> columnNames = Arrays.asList("column1", "column2",
"column3", "column4");
+ Collection<String> uniqueKeyColumnNames = Arrays.asList("column2",
"column4");
+
+ String expected = "INSERT INTO" +
+ " table(column1, column2, column3, column4) VALUES (?, ?, ?,
?)" +
+ " ON DUPLICATE KEY UPDATE" +
+ " column1 = ?, column2 = ?, column3 = ?, column4 = ?";
+
+ // WHEN
+ // THEN
+ testGetUpsertStatement(tableName, columnNames, uniqueKeyColumnNames,
expected);
+ }
+
+ private void testGetUpsertStatement(String tableName, List<String>
columnNames, Collection<String> uniqueKeyColumnNames, IllegalArgumentException
expected) {
+ try {
+ testGetUpsertStatement(tableName, columnNames,
uniqueKeyColumnNames, (String) null);
+ fail();
+ } catch (IllegalArgumentException e) {
+ assertEquals(expected.getMessage(), e.getMessage());
+ }
+ }
+
+ private void testGetUpsertStatement(String tableName, List<String>
columnNames, Collection<String> uniqueKeyColumnNames, String expected) {
+ // WHEN
+ String actual = testSubject.getUpsertStatement(tableName, columnNames,
uniqueKeyColumnNames);
+
+ // THEN
+ assertEquals(expected, actual);
+ }
+
+}
\ No newline at end of file
