[Improve] revert pr 479 to remove resolvedFilterQuery in DorisSource (#…

…497)
apache · Oct 11, 2024 · 3d13c7f · 3d13c7f
1 parent 66ce015
commit 3d13c7f
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 23 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/source/DorisSource.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/source/DorisSource.java
@@ -27,7 +27,6 @@
 import org.apache.flink.api.connector.source.SplitEnumeratorContext;
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
 import org.apache.flink.core.io.SimpleVersionedSerializer;
-import org.apache.flink.util.StringUtils;
 
 import org.apache.doris.flink.cfg.DorisOptions;
 import org.apache.doris.flink.cfg.DorisReadOptions;
@@ -65,18 +64,14 @@ public class DorisSource<OUT>
     private final Boundedness boundedness;
     private final DorisDeserializationSchema<OUT> deserializer;
 
-    private final List<String> resolvedFilterQuery;
-
     public DorisSource(
             DorisOptions options,
             DorisReadOptions readOptions,
             Boundedness boundedness,
-            List<String> resolvedFilterQuery,
             DorisDeserializationSchema<OUT> deserializer) {
         this.options = options;
         this.readOptions = readOptions;
         this.boundedness = boundedness;
-        this.resolvedFilterQuery = resolvedFilterQuery;
         this.deserializer = deserializer;
     }
 
@@ -100,15 +95,6 @@ public SourceReader<OUT, DorisSourceSplit> createReader(SourceReaderContext read
     public SplitEnumerator<DorisSourceSplit, PendingSplitsCheckpoint> createEnumerator(
             SplitEnumeratorContext<DorisSourceSplit> context) throws Exception {
         List<DorisSourceSplit> dorisSourceSplits = new ArrayList<>();
-        if (!resolvedFilterQuery.isEmpty()) {
-            String filterQuery = String.join(" AND ", resolvedFilterQuery);
-            if (StringUtils.isNullOrWhitespaceOnly(readOptions.getFilterQuery())) {
-                readOptions.setFilterQuery(filterQuery);
-            } else {
-                readOptions.setFilterQuery(
-                        String.join(" AND ", readOptions.getFilterQuery(), filterQuery));
-            }
-        }
         List<PartitionDefinition> partitions =
                 RestService.findPartitions(options, readOptions, LOG);
         for (int index = 0; index < partitions.size(); index++) {
@@ -162,7 +148,6 @@ public static class DorisSourceBuilder<OUT> {
         // Boundedness
         private Boundedness boundedness;
         private DorisDeserializationSchema<OUT> deserializer;
-        private List<String> resolvedFilterQuery = new ArrayList<>();
 
         DorisSourceBuilder() {
             boundedness = Boundedness.BOUNDED;
@@ -208,11 +193,6 @@ public DorisSourceBuilder<OUT> setDeserializer(
             return this;
         }
 
-        public DorisSourceBuilder<OUT> setResolvedFilterQuery(List<String> resolvedFilterQuery) {
-            this.resolvedFilterQuery = resolvedFilterQuery;
-            return this;
-        }
-
         /**
          * Build the {@link DorisSource}.
          *
@@ -222,8 +202,7 @@ public DorisSource<OUT> build() {
             if (readOptions == null) {
                 readOptions = DorisReadOptions.builder().build();
             }
-            return new DorisSource<>(
-                    options, readOptions, boundedness, resolvedFilterQuery, deserializer);
+            return new DorisSource<>(options, readOptions, boundedness, deserializer);
         }
     }
 }
diff --git a/...k-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java b/...k-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java
@@ -90,6 +90,11 @@ public ChangelogMode getChangelogMode() {
 
     @Override
     public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
+        if (StringUtils.isNullOrWhitespaceOnly(readOptions.getFilterQuery())) {
+            String filterQuery = resolvedFilterQuery.stream().collect(Collectors.joining(" AND "));
+            readOptions.setFilterQuery(filterQuery);
+        }
+
         if (StringUtils.isNullOrWhitespaceOnly(readOptions.getReadFields())) {
             String[] selectFields =
                     DataType.getFieldNames(physicalRowDataType).toArray(new String[0]);
@@ -123,7 +128,6 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderCon
                     DorisSource.<RowData>builder()
                             .setDorisReadOptions(readOptions)
                             .setDorisOptions(options)
-                            .setResolvedFilterQuery(resolvedFilterQuery)
                             .setDeserializer(
                                     new RowDataDeserializationSchema(
                                             (RowType) physicalRowDataType.getLogicalType()))