fix(compile): preserve remote pipeline receiver ownership

pkg/sql/compile/compile.go

@@ -5754,6 +5754,7 @@ func (c *Compile) newEmptyMergeScope() *Scope {
 
 func (c *Compile) newMergeScope(ss []*Scope) *Scope {
 	rs := c.newEmptyMergeScope()
+	ss = c.groupRemoteRunDependenciesByCNIfNeeded(ss, rs.NodeInfo)
 	rs.PreScopes = ss
 
 	rs.Proc = c.proc.NewNoContextChildProc(len(ss))
@@ -5992,6 +5993,32 @@ func (c *Compile) mergeShuffleScopesIfNeeded(ss []*Scope, force bool) []*Scope {
 	return rs
 }
 
+// groupRemoteRunDependenciesByCNIfNeeded preserves the ownership boundary of
+// in-process dispatch and connector receivers when a local merge makes its
+// inputs separate RemoteRun units. A scope that targets a receiver owned by a
+// sibling scope cannot execute remotely on its own; wrapping all inputs from
+// the same CN in one merge scope keeps those dependencies in one serialized
+// tree. Only a non-local invalid input triggers regrouping; once triggered, the
+// whole input stage is grouped consistently by CN. Independent input stages
+// retain the direct fast path.
+func (c *Compile) groupRemoteRunDependenciesByCNIfNeeded(
+	ss []*Scope,
+	mergeNode engine.Node,
+) []*Scope {
+	stageNodes := shuffleBucketStageNodes(ss)
+	if len(ss) <= len(stageNodes) {
+		return ss
+	}
+
+	for _, scope := range ss {
+		if !sameExecutionNode(scope.NodeInfo, mergeNode) &&
+			findPipelineExternalLocalReceiver(scope) != nil {
+			return c.mergeScopesByStageNodes(ss, stageNodes)
+		}
+	}
+	return ss
+}
+
 // shuffleBucketsNeedPerCNGrouping reports whether a dispatch in one top-level
 // bucket tree targets a local receiver owned only by a sibling bucket tree on
 // the same CN. Such a tree is not independently executable by RemoteRun and

pkg/sql/compile/merge_scope_placement_test.go

@@ -0,0 +1,177 @@
+// Copyright 2026 Matrix Origin
+//
+// Licensed 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.
+
+package compile
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/pb/plan"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec/connector"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec/dispatch"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec/merge"
+	plan2 "github.com/matrixorigin/matrixone/pkg/sql/plan"
+	"github.com/matrixorigin/matrixone/pkg/sql/plan/function"
+	"github.com/matrixorigin/matrixone/pkg/vm"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine"
+	"github.com/matrixorigin/matrixone/pkg/vm/process"
+)
+
+func TestNewMergeScopeGroupsRemoteCrossTreeDependenciesByCN(t *testing.T) {
+	nodes := engine.Nodes{
+		{Id: "cn-local", Addr: "cn-local:6001", Mcpu: 2},
+		{Id: "cn-remote", Addr: "cn-remote:6001", Mcpu: 2},
+	}
+
+	tests := []struct {
+		name       string
+		operator   vm.OpType
+		node       engine.Node
+		wantInputs int
+	}{
+		{
+			name:       "remote dispatch dependency",
+			operator:   vm.Dispatch,
+			node:       nodes[1],
+			wantInputs: 1,
+		},
+		{
+			name:       "remote connector dependency",
+			operator:   vm.Connector,
+			node:       nodes[1],
+			wantInputs: 1,
+		},
+		{
+			name:       "local dependency needs no remote container",
+			operator:   vm.Dispatch,
+			node:       nodes[0],
+			wantInputs: 2,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			c := newCompileForShuffleJoinTest(t, nodes)
+			owner := newRemoteMergeInputForTest(c, test.node, 1)
+			producer := newRemoteMergeInputForTest(c, test.node, 0)
+			dependency := &Scope{
+				Magic:    Remote,
+				NodeInfo: scopeNodeWithMcpu(test.node, 1),
+				Proc:     c.proc.NewNoContextChildProc(0),
+			}
+			switch test.operator {
+			case vm.Dispatch:
+				op := dispatch.NewArgument()
+				op.LocalRegs = []*process.WaitRegister{owner.Proc.Reg.MergeReceivers[0]}
+				dependency.setRootOperator(op)
+			case vm.Connector:
+				dependency.setRootOperator(connector.NewArgument().WithReg(owner.Proc.Reg.MergeReceivers[0]))
+			default:
+				t.Fatalf("unsupported dependency operator %s", test.operator)
+			}
+			producer.PreScopes = append(producer.PreScopes, dependency)
+
+			result := c.newMergeScope([]*Scope{producer, owner})
+			require.Len(t, result.PreScopes, test.wantInputs)
+			if test.wantInputs == 1 {
+				cnGroup := result.PreScopes[0]
+				require.Len(t, cnGroup.PreScopes, 2)
+				cnGroup.Proc.Base.TxnOperator = fakeTxnOperator{}
+				require.True(t, checkPipelineStandaloneExecutableAtRemote(cnGroup))
+			}
+		})
+	}
+}
+
+func TestNewMergeScopePreservesIndependentRemoteInputs(t *testing.T) {
+	nodes := engine.Nodes{
+		{Id: "cn-local", Addr: "cn-local:6001", Mcpu: 2},
+		{Id: "cn-remote", Addr: "cn-remote:6001", Mcpu: 2},
+	}
+	c := newCompileForShuffleJoinTest(t, nodes)
+	inputs := []*Scope{
+		newRemoteMergeInputForTest(c, nodes[1], 0),
+		newRemoteMergeInputForTest(c, nodes[1], 0),
+	}
+
+	result := c.newMergeScope(inputs)
+
+	require.Equal(t, inputs, result.PreScopes,
+		"independent remote pipelines should not pay for an extra per-CN merge")
+}
+
+func TestCompileWindowKeepsDistributedShuffleJoinTreesStandalone(t *testing.T) {
+	const dop = int32(2)
+	nodes := engine.Nodes{
+		{Id: "cn-local", Addr: "cn-local:6001", Mcpu: int(dop)},
+		{Id: "cn-remote", Addr: "cn-remote:6001", Mcpu: int(dop)},
+	}
+	c := newCompileForShuffleJoinTest(t, nodes)
+	c.execType = plan2.ExecTypeAP_MULTICN
+
+	joinNode := newShuffleJoinTestNode(dop)
+	joinNode.JoinType = plan.Node_OUTER
+	joinNode.IsRightJoin = true
+	joinNode.Stats.HashmapStats.ShuffleMethod = plan.ShuffleMethod_Normal
+	left := &plan.Node{Stats: &plan.Stats{Dop: dop}}
+	right := &plan.Node{Stats: &plan.Stats{Dop: dop}}
+	probeScopes := []*Scope{
+		newShuffleJoinTestScope(t, nodes[0], 1),
+		newShuffleJoinTestScope(t, nodes[1], 1),
+	}
+	buildScopes := []*Scope{
+		newShuffleJoinTestScope(t, nodes[0], 1),
+		newShuffleJoinTestScope(t, nodes[1], 1),
+	}
+
+	buckets := c.compileShuffleJoin(joinNode, left, right, probeScopes, buildScopes)
+	require.Len(t, buckets, len(nodes)*int(dop))
+
+	windowScopes := c.compileWin(newRowNumberWindowNodeForTest(), buckets)
+	require.Len(t, windowScopes, 1)
+	require.Len(t, windowScopes[0].PreScopes, len(nodes),
+		"a global window must send one standalone shuffle tree per CN")
+	for _, pre := range windowScopes[0].PreScopes {
+		pre.Proc.Base.TxnOperator = fakeTxnOperator{}
+		require.True(t, checkPipelineStandaloneExecutableAtRemote(pre),
+			"window input on %s retains an out-of-tree local receiver", pre.NodeInfo.Addr)
+	}
+}
+
+func newRemoteMergeInputForTest(c *Compile, node engine.Node, receivers int) *Scope {
+	s := &Scope{
+		Magic:    Remote,
+		NodeInfo: scopeNodeWithMcpu(node, 1),
+		Proc:     c.proc.NewNoContextChildProc(receivers),
+	}
+	s.setRootOperator(merge.NewArgument())
+	return s
+}
+
+func newRowNumberWindowNodeForTest() *plan.Node {
+	return &plan.Node{WinSpecList: []*plan.Expr{{
+		Typ: plan.Type{Id: int32(types.T_int64)},
+		Expr: &plan.Expr_W{W: &plan.WindowSpec{
+			WindowFunc: &plan.Expr{Expr: &plan.Expr_F{F: &plan.Function{
+				Func: &plan.ObjectRef{
+					Obj:     function.EncodeOverloadID(function.ROW_NUMBER, 0),
+					ObjName: "row_number",
+				},
+			}}},
+		}},
+	}}}
+}

pkg/sql/compile/remoterunClient.go

@@ -115,6 +115,35 @@ func (s *Scope) remoteRun(c *Compile) (sender *messageSenderOnClient, err error)
 //
 // it returns true if the pipeline has only the root operator capable of sending data to other outer pipeline.
 func checkPipelineStandaloneExecutableAtRemote(s *Scope) bool {
+	offender := findPipelineExternalLocalReceiver(s)
+	if offender == nil {
+		return true
+	}
+
+	switch offender.OpType() {
+	case vm.Dispatch:
+		s.Proc.Infof(
+			s.Proc.Ctx,
+			"txn id : %s, the pipeline %p cannot execute remotely because its dispatch operator targets another local pipeline tree.",
+			s.Proc.GetTxnOperator().Txn().ID, s)
+	case vm.Connector:
+		s.Proc.Infof(
+			s.Proc.Ctx,
+			"txn id : %s, the pipeline %p cannot execute remotely because its connector targets another local pipeline tree.",
+			s.Proc.GetTxnOperator().Txn().ID, s)
+	}
+	return false
+}
+
+// findPipelineExternalLocalReceiver returns the first non-root output operator
+// whose in-process receiver is not owned by the scope tree. The root output is
+// intentionally excluded: RemoteRun retains it on the caller and forwards the
+// remotely executed child tree back through that output.
+func findPipelineExternalLocalReceiver(s *Scope) vm.Operator {
+	if s == nil {
+		return nil
+	}
+
 	var regs = make(map[*process.WaitRegister]struct{})
 	var toScan []*Scope
 	// record which mergeReceivers this scope tree holds.
@@ -123,13 +152,18 @@ func checkPipelineStandaloneExecutableAtRemote(s *Scope) bool {
 		for len(toScan) > 0 {
 			node := toScan[len(toScan)-1]
 			toScan = toScan[:len(toScan)-1]
+			if node == nil {
+				continue
+			}
 
 			if len(node.PreScopes) > 0 {
 				toScan = append(toScan, node.PreScopes...)
 			}
 
-			for i := range node.Proc.Reg.MergeReceivers {
-				regs[node.Proc.Reg.MergeReceivers[i]] = struct{}{}
+			if node.Proc != nil {
+				for i := range node.Proc.Reg.MergeReceivers {
+					regs[node.Proc.Reg.MergeReceivers[i]] = struct{}{}
+				}
 			}
 		}
 	}
@@ -143,41 +177,37 @@ func checkPipelineStandaloneExecutableAtRemote(s *Scope) bool {
 		for len(toScan) > 0 {
 			node := toScan[len(toScan)-1]
 			toScan = toScan[:len(toScan)-1]
+			if node == nil {
+				continue
+			}
 
 			if len(node.PreScopes) > 0 {
 				toScan = append(toScan, node.PreScopes...)
 			}
+			if node.RootOp == nil {
+				continue
+			}
 
 			if node.RootOp.OpType() == vm.Dispatch {
 				t := node.RootOp.(*dispatch.Dispatch)
 				for i := range t.LocalRegs {
 					if _, ok := regs[t.LocalRegs[i]]; !ok {
-						s.Proc.Infof(
-							s.Proc.Ctx,
-							"txn id : %s, the pipeline %p cannot execute remotely because its dispatch operator targets another local pipeline tree.",
-							s.Proc.GetTxnOperator().Txn().ID, s)
-
-						return false
+						return node.RootOp
 					}
 				}
 				continue
 			}
 			if node.RootOp.OpType() == vm.Connector {
 				t := node.RootOp.(*connector.Connector)
 				if _, ok := regs[t.Reg]; !ok {
-					s.Proc.Infof(
-						s.Proc.Ctx,
-						"txn id : %s, the pipeline %p cannot execute remotely because its connector targets another local pipeline tree.",
-						s.Proc.GetTxnOperator().Txn().ID, s)
-
-					return false
+					return node.RootOp
 				}
 				continue
 			}
 		}
 	}
 
-	return true
+	return nil
 }
 
 func prepareRemoteRunSendingData(sqlStr string, s *Scope, proc *process.Process) (scopeData []byte, withoutOutput bool, processData []byte, folded bool, err error) {

test/distributed/cases/optimizer/shuffle.result

@@ -278,6 +278,35 @@ Project  𝄀
 select count(*) from t1 full outer join t2 on t1.c2=t2.c2;
 ➤ count(*)[-5,64,0]  𝄀
 5000000
+with
+l as (
+select c2 as k from t1 where c2 % 100 = 0
+),
+r as (
+select c2 as k from t2 where c2 % 100 = 0
+),
+joined as (
+select case
+when l.k is not null and r.k is not null then 'both'
+when l.k is not null then 'left_only'
+else 'right_only'
+end as join_side,
+coalesce(l.k, r.k) as k
+from l full outer join r on l.k = r.k
+),
+ranked as (
+select join_side,
+row_number() over (partition by join_side order by k) as rn
+from joined
+)
+select join_side, count(*) as cnt
+from ranked
+where rn <= 2
+group by join_side
+order by join_side;
+➤ join_side[12,-1,0]  ¦  cnt[-5,64,0]  𝄀
+both  ¦  2  𝄀
+left_only  ¦  2
 create table t3(c1 int not null, c2 int not null)cluster by c1;
 insert into t3 select *,* from generate_series(1,1000000)g;
 select mo_ctl('dn', 'flush', 'd1.t3');