in flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/utils/PostgresTypeUtils.java [79:164]
private static DataType convertFromColumn(Column column) {
String typeName = column.typeName();
int precision = column.length();
int scale = column.scale().orElse(0);
switch (typeName) {
case PG_BOOLEAN:
return DataTypes.BOOLEAN();
case PG_BOOLEAN_ARRAY:
return DataTypes.ARRAY(DataTypes.BOOLEAN());
case PG_BYTEA:
return DataTypes.BYTES();
case PG_BYTEA_ARRAY:
return DataTypes.ARRAY(DataTypes.BYTES());
case PG_SMALLINT:
case PG_SMALLSERIAL:
return DataTypes.SMALLINT();
case PG_SMALLINT_ARRAY:
return DataTypes.ARRAY(DataTypes.SMALLINT());
case PG_INTEGER:
case PG_SERIAL:
return DataTypes.INT();
case PG_INTEGER_ARRAY:
return DataTypes.ARRAY(DataTypes.INT());
case PG_BIGINT:
case PG_BIGSERIAL:
return DataTypes.BIGINT();
case PG_BIGINT_ARRAY:
return DataTypes.ARRAY(DataTypes.BIGINT());
case PG_REAL:
return DataTypes.FLOAT();
case PG_REAL_ARRAY:
return DataTypes.ARRAY(DataTypes.FLOAT());
case PG_DOUBLE_PRECISION:
return DataTypes.DOUBLE();
case PG_DOUBLE_PRECISION_ARRAY:
return DataTypes.ARRAY(DataTypes.DOUBLE());
case PG_NUMERIC:
// see SPARK-26538: handle numeric without explicit precision and scale.
if (precision > 0) {
return DataTypes.DECIMAL(precision, scale);
}
return DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 18);
case PG_NUMERIC_ARRAY:
// see SPARK-26538: handle numeric without explicit precision and scale.
if (precision > 0) {
return DataTypes.ARRAY(DataTypes.DECIMAL(precision, scale));
}
return DataTypes.ARRAY(DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 18));
case PG_CHAR:
case PG_CHARACTER:
return DataTypes.CHAR(precision);
case PG_CHAR_ARRAY:
case PG_CHARACTER_ARRAY:
return DataTypes.ARRAY(DataTypes.CHAR(precision));
case PG_CHARACTER_VARYING:
return DataTypes.VARCHAR(precision);
case PG_CHARACTER_VARYING_ARRAY:
return DataTypes.ARRAY(DataTypes.VARCHAR(precision));
case PG_TEXT:
case PG_UUID:
return DataTypes.STRING();
case PG_TEXT_ARRAY:
return DataTypes.ARRAY(DataTypes.STRING());
case PG_TIMESTAMP:
return DataTypes.TIMESTAMP(scale);
case PG_TIMESTAMP_ARRAY:
return DataTypes.ARRAY(DataTypes.TIMESTAMP(scale));
case PG_TIMESTAMPTZ:
return DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(scale);
case PG_TIMESTAMPTZ_ARRAY:
return DataTypes.ARRAY(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(scale));
case PG_TIME:
return DataTypes.TIME(scale);
case PG_TIME_ARRAY:
return DataTypes.ARRAY(DataTypes.TIME(scale));
case PG_DATE:
return DataTypes.DATE();
case PG_DATE_ARRAY:
return DataTypes.ARRAY(DataTypes.DATE());
default:
throw new UnsupportedOperationException(
String.format("Doesn't support Postgres type '%s' yet", typeName));
}
}