unified-storage: move rvmanager into its own package (#115445)

* unified-storage: move rvmanager into its own package so it can be reused with sqlkv later

pkg/storage/unified/sql/rvmanager/data/resource_history_update_rv.sql

@@ -0,0 +1,31 @@
+UPDATE {{ .Ident "resource_history" }}
+SET {{ .Ident "resource_version" }} = (
+    CASE
+    {{ range $guid, $rv := .GUIDToRV }}
+    WHEN {{ $.Ident "guid" }} = {{ $.Arg $guid }} THEN CAST({{ $.Arg $rv }} AS {{ if eq $.DialectName "postgres" }}BIGINT{{ else }}SIGNED{{ end }})
+    {{ end }}
+    END
+), {{ .Ident "key_path" }} = (
+    CASE
+    {{ range $guid, $snowflakeRv := .GUIDToSnowflakeRV }}
+    WHEN {{ $.Ident "guid" }} = {{ $.Arg $guid }} THEN CONCAT(
+      'unified', {{ $.SlashFunc }}, 'data', {{ $.SlashFunc }},
+      {{ $.Ident "group" }}, {{ $.SlashFunc }},
+      {{ $.Ident "resource" }}, {{ $.SlashFunc }},
+      {{ $.Ident "namespace" }}, {{ $.SlashFunc }},
+      {{ $.Ident "name" }}, {{ $.SlashFunc }},
+      CAST({{ $.Arg $snowflakeRv }} AS {{ if eq $.DialectName "postgres" }}BIGINT{{ else }}SIGNED{{ end }}),
+      {{ $.TildeFunc }},
+      CASE {{ $.Ident "action" }}
+        WHEN 1 THEN 'created'
+        WHEN 2 THEN 'updated'
+        WHEN 3 THEN 'deleted'
+      END, {{ $.TildeFunc }},
+      COALESCE({{ $.Ident "folder" }}, ''))
+    {{ end }}
+    END
+)
+WHERE {{ .Ident "guid" }} IN (
+    {{$first := true}}
+    {{ range $guid, $rv := .GUIDToRV }}{{if $first}}{{$first = false}}{{else}}, {{end}}{{ $.Arg $guid }}{{ end }}
+);

pkg/storage/unified/sql/rvmanager/data/resource_update_rv.sql

@@ -0,0 +1,12 @@
+UPDATE {{ .Ident "resource" }}
+SET {{ .Ident "resource_version" }} = (
+    CASE
+    {{ range $guid, $rv := .GUIDToRV }}
+    WHEN {{ $.Ident "guid" }} = {{ $.Arg $guid }} THEN CAST({{ $.Arg $rv }} AS {{ if eq $.DialectName "postgres" }}BIGINT{{ else }}SIGNED{{ end }})
+    {{ end }}
+    END
+)
+WHERE {{ .Ident "guid" }} IN (
+    {{$first := true}}
+    {{ range $guid, $rv := .GUIDToRV }}{{if $first}}{{$first = false}}{{else}}, {{end}}{{ $.Arg $guid }}{{ end }}
+);

pkg/storage/unified/sql/rvmanager/data/resource_version_get.sql

@@ -0,0 +1,11 @@
+SELECT
+        {{ .Ident "resource_version" | .Into .Response.ResourceVersion }},
+        {{ .CurrentEpoch | .Into .Response.CurrentEpoch }}
+    FROM {{ .Ident "resource_version" }}
+    WHERE 1 = 1
+        AND {{ .Ident "group" }}    = {{ .Arg .Group }}
+        AND {{ .Ident "resource" }} = {{ .Arg .Resource }}
+    {{ if not .ReadOnly }}
+    {{ .SelectFor "UPDATE" }}
+    {{ end}}
+;

pkg/storage/unified/sql/rvmanager/data/resource_version_insert.sql

@@ -0,0 +1,13 @@
+INSERT INTO {{ .Ident "resource_version" }}
+    (
+        {{ .Ident "group" }},
+        {{ .Ident "resource" }},
+        {{ .Ident "resource_version" }}
+    )
+
+    VALUES (
+        {{ .Arg .Group }},
+        {{ .Arg .Resource }},
+        {{ .CurrentEpoch }}
+    )
+;

pkg/storage/unified/sql/rvmanager/data/resource_version_update.sql

@@ -0,0 +1,7 @@
+UPDATE {{ .Ident "resource_version" }}
+SET
+    {{ .Ident "resource_version" }} = {{ .Arg .ResourceVersion }}
+WHERE 1 = 1
+    AND {{ .Ident "group" }}    = {{ .Arg .Group }}
+    AND {{ .Ident "resource" }} = {{ .Arg .Resource }}
+;

pkg/storage/unified/sql/rvmanager/queries.go

@@ -0,0 +1,84 @@
+package rvmanager
+
+import (
+	"github.com/grafana/grafana/pkg/storage/unified/sql/sqltemplate"
+)
+
+type SqlResourceUpdateRVRequest struct {
+	sqltemplate.SQLTemplate
+	GUIDToRV          map[string]int64
+	GUIDToSnowflakeRV map[string]int64
+}
+
+func (r SqlResourceUpdateRVRequest) Validate() error {
+	return nil // TODO
+}
+
+func (r SqlResourceUpdateRVRequest) SlashFunc() string {
+	if r.DialectName() == "postgres" {
+		return "CHR(47)"
+	}
+
+	return "CHAR(47)"
+}
+
+func (r SqlResourceUpdateRVRequest) TildeFunc() string {
+	if r.DialectName() == "postgres" {
+		return "CHR(126)"
+	}
+
+	return "CHAR(126)"
+}
+
+type ResourceVersionResponse struct {
+	ResourceVersion int64
+	CurrentEpoch    int64
+}
+
+func (r *ResourceVersionResponse) Results() (*ResourceVersionResponse, error) {
+	return r, nil
+}
+
+type sqlResourceVersionGetRequest struct {
+	sqltemplate.SQLTemplate
+	Group, Resource string
+	ReadOnly        bool
+	Response        *ResourceVersionResponse
+}
+
+func (r sqlResourceVersionGetRequest) Validate() error {
+	return nil // TODO
+}
+func (r sqlResourceVersionGetRequest) Results() (*ResourceVersionResponse, error) {
+	return &ResourceVersionResponse{
+		ResourceVersion: r.Response.ResourceVersion,
+		CurrentEpoch:    r.Response.CurrentEpoch,
+	}, nil
+}
+
+type SqlResourceVersionUpsertRequest struct {
+	sqltemplate.SQLTemplate
+	Group, Resource string
+	ResourceVersion int64
+}
+
+func (r SqlResourceVersionUpsertRequest) Validate() error {
+	return nil // TODO
+}
+
+type SqlResourceVersionGetRequest struct {
+	sqltemplate.SQLTemplate
+	Group, Resource string
+	ReadOnly        bool
+	Response        *ResourceVersionResponse
+}
+
+func (r SqlResourceVersionGetRequest) Validate() error {
+	return nil // TODO
+}
+func (r SqlResourceVersionGetRequest) Results() (*ResourceVersionResponse, error) {
+	return &ResourceVersionResponse{
+		ResourceVersion: r.Response.ResourceVersion,
+		CurrentEpoch:    r.Response.CurrentEpoch,
+	}, nil
+}

pkg/storage/unified/sql/rvmanager/rv_manager.go

@@ -0,0 +1,411 @@
+package rvmanager
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/bwmarrin/snowflake"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
+
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
+	"github.com/grafana/grafana/pkg/storage/unified/sql/db"
+	"github.com/grafana/grafana/pkg/storage/unified/sql/dbutil"
+	"github.com/grafana/grafana/pkg/storage/unified/sql/sqltemplate"
+)
+
+var tracer = otel.Tracer("github.com/grafana/grafana/pkg/storage/unified/sql/rvmanager")
+
+var (
+	rvmWriteDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:                        "rvmanager_write_duration_seconds",
+		Help:                        "Duration of ResourceVersionManager write operations",
+		Namespace:                   "grafana",
+		NativeHistogramBucketFactor: 1.1,
+	}, []string{"group", "resource", "status"})
+
+	rvmExecBatchDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:                        "rvmanager_exec_batch_duration_seconds",
+		Help:                        "Duration of ResourceVersionManager batch operations",
+		Namespace:                   "grafana",
+		NativeHistogramBucketFactor: 1.1,
+	}, []string{"group", "resource", "status"})
+
+	rvmExecBatchPhaseDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:                        "rvmanager_exec_batch_phase_duration_seconds",
+		Help:                        "Duration of batch operation phases",
+		Namespace:                   "grafana",
+		NativeHistogramBucketFactor: 1.1,
+	}, []string{"group", "resource", "phase"})
+
+	rvmInflightWrites = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name:      "rvmanager_inflight_writes",
+		Help:      "Number of concurrent write operations",
+		Namespace: "grafana",
+	}, []string{"group", "resource"})
+
+	rvmBatchSize = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:                        "rvmanager_batch_size",
+		Help:                        "Number of write operations per batch",
+		Namespace:                   "grafana",
+		NativeHistogramBucketFactor: 1.1,
+	}, []string{"group", "resource"})
+)
+
+const (
+	defaultMaxBatchSize     = 25
+	defaultMaxBatchWaitTime = 100 * time.Millisecond
+	defaultBatchTimeout     = 5 * time.Second
+)
+
+// ResourceVersionManager handles resource version operations
+type ResourceVersionManager struct {
+	dialect    sqltemplate.Dialect
+	db         db.DB
+	batchMu    sync.RWMutex
+	batchChMap map[string]chan *writeOp
+
+	maxBatchSize     int           // The maximum number of operations to batch together
+	maxBatchWaitTime time.Duration // The maximum time to wait for a batch to be ready
+}
+
+type writeOpResult struct {
+	guid           string
+	rv             int64
+	err            error
+	batchTraceLink trace.Link
+}
+
+// writeOp is a write operation that is executed with an incremented resource version
+type writeOp struct {
+	key  *resourcepb.ResourceKey // The key of the resource
+	fn   WriteEventFunc          // The function to execute to write the event
+	done chan writeOpResult      // A channel informing the operation is done
+}
+
+// WriteEventFunc is a function that writes a resource to the database
+// It returns the GUID of the created resource
+// The GUID is used to update the resource version for the resource in the same transaction.
+type WriteEventFunc func(tx db.Tx) (guid string, err error)
+
+type ResourceManagerOptions struct {
+	Dialect          sqltemplate.Dialect // The dialect to use for the database
+	DB               db.DB               // The database to use
+	MaxBatchSize     int                 // The maximum number of operations to batch together
+	MaxBatchWaitTime time.Duration       // The maximum time to wait for a batch to be ready
+}
+
+// NewResourceVersionManager creates a new ResourceVersionManager
+func NewResourceVersionManager(opts ResourceManagerOptions) (*ResourceVersionManager, error) {
+	if opts.MaxBatchSize == 0 {
+		opts.MaxBatchSize = defaultMaxBatchSize
+	}
+	if opts.MaxBatchWaitTime == 0 {
+		opts.MaxBatchWaitTime = defaultMaxBatchWaitTime
+	}
+	if opts.Dialect == nil {
+		return nil, errors.New("dialect is required")
+	}
+	if opts.DB == nil {
+		return nil, errors.New("db is required")
+	}
+	return &ResourceVersionManager{
+		dialect:          opts.Dialect,
+		db:               opts.DB,
+		batchChMap:       make(map[string]chan *writeOp),
+		maxBatchSize:     opts.MaxBatchSize,
+		maxBatchWaitTime: opts.MaxBatchWaitTime,
+	}, nil
+}
+
+// ExecWithRV executes the given function with an incremented resource version
+func (m *ResourceVersionManager) ExecWithRV(ctx context.Context, key *resourcepb.ResourceKey, fn WriteEventFunc) (rv int64, err error) {
+	rvmInflightWrites.WithLabelValues(key.Group, key.Resource).Inc()
+	defer rvmInflightWrites.WithLabelValues(key.Group, key.Resource).Dec()
+
+	var status string
+	timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
+		status = "success"
+		if err != nil {
+			status = "error"
+		}
+		rvmWriteDuration.WithLabelValues(key.Group, key.Resource, status).Observe(v)
+	}))
+	defer timer.ObserveDuration()
+
+	ctx, span := tracer.Start(ctx, "sql.resourceVersionManager.ExecWithRV")
+	defer span.End()
+
+	span.SetAttributes(
+		attribute.String("group", key.Group),
+		attribute.String("resource", key.Resource),
+	)
+	op := writeOp{key: key, fn: fn, done: make(chan writeOpResult, 1)}
+	batchKey := fmt.Sprintf("%s/%s", key.Group, key.Resource)
+
+	m.batchMu.Lock()
+	ch, ok := m.batchChMap[batchKey]
+	if !ok {
+		ch = make(chan *writeOp, m.maxBatchSize)
+		m.batchChMap[batchKey] = ch
+		go m.startBatchProcessor(key.Group, key.Resource)
+	}
+	m.batchMu.Unlock()
+	select {
+	case ch <- &op:
+	case <-ctx.Done():
+		return 0, ctx.Err()
+	}
+
+	select {
+	case res := <-op.done:
+		if res.err != nil {
+			span.RecordError(res.err)
+		}
+		span.SetAttributes(
+			attribute.String("guid", res.guid),
+			attribute.Int64("resource_version", res.rv),
+		)
+		span.AddLink(res.batchTraceLink)
+
+		return res.rv, res.err
+	case <-ctx.Done():
+		return 0, ctx.Err()
+	}
+}
+
+// startBatchProcessor is responsible for processing batches of write operations
+func (m *ResourceVersionManager) startBatchProcessor(group, resource string) {
+	ctx := context.TODO()
+	batchKey := fmt.Sprintf("%s/%s", group, resource)
+
+	m.batchMu.Lock()
+	ch, ok := m.batchChMap[batchKey]
+	if !ok {
+		m.batchMu.Unlock()
+		return
+	}
+	m.batchMu.Unlock()
+
+	for {
+		batch := make([]writeOp, 0, m.maxBatchSize)
+		// wait for a new writeOp
+		select {
+		case op := <-ch:
+			batch = append(batch, *op)
+		case <-ctx.Done():
+			return
+		}
+
+	prepare:
+		for len(batch) < m.maxBatchSize {
+			select {
+			case op := <-ch:
+				batch = append(batch, *op)
+			default:
+				break prepare
+			}
+		}
+
+		rvmBatchSize.WithLabelValues(group, resource).Observe(float64(len(batch)))
+		m.execBatch(ctx, group, resource, batch)
+	}
+}
+
+var readCommitted = &sql.TxOptions{
+	Isolation: sql.LevelReadCommitted,
+}
+
+func (m *ResourceVersionManager) execBatch(ctx context.Context, group, resource string, batch []writeOp) {
+	ctx, span := tracer.Start(ctx, "sql.resourceVersionManager.execBatch")
+	defer span.End()
+
+	// Add batch size attribute
+	span.SetAttributes(
+		attribute.Int("batch_size", len(batch)),
+		attribute.String("group", group),
+		attribute.String("resource", resource),
+	)
+
+	var err error
+	timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
+		status := "success"
+		if err != nil {
+			status = "error"
+		}
+		rvmExecBatchDuration.WithLabelValues(group, resource, status).Observe(v)
+	}))
+	defer timer.ObserveDuration()
+
+	ctx, cancel := context.WithTimeout(ctx, defaultBatchTimeout)
+	defer cancel()
+
+	guidToRV := make(map[string]int64, len(batch))
+	guidToSnowflakeRV := make(map[string]int64, len(batch))
+	guids := make([]string, len(batch)) // The GUIDs of the created resources in the same order as the batch
+	rvs := make([]int64, len(batch))    // The RVs of the created resources in the same order as the batch
+
+	err = m.db.WithTx(ctx, readCommitted, func(ctx context.Context, tx db.Tx) error {
+		span.AddEvent("starting_batch_transaction")
+
+		writeTimer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
+			rvmExecBatchPhaseDuration.WithLabelValues(group, resource, "write_ops").Observe(v)
+		}))
+		for i := range batch {
+			guid, err := batch[i].fn(tx)
+			if err != nil {
+				span.AddEvent("batch_operation_failed", trace.WithAttributes(
+					attribute.Int("operation_index", i),
+					attribute.String("error", err.Error()),
+				))
+				return err
+			}
+			guids[i] = guid
+		}
+		writeTimer.ObserveDuration()
+		span.AddEvent("batch_operations_completed")
+
+		lockTimer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
+			rvmExecBatchPhaseDuration.WithLabelValues(group, resource, "waiting_for_lock").Observe(v)
+		}))
+		rv, err := m.Lock(ctx, tx, group, resource)
+		lockTimer.ObserveDuration()
+		if err != nil {
+			span.AddEvent("resource_version_lock_failed", trace.WithAttributes(
+				attribute.String("error", err.Error()),
+			))
+			return fmt.Errorf("failed to increment resource version: %w", err)
+		}
+		span.AddEvent("resource_version_locked", trace.WithAttributes(
+			attribute.Int64("initial_rv", rv),
+		))
+
+		rvUpdateTimer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
+			rvmExecBatchPhaseDuration.WithLabelValues(group, resource, "update_resource_versions").Observe(v)
+		}))
+		defer rvUpdateTimer.ObserveDuration()
+		// Allocate the RVs
+		for i, guid := range guids {
+			guidToRV[guid] = rv
+			guidToSnowflakeRV[guid] = snowflakeFromRv(rv)
+			rvs[i] = rv
+			rv++
+		}
+		// Update the resource version for the created resources in both the resource and the resource history
+		if _, err := dbutil.Exec(ctx, tx, SqlResourceUpdateRV, SqlResourceUpdateRVRequest{
+			SQLTemplate: sqltemplate.New(m.dialect),
+			GUIDToRV:    guidToRV,
+		}); err != nil {
+			span.AddEvent("resource_update_rv_failed", trace.WithAttributes(
+				attribute.String("error", err.Error()),
+			))
+			return fmt.Errorf("update resource version: %w", err)
+		}
+		span.AddEvent("resource_versions_updated")
+
+		if _, err := dbutil.Exec(ctx, tx, SqlResourceHistoryUpdateRV, SqlResourceUpdateRVRequest{
+			SQLTemplate:       sqltemplate.New(m.dialect),
+			GUIDToRV:          guidToRV,
+			GUIDToSnowflakeRV: guidToSnowflakeRV,
+		}); err != nil {
+			span.AddEvent("resource_history_update_rv_failed", trace.WithAttributes(
+				attribute.String("error", err.Error()),
+			))
+			return fmt.Errorf("update resource history version: %w", err)
+		}
+		span.AddEvent("resource_history_versions_updated")
+
+		// Record the latest RV in the resource version table
+		err = m.SaveRV(ctx, tx, group, resource, rv)
+		if err != nil {
+			span.AddEvent("save_rv_failed", trace.WithAttributes(
+				attribute.String("error", err.Error()),
+			))
+		}
+		return err
+	})
+
+	if err != nil {
+		span.AddEvent("batch_transaction_failed", trace.WithAttributes(
+			attribute.String("error", err.Error()),
+		))
+	} else {
+		span.AddEvent("batch_transaction_completed")
+	}
+
+	// notify the caller that the operations are done
+	for i := range batch {
+		batch[i].done <- writeOpResult{
+			guid:           guids[i],
+			rv:             rvs[i],
+			err:            err,
+			batchTraceLink: trace.LinkFromContext(ctx),
+		}
+	}
+}
+
+// takes a unix microsecond rv and transforms into a snowflake format. The timestamp is converted from microsecond to
+// millisecond (the integer division) and the remainder is saved in the stepbits section. machine id is always 0
+func snowflakeFromRv(rv int64) int64 {
+	return (((rv / 1000) - snowflake.Epoch) << (snowflake.NodeBits + snowflake.StepBits)) + (rv % 1000)
+}
+
+// Lock locks the resource version for the given key
+func (m *ResourceVersionManager) Lock(ctx context.Context, x db.ContextExecer, group, resource string) (nextRV int64, err error) {
+	// 1. Lock the row and prevent concurrent updates until the transaction is committed
+	res, err := dbutil.QueryRow(ctx, x, SqlResourceVersionGet, sqlResourceVersionGetRequest{
+		SQLTemplate: sqltemplate.New(m.dialect),
+		Group:       group,
+		Resource:    resource,
+		Response:    new(ResourceVersionResponse),
+		ReadOnly:    false, // Lock the row for update
+	})
+
+	if errors.Is(err, sql.ErrNoRows) {
+		// If there wasn't a row for this resource, create it
+		if _, err = dbutil.Exec(ctx, x, SqlResourceVersionInsert, SqlResourceVersionUpsertRequest{
+			SQLTemplate: sqltemplate.New(m.dialect),
+			Group:       group,
+			Resource:    resource,
+		}); err != nil {
+			return 0, fmt.Errorf("insert into resource_version: %w", err)
+		}
+
+		// Fetch the newly created resource version
+		res, err = dbutil.QueryRow(ctx, x, SqlResourceVersionGet, sqlResourceVersionGetRequest{
+			SQLTemplate: sqltemplate.New(m.dialect),
+			Group:       group,
+			Resource:    resource,
+			Response:    new(ResourceVersionResponse),
+			ReadOnly:    true,
+		})
+		if err != nil {
+			return 0, fmt.Errorf("fetching RV after insert: %w", err)
+		}
+		return res.ResourceVersion, nil
+	} else if err != nil {
+		return 0, fmt.Errorf("lock the resource version: %w", err)
+	}
+
+	return max(res.CurrentEpoch, res.ResourceVersion+1), nil
+}
+
+func (m *ResourceVersionManager) SaveRV(ctx context.Context, x db.ContextExecer, group, resource string, rv int64) error {
+	_, err := dbutil.Exec(ctx, x, SqlResourceVersionUpdate, SqlResourceVersionUpsertRequest{
+		SQLTemplate:     sqltemplate.New(m.dialect),
+		Group:           group,
+		Resource:        resource,
+		ResourceVersion: rv,
+	})
+	if err != nil {
+		return fmt.Errorf("save resource version: %w", err)
+	}
+	return nil
+}

pkg/storage/unified/sql/rvmanager/rv_manager_test.go

@@ -0,0 +1,65 @@
+package rvmanager
+
+import (
+	"testing"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
+	"github.com/grafana/grafana/pkg/storage/unified/sql/db"
+	"github.com/grafana/grafana/pkg/storage/unified/sql/sqltemplate"
+	"github.com/grafana/grafana/pkg/storage/unified/sql/test"
+	"github.com/grafana/grafana/pkg/util/testutil"
+)
+
+func expectSuccessfulResourceVersionLock(t *testing.T, dbp test.TestDBProvider, rv int64, timestamp int64) {
+	dbp.SQLMock.ExpectQuery("select resource_version, unix_timestamp for update").
+		WillReturnRows(sqlmock.NewRows([]string{"resource_version", "unix_timestamp"}).
+			AddRow(rv, timestamp))
+}
+
+func expectSuccessfulResourceVersionSaveRV(t *testing.T, dbp test.TestDBProvider) {
+	dbp.SQLMock.ExpectExec("update resource set resource_version").WillReturnResult(sqlmock.NewResult(1, 1))
+	dbp.SQLMock.ExpectExec("update resource_history set resource_version").WillReturnResult(sqlmock.NewResult(1, 1))
+	dbp.SQLMock.ExpectExec("update resource_version set resource_version").WillReturnResult(sqlmock.NewResult(1, 1))
+}
+
+func expectSuccessfulResourceVersionExec(t *testing.T, dbp test.TestDBProvider, cbs ...func()) {
+	for _, cb := range cbs {
+		cb()
+	}
+	expectSuccessfulResourceVersionLock(t, dbp, 100, 200)
+	expectSuccessfulResourceVersionSaveRV(t, dbp)
+}
+
+func TestResourceVersionManager(t *testing.T) {
+	ctx := testutil.NewDefaultTestContext(t)
+	dbp := test.NewDBProviderMatchWords(t)
+	dialect := sqltemplate.DialectForDriver(dbp.DB.DriverName())
+	manager, err := NewResourceVersionManager(ResourceManagerOptions{
+		DB:      dbp.DB,
+		Dialect: dialect,
+	})
+	require.NoError(t, err)
+	require.NotNil(t, manager)
+
+	t.Run("should handle single operation", func(t *testing.T) {
+		key := &resourcepb.ResourceKey{
+			Group:    "test-group",
+			Resource: "test-resource",
+		}
+		dbp.SQLMock.ExpectBegin()
+		expectSuccessfulResourceVersionExec(t, dbp, func() {
+			dbp.SQLMock.ExpectExec("select 1").WillReturnResult(sqlmock.NewResult(1, 1))
+		})
+		dbp.SQLMock.ExpectCommit()
+
+		rv, err := manager.ExecWithRV(ctx, key, func(tx db.Tx) (string, error) {
+			_, err := tx.ExecContext(ctx, "select 1")
+			return "1234", err
+		})
+		require.NoError(t, err)
+		require.Equal(t, rv, int64(200))
+	})
+}

pkg/storage/unified/sql/rvmanager/templates.go

@@ -0,0 +1,30 @@
+package rvmanager
+
+import (
+	"embed"
+	"fmt"
+	"text/template"
+)
+
+// Templates setup.
+var (
+	//go:embed data/*.sql
+	sqlTemplatesFS embed.FS
+
+	sqlTemplates = template.Must(template.New("sql").ParseFS(sqlTemplatesFS, `data/*.sql`))
+)
+
+func mustTemplate(filename string) *template.Template {
+	if t := sqlTemplates.Lookup(filename); t != nil {
+		return t
+	}
+	panic(fmt.Sprintf("template file not found: %s", filename))
+}
+
+var (
+	SqlResourceUpdateRV        = mustTemplate("resource_update_rv.sql")
+	SqlResourceHistoryUpdateRV = mustTemplate("resource_history_update_rv.sql")
+	SqlResourceVersionGet      = mustTemplate("resource_version_get.sql")
+	SqlResourceVersionUpdate   = mustTemplate("resource_version_update.sql")
+	SqlResourceVersionInsert   = mustTemplate("resource_version_insert.sql")
+)