fix oracle cdb+pdb model

JNSimba · Jul 11, 2024 · de2450c · de2450c
1 parent e8ae64e
commit de2450c
Showing 1 changed file with 21 additions and 0 deletions.
diff --git a/...s-connector/src/main/java/org/apache/doris/flink/tools/cdc/oracle/OracleDatabaseSync.java b/...s-connector/src/main/java/org/apache/doris/flink/tools/cdc/oracle/OracleDatabaseSync.java
@@ -44,6 +44,7 @@
 import java.sql.DriverManager;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -63,6 +64,7 @@ public class OracleDatabaseSync extends DatabaseSync {
     private static final Logger LOG = LoggerFactory.getLogger(OracleDatabaseSync.class);
 
     private static final String JDBC_URL = "jdbc:oracle:thin:@%s:%d:%s";
+    private static final String PDB_KEY = "debezium.database.pdb.name";
 
     public OracleDatabaseSync() throws SQLException {
         super();
@@ -108,9 +110,11 @@ public Connection getConnection() throws SQLException {
     public List<SourceSchema> getSchemaList() throws Exception {
         String databaseName = config.get(OracleSourceOptions.DATABASE_NAME);
         String schemaName = config.get(OracleSourceOptions.SCHEMA_NAME);
+
         List<SourceSchema> schemaList = new ArrayList<>();
         LOG.info("database-name {}, schema-name {}", databaseName, schemaName);
         try (Connection conn = getConnection()) {
+            setSessionToPdb(conn);
             DatabaseMetaData metaData = conn.getMetaData();
             try (ResultSet tables =
                     metaData.getTables(databaseName, schemaName, "%", new String[] {"TABLE"})) {
@@ -134,6 +138,23 @@ public List<SourceSchema> getSchemaList() throws Exception {
         return schemaList;
     }
 
+    private void setSessionToPdb(Connection conn) throws SQLException {
+        String pdbName = null;
+        for (Map.Entry<String, String> entry : config.toMap().entrySet()) {
+            String key = entry.getKey();
+            if (key.equals(PDB_KEY)) {
+                pdbName = entry.getValue();
+                break;
+            }
+        }
+        if (!StringUtils.isNullOrWhitespaceOnly(pdbName)) {
+            LOG.info("Found pdb name in config, set session to pdb to {}", pdbName);
+            try (Statement statement = conn.createStatement()) {
+                statement.execute("alter session set container=" + pdbName);
+            }
+        }
+    }
+
     @Override
     public DataStreamSource<String> buildCdcSource(StreamExecutionEnvironment env) {
         Properties debeziumProperties = new Properties();