Merge branch 'master' into params-dict

.github/workflows/slo.yml

@@ -39,7 +39,7 @@ jobs:
           s3_images_folder: ${{ vars.SLO_S3_IMAGES_FOLDER }}
           grafana_domain: ${{ vars.SLO_GRAFANA_DOMAIN }}
           grafana_dashboard: ${{ vars.SLO_GRAFANA_DASHBOARD }}
-          ydb_version: 'newest'
+          ydb_version: '24.1.7'
           timeBetweenPhases: 30
           shutdownTime: 30
 

CHANGELOG.md

@@ -1,4 +1,19 @@
 * Added `Dict` support for `ydb.ParamsBuilder()`
+
+## v3.57.3
+* Added metrics over query service internals
+* Added session create and delete events into `trace.Query`
+* Moved public type `query.SessionStatus` into `internal/query` package
+
+## v3.57.2
+* Fixed cases when some option is nil
+
+## v3.57.1
+* Added logs over query service internals
+* Changed `trace.Query` events
+* Changed visibility of `query.{Do,DoTx}Options` from public to private
+
+## v3.57.0
 * Added experimental implementation of query service client
 * Fixed sometime panic on topic writer closing
 * Added experimental query parameters builder `ydb.ParamsBuilder()`

balancers/config.go

@@ -116,9 +116,9 @@ func FromConfig(config string, opts ...fromConfigOption) *balancerConfig.Config
 		b   *balancerConfig.Config
 		err error
 	)
-	for _, o := range opts {
-		if o != nil {
-			o(&h)
+	for _, opt := range opts {
+		if opt != nil {
+			opt(&h)
 		}
 	}
 

config/config.go

@@ -268,9 +268,9 @@ func ExcludeGRPCCodesForPessimization(codes ...grpcCodes.Code) Option {
 func New(opts ...Option) *Config {
 	c := defaultConfig()
 
-	for _, o := range opts {
-		if o != nil {
-			o(c)
+	for _, opt := range opts {
+		if opt != nil {
+			opt(c)
 		}
 	}
 
@@ -281,9 +281,9 @@ func New(opts ...Option) *Config {
 
 // With makes copy of current Config with specified options
 func (c *Config) With(opts ...Option) *Config {
-	for _, o := range opts {
-		if o != nil {
-			o(c)
+	for _, opt := range opts {
+		if opt != nil {
+			opt(c)
 		}
 	}
 	c.meta = meta.New(

internal/backoff/backoff.go

@@ -86,9 +86,9 @@ func New(opts ...option) logBackoff {
 	b := logBackoff{
 		r: xrand.New(xrand.WithLock()),
 	}
-	for _, o := range opts {
-		if o != nil {
-			o(&b)
+	for _, opt := range opts {
+		if opt != nil {
+			opt(&b)
 		}
 	}
 

internal/conn/conn.go

@@ -500,9 +500,9 @@ func newConn(e endpoint.Endpoint, config Config, opts ...option) *conn {
 		done:     make(chan struct{}),
 	}
 	c.state.Store(uint32(Created))
-	for _, o := range opts {
-		if o != nil {
-			o(c)
+	for _, opt := range opts {
+		if opt != nil {
+			opt(c)
 		}
 	}
 

internal/conn/grpc_client_stream.go

@@ -135,9 +135,10 @@ func (s *grpcClientStream) wrapError(err error) error {
 		return nil
 	}
 
-	nodeErr := newConnError(s.c.endpoint.NodeID(), s.c.endpoint.Address(), err)
-
-	return xerrors.WithStackTrace(nodeErr, xerrors.WithSkipDepth(1))
+	return xerrors.WithStackTrace(
+		newConnError(s.c.endpoint.NodeID(), s.c.endpoint.Address(), err),
+		xerrors.WithSkipDepth(1),
+	)
 }
 
 func createPinger(c *conn) context.CancelFunc {

internal/coordination/config/config.go

@@ -37,9 +37,9 @@ func New(opts ...Option) Config {
 	c := Config{
 		trace: &trace.Coordination{},
 	}
-	for _, o := range opts {
-		if o != nil {
-			o(&c)
+	for _, opt := range opts {
+		if opt != nil {
+			opt(&c)
 		}
 	}
 

internal/credentials/access_error.go

@@ -94,10 +94,12 @@ func AccessError(msg string, err error, opts ...authErrorOption) error {
 	buffer.WriteString(msg)
 	buffer.WriteString(" (")
 	for i, opt := range opts {
-		if i != 0 {
-			buffer.WriteString(",")
+		if opt != nil {
+			if i != 0 {
+				buffer.WriteString(",")
+			}
+			opt.applyAuthErrorOption(buffer)
 		}
-		opt.applyAuthErrorOption(buffer)
 	}
 	buffer.WriteString("): %w")
 

internal/credentials/access_token.go

@@ -32,7 +32,9 @@ func NewAccessTokenCredentials(token string, opts ...AccessTokenCredentialsOptio
 		sourceInfo: stack.Record(1),
 	}
 	for _, opt := range opts {
-		opt.ApplyAccessTokenCredentialsOption(c)
+		if opt != nil {
+			opt.ApplyAccessTokenCredentialsOption(c)
+		}
 	}
 
 	return c

internal/credentials/anonymous.go

@@ -28,7 +28,9 @@ func NewAnonymousCredentials(opts ...AnonymousCredentialsOption) *Anonymous {
 		sourceInfo: stack.Record(1),
 	}
 	for _, opt := range opts {
-		opt.ApplyAnonymousCredentialsOption(c)
+		if opt != nil {
+			opt.ApplyAnonymousCredentialsOption(c)
+		}
 	}
 
 	return c

internal/credentials/static.go

@@ -47,7 +47,9 @@ func NewStaticCredentials(user, password, endpoint string, opts ...StaticCredent
 		sourceInfo: stack.Record(1),
 	}
 	for _, opt := range opts {
-		opt.ApplyStaticCredentialsOption(c)
+		if opt != nil {
+			opt.ApplyStaticCredentialsOption(c)
+		}
 	}
 
 	return c

internal/discovery/config/config.go

@@ -29,9 +29,9 @@ func New(opts ...Option) *Config {
 		interval: DefaultInterval,
 		trace:    &trace.Discovery{},
 	}
-	for _, o := range opts {
-		if o != nil {
-			o(c)
+	for _, opt := range opts {
+		if opt != nil {
+			opt(c)
 		}
 	}
 

internal/endpoint/endpoint.go

@@ -110,13 +110,10 @@ func (e *endpoint) LastUpdated() time.Time {
 func (e *endpoint) Touch(opts ...Option) {
 	e.mu.Lock()
 	defer e.mu.Unlock()
-	for _, o := range append(
-		[]Option{
-			withLastUpdated(time.Now()),
-		},
-		opts...,
-	) {
-		o(e)
+	for _, opt := range append([]Option{withLastUpdated(time.Now())}, opts...) {
+		if opt != nil {
+			opt(e)
+		}
 	}
 }
 
@@ -163,9 +160,9 @@ func New(address string, opts ...Option) *endpoint {
 		address:     address,
 		lastUpdated: time.Now(),
 	}
-	for _, o := range opts {
-		if o != nil {
-			o(e)
+	for _, opt := range opts {
+		if opt != nil {
+			opt(e)
 		}
 	}
 

internal/meta/meta.go

@@ -24,9 +24,9 @@ func New(
 		credentials: credentials,
 		database:    database,
 	}
-	for _, o := range opts {
-		if o != nil {
-			o(m)
+	for _, opt := range opts {
+		if opt != nil {
+			opt(m)
 		}
 	}
 

internal/meta/trace_id.go

@@ -19,7 +19,9 @@ func TraceID(ctx context.Context, opts ...func(opts *newTraceIDOpts)) (context.C
 	}
 	options := newTraceIDOpts{newRandom: uuid.NewRandom}
 	for _, opt := range opts {
-		opt(&options)
+		if opt != nil {
+			opt(&options)
+		}
 	}
 	uuid, err := options.newRandom()
 	if err != nil {

internal/params/list_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/allocator"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/value"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xtest"
 )
 
@@ -639,6 +640,39 @@ func TestList(t *testing.T) {
 				},
 			},
 		},
+		{
+			name:    xtest.CurrentFileLine(),
+			builder: Builder{}.Param("$x").List().AddItems(value.Uint64Value(123), value.Uint64Value(321)).Build(),
+			params: map[string]*Ydb.TypedValue{
+				"$x": {
+					Type: &Ydb.Type{
+						Type: &Ydb.Type_ListType{
+							ListType: &Ydb.ListType{
+								Item: &Ydb.Type{
+									Type: &Ydb.Type_TypeId{
+										TypeId: Ydb.Type_UINT64,
+									},
+								},
+							},
+						},
+					},
+					Value: &Ydb.Value{
+						Items: []*Ydb.Value{
+							{
+								Value: &Ydb.Value_Uint64Value{
+									Uint64Value: 123,
+								},
+							},
+							{
+								Value: &Ydb.Value_Uint64Value{
+									Uint64Value: 321,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
 			a := allocator.New()

internal/params/set.go

@@ -24,6 +24,12 @@ func (s *set) AddItem() *setItem {
 	}
 }
 
+func (s *set) AddItems(items ...value.Value) *set {
+	s.values = append(s.values, items...)
+
+	return s
+}
+
 func (s *set) Build() Builder {
 	s.parent.params = append(s.parent.params, &Parameter{
 		parent: s.parent,

internal/params/set_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/allocator"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/value"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xtest"
 )
 
@@ -816,6 +817,52 @@ func TestSet(t *testing.T) {
 				},
 			},
 		},
+		{
+			name:    xtest.CurrentFileLine(),
+			builder: Builder{}.Param("$x").Set().AddItems(value.Uint64Value(123), value.Uint64Value(321)).Build(),
+			params: map[string]*Ydb.TypedValue{
+				"$x": {
+					Type: &Ydb.Type{
+						Type: &Ydb.Type_DictType{
+							DictType: &Ydb.DictType{
+								Key: &Ydb.Type{
+									Type: &Ydb.Type_TypeId{
+										TypeId: Ydb.Type_UINT64,
+									},
+								},
+								Payload: &Ydb.Type{
+									Type: &Ydb.Type_VoidType{},
+								},
+							},
+						},
+					},
+					Value: &Ydb.Value{
+						Pairs: []*Ydb.ValuePair{
+							{
+								Key: &Ydb.Value{
+									Value: &Ydb.Value_Uint64Value{
+										Uint64Value: 123,
+									},
+								},
+								Payload: &Ydb.Value{
+									Value: &Ydb.Value_NullFlagValue{},
+								},
+							},
+							{
+								Key: &Ydb.Value{
+									Value: &Ydb.Value_Uint64Value{
+										Uint64Value: 321,
+									},
+								},
+								Payload: &Ydb.Value{
+									Value: &Ydb.Value_NullFlagValue{},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
 			a := allocator.New()

internal/pool/pool.go

@@ -66,7 +66,9 @@ func New[T any](
 		done: make(chan struct{}),
 	}
 	for _, opt := range opts {
-		opt(p)
+		if opt != nil {
+			opt(p)
+		}
 	}
 
 	return p
@@ -107,15 +109,15 @@ func (p *Pool[T]) try(ctx context.Context, f func(ctx context.Context, item *T)
 	return nil
 }
 
-func (p *Pool[T]) With(ctx context.Context, f func(ctx context.Context, item *T) error) error {
+func (p *Pool[T]) With(ctx context.Context, f func(ctx context.Context, item *T) error, opts ...retry.Option) error {
 	err := retry.Retry(ctx, func(ctx context.Context) error {
 		err := p.try(ctx, f)
 		if err != nil {
 			return xerrors.WithStackTrace(err)
 		}
 
 		return nil
-	})
+	}, opts...)
 	if err != nil {
 		return xerrors.WithStackTrace(err)
 	}