diff --git a/data/test/vtgate/select_cases.txt b/data/test/vtgate/select_cases.txt index cfc528b251e..92da4638a33 100644 --- a/data/test/vtgate/select_cases.txt +++ b/data/test/vtgate/select_cases.txt @@ -105,6 +105,68 @@ } } +# select with partial scatter directive +"select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ * from user" +{ + "Original": "select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ * from user", + "Instructions": { + "Opcode": "SelectScatter", + "Keyspace": { + "Name": "user", + "Sharded": true + }, + "Query": "select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ * from user", + "FieldQuery": "select * from user where 1 != 1", + "ScatterErrorsAsWarnings": true + } +} + +# select aggregation with partial scatter directive +"select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ count(*) from user" +{ + "Original": "select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ count(*) from user", + "Instructions": { + "Aggregates": [ + { + "Opcode": "count", + "Col": 0 + } + ], + "Keys": null, + "Input": { + "Opcode": "SelectScatter", + "Keyspace": { + "Name": "user", + "Sharded": true + }, + "Query": "select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ count(*) from user", + "FieldQuery": "select count(*) from user where 1 != 1", + "ScatterErrorsAsWarnings": true + } + } +} + +# select limit with partial scatter directive +"select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ * from user limit 10" +{ + "Original": "select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ * from user limit 10", + "Instructions": { + "Opcode": "Limit", + "Count": 10, + "Offset": null, + "Input": { + "Opcode": "SelectScatter", + "Keyspace": { + "Name": "user", + "Sharded": true + }, + "Query": "select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ * from user limit :__upper_limit", + "FieldQuery": "select * from user where 1 != 1", + "ScatterErrorsAsWarnings": true + } + } +} + # qualified '*' expression for simple route "select user.* from user" { diff --git a/go/vt/concurrency/error_recorder.go b/go/vt/concurrency/error_recorder.go index 6087a55158d..f0a2015cd32 100644 --- a/go/vt/concurrency/error_recorder.go +++ b/go/vt/concurrency/error_recorder.go @@ -135,3 +135,13 @@ func (aer *AllErrorRecorder) ErrorStrings() []string { } return errs } + +// GetErrors returns a reference to the internal errors array. +// +// Note that the array is not copied, so this should only be used +// once the recording is complete. +func (aer *AllErrorRecorder) GetErrors() []error { + aer.mu.Lock() + defer aer.mu.Unlock() + return aer.Errors +} diff --git a/go/vt/sqlparser/comments.go b/go/vt/sqlparser/comments.go index 94bfb7aaadf..06f6307da5c 100644 --- a/go/vt/sqlparser/comments.go +++ b/go/vt/sqlparser/comments.go @@ -30,6 +30,8 @@ const ( DirectiveSkipQueryPlanCache = "SKIP_QUERY_PLAN_CACHE" // DirectiveQueryTimeout sets a query timeout in vtgate. Only supported for SELECTS. DirectiveQueryTimeout = "QUERY_TIMEOUT_MS" + // DirectiveScatterErrorsAsWarnings enables partial success scatter select queries + DirectiveScatterErrorsAsWarnings = "SCATTER_ERRORS_AS_WARNINGS" ) func isNonSpace(r rune) bool { diff --git a/go/vt/vtgate/engine/delete.go b/go/vt/vtgate/engine/delete.go index 73989609676..4930d0933fc 100644 --- a/go/vt/vtgate/engine/delete.go +++ b/go/vt/vtgate/engine/delete.go @@ -238,5 +238,6 @@ func (del *Delete) execDeleteByDestination(vcursor VCursor, bindVars map[string] } } autocommit := (len(rss) == 1 || del.MultiShardAutocommit) && vcursor.AutocommitApproval() - return vcursor.ExecuteMultiShard(rss, queries, true /* isDML */, autocommit) + res, errs := vcursor.ExecuteMultiShard(rss, queries, true /* isDML */, autocommit) + return res, vterrors.Aggregate(errs) } diff --git a/go/vt/vtgate/engine/fake_vcursor_test.go b/go/vt/vtgate/engine/fake_vcursor_test.go index 17d0699b9d5..f586d664a0c 100644 --- a/go/vt/vtgate/engine/fake_vcursor_test.go +++ b/go/vt/vtgate/engine/fake_vcursor_test.go @@ -54,7 +54,7 @@ func (t noopVCursor) ExecuteAutocommit(method string, query string, bindvars map panic("unimplemented") } -func (t noopVCursor) ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, isDML, autocommit bool) (*sqltypes.Result, error) { +func (t noopVCursor) ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, isDML, autocommit bool) (*sqltypes.Result, []error) { panic("unimplemented") } @@ -88,6 +88,10 @@ type loggingVCursor struct { curResult int resultErr error + // Optional errors that can be returned from nextResult() alongside the results for + // multi-shard queries + multiShardErrs []error + log []string } @@ -103,9 +107,14 @@ func (f *loggingVCursor) Execute(method string, query string, bindvars map[strin return f.nextResult() } -func (f *loggingVCursor) ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, isDML, canAutocommit bool) (*sqltypes.Result, error) { +func (f *loggingVCursor) ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, isDML, canAutocommit bool) (*sqltypes.Result, []error) { f.log = append(f.log, fmt.Sprintf("ExecuteMultiShard %v%v %v", printResolvedShardQueries(rss, queries), isDML, canAutocommit)) - return f.nextResult() + res, err := f.nextResult() + if err != nil { + return nil, []error{err} + } + + return res, f.multiShardErrs } func (f *loggingVCursor) AutocommitApproval() bool { diff --git a/go/vt/vtgate/engine/insert.go b/go/vt/vtgate/engine/insert.go index 2c169d7e3c7..72d5e20da56 100644 --- a/go/vt/vtgate/engine/insert.go +++ b/go/vt/vtgate/engine/insert.go @@ -220,9 +220,9 @@ func (ins *Insert) execInsertSharded(vcursor VCursor, bindVars map[string]*query } autocommit := (len(rss) == 1 || ins.MultiShardAutocommit) && vcursor.AutocommitApproval() - result, err := vcursor.ExecuteMultiShard(rss, queries, true /* isDML */, autocommit) - if err != nil { - return nil, vterrors.Wrap(err, "execInsertSharded") + result, errs := vcursor.ExecuteMultiShard(rss, queries, true /* isDML */, autocommit) + if errs != nil { + return nil, vterrors.Wrap(vterrors.Aggregate(errs), "execInsertSharded") } if insertID != 0 { diff --git a/go/vt/vtgate/engine/primitive.go b/go/vt/vtgate/engine/primitive.go index a29fb7443d1..a5c079d65a8 100644 --- a/go/vt/vtgate/engine/primitive.go +++ b/go/vt/vtgate/engine/primitive.go @@ -52,7 +52,7 @@ type VCursor interface { AutocommitApproval() bool // Shard-level functions. - ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, isDML, canAutocommit bool) (*sqltypes.Result, error) + ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, isDML, canAutocommit bool) (*sqltypes.Result, []error) ExecuteStandalone(query string, bindvars map[string]*querypb.BindVariable, rs *srvtopo.ResolvedShard) (*sqltypes.Result, error) StreamExecuteMulti(query string, rss []*srvtopo.ResolvedShard, bindVars []map[string]*querypb.BindVariable, callback func(reply *sqltypes.Result) error) error diff --git a/go/vt/vtgate/engine/route.go b/go/vt/vtgate/engine/route.go index 0e9521cd4ee..3efb775c36e 100644 --- a/go/vt/vtgate/engine/route.go +++ b/go/vt/vtgate/engine/route.go @@ -24,6 +24,7 @@ import ( "vitess.io/vitess/go/jsonutil" "vitess.io/vitess/go/sqltypes" + "vitess.io/vitess/go/stats" "vitess.io/vitess/go/vt/key" "vitess.io/vitess/go/vt/srvtopo" "vitess.io/vitess/go/vt/vterrors" @@ -71,6 +72,9 @@ type Route struct { // QueryTimeout contains the optional timeout (in milliseconds) to apply to this query QueryTimeout int + + // ScatterErrorsAsWarnings is true if results should be returned even if some shards have an error + ScatterErrorsAsWarnings bool } // OrderbyParams specifies the parameters for ordering. @@ -88,25 +92,27 @@ func (route *Route) MarshalJSON() ([]byte, error) { vindexName = route.Vindex.String() } marshalRoute := struct { - Opcode RouteOpcode - Keyspace *vindexes.Keyspace `json:",omitempty"` - Query string `json:",omitempty"` - FieldQuery string `json:",omitempty"` - Vindex string `json:",omitempty"` - Values []sqltypes.PlanValue `json:",omitempty"` - OrderBy []OrderbyParams `json:",omitempty"` - TruncateColumnCount int `json:",omitempty"` - QueryTimeout int `json:",omitempty"` + Opcode RouteOpcode + Keyspace *vindexes.Keyspace `json:",omitempty"` + Query string `json:",omitempty"` + FieldQuery string `json:",omitempty"` + Vindex string `json:",omitempty"` + Values []sqltypes.PlanValue `json:",omitempty"` + OrderBy []OrderbyParams `json:",omitempty"` + TruncateColumnCount int `json:",omitempty"` + QueryTimeout int `json:",omitempty"` + ScatterErrorsAsWarnings bool `json:",omitempty"` }{ - Opcode: route.Opcode, - Keyspace: route.Keyspace, - Query: route.Query, - FieldQuery: route.FieldQuery, - Vindex: vindexName, - Values: route.Values, - OrderBy: route.OrderBy, - TruncateColumnCount: route.TruncateColumnCount, - QueryTimeout: route.QueryTimeout, + Opcode: route.Opcode, + Keyspace: route.Keyspace, + Query: route.Query, + FieldQuery: route.FieldQuery, + Vindex: vindexName, + Values: route.Values, + OrderBy: route.OrderBy, + TruncateColumnCount: route.TruncateColumnCount, + QueryTimeout: route.QueryTimeout, + ScatterErrorsAsWarnings: route.ScatterErrorsAsWarnings, } return jsonutil.MarshalNoEscape(marshalRoute) } @@ -152,6 +158,10 @@ var routeName = map[RouteOpcode]string{ SelectDBA: "SelectDBA", } +var ( + partialSuccessScatterQueries = stats.NewCounter("PartialSuccessScatterQueries", "Count of partially successful scatter queries") +) + // MarshalJSON serializes the RouteOpcode as a JSON string. // It's used for testing and diagnostics. func (code RouteOpcode) MarshalJSON() ([]byte, error) { @@ -209,9 +219,15 @@ func (route *Route) execute(vcursor VCursor, bindVars map[string]*querypb.BindVa } queries := getQueries(route.Query, bvs) - result, err := vcursor.ExecuteMultiShard(rss, queries, false /* isDML */, false /* autocommit */) - if err != nil { - return nil, err + result, errs := vcursor.ExecuteMultiShard(rss, queries, false /* isDML */, false /* autocommit */) + + if errs != nil { + if route.ScatterErrorsAsWarnings { + partialSuccessScatterQueries.Add(1) + // fall through + } else { + return nil, vterrors.Aggregate(errs) + } } if len(route.OrderBy) == 0 { return result, nil @@ -421,12 +437,13 @@ func execAnyShard(vcursor VCursor, query string, bindVars map[string]*querypb.Bi func execShard(vcursor VCursor, query string, bindVars map[string]*querypb.BindVariable, rs *srvtopo.ResolvedShard, isDML, canAutocommit bool) (*sqltypes.Result, error) { autocommit := canAutocommit && vcursor.AutocommitApproval() - return vcursor.ExecuteMultiShard([]*srvtopo.ResolvedShard{rs}, []*querypb.BoundQuery{ + result, errs := vcursor.ExecuteMultiShard([]*srvtopo.ResolvedShard{rs}, []*querypb.BoundQuery{ { Sql: query, BindVariables: bindVars, }, }, isDML, autocommit) + return result, vterrors.Aggregate(errs) } func getQueries(query string, bvs []map[string]*querypb.BindVariable) []*querypb.BoundQuery { diff --git a/go/vt/vtgate/engine/route_test.go b/go/vt/vtgate/engine/route_test.go index ad77cdb9f2d..9acb3dc8b04 100644 --- a/go/vt/vtgate/engine/route_test.go +++ b/go/vt/vtgate/engine/route_test.go @@ -782,6 +782,7 @@ func TestParamsFail(t *testing.T) { } func TestExecFail(t *testing.T) { + // Unsharded error sel := &Route{ Opcode: SelectUnsharded, Keyspace: &vindexes.Keyspace{ @@ -799,4 +800,94 @@ func TestExecFail(t *testing.T) { vc.Rewind() _, err = wrapStreamExecute(sel, vc, map[string]*querypb.BindVariable{}, false) expectError(t, "sel.StreamExecute err", err, "result error") + + // Scatter fails if one of N fails without ScatterErrorsAsWarnings + sel = &Route{ + Opcode: SelectScatter, + Keyspace: &vindexes.Keyspace{ + Name: "ks", + Sharded: true, + }, + Query: "dummy_select", + FieldQuery: "dummy_select_field", + } + + vc = &loggingVCursor{ + shards: []string{"-20", "20-"}, + results: []*sqltypes.Result{defaultSelectResult}, + multiShardErrs: []error{ + errors.New("result error -20"), + }, + } + _, err = sel.Execute(vc, map[string]*querypb.BindVariable{}, false) + expectError(t, "sel.Execute err", err, "result error -20") + vc.ExpectLog(t, []string{ + `ResolveDestinations ks [] Destinations:DestinationAllShards()`, + `ExecuteMultiShard ks.-20: dummy_select {} ks.20-: dummy_select {} false false`, + }) + + vc.Rewind() + + // Scatter succeeds if all shards fail with ScatterErrorsAsWarnings + sel = &Route{ + Opcode: SelectScatter, + Keyspace: &vindexes.Keyspace{ + Name: "ks", + Sharded: true, + }, + Query: "dummy_select", + FieldQuery: "dummy_select_field", + ScatterErrorsAsWarnings: true, + } + + vc = &loggingVCursor{ + shards: []string{"-20", "20-"}, + results: []*sqltypes.Result{defaultSelectResult}, + multiShardErrs: []error{ + errors.New("result error -20"), + errors.New("result error 20-"), + }, + } + _, err = sel.Execute(vc, map[string]*querypb.BindVariable{}, false) + if err != nil { + t.Errorf("unexpected ScatterErrorsAsWarnings error %v", err) + } + vc.ExpectLog(t, []string{ + `ResolveDestinations ks [] Destinations:DestinationAllShards()`, + `ExecuteMultiShard ks.-20: dummy_select {} ks.20-: dummy_select {} false false`, + }) + + vc.Rewind() + + // Scatter succeeds if one of N fails with ScatterErrorsAsWarnings + sel = &Route{ + Opcode: SelectScatter, + Keyspace: &vindexes.Keyspace{ + Name: "ks", + Sharded: true, + }, + Query: "dummy_select", + FieldQuery: "dummy_select_field", + ScatterErrorsAsWarnings: true, + } + + vc = &loggingVCursor{ + shards: []string{"-20", "20-"}, + results: []*sqltypes.Result{defaultSelectResult}, + multiShardErrs: []error{ + errors.New("result error -20"), + nil, + }, + } + result, err := sel.Execute(vc, map[string]*querypb.BindVariable{}, false) + if err != nil { + t.Errorf("unexpected ScatterErrorsAsWarnings error %v", err) + } + vc.ExpectLog(t, []string{ + `ResolveDestinations ks [] Destinations:DestinationAllShards()`, + `ExecuteMultiShard ks.-20: dummy_select {} ks.20-: dummy_select {} false false`, + }) + expectResult(t, "sel.Execute", result, defaultSelectResult) + + vc.Rewind() } diff --git a/go/vt/vtgate/engine/update.go b/go/vt/vtgate/engine/update.go index fb90c01daa4..6f013541b3a 100644 --- a/go/vt/vtgate/engine/update.go +++ b/go/vt/vtgate/engine/update.go @@ -251,5 +251,6 @@ func (upd *Update) execUpdateByDestination(vcursor VCursor, bindVars map[string] } } autocommit := (len(rss) == 1 || upd.MultiShardAutocommit) && vcursor.AutocommitApproval() - return vcursor.ExecuteMultiShard(rss, queries, true /* isDML */, autocommit) + result, errs := vcursor.ExecuteMultiShard(rss, queries, true /* isDML */, autocommit) + return result, vterrors.Aggregate(errs) } diff --git a/go/vt/vtgate/executor_select_test.go b/go/vt/vtgate/executor_select_test.go index c3c9dc1d199..0499b0d60b6 100644 --- a/go/vt/vtgate/executor_select_test.go +++ b/go/vt/vtgate/executor_select_test.go @@ -31,6 +31,7 @@ import ( querypb "vitess.io/vitess/go/vt/proto/query" topodatapb "vitess.io/vitess/go/vt/proto/topodata" vtgatepb "vitess.io/vitess/go/vt/proto/vtgate" + vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc" ) func TestSelectNext(t *testing.T) { @@ -782,6 +783,67 @@ func TestSelectScatter(t *testing.T) { testQueryLog(t, logChan, "TestExecute", "SELECT", wantQueries[0].Sql, 8) } +func TestSelectScatterPartial(t *testing.T) { + // Special setup: Don't use createExecutorEnv. + cell := "aa" + hc := discovery.NewFakeHealthCheck() + s := createSandbox("TestExecutor") + s.VSchema = executorVSchema + getSandbox(KsTestUnsharded).VSchema = unshardedVSchema + serv := new(sandboxTopo) + resolver := newTestResolver(hc, serv, cell) + shards := []string{"-20", "20-40", "40-60", "60-80", "80-a0", "a0-c0", "c0-e0", "e0-"} + var conns []*sandboxconn.SandboxConn + for _, shard := range shards { + sbc := hc.AddTestTablet(cell, shard, 1, "TestExecutor", shard, topodatapb.TabletType_MASTER, true, 1, nil) + conns = append(conns, sbc) + } + + executor := NewExecutor(context.Background(), serv, cell, "", resolver, false, testBufferSize, testCacheSize, false) + logChan := QueryLogger.Subscribe("Test") + defer QueryLogger.Unsubscribe(logChan) + + // Fail 1 of N without the directive fails the whole operation + conns[2].MustFailCodes[vtrpcpb.Code_RESOURCE_EXHAUSTED] = 1000 + results, err := executorExec(executor, "select id from user", nil) + wantErr := "target: TestExecutor.40-60.master, used tablet: aa-0 (40-60), RESOURCE_EXHAUSTED error" + if err == nil || err.Error() != wantErr { + t.Errorf("want error %v, got %v", wantErr, err) + } + if results != nil { + t.Errorf("want nil results, got %v", results) + } + testQueryLog(t, logChan, "TestExecute", "SELECT", "select id from user", 8) + + // Fail 1 of N with the directive succeeds with 7 rows + results, err = executorExec(executor, "select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ id from user", nil) + if err != nil { + t.Error(err) + } + if results == nil || len(results.Rows) != 7 { + t.Errorf("want 7 results, got %v", results) + } + testQueryLog(t, logChan, "TestExecute", "SELECT", "select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ id from user", 8) + + // Even if all shards fail the operation succeeds with 0 rows + conns[0].MustFailCodes[vtrpcpb.Code_RESOURCE_EXHAUSTED] = 1000 + conns[1].MustFailCodes[vtrpcpb.Code_RESOURCE_EXHAUSTED] = 1000 + conns[3].MustFailCodes[vtrpcpb.Code_RESOURCE_EXHAUSTED] = 1000 + conns[4].MustFailCodes[vtrpcpb.Code_RESOURCE_EXHAUSTED] = 1000 + conns[5].MustFailCodes[vtrpcpb.Code_RESOURCE_EXHAUSTED] = 1000 + conns[6].MustFailCodes[vtrpcpb.Code_RESOURCE_EXHAUSTED] = 1000 + conns[7].MustFailCodes[vtrpcpb.Code_RESOURCE_EXHAUSTED] = 1000 + + results, err = executorExec(executor, "select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ id from user", nil) + if err != nil { + t.Error(err) + } + if results == nil || len(results.Rows) != 0 { + t.Errorf("want 0 result rows, got %v", results) + } + testQueryLog(t, logChan, "TestExecute", "SELECT", "select /*vt+ SCATTER_ERRORS_AS_WARNINGS=1 */ id from user", 8) +} + func TestStreamSelectScatter(t *testing.T) { // Special setup: Don't use createExecutorEnv. cell := "aa" diff --git a/go/vt/vtgate/gateway/discoverygateway.go b/go/vt/vtgate/gateway/discoverygateway.go index b8950a22e9d..6d5d4ed7522 100644 --- a/go/vt/vtgate/gateway/discoverygateway.go +++ b/go/vt/vtgate/gateway/discoverygateway.go @@ -235,7 +235,7 @@ func (dg *discoveryGateway) CacheStatus() TabletCacheStatusList { // the middle of a transaction. While returning the error check if it maybe a result of // a resharding event, and set the re-resolve bit and let the upper layers // re-resolve and retry. -func (dg *discoveryGateway) withRetry(ctx context.Context, target *querypb.Target, conn queryservice.QueryService, name string, inTransaction bool, inner func(ctx context.Context, target *querypb.Target, conn queryservice.QueryService) (error, bool)) error { +func (dg *discoveryGateway) withRetry(ctx context.Context, target *querypb.Target, unused queryservice.QueryService, name string, inTransaction bool, inner func(ctx context.Context, target *querypb.Target, conn queryservice.QueryService) (error, bool)) error { var tabletLastUsed *topodatapb.Tablet var err error invalidTablets := make(map[string]bool) diff --git a/go/vt/vtgate/planbuilder/select.go b/go/vt/vtgate/planbuilder/select.go index 74c5ed8a6e6..19a3401f931 100644 --- a/go/vt/vtgate/planbuilder/select.go +++ b/go/vt/vtgate/planbuilder/select.go @@ -82,7 +82,12 @@ func (pb *primitiveBuilder) processSelect(sel *sqlparser.Select, outer *symtab) if rb.ERoute.TargetDestination != nil { return errors.New("unsupported: SELECT with a target destination") } + + if directives.IsSet(sqlparser.DirectiveScatterErrorsAsWarnings) { + rb.ERoute.ScatterErrorsAsWarnings = true + } } + // Set the outer symtab after processing of FROM clause. // This is because correlation is not allowed there. pb.st.Outer = outer diff --git a/go/vt/vtgate/scatter_conn.go b/go/vt/vtgate/scatter_conn.go index 896f2062528..32283678d33 100644 --- a/go/vt/vtgate/scatter_conn.go +++ b/go/vt/vtgate/scatter_conn.go @@ -132,7 +132,7 @@ func (stc *ScatterConn) Execute( var mu sync.Mutex qr := new(sqltypes.Result) - err := stc.multiGoTransaction( + allErrors := stc.multiGoTransaction( ctx, "Execute", rss, @@ -160,11 +160,16 @@ func (stc *ScatterConn) Execute( qr.AppendResult(innerqr) return transactionID, nil }) - return qr, err + + return qr, allErrors.AggrError(vterrors.Aggregate) } // ExecuteMultiShard is like Execute, // but each shard gets its own Sql Queries and BindVariables. +// +// It always returns a non-nil query result and an array of +// shard errors which may be nil so that callers can optionally +// process a partially-successful operation. func (stc *ScatterConn) ExecuteMultiShard( ctx context.Context, rss []*srvtopo.ResolvedShard, @@ -173,13 +178,13 @@ func (stc *ScatterConn) ExecuteMultiShard( session *SafeSession, notInTransaction bool, autocommit bool, -) (*sqltypes.Result, error) { +) (qr *sqltypes.Result, errs []error) { // mu protects qr var mu sync.Mutex - qr := new(sqltypes.Result) + qr = new(sqltypes.Result) - err := stc.multiGoTransaction( + allErrors := stc.multiGoTransaction( ctx, "Execute", rss, @@ -213,7 +218,8 @@ func (stc *ScatterConn) ExecuteMultiShard( qr.AppendResult(innerqr) return transactionID, nil }) - return qr, err + + return qr, allErrors.GetErrors() } func (stc *ScatterConn) executeAutocommit(ctx context.Context, rs *srvtopo.ResolvedShard, sql string, bindVariables map[string]*querypb.BindVariable, options *querypb.ExecuteOptions) (*sqltypes.Result, error) { @@ -246,7 +252,7 @@ func (stc *ScatterConn) ExecuteEntityIds( var mu sync.Mutex qr := new(sqltypes.Result) - err := stc.multiGoTransaction( + allErrors := stc.multiGoTransaction( ctx, "ExecuteEntityIds", rss, @@ -271,7 +277,8 @@ func (stc *ScatterConn) ExecuteEntityIds( qr.AppendResult(innerqr) return transactionID, nil }) - return qr, err + + return qr, allErrors.AggrError(vterrors.Aggregate) } // scatterBatchRequest needs to be built to perform a scatter batch query. @@ -359,6 +366,7 @@ func (stc *ScatterConn) ExecuteBatch( asTransaction bool, session *SafeSession, options *querypb.ExecuteOptions) (qrs []sqltypes.Result, err error) { + allErrors := new(concurrency.AllErrorRecorder) results := make([]sqltypes.Result, batchRequest.length) @@ -759,6 +767,10 @@ func (stc *ScatterConn) multiGo( // and updates the Session with the transaction id. If the session already // contains a transaction id for the shard, it reuses it. // The action function must match the shardActionTransactionFunc signature. +// +// It returns an error recorder in which each shard error is recorded positionally, +// i.e. if rss[2] had an error, then the error recorder will store that error +// in the second position. func (stc *ScatterConn) multiGoTransaction( ctx context.Context, name string, @@ -767,12 +779,14 @@ func (stc *ScatterConn) multiGoTransaction( session *SafeSession, notInTransaction bool, action shardActionTransactionFunc, -) error { - if len(rss) == 0 { - return nil - } +) (allErrors *concurrency.AllErrorRecorder) { - allErrors := new(concurrency.AllErrorRecorder) + numShards := len(rss) + allErrors = new(concurrency.AllErrorRecorder) + + if numShards == 0 { + return allErrors + } oneShard := func(rs *srvtopo.ResolvedShard, i int) { var err error startTime, statsKey := stc.startAction(name, rs.Target) @@ -791,7 +805,7 @@ func (stc *ScatterConn) multiGoTransaction( } var wg sync.WaitGroup - if len(rss) == 1 { + if numShards == 1 { // only one shard, do it synchronously. for i, rs := range rss { oneShard(rs, i) @@ -812,10 +826,7 @@ end: if session.MustRollback() { stc.txConn.Rollback(ctx, session) } - if allErrors.HasErrors() { - return allErrors.AggrError(vterrors.Aggregate) - } - return nil + return allErrors } // transactionInfo looks at the current session, and returns: diff --git a/go/vt/vtgate/scatter_conn_test.go b/go/vt/vtgate/scatter_conn_test.go index 91f4931504f..50d52060ce6 100644 --- a/go/vt/vtgate/scatter_conn_test.go +++ b/go/vt/vtgate/scatter_conn_test.go @@ -68,7 +68,8 @@ func TestScatterConnExecuteMulti(t *testing.T) { } } - return sc.ExecuteMultiShard(context.Background(), rss, queries, topodatapb.TabletType_REPLICA, nil, false, false) + qr, errs := sc.ExecuteMultiShard(context.Background(), rss, queries, topodatapb.TabletType_REPLICA, nil, false, false) + return qr, vterrors.Aggregate(errs) }) } diff --git a/go/vt/vtgate/vcursor_impl.go b/go/vt/vtgate/vcursor_impl.go index 60b7328231d..28d0b5703bb 100644 --- a/go/vt/vtgate/vcursor_impl.go +++ b/go/vt/vtgate/vcursor_impl.go @@ -152,13 +152,14 @@ func (vc *vcursorImpl) ExecuteAutocommit(method string, query string, BindVars m } // ExecuteMultiShard is part of the engine.VCursor interface. -func (vc *vcursorImpl) ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, isDML, autocommit bool) (*sqltypes.Result, error) { +func (vc *vcursorImpl) ExecuteMultiShard(rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, isDML, autocommit bool) (*sqltypes.Result, []error) { atomic.AddUint32(&vc.logStats.ShardQueries, uint32(len(queries))) - qr, err := vc.executor.scatterConn.ExecuteMultiShard(vc.ctx, rss, commentedShardQueries(queries, vc.marginComments), vc.tabletType, vc.safeSession, false, autocommit) - if err == nil { + qr, errs := vc.executor.scatterConn.ExecuteMultiShard(vc.ctx, rss, commentedShardQueries(queries, vc.marginComments), vc.tabletType, vc.safeSession, false, autocommit) + + if errs == nil { vc.hasPartialDML = true } - return qr, err + return qr, errs } // AutocommitApproval is part of the engine.VCursor interface. @@ -177,7 +178,8 @@ func (vc *vcursorImpl) ExecuteStandalone(query string, bindVars map[string]*quer } // The autocommit flag is always set to false because we currently don't // execute DMLs through ExecuteStandalone. - return vc.executor.scatterConn.ExecuteMultiShard(vc.ctx, rss, bqs, vc.tabletType, NewAutocommitSession(vc.safeSession.Session), false, false /* autocommit */) + qr, errs := vc.executor.scatterConn.ExecuteMultiShard(vc.ctx, rss, bqs, vc.tabletType, NewAutocommitSession(vc.safeSession.Session), false, false /* autocommit */) + return qr, vterrors.Aggregate(errs) } // StreamExeculteMulti is the streaming version of ExecuteMultiShard. diff --git a/go/vt/vtgate/vtgate_test.go b/go/vt/vtgate/vtgate_test.go index d0d372a2a5c..e77270bb42d 100644 --- a/go/vt/vtgate/vtgate_test.go +++ b/go/vt/vtgate/vtgate_test.go @@ -2032,7 +2032,7 @@ func testErrorPropagation(t *testing.T, sbcs []*sandboxconn.SandboxConn, before } else { ec := vterrors.Code(err) if ec != expected { - t.Errorf("unexpected error, got %v want %v: %v", ec, expected, err) + t.Errorf("unexpected error, got code %v err %v, want %v", ec, err, expected) } } for _, sbc := range sbcs {