Merge #149621 #149622 #150319 #150370 #150436

149621: sql: validate trigger backreferences in table descriptors r=spilchen a=spilchen

Previously, descriptor validation only accounted for backreferences from sequences and views. However, the introduction of triggers introduces table-to-table references, which were not being validated. This change extends the validation logic to cover these cases, improving correctness and robustness of descriptor validation involving triggers.

Fixes: #148167
Epic: CRDB-42942

Release note: None

149622: changefeedccl: support resolved option with protobuf format r=asg0451 a=elizaMkraule

This change introduces support for the resolved option
with protobuf format for changefeeds

Release note (general change): the changefeeds with protobuf
format now supports the resolved option for emitting
resolved timestamps.

Fixes #148934

150319: kv: remove some tautological conditionals r=tbg a=stevendanna

Epic: none
Release note: None

150370: pcr: add debug benchmarking settings to discard data r=dt a=dt

Release note: none.
Epic: none.

150436: upgradeccl: ensure multiple releases for TestTenantUpgradeFailure test r=rail a=celiala

TestTenantUpgradeFailure can only run when we support more than one previous release. This PR ensures that we setup multple releases for this test.

This unblocks the PR needed to mint release-25.3:
#150211 (comment)
- Once 150436 is backported onto release-25.3, I'll remove a38ea8f from #150211, and just rebase on top of this backported fix.



Epic: None
Release note: None
Release justification: test-only fix.

Co-authored-by: Matt Spilchen <[email protected]>
Co-authored-by: Eliza Kraule <[email protected]>
Co-authored-by: Steven Danna <[email protected]>
Co-authored-by: David Taylor <[email protected]>
Co-authored-by: Celia La <[email protected]>

Author: 6 people

pkg/ccl/changefeedccl/encoder_protobuf.go

@@ -15,7 +15,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedpb"
 	"github.com/cockroachdb/cockroach/pkg/geo"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
-	"github.com/cockroachdb/cockroach/pkg/util/errorutil/unimplemented"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/errors"
@@ -77,9 +76,29 @@ func (e *protobufEncoder) EncodeValue(
 
 // EncodeResolvedTimestamp encodes a resolved timestamp message for the specified topic.
 func (e *protobufEncoder) EncodeResolvedTimestamp(
-	context.Context, string, hlc.Timestamp,
+	ctx context.Context, topic string, ts hlc.Timestamp,
 ) ([]byte, error) {
-	return nil, unimplemented.NewWithIssuef(148934, "protobuf encoder does not support resolved timestamps yet")
+	var msg *changefeedpb.Message
+	if e.envelopeType == changefeedbase.OptEnvelopeBare {
+		msg = &changefeedpb.Message{
+			Data: &changefeedpb.Message_BareResolved{
+				BareResolved: &changefeedpb.BareResolved{
+					XCrdb__: &changefeedpb.Resolved{
+						Resolved: ts.AsOfSystemTime(),
+					},
+				},
+			},
+		}
+	} else {
+		msg = &changefeedpb.Message{
+			Data: &changefeedpb.Message_Resolved{
+				Resolved: &changefeedpb.Resolved{
+					Resolved: ts.AsOfSystemTime(),
+				},
+			},
+		}
+	}
+	return protoutil.Marshal(msg)
 }
 
 // buildBare constructs a BareEnvelope with optional metadata and serializes it.

pkg/ccl/changefeedccl/encoder_protobuf_test.go

@@ -70,3 +70,60 @@ func TestProtoEncoder_BareEnvelope_WithMetadata(t *testing.T) {
 	require.Equal(t, int64(1), bare.Values["id"].GetInt64Value())
 	require.Equal(t, "Alice", bare.Values["name"].GetStringValue())
 }
+
+func TestProtoEncoder_ResolvedEnvelope(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	tableDesc, err := parseTableDesc(`CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
+	require.NoError(t, err)
+	targets := mkTargets(tableDesc)
+
+	ts := hlc.Timestamp{WallTime: 123, Logical: 456}
+
+	tests := []struct {
+		name          string
+		envelopeType  changefeedbase.EnvelopeType
+		expectWrapped bool
+	}{
+		{
+			name:         "wrapped envelope",
+			envelopeType: changefeedbase.OptEnvelopeWrapped,
+		},
+		{
+			name:         "bare envelope",
+			envelopeType: changefeedbase.OptEnvelopeBare,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			opts := changefeedbase.EncodingOptions{
+				Envelope: tc.envelopeType,
+				Format:   changefeedbase.OptFormatProtobuf,
+			}
+
+			enc, err := getEncoder(context.Background(), opts, targets, false, nil, nil, nil)
+			require.NoError(t, err)
+
+			b, err := enc.EncodeResolvedTimestamp(context.Background(), "test-topic", ts)
+			require.NoError(t, err)
+
+			var msg changefeedpb.Message
+			require.NoError(t, protoutil.Unmarshal(b, &msg))
+
+			switch tc.envelopeType {
+			case changefeedbase.OptEnvelopeWrapped:
+				res := msg.GetResolved()
+				require.NotNil(t, res, "wrapped envelope should populate Resolved field")
+				require.Equal(t, ts.AsOfSystemTime(), res.Resolved)
+			case changefeedbase.OptEnvelopeBare:
+				res := msg.GetBareResolved()
+				require.NotNil(t, res, "bare envelope should populate BareResolved field")
+				require.Equal(t, ts.AsOfSystemTime(), res.XCrdb__.Resolved)
+			default:
+				t.Fatalf("unexpected envelope type: %v", tc.envelopeType)
+			}
+		})
+	}
+}

pkg/ccl/kvccl/kvtenantccl/upgradeccl/tenant_upgrade_test.go

@@ -346,20 +346,21 @@ func TestTenantUpgradeFailure(t *testing.T) {
 
 	v0 := clusterversion.MinSupported.Version()
 	v2 := clusterversion.Latest.Version()
-	// v1 needs to be between v0 and v2. Set it to the minor release
-	// after v0 and before v2.
+	// v1 needs to be between v0 and v2. Set it to the first version with a
+	// different major/minor from v0.
 	var v1 roachpb.Version
 	for _, version := range clusterversion.ListBetween(v0, v2) {
-		if version.Minor != v0.Minor {
+		if version.Major != v0.Major && version.Minor != v0.Minor {
 			v1 = version
 			break
 		}
 	}
-	if v1 == (roachpb.Version{}) {
+	if v1 == (roachpb.Version{}) || v1 == v2 {
 		// There is no in-between version supported; skip this test.
 		skip.IgnoreLint(t, "test can only run when we support two previous releases")
 	}
 
+	t.Logf("v0=%s  v1=%s  v2=%s", v0, v1, v2)
 	t.Log("starting server")
 	ctx := context.Background()
 	settings := cluster.MakeTestingClusterSettingsWithVersions(

pkg/crosscluster/producer/event_stream.go

@@ -366,15 +366,27 @@ func (s *eventStream) maybeFlushBatch(ctx context.Context) error {
 	return nil
 }
 
+var debugSettingDropData = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"physical_replication.producer.unsafe_debug.discard_all_data.enabled",
+	"discard all row data during cluster replication (for experimental debugging purposes only)",
+	false,
+	settings.WithUnsafe,
+)
+
 func (s *eventStream) flushBatch(ctx context.Context, reason streampb.FlushReason) error {
 	if s.seb.size == 0 {
 		return nil
 	}
+	defer s.seb.reset()
+
+	if debugSettingDropData.Get(s.execCfg.SV()) {
+		return nil
+	}
+
 	s.seqNum++
 	s.debug.Flushed(int64(s.seb.size), reason, s.seqNum)
 
-	defer s.seb.reset()
-
 	return s.sendFlush(ctx, &streampb.StreamEvent{StreamSeq: s.seqNum, Batch: &s.seb.batch})
 }
 func (s *eventStream) sendFlush(ctx context.Context, event *streampb.StreamEvent) error {

pkg/kv/bulk/sst_batcher.go

@@ -636,6 +636,14 @@ func (b *SSTBatcher) syncFlush() error {
 	return flushErr
 }
 
+var debugDropSSTOnFlush = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"bulkio.ingest.unsafe_debug.drop_sst_on_flush.enabled",
+	"if set, the SSTBatcher will simply discard data instead of flushing it (destroys data; for performance debugging experiments only)",
+	false,
+	settings.WithUnsafe,
+)
+
 // startFlush starts a flush of the current batch. If it encounters any errors
 // the errors are reported by the call to `syncFlush`.
 //
@@ -825,9 +833,13 @@ func (b *SSTBatcher) startFlush(ctx context.Context, reason int) {
 	b.asyncAddSSTs.GoCtx(func(ctx context.Context) error {
 		defer res.Release()
 		defer b.mem.Shrink(ctx, reserved)
-		results, err := b.adder.AddSSTable(ctx, batchTS, start, end, data, mvccStats, performanceStats)
-		if err != nil {
-			return err
+
+		var results []addSSTResult
+		if !debugDropSSTOnFlush.Get(&b.settings.SV) {
+			results, err = b.adder.AddSSTable(ctx, batchTS, start, end, data, mvccStats, performanceStats)
+			if err != nil {
+				return err
+			}
 		}
 
 		// Now that we have completed ingesting the SSTables we take a lock and

pkg/kv/client_test.go

@@ -874,10 +874,8 @@ func TestTxn_ReverseScan(t *testing.T) {
 			t.Errorf("expected empty, got %v", rows)
 		}
 		return err
-	}); err != nil {
-		if err == nil {
-			t.Errorf("expected a truncation error, got %s", err)
-		}
+	}); err == nil || !strings.Contains(err.Error(), "must be greater than start") {
+		t.Errorf("expected a truncation error, got %s", err)
 	}
 
 	// Try reverse scan with non-existent key.

pkg/kv/txn_test.go

@@ -119,9 +119,7 @@ func newTestTxnFactory(
 			}
 			if ba.Txn != nil && br.Txn == nil {
 				br.Txn = ba.Txn.Clone()
-				if pErr == nil {
-					br.Txn.Status = status
-				}
+				br.Txn.Status = status
 				// Update the MockTxnSender's proto.
 				*txn = *br.Txn
 			}

pkg/sql/catalog/tabledesc/validate.go

@@ -361,8 +361,6 @@ func (desc *wrapper) ValidateBackReferences(
 		}
 		switch depDesc.DescriptorType() {
 		case catalog.Table:
-			// If this is a table, it may be referenced by a view, otherwise if this
-			// is a sequence, then it may be also be referenced by a table.
 			vea.Report(desc.validateInboundTableRef(by, vdg))
 		case catalog.Function:
 			// This relation may be referenced by a function.
@@ -504,49 +502,76 @@ func (desc *wrapper) validateInboundTableRef(
 			backReferencedTable.GetName(), backReferencedTable.GetID())
 	}
 	if desc.IsSequence() {
-		// The ColumnIDs field takes a different meaning when the validated
-		// descriptor is for a sequence. In this case, they refer to the columns
-		// in the referenced descriptor instead.
-		for _, colID := range by.ColumnIDs {
-			// Skip this check if the column ID is zero. This can happen due to
-			// bugs in 20.2.
-			//
-			// TODO(ajwerner): Make sure that a migration in 22.2 fixes this issue.
-			if colID == 0 {
-				continue
-			}
-			col := catalog.FindColumnByID(backReferencedTable, colID)
-			if col == nil {
-				return errors.AssertionFailedf("depended-on-by relation %q (%d) does not have a column with ID %d",
-					backReferencedTable.GetName(), by.ID, colID)
+		if err := validateSequenceColumnBackrefs(desc, backReferencedTable, by); err != nil {
+			return err
+		}
+	}
+
+	// View back-references need corresponding forward reference.
+	if backReferencedTable.IsView() {
+		for _, id := range backReferencedTable.TableDesc().DependsOn {
+			if id == desc.GetID() {
+				return nil
 			}
-			var found bool
-			for i := 0; i < col.NumUsesSequences(); i++ {
-				if col.GetUsesSequenceID(i) == desc.GetID() {
-					found = true
-					break
+		}
+		return errors.AssertionFailedf("depended-on-by view %q (%d) has no corresponding depends-on forward reference",
+			backReferencedTable.GetName(), by.ID)
+	}
+
+	// Table to table back-references must have a trigger reference.
+	if backReferencedTable.IsTable() && desc.IsTable() {
+		for _, trigger := range backReferencedTable.TableDesc().Triggers {
+			for _, id := range trigger.DependsOn {
+				if id == desc.GetID() {
+					return nil
 				}
 			}
-			if found {
-				continue
-			}
-			return errors.AssertionFailedf(
-				"depended-on-by relation %q (%d) has no reference to this sequence in column %q (%d)",
-				backReferencedTable.GetName(), by.ID, col.GetName(), col.GetID())
 		}
-	}
 
-	// View back-references need corresponding forward reference.
-	if !backReferencedTable.IsView() {
-		return nil
+		// No valid forward reference found to justify the backref.
+		return errors.AssertionFailedf(
+			"table %q (%d) does not have a forward reference to descriptor %q (%d)",
+			backReferencedTable.GetName(), by.ID, desc.GetName(), desc.GetID())
 	}
-	for _, id := range backReferencedTable.TableDesc().DependsOn {
-		if id == desc.GetID() {
-			return nil
+	return nil
+}
+
+func validateSequenceColumnBackrefs(
+	seq catalog.Descriptor,
+	backReferencedTable catalog.TableDescriptor,
+	by descpb.TableDescriptor_Reference,
+) error {
+	// The ColumnIDs field takes a different meaning when the validated
+	// descriptor is for a sequence. In this case, they refer to the columns
+	// in the referenced descriptor instead.
+	for _, colID := range by.ColumnIDs {
+		// Skip this check if the column ID is zero. This can happen due to
+		// bugs in 20.2.
+		//
+		// TODO(ajwerner): Make sure that a migration in 22.2 fixes this issue.
+		if colID == 0 {
+			continue
 		}
+		col := catalog.FindColumnByID(backReferencedTable, colID)
+		if col == nil {
+			return errors.AssertionFailedf("depended-on-by relation %q (%d) does not have a column with ID %d",
+				backReferencedTable.GetName(), by.ID, colID)
+		}
+		var found bool
+		for i := 0; i < col.NumUsesSequences(); i++ {
+			if col.GetUsesSequenceID(i) == seq.GetID() {
+				found = true
+				break
+			}
+		}
+		if found {
+			continue
+		}
+		return errors.AssertionFailedf(
+			"depended-on-by relation %q (%d) has no reference to this sequence in column %q (%d)",
+			backReferencedTable.GetName(), by.ID, col.GetName(), col.GetID())
 	}
-	return errors.AssertionFailedf("depended-on-by view %q (%d) has no corresponding depends-on forward reference",
-		backReferencedTable.GetName(), by.ID)
+	return nil
 }
 
 // validateFK asserts that references to desc from inbound and outbound FKs are