sql: use txn timestamp when comparing time in table leases

Prior to this change we used the clocks current time when
deciding if a lease expired. We would also store the current
time as the modification time for the table descriptor.
It was much harder to see if we could hit weird race conditions
and it was almost impossible to reasona around correctness.

This change moves all the table lease logic to use the txn
timestamp when writing timestamps and when comparing timestamps
for expiration. It is a lot easier to reason about the code
after this change.

This change also returns a retryable error when a transaction
encounters two different versions of the same table descriptor.

fixes cockroachdb#2948

pkg/internal/client/txn.go

@@ -980,7 +980,7 @@ func firstWriteIndex(ba roachpb.BatchRequest) (int, *roachpb.Error) {
 // the txn. If the error is not a RetryableTxnError, then this is a no-op. For a
 // retryable error, the Transaction proto is either initialized with the updated
 // proto from the error, or a new Transaction proto is initialized.
-// This needs to be called when a SQL hits a retryable error outside of the
+// This needs to be called when a SQL statement hits a retryable error outside of the
 // TxnCoordSender, like in distSQL or retryable errors in SQL logic.
 func (txn *Txn) UpdateStateOnRetryableErr(ctx context.Context, pErr roachpb.Error) {
 	txn.mu.Lock()

pkg/kv/txn_coord_sender.go

@@ -886,6 +886,8 @@ func (tc *TxnCoordSender) updateState(
 					tc.metrics.RestartsSerializable.Inc(1)
 				case roachpb.RETRY_POSSIBLE_REPLAY:
 					tc.metrics.RestartsPossibleReplay.Inc(1)
+				case roachpb.RETRY_TABLE_FROM_FUTURE, roachpb.RETRY_TABLE_VERSION_CHANGE:
+					panic("illegal transaction retry error")
 				}
 			}
 			newTxn = roachpb.PrepareTransactionForRetry(ctx, pErr, ba.UserPriority)

pkg/roachpb/errors.proto

@@ -126,6 +126,11 @@ enum TransactionRetryReason {
   // that the write timestamp on the table descriptor is after the
   // transaction timestamp.
   RETRY_TABLE_FROM_FUTURE = 5;
+  // A transaction using a table descriptor with a specific version
+  // for some commands, sees a new table version that it cannot
+  // use for more commands. Retry until it can run the entire
+  // transaction with one table version.
+  RETRY_TABLE_VERSION_CHANGE = 6;
 }
 
 // A TransactionRetryError indicates that the transaction must be

pkg/sql/crdb_internal.go

@@ -221,7 +221,7 @@ CREATE TABLE crdb_internal.leases (
   parent_id   INT NOT NULL,
   expiration  TIMESTAMP NOT NULL,
   released    BOOL NOT NULL,
-  deleted     BOOL NOT NULL
+  dropped     BOOL NOT NULL
 );
 `,
 	populate: func(_ context.Context, p *planner, addRow func(...parser.Datum) error) error {
@@ -238,7 +238,7 @@ CREATE TABLE crdb_internal.leases (
 				ts.mu.Lock()
 				defer ts.mu.Unlock()
 
-				deleted := parser.MakeDBool(parser.DBool(ts.deleted))
+				dropped := parser.MakeDBool(parser.DBool(ts.dropped))
 
 				for _, state := range ts.active.data {
 					if !userCanSeeDescriptor(&state.TableDescriptor, p.session.User) {
@@ -252,7 +252,7 @@ CREATE TABLE crdb_internal.leases (
 						parser.NewDInt(parser.DInt(int64(state.ParentID))),
 						&expCopy,
 						parser.MakeDBool(parser.DBool(state.released)),
-						deleted,
+						dropped,
 					); err != nil {
 						return err
 					}

pkg/sql/drop.go

@@ -739,7 +739,6 @@ func (n *dropTableNode) Start(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
-
 		// Log a Drop Table event for this table. This is an auditable log event
 		// and is recorded in the same transaction as the table descriptor
 		// update.