sql: add SHOW COMMIT TIMESTAMP to retrieve a causality token

Fixes cockroachdb#79591

Relates to cockroachdb#7945

Release note (sql change): A new sql statement `SHOW COMMIT TIMESTAMP` has been
added. This statement can be used to retrieve the commit timestamp of the
current explicit transaction, current multi-statement implicit transaction, or
previous transaction. The statement may be used in a variety of settings to
maximize its utility in the face of connection pooling.

When used as a part of an explicit transaction, the statement implicitly
commits the transaction internally before being able to return a causality
token. This is similar to the `RELEASE cockroach_restart` behavior; after
issuing this statement, commands to the transaction will be rejected until
`COMMIT` is issued.

When used as part of a multi-statement implicit transaction, the statement
must be the final statement. If it occurs in the middle of a multi-statement
implicit transaction, it will be rejected with an error.

When sent as a stand-alone single-statement implicit transaction, it will
return the commit timestamp of the previously committed transaction. If there
was no transaction on the connection, or the previous transaction did not
commit (either rolled back or encountered an error), the command will fail with
an error code 25000 (`InvalidTransactionState`).

docs/generated/sql/bnf/BUILD.bazel

@@ -208,6 +208,7 @@ FILES = [
     "show_backup",
     "show_cluster_setting",
     "show_columns_stmt",
+    "show_commit_timestamp_stmt",
     "show_constraints_stmt",
     "show_create_stmt",
     "show_create_schedules_stmt",

docs/generated/sql/bnf/show_commit_timestamp_stmt.bnf

@@ -0,0 +1,2 @@
+show_commit_timestamp_stmt ::=
+	'SHOW' 'COMMIT' 'TIMESTAMP'

docs/generated/sql/bnf/stmt_block.bnf

@@ -30,6 +30,7 @@ stmt_without_legacy_transaction ::=
 	| fetch_cursor_stmt
 	| move_cursor_stmt
 	| unlisten_stmt
+	| show_commit_timestamp_stmt
 
 legacy_transaction_stmt ::=
 	legacy_begin_stmt
@@ -160,6 +161,9 @@ unlisten_stmt ::=
 	'UNLISTEN' type_name
 	| 'UNLISTEN' '*'
 
+show_commit_timestamp_stmt ::=
+	'SHOW' 'COMMIT' 'TIMESTAMP'
+
 legacy_begin_stmt ::=
 	'BEGIN' opt_transaction begin_transaction
 

docs/generated/sql/bnf/stmt_without_legacy_transaction.bnf

@@ -21,3 +21,4 @@ stmt_without_legacy_transaction ::=
 	| fetch_cursor_stmt
 	| move_cursor_stmt
 	| unlisten_stmt
+	| show_commit_timestamp_stmt

pkg/gen/bnf.bzl

@@ -208,6 +208,7 @@ BNF_SRCS = [
   "//docs/generated/sql/bnf:show_backup.bnf",
   "//docs/generated/sql/bnf:show_cluster_setting.bnf",
   "//docs/generated/sql/bnf:show_columns_stmt.bnf",
+  "//docs/generated/sql/bnf:show_commit_timestamp_stmt.bnf",
   "//docs/generated/sql/bnf:show_constraints_stmt.bnf",
   "//docs/generated/sql/bnf:show_create_external_connections_stmt.bnf",
   "//docs/generated/sql/bnf:show_create_schedules_stmt.bnf",

pkg/gen/diagrams.bzl

@@ -206,6 +206,7 @@ DIAGRAMS_SRCS = [
   "//docs/generated/sql/bnf:show_backup.html",
   "//docs/generated/sql/bnf:show_cluster_setting.html",
   "//docs/generated/sql/bnf:show_columns.html",
+  "//docs/generated/sql/bnf:show_commit_timestamp.html",
   "//docs/generated/sql/bnf:show_constraints.html",
   "//docs/generated/sql/bnf:show_create.html",
   "//docs/generated/sql/bnf:show_create_external_connections.html",

pkg/gen/docs.bzl

@@ -220,6 +220,7 @@ DOCS_SRCS = [
   "//docs/generated/sql/bnf:show_backup.bnf",
   "//docs/generated/sql/bnf:show_cluster_setting.bnf",
   "//docs/generated/sql/bnf:show_columns_stmt.bnf",
+  "//docs/generated/sql/bnf:show_commit_timestamp_stmt.bnf",
   "//docs/generated/sql/bnf:show_constraints_stmt.bnf",
   "//docs/generated/sql/bnf:show_create_external_connections_stmt.bnf",
   "//docs/generated/sql/bnf:show_create_schedules_stmt.bnf",

pkg/sql/BUILD.bazel

@@ -44,6 +44,7 @@ go_library(
         "conn_executor_exec.go",
         "conn_executor_prepare.go",
         "conn_executor_savepoints.go",
+        "conn_executor_show_commit_timestamp.go",
         "conn_fsm.go",
         "conn_io.go",
         "constraint.go",

pkg/sql/catalog/colinfo/result_columns.go

@@ -142,6 +142,11 @@ var ExplainPlanColumns = ResultColumns{
 	{Name: "info", Typ: types.String},
 }
 
+// ShowCommitTimestampColumns are the result columns of SHOW COMMIT TIMESTAMP.
+var ShowCommitTimestampColumns = ResultColumns{
+	{Name: "commit_timestamp", Typ: types.Decimal},
+}
+
 // ShowTraceColumns are the result columns of a SHOW [KV] TRACE statement.
 var ShowTraceColumns = ResultColumns{
 	{Name: "timestamp", Typ: types.TimestampTZ},

pkg/sql/conn_executor.go

@@ -1357,7 +1357,7 @@ type connExecutor struct {
 		txnFinishClosure struct {
 			// txnStartTime is the time that the transaction started.
 			txnStartTime time.Time
-			// implicit is whether or not the transaction was implicit.
+			// implicit is whether the transaction was implicit.
 			implicit bool
 		}
 
@@ -1413,6 +1413,13 @@ type connExecutor struct {
 		rowsWrittenLogged bool
 		rowsReadLogged    bool
 
+		// shouldAcceptReleaseCockroachRestartInCommitWait is set to true if the
+		// transaction had SAVEPOINT cockroach_restart installed at the time that
+		// SHOW COMMIT TIMESTAMP was executed to commit the transaction. If so, the
+		// connExecutor will permit one invocation of RELEASE SAVEPOINT
+		// cockroach_restart while in the CommitWait state.
+		shouldAcceptReleaseCockroachRestartInCommitWait bool
+
 		// hasAdminRole is used to cache if the user running the transaction
 		// has admin privilege. hasAdminRoleCache is set for the first statement
 		// in a transaction.
@@ -1537,6 +1544,11 @@ type connExecutor struct {
 	// totalActiveTimeStopWatch tracks the total active time of the session.
 	// This is defined as the time spent executing transactions and statements.
 	totalActiveTimeStopWatch *timeutil.StopWatch
+
+	// previousTransactionCommitTimestamp is the timestamp of the previous
+	// transaction which committed. It is zero-valued when there is a transaction
+	// open or the previous transaction did not successfully commit.
+	previousTransactionCommitTimestamp hlc.Timestamp
 }
 
 // ctxHolder contains a connection's context and, while session tracing is
@@ -1923,7 +1935,9 @@ func (ex *connExecutor) execCmd() error {
 			// The behavior is configurable, in case users want to preserve the
 			// behavior from v21.2 and earlier.
 			implicitTxnForBatch := ex.sessionData().EnableImplicitTransactionForBatchStatements
-			canAutoCommit := ex.implicitTxn() && (tcmd.LastInBatch || !implicitTxnForBatch)
+			canAutoCommit := ex.implicitTxn() &&
+				(tcmd.LastInBatchBeforeShowCommitTimestamp ||
+					tcmd.LastInBatch || !implicitTxnForBatch)
 			ev, payload, err = ex.execStmt(
 				ctx, tcmd.Statement, nil /* prepared */, nil /* pinfo */, stmtRes, canAutoCommit,
 			)

pkg/sql/conn_executor_exec.go

@@ -120,7 +120,7 @@ func (ex *connExecutor) execStmt(
 		// Note: when not using explicit transactions, we go through this transition
 		// for every statement. It is important to minimize the amount of work and
 		// allocations performed up to this point.
-		ev, payload = ex.execStmtInNoTxnState(ctx, ast)
+		ev, payload = ex.execStmtInNoTxnState(ctx, ast, res)
 
 	case stateOpen:
 		err = ex.execWithProfiling(ctx, ast, prepared, func(ctx context.Context) error {
@@ -587,6 +587,10 @@ func (ex *connExecutor) execStmtInOpenState(
 		ev, payload := ex.execRollbackToSavepointInOpenState(ctx, s, res)
 		return ev, payload, nil
 
+	case *tree.ShowCommitTimestamp:
+		ev, payload := ex.execShowCommitTimestampInOpenState(ctx, s, res, canAutoCommit)
+		return ev, payload, nil
+
 	case *tree.Prepare:
 		// This is handling the SQL statement "PREPARE". See execPrepare for
 		// handling of the protocol-level command for preparing statements.
@@ -774,6 +778,9 @@ func (ex *connExecutor) execStmtInOpenState(
 // the timestamps of the transaction accordingly.
 func (ex *connExecutor) handleAOST(ctx context.Context, stmt tree.Statement) error {
 	if _, isNoTxn := ex.machine.CurState().(stateNoTxn); isNoTxn {
+		if _, ok := stmt.(*tree.ShowCommitTimestamp); ok {
+			return nil
+		}
 		return errors.AssertionFailedf(
 			"cannot handle AOST clause without a transaction",
 		)
@@ -1637,7 +1644,7 @@ var eventStartExplicitTxn fsm.Event = eventTxnStart{ImplicitTxn: fsm.False}
 // the cursor is not advanced. This means that the statement will run again in
 // stateOpen, at each point its results will also be flushed.
 func (ex *connExecutor) execStmtInNoTxnState(
-	ctx context.Context, ast tree.Statement,
+	ctx context.Context, ast tree.Statement, res RestrictedCommandResult,
 ) (_ fsm.Event, payload fsm.EventPayload) {
 	switch s := ast.(type) {
 	case *tree.BeginTransaction:
@@ -1660,6 +1667,8 @@ func (ex *connExecutor) execStmtInNoTxnState(
 				historicalTs,
 				ex.transitionCtx,
 				ex.QualityOfService())
+	case *tree.ShowCommitTimestamp:
+		return ex.execShowCommitTimestampInNoTxnState(ctx, s, res)
 	case *tree.CommitTransaction, *tree.ReleaseSavepoint,
 		*tree.RollbackTransaction, *tree.SetTransaction, *tree.Savepoint:
 		return ex.makeErrEvent(errNoTransactionInProgress, ast)
@@ -1782,7 +1791,14 @@ func (ex *connExecutor) execStmtInCommitWaitState(
 			ex.incrementExecutedStmtCounter(ast)
 		}
 	}()
-	switch ast.(type) {
+	acceptReleaseSavepoint := ex.shouldAcceptReleaseSavepointInCommitWait()
+	switch s := ast.(type) {
+	case *tree.ReleaseSavepoint:
+		if acceptReleaseSavepoint && s.Savepoint == commitOnReleaseSavepointName {
+			return nil, nil
+		}
+	case *tree.ShowCommitTimestamp:
+		return ex.execShowCommitTimestampInCommitWaitState(ctx, s, res)
 	case *tree.CommitTransaction, *tree.RollbackTransaction:
 		// Reply to a rollback with the COMMIT tag, by analogy to what we do when we
 		// get a COMMIT in state Aborted.
@@ -1795,13 +1811,23 @@ func (ex *connExecutor) execStmtInCommitWaitState(
 				return nil
 			},
 		)
-	default:
-		ev = eventNonRetriableErr{IsCommit: fsm.False}
-		payload = eventNonRetriableErrPayload{
+	}
+	return eventNonRetriableErr{IsCommit: fsm.False},
+		eventNonRetriableErrPayload{
 			err: sqlerrors.NewTransactionCommittedError(),
 		}
-		return ev, payload
-	}
+}
+
+// shouldAcceptReleaseCockroachRestartInCommitWait check if the statement
+// RELEASE SAVEPOINT cockroach_restart should be accepted while in the
+// CommitWait state because such a savepoint had been installed when
+// the transaction was committed via SHOW COMMIT TIMESTAMP. The act of
+// checking whether the statement should be accepted also prevents any
+// future such statement from being accepted.
+func (ex *connExecutor) shouldAcceptReleaseSavepointInCommitWait() (shouldAccept bool) {
+	shouldAccept = ex.extraTxnState.shouldAcceptReleaseCockroachRestartInCommitWait
+	ex.extraTxnState.shouldAcceptReleaseCockroachRestartInCommitWait = false
+	return shouldAccept
 }
 
 // runObserverStatement executes the given observer statement.
@@ -2224,6 +2250,8 @@ func (ex *connExecutor) onTxnFinish(ctx context.Context, ev txnEvent) {
 			}
 			ex.server.ServerMetrics.StatsMetrics.DiscardedStatsCount.Inc(1)
 		}
+		// If we have a commitTimestamp, we should use it.
+		ex.previousTransactionCommitTimestamp.Forward(ev.commitTimestamp)
 	}
 }
 
@@ -2287,6 +2315,8 @@ func (ex *connExecutor) recordTransactionStart(txnID uuid.UUID) {
 	ex.extraTxnState.shouldExecuteOnTxnRestart = true
 
 	ex.statsCollector.StartTransaction()
+
+	ex.previousTransactionCommitTimestamp = hlc.Timestamp{}
 }
 
 func (ex *connExecutor) recordTransactionFinish(

pkg/sql/conn_executor_prepare.go

@@ -40,7 +40,13 @@ func (ex *connExecutor) execPrepare(
 	// descriptors for type checking. This implicit txn will be open until
 	// the Sync message is handled.
 	if _, isNoTxn := ex.machine.CurState().(stateNoTxn); isNoTxn {
-		return ex.beginImplicitTxn(ctx, parseCmd.AST)
+		// The one exception is that we must not open a new transaction when
+		// preparing SHOW COMMIT TIMESTAMP. If we did, it would destroy the
+		// information about the previous transaction. We expect to execute
+		// this command in NoTxn.
+		if _, ok := parseCmd.AST.(*tree.ShowCommitTimestamp); !ok {
+			return ex.beginImplicitTxn(ctx, parseCmd.AST)
+		}
 	} else if _, isAbortedTxn := ex.machine.CurState().(stateAborted); isAbortedTxn {
 		if !ex.isAllowedInAbortedTxn(parseCmd.AST) {
 			return retErr(sqlerrors.NewTransactionAbortedError("" /* customMsg */))
@@ -315,7 +321,12 @@ func (ex *connExecutor) execBind(
 	// SQL EXECUTE command (which also needs to bind and resolve types) is
 	// handled separately in conn_executor_exec.
 	if _, isNoTxn := ex.machine.CurState().(stateNoTxn); isNoTxn {
-		return ex.beginImplicitTxn(ctx, ps.AST)
+		// The one critical exception is that we must not open a transaction when
+		// executing SHOW COMMIT TIMESTAMP as it would destroy the information
+		// about the previously committed transaction.
+		if _, ok := ps.AST.(*tree.ShowCommitTimestamp); !ok {
+			return ex.beginImplicitTxn(ctx, ps.AST)
+		}
 	} else if _, isAbortedTxn := ex.machine.CurState().(stateAborted); isAbortedTxn {
 		if !ex.isAllowedInAbortedTxn(ps.AST) {
 			return retErr(sqlerrors.NewTransactionAbortedError("" /* customMsg */))