Skip to content

Commit fa5b985

Browse files
msbutlermsbutler
authored
Merge pull request #153878 from msbutler/backport24.3-153433
release-24.3: crosscluster/physical: add reader tenant system table id offset setting
2 parents 6d7556c + e8a778d commit fa5b985

File tree

17 files changed

+222
-49
lines changed

17 files changed

+222
-49
lines changed

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

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

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

119233
var actualQueryResult int
@@ -174,6 +288,9 @@ func TestReaderTenantCutover(t *testing.T) {
174288
c, cleanup := replicationtestutils.CreateTenantStreamingClusters(ctx, t, args)
175289
defer cleanup()
176290

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

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

‎pkg/ccl/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/crosscluster"
1213
"github.com/cockroachdb/cockroach/pkg/ccl/crosscluster/streamclient"
@@ -18,6 +19,7 @@ import (
1819
"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
1920
"github.com/cockroachdb/cockroach/pkg/roachpb"
2021
"github.com/cockroachdb/cockroach/pkg/server/telemetry"
22+
"github.com/cockroachdb/cockroach/pkg/settings"
2123
"github.com/cockroachdb/cockroach/pkg/sql"
2224
"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
2325
"github.com/cockroachdb/cockroach/pkg/sql/exprutil"
@@ -33,6 +35,16 @@ import (
3335
// replicated data will be retained.
3436
const defaultRetentionTTLSeconds = int32(4 * 60 * 60)
3537

38+
var readerTenantSystemTableIDOffset = settings.RegisterIntSetting(
39+
settings.ApplicationLevel,
40+
"physical_cluster_replication.reader_system_table_id_offset",
41+
"the offset added to dynamically allocated system table IDs in the reader tenant",
42+
0,
43+
// Max offset is 1000 less than MaxUint32 to leave room 1000 dynamically
44+
// allocated system table ids. Hope that never happens.
45+
settings.NonNegativeIntWithMaximum(math.MaxUint32-1000),
46+
)
47+
3648
// CannotSetExpirationWindowErr get returned if the user attempts to specify the
3749
// EXPIRATION WINDOW option to create a replication stream, as this job setting
3850
// should only be set from the producer cluster.
@@ -333,7 +345,8 @@ func createReaderTenant(
333345
}
334346

335347
readerInfo.ID = readerID.ToUint64()
336-
_, err = sql.BootstrapTenant(ctx, p.ExecCfg(), p.Txn(), readerInfo, readerZcfg)
348+
systemTableIDOffset := readerTenantSystemTableIDOffset.Get(&p.ExecCfg().Settings.SV)
349+
_, err = sql.BootstrapTenant(ctx, p.ExecCfg(), p.Txn(), readerInfo, readerZcfg, uint32(systemTableIDOffset))
337350
if err != nil {
338351
return readerID, err
339352
}

‎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/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
@@ -459,10 +459,10 @@ func allocateStoreIDs(
459459

460460
// GetBootstrapSchema returns the schema which will be used to bootstrap a new
461461
// server.
462-
func GetBootstrapSchema(
462+
func GetBootstrapSchemaForTest(
463463
defaultZoneConfig *zonepb.ZoneConfig, defaultSystemZoneConfig *zonepb.ZoneConfig,
464464
) bootstrap.MetadataSchema {
465-
return bootstrap.MakeMetadataSchema(keys.SystemSQLCodec, defaultZoneConfig, defaultSystemZoneConfig)
465+
return bootstrap.MakeMetadataSchema(keys.SystemSQLCodec, defaultZoneConfig, defaultSystemZoneConfig, bootstrap.NoOffset)
466466
}
467467

468468
// 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
@@ -102,7 +102,7 @@ func TestBootstrapCluster(t *testing.T) {
102102
}
103103

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

‎pkg/server/testserver.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1789,7 +1789,7 @@ func ExpectedInitialRangeCount(
17891789
defaultZoneConfig *zonepb.ZoneConfig,
17901790
defaultSystemZoneConfig *zonepb.ZoneConfig,
17911791
) (int, error) {
1792-
_, splits := bootstrap.MakeMetadataSchema(codec, defaultZoneConfig, defaultSystemZoneConfig).GetInitialValues()
1792+
_, splits := bootstrap.MakeMetadataSchema(codec, defaultZoneConfig, defaultSystemZoneConfig, bootstrap.NoOffset).GetInitialValues()
17931793
// N splits means N+1 ranges.
17941794
return len(config.StaticSplits()) + len(splits) + 1, nil
17951795
}

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,7 @@ go_test(
6161
"//pkg/security/securityassets",
6262
"//pkg/security/securitytest",
6363
"//pkg/server",
64+
"//pkg/sql/catalog/descpb",
6465
"//pkg/testutils",
6566
"//pkg/testutils/datapathutils",
6667
"//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
@@ -82,7 +83,7 @@ var initialValuesFactoryByKey = map[clusterversion.Key]initialValuesFactoryFn{
8283
func buildLatestInitialValues(
8384
opts InitialValuesOpts,
8485
) (kvs []roachpb.KeyValue, splits []roachpb.RKey, _ error) {
85-
schema := MakeMetadataSchema(opts.Codec, opts.DefaultZoneConfig, opts.DefaultSystemZoneConfig)
86+
schema := MakeMetadataSchema(opts.Codec, opts.DefaultZoneConfig, opts.DefaultSystemZoneConfig, opts.DynamicSystemTableIDOffset)
8687
kvs, splits = schema.GetInitialValues()
8788
return kvs, splits, nil
8889
}

0 commit comments

Comments
 (0)