splitted trace.BulkUpsert{Start,Done}Info for client and session

Author: asmyasnikov

internal/table/client.go

@@ -29,7 +29,6 @@ func New(ctx context.Context, cc grpc.ClientConnInterface, config *config.Config
 	return &Client{
 		clock:  config.Clock(),
 		config: config,
-		client: Ydb_Table_V1.NewTableServiceClient(cc),
 		cc:     cc,
 		build: func(ctx context.Context) (s *session, err error) {
 			return newSession(ctx, cc, config)
@@ -89,7 +88,6 @@ func New(ctx context.Context, cc grpc.ClientConnInterface, config *config.Config
 type Client struct {
 	// read-only fields
 	config *config.Config
-	client Ydb_Table_V1.TableServiceClient
 	build  sessionBuilder
 	cc     grpc.ClientConnInterface
 	clock  clockwork.Clock
@@ -309,10 +307,12 @@ func (c *Client) BulkUpsert(
 		return xerrors.WithStackTrace(err)
 	}
 
+	client := Ydb_Table_V1.NewTableServiceClient(c.cc)
+
 	err = retry.Retry(ctx,
 		func(ctx context.Context) (err error) {
 			attempts++
-			_, err = c.client.BulkUpsert(ctx, request)
+			_, err = client.BulkUpsert(ctx, request)
 
 			return err
 		},

internal/table/session.go

@@ -1265,7 +1265,7 @@ func (s *session) BulkUpsert(ctx context.Context, table string, rows value.Value
 	)
 	defer func() {
 		defer a.Free()
-		onDone(err, 1)
+		onDone(err)
 	}()
 
 	for _, opt := range opts {

table/table.go

@@ -743,7 +743,9 @@ func BulkUpsertDataArrow(data []byte, opts ...arrowFormatOption) bulkUpsertArrow
 
 type arrowSchemaOption []byte
 
-func (schema arrowSchemaOption) applyArrowFormatOption(dataFormat *Ydb_Table.BulkUpsertRequest_ArrowBatchSettings) error {
+func (schema arrowSchemaOption) applyArrowFormatOption(
+	dataFormat *Ydb_Table.BulkUpsertRequest_ArrowBatchSettings,
+) error {
 	dataFormat.ArrowBatchSettings.Schema = schema
 
 	return nil

trace/table.go

@@ -35,7 +35,7 @@ type (
 		OnSessionKeepAlive func(TableKeepAliveStartInfo) func(TableKeepAliveDoneInfo)
 		// Query events
 		// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
-		OnSessionBulkUpsert func(TableBulkUpsertStartInfo) func(TableBulkUpsertDoneInfo)
+		OnSessionBulkUpsert func(TableSessionBulkUpsertStartInfo) func(TableSessionBulkUpsertDoneInfo)
 		// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
 		OnSessionQueryPrepare func(TablePrepareDataQueryStartInfo) func(TablePrepareDataQueryDoneInfo)
 		// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
@@ -151,6 +151,19 @@ type (
 		Attempts int
 	}
 	// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
+	TableSessionBulkUpsertStartInfo struct {
+		// Context make available context in trace callback function.
+		// Pointer to context provide replacement of context in trace callback function.
+		// Warning: concurrent access to pointer on client side must be excluded.
+		// Safe replacement of context are provided only inside callback function
+		Context *context.Context
+		Call    call
+	}
+	// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
+	TableSessionBulkUpsertDoneInfo struct {
+		Error error
+	}
+	// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
 	TableSessionDeleteStartInfo struct {
 		// Context make available context in trace callback function.
 		// Pointer to context provide replacement of context in trace callback function.