Repository: matrixorigin/matrixone
Pull request #11627 (2023-09-04 09:12:23)

remove an useless log and add test
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #

## What this PR does / why we need it:


Patch:
diff --git a/pkg/sql/compile/scopeRemoteRunTypes.go b/pkg/sql/compile/scopeRemoteRunTypes.go
index d84389932..61e507911 100644
--- a/pkg/sql/compile/scopeRemoteRunTypes.go
+++ b/pkg/sql/compile/scopeRemoteRunTypes.go
@@ -144,8 +144,8 @@ func (sender *messageSenderOnClient) send(
 func (sender *messageSenderOnClient) receiveMessage() (morpc.Message, error) {
 	select {
 	case <-sender.ctx.Done():
-		logutil.Errorf("sender ctx done during receive")
 		return nil, nil
+
 	case val, ok := <-sender.receiveCh:
 		if !ok || val == nil {
 			// ch close


Test patch:
diff --git a/pkg/sql/compile/scope_test.go b/pkg/sql/compile/scope_test.go
index ae88eac23..61c578e8f 100644
--- a/pkg/sql/compile/scope_test.go
+++ b/pkg/sql/compile/scope_test.go
@@ -17,7 +17,10 @@ package compile
 import (
 	"context"
 	"fmt"
+	"github.com/matrixorigin/matrixone/pkg/common/morpc"
+	"github.com/matrixorigin/matrixone/pkg/pb/pipeline"
 	"testing"
+	"time"
 
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
@@ -95,3 +98,69 @@ func generateScopeCases(t *testing.T, testCases []string) []*Scope {
 	}
 	return result
 }
+
+func TestMessageSenderOnClientReceive(t *testing.T) {
+	sender := new(messageSenderOnClient)
+	sender.receiveCh = make(chan morpc.Message, 1)
+
+	// case 1: use source context, and source context is canceled
+	{
+		sourceCtx, sourceCancel := context.WithCancel(context.Background())
+		sender.ctx = sourceCtx
+		sender.ctxCancel = sourceCancel
+		sourceCancel()
+		v, err := sender.receiveMessage()
+		require.NoError(t, err)
+		require.Equal(t, nil, v)
+	}
+
+	// case 2: use derived context, and source context is canceled
+	{
+		sourceCtx, sourceCancel := context.WithCancel(context.Background())
+		receiveCtx, receiveCancel := context.WithTimeout(sourceCtx, 3*time.Second)
+		sender.ctx = receiveCtx
+		sender.ctxCancel = receiveCancel
+		sourceCancel()
+
+		startTime := time.Now()
+		v, err := sender.receiveMessage()
+		require.NoError(t, err)
+		require.Equal(t, nil, v)
+		require.True(t, time.Since(startTime) < 3*time.Second)
+		receiveCancel()
+	}
+
+	// case 3: receive a nil message
+	{
+		sourceCtx, sourceCancel := context.WithCancel(context.Background())
+		sender.ctx = sourceCtx
+		sender.ctxCancel = sourceCancel
+		sender.receiveCh <- nil
+		_, err := sender.receiveMessage()
+		require.NotNil(t, err)
+		sourceCancel()
+	}
+
+	// case 4: receive a message
+	{
+		sourceCtx, sourceCancel := context.WithCancel(context.Background())
+		sender.ctx = sourceCtx
+		sender.ctxCancel = sourceCancel
+		data := &pipeline.Message{}
+		sender.receiveCh <- data
+		v, err := sender.receiveMessage()
+		require.NoError(t, err)
+		require.Equal(t, data, v)
+		sourceCancel()
+	}
+
+	// case 5: channel is closed
+	{
+		sourceCtx, sourceCancel := context.WithCancel(context.Background())
+		sender.ctx = sourceCtx
+		sender.ctxCancel = sourceCancel
+		close(sender.receiveCh)
+		_, err := sender.receiveMessage()
+		require.NotNil(t, err)
+	}
+}