NIFI-12530: Support CREATE TABLE in Oracle database adapters

mattyb149 · Dec 21, 2023 · 4b9fbfb · 4b9fbfb
1 parent c41b273
commit 4b9fbfb
Show file tree

Hide file tree

Showing 3 changed files with 104 additions and 0 deletions.
diff --git a/...rs/src/main/java/org/apache/nifi/processors/standard/db/impl/Oracle12DatabaseAdapter.java b/...rs/src/main/java/org/apache/nifi/processors/standard/db/impl/Oracle12DatabaseAdapter.java
@@ -22,11 +22,13 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.nifi.processors.standard.db.ColumnDescription;
 import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+import org.apache.nifi.processors.standard.db.TableSchema;
 
 import static java.sql.Types.CHAR;
 import static java.sql.Types.CLOB;
@@ -253,4 +255,43 @@ public String getSQLForDataType(int sqlType) {
         }
     }
 
+    @Override
+    public boolean supportsCreateTableIfNotExists() {
+        return true;
+    }
+
+    /**
+     * Generates a CREATE TABLE statement using the specified table schema
+     * @param tableSchema The table schema including column information
+     * @param quoteTableName Whether to quote the table name in the generated DDL
+     * @param quoteColumnNames Whether to quote column names in the generated DDL
+     * @return A String containing DDL to create the specified table
+     */
+    @Override
+    public String getCreateTableStatement(TableSchema tableSchema, boolean quoteTableName, boolean quoteColumnNames) {
+        StringBuilder createTableStatement = new StringBuilder();
+
+        List<ColumnDescription> columns = tableSchema.getColumnsAsList();
+        List<String> columnsAndDatatypes = new ArrayList<>(columns.size());
+        Set<String> primaryKeyColumnNames = tableSchema.getPrimaryKeyColumnNames();
+        for (ColumnDescription column : columns) {
+            String sb = (quoteColumnNames ? getColumnQuoteString() : "")
+                    + column.getColumnName()
+                    + (quoteColumnNames ? getColumnQuoteString() : "")
+                    + " " + getSQLForDataType(column.getDataType())
+                    + (column.isNullable() ? "" : " NOT NULL")
+                    + (primaryKeyColumnNames != null && primaryKeyColumnNames.contains(column.getColumnName()) ? " PRIMARY KEY" : "");
+            columnsAndDatatypes.add(sb);
+        }
+
+        createTableStatement
+                .append("DECLARE\n\tsql_stmt long;\nBEGIN\n\tsql_stmt:='CREATE TABLE ")
+                .append(generateTableName(quoteTableName, tableSchema.getCatalogName(), tableSchema.getSchemaName(), tableSchema.getTableName(), tableSchema))
+                .append(" (")
+                .append(String.join(", ", columnsAndDatatypes))
+                .append(")';\nEXECUTE IMMEDIATE sql_stmt;\nEXCEPTION\n\tWHEN OTHERS THEN\n\t\tIF SQLCODE = -955 THEN\n\t\t\t")
+                .append("NULL;\n\t\tELSE\n\t\t\tRAISE;\n\t\tEND IF;\nEND;");
+
+        return createTableStatement.toString();
+    }
 }
diff --git a/...sors/src/main/java/org/apache/nifi/processors/standard/db/impl/OracleDatabaseAdapter.java b/...sors/src/main/java/org/apache/nifi/processors/standard/db/impl/OracleDatabaseAdapter.java
@@ -19,12 +19,14 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.nifi.processors.standard.db.ColumnDescription;
 import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+import org.apache.nifi.processors.standard.db.TableSchema;
 
 import java.sql.JDBCType;
 import java.sql.Types;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Set;
 
 import static java.sql.Types.CHAR;
 import static java.sql.Types.CLOB;
@@ -174,4 +176,44 @@ public String getSQLForDataType(int sqlType) {
                 return JDBCType.valueOf(sqlType).getName();
         }
     }
+
+    @Override
+    public boolean supportsCreateTableIfNotExists() {
+        return true;
+    }
+
+    /**
+     * Generates a CREATE TABLE statement using the specified table schema
+     * @param tableSchema The table schema including column information
+     * @param quoteTableName Whether to quote the table name in the generated DDL
+     * @param quoteColumnNames Whether to quote column names in the generated DDL
+     * @return A String containing DDL to create the specified table
+     */
+    @Override
+    public String getCreateTableStatement(TableSchema tableSchema, boolean quoteTableName, boolean quoteColumnNames) {
+        StringBuilder createTableStatement = new StringBuilder();
+
+        List<ColumnDescription> columns = tableSchema.getColumnsAsList();
+        List<String> columnsAndDatatypes = new ArrayList<>(columns.size());
+        Set<String> primaryKeyColumnNames = tableSchema.getPrimaryKeyColumnNames();
+        for (ColumnDescription column : columns) {
+            String sb = (quoteColumnNames ? getColumnQuoteString() : "")
+                    + column.getColumnName()
+                    + (quoteColumnNames ? getColumnQuoteString() : "")
+                    + " " + getSQLForDataType(column.getDataType())
+                    + (column.isNullable() ? "" : " NOT NULL")
+                    + (primaryKeyColumnNames != null && primaryKeyColumnNames.contains(column.getColumnName()) ? " PRIMARY KEY" : "");
+            columnsAndDatatypes.add(sb);
+        }
+
+        createTableStatement
+                .append("DECLARE\n\tsql_stmt long;\nBEGIN\n\tsql_stmt:='CREATE TABLE ")
+                .append(generateTableName(quoteTableName, tableSchema.getCatalogName(), tableSchema.getSchemaName(), tableSchema.getTableName(), tableSchema))
+                .append(" (")
+                .append(String.join(", ", columnsAndDatatypes))
+                .append(")';\nEXECUTE IMMEDIATE sql_stmt;\nEXCEPTION\n\tWHEN OTHERS THEN\n\t\tIF SQLCODE = -955 THEN\n\t\t\t")
+                .append("NULL;\n\t\tELSE\n\t\t\tRAISE;\n\t\tEND IF;\nEND;");
+
+        return createTableStatement.toString();
+    }
 }
diff --git a/...rc/test/java/org/apache/nifi/processors/standard/db/impl/TestOracle12DatabaseAdapter.java b/...rc/test/java/org/apache/nifi/processors/standard/db/impl/TestOracle12DatabaseAdapter.java
@@ -16,12 +16,15 @@
  */
 package org.apache.nifi.processors.standard.db.impl;
 
+import java.sql.Types;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 
+import org.apache.nifi.processors.standard.db.ColumnDescription;
 import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+import org.apache.nifi.processors.standard.db.TableSchema;
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -150,6 +153,24 @@ public void testGetUpsertStatement() {
         testGetUpsertStatement(tableName, columnNames, uniqueKeyColumnNames, expected);
     }
 
+    @Test
+    public void testGetCreateTableStatement() {
+        assertTrue(db.supportsCreateTableIfNotExists());
+        final List<ColumnDescription> columns = Arrays.asList(
+                new ColumnDescription("col1", Types.INTEGER, true, 4, false),
+                new ColumnDescription("col2", Types.VARCHAR, false, 2000, true)
+        );
+        TableSchema tableSchema = new TableSchema("USERS", null, "TEST_TABLE", columns, true, Collections.singleton("COL1"), db.getColumnQuoteString());
+
+        String expectedStatement = "DECLARE\n\tsql_stmt long;\nBEGIN\n\tsql_stmt:='CREATE TABLE "
+                // Strings are returned as VARCHAR2(2000) regardless of reported size and that VARCHAR2 is not in java.sql.Types
+                + "\"USERS\".\"TEST_TABLE\" (\"col1\" INTEGER NOT NULL, \"col2\" VARCHAR2(2000))';"
+                + "\nEXECUTE IMMEDIATE sql_stmt;\nEXCEPTION\n\tWHEN OTHERS THEN\n\t\tIF SQLCODE = -955 THEN\n\t\t\t"
+                + "NULL;\n\t\tELSE\n\t\t\tRAISE;\n\t\tEND IF;\nEND;";
+        String actualStatement = db.getCreateTableStatement(tableSchema, true, true);
+        assertEquals(expectedStatement, actualStatement);
+    }
+
     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);