opt: lease descriptors with dependency digest

Previously, we completed avoided leasing when checking if the plan was
safe to use using the dependency digest. This is safe while no schema
change is occurring, however in middle of active schema changes the lack
of leasing can cause executing queries to make progress with stale
descriptors. If an add column is happening for example, we could be
writing to the wrong index for a given point in time. To address this,
this patch makes dependency digest code path to lease the required tables.

Fixes: #139800

Release note: None

pkg/sql/opt/BUILD.bazel

@@ -31,6 +31,7 @@ go_library(
         "//pkg/sql/catalog/colinfo",
         "//pkg/sql/catalog/descpb",
         "//pkg/sql/catalog/multiregion",
+        "//pkg/sql/catalog/typedesc",
         "//pkg/sql/opt/cat",
         "//pkg/sql/opt/partition",
         "//pkg/sql/pgwire/pgcode",

pkg/sql/opt/cat/catalog.go

@@ -256,6 +256,10 @@ type Catalog interface {
 	// exactly the same.
 	GetDependencyDigest() DependencyDigest
 
+	// LeaseByStableID leases out a descriptor by the stable ID, which will block
+	// schema changes. The underlying schema object is not returned.
+	LeaseByStableID(ctx context.Context, id StableID) error
+
 	// GetRoutineOwner returns the username.SQLUsername of the routine's
 	// (specified by routineOid) owner.
 	GetRoutineOwner(ctx context.Context, routineOid oid.Oid) (username.SQLUsername, error)

pkg/sql/opt/metadata.go

@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/multiregion"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/typedesc"
 	"github.com/cockroachdb/cockroach/pkg/sql/opt/cat"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
@@ -376,6 +377,33 @@ func (md *Metadata) dependencyDigestEquals(currentDigest *cat.DependencyDigest)
 	return currentDigest.Equal(&md.digest.depDigest)
 }
 
+// leaseObjectsInMetaData ensures that all references within this metadata
+// are leased to prevent schema changes from modifying the underlying objects
+// excessively.
+func (md *Metadata) leaseObjectsInMetaData(ctx context.Context, optCatalog cat.Catalog) error {
+	for id := range md.dataSourceDeps {
+		if err := optCatalog.LeaseByStableID(ctx, id); err != nil {
+			return err
+		}
+	}
+	for id := range md.routineDeps {
+		if err := optCatalog.LeaseByStableID(ctx, id); err != nil {
+			return err
+		}
+	}
+	for oid := range md.userDefinedTypes {
+		id := typedesc.UserDefinedTypeOIDToID(oid)
+		// Not a user defined type.
+		if id == catid.InvalidDescID {
+			continue
+		}
+		if err := optCatalog.LeaseByStableID(ctx, cat.StableID(id)); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // CheckDependencies resolves (again) each database object on which this
 // metadata depends, in order to check the following conditions:
 //  1. The object has not been modified.
@@ -402,7 +430,12 @@ func (md *Metadata) CheckDependencies(
 		evalCtx.AsOfSystemTime == nil &&
 		!evalCtx.Txn.ReadTimestampFixed() &&
 		md.dependencyDigestEquals(&currentDigest) {
-		return true, nil
+		// Lease the underlying descriptors for this metadata. If we fail to lease
+		// any descriptors attempt to resolve them by name through the more expensive
+		// code path below.
+		if err := md.leaseObjectsInMetaData(ctx, optCatalog); err == nil {
+			return true, nil
+		}
 	}
 
 	// Check that no referenced data sources have changed.

pkg/sql/opt/testutils/testcat/test_catalog.go

@@ -480,6 +480,11 @@ func (tc *Catalog) GetDependencyDigest() cat.DependencyDigest {
 	}
 }
 
+// LeaseByStableID does not do anything since no leasing is used here.
+func (tc *Catalog) LeaseByStableID(_ context.Context, _ cat.StableID) error {
+	return nil
+}
+
 // ExecuteMultipleDDL parses the given semicolon-separated DDL SQL statements
 // and applies each of them to the test catalog.
 func (tc *Catalog) ExecuteMultipleDDL(sql string) error {

pkg/sql/opt_catalog.go

@@ -510,6 +510,14 @@ func (oc *optCatalog) GetCurrentUser() username.SQLUsername {
 	return oc.planner.User()
 }
 
+// LeaseByStableID is part of the cat.Catalog interface.
+func (oc *optCatalog) LeaseByStableID(ctx context.Context, stableID cat.StableID) error {
+	// Lease all the data sources, so that schema changes cannot move forward
+	// on the current version.
+	_, err := oc.planner.LookupTableByID(ctx, descpb.ID(stableID))
+	return err
+}
+
 // GetDependencyDigest is part of the cat.Catalog interface.
 func (oc *optCatalog) GetDependencyDigest() cat.DependencyDigest {
 	// The stats cache may not be setup in some tests like