feat(csharp): honor cloudfetch.lz4.enabled on SEA path (PECO-3056)

csharp/src/DatabricksParameters.cs

@@ -34,12 +34,28 @@ public class DatabricksParameters : SparkParameters
         /// <summary>
         /// Whether to use CloudFetch for retrieving results.
         /// Default value is true if not specified.
+        ///
+        /// Honored by both protocols:
+        ///   - Thrift: controls the canUseCloudFetch flag on TFetchResultsReq.
+        ///   - SEA (REST): not yet honored; the param is accepted but ignored.
+        ///     Disabling CloudFetch on SEA requires the server to support
+        ///     <c>ARROW_STREAM</c> results with <c>INLINE</c> disposition.
+        ///     Support is deferred until that server-side capability lands.
+        ///
+        /// Note: Reyden does not generate external links and will coerce results
+        /// to INLINE on the server side regardless of what the driver requests.
         /// </summary>
         public const string UseCloudFetch = "adbc.databricks.cloudfetch.enabled";
 
         /// <summary>
         /// Whether the client can decompress LZ4 compressed results.
         /// Default value is true if not specified.
+        ///
+        /// Honored by both protocols:
+        ///   - Thrift: controls canDecompressLZ4Result on TFetchResultsReq.
+        ///   - SEA (REST): when false, clears <c>result_compression</c> on the
+        ///     ExecuteStatement request (overriding any <see cref="ResultCompression"/>
+        ///     value). Mirrors JDBC's <c>CompressionCodec.NONE</c> branch.
         /// </summary>
         public const string CanDecompressLz4 = "adbc.databricks.cloudfetch.lz4.enabled";
 

csharp/src/StatementExecution/StatementExecutionConnection.cs

@@ -326,9 +326,13 @@ private void ValidateProperties()
             properties.TryGetValue(AdbcOptions.Connection.CurrentDbSchema, out _schema);
 
             // Result configuration.
+            // The driver only implements LZ4_FRAME decompression; gzip is not supported (PECO-3056).
+            // cloudfetch.lz4.enabled=true (default) → request LZ4_FRAME compression.
+            // cloudfetch.lz4.enabled=false → null (server treats as no compression).
             _resultDisposition = PropertyHelper.GetStringProperty(properties, DatabricksParameters.ResultDisposition, "INLINE_OR_EXTERNAL_LINKS");
             _resultFormat = PropertyHelper.GetStringProperty(properties, DatabricksParameters.ResultFormat, "ARROW_STREAM");
-            properties.TryGetValue(DatabricksParameters.ResultCompression, out _resultCompression);
+            bool canDecompressLz4 = PropertyHelper.GetBooleanPropertyWithValidation(properties, DatabricksParameters.CanDecompressLz4, true);
+            _resultCompression = canDecompressLz4 ? "LZ4_FRAME" : null;
 
             _waitTimeoutSeconds = PropertyHelper.GetIntPropertyWithValidation(properties, DatabricksParameters.WaitTimeout, 10);
             if (properties.TryGetValue(DatabricksParameters.EnableDirectResults, out var directResults) &&

csharp/src/StatementExecution/StatementExecutionStatement.cs

@@ -76,6 +76,13 @@ internal class StatementExecutionStatement : TracingStatement
         private string? _currentStatementId;
         private string? _sqlQuery;
 
+        // Internal test seam: captures the last request built by this statement so
+        // E2E tests can assert the exact field values sent on the wire without HTTP
+        // interception. There is no behavioral substitute — a query succeeding does
+        // not prove a specific compression or disposition value was requested.
+        private ExecuteStatementRequest? _lastExecuteRequest;
+        internal ExecuteStatementRequest? LastExecuteRequest => _lastExecuteRequest;
+
         // Cancel support
         private readonly object _cancelLock = new();
         private CancellationTokenSource? _executeCts;
@@ -340,6 +347,7 @@ private async Task<QueryResult> ExecuteQueryInternalAsync(CancellationToken canc
                 IsMetadata = isMetadataExecution,
                 QueryTags = ParseQueryTags(_queryTags)
             };
+            _lastExecuteRequest = request;
 
             // Execute the statement
             var response = await _client.ExecuteStatementAsync(request, cancellationToken).ConfigureAwait(false);
@@ -680,6 +688,7 @@ private async Task<UpdateResult> ExecuteUpdateInternalAsync(CancellationToken ca
                 IsMetadata = false,
                 QueryTags = ParseQueryTags(_queryTags)
             };
+            _lastExecuteRequest = request;
 
             // Execute the statement
             var response = await _client.ExecuteStatementAsync(request, cancellationToken).ConfigureAwait(false);

csharp/test/E2E/StatementExecution/SeaCloudFetchParamsE2ETests.cs

@@ -0,0 +1,155 @@
+/*
+* Copyright (c) 2025 ADBC Drivers Contributors
+*
+* 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.
+*/
+
+using System.Collections.Generic;
+using AdbcDrivers.Databricks.StatementExecution;
+using AdbcDrivers.HiveServer2.Spark;
+using Apache.Arrow.Adbc;
+using Apache.Arrow.Adbc.Tests;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace AdbcDrivers.Databricks.Tests.E2E.StatementExecution
+{
+    /// <summary>
+    /// E2E tests proving that <c>adbc.databricks.cloudfetch.lz4.enabled</c> is
+    /// honored on the SEA path (PECO-3056).
+    ///
+    /// Tests assert on the <c>result_compression</c> field of the request the
+    /// driver actually built, exposed via the internal
+    /// <see cref="StatementExecutionStatement.LastExecuteRequest"/> test seam.
+    /// A behavioral "query succeeds" test is not a substitute: a query can succeed
+    /// regardless of what compression value was sent, so the seam is the only way
+    /// to verify the driver's intent without full HTTP interception.
+    ///
+    /// Note: <c>cloudfetch.enabled=false</c> is not yet honored on SEA (requires
+    /// server-side support for ARROW_STREAM with INLINE disposition) and is
+    /// intentionally left as a silent no-op.
+    /// </summary>
+    public class SeaCloudFetchParamsE2ETests : TestBase<DatabricksTestConfiguration, DatabricksTestEnvironment>
+    {
+        public SeaCloudFetchParamsE2ETests(ITestOutputHelper? outputHelper)
+            : base(outputHelper, new DatabricksTestEnvironment.Factory())
+        {
+        }
+
+        private void SkipIfNotConfigured()
+        {
+            Skip.IfNot(Utils.CanExecuteTestConfig(TestConfigVariable), "Test configuration not available");
+        }
+
+        private AdbcConnection CreateRestConnection(Dictionary<string, string> extraProperties)
+        {
+            var properties = new Dictionary<string, string>
+            {
+                [DatabricksParameters.Protocol] = "rest",
+            };
+
+            if (!string.IsNullOrEmpty(TestConfiguration.Uri))
+            {
+                properties[AdbcOptions.Uri] = TestConfiguration.Uri;
+            }
+            else
+            {
+                if (!string.IsNullOrEmpty(TestConfiguration.HostName))
+                    properties[SparkParameters.HostName] = TestConfiguration.HostName;
+                if (!string.IsNullOrEmpty(TestConfiguration.Path))
+                    properties[SparkParameters.Path] = TestConfiguration.Path;
+            }
+
+            if (!string.IsNullOrEmpty(TestConfiguration.Token))
+                properties[SparkParameters.Token] = TestConfiguration.Token;
+            if (!string.IsNullOrEmpty(TestConfiguration.AccessToken))
+                properties[SparkParameters.AccessToken] = TestConfiguration.AccessToken;
+
+            foreach (var kvp in extraProperties)
+                properties[kvp.Key] = kvp.Value;
+
+            var driver = new DatabricksDriver();
+            var database = driver.Open(properties);
+            return database.Connect(null);
+        }
+
+        private static void DrainStream(QueryResult result)
+        {
+            using var stream = result.Stream;
+            while (stream != null && stream.ReadNextRecordBatchAsync().Result != null) { }
+        }
+
+        /// <summary>
+        /// With no explicit params the driver uses the default disposition
+        /// <c>INLINE_OR_EXTERNAL_LINKS</c>. Locks in the no-regression contract
+        /// for the default path.
+        /// </summary>
+        [SkippableFact]
+        public void DefaultConfig_UsesDefaultDisposition()
+        {
+            SkipIfNotConfigured();
+
+            using var connection = CreateRestConnection(new Dictionary<string, string>());
+            using var statement = connection.CreateStatement();
+            statement.SqlQuery = "SELECT 1 AS value";
+            DrainStream(statement.ExecuteQuery());
+
+            var seaStmt = Assert.IsType<StatementExecutionStatement>(statement);
+            Assert.NotNull(seaStmt.LastExecuteRequest);
+            Assert.Equal("INLINE_OR_EXTERNAL_LINKS", seaStmt.LastExecuteRequest!.Disposition);
+        }
+
+        /// <summary>
+        /// With <c>cloudfetch.lz4.enabled=false</c> the driver must clear
+        /// <c>result_compression</c> on the wire (null / unset).
+        /// </summary>
+        [SkippableFact]
+        public void Lz4EnabledFalse_ClearsResultCompression()
+        {
+            SkipIfNotConfigured();
+
+            var extras = new Dictionary<string, string>
+            {
+                [DatabricksParameters.CanDecompressLz4] = "false",
+            };
+
+            using var connection = CreateRestConnection(extras);
+            using var statement = connection.CreateStatement();
+            statement.SqlQuery = "SELECT 1 AS value";
+            DrainStream(statement.ExecuteQuery());
+
+            var seaStmt = Assert.IsType<StatementExecutionStatement>(statement);
+            Assert.NotNull(seaStmt.LastExecuteRequest);
+            Assert.Null(seaStmt.LastExecuteRequest!.ResultCompression);
+        }
+
+        /// <summary>
+        /// With <c>cloudfetch.lz4.enabled=true</c> (default) the driver must
+        /// request <c>LZ4_FRAME</c> compression on the wire.
+        /// </summary>
+        [SkippableFact]
+        public void Lz4EnabledTrue_RequestsLz4Compression()
+        {
+            SkipIfNotConfigured();
+
+            using var connection = CreateRestConnection(new Dictionary<string, string>());
+            using var statement = connection.CreateStatement();
+            statement.SqlQuery = "SELECT 1 AS value";
+            DrainStream(statement.ExecuteQuery());
+
+            var seaStmt = Assert.IsType<StatementExecutionStatement>(statement);
+            Assert.NotNull(seaStmt.LastExecuteRequest);
+            Assert.Equal("LZ4_FRAME", seaStmt.LastExecuteRequest!.ResultCompression);
+        }
+    }
+}