feat: add row-based access of results to connection (#18)

This pull request includes new methods for acquiring query status and
"row-based" result sets.

---------

Co-authored-by: Kaviarasu <ksakthivadivel@salesforce.com>

Author: jschneidereit

README.md

@@ -75,7 +75,7 @@ properties.put("clientSecret", "${clientSecret}");
 
 The documentation for jwt authentication can be found [here][jwt flow].
 
-Instuctions to generate a private key can be found [here](#generating-a-private-key-for-jwt-authentication)
+Instructions to generate a private key can be found [here](#generating-a-private-key-for-jwt-authentication)
 
 ```java
 Properties properties = new Properties();

jdbc-slim/src/main/java/com/salesforce/datacloud/jdbc/core/DataCloudConnection.java

@@ -22,13 +22,15 @@
 import com.salesforce.datacloud.jdbc.auth.AuthenticationSettings;
 import com.salesforce.datacloud.jdbc.auth.DataCloudTokenProcessor;
 import com.salesforce.datacloud.jdbc.auth.TokenProcessor;
+import com.salesforce.datacloud.jdbc.core.partial.RowBased;
 import com.salesforce.datacloud.jdbc.exception.DataCloudJDBCException;
 import com.salesforce.datacloud.jdbc.http.ClientBuilder;
 import com.salesforce.datacloud.jdbc.interceptor.AuthorizationHeaderInterceptor;
 import com.salesforce.datacloud.jdbc.interceptor.DataspaceHeaderInterceptor;
 import com.salesforce.datacloud.jdbc.interceptor.HyperExternalClientContextHeaderInterceptor;
 import com.salesforce.datacloud.jdbc.interceptor.HyperWorkloadHeaderInterceptor;
 import com.salesforce.datacloud.jdbc.interceptor.TracingHeadersInterceptor;
+import com.salesforce.datacloud.jdbc.util.Unstable;
 import io.grpc.ClientInterceptor;
 import io.grpc.ManagedChannelBuilder;
 import java.sql.Array;
@@ -173,6 +175,33 @@ private DataCloudPreparedStatement getQueryPreparedStatement(String sql) {
         return new DataCloudPreparedStatement(this, sql, new DefaultParameterManager());
     }
 
+    /**
+     * Retrieves a collection of rows for the specified query once it is ready.
+     * Use {@link #getQueryStatus(String)} to check if the query has produced results or finished execution before calling this method.
+     * <p>
+     * When using {@link RowBased.Mode#FULL_RANGE}, this method does not handle pagination near the end of available rows.
+     * The caller is responsible for calculating the correct offset and limit to avoid out-of-range errors.
+     *
+     * @param queryId The identifier of the query to fetch results for.
+     * @param offset  The starting row offset.
+     * @param limit   The maximum number of rows to retrieve.
+     * @param mode    The fetching mode—either {@link RowBased.Mode#SINGLE_RPC} for a single request or
+     *                {@link RowBased.Mode#FULL_RANGE} to iterate through all available rows.
+     * @return A {@link DataCloudResultSet} containing the query results.
+     */
+    public DataCloudResultSet getRowBasedResultSet(String queryId, long offset, long limit, RowBased.Mode mode) {
+        val iterator = RowBased.of(executor, queryId, offset, limit, mode);
+        return StreamingResultSet.of(queryId, executor, iterator);
+    }
+
+    /**
+     * Use this to determine when a given query is complete by filtering the responses and a subsequent findFirst()
+     */
+    @Unstable
+    public Stream<DataCloudQueryStatus> getQueryStatus(String queryId) {
+        return executor.getQueryStatus(queryId);
+    }
+
     @Override
     public CallableStatement prepareCall(String sql) {
         return null;

jdbc-slim/src/main/java/com/salesforce/datacloud/jdbc/core/DataCloudQueryStatus.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright (c) 2024, Salesforce, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.salesforce.datacloud.jdbc.core;
+
+import java.util.Optional;
+import lombok.Value;
+import lombok.val;
+import salesforce.cdp.hyperdb.v1.QueryInfo;
+import salesforce.cdp.hyperdb.v1.QueryStatus;
+
+/**
+ * Represents the status of a query.
+ * The {@link CompletionStatus} enum defines the possible states of the query, which are:
+ * <ul>
+ *   <li><b>RUNNING</b>: The query is still running or its status is unspecified.</li>
+ *   <li><b>RESULTS_PRODUCED</b>: The query has completed, and the results are ready for retrieval.</li>
+ *   <li><b>FINISHED</b>: The query has finished execution and its results have been persisted, guaranteed to be available until the expiration time.</li>
+ * </ul>
+ */
+@Value
+public class DataCloudQueryStatus {
+    public enum CompletionStatus {
+        RUNNING,
+        RESULTS_PRODUCED,
+        FINISHED
+    }
+
+    String queryId;
+
+    long chunkCount;
+
+    long rowCount;
+
+    double progress;
+
+    CompletionStatus completionStatus;
+
+    /**
+     * Checks if the query's results have been produced.
+     *
+     * @return {@code true} if the query's results are available for retrieval, otherwise {@code false}.
+     */
+    public boolean isResultProduced() {
+        return completionStatus == CompletionStatus.RESULTS_PRODUCED;
+    }
+
+    /**
+     * Checks if the query execution is finished.
+     *
+     * @return {@code true} if the query has completed execution and results have been persisted, otherwise {@code false}.
+     */
+    public boolean isExecutionFinished() {
+        return completionStatus == CompletionStatus.FINISHED;
+    }
+
+    static Optional<DataCloudQueryStatus> of(QueryInfo queryInfo) {
+        return Optional.ofNullable(queryInfo).map(QueryInfo::getQueryStatus).map(DataCloudQueryStatus::of);
+    }
+
+    private static DataCloudQueryStatus of(QueryStatus s) {
+        val completionStatus = of(s.getCompletionStatus());
+        return new DataCloudQueryStatus(
+                s.getQueryId(), s.getChunkCount(), s.getRowCount(), s.getProgress(), completionStatus);
+    }
+
+    private static CompletionStatus of(QueryStatus.CompletionStatus completionStatus) {
+        switch (completionStatus) {
+            case RUNNING_OR_UNSPECIFIED:
+                return CompletionStatus.RUNNING;
+            case RESULTS_PRODUCED:
+                return CompletionStatus.RESULTS_PRODUCED;
+            case FINISHED:
+                return CompletionStatus.FINISHED;
+            default:
+                throw new IllegalArgumentException("Unknown completion status. status=" + completionStatus);
+        }
+    }
+}

jdbc-slim/src/main/java/com/salesforce/datacloud/jdbc/core/DataCloudStatement.java

@@ -25,6 +25,7 @@
 import com.salesforce.datacloud.jdbc.exception.DataCloudJDBCException;
 import com.salesforce.datacloud.jdbc.util.Constants;
 import com.salesforce.datacloud.jdbc.util.SqlErrorCodes;
+import com.salesforce.datacloud.jdbc.util.Unstable;
 import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -73,16 +74,27 @@ protected HyperGrpcClientExecutor getQueryExecutor(QueryParam additionalQueryPar
         return clientBuilder.queryTimeout(getQueryTimeout()).build();
     }
 
-    private void assertQueryReady() throws SQLException {
+    private void assertQueryExecuted() throws SQLException {
         if (listener == null) {
             throw new DataCloudJDBCException("a query was not executed before attempting to access results");
         }
+    }
+
+    private void assertQueryReady() throws SQLException {
+        assertQueryExecuted();
 
         if (!listener.isReady()) {
             throw new DataCloudJDBCException("query results were not ready");
         }
     }
 
+    @Unstable
+    public String getQueryId() throws SQLException {
+        assertQueryExecuted();
+
+        return listener.getQueryId();
+    }
+
     public boolean isReady() {
         return listener.isReady();
     }

jdbc-slim/src/main/java/com/salesforce/datacloud/jdbc/core/HyperGrpcClientExecutor.java

@@ -20,6 +20,8 @@
 import com.salesforce.datacloud.jdbc.config.DriverVersion;
 import com.salesforce.datacloud.jdbc.interceptor.QueryIdHeaderInterceptor;
 import com.salesforce.datacloud.jdbc.util.PropertiesExtensions;
+import com.salesforce.datacloud.jdbc.util.StreamUtilities;
+import com.salesforce.datacloud.jdbc.util.Unstable;
 import io.grpc.ClientInterceptor;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
@@ -29,8 +31,10 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
 import lombok.AccessLevel;
 import lombok.Builder;
 import lombok.Getter;
@@ -45,6 +49,7 @@
 import salesforce.cdp.hyperdb.v1.QueryParam;
 import salesforce.cdp.hyperdb.v1.QueryResult;
 import salesforce.cdp.hyperdb.v1.QueryResultParam;
+import salesforce.cdp.hyperdb.v1.ResultRange;
 
 @Slf4j
 @Builder(toBuilder = true)
@@ -135,6 +140,29 @@ public Iterator<QueryInfo> getQueryInfoStreaming(String queryId) {
         return getStub(queryId).getQueryInfo(param);
     }
 
+    @Unstable
+    public Stream<DataCloudQueryStatus> getQueryStatus(String queryId) {
+        val iterator = getQueryInfo(queryId);
+        return StreamUtilities.toStream(iterator)
+                .map(DataCloudQueryStatus::of)
+                .filter(Optional::isPresent)
+                .map(Optional::get);
+    }
+
+    public Iterator<QueryResult> getQueryResult(String queryId, long offset, long limit, boolean omitSchema) {
+        val rowRange =
+                ResultRange.newBuilder().setRowOffset(offset).setRowLimit(limit).setByteLimit(1024);
+
+        final QueryResultParam param = QueryResultParam.newBuilder()
+                .setQueryId(queryId)
+                .setResultRange(rowRange)
+                .setOmitSchema(omitSchema)
+                .setOutputFormat(OutputFormat.ARROW_IPC)
+                .build();
+
+        return getStub(queryId).getQueryResult(param);
+    }
+
     public Iterator<QueryResult> getQueryResult(String queryId, long chunkId, boolean omitSchema) {
         val param = getQueryResultParam(queryId, chunkId, omitSchema);
         return getStub(queryId).getQueryResult(param);
@@ -161,12 +189,9 @@ private QueryResultParam getQueryResultParam(String queryId, long chunkId, boole
         val builder = QueryResultParam.newBuilder()
                 .setQueryId(queryId)
                 .setChunkId(chunkId)
+                .setOmitSchema(omitSchema)
                 .setOutputFormat(OutputFormat.ARROW_IPC);
 
-        if (omitSchema) {
-            builder.setOmitSchema(true);
-        }
-
         return builder.build();
     }
 

jdbc-slim/src/main/java/com/salesforce/datacloud/jdbc/core/StreamingResultSet.java

@@ -18,11 +18,15 @@
 import com.salesforce.datacloud.jdbc.core.listener.QueryStatusListener;
 import com.salesforce.datacloud.jdbc.exception.QueryExceptionHandler;
 import com.salesforce.datacloud.jdbc.util.ArrowUtils;
+import com.salesforce.datacloud.jdbc.util.StreamUtilities;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.TimeZone;
+import java.util.stream.Stream;
 import lombok.SneakyThrows;
+import lombok.Value;
 import lombok.extern.slf4j.Slf4j;
 import lombok.val;
 import org.apache.arrow.memory.RootAllocator;
@@ -32,6 +36,7 @@
 import org.apache.calcite.avatica.AvaticaStatement;
 import org.apache.calcite.avatica.Meta;
 import org.apache.calcite.avatica.QueryState;
+import salesforce.cdp.hyperdb.v1.QueryResult;
 
 @Slf4j
 public class StreamingResultSet extends AvaticaResultSet implements DataCloudResultSet {
@@ -51,6 +56,7 @@ private StreamingResultSet(
         this.listener = listener;
     }
 
+    @Deprecated
     @SneakyThrows
     public static StreamingResultSet of(String sql, QueryStatusListener listener) {
         try {
@@ -73,6 +79,30 @@ public static StreamingResultSet of(String sql, QueryStatusListener listener) {
         }
     }
 
+    @SneakyThrows
+    public static StreamingResultSet of(
+            String queryId, HyperGrpcClientExecutor client, Iterator<QueryResult> iterator) {
+        try {
+            val channel = ExecuteQueryResponseChannel.of(StreamUtilities.toStream(iterator));
+            val reader = new ArrowStreamReader(channel, new RootAllocator(ROOT_ALLOCATOR_MB_FROM_V2));
+            val schemaRoot = reader.getVectorSchemaRoot();
+            val columns = ArrowUtils.toColumnMetaData(schemaRoot.getSchema().getFields());
+            val timezone = TimeZone.getDefault();
+            val state = new QueryState();
+            val signature = new Meta.Signature(
+                    columns, null, Collections.emptyList(), Collections.emptyMap(), null, Meta.StatementType.SELECT);
+            val metadata = new AvaticaResultSetMetaData(null, null, signature);
+            val listener = new AlreadyReadyNoopListener(queryId);
+            val result = new StreamingResultSet(listener, null, state, signature, metadata, timezone, null);
+            val cursor = new ArrowStreamReaderCursor(reader);
+            result.execute2(cursor, columns);
+
+            return result;
+        } catch (Exception ex) {
+            throw QueryExceptionHandler.createException(QUERY_FAILURE + queryId, ex);
+        }
+    }
+
     @Override
     public String getQueryId() {
         return listener.getQueryId();
@@ -87,4 +117,24 @@ public String getStatus() {
     public boolean isReady() {
         return listener.isReady();
     }
+
+    private static final String QUERY_FAILURE = "Failed to execute query: ";
+
+    @Value
+    private static class AlreadyReadyNoopListener implements QueryStatusListener {
+        String queryId;
+        String status = "Status should be determined via DataCloudConnection::getStatus";
+        String query = null;
+        boolean ready = true;
+
+        @Override
+        public DataCloudResultSet generateResultSet() {
+            return null;
+        }
+
+        @Override
+        public Stream<QueryResult> stream() {
+            return Stream.empty();
+        }
+    }
 }

jdbc-slim/src/main/java/com/salesforce/datacloud/jdbc/core/listener/QueryStatusListener.java

@@ -20,6 +20,7 @@
 import java.util.stream.Stream;
 import salesforce.cdp.hyperdb.v1.QueryResult;
 
+@Deprecated
 public interface QueryStatusListener {
     String BEFORE_READY = "Results were requested before ready";