GH-46315 [C#] Apache Arrow Flight Middleware

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

@@ -0,0 +1,54 @@
+// 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.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()
+    {
+        if (await Reader.MoveNext().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/FlightCallOptions.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/FlightExtensions.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;
+    }
+}

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

@@ -285,6 +285,7 @@ public override Task DoGet(FlightTicket ticket, FlightServerRecordBatchStreamWri
     public override Task DoAction(FlightAction action, IAsyncStreamWriter<FlightResult> responseStream, ServerCallContext context)
     {
         Logger?.LogTrace("Executing Flight SQL DoAction: {ActionType}", action.Type);
+
         switch (action.Type)
         {
             case SqlAction.CreateRequest: