GH-44800: [C#] Implement Flight SQL Client (#44783)

GH-44800: [C#] Implement Flight SQL Client

## Rationale for this Change

This pull request introduces a **new implementation of `FlightSqlClient` and `PreparedStatement` in C#**. Previously, there was no C# client for Flight SQL, leaving a significant gap for .NET developers who wished to interact with Flight SQL servers.

The implementation aligns with the existing **C++ Flight SQL client** API, ensuring consistent and familiar behavior across languages and providing a robust client for the Apache Arrow ecosystem in .NET.

---

## What's Included in this PR?

### Key Features

1. **`FlightSqlClient`**:
   - Provides query execution (`ExecuteAsync`, `ExecuteUpdateAsync`) and schema retrieval (`GetCatalogsAsync`, `GetDbSchemasAsync`, etc.).
   - Implements metadata operations for catalogs, schemas, tables, and more.
   - Fully integrated with gRPC and Apache Arrow ecosystems.
   - Supports extensibility for advanced features like transactions.

2. **`PreparedStatement`**:
   - Implements parameterized query execution (`SetParameters`, `ExecuteAsync`, and `ExecuteUpdateAsync`).
   - Supports lifecycle management (`CloseAsync`) for effective resource handling.
   - Aligns with the prepared statement design in the C++ client.

### API Consistency

This implementation mirrors the **C++ Flight SQL client** to ensure API alignment across supported languages:
- Consistent naming conventions and parameter semantics.
- Ensures .NET developers can work seamlessly with existing Flight SQL servers.

---

## Are These Changes Tested?

### Testing Overview

1. **Unit Tests**:
   - Added tests for query execution and parameter binding in `PreparedStatement`.
   - Verified schema retrieval methods like `GetCatalogsAsync` and `GetDbSchemasAsync`.

2. **Integration Tests**:
   - Tested against a live Flight SQL server to validate query execution, schema retrieval, and metadata operations.

3. **End-to-End Tests**:
   - Covered real-world scenarios for parameterized updates and queries, ensuring robustness.

### Example Test Cases
- Verify that parameterized queries return correct results with valid input.
- Ensure schema retrieval throws appropriate exceptions for invalid descriptors.
- Validate row counts after `ExecuteUpdateAsync`.

---

## Are There Any Breaking Changes?

This PR introduces **new functionality** and does not affect any existing features. There are **no breaking changes**.

---

## Are There Any User-Facing Changes?

### New Capabilities
1. **FlightSqlClient**:
   - Query execution and schema retrieval for SQL queries on Flight SQL servers.
   - Metadata retrieval for catalogs, schemas, tables, and more.

2. **PreparedStatement**:
   - Supports parameterized queries with proper parameter binding.
   - Provides robust lifecycle management and execution.

### API Consistency
- Aligns with the C++ Flight SQL client for interoperability and familiar API design.

---

## Additional Notes

- This PR **does not include transaction support** at this stage, as it requires additional server-side capabilities.
- All methods follow idiomatic C# practices, including `async/await` for non-blocking operations.
- Extensible for future enhancements, including advanced features like savepoints.

---

## Resources

- **C++ Flight SQL Client Reference**: [Apache Arrow Flight SQL Documentation](https://arrow.apache.org/docs/)
- **Apache Arrow Contribution Guide**: [Contributing to Apache Arrow](https://arrow.apache.org/docs/dev/developers/guide/)

---

## Feedback and Suggestions

Thank you for reviewing this contribution! Suggestions and feedback are welcome to ensure the implementation meets the project's standards and requirements.
* GitHub Issue: #44800

Lead-authored-by: HackPoint <[email protected]>
Co-authored-by: Genady Shmunik <[email protected]>
Co-authored-by: HackP0!nt <[email protected]>
Co-authored-by: HackPoint <[email protected]>
Co-authored-by: Curt Hagenlocher <[email protected]>
Signed-off-by: Curt Hagenlocher <[email protected]>

Author: 3 people

csharp/Apache.Arrow.sln

@@ -17,8 +17,6 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow.Flight", "src\
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow.Flight.AspNetCore", "src\Apache.Arrow.Flight.AspNetCore\Apache.Arrow.Flight.AspNetCore.csproj", "{E4F74938-E8FF-4AC1-A495-FEE95FC1EFDF}"
 EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow.IntegrationTest", "test\Apache.Arrow.IntegrationTest\Apache.Arrow.IntegrationTest.csproj", "{E8264B7F-B680-4A55-939B-85DB628164BB}"
-EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow.Compression", "src\Apache.Arrow.Compression\Apache.Arrow.Compression.csproj", "{B62E77D2-D0B0-4C0C-BA78-1C117DE4C299}"
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow.Compression.Tests", "test\Apache.Arrow.Compression.Tests\Apache.Arrow.Compression.Tests.csproj", "{5D7FF380-B7DF-4752-B415-7C08C70C9F06}"
@@ -29,6 +27,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Apache.Arrow.Flight.Sql", "
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Apache.Arrow.Flight.IntegrationTest", "test\Apache.Arrow.Flight.IntegrationTest\Apache.Arrow.Flight.IntegrationTest.csproj", "{7E66CBB4-D921-41E7-A98A-7C6DEA521696}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Apache.Arrow.IntegrationTest", "test\Apache.Arrow.IntegrationTest\Apache.Arrow.IntegrationTest.csproj", "{E8264B7F-B680-4A55-939B-85DB628164BB}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU

csharp/src/Apache.Arrow.Flight.Sql/Client/FlightSqlClient.cs

@@ -0,0 +1,55 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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.Threading;
+using System.Threading.Tasks;
+using Apache.Arrow.Flight.Client;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class DoPutResult
+{
+    public FlightClientRecordBatchStreamWriter Writer { get; }
+    public IAsyncStreamReader<FlightPutResult> Reader { get; }
+
+    public DoPutResult(FlightClientRecordBatchStreamWriter writer, IAsyncStreamReader<FlightPutResult> reader)
+    {
+        Writer = writer;
+        Reader = reader;
+    }
+    
+    /// <summary>
+    /// Reads the metadata asynchronously from the reader.
+    /// </summary>
+    /// <returns>A ByteString containing the metadata read from the reader.</returns>
+    public async Task<Google.Protobuf.ByteString> ReadMetadataAsync(CancellationToken cancellationToken = default)
+    {
+        if (await Reader.MoveNext(cancellationToken).ConfigureAwait(false))
+        {
+            return Reader.Current.ApplicationMetadata;
+        }
+        throw new RpcException(new Status(StatusCode.Internal, "No metadata available in the response stream."));
+    }
+    
+    /// <summary>
+    /// Completes the writer by signaling the end of the writing process.
+    /// </summary>
+    /// <returns>A Task representing the completion of the writer.</returns>
+    public async Task CompleteAsync()
+    {
+        await Writer.CompleteAsync().ConfigureAwait(false);
+    }
+}

csharp/src/Apache.Arrow.Flight.Sql/DoPutResult.cs

@@ -0,0 +1,38 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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.Buffers;
+using System.Threading;
+using Grpc.Core;
+
+namespace Apache.Arrow.Flight.Sql;
+
+public class FlightCallOptions
+{
+    public FlightCallOptions()
+    {
+        Timeout = TimeSpan.FromSeconds(-1);
+    }
+    
+    // Implement any necessary options for RPC calls
+    public Metadata Headers { get; set; } = new();
+
+    /// <summary>
+    /// Gets or sets the optional timeout for this call.
+    /// Negative durations mean an implementation-defined default behavior will be used instead.
+    /// </summary>
+    public TimeSpan Timeout { get; set; }
+}

csharp/src/Apache.Arrow.Flight.Sql/FlightCallOptions.cs

@@ -0,0 +1,41 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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 Google.Protobuf;
+using Google.Protobuf.WellKnownTypes;
+
+namespace Apache.Arrow.Flight.Sql;
+
+internal static class FlightExtensions
+{
+    public static byte[] PackAndSerialize(this IMessage command) => Any.Pack(command).ToByteArray();
+
+    public static T ParseAndUnpack<T>(this ByteString source) where T : IMessage<T>, new() =>
+        Any.Parser.ParseFrom(source).Unpack<T>();
+
+    public static int ExtractRowCount(this RecordBatch batch)
+    {
+        if (batch.ColumnCount == 0) return 0;
+        int length = batch.Column(0).Length;
+        foreach (var column in batch.Arrays)
+        {
+            if (column.Length != length)
+                throw new InvalidOperationException("Inconsistent column lengths in RecordBatch.");
+        }
+
+        return length;
+    }
+}