Merge #157153 #158065 #158089

157153: spanset: assert that batches don't access store local and unreplicated RangeID local keys r=iskettaneh a=iskettaneh

This PR adds the following test-only assertions:
1) Generated batches don't touch store-local keys.
2) Generated batches don't touch unreplicated RangeID local keys.

We disable the check in exactly 3 locations we know that we currently
touch those keys.

Fixes: #156537

Release note: None

158065: sql: use new increment when restarting sequence r=bghal a=bghal

Previously, if a sequence were restarted and its increment updated in
the same transaction (command), the old increment would be used to set
the value in KV.
This change corrects the `ALTER` DDL to use the new increment value.

Informs: #142914
Informs: #21564

Epic: CRDB-31283

Release note (sql change): Restarting a sequence with an updated
increment has the expected initial value.


158089: ioctx: add ReaderAtSeekerAdapter for random access support r=sravotto a=sravotto

This commit adds support for adapting context-aware readers that implement ReaderAtCtx and SeekerCtx into the standard io.ReaderAt and io.Seeker interfaces. This is needed for integrating cloud storage readers with libraries that require random access (e.g., Apache Parquet).

The adapter:
- Captures a context at construction time and uses it for all operations
- Validates that the underlying reader actually supports seeking
- Provides io.Reader, io.ReaderAt, io.Seeker, and io.Closer interfaces
- Enables efficient random access to cloud-stored files without buffering

Release note: None

Epic: CRDB-23802

Co-authored-by: iskettaneh <[email protected]>
Co-authored-by: Brendan Gerrity <[email protected]>
Co-authored-by: Silvano Ravotto <[email protected]>

Author: 4 people

pkg/BUILD.bazel

@@ -771,6 +771,7 @@ ALL_TESTS = [
     "//pkg/util/interval/generic:generic_test",
     "//pkg/util/interval:interval_test",
     "//pkg/util/intsets:intsets_test",
+    "//pkg/util/ioctx:ioctx_test",
     "//pkg/util/ipaddr:ipaddr_test",
     "//pkg/util/iterutil:iterutil_test",
     "//pkg/util/json/tokenizer:tokenizer_test",
@@ -2692,6 +2693,7 @@ GO_TARGETS = [
     "//pkg/util/intsets:intsets",
     "//pkg/util/intsets:intsets_test",
     "//pkg/util/ioctx:ioctx",
+    "//pkg/util/ioctx:ioctx_test",
     "//pkg/util/ipaddr:ipaddr",
     "//pkg/util/ipaddr:ipaddr_test",
     "//pkg/util/iterutil:iterutil",

pkg/keys/constants.go

@@ -163,6 +163,7 @@ var (
 	//
 	// LocalStorePrefix is the prefix identifying per-store data.
 	LocalStorePrefix = makeKey(LocalPrefix, roachpb.Key("s"))
+	LocalStoreMax    = roachpb.Key(LocalStorePrefix).PrefixEnd()
 	// localStoreClusterVersionSuffix stores the cluster-wide version
 	// information for this store, updated any time the operator
 	// updates the minimum cluster version.

pkg/keys/keys.go

@@ -308,6 +308,17 @@ func DecodeRangeIDKey(
 	return roachpb.RangeID(rangeInt), infix, suffix, b, nil
 }
 
+// DecodeRangeIDPrefix parses a local range ID prefix into range ID.
+func DecodeRangeIDPrefix(key roachpb.Key) (roachpb.RangeID, error) {
+	if !bytes.HasPrefix(key, LocalRangeIDPrefix) {
+		return 0, errors.Errorf("key %s does not have %s prefix", key, LocalRangeIDPrefix)
+	}
+	// Cut the prefix, the Range ID, and the infix specifier.
+	b := key[len(LocalRangeIDPrefix):]
+	_, rangeInt, err := encoding.DecodeUvarintAscending(b)
+	return roachpb.RangeID(rangeInt), err
+}
+
 // AbortSpanKey returns a range-local key by Range ID for an
 // AbortSpan entry, with detail specified by encoding the
 // supplied transaction ID.

pkg/kv/kvserver/batcheval/cmd_add_sstable.go

@@ -337,7 +337,7 @@ func EvalAddSSTable(
 	// addition, and instead just use this key-only iterator. If a caller actually
 	// needs to know what data is there, it must issue its own real Scan.
 	if args.ReturnFollowingLikelyNonEmptySpanStart {
-		existingIter, err := spanset.DisableReaderAssertions(readWriter).NewMVCCIterator(
+		existingIter, err := spanset.DisableUndeclaredSpanAssertions(readWriter).NewMVCCIterator(
 			ctx,
 			storage.MVCCKeyIterKind, // don't care if it is committed or not, just that it isn't empty.
 			storage.IterOptions{

pkg/kv/kvserver/batcheval/cmd_delete.go

@@ -58,7 +58,7 @@ func Delete(
 	// If requested, replace point tombstones with range tombstones.
 	if cArgs.EvalCtx.EvalKnobs().UseRangeTombstonesForPointDeletes && h.Txn == nil {
 		if err := storage.ReplacePointTombstonesWithRangeTombstones(
-			ctx, spanset.DisableReadWriterAssertions(readWriter),
+			ctx, spanset.DisableUndeclaredSpanAssertions(readWriter),
 			cArgs.Stats, args.Key, args.EndKey); err != nil {
 			return result.Result{}, err
 		}

pkg/kv/kvserver/batcheval/cmd_delete_range.go

@@ -302,7 +302,7 @@ func deleteRangeTransactional(
 	// If requested, replace point tombstones with range tombstones.
 	if cArgs.EvalCtx.EvalKnobs().UseRangeTombstonesForPointDeletes && h.Txn == nil {
 		if err := storage.ReplacePointTombstonesWithRangeTombstones(
-			ctx, spanset.DisableReadWriterAssertions(readWriter),
+			ctx, spanset.DisableUndeclaredSpanAssertions(readWriter),
 			cArgs.Stats, args.Key, args.EndKey); err != nil {
 			return result.Result{}, err
 		}

pkg/kv/kvserver/batcheval/cmd_end_transaction.go

@@ -719,7 +719,7 @@ func resolveLocalLocksWithPagination(
 				// If requested, replace point tombstones with range tombstones.
 				if ok && evalCtx.EvalKnobs().UseRangeTombstonesForPointDeletes {
 					if err := storage.ReplacePointTombstonesWithRangeTombstones(
-						ctx, spanset.DisableReadWriterAssertions(readWriter),
+						ctx, spanset.DisableUndeclaredSpanAssertions(readWriter),
 						ms, update.Key, update.EndKey); err != nil {
 						return 0, 0, 0, errors.Wrapf(err,
 							"replacing point tombstones with range tombstones for write intent at %s on end transaction [%s]",
@@ -757,7 +757,7 @@ func resolveLocalLocksWithPagination(
 			// If requested, replace point tombstones with range tombstones.
 			if evalCtx.EvalKnobs().UseRangeTombstonesForPointDeletes {
 				if err := storage.ReplacePointTombstonesWithRangeTombstones(
-					ctx, spanset.DisableReadWriterAssertions(readWriter),
+					ctx, spanset.DisableUndeclaredSpanAssertions(readWriter),
 					ms, update.Key, update.EndKey); err != nil {
 					return 0, 0, 0, errors.Wrapf(err,
 						"replacing point tombstones with range tombstones for write intent range at %s on end transaction [%s]",
@@ -1340,8 +1340,10 @@ func splitTriggerHelper(
 	if err != nil {
 		return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to fetch last replica GC timestamp")
 	}
+
 	if err := storage.MVCCPutProto(
-		ctx, batch, keys.RangeLastReplicaGCTimestampKey(split.RightDesc.RangeID), hlc.Timestamp{},
+		ctx, spanset.DisableForbiddenSpanAssertions(batch),
+		keys.RangeLastReplicaGCTimestampKey(split.RightDesc.RangeID), hlc.Timestamp{},
 		&replicaGCTS, storage.MVCCWriteOptions{Category: fs.BatchEvalReadCategory}); err != nil {
 		return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to copy last replica GC timestamp")
 	}
@@ -1541,7 +1543,8 @@ func splitTriggerHelper(
 		// as all replicas will be responsible for writing it locally before
 		// applying the split.
 		if !rec.ClusterSettings().Version.IsActive(ctx, clusterversion.V25_4_WriteInitialTruncStateBeforeSplitApplication) {
-			if err := kvstorage.WriteInitialTruncState(ctx, batch, split.RightDesc.RangeID); err != nil {
+			if err := kvstorage.WriteInitialTruncState(ctx,
+				spanset.DisableForbiddenSpanAssertions(batch), split.RightDesc.RangeID); err != nil {
 				return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to write initial Replica state")
 			}
 		}

pkg/kv/kvserver/batcheval/cmd_resolve_intent.go

@@ -129,7 +129,7 @@ func ResolveIntent(
 	// If requested, replace point tombstones with range tombstones.
 	if ok && cArgs.EvalCtx.EvalKnobs().UseRangeTombstonesForPointDeletes {
 		if err := storage.ReplacePointTombstonesWithRangeTombstones(ctx,
-			spanset.DisableReadWriterAssertions(readWriter), ms, update.Key, update.EndKey); err != nil {
+			spanset.DisableUndeclaredSpanAssertions(readWriter), ms, update.Key, update.EndKey); err != nil {
 			return result.Result{}, err
 		}
 	}

pkg/kv/kvserver/batcheval/cmd_resolve_intent_range.go

@@ -99,7 +99,7 @@ func ResolveIntentRange(
 	// If requested, replace point tombstones with range tombstones.
 	if cArgs.EvalCtx.EvalKnobs().UseRangeTombstonesForPointDeletes {
 		if err := storage.ReplacePointTombstonesWithRangeTombstones(ctx,
-			spanset.DisableReadWriterAssertions(readWriter), ms, args.Key, args.EndKey); err != nil {
+			spanset.DisableUndeclaredSpanAssertions(readWriter), ms, args.Key, args.EndKey); err != nil {
 			return result.Result{}, err
 		}
 	}

pkg/kv/kvserver/replica_eval_context_span.go

@@ -110,7 +110,7 @@ func (rec *SpanSetReplicaEvalContext) IsFirstRange() bool {
 func (rec SpanSetReplicaEvalContext) Desc() *roachpb.RangeDescriptor {
 	desc := rec.i.Desc()
 	rec.ss.AssertAllowed(spanset.SpanReadOnly,
-		roachpb.Span{Key: keys.RangeDescriptorKey(desc.StartKey)},
+		spanset.TrickySpan{Key: keys.RangeDescriptorKey(desc.StartKey)},
 	)
 	return desc
 }
@@ -151,7 +151,7 @@ func (rec SpanSetReplicaEvalContext) CanCreateTxnRecord(
 	ctx context.Context, txnID uuid.UUID, txnKey []byte, txnMinTS hlc.Timestamp,
 ) (bool, kvpb.TransactionAbortedReason) {
 	rec.ss.AssertAllowed(spanset.SpanReadOnly,
-		roachpb.Span{Key: keys.TransactionKey(txnKey, txnID)},
+		spanset.TrickySpan{Key: keys.TransactionKey(txnKey, txnID)},
 	)
 	return rec.i.CanCreateTxnRecord(ctx, txnID, txnKey, txnMinTS)
 }
@@ -163,7 +163,7 @@ func (rec SpanSetReplicaEvalContext) MinTxnCommitTS(
 	ctx context.Context, txnID uuid.UUID, txnKey []byte,
 ) hlc.Timestamp {
 	rec.ss.AssertAllowed(spanset.SpanReadOnly,
-		roachpb.Span{Key: keys.TransactionKey(txnKey, txnID)},
+		spanset.TrickySpan{Key: keys.TransactionKey(txnKey, txnID)},
 	)
 	return rec.i.MinTxnCommitTS(ctx, txnID, txnKey)
 }
@@ -173,7 +173,7 @@ func (rec SpanSetReplicaEvalContext) MinTxnCommitTS(
 // not be served.
 func (rec SpanSetReplicaEvalContext) GetGCThreshold() hlc.Timestamp {
 	rec.ss.AssertAllowed(spanset.SpanReadOnly,
-		roachpb.Span{Key: keys.RangeGCThresholdKey(rec.GetRangeID())},
+		spanset.TrickySpan{Key: keys.RangeGCThresholdKey(rec.GetRangeID())},
 	)
 	return rec.i.GetGCThreshold()
 }
@@ -197,7 +197,7 @@ func (rec SpanSetReplicaEvalContext) GetLastReplicaGCTimestamp(
 	ctx context.Context,
 ) (hlc.Timestamp, error) {
 	if err := rec.ss.CheckAllowed(spanset.SpanReadOnly,
-		roachpb.Span{Key: keys.RangeLastReplicaGCTimestampKey(rec.GetRangeID())},
+		spanset.TrickySpan{Key: keys.RangeLastReplicaGCTimestampKey(rec.GetRangeID())},
 	); err != nil {
 		return hlc.Timestamp{}, err
 	}
@@ -228,11 +228,11 @@ func (rec *SpanSetReplicaEvalContext) GetCurrentReadSummary(ctx context.Context)
 	// To capture a read summary over the range, all keys must be latched for
 	// writing to prevent any concurrent reads or writes.
 	desc := rec.i.Desc()
-	rec.ss.AssertAllowed(spanset.SpanReadWrite, roachpb.Span{
+	rec.ss.AssertAllowed(spanset.SpanReadWrite, spanset.TrickySpan{
 		Key:    keys.MakeRangeKeyPrefix(desc.StartKey),
 		EndKey: keys.MakeRangeKeyPrefix(desc.EndKey),
 	})
-	rec.ss.AssertAllowed(spanset.SpanReadWrite, roachpb.Span{
+	rec.ss.AssertAllowed(spanset.SpanReadWrite, spanset.TrickySpan{
 		Key:    desc.StartKey.AsRawKey(),
 		EndKey: desc.EndKey.AsRawKey(),
 	})