From 568537814f4d6437b782f06d765c07d8704aef00 Mon Sep 17 00:00:00 2001
From: wudi <676366545@qq.com>
Date: Wed, 30 Oct 2024 14:36:59 +0800
Subject: [PATCH] fix oracle regular expresion too long

---
 .../tools/cdc/oracle/OracleDatabaseSync.java      | 15 ++++++++++-----
 1 file changed, 10 insertions(+), 5 deletions(-)

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/oracle/OracleDatabaseSync.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/oracle/OracleDatabaseSync.java
index 9b6844544..1a6a41877 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/oracle/OracleDatabaseSync.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/oracle/OracleDatabaseSync.java
@@ -164,11 +164,16 @@ public DataStreamSource<String> buildCdcSource(StreamExecutionEnvironment env) {
         Preconditions.checkNotNull(databaseName, "database-name in oracle is required");
         Preconditions.checkNotNull(schemaName, "schema-name in oracle is required");
         String tableName = config.get(OracleSourceOptions.TABLE_NAME);
-        // When debezium incrementally reads, it will be judged based on regexp_like.
-        // When the regular length exceeds 512, an error will be reported,
-        // like ORA-12733: regular expression too long
-        if (tableName.length() > 512) {
-            tableName = StringUtils.isNullOrWhitespaceOnly(includingTables) ? ".*" : tableName;
+        // When debezium incrementally reads (refer LogMinerQueryBuilder.listOfPatternsToSql),
+        // it will be judged based on regexp_like. When the regular length exceeds 512, an error
+        // will be reported, like ORA-12733: regular expression too long
+        if (tableName.length() > 450) {
+            // REGEXP_LIKE('^SCHEMA.(TBL1|TBL2)$')
+            if (StringUtils.isNullOrWhitespaceOnly(excludingTables)
+                    && (StringUtils.isNullOrWhitespaceOnly(includingTables)
+                            || ".*".equals(includingTables))) {
+                tableName = ".*";
+            }
         }
 
         String url = config.get(OracleSourceOptions.URL);