changefeedccl: enable frontier quantization #138328

wenyihu6 · 2025-01-06T17:18:55Z

Previously, all checkpoint timestamps were forwarded precisely to the resolved
timestamp. To improve merging of adjacent frontier spans, this patch introduces
a frontier quantization mode. In this mode, resolved timestamps from rangefeeds
are rounded down to the nearest multiple of the quantization granularity.
Rounding down ensures no progress is missed and that we consistently track the
lowest possible resolved timestamp.

Resolves: #125723
Release note: changefeed.frontier.timestamp_granularity (default off)
has been introduced to enable more efficient tracking of resolved timestamps.

cockroach-teamcity · 2025-01-06T17:19:05Z

This change is

asg0451 · 2025-01-06T20:19:03Z

pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go

@@ -128,7 +130,12 @@ func (p *rangefeed) addEventsToBuffer(ctx context.Context) error {
 				}
 				stop()
 			case *kvpb.RangeFeedCheckpoint:
-				if !t.ResolvedTS.IsEmpty() && t.ResolvedTS.Less(p.cfg.Frontier) {
+				ev := e.RangeFeedEvent


while this is a pretty simple implementation, would you mind investigating how much work it would be to switch to using the higher-level rangefeed client implementation that LDR uses? it already includes an implementation of quantization and more things that may prove useful. see https://github.com/asg0451/cockroach/blob/ed5ed37ea91b04fba061084ba73fe9225c018288/pkg/crosscluster/producer/event_stream.go#L147 &/ talk to me about it

if it's too much work / too incompatible an api, then maybe dont. however i think it would be beneficial if it's low to moderate effort

I will look into the refactoring. Can we try merging this PR anyways since this is pretty straightforward and already up? I will look into the refactoring as part of the follow up work.

I put up #138346 and assigned myself.

thanks. lmk how it goes

I would also be in favor of using the already-built frontier quantization rangefeed client code if it's not too complicated. Maybe you could try prototyping it and seeing how involved it would be?

Ack, I'm exploring this option now and see how much work this turns out to be. I would still prefer to drive this PR forward given this is already up, but I'm actively prototyping for this refactoring.

asg0451

looks good but i have some questions about the tests.

pkg/ccl/changefeedccl/kvfeed/kv_feed_test.go

asg0451 · 2025-01-06T21:42:30Z

pkg/ccl/changefeedccl/kvfeed/kv_feed_test.go

+				require.NoError(t, err)
+			}
+			frontier.Entries(func(sp roachpb.Span, ts hlc.Timestamp) span.OpResult {
+				t.Logf("span: %v, ts: %v\n", sp, ts)


I thought it might be worth leaving it in since it was helpful for debugging test failures wdyt?

asg0451 · 2025-01-06T21:55:14Z

pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go

@@ -128,7 +130,12 @@ func (p *rangefeed) addEventsToBuffer(ctx context.Context) error {
 				}
 				stop()
 			case *kvpb.RangeFeedCheckpoint:
-				if !t.ResolvedTS.IsEmpty() && t.ResolvedTS.Less(p.cfg.Frontier) {
+				ev := e.RangeFeedEvent


thanks. lmk how it goes

asg0451 · 2025-01-06T21:56:17Z

pkg/ccl/changefeedccl/kvfeed/kv_feed_test.go

+			const quantize = time.Duration(10)
+			for _, e := range tc.events {
+				if tc.withFrontierQuantization {
+					e.Checkpoint.ResolvedTS = mockQuantization(e.Checkpoint.ResolvedTS, quantize)


i'm not really sure what this test is testing. just your mockQuantization function? can you make sure it tests the actual implementation?

Good idea, I refactored the actual quantization to a helper function and made tests call into the helper function instead.

asg0451 · 2025-01-06T21:56:34Z

pkg/ccl/changefeedccl/kvfeed/kv_feed_test.go

+	mockQuantization := func(ts hlc.Timestamp, quantize time.Duration) hlc.Timestamp {
+		ts.Logical = 0
+		ts.WallTime -= ts.WallTime % int64(quantize)
+		return ts
+	}
+
+	makeCheckpointEvent := func(key []byte, endKey []byte, ts int, logical int32) *kvpb.RangeFeedEvent {
+		return &kvpb.RangeFeedEvent{
+			Checkpoint: &kvpb.RangeFeedCheckpoint{
+				Span:       roachpb.Span{Key: key, EndKey: endKey},
+				ResolvedTS: hlc.Timestamp{WallTime: int64(ts), Logical: logical},
+			},
+		}
+	}


nit these can be pulled out between these two tests

asg0451 · 2025-01-06T21:57:11Z

pkg/ccl/changefeedccl/kvfeed/kv_feed_test.go

+	quantizedFrontier, err := span.MakeFrontier([]roachpb.Span{{Key: keys.MinKey, EndKey: keys.MaxKey}}...)
+	require.NoError(t, err)
+	for _, e := range events {
+		quantizedTs := mockQuantization(e.Checkpoint.ResolvedTS, quantize)


ditto on the "what does this test" question

I added more comments for these tests.

pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go

andyyang890

Nice work! Left some comments.

andyyang890 · 2025-01-06T22:36:50Z

pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go

+				if quantize := p.cfg.WithFrontierQuantize; quantize != 0 {
+					ev.Checkpoint.ResolvedTS = quantizeTS(ev.Checkpoint.ResolvedTS, quantize)
+				}
+				if resolvedTs := ev.Checkpoint.ResolvedTS; resolvedTs.IsEmpty() && resolvedTs.Less(p.cfg.Frontier) {


This code used to check that the timestamp wasn't empty. Is the change intentional?

Good catch, done.

andyyang890 · 2025-01-06T22:47:41Z

pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go

@@ -128,7 +130,12 @@ func (p *rangefeed) addEventsToBuffer(ctx context.Context) error {
 				}
 				stop()
 			case *kvpb.RangeFeedCheckpoint:
-				if !t.ResolvedTS.IsEmpty() && t.ResolvedTS.Less(p.cfg.Frontier) {
+				ev := e.RangeFeedEvent


I would also be in favor of using the already-built frontier quantization rangefeed client code if it's not too complicated. Maybe you could try prototyping it and seeing how involved it would be?

andyyang890 · 2025-01-06T23:07:14Z

pkg/ccl/changefeedccl/kvfeed/kv_feed.go

@@ -126,6 +126,7 @@ func Run(ctx context.Context, cfg Config) error {
 		cfg.Writer, cfg.Spans, cfg.CheckpointSpans, cfg.CheckpointTimestamp,
 		cfg.SchemaChangeEvents, cfg.SchemaChangePolicy,
 		cfg.NeedsInitialScan, cfg.WithDiff, cfg.WithFiltering,
+		changefeedbase.Quantize.Get(&cfg.Settings.SV),


I wonder if we should pass the quantization factor in the processor specs so that we don't end up in a situation where multiple change aggregators have different quantization factors because of a race condition between changing the cluster setting and (re)starting a changefeed. WDYT?

Good point. Let's discuss this further offline. At first glance, having different granularities across aggregators doesn’t seem like a major issue. It might reduce quantization but may not justify the complexity of passing it to processor specs. We can discuss this more offline, and I'm looking into this now.

andyyang890 · 2025-01-06T23:15:34Z

pkg/ccl/changefeedccl/changefeedbase/settings.go

+
+var Quantize = settings.RegisterDurationSettingWithExplicitUnit(
+	settings.ApplicationLevel,
+	"changefeed.frontier.timestamp_granularity",


I'd prefer changefeed.resolved_timestamp.granularity since ultimately this will be surfaced to the user via quantized resolved timestamps. WDYT?

andyyang890 · 2025-01-06T23:20:54Z

pkg/ccl/changefeedccl/kvfeed/kv_feed_test.go

-				_, err := buf.Get(ctx)
+			for eventIdx := 0; eventIdx < tc.expEventsCount; eventIdx++ {
+				e, err := buf.Get(ctx)
+				if tc.expEvents != nil {


nit: this if branch should go under the assert.NoError

andyyang890 · 2025-01-06T23:26:44Z

pkg/ccl/changefeedccl/kvfeed/kv_feed_test.go

+	frontier, err := span.MakeFrontier([]roachpb.Span{{Key: keys.MinKey, EndKey: keys.MaxKey}}...)
+	require.NoError(t, err)
+	for _, e := range events {
+		_, err := frontier.Forward(e.Checkpoint.Span, e.Checkpoint.ResolvedTS)


nit: create both frontiers at the same time and have a single loop where you forward the events to both frontiers

Previously, all checkpoint timestamps were forwarded precisely to the resolved timestamp. To improve merging of adjacent frontier spans, this patch introduces a frontier quantization mode. In this mode, resolved timestamps from rangefeeds are rounded down to the nearest multiple of the quantization granularity. Rounding down ensures no progress is missed and that we consistently track the lowest possible resolved timestamp. Resolves: cockroachdb#125723 Release note: changefeed.frontier.timestamp_granularity (default off) has been introduced to enable more efficient tracking of resolved timestamps.

wenyihu6 marked this pull request as ready for review January 6, 2025 17:20

wenyihu6 requested a review from a team as a code owner January 6, 2025 17:20

wenyihu6 requested review from asg0451, rharding6373 and andyyang890 and removed request for a team January 6, 2025 17:20

wenyihu6 force-pushed the frontierquantize branch 2 times, most recently from ecc531b to 8eae6cf Compare January 6, 2025 18:07

asg0451 reviewed Jan 6, 2025

View reviewed changes

wenyihu6 force-pushed the frontierquantize branch from 8eae6cf to d165a8c Compare January 6, 2025 20:37

asg0451 reviewed Jan 6, 2025

View reviewed changes

wenyihu6 force-pushed the frontierquantize branch 2 times, most recently from cd1cba8 to 8bcc47f Compare January 6, 2025 22:16

dt reviewed Jan 6, 2025

View reviewed changes

pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go Show resolved Hide resolved

wenyihu6 force-pushed the frontierquantize branch 2 times, most recently from 64820b0 to 3f5a7ef Compare January 6, 2025 22:24

andyyang890 reviewed Jan 6, 2025

View reviewed changes

wenyihu6 force-pushed the frontierquantize branch from 3f5a7ef to bfe3ed5 Compare January 7, 2025 16:32

wenyihu6 requested review from asg0451 and andyyang890 January 7, 2025 16:34

wenyihu6 force-pushed the frontierquantize branch from bfe3ed5 to 30a63fc Compare January 7, 2025 18:25

wenyihu6 force-pushed the frontierquantize branch from 30a63fc to 592c7b6 Compare January 7, 2025 21:27

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

changefeedccl: enable frontier quantization #138328

changefeedccl: enable frontier quantization #138328

wenyihu6 commented Jan 6, 2025 •

edited

Loading

cockroach-teamcity commented Jan 6, 2025

asg0451 Jan 6, 2025

asg0451 Jan 6, 2025

wenyihu6 Jan 6, 2025

wenyihu6 Jan 6, 2025

asg0451 Jan 6, 2025

andyyang890 Jan 6, 2025

wenyihu6 Jan 7, 2025

asg0451 left a comment

asg0451 Jan 6, 2025

wenyihu6 Jan 6, 2025

asg0451 Jan 6, 2025

asg0451 Jan 6, 2025

wenyihu6 Jan 6, 2025

asg0451 Jan 6, 2025

wenyihu6 Jan 6, 2025

asg0451 Jan 6, 2025

wenyihu6 Jan 6, 2025

andyyang890 left a comment

andyyang890 Jan 6, 2025

wenyihu6 Jan 7, 2025

andyyang890 Jan 6, 2025

andyyang890 Jan 6, 2025

wenyihu6 Jan 7, 2025 •

edited

Loading

andyyang890 Jan 6, 2025

wenyihu6 Jan 7, 2025

andyyang890 Jan 6, 2025

wenyihu6 Jan 7, 2025

andyyang890 Jan 6, 2025

wenyihu6 Jan 7, 2025

changefeedccl: enable frontier quantization #138328

Are you sure you want to change the base?

changefeedccl: enable frontier quantization #138328

Conversation

wenyihu6 commented Jan 6, 2025 • edited Loading

cockroach-teamcity commented Jan 6, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asg0451 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andyyang890 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wenyihu6 Jan 7, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wenyihu6 commented Jan 6, 2025 •

edited

Loading

wenyihu6 Jan 7, 2025 •

edited

Loading