Skip to content

Commit 35d60ba

Browse files
asmyasnikovasmyasnikov
committed
* Added experimental ydb.{Register,Unregister}DsnParser global funcs for register/unregister external custom DSN parser for ydb.Open and sql.Open driver constructor
1 parent d6d2af1 commit 35d60ba

15 files changed

+507
-390
lines changed

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
* Added experimental `ydb.{Register,Unregister}DsnParser` global funcs for register/unregister external custom DSN parser for `ydb.Open` and `sql.Open` driver constructor
12
* Simple implement option WithReaderWithoutConsumer
23
* Fixed bug: topic didn't send specified partition number to a server
34

‎driver.go‎

Lines changed: 15 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package ydb
33
import (
44
"context"
55
"errors"
6+
"fmt"
67
"os"
78
"sync"
89

@@ -234,13 +235,20 @@ func (d *Driver) Topic() topic.Client {
234235
// See sugar.DSN helper for make dsn from endpoint and database
235236
//
236237
//nolint:nonamedreturns
237-
func Open(ctx context.Context, dsn string, opts ...Option) (_ *Driver, err error) {
238-
d, err := newConnectionFromOptions(ctx, append(
239-
[]Option{
240-
WithConnectionString(dsn),
241-
},
242-
opts...,
243-
)...)
238+
func Open(ctx context.Context, dsn string, opts ...Option) (_ *Driver, _ error) {
239+
opts = append(append(make([]Option, 0, len(opts)+1), WithConnectionString(dsn)), opts...)
240+
241+
for parserIdx := range dsnParsers {
242+
if parser := dsnParsers[parserIdx]; parser != nil {
243+
optsFromParser, err := parser(dsn)
244+
if err != nil {
245+
return nil, xerrors.WithStackTrace(fmt.Errorf("data source name '%s' wrong: %w", dsn, err))
246+
}
247+
opts = append(opts, optsFromParser...)
248+
}
249+
}
250+
251+
d, err := newConnectionFromOptions(ctx, opts...)
244252
if err != nil {
245253
return nil, xerrors.WithStackTrace(err)
246254
}

‎dsn.go‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
package ydb
2+
3+
import (
4+
"errors"
5+
"fmt"
6+
"regexp"
7+
"strings"
8+
9+
"github.com/ydb-platform/ydb-go-sdk/v3/balancers"
10+
"github.com/ydb-platform/ydb-go-sdk/v3/credentials"
11+
"github.com/ydb-platform/ydb-go-sdk/v3/internal/bind"
12+
"github.com/ydb-platform/ydb-go-sdk/v3/internal/dsn"
13+
"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
14+
"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsql"
15+
)
16+
17+
const tablePathPrefixTransformer = "table_path_prefix"
18+
19+
var dsnParsers = []func(dsn string) (opts []Option, _ error){
20+
func(dsn string) ([]Option, error) {
21+
opts, err := parseConnectionString(dsn)
22+
if err != nil {
23+
return nil, xerrors.WithStackTrace(err)
24+
}
25+
26+
return opts, nil
27+
},
28+
}
29+
30+
// RegisterDsnParser registers DSN parser for ydb.Open and sql.Open driver constructors
31+
//
32+
// Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
33+
func RegisterDsnParser(parser func(dsn string) (opts []Option, _ error)) (registrationID int) {
34+
dsnParsers = append(dsnParsers, parser)
35+
36+
return len(dsnParsers) - 1
37+
}
38+
39+
// UnregisterDsnParser unregisters DSN parser by key
40+
//
41+
// Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
42+
func UnregisterDsnParser(registrationID int) {
43+
dsnParsers[registrationID] = nil
44+
}
45+
46+
func parseConnectionString(dataSourceName string) (opts []Option, _ error) {
47+
info, err := dsn.Parse(dataSourceName)
48+
if err != nil {
49+
return nil, xerrors.WithStackTrace(err)
50+
}
51+
opts = append(opts, With(info.Options...))
52+
if token := info.Params.Get("token"); token != "" {
53+
opts = append(opts, WithCredentials(credentials.NewAccessTokenCredentials(token)))
54+
}
55+
if balancer := info.Params.Get("go_balancer"); balancer != "" {
56+
opts = append(opts, WithBalancer(balancers.FromConfig(balancer)))
57+
} else if balancer := info.Params.Get("balancer"); balancer != "" {
58+
opts = append(opts, WithBalancer(balancers.FromConfig(balancer)))
59+
}
60+
if queryMode := info.Params.Get("go_query_mode"); queryMode != "" {
61+
mode := xsql.QueryModeFromString(queryMode)
62+
if mode == xsql.UnknownQueryMode {
63+
return nil, xerrors.WithStackTrace(fmt.Errorf("unknown query mode: %s", queryMode))
64+
}
65+
opts = append(opts, withConnectorOptions(xsql.WithDefaultQueryMode(mode)))
66+
} else if queryMode := info.Params.Get("query_mode"); queryMode != "" {
67+
mode := xsql.QueryModeFromString(queryMode)
68+
if mode == xsql.UnknownQueryMode {
69+
return nil, xerrors.WithStackTrace(fmt.Errorf("unknown query mode: %s", queryMode))
70+
}
71+
opts = append(opts, withConnectorOptions(xsql.WithDefaultQueryMode(mode)))
72+
}
73+
if fakeTx := info.Params.Get("go_fake_tx"); fakeTx != "" {
74+
for _, queryMode := range strings.Split(fakeTx, ",") {
75+
mode := xsql.QueryModeFromString(queryMode)
76+
if mode == xsql.UnknownQueryMode {
77+
return nil, xerrors.WithStackTrace(fmt.Errorf("unknown query mode: %s", queryMode))
78+
}
79+
opts = append(opts, withConnectorOptions(xsql.WithFakeTx(mode)))
80+
}
81+
}
82+
if info.Params.Has("go_query_bind") {
83+
var binders []xsql.ConnectorOption
84+
queryTransformers := strings.Split(info.Params.Get("go_query_bind"), ",")
85+
for _, transformer := range queryTransformers {
86+
switch transformer {
87+
case "declare":
88+
binders = append(binders, xsql.WithQueryBind(bind.AutoDeclare{}))
89+
case "positional":
90+
binders = append(binders, xsql.WithQueryBind(bind.PositionalArgs{}))
91+
case "numeric":
92+
binders = append(binders, xsql.WithQueryBind(bind.NumericArgs{}))
93+
default:
94+
if strings.HasPrefix(transformer, tablePathPrefixTransformer) {
95+
prefix, err := extractTablePathPrefixFromBinderName(transformer)
96+
if err != nil {
97+
return nil, xerrors.WithStackTrace(err)
98+
}
99+
binders = append(binders, xsql.WithTablePathPrefix(prefix))
100+
} else {
101+
return nil, xerrors.WithStackTrace(
102+
fmt.Errorf("unknown query rewriter: %s", transformer),
103+
)
104+
}
105+
}
106+
}
107+
opts = append(opts, withConnectorOptions(binders...))
108+
}
109+
110+
return opts, nil
111+
}
112+
113+
var (
114+
tablePathPrefixRe = regexp.MustCompile(tablePathPrefixTransformer + "\\((.*)\\)")
115+
errWrongTablePathPrefix = errors.New("wrong '" + tablePathPrefixTransformer + "' query transformer")
116+
)
117+
118+
func extractTablePathPrefixFromBinderName(binderName string) (string, error) {
119+
ss := tablePathPrefixRe.FindAllStringSubmatch(binderName, -1)
120+
if len(ss) != 1 || len(ss[0]) != 2 || ss[0][1] == "" {
121+
return "", xerrors.WithStackTrace(fmt.Errorf("%w: %s", errWrongTablePathPrefix, binderName))
122+
}
123+
124+
return ss[0][1], nil
125+
}

‎internal/xsql/dsn_test.go‎ renamed to ‎dsn_test.go‎

Lines changed: 39 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,19 @@
1-
package xsql
1+
package ydb
22

33
import (
4+
"context"
45
"testing"
56

67
"github.com/stretchr/testify/require"
78

89
"github.com/ydb-platform/ydb-go-sdk/v3/config"
910
"github.com/ydb-platform/ydb-go-sdk/v3/internal/bind"
11+
"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsql"
1012
)
1113

1214
func TestParse(t *testing.T) {
13-
newConnector := func(opts ...ConnectorOption) *Connector {
14-
c := &Connector{}
15+
newConnector := func(opts ...xsql.ConnectorOption) *xsql.Connector {
16+
c := &xsql.Connector{}
1517
for _, opt := range opts {
1618
if opt != nil {
1719
if err := opt.Apply(c); err != nil {
@@ -30,7 +32,7 @@ func TestParse(t *testing.T) {
3032
for _, tt := range []struct {
3133
dsn string
3234
opts []config.Option
33-
connectorOpts []ConnectorOption
35+
connectorOpts []xsql.ConnectorOption
3436
err error
3537
}{
3638
{
@@ -40,9 +42,9 @@ func TestParse(t *testing.T) {
4042
config.WithEndpoint("localhost:2135"),
4143
config.WithDatabase("/local"),
4244
},
43-
connectorOpts: []ConnectorOption{
44-
WithFakeTx(ScriptingQueryMode),
45-
WithFakeTx(SchemeQueryMode),
45+
connectorOpts: []xsql.ConnectorOption{
46+
xsql.WithFakeTx(xsql.ScriptingQueryMode),
47+
xsql.WithFakeTx(xsql.SchemeQueryMode),
4648
},
4749
err: nil,
4850
},
@@ -73,8 +75,8 @@ func TestParse(t *testing.T) {
7375
config.WithEndpoint("localhost:2135"),
7476
config.WithDatabase("/local"),
7577
},
76-
connectorOpts: []ConnectorOption{
77-
WithDefaultQueryMode(ScriptingQueryMode),
78+
connectorOpts: []xsql.ConnectorOption{
79+
xsql.WithDefaultQueryMode(xsql.ScriptingQueryMode),
7880
},
7981
err: nil,
8082
},
@@ -85,9 +87,9 @@ func TestParse(t *testing.T) {
8587
config.WithEndpoint("localhost:2135"),
8688
config.WithDatabase("/local"),
8789
},
88-
connectorOpts: []ConnectorOption{
89-
WithDefaultQueryMode(ScriptingQueryMode),
90-
WithTablePathPrefix("path/to/tables"),
90+
connectorOpts: []xsql.ConnectorOption{
91+
xsql.WithDefaultQueryMode(xsql.ScriptingQueryMode),
92+
xsql.WithTablePathPrefix("path/to/tables"),
9193
},
9294
err: nil,
9395
},
@@ -98,10 +100,10 @@ func TestParse(t *testing.T) {
98100
config.WithEndpoint("localhost:2135"),
99101
config.WithDatabase("/local"),
100102
},
101-
connectorOpts: []ConnectorOption{
102-
WithDefaultQueryMode(ScriptingQueryMode),
103-
WithTablePathPrefix("path/to/tables"),
104-
WithQueryBind(bind.NumericArgs{}),
103+
connectorOpts: []xsql.ConnectorOption{
104+
xsql.WithDefaultQueryMode(xsql.ScriptingQueryMode),
105+
xsql.WithTablePathPrefix("path/to/tables"),
106+
xsql.WithQueryBind(bind.NumericArgs{}),
105107
},
106108
err: nil,
107109
},
@@ -112,10 +114,10 @@ func TestParse(t *testing.T) {
112114
config.WithEndpoint("localhost:2135"),
113115
config.WithDatabase("/local"),
114116
},
115-
connectorOpts: []ConnectorOption{
116-
WithDefaultQueryMode(ScriptingQueryMode),
117-
WithTablePathPrefix("path/to/tables"),
118-
WithQueryBind(bind.PositionalArgs{}),
117+
connectorOpts: []xsql.ConnectorOption{
118+
xsql.WithDefaultQueryMode(xsql.ScriptingQueryMode),
119+
xsql.WithTablePathPrefix("path/to/tables"),
120+
xsql.WithQueryBind(bind.PositionalArgs{}),
119121
},
120122
err: nil,
121123
},
@@ -126,10 +128,10 @@ func TestParse(t *testing.T) {
126128
config.WithEndpoint("localhost:2135"),
127129
config.WithDatabase("/local"),
128130
},
129-
connectorOpts: []ConnectorOption{
130-
WithDefaultQueryMode(ScriptingQueryMode),
131-
WithTablePathPrefix("path/to/tables"),
132-
WithQueryBind(bind.AutoDeclare{}),
131+
connectorOpts: []xsql.ConnectorOption{
132+
xsql.WithDefaultQueryMode(xsql.ScriptingQueryMode),
133+
xsql.WithTablePathPrefix("path/to/tables"),
134+
xsql.WithQueryBind(bind.AutoDeclare{}),
133135
},
134136
err: nil,
135137
},
@@ -140,9 +142,9 @@ func TestParse(t *testing.T) {
140142
config.WithEndpoint("localhost:2135"),
141143
config.WithDatabase("/local"),
142144
},
143-
connectorOpts: []ConnectorOption{
144-
WithDefaultQueryMode(ScriptingQueryMode),
145-
WithTablePathPrefix("path/to/tables"),
145+
connectorOpts: []xsql.ConnectorOption{
146+
xsql.WithDefaultQueryMode(xsql.ScriptingQueryMode),
147+
xsql.WithTablePathPrefix("path/to/tables"),
146148
},
147149
err: nil,
148150
},
@@ -153,23 +155,25 @@ func TestParse(t *testing.T) {
153155
config.WithEndpoint("localhost:2135"),
154156
config.WithDatabase("/local"),
155157
},
156-
connectorOpts: []ConnectorOption{
157-
WithDefaultQueryMode(ScriptingQueryMode),
158-
WithTablePathPrefix("path/to/tables"),
159-
WithQueryBind(bind.PositionalArgs{}),
160-
WithQueryBind(bind.AutoDeclare{}),
158+
connectorOpts: []xsql.ConnectorOption{
159+
xsql.WithDefaultQueryMode(xsql.ScriptingQueryMode),
160+
xsql.WithTablePathPrefix("path/to/tables"),
161+
xsql.WithQueryBind(bind.PositionalArgs{}),
162+
xsql.WithQueryBind(bind.AutoDeclare{}),
161163
},
162164
err: nil,
163165
},
164166
} {
165167
t.Run("", func(t *testing.T) {
166-
opts, connectorOpts, err := Parse(tt.dsn)
168+
opts, err := parseConnectionString(tt.dsn)
167169
if tt.err != nil {
168170
require.ErrorIs(t, err, tt.err)
169171
} else {
170172
require.NoError(t, err)
171-
require.Equal(t, newConnector(tt.connectorOpts...), newConnector(connectorOpts...))
172-
compareConfigs(t, config.New(tt.opts...), config.New(opts...))
173+
d, err := newConnectionFromOptions(context.Background(), opts...)
174+
require.NoError(t, err)
175+
require.Equal(t, newConnector(tt.connectorOpts...), newConnector(d.databaseSQLOptions...))
176+
compareConfigs(t, config.New(tt.opts...), d.config)
173177
}
174178
})
175179
}

0 commit comments

Comments
 (0)