Added executeScanQueryRaw method

Author: alex268

table/src/main/java/tech/ydb/table/Session.java

@@ -9,6 +9,7 @@
 import tech.ydb.core.Result;
 import tech.ydb.core.Status;
 import tech.ydb.core.grpc.GrpcReadStream;
+import tech.ydb.proto.ValueProtos;
 import tech.ydb.table.description.TableDescription;
 import tech.ydb.table.description.TableOptionDescription;
 import tech.ydb.table.query.BulkUpsertData;
@@ -19,6 +20,7 @@
 import tech.ydb.table.query.ReadRowsResult;
 import tech.ydb.table.query.ReadTablePart;
 import tech.ydb.table.result.ResultSetReader;
+import tech.ydb.table.result.impl.ProtoValueReaders;
 import tech.ydb.table.settings.AlterTableSettings;
 import tech.ydb.table.settings.BeginTxSettings;
 import tech.ydb.table.settings.BulkUpsertSettings;
@@ -147,7 +149,24 @@ default CompletableFuture<Result<TableTransaction>> beginTransaction(TxMode txMo
 
     GrpcReadStream<ReadTablePart> executeReadTable(String tablePath, ReadTableSettings settings);
 
-    GrpcReadStream<ResultSetReader> executeScanQuery(String query, Params params, ExecuteScanQuerySettings settings);
+    GrpcReadStream<ValueProtos.ResultSet> executeScanQueryRaw(String query, Params params,
+            ExecuteScanQuerySettings settings);
+
+    default GrpcReadStream<ResultSetReader> executeScanQuery(String query, Params params,
+            ExecuteScanQuerySettings settings) {
+        GrpcReadStream<ValueProtos.ResultSet> stream = executeScanQueryRaw(query, params, settings);
+        return new GrpcReadStream<ResultSetReader>() {
+            @Override
+            public CompletableFuture<Status> start(GrpcReadStream.Observer<ResultSetReader> observer) {
+                return stream.start(part -> observer.onNext(ProtoValueReaders.forResultSet(part)));
+            }
+
+            @Override
+            public void cancel() {
+                stream.cancel();
+            }
+        };
+    }
 
     @Deprecated
     default GrpcReadStream<ResultSetReader> readTable(String tablePath, ReadTableSettings settings) {

table/src/main/java/tech/ydb/table/impl/BaseSession.java

@@ -64,8 +64,6 @@
 import tech.ydb.table.query.Params;
 import tech.ydb.table.query.ReadRowsResult;
 import tech.ydb.table.query.ReadTablePart;
-import tech.ydb.table.result.ResultSetReader;
-import tech.ydb.table.result.impl.ProtoValueReaders;
 import tech.ydb.table.rpc.TableRpc;
 import tech.ydb.table.settings.AlterTableSettings;
 import tech.ydb.table.settings.AutoPartitioningPolicy;
@@ -1315,7 +1313,7 @@ ReadTablePart readValue(YdbTable.ReadTableResponse message) {
     }
 
     @Override
-    public GrpcReadStream<ResultSetReader> executeScanQuery(String query, Params params,
+    public GrpcReadStream<ValueProtos.ResultSet> executeScanQueryRaw(String query, Params params,
                                                             ExecuteScanQuerySettings settings) {
         YdbTable.ExecuteScanQueryRequest req = YdbTable.ExecuteScanQueryRequest.newBuilder()
                 .setQuery(YdbTable.Query.newBuilder().setYqlText(query))
@@ -1330,7 +1328,7 @@ public GrpcReadStream<ResultSetReader> executeScanQuery(String query, Params par
         }
 
         GrpcReadStream<YdbTable.ExecuteScanQueryPartialResponse> origin = rpc.streamExecuteScanQuery(req, opts.build());
-        return new ProxyStream<YdbTable.ExecuteScanQueryPartialResponse, ResultSetReader>(origin) {
+        return new ProxyStream<YdbTable.ExecuteScanQueryPartialResponse, ValueProtos.ResultSet>(origin) {
             @Override
             StatusIds.StatusCode readStatusCode(YdbTable.ExecuteScanQueryPartialResponse message) {
                 return message.getStatus();
@@ -1342,8 +1340,8 @@ List<YdbIssueMessage.IssueMessage> readIssues(YdbTable.ExecuteScanQueryPartialRe
             }
 
             @Override
-            ResultSetReader readValue(YdbTable.ExecuteScanQueryPartialResponse message) {
-                return ProtoValueReaders.forResultSet(message.getResult().getResultSet());
+            ValueProtos.ResultSet readValue(YdbTable.ExecuteScanQueryPartialResponse message) {
+                return message.getResult().getResultSet();
             }
         };
     }

table/src/test/java/tech/ydb/table/SessionStub.java

@@ -8,6 +8,7 @@
 import tech.ydb.core.Status;
 import tech.ydb.core.grpc.GrpcReadStream;
 import tech.ydb.core.utils.FutureTools;
+import tech.ydb.proto.ValueProtos;
 import tech.ydb.table.description.TableDescription;
 import tech.ydb.table.description.TableOptionDescription;
 import tech.ydb.table.query.BulkUpsertData;
@@ -17,7 +18,6 @@
 import tech.ydb.table.query.Params;
 import tech.ydb.table.query.ReadRowsResult;
 import tech.ydb.table.query.ReadTablePart;
-import tech.ydb.table.result.ResultSetReader;
 import tech.ydb.table.settings.AlterTableSettings;
 import tech.ydb.table.settings.BeginTxSettings;
 import tech.ydb.table.settings.BulkUpsertSettings;
@@ -158,7 +158,7 @@ public GrpcReadStream<ReadTablePart> executeReadTable(String tablePath, ReadTabl
     }
 
     @Override
-    public GrpcReadStream<ResultSetReader> executeScanQuery(String query, Params params, ExecuteScanQuerySettings settings) {
+    public GrpcReadStream<ValueProtos.ResultSet> executeScanQueryRaw(String query, Params params, ExecuteScanQuerySettings settings) {
         throw new UnsupportedOperationException("executeScanQuery not implemented");
     }