Merge pull request #335 from ydb-platform/retry-do-tx

* Added `retry.DoTx` helper for retrying `database/sql` transactions

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Added `retry.DoTx` helper for retrying `database/sql` transactions 
 * Implemented `database/sql` driver over `ydb-go-sdk`
 * Marked as deprecated `trace.Table.OnPoolSessionNew` and `trace.Table.OnPoolSessionClose` events
 * Added `trace.Table.OnPoolSessionAdd` and `trace.Table.OnPoolSessionRemove` events

internal/scheme/client.go

@@ -153,14 +153,20 @@ func (c *Client) DescribePath(ctx context.Context, path string) (e scheme.Entry,
 	}
 	call := func(ctx context.Context) error {
 		e, err = c.describePath(ctx, path)
-		return xerrors.WithStackTrace(err)
+		if err != nil {
+			return xerrors.WithStackTrace(err)
+		}
+		return nil
 	}
 	if !c.config.AutoRetry() {
 		err = call(ctx)
 		return
 	}
 	err = retry.Retry(ctx, call, retry.WithIdempotent(true), retry.WithStackTrace())
-	return
+	if err != nil {
+		return e, xerrors.WithStackTrace(err)
+	}
+	return e, nil
 }
 
 func (c *Client) describePath(ctx context.Context, path string) (e scheme.Entry, err error) {
@@ -223,7 +229,10 @@ func (c *Client) modifyPermissions(ctx context.Context, path string, opts ...sch
 			),
 		},
 	)
-	return xerrors.WithStackTrace(err)
+	if err != nil {
+		return xerrors.WithStackTrace(err)
+	}
+	return nil
 }
 
 func putEntry(dst []scheme.Entry, src []*Ydb_Scheme.Entry) {

internal/table/statement.go

@@ -18,6 +18,10 @@ type statement struct {
 	params  map[string]*Ydb.Type
 }
 
+func Params(s table.Statement) map[string]*Ydb.Type {
+	return s.(*statement).params
+}
+
 // Execute executes prepared data query.
 func (s *statement) Execute(
 	ctx context.Context, tx *table.TransactionControl,

internal/xsql/conn.go

@@ -7,7 +7,9 @@ import (
 	"fmt"
 	"sync/atomic"
 
+	internal "github.com/ydb-platform/ydb-go-sdk/v3/internal/table"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsql/isolation"
 	"github.com/ydb-platform/ydb-go-sdk/v3/retry"
 	"github.com/ydb-platform/ydb-go-sdk/v3/table"
 	"github.com/ydb-platform/ydb-go-sdk/v3/table/options"
@@ -114,17 +116,17 @@ func (c *conn) PrepareContext(ctx context.Context, query string) (driver.Stmt, e
 		return nil, c.checkClosed(err)
 	}
 	return &stmt{
-		conn:      c,
-		statement: s,
-		query:     query,
+		conn:   c,
+		params: internal.Params(s),
+		query:  query,
 	}, nil
 }
 
-func (c *conn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {
+func (c *conn) BeginTx(ctx context.Context, txOptions driver.TxOptions) (driver.Tx, error) {
 	if c.isClosed() {
 		return nil, errClosedConn
 	}
-	txSettings, err := mapIsolation(opts)
+	txSettings, err := isolation.ToYDB(txOptions)
 	if err != nil {
 		return nil, xerrors.WithStackTrace(err)
 	}

internal/xsql/dsn.go

@@ -0,0 +1,28 @@
+package xsql
+
+import (
+	"fmt"
+	"net/url"
+
+	"github.com/ydb-platform/ydb-go-sdk/v3/config"
+	"github.com/ydb-platform/ydb-go-sdk/v3/credentials"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
+)
+
+func Parse(dataSourceName string) (opts []config.Option, connectorOpts []ConnectorOption, err error) {
+	uri, err := url.Parse(dataSourceName)
+	if err != nil {
+		return nil, nil, xerrors.WithStackTrace(err)
+	}
+	if token := uri.Query().Get("token"); token != "" {
+		opts = append(opts, config.WithCredentials(credentials.NewAccessTokenCredentials(token)))
+	}
+	if queryMode := uri.Query().Get("query_mode"); queryMode != "" {
+		mode := QueryModeFromString(queryMode)
+		if mode == UnknownQueryMode {
+			return nil, nil, xerrors.WithStackTrace(fmt.Errorf("unknown query mode: %s", queryMode))
+		}
+		connectorOpts = append(connectorOpts, WithDefaultQueryMode(mode))
+	}
+	return opts, connectorOpts, nil
+}

internal/xsql/isolation.go

@@ -0,0 +1,84 @@
+package isolation
+
+import (
+	"database/sql"
+	"database/sql/driver"
+	"errors"
+	"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"
+)
+
+var errUnsupportedReadOnlyMode = errors.New("unsupported read-only mode")
+
+// ToYDB maps driver transaction options to ydb transaction Option or query transaction control.
+// 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) (txSettings *table.TransactionSettings, err error) {
+	level := sql.IsolationLevel(opts.Isolation)
+	if !opts.ReadOnly {
+		switch level {
+		case sql.LevelDefault,
+			sql.LevelSerializable:
+			return table.TxSettings(table.WithSerializableReadWrite()), nil
+		default:
+			return nil, xerrors.WithStackTrace(fmt.Errorf(
+				"ydb: unsupported transaction options: isolation=%s read_only=%t",
+				nameIsolationLevel(level), opts.ReadOnly,
+			))
+		}
+	}
+	return nil, xerrors.WithStackTrace(errUnsupportedReadOnlyMode)
+}
+
+func nameIsolationLevel(x sql.IsolationLevel) string {
+	if int(x) < len(isolationLevelName) {
+		return isolationLevelName[x]
+	}
+	return "unknown_isolation"
+}
+
+var isolationLevelName = [...]string{
+	sql.LevelDefault:         "default",
+	sql.LevelReadUncommitted: "read_uncommitted",
+	sql.LevelReadCommitted:   "read_committed",
+	sql.LevelWriteCommitted:  "write_committed",
+	sql.LevelRepeatableRead:  "repeatable_read",
+	sql.LevelSnapshot:        "snapshot",
+	sql.LevelSerializable:    "serializable",
+	sql.LevelLinearizable:    "linearizable",
+}
+
+// FromYDB maps table transaction settings to driver transaction options
+func FromYDB(txSettings *table.TransactionSettings) (txOptions *sql.TxOptions, err error) {
+	switch txMode := txSettings.Settings().TxMode.(type) {
+	case *Ydb_Table.TransactionSettings_SerializableReadWrite:
+		return &sql.TxOptions{
+			Isolation: sql.LevelSerializable,
+			ReadOnly:  false,
+		}, nil
+	case *Ydb_Table.TransactionSettings_OnlineReadOnly:
+		txOptions = &sql.TxOptions{
+			Isolation: sql.LevelReadCommitted,
+			ReadOnly:  true,
+		}
+		if txMode.OnlineReadOnly.AllowInconsistentReads {
+			txOptions.Isolation = sql.LevelReadUncommitted
+		}
+		return txOptions, nil
+	case *Ydb_Table.TransactionSettings_StaleReadOnly:
+		return &sql.TxOptions{
+			Isolation: sql.LevelReadCommitted,
+			ReadOnly:  true,
+		}, nil
+	default:
+		return nil, xerrors.WithStackTrace(fmt.Errorf(
+			"ydb: unsupported transaction options: txMode=%T(%v)",
+			txMode, txMode,
+		))
+	}
+}