NCO-57: Async Analytics Query

src/Couchbase.Analytics/Async/QueryHandle.cs

@@ -39,14 +39,14 @@ public class QueryHandle
     internal AsyncQueryMetrics? Metrics { get; }
 
     /// <summary>
-    /// The query handle string used to poll for the result handle.
+    /// The query handle string used to poll for the query status.
     /// </summary>
-    public string Handle { get; }
+    internal string Handle { get; }
 
     /// <summary>
     /// The request ID assigned by the server when the query was submitted.
     /// </summary>
-    public string RequestId { get; }
+    internal string RequestId { get; }
 
     internal QueryHandle(string handle, string requestId, JsonElement root, IAnalyticsService analyticsService)
     {
@@ -67,25 +67,25 @@ internal QueryHandle(string handle, string requestId, JsonElement root, IAnalyti
     }
 
     /// <summary>
-    /// Fetches the result handle of the asynchronous query from the server.
+    /// Fetches the current status of the asynchronous query from the server.
     /// </summary>
-    /// <param name="options">Options for fetching the result handle.</param>
+    /// <param name="options">Options for fetching the status.</param>
     /// <param name="cancellationToken">A cancellation token.</param>
-    /// <returns>A <see cref="QueryResultHandle"/> if results are ready, otherwise null.</returns>
-    public Task<QueryResultHandle?> FetchResultHandleAsync(FetchResultHandleOptions? options = null, CancellationToken cancellationToken = default)
+    /// <returns>A <see cref="QueryStatus"/> representing the current state of the query.</returns>
+    public Task<QueryStatus> FetchStatusAsync(FetchStatusOptions? options = null, CancellationToken cancellationToken = default)
     {
-        options ??= new FetchResultHandleOptions();
-        return _analyticsService.FetchResultHandleAsync(this, options, cancellationToken);
+        options ??= new FetchStatusOptions();
+        return _analyticsService.FetchStatusAsync(this, options, cancellationToken);
     }
 
     /// <summary>
-    /// Fetches the result handle of the asynchronous query from the server.
+    /// Fetches the current status of the asynchronous query from the server.
     /// </summary>
-    public Task<QueryResultHandle?> FetchResultHandleAsync(Func<FetchResultHandleOptions, FetchResultHandleOptions> options, CancellationToken cancellationToken = default)
+    public Task<QueryStatus> FetchStatusAsync(Func<FetchStatusOptions, FetchStatusOptions> options, CancellationToken cancellationToken = default)
     {
-        var fetchOptions = new FetchResultHandleOptions();
+        var fetchOptions = new FetchStatusOptions();
         fetchOptions = options.Invoke(fetchOptions);
-        return FetchResultHandleAsync(fetchOptions, cancellationToken);
+        return FetchStatusAsync(fetchOptions, cancellationToken);
     }
 
     /// <summary>

src/Couchbase.Analytics/Async/QueryStatus.cs

@@ -0,0 +1,170 @@
+#region License
+/* ************************************************************
+ *
+ *    @author Couchbase <info@couchbase.com>
+ *    @copyright 2025 Couchbase, 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.
+ *
+ * ************************************************************/
+#endregion
+
+using System.Text.Json;
+using Couchbase.AnalyticsClient.Internal;
+using Couchbase.AnalyticsClient.Query;
+
+namespace Couchbase.AnalyticsClient.Async;
+
+/// <summary>
+/// Represents the status of a server-side asynchronous query.
+/// Obtained from <see cref="QueryHandle.FetchStatusAsync"/>.
+/// </summary>
+public class QueryStatus
+{
+    private readonly string? _resultHandlePath;
+    private readonly string _requestId;
+    private readonly JsonElement _root;
+    private readonly IAnalyticsService _analyticsService;
+
+    internal string? Status { get; }
+
+    internal AsyncQueryMetrics? Metrics { get; }
+
+    internal int? ResultCount { get; }
+
+    internal bool? ResultSetOrdered { get; }
+
+    internal string? CreatedAt { get; }
+
+    internal int PartitionCount { get; }
+
+    /// <summary>
+    /// Returns <c>true</c> if and only if the server response provides a handle
+    /// for the SDK to retrieve the query results. Otherwise returns <c>false</c>.
+    /// This property does not perform network calls.
+    /// </summary>
+    public bool ResultsReady { get; }
+
+    internal QueryStatus(string requestId, JsonElement root, IAnalyticsService analyticsService)
+    {
+        _requestId = requestId ?? throw new ArgumentNullException(nameof(requestId));
+        _analyticsService = analyticsService ?? throw new ArgumentNullException(nameof(analyticsService));
+
+        if (root.ValueKind is JsonValueKind.Undefined or JsonValueKind.Null)
+        {
+            throw new ArgumentException("The JSON response element must not be empty or null.", nameof(root));
+        }
+
+        _root = root;
+
+        // ── Core fields ──
+        Status = root.TryGetProperty("status", out var statusProp) ? statusProp.GetString() : null;
+
+        if (root.TryGetProperty("metrics", out var metricsElement))
+        {
+            Metrics = JsonSerializer.Deserialize<AsyncQueryMetrics>(metricsElement.GetRawText());
+        }
+
+        // ResultsReady is true when the server provides a result handle path
+        _resultHandlePath = root.TryGetProperty("handle", out var handleProp) ? handleProp.GetString() : null;
+        ResultsReady = !string.IsNullOrWhiteSpace(_resultHandlePath);
+
+        // ── Optional diagnostic fields (may be absent depending on server version or query state) ──
+        if (root.TryGetProperty("resultCount", out var resultCountProp) && resultCountProp.TryGetInt32(out var resultCount))
+        {
+            ResultCount = resultCount;
+        }
+
+        if (root.TryGetProperty("resultSetOrdered", out var orderedProp) && orderedProp.ValueKind is JsonValueKind.True or JsonValueKind.False)
+        {
+            ResultSetOrdered = orderedProp.GetBoolean();
+        }
+
+        CreatedAt = root.TryGetProperty("createdAt", out var createdProp) ? createdProp.GetString() : null;
+
+        if (root.TryGetProperty("partitions", out var partProp) && partProp.ValueKind == JsonValueKind.Array)
+        {
+            PartitionCount = partProp.GetArrayLength();
+        }
+    }
+
+    /// <summary>
+    /// Returns a new <see cref="QueryResultHandle"/> instance.
+    /// This method does not perform network calls.
+    /// </summary>
+    /// <returns>A <see cref="QueryResultHandle"/> that can be used to fetch or discard results.</returns>
+    /// <exception cref="InvalidOperationException">Thrown if results are not yet ready.</exception>
+    public QueryResultHandle ResultHandle()
+    {
+        if (!ResultsReady)
+        {
+            throw new InvalidOperationException(
+                $"Results are not ready. Current status: {Status ?? "unknown"}. " +
+                "Poll again with FetchStatusAsync() until ResultsReady is true.");
+        }
+
+        return new QueryResultHandle(_resultHandlePath!, _requestId, _root, _analyticsService);
+    }
+
+    /// <inheritdoc />
+    public override string ToString()
+    {
+        var parts = new List<string>
+        {
+            $"Status={Status ?? "unknown"}",
+            $"ResultsReady={ResultsReady}"
+        };
+
+        if (ResultCount.HasValue)
+        {
+            parts.Add($"ResultCount={ResultCount}");
+        }
+
+        if (ResultSetOrdered.HasValue)
+        {
+            parts.Add($"ResultSetOrdered={ResultSetOrdered}");
+        }
+
+        if (PartitionCount > 0)
+        {
+            parts.Add($"Partitions={PartitionCount}");
+        }
+
+        if (CreatedAt is not null)
+        {
+            parts.Add($"CreatedAt={CreatedAt}");
+        }
+
+        if (Metrics is not null)
+        {
+            var metricParts = new List<string>();
+
+            if (Metrics.ElapsedTime.HasValue)
+                metricParts.Add($"ElapsedTime={Metrics.ElapsedTime.Value.TotalMilliseconds:F1}ms");
+            if (Metrics.ExecutionTime.HasValue)
+                metricParts.Add($"ExecutionTime={Metrics.ExecutionTime.Value.TotalMilliseconds:F1}ms");
+            if (Metrics.CompileTime.HasValue)
+                metricParts.Add($"CompileTime={Metrics.CompileTime.Value.TotalMilliseconds:F1}ms");
+            if (Metrics.QueueWaitTime.HasValue)
+                metricParts.Add($"QueueWaitTime={Metrics.QueueWaitTime.Value.TotalMilliseconds:F1}ms");
+            if (Metrics.ProcessedObjects.HasValue)
+                metricParts.Add($"ProcessedObjects={Metrics.ProcessedObjects}");
+
+            parts.Add(metricParts.Count > 0
+                ? $"Metrics={{{string.Join(", ", metricParts)}}}"
+                : "Metrics={none}");
+        }
+
+        return $"QueryStatus [{string.Join(", ", parts)}]";
+    }
+}

src/Couchbase.Analytics/Internal/AnalyticsService.cs

@@ -247,8 +247,8 @@ public async Task<QueryHandle> StartQueryAsync(string statement, StartQueryOptio
                 errorContext.StatusCode = response.StatusCode;
 
                 var responseBody = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
-                var json = JsonDocument.Parse(responseBody);
-                var root = json.RootElement;
+                using var json = JsonDocument.Parse(responseBody);
+                var root = json.RootElement.Clone();
 
                 if (!response.IsSuccessStatusCode)
                 {
@@ -325,7 +325,7 @@ public async Task<QueryHandle> StartQueryAsync(string statement, StartQueryOptio
         throw lastException ?? ThrowTooManyRetries(errorContext);
     }
 
-    public async Task<QueryResultHandle?> FetchResultHandleAsync(QueryHandle handle, FetchResultHandleOptions options, CancellationToken cancellationToken = default)
+    public async Task<QueryStatus> FetchStatusAsync(QueryHandle handle, FetchStatusOptions options, CancellationToken cancellationToken = default)
     {
         var timeout = _clusterOptions.TimeoutOptions.DispatchTimeout;
         var httpClient = CreateHttpClient(timeout);
@@ -336,7 +336,7 @@ public async Task<QueryHandle> StartQueryAsync(string statement, StartQueryOptio
 
         var request = new HttpRequestMessage(HttpMethod.Get, statusUri);
 
-        LogFetchResultHandleRequest(_logger, _redactor.SystemData(statusUri), _redactor.SystemData(handle.Handle));
+        LogFetchStatusRequest(_logger, _redactor.SystemData(statusUri), _redactor.SystemData(handle.Handle));
 
         try
         {
@@ -349,8 +349,8 @@ public async Task<QueryHandle> StartQueryAsync(string statement, StartQueryOptio
             }
 
             var responseBody = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
-            var json = JsonDocument.Parse(responseBody);
-            var root = json.RootElement;
+            using var json = JsonDocument.Parse(responseBody);
+            var root = json.RootElement.Clone();
 
             var status = root.TryGetProperty("status", out var statusProp) ? statusProp.GetString() : null;
 
@@ -361,7 +361,7 @@ public async Task<QueryHandle> StartQueryAsync(string statement, StartQueryOptio
 
             if (!response.IsSuccessStatusCode)
             {
-                LogFetchResultHandleUnexpectedHttp(_logger, _redactor.SystemData(handle.Handle), (int)response.StatusCode);
+                LogFetchStatusUnexpectedHttp(_logger, _redactor.SystemData(handle.Handle), (int)response.StatusCode);
             }
 
             IReadOnlyList<QueryError>? errors = null;
@@ -381,54 +381,40 @@ public async Task<QueryHandle> StartQueryAsync(string statement, StartQueryOptio
                 throw new AnalyticsException($"Query status fetch failed with HTTP {(int)response.StatusCode} and status: {status}", errorContext);
             }
 
-            LogFetchResultHandleResponse(_logger, _redactor.SystemData(handle.Handle), status, (int)response.StatusCode);
+            LogFetchStatusResponse(_logger, _redactor.SystemData(handle.Handle), status, (int)response.StatusCode);
 
-            if (string.Equals(status, "running", StringComparison.OrdinalIgnoreCase))
+            // Handle error statuses that come back with HTTP 200 but indicate failure
+            if (string.Equals(status, "stopped", StringComparison.OrdinalIgnoreCase) ||
+                string.Equals(status, "aborted", StringComparison.OrdinalIgnoreCase) ||
+                string.Equals(status, "closed", StringComparison.OrdinalIgnoreCase))
             {
-                return null;
+                throw new QueryNotFoundException($"Query has been discarded or canceled (status: {status}).");
             }
 
-            if (!string.Equals(status, "success", StringComparison.OrdinalIgnoreCase))
+            if (string.Equals(status, "timeout", StringComparison.OrdinalIgnoreCase))
             {
-                if (string.Equals(status, "stopped", StringComparison.OrdinalIgnoreCase) ||
-                    string.Equals(status, "aborted", StringComparison.OrdinalIgnoreCase) ||
-                    string.Equals(status, "closed", StringComparison.OrdinalIgnoreCase))
-                {
-                    throw new QueryNotFoundException($"Query has been discarded or canceled (status: {status}).");
-                }
-
-                if (string.Equals(status, "timeout", StringComparison.OrdinalIgnoreCase))
-                {
-                    throw new AnalyticsTimeoutException("The query evaluation timed out on the server.");
-                }
-
-                if (string.Equals(status, "fatal", StringComparison.OrdinalIgnoreCase) ||
-                    string.Equals(status, "failed", StringComparison.OrdinalIgnoreCase) ||
-                    string.Equals(status, "errors", StringComparison.OrdinalIgnoreCase))
-                {
-                    if (errors is { Count: > 0 })
-                    {
-                        var errorContext = new ErrorContext(string.Empty, LightweightStopwatch.StartNew(), timeout);
-                        errorContext.StatusCode = response.StatusCode;
-                        throw AnalyticsErrorMapper.MapServiceErrors(errors, errorContext);
-                    }
-                    throw new AnalyticsException($"Query execution failed on the server (status: {status}).");
-                }
-
-                throw new AnalyticsException($"Query status fetch failed with unrecognized status: {status}");
+                throw new AnalyticsTimeoutException("The query evaluation timed out on the server.");
             }
 
-            var resultHandle = root.TryGetProperty("handle", out var handleProp) ? handleProp.GetString() : null;
-            if (string.IsNullOrWhiteSpace(resultHandle))
+            if (string.Equals(status, "fatal", StringComparison.OrdinalIgnoreCase) ||
+                string.Equals(status, "failed", StringComparison.OrdinalIgnoreCase) ||
+                string.Equals(status, "errors", StringComparison.OrdinalIgnoreCase))
             {
-                throw new InvalidOperationException("Query status indicates success but no result handle was provided by the server.");
+                if (errors is { Count: > 0 })
+                {
+                    var errorContext = new ErrorContext(string.Empty, LightweightStopwatch.StartNew(), timeout);
+                    errorContext.StatusCode = response.StatusCode;
+                    throw AnalyticsErrorMapper.MapServiceErrors(errors, errorContext);
+                }
+                throw new AnalyticsException($"Query execution failed on the server (status: {status}).");
             }
 
-            return new QueryResultHandle(resultHandle, handle.RequestId, root, this);
+            // For queued, running, and success — return a QueryStatus
+            return new QueryStatus(handle.RequestId, root, this);
         }
         catch (TaskCanceledException taskCanceledEx)
         {
-            throw new AnalyticsTimeoutException("The FetchResultHandle request was canceled.", taskCanceledEx);
+            throw new AnalyticsTimeoutException("The FetchStatus request was canceled.", taskCanceledEx);
         }
     }
 
@@ -608,14 +594,14 @@ private static Exception ThrowTooManyRetries(ErrorContext errorContext)
     [LoggerMessage(9, LogLevel.Debug, "Async StartQuery succeeded for {ClientContextId}. Handle={Handle}, RequestId={RequestId} (HTTP {StatusCode})")]
     private static partial void LogAsyncStartQuerySucceeded(ILogger logger, string? clientContextId, Redacted<string> handle, Redacted<string> requestId, int statusCode);
 
-    [LoggerMessage(10, LogLevel.Debug, "FetchResultHandle sending GET to {Uri} for handle {Handle}")]
-    private static partial void LogFetchResultHandleRequest(ILogger logger, Redacted<Uri> uri, Redacted<string> handle);
+    [LoggerMessage(10, LogLevel.Debug, "FetchStatus sending GET to {Uri} for handle {Handle}")]
+    private static partial void LogFetchStatusRequest(ILogger logger, Redacted<Uri> uri, Redacted<string> handle);
 
-    [LoggerMessage(11, LogLevel.Debug, "FetchResultHandle for handle {Handle} returned status={Status} (HTTP {StatusCode})")]
-    private static partial void LogFetchResultHandleResponse(ILogger logger, Redacted<string> handle, string status, int statusCode);
+    [LoggerMessage(11, LogLevel.Debug, "FetchStatus for handle {Handle} returned status={Status} (HTTP {StatusCode})")]
+    private static partial void LogFetchStatusResponse(ILogger logger, Redacted<string> handle, string status, int statusCode);
 
-    [LoggerMessage(12, LogLevel.Warning, "FetchResultHandle for handle {Handle} returned unexpected HTTP {StatusCode}")]
-    private static partial void LogFetchResultHandleUnexpectedHttp(ILogger logger, Redacted<string> handle, int statusCode);
+    [LoggerMessage(12, LogLevel.Warning, "FetchStatus for handle {Handle} returned unexpected HTTP {StatusCode}")]
+    private static partial void LogFetchStatusUnexpectedHttp(ILogger logger, Redacted<string> handle, int statusCode);
 
     [LoggerMessage(13, LogLevel.Debug, "FetchResults sending GET to {Uri} for handle {Handle}")]
     private static partial void LogFetchResultsRequest(ILogger logger, Redacted<Uri> uri, Redacted<string> handle);

src/Couchbase.Analytics/Internal/IAnalyticsService.cs

@@ -43,7 +43,7 @@ internal interface IAnalyticsService
     /// Fetches the status of an async query from the server.
     /// Sends GET to /api/v1/request/status/{requestID}/{handleID}.
     /// </summary>
-    Task<QueryResultHandle?> FetchResultHandleAsync(QueryHandle handle, FetchResultHandleOptions options, CancellationToken cancellationToken = default);
+    Task<QueryStatus> FetchStatusAsync(QueryHandle handle, FetchStatusOptions options, CancellationToken cancellationToken = default);
 
     /// <summary>
     /// Fetches the results of a completed async query from the server.

src/Couchbase.Analytics/Internal/Results/AnalyticsResultBase.cs

@@ -65,4 +65,15 @@ public virtual void Dispose()
         ResponseStream?.Dispose();
         _ownedForCleanup?.Dispose();
     }
+
+    public virtual async ValueTask DisposeAsync()
+    {
+        if (_disposed) return;
+        _disposed = true;
+        if (ResponseStream != null)
+        {
+            await ResponseStream.DisposeAsync().ConfigureAwait(false);
+        }
+        _ownedForCleanup?.Dispose();
+    }
 }