fix(csharp): update StatementExecutionStatement to use protocol-agnostic CloudFetch pattern

Update CreateExternalLinksReader to use the protocol-agnostic CloudFetch pattern:
- Use StatementExecutionResultFetcher with 3-parameter constructor (no manager resources)
- Use CloudFetchConfiguration.FromProperties to parse config
- Use CloudFetchDownloadManager with config (calls Initialize() on fetcher)
- Use CloudFetchReader with ITracingStatement and nullable IResponse

Also fix StatementExecutionResultFetcher constructor that was reverted during conflict resolution:
- Remove memoryManager and downloadQueue parameters
- Pass null to base constructor for late initialization
- Resources will be injected via Initialize() by CloudFetchDownloadManager

This completes the CloudFetch REST API integration using the unified protocol-agnostic pattern.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: Jade Wang

csharp/src/Reader/CloudFetch/StatementExecutionResultFetcher.cs

@@ -45,19 +45,17 @@ internal class StatementExecutionResultFetcher : BaseResultFetcher
 
         /// <summary>
         /// Initializes a new instance of the <see cref="StatementExecutionResultFetcher"/> class.
+        /// Resources (memoryManager, downloadQueue) will be initialized by CloudFetchDownloadManager
+        /// via the Initialize() method.
         /// </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="memoryManager">The memory buffer manager.</param>
-        /// <param name="downloadQueue">The queue to add download tasks to.</param>
         public StatementExecutionResultFetcher(
             IStatementExecutionClient client,
             string statementId,
-            ResultManifest manifest,
-            ICloudFetchMemoryBufferManager memoryManager,
-            BlockingCollection<IDownloadResult> downloadQueue)
-            : base(memoryManager, downloadQueue)
+            ResultManifest manifest)
+            : base(null, null)  // Resources will be injected via Initialize()
         {
             _client = client ?? throw new ArgumentNullException(nameof(client));
             _statementId = statementId ?? throw new ArgumentNullException(nameof(statementId));

csharp/src/StatementExecution/StatementExecutionStatement.cs

@@ -365,84 +365,36 @@ private IArrowArrayStream CreateExternalLinksReader(GetStatementResponse respons
             // Determine compression
             bool isLz4Compressed = response.Manifest.ResultCompression?.Equals("lz4", StringComparison.OrdinalIgnoreCase) == true;
 
-            // Create memory manager
-            int memoryBufferSizeMB = int.Parse(GetPropertyOrDefault(DatabricksParameters.CloudFetchMemoryBufferSize, "200"));
-            var memoryManager = new CloudFetchMemoryBufferManager(memoryBufferSizeMB);
-
-            // Create download and result queues
-            var downloadQueue = new BlockingCollection<IDownloadResult>(new ConcurrentQueue<IDownloadResult>(), 10);
-            var resultQueue = new BlockingCollection<IDownloadResult>(new ConcurrentQueue<IDownloadResult>(), 10);
-
-            // If Result field has external links, add them to the download queue first
-            // (Result contains the first chunk, Manifest may not include it for large results)
-            if (response.Result?.ExternalLinks != null && response.Result.ExternalLinks.Any())
-            {
-                foreach (var link in response.Result.ExternalLinks)
-                {
-                    var expirationTime = DateTime.UtcNow.AddHours(1);
-                    if (!string.IsNullOrEmpty(link.Expiration))
-                    {
-                        try
-                        {
-                            expirationTime = DateTime.Parse(link.Expiration, System.Globalization.CultureInfo.InvariantCulture, System.Globalization.DateTimeStyles.RoundtripKind);
-                        }
-                        catch (FormatException) { }
-                    }
-
-                    var downloadResult = new DownloadResult(
-                        chunkIndex: link.ChunkIndex,
-                        fileUrl: link.ExternalLinkUrl,
-                        startRowOffset: link.RowOffset,
-                        rowCount: link.RowCount,
-                        byteCount: link.ByteCount,
-                        expirationTime: expirationTime,
-                        memoryManager: memoryManager,
-                        httpHeaders: link.HttpHeaders);
-
-                    downloadQueue.Add(downloadResult);
-                }
-            }
-
-            // Create result fetcher
+            // 1. Create REST-specific result fetcher
+            // Resources (memory manager, download queue) will be initialized by CloudFetchDownloadManager
             var resultFetcher = new StatementExecutionResultFetcher(
                 _client,
                 response.StatementId,
-                response.Manifest,
-                memoryManager,
-                downloadQueue);
-
-            // Create downloader with correct parameters
-            int parallelDownloads = int.Parse(GetPropertyOrDefault(DatabricksParameters.CloudFetchParallelDownloads, "3"));
-            int maxRetries = int.Parse(GetPropertyOrDefault(DatabricksParameters.CloudFetchMaxRetries, "3"));
-            int retryDelayMs = int.Parse(GetPropertyOrDefault(DatabricksParameters.CloudFetchRetryDelayMs, "500"));
-            int urlExpirationBufferSeconds = int.Parse(GetPropertyOrDefault(DatabricksParameters.CloudFetchUrlExpirationBufferSeconds, "60"));
-            int maxUrlRefreshAttempts = int.Parse(GetPropertyOrDefault(DatabricksParameters.CloudFetchMaxUrlRefreshAttempts, "3"));
-
-            var downloader = new CloudFetchDownloader(
-                this, // Pass this as ITracingStatement
-                downloadQueue,
-                resultQueue,
-                memoryManager,
-                _httpClient,
-                resultFetcher,
-                parallelDownloads,
-                isLz4Compressed,
-                maxRetries,
-                retryDelayMs,
-                maxUrlRefreshAttempts,
-                urlExpirationBufferSeconds);
-
-            // Create download manager using test constructor (for REST API)
-            var downloadManager = new CloudFetchDownloadManager(
+                response.Manifest);
+
+            // 2. Parse configuration from REST properties (unified properties work for both Thrift and REST)
+            var config = CloudFetchConfiguration.FromProperties(
+                _properties,
                 schema,
-                resultFetcher,
-                downloader);
+                isLz4Compressed);
 
-            // Start the download manager
+            // 3. Create protocol-agnostic download manager
+            // Manager creates shared resources and calls Initialize() on the fetcher
+            var downloadManager = new CloudFetchDownloadManager(
+                resultFetcher,        // Protocol-specific fetcher
+                _httpClient,
+                config,
+                this);                // ITracingStatement for tracing
+
+            // 4. Start the manager
             downloadManager.StartAsync().GetAwaiter().GetResult();
 
-            // Create and return a simple reader that uses the download manager
-            return new SimpleCloudFetchReader(downloadManager, response.Manifest.ResultCompression, schema);
+            // 5. Create protocol-agnostic reader
+            return new CloudFetchReader(
+                this,                 // ITracingStatement (both Thrift and REST implement this)
+                schema,
+                null,                 // IResponse (REST doesn't use IResponse)
+                downloadManager);
         }
 
         /// <summary>