feat(csharp): implement EnableComplexDatatypeSupport for consistent complex type behavior (PECO-2938)

csharp/src/ComplexTypeSerializingStream.cs

@@ -0,0 +1,182 @@
+/*
+* 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;
+using System.Collections.Generic;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Apache.Arrow;
+using Apache.Arrow.Adbc.Extensions;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Types;
+
+namespace AdbcDrivers.Databricks
+{
+    /// <summary>
+    /// Wraps an <see cref="IArrowArrayStream"/> and converts columns of complex Arrow types
+    /// (LIST, MAP represented as LIST of STRUCTs, STRUCT) into STRING columns containing
+    /// their JSON representation.
+    ///
+    /// This is applied when EnableComplexDatatypeSupport=false (the default), so that SEA
+    /// results match the legacy Thrift behavior of returning JSON strings for complex types.
+    /// </summary>
+    internal sealed class ComplexTypeSerializingStream : IArrowArrayStream
+    {
+        private readonly IArrowArrayStream _inner;
+        private readonly Schema _schema;
+        private readonly HashSet<int> _complexColumnIndices;
+
+        public ComplexTypeSerializingStream(IArrowArrayStream inner)
+        {
+            _inner = inner ?? throw new ArgumentNullException(nameof(inner));
+            (_schema, _complexColumnIndices) = BuildStringSchema(inner.Schema);
+        }
+
+        public Schema Schema => _schema;
+
+        public async ValueTask<RecordBatch?> ReadNextRecordBatchAsync(CancellationToken cancellationToken = default)
+        {
+            var batch = await _inner.ReadNextRecordBatchAsync(cancellationToken).ConfigureAwait(false);
+            if (batch == null)
+                return null;
+
+            if (_complexColumnIndices.Count == 0)
+                return batch;
+
+            return ConvertComplexColumns(batch);
+        }
+
+        public void Dispose() => _inner.Dispose();
+
+        private RecordBatch ConvertComplexColumns(RecordBatch batch)
+        {
+            var arrays = new IArrowArray[batch.ColumnCount];
+            for (int i = 0; i < batch.ColumnCount; i++)
+            {
+                arrays[i] = _complexColumnIndices.Contains(i) ? SerializeToStringArray(batch.Column(i)) : batch.Column(i);
+            }
+            return new RecordBatch(_schema, arrays, batch.Length);
+        }
+
+        private static StringArray SerializeToStringArray(IArrowArray array)
+        {
+            var builder = new StringArray.Builder();
+            for (int i = 0; i < array.Length; i++)
+            {
+                if (array.IsNull(i))
+                    builder.AppendNull();
+                else
+                    builder.Append(JsonSerializer.Serialize(ToObject(array, i)));
+            }
+            return builder.Build();
+        }
+
+        /// <summary>
+        /// Builds a new schema where all complex-type fields are replaced with StringType,
+        /// and returns the set of column indices that were converted.
+        /// </summary>
+        private static (Schema schema, HashSet<int> complexIndices) BuildStringSchema(Schema original)
+        {
+            var fields = new List<Field>(original.FieldsList.Count);
+            var indices = new HashSet<int>();
+
+            for (int i = 0; i < original.FieldsList.Count; i++)
+            {
+                var field = original.FieldsList[i];
+                if (IsComplexType(field.DataType))
+                {
+                    fields.Add(new Field(field.Name, StringType.Default, nullable: true, field.Metadata));
+                    indices.Add(i);
+                }
+                else
+                {
+                    fields.Add(field);
+                }
+            }
+
+            return (new Schema(fields, original.Metadata), indices);
+        }
+
+        private static bool IsComplexType(IArrowType type) =>
+            type is ListType || type is MapType || type is StructType;
+
+        // --- JSON serialization helpers ---
+
+        private static object? ToObject(IArrowArray array, int index)
+        {
+            if (array.IsNull(index))
+                return null;
+
+            // Handle complex types with recursive traversal, and types needing specific
+            // string formatting. All other primitives delegate to ValueAt().
+            return array switch
+            {
+                ListArray la => ToListOrMap(la, index),
+                StructArray sa => ToDict(sa, index),
+                Decimal128Array dec => dec.GetString(index),            // preserve precision as string
+                Date32Array d32 => d32.GetDateTime(index)?.ToString("yyyy-MM-dd"),
+                _ => array.ValueAt(index, StructResultType.Object)      // int, long, float, bool, string, timestamp, etc.
+            };
+        }
+
+        private static object ToListOrMap(ListArray listArray, int index)
+        {
+            var values = listArray.Values;
+            int start = (int)listArray.ValueOffsets[index];
+            int end = (int)listArray.ValueOffsets[index + 1];
+
+            // Arrow MAP is stored as List<Struct<key, value>>
+            if (values is StructArray structValues && IsMapStruct(structValues))
+                return ToMapDict(structValues, start, end);
+
+            var list = new List<object?>();
+            for (int i = start; i < end; i++)
+                list.Add(ToObject(values, i));
+            return list;
+        }
+
+        private static bool IsMapStruct(StructArray structArray)
+        {
+            var type = (StructType)structArray.Data.DataType;
+            return type.Fields.Count == 2 &&
+                   type.Fields[0].Name == "key" &&
+                   type.Fields[1].Name == "value";
+        }
+
+        private static Dictionary<string, object?> ToMapDict(StructArray entries, int start, int end)
+        {
+            var keyArray = entries.Fields[0];
+            var valueArray = entries.Fields[1];
+            var result = new Dictionary<string, object?>();
+            for (int i = start; i < end; i++)
+            {
+                string key = keyArray is StringArray sa ? sa.GetString(i) ?? "null" : "null";
+                result[key] = ToObject(valueArray, i);
+            }
+            return result;
+        }
+
+        private static Dictionary<string, object?> ToDict(StructArray structArray, int index)
+        {
+            var type = (StructType)structArray.Data.DataType;
+            var dict = new Dictionary<string, object?>();
+            for (int i = 0; i < type.Fields.Count; i++)
+                dict[type.Fields[i].Name] = ToObject(structArray.Fields[i], index);
+            return dict;
+        }
+    }
+}

csharp/src/DatabricksConnection.cs

@@ -66,6 +66,7 @@ internal class DatabricksConnection : SparkHttpConnection
         private bool _enableMultipleCatalogSupport = true;
         private bool _enablePKFK = true;
         private bool _runAsyncInThrift = true;
+        private bool _enableComplexDatatypeSupport = false;
 
         // DirectQuery configuration
         private const long DefaultDirectResultMaxBytes = 10 * 1024 * 1024; // 10MB for direct query results size limit
@@ -183,6 +184,7 @@ private void ValidateProperties()
             _canDecompressLz4 = PropertyHelper.GetBooleanPropertyWithValidation(Properties, DatabricksParameters.CanDecompressLz4, _canDecompressLz4);
             _useDescTableExtended = PropertyHelper.GetBooleanPropertyWithValidation(Properties, DatabricksParameters.UseDescTableExtended, _useDescTableExtended);
             _runAsyncInThrift = PropertyHelper.GetBooleanPropertyWithValidation(Properties, DatabricksParameters.EnableRunAsyncInThriftOp, _runAsyncInThrift);
+            _enableComplexDatatypeSupport = PropertyHelper.GetBooleanPropertyWithValidation(Properties, DatabricksParameters.EnableComplexDatatypeSupport, _enableComplexDatatypeSupport);
 
             if (Properties.ContainsKey(DatabricksParameters.MaxBytesPerFile))
             {
@@ -358,6 +360,11 @@ protected internal override bool TrySetGetDirectResults(IRequest request)
         /// </summary>
         public bool RunAsyncInThrift => _runAsyncInThrift;
 
+        /// <summary>
+        /// Whether to return complex types as native Arrow types (true) or JSON strings (false).
+        /// </summary>
+        internal bool EnableComplexDatatypeSupport => _enableComplexDatatypeSupport;
+
         /// <summary>
         /// Gets a value indicating whether to retry requests that receive retryable responses (408, 502, 503, 504) .
         /// </summary>

csharp/src/DatabricksParameters.cs

@@ -376,6 +376,15 @@ public class DatabricksParameters : SparkParameters
         /// Default value is 900 seconds (15 minutes) if not specified.
         /// </summary>
         public const string FeatureFlagCacheTtlSeconds = "adbc.databricks.feature_flag_cache_ttl_seconds";
+
+        /// <summary>
+        /// Whether to return complex types (ARRAY, MAP, STRUCT) as native Arrow types.
+        /// When false (default): complex types are serialized to JSON strings, matching legacy behavior.
+        /// When true: complex types are returned as native Arrow types (ListType, MapType, StructType).
+        /// This applies to both Thrift and SEA protocols, providing consistent behavior across protocols.
+        /// Default value is false if not specified.
+        /// </summary>
+        public const string EnableComplexDatatypeSupport = "adbc.databricks.enable_complex_datatype_support";
     }
 
     /// <summary>

csharp/src/DatabricksStatement.cs

@@ -57,6 +57,7 @@ internal class DatabricksStatement : SparkStatement, IHiveServer2Statement
         private bool enableMultipleCatalogSupport;
         private bool enablePKFK;
         private bool runAsyncInThrift;
+        private bool enableComplexDatatypeSupport;
         private Dictionary<string, string>? confOverlay;
 
         public override long BatchSize { get; protected set; } = DatabricksBatchSizeDefault;
@@ -84,6 +85,7 @@ public DatabricksStatement(DatabricksConnection connection)
             enablePKFK = connection.EnablePKFK;
 
             runAsyncInThrift = connection.RunAsyncInThrift;
+            enableComplexDatatypeSupport = connection.EnableComplexDatatypeSupport;
 
             // Override the Apache base default (500ms) with Databricks-specific poll interval (100ms)
             if (!connection.Properties.ContainsKey(ApacheParameters.PollTimeMilliseconds))
@@ -136,11 +138,7 @@ protected override void SetStatementProperties(TExecuteStatementReq statement)
             {
                 TimestampAsArrow = true,
                 DecimalAsArrow = true,
-
-                // set to false so they return as string
-                // otherwise, they return as ARRAY_TYPE but you can't determine
-                // the object type of the items in the array
-                ComplexTypesAsArrow = false,
+                ComplexTypesAsArrow = enableComplexDatatypeSupport,
                 IntervalTypesAsArrow = false,
             };
 

csharp/src/StatementExecution/StatementExecutionConnection.cs

@@ -63,6 +63,7 @@ internal class StatementExecutionConnection : TracingConnection
         private readonly bool _tracePropagationEnabled;
         private readonly string _traceParentHeaderName;
         private readonly bool _traceStateEnabled;
+        private readonly bool _enableComplexDatatypeSupport;
 
         // Authentication support
         private readonly string? _identityFederationClientId;
@@ -200,6 +201,7 @@ private StatementExecutionConnection(
             _tracePropagationEnabled = PropertyHelper.GetBooleanPropertyWithValidation(properties, DatabricksParameters.TracePropagationEnabled, true);
             _traceParentHeaderName = PropertyHelper.GetStringProperty(properties, DatabricksParameters.TraceParentHeaderName, "traceparent");
             _traceStateEnabled = PropertyHelper.GetBooleanPropertyWithValidation(properties, DatabricksParameters.TraceStateEnabled, false);
+            _enableComplexDatatypeSupport = PropertyHelper.GetBooleanPropertyWithValidation(properties, DatabricksParameters.EnableComplexDatatypeSupport, false);
 
             // Authentication configuration
             if (properties.TryGetValue(DatabricksParameters.IdentityFederationClientId, out string? identityFederationClientId))
@@ -450,6 +452,8 @@ public override void Dispose()
         }
 
         // TracingConnection provides IActivityTracer implementation
+        internal bool EnableComplexDatatypeSupport => _enableComplexDatatypeSupport;
+
         public override string AssemblyVersion => GetType().Assembly.GetName().Version?.ToString() ?? "1.0.0";
         public override string AssemblyName => "AdbcDrivers.Databricks";
     }

csharp/src/StatementExecution/StatementExecutionStatement.cs

@@ -21,6 +21,7 @@
 using System.Net.Http;
 using System.Threading;
 using System.Threading.Tasks;
+using AdbcDrivers.Databricks;
 using AdbcDrivers.Databricks.Reader.CloudFetch;
 using Apache.Arrow;
 using Apache.Arrow.Adbc;
@@ -60,6 +61,9 @@ internal class StatementExecutionStatement : TracingStatement
         // HTTP client for CloudFetch downloads
         private readonly HttpClient _httpClient;
 
+        // Complex type configuration
+        private readonly bool _enableComplexDatatypeSupport;
+
         // Statement state
         private string? _currentStatementId;
         private string? _sqlQuery;
@@ -96,6 +100,7 @@ public StatementExecutionStatement(
             _recyclableMemoryStreamManager = recyclableMemoryStreamManager ?? throw new ArgumentNullException(nameof(recyclableMemoryStreamManager));
             _lz4BufferPool = lz4BufferPool ?? throw new ArgumentNullException(nameof(lz4BufferPool));
             _httpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient));
+            _enableComplexDatatypeSupport = connection.EnableComplexDatatypeSupport;
         }
 
         /// <summary>
@@ -189,6 +194,13 @@ public async Task<QueryResult> ExecuteQueryAsync(CancellationToken cancellationT
             // Create appropriate reader based on result disposition
             IArrowArrayStream reader = CreateReader(response, cancellationToken);
 
+            // When EnableComplexDatatypeSupport=false (default), serialize complex Arrow types to JSON strings
+            // so that SEA behavior matches Thrift (which sets ComplexTypesAsArrow=false).
+            if (!_enableComplexDatatypeSupport)
+            {
+                reader = new ComplexTypeSerializingStream(reader);
+            }
+
             // Get schema from reader
             var schema = reader.Schema;