Repository: vitessio/vitess
Pull request #8018 (2021-05-02 15:05:14)

Scatter errors as warning in olap query
<!--
  Thank you for your contribution to the Vitess project.
  How to contribute: https://vitess.io/docs/contributing/
  Please first make sure there is an open Issue to discuss the feature/fix suggested in this PR.
  If this is a new feature, please mark the Issue as "RFC".
 -->

<!-- if this PR is Work in Progress please create it as a Draft Pull Request -->

## Description
<!-- A few sentences describing the overall goals of the pull request's commits. -->
<!-- If this is a bug fix and you think the fix should be backported, please write so. -->

## Related Issue(s)
<!-- List related issues and pull requests. If this PR fixes an issue, please add it using Fixes #????  -->

Closes https://github.com/vitessio/vitess/issues/7919

## Checklist
- [X] Tests were added or are not required
- [X] Documentation was added or is not required

## Deployment Notes
<!-- Notes regarding deployment of the contained body of work. These should note any db migrations, etc. -->


Patch:
diff --git a/go/vt/vtgate/engine/route.go b/go/vt/vtgate/engine/route.go
index aecc6094c5..7c66da87c2 100644
--- a/go/vt/vtgate/engine/route.go
+++ b/go/vt/vtgate/engine/route.go
@@ -347,12 +347,25 @@ func (route *Route) StreamExecute(vcursor VCursor, bindVars map[string]*querypb.
 	}
 
 	if len(route.OrderBy) == 0 {
-		return vcursor.StreamExecuteMulti(route.Query, rss, bvs, func(qr *sqltypes.Result) error {
+		err = vcursor.StreamExecuteMulti(route.Query, rss, bvs, func(qr *sqltypes.Result) error {
 			return callback(qr.Truncate(route.TruncateColumnCount))
 		})
+		if err != nil {
+			if !route.ScatterErrorsAsWarnings {
+				return err
+			}
+			partialSuccessScatterQueries.Add(1)
+			sErr := mysql.NewSQLErrorFromError(err).(*mysql.SQLError)
+			vcursor.Session().RecordWarning(&querypb.QueryWarning{Code: uint32(sErr.Num), Message: err.Error()})
+		}
+		return nil
 	}
 
 	// There is an order by. We have to merge-sort.
+	return route.mergeSort(vcursor, bindVars, wantfields, callback, rss, bvs)
+}
+
+func (route *Route) mergeSort(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error, rss []*srvtopo.ResolvedShard, bvs []map[string]*querypb.BindVariable) error {
 	prims := make([]StreamExecutor, 0, len(rss))
 	for i, rs := range rss {
 		prims = append(prims, &shardRoute{
@@ -365,9 +378,18 @@ func (route *Route) StreamExecute(vcursor VCursor, bindVars map[string]*querypb.
 		Primitives: prims,
 		OrderBy:    route.OrderBy,
 	}
-	return ms.StreamExecute(vcursor, bindVars, wantfields, func(qr *sqltypes.Result) error {
+	err := ms.StreamExecute(vcursor, bindVars, wantfields, func(qr *sqltypes.Result) error {
 		return callback(qr.Truncate(route.TruncateColumnCount))
 	})
+	if err != nil {
+		if !route.ScatterErrorsAsWarnings {
+			return err
+		}
+		partialSuccessScatterQueries.Add(1)
+		sErr := mysql.NewSQLErrorFromError(err).(*mysql.SQLError)
+		vcursor.Session().RecordWarning(&querypb.QueryWarning{Code: uint32(sErr.Num), Message: err.Error()})
+	}
+	return nil
 }
 
 // GetFields fetches the field info.


Test patch:
diff --git a/go/vt/vtgate/engine/fake_vcursor_test.go b/go/vt/vtgate/engine/fake_vcursor_test.go
index a85dcd78cd..d5a2cbc7d8 100644
--- a/go/vt/vtgate/engine/fake_vcursor_test.go
+++ b/go/vt/vtgate/engine/fake_vcursor_test.go
@@ -23,6 +23,7 @@ import (
 	"reflect"
 	"sort"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -273,6 +274,7 @@ type loggingVCursor struct {
 	multiShardErrs []error
 
 	log []string
+	mu  sync.Mutex
 
 	resolvedTargetTabletType topodatapb.TabletType
 
@@ -400,8 +402,10 @@ func (f *loggingVCursor) ExecuteStandalone(query string, bindvars map[string]*qu
 }
 
 func (f *loggingVCursor) StreamExecuteMulti(query string, rss []*srvtopo.ResolvedShard, bindVars []map[string]*querypb.BindVariable, callback func(reply *sqltypes.Result) error) error {
+	f.mu.Lock()
 	f.log = append(f.log, fmt.Sprintf("StreamExecuteMulti %s %s", query, printResolvedShardsBindVars(rss, bindVars)))
 	r, err := f.nextResult()
+	f.mu.Unlock()
 	if err != nil {
 		return err
 	}
diff --git a/go/vt/vtgate/engine/route_test.go b/go/vt/vtgate/engine/route_test.go
index 0b7773dc74..2bb18aee82 100644
--- a/go/vt/vtgate/engine/route_test.go
+++ b/go/vt/vtgate/engine/route_test.go
@@ -1184,7 +1184,6 @@ func TestExecFail(t *testing.T) {
 		`ExecuteMultiShard ks.-20: dummy_select {} ks.20-: dummy_select {} false false`,
 	})
 
-	vc.Rewind()
 	// Scatter succeeds if all shards fail with ScatterErrorsAsWarnings
 	sel = NewRoute(
 		SelectScatter,
@@ -1220,6 +1219,11 @@ func TestExecFail(t *testing.T) {
 	})
 
 	vc.Rewind()
+	vc.results = nil
+	vc.resultErr = mysql.NewSQLError(mysql.ERQueryInterrupted, "", "query timeout -20")
+	_, err = wrapStreamExecute(sel, vc, map[string]*querypb.BindVariable{}, false)
+	require.NoError(t, err, "unexpected ScatterErrorsAsWarnings error %v", err)
+	vc.ExpectWarnings(t, []*querypb.QueryWarning{{Code: mysql.ERQueryInterrupted, Message: "query timeout -20 (errno 1317) (sqlstate HY000)"}})
 
 	// Scatter succeeds if one of N fails with ScatterErrorsAsWarnings
 	sel = NewRoute(
@@ -1250,4 +1254,13 @@ func TestExecFail(t *testing.T) {
 	expectResult(t, "sel.Execute", result, defaultSelectResult)
 
 	vc.Rewind()
+	vc.resultErr = mysql.NewSQLError(mysql.ERQueryInterrupted, "", "query timeout -20")
+	// test when there is order by column
+	sel.OrderBy = []OrderbyParams{{
+		WeightStringCol: -1,
+		Col:             0,
+	}}
+	_, err = wrapStreamExecute(sel, vc, map[string]*querypb.BindVariable{}, false)
+	require.NoError(t, err, "unexpected ScatterErrorsAsWarnings error %v", err)
+	vc.ExpectWarnings(t, []*querypb.QueryWarning{{Code: mysql.ERQueryInterrupted, Message: "query timeout -20 (errno 1317) (sqlstate HY000)"}})
 }