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

Author: Hisoka-X

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/main/java/org/apache/flink/cdc/connectors/oracle/utils/OracleSchemaUtils.java

@@ -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();
@@ -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;
                     });

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OraclePipelineITCase.java

@@ -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 =
@@ -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());
     }
 

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OracleE2eITCase.java

@@ -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(
@@ -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(
@@ -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(