diff --git a/csharp/src/DatabricksParameters.cs b/csharp/src/DatabricksParameters.cs index 891c82c8c..25ac5093c 100644 --- a/csharp/src/DatabricksParameters.cs +++ b/csharp/src/DatabricksParameters.cs @@ -34,12 +34,28 @@ public class DatabricksParameters : SparkParameters /// /// 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 + /// ARROW_STREAM results with INLINE 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. /// public const string UseCloudFetch = "adbc.databricks.cloudfetch.enabled"; /// /// 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 result_compression on the + /// ExecuteStatement request (overriding any + /// value). Mirrors JDBC's CompressionCodec.NONE branch. /// public const string CanDecompressLz4 = "adbc.databricks.cloudfetch.lz4.enabled"; diff --git a/csharp/src/StatementExecution/StatementExecutionConnection.cs b/csharp/src/StatementExecution/StatementExecutionConnection.cs index bd57dd1de..0d4471cb1 100644 --- a/csharp/src/StatementExecution/StatementExecutionConnection.cs +++ b/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) && diff --git a/csharp/src/StatementExecution/StatementExecutionStatement.cs b/csharp/src/StatementExecution/StatementExecutionStatement.cs index b910097af..7aaaa85be 100644 --- a/csharp/src/StatementExecution/StatementExecutionStatement.cs +++ b/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 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 ExecuteUpdateInternalAsync(CancellationToken ca IsMetadata = false, QueryTags = ParseQueryTags(_queryTags) }; + _lastExecuteRequest = request; // Execute the statement var response = await _client.ExecuteStatementAsync(request, cancellationToken).ConfigureAwait(false); diff --git a/csharp/test/E2E/StatementExecution/SeaCloudFetchParamsE2ETests.cs b/csharp/test/E2E/StatementExecution/SeaCloudFetchParamsE2ETests.cs new file mode 100644 index 000000000..d04e1f343 --- /dev/null +++ b/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 +{ + /// + /// E2E tests proving that adbc.databricks.cloudfetch.lz4.enabled is + /// honored on the SEA path (PECO-3056). + /// + /// Tests assert on the result_compression field of the request the + /// driver actually built, exposed via the internal + /// 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: cloudfetch.enabled=false 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. + /// + public class SeaCloudFetchParamsE2ETests : TestBase + { + 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 extraProperties) + { + var properties = new Dictionary + { + [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) { } + } + + /// + /// With no explicit params the driver uses the default disposition + /// INLINE_OR_EXTERNAL_LINKS. Locks in the no-regression contract + /// for the default path. + /// + [SkippableFact] + public void DefaultConfig_UsesDefaultDisposition() + { + SkipIfNotConfigured(); + + using var connection = CreateRestConnection(new Dictionary()); + using var statement = connection.CreateStatement(); + statement.SqlQuery = "SELECT 1 AS value"; + DrainStream(statement.ExecuteQuery()); + + var seaStmt = Assert.IsType(statement); + Assert.NotNull(seaStmt.LastExecuteRequest); + Assert.Equal("INLINE_OR_EXTERNAL_LINKS", seaStmt.LastExecuteRequest!.Disposition); + } + + /// + /// With cloudfetch.lz4.enabled=false the driver must clear + /// result_compression on the wire (null / unset). + /// + [SkippableFact] + public void Lz4EnabledFalse_ClearsResultCompression() + { + SkipIfNotConfigured(); + + var extras = new Dictionary + { + [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(statement); + Assert.NotNull(seaStmt.LastExecuteRequest); + Assert.Null(seaStmt.LastExecuteRequest!.ResultCompression); + } + + /// + /// With cloudfetch.lz4.enabled=true (default) the driver must + /// request LZ4_FRAME compression on the wire. + /// + [SkippableFact] + public void Lz4EnabledTrue_RequestsLz4Compression() + { + SkipIfNotConfigured(); + + using var connection = CreateRestConnection(new Dictionary()); + using var statement = connection.CreateStatement(); + statement.SqlQuery = "SELECT 1 AS value"; + DrainStream(statement.ExecuteQuery()); + + var seaStmt = Assert.IsType(statement); + Assert.NotNull(seaStmt.LastExecuteRequest); + Assert.Equal("LZ4_FRAME", seaStmt.LastExecuteRequest!.ResultCompression); + } + } +}