This is an automated email from the ASF dual-hosted git repository. mattyb149 pushed a commit to branch support/nifi-1.x in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push: new 4fa47ecc2a NIFI-5151: Add UPSERT support for Apache Phoenix 4fa47ecc2a is described below commit 4fa47ecc2ad02fe266bcfee38b954d00e7aeddf8 Author: Lehel Boér <lehe...@hotmail.com> AuthorDate: Thu May 18 16:52:51 2023 +0200 NIFI-5151: Add UPSERT support for Apache Phoenix Signed-off-by: Matthew Burgess <mattyb...@apache.org> --- .../standard/db/impl/PhoenixDatabaseAdapter.java | 33 ++++++ ...che.nifi.processors.standard.db.DatabaseAdapter | 3 +- .../db/impl/TestPhoenixDatabaseAdapter.java | 118 +++++++++++++++++++++ 3 files changed, 153 insertions(+), 1 deletion(-) diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/impl/PhoenixDatabaseAdapter.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/impl/PhoenixDatabaseAdapter.java index c19efe51ca..86d51e7f38 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/impl/PhoenixDatabaseAdapter.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/impl/PhoenixDatabaseAdapter.java @@ -23,8 +23,10 @@ import org.apache.nifi.processors.standard.db.DatabaseAdapter; import java.sql.JDBCType; import java.sql.Types; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.List; +import java.util.stream.Collectors; import static java.sql.Types.CHAR; import static java.sql.Types.CLOB; @@ -106,6 +108,37 @@ public final class PhoenixDatabaseAdapter implements DatabaseAdapter { return query.toString(); } + @Override + public String getUpsertStatement(String table, List<String> columnNames, Collection<String> uniqueKeyColumnNames) { + if (org.apache.nifi.util.StringUtils.isEmpty(table)) { + throw new IllegalArgumentException("Table name cannot be null or blank"); + } + if (columnNames == null || columnNames.isEmpty()) { + throw new IllegalArgumentException("Column names cannot be null or empty"); + } + if (uniqueKeyColumnNames == null || uniqueKeyColumnNames.isEmpty()) { + throw new IllegalArgumentException("Key column names cannot be null or empty"); + } + + String columns = String.join(", ", columnNames); + + String parameterizedUpsertValues = columnNames.stream() + .map(columnName -> "?") + .collect(Collectors.joining(", ")); + + StringBuilder statementStringBuilder = new StringBuilder("UPSERT INTO ") + .append(table) + .append("(").append(columns).append(")") + .append(" VALUES ") + .append("(").append(parameterizedUpsertValues).append(")"); + return statementStringBuilder.toString(); + } + + @Override + public boolean supportsUpsert() { + return true; + } + @Override public boolean supportsCreateTableIfNotExists() { return true; diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processors.standard.db.DatabaseAdapter b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processors.standard.db.DatabaseAdapter index f104782c5b..641223d21b 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processors.standard.db.DatabaseAdapter +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processors.standard.db.DatabaseAdapter @@ -18,4 +18,5 @@ org.apache.nifi.processors.standard.db.impl.Oracle12DatabaseAdapter org.apache.nifi.processors.standard.db.impl.MSSQLDatabaseAdapter org.apache.nifi.processors.standard.db.impl.MSSQL2008DatabaseAdapter org.apache.nifi.processors.standard.db.impl.MySQLDatabaseAdapter -org.apache.nifi.processors.standard.db.impl.PostgreSQLDatabaseAdapter \ No newline at end of file +org.apache.nifi.processors.standard.db.impl.PostgreSQLDatabaseAdapter +org.apache.nifi.processors.standard.db.impl.PhoenixDatabaseAdapter \ No newline at end of file diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/db/impl/TestPhoenixDatabaseAdapter.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/db/impl/TestPhoenixDatabaseAdapter.java new file mode 100644 index 0000000000..7234fe57e9 --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/db/impl/TestPhoenixDatabaseAdapter.java @@ -0,0 +1,118 @@ +/* + * 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.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class TestPhoenixDatabaseAdapter { + private PhoenixDatabaseAdapter testSubject; + + @BeforeEach + public void setUp() throws Exception { + testSubject = new PhoenixDatabaseAdapter(); + } + + @Test + void testSupportsUpsert() { + assertTrue(testSubject.supportsUpsert(), testSubject.getClass().getSimpleName() + " should support upsert"); + } + + @Test + void testGetUpsertStatementWithNullTableName() { + testGetUpsertStatement(null, Arrays.asList("notEmpty"), Arrays.asList("notEmpty"), new IllegalArgumentException("Table name cannot be null or blank")); + } + + @Test + void testGetUpsertStatementWithBlankTableName() { + testGetUpsertStatement("", Arrays.asList("notEmpty"), Arrays.asList("notEmpty"), new IllegalArgumentException("Table name cannot be null or blank")); + } + + @Test + void testGetUpsertStatementWithNullColumnNames() { + testGetUpsertStatement("notEmpty", null, Arrays.asList("notEmpty"), new IllegalArgumentException("Column names cannot be null or empty")); + } + + @Test + void testGetUpsertStatementWithEmptyColumnNames() { + testGetUpsertStatement("notEmpty", Collections.emptyList(), Arrays.asList("notEmpty"), new IllegalArgumentException("Column names cannot be null or empty")); + } + + @Test + void testGetUpsertStatementWithNullKeyColumnNames() { + testGetUpsertStatement("notEmpty", Arrays.asList("notEmpty"), null, new IllegalArgumentException("Key column names cannot be null or empty")); + } + + @Test + void testGetUpsertStatementWithEmptyKeyColumnNames() { + testGetUpsertStatement("notEmpty", Arrays.asList("notEmpty"), Collections.emptyList(), new IllegalArgumentException("Key column names cannot be null or empty")); + } + + @Test + void testGetUpsertStatement() { + // GIVEN + String tableName = "table"; + List<String> columnNames = Arrays.asList("column1", "column2", "column3", "column4"); + Collection<String> uniqueKeyColumnNames = Arrays.asList("column2", "column4"); + + String expected = "UPSERT INTO" + + " table(column1, column2, column3, column4) VALUES (?, ?, ?, ?)"; + + // WHEN + // THEN + testGetUpsertStatement(tableName, columnNames, uniqueKeyColumnNames, expected); + } + + private void testGetUpsertStatement(String tableName, List<String> columnNames, Collection<String> uniqueKeyColumnNames, IllegalArgumentException expected) { + final IllegalArgumentException e = assertThrows(IllegalArgumentException.class, () -> { + testGetUpsertStatement(tableName, columnNames, uniqueKeyColumnNames, (String) null); + }); + 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); + } + + @Test + public void testGetUpsertStatementQuoted() { + // GIVEN + String tableName = "\"table\""; + List<String> columnNames = Arrays.asList("column1", "\"column2\"", "column3", "column4"); + Collection<String> uniqueKeyColumnNames = Arrays.asList("\"column2\"", "column4"); + + String expected = "UPSERT INTO" + + " \"table\"(column1, \"column2\", column3, column4) VALUES (?, ?, ?, ?)"; + + // WHEN + // THEN + testGetUpsertStatement(tableName, columnNames, uniqueKeyColumnNames, expected); + } +}