[FLINK-39171][oracle-cdc] Extract removeQuotes as utility method for Oracle column name handling

Extract the removeQuotes method to OracleSchemaUtils to centralize Oracle table/column
name case handling logic. This method follows Oracle naming rules: names enclosed in
double quotes retain their original case, otherwise they are converted to uppercase.

This change improves code reuse between OracleSchemaUtils and BaseParserListener,
making it easier to add configuration options for case handling in the future.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: lvyanquan

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/main/java/org/apache/flink/cdc/connectors/oracle/source/parser/BaseParserListener.java

@@ -17,6 +17,8 @@
 
 package org.apache.flink.cdc.connectors.oracle.source.parser;
 
+import org.apache.flink.cdc.connectors.oracle.utils.OracleSchemaUtils;
+
 import io.debezium.ddl.parser.oracle.generated.PlSqlParser;
 import io.debezium.ddl.parser.oracle.generated.PlSqlParserBaseListener;
 
@@ -76,21 +78,6 @@ public String getColumnName(final PlSqlParser.New_column_nameContext ctx) {
      * @return parsed table or column name from the supplied name argument
      */
     private static String getTableOrColumnName(String name) {
-        return removeQuotes(name, true);
-    }
-
-    /**
-     * Removes leading and trailing double quote characters from the provided string.
-     *
-     * @param text value to have double quotes removed
-     * @param upperCaseIfNotQuoted control if returned string is upper-cased if not quoted
-     * @return string that has had quotes removed
-     */
-    @SuppressWarnings("SameParameterValue")
-    private static String removeQuotes(String text, boolean upperCaseIfNotQuoted) {
-        if (text != null && text.length() > 2 && text.startsWith("\"") && text.endsWith("\"")) {
-            return text.substring(1, text.length() - 1);
-        }
-        return upperCaseIfNotQuoted ? text.toUpperCase() : text;
+        return OracleSchemaUtils.removeQuotes(name);
     }
 }

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

@@ -46,6 +46,25 @@ public class OracleSchemaUtils {
 
     private static final Logger LOG = LoggerFactory.getLogger(OracleSchemaUtils.class);
 
+    /**
+     * Removes leading and trailing double quote characters from the provided string.
+     *
+     * <p>Oracle table and column names are inherently stored in upper-case; however, if the objects
+     * are created using double-quotes, the case of the object name is retained. This method will
+     * adhere to those rules and will always return the name in upper-case unless the provided name
+     * is double-quoted in which the returned value will have the double-quotes removed and case
+     * retained.
+     *
+     * @param text value to have double quotes removed
+     * @return string that has had quotes removed
+     */
+    public static String removeQuotes(String text) {
+        if (text != null && text.length() > 2 && text.startsWith("\"") && text.endsWith("\"")) {
+            return text.substring(1, text.length() - 1);
+        }
+        return text.toUpperCase();
+    }
+
     public static List<TableId> listTables(
             OracleSourceConfig sourceConfig, @Nullable String dbName) {
         try (JdbcConnection jdbc = createOracleConnection(sourceConfig)) {
@@ -127,16 +146,23 @@ public static Schema toSchema(Table table) {
                         .map(OracleSchemaUtils::toColumn)
                         .collect(Collectors.toList());
 
+        List<String> primaryKeys =
+                table.primaryKeyColumnNames().stream()
+                        .map(OracleSchemaUtils::removeQuotes)
+                        .collect(Collectors.toList());
+
         return Schema.newBuilder()
                 .setColumns(columns)
-                .primaryKey(table.primaryKeyColumnNames())
+                .primaryKey(primaryKeys)
                 .comment(table.comment())
                 .build();
     }
 
     public static Column toColumn(io.debezium.relational.Column column) {
         return Column.physicalColumn(
-                column.name(), OracleTypeUtils.fromDbzColumn(column), column.comment());
+                removeQuotes(column.name()),
+                OracleTypeUtils.fromDbzColumn(column),
+                column.comment());
     }
 
     public static io.debezium.relational.TableId toDbzTableId(TableId tableId) {
@@ -158,7 +184,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(), dataType);
+                            removeQuotes(column.name()), dataType);
             list.add(cdcColumn);
         }
         return Schema.newBuilder().setColumns(list).primaryKey(pks).build();
@@ -181,7 +207,7 @@ public static List<String> getTablePks(
                         while (rs.next()) {
                             String columnName;
                             columnName = rs.getString(1);
-                            list.add(columnName);
+                            list.add(removeQuotes(columnName));
                         }
                         return list;
                     });