Merge #154186

154186: catalog/lease: fix a bug preventing timestamp refreshes r=fqazi a=fqazi

Previously, when we added support for locked leased manager timestamps, which are disabled by default we introduced a new bug that prevented the timestamp from moving forward (even with this feature disabled). To address this, this patch modifies the leased descriptors to allow the timestamps to move forward again.

Note: This patch also temporarily skips the flaky test: TestLeaseManagerLockedTimestampBasic

Fixes: #153830
Fixes: #153834
Fixes: #153835
Fixes: #154118
Fixes: #154124

Release note: None

Co-authored-by: Faizan Qazi <faizan@cockroachlabs.com>

Author: craig[bot]

pkg/sql/catalog/descs/leased_descriptors.go

@@ -181,7 +181,12 @@ func (ld *leasedDescriptors) maybeAssertExternalRowDataTS(desc catalog.Descripto
 
 // maybeInitReadTimestamp selects a read timestamp for the lease manager.
 func (ld *leasedDescriptors) maybeInitReadTimestamp(txn deadlineHolder) {
-	if ld.leaseTimestampSet {
+	// Refresh the leased timestamp if the read timestamp has changed on us
+	// or if it hasn't been populated yet.
+	// TODO (fqazi): For locked read timestamps inside leasing,
+	// we will need to have extra logic to ensure this is safe.
+	if ld.leaseTimestampSet &&
+		ld.leaseTimestamp.GetBaseTimestamp() == txn.ReadTimestamp() {
 		return
 	}
 	readTimestamp := txn.ReadTimestamp()

pkg/sql/catalog/lease/lease_internal_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness/sqllivenesstestutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/admission"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
@@ -1739,6 +1740,8 @@ CREATE TABLE t.test (k CHAR PRIMARY KEY, v CHAR);
 func TestLeaseManagerLockedTimestampBasic(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
+	// Skip until future PRs fix this test.
+	skip.WithIssue(t, 153826)
 
 	var blockUpdates atomic.Bool
 	var blockCheckPoint atomic.Bool