-
Notifications
You must be signed in to change notification settings - Fork 3.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
changefeedccl: move checkpoint creation code into separate package
This patch moves the changefeed checkpoint creation code into a separate package to prevent a dependency cycle when the resolved span frontiers are moved into its own package in a later commit. Given the upcoming checkpoint improvement work, this is also a good opportunity to clean up the checkpoint creation API. Finally, a unit test covering checkpoint creation has been added. Release note: None
- Loading branch information
1 parent
8057da9
commit c0e886c
Showing
9 changed files
with
308 additions
and
154 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test") | ||
|
|
||
| go_library( | ||
| name = "checkpoint", | ||
| srcs = ["checkpoint.go"], | ||
| importpath = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/checkpoint", | ||
| visibility = ["//visibility:public"], | ||
| deps = [ | ||
| "//pkg/jobs/jobspb", | ||
| "//pkg/roachpb", | ||
| "//pkg/util/hlc", | ||
| "//pkg/util/span", | ||
| ], | ||
| ) | ||
|
|
||
| go_test( | ||
| name = "checkpoint_test", | ||
| srcs = ["checkpoint_test.go"], | ||
| deps = [ | ||
| ":checkpoint", | ||
| "//pkg/ccl/changefeedccl/changefeedbase", | ||
| "//pkg/jobs/jobspb", | ||
| "//pkg/kv/kvserver/rangefeed", | ||
| "//pkg/roachpb", | ||
| "//pkg/util/hlc", | ||
| "//pkg/util/leaktest", | ||
| "//pkg/util/log", | ||
| "//pkg/util/randutil", | ||
| "//pkg/util/shuffle", | ||
| "//pkg/util/span", | ||
| "@com_github_stretchr_testify//require", | ||
| ], | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| // Copyright 2024 The Cockroach Authors. | ||
| // | ||
| // Use of this software is governed by the CockroachDB Software License | ||
| // included in the /LICENSE file. | ||
|
|
||
| // Package checkpoint contains code responsible for handling changefeed | ||
| // checkpoints. | ||
| package checkpoint | ||
|
|
||
| import ( | ||
| "github.com/cockroachdb/cockroach/pkg/jobs/jobspb" | ||
| "github.com/cockroachdb/cockroach/pkg/roachpb" | ||
| "github.com/cockroachdb/cockroach/pkg/util/hlc" | ||
| "github.com/cockroachdb/cockroach/pkg/util/span" | ||
| ) | ||
|
|
||
| // SpanIter is an iterator over a collection of spans. | ||
| type SpanIter func(forEachSpan span.Operation) | ||
|
|
||
| // Make creates a checkpoint with as many spans that should be checkpointed (are | ||
| // above the highwater mark) as can fit in maxBytes, along with the earliest | ||
| // timestamp of the checkpointed spans. A SpanGroup is used to merge adjacent | ||
| // spans above the high-water mark. | ||
| func Make( | ||
| frontier hlc.Timestamp, forEachSpan SpanIter, maxBytes int64, | ||
| ) jobspb.ChangefeedProgress_Checkpoint { | ||
| // Collect leading spans into a SpanGroup to merge adjacent spans and store | ||
| // the lowest timestamp found | ||
| var checkpointSpanGroup roachpb.SpanGroup | ||
| checkpointFrontier := hlc.Timestamp{} | ||
| forEachSpan(func(s roachpb.Span, ts hlc.Timestamp) span.OpResult { | ||
| if frontier.Less(ts) { | ||
| checkpointSpanGroup.Add(s) | ||
| if checkpointFrontier.IsEmpty() || ts.Less(checkpointFrontier) { | ||
| checkpointFrontier = ts | ||
| } | ||
| } | ||
| return span.ContinueMatch | ||
| }) | ||
|
|
||
| // Ensure we only return up to maxBytes spans | ||
| var checkpointSpans []roachpb.Span | ||
| var used int64 | ||
| for _, span := range checkpointSpanGroup.Slice() { | ||
| used += int64(len(span.Key)) + int64(len(span.EndKey)) | ||
| if used > maxBytes { | ||
| break | ||
| } | ||
| checkpointSpans = append(checkpointSpans, span) | ||
| } | ||
|
|
||
| return jobspb.ChangefeedProgress_Checkpoint{ | ||
| Spans: checkpointSpans, | ||
| Timestamp: checkpointFrontier, | ||
| } | ||
| } |
Oops, something went wrong.