[BUGFIX][Catalog] oracle catalog create table repeat and oracle pg null point

.github/workflows/backend.yml

@@ -869,6 +869,31 @@ jobs:
         env:
           MAVEN_OPTS: -Xmx4096m
 
+  jdbc-connectors-it-part-6:
+    needs: [ changes, sanity-check ]
+    if: needs.changes.outputs.api == 'true'
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        java: [ '8', '11' ]
+        os: [ 'ubuntu-latest' ]
+    timeout-minutes: 90
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up JDK ${{ matrix.java }}
+        uses: actions/setup-java@v3
+        with:
+          java-version: ${{ matrix.java }}
+          distribution: 'temurin'
+          cache: 'maven'
+      - name: run jdbc connectors integration test (part-6)
+        if: needs.changes.outputs.api == 'true'
+        run: |
+          ./mvnw -B -T 1C verify -DskipUT=true -DskipIT=false -D"license.skipAddThirdParty"=true --no-snapshot-updates -pl :connector-jdbc-e2e-part-6 -am -Pci
+        env:
+          MAVEN_OPTS: -Xmx4096m
+
+
   kafka-connector-it:
     needs: [ changes, sanity-check ]
     if: needs.changes.outputs.api == 'true'

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/AbstractJdbcCatalog.java

@@ -539,6 +539,7 @@ protected List<String> queryString(String url, String sql, ResultSetConsumer<Str
     // If sql is DDL, the execute() method always returns false, so the return value
     // should not be used to determine whether changes were made in database.
     protected boolean executeInternal(String url, String sql) throws SQLException {
+        LOG.info("create table sql is: {}", sql);
         try (PreparedStatement ps = getConnection(url).prepareStatement(sql)) {
             return ps.execute();
         }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oracle/OracleCatalog.java

@@ -129,12 +129,7 @@ protected String getCreateTableSql(TablePath tablePath, CatalogTable table) {
 
     @Override
     protected String getDropTableSql(TablePath tablePath) {
-        return String.format("DROP TABLE %s", getTableName(tablePath));
-    }
-
-    @Override
-    protected String getTableName(TablePath tablePath) {
-        return tablePath.getSchemaAndTableName().toUpperCase();
+        return String.format("DROP TABLE %s", tablePath.getSchemaAndTableName("\""));
     }
 
     @Override

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oracle/OracleCreateTableSqlBuilder.java

@@ -91,7 +91,7 @@ private String buildColumnSql(Column column) {
         columnSql.append("\"").append(column.getName()).append("\" ");
 
         String columnType =
-                sourceCatalogName.equals("oracle")
+                StringUtils.equalsIgnoreCase("oracle", sourceCatalogName)
                         ? column.getSourceType()
                         : buildColumnType(column);
         columnSql.append(columnType);

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/psql/PostgresCreateTableSqlBuilder.java

@@ -89,7 +89,7 @@ private String buildColumnSql(Column column) {
 
         // For simplicity, assume the column type in SeaTunnelDataType is the same as in PostgreSQL
         String columnType =
-                sourceCatalogName.equals("postgres")
+                StringUtils.equalsIgnoreCase("postgres", sourceCatalogName)
                         ? column.getSourceType()
                         : buildColumnType(column);
         columnSql.append(columnType);

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkFactory.java

@@ -102,6 +102,7 @@ public TableSink createSink(TableFactoryContext context) {
                                 catalogTable.getTableSchema(),
                                 catalogTable.getOptions(),
                                 catalogTable.getPartitionKeys(),
+                                catalogTable.getComment(),
                                 catalogTable.getCatalogName());
             }
             Map<String, String> map = config.toMap();

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-6/pom.xml

@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+    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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.seatunnel</groupId>
+        <artifactId>connector-jdbc-e2e</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>connector-jdbc-e2e-part-6</artifactId>
+    <name>SeaTunnel : E2E : Connector V2 : Jdbc : Part 2</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-jdbc-e2e-common</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>oracle-xe</artifactId>
+            <version>${testcontainer.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <!-- drivers -->
+        <dependency>
+            <groupId>com.oracle.database.jdbc</groupId>
+            <artifactId>ojdbc8</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-6/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcOracleLowercaseTableIT.java

@@ -0,0 +1,217 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.jdbc;
+
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.oracle.OracleCatalog;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.oracle.OracleURLParser;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.OracleContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.DockerLoggerFactory;
+import org.testcontainers.utility.MountableFile;
+
+import com.google.common.collect.Lists;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class JdbcOracleLowercaseTableIT extends AbstractJdbcIT {
+
+    private static final String ORACLE_IMAGE = "gvenzl/oracle-xe:21-slim-faststart";
+    private static final String ORACLE_NETWORK_ALIASES = "e2e_oracleDb";
+    private static final String DRIVER_CLASS = "oracle.jdbc.OracleDriver";
+    private static final int ORACLE_PORT = 1521;
+    private static final String ORACLE_URL = "jdbc:oracle:thin:@" + HOST + ":%s/%s";
+    private static final String USERNAME = "TESTUSER";
+    private static final String PASSWORD = "testPassword";
+    private static final String DATABASE = "XE";
+    private static final String SCHEMA = USERNAME;
+    private static final String SOURCE_TABLE = "E2E_TABLE_SOURCE_LOWER";
+    private static final String SINK_TABLE = "E2E_TABLE_SINK_LOWER";
+    private static final String CATALOG_TABLE = "e2e_table_catalog_lower";
+    // no execute conf just test lower oracle create table
+    private static final List<String> CONFIG_FILE = Lists.newArrayList();
+
+    private static final String CREATE_SQL =
+            "create table %s\n"
+                    + "(\n"
+                    + "    VARCHAR_10_COL                varchar2(10),\n"
+                    + "    CHAR_10_COL                   char(10),\n"
+                    + "    CLOB_COL                      clob,\n"
+                    + "    NUMBER_3_SF_2_DP              number(3, 2),\n"
+                    + "    INTEGER_COL                   integer,\n"
+                    + "    FLOAT_COL                     float(10),\n"
+                    + "    REAL_COL                      real,\n"
+                    + "    BINARY_FLOAT_COL              binary_float,\n"
+                    + "    BINARY_DOUBLE_COL             binary_double,\n"
+                    + "    DATE_COL                      date,\n"
+                    + "    TIMESTAMP_WITH_3_FRAC_SEC_COL timestamp(3),\n"
+                    + "    TIMESTAMP_WITH_LOCAL_TZ       timestamp with local time zone\n"
+                    + ")";
+
+    @Override
+    JdbcCase getJdbcCase() {
+        Map<String, String> containerEnv = new HashMap<>();
+        containerEnv.put("ORACLE_PASSWORD", PASSWORD);
+        containerEnv.put("APP_USER", USERNAME);
+        containerEnv.put("APP_USER_PASSWORD", PASSWORD);
+        String jdbcUrl = String.format(ORACLE_URL, ORACLE_PORT, SCHEMA);
+        Pair<String[], List<SeaTunnelRow>> testDataSet = initTestData();
+        String[] fieldNames = testDataSet.getKey();
+
+        String insertSql = insertTable(SCHEMA, SOURCE_TABLE, fieldNames);
+
+        return JdbcCase.builder()
+                .dockerImage(ORACLE_IMAGE)
+                .networkAliases(ORACLE_NETWORK_ALIASES)
+                .containerEnv(containerEnv)
+                .driverClass(DRIVER_CLASS)
+                .host(HOST)
+                .port(ORACLE_PORT)
+                .localPort(ORACLE_PORT)
+                .jdbcTemplate(ORACLE_URL)
+                .jdbcUrl(jdbcUrl)
+                .userName(USERNAME)
+                .password(PASSWORD)
+                .database(DATABASE)
+                .schema(SCHEMA)
+                .sourceTable(SOURCE_TABLE)
+                .sinkTable(SINK_TABLE)
+                .catalogDatabase(DATABASE)
+                .catalogSchema(SCHEMA)
+                .catalogTable(CATALOG_TABLE)
+                .createSql(CREATE_SQL)
+                .configFile(CONFIG_FILE)
+                .insertSql(insertSql)
+                .testData(testDataSet)
+                .build();
+    }
+
+    @Override
+    void compareResult() {}
+
+    @Override
+    String driverUrl() {
+        return "https://repo1.maven.org/maven2/com/oracle/database/jdbc/ojdbc8/12.2.0.1/ojdbc8-12.2.0.1.jar";
+    }
+
+    @Override
+    Pair<String[], List<SeaTunnelRow>> initTestData() {
+        String[] fieldNames =
+                new String[] {
+                    "VARCHAR_10_COL",
+                    "CHAR_10_COL",
+                    "CLOB_COL",
+                    "NUMBER_3_SF_2_DP",
+                    "INTEGER_COL",
+                    "FLOAT_COL",
+                    "REAL_COL",
+                    "BINARY_FLOAT_COL",
+                    "BINARY_DOUBLE_COL",
+                    "DATE_COL",
+                    "TIMESTAMP_WITH_3_FRAC_SEC_COL",
+                    "TIMESTAMP_WITH_LOCAL_TZ"
+                };
+
+        List<SeaTunnelRow> rows = new ArrayList<>();
+        for (int i = 0; i < 100; i++) {
+            SeaTunnelRow row =
+                    new SeaTunnelRow(
+                            new Object[] {
+                                String.format("f%s", i),
+                                String.format("f%s", i),
+                                String.format("f%s", i),
+                                BigDecimal.valueOf(1.1),
+                                i,
+                                Float.parseFloat("2.2"),
+                                Float.parseFloat("2.2"),
+                                Float.parseFloat("22.2"),
+                                Double.parseDouble("2.2"),
+                                Date.valueOf(LocalDate.now()),
+                                Timestamp.valueOf(LocalDateTime.now()),
+                                Timestamp.valueOf(LocalDateTime.now())
+                            });
+            rows.add(row);
+        }
+
+        return Pair.of(fieldNames, rows);
+    }
+
+    @Override
+    GenericContainer<?> initContainer() {
+        DockerImageName imageName = DockerImageName.parse(ORACLE_IMAGE);
+
+        GenericContainer<?> container =
+                new OracleContainer(imageName)
+                        .withDatabaseName(SCHEMA)
+                        .withCopyFileToContainer(
+                                MountableFile.forClasspathResource("sql/oracle_init.sql"),
+                                "/container-entrypoint-startdb.d/init.sql")
+                        .withNetwork(NETWORK)
+                        .withNetworkAliases(ORACLE_NETWORK_ALIASES)
+                        .withExposedPorts(ORACLE_PORT)
+                        .withLogConsumer(
+                                new Slf4jLogConsumer(DockerLoggerFactory.getLogger(ORACLE_IMAGE)));
+
+        container.setPortBindings(
+                Lists.newArrayList(String.format("%s:%s", ORACLE_PORT, ORACLE_PORT)));
+
+        return container;
+    }
+
+    @Override
+    public String quoteIdentifier(String field) {
+        return "\"" + field + "\"";
+    }
+
+    @Override
+    protected void clearTable(String database, String schema, String table) {
+        clearTable(schema, table);
+    }
+
+    @Override
+    protected String buildTableInfoWithSchema(String database, String schema, String table) {
+        return buildTableInfoWithSchema(schema, table);
+    }
+
+    @Override
+    protected void initCatalog() {
+        String jdbcUrl = jdbcCase.getJdbcUrl().replace(HOST, dbServer.getHost());
+        catalog =
+                new OracleCatalog(
+                        "oracle",
+                        jdbcCase.getUserName(),
+                        jdbcCase.getPassword(),
+                        OracleURLParser.parse(jdbcUrl),
+                        SCHEMA);
+        catalog.open();
+    }
+}

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-6/src/test/resources/sql/oracle_init.sql

@@ -0,0 +1,22 @@
+--
+-- 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.
+--
+
+ALTER SESSION SET CONTAINER = TESTUSER;
+
+CREATE USER TESTUSER IDENTIFIED BY testPassword;
+
+GRANT DBA TO TESTUSER;

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/pom.xml

@@ -33,6 +33,7 @@
         <module>connector-jdbc-e2e-part-3</module>
         <module>connector-jdbc-e2e-part-4</module>
         <module>connector-jdbc-e2e-part-5</module>
+        <module>connector-jdbc-e2e-part-6</module>
     </modules>
 
     <dependencyManagement>