[fix](fe) Deduplicate recursive CTE fragment reset requests

fe/fe-core/src/main/java/org/apache/doris/qe/runtime/ThriftPlansBuilder.java

@@ -822,6 +822,7 @@ private static Set<Integer> setParamsForRecursiveCteNode(List<PipelineDistribute
                 List<TRecCTETarget> targets = new ArrayList<>();
                 // reset infos for all instances of child fragments (used to reset state)
                 List<TRecCTEResetInfo> fragmentsToReset = new ArrayList<>();
+                Set<String> resetFragmentKeys = new HashSet<>();
                 // The recursive side is under the right child; collect all fragments
                 List<PlanFragment> childFragments = new ArrayList<>();
                 recursiveCteNode.getChild(1).getChild(0).getFragment().collectAll(PlanFragment.class::isInstance,
@@ -845,6 +846,11 @@ private static Set<Integer> setParamsForRecursiveCteNode(List<PipelineDistribute
                                 String.format("can't find TNetworkAddress for fragment %d", childFragmentId));
                     }
                     for (TNetworkAddress address : tNetworkAddresses) {
+                        String resetFragmentKey = childFragmentId.asInt() + "@"
+                                + address.getHostname() + ":" + address.getPort();
+                        if (!resetFragmentKeys.add(resetFragmentKey)) {
+                            continue;
+                        }
                         TRecCTEResetInfo tRecCTEResetInfo = new TRecCTEResetInfo();
                         tRecCTEResetInfo.setFragmentId(childFragmentId.asInt());
                         tRecCTEResetInfo.setAddr(address);

regression-test/data/recursive_cte/shared_cte_reset_test.out

@@ -0,0 +1,8 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !shared_cte_reset --
+1	MP:a
+2	MP:a
+3	MP:b
+a	MP:a
+b	MP:b
+

regression-test/suites/recursive_cte/shared_cte_reset_test.groovy

@@ -0,0 +1,67 @@
+// 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("shared_cte_reset_test", "rec_cte") {
+    qt_shared_cte_reset """
+        WITH RECURSIVE
+        org_mp_bridge AS (
+            SELECT '1' AS organization_id, 'a' AS marketplace_account_id
+            UNION ALL SELECT '2', 'a'
+            UNION ALL SELECT '3', 'b'
+        ),
+        edges AS (
+            SELECT CONCAT('ORG:', organization_id) AS src,
+                   CONCAT('MP:', marketplace_account_id) AS dst
+            FROM org_mp_bridge
+            UNION
+            SELECT CONCAT('MP:', marketplace_account_id),
+                   CONCAT('ORG:', organization_id)
+            FROM org_mp_bridge
+        ),
+        nodes AS (
+            SELECT src AS node FROM edges
+            UNION
+            SELECT dst FROM edges
+        ),
+        reach (start_node, node) AS (
+            SELECT node, node FROM nodes
+            UNION
+            SELECT r.start_node, e.dst
+            FROM reach r
+            JOIN edges e ON r.node = e.src
+        ),
+        node_group AS (
+            SELECT node, MIN(start_node) AS reconciliation_group_id
+            FROM reach
+            GROUP BY node
+        ),
+        org_group AS (
+            SELECT DISTINCT b.organization_id, g.reconciliation_group_id
+            FROM org_mp_bridge b
+            JOIN node_group g ON g.node = CONCAT('ORG:', b.organization_id)
+        ),
+        mp_group AS (
+            SELECT DISTINCT b.marketplace_account_id, g.reconciliation_group_id
+            FROM org_mp_bridge b
+            JOIN node_group g ON g.node = CONCAT('MP:', b.marketplace_account_id)
+        )
+        SELECT organization_id, reconciliation_group_id FROM org_group
+        UNION ALL
+        SELECT marketplace_account_id, reconciliation_group_id FROM mp_group
+        ORDER BY organization_id, reconciliation_group_id
+    """
+}