feat: expose query warnings

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Added method `query.WithIssuesHandler` to get query issues  
+
 ## v3.117.1
 * Fixed scan a column of type `Decimal(precision,scale)` into a struct field of type `types.Decimal{}` using `ScanStruct()`
 * Fixed race in integration test `TestTopicWriterLogMessagesWithoutData`

internal/query/client.go

@@ -357,6 +357,11 @@ func clientExec(ctx context.Context, pool sessionPool, q string, opts ...options
 		}()
 
 		err = readAll(ctx, streamResult)
+		if len(streamResult.issuesList) > 0 {
+			if issueHandler := settings.IssuesOpts(); issueHandler != nil {
+				issueHandler(streamResult.issuesList)
+			}
+		}
 		if err != nil {
 			return xerrors.WithStackTrace(err)
 		}
@@ -397,7 +402,7 @@ func clientQuery(ctx context.Context, pool sessionPool, q string, opts ...option
 ) {
 	settings := options.ExecuteSettings(opts...)
 	err = do(ctx, pool, func(ctx context.Context, s *Session) (err error) {
-		streamResult, err := s.execute(ctx, q, options.ExecuteSettings(opts...), withStreamResultTrace(s.trace))
+		streamResult, err := s.execute(ctx, q, settings, withStreamResultTrace(s.trace))
 		if err != nil {
 			return xerrors.WithStackTrace(err)
 		}
@@ -406,6 +411,11 @@ func clientQuery(ctx context.Context, pool sessionPool, q string, opts ...option
 		}()
 
 		r, err = resultToMaterializedResult(ctx, streamResult)
+		if len(streamResult.issuesList) > 0 {
+			if issueHandler := settings.IssuesOpts(); issueHandler != nil {
+				issueHandler(streamResult.issuesList)
+			}
+		}
 		if err != nil {
 			return xerrors.WithStackTrace(err)
 		}

internal/query/options/execute.go

@@ -1,6 +1,7 @@
 package options
 
 import (
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Issue"
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Query"
 	"google.golang.org/grpc"
 
@@ -42,6 +43,7 @@ type (
 		callOptions            []grpc.CallOption
 		txControl              *tx.Control
 		retryOptions           []retry.Option
+		issueCallback          func(issues []*Ydb_Issue.IssueMessage)
 		responsePartLimitBytes int64
 		label                  string
 	}
@@ -70,6 +72,9 @@ type (
 	}
 	execModeOption         = ExecMode
 	responsePartLimitBytes int64
+	issuesOption           struct {
+		callback func([]*Ydb_Issue.IssueMessage)
+	}
 )
 
 func (poolID resourcePool) applyExecuteOption(s *executeSettings) {
@@ -80,6 +85,10 @@ func (s *executeSettings) RetryOpts() []retry.Option {
 	return s.retryOptions
 }
 
+func (s *executeSettings) IssuesOpts() func([]*Ydb_Issue.IssueMessage) {
+	return s.issueCallback
+}
+
 func (s *executeSettings) StatsCallback() func(stats.QueryStats) {
 	return s.statsCallback
 }
@@ -119,6 +128,10 @@ func (mode ExecMode) applyExecuteOption(s *executeSettings) {
 	s.execMode = mode
 }
 
+func (opts issuesOption) applyExecuteOption(s *executeSettings) {
+	s.issueCallback = opts.callback
+}
+
 const (
 	ExecModeParse    = ExecMode(Ydb_Query.ExecMode_EXEC_MODE_PARSE)
 	ExecModeValidate = ExecMode(Ydb_Query.ExecMode_EXEC_MODE_VALIDATE)
@@ -244,6 +257,12 @@ func WithStatsMode(mode StatsMode, callback func(stats.QueryStats)) statsModeOpt
 	}
 }
 
+func WithIssuesHandler(callback func(issues []*Ydb_Issue.IssueMessage)) issuesOption {
+	return issuesOption{
+		callback: callback,
+	}
+}
+
 func WithCallOptions(opts ...grpc.CallOption) callOptionsOption {
 	return opts
 }

internal/query/result.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/ydb-platform/ydb-go-genproto/Ydb_Query_V1"
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Issue"
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Query"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/query/result"
@@ -38,8 +39,10 @@ type (
 		resultSetIndex int64
 		trace          *trace.Query
 		statsCallback  func(queryStats stats.QueryStats)
+		issuesCallback func(issues []*Ydb_Issue.IssueMessage)
 		onNextPartErr  []func(err error)
 		onTxMeta       []func(txMeta *Ydb_Query.TransactionMeta)
+		issuesList     []*Ydb_Issue.IssueMessage
 		closeTimeout   time.Duration
 	}
 	resultOption func(s *streamResult)
@@ -131,6 +134,7 @@ func newResult(
 		stream:         stream,
 		closer:         NewResultCloser(),
 		resultSetIndex: -1,
+		issuesList:     make([]*Ydb_Issue.IssueMessage, 0),
 	}
 
 	for _, opt := range opts {
@@ -162,6 +166,13 @@ func newResult(
 		if part.GetExecStats() != nil && r.statsCallback != nil {
 			r.statsCallback(stats.FromQueryStats(part.GetExecStats()))
 		}
+		if r.issuesCallback != nil {
+			if r.lastPart != nil {
+				r.issuesCallback(r.lastPart.GetIssues())
+			} else {
+				r.issuesCallback(make([]*Ydb_Issue.IssueMessage, 0))
+			}
+		}
 
 		return &r, nil
 	}
@@ -197,6 +208,9 @@ func (r *streamResult) nextPart(ctx context.Context) (
 		}()
 
 		part, err = nextPart(r.stream)
+		if part != nil {
+			r.issuesList = append(r.issuesList, part.GetIssues()...)
+		}
 		if err != nil {
 			for _, callback := range r.onNextPartErr {
 				callback(err)

internal/query/result/result.go

@@ -41,4 +41,8 @@ type (
 		ScanNamed(dst ...scanner.NamedDestination) error
 		ScanStruct(dst interface{}, opts ...scanner.ScanStructOption) error
 	}
+	resultOption func(s *Result)
+	Option       interface {
+		ApplyResultOption(opts *resultOption)
+	}
 )

query/execute_options.go

@@ -1,6 +1,7 @@
 package query
 
 import (
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Issue"
 	"google.golang.org/grpc"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/params"
@@ -57,6 +58,10 @@ func WithStatsMode(mode options.StatsMode, callback func(Stats)) ExecuteOption {
 	return options.WithStatsMode(mode, callback)
 }
 
+func WithIssuesHandler(callback func(issues []*Ydb_Issue.IssueMessage)) ExecuteOption {
+	return options.WithIssuesHandler(callback)
+}
+
 // WithResponsePartLimitSizeBytes limit size of each part (data portion) in stream for query service resoponse
 // it isn't limit total size of answer
 func WithResponsePartLimitSizeBytes(size int64) ExecuteOption {