feat(csharp/src/Drivers/Databricks): Optimize GetColumnsExtendedAsync via DESC TABLE EXTENDED (#2953)

# Optimize GetColumnsExtendedAsync in Databricks via `DESC TABLE
EXTENDED`

## Motivation

Currently, `HiveServer2Statement.GetColumnsExtendedAsync` will make 3
thrift calls `GetColumns`, `GetPrimaryKeys` and `GetCrossReference` to
get all the information and then join them into one result set. Now
Databricks introduces a SQL `DESC TABLE EXTENDED <table> AS JSON` that
will return all of these information in one query, this can save 2 extra
roundtrips and improve the performance.


### Description

Override `HiveServer2Statement.GetColumnsExtendedAsync` in
`DatabricksStatement` by executing single SQL `DESC TABLE EXTENDED
<table> AS JSON` to get all the required column info for
`GetColumnsExtendedAsync` then combine and join these info into the
result. As this SQL `DESC TABLE EXTENDED <table> AS JSON` is only
available in Databricks Runtime 16.2 or above, it will check the
`ServerProtocolVersion`. if it does not meet the requirement, it will
fallback the implementation of base class.

### Change
- Added `DescTableExtendedResult` that represents the result of `DESC
TABLE EXTENDED <table> AS JSON` (see format
[here](https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-aux-describe-table#json-formatted-output)).
It also
- Parses the string-based `table_constraints` to the structured primary
key and foreign key constraints
- Adds calculated properties `DataType`, `ColumnSize`, `FullTypeName`
which are calculated from the column type and type specific properties
- Changed the access modifiers of some properties and methods in
`HiveServer2Statement` to protected so that they can be used/overridden
in subclass `DatabricksStatement`
   - `PrimaryKeyFields`
   - `ForeignKeyFields`
   - `PrimaryKeyPrefix`
   - `ForeignKeyPrefix`
   - `GetColumnsExtendedAsync`
   - `CreateEmptyExtendedColumnsResult`
- Updated `DatabricksStatement` with the changes below
- Added `GetColumnsExtendedAsync` in `DatabricksStatement`, it executes
query `DESC TABLE EXTENDED <table> AS JSON` to get all the required info
and then json them into the QueryResult
- Moved the column metadata schema creation logic from `GetColumnsAsync`
to a reusable method `CreateColumnMetadataSchema`
- Moved the column metadata data array creation logic from
`GetColumnsAsync` to a reusable method `CreateColumnMetadataEmptyArray`
- Added a Databricks connection Parameter
`adbc.databricks.use_desc_table_extended`
- Added an internal calculated property `CanUseDescTableExtended` in
`DatabricksConnection`, `DatabricksStatement` will call it to decide if
it will override `GetColumnsExtendedAsync`
- Improved the Databricks driver integration test
`StatementTest:CanGetColumnsExtended`
- Setup the required table resources during the test instead of
depending on manual setup before the test
  - Support running multiple test cases from the inputs
- Add a deep result check to make sure all the properties of all the
columns are set correctly

### Testing
- Added Test class `DescTableExtendedResultTest` to cover all the
deserialization and parsing cases from raw result of `DESC TABLE
EXTENDED <table> AS JSON`
- Tested all the `ColumnsExtended` relevant test cases in
`Databricks/StatementTest`

Author: jackyhu-db

csharp/src/Drivers/Apache/Hive2/HiveServer2Statement.cs

@@ -47,12 +47,12 @@ internal class HiveServer2Statement : TracingStatement
             GetColumnsExtendedCommandName;
 
         // Add constants for PK and FK field names and prefixes
-        private static readonly string[] PrimaryKeyFields = new[] { "COLUMN_NAME" };
-        private static readonly string[] ForeignKeyFields = new[] { "PKCOLUMN_NAME", "PKTABLE_CAT", "PKTABLE_SCHEM", "PKTABLE_NAME", "FKCOLUMN_NAME", "FK_NAME", "KEQ_SEQ" };
+        protected static readonly string[] PrimaryKeyFields = new[] { "COLUMN_NAME" };
+        protected static readonly string[] ForeignKeyFields = new[] { "PKCOLUMN_NAME", "PKTABLE_CAT", "PKTABLE_SCHEM", "PKTABLE_NAME", "FKCOLUMN_NAME", "FK_NAME", "KEQ_SEQ" };
         private static readonly string s_assemblyName = ApacheUtility.GetAssemblyName(typeof(HiveServer2Statement));
         private static readonly string s_assemblyVersion = ApacheUtility.GetAssemblyVersion(typeof(HiveServer2Statement));
-        private const string PrimaryKeyPrefix = "PK_";
-        private const string ForeignKeyPrefix = "FK_";
+        protected const string PrimaryKeyPrefix = "PK_";
+        protected const string ForeignKeyPrefix = "FK_";
 
         internal HiveServer2Statement(HiveServer2Connection connection)
             : base(connection)
@@ -704,7 +704,7 @@ protected internal QueryResult EnhanceGetColumnsResult(Schema originalSchema, IR
             return (batches, schema, totalRows);
         }
 
-        private async Task<QueryResult> GetColumnsExtendedAsync(CancellationToken cancellationToken = default)
+        protected virtual async Task<QueryResult> GetColumnsExtendedAsync(CancellationToken cancellationToken = default)
         {
             // 1. Get all three results at once
             var columnsResult = await GetColumnsAsync(cancellationToken);
@@ -797,7 +797,7 @@ await ProcessRelationshipDataSafe(fkResult, ForeignKeyPrefix, "FKCOLUMN_NAME",
         }
 
         // Helper method to create an empty result with the complete extended columns schema
-        private QueryResult CreateEmptyExtendedColumnsResult(Schema baseSchema)
+        protected QueryResult CreateEmptyExtendedColumnsResult(Schema baseSchema)
         {
             // Create the complete schema with all fields
             var allFields = new List<Field>(baseSchema.FieldsList);
@@ -809,7 +809,8 @@ private QueryResult CreateEmptyExtendedColumnsResult(Schema baseSchema)
             // Add FK fields
             foreach (var field in ForeignKeyFields)
             {
-                allFields.Add(new Field(ForeignKeyPrefix + field, StringType.Default, true));
+                IArrowType fieldType = field != "KEQ_SEQ" ? StringType.Default : Int32Type.Default;
+                allFields.Add(new Field(ForeignKeyPrefix + field, fieldType, true));
             }
 
             var combinedSchema = new Schema(allFields, baseSchema.Metadata);

csharp/src/Drivers/Databricks/DatabricksConnection.cs

@@ -56,6 +56,7 @@ internal class DatabricksConnection : SparkHttpConnection
         private long _maxBytesPerFile = DefaultMaxBytesPerFile;
         private const bool DefaultRetryOnUnavailable = true;
         private const int DefaultTemporarilyUnavailableRetryTimeout = 900;
+        private bool _useDescTableExtended = true;
 
         // Default namespace
         private TNamespace? _defaultNamespace;
@@ -145,6 +146,18 @@ private void ValidateProperties()
                 }
             }
 
+            if (Properties.TryGetValue(DatabricksParameters.UseDescTableExtended, out string? useDescTableExtendedStr))
+            {
+                if (bool.TryParse(useDescTableExtendedStr, out bool useDescTableExtended))
+                {
+                    _useDescTableExtended = useDescTableExtended;
+                }
+                else
+                {
+                    throw new ArgumentException($"Parameter '{DatabricksParameters.UseDescTableExtended}' value '{useDescTableExtendedStr}' could not be parsed. Valid values are 'true' and 'false'.");
+                }
+            }
+
             if (Properties.TryGetValue(DatabricksParameters.MaxBytesPerFile, out string? maxBytesPerFileStr))
             {
                 if (!long.TryParse(maxBytesPerFileStr, out long maxBytesPerFileValue))
@@ -223,6 +236,11 @@ private void ValidateProperties()
         /// </summary>
         internal bool EnableMultipleCatalogSupport => _enableMultipleCatalogSupport;
 
+        /// <summary>
+        /// Check if current connection can use `DESC TABLE EXTENDED` query
+        /// </summary>
+        internal bool CanUseDescTableExtended => _useDescTableExtended && ServerProtocolVersion != null && ServerProtocolVersion >= TProtocolVersion.SPARK_CLI_SERVICE_PROTOCOL_V7;
+
         /// <summary>
         /// Gets whether PK/FK metadata call is enabled
         /// </summary>

csharp/src/Drivers/Databricks/DatabricksParameters.cs

@@ -173,6 +173,12 @@ public class DatabricksParameters : SparkParameters
         /// Default value is true if not specified.
         /// </summary>
         public const string EnablePKFK = "adbc.databricks.enable_pk_fk";
+
+        /// <summary>
+        /// Whether to use query DESC TABLE EXTENDED to get extended column metadata when the current DBR supports it
+        /// Default value is true if not specified.
+        /// </summary>
+        public const string UseDescTableExtended = "adbc.databricks.use_desc_table_extended";
     }
 
     /// <summary>