Skip to content

Commit 85f2c07

Browse files
jgao54jgao54
authored
update default settings for snapshotting (#4217)
Can remove the MongoDB setting completely in a few weeks.
1 parent 905df36 commit 85f2c07

4 files changed

Lines changed: 6 additions & 29 deletions

File tree

‎flow/connectors/postgres/qrep_partition_test.go‎

Lines changed: 4 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,10 @@ func newTestCaseForCTID(schema string, name string, rows uint32, expectedNum int
6565
Query: query,
6666
WatermarkTable: schemaQualifiedTable,
6767
WatermarkColumn: ctidColumnName,
68+
// The test table fits in a single 8 KB page, so block-based CTID partitioning only emit 1 partition.
69+
// Override with PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE = false to test legacy behavior
70+
// with NTileBucketPartitioning. We have separate tests below for testing CtidBlockPartitioning logic.
71+
Env: map[string]string{"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "false"},
6872
},
6973
expectedNumPartitions: expectedNum,
7074
}
@@ -320,7 +324,6 @@ func TestCTIDPartitioningOnPartitionedTable(t *testing.T) {
320324
Query: query,
321325
WatermarkTable: parentTable,
322326
WatermarkColumn: "ctid",
323-
Env: map[string]string{"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "true"},
324327
}, nil)
325328
require.NoError(t, err)
326329
require.NotEmpty(t, partitions)
@@ -409,7 +412,6 @@ func TestCTIDPartitioningOnMultiLevelPartitionedTable(t *testing.T) {
409412
Query: query,
410413
WatermarkTable: rootTable,
411414
WatermarkColumn: "ctid",
412-
Env: map[string]string{"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "true"},
413415
}, nil)
414416
require.NoError(t, err)
415417
require.NotEmpty(t, partitions)
@@ -471,7 +473,6 @@ func TestCTIDPartitioningOnEmptyPartitionedTable(t *testing.T) {
471473
Query: query,
472474
WatermarkTable: parentTable,
473475
WatermarkColumn: "ctid",
474-
Env: map[string]string{"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "true"},
475476
}, nil)
476477
require.NoError(t, err)
477478
require.Empty(t, partitions)
@@ -529,7 +530,6 @@ func TestCTIDPartitioningGroupingWhenChildrenExceedBudget(t *testing.T) {
529530
Query: query,
530531
WatermarkTable: parentTable,
531532
WatermarkColumn: "ctid",
532-
Env: map[string]string{"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "true"},
533533
}, nil)
534534
require.NoError(t, err)
535535
require.NotEmpty(t, partitions)
@@ -637,7 +637,6 @@ func TestCTIDPartitioningOnInheritedTable(t *testing.T) {
637637
NumPartitionsOverride: 8,
638638
WatermarkTable: parentTable,
639639
WatermarkColumn: "ctid",
640-
Env: map[string]string{"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "true"},
641640
}, nil)
642641
require.NoError(t, err)
643642
require.NotEmpty(t, partitions)
@@ -700,7 +699,6 @@ func TestCTIDPartitioningOnMultiLevelInheritedTable(t *testing.T) {
700699
NumPartitionsOverride: 10,
701700
WatermarkTable: grandparent,
702701
WatermarkColumn: "ctid",
703-
Env: map[string]string{"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "true"},
704702
}, nil)
705703
require.NoError(t, err)
706704
require.NotEmpty(t, partitions)
@@ -745,7 +743,6 @@ func TestCTIDPartitioningOnEmptyInheritedTable(t *testing.T) {
745743
NumPartitionsOverride: 4,
746744
WatermarkTable: parentTable,
747745
WatermarkColumn: "ctid",
748-
Env: map[string]string{"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "true"},
749746
}, nil)
750747
require.NoError(t, err)
751748
require.Empty(t, partitions)

‎flow/e2e/clickhouse_test.go‎

Lines changed: 0 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -3239,9 +3239,6 @@ func (s ClickHouseSuite) Test_CTID_Partitioned_Table() {
32393239
flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
32403240
flowConnConfig.DoInitialSnapshot = true
32413241
flowConnConfig.SnapshotNumPartitionsOverride = 10
3242-
flowConnConfig.Env = map[string]string{
3243-
"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "true",
3244-
}
32453242

32463243
tc := NewTemporalClient(s.t)
32473244
env := ExecutePeerflow(s.t, tc, flowConnConfig)
@@ -3311,9 +3308,6 @@ func (s ClickHouseSuite) Test_CTID_Multi_Level_Partitioned_Table() {
33113308
flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
33123309
flowConnConfig.DoInitialSnapshot = true
33133310
flowConnConfig.SnapshotNumPartitionsOverride = 6
3314-
flowConnConfig.Env = map[string]string{
3315-
"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "true",
3316-
}
33173311

33183312
tc := NewTemporalClient(s.t)
33193313
env := ExecutePeerflow(s.t, tc, flowConnConfig)
@@ -3370,9 +3364,6 @@ func (s ClickHouseSuite) Test_CTID_Inherited_Table() {
33703364
flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
33713365
flowConnConfig.DoInitialSnapshot = true
33723366
flowConnConfig.SnapshotNumPartitionsOverride = 8
3373-
flowConnConfig.Env = map[string]string{
3374-
"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "true",
3375-
}
33763367

33773368
tc := NewTemporalClient(s.t)
33783369
env := ExecutePeerflow(s.t, tc, flowConnConfig)
@@ -3433,9 +3424,6 @@ func (s ClickHouseSuite) Test_CTID_Multi_Level_Inherited_Table() {
34333424
flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
34343425
flowConnConfig.DoInitialSnapshot = true
34353426
flowConnConfig.SnapshotNumPartitionsOverride = 10
3436-
flowConnConfig.Env = map[string]string{
3437-
"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "true",
3438-
}
34393427

34403428
tc := NewTemporalClient(s.t)
34413429
env := ExecutePeerflow(s.t, tc, flowConnConfig)
@@ -3500,9 +3488,6 @@ func (s ClickHouseSuite) Test_CTID_Inherited_Table_Extra_Columns() {
35003488
flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
35013489
flowConnConfig.DoInitialSnapshot = true
35023490
flowConnConfig.SnapshotNumPartitionsOverride = 6
3503-
flowConnConfig.Env = map[string]string{
3504-
"PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE": "true",
3505-
}
35063491

35073492
tc := NewTemporalClient(s.t)
35083493
env := ExecutePeerflow(s.t, tc, flowConnConfig)

‎flow/e2e/mongo_test.go‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -109,7 +109,6 @@ func (s MongoClickhouseSuite) Test_Simple_Flow_Partitioned() {
109109
flowConnConfig := s.generateFlowConnectionConfigsDefaultEnv(connectionGen)
110110
flowConnConfig.DoInitialSnapshot = true
111111
flowConnConfig.SnapshotNumRowsPerPartition = 10
112-
flowConnConfig.Env["PEERDB_MONGODB_PARALLEL_SNAPSHOTTING"] = "true"
113112

114113
adminClient := s.Source().(*MongoSource).AdminClient()
115114
collection := adminClient.Database(srcDatabase).Collection(srcTable)
@@ -171,7 +170,6 @@ func (s MongoClickhouseSuite) Test_Snapshot_Collection_With_Single_Document() {
171170
flowConnConfig := s.generateFlowConnectionConfigsDefaultEnv(connectionGen)
172171
flowConnConfig.DoInitialSnapshot = true
173172
flowConnConfig.SnapshotNumRowsPerPartition = 1
174-
flowConnConfig.Env["PEERDB_MONGODB_PARALLEL_SNAPSHOTTING"] = "true"
175173

176174
adminClient := s.Source().(*MongoSource).AdminClient()
177175
collection := adminClient.Database(srcDatabase).Collection(srcTable)
@@ -217,7 +215,6 @@ func (s MongoClickhouseSuite) Test_Snapshot_Empty_Collection() {
217215
flowConnConfig := s.generateFlowConnectionConfigsDefaultEnv(connectionGen)
218216
flowConnConfig.DoInitialSnapshot = true
219217
flowConnConfig.SnapshotNumRowsPerPartition = 1
220-
flowConnConfig.Env["PEERDB_MONGODB_PARALLEL_SNAPSHOTTING"] = "true"
221218

222219
adminClient := s.Source().(*MongoSource).AdminClient()
223220
err := adminClient.Database(srcDatabase).CreateCollection(t.Context(), srcTable)
@@ -251,7 +248,6 @@ func (s MongoClickhouseSuite) Test_Snapshot_Non_ObjectID_Falls_Back_To_Single_Pa
251248
flowConnConfig := s.generateFlowConnectionConfigsDefaultEnv(connectionGen)
252249
flowConnConfig.DoInitialSnapshot = true
253250
flowConnConfig.SnapshotNumRowsPerPartition = 5
254-
flowConnConfig.Env["PEERDB_MONGODB_PARALLEL_SNAPSHOTTING"] = "true"
255251

256252
adminClient := s.Source().(*MongoSource).AdminClient()
257253
collection := adminClient.Database(srcDatabase).Collection(srcTable)
@@ -297,7 +293,6 @@ func (s MongoClickhouseSuite) Test_Snapshot_Mixed_ObjectID_Falls_Back_To_Single_
297293
flowConnConfig := s.generateFlowConnectionConfigsDefaultEnv(connectionGen)
298294
flowConnConfig.DoInitialSnapshot = true
299295
flowConnConfig.SnapshotNumRowsPerPartition = 1
300-
flowConnConfig.Env["PEERDB_MONGODB_PARALLEL_SNAPSHOTTING"] = "true"
301296

302297
adminClient := s.Source().(*MongoSource).AdminClient()
303298
collection := adminClient.Database(srcDatabase).Collection(srcTable)

‎flow/internal/dynamicconf.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -422,15 +422,15 @@ var DynamicSettings = [...]*protos.DynamicSetting{
422422
{
423423
Name: "PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE",
424424
Description: "Use CTID block partitioning for initial snapshot if watermark column is ctid",
425-
DefaultValue: "false",
425+
DefaultValue: "true",
426426
ValueType: protos.DynconfValueType_BOOL,
427427
ApplyMode: protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
428428
TargetForSetting: protos.DynconfTarget_ALL,
429429
},
430430
{
431431
Name: "PEERDB_MONGODB_PARALLEL_SNAPSHOTTING",
432432
Description: "Enable parallel partitioning for MongoDB initial snapshot",
433-
DefaultValue: "false",
433+
DefaultValue: "true",
434434
ValueType: protos.DynconfValueType_BOOL,
435435
ApplyMode: protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
436436
TargetForSetting: protos.DynconfTarget_ALL,

0 commit comments

Comments
 (0)