Fixed inconsistent transaction state

Author: alex268

jdbc/src/main/java/tech/ydb/jdbc/context/QueryServiceExecutor.java

@@ -212,16 +212,17 @@ public YdbQueryResult executeDataQuery(
         }
 
         String msg = "STREAM_QUERY >>\n" + yql;
-        QueryStream stream = tx.createQuery(yql, isAutoCommit, params, settings);
-        StreamQueryResult result = new StreamQueryResult(msg, statement, query, stream::cancel);
-
-        result.start(stream).thenRun(() -> {
+        try {
+            return validator.call(msg, () -> {
+                QueryStream stream = tx.createQuery(yql, isAutoCommit, params, settings);
+                StreamQueryResult result = new StreamQueryResult(msg, statement, query, stream::cancel);
+                return result.execute(stream);
+            });
+        } finally {
             if (!tx.isActive()) {
                 cleanTx();
             }
-        });
-
-        return result;
+        }
     }
 
     @Override

jdbc/src/main/java/tech/ydb/jdbc/context/StreamQueryResult.java

@@ -40,7 +40,9 @@ public class StreamQueryResult implements YdbQueryResult {
     private final String msg;
     private final YdbStatement statement;
     private final Runnable stopRunnable;
-    private final CompletableFuture<Status> statusFuture = new CompletableFuture<>();
+
+    private final CompletableFuture<Status> finishFuture = new CompletableFuture<>();
+    private final CompletableFuture<Result<StreamQueryResult>> startFuture = new CompletableFuture<>();
 
     private final int[] resultIndexes;
     private final List<CompletableFuture<Result<LazyResultSet>>> resultFutures = new ArrayList<>();
@@ -72,31 +74,34 @@ public StreamQueryResult(String msg, YdbStatement statement, YdbQuery query, Run
         }
     }
 
-    public CompletableFuture<Status> start(QueryStream stream) {
-        return stream
-                .execute(new QueryPartsHandler())
+    public CompletableFuture<Result<StreamQueryResult>> execute(QueryStream stream) {
+        stream.execute(new QueryPartsHandler())
                 .thenApply(Result::getStatus)
-                .whenComplete(this::onStreamFinish);
+                .whenComplete(this::onStreamFinished);
+        return startFuture;
     }
 
-    private void onStreamFinish(Status status, Throwable th) {
+    private void onStreamFinished(Status status, Throwable th) {
         if (th != null) {
-            statusFuture.completeExceptionally(th);
+            finishFuture.completeExceptionally(th);
             for (CompletableFuture<Result<LazyResultSet>> future: resultFutures) {
                 future.completeExceptionally(th);
             }
+            startFuture.completeExceptionally(th);
         }
 
         if (status != null) {
-            statusFuture.complete(status);
+            finishFuture.complete(status);
             if (status.isSuccess()) {
                 for (CompletableFuture<Result<LazyResultSet>> future: resultFutures) {
                     future.complete(Result.success(new LazyResultSet(statement, new ColumnInfo[0]), status));
                 }
+                startFuture.complete(Result.success(this));
             } else {
                 for (CompletableFuture<Result<LazyResultSet>> future: resultFutures) {
                     future.complete(Result.fail(status));
                 }
+                startFuture.complete(Result.fail(status));
             }
         }
 
@@ -112,7 +117,7 @@ private void onStreamFinish(Status status, Throwable th) {
 
     @Override
     public void close() throws SQLException {
-        Status status = statusFuture.join();
+        Status status = finishFuture.join();
         if (!status.isSuccess()) {
             throw ExceptionFactory.createException("Cannot execute '" + msg + "' with " + status,
                     new UnexpectedResultException("Unexpected status", status)
@@ -218,11 +223,13 @@ private void onResultSet(int index, ResultSetReader rsr) {
     private class QueryPartsHandler implements QueryStream.PartsHandler {
         @Override
         public void onIssues(Issue[] issues) {
+            startFuture.complete(Result.success(StreamQueryResult.this));
             statement.getValidator().addStatusIssues(Arrays.asList(issues));
         }
 
         @Override
         public void onNextPart(QueryResultPart part) {
+            startFuture.complete(Result.success(StreamQueryResult.this));
             onResultSet((int) part.getResultSetIndex(), part.getResultSetReader());
         }
     }