diff --git a/go/mysql/query.go b/go/mysql/query.go index 26582cecbd9..720a533a772 100644 --- a/go/mysql/query.go +++ b/go/mysql/query.go @@ -404,6 +404,7 @@ func (c *Conn) ReadQueryResult(maxrows int, wantfields bool) (*sqltypes.Result, return &sqltypes.Result{ RowsAffected: packetOk.affectedRows, InsertID: packetOk.lastInsertID, + InsertIDChanged: packetOk.lastInsertID > 0, SessionStateChanges: packetOk.sessionStateData, StatusFlags: packetOk.statusFlags, Info: packetOk.info, diff --git a/go/vt/vtgate/executor.go b/go/vt/vtgate/executor.go index d7f31b4b9a9..2d4853f86cd 100644 --- a/go/vt/vtgate/executor.go +++ b/go/vt/vtgate/executor.go @@ -680,7 +680,7 @@ func (e *Executor) executeSPInAllSessions(ctx context.Context, safeSession *econ }) queries = append(queries, &querypb.BoundQuery{Sql: sql}) } - qr, errs = e.ExecuteMultiShard(ctx, nil, rss, queries, safeSession, false /*autocommit*/, ignoreMaxMemoryRows, nullResultsObserver{}) + qr, errs = e.ExecuteMultiShard(ctx, nil, rss, queries, safeSession, false /*autocommit*/, ignoreMaxMemoryRows, nullResultsObserver{}, false) err := vterrors.Aggregate(errs) if err != nil { return nil, err @@ -1486,8 +1486,8 @@ func parseAndValidateQuery(query string, parser *sqlparser.Parser) (sqlparser.St } // ExecuteMultiShard implements the IExecutor interface -func (e *Executor) ExecuteMultiShard(ctx context.Context, primitive engine.Primitive, rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, session *econtext.SafeSession, autocommit bool, ignoreMaxMemoryRows bool, resultsObserver econtext.ResultsObserver) (qr *sqltypes.Result, errs []error) { - return e.scatterConn.ExecuteMultiShard(ctx, primitive, rss, queries, session, autocommit, ignoreMaxMemoryRows, resultsObserver) +func (e *Executor) ExecuteMultiShard(ctx context.Context, primitive engine.Primitive, rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, session *econtext.SafeSession, autocommit bool, ignoreMaxMemoryRows bool, resultsObserver econtext.ResultsObserver, fetchLastInsertID bool) (qr *sqltypes.Result, errs []error) { + return e.scatterConn.ExecuteMultiShard(ctx, primitive, rss, queries, session, autocommit, ignoreMaxMemoryRows, resultsObserver, fetchLastInsertID) } // StreamExecuteMulti implements the IExecutor interface diff --git a/go/vt/vtgate/executorcontext/vcursor_impl.go b/go/vt/vtgate/executorcontext/vcursor_impl.go index 08fb89366c2..3ac8e3d7995 100644 --- a/go/vt/vtgate/executorcontext/vcursor_impl.go +++ b/go/vt/vtgate/executorcontext/vcursor_impl.go @@ -95,7 +95,7 @@ type ( // vcursor_impl needs these facilities to be able to be able to execute queries for vindexes iExecute interface { Execute(ctx context.Context, mysqlCtx vtgateservice.MySQLConnection, method string, session *SafeSession, s string, vars map[string]*querypb.BindVariable) (*sqltypes.Result, error) - ExecuteMultiShard(ctx context.Context, primitive engine.Primitive, rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, session *SafeSession, autocommit bool, ignoreMaxMemoryRows bool, resultsObserver ResultsObserver) (qr *sqltypes.Result, errs []error) + ExecuteMultiShard(ctx context.Context, primitive engine.Primitive, rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, session *SafeSession, autocommit bool, ignoreMaxMemoryRows bool, resultsObserver ResultsObserver, fetchLastInsertID bool) (qr *sqltypes.Result, errs []error) StreamExecuteMulti(ctx context.Context, primitive engine.Primitive, query string, rss []*srvtopo.ResolvedShard, vars []map[string]*querypb.BindVariable, session *SafeSession, autocommit bool, callback func(reply *sqltypes.Result) error, observer ResultsObserver) []error ExecuteLock(ctx context.Context, rs *srvtopo.ResolvedShard, query *querypb.BoundQuery, session *SafeSession, lockFuncType sqlparser.LockingFuncType) (*sqltypes.Result, error) Commit(ctx context.Context, safeSession *SafeSession) error @@ -761,7 +761,7 @@ func (vc *VCursorImpl) ExecuteMultiShard(ctx context.Context, primitive engine.P return nil, []error{err} } - qr, errs := vc.executor.ExecuteMultiShard(ctx, primitive, rss, commentedShardQueries(queries, vc.marginComments), vc.SafeSession, canAutocommit, vc.ignoreMaxMemoryRows, vc.observer) + qr, errs := vc.executor.ExecuteMultiShard(ctx, primitive, rss, commentedShardQueries(queries, vc.marginComments), vc.SafeSession, canAutocommit, vc.ignoreMaxMemoryRows, vc.observer, fetchLastInsertID) vc.setRollbackOnPartialExecIfRequired(len(errs) != len(rss), rollbackOnError) vc.logShardsQueried(primitive, len(rss)) return qr, errs @@ -801,7 +801,7 @@ func (vc *VCursorImpl) ExecuteStandalone(ctx context.Context, primitive engine.P } // The autocommit flag is always set to false because we currently don't // execute DMLs through ExecuteStandalone. - qr, errs := vc.executor.ExecuteMultiShard(ctx, primitive, rss, bqs, NewAutocommitSession(vc.SafeSession.Session), false /* autocommit */, vc.ignoreMaxMemoryRows, vc.observer) + qr, errs := vc.executor.ExecuteMultiShard(ctx, primitive, rss, bqs, NewAutocommitSession(vc.SafeSession.Session), false /* autocommit */, vc.ignoreMaxMemoryRows, vc.observer, false /* fetchLastInsertID */) vc.logShardsQueried(primitive, len(rss)) return qr, vterrors.Aggregate(errs) } diff --git a/go/vt/vtgate/executorcontext/vcursor_impl_test.go b/go/vt/vtgate/executorcontext/vcursor_impl_test.go index 16d2c03bf1c..b5245c0801e 100644 --- a/go/vt/vtgate/executorcontext/vcursor_impl_test.go +++ b/go/vt/vtgate/executorcontext/vcursor_impl_test.go @@ -389,7 +389,7 @@ func (f fakeExecutor) Execute(ctx context.Context, mysqlCtx vtgateservice.MySQLC panic("implement me") } -func (f fakeExecutor) ExecuteMultiShard(ctx context.Context, primitive engine.Primitive, rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, session *SafeSession, autocommit bool, ignoreMaxMemoryRows bool, resultsObserver ResultsObserver) (qr *sqltypes.Result, errs []error) { +func (f fakeExecutor) ExecuteMultiShard(ctx context.Context, primitive engine.Primitive, rss []*srvtopo.ResolvedShard, queries []*querypb.BoundQuery, session *SafeSession, autocommit bool, ignoreMaxMemoryRows bool, resultsObserver ResultsObserver, fetchLastInsertID bool) (qr *sqltypes.Result, errs []error) { // TODO implement me panic("implement me") } diff --git a/go/vt/vtgate/legacy_scatter_conn_test.go b/go/vt/vtgate/legacy_scatter_conn_test.go index 0d49e7b7bd9..cd711a14d08 100644 --- a/go/vt/vtgate/legacy_scatter_conn_test.go +++ b/go/vt/vtgate/legacy_scatter_conn_test.go @@ -101,7 +101,7 @@ func TestLegacyExecuteFailOnAutocommit(t *testing.T) { }, Autocommit: false, } - _, errs := sc.ExecuteMultiShard(ctx, nil, rss, queries, econtext.NewSafeSession(session), true /*autocommit*/, false, nullResultsObserver{}) + _, errs := sc.ExecuteMultiShard(ctx, nil, rss, queries, econtext.NewSafeSession(session), true /*autocommit*/, false, nullResultsObserver{}, false) err := vterrors.Aggregate(errs) require.Error(t, err) require.Contains(t, err.Error(), "in autocommit mode, transactionID should be zero but was: 123") @@ -125,7 +125,7 @@ func TestScatterConnExecuteMulti(t *testing.T) { } } - qr, errs := sc.ExecuteMultiShard(ctx, nil, rss, queries, econtext.NewSafeSession(nil), false /*autocommit*/, false, nullResultsObserver{}) + qr, errs := sc.ExecuteMultiShard(ctx, nil, rss, queries, econtext.NewSafeSession(nil), false /*autocommit*/, false, nullResultsObserver{}, false) return qr, vterrors.Aggregate(errs) }) } @@ -312,7 +312,7 @@ func TestMaxMemoryRows(t *testing.T) { sbc0.SetResults([]*sqltypes.Result{tworows, tworows}) sbc1.SetResults([]*sqltypes.Result{tworows, tworows}) - _, errs := sc.ExecuteMultiShard(ctx, nil, rss, queries, session, false, test.ignoreMaxMemoryRows, nullResultsObserver{}) + _, errs := sc.ExecuteMultiShard(ctx, nil, rss, queries, session, false, test.ignoreMaxMemoryRows, nullResultsObserver{}, false) if test.ignoreMaxMemoryRows { require.NoError(t, err) } else { @@ -344,7 +344,7 @@ func TestLegaceHealthCheckFailsOnReservedConnections(t *testing.T) { }) } - _, errs := sc.ExecuteMultiShard(ctx, nil, rss, queries, session, false, false, nullResultsObserver{}) + _, errs := sc.ExecuteMultiShard(ctx, nil, rss, queries, session, false, false, nullResultsObserver{}, false) require.Error(t, vterrors.Aggregate(errs)) } @@ -367,7 +367,7 @@ func executeOnShardsReturnsErr(t *testing.T, ctx context.Context, res *srvtopo.R }) } - _, errs := sc.ExecuteMultiShard(ctx, nil, rss, queries, session, false, false, nullResultsObserver{}) + _, errs := sc.ExecuteMultiShard(ctx, nil, rss, queries, session, false, false, nullResultsObserver{}, false) return vterrors.Aggregate(errs) } @@ -432,7 +432,7 @@ func TestMultiExecs(t *testing.T) { observer := recordingResultsObserver{} session := econtext.NewSafeSession(&vtgatepb.Session{}) - _, err := sc.ExecuteMultiShard(ctx, nil, rss, queries, session, false, false, &observer) + _, err := sc.ExecuteMultiShard(ctx, nil, rss, queries, session, false, false, &observer, false) require.NoError(t, vterrors.Aggregate(err)) if len(sbc0.Queries) == 0 || len(sbc1.Queries) == 0 { t.Fatalf("didn't get expected query") @@ -515,27 +515,27 @@ func TestScatterConnSingleDB(t *testing.T) { // TransactionMode_SINGLE in session session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true, TransactionMode: vtgatepb.TransactionMode_SINGLE}) queries := []*querypb.BoundQuery{{Sql: "query1"}} - _, errors := sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + _, errors := sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) require.Empty(t, errors) - _, errors = sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}) + _, errors = sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}, false) require.Error(t, errors[0]) assert.Contains(t, errors[0].Error(), want) // TransactionMode_SINGLE in txconn sc.txConn.mode = vtgatepb.TransactionMode_SINGLE session = econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - _, errors = sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + _, errors = sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) require.Empty(t, errors) - _, errors = sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}) + _, errors = sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}, false) require.Error(t, errors[0]) assert.Contains(t, errors[0].Error(), want) // TransactionMode_MULTI in txconn. Should not fail. sc.txConn.mode = vtgatepb.TransactionMode_MULTI session = econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - _, errors = sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + _, errors = sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) require.Empty(t, errors) - _, errors = sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}) + _, errors = sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}, false) require.Empty(t, errors) } diff --git a/go/vt/vtgate/scatter_conn.go b/go/vt/vtgate/scatter_conn.go index 6e2cf9ad8ba..37d6e22eea3 100644 --- a/go/vt/vtgate/scatter_conn.go +++ b/go/vt/vtgate/scatter_conn.go @@ -152,6 +152,7 @@ func (stc *ScatterConn) ExecuteMultiShard( autocommit bool, ignoreMaxMemoryRows bool, resultsObserver econtext.ResultsObserver, + fetchLastInsertID bool, ) (qr *sqltypes.Result, errs []error) { if len(rss) != len(queries) { @@ -187,6 +188,9 @@ func (stc *ScatterConn) ExecuteMultiShard( opts = session.Session.Options } + // TODO reset this? + opts.FetchLastInsertId = fetchLastInsertID + if autocommit { // As this is auto-commit, the transactionID is supposed to be zero. if transactionID != int64(0) { diff --git a/go/vt/vtgate/scatter_conn_test.go b/go/vt/vtgate/scatter_conn_test.go index ab8680ca5e6..e6c976c7b74 100644 --- a/go/vt/vtgate/scatter_conn_test.go +++ b/go/vt/vtgate/scatter_conn_test.go @@ -101,7 +101,7 @@ func TestExecuteFailOnAutocommit(t *testing.T) { }, Autocommit: false, } - _, errs := sc.ExecuteMultiShard(ctx, nil, rss, queries, econtext.NewSafeSession(session), true /*autocommit*/, false, nullResultsObserver{}) + _, errs := sc.ExecuteMultiShard(ctx, nil, rss, queries, econtext.NewSafeSession(session), true /*autocommit*/, false, nullResultsObserver{}, false) err := vterrors.Aggregate(errs) require.Error(t, err) require.Contains(t, err.Error(), "in autocommit mode, transactionID should be zero but was: 123") @@ -184,7 +184,7 @@ func TestExecutePanic(t *testing.T) { require.Contains(t, logMessage, "(*ScatterConn).multiGoTransaction") }() - _, _ = sc.ExecuteMultiShard(ctx, nil, rss, queries, econtext.NewSafeSession(session), true /*autocommit*/, false, nullResultsObserver{}) + _, _ = sc.ExecuteMultiShard(ctx, nil, rss, queries, econtext.NewSafeSession(session), true /*autocommit*/, false, nullResultsObserver{}, false) } diff --git a/go/vt/vtgate/tx_conn_test.go b/go/vt/vtgate/tx_conn_test.go index 333094569c8..d96f0b8fccf 100644 --- a/go/vt/vtgate/tx_conn_test.go +++ b/go/vt/vtgate/tx_conn_test.go @@ -58,7 +58,7 @@ func TestTxConnBegin(t *testing.T) { require.NoError(t, err) wantSession := vtgatepb.Session{InTransaction: true} utils.MustMatch(t, &wantSession, session, "Session") - _, errors := sc.ExecuteMultiShard(ctx, nil, rss0, queries, safeSession, false, false, nullResultsObserver{}) + _, errors := sc.ExecuteMultiShard(ctx, nil, rss0, queries, safeSession, false, false, nullResultsObserver{}, false) require.Empty(t, errors) // Begin again should cause a commit and a new begin. @@ -78,7 +78,7 @@ func TestTxConnCommitFailure(t *testing.T) { // Sequence the executes to ensure commit order session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rssm[0], queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rssm[0], queries, session, false, false, nullResultsObserver{}, false) wantSession := vtgatepb.Session{ InTransaction: true, ShardSessions: []*vtgatepb.Session_ShardSession{{ @@ -93,7 +93,7 @@ func TestTxConnCommitFailure(t *testing.T) { } utils.MustMatch(t, &wantSession, session.Session, "Session") - sc.ExecuteMultiShard(ctx, nil, rssm[1], queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rssm[1], queries, session, false, false, nullResultsObserver{}, false) wantSession = vtgatepb.Session{ InTransaction: true, ShardSessions: []*vtgatepb.Session_ShardSession{{ @@ -116,7 +116,7 @@ func TestTxConnCommitFailure(t *testing.T) { } utils.MustMatch(t, &wantSession, session.Session, "Session") - sc.ExecuteMultiShard(ctx, nil, rssa, threeQueries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rssa, threeQueries, session, false, false, nullResultsObserver{}, false) wantSession = vtgatepb.Session{ InTransaction: true, ShardSessions: []*vtgatepb.Session_ShardSession{{ @@ -185,7 +185,7 @@ func TestTxConnCommitFailureAfterNonAtomicCommitMaxShards(t *testing.T) { } for i := 0; i < 18; i++ { - sc.ExecuteMultiShard(ctx, nil, rssm[i], queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rssm[i], queries, session, false, false, nullResultsObserver{}, false) wantSession.ShardSessions = append(wantSession.ShardSessions, &vtgatepb.Session_ShardSession{ Target: &querypb.Target{ Keyspace: "TestTxConn", @@ -239,7 +239,7 @@ func TestTxConnCommitFailureBeforeNonAtomicCommitMaxShards(t *testing.T) { } for i := 0; i < 17; i++ { - sc.ExecuteMultiShard(ctx, nil, rssm[i], queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rssm[i], queries, session, false, false, nullResultsObserver{}, false) wantSession.ShardSessions = append(wantSession.ShardSessions, &vtgatepb.Session_ShardSession{ Target: &querypb.Target{ Keyspace: "TestTxConn", @@ -285,7 +285,7 @@ func TestTxConnCommitSuccess(t *testing.T) { // Sequence the executes to ensure commit order session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) wantSession := vtgatepb.Session{ InTransaction: true, ShardSessions: []*vtgatepb.Session_ShardSession{{ @@ -299,7 +299,7 @@ func TestTxConnCommitSuccess(t *testing.T) { }}, } utils.MustMatch(t, &wantSession, session.Session, "Session") - sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}, false) wantSession = vtgatepb.Session{ InTransaction: true, ShardSessions: []*vtgatepb.Session_ShardSession{{ @@ -338,7 +338,7 @@ func TestTxConnReservedCommitSuccess(t *testing.T) { // Sequence the executes to ensure commit order session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true, InReservedConn: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) wantSession := vtgatepb.Session{ InTransaction: true, InReservedConn: true, @@ -354,7 +354,7 @@ func TestTxConnReservedCommitSuccess(t *testing.T) { }}, } utils.MustMatch(t, &wantSession, session.Session, "Session") - sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}, false) wantSession = vtgatepb.Session{ InTransaction: true, InReservedConn: true, @@ -425,9 +425,9 @@ func TestTxConnReservedOn2ShardTxOn1ShardAndCommit(t *testing.T) { session := econtext.NewSafeSession(&vtgatepb.Session{InReservedConn: true}) // this will create reserved connections against all tablets - _, errs := sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}) + _, errs := sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}, false) require.Empty(t, errs) - _, errs = sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + _, errs = sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) require.Empty(t, errs) wantSession := vtgatepb.Session{ @@ -455,7 +455,7 @@ func TestTxConnReservedOn2ShardTxOn1ShardAndCommit(t *testing.T) { session.Session.InTransaction = true // start a transaction against rss0 - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) wantSession = vtgatepb.Session{ InTransaction: true, InReservedConn: true, @@ -520,9 +520,9 @@ func TestTxConnReservedOn2ShardTxOn1ShardAndRollback(t *testing.T) { session := econtext.NewSafeSession(&vtgatepb.Session{InReservedConn: true}) // this will create reserved connections against all tablets - _, errs := sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}) + _, errs := sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}, false) require.Empty(t, errs) - _, errs = sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + _, errs = sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) require.Empty(t, errs) wantSession := vtgatepb.Session{ @@ -550,7 +550,7 @@ func TestTxConnReservedOn2ShardTxOn1ShardAndRollback(t *testing.T) { session.Session.InTransaction = true // start a transaction against rss0 - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) wantSession = vtgatepb.Session{ InTransaction: true, InReservedConn: true, @@ -614,13 +614,13 @@ func TestTxConnCommitOrderFailure1(t *testing.T) { // Sequence the executes to ensure commit order session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) session.SetCommitOrder(vtgatepb.CommitOrder_PRE) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) session.SetCommitOrder(vtgatepb.CommitOrder_POST) - sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}, false) sbc0.MustFailCodes[vtrpcpb.Code_INVALID_ARGUMENT] = 1 err := sc.txConn.Commit(ctx, session) @@ -649,13 +649,13 @@ func TestTxConnCommitOrderFailure2(t *testing.T) { // Sequence the executes to ensure commit order session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(context.Background(), nil, rss1, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(context.Background(), nil, rss1, queries, session, false, false, nullResultsObserver{}, false) session.SetCommitOrder(vtgatepb.CommitOrder_PRE) - sc.ExecuteMultiShard(context.Background(), nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(context.Background(), nil, rss0, queries, session, false, false, nullResultsObserver{}, false) session.SetCommitOrder(vtgatepb.CommitOrder_POST) - sc.ExecuteMultiShard(context.Background(), nil, rss1, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(context.Background(), nil, rss1, queries, session, false, false, nullResultsObserver{}, false) sbc1.MustFailCodes[vtrpcpb.Code_INVALID_ARGUMENT] = 1 err := sc.txConn.Commit(ctx, session) @@ -683,13 +683,13 @@ func TestTxConnCommitOrderFailure3(t *testing.T) { // Sequence the executes to ensure commit order session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) session.SetCommitOrder(vtgatepb.CommitOrder_PRE) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) session.SetCommitOrder(vtgatepb.CommitOrder_POST) - sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}, false) sbc1.MustFailCodes[vtrpcpb.Code_INVALID_ARGUMENT] = 1 require.NoError(t, @@ -725,7 +725,7 @@ func TestTxConnCommitOrderSuccess(t *testing.T) { // Sequence the executes to ensure commit order session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) wantSession := vtgatepb.Session{ InTransaction: true, ShardSessions: []*vtgatepb.Session_ShardSession{{ @@ -741,7 +741,7 @@ func TestTxConnCommitOrderSuccess(t *testing.T) { utils.MustMatch(t, &wantSession, session.Session, "Session") session.SetCommitOrder(vtgatepb.CommitOrder_PRE) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) wantSession = vtgatepb.Session{ InTransaction: true, PreSessions: []*vtgatepb.Session_ShardSession{{ @@ -766,7 +766,7 @@ func TestTxConnCommitOrderSuccess(t *testing.T) { utils.MustMatch(t, &wantSession, session.Session, "Session") session.SetCommitOrder(vtgatepb.CommitOrder_POST) - sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}, false) wantSession = vtgatepb.Session{ InTransaction: true, PreSessions: []*vtgatepb.Session_ShardSession{{ @@ -800,7 +800,7 @@ func TestTxConnCommitOrderSuccess(t *testing.T) { utils.MustMatch(t, &wantSession, session.Session, "Session") // Ensure nothing changes if we reuse a transaction. - sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}, false) utils.MustMatch(t, &wantSession, session.Session, "Session") require.NoError(t, @@ -823,7 +823,7 @@ func TestTxConnReservedCommitOrderSuccess(t *testing.T) { // Sequence the executes to ensure commit order session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true, InReservedConn: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) wantSession := vtgatepb.Session{ InTransaction: true, InReservedConn: true, @@ -841,7 +841,7 @@ func TestTxConnReservedCommitOrderSuccess(t *testing.T) { utils.MustMatch(t, &wantSession, session.Session, "Session") session.SetCommitOrder(vtgatepb.CommitOrder_PRE) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) wantSession = vtgatepb.Session{ InTransaction: true, InReservedConn: true, @@ -869,7 +869,7 @@ func TestTxConnReservedCommitOrderSuccess(t *testing.T) { utils.MustMatch(t, &wantSession, session.Session, "Session") session.SetCommitOrder(vtgatepb.CommitOrder_POST) - sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}, false) wantSession = vtgatepb.Session{ InTransaction: true, InReservedConn: true, @@ -907,7 +907,7 @@ func TestTxConnReservedCommitOrderSuccess(t *testing.T) { utils.MustMatch(t, &wantSession, session.Session, "Session") // Ensure nothing changes if we reuse a transaction. - sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}, false) utils.MustMatch(t, &wantSession, session.Session, "Session") require.NoError(t, @@ -960,8 +960,8 @@ func TestTxConnCommit2PC(t *testing.T) { sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, ctx, "TestTxConnCommit2PC") session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) - sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) + sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}, false) session.TransactionMode = vtgatepb.TransactionMode_TWOPC require.NoError(t, sc.txConn.Commit(ctx, session)) @@ -977,7 +977,7 @@ func TestTxConnCommit2PCOneParticipant(t *testing.T) { sc, sbc0, _, rss0, _, _ := newTestTxConnEnv(t, ctx, "TestTxConnCommit2PCOneParticipant") session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) session.TransactionMode = vtgatepb.TransactionMode_TWOPC require.NoError(t, sc.txConn.Commit(ctx, session)) @@ -990,8 +990,8 @@ func TestTxConnCommit2PCCreateTransactionFail(t *testing.T) { sc, sbc0, sbc1, rss0, rss1, _ := newTestTxConnEnv(t, ctx, "TestTxConnCommit2PCCreateTransactionFail") session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) - sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) + sc.ExecuteMultiShard(ctx, nil, rss1, queries, session, false, false, nullResultsObserver{}, false) sbc0.MustFailCreateTransaction = 1 session.TransactionMode = vtgatepb.TransactionMode_TWOPC @@ -1012,8 +1012,8 @@ func TestTxConnCommit2PCPrepareFail(t *testing.T) { sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, ctx, "TestTxConnCommit2PCPrepareFail") session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) - sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) + sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}, false) sbc1.MustFailPrepare = 1 session.TransactionMode = vtgatepb.TransactionMode_TWOPC @@ -1038,8 +1038,8 @@ func TestTxConnCommit2PCStartCommitFail(t *testing.T) { sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, ctx, "TestTxConnCommit2PCStartCommitFail") session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) - sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) + sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}, false) sbc0.MustFailStartCommit = 1 session.TransactionMode = vtgatepb.TransactionMode_TWOPC @@ -1057,8 +1057,8 @@ func TestTxConnCommit2PCStartCommitFail(t *testing.T) { sbc1.ResetCounter() session = econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) - sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) + sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}, false) // Here the StartCommit failure is in uncertain state so rollback is not called and neither conclude. sbc0.MustFailStartCommitUncertain = 1 @@ -1080,8 +1080,8 @@ func TestTxConnCommit2PCCommitPreparedFail(t *testing.T) { sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, ctx, "TestTxConnCommit2PCCommitPreparedFail") session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) - sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) + sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}, false) sbc1.MustFailCommitPrepared = 1 session.TransactionMode = vtgatepb.TransactionMode_TWOPC @@ -1100,8 +1100,8 @@ func TestTxConnCommit2PCConcludeTransactionFail(t *testing.T) { sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, ctx, "TestTxConnCommit2PCConcludeTransactionFail") session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) - sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) + sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}, false) sbc0.MustFailConcludeTransaction = 1 session.TransactionMode = vtgatepb.TransactionMode_TWOPC @@ -1120,8 +1120,8 @@ func TestTxConnRollback(t *testing.T) { sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, ctx, "TxConnRollback") session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) - sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) + sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}, false) require.NoError(t, sc.txConn.Rollback(ctx, session)) wantSession := vtgatepb.Session{} @@ -1136,8 +1136,8 @@ func TestTxConnReservedRollback(t *testing.T) { sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, ctx, "TxConnReservedRollback") session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true, InReservedConn: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) - sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) + sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}, false) require.NoError(t, sc.txConn.Rollback(ctx, session)) wantSession := vtgatepb.Session{ @@ -1173,8 +1173,8 @@ func TestTxConnReservedRollbackFailure(t *testing.T) { sc, sbc0, sbc1, rss0, rss1, rss01 := newTestTxConnEnv(t, ctx, "TxConnReservedRollback") session := econtext.NewSafeSession(&vtgatepb.Session{InTransaction: true, InReservedConn: true}) - sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}) - sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}) + sc.ExecuteMultiShard(ctx, nil, rss0, queries, session, false, false, nullResultsObserver{}, false) + sc.ExecuteMultiShard(ctx, nil, rss01, twoQueries, session, false, false, nullResultsObserver{}, false) sbc1.MustFailCodes[vtrpcpb.Code_INVALID_ARGUMENT] = 1 assert.Error(t, diff --git a/go/vt/vttablet/tabletserver/query_executor.go b/go/vt/vttablet/tabletserver/query_executor.go index 88c75a2866d..407ce662235 100644 --- a/go/vt/vttablet/tabletserver/query_executor.go +++ b/go/vt/vttablet/tabletserver/query_executor.go @@ -70,7 +70,7 @@ type QueryExecutor struct { const ( streamRowsSize = 256 - resetLastIDQuery = "select last_insert_id(-4200)" + resetLastIDQuery = "select last_insert_id(18446744073709547416)" resetLastIDValue = 18446744073709547416 ) @@ -1113,71 +1113,73 @@ func (qre *QueryExecutor) getSelectLimit() int64 { func (qre *QueryExecutor) execDBConn(conn *connpool.Conn, sql string, wantfields bool) (*sqltypes.Result, error) { span, ctx := trace.NewSpan(qre.ctx, "QueryExecutor.execDBConn") defer span.Finish() - defer qre.logStats.AddRewrittenSQL(sql, time.Now()) qd := NewQueryDetail(qre.logStats.Ctx, conn) - err := qre.tsv.statelessql.Add(qd) - if err != nil { + + if err := qre.tsv.statelessql.Add(qd); err != nil { return nil, err } defer qre.tsv.statelessql.Remove(qd) - if qre.options.ShouldFetchLastInsertID() { - _, err = conn.Exec(ctx, resetLastIDQuery, 1, false) - if err != nil { - return nil, err - } + if err := qre.resetLastInsertIDIfNeeded(ctx, conn); err != nil { + return nil, err } exec, err := conn.Exec(ctx, sql, int(qre.tsv.qe.maxResultSize.Load()), wantfields) if err != nil { return nil, err } - if qre.options.ShouldFetchLastInsertID() { - err := qre.fetchLastInsertID(ctx, conn, exec) - if err != nil { - return nil, err - } + + if err := qre.fetchLastInsertID(ctx, conn, exec); err != nil { + return nil, err } + return exec, nil } func (qre *QueryExecutor) execStatefulConn(conn *StatefulConnection, sql string, wantfields bool) (*sqltypes.Result, error) { span, ctx := trace.NewSpan(qre.ctx, "QueryExecutor.execStatefulConn") defer span.Finish() - defer qre.logStats.AddRewrittenSQL(sql, time.Now()) qd := NewQueryDetail(qre.logStats.Ctx, conn) - err := qre.tsv.statefulql.Add(qd) - if err != nil { + + if err := qre.tsv.statefulql.Add(qd); err != nil { return nil, err } defer qre.tsv.statefulql.Remove(qd) - if qre.options.ShouldFetchLastInsertID() { - _, err = conn.Exec(ctx, resetLastIDQuery, 1, false) - if err != nil { - return nil, err - } + if err := qre.resetLastInsertIDIfNeeded(ctx, conn.UnderlyingDBConn().Conn); err != nil { + return nil, err } exec, err := conn.Exec(ctx, sql, int(qre.tsv.qe.maxResultSize.Load()), wantfields) if err != nil { return nil, err } + + if err := qre.fetchLastInsertID(ctx, conn.UnderlyingDBConn().Conn, exec); err != nil { + return nil, err + } + + return exec, nil +} + +func (qre *QueryExecutor) resetLastInsertIDIfNeeded(ctx context.Context, conn *connpool.Conn) error { if qre.options.ShouldFetchLastInsertID() { - err = qre.fetchLastInsertID(ctx, conn.UnderlyingDBConn().Conn, exec) + // if the query contains a last_insert_id(x) function, + // we need to reset the last insert id to check if it was set by the query or not + _, err := conn.Exec(ctx, resetLastIDQuery, 1, false) if err != nil { - return nil, err + return err } } - return exec, nil + return nil } func (qre *QueryExecutor) fetchLastInsertID(ctx context.Context, conn *connpool.Conn, exec *sqltypes.Result) error { - if exec.InsertID != 0 { + if exec.InsertID != 0 || !qre.options.ShouldFetchLastInsertID() { return nil }