* Supported read-only sql.LevelSnapshot isolation with fake transaction and OnlineReadOnly transaction control (transient, while YDB clusters are not updated with true snapshot isolation mode)

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Supported read-only `sql.LevelSnapshot` isolation with fake transaction and `OnlineReadOnly` transaction control (transient, while YDB clusters are not updated with true snapshot isolation mode)
 * Supported the `*sql.Conn` as input type `ydb.Unwrap` helper for go's 1.18
 
 ## v3.36.2

SQL.md

@@ -87,8 +87,9 @@ if err = rows.Err(); err != nil { // always check final rows err
 
 ### With transaction <a name="queries-tx"></a>
 Supports only `default` transaction options which mapped to `YDB` `SerializableReadWrite` transaction settings.
+
 `YDB`'s `OnlineReadOnly` and `StaleReadOnly` transaction settings are not compatible with interactive transactions such as `database/sql`'s `*sql.Tx`.
-`YDB`'s `OnlineReadOnly` and `StaleReadOnly` transaction settings can be explicitly applied to each query outside interactive transaction (see more in [Isolation levels support](#tx-control))
+That's why `ydb-go-sdk` implements read-only `sql.LevelSnapshot` with fake transaction (transient, while YDB clusters are not updated with true snapshot isolation mode)
 ```go
 tx, err := db.BeginTx(ctx, sql.TxOptions{})
 if err != nil {
@@ -191,14 +192,19 @@ import (
 )
 ...
 err := retry.DoTx(context.TODO(), db, func(ctx context.Context, tx *sql.Tx) error {
-   // work with tx
-   rows, err := tx.QueryContext(ctx, "SELECT 1;")
-   if err != nil {
-       return err // return err to retryer
-   }
-   ...
-   return nil // good final of retry tx operation
-}, retry.WithDoTxRetryOptions(retry.WithIdempotent(true)))
+    // work with tx
+    rows, err := tx.QueryContext(ctx, "SELECT 1;")
+    if err != nil {
+        return err // return err to retryer
+    }
+    ...
+    return nil // good final of retry tx operation
+}, retry.WithDoTxRetryOptions(
+    retry.WithIdempotent(true),
+), retry.WithTxOptions(&sql.TxOptions{
+    Isolation: sql.LevelSnapshot,
+    ReadOnly:  true,
+}))
 ```
 
 ## Query args types <a name="arg-types"></a>

internal/xsql/conn.go

@@ -62,7 +62,14 @@ type conn struct {
 
 	scanOpts []options.ExecuteScanQueryOption
 
-	currentTx *tx
+	currentTx currentTx
+}
+
+type currentTx interface {
+	driver.Tx
+	driver.ExecerContext
+	driver.QueryerContext
+	table.TransactionIdentifier
 }
 
 var (
@@ -309,6 +316,19 @@ func (c *conn) BeginTx(ctx context.Context, txOptions driver.TxOptions) (_ drive
 			fmt.Errorf("conn already have an opened currentTx: %s", c.currentTx.ID()),
 		)
 	}
+	// TODO: replace with true transaction with snapshot read-only isolation after implementing it on server-side
+	//nolint:godox
+	if txOptions.ReadOnly && txOptions.Isolation == driver.IsolationLevel(sql.LevelSnapshot) {
+		c.currentTx = &fakeTx{
+			conn: c,
+			txControl: table.TxControl(
+				table.BeginTx(table.WithSerializableReadWrite()),
+				table.CommitTx(),
+			),
+			ctx: ctx,
+		}
+		return c.currentTx, nil
+	}
 	var txc table.TxOption
 	txc, err = isolation.ToYDB(txOptions)
 	if err != nil {

internal/xsql/fake_tx.go

@@ -0,0 +1,67 @@
+package xsql
+
+// TODO: drop this after implementing it on server-side
+//nolint:godox
+
+import (
+	"context"
+	"database/sql/driver"
+
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/retry"
+	"github.com/ydb-platform/ydb-go-sdk/v3/table"
+	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
+)
+
+type fakeTx struct {
+	txControl *table.TransactionControl
+	conn      *conn
+	ctx       context.Context
+}
+
+func (tx *fakeTx) ID() string {
+	return "fakeTx"
+}
+
+func (tx *fakeTx) Commit() (err error) {
+	onDone := trace.DatabaseSQLOnTxCommit(tx.conn.trace, &tx.ctx, tx)
+	defer func() {
+		onDone(err)
+	}()
+	if tx.conn.isClosed() {
+		return errClosedConn
+	}
+	defer func() {
+		tx.conn.currentTx = nil
+	}()
+	return nil
+}
+
+func (tx *fakeTx) Rollback() (err error) {
+	onDone := trace.DatabaseSQLOnTxRollback(tx.conn.trace, &tx.ctx, tx)
+	defer func() {
+		onDone(err)
+	}()
+	if tx.conn.isClosed() {
+		return errClosedConn
+	}
+	defer func() {
+		tx.conn.currentTx = nil
+	}()
+	return nil
+}
+
+func (tx *fakeTx) QueryContext(ctx context.Context, q string, args []driver.NamedValue) (_ driver.Rows, err error) {
+	onDone := trace.DatabaseSQLOnTxQuery(tx.conn.trace, &ctx, tx.ctx, tx, q, retry.IsIdempotent(ctx))
+	defer func() {
+		onDone(err)
+	}()
+	return tx.conn.QueryContext(WithTxControl(ctx, tx.txControl), q, args)
+}
+
+func (tx *fakeTx) ExecContext(ctx context.Context, q string, args []driver.NamedValue) (_ driver.Result, err error) {
+	onDone := trace.DatabaseSQLOnTxExec(tx.conn.trace, &ctx, tx.ctx, tx, q, retry.IsIdempotent(ctx))
+	defer func() {
+		onDone(err)
+	}()
+	return tx.conn.ExecContext(WithTxControl(ctx, tx.txControl), q, args)
+}

internal/xsql/isolation/isolation.go

@@ -5,8 +5,6 @@ import (
 	"database/sql/driver"
 	"fmt"
 
-	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Table"
-
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
 	"github.com/ydb-platform/ydb-go-sdk/v3/table"
 )
@@ -15,28 +13,12 @@ import (
 // This caused by ydb logic that prevents start actual transaction with OnlineReadOnly mode and ReadCommitted
 // and ReadUncommitted isolation levels should use tx_control in every query request.
 // It returns error on unsupported options.
-func ToYDB(opts driver.TxOptions) (txControl table.TxOption, err error) {
+func ToYDB(opts driver.TxOptions) (txcControl table.TxOption, err error) {
 	level := sql.IsolationLevel(opts.Isolation)
 	if !opts.ReadOnly && level == sql.LevelDefault {
 		return table.WithSerializableReadWrite(), nil
 	}
 	return nil, xerrors.WithStackTrace(fmt.Errorf(
-		"ydb: unsupported transaction options: isolation='%s' read_only='%t'",
-		level.String(), opts.ReadOnly,
+		"ydb: unsupported transaction options: %+v", opts,
 	))
 }
-
-// FromYDB maps table transaction settings to driver transaction options
-func FromYDB(txSettings *table.TransactionSettings) (txOptions *sql.TxOptions, err error) {
-	switch txSettings.Settings().TxMode.(type) {
-	case *Ydb_Table.TransactionSettings_SerializableReadWrite:
-		return &sql.TxOptions{
-			Isolation: sql.LevelDefault,
-			ReadOnly:  false,
-		}, nil
-	default:
-		return nil, xerrors.WithStackTrace(
-			fmt.Errorf("ydb: unsupported transaction settings: %+v", txSettings),
-		)
-	}
-}

internal/xsql/isolation/isolation_test.go

@@ -148,43 +148,3 @@ func TestToYDB(t *testing.T) {
 		})
 	}
 }
-
-func TestFromYDB(t *testing.T) {
-	for _, tt := range []struct {
-		txControl table.TxOption
-		txOptions *sql.TxOptions
-		err       bool
-	}{
-		{
-			txControl: table.WithSerializableReadWrite(),
-			txOptions: &sql.TxOptions{
-				Isolation: sql.LevelDefault,
-				ReadOnly:  false,
-			},
-		},
-		{
-			txControl: table.WithOnlineReadOnly(),
-			err:       true,
-		},
-		{
-			txControl: table.WithOnlineReadOnly(table.WithInconsistentReads()),
-			err:       true,
-		},
-		{
-			txControl: table.WithStaleReadOnly(),
-			err:       true,
-		},
-	} {
-		t.Run(fmt.Sprintf("%+v", table.TxSettings(tt.txControl).Settings()), func(t *testing.T) {
-			fromYDB, err := FromYDB(table.TxSettings(tt.txControl))
-			if tt.err {
-				require.Error(t, err)
-			} else {
-				require.NoError(t, err)
-				if *fromYDB != *tt.txOptions {
-					t.Errorf("%+v != %+v", *fromYDB, *tt.txOptions)
-				}
-			}
-		})
-	}
-}

retry/sql.go

@@ -57,6 +57,7 @@ func Do(ctx context.Context, db *sql.DB, f func(ctx context.Context, cc *sql.Con
 }
 
 type doTxOptions struct {
+	txOptions    *sql.TxOptions
 	retryOptions []retryOption
 }
 
@@ -71,14 +72,23 @@ func WithDoTxRetryOptions(opts ...retryOption) doTxOption {
 	}
 }
 
+// WithTxOptions specified transaction options
+func WithTxOptions(txOptions *sql.TxOptions) doTxOption {
+	return func(o *doTxOptions) error {
+		o.txOptions = txOptions
+		return nil
+	}
+}
+
 // DoTx is a retryer of database/sql transactions with fallbacks on errors
 func DoTx(ctx context.Context, db *sql.DB, f func(context.Context, *sql.Tx) error, opts ...doTxOption) error {
 	var (
-		txOptions = &sql.TxOptions{
-			Isolation: sql.LevelDefault,
-			ReadOnly:  false,
+		options = doTxOptions{
+			txOptions: &sql.TxOptions{
+				Isolation: sql.LevelDefault,
+				ReadOnly:  false,
+			},
 		}
-		options  = doTxOptions{}
 		attempts = 0
 	)
 	for _, o := range opts {
@@ -88,7 +98,7 @@ func DoTx(ctx context.Context, db *sql.DB, f func(context.Context, *sql.Tx) erro
 	}
 	err := Retry(ctx, func(ctx context.Context) (err error) {
 		attempts++
-		tx, err := db.BeginTx(ctx, txOptions)
+		tx, err := db.BeginTx(ctx, options.txOptions)
 		if err != nil {
 			return unwrapErrBadConn(xerrors.WithStackTrace(err))
 		}

sql_e2e_test.go

@@ -165,31 +165,38 @@ func TestDatabaseSql(t *testing.T) {
 		if err != nil {
 			t.Fatalf("begin tx failed: %v\n", err)
 		}
-		err = retry.DoTx(ctx, db, func(ctx context.Context, tx *sql.Tx) error {
-			row := tx.QueryRowContext(ctx,
-				render(
-					querySelect,
-					templateConfig{
-						TablePathPrefix: path.Join(cc.Name(), folder),
-					},
-				),
-				sql.Named("seriesID", uint64(1)),
-				sql.Named("seasonID", uint64(1)),
-				sql.Named("episodeID", uint64(1)),
-			)
-			var views sql.NullFloat64
-			if err = row.Scan(&views); err != nil {
-				return fmt.Errorf("cannot select current views: %w", err)
-			}
-			if !views.Valid {
-				return fmt.Errorf("unexpected invalid views: %v", views)
-			}
-			t.Logf("views = %v", views)
-			if views.Float64 != 1 {
-				return fmt.Errorf("unexpected views value: %v", views)
-			}
-			return nil
-		}, retry.WithDoTxRetryOptions(retry.WithIdempotent(true)))
+		err = retry.DoTx(ctx, db,
+			func(ctx context.Context, tx *sql.Tx) error {
+				row := tx.QueryRowContext(ctx,
+					render(
+						querySelect,
+						templateConfig{
+							TablePathPrefix: path.Join(cc.Name(), folder),
+						},
+					),
+					sql.Named("seriesID", uint64(1)),
+					sql.Named("seasonID", uint64(1)),
+					sql.Named("episodeID", uint64(1)),
+				)
+				var views sql.NullFloat64
+				if err = row.Scan(&views); err != nil {
+					return fmt.Errorf("cannot select current views: %w", err)
+				}
+				if !views.Valid {
+					return fmt.Errorf("unexpected invalid views: %v", views)
+				}
+				t.Logf("views = %v", views)
+				if views.Float64 != 1 {
+					return fmt.Errorf("unexpected views value: %v", views)
+				}
+				return nil
+			},
+			retry.WithDoTxRetryOptions(retry.WithIdempotent(true)),
+			retry.WithTxOptions(&sql.TxOptions{
+				Isolation: sql.LevelSnapshot,
+				ReadOnly:  true,
+			}),
+		)
 		if err != nil {
 			t.Fatalf("begin tx failed: %v\n", err)
 		}

table/table.go

@@ -364,6 +364,15 @@ func DefaultTxControl() *TransactionControl {
 	)
 }
 
+// SerializableReadWriteTxControl returns transaction control with serializable read-write isolation mode
+func SerializableReadWriteTxControl(opts ...TxControlOption) *TransactionControl {
+	return TxControl(
+		append([]TxControlOption{
+			BeginTx(WithSerializableReadWrite()),
+		}, opts...)...,
+	)
+}
+
 // OnlineReadOnlyTxControl returns online read-only transaction control
 func OnlineReadOnlyTxControl(opts ...TxOnlineReadOnlyOption) *TransactionControl {
 	return TxControl(