Merge pull request #351 from ydb-platform/tx-ctx

* Added some description to error message from table pool get

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,12 @@
+* Added some description to error message from table pool get
+* Moved implementation `sugar.GenerateDeclareSection` to `internal/table`
+* Added transaction trace callbacks and internal logging with them
+* Stored context from `BeginTx` to `internal/xsql` transaction
+* Added automatically generated declare section to query text in `database/sql` usage 
+* Removed supports `sql.LevelSerializable`
+* Added `retry.Do` helper for retry custom lambda with `database/sql` without transactions
+* Removed `retry.WithTxOptions` option (only default isolation supports)
+
 ## v3.34.1
 * Changed `database/sql` driver `prepare` behaviour to `nop` with proxing call to conn exec/query with keep-in-cache flag
 * Added metadata to `trace.Driver.OnInvoke` and `trace.Driver.OnNewStream` done events

internal/table/client.go

@@ -323,8 +323,9 @@ func (c *Client) internalPoolGet(ctx context.Context, opts ...getOption) (s *ses
 	}
 
 	var (
-		i = 0
-		o = getOptions{t: c.config.Trace()}
+		start = time.Now()
+		i     = 0
+		o     = getOptions{t: c.config.Trace()}
 	)
 	for _, opt := range opts {
 		opt(&o)
@@ -375,11 +376,24 @@ func (c *Client) internalPoolGet(ctx context.Context, opts ...getOption) (s *ses
 		}
 	}
 	if s == nil && err == nil {
-		err = xerrors.WithStackTrace(fmt.Errorf("%w: attempts=%d", errNoProgress, i))
+		err = xerrors.WithStackTrace(errNoProgress)
 	}
 	if err != nil {
+		var (
+			index            int
+			idle             int
+			createInProgress int
+		)
+		c.mu.WithLock(func() {
+			index = len(c.index)
+			idle = c.idle.Len()
+			createInProgress = c.createInProgress
+		})
 		return s, xerrors.WithStackTrace(
-			fmt.Errorf("%w: attempts=%d", err, i),
+			fmt.Errorf("failed to get session from pool ("+
+				"attempts: %d, latency: %v, stats: {index: %d, idle: %d, create_in_progress: %d}"+
+				"): %w", i, time.Since(start), index, idle, createInProgress, err,
+			),
 		)
 	}
 	return s, nil

internal/table/params.go

@@ -0,0 +1,34 @@
+package table
+
+import (
+	"bytes"
+	"fmt"
+	"sort"
+
+	"github.com/ydb-platform/ydb-go-sdk/v3/table"
+	"github.com/ydb-platform/ydb-go-sdk/v3/table/types"
+)
+
+// GenerateDeclareSection generates DECLARE section text in YQL query by params
+//
+// Warning: This is an experimental feature and could change at any time
+func GenerateDeclareSection(params *table.QueryParameters) string {
+	var (
+		buf      bytes.Buffer
+		names    []string
+		declares = make(map[string]string, len(params.Params()))
+	)
+	params.Each(func(name string, v types.Value) {
+		names = append(names, name)
+		declares[name] = fmt.Sprintf(
+			"DECLARE %s AS %s;\n",
+			name,
+			v.Type().String(),
+		)
+	})
+	sort.Strings(names)
+	for _, name := range names {
+		buf.WriteString(declares[name])
+	}
+	return buf.String()
+}

internal/table/session.go

@@ -666,7 +666,7 @@ func (s *session) Execute(
 		optsResult.QueryCachePolicy.GetKeepInCache(),
 	)
 	defer func() {
-		onDone(txr, true, r, err)
+		onDone(txr, false, r, err)
 	}()
 
 	request, result, err := s.executeDataQuery(ctx, tx, q, params, opts...)

internal/table/transaction.go

@@ -46,6 +46,28 @@ func (tx *transaction) Execute(
 	query string, params *table.QueryParameters,
 	opts ...options.ExecuteDataQueryOption,
 ) (r result.Result, err error) {
+	q := new(dataQuery)
+	q.initFromText(query)
+
+	if params == nil {
+		params = table.NewQueryParameters()
+	}
+	var optsResult options.ExecuteDataQueryDesc
+	for _, f := range opts {
+		f(&optsResult)
+	}
+	onDone := trace.TableOnSessionTransactionExecute(
+		tx.s.config.Trace(),
+		&ctx,
+		tx.s,
+		tx,
+		q,
+		params,
+		optsResult.QueryCachePolicy.GetKeepInCache(),
+	)
+	defer func() {
+		onDone(r, err)
+	}()
 	_, r, err = tx.s.Execute(ctx, tx.txc(), query, params, opts...)
 	return
 }
@@ -56,6 +78,23 @@ func (tx *transaction) ExecuteStatement(
 	stmt table.Statement, params *table.QueryParameters,
 	opts ...options.ExecuteDataQueryOption,
 ) (r result.Result, err error) {
+	if params == nil {
+		params = table.NewQueryParameters()
+	}
+	var optsResult options.ExecuteDataQueryDesc
+	for _, f := range opts {
+		f(&optsResult)
+	}
+	onDone := trace.TableOnSessionTransactionExecuteStatement(
+		tx.s.config.Trace(),
+		&ctx,
+		tx.s,
+		tx,
+		params,
+	)
+	defer func() {
+		onDone(r, err)
+	}()
 	_, r, err = stmt.Execute(ctx, tx.txc(), params, opts...)
 	return
 }

internal/xsql/badconn/badconn_go1.18.go

@@ -15,7 +15,7 @@ type badConnError struct {
 }
 
 func (e badConnError) Error() string {
-	return "ydbsql: bad connection: " + e.err.Error()
+	return e.err.Error()
 }
 
 func (e badConnError) Is(err error) bool {