Skip to content

Commit

Permalink
[Feature][JDBC source] pg support char types (#8420)
Browse files Browse the repository at this point in the history
  • Loading branch information
@xyq2834646405
xyq2834646405 authored Jan 8, 2025
1 parent b8b8ca1 commit 776ac94
Show file tree
Hide file tree
Showing 9 changed files with 35 additions and 19 deletions.
4 changes: 3 additions & 1 deletion ...ache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/psql/PostgresTypeConverter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,8 @@ public class PostgresTypeConverter implements TypeConverter<BasicTypeDefine> {
public static final String PG_MONEY = "money";

// char <=> character <=> bpchar
public static final String PG_CHAR = "bpchar";
public static final String PG_CHAR = "char";
public static final String PG_BPCHAR = "bpchar";
public static final String PG_CHARACTER = "character";
// char[] <=> _character <=> _bpchar
public static final String PG_CHAR_ARRAY = "_bpchar";
Expand Down Expand Up @@ -189,6 +190,7 @@ public Column convert(BasicTypeDefine typeDefine) {
builder.scale(2);
break;
case PG_CHAR:
case PG_BPCHAR:
case PG_CHARACTER:
builder.dataType(BasicType.STRING_TYPE);
if (typeDefine.getLength() == null || typeDefine.getLength() <= 0) {
Expand Down
18 changes: 12 additions & 6 deletions ...rc/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcPostgresIdentifierIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,8 @@ public class JdbcPostgresIdentifierIT extends TestSuiteBase implements TestResou
+ " multipolygon geometry(MULTIPOLYGON, 4326),\n"
+ " geometrycollection geometry(GEOMETRYCOLLECTION, 4326),\n"
+ " geog geography(POINT, 4326),\n"
+ " inet_col INET\n"
+ " inet_col INET,\n"
+ " char_one_col CHAR(1)\n"
+ ")";
private static final String PG_SINK_DDL =
"CREATE TABLE IF NOT EXISTS test.public.\"PG_IDE_SINK_TABLE\" (\n"
Expand Down Expand Up @@ -125,7 +126,8 @@ public class JdbcPostgresIdentifierIT extends TestSuiteBase implements TestResou
+ " \"MULTIPOLYGON\" varchar(2000) NULL,\n"
+ " \"GEOMETRYCOLLECTION\" varchar(2000) NULL,\n"
+ " \"GEOG\" varchar(2000) NULL,\n"
+ " \"INET_COL\" INET NULL\n"
+ " \"INET_COL\" INET NULL,\n"
+ " \"CHAR_ONE_COL\" CHAR(1) NULL\n"
+ " )";

private static final String SOURCE_SQL =
Expand Down Expand Up @@ -158,7 +160,8 @@ public class JdbcPostgresIdentifierIT extends TestSuiteBase implements TestResou
+ "multipolygon,\n"
+ "geometrycollection,\n"
+ "geog,\n"
+ "inet_col\n"
+ "inet_col,\n"
+ "char_one_col\n"
+ " from pg_ide_source_table";
private static final String SINK_SQL =
"SELECT\n"
Expand Down Expand Up @@ -190,7 +193,8 @@ public class JdbcPostgresIdentifierIT extends TestSuiteBase implements TestResou
+ " CAST(\"MULTIPOLYGON\" AS GEOMETRY) AS MULTILINESTRING,\n"
+ " CAST(\"GEOMETRYCOLLECTION\" AS GEOMETRY) AS GEOMETRYCOLLECTION,\n"
+ " CAST(\"GEOG\" AS GEOGRAPHY) AS GEOG,\n"
+ " \"INET_COL\"\n"
+ " \"INET_COL\",\n"
+ " \"CHAR_ONE_COL\"\n"
+ "FROM\n"
+ " \"PG_IDE_SINK_TABLE\";";

Expand Down Expand Up @@ -282,7 +286,8 @@ private void initializeJdbcTable() {
+ " multipolygon,\n"
+ " geometrycollection,\n"
+ " geog,\n"
+ " inet_col\n"
+ " inet_col,\n"
+ " char_one_col\n"
+ " )\n"
+ "VALUES\n"
+ " (\n"
Expand Down Expand Up @@ -334,7 +339,8 @@ private void initializeJdbcTable() {
+ " 4326\n"
+ " ),\n"
+ " ST_GeographyFromText('POINT(-122.3452 47.5925)'),\n"
+ " '192.168.1.1'\n"
+ " '192.168.1.1',\n"
+ " 'T'\n"
+ " )");
}

Expand Down
2 changes: 1 addition & 1 deletion ...c-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_postgres_ide_source_and_sink.conf
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ source{
password = "test"
query ="""select gid, text_col, varchar_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col, double_precision_col,
smallserial_col, serial_col, bigserial_col, date_col, timestamp_col, bpchar_col, age, name, point, linestring, polygon_colums, multipoint,
multilinestring, multipolygon, geometrycollection, geog,inet_col from pg_ide_source_table"""
multilinestring, multipolygon, geometrycollection, geog,inet_col,char_one_col from pg_ide_source_table"""
}
}

Expand Down
8 changes: 8 additions & 0 deletions ...e-part-3/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcPostgresIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ public class JdbcPostgresIT extends TestSuiteBase implements TestResource {
+ " uuid_col UUID,\n"
+ " text_col TEXT,\n"
+ " varchar_col VARCHAR(255),\n"
+ " char_one_col CHAR(1),\n"
+ " char_col CHAR(10),\n"
+ " boolean_col bool,\n"
+ " smallint_col int2,\n"
Expand Down Expand Up @@ -121,6 +122,7 @@ public class JdbcPostgresIT extends TestSuiteBase implements TestResource {
+ " uuid_col UUID,\n"
+ " text_col TEXT,\n"
+ " varchar_col VARCHAR(255),\n"
+ " char_one_col CHAR(1),\n"
+ " char_col CHAR(10),\n"
+ " boolean_col bool,\n"
+ " smallint_col int2,\n"
Expand Down Expand Up @@ -156,6 +158,7 @@ public class JdbcPostgresIT extends TestSuiteBase implements TestResource {
+ "uuid_col, \n"
+ "text_col,\n"
+ "varchar_col,\n"
+ "char_one_col,\n"
+ "char_col,\n"
+ "boolean_col,\n"
+ "smallint_col,\n"
Expand Down Expand Up @@ -191,6 +194,7 @@ public class JdbcPostgresIT extends TestSuiteBase implements TestResource {
+ "uuid_col, \n"
+ " text_col,\n"
+ " varchar_col,\n"
+ " char_one_col,\n"
+ " char_col,\n"
+ " boolean_col,\n"
+ " smallint_col,\n"
Expand Down Expand Up @@ -392,6 +396,7 @@ private void initializeJdbcTable() {
+ " uuid_col,\n"
+ " text_col,\n"
+ " varchar_col,\n"
+ " char_one_col,\n"
+ " char_col,\n"
+ " boolean_col,\n"
+ " smallint_col,\n"
Expand Down Expand Up @@ -429,6 +434,7 @@ private void initializeJdbcTable() {
+ " gen_random_uuid(),\n"
+ " 'Hello World',\n"
+ " 'Test',\n"
+ " 'T',\n"
+ " 'Testing',\n"
+ " true,\n"
+ " 10,\n"
Expand Down Expand Up @@ -562,6 +568,7 @@ public void testCatalogForSaveMode() {
+ " pg_ide_sink_table_2 (gid,\n"
+ " text_col,\n"
+ " varchar_col,\n"
+ " char_one_col,\n"
+ " char_col,\n"
+ " boolean_col,\n"
+ " smallint_col,\n"
Expand Down Expand Up @@ -598,6 +605,7 @@ public void testCatalogForSaveMode() {
+ "',\n"
+ " 'Hello World',\n"
+ " 'Test',\n"
+ " 'T',\n"
+ " 'Testing',\n"
+ " true,\n"
+ " 10,\n"
Expand Down
2 changes: 1 addition & 1 deletion ...-jdbc-e2e/connector-jdbc-e2e-part-3/src/test/resources/jdbc_postgres_source_and_sink.conf
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ source{
driver = "org.postgresql.Driver"
user = "test"
password = "test"
query ="""select gid, uuid_col, text_col, varchar_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col, double_precision_col,
query ="""select gid, uuid_col, text_col, varchar_col, char_one_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col, double_precision_col,
smallserial_col, serial_col, bigserial_col, date_col, timestamp_col, bpchar_col, age, name, point, linestring, polygon_colums, multipoint,
multilinestring, multipolygon, geometrycollection, geog, json_col, jsonb_col,xml_col from pg_e2e_source_table"""
partition_column = "varchar_col"
Expand Down
2 changes: 1 addition & 1 deletion ...connector-jdbc-e2e-part-3/src/test/resources/jdbc_postgres_source_and_sink_copy_stmt.conf
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ source{
driver = "org.postgresql.Driver"
user = "test"
password = "test"
query ="""select gid, uuid_col, text_col, varchar_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col, double_precision_col,
query ="""select gid, uuid_col, text_col, varchar_col, char_one_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col, double_precision_col,
smallserial_col, serial_col, bigserial_col, date_col, timestamp_col, bpchar_col, age, name, point, linestring, polygon_colums, multipoint,
multilinestring, multipolygon, geometrycollection, geog, json_col, jsonb_col,xml_col from pg_e2e_source_table"""
partition_column = "varchar_col"
Expand Down
6 changes: 3 additions & 3 deletions .../connector-jdbc-e2e-part-3/src/test/resources/jdbc_postgres_source_and_sink_parallel.conf
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ source{
driver = "org.postgresql.Driver"
user = "test"
password = "test"
query ="""select gid, uuid_col, text_col, varchar_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col, double_precision_col,
query ="""select gid, uuid_col, text_col, varchar_col, char_one_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col, double_precision_col,
smallserial_col, serial_col, bigserial_col, date_col, timestamp_col, bpchar_col, age, name, point, linestring, polygon_colums, multipoint,
multilinestring, multipolygon, geometrycollection, geog, json_col, jsonb_col,xml_col from pg_e2e_source_table"""
partition_column= "gid"
Expand All @@ -45,9 +45,9 @@ sink {
user = "test"
password = "test"
connection_check_timeout_sec = 100
query ="""INSERT INTO pg_e2e_sink_table ( gid, uuid_col, text_col, varchar_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col,
query ="""INSERT INTO pg_e2e_sink_table ( gid, uuid_col, text_col, varchar_col, char_one_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col,
double_precision_col, smallserial_col, serial_col, bigserial_col, date_col, timestamp_col, bpchar_col, age, name, point,
linestring, polygon_colums, multipoint, multilinestring, multipolygon, geometrycollection, geog, json_col, jsonb_col,xml_col)
VALUES( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,? )"""
VALUES( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,? )"""
}
}
6 changes: 3 additions & 3 deletions ...dbc-e2e-part-3/src/test/resources/jdbc_postgres_source_and_sink_parallel_upper_lower.conf
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ source{
driver = "org.postgresql.Driver"
user = "test"
password = "test"
query ="""select gid, uuid_col, text_col, varchar_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col, double_precision_col,
query ="""select gid, uuid_col, text_col, varchar_col, char_one_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col, double_precision_col,
smallserial_col, serial_col, bigserial_col, date_col, timestamp_col, bpchar_col, age, name, point, linestring, polygon_colums, multipoint,
multilinestring, multipolygon, geometrycollection, geog, json_col, jsonb_col,xml_col from pg_e2e_source_table"""
partition_column= "gid"
Expand All @@ -49,9 +49,9 @@ sink {
user = "test"
password = "test"
connection_check_timeout_sec = 100
query ="""INSERT INTO pg_e2e_sink_table ( gid, uuid_col, text_col, varchar_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col,
query ="""INSERT INTO pg_e2e_sink_table ( gid, uuid_col, text_col, varchar_col, char_one_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col,
double_precision_col, smallserial_col, serial_col, bigserial_col, date_col, timestamp_col, bpchar_col, age, name, point,
linestring, polygon_colums, multipoint, multilinestring, multipolygon, geometrycollection, geog, json_col, jsonb_col,xml_col )
VALUES( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? ,?)"""
VALUES( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? ,?)"""
}
}
6 changes: 3 additions & 3 deletions ...bc-e2e/connector-jdbc-e2e-part-3/src/test/resources/jdbc_postgres_source_and_sink_xa.conf
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ source {
driver = "org.postgresql.Driver"
user = "test"
password = "test"
query ="""select gid, uuid_col, text_col, varchar_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col, double_precision_col,
query ="""select gid, uuid_col, text_col, varchar_col, char_one_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col, double_precision_col,
smallserial_col, serial_col, bigserial_col, date_col, timestamp_col, bpchar_col, age, name, point, linestring, polygon_colums, multipoint,
multilinestring, multipolygon, geometrycollection, geog, json_col, jsonb_col ,xml_col from pg_e2e_source_table"""
}
Expand All @@ -42,10 +42,10 @@ sink {
user = "test"
password = "test"
max_retries = 0
query ="""INSERT INTO pg_e2e_sink_table ( gid, uuid_col, text_col, varchar_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col,
query ="""INSERT INTO pg_e2e_sink_table ( gid, uuid_col, text_col, varchar_col, char_one_col, char_col, boolean_col, smallint_col, integer_col, bigint_col, decimal_col, numeric_col, real_col,
double_precision_col, smallserial_col, serial_col, bigserial_col, date_col, timestamp_col, bpchar_col, age, name, point,
linestring, polygon_colums, multipoint, multilinestring, multipolygon, geometrycollection, geog, json_col, jsonb_col ,xml_col)
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?)"""
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?)"""

is_exactly_once = "true"

Expand Down

0 comments on commit 776ac94

Please sign in to comment.