Storage: Add ascending order support for NotOlderThan queries and introduce ResourceVersionMatch_Unset as default (#102505)
* Add support for ASC ordering and introduce ResourceVersionMatch_Unset as default Add SortAscending to continue token and add integration test for pagination. * Change protobuf order * Make backwards compatible * Update pkg/storage/unified/sql/backend.go Co-authored-by: Jean-Philippe Quéméner <JohnnyQQQQ@users.noreply.github.com> --------- Co-authored-by: Marco de Abreu <18629099+marcoabreu@users.noreply.github.com> Co-authored-by: Jean-Philippe Quéméner <JohnnyQQQQ@users.noreply.github.com>
This commit is contained in:
Marco de Abreu
co-authored by
Jean-Philippe Quéméner
Marco de Abreu
Jean-Philippe Quéméner
Marco de Abreu
parent
7d435e5d84
commit
6d570db312
@@ -544,9 +544,10 @@ func (b *backend) ListIterator(ctx context.Context, req *resource.ListRequest, c
|
||||
}
|
||||
|
||||
type listIter struct {
|
||||
rows db.Rows
|
||||
offset int64
|
||||
listRV int64
|
||||
rows db.Rows
|
||||
offset int64
|
||||
listRV int64
|
||||
sortAsc bool
|
||||
|
||||
// any error
|
||||
err error
|
||||
@@ -561,11 +562,11 @@ type listIter struct {
|
||||
|
||||
// ContinueToken implements resource.ListIterator.
|
||||
func (l *listIter) ContinueToken() string {
|
||||
return resource.ContinueToken{ResourceVersion: l.listRV, StartOffset: l.offset}.String()
|
||||
return resource.ContinueToken{ResourceVersion: l.listRV, StartOffset: l.offset, SortAscending: l.sortAsc}.String()
|
||||
}
|
||||
|
||||
func (l *listIter) ContinueTokenWithCurrentRV() string {
|
||||
return resource.ContinueToken{ResourceVersion: l.rv, StartOffset: l.offset}.String()
|
||||
return resource.ContinueToken{ResourceVersion: l.rv, StartOffset: l.offset, SortAscending: l.sortAsc}.String()
|
||||
}
|
||||
|
||||
func (l *listIter) Error() error {
|
||||
@@ -615,7 +616,7 @@ func (b *backend) listLatest(ctx context.Context, req *resource.ListRequest, cb
|
||||
return 0, fmt.Errorf("only works for the 'latest' resource version")
|
||||
}
|
||||
|
||||
iter := &listIter{}
|
||||
iter := &listIter{sortAsc: false}
|
||||
err := b.db.WithTx(ctx, ReadCommittedRO, func(ctx context.Context, tx db.Tx) error {
|
||||
var err error
|
||||
iter.listRV, err = fetchLatestRV(ctx, tx, b.dialect, req.Options.Key.Group, req.Options.Key.Resource)
|
||||
@@ -650,7 +651,7 @@ func (b *backend) listLatest(ctx context.Context, req *resource.ListRequest, cb
|
||||
// listAtRevision fetches the resources from the resource_history table at a specific revision.
|
||||
func (b *backend) listAtRevision(ctx context.Context, req *resource.ListRequest, cb func(resource.ListIterator) error) (int64, error) {
|
||||
// Get the RV
|
||||
iter := &listIter{listRV: req.ResourceVersion}
|
||||
iter := &listIter{listRV: req.ResourceVersion, sortAsc: false}
|
||||
if req.NextPageToken != "" {
|
||||
continueToken, err := resource.GetContinueToken(req.NextPageToken)
|
||||
if err != nil {
|
||||
@@ -700,14 +701,37 @@ func (b *backend) listAtRevision(ctx context.Context, req *resource.ListRequest,
|
||||
return iter.listRV, err
|
||||
}
|
||||
|
||||
// listLatest fetches the resources from the resource table.
|
||||
// getHistory fetches the resources from the resource table.
|
||||
func (b *backend) getHistory(ctx context.Context, req *resource.ListRequest, cb func(resource.ListIterator) error) (int64, error) {
|
||||
// Backwards compatibility for ResourceVersionMatch
|
||||
if req.VersionMatch != nil && req.GetVersionMatchV2() == resource.ResourceVersionMatchV2_UNKNOWN {
|
||||
switch req.GetVersionMatch() {
|
||||
case resource.ResourceVersionMatch_DEPRECATED_NotOlderThan:
|
||||
// This is not a typo. The old implementation actually did behave like Unset.
|
||||
req.VersionMatchV2 = resource.ResourceVersionMatchV2_Unset
|
||||
case resource.ResourceVersionMatch_DEPRECATED_Exact:
|
||||
req.VersionMatchV2 = resource.ResourceVersionMatchV2_Exact
|
||||
default:
|
||||
return 0, fmt.Errorf("unknown version match: %v", req.GetVersionMatch())
|
||||
}
|
||||
|
||||
// Log the migration for debugging purposes
|
||||
b.log.Debug("Old client request received, migrating from version_match to version_match_v2",
|
||||
"oldValue", req.GetVersionMatch(),
|
||||
"newValue", req.GetVersionMatchV2())
|
||||
}
|
||||
|
||||
listReq := sqlGetHistoryRequest{
|
||||
SQLTemplate: sqltemplate.New(b.dialect),
|
||||
Key: req.Options.Key,
|
||||
Trash: req.Source == resource.ListRequest_TRASH,
|
||||
}
|
||||
|
||||
// We are assuming that users want history in ascending order
|
||||
// when they are using NotOlderThan matching, and descending order
|
||||
// for Unset (default) and Exact matching.
|
||||
listReq.SortAscending = req.GetVersionMatchV2() == resource.ResourceVersionMatchV2_NotOlderThan
|
||||
|
||||
iter := &listIter{}
|
||||
if req.NextPageToken != "" {
|
||||
continueToken, err := resource.GetContinueToken(req.NextPageToken)
|
||||
@@ -715,10 +739,12 @@ func (b *backend) getHistory(ctx context.Context, req *resource.ListRequest, cb
|
||||
return 0, fmt.Errorf("get continue token: %w", err)
|
||||
}
|
||||
listReq.StartRV = continueToken.ResourceVersion
|
||||
listReq.SortAscending = continueToken.SortAscending
|
||||
}
|
||||
iter.sortAsc = listReq.SortAscending
|
||||
|
||||
// Set ExactRV when using Exact matching
|
||||
if req.VersionMatch == resource.ResourceVersionMatch_Exact {
|
||||
if req.VersionMatchV2 == resource.ResourceVersionMatchV2_Exact {
|
||||
if req.ResourceVersion <= 0 {
|
||||
return 0, fmt.Errorf("expecting an explicit resource version query when using Exact matching")
|
||||
}
|
||||
@@ -726,7 +752,7 @@ func (b *backend) getHistory(ctx context.Context, req *resource.ListRequest, cb
|
||||
}
|
||||
|
||||
// Set MinRV when using NotOlderThan matching to filter at the database level
|
||||
if req.ResourceVersion > 0 && req.VersionMatch == resource.ResourceVersionMatch_NotOlderThan {
|
||||
if req.ResourceVersion > 0 && req.VersionMatchV2 == resource.ResourceVersionMatchV2_NotOlderThan {
|
||||
listReq.MinRV = req.ResourceVersion
|
||||
}
|
||||
|
||||
|
||||
@@ -391,7 +391,7 @@ func TestBackend_getHistory(t *testing.T) {
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
versionMatch resource.ResourceVersionMatch
|
||||
versionMatch resource.ResourceVersionMatchV2
|
||||
resourceVersion int64
|
||||
expectedVersions []int64
|
||||
expectedListRv int64
|
||||
@@ -400,30 +400,37 @@ func TestBackend_getHistory(t *testing.T) {
|
||||
}{
|
||||
{
|
||||
name: "with ResourceVersionMatch_NotOlderThan",
|
||||
versionMatch: resource.ResourceVersionMatch_NotOlderThan,
|
||||
versionMatch: resource.ResourceVersionMatchV2_NotOlderThan,
|
||||
resourceVersion: rv2,
|
||||
expectedVersions: []int64{rv3, rv2},
|
||||
expectedVersions: []int64{rv2, rv3}, // Should be in ASC order due to NotOlderThan
|
||||
expectedListRv: rv3,
|
||||
expectedRowsCount: 2,
|
||||
},
|
||||
{
|
||||
name: "with ResourceVersionMatch_NotOlderThan and ResourceVersion=0",
|
||||
versionMatch: resource.ResourceVersionMatchV2_NotOlderThan,
|
||||
resourceVersion: 0,
|
||||
expectedVersions: []int64{rv1, rv2, rv3}, // Should be in ASC order due to NotOlderThan
|
||||
expectedListRv: rv3,
|
||||
expectedRowsCount: 3,
|
||||
},
|
||||
{
|
||||
name: "with ResourceVersionMatch_Exact",
|
||||
versionMatch: resource.ResourceVersionMatch_Exact,
|
||||
versionMatch: resource.ResourceVersionMatchV2_Exact,
|
||||
resourceVersion: rv2,
|
||||
expectedVersions: []int64{rv2},
|
||||
expectedListRv: rv3,
|
||||
expectedRowsCount: 1,
|
||||
},
|
||||
{
|
||||
name: "without version matcher",
|
||||
versionMatch: resource.ResourceVersionMatch_NotOlderThan,
|
||||
expectedVersions: []int64{rv3, rv2, rv1},
|
||||
name: "with ResourceVersionMatch_Unset (default)",
|
||||
expectedVersions: []int64{rv3, rv2, rv1}, // Should be in DESC order by default
|
||||
expectedListRv: rv3,
|
||||
expectedRowsCount: 3,
|
||||
},
|
||||
{
|
||||
name: "error with ResourceVersionMatch_Exact and ResourceVersion <= 0",
|
||||
versionMatch: resource.ResourceVersionMatch_Exact,
|
||||
versionMatch: resource.ResourceVersionMatchV2_Exact,
|
||||
resourceVersion: 0,
|
||||
expectedErr: "expecting an explicit resource version query when using Exact matching",
|
||||
},
|
||||
@@ -439,7 +446,7 @@ func TestBackend_getHistory(t *testing.T) {
|
||||
req := &resource.ListRequest{
|
||||
Options: &resource.ListOptions{Key: key},
|
||||
ResourceVersion: tc.resourceVersion,
|
||||
VersionMatch: tc.versionMatch,
|
||||
VersionMatchV2: tc.versionMatch,
|
||||
Source: resource.ListRequest_HISTORY,
|
||||
}
|
||||
|
||||
@@ -454,14 +461,18 @@ func TestBackend_getHistory(t *testing.T) {
|
||||
// Callback that tracks returned items
|
||||
callback := func(iter resource.ListIterator) error {
|
||||
count := 0
|
||||
var seenVersions []int64
|
||||
for iter.Next() {
|
||||
count++
|
||||
currentRV := iter.ResourceVersion()
|
||||
seenVersions = append(seenVersions, currentRV)
|
||||
expectedValue, ok := expectedValues[currentRV]
|
||||
require.True(t, ok, "Got unexpected RV: %d", currentRV)
|
||||
require.Equal(t, expectedValue, string(iter.Value()))
|
||||
}
|
||||
require.Equal(t, tc.expectedRowsCount, count)
|
||||
// Verify the order matches what we expect
|
||||
require.Equal(t, tc.expectedVersions, seenVersions, "Resource versions returned in incorrect order")
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -503,3 +514,153 @@ func TestBackend_getHistory(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestBackend_getHistoryPagination tests the ordering behavior for ResourceVersionMatch_NotOlderThan
|
||||
// when using pagination, ensuring entries are returned in oldest-to-newest order.
|
||||
func TestBackend_getHistoryPagination(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// Common setup
|
||||
key := &resource.ResourceKey{
|
||||
Namespace: "ns",
|
||||
Group: "gr",
|
||||
Resource: "rs",
|
||||
Name: "nm",
|
||||
}
|
||||
|
||||
// Create resource versions that will be returned in our test
|
||||
versions := make([]int64, 10)
|
||||
for i := range versions {
|
||||
versions[i] = int64(51 + i)
|
||||
}
|
||||
rv51, rv52, rv53, rv54, rv55, rv56, rv57, rv58, rv59, rv60 := versions[0], versions[1], versions[2], versions[3], versions[4], versions[5], versions[6], versions[7], versions[8], versions[9]
|
||||
|
||||
t.Run("pagination with NotOlderThan should return entries from oldest to newest", func(t *testing.T) {
|
||||
b, ctx := setupBackendTest(t)
|
||||
|
||||
// Define all pages we want to test
|
||||
pages := []struct {
|
||||
versions []int64
|
||||
token *resource.ContinueToken
|
||||
}{
|
||||
{
|
||||
versions: []int64{rv51, rv52, rv53, rv54},
|
||||
token: nil,
|
||||
},
|
||||
{
|
||||
versions: []int64{rv55, rv56, rv57, rv58},
|
||||
token: &resource.ContinueToken{
|
||||
ResourceVersion: rv54,
|
||||
StartOffset: 4,
|
||||
SortAscending: true,
|
||||
},
|
||||
},
|
||||
{
|
||||
versions: []int64{rv59, rv60},
|
||||
token: &resource.ContinueToken{
|
||||
ResourceVersion: rv58,
|
||||
StartOffset: 8,
|
||||
SortAscending: true,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
var allItems []int64
|
||||
initialRV := rv51
|
||||
|
||||
// Test each page
|
||||
for _, page := range pages {
|
||||
req := &resource.ListRequest{
|
||||
Options: &resource.ListOptions{Key: key},
|
||||
ResourceVersion: initialRV,
|
||||
VersionMatchV2: resource.ResourceVersionMatchV2_NotOlderThan,
|
||||
Source: resource.ListRequest_HISTORY,
|
||||
Limit: 4,
|
||||
}
|
||||
if page.token != nil {
|
||||
req.NextPageToken = page.token.String()
|
||||
}
|
||||
|
||||
items := make([]int64, 0)
|
||||
callback := func(iter resource.ListIterator) error {
|
||||
for iter.Next() {
|
||||
items = append(items, iter.ResourceVersion())
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
b.SQLMock.ExpectBegin()
|
||||
historyRows := setupHistoryTest(b, page.versions, rv60)
|
||||
b.SQLMock.ExpectQuery("SELECT .* FROM resource_history").WillReturnRows(historyRows)
|
||||
b.SQLMock.ExpectCommit()
|
||||
|
||||
listRv, err := b.getHistory(ctx, req, callback)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, rv60, listRv, "Head version should be the latest resource version (rv60)")
|
||||
require.Equal(t, page.versions, items, "Items should be in ASC order")
|
||||
|
||||
allItems = append(allItems, items...)
|
||||
}
|
||||
|
||||
// Verify complete sequence
|
||||
expectedAllItems := []int64{rv51, rv52, rv53, rv54, rv55, rv56, rv57, rv58, rv59, rv60}
|
||||
require.Equal(t, expectedAllItems, allItems)
|
||||
})
|
||||
|
||||
t.Run("pagination with ResourceVersion=0 and NotOlderThan should return entries in ASC order", func(t *testing.T) {
|
||||
b, ctx := setupBackendTest(t)
|
||||
|
||||
req := &resource.ListRequest{
|
||||
Options: &resource.ListOptions{Key: key},
|
||||
ResourceVersion: 0,
|
||||
VersionMatchV2: resource.ResourceVersionMatchV2_NotOlderThan,
|
||||
Source: resource.ListRequest_HISTORY,
|
||||
Limit: 4,
|
||||
}
|
||||
|
||||
// First batch of items we expect, in ASC order (because of NotOlderThan flag)
|
||||
// Even with ResourceVersion=0, the order is ASC because we use SortAscending=true
|
||||
expectedVersions := []int64{rv51, rv52, rv53, rv54}
|
||||
items := make([]int64, 0)
|
||||
|
||||
callback := func(iter resource.ListIterator) error {
|
||||
for iter.Next() {
|
||||
items = append(items, iter.ResourceVersion())
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
b.SQLMock.ExpectBegin()
|
||||
historyRows := setupHistoryTest(b, expectedVersions, rv60)
|
||||
b.SQLMock.ExpectQuery("SELECT .* FROM resource_history").WillReturnRows(historyRows)
|
||||
b.SQLMock.ExpectCommit()
|
||||
|
||||
listRv, err := b.getHistory(ctx, req, callback)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, rv60, listRv, "Head version should be the latest resource version (rv60)")
|
||||
require.Equal(t, expectedVersions, items, "Items should be in ASC order even with ResourceVersion=0")
|
||||
})
|
||||
}
|
||||
|
||||
// setupHistoryTest creates the necessary mock expectations for a history test
|
||||
func setupHistoryTest(b testBackend, resourceVersions []int64, latestRV int64) *sqlmock.Rows {
|
||||
// Expect fetch latest RV call - set to the highest resource version
|
||||
latestRVRows := sqlmock.NewRows([]string{"resource_version", "unix_timestamp"}).
|
||||
AddRow(latestRV, 0)
|
||||
b.SQLMock.ExpectQuery("SELECT .* FROM resource_version").WillReturnRows(latestRVRows)
|
||||
|
||||
// Create the mock rows for the history items
|
||||
cols := []string{"resource_version", "namespace", "name", "folder", "value"}
|
||||
historyRows := sqlmock.NewRows(cols)
|
||||
for _, rv := range resourceVersions {
|
||||
historyRows.AddRow(
|
||||
rv, // resource_version
|
||||
"ns", // namespace
|
||||
"nm", // name
|
||||
"folder", // folder
|
||||
[]byte(fmt.Sprintf("rv-%d", rv)), // value
|
||||
)
|
||||
}
|
||||
|
||||
return historyRows
|
||||
}
|
||||
|
||||
@@ -16,12 +16,20 @@ WHERE 1 = 1
|
||||
AND {{ .Ident "action" }} = 3
|
||||
{{ end }}
|
||||
{{ if (gt .StartRV 0) }}
|
||||
{{ if .SortAscending }}
|
||||
AND {{ .Ident "resource_version" }} > {{ .Arg .StartRV }}
|
||||
{{ else }}
|
||||
AND {{ .Ident "resource_version" }} < {{ .Arg .StartRV }}
|
||||
{{ end }}
|
||||
{{ end }}
|
||||
{{ if (gt .MinRV 0) }}
|
||||
AND {{ .Ident "resource_version" }} >= {{ .Arg .MinRV }}
|
||||
{{ end }}
|
||||
{{ if (gt .ExactRV 0) }}
|
||||
AND {{ .Ident "resource_version" }} = {{ .Arg .ExactRV }}
|
||||
{{ end }}
|
||||
{{ if .SortAscending }}
|
||||
ORDER BY resource_version ASC
|
||||
{{ else }}
|
||||
ORDER BY resource_version DESC
|
||||
{{ end }}
|
||||
|
||||
@@ -243,11 +243,12 @@ func (r *sqlResourceHistoryDeleteRequest) Validate() error {
|
||||
|
||||
type sqlGetHistoryRequest struct {
|
||||
sqltemplate.SQLTemplate
|
||||
Key *resource.ResourceKey
|
||||
Trash bool // only deleted items
|
||||
StartRV int64 // from NextPageToken
|
||||
MinRV int64 // minimum resource version for NotOlderThan
|
||||
ExactRV int64 // exact resource version for Exact
|
||||
Key *resource.ResourceKey
|
||||
Trash bool // only deleted items
|
||||
StartRV int64 // from NextPageToken
|
||||
MinRV int64 // minimum resource version for NotOlderThan
|
||||
ExactRV int64 // exact resource version for Exact
|
||||
SortAscending bool // if true, sort by resource_version ASC, otherwise DESC
|
||||
}
|
||||
|
||||
func (r sqlGetHistoryRequest) Validate() error {
|
||||
|
||||
Reference in New Issue
Block a user