Storage: Add resource version matching in unified storage API (#102417)

Add NotOlderThan support to getHistory

Add support for Exact

Add tests

Refactor tests

Add error test

Co-authored-by: Marco de Abreu <18629099+marcoabreu@users.noreply.github.com>

pkg/storage/unified/apistore/util.go

@@ -31,6 +31,25 @@ func toListRequest(k *resource.ResourceKey, opts storage.ListOptions) (*resource
 		NextPageToken: predicate.Continue,
 	}
 
+	if opts.ResourceVersion != "" {
+		rv, err := strconv.ParseInt(opts.ResourceVersion, 10, 64)
+		if err != nil {
+			return nil, predicate, apierrors.NewBadRequest(fmt.Sprintf("invalid resource version: %s", opts.ResourceVersion))
+		}
+		req.ResourceVersion = rv
+	}
+
+	switch opts.ResourceVersionMatch {
+	case "", metav1.ResourceVersionMatchNotOlderThan:
+		req.VersionMatch = resource.ResourceVersionMatch_NotOlderThan
+	case metav1.ResourceVersionMatchExact:
+		req.VersionMatch = resource.ResourceVersionMatch_Exact
+	default:
+		return nil, predicate, apierrors.NewBadRequest(
+			fmt.Sprintf("unsupported version match: %v", opts.ResourceVersionMatch),
+		)
+	}
+
 	if opts.Predicate.Label != nil && !opts.Predicate.Label.Empty() {
 		requirements, selectable := opts.Predicate.Label.Requirements()
 		if !selectable {
@@ -91,25 +110,6 @@ func toListRequest(k *resource.ResourceKey, opts storage.ListOptions) (*resource
 		}
 	}
 
-	if opts.ResourceVersion != "" {
-		rv, err := strconv.ParseInt(opts.ResourceVersion, 10, 64)
-		if err != nil {
-			return nil, predicate, apierrors.NewBadRequest(fmt.Sprintf("invalid resource version: %s", opts.ResourceVersion))
-		}
-		req.ResourceVersion = rv
-	}
-
-	switch opts.ResourceVersionMatch {
-	case "", metav1.ResourceVersionMatchNotOlderThan:
-		req.VersionMatch = resource.ResourceVersionMatch_NotOlderThan
-	case metav1.ResourceVersionMatchExact:
-		req.VersionMatch = resource.ResourceVersionMatch_Exact
-	default:
-		return nil, predicate, apierrors.NewBadRequest(
-			fmt.Sprintf("unsupported version match: %v", opts.ResourceVersionMatch),
-		)
-	}
-
 	return req, predicate, nil
 }
 

pkg/storage/unified/sql/backend.go

@@ -798,6 +798,19 @@ func (b *backend) getHistory(ctx context.Context, req *resource.ListRequest, cb
 		listReq.StartRV = continueToken.ResourceVersion
 	}
 
+	// Set ExactRV when using Exact matching
+	if req.VersionMatch == resource.ResourceVersionMatch_Exact {
+		if req.ResourceVersion <= 0 {
+			return 0, fmt.Errorf("expecting an explicit resource version query when using Exact matching")
+		}
+		listReq.ExactRV = req.ResourceVersion
+	}
+
+	// Set MinRV when using NotOlderThan matching to filter at the database level
+	if req.ResourceVersion > 0 && req.VersionMatch == resource.ResourceVersionMatch_NotOlderThan {
+		listReq.MinRV = req.ResourceVersion
+	}
+
 	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)

pkg/storage/unified/sql/backend_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql/driver"
 	"errors"
+	"fmt"
 	"testing"
 
 	sqlmock "github.com/DATA-DOG/go-sqlmock"
@@ -456,3 +457,131 @@ func TestBackend_restore(t *testing.T) {
 		require.ErrorContains(t, err, "update history uid")
 	})
 }
+
+func TestBackend_getHistory(t *testing.T) {
+	t.Parallel()
+
+	// Common setup
+	key := &resource.ResourceKey{
+		Namespace: "ns",
+		Group:     "gr",
+		Resource:  "rs",
+		Name:      "nm",
+	}
+	rv1, rv2, rv3 := int64(100), int64(200), int64(300)
+	cols := []string{"resource_version", "namespace", "name", "folder", "value"}
+
+	tests := []struct {
+		name              string
+		versionMatch      resource.ResourceVersionMatch
+		resourceVersion   int64
+		expectedVersions  []int64
+		expectedListRv    int64
+		expectedRowsCount int
+		expectedErr       string
+	}{
+		{
+			name:              "with ResourceVersionMatch_NotOlderThan",
+			versionMatch:      resource.ResourceVersionMatch_NotOlderThan,
+			resourceVersion:   rv2,
+			expectedVersions:  []int64{rv3, rv2},
+			expectedListRv:    rv3,
+			expectedRowsCount: 2,
+		},
+		{
+			name:              "with ResourceVersionMatch_Exact",
+			versionMatch:      resource.ResourceVersionMatch_Exact,
+			resourceVersion:   rv2,
+			expectedVersions:  []int64{rv2},
+			expectedListRv:    rv3,
+			expectedRowsCount: 1,
+		},
+		{
+			name:              "without version matcher",
+			versionMatch:      resource.ResourceVersionMatch_NotOlderThan,
+			expectedVersions:  []int64{rv3, rv2, rv1},
+			expectedListRv:    rv3,
+			expectedRowsCount: 3,
+		},
+		{
+			name:            "error with ResourceVersionMatch_Exact and ResourceVersion <= 0",
+			versionMatch:    resource.ResourceVersionMatch_Exact,
+			resourceVersion: 0,
+			expectedErr:     "expecting an explicit resource version query when using Exact matching",
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			b, ctx := setupBackendTest(t)
+
+			// Build request with appropriate matcher
+			req := &resource.ListRequest{
+				Options:         &resource.ListOptions{Key: key},
+				ResourceVersion: tc.resourceVersion,
+				VersionMatch:    tc.versionMatch,
+				Source:          resource.ListRequest_HISTORY,
+			}
+
+			// Set up mock expectations only if we don't expect an error
+			if tc.expectedErr == "" {
+				// Build expected values map
+				expectedValues := make(map[int64]string)
+				for _, rv := range tc.expectedVersions {
+					expectedValues[rv] = fmt.Sprintf("rv-%d", rv)
+				}
+
+				// Callback that tracks returned items
+				callback := func(iter resource.ListIterator) error {
+					count := 0
+					for iter.Next() {
+						count++
+						currentRV := iter.ResourceVersion()
+						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)
+					return nil
+				}
+
+				b.SQLMock.ExpectBegin()
+
+				// Expect fetch latest RV call
+				latestRVRows := sqlmock.NewRows([]string{"resource_version", "unix_timestamp"}).
+					AddRow(rv3, 0)
+				b.SQLMock.ExpectQuery("SELECT .* FROM resource_version").WillReturnRows(latestRVRows)
+
+				// Expect history query
+				historyRows := sqlmock.NewRows(cols)
+				for _, rv := range tc.expectedVersions {
+					historyRows.AddRow(
+						rv,                               // resource_version
+						"ns",                             // namespace
+						"nm",                             // name
+						"folder",                         // folder
+						[]byte(fmt.Sprintf("rv-%d", rv)), // value
+					)
+				}
+				b.SQLMock.ExpectQuery("SELECT .* FROM resource_history").WillReturnRows(historyRows)
+				b.SQLMock.ExpectCommit()
+
+				// Execute the test
+				listRv, err := b.getHistory(ctx, req, callback)
+				require.NoError(t, err)
+				require.Equal(t, tc.expectedListRv, listRv)
+			} else {
+				// For error cases, we use a simple empty callback
+				callback := func(iter resource.ListIterator) error { return nil }
+
+				// Execute the test expecting an error
+				listRv, err := b.getHistory(ctx, req, callback)
+				require.Zero(t, listRv)
+				require.Error(t, err)
+				require.ErrorContains(t, err, tc.expectedErr)
+			}
+		})
+	}
+}

pkg/storage/unified/sql/data/resource_history_get.sql

@@ -18,4 +18,10 @@ WHERE 1 = 1
   {{ if (gt .StartRV 0) }}
   AND {{ .Ident "resource_version" }} < {{ .Arg .StartRV }}
   {{ end }}
+  {{ if (gt .MinRV 0) }}
+  AND {{ .Ident "resource_version" }} >= {{ .Arg .MinRV }}
+  {{ end }}
+  {{ if (gt .ExactRV 0) }}
+  AND {{ .Ident "resource_version" }} = {{ .Arg .ExactRV }}
+  {{ end }}
 ORDER BY resource_version DESC

pkg/storage/unified/sql/queries.go

@@ -247,6 +247,8 @@ type sqlGetHistoryRequest struct {
 	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
 }
 
 func (r sqlGetHistoryRequest) Validate() error {