Merge #156715

156715: logical: fix handling of tenants r=jeffswenson a=jeffswenson

The legacy sql writer and the kv writer both have this logic for stripping tenant prefixes before decoding the key. This was uncovered because #156510 enabled metamorphic multi-tenancy for the LDR tests.

Release note: none
Fixes: #156640

Co-authored-by: Jeff Swenson <[email protected]>

Author: craig[bot]

pkg/crosscluster/logical/event_decoder.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcevent"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
@@ -136,6 +137,13 @@ func (d *eventDecoder) decodeAndCoalesceEvents(
 		if discard == jobspb.LogicalReplicationDetails_DiscardAllDeletes && event.KeyValue.Value.RawBytes == nil {
 			continue
 		}
+
+		var err error
+		event.KeyValue.Key, err = keys.StripTenantPrefix(event.KeyValue.Key)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to strip tenant prefix")
+		}
+
 		toDecode = append(toDecode, event)
 	}
 

pkg/sql/sqlclustersettings/clustersettings.go

@@ -145,7 +145,7 @@ var LDRImmediateModeWriter = settings.RegisterStringSetting(
 	"logical_replication.consumer.immediate_mode_writer",
 	"the writer to use when in immediate mode",
 	// TODO(jeffswenson): make the crud writer the default
-	metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(LDRWriterTypeLegacyKV), string(LDRWriterTypeSQL)),
+	metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(LDRWriterTypeLegacyKV), string(LDRWriterTypeCRUD), string(LDRWriterTypeSQL)),
 	settings.WithValidateString(func(sv *settings.Values, val string) error {
 		if val != string(LDRWriterTypeSQL) && val != string(LDRWriterTypeLegacyKV) && val != string(LDRWriterTypeCRUD) {
 			return errors.Newf("immediate mode writer must be either 'sql', 'legacy-kv', or 'crud', got '%s'", val)