Skip to content

[FLINK-39071][oracle-cdc] Fix Oracle pipeline connector wrong field case #4292

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Hisoka-X wants to merge 1 commit into
base: master
Choose a base branch
from
+14 −14
Open

[FLINK-39071][oracle-cdc] Fix Oracle pipeline connector wrong field case #4292

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions ...-oracle/src/main/java/org/apache/flink/cdc/connectors/oracle/utils/OracleSchemaUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,7 @@ public static Schema getSchema(
DataType dataType = OracleTypeUtils.fromDbzColumn(column);
org.apache.flink.cdc.common.schema.Column cdcColumn =
org.apache.flink.cdc.common.schema.Column.physicalColumn(
column.name().toLowerCase(Locale.ROOT), dataType);
column.name(), dataType);
list.add(cdcColumn);
}
return Schema.newBuilder().setColumns(list).primaryKey(pks).build();
Expand All @@ -181,7 +181,7 @@ public static List<String> getTablePks(
while (rs.next()) {
String columnName;
columnName = rs.getString(1);
list.add(columnName.toLowerCase(Locale.ROOT));
list.add(columnName);
}
return list;
});
Expand Down
18 changes: 9 additions & 9 deletions ...cle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OraclePipelineITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -981,10 +981,10 @@ public void testGeometryType() throws Exception {
new CreateTableEvent(
TableId.tableId("DEBEZIUM", "MYLAKE"),
Schema.newBuilder()
.physicalColumn("feature_id", DataTypes.BIGINT().notNull())
.physicalColumn("name", DataTypes.VARCHAR(32))
.physicalColumn("shape", DataTypes.STRING())
.primaryKey(Arrays.asList("feature_id"))
.physicalColumn("FEATURE_ID", DataTypes.BIGINT().notNull())
.physicalColumn("NAME", DataTypes.VARCHAR(32))
.physicalColumn("SHAPE", DataTypes.STRING())
.primaryKey(Arrays.asList("FEATURE_ID"))
.build());

RowType rowType =
Expand Down Expand Up @@ -1568,11 +1568,11 @@ private CreateTableEvent getProductsCreateTableEvent(TableId tableId) {
return new CreateTableEvent(
tableId,
Schema.newBuilder()
.physicalColumn("id", DataTypes.BIGINT().notNull())
.physicalColumn("name", DataTypes.VARCHAR(255).notNull())
.physicalColumn("description", DataTypes.VARCHAR(512))
.physicalColumn("weight", DataTypes.FLOAT())
.primaryKey(Collections.singletonList("id"))
.physicalColumn("ID", DataTypes.BIGINT().notNull())
.physicalColumn("NAME", DataTypes.VARCHAR(255).notNull())
.physicalColumn("DESCRIPTION", DataTypes.VARCHAR(512))
.physicalColumn("WEIGHT", DataTypes.FLOAT())
.primaryKey(Collections.singletonList("ID"))
.build());
}

Expand Down
6 changes: 3 additions & 3 deletions ...pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OracleE2eITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ void testSyncWholeDatabase() throws Exception {
Statement stat = conn.createStatement()) {

waitUntilSpecificEvent(
"CreateTableEvent{tableId=DEBEZIUM.PRODUCTS, schema=columns={`id` BIGINT NOT NULL,`name` VARCHAR(255) NOT NULL,`description` VARCHAR(512),`weight` FLOAT}, primaryKeys=id, options=()}");
"CreateTableEvent{tableId=DEBEZIUM.PRODUCTS, schema=columns={`ID` BIGINT NOT NULL,`NAME` VARCHAR(255) NOT NULL,`DESCRIPTION` VARCHAR(512),`WEIGHT` FLOAT}, primaryKeys=ID, options=()}");
waitUntilSpecificEvent(
"DataChangeEvent{tableId=DEBEZIUM.PRODUCTS, before=[], after=[109, spare tire, 24 inch spare tire, 22.2], op=INSERT, meta=()}");
waitUntilSpecificEvent(
Expand All @@ -171,7 +171,7 @@ void testSyncWholeDatabase() throws Exception {
"DataChangeEvent{tableId=DEBEZIUM.PRODUCTS, before=[], after=[105, hammer, 14oz carpenters hammer, 0.875], op=INSERT, meta=()}");

waitUntilSpecificEvent(
"CreateTableEvent{tableId=DEBEZIUM.CUSTOMERS, schema=columns={`id` BIGINT NOT NULL,`name` VARCHAR(255) NOT NULL,`address` VARCHAR(1024),`phone_number` VARCHAR(512)}, primaryKeys=id, options=()}");
"CreateTableEvent{tableId=DEBEZIUM.CUSTOMERS, schema=columns={`ID` BIGINT NOT NULL,`NAME` VARCHAR(255) NOT NULL,`ADDRESS` VARCHAR(1024),`PHONE_NUMBER` VARCHAR(512)}, primaryKeys=ID, options=()}");
waitUntilSpecificEvent(
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[171798691841, user_1, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
Expand Down Expand Up @@ -225,7 +225,7 @@ void testSyncWholeDatabase() throws Exception {
waitUntilSpecificEvent(
"DataChangeEvent{tableId=DEBEZIUM.PRODUCTS, before=[107, rocks, box of assorted rocks, 5.3], after=[107, rocks, box of assorted rocks, 5.1], op=UPDATE, meta=()}");
waitUntilSpecificEvent(
"CreateTableEvent{tableId=DEBEZIUM.CUSTOMERS_1, schema=columns={`id` BIGINT NOT NULL,`name` VARCHAR(255) NOT NULL,`address` VARCHAR(1024),`phone_number` VARCHAR(512)}, primaryKeys=id, options=()}");
"CreateTableEvent{tableId=DEBEZIUM.CUSTOMERS_1, schema=columns={`ID` BIGINT NOT NULL,`NAME` VARCHAR(255) NOT NULL,`ADDRESS` VARCHAR(1024),`PHONE_NUMBER` VARCHAR(512)}, primaryKeys=ID, options=()}");
waitUntilSpecificEvent(
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[171798691842, user_10, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
Expand Down
Loading