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

### What problem does this PR solve?

fix legacy planner compute wrong result when parallel_pipeline_task_num
= 1

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [x] Regression test
    - [ ] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [x] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

Author: 924060929

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)
+    }
+}