-
Notifications
You must be signed in to change notification settings - Fork 3.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
sql: add mechanism to retrieve causality token
Fixes #79591 Relates to #7945 Release note (<category, see below>): A new adds a new special sentinel builtin `crdb_internal.commit_with_causality_token` has been added which can be used exclusively as the only expression in a `SELECT` statement at the end of an explicit transaction for the purpose of committing the the transaction and retrieving the now committed transaction's commit timestamp. This timestamp can then be used as the input to, say, an AS OF SYSTEM TIME clause or a changefeed in order to observe the exact snapshot at which the transaction committed. Note that after issuing such a statement successfully, the client must still issue a COMMIT to reset the connection state for the next transaction.
- Loading branch information
Showing
12 changed files
with
483 additions
and
9 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| // Copyright 2022 The Cockroach Authors. | ||
| // | ||
| // Use of this software is governed by the Business Source License | ||
| // included in the file licenses/BSL.txt. | ||
| // | ||
| // As of the Change Date specified in that file, in accordance with | ||
| // the Business Source License, use of this software will be governed | ||
| // by the Apache License, Version 2.0, included in the file | ||
| // licenses/APL.txt. | ||
|
|
||
| package sql | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/cockroachdb/cockroach/pkg/sql/parser" | ||
| "github.com/cockroachdb/cockroach/pkg/util/leaktest" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| // TestIsSelectWithCausalityToken tests whether given sql statements match the | ||
| // criteria for isSelectWithCausalityToken. | ||
| func TestIsSelectWithCausalityToken(t *testing.T) { | ||
| defer leaktest.AfterTest(t)() | ||
|
|
||
| for _, tc := range []struct { | ||
| in string | ||
| exp bool | ||
| }{ | ||
| {"Select crdb_internal.commit_with_causality_token()", true}, | ||
| {"select crdb_inteRnal.comMit_wiTh_cauSality_toKen()", true}, | ||
| {`select "crdb_internal".comMit_wiTh_cauSality_toKen()`, true}, | ||
| {"select crdb_inteRnal.comMit_wiTh_cauSality_toKen(), 1", false}, | ||
| {"select crdb_internal.commit_with_causality_token() from crdb_internal.ranges_no_leases", false}, | ||
| {"select crdb_internal.commit_with_causality_token() from generate_series(1, 100)", false}, | ||
| {`select distinct "crdb_internal".comMit_wiTh_cauSality_toKen()`, false}, | ||
| {`select "crdb_inteRnal".comMit_wiTh_cauSality_toKen()`, false}, | ||
| {"(select crdb_internal.commit_with_causality_token())", false}, | ||
| {"(select crdb_inteRnal.comMit_wiTh_cauSality_toKen())", false}, | ||
| {`(select "crdb_internal".comMit_wiTh_cauSality_toKen())`, false}, | ||
| {`(select "crdb_inteRnal".comMit_wiTh_cauSality_toKen())`, false}, | ||
| {`((select "crdb_internal".comMit_wiTh_cauSality_toKen()))`, false}, | ||
| {`SELECT ((select "crdb_internal".comMit_wiTh_cauSality_toKen()))`, false}, | ||
| {"SELECT crdb_internal.commit_with_causality_token() FOR UPDATE ", false}, | ||
| {"Select crdb_internal.commit_with_causality_token", false}, | ||
| {"with a as (select 1) select crdb_internal.commit_with_causality_token()", false}, | ||
| {"(select crdb_internal.commit_with_causality_token() limit 0)", false}, | ||
| {"(select crdb_internal.commit_with_causality_token()) limit 0", false}, | ||
| {"select crdb_internal.commit_with_causality_token() limit 0", false}, | ||
| {"(select crdb_internal.commit_with_causality_token() where true)", false}, | ||
| {"select crdb_internal.commit_with_causality_token() where true", false}, | ||
| {"select crdb_internal.commit_with_causality_token() having true", false}, | ||
| {"select crdb_internal.commit_with_causality_token() order by 1", false}, | ||
| {"(select crdb_internal.commit_with_causality_token()) order by 1", false}, | ||
| {"(select crdb_internal.commit_with_causality_token() order by 1)", false}, | ||
| } { | ||
| t.Run(tc.in, func(t *testing.T) { | ||
| stmts, err := parser.Parse(tc.in) | ||
| require.NoError(t, err) | ||
| require.Len(t, stmts, 1) | ||
| require.Equalf( | ||
| t, tc.exp, isSelectCommitWithCausalityToken(stmts[0].AST), | ||
| "%s", stmts[0].SQL, | ||
| ) | ||
| }) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,114 @@ | ||
| // Copyright 2022 The Cockroach Authors. | ||
| // | ||
| // Use of this software is governed by the Business Source License | ||
| // included in the file licenses/BSL.txt. | ||
| // | ||
| // As of the Change Date specified in that file, in accordance with | ||
| // the Business Source License, use of this software will be governed | ||
| // by the Apache License, Version 2.0, included in the file | ||
| // licenses/APL.txt. | ||
|
|
||
| package sql | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo" | ||
| "github.com/cockroachdb/cockroach/pkg/sql/sem/builtins" | ||
| "github.com/cockroachdb/cockroach/pkg/sql/sem/catconstants" | ||
| "github.com/cockroachdb/cockroach/pkg/sql/sem/eval" | ||
| "github.com/cockroachdb/cockroach/pkg/sql/sem/tree" | ||
| "github.com/cockroachdb/cockroach/pkg/sql/types" | ||
| "github.com/cockroachdb/cockroach/pkg/util/fsm" | ||
| ) | ||
|
|
||
| // handleCommitWithCausalityToken deals with the special statement | ||
| // SELECT commit_with_causality_token while in the OpenState of an | ||
| // explicit transaction. | ||
| func (ex *connExecutor) handleCommitWithCausalityToken( | ||
| ctx context.Context, res RestrictedCommandResult, | ||
| ) (fsm.Event, fsm.EventPayload, error) { | ||
| res.ResetStmtType((*tree.CommitTransaction)(nil)) | ||
| err := ex.commitSQLTransactionInternal(ctx) | ||
| if err == nil { | ||
|
|
||
| res.SetColumns(ctx, colinfo.ResultColumns{ | ||
| { | ||
| Name: "causality_token", | ||
| Typ: types.Decimal, | ||
| Hidden: false, | ||
| }, | ||
| }) | ||
| if err := res.AddRow(ctx, tree.Datums{ | ||
| eval.TimestampToDecimalDatum(ex.planner.Txn().CommitTimestamp()), | ||
| }); err != nil { | ||
| return nil, nil, err | ||
| } | ||
|
|
||
| // If we have a SAVEPOINT cockroach_restart, then we need to note that | ||
| // fact now, as the SAVEPOINT stack will be destroyed as the state | ||
| // machine moves into COMMIT. This state in extraTxnState will be cleaned | ||
| // up as we process any statement in CommitWait. | ||
| if entry, _ := ex.extraTxnState.savepoints.find( | ||
| commitOnReleaseSavepointName, | ||
| ); entry != nil && entry.commitOnRelease { | ||
| ex.extraTxnState.shouldAcceptReleaseSavepointCockroachRestart = true | ||
| } | ||
|
|
||
| return eventTxnCommittedWithCausalityToken{}, nil, nil | ||
| } | ||
|
|
||
| // Committing the transaction failed. We'll go to state RestartWait if | ||
| // it's a retriable error, or to state RollbackWait otherwise. | ||
| if errIsRetriable(err) { | ||
| rc, canAutoRetry := ex.getRewindTxnCapability() | ||
| ev := eventRetriableErr{ | ||
| IsCommit: fsm.FromBool(false /* isCommit */), | ||
| CanAutoRetry: fsm.FromBool(canAutoRetry), | ||
| } | ||
| payload := eventRetriableErrPayload{err: err, rewCap: rc} | ||
| return ev, payload, nil | ||
| } | ||
|
|
||
| ev := eventNonRetriableErr{IsCommit: fsm.FromBool(false)} | ||
| payload := eventNonRetriableErrPayload{err: err} | ||
| return ev, payload, nil | ||
| } | ||
|
|
||
| // isSelectWithCausalityToken returns true if the statement is exactly the | ||
| // following, modulo capitalization: | ||
| // | ||
| // SELECT crdb_internal.commit_with_causality_token | ||
| // | ||
| func isSelectCommitWithCausalityToken(ast tree.Statement) bool { | ||
| sel, ok := ast.(*tree.Select) | ||
| if !ok { | ||
| return false | ||
| } | ||
| selStmt := sel.Select | ||
| if sel.With != nil || sel.Locking != nil || sel.Limit != nil || sel.OrderBy != nil { | ||
| return false | ||
| } | ||
| // We intentionally don't open up ParenSelect clauses. | ||
| sc, ok := selStmt.(*tree.SelectClause) | ||
| if !ok { | ||
| return false | ||
| } | ||
| // TODO(ajwerner): Find a more exhaustive way to do this. | ||
| if len(sc.From.Tables) != 0 || len(sc.Exprs) != 1 || sc.Distinct || | ||
| sc.Where != nil || sc.GroupBy != nil || sc.Having != nil || sc.Window != nil || | ||
| sc.From.AsOf.Expr != nil { | ||
| return false | ||
| } | ||
| funcExpr, isFuncExpr := sc.Exprs[0].Expr.(*tree.FuncExpr) | ||
| if !isFuncExpr || len(funcExpr.Exprs) != 0 { | ||
| return false | ||
| } | ||
| name, isName := funcExpr.Func.FunctionReference.(*tree.UnresolvedName) | ||
| if !isName || name.NumParts != 2 || | ||
| name.Parts[1] != catconstants.CRDBInternalSchemaName || | ||
| name.Parts[0] != builtins.CommitWithCausalityTokenName { | ||
| return false | ||
| } | ||
| return true | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.