[hotfix][Connector/JDBC] Support PostgreSQL uuid type

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 == null ? null: 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,6 +28,7 @@
 import org.apache.flink.table.types.logical.utils.LogicalTypeUtils;
 
 import java.lang.reflect.Array;
+
 /**
  * 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

@@ -201,9 +201,7 @@ void testUuidTypes() {
                         tEnv.sqlQuery(String.format("select * from %s", TABLE_UUID_TYPE))
                                 .execute()
                                 .collect());
-        assertThat(results)
-                .hasToString(
-                        "[+I[1, 123e4567-e89b-12d3-a456-426614174000]]");
+        assertThat(results).hasToString("[+I[1, 123e4567-e89b-12d3-a456-426614174000]]");
     }
 
     @Test
@@ -213,8 +211,6 @@ void testNullUuidTypes() {
                         tEnv.sqlQuery(String.format("select * from %s", TABLE_UUID_TYPE2))
                                 .execute()
                                 .collect());
-        assertThat(results)
-                .hasToString(
-                        "[+I[1, NULL]]");
+        assertThat(results).hasToString("[+I[1, NULL]]");
     }
 }

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

@@ -192,8 +192,7 @@ public void testSerialDataTypes() throws TableNotExistException {
     @Test
     void testUuidDataTypes() throws TableNotExistException {
         CatalogBaseTable table =
-                catalog.getTable(
-                        new ObjectPath(PostgresCatalog.DEFAULT_DATABASE, TABLE_UUID_TYPE));
+                catalog.getTable(new ObjectPath(PostgresCatalog.DEFAULT_DATABASE, TABLE_UUID_TYPE));
         assertThat(table.getUnresolvedSchema()).isEqualTo(getUuidTable().schema);
     }
 

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

@@ -111,8 +111,7 @@ static void init() throws SQLException {
                 PostgresTablePath.fromFlinkTableName(TABLE_SERIAL_TYPE),
                 getSerialTable().pgSchemaSql);
         createTable(
-                PostgresTablePath.fromFlinkTableName(TABLE_UUID_TYPE),
-                getUuidTable().pgSchemaSql);
+                PostgresTablePath.fromFlinkTableName(TABLE_UUID_TYPE), getUuidTable().pgSchemaSql);
         createTable(
                 PostgresTablePath.fromFlinkTableName(TABLE_UUID_TYPE2),
                 getNullUuidTable().pgSchemaSql);
@@ -141,7 +140,8 @@ static void init() throws SQLException {
         executeSQL(
                 PostgresCatalog.DEFAULT_DATABASE,
                 String.format(
-                        "insert into %s values (%s);", TABLE_UUID_TYPE2, getNullUuidTable().values));
+                        "insert into %s values (%s);",
+                        TABLE_UUID_TYPE2, getNullUuidTable().values));
     }
 
     @AfterAll
@@ -425,24 +425,18 @@ public static TestTable getUuidTable() {
                         .column("id", DataTypes.INT())
                         .column("uid_col", DataTypes.VARCHAR(36))
                         .build(),
-                "id INT, "
-                        + "uid_col UUID",
-                        String.format("1, '%s'", uuid1)
-
-        );
+                "id INT, " + "uid_col UUID",
+                String.format("1, '%s'", uuid1));
     }
 
-    public static TestTable getNullUuidTable(){
+    public static TestTable getNullUuidTable() {
 
         return new TestTable(
                 Schema.newBuilder()
                         .column("id", DataTypes.INT())
                         .column("uid_col", DataTypes.VARCHAR(36))
                         .build(),
-                "id INT, "
-                        + "uid_col UUID",
-                "1, NULL"
-
-        );
+                "id INT, " + "uid_col UUID",
+                "1, NULL");
     }
 }

flink-connector-jdbc-postgres/src/test/java/org/apache/flink/connector/jdbc/postgres/table/PostgresDynamicTableSourceITCase.java

@@ -46,7 +46,6 @@ protected TableRow createInputTable() {
                 field("id", DataTypes.BIGINT().notNull()),
                 // uuid test field
                 field("uid_col", dbType("uuid"), DataTypes.STRING().notNull()),
-
                 field("decimal_col", DataTypes.DECIMAL(10, 4)),
                 field("timestamp6_col", DataTypes.TIMESTAMP(6)),
                 // other fields

flink-connector-jdbc-postgres/src/test/java/org/apache/flink/connector/jdbc/postgres/testutils/PostgresMetadata.java

@@ -43,7 +43,10 @@ public PostgresMetadata(JdbcDatabaseContainer<?> container, boolean hasXaEnabled
         this.username = container.getUsername();
         this.password = container.getPassword();
         String baseUrl = container.getJdbcUrl();
-        this.url = baseUrl.contains("?") ? baseUrl + "&stringtype=unspecified" : baseUrl + "?stringtype=unspecified";
+        this.url =
+                baseUrl.contains("?")
+                        ? baseUrl + "&stringtype=unspecified"
+                        : baseUrl + "?stringtype=unspecified";
         this.driver = container.getDriverClassName();
         this.version = container.getDockerImageName();
         this.xaEnabled = hasXaEnabled;