BE-636 | Add pipeline package #570
Conversation
Adds pipeline package allowing to manipulate data in various ways including filtering, sorting, paginating and iterating.
deividaspetraitis
requested review from
cryptomatictrader,
PaddyMc and
p0mvn
|
Caution Review failedThe pull request is closed. WalkthroughThe changes introduce several new functionalities within the Changes
Possibly related PRs
Suggested reviewers
Poem
Warning There were issues while running some tools. Please review the errors and either fix the tool’s configuration or disable the tool if it’s a critical failure. 🔧 buf (1.47.2)sqsdomain/proto/ingest.protoAn unexpected error occurred while running buf. 🔧 golangci-lint (1.62.2)level=warning msg="The linter 'exportloopref' is deprecated (since v1.60.2) due to: Since Go1.22 (loopvar) this linter is no longer relevant. Replaced by copyloopvar." Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
Documentation and Community
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
🧹 Outside diff range and nitpick comments (4)
domain/pipeline/iterator.go (1)
22-27: Consider thread safety for index fieldThe
indexfield could be accessed concurrently throughNext(),SetOffset(), andReset()methods. Consider using atomic operations or mutex protection.type SyncMapIterator[K, V any] struct { data *sync.Map keys []K - index int + mu sync.RWMutex + index atomic.Int64 }domain/pipeline/paginator.go (1)
19-20: Update function names in the comment to match the actual methodsThe comment mentions
GetPageBasedPageandGetCursorBasedPage, but the actual method names areFetchPageByPageNumberandFetchPageByCursor. Please update the comment for consistency.Apply this diff to correct the comment:
-// Under the hood it calls either GetPageBasedPage or GetCursorBasedPage. +// Under the hood it calls either FetchPageByPageNumber or FetchPageByCursor.domain/pipeline/transformer.go (2)
74-96: Refine sorting logic to properly handle multiple criteriaIn the
Sortmethod, when multiple comparison functions are provided, the current implementation returns based on the first criterion only. This may not correctly handle secondary sorting conditions.Modify the sorting logic to consider all provided criteria:
sort.Slice(dt.keys, func(i, j int) bool { vi, _ := dt.load(dt.keys[i]) vj, _ := dt.load(dt.keys[j]) for _, lessFn := range less { - return lessFn(vi, vj) + if lessFn(vi, vj) { + return true + } + if lessFn(vj, vi) { + return false + } } return false })This approach checks each sorting criterion in order and only moves to the next criterion if the current one deems the elements equal.
125-137: Simplify theloadmethod using inline type assertionsThe
loadmethod can be simplified for clarity and conciseness.Simplify the method as follows:
func (dt *SyncMapTransformer[K, V]) load(key K) (V, bool) { - mv, ok := dt.data.Load(key) - if !ok { - return *new(V), false - } - v, ok := mv.(V) - if !ok { - return *new(V), false - } - return v, true + if mv, ok := dt.data.Load(key); ok { + if v, ok := mv.(V); ok { + return v, true + } + } + var zeroValue V + return zeroValue, false }This makes the method more readable by reducing nesting and using inline type assertions.
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
⛔ Files ignored due to path filters (3)
domain/pipeline/iterator_test.gois excluded by!**/*_test.godomain/pipeline/paginator_test.gois excluded by!**/*_test.godomain/pipeline/transformer_test.gois excluded by!**/*_test.go
📒 Files selected for processing (3)
domain/pipeline/iterator.go(1 hunks)domain/pipeline/paginator.go(1 hunks)domain/pipeline/transformer.go(1 hunks)
🧰 Additional context used
🪛 GitHub Check: build
domain/pipeline/paginator.go
[failure] 3-3:
no required module provides package github.com/osmosis-labs/sqs/pkg/api/v1beta1; to add it:
🔇 Additional comments (2)
domain/pipeline/iterator.go (2)
1-3: LGTM!
Package name and import are appropriate for the functionality.
1-65: Verify integration with pagination package
Since this iterator is part of a larger pagination system, we should verify its integration with the paginator package.
✅ Verification successful
Let me search for more context about the pagination integration.
Let me check the paginator implementation to verify the integration.
Integration with pagination package is properly implemented
The Iterator interface is correctly integrated with the pagination system:
- The
Paginatorstruct properly composes theIteratorinterface - Both page-based and cursor-based pagination strategies utilize all required Iterator methods (Next, HasNext, SetOffset, Reset)
- Comprehensive test coverage demonstrates the integration works as expected with different pagination scenarios
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Search for usage of Iterator interface in paginator package
ast-grep --pattern 'type $_ struct {
$$$
iterator Iterator[$_, $_]
$$$
}'
# Search for SyncMapIterator instantiation
rg -A 5 'NewSyncMapIterator'
Length of output: 1117
Script:
#!/bin/bash
# Search for files containing "paginator" or "pagination"
fd -t f -e go . | xargs rg -l "pagin"
# Search for any struct implementing the Iterator interface
ast-grep --pattern 'type $name struct {
$$$
}
func ($recv *$name) Next() ($_, bool) { $$$ }
func ($recv *$name) HasNext() bool { $$$ }
func ($recv *$name) SetOffset(offset int) { $$$ }
func ($recv *$name) Reset() { $$$ }'
# Search for any usage of Iterator interface
rg "Iterator\[.*\]"
Length of output: 1534
Script:
#!/bin/bash
# Check paginator implementation
cat domain/pipeline/paginator.go
# Check paginator tests
cat domain/pipeline/paginator_test.go
Length of output: 6206
|
Note: Sonarcloud is complaining regarding coverage because of generated |
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Outside diff range and nitpick comments (6)
proto/buf.yaml (2)
4-8: Consider pinning specific versions for all dependenciesWhile
cometbftis pinned to a specific commit, other dependencies likegogo-proto,cosmos-proto, andgoogleapisdon't specify versions. This could lead to unexpected breaking changes if these dependencies are updated.Consider pinning specific versions for all dependencies. For example:
- buf.build/cometbft/cometbft:4a62c99d422068a5165429b62a7eb824df46cca9 - - buf.build/cosmos/gogo-proto + - buf.build/cosmos/gogo-proto:v1.3.2 - buf.build/cosmos/cosmos-sdk:v0.47.0 - - buf.build/cosmos/cosmos-proto - - buf.build/googleapis/googleapis + - buf.build/cosmos/cosmos-proto:v1.0.0 + - buf.build/googleapis/googleapis:v0.1.0🧰 Tools
🪛 yamllint (1.35.1)
[error] 4-4: trailing spaces
(trailing-spaces)
17-34: Consider documenting the rationale for lint exceptionsThe configuration disables numerous lint rules. While these exceptions might be necessary, it would be valuable to document why these specific rules are being disabled to help maintain the codebase's quality standards in the long term.
Consider either:
- Adding comments explaining the rationale for each exception group, or
- Creating a separate documentation file explaining the protobuf style guide and lint configuration decisions.
This will help future contributors understand the reasoning behind these exceptions and prevent unnecessary rule re-enabling.
proto/sqs/query/v1beta1/pagination.proto (1)
43-54: Review numeric type consistencyThere's an inconsistency in numeric types:
next_cursorusesint64(signed)- Other pagination fields use
uint64(unsigned)While using -1 as a sentinel value for
next_cursoris a valid pattern, consider:
- Using a separate boolean field
has_moreinstead- Using consistent unsigned types throughout
- Documenting this design decision
message PaginationResponse { - int64 next_cursor = 1; + uint64 next_cursor = 1; + bool has_more = 3; // true if more results are available uint64 total_items = 2; }pkg/api/v1beta1/pagination.go (3)
49-49: Typo in method comment: 'imlpements' should be 'implements'The comment for
UnmarshalHTTPRequestcontains a typographical error.Apply this diff to correct the typo:
-// UnmarshalHTTPRequest imlpements RequestUnmarshaler interface. +// UnmarshalHTTPRequest implements RequestUnmarshaler interface.
58-77: Refactor repetitive code when parsing query parametersThe code for parsing
pageParam,limitParam, andcursorParamis repetitive. Refactoring it can reduce duplication and enhance maintainability.You can create a helper function:
func parseUintQueryParam(c echo.Context, paramName string) (uint64, error) { param := c.QueryParam(paramName) if param != "" { return strconv.ParseUint(param, 10, 64) } return 0, nil }Modify
UnmarshalHTTPRequestaccordingly:func (r *PaginationRequest) UnmarshalHTTPRequest(c echo.Context) error { var err error - // Fetch query parameters - pageParam := c.QueryParam(queryPageNumber) - limitParam := c.QueryParam(queryPageSize) - cursorParam := c.QueryParam(queryPageCursor) - - if pageParam != "" { - r.Page, err = strconv.ParseUint(pageParam, 10, 64) - if err != nil { - return err - } - } - - if limitParam != "" { - r.Limit, err = strconv.ParseUint(limitParam, 10, 64) - if err != nil { - return err - } - } - - if cursorParam != "" { - r.Cursor, err = strconv.ParseUint(cursorParam, 10, 64) - if err != nil { - return err - } - } + if r.Page, err = parseUintQueryParam(c, queryPageNumber); err != nil { + return fmt.Errorf("invalid value for parameter %s: %w", queryPageNumber, err) + } + if r.Limit, err = parseUintQueryParam(c, queryPageSize); err != nil { + return fmt.Errorf("invalid value for parameter %s: %w", queryPageSize, err) + } + if r.Cursor, err = parseUintQueryParam(c, queryPageCursor); err != nil { + return fmt.Errorf("invalid value for parameter %s: %w", queryPageCursor, err) + }This refactoring reduces duplication and enhances error messages.
60-61: Include parameter names in parsing error messagesWhen parsing query parameters, wrapping errors with parameter names provides better context for debugging.
Apply this diff to improve error messages:
if pageParam != "" { r.Page, err = strconv.ParseUint(pageParam, 10, 64) if err != nil { - return err + return fmt.Errorf("invalid value for parameter %s: %w", queryPageNumber, err) } } if limitParam != "" { r.Limit, err = strconv.ParseUint(limitParam, 10, 64) if err != nil { - return err + return fmt.Errorf("invalid value for parameter %s: %w", queryPageSize, err) } } if cursorParam != "" { r.Cursor, err = strconv.ParseUint(cursorParam, 10, 64) if err != nil { - return err + return fmt.Errorf("invalid value for parameter %s: %w", queryPageCursor, err) } }This enhancement provides clearer error information to the caller.
Also applies to: 67-68, 74-75
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
⛔ Files ignored due to path filters (3)
pkg/api/v1beta1/pagination.pb.gois excluded by!**/*.pb.gopkg/api/v1beta1/pagination_test.gois excluded by!**/*_test.goproto/buf.lockis excluded by!**/*.lock
📒 Files selected for processing (6)
pkg/api/v1beta1/pagination.go(1 hunks)proto/Dockerfile(1 hunks)proto/buf.gen.gogo.yaml(1 hunks)proto/buf.gen.swagger.yaml(1 hunks)proto/buf.yaml(1 hunks)proto/sqs/query/v1beta1/pagination.proto(1 hunks)
✅ Files skipped from review due to trivial changes (3)
- proto/Dockerfile
- proto/buf.gen.gogo.yaml
- proto/buf.gen.swagger.yaml
🧰 Additional context used
🪛 yamllint (1.35.1)
proto/buf.yaml
[error] 4-4: trailing spaces
(trailing-spaces)
🔇 Additional comments (3)
proto/sqs/query/v1beta1/pagination.proto (3)
1-4: LGTM! Package declaration follows best practices.
The package declaration and versioning strategy (v1beta1) align well with protocol buffer best practices.
18-22: Clarify cursor invalidation strategy
The comment mentions that clients should re-request all data if it changes, but doesn't specify:
- How clients detect that data has changed
- How cursor invalidation is communicated to clients
- Whether there's a mechanism to prevent stale cursors from being used
Consider adding these details to the documentation or implementing a cursor validation mechanism.
1-54: Verify consistent pagination implementation
Let's verify that:
- All endpoints using pagination implement both strategies consistently
- Error handling for invalid pagination parameters is implemented
- Documentation exists for pagination usage
✅ Verification successful
Based on the analysis of the codebase through the executed scripts, I can now provide a final response:
Pagination implementation is well-structured and consistent
The codebase shows a robust pagination implementation with:
- Comprehensive validation in
PaginationRequest.Validate()that checks for invalid page numbers, limits, and strategies - Consistent handling of both PAGE and CURSOR strategies in the
Paginatorimplementation - Thorough test coverage demonstrating proper handling of edge cases
- Clean integration with the Echo web framework for HTTP request parsing
The implementation includes proper error handling for:
- Invalid page numbers
- Exceeding maximum page/limit values
- Unsupported pagination strategies
- Out-of-range cursor values
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Check for consistent pagination usage
echo "Checking endpoint implementations..."
rg -A 10 "PaginationRequest|PaginationStrategy" --type go
echo "Checking error handling..."
rg -A 5 "invalid.*pagination|pagination.*invalid" --type go
echo "Checking documentation..."
fd -e md -x grep -l "pagination" {}
Length of output: 40338
Script:
#!/bin/bash
# Let's check for actual usage of pagination in API endpoints
echo "Checking API endpoints using pagination..."
rg -A 5 "func.*\(.*PaginationRequest" --type go
echo "Checking for pagination validation..."
ast-grep --pattern 'func $_(p *PaginationRequest) Validate() error {
$$$
}'
echo "Checking for pagination documentation in API docs..."
fd -g "*.md" -x cat {} | grep -A 5 -B 5 "pagination"
Length of output: 7446
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Outside diff range and nitpick comments (5)
domain/pipeline/paginator.go (2)
1-68: Improve error handling throughout the packageConsider returning errors instead of using panic for validation failures. This would allow better error handling by the caller.
Would you like me to provide an example of how to implement proper error handling throughout the package?
1-68: Add comprehensive test coverageAs this is a critical pagination component, please ensure comprehensive test coverage including:
- Edge cases for both pagination strategies
- Validation error scenarios
- Iterator error handling
- Boundary conditions for page/cursor/limit values
Would you like me to help generate a test suite for this package?
pkg/api/v1beta1/pagination.go (2)
5-5: Remove unnecessary alias for themathpackageThe alias
math "math"is unnecessary since it doesn't change the package name. You can import the package directly without an alias.Apply this diff to remove the alias:
-import ( - "fmt" - math "math" - "strconv" - - "github.com/labstack/echo/v4" -) +import ( + "fmt" + "math" + "strconv" + + "github.com/labstack/echo/v4" +)
50-50: Fix typo in comment: 'imlpements' should be 'implements'There is a typographical error in the comment.
Apply this diff to correct the typo:
-// UnmarshalHTTPRequest imlpements RequestUnmarshaler interface. +// UnmarshalHTTPRequest implements RequestUnmarshaler interface.domain/pipeline/iterator.go (1)
10-10: Update the comment forNext()to reflect return valuesThe comment for the
Next()method states that it returns "a value and a bool indicating if it's valid," but the method signature returns(V, error). Please update the comment to accurately describe the return types.
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
⛔ Files ignored due to path filters (2)
domain/pipeline/iterator_test.gois excluded by!**/*_test.goingest/usecase/plugins/orderbook/claimbot/export_test.gois excluded by!**/*_test.go
📒 Files selected for processing (4)
domain/pipeline/iterator.go(1 hunks)domain/pipeline/paginator.go(1 hunks)domain/pipeline/transformer.go(1 hunks)pkg/api/v1beta1/pagination.go(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- domain/pipeline/transformer.go
🔇 Additional comments (8)
domain/pipeline/paginator.go (3)
3-3: Fix the import path or add the missing module dependency
The package github.com/osmosis-labs/sqs/pkg/api/v1beta1 appears to be missing from the module dependencies.
6-15: LGTM! Well-structured initialization with proper nil handling
The function is well-implemented with:
- Proper nil check for pagination request
- Clean use of generics
- Clear initialization pattern
17-21: LGTM! Clean struct definition with good documentation
The struct is well-designed with appropriate use of generics and clear field definitions.
pkg/api/v1beta1/pagination.go (3)
66-71: Set a default limit when none is provided
If the limit parameter is not specified, r.Limit remains zero, causing Validate to return ErrLimitNotValid. Consider setting a default limit to improve user experience.
The previous review comment regarding setting a default limit when none is provided is still applicable.
81-87: Handle conflicting pagination parameters
Currently, if both cursorParam and pageParam are provided, the PaginationStrategy defaults to CURSOR. This may lead to unexpected behavior if the client unintentionally provides both.
Consider adding validation to ensure only one pagination strategy is used at a time.
if cursorParam != "" && pageParam != "" {
return fmt.Errorf("cannot use both cursor and page pagination simultaneously")
}Alternatively, prioritize one strategy over the other but document this behavior clearly.
132-134:
Avoid potential overflow when converting uint64 to int64
Converting uint64 to int64 may result in negative values if the value exceeds math.MaxInt64. This could cause unexpected behavior when nextCursor becomes negative unintentionally.
Consider changing the return type of CalculateNextCursor to uint64 and using a pointer or an additional boolean to indicate the absence of the next cursor:
-func (r *PaginationRequest) CalculateNextCursor(totalItems uint64) (nextCursor int64) {
+func (r *PaginationRequest) CalculateNextCursor(totalItems uint64) (nextCursor uint64, hasNext bool) {
// Existing checks...
- nextCursor = int64(r.Cursor + r.Limit)
+ nextCursor = r.Cursor + r.Limit
+ hasNext = true
return nextCursor
}Update the calling code to handle the new return values accordingly.
Alternatively, ensure that r.Cursor + r.Limit does not exceed math.MaxInt64 before conversion:
if r.Cursor + r.Limit > math.MaxInt64 {
return -1 // value too large for int64
}
nextCursor = int64(r.Cursor + r.Limit)Likely invalid or redundant comment.
domain/pipeline/iterator.go (2)
9-14: Consider adding Length() method to support pagination
To facilitate effective pagination, clients might need to know the total number of items. Consider adding a Length() int method to the Iterator interface.
Here's how you might modify the interface:
type Iterator[K, V any] interface {
Next() (V, error) // Retrieves the next element and an error if any
HasNext() bool // Checks if there are more elements
SetOffset(offset int) // Sets the offset for starting point of iteration
Reset() // Resets the iterator to the start
+ Length() int // Returns total number of items
}And implement it in SyncMapIterator:
type SyncMapIterator[K, V any] struct {
data *sync.Map
keys []K
index int
}
+// Length returns the total number of items in the iterator
+func (it *SyncMapIterator[K, V]) Length() int {
+ return len(it.keys)
+}17-23: 🛠️ Refactor suggestion
Add input validation in NewSyncMapIterator
The constructor should validate that neither the data map nor the keys slice is nil to prevent potential panics during iteration.
Here is how you might add the input validation:
func NewSyncMapIterator[K, V any](data *sync.Map, keys []K) *SyncMapIterator[K, V] {
+ if data == nil {
+ panic("data map cannot be nil")
+ }
+ if keys == nil {
+ panic("keys slice cannot be nil")
+ }
return &SyncMapIterator[K, V]{
data: data,
keys: keys,
index: 0,
}
}
|
This PR represents a smaller, more focused part of bigger PR #553 implementing sorting, filtering, pagination and search functionality for
/poolsendpoint. For bigger implementation picture refer to mentioned PR and for more context please refer to BE-636.PR introduces pipeline package allowing to manipulate data in various ways including filtering, sorting, paginating and iterating. Such functionality originally is utilized in
/poolsendpoint used here.Additionally this PR as dependency introduces some
protofiles, specifically for generatingPaginationRequestused bypipeline.Paginator.Summary by CodeRabbit
New Features
Iteratorinterface for structured data iteration.Paginatorfor flexible paginated data retrieval.Transformerframework for data filtering, sorting, and key retrieval.Improvements
SyncMapIteratorandSyncMapTransformer.