Merge pull request #504 from ydb-platform/fix-with-commit

Reverted tx.WithCommit() and added options.WithCommit() option of ExecuteDataQuery

Author: asmyasnikov

.golangci.yml

@@ -127,7 +127,6 @@ linters:
   disable-all: true
   enable:
 #    - cyclop
-    - deadcode
     - depguard
     - dogsled
 #    - dupl
@@ -173,7 +172,6 @@ linters:
     - revive
     - staticcheck
     - stylecheck
-    - structcheck
 #    - tagliatelle
 #    - testpackage
 #    - thelper
@@ -183,7 +181,6 @@ linters:
     - unparam
     - unused
 #    - varnamelen
-    - varcheck
     - whitespace
 #    - wrapcheck
 #    - wsl

CHANGELOG.md

@@ -1,6 +1,9 @@
 * Added retry policy options for topics: `topic/topicoptions.WithReaderCheckRetryErrorFunction`, `topic/topicoptions.WithReaderStartTimeout`, `topic/topicoptions.WithWriterCheckRetryErrorFunction`, `topic/topicoptions.WithWriterStartTimeout`
-* Refactored `internal/conn` middlewares for `DRY`
+* Refactored `internal/conn` middlewares
 * Added `trace.tableSessionInfo.LastUsage()` method for get last usage timestamp
+* Reverted `tx.WithCommit()` changes for fix unstable behaviour of lazy transactions
+* Added `options.WithCommit()` option for execute query with auto-commit flag
+* Removed `trace.TableTransactionExecuteStartInfo.KeepInCache` field as redundant
 
 ## v3.41.0
 * Added option for set interval of auth token update in topic streams

internal/cmd/gtrace/writer.go

@@ -782,7 +782,6 @@ func (f flags) has(x flags) bool {
 }
 
 const (
-	//nolint
 	zeroFlags flags = 1 << iota >> 1
 	docs
 )

internal/table/session.go

@@ -671,25 +671,31 @@ func (s *session) Execute(
 		return nil, nil, xerrors.WithStackTrace(err)
 	}
 
-	return s.executeQueryResult(result, txControl)
+	return s.executeQueryResult(result, request.TxControl)
 }
 
 // executeQueryResult returns Transaction and result built from received
 // result.
-func (s *session) executeQueryResult(res *Ydb_Table.ExecuteQueryResult, txControl *table.TransactionControl) (
+func (s *session) executeQueryResult(
+	res *Ydb_Table.ExecuteQueryResult, txControl *Ydb_Table.TransactionControl,
+) (
 	table.Transaction, result.Result, error,
 ) {
-	t := &transaction{
+	tx := &transaction{
 		id: res.GetTxMeta().GetId(),
 		s:  s,
-		c:  txControl,
 	}
-	r := scanner.NewUnary(
+	if txControl.CommitTx {
+		tx.state = txStateCommitted
+	} else {
+		tx.state = txStateInitialized
+		tx.control = table.TxControl(table.WithTxID(tx.id))
+	}
+	return tx, scanner.NewUnary(
 		res.GetResultSets(),
 		res.GetQueryStats(),
 		scanner.WithIgnoreTruncated(s.config.IgnoreTruncated()),
-	)
-	return t, r, nil
+	), nil
 }
 
 // executeDataQuery executes data query.
@@ -1084,8 +1090,9 @@ func (s *session) BeginTransaction(
 		return
 	}
 	return &transaction{
-		id: result.GetTxMeta().GetId(),
-		s:  s,
-		c:  table.TxControl(table.WithTxID(result.GetTxMeta().GetId())),
+		id:      result.GetTxMeta().GetId(),
+		state:   txStateInitialized,
+		s:       s,
+		control: table.TxControl(table.WithTxID(result.GetTxMeta().GetId())),
 	}, nil
 }

internal/table/statement.go

@@ -63,21 +63,20 @@ func (s *statement) Execute(
 		onDone(txr, true, r, err)
 	}()
 
-	return s.execute(ctx, a, request, txControl)
+	return s.execute(ctx, a, request, request.TxControl)
 }
 
 // execute executes prepared query without any tracing.
 func (s *statement) execute(
 	ctx context.Context, a *allocator.Allocator,
-	request *Ydb_Table.ExecuteDataQueryRequest, txControl *table.TransactionControl,
+	request *Ydb_Table.ExecuteDataQueryRequest, txControl *Ydb_Table.TransactionControl,
 ) (
 	txr table.Transaction, r result.Result, err error,
 ) {
 	res, err := s.session.executeDataQuery(ctx, a, request)
 	if err != nil {
 		return nil, nil, xerrors.WithStackTrace(err)
 	}
-
 	return s.session.executeQueryResult(res, txControl)
 }