convert some pipelines to run locally, as they will cause hung due to…

… receiver lost. (#19467)

由于某些pipeline需要往其他pipeline发送数据（例如insert到唯一约束的列，需要发送一份数据到去重pipeline），无法发送到远程单独执行，这会导致找不到dispatch算子的数据接收者而卡死的问题。

对于这些pipeline在记录info级别的log后转换为本地执行。

Approved by: @ouyuanning

pkg/sql/compile/remoterun.go

@@ -511,7 +511,6 @@ func convertToPipelineInstruction(op vm.Operator, proc *process.Process, ctx *sc
 				PipelineId:     ctx0.id,
 			}
 		}
-
 		if len(t.RemoteRegs) > 0 {
 			in.Dispatch.RemoteConnector = make([]*pipeline.WrapNode, len(t.RemoteRegs))
 			for i := range t.RemoteRegs {

pkg/sql/compile/remoterunClient.go

@@ -92,6 +92,76 @@ func (s *Scope) remoteRun(c *Compile) (sender *messageSenderOnClient, err error)
 	return sender, err
 }
 
+// checkPipelineStandaloneExecutableAtRemote is responsible for checking the standalone excitability of the pipeline
+// once it was sent to other remote node.
+//
+// it returns true if the pipeline has only the root operator capable of sending data to other outer pipeline.
+func checkPipelineStandaloneExecutableAtRemote(s *Scope) bool {
+	var regs = make(map[*process.WaitRegister]struct{})
+	var toScan []*Scope
+	// record which mergeReceivers this scope tree holds.
+	{
+		toScan = append(toScan, s)
+		for len(toScan) > 0 {
+			node := toScan[len(toScan)-1]
+			toScan = toScan[:len(toScan)-1]
+
+			if len(node.PreScopes) > 0 {
+				toScan = append(toScan, node.PreScopes...)
+			}
+
+			for i := range node.Proc.Reg.MergeReceivers {
+				regs[node.Proc.Reg.MergeReceivers[i]] = struct{}{}
+			}
+		}
+	}
+
+	// check if there are target channels from other trees.
+	{
+		if len(s.PreScopes) > 0 {
+			toScan = append(toScan, s.PreScopes...)
+		}
+
+		for len(toScan) > 0 {
+			node := toScan[len(toScan)-1]
+			toScan = toScan[:len(toScan)-1]
+
+			if len(node.PreScopes) > 0 {
+				toScan = append(toScan, node.PreScopes...)
+			}
+
+			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 convert to execute locally because it holds a dispatch operator will send data to other local pipeline tree.",
+							s.Proc.GetTxnOperator().Txn().ID, s)
+
+						return false
+					}
+				}
+				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 convert to execute locally because it holds a connector operator will send data to other local pipeline tree.",
+						s.Proc.GetTxnOperator().Txn().ID, s)
+
+					return false
+				}
+				continue
+			}
+		}
+	}
+
+	return true
+}
+
 func prepareRemoteRunSendingData(sqlStr string, s *Scope) (scopeData []byte, withoutOutput bool, processData []byte, err error) {
 	// if simpleRun is true, it indicates that this pipeline will not produce any output.
 	withoutOutput = true

pkg/sql/compile/remoterun_test.go

@@ -502,6 +502,12 @@ type fakeTxnOperator struct {
 	client.TxnOperator
 }
 
+func (f fakeTxnOperator) Txn() txn.TxnMeta {
+	return txn.TxnMeta{
+		ID: []byte("test"),
+	}
+}
+
 func (f fakeTxnOperator) Snapshot() (txn.CNTxnSnapshot, error) {
 	return txn.CNTxnSnapshot{}, nil
 }
@@ -652,3 +658,102 @@ func Test_ReceiveMessageFromCnServer(t *testing.T) {
 		require.NotNil(t, receiveMessageFromCnServer(s4, false, &sender))
 	}
 }
+
+func Test_checkPipelineStandaloneExecutableAtRemote(t *testing.T) {
+	proc := testutil.NewProcess()
+	proc.Base.TxnOperator = fakeTxnOperator{}
+	// a standalone pipeline tree should return true.
+	{
+		// s0, pre: s1, s2
+		s0 := &Scope{
+			Proc:   proc.NewContextChildProc(2),
+			RootOp: dispatch.NewArgument(),
+		}
+
+		s1 := &Scope{
+			Proc: proc.NewContextChildProc(0),
+		}
+		op1 := connector.NewArgument()
+		op1.Reg = s0.Proc.Reg.MergeReceivers[0]
+		s1.RootOp = op1
+
+		s2 := &Scope{
+			Proc: proc.NewContextChildProc(0),
+		}
+		op2 := dispatch.NewArgument()
+		op2.LocalRegs = []*process.WaitRegister{s0.Proc.Reg.MergeReceivers[1]}
+		s2.RootOp = op2
+
+		s0.PreScopes = append(s0.PreScopes, s1, s2)
+
+		require.True(t, checkPipelineStandaloneExecutableAtRemote(s0))
+	}
+
+	// a pipeline holds an invalid dispatch should return false.
+	{
+		// s0, pre: s1
+		s0 := &Scope{
+			Proc:   proc.NewContextChildProc(1),
+			RootOp: dispatch.NewArgument(),
+		}
+
+		s1 := &Scope{
+			Proc: proc.NewContextChildProc(0),
+		}
+		op1 := dispatch.NewArgument()
+		op1.LocalRegs = []*process.WaitRegister{{}}
+		s1.RootOp = op1
+
+		s0.PreScopes = append(s0.PreScopes, s1)
+
+		require.False(t, checkPipelineStandaloneExecutableAtRemote(s0))
+	}
+
+	// a pipeline holds an invalid connector should return false.
+	{
+		// s0, pre: s1
+		s0 := &Scope{
+			Proc:   proc.NewContextChildProc(1),
+			RootOp: dispatch.NewArgument(),
+		}
+
+		s1 := &Scope{
+			Proc: proc.NewContextChildProc(0),
+		}
+		op1 := connector.NewArgument()
+		op1.Reg = &process.WaitRegister{}
+		s1.RootOp = op1
+
+		s0.PreScopes = append(s0.PreScopes, s1)
+
+		require.False(t, checkPipelineStandaloneExecutableAtRemote(s0))
+	}
+
+	// depth more than 2.
+	{
+		// s0, pre: s1, pre: s2.
+		s0 := &Scope{
+			Proc:   proc.NewContextChildProc(1),
+			RootOp: dispatch.NewArgument(),
+		}
+
+		s1 := &Scope{
+			Proc: proc.NewContextChildProc(1),
+		}
+		op1 := connector.NewArgument()
+		op1.Reg = s0.Proc.Reg.MergeReceivers[0]
+		s1.RootOp = op1
+
+		s2 := &Scope{
+			Proc: proc.NewContextChildProc(0),
+		}
+		op2 := connector.NewArgument()
+		op2.Reg = &process.WaitRegister{}
+		s2.RootOp = op2
+
+		s0.PreScopes = append(s0.PreScopes, s1)
+		s1.PreScopes = append(s1.PreScopes, s2)
+
+		require.False(t, checkPipelineStandaloneExecutableAtRemote(s0))
+	}
+}

pkg/sql/compile/scope.go

@@ -341,6 +341,9 @@ func (s *Scope) RemoteRun(c *Compile) error {
 	if !s.canRemote(c, true) {
 		return s.MergeRun(c)
 	}
+	if !checkPipelineStandaloneExecutableAtRemote(s) {
+		return s.MergeRun(c)
+	}
 
 	runtime.ServiceRuntime(s.Proc.GetService()).Logger().
 		Debug("remote run pipeline",