-
Notifications
You must be signed in to change notification settings - Fork 887
Part 1: Bigtable client and Kafka ingestion #3776
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
Kbhat1
wants to merge
2
commits into
main
Choose a base branch
from
mvcc-bigtable-pt1
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Open
Changes from all commits
Commits
Show all changes
2 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 hidden or 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,57 @@ | ||
| # Historical State Offload (Bigtable) | ||
|
|
||
| Bigtable holds immutable MVCC mutation rows for history that local SS has | ||
| pruned. The shape is narrow: | ||
|
|
||
| - local SS remains the hot store for recent state, writes, imports, pruning, and iterators | ||
| - Bigtable keeps immutable MVCC mutation rows for older history | ||
| - reads below local SS retention can fall back to Bigtable for `Get` and `Has` | ||
|
|
||
| Row keys are salted with an inverted height suffix: | ||
|
|
||
| ```text | ||
| m | shard(store,key) | store_name | state_key | inverted_height | ||
| ``` | ||
|
|
||
| Reads scan from `inverted(target_height)` and stop after the first row, giving | ||
| the latest write at or before the requested height. Ordered prefix iteration is | ||
| intentionally not served from the offload store. | ||
|
|
||
| ## Consumer | ||
|
|
||
| The consumer reads historical offload changelog messages from Kafka and writes | ||
| them into Bigtable. Kafka offsets are committed only after the sink write | ||
| succeeds. Mutation rows are written before the version marker. | ||
|
|
||
| ```bash | ||
| cbt -project my-gcp-project -instance sei-history createtable state_mutations | ||
| cbt -project my-gcp-project -instance sei-history createfamily state_mutations state | ||
|
|
||
| go run ./sei-db/state_db/ss/offload/consumer/cmd/historical-offload-consumer \ | ||
| ./sei-db/state_db/ss/offload/consumer/config/example-bigtable.json | ||
| ``` | ||
|
|
||
| The example config is a local/dev placeholder. Set real Kafka brokers and | ||
| Bigtable credentials/config in your own config. | ||
|
|
||
| For Google Cloud Managed Service for Apache Kafka, connect with TLS plus | ||
| SASL/PLAIN using service-account credentials: | ||
|
|
||
| ```json | ||
| "Kafka": { | ||
| "Brokers": ["bootstrap.CLUSTER.REGION.managedkafka.PROJECT.cloud.goog:9092"], | ||
| "TLSEnabled": true, | ||
| "SASLMechanism": "plain", | ||
| "Username": "kafka-client@PROJECT.iam.gserviceaccount.com", | ||
| "Password": "<base64-encoded service account key JSON>" | ||
| } | ||
| ``` | ||
|
|
||
| ## Current Limits | ||
|
|
||
| - The node-side read fallback lands in part 2; this part is the client library | ||
| and the ingestion pipeline. | ||
| - No cross-row transaction on ingest; mutation rows are written first and the | ||
| version marker is written last, so replay is idempotent after partial failure. | ||
| - No automatic table creation from the binary. | ||
| - No backfill tooling; coverage starts when ingestion starts. |
This file contains hidden or 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,233 @@ | ||
| package consumer | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "sort" | ||
| "strconv" | ||
| "time" | ||
|
|
||
| "github.com/sei-protocol/sei-chain/sei-db/proto" | ||
| "github.com/sei-protocol/sei-chain/sei-db/state_db/ss/offload/historical" | ||
| "golang.org/x/sync/errgroup" | ||
| ) | ||
|
|
||
| type BigtableConfig = historical.BigtableConfig | ||
|
|
||
| const ( | ||
| defaultBigtableMutationChunkRows = 1024 | ||
| defaultBigtableMutationChunkConcurrency = 8 | ||
| ) | ||
|
|
||
| type bigtableSink struct { | ||
| client *historical.BigtableClient | ||
| applyBulk historical.BigtableApplyBulkFunc | ||
| family string | ||
| shards int | ||
| bulkChunkRows int | ||
| bulkChunkWorkers int | ||
| } | ||
|
|
||
| var _ Sink = (*bigtableSink)(nil) | ||
|
|
||
| func NewBigtableSink(cfg BigtableConfig) (Sink, error) { | ||
| cfg.ApplyDefaults() | ||
| if err := cfg.Validate(); err != nil { | ||
| return nil, err | ||
| } | ||
| ctx := context.Background() | ||
| client, err := historical.OpenBigtableClient(ctx, cfg) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &bigtableSink{ | ||
| client: client, | ||
| applyBulk: client.ApplyBulk, | ||
| family: cfg.Family, | ||
| shards: cfg.Shards, | ||
| bulkChunkRows: defaultBigtableMutationChunkRows, | ||
| bulkChunkWorkers: defaultBigtableMutationChunkConcurrency, | ||
| }, nil | ||
| } | ||
|
|
||
| func (s *bigtableSink) Close() error { | ||
| if s.client != nil { | ||
| return s.client.Close() | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (s *bigtableSink) WriteBatch(ctx context.Context, records []Record) error { | ||
| records = compactRecords(records) | ||
| if len(records) == 0 { | ||
| return nil | ||
| } | ||
| if err := s.writeRecordRows(ctx, records); err != nil { | ||
| return err | ||
| } | ||
| return s.writeVersionMarkers(ctx, records) | ||
| } | ||
|
|
||
| func (s *bigtableSink) writeRecordRows(ctx context.Context, records []Record) error { | ||
| rows := make([]historical.BigtableRowMutation, 0, bigtableRowMutationCount(records)) | ||
| for _, rec := range records { | ||
| rows = s.appendRecordRowMutations(rows, rec.Entry) | ||
| } | ||
| if len(rows) == 0 { | ||
| return nil | ||
| } | ||
| return s.applyRecordRowMutations(ctx, rows) | ||
| } | ||
|
|
||
| func (s *bigtableSink) applyRecordRowMutations(ctx context.Context, rows []historical.BigtableRowMutation) error { | ||
| chunks := bigtableRowMutationChunks(rows, s.bulkChunkRows) | ||
| g, gctx := errgroup.WithContext(ctx) | ||
| g.SetLimit(s.bulkChunkWorkers) | ||
| for _, chunk := range chunks { | ||
| chunk := chunk | ||
| g.Go(func() error { | ||
| errs, err := s.applyBulk(gctx, chunk) | ||
| return bigtableBulkError(chunk, errs, err) | ||
| }) | ||
| } | ||
| return g.Wait() | ||
| } | ||
|
|
||
| func (s *bigtableSink) appendRecordRowMutations(rows []historical.BigtableRowMutation, entry *proto.ChangelogEntry) []historical.BigtableRowMutation { | ||
| for _, mutation := range compactMutations(entry) { | ||
| rows = append(rows, s.mutationRow(entry.Version, mutation.storeName, mutation.pair)) | ||
| } | ||
| for _, up := range entry.Upgrades { | ||
| rows = append(rows, s.upgradeRow(entry.Version, up)) | ||
| } | ||
| return rows | ||
| } | ||
|
|
||
| // mutationRow writes value+deleted cells for live pairs but only a deleted | ||
| // cell for tombstones, saving a cell per delete. Readers must therefore check | ||
| // the deleted column before trusting any value cell — a replayed live write | ||
| // followed by a tombstone leaves both cells on the row. | ||
| func (s *bigtableSink) mutationRow(version int64, storeName string, pair *proto.KVPair) historical.BigtableRowMutation { | ||
| ts := historical.BigtableTimestamp(version) | ||
| deleted := pair.Delete || pair.Value == nil | ||
| rowKey := historical.BigtableMutationRowKey(storeName, pair.Key, version, s.shards) | ||
| if deleted { | ||
| return historical.BigtableRowMutation{ | ||
| RowKey: rowKey, | ||
| SetCells: []historical.BigtableSetCell{{ | ||
| Family: s.family, | ||
| Qualifier: historical.BigtableDeletedColumn, | ||
| TimestampMicros: ts, | ||
| Value: boolByte(true), | ||
| }}, | ||
| } | ||
| } | ||
| return historical.BigtableRowMutation{ | ||
| RowKey: rowKey, | ||
| SetCells: []historical.BigtableSetCell{ | ||
| {Family: s.family, Qualifier: historical.BigtableValueColumn, TimestampMicros: ts, Value: pair.Value}, | ||
| {Family: s.family, Qualifier: historical.BigtableDeletedColumn, TimestampMicros: ts, Value: boolByte(false)}, | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| func (s *bigtableSink) upgradeRow(version int64, up *proto.TreeNameUpgrade) historical.BigtableRowMutation { | ||
| ts := historical.BigtableTimestamp(version) | ||
| return historical.BigtableRowMutation{ | ||
| RowKey: historical.BigtableUpgradeRowKey(version, up.Name), | ||
| SetCells: []historical.BigtableSetCell{ | ||
| {Family: s.family, Qualifier: "rename_from", TimestampMicros: ts, Value: []byte(up.RenameFrom)}, | ||
| {Family: s.family, Qualifier: historical.BigtableDeletedColumn, TimestampMicros: ts, Value: boolByte(up.Delete)}, | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| func (s *bigtableSink) writeVersionMarkers(ctx context.Context, records []Record) error { | ||
| rows := make([]historical.BigtableRowMutation, 0, len(records)) | ||
| ingestedAt := []byte(strconv.FormatInt(time.Now().UnixNano(), 10)) | ||
| for _, rec := range records { | ||
| version := rec.Entry.Version | ||
| ts := historical.BigtableTimestamp(version) | ||
| rows = append(rows, historical.BigtableRowMutation{ | ||
| RowKey: historical.BigtableVersionRowKey(version), | ||
| SetCells: []historical.BigtableSetCell{ | ||
| {Family: s.family, Qualifier: "topic", TimestampMicros: ts, Value: []byte(rec.Topic)}, | ||
| {Family: s.family, Qualifier: "partition", TimestampMicros: ts, Value: []byte(strconv.Itoa(rec.Partition))}, | ||
| {Family: s.family, Qualifier: "offset", TimestampMicros: ts, Value: []byte(strconv.FormatInt(rec.Offset, 10))}, | ||
| {Family: s.family, Qualifier: "ingested_at_unix_nano", TimestampMicros: ts, Value: ingestedAt}, | ||
| }, | ||
| }) | ||
| } | ||
| errs, err := s.applyBulk(ctx, rows) | ||
| if err := bigtableBulkError(rows, errs, err); err != nil { | ||
| return fmt.Errorf("insert bigtable version markers: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func bigtableRowMutationCount(records []Record) int { | ||
| total := 0 | ||
| for _, rec := range records { | ||
| total += entryMutationCapacity(rec.Entry) + len(rec.Entry.Upgrades) | ||
| } | ||
| return total | ||
| } | ||
|
|
||
| func bigtableRowMutationChunks(rows []historical.BigtableRowMutation, maxRows int) [][]historical.BigtableRowMutation { | ||
| if len(rows) == 0 { | ||
| return nil | ||
| } | ||
| if maxRows <= 0 { | ||
| maxRows = len(rows) | ||
| } | ||
| sort.Slice(rows, func(i, j int) bool { | ||
| return rows[i].RowKey < rows[j].RowKey | ||
| }) | ||
|
|
||
| chunks := make([][]historical.BigtableRowMutation, 0, (len(rows)+maxRows-1)/maxRows) | ||
| start := 0 | ||
| startLocality := bigtableRowLocality(rows[0].RowKey) | ||
| for i := 1; i < len(rows); i++ { | ||
| locality := bigtableRowLocality(rows[i].RowKey) | ||
| if i-start >= maxRows || locality != startLocality { | ||
| chunks = append(chunks, rows[start:i]) | ||
| start = i | ||
| startLocality = locality | ||
| } | ||
| } | ||
| return append(chunks, rows[start:]) | ||
| } | ||
|
|
||
| func bigtableRowLocality(rowKey string) string { | ||
| // Mutation row keys are m|shard|store|key|version; keep chunks inside one | ||
| // shard prefix so separate chunks can hit separate Bigtable tablets. | ||
| if len(rowKey) >= 3 && rowKey[0] == 'm' { | ||
| return rowKey[:3] | ||
| } | ||
| if len(rowKey) > 0 { | ||
| return rowKey[:1] | ||
| } | ||
| return rowKey | ||
| } | ||
|
|
||
| func bigtableBulkError(rows []historical.BigtableRowMutation, errs []error, err error) error { | ||
| if err != nil { | ||
| return err | ||
| } | ||
| if len(errs) != len(rows) { | ||
| return fmt.Errorf("bigtable returned %d mutation results for %d rows", len(errs), len(rows)) | ||
| } | ||
| for i, rowErr := range errs { | ||
| if rowErr != nil { | ||
| return fmt.Errorf("row %q: %w", rows[i].RowKey, rowErr) | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func boolByte(v bool) []byte { | ||
| if v { | ||
| return []byte{1} | ||
| } | ||
| return []byte{0} | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
[nit]
deleted := pair.Delete || pair.Value == nilconflates a live write of a nil value with a tombstone, so a genuine nil-value write is stored (and later read) as absent.bigtableValueFromRowmirrors this (value == nil→ ErrNotFound), so it's self-consistent, but worth a comment confirming SS never emits a live nil-value pair — otherwise such writes silently disappear.