Add smallserial and apply changes after refactoring pgjdbc/pgjdbc#1194

snuyanzin · Apr 5, 2022 · dd7729e · dd7729e
1 parent 450a142
commit dd7729e
Show file tree

Hide file tree

Showing 4 changed files with 10 additions and 3 deletions.
diff --git a/flink-connectors/flink-connector-jdbc/pom.xml b/flink-connectors/flink-connector-jdbc/pom.xml
@@ -36,7 +36,7 @@ under the License.
 	<packaging>jar</packaging>
 
 	<properties>
-		<postgres.version>42.2.10</postgres.version>
+		<postgres.version>42.3.3</postgres.version>
 		<oracle.version>19.3.0.0</oracle.version>
 	</properties>
 

diff --git a/...r-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/psql/PostgresTypeMapper.java b/...r-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/psql/PostgresTypeMapper.java
@@ -49,6 +49,8 @@ public class PostgresTypeMapper implements JdbcDialectTypeMapper {
     // float <=> float8
     // boolean <=> bool
     // decimal <=> numeric
+    private static final String PG_SMALLSERIAL = "smallserial";
+    private static final String PG_SMALLSERIAL_ARRAY = "_smallserial";
     private static final String PG_SERIAL = "serial";
     private static final String PG_BIGSERIAL = "bigserial";
     private static final String PG_BYTEA = "bytea";
@@ -102,7 +104,9 @@ public DataType mapping(ObjectPath tablePath, ResultSetMetaData metadata, int co
             case PG_BYTEA_ARRAY:
                 return DataTypes.ARRAY(DataTypes.BYTES());
             case PG_SMALLINT:
+            case PG_SMALLSERIAL:
                 return DataTypes.SMALLINT();
+            case PG_SMALLSERIAL_ARRAY:
             case PG_SMALLINT_ARRAY:
                 return DataTypes.ARRAY(DataTypes.SMALLINT());
             case PG_INTEGER:

diff --git a/...rc/main/java/org/apache/flink/connector/jdbc/internal/converter/PostgresRowConverter.java b/...rc/main/java/org/apache/flink/connector/jdbc/internal/converter/PostgresRowConverter.java
@@ -91,7 +91,10 @@ private JdbcDeserializationConverter createPostgresArrayConverter(ArrayType arra
                 final Object[] array = (Object[]) Array.newInstance(elementClass, in.length);
                 for (int i = 0; i < in.length; i++) {
                     array[i] =
-                            elementConverter.deserialize(((PGobject) in[i]).getValue().getBytes());
+                            elementConverter.deserialize(
+                                    in[i] instanceof byte[]
+                                            ? in[i]
+                                            : ((PGobject) in[i]).getValue().getBytes());
                 }
                 return new GenericArrayData(array);
             };

diff --git a/...tor-jdbc/src/test/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalogITCase.java b/...tor-jdbc/src/test/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalogITCase.java
@@ -154,7 +154,7 @@ public void testArrayTypes() {
         assertEquals(
                 "[+I["
                         + "[1, 2, 3], "
-                        + "[[92, 120, 51, 50], [92, 120, 51, 51], [92, 120, 51, 52]], "
+                        + "[[50], [51], [52]], "
                         + "[3, 4, 5], "
                         + "[4, 5, 6], "
                         + "[5.5, 6.6, 7.7], "