🔥 : improve : support jdbc db options

- Apply review feedback
- Add null data test case
- Remove unused import

Author: och5351

flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/database/dialect/AbstractDialectConverter.java

@@ -173,7 +173,7 @@ protected JdbcDeserializationConverter createInternalConverter(LogicalType type)
                                 : TimestampData.fromTimestamp((Timestamp) val);
             case CHAR:
             case VARCHAR:
-                return val -> StringData.fromString(val.toString());
+                return val -> StringData.fromString(val == null ? null: val.toString());
             case BINARY:
             case VARBINARY:
                 return val -> val;

flink-connector-jdbc-postgres/src/main/java/org/apache/flink/connector/jdbc/postgres/database/dialect/PostgresDialectConverter.java

@@ -28,7 +28,6 @@
 import org.apache.flink.table.types.logical.utils.LogicalTypeUtils;
 
 import java.lang.reflect.Array;
-import java.util.UUID;
 /**
  * Runtime converter that responsible to convert between JDBC object and Flink internal object for
  * PostgreSQL.

flink-connector-jdbc-postgres/src/test/java/org/apache/flink/connector/jdbc/postgres/database/catalog/PostgresCatalogITCase.java

@@ -205,4 +205,16 @@ void testUuidTypes() {
                 .hasToString(
                         "[+I[1, 123e4567-e89b-12d3-a456-426614174000]]");
     }
+
+    @Test
+    void testNullUuidTypes() {
+        List<Row> results =
+                CollectionUtil.iteratorToList(
+                        tEnv.sqlQuery(String.format("select * from %s", TABLE_UUID_TYPE2))
+                                .execute()
+                                .collect());
+        assertThat(results)
+                .hasToString(
+                        "[+I[1, NULL]]");
+    }
 }

flink-connector-jdbc-postgres/src/test/java/org/apache/flink/connector/jdbc/postgres/database/catalog/PostgresCatalogTest.java

@@ -74,7 +74,8 @@ void testListTables() throws DatabaseNotExistException {
                                 "public.t1",
                                 "public.t4",
                                 "public.t5",
-                                "public.uuid_table"));
+                                "public.uuid_table",
+                                "public.uuid_table2"));
 
         actual = catalog.listTables(TEST_DB);
 
@@ -195,4 +196,12 @@ void testUuidDataTypes() throws TableNotExistException {
                         new ObjectPath(PostgresCatalog.DEFAULT_DATABASE, TABLE_UUID_TYPE));
         assertThat(table.getUnresolvedSchema()).isEqualTo(getUuidTable().schema);
     }
+
+    @Test
+    void testNullUuidDataTypes() throws TableNotExistException {
+        CatalogBaseTable table =
+                catalog.getTable(
+                        new ObjectPath(PostgresCatalog.DEFAULT_DATABASE, TABLE_UUID_TYPE2));
+        assertThat(table.getUnresolvedSchema()).isEqualTo(getNullUuidTable().schema);
+    }
 }

flink-connector-jdbc-postgres/src/test/java/org/apache/flink/connector/jdbc/postgres/database/catalog/PostgresCatalogTestBase.java

@@ -56,6 +56,7 @@ private static DatabaseMetadata getStaticMetadata() {
     protected static final String TABLE_ARRAY_TYPE = "array_table";
     protected static final String TABLE_SERIAL_TYPE = "serial_table";
     protected static final String TABLE_UUID_TYPE = "uuid_table";
+    protected static final String TABLE_UUID_TYPE2 = "uuid_table2";
 
     protected static String baseUrl;
     protected static PostgresCatalog catalog;
@@ -112,6 +113,9 @@ static void init() throws SQLException {
         createTable(
                 PostgresTablePath.fromFlinkTableName(TABLE_UUID_TYPE),
                 getUuidTable().pgSchemaSql);
+        createTable(
+                PostgresTablePath.fromFlinkTableName(TABLE_UUID_TYPE2),
+                getNullUuidTable().pgSchemaSql);
 
         executeSQL(
                 PostgresCatalog.DEFAULT_DATABASE,
@@ -134,6 +138,10 @@ static void init() throws SQLException {
                 PostgresCatalog.DEFAULT_DATABASE,
                 String.format(
                         "insert into %s values (%s);", TABLE_UUID_TYPE, getUuidTable().values));
+        executeSQL(
+                PostgresCatalog.DEFAULT_DATABASE,
+                String.format(
+                        "insert into %s values (%s);", TABLE_UUID_TYPE2, getNullUuidTable().values));
     }
 
     @AfterAll
@@ -174,6 +182,10 @@ static void afterAll() throws SQLException {
                 PostgresCatalog.DEFAULT_DATABASE,
                 String.format(
                         "DROP TABLE %s ", PostgresTablePath.fromFlinkTableName(TABLE_UUID_TYPE)));
+        executeSQL(
+                PostgresCatalog.DEFAULT_DATABASE,
+                String.format(
+                        "DROP TABLE %s ", PostgresTablePath.fromFlinkTableName(TABLE_UUID_TYPE2)));
     }
 
     public static void createTable(PostgresTablePath tablePath, String tableSchemaSql)
@@ -408,6 +420,19 @@ public static TestTable getSerialTable() {
 
     public static TestTable getUuidTable() {
         String uuid1 = "123e4567-e89b-12d3-a456-426614174000";
+        return new TestTable(
+                Schema.newBuilder()
+                        .column("id", DataTypes.INT())
+                        .column("uid_col", DataTypes.VARCHAR(36))
+                        .build(),
+                "id INT, "
+                        + "uid_col UUID",
+                        String.format("1, '%s'", uuid1)
+
+        );
+    }
+
+    public static TestTable getNullUuidTable(){
 
         return new TestTable(
                 Schema.newBuilder()
@@ -416,9 +441,8 @@ public static TestTable getUuidTable() {
                         .build(),
                 "id INT, "
                         + "uid_col UUID",
-                String.format(
-                        "1, '%s'",
-                        uuid1)
+                "1, NULL"
+
         );
     }
 }