[FLINK-17622][connectors/jdbc] Remove useless switch for decimal in P…

…ostgresCatalog This closes apache#12090
derekm · May 20, 2020 · d7fc0d0 · d7fc0d0
1 parent df16e42
commit d7fc0d0
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 5 deletions.
diff --git a/...connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalog.java b/...connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalog.java
@@ -214,6 +214,14 @@ public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistExcep
  }
  }
 
+ // Postgres jdbc driver maps several alias to real type, we use real type rather than alias:
+ // smallint <=> int2
+ // integer <=> int4
+ // int <=> int4
+ // bigint <=> int8
+ // float <=> float8
+ // boolean <=> bool
+ // decimal <=> numeric
  public static final String PG_BYTEA = "bytea";
  public static final String PG_BYTEA_ARRAY = "_bytea";
  public static final String PG_SMALLINT = "int2";
@@ -224,8 +232,6 @@ public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistExcep
  public static final String PG_BIGINT_ARRAY = "_int8";
  public static final String PG_REAL = "float4";
  public static final String PG_REAL_ARRAY = "_float4";
- public static final String PG_DECIMAL = "decimal";
- public static final String PG_DECIMAL_ARRAY = "_decimal";
  public static final String PG_DOUBLE_PRECISION = "float8";
  public static final String PG_DOUBLE_PRECISION_ARRAY = "_float8";
  public static final String PG_NUMERIC = "numeric";
@@ -249,12 +255,19 @@ public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistExcep
  public static final String PG_CHARACTER_VARYING = "varchar";
  public static final String PG_CHARACTER_VARYING_ARRAY = "_varchar";
 
+ /**
+ * Converts Postgres type to Flink {@link DataType}.
+ *
+ * @see org.postgresql.jdbc.TypeInfoCache
+ */
  private DataType fromJDBCType(ResultSetMetaData metadata, int colIndex) throws SQLException {
  String pgType = metadata.getColumnTypeName(colIndex);
 
  int precision = metadata.getPrecision(colIndex);
  int scale = metadata.getScale(colIndex);
 
+ // pg types that gets replaced by jdbc driver:
+ // - decimal => numeric
  switch (pgType) {
  case PG_BOOLEAN:
  return DataTypes.BOOLEAN();
@@ -284,14 +297,12 @@ private DataType fromJDBCType(ResultSetMetaData metadata, int colIndex) throws S
  return DataTypes.DOUBLE();
  case PG_DOUBLE_PRECISION_ARRAY:
  return DataTypes.ARRAY(DataTypes.DOUBLE());
- case PG_DECIMAL:
  case PG_NUMERIC:
  // see SPARK-26538: handle numeric without explicit precision and scale.
  if (precision > 0) {
  return DataTypes.DECIMAL(precision, metadata.getScale(colIndex));
  }
  return DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 18);
- case PG_DECIMAL_ARRAY:
  case PG_NUMERIC_ARRAY:
  // see SPARK-26538: handle numeric without explicit precision and scale.
  if (precision > 0) {

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
@@ -88,7 +88,7 @@ public void testPrimitiveTypes() throws Exception {
  List<Row> results = Lists.newArrayList(
  tEnv.sqlQuery(String.format("select * from %s", TABLE_PRIMITIVE_TYPE)).execute().collect());
 
- assertEquals("[1,[50],3,4,5.5,6.6,7.70000,true,a,b,c ,d,2016-06-22T19:10:25,2015-01-01,00:51:03,500.000000000000000000]", results.toString());
+ assertEquals("[1,[50],3,4,5.5,6.6,7.70000,8.8,true,a,b,c ,d,2016-06-22T19:10:25,2015-01-01,00:51:03,500.000000000000000000]", results.toString());
  }
 
  @Test
@@ -107,6 +107,7 @@ public void testArrayTypes() throws Exception {
  "[6.6, 7.7, 8.8]," +
  "[7.70000, 8.80000, 9.90000]," +
  "[8.800000000000000000, 9.900000000000000000, 10.100000000000000000]," +
+ "[9.90, 10.10, 11.11]," +
  "[true, false, true]," +
  "[a, b, c]," +
  "[b, c, d]," +

diff --git a/...r-jdbc/src/test/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalogTestBase.java b/...r-jdbc/src/test/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalogTestBase.java
@@ -164,6 +164,7 @@ public static TestTable getPrimitiveTable() {
  .field("real", DataTypes.FLOAT())
  .field("double_precision", DataTypes.DOUBLE())
  .field("numeric", DataTypes.DECIMAL(10, 5))
+ .field("decimal", DataTypes.DECIMAL(10, 1))
  .field("boolean", DataTypes.BOOLEAN())
  .field("text", DataTypes.STRING())
  .field("char", DataTypes.CHAR(1))
@@ -182,6 +183,7 @@ public static TestTable getPrimitiveTable() {
  "real real, " +
  "double_precision double precision, " +
  "numeric numeric(10, 5), " +
+ "decimal decimal(10, 1), " +
  "boolean boolean, " +
  "text text, " +
  "char char, " +
@@ -199,6 +201,7 @@ public static TestTable getPrimitiveTable() {
  "5.5," +
  "6.6," +
  "7.7," +
+ "8.8," +
  "true," +
  "'a'," +
  "'b'," +
@@ -224,6 +227,7 @@ public static TestTable getArrayTable() {
  .field("double_precision_arr", DataTypes.ARRAY(DataTypes.DOUBLE()))
  .field("numeric_arr", DataTypes.ARRAY(DataTypes.DECIMAL(10, 5)))
  .field("numeric_arr_default", DataTypes.ARRAY(DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 18)))
+ .field("decimal_arr", DataTypes.ARRAY(DataTypes.DECIMAL(10, 2)))
  .field("boolean_arr", DataTypes.ARRAY(DataTypes.BOOLEAN()))
  .field("text_arr", DataTypes.ARRAY(DataTypes.STRING()))
  .field("char_arr", DataTypes.ARRAY(DataTypes.CHAR(1)))
@@ -243,6 +247,7 @@ public static TestTable getArrayTable() {
  "double_precision_arr double precision[], " +
  "numeric_arr numeric(10, 5)[], " +
  "numeric_arr_default numeric[], " +
+ "decimal_arr decimal(10,2)[], " +
  "boolean_arr boolean[], " +
  "text_arr text[], " +
  "char_arr char[], " +
@@ -261,6 +266,7 @@ public static TestTable getArrayTable() {
  "'{6.6,7.7,8.8}'," +
  "'{7.7,8.8,9.9}'," +
  "'{8.8,9.9,10.10}'," +
+ "'{9.9,10.10,11.11}'," +
  "'{true,false,true}'," +
  "'{a,b,c}'," +
  "'{b,c,d}'," +