(seatunnel) branch dev updated: [Fix][Connector-V2] Fix Oracle CDC schema cache matching (#11110)

[lidongdai]

This is an automated email from the ASF dual-hosted git repository.

davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git


The following commit(s) were added to refs/heads/dev by this push:
     new c698247699 [Fix][Connector-V2] Fix Oracle CDC schema cache matching 
(#11110)
c698247699 is described below

commit c69824769969c52754da3c8f0ac58cf22e15ce24
Author: Jast <shengh...@apache.org>
AuthorDate: Thu Jun 18 08:26:13 2026 +0800

    [Fix][Connector-V2] Fix Oracle CDC schema cache matching (#11110)
    
    Co-authored-by: zhangshenghang <zhangshengh...@apache.org>
---
 .../seatunnel/cdc/oracle/utils/OracleSchema.java   | 22 +++++++++++++++++++---
 .../cdc/oracle/utils/OracleUtilsTest.java          | 14 ++++++++++++++
 2 files changed, 33 insertions(+), 3 deletions(-)

diff --git 
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleSchema.java
 
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleSchema.java
index f2713e3481..a2db1d9ceb 100644
--- 
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleSchema.java
+++ 
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleSchema.java
@@ -74,14 +74,15 @@ public class OracleSchema {
                     null,
                     false);
             for (TableId id : tables.tableIds()) {
-                if (tableMap.containsKey(id)) {
+                TableId tableMapId = resolveTableId(id, tableId, tableMap);
+                if (tableMap.containsKey(tableMapId)) {
                     Table table =
                             CatalogTableUtils.mergeCatalogTableConfig(
-                                    tables.forTable(id), tableMap.get(id));
+                                    tables.forTable(id), 
tableMap.get(tableMapId));
                     TableChanges.TableChange tableChange =
                             new TableChanges.TableChange(
                                     TableChanges.TableChangeType.CREATE, 
table);
-                    schemasByTableId.put(id, tableChange);
+                    schemasByTableId.put(tableMapId, tableChange);
                 }
             }
         } catch (SQLException e) {
@@ -96,4 +97,19 @@ public class OracleSchema {
 
         return schemasByTableId.get(tableId);
     }
+
+    static TableId resolveTableId(
+            TableId readTableId, TableId requestedTableId, Map<TableId, ?> 
tableMap) {
+        if (tableMap.containsKey(readTableId)) {
+            return readTableId;
+        }
+
+        TableId readTableIdWithRequestedCatalog =
+                new TableId(requestedTableId.catalog(), readTableId.schema(), 
readTableId.table());
+        if (tableMap.containsKey(readTableIdWithRequestedCatalog)) {
+            return readTableIdWithRequestedCatalog;
+        }
+
+        return readTableId;
+    }
 }
diff --git 
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleUtilsTest.java
 
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleUtilsTest.java
index 10c253da83..a3856a7ff7 100644
--- 
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleUtilsTest.java
+++ 
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleUtilsTest.java
@@ -26,6 +26,9 @@ import org.junit.jupiter.api.Test;
 
 import io.debezium.relational.TableId;
 
+import java.util.Collections;
+import java.util.Map;
+
 public class OracleUtilsTest {
     @Test
     public void testSplitScanQuery() {
@@ -70,4 +73,15 @@ public class OracleUtilsTest {
         Assertions.assertEquals(
                 "SELECT * FROM \"schema1\".\"table1\" WHERE \"id\" >= ?", 
splitScanSQL);
     }
+
+    @Test
+    public void testResolveTableIdWithRequestedCatalog() {
+        TableId requestedTableId = TableId.parse("ORCLPDB.LIB_B.T_B1");
+        TableId readTableIdWithoutCatalog = new TableId(null, "LIB_B", "T_B1");
+        Map<TableId, ?> tableMap = Collections.singletonMap(requestedTableId, 
null);
+
+        Assertions.assertEquals(
+                requestedTableId,
+                OracleSchema.resolveTableId(readTableIdWithoutCatalog, 
requestedTableId, tableMap));
+    }
 }