[Storage] Refactor stored chunk data pack #7983
+1,478
−501
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 | ||||||
|---|---|---|---|---|---|---|---|---|
|
|
@@ -87,16 +87,14 @@ func runE(*cobra.Command, []string) error { | |||||||
| return fmt.Errorf("could not open chunk data pack DB at %v: %w", flagChunkDataPackDir, err) | ||||||||
| } | ||||||||
| chunkDataPacksDB := pebbleimpl.ToDB(chunkDataPacksPebbleDB) | ||||||||
| storedChunkDataPacks := store.NewStoredChunkDataPacks(metrics, chunkDataPacksDB, 1000) | ||||||||
| chunkDataPacks := store.NewChunkDataPacks(metrics, chunkDataPacksDB, storedChunkDataPacks, collections, 1000) | ||||||||
| protocolDBBatch := db.NewBatch() | ||||||||
| defer protocolDBBatch.Close() | ||||||||
|
|
||||||||
| // collect chunk IDs to be removed | ||||||||
| chunkIDs, err := removeExecutionResultsFromHeight( | ||||||||
| protocolDBBatch, | ||||||||
| state, | ||||||||
| transactionResults, | ||||||||
| commits, | ||||||||
|
|
@@ -112,12 +110,19 @@ func runE(*cobra.Command, []string) error { | |||||||
| } | ||||||||
|
|
||||||||
| // remove chunk data packs first, because otherwise the index to find chunk data pack will be removed. | ||||||||
| if len(chunkIDs) > 0 { | ||||||||
| chunkDataPackIDs, err := chunkDataPacks.BatchRemove(chunkIDs, protocolDBBatch) | ||||||||
| if err != nil { | ||||||||
| return fmt.Errorf("could not remove chunk data packs at %v: %w", flagHeight, err) | ||||||||
| } | ||||||||
|
|
||||||||
| err = storedChunkDataPacks.Remove(chunkDataPackIDs) | ||||||||
| if err != nil { | ||||||||
| return fmt.Errorf("could not commit chunk batch at %v: %w", flagHeight, err) | ||||||||
| } | ||||||||
| } | ||||||||
|
|
||||||||
| err = protocolDBBatch.Commit() | ||||||||
| if err != nil { | ||||||||
| return fmt.Errorf("could not flush write batch at %v: %w", flagHeight, err) | ||||||||
| } | ||||||||
|
|
@@ -141,8 +146,7 @@ func runE(*cobra.Command, []string) error { | |||||||
| // use badger instances directly instead of stroage interfaces so that the interface don't | ||||||||
| // need to include the Remove methods | ||||||||
|
Comment on lines
146
to
147
There was a problem hiding this comment. this doc could use an update, please. |
||||||||
| func removeExecutionResultsFromHeight( | ||||||||
| protocolDBBatch storage.Batch, | ||||||||
| protoState protocol.State, | ||||||||
| transactionResults storage.TransactionResults, | ||||||||
| commits storage.Commits, | ||||||||
|
|
@@ -151,40 +155,42 @@ func removeExecutionResultsFromHeight( | |||||||
| myReceipts storage.MyExecutionReceipts, | ||||||||
| events storage.Events, | ||||||||
| serviceEvents storage.ServiceEvents, | ||||||||
| fromHeight uint64) ([]flow.Identifier, error) { | ||||||||
|
There was a problem hiding this comment. I prefer to put the return value on the next line. Thereby, the function body is visually separated from eh header via a line without indentation. It helps human brains to quicker structure the visually seen data according to broad structural information (color, indentation, etc)
Suggested change
|
||||||||
| log.Info().Msgf("removing results for blocks from height: %v", fromHeight) | ||||||||
|
|
||||||||
| root := protoState.Params().FinalizedRoot() | ||||||||
|
|
||||||||
| if fromHeight <= root.Height { | ||||||||
| return nil, fmt.Errorf("can only remove results for block above root block. fromHeight: %v, rootHeight: %v", fromHeight, root.Height) | ||||||||
| } | ||||||||
|
|
||||||||
| final, err := protoState.Final().Head() | ||||||||
| if err != nil { | ||||||||
| return nil, fmt.Errorf("could get not finalized height: %w", err) | ||||||||
| } | ||||||||
|
|
||||||||
| if fromHeight > final.Height { | ||||||||
| return nil, fmt.Errorf("could not remove results for unfinalized height: %v, finalized height: %v", fromHeight, final.Height) | ||||||||
| } | ||||||||
|
|
||||||||
| finalRemoved := 0 | ||||||||
| total := int(final.Height-fromHeight) + 1 | ||||||||
| var allChunkIDs []flow.Identifier | ||||||||
|
|
||||||||
| // removing for finalized blocks | ||||||||
| for height := fromHeight; height <= final.Height; height++ { | ||||||||
| head, err := protoState.AtHeight(height).Head() | ||||||||
| if err != nil { | ||||||||
| return nil, fmt.Errorf("could not get header at height: %w", err) | ||||||||
| } | ||||||||
|
|
||||||||
| blockID := head.ID() | ||||||||
|
|
||||||||
| chunkIDs, err := removeForBlockID(protocolDBBatch, commits, transactionResults, results, chunkDataPacks, myReceipts, events, serviceEvents, blockID) | ||||||||
| if err != nil { | ||||||||
| return nil, fmt.Errorf("could not remove result for finalized block: %v, %w", blockID, err) | ||||||||
| } | ||||||||
| allChunkIDs = append(allChunkIDs, chunkIDs...) | ||||||||
|
|
||||||||
| finalRemoved++ | ||||||||
| log.Info().Msgf("result at height %v has been removed. progress (%v/%v)", height, finalRemoved, total) | ||||||||
|
|
@@ -193,18 +199,18 @@ func removeExecutionResultsFromHeight( | |||||||
| // removing for pending blocks | ||||||||
| pendings, err := protoState.Final().Descendants() | ||||||||
| if err != nil { | ||||||||
| return nil, fmt.Errorf("could not get pending block: %w", err) | ||||||||
| } | ||||||||
|
|
||||||||
| pendingRemoved := 0 | ||||||||
| total = len(pendings) | ||||||||
|
|
||||||||
| for _, pending := range pendings { | ||||||||
| chunkIDs, err := removeForBlockID(protocolDBBatch, commits, transactionResults, results, chunkDataPacks, myReceipts, events, serviceEvents, pending) | ||||||||
| if err != nil { | ||||||||
| return nil, fmt.Errorf("could not remove result for pending block %v: %w", pending, err) | ||||||||
| } | ||||||||
| allChunkIDs = append(allChunkIDs, chunkIDs...) | ||||||||
|
|
||||||||
| pendingRemoved++ | ||||||||
| log.Info().Msgf("result for pending block %v has been removed. progress (%v/%v) ", pending, pendingRemoved, total) | ||||||||
|
|
@@ -213,15 +219,14 @@ func removeExecutionResultsFromHeight( | |||||||
| log.Info().Msgf("removed height from %v. removed for %v finalized blocks, and %v pending blocks", | ||||||||
| fromHeight, finalRemoved, pendingRemoved) | ||||||||
|
|
||||||||
| return allChunkIDs, nil | ||||||||
| } | ||||||||
|
|
||||||||
| // removeForBlockID remove block execution related data for a given block. | ||||||||
| // All data to be removed will be removed in a batch write. | ||||||||
| // It bubbles up any error encountered | ||||||||
| func removeForBlockID( | ||||||||
| protocolDBBatch storage.Batch, | ||||||||
| commits storage.Commits, | ||||||||
| transactionResults storage.TransactionResults, | ||||||||
| results storage.ExecutionResults, | ||||||||
|
|
@@ -230,74 +235,70 @@ func removeForBlockID( | |||||||
| events storage.Events, | ||||||||
| serviceEvents storage.ServiceEvents, | ||||||||
| blockID flow.Identifier, | ||||||||
| ) ([]flow.Identifier, error) { | ||||||||
| result, err := results.ByBlockID(blockID) | ||||||||
| if errors.Is(err, storage.ErrNotFound) { | ||||||||
| log.Info().Msgf("result not found for block %v", blockID) | ||||||||
| return nil, nil | ||||||||
| } | ||||||||
|
|
||||||||
| if err != nil { | ||||||||
| return nil, fmt.Errorf("could not find result for block %v: %w", blockID, err) | ||||||||
| } | ||||||||
|
|
||||||||
| chunkIDs := make([]flow.Identifier, 0, len(result.Chunks)) | ||||||||
| for _, chunk := range result.Chunks { | ||||||||
| chunkID := chunk.ID() | ||||||||
| chunkIDs = append(chunkIDs, chunkID) | ||||||||
| } | ||||||||
|
|
||||||||
| // remove commits | ||||||||
| err = commits.BatchRemoveByBlockID(blockID, protocolDBBatch) | ||||||||
| if err != nil { | ||||||||
| if errors.Is(err, storage.ErrNotFound) { | ||||||||
| return nil, fmt.Errorf("could not remove by block ID %v: %w", blockID, err) | ||||||||
| } | ||||||||
|
|
||||||||
| log.Warn().Msgf("statecommitment not found for block %v", blockID) | ||||||||
| } | ||||||||
|
|
||||||||
| // remove transaction results | ||||||||
| err = transactionResults.BatchRemoveByBlockID(blockID, protocolDBBatch) | ||||||||
| if err != nil { | ||||||||
| return nil, fmt.Errorf("could not remove transaction results by BlockID %v: %w", blockID, err) | ||||||||
| } | ||||||||
|
|
||||||||
| // remove own execution results index | ||||||||
| err = myReceipts.BatchRemoveIndexByBlockID(blockID, protocolDBBatch) | ||||||||
| if err != nil { | ||||||||
| if !errors.Is(err, storage.ErrNotFound) { | ||||||||
| return nil, fmt.Errorf("could not remove own receipt by BlockID %v: %w", blockID, err) | ||||||||
| } | ||||||||
|
|
||||||||
| log.Warn().Msgf("own receipt not found for block %v", blockID) | ||||||||
| } | ||||||||
|
|
||||||||
| // remove events | ||||||||
| err = events.BatchRemoveByBlockID(blockID, protocolDBBatch) | ||||||||
| if err != nil { | ||||||||
| return nil, fmt.Errorf("could not remove events by BlockID %v: %w", blockID, err) | ||||||||
| } | ||||||||
|
|
||||||||
| // remove service events | ||||||||
| err = serviceEvents.BatchRemoveByBlockID(blockID, protocolDBBatch) | ||||||||
| if err != nil { | ||||||||
| return nil, fmt.Errorf("could not remove service events by blockID %v: %w", blockID, err) | ||||||||
| } | ||||||||
|
|
||||||||
| // remove execution result index | ||||||||
| err = results.BatchRemoveIndexByBlockID(blockID, protocolDBBatch) | ||||||||
| if err != nil { | ||||||||
| if !errors.Is(err, storage.ErrNotFound) { | ||||||||
| return nil, fmt.Errorf("could not remove result by BlockID %v: %w", blockID, err) | ||||||||
| } | ||||||||
|
|
||||||||
| log.Warn().Msgf("result not found for block %v", blockID) | ||||||||
| } | ||||||||
|
|
||||||||
|
There was a problem hiding this comment. [This might be more of a suggestion for your follow up PR you mentioned above.] I would include the chunk data pack index here:
|
||||||||
| return chunkIDs, nil | ||||||||
| } | ||||||||
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.
chunkDataPacks.BatchRemoveinternally also callsstoredChunkDataPacks.Remove