Merge #35300 #35682 #35689

35300: sql: add checks after all referenced columns have been backfilled r=lucy-zhang a=lucy-zhang

Previously, if a column was added with a check constraint that also referenced
another column that was public, writes to that public column would erroneously
fail (and, in the worst case, result in a panic) while the column being added
was not yet public. With this change, the schema changer will now wait to add
the check constraint to the table descriptor until after all the columns that
were added in the same transaction have been backfilled.

A new optional field has been added to `ConstraintToValidate` for the check
constraint itself, so that it can be added to the table descriptor at the
correct step in the schema change process.

I ended up adding this field to the existing mutation instead of creating a new
type of mutation to add the constraint to the descriptor, since it ultimately
seemed to me that a mutation that simply adds a schema element in its backfill
step would be too inconsistent with what mutations are, especially since all
the mutations for a single transaction are basically executed at the same time.
To support NOT VALID in the future, we could add more flags to the protobuf to
indicate that either the addition of the constraint or the validation should be
skipped, so that they can be executed separately.

Fixes #35258, fixes #35193

Release note: None

35682: engineccl/mvcc: fix time-bound iterator's interaction with moving intents r=nvanbenschoten a=nvanbenschoten

Fixes #34819.

349ff61 introduced a workaround for #28358 into MVCCIncrementalIterator.
This workaround created a second (non-time-bound) iterator to verify
possibly-phantom MVCCMetadata keys during iteration.

We found in #34819 that it is necessary for correctness that sanityIter
be created before the original iter. This is because the provided Reader
that both iterators are created from may not be a consistent snapshot, so
the two iterators could end up observing different information. The hack
around sanityCheckMetadataKey only works properly if all possible
discrepancies between the two iterators lead to intents and values falling
outside of the timestamp range **from the time-bound iterator's perspective**.
This allows us to simply ignore discrepancies that we notice in advance().

This commit makes this change. It also adds a test that failed regularly
before the fix under stress and no longer fails after the fix.

Release note: None

35689: roachtest: add large node kv tests and batching kv tests r=nvanbenschoten a=nvanbenschoten

This commit adds support for running `kv` with a `--batch` parameter. It
then adds the following new roachtest configurations:
- kv0/enc=false/nodes=3/batch=16
- kv95/enc=false/nodes=3/batch=16
- kv0/enc=false/nodes=3/cpu=96
- kv95/enc=false/nodes=3/cpu=96
- kv50/enc=false/nodes=4/cpu=96/batch=64

The last test is currently skipped because of #34241. I confirmed that
it triggers the corresponding assertion on both AWS and GCE.

My request for more m5d.24xlarge quota just succeeded, but I may need to
request more quota for n1-highcpu-96 VMs for these to run nightly.

Release note: None

Co-authored-by: Lucy Zhang <[email protected]>
Co-authored-by: Nathan VanBenschoten <[email protected]>

pkg/ccl/storageccl/engineccl/mvcc.go

@@ -81,6 +81,17 @@ func NewMVCCIncrementalIterator(e engine.Reader, opts IterOptions) *MVCCIncremen
 	return &MVCCIncrementalIterator{
 		e:          e,
 		upperBound: opts.UpperBound,
+		// It is necessary for correctness that sanityIter be created before
+		// iter. This is because the provided Reader may not be a consistent
+		// snapshot, so the two could end up observing different information.
+		// The hack around sanityCheckMetadataKey only works properly if all
+		// possible discrepancies between the two iterators lead to intents
+		// and values falling outside of the timestamp range **from iter's
+		// perspective**. This allows us to simply ignore discrepancies that
+		// we notice in advance(). See #34819.
+		sanityIter: e.NewIterator(engine.IterOptions{
+			UpperBound: opts.UpperBound,
+		}),
 		iter: e.NewIterator(engine.IterOptions{
 			// The call to startTime.Next() converts our exclusive start bound into
 			// the inclusive start bound that MinTimestampHint expects. This is
@@ -214,11 +225,6 @@ func (i *MVCCIncrementalIterator) advance() {
 // sees that exact key. Otherwise, it returns false. It's used in the workaround
 // in `advance` for a time-bound iterator bug.
 func (i *MVCCIncrementalIterator) sanityCheckMetadataKey() ([]byte, bool, error) {
-	if i.sanityIter == nil {
-		// The common case is that we'll won't need the sanityIter for a given
-		// MVCCIncrementalIterator, so create it lazily.
-		i.sanityIter = i.e.NewIterator(engine.IterOptions{UpperBound: i.upperBound})
-	}
 	unsafeKey := i.iter.UnsafeKey()
 	i.sanityIter.Seek(unsafeKey)
 	if ok, err := i.sanityIter.Valid(); err != nil {

pkg/ccl/storageccl/engineccl/mvcc_test.go

@@ -24,7 +24,9 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
+	"github.com/pkg/errors"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
 )
 
 func iterateExpectErr(
@@ -486,6 +488,28 @@ func TestMVCCIncrementalIteratorIntentStraddlesSStables(t *testing.T) {
 	}
 }
 
+func slurpKVsInTimeRange(
+	e engine.Reader, prefix roachpb.Key, startTime, endTime hlc.Timestamp,
+) ([]engine.MVCCKeyValue, error) {
+	endKey := prefix.PrefixEnd()
+	iter := NewMVCCIncrementalIterator(e, IterOptions{
+		StartTime:  startTime,
+		EndTime:    endTime,
+		UpperBound: endKey,
+	})
+	defer iter.Close()
+	var kvs []engine.MVCCKeyValue
+	for iter.Seek(engine.MakeMVCCMetadataKey(prefix)); ; iter.Next() {
+		if ok, err := iter.Valid(); err != nil {
+			return nil, err
+		} else if !ok || iter.UnsafeKey().Key.Compare(endKey) >= 0 {
+			break
+		}
+		kvs = append(kvs, engine.MVCCKeyValue{Key: iter.Key(), Value: iter.Value()})
+	}
+	return kvs, nil
+}
+
 // TestMVCCIncrementalIteratorIntentDeletion checks a workaround in
 // MVCCIncrementalIterator for a bug in time-bound iterators, where an intent
 // has been deleted, but the time-bound iterator doesn't see the deletion.
@@ -510,27 +534,6 @@ func TestMVCCIncrementalIteratorIntentDeletion(t *testing.T) {
 			Status: roachpb.COMMITTED,
 		}
 	}
-	slurpKVs := func(
-		e engine.Reader, prefix roachpb.Key, startTime, endTime hlc.Timestamp,
-	) ([]engine.MVCCKeyValue, error) {
-		endKey := prefix.PrefixEnd()
-		iter := NewMVCCIncrementalIterator(e, IterOptions{
-			StartTime:  startTime,
-			EndTime:    endTime,
-			UpperBound: endKey,
-		})
-		defer iter.Close()
-		var kvs []engine.MVCCKeyValue
-		for iter.Seek(engine.MakeMVCCMetadataKey(prefix)); ; iter.Next() {
-			if ok, err := iter.Valid(); err != nil {
-				return nil, err
-			} else if !ok || iter.UnsafeKey().Key.Compare(endKey) >= 0 {
-				break
-			}
-			kvs = append(kvs, engine.MVCCKeyValue{Key: iter.Key(), Value: iter.Value()})
-		}
-		return kvs, nil
-	}
 
 	ctx := context.Background()
 	kA := roachpb.Key("kA")
@@ -592,32 +595,102 @@ func TestMVCCIncrementalIteratorIntentDeletion(t *testing.T) {
 
 	// The kA ts1 intent has been resolved. There's now a new intent on kA, but
 	// the timestamp (ts3) is too new so it should be ignored.
-	kvs, err := slurpKVs(db, kA, ts0, ts1)
+	kvs, err := slurpKVsInTimeRange(db, kA, ts0, ts1)
 	require.NoError(t, err)
 	require.Equal(t, []engine.MVCCKeyValue{
 		{Key: engine.MVCCKey{Key: kA, Timestamp: ts1}, Value: vA1.RawBytes},
 	}, kvs)
 	// kA has a value at ts2. Again the intent is too new (ts3), so ignore.
-	kvs, err = slurpKVs(db, kA, ts0, ts2)
+	kvs, err = slurpKVsInTimeRange(db, kA, ts0, ts2)
 	require.NoError(t, err)
 	require.Equal(t, []engine.MVCCKeyValue{
 		{Key: engine.MVCCKey{Key: kA, Timestamp: ts2}, Value: vA2.RawBytes},
 		{Key: engine.MVCCKey{Key: kA, Timestamp: ts1}, Value: vA1.RawBytes},
 	}, kvs)
 	// At ts3, we should see the new intent
-	_, err = slurpKVs(db, kA, ts0, ts3)
+	_, err = slurpKVsInTimeRange(db, kA, ts0, ts3)
 	require.EqualError(t, err, `conflicting intents on "kA"`)
 
 	// Similar to the kA ts1 check, but there is no newer intent. We expect to
 	// pick up the intent deletion and it should cancel out the intent, leaving
 	// only the value at ts1.
-	kvs, err = slurpKVs(db, kB, ts0, ts1)
+	kvs, err = slurpKVsInTimeRange(db, kB, ts0, ts1)
 	require.NoError(t, err)
 	require.Equal(t, []engine.MVCCKeyValue{
 		{Key: engine.MVCCKey{Key: kB, Timestamp: ts1}, Value: vB1.RawBytes},
 	}, kvs)
 
 	// Sanity check that we see the still unresolved intent for kC ts1.
-	_, err = slurpKVs(db, kC, ts0, ts1)
+	_, err = slurpKVsInTimeRange(db, kC, ts0, ts1)
 	require.EqualError(t, err, `conflicting intents on "kC"`)
 }
+
+// TestMVCCIncrementalIteratorIntentRewrittenConcurrently verifies that the
+// workaround in MVCCIncrementalIterator to double-check for deleted intents
+// properly handles cases where an intent originally in a time-bound iterator's
+// time range is rewritten at a timestamp outside of its time range.
+func TestMVCCIncrementalIteratorIntentRewrittenConcurrently(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	// Create a DB containing a single intent.
+	ctx := context.Background()
+	db := engine.NewInMem(roachpb.Attributes{}, 10<<20 /* 10 MB */)
+	defer db.Close()
+
+	kA := roachpb.Key("kA")
+	vA1 := roachpb.MakeValueFromString("vA1")
+	vA2 := roachpb.MakeValueFromString("vA2")
+	ts0 := hlc.Timestamp{WallTime: 0}
+	ts1 := hlc.Timestamp{WallTime: 1}
+	ts2 := hlc.Timestamp{WallTime: 2}
+	ts3 := hlc.Timestamp{WallTime: 3}
+	txn := &roachpb.Transaction{
+		TxnMeta: enginepb.TxnMeta{
+			Key:       roachpb.Key("b"),
+			ID:        uuid.MakeV4(),
+			Epoch:     1,
+			Timestamp: ts1,
+			Sequence:  1,
+		},
+		OrigTimestamp: ts1,
+	}
+	if err := engine.MVCCPut(ctx, db, nil, kA, ts1, vA1, txn); err != nil {
+		t.Fatal(err)
+	}
+
+	// Concurrently iterate over the intent using a time-bound iterator and move
+	// the intent out of the time-bound iterator's time range by writing to it
+	// again at a higher timestamp.
+	g, _ := errgroup.WithContext(ctx)
+	g.Go(func() error {
+		// Re-write the intent with a higher timestamp.
+		txn.Timestamp = ts3
+		txn.Sequence = 2
+		return engine.MVCCPut(ctx, db, nil, kA, ts1, vA2, txn)
+	})
+	g.Go(func() error {
+		// Iterate with a time range that includes the initial intent but does
+		// not include the new intent.
+		kvs, err := slurpKVsInTimeRange(db, kA, ts0, ts2)
+
+		// There are two permissible outcomes from the scan. If the iteration
+		// wins the race with the put that moves the intent then it should
+		// observe the intent and return a write intent error. If the iteration
+		// loses the race with the put that moves the intent then it should
+		// observe and return nothing because there will be no committed or
+		// provisional keys in its time range.
+		if err != nil {
+			if !testutils.IsError(err, `conflicting intents on "kA"`) {
+				return err
+			}
+		} else {
+			if len(kvs) != 0 {
+				return errors.Errorf(`unexpected kvs: %v`, kvs)
+			}
+		}
+		return nil
+	})
+	if err := g.Wait(); err != nil {
+		t.Fatal(err)
+	}
+}

pkg/cmd/roachtest/cluster.go

@@ -403,8 +403,13 @@ func awsMachineType(cpus int) string {
 		return "c5d.4xlarge"
 	case cpus <= 36:
 		return "c5d.9xlarge"
-	default:
+	case cpus <= 72:
 		return "c5d.18xlarge"
+	case cpus <= 96:
+		// There is no c5d.24xlarge.
+		return "m5d.24xlarge"
+	default:
+		panic(fmt.Sprintf("no aws machine type with %d cpus", cpus))
 	}
 }
 

pkg/cmd/roachtest/kv.go

@@ -34,6 +34,7 @@ func registerKV(r *registry) {
 		nodes       int
 		cpus        int
 		readPercent int
+		batchSize   int
 		blockSize   int
 		encryption  bool
 	}
@@ -49,6 +50,12 @@ func registerKV(r *registry) {
 			concurrency := ifLocal("", " --concurrency="+fmt.Sprint(nodes*64))
 			duration := " --duration=" + ifLocal("10s", "10m")
 			readPercent := fmt.Sprintf(" --read-percent=%d", opts.readPercent)
+
+			var batchSize string
+			if opts.batchSize > 0 {
+				batchSize = fmt.Sprintf(" --batch=%d", opts.batchSize)
+			}
+
 			var blockSize string
 			if opts.blockSize > 0 {
 				blockSize = fmt.Sprintf(" --min-block-bytes=%d --max-block-bytes=%d",
@@ -57,7 +64,7 @@ func registerKV(r *registry) {
 
 			cmd := fmt.Sprintf(
 				"./workload run kv --init --splits=1000 --histograms=logs/stats.json"+
-					concurrency+duration+readPercent+blockSize+" {pgurl:1-%d}",
+					concurrency+duration+readPercent+batchSize+blockSize+" {pgurl:1-%d}",
 				nodes)
 			c.Run(ctx, c.Node(nodes+1), cmd)
 			return nil
@@ -86,6 +93,16 @@ func registerKV(r *registry) {
 		{nodes: 3, cpus: 32, readPercent: 0, blockSize: 1 << 16 /* 64 KB */},
 		{nodes: 3, cpus: 32, readPercent: 95, blockSize: 1 << 16 /* 64 KB */},
 
+		// Configs with large batch sizes.
+		{nodes: 3, cpus: 8, readPercent: 0, batchSize: 16},
+		{nodes: 3, cpus: 8, readPercent: 95, batchSize: 16},
+
+		// Configs with large nodes.
+		{nodes: 3, cpus: 96, readPercent: 0},
+		{nodes: 3, cpus: 96, readPercent: 95},
+		// Skipped: https://github.com/cockroachdb/cockroach/issues/34241.
+		// {nodes: 4, cpus: 96, readPercent: 50, batchSize: 64},
+
 		// Configs with encryption.
 		{nodes: 1, cpus: 8, readPercent: 0, encryption: true},
 		{nodes: 1, cpus: 8, readPercent: 95, encryption: true},
@@ -101,6 +118,9 @@ func registerKV(r *registry) {
 		if opts.cpus != 8 { // support legacy test name which didn't include cpu
 			nameParts = append(nameParts, fmt.Sprintf("cpu=%d", opts.cpus))
 		}
+		if opts.batchSize != 0 { // support legacy test name which didn't include batch size
+			nameParts = append(nameParts, fmt.Sprintf("batch=%d", opts.batchSize))
+		}
 		if opts.blockSize != 0 { // support legacy test name which didn't include block size
 			nameParts = append(nameParts, fmt.Sprintf("size=%dkb", opts.blockSize>>10))
 		}

pkg/sql/alter_table.go

@@ -217,10 +217,7 @@ func (n *alterTableNode) startExec(params runParams) error {
 					return err
 				}
 				ck.Validity = sqlbase.ConstraintValidity_Validating
-				n.tableDesc.Checks = append(n.tableDesc.Checks, ck)
-				descriptorChanged = true
-
-				n.tableDesc.AddCheckValidationMutation(ck.Name)
+				n.tableDesc.AddCheckValidationMutation(ck)
 
 			case *tree.ForeignKeyConstraintTableDef:
 				for _, colName := range d.FromCols {
@@ -399,7 +396,7 @@ func (n *alterTableNode) startExec(params runParams) error {
 
 			// Drop check constraints which reference the column.
 			validChecks := n.tableDesc.Checks[:0]
-			for _, check := range n.tableDesc.Checks {
+			for _, check := range n.tableDesc.AllActiveAndInactiveChecks() {
 				if used, err := check.UsesColumn(n.tableDesc.TableDesc(), col.ID); err != nil {
 					return err
 				} else if used {
@@ -482,10 +479,6 @@ func (n *alterTableNode) startExec(params runParams) error {
 					}
 				}
 				if !found {
-					panic("constraint returned by GetConstraintInfo not found")
-				}
-
-				if n.tableDesc.Checks[idx].Validity == sqlbase.ConstraintValidity_Validating {
 					return fmt.Errorf("constraint %q in the middle of being added, try again later", t.Constraint)
 				}