Merge #145462 #145494

145462: catalog/lease: purge old versions on range feed recovery r=fqazi a=fqazi

catalog/lease: purge old versions on range feed recovery

Previously, the lease manager range feed recovery logic would pick up
newer versions of descriptors after a range feed failure. However, it
would not purge old versions, which could lead to old versions still
being in use. To address this, this patch always purges old versions
when refreshing during a range feed failure.

Fixes: #145422

Release note (bug fix): Addresses a bug that can lead to hung schema
changes after recovery from availability issues.

This patch also fixes TestLeaseDescriptorRangeFeedFailure which did not work correctly when the range feed recovery logic was added.

145494: sql: fix telemetry for CREATE STATISTICS r=DrewKimball,msbutler a=michae2

My change in #144316 accidentally coupled telemetry for CREATE STATISTICS with cluster setting
`sql.stats.automatic_job_check_before_creating_job`. Michael even pointed this out, but I failed to understand! (Sorry Michael!)

Decouple telemetry from the cluster setting, so that telemetry behavior matches how it was before #144316.

Epic: None

Release note: None

Co-authored-by: Faizan Qazi <[email protected]>
Co-authored-by: Michael Erickson <[email protected]>

Author: 3 people

pkg/sql/catalog/lease/helpers_test.go

@@ -290,7 +290,7 @@ func (m *Manager) Publish(
 }
 
 func (m *Manager) TestingRefreshSomeLeases(ctx context.Context) {
-	m.refreshSomeLeases(ctx, false /*refreshAll*/)
+	m.refreshSomeLeases(ctx, false /*refreshAndPurgeAllDescriptors*/)
 }
 
 func (m *Manager) TestingDescriptorStateIsNil(id descpb.ID) bool {

pkg/sql/catalog/lease/lease.go

@@ -1886,7 +1886,7 @@ func (m *Manager) RunBackgroundLeasingTask(ctx context.Context) {
 					// If the range feed recovers after a failure, re-read all
 					// descriptors.
 					if refreshAllDescriptors {
-						m.refreshSomeLeases(ctx, true /*refreshAll*/)
+						m.refreshSomeLeases(ctx, true /*refreshAndPurgeAllDescriptors*/)
 					}
 				}
 				rangeFeedProgressWatchDogTimeout,
@@ -1895,7 +1895,7 @@ func (m *Manager) RunBackgroundLeasingTask(ctx context.Context) {
 			case err := <-m.rangefeedErrCh:
 				log.Warningf(ctx, "lease rangefeed failed with error: %s", err.Error())
 				m.handleRangeFeedError(ctx)
-				m.refreshSomeLeases(ctx, true /*refreshAll*/)
+				m.refreshSomeLeases(ctx, true /*refreshAndPurgeAllDescriptors*/)
 			case <-refreshTimer.C:
 				refreshTimer.Reset(getRefreshTimerDuration() / 2)
 
@@ -1999,8 +1999,9 @@ func (m *Manager) cleanupExpiredSessionLeases(ctx context.Context) {
 	}
 }
 
-// Refresh some of the current leases.
-func (m *Manager) refreshSomeLeases(ctx context.Context, includeAll bool) {
+// Refresh some of the current leases. If refreshAndPurgeAllDescriptors is set,
+// then all descriptors are refreshed, and old versions are purged.
+func (m *Manager) refreshSomeLeases(ctx context.Context, refreshAndPurgeAllDescriptors bool) {
 	limit := leaseRefreshLimit.Get(&m.storage.settings.SV)
 	if limit <= 0 {
 		return
@@ -2013,7 +2014,7 @@ func (m *Manager) refreshSomeLeases(ctx context.Context, includeAll bool) {
 		ids := make([]descpb.ID, 0, len(m.mu.descriptors))
 		var i int64
 		for k, desc := range m.mu.descriptors {
-			if i++; i > limit && !includeAll {
+			if i++; i > limit && !refreshAndPurgeAllDescriptors {
 				break
 			}
 			takenOffline := func() bool {
@@ -2048,7 +2049,6 @@ func (m *Manager) refreshSomeLeases(ctx context.Context, includeAll bool) {
 						return
 					}
 				}
-
 				if _, err := acquireNodeLease(ctx, m, id, AcquireBackground); err != nil {
 					log.Errorf(ctx, "refreshing descriptor: %d lease failed: %s", id, err)
 
@@ -2057,7 +2057,7 @@ func (m *Manager) refreshSomeLeases(ctx context.Context, includeAll bool) {
 						if err := purgeOldVersions(
 							ctx, m.storage.db.KV(), id, true /* dropped */, 0 /* minVersion */, m,
 						); err != nil {
-							log.Warningf(ctx, "error purging leases for descriptor %d: %s",
+							log.Warningf(ctx, "error purging leases for descriptor %d: %v",
 								id, err)
 						}
 						func() {
@@ -2067,6 +2067,15 @@ func (m *Manager) refreshSomeLeases(ctx context.Context, includeAll bool) {
 						}()
 					}
 				}
+				if refreshAndPurgeAllDescriptors {
+					// If we are refreshing all descriptors, then we want to purge older versions as
+					// we are doing this operation.
+					err := purgeOldVersions(ctx, m.storage.db.KV(), id, false /* dropped */, 0 /* minVersion */, m)
+					if err != nil {
+						log.Warningf(ctx, "error purging leases for descriptor %d: %v",
+							id, err)
+					}
+				}
 			}); err != nil {
 			log.Infof(ctx, "didnt refresh descriptor: %d lease: %s", id, err)
 			wg.Done()

pkg/sql/catalog/lease/lease_test.go

@@ -3200,30 +3200,31 @@ func TestLeaseDescriptorRangeFeedFailure(t *testing.T) {
 	var srv serverutils.TestClusterInterface
 	var enableAfterStageKnob atomic.Bool
 	var rangeFeedResetChan chan struct{}
+	grp := ctxgroup.WithContext(ctx)
 	srv = serverutils.StartCluster(t, 3, base.TestClusterArgs{
 		ServerArgs: base.TestServerArgs{
 			Settings: settings,
 			Knobs: base.TestingKnobs{
 				SQLDeclarativeSchemaChanger: &scexec.TestingKnobs{
 					BeforeStage: func(p scplan.Plan, stageIdx int) error {
-						// Once this stage completes, we can "resume" the range feed,
-						// so the update is detected.
-						if p.Params.ExecutionPhase == scop.PostCommitPhase &&
-							enableAfterStageKnob.Load() &&
-							strings.Contains(p.Statements[0].Statement, "ALTER TABLE t1 ADD COLUMN j INT DEFAULT 64") {
-							rangeFeedResetChan = srv.ApplicationLayer(1).LeaseManager().(*lease.Manager).TestingSetDisableRangeFeedCheckpointFn(true)
+						// Skip if the knob is disabled
+						if !enableAfterStageKnob.Load() ||
+							!strings.Contains(p.Statements[0].Statement, "ALTER TABLE t1 ADD COLUMN j INT8 DEFAULT 64") ||
+							p.Params.ExecutionPhase != scop.PostCommitPhase {
+							return nil
 						}
-						return nil
-					},
-					AfterStage: func(p scplan.Plan, stageIdx int) error {
 						// Once this stage completes, we can "resume" the range feed,
 						// so the update is detected.
-						if p.Params.ExecutionPhase == scop.PostCommitPhase &&
-							enableAfterStageKnob.Load() &&
-							strings.Contains(p.Statements[0].Statement, "ALTER TABLE t1 ADD COLUMN j INT DEFAULT 64") {
-							<-rangeFeedResetChan
-							srv.ApplicationLayer(1).LeaseManager().(*lease.Manager).TestingSetDisableRangeFeedCheckpointFn(false)
+						if stageIdx == 0 {
+							rangeFeedResetChan = srv.ApplicationLayer(1).LeaseManager().(*lease.Manager).TestingSetDisableRangeFeedCheckpointFn(true)
 							enableAfterStageKnob.Swap(false)
+							grp.Go(func() error {
+								// Once this channel is closed we know for certain the range feed has
+								// recovered. Allow updates so that descriptors are refreshed and purged.
+								<-rangeFeedResetChan
+								srv.ApplicationLayer(1).LeaseManager().(*lease.Manager).TestingSetDisableRangeFeedCheckpointFn(false)
+								return nil
+							})
 						}
 						return nil
 					},
@@ -3260,6 +3261,8 @@ func TestLeaseDescriptorRangeFeedFailure(t *testing.T) {
 			t.Fatal("no error encountered")
 		}
 	}
+	require.NoError(t, grp.Wait())
+	require.Falsef(t, enableAfterStageKnob.Load(), "testing knob was not hit")
 }
 
 // TestLeaseTableWriteFailure is used to ensure that sqlliveness heart-beating

pkg/sql/create_stats.go

@@ -156,16 +156,18 @@ func (n *createStatsNode) runJob(ctx context.Context) error {
 	details := record.Details.(jobspb.CreateStatsDetails)
 
 	jobCheckBefore := automaticJobCheckBeforeCreatingJob.Get(n.p.ExecCfg().SV())
-	if (n.Name == jobspb.AutoStatsName || n.Name == jobspb.AutoPartialStatsName) && jobCheckBefore {
-		// Don't start the job if there is already a CREATE STATISTICS job running.
-		// (To handle race conditions we check this again after the job starts,
-		// but this check is used to prevent creating a large number of jobs that
-		// immediately fail).
-		if err := checkRunningJobs(
-			ctx, nil /* job */, n.p, n.Name == jobspb.AutoPartialStatsName, n.p.ExecCfg().JobRegistry,
-			details.Table.ID,
-		); err != nil {
-			return err
+	if n.Name == jobspb.AutoStatsName || n.Name == jobspb.AutoPartialStatsName {
+		if jobCheckBefore {
+			// Don't start the job if there is already a CREATE STATISTICS job running.
+			// (To handle race conditions we check this again after the job starts,
+			// but this check is used to prevent creating a large number of jobs that
+			// immediately fail).
+			if err := checkRunningJobs(
+				ctx, nil /* job */, n.p, n.Name == jobspb.AutoPartialStatsName, n.p.ExecCfg().JobRegistry,
+				details.Table.ID,
+			); err != nil {
+				return err
+			}
 		}
 	} else {
 		telemetry.Inc(sqltelemetry.CreateStatisticsUseCounter)