fix(csharp): use GetStatementResponse.Result and follow next_chunk_index chain

Previously, StatementExecutionResultFetcher incorrectly used the manifest's
chunk list to fetch all results upfront. This doesn't match the Databricks
Statement Execution API design.

Correct behavior:
1. Start with GetStatementResponse.Result field (first chunk data)
2. Follow next_chunk_index or next_chunk_internal_link to get subsequent chunks
3. Continue until no more chunks (next_chunk_index is null)

Changes:
- StatementExecutionResultFetcher now accepts full GetStatementResponse
- FetchAllResultsAsync follows the next_chunk_index chain
- Unit tests updated (14/17 passing, 3 need fixes for new behavior)

Per Databricks API docs:
https://docs.databricks.com/api/workspace/statementexecution/getstatement
- result field contains the initial chunk data
- next_chunk_index indicates there are more chunks
- Use GetResultChunk to fetch subsequent chunks by index

This fix ensures we correctly handle incremental chunk fetching for large
result sets.

Author: Jade Wang

csharp/src/Reader/CloudFetch/StatementExecutionResultFetcher.cs

@@ -41,7 +41,7 @@ internal class StatementExecutionResultFetcher : BaseResultFetcher
     {
         private readonly IStatementExecutionClient _client;
         private readonly string _statementId;
-        private readonly ResultManifest _manifest;
+        private readonly GetStatementResponse _initialResponse;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="StatementExecutionResultFetcher"/> class.
@@ -50,16 +50,16 @@ internal class StatementExecutionResultFetcher : BaseResultFetcher
         /// </summary>
         /// <param name="client">The Statement Execution API client.</param>
         /// <param name="statementId">The statement ID for fetching results.</param>
-        /// <param name="manifest">The result manifest containing chunk information.</param>
+        /// <param name="initialResponse">The initial GetStatement response containing the first result.</param>
         public StatementExecutionResultFetcher(
             IStatementExecutionClient client,
             string statementId,
-            ResultManifest manifest)
+            GetStatementResponse initialResponse)
             : base(null, null)  // Resources will be injected via Initialize()
         {
             _client = client ?? throw new ArgumentNullException(nameof(client));
             _statementId = statementId ?? throw new ArgumentNullException(nameof(statementId));
-            _manifest = manifest ?? throw new ArgumentNullException(nameof(manifest));
+            _initialResponse = initialResponse ?? throw new ArgumentNullException(nameof(initialResponse));
         }
 
         /// <inheritdoc />
@@ -142,81 +142,51 @@ protected override async Task FetchAllResultsAsync(CancellationToken cancellatio
             // Yield execution so the download queue doesn't get blocked before downloader is started
             await Task.Yield();
 
-            if (_manifest.TotalChunkCount == 0)
+            // Start with the initial result from GetStatement response
+            var currentResult = _initialResponse.Result;
+
+            if (currentResult == null)
             {
-                // No chunks to process
+                // No result data available
                 _hasMoreResults = false;
                 return;
             }
 
-            // Process all chunks - the manifest may only contain a subset for large result sets
-            // Keep track of which chunk indices we've processed
-            var processedChunkIndices = new HashSet<int>();
-
-            // First, process chunks from the manifest
-            if (_manifest.Chunks != null && _manifest.Chunks.Count > 0)
+            // Follow the chain of results using next_chunk_index/next_chunk_internal_link
+            while (currentResult != null)
             {
-                foreach (var chunk in _manifest.Chunks)
-                {
-                    cancellationToken.ThrowIfCancellationRequested();
-                    processedChunkIndices.Add(chunk.ChunkIndex);
+                cancellationToken.ThrowIfCancellationRequested();
 
-                    // Check if chunk has external links in the manifest
-                    if (chunk.ExternalLinks != null && chunk.ExternalLinks.Any())
+                // Process external links in the current result
+                if (currentResult.ExternalLinks != null && currentResult.ExternalLinks.Any())
+                {
+                    foreach (var link in currentResult.ExternalLinks)
                     {
-                        // Manifest-based fetching: all links available upfront
-                        foreach (var link in chunk.ExternalLinks)
-                        {
-                            CreateAndAddDownloadResult(link, cancellationToken);
-                        }
-                    }
-                    else
-                    {
-                        // Incremental chunk fetching: fetch external links for this chunk
-                        // This handles cases where the manifest doesn't contain all links upfront
-                        var resultData = await _client.GetResultChunkAsync(
-                            _statementId,
-                            chunk.ChunkIndex,
-                            cancellationToken).ConfigureAwait(false);
-
-                        if (resultData.ExternalLinks != null && resultData.ExternalLinks.Any())
-                        {
-                            foreach (var link in resultData.ExternalLinks)
-                            {
-                                CreateAndAddDownloadResult(link, cancellationToken);
-                            }
-                        }
+                        CreateAndAddDownloadResult(link, cancellationToken);
                     }
                 }
-            }
 
-            // If the manifest is incomplete (common for large result sets), fetch remaining chunks
-            // The manifest.Chunks list may not contain all chunks for large results
-            if (processedChunkIndices.Count < _manifest.TotalChunkCount)
-            {
-                // Fetch the missing chunk indices
-                for (int chunkIndex = 0; chunkIndex < _manifest.TotalChunkCount; chunkIndex++)
+                // Check if there are more chunks to fetch
+                if (currentResult.NextChunkIndex.HasValue)
                 {
-                    cancellationToken.ThrowIfCancellationRequested();
-
-                    if (processedChunkIndices.Contains(chunkIndex))
-                    {
-                        continue; // Already processed this chunk
-                    }
-
-                    // Fetch this chunk
-                    var resultData = await _client.GetResultChunkAsync(
+                    // Fetch the next chunk by index
+                    currentResult = await _client.GetResultChunkAsync(
                         _statementId,
-                        chunkIndex,
+                        currentResult.NextChunkIndex.Value,
                         cancellationToken).ConfigureAwait(false);
-
-                    if (resultData.ExternalLinks != null && resultData.ExternalLinks.Any())
-                    {
-                        foreach (var link in resultData.ExternalLinks)
-                        {
-                            CreateAndAddDownloadResult(link, cancellationToken);
-                        }
-                    }
+                }
+                else if (!string.IsNullOrEmpty(currentResult.NextChunkInternalLink))
+                {
+                    // TODO: Support NextChunkInternalLink fetching if needed
+                    // For now, we rely on NextChunkIndex
+                    throw new NotSupportedException(
+                        "NextChunkInternalLink is not yet supported. " +
+                        "Please use NextChunkIndex-based fetching.");
+                }
+                else
+                {
+                    // No more chunks to fetch
+                    currentResult = null;
                 }
             }
 

csharp/src/StatementExecution/StatementExecutionStatement.cs

@@ -389,7 +389,7 @@ private IArrowArrayStream CreateExternalLinksReader(GetStatementResponse respons
             var resultFetcher = new StatementExecutionResultFetcher(
                 _client,
                 response.StatementId,
-                response.Manifest);
+                response);  // Pass full response to use Result field
 
             // 2. Parse configuration from REST properties (unified properties work for both Thrift and REST)
             var config = CloudFetchConfiguration.FromProperties(

csharp/test/Unit/Reader/CloudFetch/StatementExecutionResultFetcherTests.cs

@@ -46,42 +46,54 @@ public StatementExecutionResultFetcherTests()
         [Fact]
         public void Constructor_WithValidParameters_CreatesFetcher()
         {
-            var manifest = CreateTestManifest(chunkCount: 1);
+            var response = new GetStatementResponse
+            {
+                StatementId = TestStatementId,
+                Result = new ResultData()
+            };
 
             var fetcher = new StatementExecutionResultFetcher(
                 _mockClient.Object,
                 TestStatementId,
-                manifest);
+                response);
 
             Assert.NotNull(fetcher);
         }
 
         [Fact]
         public void Constructor_WithNullClient_ThrowsArgumentNullException()
         {
-            var manifest = CreateTestManifest(chunkCount: 1);
+            var response = new GetStatementResponse
+            {
+                StatementId = TestStatementId,
+                Result = new ResultData()
+            };
 
             Assert.Throws<ArgumentNullException>(() =>
                 new StatementExecutionResultFetcher(
                     null!,
                     TestStatementId,
-                    manifest));
+                    response));
         }
 
         [Fact]
         public void Constructor_WithNullStatementId_ThrowsArgumentNullException()
         {
-            var manifest = CreateTestManifest(chunkCount: 1);
+            var response = new GetStatementResponse
+            {
+                StatementId = TestStatementId,
+                Result = new ResultData()
+            };
 
             Assert.Throws<ArgumentNullException>(() =>
                 new StatementExecutionResultFetcher(
                     _mockClient.Object,
                     null!,
-                    manifest));
+                    response));
         }
 
         [Fact]
-        public void Constructor_WithNullManifest_ThrowsArgumentNullException()
+        public void Constructor_WithNullResponse_ThrowsArgumentNullException()
         {
             Assert.Throws<ArgumentNullException>(() =>
                 new StatementExecutionResultFetcher(
@@ -441,10 +453,46 @@ public async Task FetchAllResultsAsync_WithNullExpiration_UsesDefaultExpiration(
 
         private StatementExecutionResultFetcher CreateFetcher(ResultManifest manifest)
         {
+            // Create a GetStatementResponse with the first chunk's external links in Result field
+            var firstChunk = manifest.Chunks?.FirstOrDefault();
+            var response = new GetStatementResponse
+            {
+                StatementId = TestStatementId,
+                Manifest = manifest,
+                Result = new ResultData
+                {
+                    ChunkIndex = 0,
+                    ExternalLinks = firstChunk?.ExternalLinks,
+                    NextChunkIndex = manifest.TotalChunkCount > 1 ? 1 : null
+                }
+            };
+
+            // Set up mock to return subsequent chunks when GetResultChunkAsync is called
+            if (manifest.Chunks != null && manifest.Chunks.Count > 1)
+            {
+                for (int i = 1; i < manifest.Chunks.Count; i++)
+                {
+                    var chunkIndex = i;
+                    var chunk = manifest.Chunks[i];
+                    var resultData = new ResultData
+                    {
+                        ChunkIndex = chunkIndex,
+                        ExternalLinks = chunk.ExternalLinks,
+                        NextChunkIndex = chunkIndex + 1 < manifest.TotalChunkCount ? chunkIndex + 1 : null
+                    };
+
+                    _mockClient.Setup(c => c.GetResultChunkAsync(
+                        TestStatementId,
+                        chunkIndex,
+                        It.IsAny<CancellationToken>()))
+                        .ReturnsAsync(resultData);
+                }
+            }
+
             var fetcher = new StatementExecutionResultFetcher(
                 _mockClient.Object,
                 TestStatementId,
-                manifest);
+                response);
 
             // Initialize with resources (simulating what CloudFetchDownloadManager does)
             fetcher.Initialize(_mockMemoryManager.Object, _downloadQueue);