From 4fc72ce60b9cf729ca0b8e1c776bb1350a46ea3c Mon Sep 17 00:00:00 2001
From: wudi <676366545@qq.com>
Date: Thu, 26 Sep 2024 18:36:14 +0800
Subject: [PATCH] fix unionall bug

---
 .../doris/flink/cfg/DorisReadOptions.java     | 18 ++++++++
 .../flink/table/DorisDynamicTableSource.java  |  3 +-
 .../doris/flink/source/DorisSourceITCase.java | 46 +++++++++++++++++++
 3 files changed, 66 insertions(+), 1 deletion(-)

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/DorisReadOptions.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/DorisReadOptions.java
index 2f6cd8a86..42cd83a84 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/DorisReadOptions.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/DorisReadOptions.java
@@ -187,6 +187,24 @@ public int hashCode() {
                 flightSqlPort);
     }
 
+    public DorisReadOptions copy(){
+        return new DorisReadOptions(
+                readFields,
+                filterQuery,
+                requestTabletSize,
+                requestConnectTimeoutMs,
+                requestReadTimeoutMs,
+                requestQueryTimeoutS,
+                requestRetries,
+                requestBatchSize,
+                execMemLimit,
+                deserializeQueueSize,
+                deserializeArrowAsync,
+                useOldApi,
+                useFlightSql,
+                flightSqlPort);
+    }
+
     /** Builder of {@link DorisReadOptions}. */
     public static class Builder {
 
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java
index e55d3631b..aaf5e9436 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java
@@ -166,9 +166,10 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
 
     @Override
     public DynamicTableSource copy() {
+        // filterQuery/readFields of readOption may be overwritten in union all sql
         DorisDynamicTableSource newSource =
                 new DorisDynamicTableSource(
-                        options, readOptions, lookupOptions, physicalSchema, physicalRowDataType);
+                        options, readOptions.copy(), lookupOptions, physicalSchema, physicalRowDataType);
         newSource.resolvedFilterQuery = new ArrayList<>(this.resolvedFilterQuery);
         return newSource;
     }
diff --git a/flink-doris-connector/src/test/java/org/apache/doris/flink/source/DorisSourceITCase.java b/flink-doris-connector/src/test/java/org/apache/doris/flink/source/DorisSourceITCase.java
index 4fb6fba8f..7176ad826 100644
--- a/flink-doris-connector/src/test/java/org/apache/doris/flink/source/DorisSourceITCase.java
+++ b/flink-doris-connector/src/test/java/org/apache/doris/flink/source/DorisSourceITCase.java
@@ -63,6 +63,8 @@ public class DorisSourceITCase extends AbstractITCaseService {
             "tbl_read_tbl_push_down_with_union_all";
     static final String TABLE_CSV_JM = "tbl_csv_jm_source";
     static final String TABLE_CSV_TM = "tbl_csv_tm_source";
+    private static final String TABLE_READ_TBL_PUSH_DOWN_WITH_UNION_ALL_NOT_EQ_FILTER =
+            "tbl_read_tbl_push_down_with_union_all_not_eq_filter";
 
     @Rule
     public final MiniClusterWithClientResource miniClusterResource =
@@ -353,6 +355,50 @@ public void testTableSourceFilterWithUnionAll() throws Exception {
         checkResultInAnyOrder("testTableSourceFilterWithUnionAll", expected, actual.toArray());
     }
 
+    @Test
+    public void testTableSourceFilterWithUnionAllNotEqualFilter() throws Exception {
+        LOG.info("starting to execute testTableSourceFilterWithUnionAllNotEqualFilter case.");
+        initializeTable(TABLE_READ_TBL_PUSH_DOWN_WITH_UNION_ALL_NOT_EQ_FILTER);
+        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        env.setParallelism(DEFAULT_PARALLELISM);
+        final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+        String sourceDDL =
+                String.format(
+                        "CREATE TABLE doris_source_filter_with_union_all ("
+                                + " name STRING,"
+                                + " age INT"
+                                + ") WITH ("
+                                + " 'connector' = '"
+                                + DorisConfigOptions.IDENTIFIER
+                                + "',"
+                                + " 'fenodes' = '%s',"
+                                + " 'table.identifier' = '%s',"
+                                + " 'username' = '%s',"
+                                + " 'password' = '%s'"
+                                + ")",
+                        getFenodes(),
+                        DATABASE + "." + TABLE_READ_TBL_PUSH_DOWN_WITH_UNION_ALL_NOT_EQ_FILTER,
+                        getDorisUsername(),
+                        getDorisPassword());
+        tEnv.executeSql(sourceDDL);
+        String querySql =
+                "  SELECT * FROM doris_source_filter_with_union_all where name = 'doris'"
+                        + " UNION ALL "
+                        + "SELECT * FROM doris_source_filter_with_union_all where name in ('error','flink')";
+        TableResult tableResult = tEnv.executeSql(querySql);
+
+        List<String> actual = new ArrayList<>();
+        try (CloseableIterator<Row> iterator = tableResult.collect()) {
+            while (iterator.hasNext()) {
+                actual.add(iterator.next().toString());
+            }
+        }
+
+        String[] expected = new String[] {"+I[flink, 10]", "+I[doris, 18]"};
+        checkResultInAnyOrder("testTableSourceFilterWithUnionAllNotEqualFilter", expected, actual.toArray());
+    }
+
     @Test
     public void testJobManagerFailoverSource() throws Exception {
         LOG.info("start to test JobManagerFailoverSource.");