Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Refactor routing primitive #9543

Merged
merged 12 commits into from
Jan 22, 2022
Merged

Refactor routing primitive #9543

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
29afd1d
refactor: move dba query routing to common routing logic
GuptaManan100 Jan 19, 2022
f8a2a8c
refactor: move unsharded and any query routing to common routing logic
GuptaManan100 Jan 19, 2022
6bde9f2
refactor: move scatter query routing to common routing logic
GuptaManan100 Jan 19, 2022
b1b7dcd
refactor: move byDestination query routing to common routing logic
GuptaManan100 Jan 19, 2022
895be38
refactor: move equal query routing to common routing logic
GuptaManan100 Jan 19, 2022
f903ee2
refactor: route and dml to use routingparameters struct
harshit-gangal Jan 20, 2022
b6b1614
refactor: dml struct as reference to update and delete primitive
harshit-gangal Jan 20, 2022
bcf98c3
refactor: moved opcode to RoutingParameters to use common opcodes, ch…
harshit-gangal Jan 20, 2022
e8328c0
refactor: moved all route routing logic to common routing
harshit-gangal Jan 21, 2022
23040f9
refactor: delete primitive to use common routing
harshit-gangal Jan 21, 2022
d7d3bea
refactor: update primitive to use common routing
harshit-gangal Jan 21, 2022
d52155e
refactor: pulled update and delete exec logic into common dml
harshit-gangal Jan 21, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 43 additions & 48 deletions go/vt/vtgate/engine/cached_size.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

164 changes: 20 additions & 144 deletions go/vt/vtgate/engine/delete.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,36 +25,25 @@ import (
topodatapb "vitess.io/vitess/go/vt/proto/topodata"

"vitess.io/vitess/go/sqltypes"
"vitess.io/vitess/go/vt/key"
"vitess.io/vitess/go/vt/srvtopo"
"vitess.io/vitess/go/vt/vterrors"
"vitess.io/vitess/go/vt/vtgate/vindexes"

querypb "vitess.io/vitess/go/vt/proto/query"
vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
)

var _ Primitive = (*Delete)(nil)

// Delete represents the instructions to perform a delete.
type Delete struct {
DML
*DML

// Delete does not take inputs
noInputs
}

var delName = map[DMLOpcode]string{
Unsharded: "DeleteUnsharded",
Equal: "DeleteEqual",
In: "DeleteIn",
Scatter: "DeleteScatter",
ByDestination: "DeleteByDestination",
}

// RouteType returns a description of the query routing type used by the primitive
func (del *Delete) RouteType() string {
return delName[del.Opcode]
return del.Opcode.String()
}

// GetKeyspaceName specifies the Keyspace that this primitive routes to.
Expand All @@ -77,25 +66,25 @@ func (del *Delete) TryExecute(vcursor VCursor, bindVars map[string]*querypb.Bind
defer cancel()
}

rss, _, err := del.findRoute(vcursor, bindVars)
if err != nil {
return nil, err
}
err = allowOnlyPrimary(rss...)
if err != nil {
return nil, err
}

switch del.Opcode {
case Unsharded:
return del.execDeleteUnsharded(vcursor, bindVars)
return del.execUnsharded(vcursor, bindVars, rss)
case Equal:
switch del.Vindex.(type) {
case vindexes.MultiColumn:
return del.execDeleteEqualMultiCol(vcursor, bindVars)
default:
return del.execDeleteEqual(vcursor, bindVars)
}
case In:
return del.execDeleteIn(vcursor, bindVars)
case Scatter:
return del.execDeleteByDestination(vcursor, bindVars, key.DestinationAllShards{})
case ByDestination:
return del.execDeleteByDestination(vcursor, bindVars, del.TargetDestination)
return del.execEqual(vcursor, bindVars, rss, del.deleteVindexEntries)
case IN, Scatter, ByDestination:
return del.execMultiDestination(vcursor, bindVars, rss, del.deleteVindexEntries)
default:
// Unreachable.
return nil, fmt.Errorf("unsupported opcode: %v", del)
return nil, fmt.Errorf("unsupported opcode: %v", del.Opcode)
}
}

Expand All @@ -113,126 +102,13 @@ func (del *Delete) GetFields(VCursor, map[string]*querypb.BindVariable) (*sqltyp
return nil, fmt.Errorf("BUG: unreachable code for %q", del.Query)
}

func (del *Delete) execDeleteUnsharded(vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
rss, _, err := vcursor.ResolveDestinations(del.Keyspace.Name, nil, []key.Destination{key.DestinationAllShards{}})
if err != nil {
return nil, err
}
if len(rss) != 1 {
return nil, vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "cannot send query to multiple shards for un-sharded database: %v", rss)
}
err = allowOnlyPrimary(rss...)
if err != nil {
return nil, err
}
return execShard(vcursor, del.Query, bindVars, rss[0], true, true /* canAutocommit */)
}

func (del *Delete) execDeleteEqual(vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
env := evalengine.EnvWithBindVars(bindVars)
key, err := env.Evaluate(del.Values[0])
if err != nil {
return nil, err
}
rs, ksid, err := resolveSingleShard(vcursor, del.Vindex.(vindexes.SingleColumn), del.Keyspace, key.Value())
if err != nil {
return nil, err
}
err = allowOnlyPrimary(rs)
if err != nil {
return nil, err
}

if len(ksid) == 0 {
return &sqltypes.Result{}, nil
}
if del.OwnedVindexQuery != "" {
err = del.deleteVindexEntries(vcursor, bindVars, []*srvtopo.ResolvedShard{rs})
if err != nil {
return nil, err
}
}
return execShard(vcursor, del.Query, bindVars, rs, true /* rollbackOnError */, true /* canAutocommit */)
}

func (del *Delete) execDeleteEqualMultiCol(vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
env := evalengine.EnvWithBindVars(bindVars)
var rowValue []sqltypes.Value
for _, rvalue := range del.Values {
v, err := env.Evaluate(rvalue)
if err != nil {
return nil, err
}
rowValue = append(rowValue, v.Value())
}
rss, _, err := resolveShardsMultiCol(vcursor, del.Vindex.(vindexes.MultiColumn), del.Keyspace, [][]sqltypes.Value{rowValue}, false /* shardIdsNeeded */)
if err != nil {
return nil, err
}
if len(rss) != 1 {
return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "vindex mapped id to multi shards: %d", len(rss))
}
err = allowOnlyPrimary(rss...)
if err != nil {
return nil, err
}
if del.OwnedVindexQuery != "" {
err = del.deleteVindexEntries(vcursor, bindVars, rss)
if err != nil {
return nil, err
}
}
return execShard(vcursor, del.Query, bindVars, rss[0], true /* rollbackOnError */, true /* canAutocommit */)
}

func (del *Delete) execDeleteIn(vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
rss, queries, err := resolveMultiValueShards(vcursor, del.Keyspace, del.Query, bindVars, del.Values, del.Vindex)
if err != nil {
return nil, err
}
err = allowOnlyPrimary(rss...)
if err != nil {
return nil, err
}

if del.OwnedVindexQuery != "" {
if err := del.deleteVindexEntries(vcursor, bindVars, rss); err != nil {
return nil, err
}
}
return execMultiShard(vcursor, rss, queries, del.MultiShardAutocommit)
}

func (del *Delete) execDeleteByDestination(vcursor VCursor, bindVars map[string]*querypb.BindVariable, dest key.Destination) (*sqltypes.Result, error) {
rss, _, err := vcursor.ResolveDestinations(del.Keyspace.Name, nil, []key.Destination{dest})
if err != nil {
return nil, err
}
err = allowOnlyPrimary(rss...)
if err != nil {
return nil, err
}

queries := make([]*querypb.BoundQuery, len(rss))
for i := range rss {
queries[i] = &querypb.BoundQuery{
Sql: del.Query,
BindVariables: bindVars,
}
}
if len(del.Table.Owned) > 0 {
err = del.deleteVindexEntries(vcursor, bindVars, rss)
if err != nil {
return nil, err
}
}
return execMultiShard(vcursor, rss, queries, del.MultiShardAutocommit)
}

// deleteVindexEntries performs an delete if table owns vindex.
// Note: the commit order may be different from the DML order because it's possible
// for DMLs to reuse existing transactions.
func (del *Delete) deleteVindexEntries(vcursor VCursor, bindVars map[string]*querypb.BindVariable, rss []*srvtopo.ResolvedShard) error {
if del.OwnedVindexQuery == "" {
return nil
}
queries := make([]*querypb.BoundQuery, len(rss))
for i := range rss {
queries[i] = &querypb.BoundQuery{Sql: del.OwnedVindexQuery, BindVariables: bindVars}
Expand Down Expand Up @@ -290,7 +166,7 @@ func (del *Delete) description() PrimitiveDescription {
}
}

func addFieldsIfNotEmpty(dml DML, other map[string]interface{}) {
func addFieldsIfNotEmpty(dml *DML, other map[string]interface{}) {
if dml.Vindex != nil {
other["Vindex"] = dml.Vindex.String()
}
Expand Down
Loading