Skip to content

Commit 40b55a6

Browse files
rekbyrekby
committed
add changelog, remove allowlazy
1 parent eaa2e0c commit 40b55a6

File tree

7 files changed

+36
-183
lines changed

7 files changed

+36
-183
lines changed

‎CHANGELOG.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
* Added write to topics within transactions
2+
13
## v3.80.10
24
* Added `ydb.WithSessionPoolSessionUsageLimit()` option for limitation max count of session usage
35
* Refactored experimental topic iterators in `topicsugar` package

‎examples/topic/table_topic_transactions/table_topic_transactions.go

Lines changed: 0 additions & 92 deletions
This file was deleted.

‎examples/topic/topicwriter/topic_writer_transaction.go

Lines changed: 34 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -11,41 +11,26 @@ import (
1111
"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicwriter"
1212
)
1313

14-
func TableAndTopicWithinTransaction(ctx context.Context, db *ydb.Driver, writer *topicwriter.Writer, id int64) error {
15-
return db.Query().DoTx(ctx, func(ctx context.Context, t query.TxActor) error {
16-
row, err := t.QueryRow(ctx, "SELECT val FROM table WHERE id=$id", query.WithParameters(
17-
ydb.ParamsBuilder().
18-
Param("$id").Int64(id).
19-
Build()))
20-
if err != nil {
21-
return err
22-
}
23-
24-
var val int64
25-
if err = row.Scan(&val); err != nil {
26-
return err
27-
}
28-
29-
err = writer.WriteWithTx(ctx, t, topicwriter.Message{
30-
Data: strings.NewReader(fmt.Sprintf("val: %v processed", val)),
31-
})
14+
func CopyMessagesBetweenTopicsTxWriter(
15+
ctx context.Context,
16+
db *ydb.Driver,
17+
reader *topicreader.Reader,
18+
topic string,
19+
) error {
20+
return db.Query().DoTx(ctx, func(ctx context.Context, tx query.TxActor) error {
21+
writer, err := db.Topic().StartTransactionalWriter(tx, topic)
3222
if err != nil {
3323
return err
3424
}
35-
return nil
36-
})
37-
}
3825

39-
func CopyMessagesBetweenTopics(ctx context.Context, db *ydb.Driver, reader *topicreader.Reader, writer *topicwriter.Writer) error {
40-
return db.Query().DoTx(ctx, func(ctx context.Context, tx query.TxActor) error {
4126
batch, err := reader.PopMessagesBatchTx(ctx, tx)
4227
if err != nil {
4328
return err
4429
}
4530

4631
for _, mess := range batch.Messages {
4732

48-
if err = writer.WriteWithTx(ctx, tx, topicwriter.Message{Data: mess}); err != nil {
33+
if err = writer.Write(ctx, topicwriter.Message{Data: mess}); err != nil {
4934
return err
5035
}
5136
}
@@ -54,25 +39,41 @@ func CopyMessagesBetweenTopics(ctx context.Context, db *ydb.Driver, reader *topi
5439
}, query.WithIdempotent())
5540
}
5641

57-
func CopyMessagesBetweenTopicsTxWriter(ctx context.Context, db *ydb.Driver, reader *topicreader.Reader, topic string) error {
58-
return db.Query().DoTx(ctx, func(ctx context.Context, tx query.TxActor) error {
59-
writer, err := db.Topic().StartTransactionalWriter(ctx, tx, topic)
42+
func TableAndTopicWithinTransaction(
43+
ctx context.Context,
44+
db *ydb.Driver,
45+
topicPath string,
46+
id int64,
47+
) error {
48+
return db.Query().DoTx(ctx, func(ctx context.Context, t query.TxActor) error {
49+
row, err := t.QueryRow(ctx, "SELECT val FROM table WHERE id=$id", query.WithParameters(
50+
ydb.ParamsBuilder().
51+
Param("$id").Int64(id).
52+
Build()))
6053
if err != nil {
6154
return err
6255
}
6356

64-
batch, err := reader.PopMessagesBatchTx(ctx, tx)
57+
var val int64
58+
if err = row.Scan(&val); err != nil {
59+
return err
60+
}
61+
62+
// the writer is dedicated for the transaction, it can't be used outside the transaction
63+
// it is no needs to close or flush the messages - it happened internally on transaction commit
64+
writer, err := db.Topic().StartTransactionalWriter(t, topicPath)
6565
if err != nil {
6666
return err
6767
}
6868

69-
for _, mess := range batch.Messages {
69+
err = writer.Write(ctx, topicwriter.Message{
70+
Data: strings.NewReader(fmt.Sprintf("val: %v processed", val)),
71+
})
7072

71-
if err = writer.Write(ctx, topicwriter.Message{Data: mess}); err != nil {
72-
return err
73-
}
73+
if err != nil {
74+
return err
7475
}
7576

7677
return nil
77-
}, query.WithIdempotent())
78+
})
7879
}

‎internal/query/execute_query.go

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@ type executeSettings interface {
3030
Params() *params.Parameters
3131
CallOptions() []grpc.CallOption
3232
RetryOpts() []retry.Option
33-
AllowLazyTx() bool
3433
}
3534

3635
type executeScriptConfig interface {

‎internal/query/options/execute.go

Lines changed: 0 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,6 @@ type (
3535
callOptions []grpc.CallOption
3636
txControl *tx.Control
3737
retryOptions []retry.Option
38-
allowLazyTx bool
3938
}
4039

4140
// Execute is an interface for execute method options
@@ -58,15 +57,8 @@ type (
5857
callback func(stats.QueryStats)
5958
}
6059
execModeOption = ExecMode
61-
lazyTxOption struct {
62-
allow bool
63-
}
6460
)
6561

66-
func (s *executeSettings) AllowLazyTx() bool {
67-
return s.allowLazyTx
68-
}
69-
7062
func (s *executeSettings) RetryOpts() []retry.Option {
7163
return s.retryOptions
7264
}
@@ -89,10 +81,6 @@ func (syntax Syntax) applyExecuteOption(s *executeSettings) {
8981
s.syntax = syntax
9082
}
9183

92-
func (lazyTx lazyTxOption) applyExecuteOption(s *executeSettings) {
93-
s.allowLazyTx = lazyTx.allow
94-
}
95-
9684
const (
9785
SyntaxYQL = Syntax(Ydb_Query.Syntax_SYNTAX_YQL_V1)
9886
SyntaxPostgreSQL = Syntax(Ydb_Query.Syntax_SYNTAX_PG)
@@ -188,7 +176,6 @@ var (
188176
_ Execute = StatsMode(0)
189177
_ Execute = txCommitOption{}
190178
_ Execute = (*txControlOption)(nil)
191-
_ Execute = lazyTxOption{}
192179
)
193180

194181
func WithCommit() txCommitOption {
@@ -203,10 +190,6 @@ func WithSyntax(syntax Syntax) syntaxOption {
203190
return syntax
204191
}
205192

206-
func WithLazyTx(allowLazy bool) lazyTxOption {
207-
return lazyTxOption{allow: allowLazy}
208-
}
209-
210193
func (opt statsModeOption) applyExecuteOption(s *executeSettings) {
211194
s.statsMode = opt.mode
212195
s.statsCallback = opt.callback

‎internal/query/transaction.go

Lines changed: 0 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -89,14 +89,6 @@ func (tx *Transaction) QueryResultSet(
8989
return nil, xerrors.WithStackTrace(errExecuteOnCompletedTx)
9090
}
9191

92-
if !options.ExecuteSettings(opts...).AllowLazyTx() {
93-
// It needs to execute before create settings because if tx must be not lazy, and it is first query
94-
// change txcontrol during unlazy
95-
if err := tx.UnLazy(ctx); err != nil {
96-
return nil, err
97-
}
98-
}
99-
10092
settings, err := tx.executeSettings(opts...)
10193
if err != nil {
10294
return nil, xerrors.WithStackTrace(err)
@@ -144,14 +136,6 @@ func (tx *Transaction) QueryRow(
144136
onDone(finalErr)
145137
}()
146138

147-
if !options.ExecuteSettings(opts...).AllowLazyTx() {
148-
// It needs to execute before create settings because if tx must be not lazy, and it is first query
149-
// change txcontrol during unlazy
150-
if err := tx.UnLazy(ctx); err != nil {
151-
return nil, err
152-
}
153-
}
154-
155139
settings := options.ExecuteSettings(
156140
append(
157141
[]options.Execute{options.WithTxControl(tx.txControl())},
@@ -219,14 +203,6 @@ func (tx *Transaction) Exec(ctx context.Context, q string, opts ...options.Execu
219203
return xerrors.WithStackTrace(errExecuteOnCompletedTx)
220204
}
221205

222-
if !options.ExecuteSettings(opts...).AllowLazyTx() {
223-
// It needs to execute before create settings because if tx must be not lazy, and it is first query
224-
// change txcontrol during unlazy
225-
if err := tx.UnLazy(ctx); err != nil {
226-
return err
227-
}
228-
}
229-
230206
settings, err := tx.executeSettings(opts...)
231207
if err != nil {
232208
return xerrors.WithStackTrace(err)
@@ -296,14 +272,6 @@ func (tx *Transaction) Query(ctx context.Context, q string, opts ...options.Exec
296272
return nil, xerrors.WithStackTrace(errExecuteOnCompletedTx)
297273
}
298274

299-
if !options.ExecuteSettings(opts...).AllowLazyTx() {
300-
// It needs to execute before create settings because if tx must be not lazy, and it is first query
301-
// change txcontrol during unlazy
302-
if err := tx.UnLazy(ctx); err != nil {
303-
return nil, err
304-
}
305-
}
306-
307275
settings, err := tx.executeSettings(opts...)
308276
if err != nil {
309277
return nil, xerrors.WithStackTrace(err)

‎query/transaction.go

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@ package query
33
import (
44
"context"
55

6-
"github.com/ydb-platform/ydb-go-sdk/v3/internal/query/options"
76
internal "github.com/ydb-platform/ydb-go-sdk/v3/internal/query/tx"
87
"github.com/ydb-platform/ydb-go-sdk/v3/internal/tx"
98
)
@@ -33,13 +32,6 @@ func WithTx(t tx.Identifier) internal.ControlOption {
3332
return internal.WithTx(t)
3433
}
3534

36-
// WithLazyTx replace default behavior for use lazy transaction or explicit start it before any query.
37-
//
38-
// Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
39-
func WithLazyTx(allowLazy bool) options.Execute {
40-
return options.WithLazyTx(allowLazy)
41-
}
42-
4335
func WithTxID(txID string) internal.ControlOption {
4436
return internal.WithTxID(txID)
4537
}

0 commit comments

Comments
 (0)