Streaming query result implementation

Author: alex268

jdbc/src/main/java/tech/ydb/jdbc/YdbStatement.java

@@ -33,7 +33,7 @@ public interface YdbStatement extends Statement {
      */
     YdbResultSet executeExplainQuery(String sql) throws SQLException;
 
-    YdbValidator getValidator() throws SQLException;
+    YdbValidator getValidator();
 
     @Override
     YdbResultSet executeQuery(String sql) throws SQLException;

jdbc/src/main/java/tech/ydb/jdbc/common/ColumnInfo.java

@@ -1,5 +1,6 @@
 package tech.ydb.jdbc.common;
 
+import tech.ydb.table.result.ResultSetReader;
 import tech.ydb.table.values.PrimitiveType;
 import tech.ydb.table.values.Type;
 
@@ -66,4 +67,13 @@ public MappingGetters.SqlType getSqlType() {
     public MappingGetters.Getters getGetters() {
         return this.getters;
     }
+
+    public static ColumnInfo[] fromResultSetReader(ResultSetReader rsr) {
+        ColumnInfo[] columns = new ColumnInfo[rsr.getColumnCount()];
+        for (int idx = 0; idx < rsr.getColumnCount(); idx += 1) {
+            columns[idx] = new ColumnInfo(rsr.getColumnName(idx), rsr.getColumnType(idx));
+        }
+        return columns;
+    }
+
 }

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

@@ -4,22 +4,18 @@
 import java.sql.SQLException;
 import java.sql.SQLFeatureNotSupportedException;
 import java.time.Duration;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import tech.ydb.common.transaction.TxMode;
 import tech.ydb.core.Issue;
 import tech.ydb.core.Result;
 import tech.ydb.core.UnexpectedResultException;
 import tech.ydb.jdbc.YdbConst;
-import tech.ydb.jdbc.YdbResultSet;
 import tech.ydb.jdbc.YdbStatement;
 import tech.ydb.jdbc.exception.ExceptionFactory;
 import tech.ydb.jdbc.impl.YdbQueryResult;
-import tech.ydb.jdbc.impl.YdbStaticResultSet;
 import tech.ydb.jdbc.query.QueryType;
 import tech.ydb.jdbc.query.YdbQuery;
 import tech.ydb.query.QueryClient;
@@ -32,7 +28,6 @@
 import tech.ydb.query.settings.ExecuteQuerySettings;
 import tech.ydb.query.settings.QueryExecMode;
 import tech.ydb.query.settings.RollbackTransactionSettings;
-import tech.ydb.query.tools.QueryReader;
 import tech.ydb.table.query.Params;
 
 /**
@@ -216,20 +211,17 @@ public YdbQueryResult executeDataQuery(
             tx = createNewQuerySession(validator).createNewTransaction(txMode);
         }
 
-        try {
-            QueryReader result = validator.call(QueryType.DATA_QUERY + " >>\n" + yql,
-                    () -> QueryReader.readFrom(tx.createQuery(yql, isAutoCommit, params, settings))
-            );
-            validator.addStatusIssues(result.getIssueList());
+        String msg = "STREAM_QUERY >>\n" + yql;
+        QueryStream stream = tx.createQuery(yql, isAutoCommit, params, settings);
+        StreamQueryResult result = new StreamQueryResult(msg, statement, query, stream::cancel);
 
-            List<YdbResultSet> rsList = new ArrayList<>();
-            result.forEach(rsr -> rsList.add(new YdbStaticResultSet(statement, rsr)));
-            return new StaticQueryResult(query, rsList);
-        } finally {
+        result.start(stream).thenRun(() -> {
             if (!tx.isActive()) {
                 cleanTx();
             }
-        }
+        });
+
+        return result;
     }
 
     @Override