[fix](planner) fix legacy planner compute wrong result when parallel_pipeline_task_num =1

fe/fe-core/src/main/java/org/apache/doris/common/TreeNode.java

@@ -251,4 +251,14 @@ public boolean anyMatch(Predicate<TreeNode<? extends NodeType>> func) {
         return false;
     }
 
+    /** foreachDown */
+    public void foreachDown(Predicate<TreeNode<NodeType>> visitor) {
+        if (!visitor.test(this)) {
+            return;
+        }
+
+        for (TreeNode<NodeType> child : getChildren()) {
+            child.foreachDown(visitor);
+        }
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java

@@ -236,8 +236,14 @@ private PlanFragment createPlanFragments(
         // move 'result' to end, it depends on all of its children
         fragments.remove(result);
         fragments.add(result);
-        if ((!isPartitioned && result.isPartitioned() && result.getPlanRoot().getNumInstances() > 1)
-                || (!(root instanceof SortNode) && root.hasOffset())) {
+        List<ScanNode> scanNodes = result.getPlanRoot().collectInCurrentFragment(p -> p instanceof ScanNode);
+        int scanRangeNum = 0;
+        for (ScanNode scanNode : scanNodes) {
+            scanRangeNum += scanNode.getScanRangeLocations(0).size();
+        }
+        boolean inputFragmentHasMultiInstances = !isPartitioned && result.isPartitioned()
+                && (result.getPlanRoot().getNumInstances() > 1 || scanRangeNum > 1);
+        if (inputFragmentHasMultiInstances || (!(root instanceof SortNode) && root.hasOffset())) {
             result = createMergeFragment(result);
             fragments.add(result);
         }

fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java

@@ -62,6 +62,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 /**
@@ -1271,4 +1273,27 @@ public void addIntermediateOutputTupleDescList(TupleDescriptor tupleDescriptor)
     public void addIntermediateProjectList(List<Expr> exprs) {
         intermediateProjectListList.add(exprs);
     }
+
+    public <T extends PlanNode> List<T> collectInCurrentFragment(Predicate<PlanNode> predicate) {
+        List<PlanNode> result = Lists.newArrayList();
+        foreachDownInCurrentFragment(child -> {
+            if (predicate.test(child)) {
+                result.add(child);
+            }
+        });
+        return (List) result;
+    }
+
+    /** foreachDownInCurrentFragment */
+    public void foreachDownInCurrentFragment(Consumer<PlanNode> visitor) {
+        int currentFragmentId = getFragmentId().asInt();
+        foreachDown(child -> {
+            PlanNode childNode = (PlanNode) child;
+            if (childNode.getFragmentId().asInt() != currentFragmentId) {
+                return false;
+            }
+            visitor.accept(childNode);
+            return true;
+        });
+    }
 }

regression-test/suites/nereids_syntax_p0/not_equal.groovy

@@ -0,0 +1,48 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("not_equal") {
+    multi_sql """
+        drop table if exists random_tbl_test;
+        CREATE TABLE `random_tbl_test` (
+          `id` int NULL
+        ) ENGINE=OLAP
+        DISTRIBUTED BY RANDOM BUCKETS 10
+        PROPERTIES (
+            "replication_allocation" = "tag.location.default: 1"
+        );
+
+        insert into random_tbl_test select * from numbers('number'='100');
+
+        set enable_nereids_planner=false;
+        set parallel_pipeline_task_num=1;
+        set enable_pipeline_engine=true;
+        set enable_pipeline_x_engine=false;
+        set enable_shared_scan=true;"""
+
+    explain {
+        sql "select * from random_tbl_test where 1 ! = 2"
+        check { String explainStr ->
+            assertEquals(2, explainStr.count("PLAN FRAGMENT"))
+        }
+    }
+
+    test {
+        sql "select * from random_tbl_test where 1 ! = 2"
+        rowNum(100)
+    }
+}