fix oracle ora-12733

JNSimba · Dec 26, 2024 · 0ccf728 · 0ccf728
1 parent 2bd1c8f
commit 0ccf728
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 11 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java
@@ -42,6 +42,7 @@
 import org.apache.doris.flink.sink.writer.serializer.DorisRecordSerializer;
 import org.apache.doris.flink.sink.writer.serializer.JsonDebeziumSchemaSerializer;
 import org.apache.doris.flink.table.DorisConfigOptions;
+import org.apache.doris.flink.tools.cdc.oracle.OracleDatabaseSync;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -58,6 +59,7 @@
 import java.util.Properties;
 import java.util.Set;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import static org.apache.flink.cdc.debezium.utils.JdbcUrlUtils.PROPERTIES_PREFIX;
 
@@ -359,7 +361,14 @@ protected boolean isSyncNeeded(String tableName) {
 
     protected String getSyncTableList(List<String> syncTables) {
         if (!singleSink) {
-            return String.format("(%s)\\.(%s)", getTableListPrefix(), String.join("|", syncTables));
+            if (this instanceof OracleDatabaseSync) {
+                return syncTables.stream()
+                        .map(v -> getTableListPrefix() + "\\." + v)
+                        .collect(Collectors.joining("|"));
+            } else {
+                return String.format(
+                        "(%s)\\.(%s)", getTableListPrefix(), String.join("|", syncTables));
+            }
         } else {
             // includingTablePattern and ^excludingPattern
             if (includingTables == null) {

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
@@ -164,16 +164,10 @@ 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 (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 = ".*";
-            }
+        // LogMinerQueryBuilder.buildTablePredicate is separated by commas to avoid
+        // the error ORA-12733 when the regexp_like regular expression exceeds 512 characters
+        if (!singleSink && tableName.length() > 256) {
+            tableName = tableName.replace("|", ",");
         }
 
         String url = config.get(OracleSourceOptions.URL);