You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2021/01/22 09:36:31 UTC
[flink] branch release-1.12 updated: [FLINK-19771][connector-jdbc]
Fix NPE when accessing array using postgres
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.12 by this push:
new d11137e [FLINK-19771][connector-jdbc] Fix NPE when accessing array using postgres
d11137e is described below
commit d11137ea1f68a0480ada713e65065c092b281e1e
Author: Dylan Forciea <df...@gmail.com>
AuthorDate: Sun Oct 25 19:01:36 2020 -0500
[FLINK-19771][connector-jdbc] Fix NPE when accessing array using postgres
Switching to handle arrays in createInternalConverter rather than
createNullableInternalConverter, so nulls are handled properly.
This closes #13787.
---
.../jdbc/internal/converter/PostgresRowConverter.java | 4 ++--
.../flink/connector/jdbc/catalog/PostgresCatalogITCase.java | 4 +++-
.../flink/connector/jdbc/catalog/PostgresCatalogTestBase.java | 10 ++++++++--
3 files changed, 13 insertions(+), 5 deletions(-)
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/converter/PostgresRowConverter.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/converter/PostgresRowConverter.java
index 2b92d9c..1b07567 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/converter/PostgresRowConverter.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/converter/PostgresRowConverter.java
@@ -50,7 +50,7 @@ public class PostgresRowConverter extends AbstractJdbcRowConverter {
}
@Override
- public JdbcDeserializationConverter createNullableInternalConverter(LogicalType type) {
+ public JdbcDeserializationConverter createInternalConverter(LogicalType type) {
LogicalTypeRoot root = type.getTypeRoot();
if (root == LogicalTypeRoot.ARRAY) {
@@ -116,6 +116,6 @@ public class PostgresRowConverter extends AbstractJdbcRowConverter {
// Have its own method so that Postgres can support primitives that super class doesn't support
// in the future
private JdbcDeserializationConverter createPrimitiveConverter(LogicalType type) {
- return super.createNullableInternalConverter(type);
+ return super.createInternalConverter(type);
}
}
diff --git a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalogITCase.java b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalogITCase.java
index b958cd1..310cfb5 100644
--- a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalogITCase.java
+++ b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalogITCase.java
@@ -173,7 +173,9 @@ public class PostgresCatalogITCase extends PostgresCatalogTestBase {
+ "[b, c, d],"
+ "[2016-06-22T19:10:25, 2019-06-22T19:10:25],"
+ "[2015-01-01, 2020-01-01],"
- + "[00:51:03, 00:59:03]]",
+ + "[00:51:03, 00:59:03],"
+ + "null,"
+ + "null]",
results.toString());
}
diff --git a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalogTestBase.java b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalogTestBase.java
index 6930ff2..7402a54 100644
--- a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalogTestBase.java
+++ b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalogTestBase.java
@@ -283,6 +283,8 @@ public class PostgresCatalogTestBase {
// DataTypes.ARRAY(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(4)))
.field("date_arr", DataTypes.ARRAY(DataTypes.DATE()))
.field("time_arr", DataTypes.ARRAY(DataTypes.TIME(0)))
+ .field("null_bytea_arr", DataTypes.ARRAY(DataTypes.BYTES()))
+ .field("null_text_arr", DataTypes.ARRAY(DataTypes.STRING()))
.build(),
"int_arr integer[], "
+ "bytea_arr bytea[], "
@@ -302,7 +304,9 @@ public class PostgresCatalogTestBase {
+
// "timestamptz_arr timestamptz(4)[], " +
"date_arr date[], "
- + "time_arr time(0)[]",
+ + "time_arr time(0)[], "
+ + "null_bytea_arr bytea[], "
+ + "null_text_arr text[]",
String.format(
"'{1,2,3}',"
+ "'{2,3,4}',"
@@ -322,7 +326,9 @@ public class PostgresCatalogTestBase {
+
// "'{\"2006-06-22 19:10:25\", \"2009-06-22 19:10:25\"}'," +
"'{\"2015-01-01\", \"2020-01-01\"}',"
- + "'{\"00:51:02.746572\", \"00:59:02.746572\"}'"));
+ + "'{\"00:51:02.746572\", \"00:59:02.746572\"}',"
+ + "NULL,"
+ + "NULL"));
}
public static TestTable getSerialTable() {