kvserver: Add migration for interleaved intents to separated

docs/generated/settings/settings-for-tenants.txt

@@ -153,4 +153,4 @@ trace.datadog.project	string	CockroachDB	the project under which traces will be
 trace.debug.enable	boolean	false	if set, traces for recent requests can be seen at https://<ui>/debug/requests
 trace.lightstep.token	string		if set, traces go to Lightstep using this token
 trace.zipkin.collector	string		if set, traces go to the given Zipkin instance (example: '127.0.0.1:9411'). Only one tracer can be configured at a time.
-version	version	21.1-126	set the active cluster version in the format '<major>.<minor>'
+version	version	21.1-130	set the active cluster version in the format '<major>.<minor>'

docs/generated/settings/settings.html

@@ -157,6 +157,6 @@
 <tr><td><code>trace.debug.enable</code></td><td>boolean</td><td><code>false</code></td><td>if set, traces for recent requests can be seen at https://<ui>/debug/requests</td></tr>
 <tr><td><code>trace.lightstep.token</code></td><td>string</td><td><code></code></td><td>if set, traces go to Lightstep using this token</td></tr>
 <tr><td><code>trace.zipkin.collector</code></td><td>string</td><td><code></code></td><td>if set, traces go to the given Zipkin instance (example: '127.0.0.1:9411'). Only one tracer can be configured at a time.</td></tr>
-<tr><td><code>version</code></td><td>version</td><td><code>21.1-126</code></td><td>set the active cluster version in the format '<major>.<minor>'</td></tr>
+<tr><td><code>version</code></td><td>version</td><td><code>21.1-130</code></td><td>set the active cluster version in the format '<major>.<minor>'</td></tr>
 </tbody>
 </table>

pkg/base/config.go

@@ -515,6 +515,9 @@ type StorageConfig struct {
 	// UseFileRegistry is true if the file registry is needed (eg: encryption-at-rest).
 	// This may force the store version to versionFileRegistry if currently lower.
 	UseFileRegistry bool
+	// DisableSeparatedIntents is true if separated intents should not be written
+	// by intent writers. Only true for tests.
+	DisableSeparatedIntents bool
 	// EncryptionOptions is a serialized protobuf set by Go CCL code and passed
 	// through to C CCL code to set up encryption-at-rest.  Must be set if and
 	// only if encryption is enabled, otherwise left empty.

pkg/cli/debug.go

@@ -24,6 +24,8 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
@@ -1307,6 +1309,99 @@ func runDebugMergeLogs(cmd *cobra.Command, args []string) error {
 	return writeLogStream(s, cmd.OutOrStdout(), o.filter, o.prefix, o.keepRedactable)
 }
 
+var debugIntentCount = &cobra.Command{
+	Use:   "intent-count <store directory>",
+	Short: "return a count of intents in directory",
+	Long: `
+Returns a count of interleaved and separated intents in the store directory.
+Used to investigate stores with lots of unresolved intents, or to confirm
+if the migration away from interleaved intents was successful.
+`,
+	Args: cobra.MinimumNArgs(1),
+	RunE: runDebugIntentCount,
+}
+
+func runDebugIntentCount(cmd *cobra.Command, args []string) error {
+	stopper := stop.NewStopper()
+	ctx := context.Background()
+	defer stopper.Stop(ctx)
+
+	db, err := OpenExistingStore(args[0], stopper, true /* readOnly */)
+	if err != nil {
+		return err
+	}
+	defer db.Close()
+
+	var interleavedIntentCount, separatedIntentCount int
+	var keysCount uint64
+	var wg sync.WaitGroup
+	closer := make(chan bool)
+
+	wg.Add(1)
+	_ = stopper.RunAsyncTask(ctx, "intent-count-progress-indicator", func(ctx context.Context) {
+		defer wg.Done()
+		ctx, cancel := stopper.WithCancelOnQuiesce(ctx)
+		defer cancel()
+
+		ticker := time.NewTicker(10 * time.Second)
+		defer ticker.Stop()
+
+		select {
+		case <-ticker.C:
+			fmt.Printf("scanned %d keys\n", atomic.LoadUint64(&keysCount))
+		case <-ctx.Done():
+			return
+		case <-closer:
+			return
+		}
+	})
+
+	iter := db.NewEngineIterator(storage.IterOptions{
+		LowerBound: roachpb.KeyMin,
+		UpperBound: roachpb.KeyMax,
+	})
+	defer iter.Close()
+	valid, err := iter.SeekEngineKeyGE(storage.EngineKey{Key: roachpb.KeyMin})
+	var meta enginepb.MVCCMetadata
+	for ; valid && err == nil; valid, err = iter.NextEngineKey() {
+		key, err := iter.EngineKey()
+		if err != nil {
+			return err
+		}
+		atomic.AddUint64(&keysCount, 1)
+		if key.IsLockTableKey() {
+			separatedIntentCount++
+			continue
+		}
+		if !key.IsMVCCKey() {
+			continue
+		}
+		mvccKey, err := key.ToMVCCKey()
+		if err != nil {
+			return err
+		}
+		if !mvccKey.Timestamp.IsEmpty() {
+			continue
+		}
+		val := iter.UnsafeValue()
+		if err := protoutil.Unmarshal(val, &meta); err != nil {
+			return err
+		}
+		if meta.IsInline() {
+			continue
+		}
+		interleavedIntentCount++
+	}
+	if err != nil {
+		return err
+	}
+	close(closer)
+	wg.Wait()
+	fmt.Printf("interleaved intents: %d\nseparated intents: %d\n",
+		interleavedIntentCount, separatedIntentCount)
+	return nil
+}
+
 // DebugCmdsForRocksDB lists debug commands that access rocksdb through the engine
 // and need encryption flags (injected by CCL code).
 // Note: do NOT include commands that just call rocksdb code without setting up an engine.
@@ -1318,6 +1413,7 @@ var DebugCmdsForRocksDB = []*cobra.Command{
 	debugRaftLogCmd,
 	debugRangeDataCmd,
 	debugRangeDescriptorsCmd,
+	debugIntentCount,
 }
 
 // All other debug commands go here.

pkg/clusterversion/cockroach_versions.go

@@ -274,6 +274,12 @@ const (
 	// AlterSystemWebSessionsCreateIndexes creates indexes on the columns revokedAt and
 	// lastUsedAt for the system.web_sessions table.
 	AlterSystemWebSessionsCreateIndexes
+	// SeparatedIntentsMigration adds the migration to move over all remaining
+	// intents to the separated lock table space.
+	SeparatedIntentsMigration
+	// PostSeparatedIntentsMigration runs a cleanup migration after the main
+	// SeparatedIntentsMigration.
+	PostSeparatedIntentsMigration
 
 	// Step (1): Add new versions here.
 )
@@ -447,6 +453,14 @@ var versionsSingleton = keyedVersions{
 		Key:     AlterSystemWebSessionsCreateIndexes,
 		Version: roachpb.Version{Major: 21, Minor: 1, Internal: 126},
 	},
+	{
+		Key:     SeparatedIntentsMigration,
+		Version: roachpb.Version{Major: 21, Minor: 1, Internal: 128},
+	},
+	{
+		Key:     PostSeparatedIntentsMigration,
+		Version: roachpb.Version{Major: 21, Minor: 1, Internal: 130},
+	},
 
 	// Step (2): Add new versions here.
 }

pkg/keys/keys.go

@@ -551,20 +551,25 @@ func MustAddr(k roachpb.Key) roachpb.RKey {
 // range-local key, which is guaranteed to be located on the same range.
 // AddrUpperBound() returns the regular key that is just to the right, which may
 // not be on the same range but is suitable for use as the EndKey of a span
-// involving a range-local key.
+// involving a range-local key. The one exception to this is the local key
+// prefix itself; that continues to return the left key as having that as the
+// upper bound excludes all local keys.
 //
 // Logically, the keys are arranged as follows:
 //
 // k1 /local/k1/KeyMin ... /local/k1/KeyMax k1\x00 /local/k1/x00/KeyMin ...
 //
 // and so any end key /local/k1/x corresponds to an address-resolved end key of
-// k1\x00.
+// k1\x00, with the exception of /local/k1 itself (no suffix) which corresponds
+// to an address-resolved end key of k1.
 func AddrUpperBound(k roachpb.Key) (roachpb.RKey, error) {
 	rk, err := Addr(k)
 	if err != nil {
 		return rk, err
 	}
-	if IsLocal(k) {
+	// If k is the RangeKeyPrefix, it excludes all range local keys under rk.
+	// The Next() is not necessary.
+	if IsLocal(k) && !k.Equal(MakeRangeKeyPrefix(rk)) {
 		// The upper bound for a range-local key that addresses to key k
 		// is the key directly after k.
 		rk = rk.Next()

pkg/keys/keys_test.go

@@ -106,7 +106,9 @@ func TestKeyAddress(t *testing.T) {
 	}{
 		{roachpb.Key{}, roachpb.RKeyMin},
 		{roachpb.Key("123"), roachpb.RKey("123")},
+		{MakeRangeKeyPrefix(roachpb.RKey("foo")), roachpb.RKey("foo")},
 		{RangeDescriptorKey(roachpb.RKey("foo")), roachpb.RKey("foo")},
+		{MakeRangeKeyPrefix(roachpb.RKey("baz")), roachpb.RKey("baz")},
 		{TransactionKey(roachpb.Key("baz"), uuid.MakeV4()), roachpb.RKey("baz")},
 		{TransactionKey(roachpb.KeyMax, uuid.MakeV4()), roachpb.RKeyMax},
 		{RangeDescriptorKey(roachpb.RKey(TransactionKey(roachpb.Key("doubleBaz"), uuid.MakeV4()))), roachpb.RKey("doubleBaz")},
@@ -121,6 +123,30 @@ func TestKeyAddress(t *testing.T) {
 	}
 }
 
+func TestKeyAddressUpperBound(t *testing.T) {
+	testCases := []struct {
+		key        roachpb.Key
+		expAddress roachpb.RKey
+	}{
+		{roachpb.Key{}, roachpb.RKeyMin},
+		{roachpb.Key("123"), roachpb.RKey("123")},
+		{MakeRangeKeyPrefix(roachpb.RKey("foo")), roachpb.RKey("foo")},
+		{RangeDescriptorKey(roachpb.RKey("foo")), roachpb.RKey("foo").Next()},
+		{MakeRangeKeyPrefix(roachpb.RKey("baz")), roachpb.RKey("baz")},
+		{TransactionKey(roachpb.Key("baz"), uuid.MakeV4()), roachpb.RKey("baz").Next()},
+		{TransactionKey(roachpb.KeyMax, uuid.MakeV4()), roachpb.RKeyMax.Next()},
+		{RangeDescriptorKey(roachpb.RKey(TransactionKey(roachpb.Key("doubleBaz"), uuid.MakeV4()))), roachpb.RKey("doubleBaz").Next()},
+		{nil, nil},
+	}
+	for i, test := range testCases {
+		if keyAddr, err := AddrUpperBound(test.key); err != nil {
+			t.Errorf("%d: %v", i, err)
+		} else if !keyAddr.Equal(test.expAddress) {
+			t.Errorf("%d: expected address for key %q doesn't match %q", i, test.key, test.expAddress)
+		}
+	}
+}
+
 func TestKeyAddressError(t *testing.T) {
 	testCases := map[string][]roachpb.Key{
 		"store-local key .* is not addressable": {

pkg/kv/batch.go

@@ -271,6 +271,8 @@ func (b *Batch) fillResults(ctx context.Context) {
 			case *roachpb.AddSSTableRequest:
 			case *roachpb.MigrateRequest:
 			case *roachpb.QueryResolvedTimestampRequest:
+			case *roachpb.BarrierRequest:
+			case *roachpb.ScanInterleavedIntentsRequest:
 			default:
 				if result.Err == nil {
 					result.Err = errors.Errorf("unsupported reply: %T for %T",
@@ -843,3 +845,45 @@ func (b *Batch) queryResolvedTimestamp(s, e interface{}) {
 	b.appendReqs(req)
 	b.initResult(1, 0, notRaw, nil)
 }
+
+func (b *Batch) scanInterleavedIntents(s, e interface{}) {
+	begin, err := marshalKey(s)
+	if err != nil {
+		b.initResult(0, 0, notRaw, err)
+		return
+	}
+	end, err := marshalKey(e)
+	if err != nil {
+		b.initResult(0, 0, notRaw, err)
+		return
+	}
+	req := &roachpb.ScanInterleavedIntentsRequest{
+		RequestHeader: roachpb.RequestHeader{
+			Key:    begin,
+			EndKey: end,
+		},
+	}
+	b.appendReqs(req)
+	b.initResult(1, 0, notRaw, nil)
+}
+
+func (b *Batch) barrier(s, e interface{}) {
+	begin, err := marshalKey(s)
+	if err != nil {
+		b.initResult(0, 0, notRaw, err)
+		return
+	}
+	end, err := marshalKey(e)
+	if err != nil {
+		b.initResult(0, 0, notRaw, err)
+		return
+	}
+	req := &roachpb.BarrierRequest{
+		RequestHeader: roachpb.RequestHeader{
+			Key:    begin,
+			EndKey: end,
+		},
+	}
+	b.appendReqs(req)
+	b.initResult(1, 0, notRaw, nil)
+}

pkg/kv/db.go

@@ -725,6 +725,51 @@ func (db *DB) QueryResolvedTimestamp(
 	return r.ResolvedTS, nil
 }
 
+// ScanInterleavedIntents is a command that returns all interleaved intents
+// encountered in the request span. A resume span is returned if the entirety
+// of the request span was not scanned.
+func (db *DB) ScanInterleavedIntents(
+	ctx context.Context, begin, end interface{}, ts hlc.Timestamp,
+) ([]roachpb.Intent, *roachpb.Span, error) {
+	b := &Batch{Header: roachpb.Header{Timestamp: ts}}
+	b.scanInterleavedIntents(begin, end)
+	result, err := getOneResult(db.Run(ctx, b), b)
+	if err != nil {
+		return nil, nil, err
+	}
+	responses := b.response.Responses
+	if len(responses) == 0 {
+		return nil, nil, errors.Errorf("unexpected empty response for ScanInterleavedIntents")
+	}
+	resp, ok := responses[0].GetInner().(*roachpb.ScanInterleavedIntentsResponse)
+	if !ok {
+		return nil, nil, errors.Errorf("unexpected response of type %T for ScanInterleavedIntents",
+			responses[0].GetInner())
+	}
+	return resp.Intents, result.ResumeSpan, nil
+}
+
+// Barrier is a command that waits for conflicting operations such as earlier
+// writes on the specified key range to finish.
+func (db *DB) Barrier(ctx context.Context, begin, end interface{}) (hlc.Timestamp, error) {
+	b := &Batch{}
+	b.barrier(begin, end)
+	err := getOneErr(db.Run(ctx, b), b)
+	if err != nil {
+		return hlc.Timestamp{}, err
+	}
+	responses := b.response.Responses
+	if len(responses) == 0 {
+		return hlc.Timestamp{}, errors.Errorf("unexpected empty response for Barrier")
+	}
+	resp, ok := responses[0].GetInner().(*roachpb.BarrierResponse)
+	if !ok {
+		return hlc.Timestamp{}, errors.Errorf("unexpected response of type %T for Barrier",
+			responses[0].GetInner())
+	}
+	return resp.Timestamp, nil
+}
+
 // sendAndFill is a helper which sends the given batch and fills its results,
 // returning the appropriate error which is either from the first failing call,
 // or an "internal" error.

pkg/kv/kvserver/batcheval/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "batcheval",
     srcs = [
         "cmd_add_sstable.go",
+        "cmd_barrier.go",
         "cmd_clear_range.go",
         "cmd_compute_checksum.go",
         "cmd_conditional_put.go",
@@ -36,6 +37,7 @@ go_library(
         "cmd_reverse_scan.go",
         "cmd_revert_range.go",
         "cmd_scan.go",
+        "cmd_scan_interleaved_intents.go",
         "cmd_subsume.go",
         "cmd_truncate_log.go",
         "command.go",