Merge #52163 #52408 #52486

52163: sql: implement ConstructOpaque in the new factory r=yuzefovich a=yuzefovich

Addresses: #47473.

Release note: None

52408: sql/catalog/lease: attempt to fix a flakey test r=lucy-zhang a=ajwerner

I think this test was going to fail.

Fixes #52385.

Release note: None

52486: sqlbase,catalogkv: move more kv interactions to catalogkv r=ajwerner a=ajwerner

At this point the only remaining kv interactions in sqlbase are during type
and table interactions though there are also usages through the protoGetter.

This is a minor change in the work to pick apart sqlbase.

Release note: None

Co-authored-by: Yahor Yuzefovich <yahor@cockroachlabs.com>
Co-authored-by: Andrew Werner <ajwerner@cockroachlabs.com>

pkg/ccl/backupccl/restore_job.go

@@ -300,7 +300,7 @@ func WriteDescriptors(
  // Depending on which cluster version we are restoring to, we decide which
  // namespace table to write the descriptor into. This may cause wrong
  // behavior if the cluster version is bumped DURING a restore.
- dKey := sqlbase.MakeDatabaseNameKey(ctx, settings, desc.GetName())
+ dKey := catalogkv.MakeDatabaseNameKey(ctx, settings, desc.GetName())
  b.CPut(dKey.Key(keys.SystemSQLCodec), desc.GetID(), nil)
  }
  for i := range tables {
@@ -344,7 +344,7 @@ func WriteDescriptors(
  // Depending on which cluster version we are restoring to, we decide which
  // namespace table to write the descriptor into. This may cause wrong
  // behavior if the cluster version is bumped DURING a restore.
- tkey := sqlbase.MakeObjectNameKey(
+ tkey := catalogkv.MakeObjectNameKey(
  ctx,
  settings,
  table.GetParentID(),
@@ -369,7 +369,7 @@ func WriteDescriptors(
  ); err != nil {
  return err
  }
- tkey := sqlbase.MakePublicTableNameKey(ctx, settings, typ.ParentID, typ.Name)
+ tkey := catalogkv.MakePublicTableNameKey(ctx, settings, typ.ParentID, typ.Name)
  b.CPut(tkey.Key(keys.SystemSQLCodec), typ.ID, nil)
  }
 
@@ -1115,7 +1115,7 @@ func (r *restoreResumer) publishDescriptors(ctx context.Context) error {
  return err
  }
  newDescriptorChangeJobs = append(newDescriptorChangeJobs, newJobs...)
- existingDescVal, err := sqlbase.ConditionalGetTableDescFromTxn(ctx, txn, r.execCfg.Codec, tbl)
+ existingDescVal, err := catalogkv.ConditionalGetTableDescFromTxn(ctx, txn, r.execCfg.Codec, tbl)
  if err != nil {
  return errors.Wrap(err, "validating table descriptor has not changed")
  }
@@ -1233,11 +1233,11 @@ func (r *restoreResumer) dropTables(ctx context.Context, jr *jobs.Registry, txn
  prev := tableToDrop.Immutable().(sqlbase.TableDescriptor)
  tableToDrop.Version++
  tableToDrop.State = descpb.TableDescriptor_DROP
- err := sqlbase.RemovePublicTableNamespaceEntry(ctx, txn, keys.SystemSQLCodec, tbl.ParentID, tbl.Name)
+ err := catalogkv.RemovePublicTableNamespaceEntry(ctx, txn, keys.SystemSQLCodec, tbl.ParentID, tbl.Name)
  if err != nil {
  return errors.Wrap(err, "dropping tables caused by restore fail/cancel from public namespace")
  }
- existingDescVal, err := sqlbase.ConditionalGetTableDescFromTxn(ctx, txn, r.execCfg.Codec, prev)
+ existingDescVal, err := catalogkv.ConditionalGetTableDescFromTxn(ctx, txn, r.execCfg.Codec, prev)
  if err != nil {
  return errors.Wrap(err, "dropping tables caused by restore fail/cancel")
  }

pkg/ccl/backupccl/restore_planning.go

@@ -297,7 +297,7 @@ func allocateDescriptorRewrites(
  if err := p.ExecCfg().DB.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
  // Check that any DBs being restored do _not_ exist.
  for name := range restoreDBNames {
- found, _, err := sqlbase.LookupDatabaseID(ctx, txn, p.ExecCfg().Codec, name)
+ found, _, err := catalogkv.LookupDatabaseID(ctx, txn, p.ExecCfg().Codec, name)
  if err != nil {
  return err
  }
@@ -352,7 +352,7 @@ func allocateDescriptorRewrites(
  } else {
  var parentID descpb.ID
  {
- found, newParentID, err := sqlbase.LookupDatabaseID(ctx, txn, p.ExecCfg().Codec, targetDB)
+ found, newParentID, err := catalogkv.LookupDatabaseID(ctx, txn, p.ExecCfg().Codec, targetDB)
  if err != nil {
  return err
  }
@@ -417,7 +417,7 @@ func allocateDescriptorRewrites(
  }
 
  // Look up the parent database's ID.
- found, parentID, err := sqlbase.LookupDatabaseID(ctx, txn, p.ExecCfg().Codec, targetDB)
+ found, parentID, err := catalogkv.LookupDatabaseID(ctx, txn, p.ExecCfg().Codec, targetDB)
  if err != nil {
  return err
  }
@@ -433,7 +433,7 @@ func allocateDescriptorRewrites(
  }
 
  // See if there is an existing type with the same name.
- found, id, err := sqlbase.LookupObjectID(ctx, txn, p.ExecCfg().Codec, parentID, keys.PublicSchemaID, typ.Name)
+ found, id, err := catalogkv.LookupObjectID(ctx, txn, p.ExecCfg().Codec, parentID, keys.PublicSchemaID, typ.Name)
  if err != nil {
  return err
  }

pkg/ccl/backupccl/targets.go

@@ -656,7 +656,7 @@ func fullClusterTargets(
 func lookupDatabaseID(
  ctx context.Context, txn *kv.Txn, codec keys.SQLCodec, name string,
 ) (descpb.ID, error) {
- found, id, err := sqlbase.LookupDatabaseID(ctx, txn, codec, name)
+ found, id, err := catalogkv.LookupDatabaseID(ctx, txn, codec, name)
  if err != nil {
  return descpb.InvalidID, err
  }
@@ -676,7 +676,7 @@ func CheckObjectExists(
  parentSchemaID descpb.ID,
  name string,
 ) error {
- found, id, err := sqlbase.LookupObjectID(ctx, txn, codec, parentID, parentSchemaID, name)
+ found, id, err := catalogkv.LookupObjectID(ctx, txn, codec, parentID, parentSchemaID, name)
  if err != nil {
  return err
  }

pkg/ccl/importccl/import_stmt.go

@@ -988,7 +988,7 @@ func prepareExistingTableDescForIngestion(
  // upgrade and downgrade, because IMPORT does not operate in mixed-version
  // states.
  // TODO(jordan,lucy): remove this comment once 19.2 is released.
- existingDesc, err := sqlbase.ConditionalGetTableDescFromTxn(ctx, txn, execCfg.Codec, existing)
+ existingDesc, err := catalogkv.ConditionalGetTableDescFromTxn(ctx, txn, execCfg.Codec, existing)
  if err != nil {
  return nil, errors.Wrap(err, "another operation is currently operating on the table")
  }
@@ -1272,7 +1272,7 @@ func (r *importResumer) publishTables(ctx context.Context, execCfg *sql.Executor
  // upgrade and downgrade, because IMPORT does not operate in mixed-version
  // states.
  // TODO(jordan,lucy): remove this comment once 19.2 is released.
- existingDesc, err := sqlbase.ConditionalGetTableDescFromTxn(ctx, txn, execCfg.Codec, prevTableDesc)
+ existingDesc, err := catalogkv.ConditionalGetTableDescFromTxn(ctx, txn, execCfg.Codec, prevTableDesc)
  if err != nil {
  return errors.Wrap(err, "publishing tables")
  }
@@ -1404,7 +1404,7 @@ func (r *importResumer) dropTables(
  // possible. This is safe since the table data was never visible to users,
  // and so we don't need to preserve MVCC semantics.
  newTableDesc.DropTime = dropTime
- if err := sqlbase.RemovePublicTableNamespaceEntry(
+ if err := catalogkv.RemovePublicTableNamespaceEntry(
  ctx, txn, execCfg.Codec, newTableDesc.ParentID, newTableDesc.Name,
  ); err != nil {
  return err
@@ -1418,7 +1418,7 @@ func (r *importResumer) dropTables(
  // upgrade and downgrade, because IMPORT does not operate in mixed-version
  // states.
  // TODO(jordan,lucy): remove this comment once 19.2 is released.
- existingDesc, err := sqlbase.ConditionalGetTableDescFromTxn(ctx, txn, execCfg.Codec, prevTableDesc)
+ existingDesc, err := catalogkv.ConditionalGetTableDescFromTxn(ctx, txn, execCfg.Codec, prevTableDesc)
  if err != nil {
  return errors.Wrap(err, "rolling back tables")
  }

pkg/sql/alter_table_set_schema.go

@@ -14,6 +14,7 @@ import (
  "context"
 
  "github.com/cockroachdb/cockroach/pkg/kv"
+ "github.com/cockroachdb/cockroach/pkg/sql/catalog/catalogkv"
  "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
  "github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
  "github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
@@ -130,7 +131,7 @@ func (n *alterTableSetSchemaNode) startExec(params runParams) error {
  return nil
  }
 
- exists, _, err = sqlbase.LookupObjectID(
+ exists, _, err = catalogkv.LookupObjectID(
  ctx, p.txn, p.ExecCfg().Codec, databaseID, desiredSchemaID, tableDesc.Name,
  )
  if err == nil && exists {
@@ -156,7 +157,7 @@ func (n *alterTableSetSchemaNode) startExec(params runParams) error {
  return err
  }
 
- newTbKey := sqlbase.MakeObjectNameKey(ctx, p.ExecCfg().Settings,
+ newTbKey := catalogkv.MakeObjectNameKey(ctx, p.ExecCfg().Settings,
  databaseID, desiredSchemaID, tableDesc.Name).Key(p.ExecCfg().Codec)
 
  b := &kv.Batch{}

pkg/sql/alter_type.go

@@ -83,7 +83,7 @@ func (p *planner) addEnumValue(
 
 func (p *planner) renameType(params runParams, n *alterTypeNode, newName string) error {
  // See if there is a name collision with the new name.
- exists, id, err := sqlbase.LookupObjectID(
+ exists, id, err := catalogkv.LookupObjectID(
  params.ctx,
  p.txn,
  p.ExecCfg().Codec,
@@ -155,7 +155,7 @@ func (p *planner) performRenameTypeDesc(
  }
  // Construct the new namespace key.
  b := p.txn.NewBatch()
- key := sqlbase.MakeObjectNameKey(
+ key := catalogkv.MakeObjectNameKey(
  ctx,
  p.ExecCfg().Settings,
  desc.ParentID,

pkg/sql/backfill.go

@@ -1347,7 +1347,7 @@ func runSchemaChangesInTxn(
  // Reclaim all the old names. Leave the data and descriptor
  // cleanup for later.
  for _, drain := range tableDesc.DrainingNames {
- err := sqlbase.RemoveObjectNamespaceEntry(ctx, planner.Txn(), planner.ExecCfg().Codec,
+ err := catalogkv.RemoveObjectNamespaceEntry(ctx, planner.Txn(), planner.ExecCfg().Codec,
  drain.ParentID, drain.ParentSchemaID, drain.Name, false /* KVTrace */)
  if err != nil {
  return err

pkg/sql/catalog/catalogkv/catalogkv.go

@@ -18,6 +18,7 @@ import (
 
  "github.com/cockroachdb/cockroach/pkg/keys"
  "github.com/cockroachdb/cockroach/pkg/kv"
+ "github.com/cockroachdb/cockroach/pkg/roachpb"
  "github.com/cockroachdb/cockroach/pkg/settings/cluster"
  "github.com/cockroachdb/cockroach/pkg/sql/catalog"
  "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
@@ -412,7 +413,7 @@ func GetDatabaseID(
  if name == sqlbase.SystemDatabaseName {
  return keys.SystemDatabaseID, nil
  }
- found, dbID, err := sqlbase.LookupDatabaseID(ctx, txn, codec, name)
+ found, dbID, err := LookupDatabaseID(ctx, txn, codec, name)
  if err != nil {
  return descpb.InvalidID, err
  }
@@ -509,3 +510,32 @@ func GetDatabaseDescriptorsFromIDs(
  }
  return results, nil
 }
+
+// ConditionalGetTableDescFromTxn validates that the supplied TableDescriptor
+// matches the one currently stored in kv. This simulates a CPut and returns a
+// ConditionFailedError on mismatch. We don't directly use CPut with protos
+// because the marshaling is not guaranteed to be stable and also because it's
+// sensitive to things like missing vs default values of fields.
+func ConditionalGetTableDescFromTxn(
+ ctx context.Context, txn *kv.Txn, codec keys.SQLCodec, expectation sqlbase.TableDescriptor,
+) ([]byte, error) {
+ key := sqlbase.MakeDescMetadataKey(codec, expectation.GetID())
+ existingKV, err := txn.Get(ctx, key)
+ if err != nil {
+ return nil, err
+ }
+ var existing *descpb.Descriptor
+ var existingTable *descpb.TableDescriptor
+ if existingKV.Value != nil {
+ existing = &descpb.Descriptor{}
+ if err := existingKV.Value.GetProto(existing); err != nil {
+ return nil, errors.Wrapf(err,
+ "decoding current table descriptor value for id: %d", expectation.GetID())
+ }
+ existingTable = sqlbase.TableFromDescriptor(existing, existingKV.Value.Timestamp)
+ }
+ if !expectation.TableDesc().Equal(existingTable) {
+ return nil, &roachpb.ConditionFailedError{ActualValue: existingKV.Value}
+ }
+ return existingKV.Value.TagAndDataBytes(), nil
+}

pkg/sql/sqlbase/namespace.go → pkg/sql/catalog/catalogkv/namespace.go

@@ -8,7 +8,7 @@
 // by the Apache License, Version 2.0, included in the file
 // licenses/APL.txt.
 
-package sqlbase
+package catalogkv
 
 import (
  "context"
@@ -18,6 +18,7 @@ import (
  "github.com/cockroachdb/cockroach/pkg/kv"
  "github.com/cockroachdb/cockroach/pkg/settings/cluster"
  "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
+ "github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
  "github.com/cockroachdb/cockroach/pkg/util/log"
 )
 
@@ -69,20 +70,20 @@ func RemoveObjectNamespaceEntry(
  KVTrace bool,
 ) error {
  b := txn.NewBatch()
- var toDelete []DescriptorKey
+ var toDelete []sqlbase.DescriptorKey
  // The (parentID, name) mapping could be in either the new system.namespace
  // or the deprecated version. Thus we try to remove the mapping from both.
  if parentID == keys.RootNamespaceID {
- toDelete = append(toDelete, NewDatabaseKey(name))
+ toDelete = append(toDelete, sqlbase.NewDatabaseKey(name))
  // TODO(solon): This can be completely removed in 20.2.
- toDelete = append(toDelete, NewDeprecatedDatabaseKey(name))
+ toDelete = append(toDelete, sqlbase.NewDeprecatedDatabaseKey(name))
  } else if parentSchemaID == keys.RootNamespaceID {
  // Schemas were introduced in 20.1.
- toDelete = append(toDelete, NewSchemaKey(parentID, name))
+ toDelete = append(toDelete, sqlbase.NewSchemaKey(parentID, name))
  } else {
- toDelete = append(toDelete, NewTableKey(parentID, parentSchemaID, name))
+ toDelete = append(toDelete, sqlbase.NewTableKey(parentID, parentSchemaID, name))
  // TODO(solon): This can be completely removed in 20.2.
- toDelete = append(toDelete, NewDeprecatedTableKey(parentID, name))
+ toDelete = append(toDelete, sqlbase.NewDeprecatedTableKey(parentID, name))
  }
  for _, delKey := range toDelete {
  if KVTrace {
@@ -128,33 +129,33 @@ func MakeObjectNameKey(
  parentID descpb.ID,
  parentSchemaID descpb.ID,
  name string,
-) DescriptorKey {
+) sqlbase.DescriptorKey {
  // TODO(solon): This if condition can be removed in 20.2
  if !settings.Version.IsActive(ctx, clusterversion.VersionNamespaceTableWithSchemas) {
- return NewDeprecatedTableKey(parentID, name)
+ return sqlbase.NewDeprecatedTableKey(parentID, name)
  }
- var key DescriptorKey
+ var key sqlbase.DescriptorKey
  if parentID == keys.RootNamespaceID {
- key = NewDatabaseKey(name)
+ key = sqlbase.NewDatabaseKey(name)
  } else if parentSchemaID == keys.RootNamespaceID {
- key = NewSchemaKey(parentID, name)
+ key = sqlbase.NewSchemaKey(parentID, name)
  } else {
- key = NewTableKey(parentID, parentSchemaID, name)
+ key = sqlbase.NewTableKey(parentID, parentSchemaID, name)
  }
  return key
 }
 
 // MakePublicTableNameKey is a wrapper around MakeObjectNameKey for public tables.
 func MakePublicTableNameKey(
  ctx context.Context, settings *cluster.Settings, parentID descpb.ID, name string,
-) DescriptorKey {
+) sqlbase.DescriptorKey {
  return MakeObjectNameKey(ctx, settings, parentID, keys.PublicSchemaID, name)
 }
 
 // MakeDatabaseNameKey is a wrapper around MakeObjectNameKey for databases.
 func MakeDatabaseNameKey(
  ctx context.Context, settings *cluster.Settings, name string,
-) DescriptorKey {
+) sqlbase.DescriptorKey {
  return MakeObjectNameKey(ctx, settings, keys.RootNamespaceID, keys.RootNamespaceID, name)
 }
 
@@ -169,13 +170,13 @@ func LookupObjectID(
  parentSchemaID descpb.ID,
  name string,
 ) (bool, descpb.ID, error) {
- var key DescriptorKey
+ var key sqlbase.DescriptorKey
  if parentID == keys.RootNamespaceID {
- key = NewDatabaseKey(name)
+ key = sqlbase.NewDatabaseKey(name)
  } else if parentSchemaID == keys.RootNamespaceID {
- key = NewSchemaKey(parentID, name)
+ key = sqlbase.NewSchemaKey(parentID, name)
  } else {
- key = NewTableKey(parentID, parentSchemaID, name)
+ key = sqlbase.NewTableKey(parentID, parentSchemaID, name)
  }
  log.Eventf(ctx, "looking up descriptor ID for name key %q", key.Key(codec))
  res, err := txn.Get(ctx, key.Key(codec))
@@ -202,11 +203,11 @@ func LookupObjectID(
  return false, descpb.InvalidID, nil
  }
 
- var dKey DescriptorKey
+ var dKey sqlbase.DescriptorKey
  if parentID == keys.RootNamespaceID {
- dKey = NewDeprecatedDatabaseKey(name)
+ dKey = sqlbase.NewDeprecatedDatabaseKey(name)
  } else {
- dKey = NewDeprecatedTableKey(parentID, name)
+ dKey = sqlbase.NewDeprecatedTableKey(parentID, name)
  }
  log.Eventf(ctx, "looking up descriptor ID for name key %q", dKey.Key(codec))
  res, err = txn.Get(ctx, dKey.Key(codec))

pkg/sql/catalog/catalogkv/physical_accessor.go

@@ -52,7 +52,7 @@ func (a UncachedPhysicalAccessor) GetDatabaseDesc(
  return sysDB, nil
  }
 
- found, descID, err := sqlbase.LookupDatabaseID(ctx, txn, codec, name)
+ found, descID, err := LookupDatabaseID(ctx, txn, codec, name)
  if err != nil {
  return nil, err
  } else if !found {
@@ -245,7 +245,7 @@ func (a UncachedPhysicalAccessor) GetObjectDesc(
  descID := sqlbase.LookupSystemTableDescriptorID(ctx, settings, codec, dbID, object)
  if descID == descpb.InvalidID {
  var found bool
- found, descID, err = sqlbase.LookupObjectID(ctx, txn, codec, dbID, schema.ID, object)
+ found, descID, err = LookupObjectID(ctx, txn, codec, dbID, schema.ID, object)
  if err != nil {
  return nil, err
  }

pkg/sql/catalog/lease/lease.go

@@ -1662,7 +1662,7 @@ func (m *Manager) resolveName(
  txn.SetFixedTimestamp(ctx, timestamp)
  var found bool
  var err error
- found, id, err = sqlbase.LookupObjectID(ctx, txn, m.storage.codec, parentID, parentSchemaID, name)
+ found, id, err = catalogkv.LookupObjectID(ctx, txn, m.storage.codec, parentID, parentSchemaID, name)
  if err != nil {
  return err
  }