Merge branch 'master' into transactions-topic-query

Author: rekby

.github/workflows/examples.yml

@@ -33,6 +33,7 @@ jobs:
         options: '-h localhost'
     env:
       OS: ubuntu-latest
+      GO: 1.22.x
       YDB_CONNECTION_STRING: grpc://localhost:2136/local
       YDB_ANONYMOUS_CREDENTIALS: 1
       YDB_VERSION: ${{ matrix.ydb-version }}

CHANGELOG.md

@@ -1,4 +1,8 @@
 * Added traces and logs for read messages from topic within transaction
+* Removed experimental method `query.Result.Err()`
+* Added the finishing reading the grpc stream on `query.Result.Close()` call
+* Renamed experimental method `query.Result.Range()` to `query.Result.ResultSets()`
+* Renamed experimental method `query.ResultSet.Range()` to `query.ResultSet.Rows()`
 * Removed support of `go1.20`
 * Added PopMessages from topic within transaction
 

example_test.go

@@ -77,7 +77,7 @@ func Example_query() {
 				}
 			}
 
-			return res.Err() // return finally result error for auto-retry with driver
+			return nil
 		},
 		query.WithIdempotent(),
 	)

examples/basic/native/query/series.go

@@ -41,7 +41,7 @@ func read(ctx context.Context, c query.Client, prefix string) error {
 				resultSet, err := result.NextResultSet(ctx)
 				if err != nil {
 					if errors.Is(err, io.EOF) {
-						return result.Err()
+						break
 					}
 
 					return err
@@ -50,7 +50,7 @@ func read(ctx context.Context, c query.Client, prefix string) error {
 					row, err := resultSet.NextRow(ctx)
 					if err != nil {
 						if errors.Is(err, io.EOF) {
-							return result.Err()
+							break
 						}
 
 						return err
@@ -68,6 +68,8 @@ func read(ctx context.Context, c query.Client, prefix string) error {
 					log.Printf("%+v", info)
 				}
 			}
+
+			return nil
 		},
 	)
 }

examples/go.mod

@@ -1,6 +1,6 @@
 module examples
 
-go 1.21
+go 1.22
 
 require (
 	github.com/golang-jwt/jwt/v4 v4.5.0

internal/query/client.go

@@ -243,10 +243,6 @@ func clientExecute(ctx context.Context,
 			return xerrors.WithStackTrace(err)
 		}
 
-		if err = r.Err(); err != nil {
-			return xerrors.WithStackTrace(err)
-		}
-
 		return nil
 	})
 	if err != nil {

internal/query/client_test.go

@@ -444,7 +444,6 @@ func TestClient(t *testing.T) {
 				require.ErrorIs(t, err, io.EOF)
 				require.Nil(t, r3)
 			}
-			require.NoError(t, r.Err())
 		})
 	})
 }

internal/query/errors.go

@@ -10,7 +10,6 @@ var (
 	ErrTransactionRollingBack  = xerrors.Wrap(errors.New("ydb: the transaction is rolling back"))
 	ErrNotImplemented          = errors.New("not implemented yet")
 	errWrongNextResultSetIndex = errors.New("wrong result set index")
-	errClosedResult            = errors.New("result closed early")
 	errWrongResultSetIndex     = errors.New("critical violation of the logic - wrong result set index")
 	errMoreThanOneRow          = errors.New("unexpected more than one row in result set")
 	errMoreThanOneResultSet    = errors.New("unexpected more than one result set")

internal/query/execute_query.go

@@ -52,25 +52,25 @@ func queryFromText(
 	return content
 }
 
-func execute(ctx context.Context, s *Session, c Ydb_Query_V1.QueryServiceClient, q string, cfg executeConfig) (
+func execute(
+	ctx context.Context, s *Session, c Ydb_Query_V1.QueryServiceClient, q string, cfg executeConfig, opts ...resultOption,
+) (
 	_ *Transaction, _ *result, finalErr error,
 ) {
 	a := allocator.New()
 	defer a.Free()
 
 	request, callOptions := executeQueryRequest(a, s.id, q, cfg)
 
-	executeCtx, cancelExecute := xcontext.WithCancel(xcontext.ValueOnly(ctx))
+	executeCtx := xcontext.ValueOnly(ctx)
 
 	stream, err := c.ExecuteQuery(executeCtx, request, callOptions...)
 	if err != nil {
 		return nil, nil, xerrors.WithStackTrace(err)
 	}
 
-	r, txID, err := newResult(ctx, stream, s.cfg.Trace(), cancelExecute)
+	r, txID, err := newResult(ctx, stream, append(opts, withTrace(s.cfg.Trace()))...)
 	if err != nil {
-		cancelExecute()
-
 		return nil, nil, xerrors.WithStackTrace(err)
 	}
 

internal/query/execute_query_test.go

@@ -457,10 +457,8 @@ func TestExecute(t *testing.T) {
 		{
 			t.Log("nextResultSet")
 			_, err := r.nextResultSet(context.Background())
-			require.ErrorIs(t, err, errClosedResult)
+			require.ErrorIs(t, err, io.EOF)
 		}
-		t.Log("check final error")
-		require.NoError(t, r.Err())
 	})
 	t.Run("TransportError", func(t *testing.T) {
 		t.Run("OnCall", func(t *testing.T) {
@@ -621,9 +619,6 @@ func TestExecute(t *testing.T) {
 					require.True(t, xerrors.IsTransportError(err, grpcCodes.Unavailable))
 				}
 			}
-			t.Log("check final error")
-			require.Error(t, r.Err())
-			require.True(t, xerrors.IsTransportError(r.Err(), grpcCodes.Unavailable))
 		})
 	})
 	t.Run("OperationError", func(t *testing.T) {
@@ -749,9 +744,6 @@ func TestExecute(t *testing.T) {
 					require.True(t, xerrors.IsOperationError(err, Ydb.StatusIds_UNAVAILABLE))
 				}
 			}
-			t.Log("check final error")
-			require.Error(t, r.Err())
-			require.True(t, xerrors.IsOperationError(r.Err(), Ydb.StatusIds_UNAVAILABLE))
 		})
 	})
 }