fix(csharp): prevent Power BI hang when warehouse stops mid-query (PECO-2971)

Root cause: When the CloudFetch fetcher encounters an error (e.g. warehouse
stopped), the downloader's download loop exits but never called
CompleteAdding() on the result queue. This caused _resultQueue.Take() to
block forever, deadlocking PBI's Mashup container thread.

Changes:
- Validate Thrift response status in ThriftResultFetcher, throw
  HiveServer2Exception on non-SUCCESS status codes
- Always call CompleteAdding() on result queue when download loop exits
  in CloudFetchDownloader (both error and happy paths)
- Add AdbcException throw in CloudFetchDownloadManager when downloader
  returns null but fetcher has error (PBI cannot handle AggregateException)
- Enhanced tracing for error propagation and completion paths
- Added tests for status validation and error propagation

Co-authored-by: Isaac

Author: eric-wang-1990

csharp/src/Reader/CloudFetch/CloudFetchDownloadManager.cs

@@ -23,8 +23,11 @@
 
 using System;
 using System.Collections.Concurrent;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
+using Apache.Arrow.Adbc;
+using Apache.Arrow.Adbc.Tracing;
 
 namespace AdbcDrivers.Databricks.Reader.CloudFetch
 {
@@ -84,14 +87,36 @@ public CloudFetchDownloadManager(
                 throw new InvalidOperationException("Download manager has not been started.");
             }
 
+            IDownloadResult? result;
             try
             {
-                return await _downloader.GetNextDownloadedFileAsync(cancellationToken).ConfigureAwait(false);
+                result = await _downloader.GetNextDownloadedFileAsync(cancellationToken).ConfigureAwait(false);
+                Activity.Current?.AddEvent("cloudfetch.download_manager_result", [
+                    new("result_is_null", result == null),
+                    new("fetcher_has_error", _resultFetcher.HasError),
+                    new("fetcher_error_message", _resultFetcher.Error?.Message ?? "(none)"),
+                    new("fetcher_is_completed", _resultFetcher.IsCompleted),
+                    new("downloader_is_completed", _downloader.IsCompleted)
+                ]);
             }
             catch (Exception ex) when (_resultFetcher.HasError)
             {
                 throw new AggregateException("Errors in download pipeline", new[] { ex, _resultFetcher.Error! });
             }
+
+            // If the downloader returned null (end of results) but the fetcher
+            // has a stored error, the fetcher failed mid-stream and the error
+            // was not propagated through the download queue. Surface it now
+            // instead of silently returning partial data.
+            if (result == null && _resultFetcher.HasError)
+            {
+                throw new AdbcException(
+                    $"Query execution failed: {_resultFetcher.Error!.Message}",
+                    AdbcStatusCode.IOError,
+                    _resultFetcher.Error!);
+            }
+
+            return result;
         }
 
         /// <inheritdoc />

csharp/src/Reader/CloudFetch/CloudFetchDownloader.cs

@@ -437,11 +437,24 @@ await _activityTracer.TraceActivityAsync(async activity =>
                         new("total_time_sec", overallStopwatch.ElapsedMilliseconds / 1000.0)
                     ]);
 
-                    // If there's an error, add the error to the result queue
+                    // Always mark the result queue as complete when the download
+                    // loop exits. Without this, a subsequent Take() call would
+                    // block forever on an empty, non-completed queue if the caller
+                    // retries after an exception (e.g. a fetcher error that the
+                    // downloader doesn't know about).
                     if (HasError)
                     {
                         CompleteWithError(activity);
                     }
+                    else
+                    {
+                        _isCompleted = true;
+                        try { _resultQueue.CompleteAdding(); }
+                        catch (InvalidOperationException ex)
+                        {
+                            activity?.AddException(ex, [new("error.context", "cloudfetch.result_queue_already_completed")]);
+                        }
+                    }
                 }
             });
         }
@@ -671,18 +684,15 @@ private void SetError(Exception ex, Activity? activity = null)
 
         private void CompleteWithError(Activity? activity = null)
         {
+            // Mark the download as completed with error
+            _isCompleted = true;
+
             try
             {
                 // Mark the result queue as completed to prevent further additions
                 _resultQueue.CompleteAdding();
-
-                // Mark the download as completed with error
-                _isCompleted = true;
-            }
-            catch (Exception ex)
-            {
-                activity?.AddException(ex, [new("error.context", "cloudfetch.complete_with_error_failed")]);
             }
+            catch (InvalidOperationException) { /* already completed */ }
         }
 
         // Helper method to sanitize URLs for logging (to avoid exposing sensitive information)

csharp/src/Reader/CloudFetch/CloudFetchReader.cs

@@ -153,7 +153,11 @@ public CloudFetchReader(
                             this.currentDownloadResult = await this.downloadManager.GetNextDownloadedFileAsync(cancellationToken);
                             if (this.currentDownloadResult == null)
                             {
-                                Activity.Current?.AddEvent("cloudfetch.reader_no_more_files");
+                                Activity.Current?.AddEvent("cloudfetch.reader_no_more_files", [
+                                    new("rows_read_so_far", _rowsRead),
+                                    new("total_expected_rows", _totalExpectedRows),
+                                    new("current_chunk_rows_read", _currentChunkRowsRead)
+                                ]);
                                 this.downloadManager.Dispose();
                                 this.downloadManager = null;
                                 // No more files

csharp/src/Reader/CloudFetch/CloudFetchResultFetcher.cs

@@ -204,13 +204,19 @@ private async Task FetchResultsAsync(CancellationToken cancellationToken)
             {
                 Activity.Current?.AddEvent("cloudfetch.fetcher_unhandled_error", [
                     new("error_message", ex.Message),
-                    new("error_type", ex.GetType().Name)
+                    new("error_type", ex.GetType().Name),
+                    new("error_stack", ex.StackTrace ?? "(null)")
                 ]);
                 _error = ex;
                 _hasMoreResults = false;
             }
             finally
             {
+                Activity.Current?.AddEvent("cloudfetch.fetcher_completing", [
+                    new("has_error", _error != null),
+                    new("error_message", _error?.Message ?? "(none)"),
+                    new("error_type", _error?.GetType().Name ?? "(none)")
+                ]);
                 // Always add the end of results guard to signal completion to the downloader.
                 // Use Add with cancellation token to exit promptly when cancelled.
                 try

csharp/src/Reader/CloudFetch/ThriftResultFetcher.cs

@@ -174,6 +174,36 @@ private async Task FetchNextResultBatchAsync(long? offset, CancellationToken can
                 throw;
             }
 
+            // Validate the response status before processing results.
+            // Without this check, a non-SUCCESS response (e.g., warehouse stopped mid-query)
+            // with no result links would be silently treated as end-of-results,
+            // causing partial data to be returned to callers like Power BI.
+            Activity.Current?.AddEvent("cloudfetch.fetch_response_status", [
+                new("status_code", response.Status.StatusCode.ToString()),
+                new("has_result_links", response.Results?.__isset.resultLinks == true),
+                new("result_links_count", response.Results?.ResultLinks?.Count ?? 0),
+                new("has_more_rows", response.HasMoreRows),
+                new("error_message", response.Status.ErrorMessage ?? "(null)"),
+                new("start_offset", startOffset)
+            ]);
+            if (response.Status.StatusCode != TStatusCode.SUCCESS_STATUS &&
+                response.Status.StatusCode != TStatusCode.SUCCESS_WITH_INFO_STATUS)
+            {
+                Activity.Current?.AddEvent("cloudfetch.fetch_error_status_detected", [
+                    new("status_code", response.Status.StatusCode.ToString()),
+                    new("error_message", response.Status.ErrorMessage ?? "(null)"),
+                    new("error_code", response.Status.ErrorCode),
+                    new("sql_state", response.Status.SqlState ?? "(null)")
+                ]);
+                _hasMoreResults = false;
+                var errorMessage = !string.IsNullOrWhiteSpace(response.Status.ErrorMessage)
+                    ? response.Status.ErrorMessage
+                    : $"Thrift server error: {response.Status.StatusCode} (ErrorCode={response.Status.ErrorCode}, SqlState={response.Status.SqlState ?? "null"})";
+                throw new DatabricksException(errorMessage)
+                    .SetNativeError(response.Status.ErrorCode)
+                    .SetSqlState(response.Status.SqlState);
+            }
+
             // Check if we have URL-based results
             if (response.Results.__isset.resultLinks &&
                 response.Results.ResultLinks != null &&

csharp/test/E2E/CloudFetch/CloudFetchResultFetcherTest.cs

@@ -27,6 +27,7 @@
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using AdbcDrivers.Databricks;
 using AdbcDrivers.Databricks.Reader.CloudFetch;
 using AdbcDrivers.HiveServer2.Hive2;
 using Apache.Hive.Service.Rpc.Thrift;
@@ -335,6 +336,93 @@ public async Task FetchResultsAsync_WithServerError_SetsErrorState()
             await _resultFetcher.StopAsync();
         }
 
+        [Fact]
+        public async Task FetchResultsAsync_WithErrorStatusCode_SetsErrorState()
+        {
+            // Arrange - simulate a warehouse stopping mid-query by returning
+            // a response with ERROR_STATUS and no result links
+            var errorResults = new TRowSet { __isset = { resultLinks = false } };
+            var errorResponse = new TFetchResultsResp
+            {
+                Status = new TStatus
+                {
+                    StatusCode = TStatusCode.ERROR_STATUS,
+                    ErrorMessage = "Query failed: warehouse stopped",
+                    ErrorCode = 500,
+                    SqlState = "HY000"
+                },
+                HasMoreRows = false,
+                Results = errorResults,
+                __isset = { results = true, hasMoreRows = true }
+            };
+
+            _mockClient.Reset();
+            _mockClient.Setup(c => c.FetchResults(It.IsAny<TFetchResultsReq>(), It.IsAny<CancellationToken>()))
+                .ReturnsAsync(errorResponse);
+
+            // Act
+            await _resultFetcher.StartAsync(CancellationToken.None);
+
+            // Wait for the fetcher to process the error
+            await Task.Delay(200);
+
+            // Assert - the fetcher should report an error, not silently succeed
+            Assert.False(_resultFetcher.HasMoreResults);
+            Assert.True(_resultFetcher.IsCompleted);
+            Assert.True(_resultFetcher.HasError, "Fetcher should have error state when server returns ERROR_STATUS");
+            Assert.NotNull(_resultFetcher.Error);
+            Assert.IsType<DatabricksException>(_resultFetcher.Error);
+            Assert.Contains("warehouse stopped", _resultFetcher.Error.Message);
+
+            // Cleanup
+            await _resultFetcher.StopAsync();
+        }
+
+        [Fact]
+        public async Task FetchResultsAsync_WithErrorAfterFirstBatch_SetsErrorState()
+        {
+            // Arrange - first batch succeeds, second batch returns error (warehouse stopped)
+            var firstBatchLinks = new List<TSparkArrowResultLink>
+            {
+                CreateTestResultLink(0, 100, "http://test.com/file1", 3600),
+                CreateTestResultLink(100, 100, "http://test.com/file2", 3600)
+            };
+
+            var errorResults = new TRowSet { __isset = { resultLinks = false } };
+            var errorResponse = new TFetchResultsResp
+            {
+                Status = new TStatus
+                {
+                    StatusCode = TStatusCode.ERROR_STATUS,
+                    ErrorMessage = "Query failed: warehouse stopped",
+                    ErrorCode = 500,
+                    SqlState = "HY000"
+                },
+                HasMoreRows = false,
+                Results = errorResults,
+                __isset = { results = true, hasMoreRows = true }
+            };
+
+            _mockClient.SetupSequence(c => c.FetchResults(It.IsAny<TFetchResultsReq>(), It.IsAny<CancellationToken>()))
+                .ReturnsAsync(CreateFetchResultsResponse(firstBatchLinks, true))  // First batch: success, more rows
+                .ReturnsAsync(errorResponse);  // Second batch: error (warehouse stopped)
+
+            // Act
+            await _resultFetcher.StartAsync(CancellationToken.None);
+
+            // Wait for the fetcher to process
+            await Task.Delay(300);
+
+            // Assert - should have error state even though first batch succeeded
+            Assert.True(_resultFetcher.IsCompleted);
+            Assert.True(_resultFetcher.HasError, "Fetcher should have error state when second batch returns ERROR_STATUS");
+            Assert.NotNull(_resultFetcher.Error);
+            Assert.IsType<DatabricksException>(_resultFetcher.Error);
+
+            // Cleanup
+            await _resultFetcher.StopAsync();
+        }
+
         [Fact]
         public async Task StopAsync_CancelsFetching()
         {