Skip to content

Execute batch v2 Implementation #2427

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

Closed
harshit-gangal wants to merge 17 commits into from
Closed

Execute batch v2 Implementation #2427

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
96e00f2
initial commit
Dec 31, 2016
4ed0e6d
notInTransaction moved from vcursor
Dec 31, 2016
5c634e4
Fixed Build
Jan 2, 2017
eebb3a1
execute batch v2 framework
Jan 3, 2017
8c0268d
fixed lint error
Jan 3, 2017
7176abd
resolved merge conflict
Jan 4, 2017
7a6fd15
initial batchParallel
Jan 5, 2017
1a29f75
fixed test
Jan 5, 2017
d79bc78
ExecuteBatch Implementation
Jan 10, 2017
59a044b
fixed test
Jan 10, 2017
eaccf6d
fixed race condition
Jan 10, 2017
a78452c
session locks
Jan 11, 2017
22e2c6a
code refactor
Jan 11, 2017
34e2cf2
removed coverage file
Jan 11, 2017
8583782
Adding Opcode
Jan 12, 2017
befeac2
batch test
Jan 17, 2017
dac782a
sync.waitGroup changes
Jan 18, 2017
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
4 changes: 2 additions & 2 deletions go/cmd/vtgateclienttest/services/callerid.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,13 +100,13 @@ func (c *callerIDClient) ExecuteEntityIds(ctx context.Context, sql string, bindV
return c.fallbackClient.ExecuteEntityIds(ctx, sql, bindVariables, keyspace, entityColumnName, entityKeyspaceIDs, tabletType, session, notInTransaction, options)
}

func (c *callerIDClient) ExecuteBatch(ctx context.Context, sqlList []string, bindVariablesList []map[string]interface{}, keyspace string, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions) ([]sqltypes.QueryResponse, error) {
func (c *callerIDClient) ExecuteBatch(ctx context.Context, sqlList []string, bindVariablesList []map[string]interface{}, keyspace string, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions, execParallel bool) ([]sqltypes.QueryResponse, error) {
if len(sqlList) == 1 {
if ok, err := c.checkCallerID(ctx, sqlList[0]); ok {
return nil, err
}
}
return c.fallbackClient.ExecuteBatch(ctx, sqlList, bindVariablesList, keyspace, tabletType, asTransaction, session, options)
return c.fallbackClient.ExecuteBatch(ctx, sqlList, bindVariablesList, keyspace, tabletType, asTransaction, session, options, execParallel)
}

func (c *callerIDClient) ExecuteBatchShards(ctx context.Context, queries []*vtgatepb.BoundShardQuery, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions) ([]sqltypes.Result, error) {
Expand Down
5 changes: 3 additions & 2 deletions go/cmd/vtgateclienttest/services/echo.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,7 @@ func (c *echoClient) ExecuteEntityIds(ctx context.Context, sql string, bindVaria
return c.fallbackClient.ExecuteEntityIds(ctx, sql, bindVariables, keyspace, entityColumnName, entityKeyspaceIDs, tabletType, session, notInTransaction, options)
}

func (c *echoClient) ExecuteBatch(ctx context.Context, sqlList []string, bindVariablesList []map[string]interface{}, keyspace string, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions) ([]sqltypes.QueryResponse, error) {
func (c *echoClient) ExecuteBatch(ctx context.Context, sqlList []string, bindVariablesList []map[string]interface{}, keyspace string, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions, execParallel bool) ([]sqltypes.QueryResponse, error) {
if len(sqlList) > 0 && strings.HasPrefix(sqlList[0], EchoPrefix) {
var queryResponse []sqltypes.QueryResponse
if bindVariablesList == nil {
Expand All @@ -197,12 +197,13 @@ func (c *echoClient) ExecuteBatch(ctx context.Context, sqlList []string, bindVar
"session": session,
"asTransaction": asTransaction,
"options": options,
"execParallel": execParallel,
})
queryResponse = append(queryResponse, sqltypes.QueryResponse{QueryResult: result, QueryError: nil})
}
return queryResponse, nil
}
return c.fallbackClient.ExecuteBatch(ctx, sqlList, bindVariablesList, keyspace, tabletType, asTransaction, session, options)
return c.fallbackClient.ExecuteBatch(ctx, sqlList, bindVariablesList, keyspace, tabletType, asTransaction, session, options, execParallel)
}

func (c *echoClient) ExecuteBatchShards(ctx context.Context, queries []*vtgatepb.BoundShardQuery, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions) ([]sqltypes.Result, error) {
Expand Down
4 changes: 2 additions & 2 deletions go/cmd/vtgateclienttest/services/errors.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,7 @@ func (c *errorClient) ExecuteEntityIds(ctx context.Context, sql string, bindVari
return c.fallbackClient.ExecuteEntityIds(ctx, sql, bindVariables, keyspace, entityColumnName, entityKeyspaceIDs, tabletType, session, notInTransaction, options)
}

func (c *errorClient) ExecuteBatch(ctx context.Context, sqlList []string, bindVariablesList []map[string]interface{}, keyspace string, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions) ([]sqltypes.QueryResponse, error) {
func (c *errorClient) ExecuteBatch(ctx context.Context, sqlList []string, bindVariablesList []map[string]interface{}, keyspace string, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions, execParallel bool) ([]sqltypes.QueryResponse, error) {
if len(sqlList) == 1 {
if err := requestToPartialError(sqlList[0], session); err != nil {
return nil, err
Expand All @@ -192,7 +192,7 @@ func (c *errorClient) ExecuteBatch(ctx context.Context, sqlList []string, bindVa
return nil, err
}
}
return c.fallbackClient.ExecuteBatch(ctx, sqlList, bindVariablesList, keyspace, tabletType, asTransaction, session, options)
return c.fallbackClient.ExecuteBatch(ctx, sqlList, bindVariablesList, keyspace, tabletType, asTransaction, session, options, execParallel)
}

func (c *errorClient) ExecuteBatchShards(ctx context.Context, queries []*vtgatepb.BoundShardQuery, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions) ([]sqltypes.Result, error) {
Expand Down
4 changes: 2 additions & 2 deletions go/cmd/vtgateclienttest/services/fallback.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,8 +47,8 @@ func (c fallbackClient) ExecuteEntityIds(ctx context.Context, sql string, bindVa
return c.fallback.ExecuteEntityIds(ctx, sql, bindVariables, keyspace, entityColumnName, entityKeyspaceIDs, tabletType, session, notInTransaction, options)
}

func (c fallbackClient) ExecuteBatch(ctx context.Context, sqlList []string, bindVariablesList []map[string]interface{}, keyspace string, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions) ([]sqltypes.QueryResponse, error) {
return c.fallback.ExecuteBatch(ctx, sqlList, bindVariablesList, keyspace, tabletType, asTransaction, session, options)
func (c fallbackClient) ExecuteBatch(ctx context.Context, sqlList []string, bindVariablesList []map[string]interface{}, keyspace string, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions, execParallel bool) ([]sqltypes.QueryResponse, error) {
return c.fallback.ExecuteBatch(ctx, sqlList, bindVariablesList, keyspace, tabletType, asTransaction, session, options, execParallel)
}

func (c fallbackClient) ExecuteBatchShards(ctx context.Context, queries []*vtgatepb.BoundShardQuery, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions) ([]sqltypes.Result, error) {
Expand Down
2 changes: 1 addition & 1 deletion go/cmd/vtgateclienttest/services/terminal.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ func (c *terminalClient) ExecuteEntityIds(ctx context.Context, sql string, bindV
return nil, errTerminal
}

func (c *terminalClient) ExecuteBatch(ctx context.Context, sqlList []string, bindVariablesList []map[string]interface{}, keyspace string, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions) ([]sqltypes.QueryResponse, error) {
func (c *terminalClient) ExecuteBatch(ctx context.Context, sqlList []string, bindVariablesList []map[string]interface{}, keyspace string, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions, execParallel bool) ([]sqltypes.QueryResponse, error) {
if len(sqlList) == 1 {
if sqlList[0] == "quit://" {
log.Fatal("Received quit:// query. Going down.")
Expand Down
Loading