From d3bd68b92dd1a5dbe01f53db93ee88a4a985bd90 Mon Sep 17 00:00:00 2001
From: irfan sharif <irfanmahmoudsharif@gmail.com>
Date: Wed, 24 Nov 2021 09:24:58 -0500
Subject: [PATCH] spanconfig/store: support applying a batch of updates

We first introduced spanconfig.StoreWriter in #70287. Here we extend the
interface to accept a batch of updates instead of just one.

    type StoreWriter interface {
        Apply(ctx context.Context, dryrun bool, updates ...Update) (
	    deleted []roachpb.Span, added []roachpb.SpanConfigEntry,
        )
    }

The implementation is subtle -- we're not processing one update at a
time. The semantics we're after is applying a batch of updates
atomically on a consistent snapshot of the underlying store. This comes
up in the upcoming spanconfig.Reconciler (#71994) -- there, following a
zone config/descriptor change, we want to update KV state in a single
request/RTT instead of an RTT per descendent table. The intended
usage is better captured in the aforementioned PR; let's now talk
about what this entails for the datastructure.

To apply a single update, we want to find all overlapping spans and
clear out just the intersections. If the update is adding a new span
config, we'll also want to add the corresponding store entry after. We
do this by deleting all overlapping spans in their entirety and
re-adding the non-overlapping segments. Pseudo-code:

  for entry in store.overlapping(update.span):
      union, intersection = union(update.span, entry), intersection(update.span, entry)
      pre  = span{union.start_key, intersection.start_key}
      post = span{intersection.end_key, union.end_key}

      delete {span=entry.span, conf=entry.conf}
      if entry.contains(update.span.start_key):
          # First entry overlapping with update.
          add {span=pre, conf=entry.conf} if non-empty
      if entry.contains(update.span.end_key):
          # Last entry overlapping with update.
          add {span=post, conf=entry.conf} if non-empty

  if adding:
      add {span=update.span, conf=update.conf} # add ourselves

When extending to a set of updates, things are more involved. Let's
assume that the updates are non-overlapping and sorted by start key. As
before, we want to delete overlapping entries in their entirety and
re-add the non-overlapping segments. With multiple updates, it's
possible that a segment being re-added will overlap another update.  If
processing one update at a time in sorted order, we want to only re-add
the gap between the consecutive updates.

    keyspace         a  b  c  d  e  f  g  h  i  j
    existing state      [--------X--------)
    updates          [--A--)           [--B--)

When processing [a,c):A, after deleting [b,h):X, it would be incorrect
to re-add [c,h):X since we're also looking to apply [g,i):B. Instead of
re-adding the trailing segment right away, we carry it forward and
process it when iterating over the second, possibly overlapping update.
In our example, when iterating over [g,i):B we can subtract the overlap
from [c,h):X and only re-add [c,g):X.

It's also possible for the segment to extend past the second update. In
the example below, when processing [d,f):B and having [b,h):X carried
over, we want to re-add [c,d):X and carry forward [f,h):X to the update
after (i.e. [g,i):C)).

    keyspace         a  b  c  d  e  f  g  h  i  j
    existing state      [--------X--------)
    updates          [--A--)  [--B--)  [--C--)

One final note: we're iterating through the updates without actually
applying any mutations. Going back to our first example, when processing
[g,i):B, retrieving the set of overlapping spans would (again) retrieve
[b,h):X -- an entry we've already encountered when processing [a,c):A.
Re-adding non-overlapping segments naively would re-add [b,g):X -- an
entry that overlaps with our last update [a,c):A. When retrieving
overlapping entries, we need to exclude any that overlap with the
segment that was carried over. Pseudo-code:

  carry-over = <empty>
  for update in updates:
      carried-over, carry-over = carry-over, <empty>
      if update.overlap(carried-over):
          # Fill in the gap between consecutive updates.
          add {span=span{carried-over.start_key, update.start_key}, conf=carried-over.conf}
          # Consider the trailing span after update; carry it forward if non-empty.
          carry-over = {span=span{update.end_key, carried-over.end_key}, conf=carried-over.conf}
      else:
          add {span=carried-over.span, conf=carried-over.conf} if non-empty

      for entry in store.overlapping(update.span):
         if entry.overlap(processed):
              continue # already processed

          union, intersection = union(update.span, entry), intersection(update.span, entry)
          pre  = span{union.start_key, intersection.start_key}
          post = span{intersection.end_key, union.end_key}

          delete {span=entry.span, conf=entry.conf}
          if entry.contains(update.span.start_key):
              # First entry overlapping with update.
              add {span=pre, conf=entry.conf} if non-empty
          if entry.contains(update.span.end_key):
              # Last entry overlapping with update.
              carry-over = {span=post, conf=entry.conf}

       if adding:
          add {span=update.span, conf=update.conf} # add ourselves

  add {span=carry-over.span, conf=carry-over.conf} if non-empty

We've extended the randomized testing suite to generate batches of
updates at a time. We've also added a few illustrated datadriven tests.

Release note: None
---
 pkg/kv/kvserver/client_spanconfigs_test.go    |   2 +-
 pkg/roachpb/span_config.go                    |  10 +
 pkg/spanconfig/spanconfig.go                  |  23 +-
 .../spanconfigkvsubscriber/kvsubscriber.go    |   7 +-
 pkg/spanconfig/spanconfigstore/store.go       | 295 +++++++++++++-----
 pkg/spanconfig/spanconfigstore/store_test.go  | 145 ++++++---
 .../spanconfigstore/testdata/batched/basic    |  92 ++++++
 .../testdata/batched/encompass                |  79 +++++
 .../spanconfigstore/testdata/batched/errors   |  55 ++++
 .../testdata/batched/overlapping              | 206 ++++++++++++
 .../spanconfigstore/testdata/batched/straddle | 135 ++++++++
 .../testdata/{ => single}/basic               |  26 +-
 .../spanconfigstore/testdata/single/errors    |  21 ++
 .../testdata/{ => single}/internal            |  12 +-
 .../testdata/{ => single}/overlap             |  22 +-
 .../spanconfigtestutils/BUILD.bazel           |   5 +-
 pkg/spanconfig/spanconfigtestutils/utils.go   |  32 ++
 17 files changed, 1020 insertions(+), 147 deletions(-)
 create mode 100644 pkg/spanconfig/spanconfigstore/testdata/batched/basic
 create mode 100644 pkg/spanconfig/spanconfigstore/testdata/batched/encompass
 create mode 100644 pkg/spanconfig/spanconfigstore/testdata/batched/errors
 create mode 100644 pkg/spanconfig/spanconfigstore/testdata/batched/overlapping
 create mode 100644 pkg/spanconfig/spanconfigstore/testdata/batched/straddle
 rename pkg/spanconfig/spanconfigstore/testdata/{ => single}/basic (82%)
 create mode 100644 pkg/spanconfig/spanconfigstore/testdata/single/errors
 rename pkg/spanconfig/spanconfigstore/testdata/{ => single}/internal (66%)
 rename pkg/spanconfig/spanconfigstore/testdata/{ => single}/overlap (89%)

diff --git a/pkg/kv/kvserver/client_spanconfigs_test.go b/pkg/kv/kvserver/client_spanconfigs_test.go
index 74efc495d655..3995e4748eaf 100644
--- a/pkg/kv/kvserver/client_spanconfigs_test.go
+++ b/pkg/kv/kvserver/client_spanconfigs_test.go
@@ -69,7 +69,7 @@ func TestSpanConfigUpdateAppliedToReplica(t *testing.T) {
 	span := repl.Desc().RSpan().AsRawSpanWithNoLocals()
 	conf := roachpb.SpanConfig{NumReplicas: 5, NumVoters: 3}
 
-	deleted, added := spanConfigStore.Apply(ctx, spanconfig.Update{Span: span, Config: conf}, false /* dryrun */)
+	deleted, added := spanConfigStore.Apply(ctx, false /* dryrun */, spanconfig.Update{Span: span, Config: conf})
 	require.Empty(t, deleted)
 	require.Len(t, added, 1)
 	require.True(t, added[0].Span.Equal(span))
diff --git a/pkg/roachpb/span_config.go b/pkg/roachpb/span_config.go
index 209149ccc1e5..f65d664e7516 100644
--- a/pkg/roachpb/span_config.go
+++ b/pkg/roachpb/span_config.go
@@ -96,6 +96,16 @@ func (c ConstraintsConjunction) String() string {
 	return sb.String()
 }
 
+// Equal compares two span config entries.
+func (s *SpanConfigEntry) Equal(o SpanConfigEntry) bool {
+	return s.Span.Equal(o.Span) && s.Config.Equal(o.Config)
+}
+
+// Empty returns true if the span config entry is empty.
+func (s *SpanConfigEntry) Empty() bool {
+	return s.Equal(SpanConfigEntry{})
+}
+
 // TestingDefaultSpanConfig exports the default span config for testing purposes.
 func TestingDefaultSpanConfig() SpanConfig {
 	return SpanConfig{
diff --git a/pkg/spanconfig/spanconfig.go b/pkg/spanconfig/spanconfig.go
index 469ca83a56aa..710573abc6d3 100644
--- a/pkg/spanconfig/spanconfig.go
+++ b/pkg/spanconfig/spanconfig.go
@@ -192,10 +192,11 @@ type Store interface {
 
 // StoreWriter is the write-only portion of the Store interface.
 type StoreWriter interface {
-	// Apply applies the given update[1]. It also returns the existing spans that
-	// were deleted and entries that were newly added to make room for the
-	// update. The deleted list can double as a list of overlapping spans in the
-	// Store, provided the update is not a no-op[2].
+	// Apply applies a batch of non-overlapping updates atomically[1] and
+	// returns (i) the existing spans that were deleted, and (ii) the entries
+	// that were newly added to make room for the batch. The deleted list can
+	// double as a list of overlapping spans in the Store[2], provided the
+	// updates aren't no-ops.
 	//
 	// Span configs are stored in non-overlapping fashion. When an update
 	// overlaps with existing configs, the existing configs are deleted. If the
@@ -203,7 +204,8 @@ type StoreWriter interface {
 	// configs are re-added. If the update itself is adding an entry, that too
 	// is added. This is best illustrated with the following example:
 	//
-	//                                         [--- X --) is a span with config X
+	//                                        [--- X --) is a span with config X
+	//                                        [xxxxxxxx) is a span being deleted
 	//
 	//  Store    | [--- A ----)[------------- B -----------)[---------- C -----)
 	//  Update   |             [------------------ D -------------)
@@ -212,6 +214,15 @@ type StoreWriter interface {
 	//  Added    |             [------------------ D -------------)[--- C -----)
 	//  Store*   | [--- A ----)[------------------ D -------------)[--- C -----)
 	//
+	// Generalizing to multiple updates:
+	//
+	//  Store    | [--- A ----)[------------- B -----------)[---------- C -----)
+	//  Updates  |             [--- D ----)        [xxxxxxxxx)       [--- E ---)
+	//           |
+	//  Deleted  |             [------------- B -----------)[---------- C -----)
+	//  Added    |             [--- D ----)[-- B --)         [-- C -)[--- E ---)
+	//  Store*   | [--- A ----)[--- D ----)[-- B --)         [-- C -)[--- E ---)
+	//
 	// TODO(irfansharif): We'll make use of the dryrun option in a future PR
 	// when wiring up the reconciliation job to use the KVAccessor. Since the
 	// KVAccessor is a "targeted" API (the spans being deleted/upserted
@@ -255,7 +266,7 @@ type StoreWriter interface {
 	//      against a StoreWriter (populated using KVAccessor contents) using
 	//      the descriptor's span config entry would return empty lists,
 	//      indicating a no-op.
-	Apply(ctx context.Context, update Update, dryrun bool) (
+	Apply(ctx context.Context, dryrun bool, updates ...Update) (
 		deleted []roachpb.Span, added []roachpb.SpanConfigEntry,
 	)
 }
diff --git a/pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go b/pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go
index c9e27df7f937..84d7746626cb 100644
--- a/pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go
+++ b/pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go
@@ -335,7 +335,10 @@ func (s *KVSubscriber) run(ctx context.Context) error {
 			events := buffer.Flush(ctx, frontierTS)
 			s.mu.Lock()
 			for _, ev := range events {
-				s.mu.internal.Apply(ctx, ev.(*bufferEvent).Update, false /* dryrun */)
+				// TODO(irfansharif): We can apply a batch of updates atomically
+				// now that the StoreWriter interface supports it; it'll let us
+				// avoid this mutex.
+				s.mu.internal.Apply(ctx, false /* dryrun */, ev.(*bufferEvent).Update)
 			}
 			handlers := s.mu.handlers
 			s.mu.Unlock()
@@ -353,7 +356,7 @@ func (s *KVSubscriber) run(ctx context.Context) error {
 			events := buffer.Flush(ctx, initialScanTS)
 			freshStore := spanconfigstore.New(s.fallback)
 			for _, ev := range events {
-				freshStore.Apply(ctx, ev.(*bufferEvent).Update, false /* dryrun */)
+				freshStore.Apply(ctx, false /* dryrun */, ev.(*bufferEvent).Update)
 			}
 
 			s.mu.Lock()
diff --git a/pkg/spanconfig/spanconfigstore/store.go b/pkg/spanconfig/spanconfigstore/store.go
index b5eb8bd50c5b..c58bd6b282fc 100644
--- a/pkg/spanconfig/spanconfigstore/store.go
+++ b/pkg/spanconfig/spanconfigstore/store.go
@@ -12,6 +12,7 @@ package spanconfigstore
 
 import (
 	"context"
+	"sort"
 
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -20,6 +21,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/interval"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
+	"github.com/cockroachdb/errors"
 )
 
 // EnabledSetting is a hidden cluster setting to enable the use of the span
@@ -130,23 +132,51 @@ func (s *Store) GetSpanConfigForKey(
 
 // Apply is part of the spanconfig.StoreWriter interface.
 func (s *Store) Apply(
-	ctx context.Context, update spanconfig.Update, dryrun bool,
+	ctx context.Context, dryrun bool, updates ...spanconfig.Update,
 ) (deleted []roachpb.Span, added []roachpb.SpanConfigEntry) {
+	deleted, added, err := s.applyInternal(dryrun, updates...)
+	if err != nil {
+		log.Fatalf(ctx, "%v", err)
+	}
+	return deleted, added
+}
+
+func (s *Store) applyInternal(
+	dryrun bool, updates ...spanconfig.Update,
+) (deleted []roachpb.Span, added []roachpb.SpanConfigEntry, err error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	if !update.Span.Valid() || len(update.Span.EndKey) == 0 {
-		log.Fatalf(ctx, "invalid span")
+	for i := range updates {
+		if !updates[i].Span.Valid() || len(updates[i].Span.EndKey) == 0 {
+			return nil, nil, errors.New("invalid span")
+		}
+	}
+
+	sorted := make([]spanconfig.Update, len(updates))
+	copy(sorted, updates)
+	sort.Slice(sorted, func(i, j int) bool {
+		return sorted[i].Span.Key.Compare(sorted[j].Span.Key) < 0
+	})
+	updates = sorted // re-use the same variable
+
+	for i := range updates {
+		if i == 0 {
+			continue
+		}
+		if updates[i].Span.Overlaps(updates[i-1].Span) {
+			return nil, nil, errors.Newf("found overlapping updates %s and %s", updates[i-1].Span, updates[i].Span)
+		}
 	}
 
-	entriesToDelete, entriesToAdd := s.accumulateOpsForLocked(update)
+	entriesToDelete, entriesToAdd := s.accumulateOpsForLocked(updates)
 
 	deleted = make([]roachpb.Span, len(entriesToDelete))
 	for i := range entriesToDelete {
 		entry := &entriesToDelete[i]
 		if !dryrun {
 			if err := s.mu.tree.Delete(entry, false); err != nil {
-				log.Fatalf(ctx, "%v", err)
+				return nil, nil, err
 			}
 		}
 		deleted[i] = entry.Span
@@ -157,101 +187,208 @@ func (s *Store) Apply(
 		entry := &entriesToAdd[i]
 		if !dryrun {
 			if err := s.mu.tree.Insert(entry, false); err != nil {
-				log.Fatalf(ctx, "%v", err)
+				return nil, nil, err
 			}
 		}
 		added[i] = entry.SpanConfigEntry
 	}
 
-	return deleted, added
+	return deleted, added, nil
 }
 
 // accumulateOpsForLocked returns the list of store entries that would be
-// deleted and added if the given update was to be applied. To apply a given
-// update, we want to find all overlapping spans and clear out just the
-// intersections. If the update is adding a new span config, we'll also want to
-// add it store entry after. We do this by deleting all overlapping spans in
-// their entirety and re-adding the non-overlapping portions, if any.
-// Pseudo-code:
+// deleted and added if the given set of updates were to be applied.
+//
+// To apply a single update, we want to find all overlapping spans and clear out
+// just the intersections. If the update is adding a new span config, we'll also
+// want to add the corresponding store entry after. We do this by deleting all
+// overlapping spans in their entirety and re-adding the non-overlapping
+// segments. Pseudo-code:
 //
-// 	for entry in store.overlapping(update.span):
-// 		union, intersection = union(update.span, entry), intersection(update.span, entry)
-// 		pre, post = span{union.start_key, intersection.start_key}, span{intersection.end_key, union.end_key}
+//   for entry in store.overlapping(update.span):
+//       union, intersection = union(update.span, entry), intersection(update.span, entry)
+//       pre  = span{union.start_key, intersection.start_key}
+//       post = span{intersection.end_key, union.end_key}
 //
-// 		delete entry
-// 		if entry.contains(update.span.start_key):
-// 			add pre=entry.conf
-// 		if entry.contains(update.span.end_key):
-// 			add post=entry.conf
+//       delete {span=entry.span, conf=entry.conf}
+//       if entry.contains(update.span.start_key):
+//           # First entry overlapping with update.
+//           add {span=pre, conf=entry.conf} if non-empty
+//       if entry.contains(update.span.end_key):
+//           # Last entry overlapping with update.
+//           add {span=post, conf=entry.conf} if non-empty
 //
 //   if adding:
-//       add update.span=update.conf
+//       add {span=update.span, conf=update.conf} # add ourselves
+//
+// When extending to a set of updates, things are more involved (but only
+// slightly!). Let's assume that the updates are non-overlapping and sorted
+// by start key. As before, we want to delete overlapping entries in their
+// entirety and re-add the non-overlapping segments. With multiple updates, it's
+// possible that a segment being re-added will overlap another update. If
+// processing one update at a time in sorted order, we want to only re-add the
+// gap between the consecutive updates.
+//
+//   keyspace         a  b  c  d  e  f  g  h  i  j
+//   existing state      [--------X--------)
+//   updates          [--A--)           [--B--)
+//
+// When processing [a,c):A, after deleting [b,h):X, it would be incorrect to
+// re-add [c,h):X since we're also looking to apply [g,i):B. Instead of
+// re-adding the trailing segment right away, we carry it forward and process it
+// when iterating over the second, possibly overlapping update. In our example,
+// when iterating over [g,i):B we can subtract the overlap from [c,h):X and only
+// re-add [c,g):X.
+//
+// It's also possible for the segment to extend past the second update. In the
+// example below, when processing [d,f):B and having [b,h):X carried over, we
+// want to re-add [c,d):X and carry forward [f,h):X to the update after (i.e.
+// [g,i):C)).
+//
+//   keyspace         a  b  c  d  e  f  g  h  i  j
+//   existing state      [--------X--------)
+//   updates          [--A--)  [--B--)  [--C--)
+//
+// One final note: we're iterating through the updates without actually applying
+// any mutations. Going back to our first example, when processing [g,i):B,
+// retrieving the set of overlapping spans would (again) retrieve [b,h):X -- an
+// entry we've already encountered when processing [a,c):A. Re-adding
+// non-overlapping segments naively would re-add [b,g):X -- an entry that
+// overlaps with our last update [a,c):A. When retrieving overlapping entries,
+// we need to exclude any that overlap with the segment that was carried over.
+// Pseudo-code:
+//
+//   carry-over = <empty>
+//   for update in updates:
+//       carried-over, carry-over = carry-over, <empty>
+//       if update.overlap(carried-over):
+//           # Fill in the gap between consecutive updates.
+//           add {span=span{carried-over.start_key, update.start_key}, conf=carried-over.conf}
+//           # Consider the trailing span after update; carry it forward if non-empty.
+//           carry-over = {span=span{update.end_key, carried-over.end_key}, conf=carried-over.conf}
+//       else:
+//           add {span=carried-over.span, conf=carried-over.conf} if non-empty
+//
+//       for entry in store.overlapping(update.span):
+//          if entry.overlap(processed):
+//               continue # already processed
+//
+//           union, intersection = union(update.span, entry), intersection(update.span, entry)
+//           pre  = span{union.start_key, intersection.start_key}
+//           post = span{intersection.end_key, union.end_key}
+//
+//           delete {span=entry.span, conf=entry.conf}
+//           if entry.contains(update.span.start_key):
+//               # First entry overlapping with update.
+//               add {span=pre, conf=entry.conf} if non-empty
+//           if entry.contains(update.span.end_key):
+//               # Last entry overlapping with update.
+//               carry-over = {span=post, conf=entry.conf}
 //
-func (s *Store) accumulateOpsForLocked(update spanconfig.Update) (toDelete, toAdd []storeEntry) {
-	for _, overlapping := range s.mu.tree.Get(update.Span.AsRange()) {
-		existing := overlapping.(*storeEntry)
-		var (
-			union = existing.Span.Combine(update.Span)
-			inter = existing.Span.Intersect(update.Span)
-
-			pre  = roachpb.Span{Key: union.Key, EndKey: inter.Key}
-			post = roachpb.Span{Key: inter.EndKey, EndKey: union.EndKey}
-		)
-
-		// Delete the existing span in its entirety. Below we'll re-add the
-		// non-intersecting parts of the span.
-		toDelete = append(toDelete, *existing)
-
-		if existing.Span.ContainsKey(update.Span.Key) { // existing entry contains the update span's start key
-			// ex:     [-----------------)
-			//
-			// up:         [-------)
-			// up:         [-------------)
-			// up:         [--------------
-			// up:     [-------)
-			// up:     [-----------------)
-			// up:     [------------------
-
-			// Re-add the non-intersecting span, if any.
-			if pre.Valid() {
-				toAdd = append(toAdd, s.makeEntryLocked(pre, existing.Config))
+//        if adding:
+//           add {span=update.span, conf=update.conf} # add ourselves
+//
+//   add {span=carry-over.span, conf=carry-over.conf} if non-empty
+//
+func (s *Store) accumulateOpsForLocked(updates []spanconfig.Update) (toDelete, toAdd []storeEntry) {
+	var carryOver roachpb.SpanConfigEntry
+	for _, update := range updates {
+		var carriedOver roachpb.SpanConfigEntry
+		carriedOver, carryOver = carryOver, roachpb.SpanConfigEntry{}
+		if update.Span.Overlaps(carriedOver.Span) {
+			gapBetweenUpdates := roachpb.Span{Key: carriedOver.Span.Key, EndKey: update.Span.Key}
+			if gapBetweenUpdates.Valid() {
+				toAdd = append(toAdd, s.makeEntryLocked(gapBetweenUpdates, carriedOver.Config))
 			}
+
+			carryOverSpanAfterUpdate := roachpb.Span{Key: update.Span.EndKey, EndKey: carriedOver.Span.EndKey}
+			if carryOverSpanAfterUpdate.Valid() {
+				carryOver = roachpb.SpanConfigEntry{
+					Span:   carryOverSpanAfterUpdate,
+					Config: carriedOver.Config,
+				}
+			}
+		} else if !carriedOver.Empty() {
+			toAdd = append(toAdd, s.makeEntryLocked(carriedOver.Span, carriedOver.Config))
 		}
 
-		if existing.Span.ContainsKey(update.Span.EndKey) { // existing entry contains the update span's end key
-			// ex:     [-----------------)
-			//
-			// up:     -------------)
-			// up:     [------------)
-			// up:        [---------)
+		skipAddingSelf := false
+		for _, overlapping := range s.mu.tree.Get(update.Span.AsRange()) {
+			existing := overlapping.(*storeEntry)
+			if existing.Span.Overlaps(carriedOver.Span) {
+				continue // we've already processed this entry above.
+			}
 
-			// Re-add the non-intersecting span.
-			toAdd = append(toAdd, s.makeEntryLocked(post, existing.Config))
+			var (
+				union = existing.Span.Combine(update.Span)
+				inter = existing.Span.Intersect(update.Span)
+
+				pre  = roachpb.Span{Key: union.Key, EndKey: inter.Key}
+				post = roachpb.Span{Key: inter.EndKey, EndKey: union.EndKey}
+			)
+
+			if update.Addition() {
+				if existing.Span.Equal(update.Span) && existing.Config.Equal(update.Config) {
+					skipAddingSelf = true
+					break // no-op; peep-hole optimization
+				}
+			}
+
+			// Delete the existing span in its entirety. Below we'll re-add the
+			// non-intersecting parts of the span.
+			toDelete = append(toDelete, *existing)
+			if existing.Span.ContainsKey(update.Span.Key) { // existing entry contains the update span's start key
+				// ex:     [-----------------)
+				//
+				// up:         [-------)
+				// up:         [-------------)
+				// up:         [--------------
+				// up:     [-------)
+				// up:     [-----------------)
+				// up:     [------------------
+
+				// Re-add the non-intersecting span, if any.
+				if pre.Valid() {
+					toAdd = append(toAdd, s.makeEntryLocked(pre, existing.Config))
+				}
+			}
+
+			if existing.Span.ContainsKey(update.Span.EndKey) { // existing entry contains the update span's end key
+				// ex:     [-----------------)
+				//
+				// up:     -------------)
+				// up:     [------------)
+				// up:        [---------)
+
+				// Carry over the non-intersecting span.
+				carryOver = roachpb.SpanConfigEntry{
+					Span:   post,
+					Config: existing.Config,
+				}
+			}
 		}
-	}
 
-	if update.Addition() {
-		if len(toDelete) == 1 &&
-			toDelete[0].Span.Equal(update.Span) &&
-			toDelete[0].Config.Equal(update.Config) {
-			// We're deleting exactly what we're going to add, this is a no-op.
-			return nil, nil
+		if update.Addition() && !skipAddingSelf {
+			// Add the update itself.
+			toAdd = append(toAdd, s.makeEntryLocked(update.Span, update.Config))
+
+			// TODO(irfansharif): If we're adding an entry, we could inspect the
+			// entries before and after and check whether either of them have
+			// the same config. If they do, we could coalesce them into a single
+			// span. Given that these boundaries determine where we split
+			// ranges, we'd be able to reduce the number of ranges drastically
+			// (think adjacent tables/indexes/partitions with the same config).
+			// This would be especially significant for secondary tenants, where
+			// we'd be able to avoid unconditionally splitting on table
+			// boundaries. We'd still want to split on tenant boundaries, so
+			// certain preconditions would need to hold. For performance
+			// reasons, we'd probably also want to offer a primitive to allow
+			// manually splitting on specific table boundaries.
 		}
+	}
 
-		// Add the update itself.
-		toAdd = append(toAdd, s.makeEntryLocked(update.Span, update.Config))
-
-		// TODO(irfansharif): If we're adding an entry, we could inspect the
-		// entries before and after and check whether either of them have the
-		// same config. If they do, we could coalesce them into a single span.
-		// Given that these boundaries determine where we split ranges, we'd be
-		// able to reduce the number of ranges drastically (think adjacent
-		// tables/indexes/partitions with the same config). This would be
-		// especially significant for secondary tenants, where we'd be able to
-		// avoid unconditionally splitting on table boundaries. We'd still want
-		// to split on tenant boundaries, so certain preconditions would need to
-		// hold. For performance reasons, we'd probably also want to offer
-		// a primitive to allow manually splitting on specific table boundaries.
+	if !carryOver.Empty() {
+		toAdd = append(toAdd, s.makeEntryLocked(carryOver.Span, carryOver.Config))
 	}
 
 	return toDelete, toAdd
diff --git a/pkg/spanconfig/spanconfigstore/store_test.go b/pkg/spanconfig/spanconfigstore/store_test.go
index 7e4ee3d45222..4454c7cc6c91 100644
--- a/pkg/spanconfig/spanconfigstore/store_test.go
+++ b/pkg/spanconfig/spanconfigstore/store_test.go
@@ -14,6 +14,7 @@ import (
 	"context"
 	"fmt"
 	"math/rand"
+	"sort"
 	"strings"
 	"testing"
 
@@ -44,6 +45,46 @@ func (s *Store) TestingGetAllOverlapping(
 	return res
 }
 
+// TestingApplyInternal exports an internal method for testing purposes.
+func (s *Store) TestingApplyInternal(
+	_ context.Context, dryrun bool, updates ...spanconfig.Update,
+) (deleted []roachpb.Span, added []roachpb.SpanConfigEntry, err error) {
+	return s.applyInternal(dryrun, updates...)
+}
+
+// TestDataDriven runs datadriven tests against the Store interface.
+// The syntax is as follows:
+//
+// 		apply
+// 		delete [a,c)
+// 		set [c,h):X
+// 		----
+// 		deleted [b,d)
+// 		deleted [e,g)
+// 		added [c,h):X
+//
+// 		get key=b
+// 		----
+// 		conf=A # or conf=FALLBACK if the key is not present
+//
+// 		needs-split span=[b,h)
+// 		----
+// 		true
+//
+// 		compute-split span=[b,h)
+// 		----
+// 		key=c
+//
+// 		overlapping span=[b,h)
+// 		----
+// 		[b,d):A
+// 		[d,f):B
+// 		[f,h):A
+//
+//
+// Text of the form [a,b) and [a,b):C correspond to spans and span config
+// entries; see spanconfigtestutils.Parse{Span,Config,SpanConfigEntry} for more
+// details.
 func TestDataDriven(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
@@ -51,33 +92,20 @@ func TestDataDriven(t *testing.T) {
 	datadriven.Walk(t, testutils.TestDataPath(t), func(t *testing.T, path string) {
 		store := New(spanconfigtestutils.ParseConfig(t, "FALLBACK"))
 		datadriven.RunTest(t, path, func(t *testing.T, d *datadriven.TestData) string {
-			var (
-				spanStr, confStr, keyStr string
-			)
+			var spanStr, keyStr string
 			switch d.Cmd {
-			case "set":
-				d.ScanArgs(t, "span", &spanStr)
-				d.ScanArgs(t, "conf", &confStr)
-				span, config := spanconfigtestutils.ParseSpan(t, spanStr), spanconfigtestutils.ParseConfig(t, confStr)
-
+			case "apply":
+				updates := spanconfigtestutils.ParseStoreApplyArguments(t, d.Input)
 				dryrun := d.HasArg("dryrun")
-				deleted, added := store.Apply(ctx, spanconfig.Update{Span: span, Config: config}, dryrun)
-
-				var b strings.Builder
-				for _, sp := range deleted {
-					b.WriteString(fmt.Sprintf("deleted %s\n", spanconfigtestutils.PrintSpan(sp)))
-				}
-				for _, ent := range added {
-					b.WriteString(fmt.Sprintf("added %s\n", spanconfigtestutils.PrintSpanConfigEntry(ent)))
+				deleted, added, err := store.TestingApplyInternal(ctx, dryrun, updates...)
+				if err != nil {
+					return fmt.Sprintf("err: %v", err)
 				}
-				return b.String()
-
-			case "delete":
-				d.ScanArgs(t, "span", &spanStr)
-				span := spanconfigtestutils.ParseSpan(t, spanStr)
 
-				dryrun := d.HasArg("dryrun")
-				deleted, added := store.Apply(ctx, spanconfig.Update{Span: span}, dryrun)
+				sort.Sort(roachpb.Spans(deleted))
+				sort.Slice(added, func(i, j int) bool {
+					return added[i].Span.Key.Compare(added[j].Span.Key) < 0
+				})
 
 				var b strings.Builder
 				for _, sp := range deleted {
@@ -119,9 +147,9 @@ func TestDataDriven(t *testing.T) {
 				return strings.Join(results, "\n")
 
 			default:
+				t.Fatalf("unknown command: %s", d.Cmd)
 			}
 
-			t.Fatalf("unknown command: %s", d.Cmd)
 			return ""
 		})
 	})
@@ -163,6 +191,47 @@ func TestRandomized(t *testing.T) {
 		return ops[rand.Intn(2)]
 	}
 
+	getRandomUpdate := func() spanconfig.Update {
+		sp, conf, op := genRandomSpan(), getRandomConf(), getRandomOp()
+		switch op {
+		case "set":
+			return spanconfig.Update{Span: sp, Config: conf}
+		case "del":
+			return spanconfig.Update{Span: sp}
+		default:
+		}
+		t.Fatalf("unexpected op: %s", op)
+		return spanconfig.Update{}
+	}
+
+	getRandomUpdates := func() []spanconfig.Update {
+		numUpdates := 1 + rand.Intn(3)
+		updates := make([]spanconfig.Update, numUpdates)
+		for {
+			for i := 0; i < numUpdates; i++ {
+				updates[i] = getRandomUpdate()
+			}
+			sort.Slice(updates, func(i, j int) bool {
+				return updates[i].Span.Key.Compare(updates[j].Span.Key) < 0
+			})
+			invalid := false
+			for i := 1; i < numUpdates; i++ {
+				if updates[i].Span.Overlaps(updates[i-1].Span) {
+					invalid = true
+				}
+			}
+
+			if invalid {
+				continue // try again
+			}
+
+			rand.Shuffle(len(updates), func(i, j int) {
+				updates[i], updates[j] = updates[j], updates[i]
+			})
+			return updates
+		}
+	}
+
 	testSpan := spanconfigtestutils.ParseSpan(t, "[f,g)") // pin a single character span to test with
 	var expConfig roachpb.SpanConfig
 	var expFound bool
@@ -170,20 +239,16 @@ func TestRandomized(t *testing.T) {
 	const numOps = 5000
 	store := New(roachpb.TestingDefaultSpanConfig())
 	for i := 0; i < numOps; i++ {
-		sp, conf, op := genRandomSpan(), getRandomConf(), getRandomOp()
-		switch op {
-		case "set":
-			store.Apply(ctx, spanconfig.Update{Span: sp, Config: conf}, false)
-			if testSpan.Overlaps(sp) {
-				expConfig, expFound = conf, true
-			}
-		case "del":
-			store.Apply(ctx, spanconfig.Update{Span: sp}, false)
-			if testSpan.Overlaps(sp) {
-				expConfig, expFound = roachpb.SpanConfig{}, false
+		updates := getRandomUpdates()
+		store.Apply(ctx, false /* dryrun */, updates...)
+		for _, update := range updates {
+			if testSpan.Overlaps(update.Span) {
+				if update.Addition() {
+					expConfig, expFound = update.Config, true
+				} else {
+					expConfig, expFound = roachpb.SpanConfig{}, false
+				}
 			}
-		default:
-			t.Fatalf("unexpected op: %s", op)
 		}
 	}
 
@@ -220,6 +285,12 @@ func TestRandomized(t *testing.T) {
 			continue
 		}
 
+		// All spans are expected to be valid.
+		require.True(t, cur.Span.Valid(),
+			"expected to only find valid spans, found %s",
+			spanconfigtestutils.PrintSpan(cur.Span),
+		)
+
 		// Span configs are returned in strictly sorted order.
 		require.True(t, last.Span.Key.Compare(cur.Span.Key) < 0,
 			"expected to find spans in strictly sorted order, found %s then %s",
diff --git a/pkg/spanconfig/spanconfigstore/testdata/batched/basic b/pkg/spanconfig/spanconfigstore/testdata/batched/basic
new file mode 100644
index 000000000000..cf04bf4441f8
--- /dev/null
+++ b/pkg/spanconfig/spanconfigstore/testdata/batched/basic
@@ -0,0 +1,92 @@
+# Test semantics of batched updates (multiple sets or deletes applied on a snapshot).
+
+# Test that dryruns don't actually mutate anything.
+apply dryrun
+set [b,d):A
+set [f,h):B
+----
+added [b,d):A
+added [f,h):B
+
+get key=b
+----
+conf=FALLBACK
+
+get key=g
+----
+conf=FALLBACK
+
+
+# Add span configs for real.
+apply
+set [b,d):A
+set [f,h):B
+----
+added [b,d):A
+added [f,h):B
+
+get key=a
+----
+conf=FALLBACK
+
+get key=b
+----
+conf=A
+
+get key=g
+----
+conf=B
+
+
+# Check that no-ops shows up as much.
+apply
+set [b,d):A
+set [f,h):B
+----
+
+
+# Check that a delete dryrun does nothing, though emitting the right operations.
+apply dryrun
+delete [f,h)
+delete [c,d)
+----
+deleted [b,d)
+deleted [f,h)
+added [b,c):A
+
+get key=f
+----
+conf=B
+
+
+# Delete a span for real.
+apply
+delete [f,h)
+delete [c,d)
+----
+deleted [b,d)
+deleted [f,h)
+added [b,c):A
+
+# Check for no-ops again.
+apply
+delete [f,g)
+delete [c,d)
+----
+
+# Check that keys are as we'd expect (including the deleted one).
+get key=b
+----
+conf=A
+
+get key=c
+----
+conf=FALLBACK
+
+get key=f
+----
+conf=FALLBACK
+
+get key=g
+----
+conf=FALLBACK
diff --git a/pkg/spanconfig/spanconfigstore/testdata/batched/encompass b/pkg/spanconfig/spanconfigstore/testdata/batched/encompass
new file mode 100644
index 000000000000..e3e44d0e6300
--- /dev/null
+++ b/pkg/spanconfig/spanconfigstore/testdata/batched/encompass
@@ -0,0 +1,79 @@
+# Test semantics of batched updates that overlap with and/or encompass
+# spans already present.
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state
+# set        [-A)  [-B|-C)  [--D--)
+# ====================================
+# result     [-A)  [-B|-C)  [--D--)
+apply
+set [b,c):A
+set [d,e):B
+set [e,f):C
+set [g,i):D
+----
+added [b,c):A
+added [d,e):B
+added [e,f):C
+added [g,i):D
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-A)  [-B|-C)  [--D--)
+# set     [--------X--------)
+# ====================================
+# result  [--------X--------|--D--)
+apply dryrun
+set [a,g):X
+----
+deleted [b,c)
+deleted [d,e)
+deleted [e,f)
+added [a,g):X
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-A)  [-B|-C)  [--D--)
+# set     [--------X--|----Y---)
+# ====================================
+# result  [--------X--|----Y---|-D)
+apply dryrun
+set [a,e):X
+set [e,h):Y
+----
+deleted [b,c)
+deleted [d,e)
+deleted [e,f)
+deleted [g,i)
+added [a,e):X
+added [e,h):Y
+added [h,i):D
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-A)  [-B|-C)  [--D--)
+# set     [--------X-----------)
+# ====================================
+# result  [--------X-----------|-D)
+apply dryrun
+set [a,h):X
+----
+deleted [b,c)
+deleted [d,e)
+deleted [e,f)
+deleted [g,i)
+added [a,h):X
+added [h,i):D
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-A)  [-B|-C)  [--D--)
+# set     [--------X-----|xxxxx)
+# ====================================
+# result  [--------X-----)     [-D)
+apply dryrun
+set [a,f):X
+delete [f,h)
+----
+deleted [b,c)
+deleted [d,e)
+deleted [e,f)
+deleted [g,i)
+added [a,f):X
+added [h,i):D
diff --git a/pkg/spanconfig/spanconfigstore/testdata/batched/errors b/pkg/spanconfig/spanconfigstore/testdata/batched/errors
new file mode 100644
index 000000000000..78bef3959d22
--- /dev/null
+++ b/pkg/spanconfig/spanconfigstore/testdata/batched/errors
@@ -0,0 +1,55 @@
+# Ensure we get an error if updates overlap.
+
+apply
+set [a,c):A
+set [a,c):B
+----
+err: found overlapping updates {a-c} and {a-c}
+
+apply
+set [a,c):A
+set [b,c):B
+----
+err: found overlapping updates {a-c} and {b-c}
+
+apply
+set [a,c):A
+set [b,d):B
+----
+err: found overlapping updates {a-c} and {b-d}
+
+apply
+delete [a,c)
+delete [a,c)
+----
+err: found overlapping updates {a-c} and {a-c}
+
+apply
+delete [a,c)
+delete [b,c)
+----
+err: found overlapping updates {a-c} and {b-c}
+
+apply
+delete [a,c)
+delete [b,d)
+----
+err: found overlapping updates {a-c} and {b-d}
+
+apply
+set [a,c):A
+delete [a,c)
+----
+err: found overlapping updates {a-c} and {a-c}
+
+apply
+delete [a,c)
+set [b,c):A
+----
+err: found overlapping updates {a-c} and {b-c}
+
+apply
+set [a,c):A
+delete [b,d)
+----
+err: found overlapping updates {a-c} and {b-d}
diff --git a/pkg/spanconfig/spanconfigstore/testdata/batched/overlapping b/pkg/spanconfig/spanconfigstore/testdata/batched/overlapping
new file mode 100644
index 000000000000..dd54c2b84a15
--- /dev/null
+++ b/pkg/spanconfig/spanconfigstore/testdata/batched/overlapping
@@ -0,0 +1,206 @@
+# Test semantics of batched updates that partially overlap with what's already
+# present.
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state
+# set        [-----X--------)
+# ====================================
+# result     [-----X--------)
+apply
+set [b,g):X
+----
+added [b,g):X
+
+overlapping span=[a,z)
+----
+[b,g):X
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set     [--A--)
+# ====================================
+# result  [--A--)[-----X----)
+apply dryrun
+set [a,c):A
+----
+deleted [b,g)
+added [a,c):A
+added [c,g):X
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set     [--A--)              [--B--)
+# ====================================
+# result  [--A--)[-----X----)  [--B--)
+apply dryrun
+set [a,c):A
+set [h,j):B
+----
+deleted [b,g)
+added [a,c):A
+added [c,g):X
+added [h,j):B
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set     [--A--)        [--B--)
+# ====================================
+# result  [--A--)[-X-----|--B--)
+apply dryrun
+set [a,c):A
+set [f,h):B
+----
+deleted [b,g)
+added [a,c):A
+added [c,f):X
+added [f,h):B
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set     [--A--)     [--B--)
+# ====================================
+# result  [--A--)[-X--|--B--)
+apply dryrun
+set [a,c):A
+set [e,g):B
+----
+deleted [b,g)
+added [a,c):A
+added [c,e):X
+added [e,g):B
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set     [--A--)  [--B--)
+# ====================================
+# result  [--A--)[X|--B--|-X)
+apply dryrun
+set [a,c):A
+set [d,f):B
+----
+deleted [b,g)
+added [a,c):A
+added [c,d):X
+added [d,f):B
+added [f,g):X
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set           [--A--|--B--)
+# ====================================
+# result     [-X|--A--|--B--)
+apply dryrun
+set [c,e):A
+set [e,g):B
+----
+deleted [b,g)
+added [b,c):X
+added [c,e):A
+added [e,g):B
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set     [--A--)  [-B)  [--D--)
+# ====================================
+# set     [--A--|-X|-B|-X|--D--)
+apply dryrun
+set [a,c):A
+set [d,e):B
+set [f,h):D
+----
+deleted [b,g)
+added [a,c):A
+added [c,d):X
+added [d,e):B
+added [e,f):X
+added [f,h):D
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set     [--A--)  [-B)  [-D)
+# ====================================
+# set     [--A--|-X|-B|-X|-D)
+apply dryrun
+set [a,c):A
+set [d,e):B
+set [f,g):D
+----
+deleted [b,g)
+added [a,c):A
+added [c,d):X
+added [d,e):B
+added [e,f):X
+added [f,g):D
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set     [--A--)  [-B|-C|--D--)
+# ====================================
+# set     [--A--|-X|-B|-C|--D--)
+apply dryrun
+set [a,c):A
+set [d,e):B
+set [e,f):C
+set [f,h):D
+----
+deleted [b,g)
+added [a,c):A
+added [c,d):X
+added [d,e):B
+added [e,f):C
+added [f,h):D
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set     [--A--)              [xxxxx)
+# ====================================
+# result  [--A--)[-----X----)
+apply dryrun
+set [a,c):A
+delete [h,j)
+----
+deleted [b,g)
+added [a,c):A
+added [c,g):X
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set     [--A--)        [xxxxx)
+# ====================================
+# result  [--A--)[-X-----)
+apply dryrun
+set [a,c):A
+delete [f,h)
+----
+deleted [b,g)
+added [a,c):A
+added [c,f):X
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set     [xxxxx)     [--B--)
+# ====================================
+# result         [-X--|--B--)
+apply dryrun
+delete [a,c)
+set [e,g):B
+----
+deleted [b,g)
+added [c,e):X
+added [e,g):B
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [-----X--------)
+# set     [--A--)  [xx)  [--D--)
+# ====================================
+# set     [--A--|-X)  [-X|--D--)
+apply dryrun
+set [a,c):A
+delete [d,e)
+set [f,h):D
+----
+deleted [b,g)
+added [a,c):A
+added [c,d):X
+added [e,f):X
+added [f,h):D
diff --git a/pkg/spanconfig/spanconfigstore/testdata/batched/straddle b/pkg/spanconfig/spanconfigstore/testdata/batched/straddle
new file mode 100644
index 000000000000..69e8011b4392
--- /dev/null
+++ b/pkg/spanconfig/spanconfigstore/testdata/batched/straddle
@@ -0,0 +1,135 @@
+# Test semantics of batched updates that partially overlap with what's already
+# present.
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state
+# set        [--A--)  [--B--)
+# ====================================
+# result     [--A--)  [--B--)
+apply
+set [b,d):A
+set [e,g):B
+----
+added [b,d):A
+added [e,g):B
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [--A--)  [--B--)
+# set           [---X----)
+# ====================================
+# result     [-A|----X---|-B)
+apply dryrun
+set [c,f):X
+----
+deleted [b,d)
+deleted [e,g)
+added [b,c):A
+added [c,f):X
+added [f,g):B
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [--A--)  [--B--)
+# set           [---X-------)
+# ====================================
+# result     [-A|----X------)
+apply dryrun
+set [c,g):X
+----
+deleted [b,d)
+deleted [e,g)
+added [b,c):A
+added [c,g):X
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [--A--)  [--B--)
+# set           [---X----|xx)
+# ====================================
+# result     [-A|----X---)
+apply dryrun
+set [c,f):X
+delete [f,g)
+----
+deleted [b,d)
+deleted [e,g)
+added [b,c):A
+added [c,f):X
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [--A--)  [--B--)
+# set           [---X----------)
+# ====================================
+# result     [-A|----X---------)
+apply dryrun
+set [c,h):X
+----
+deleted [b,d)
+deleted [e,g)
+added [b,c):A
+added [c,h):X
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [--A--)  [--B--)
+# set        [---X-------)
+# ====================================
+# result     [-------X---|-B)
+apply dryrun
+set [b,f):X
+----
+deleted [b,d)
+deleted [e,g)
+added [b,f):X
+added [f,g):B
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [--A--)  [--B--)
+# set     [------X-------)
+# ====================================
+# result  [----------X---|-B)
+apply dryrun
+set [a,f):X
+----
+deleted [b,d)
+deleted [e,g)
+added [a,f):X
+added [f,g):B
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [--A--)  [--B--)
+# set           [xxxxxxxx)
+# ====================================
+# result     [-A)        [-B)
+apply dryrun
+delete [c,f)
+----
+deleted [b,d)
+deleted [e,g)
+added [b,c):A
+added [f,g):B
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [--A--)  [--B--)
+# set     [xxxxx|---X----------)
+# ====================================
+# result        [----X---------)
+apply dryrun
+delete [a,c)
+set [c,h):X
+----
+deleted [b,d)
+deleted [e,g)
+added [c,h):X
+
+# keys    a  b  c  d  e  f  g  h  i  j
+# state      [--A--)  [--B--)
+# set        [--X--|xxxxx|-Y)
+# ====================================
+# result     [--X--)     [-Y)
+apply dryrun
+set [b,d):X
+delete [d,f)
+set [f,g):Y
+----
+deleted [b,d)
+deleted [e,g)
+added [b,d):X
+added [f,g):Y
diff --git a/pkg/spanconfig/spanconfigstore/testdata/basic b/pkg/spanconfig/spanconfigstore/testdata/single/basic
similarity index 82%
rename from pkg/spanconfig/spanconfigstore/testdata/basic
rename to pkg/spanconfig/spanconfigstore/testdata/single/basic
index 2f894d29d056..78195d9f3dfd 100644
--- a/pkg/spanconfig/spanconfigstore/testdata/basic
+++ b/pkg/spanconfig/spanconfigstore/testdata/single/basic
@@ -1,6 +1,6 @@
 # Test basic get/set/delete operations where the spans retrieved are identical
 # to the ones being added/deleted, and are non-overlapping with respect to one
-# another.
+# another. Only a single update is applied at a time.
 
 # Check that missing keys fallback to a static config.
 get key=b
@@ -9,7 +9,8 @@ conf=FALLBACK
 
 
 # Test that dryruns don't actually mutate anything.
-set span=[b,d) conf=A dryrun
+apply dryrun
+set [b,d):A
 ----
 added [b,d):A
 
@@ -19,17 +20,20 @@ conf=FALLBACK
 
 
 # Add span configs for real.
-set span=[b,d) conf=A
+apply
+set [b,d):A
 ----
 added [b,d):A
 
-set span=[f,h) conf=B
+apply
+set [f,h):B
 ----
 added [f,h):B
 
 
 # Check that a no-op operation shows up as much.
-set span=[f,h) conf=B
+apply
+set [f,h):B
 ----
 
 
@@ -56,7 +60,8 @@ conf=FALLBACK
 
 
 # Check that a delete dryrun does nothing.
-delete span=[f,h) dryrun
+apply dryrun
+delete [f,h)
 ----
 deleted [f,h)
 
@@ -66,15 +71,18 @@ conf=B
 
 
 # Delete a span for real.
-delete span=[f,h)
+apply
+delete [f,h)
 ----
 deleted [f,h)
 
 # Check that a no-op operation does nothing.
-delete span=[f,g)
+apply
+delete [f,g)
 ----
 
-delete span=[f,h)
+apply
+delete [f,h)
 ----
 
 # Check that keys are as we'd expect (including the deleted one).
diff --git a/pkg/spanconfig/spanconfigstore/testdata/single/errors b/pkg/spanconfig/spanconfigstore/testdata/single/errors
new file mode 100644
index 000000000000..7eabee154dec
--- /dev/null
+++ b/pkg/spanconfig/spanconfigstore/testdata/single/errors
@@ -0,0 +1,21 @@
+# Ensure we get an error if updates are over invalid spans.
+
+apply
+set [c,a):A
+----
+err: invalid span
+
+apply
+set [c,c):A
+----
+err: invalid span
+
+apply
+delete [c,a)
+----
+err: invalid span
+
+apply
+delete [c,c)
+----
+err: invalid span
diff --git a/pkg/spanconfig/spanconfigstore/testdata/internal b/pkg/spanconfig/spanconfigstore/testdata/single/internal
similarity index 66%
rename from pkg/spanconfig/spanconfigstore/testdata/internal
rename to pkg/spanconfig/spanconfigstore/testdata/single/internal
index 5cffaf80713c..e84be82d1e31 100644
--- a/pkg/spanconfig/spanconfigstore/testdata/internal
+++ b/pkg/spanconfig/spanconfigstore/testdata/single/internal
@@ -1,13 +1,16 @@
-# Test the store's internal view of overlapping span configs.
+# Test the store's internal view of overlapping span configs. Only a single
+# update is applied at a time.
 
 overlapping span=[a,z)
 ----
 
-set span=[b,d) conf=A
+apply
+set [b,d):A
 ----
 added [b,d):A
 
-set span=[f,g) conf=B
+apply
+set [f,g):B
 ----
 added [f,g):B
 
@@ -30,7 +33,8 @@ overlapping span=[a,j)
 [b,d):A
 [f,g):B
 
-delete span=[f,g)
+apply
+delete [f,g)
 ----
 deleted [f,g)
 
diff --git a/pkg/spanconfig/spanconfigstore/testdata/overlap b/pkg/spanconfig/spanconfigstore/testdata/single/overlap
similarity index 89%
rename from pkg/spanconfig/spanconfigstore/testdata/overlap
rename to pkg/spanconfig/spanconfigstore/testdata/single/overlap
index 486357ed7e7f..2f5f92433960 100644
--- a/pkg/spanconfig/spanconfigstore/testdata/overlap
+++ b/pkg/spanconfig/spanconfigstore/testdata/single/overlap
@@ -1,18 +1,21 @@
-# Test operations where the spans overlap with the existing ones.
+# Test operations where the spans overlap with the existing ones. Only a single
+# update is applied at a time.
 
-set span=[b,h) conf=A
+apply
+set [b,h):A
 ----
 added [b,h):A
 
 
 # Check that writing a span with a partial overlap first deletes the existing
 # entry and adds three new ones.
-set span=[d,f) conf=B
+apply
+set [d,f):B
 ----
 deleted [b,h)
 added [b,d):A
-added [f,h):A
 added [d,f):B
+added [f,h):A
 
 overlapping span=[b,h)
 ----
@@ -24,13 +27,14 @@ overlapping span=[b,h)
 # Check that writing a span that partially overlaps with multiple existing
 # entries deletes all of them, and re-adds the right non-overlapping fragments
 # with the right configs.
-set span=[c,e) conf=C
+apply
+set [c,e):C
 ----
 deleted [b,d)
 deleted [d,f)
 added [b,c):A
-added [e,f):B
 added [c,e):C
+added [e,f):B
 
 overlapping span=[b,h)
 ----
@@ -41,7 +45,8 @@ overlapping span=[b,h)
 
 # Check that when a span being written to entirely envelopes an existing entry,
 # that entry is deleted in its entirety.
-delete span=[d,g)
+apply
+delete [d,g)
 ----
 deleted [c,e)
 deleted [e,f)
@@ -64,7 +69,8 @@ compute-split span=[b,h)
 ----
 key=c
 
-set span=[b,g) conf=A
+apply
+set [b,g):A
 ----
 deleted [b,c)
 deleted [c,d)
diff --git a/pkg/spanconfig/spanconfigtestutils/BUILD.bazel b/pkg/spanconfig/spanconfigtestutils/BUILD.bazel
index 89ce9da48b86..355d3022c600 100644
--- a/pkg/spanconfig/spanconfigtestutils/BUILD.bazel
+++ b/pkg/spanconfig/spanconfigtestutils/BUILD.bazel
@@ -5,7 +5,10 @@ go_library(
     srcs = ["utils.go"],
     importpath = "github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigtestutils",
     visibility = ["//visibility:public"],
-    deps = ["//pkg/roachpb:with-mocks"],
+    deps = [
+        "//pkg/roachpb:with-mocks",
+        "//pkg/spanconfig",
+    ],
 )
 
 go_test(
diff --git a/pkg/spanconfig/spanconfigtestutils/utils.go b/pkg/spanconfig/spanconfigtestutils/utils.go
index 314010602986..aa0e072171bb 100644
--- a/pkg/spanconfig/spanconfigtestutils/utils.go
+++ b/pkg/spanconfig/spanconfigtestutils/utils.go
@@ -17,6 +17,7 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/spanconfig"
 )
 
 // spanRe matches strings of the form "[start, end)", capturing both the "start"
@@ -137,6 +138,37 @@ func ParseKVAccessorUpdateArguments(
 	return toDelete, toUpsert
 }
 
+// ParseStoreApplyArguments is a helper function that parses datadriven
+// store update arguments. The input is of the following form:
+//
+//      delete [c,e)
+//      set [c,d):C
+//      set [d,e):D
+//
+func ParseStoreApplyArguments(t *testing.T, input string) (updates []spanconfig.Update) {
+	for _, line := range strings.Split(input, "\n") {
+		line = strings.TrimSpace(line)
+		if line == "" {
+			continue
+		}
+
+		const setPrefix, deletePrefix = "set ", "delete "
+		switch {
+		case strings.HasPrefix(line, deletePrefix):
+			line = strings.TrimPrefix(line, line[:len(deletePrefix)])
+			updates = append(updates, spanconfig.Update{Span: ParseSpan(t, line)})
+		case strings.HasPrefix(line, setPrefix):
+			line = strings.TrimPrefix(line, line[:len(setPrefix)])
+			entry := ParseSpanConfigEntry(t, line)
+			updates = append(updates, spanconfig.Update{Span: entry.Span, Config: entry.Config})
+		default:
+			t.Fatalf("malformed line %q, expected to find prefix %q or %q",
+				line, setPrefix, deletePrefix)
+		}
+	}
+	return updates
+}
+
 // PrintSpan is a helper function that transforms roachpb.Span into a string of
 // the form "[start,end)". The span is assumed to have been constructed by the
 // ParseSpan helper above.