Skip to content

Commit 8fd64d3

Browse files
msbutlermsbutler
authored
Merge pull request #153871 from msbutler/backport25.2-153433
release-25.2: crosscluster/physical: add reader tenant system table id offset setting
2 parents 911daea + 1b300d8 commit 8fd64d3

File tree

17 files changed

+222
-50
lines changed

17 files changed

+222
-50
lines changed

‎pkg/config/system_test.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -203,7 +203,7 @@ func TestGetLargestID(t *testing.T) {
203203

204204
// Real SQL layout.
205205
func() testCase {
206-
ms := bootstrap.MakeMetadataSchema(keys.SystemSQLCodec, zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef())
206+
ms := bootstrap.MakeMetadataSchema(keys.SystemSQLCodec, zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef(), bootstrap.NoOffset)
207207
descIDs := ms.DescriptorIDs()
208208
maxDescID := config.ObjectID(descIDs[len(descIDs)-1])
209209
kvs, _ /* splits */ := ms.GetInitialValues()
@@ -321,7 +321,7 @@ func TestComputeSplitKeySystemRanges(t *testing.T) {
321321

322322
cfg := config.NewSystemConfig(zonepb.DefaultZoneConfigRef())
323323
kvs, _ /* splits */ := bootstrap.MakeMetadataSchema(
324-
keys.SystemSQLCodec, cfg.DefaultZoneConfig, zonepb.DefaultSystemZoneConfigRef(),
324+
keys.SystemSQLCodec, cfg.DefaultZoneConfig, zonepb.DefaultSystemZoneConfigRef(), bootstrap.NoOffset,
325325
).GetInitialValues()
326326
cfg.SystemConfigEntries = config.SystemConfigEntries{
327327
Values: kvs,
@@ -353,7 +353,7 @@ func TestComputeSplitKeyTableIDs(t *testing.T) {
353353
minKey := roachpb.RKey(keys.TimeseriesPrefix.PrefixEnd())
354354

355355
schema := bootstrap.MakeMetadataSchema(
356-
keys.SystemSQLCodec, zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef(),
356+
keys.SystemSQLCodec, zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef(), bootstrap.NoOffset,
357357
)
358358
// Real system tables only.
359359
baseSql, _ /* splits */ := schema.GetInitialValues()
@@ -460,7 +460,7 @@ func TestComputeSplitKeyTenantBoundaries(t *testing.T) {
460460
minTenID, maxTenID := roachpb.MinTenantID.ToUint64(), roachpb.MaxTenantID.ToUint64()
461461

462462
schema := bootstrap.MakeMetadataSchema(
463-
keys.SystemSQLCodec, zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef(),
463+
keys.SystemSQLCodec, zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef(), bootstrap.NoOffset,
464464
)
465465
minKey := tkey(bootstrap.TestingUserDescID(0))
466466

@@ -599,7 +599,7 @@ func TestGetZoneConfigForKey(t *testing.T) {
599599
cfg := config.NewSystemConfig(zonepb.DefaultZoneConfigRef())
600600

601601
kvs, _ /* splits */ := bootstrap.MakeMetadataSchema(
602-
keys.SystemSQLCodec, cfg.DefaultZoneConfig, zonepb.DefaultSystemZoneConfigRef(),
602+
keys.SystemSQLCodec, cfg.DefaultZoneConfig, zonepb.DefaultSystemZoneConfigRef(), bootstrap.NoOffset,
603603
).GetInitialValues()
604604
cfg.SystemConfigEntries = config.SystemConfigEntries{
605605
Values: kvs,

‎pkg/crosscluster/physical/standby_read_ts_poller_job_test.go‎

Lines changed: 119 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,11 @@ func TestStandbyReadTSPollerJob(t *testing.T) {
3737
c, cleanup := replicationtestutils.CreateTenantStreamingClusters(ctx, t, args)
3838
defer cleanup()
3939

40+
c.SrcTenantSQL.Exec(t, `CREATE TABLE foo (i INT PRIMARY KEY)`)
41+
c.SrcTenantSQL.Exec(t, `CREATE TABLE bar (i INT PRIMARY KEY)`)
42+
43+
offset, offsetChecksInReaderTenant := maybeOffsetReaderTenantSystemTables(t, c)
44+
4045
producerJobID, ingestionJobID := c.StartStreamReplication(ctx)
4146

4247
jobutils.WaitForJobToRun(c.T, c.SrcSysSQL, jobspb.JobID(producerJobID))
@@ -68,6 +73,11 @@ INSERT INTO a VALUES (1);
6873
waitForPollerJobToStartDest(t, c, ingestionJobID)
6974
observeValueInReaderTenant(t, c.ReaderTenantSQL)
7075

76+
var idWithOffsetCount int
77+
c.ReaderTenantSQL.QueryRow(t, fmt.Sprintf("SELECT count(*) FROM system.namespace where id = %d", 50+offset)).Scan(&idWithOffsetCount)
78+
require.Equal(t, 1, idWithOffsetCount, "expected to find namespace entry for table a with offset applied")
79+
offsetChecksInReaderTenant(c.ReaderTenantSQL)
80+
7181
// Failback and setup stanby reader tenant on the og source.
7282
{
7383
c.Cutover(ctx, producerJobID, ingestionJobID, srcTime.GoTime(), false)
@@ -102,7 +112,111 @@ INSERT INTO a VALUES (1);
102112
var numTables int
103113
srcReaderSQL.QueryRow(t, `SELECT count(*) FROM [SHOW TABLES]`).Scan(&numTables)
104114
observeValueInReaderTenant(t, srcReaderSQL)
115+
offsetChecksInReaderTenant(srcReaderSQL)
116+
}
117+
}
118+
119+
func maybeOffsetReaderTenantSystemTables(
120+
t *testing.T, c *replicationtestutils.TenantStreamingClusters,
121+
) (int, func(sql *sqlutils.SQLRunner)) {
122+
if c.Rng.Intn(2) == 0 {
123+
return 0, func(sql *sqlutils.SQLRunner) {}
124+
}
125+
offset := 100000
126+
c.DestSysSQL.Exec(t, fmt.Sprintf(`SET CLUSTER SETTING physical_cluster_replication.reader_system_table_id_offset = %d`, offset))
127+
// Set on source to ensure failback works well too.
128+
c.SrcSysSQL.Exec(t, fmt.Sprintf(`SET CLUSTER SETTING physical_cluster_replication.reader_system_table_id_offset = %d`, offset))
129+
130+
// swap a system table ID and a user table ID to simulate a cluster that has interleaving user and system table ids.
131+
scaryTableIDRemapFunc := `
132+
CREATE OR REPLACE FUNCTION renumber_desc(oldID INT, newID INT) RETURNS BOOL AS
133+
$$
134+
BEGIN
135+
-- Rewrite the ID within the descriptor
136+
SELECT crdb_internal.unsafe_upsert_descriptor(
137+
newid,
138+
crdb_internal.json_to_pb(
139+
'cockroach.sql.sqlbase.Descriptor',
140+
d
141+
),
142+
true
143+
)
144+
FROM (
145+
SELECT id,
146+
json_set(
147+
json_set(
148+
crdb_internal.pb_to_json(
149+
'cockroach.sql.sqlbase.Descriptor',
150+
descriptor,
151+
false
152+
),
153+
ARRAY['table', 'id'],
154+
newid::STRING::JSONB
155+
),
156+
ARRAY['table', 'modificationTime'],
157+
json_build_object(
158+
'wallTime',
159+
(
160+
(
161+
extract('epoch', now())
162+
* 1000000
163+
)::INT8
164+
* 1000
165+
)::STRING
166+
)
167+
) AS d
168+
FROM system.descriptor
169+
WHERE id IN (oldid,)
170+
);
171+
-- Update the namespace entry and delete the old descriptor.
172+
SELECT crdb_internal.unsafe_upsert_namespace_entry("parentID", "parentSchemaID", name, newID, true) FROM (SELECT "parentID", "parentSchemaID", name, id FROM system.namespace where id =oldID) UNION ALL
173+
SELECT crdb_internal.unsafe_delete_descriptor(oldID, true);
174+
175+
RETURN true;
176+
177+
END
178+
$$ LANGUAGE PLpgSQL;`
179+
180+
c.SrcTenantSQL.Exec(t, scaryTableIDRemapFunc)
181+
var txnInsightsID, privilegesID int
182+
c.SrcTenantSQL.QueryRow(t, `SELECT id FROM system.namespace WHERE name = 'transaction_execution_insights'`).Scan(&txnInsightsID)
183+
c.SrcTenantSQL.QueryRow(t, `SELECT id FROM system.namespace WHERE name = 'privileges'`).Scan(&privilegesID)
184+
require.NotEqual(t, 0, txnInsightsID)
185+
require.NotEqual(t, 0, privilegesID)
186+
187+
// renumber these two priv tables to be out of the way
188+
txnInsightIDRemapedID := txnInsightsID + 1000
189+
privilegesIDRemapedID := privilegesID + 1000
190+
c.SrcTenantSQL.Exec(t, `SELECT renumber_desc($1, $2)`, txnInsightsID, txnInsightIDRemapedID)
191+
c.SrcTenantSQL.Exec(t, `SELECT renumber_desc($1, $2)`, privilegesID, privilegesIDRemapedID)
192+
193+
// create two user tables on the source and interleave them with system table ids
194+
var fooID, barID int
195+
c.SrcTenantSQL.QueryRow(t, `SELECT id FROM system.namespace WHERE name = 'foo'`).Scan(&fooID)
196+
c.SrcTenantSQL.QueryRow(t, `SELECT id FROM system.namespace WHERE name = 'bar'`).Scan(&barID)
197+
require.NotEqual(t, 0, fooID)
198+
require.NotEqual(t, 0, barID)
199+
200+
c.SrcTenantSQL.Exec(t, `SELECT renumber_desc($1, $2)`, fooID, txnInsightsID)
201+
c.SrcTenantSQL.Exec(t, `SELECT renumber_desc($1, $2)`, barID, privilegesID)
202+
203+
// Drop the function, to avoid hitting 152978
204+
c.SrcTenantSQL.Exec(t, `DROP FUNCTION renumber_desc`)
205+
206+
offsetChecksInReaderTenant := func(sql *sqlutils.SQLRunner) {
207+
// Check that txn execution insights table is not at the same id as source as it's not replicated.
208+
sql.QueryRow(t, `SELECT id FROM system.namespace WHERE name = 'transaction_execution_insights'`).Scan(&txnInsightsID)
209+
require.NotEqual(t, txnInsightIDRemapedID, txnInsightsID)
210+
211+
// On 25.3, the privs table is not replicated so the ids should differ.
212+
sql.QueryRow(t, `SELECT id FROM system.namespace WHERE name = 'privileges'`).Scan(&privilegesID)
213+
require.NotEqual(t, privilegesIDRemapedID, privilegesID)
214+
var count int
215+
sql.QueryRow(t, `SELECT count(*) FROM system.namespace WHERE name = 'privileges'`).Scan(&count)
216+
require.Equal(t, 1, count)
105217
}
218+
219+
return offset, offsetChecksInReaderTenant
106220
}
107221

108222
func observeValueInReaderTenant(t *testing.T, readerSQL *sqlutils.SQLRunner) {
@@ -113,8 +227,8 @@ func observeValueInReaderTenant(t *testing.T, readerSQL *sqlutils.SQLRunner) {
113227
var numTables int
114228
readerSQL.QueryRow(t, `SELECT count(*) FROM [SHOW TABLES]`).Scan(&numTables)
115229

116-
if numTables != 1 {
117-
return errors.Errorf("expected 1 table to be present in reader tenant, but got %d instead", numTables)
230+
if numTables != 3 {
231+
return errors.Errorf("expected 3 tables to be present in reader tenant, but got %d instead", numTables)
118232
}
119233

120234
var actualQueryResult int
@@ -175,6 +289,9 @@ func TestReaderTenantCutover(t *testing.T) {
175289
c, cleanup := replicationtestutils.CreateTenantStreamingClusters(ctx, t, args)
176290
defer cleanup()
177291

292+
c.SrcTenantSQL.Exec(t, `CREATE TABLE foo (i INT PRIMARY KEY)`)
293+
c.SrcTenantSQL.Exec(t, `CREATE TABLE bar (i INT PRIMARY KEY)`)
294+
178295
producerJobID, ingestionJobID := c.StartStreamReplication(ctx)
179296

180297
jobutils.WaitForJobToRun(c.T, c.SrcSysSQL, jobspb.JobID(producerJobID))

‎pkg/crosscluster/physical/stream_ingestion_planning.go‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ package physical
77

88
import (
99
"context"
10+
"math"
1011

1112
"github.com/cockroachdb/cockroach/pkg/ccl/utilccl"
1213
"github.com/cockroachdb/cockroach/pkg/crosscluster/streamclient"
@@ -17,6 +18,7 @@ import (
1718
"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
1819
"github.com/cockroachdb/cockroach/pkg/roachpb"
1920
"github.com/cockroachdb/cockroach/pkg/server/telemetry"
21+
"github.com/cockroachdb/cockroach/pkg/settings"
2022
"github.com/cockroachdb/cockroach/pkg/sql"
2123
"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
2224
"github.com/cockroachdb/cockroach/pkg/sql/exprutil"
@@ -34,6 +36,16 @@ import (
3436
// replicated data will be retained.
3537
const defaultRetentionTTLSeconds = int32(4 * 60 * 60)
3638

39+
var readerTenantSystemTableIDOffset = settings.RegisterIntSetting(
40+
settings.ApplicationLevel,
41+
"physical_cluster_replication.reader_system_table_id_offset",
42+
"the offset added to dynamically allocated system table IDs in the reader tenant",
43+
0,
44+
// Max offset is 1000 less than MaxUint32 to leave room 1000 dynamically
45+
// allocated system table ids. Hope that never happens.
46+
settings.NonNegativeIntWithMaximum(math.MaxUint32-1000),
47+
)
48+
3749
func streamIngestionJobDescription(
3850
p sql.PlanHookState,
3951
source streamclient.ConfigUri,
@@ -331,7 +343,8 @@ func createReaderTenant(
331343
}
332344

333345
readerInfo.ID = readerID.ToUint64()
334-
_, err = sql.BootstrapTenant(ctx, p.ExecCfg(), p.Txn(), readerInfo, readerZcfg)
346+
systemTableIDOffset := readerTenantSystemTableIDOffset.Get(&p.ExecCfg().Settings.SV)
347+
_, err = sql.BootstrapTenant(ctx, p.ExecCfg(), p.Txn(), readerInfo, readerZcfg, uint32(systemTableIDOffset))
335348
if err != nil {
336349
return readerID, err
337350
}

‎pkg/kv/kvserver/store_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -126,7 +126,7 @@ type testStoreOpts struct {
126126

127127
func (opts *testStoreOpts) splits() (_kvs []roachpb.KeyValue, _splits []roachpb.RKey) {
128128
kvs, splits := bootstrap.MakeMetadataSchema(
129-
keys.SystemSQLCodec, zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef(),
129+
keys.SystemSQLCodec, zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef(), bootstrap.NoOffset,
130130
).GetInitialValues()
131131
if !opts.createSystemRanges {
132132
return kvs, nil

‎pkg/server/node.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -444,10 +444,10 @@ func allocateStoreIDs(
444444

445445
// GetBootstrapSchema returns the schema which will be used to bootstrap a new
446446
// server.
447-
func GetBootstrapSchema(
447+
func GetBootstrapSchemaForTest(
448448
defaultZoneConfig *zonepb.ZoneConfig, defaultSystemZoneConfig *zonepb.ZoneConfig,
449449
) bootstrap.MetadataSchema {
450-
return bootstrap.MakeMetadataSchema(keys.SystemSQLCodec, defaultZoneConfig, defaultSystemZoneConfig)
450+
return bootstrap.MakeMetadataSchema(keys.SystemSQLCodec, defaultZoneConfig, defaultSystemZoneConfig, bootstrap.NoOffset)
451451
}
452452

453453
// bootstrapCluster initializes the passed-in engines for a new cluster.

‎pkg/server/node_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -104,7 +104,7 @@ func TestBootstrapCluster(t *testing.T) {
104104
}
105105

106106
// Add the initial keys for sql.
107-
kvs, tableSplits := GetBootstrapSchema(
107+
kvs, tableSplits := GetBootstrapSchemaForTest(
108108
zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef(),
109109
).GetInitialValues()
110110
for _, kv := range kvs {

‎pkg/server/testserver.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1882,7 +1882,7 @@ func ExpectedInitialRangeCount(
18821882
defaultZoneConfig *zonepb.ZoneConfig,
18831883
defaultSystemZoneConfig *zonepb.ZoneConfig,
18841884
) (int, error) {
1885-
_, splits := bootstrap.MakeMetadataSchema(codec, defaultZoneConfig, defaultSystemZoneConfig).GetInitialValues()
1885+
_, splits := bootstrap.MakeMetadataSchema(codec, defaultZoneConfig, defaultSystemZoneConfig, bootstrap.NoOffset).GetInitialValues()
18861886
// N splits means N+1 ranges.
18871887
return len(config.StaticSplits()) + len(splits) + 1, nil
18881888
}

‎pkg/sql/catalog/bootstrap/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,7 @@ go_test(
6262
"//pkg/security/securityassets",
6363
"//pkg/security/securitytest",
6464
"//pkg/server",
65+
"//pkg/sql/catalog/descpb",
6566
"//pkg/testutils",
6667
"//pkg/testutils/datapathutils",
6768
"//pkg/testutils/serverutils",

‎pkg/sql/catalog/bootstrap/bootstrap_test.go‎

Lines changed: 24 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import (
1212
"github.com/cockroachdb/cockroach/pkg/config/zonepb"
1313
"github.com/cockroachdb/cockroach/pkg/keys"
1414
"github.com/cockroachdb/cockroach/pkg/roachpb"
15+
"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
1516
"github.com/cockroachdb/cockroach/pkg/testutils/datapathutils"
1617
"github.com/cockroachdb/cockroach/pkg/util/leaktest"
1718
"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -132,5 +133,27 @@ func makeMetadataSchema(tenantID uint64) MetadataSchema {
132133
if tenantID > 0 {
133134
codec = keys.MakeSQLCodec(roachpb.MustMakeTenantID(tenantID))
134135
}
135-
return MakeMetadataSchema(codec, zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef())
136+
return MakeMetadataSchema(codec, zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef(), NoOffset)
137+
}
138+
139+
func TestDynamicSystemTableIDOffset(t *testing.T) {
140+
defer leaktest.AfterTest(t)()
141+
defer log.Scope(t).Close(t)
142+
143+
offset := uint32(1000)
144+
145+
defaultMetadata := MakeMetadataSchema(keys.SystemSQLCodec, zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef(), NoOffset)
146+
offsetMetadata := MakeMetadataSchema(keys.SystemSQLCodec, zonepb.DefaultZoneConfigRef(), zonepb.DefaultSystemZoneConfigRef(), offset)
147+
148+
require.Len(t, defaultMetadata.descs, len(offsetMetadata.descs))
149+
150+
for i := range defaultMetadata.descs {
151+
defaultID := defaultMetadata.descs[i].GetID()
152+
if defaultID <= keys.MaxReservedDescID {
153+
// Reserved IDs are not offset.
154+
require.Equal(t, defaultID, offsetMetadata.descs[i].GetID())
155+
} else {
156+
require.Equal(t, defaultMetadata.descs[i].GetID()+descpb.ID(offset), offsetMetadata.descs[i].GetID())
157+
}
158+
}
136159
}

‎pkg/sql/catalog/bootstrap/initial_values.go‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,10 +23,11 @@ import (
2323
// InitialValuesOpts is used to get initial values for system/secondary tenants
2424
// and allows overriding initial values with ones from previous releases.
2525
type InitialValuesOpts struct {
26-
DefaultZoneConfig *zonepb.ZoneConfig
27-
DefaultSystemZoneConfig *zonepb.ZoneConfig
28-
OverrideKey clusterversion.Key
29-
Codec keys.SQLCodec
26+
DefaultZoneConfig *zonepb.ZoneConfig
27+
DefaultSystemZoneConfig *zonepb.ZoneConfig
28+
OverrideKey clusterversion.Key
29+
Codec keys.SQLCodec
30+
DynamicSystemTableIDOffset uint32
3031
}
3132

3233
// GenerateInitialValues generates the initial values with which to bootstrap a
@@ -84,7 +85,7 @@ var initialValuesFactoryByKey = map[clusterversion.Key]initialValuesFactoryFn{
8485
func buildLatestInitialValues(
8586
opts InitialValuesOpts,
8687
) (kvs []roachpb.KeyValue, splits []roachpb.RKey, _ error) {
87-
schema := MakeMetadataSchema(opts.Codec, opts.DefaultZoneConfig, opts.DefaultSystemZoneConfig)
88+
schema := MakeMetadataSchema(opts.Codec, opts.DefaultZoneConfig, opts.DefaultSystemZoneConfig, opts.DynamicSystemTableIDOffset)
8889
kvs, splits = schema.GetInitialValues()
8990
return kvs, splits, nil
9091
}

0 commit comments

Comments
 (0)