fix(csharp/src/Client): stop blocking on async calls in the ADO.NET wrapper

csharp/src/Client/AdbcCommand.cs

@@ -23,6 +23,7 @@
 using System.Data.SqlTypes;
 using System.Globalization;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Apache.Arrow.Types;
 
@@ -209,6 +210,18 @@ protected override DbDataReader ExecuteDbDataReader(CommandBehavior behavior)
             return ExecuteReader(behavior);
         }
 
+        protected override async Task<DbDataReader> ExecuteDbDataReaderAsync(CommandBehavior behavior, CancellationToken cancellationToken)
+        {
+            bool closeConnection = ValidateReaderBehavior(behavior);
+
+            cancellationToken.ThrowIfCancellationRequested();
+
+            BindParameters();
+            QueryResult result = await AdbcStatement.ExecuteQueryAsync().ConfigureAwait(false);
+
+            return new AdbcDataReader(this, result, this.DecimalBehavior, this.StructBehavior, closeConnection);
+        }
+
         /// <summary>
         /// Executes the reader with the default behavior.
         /// </summary>
@@ -226,21 +239,33 @@ protected override DbDataReader ExecuteDbDataReader(CommandBehavior behavior)
         /// </param>
         /// <returns><see cref="AdbcDataReader"/></returns>
         public new AdbcDataReader ExecuteReader(CommandBehavior behavior)
+        {
+            bool closeConnection = ValidateReaderBehavior(behavior);
+            QueryResult result = this.ExecuteQuery();
+
+            return new AdbcDataReader(this, result, this.DecimalBehavior, this.StructBehavior, closeConnection);
+        }
+
+        /// <summary>
+        /// Validates the behavior and reports whether the connection should be closed
+        /// when the reader is disposed.
+        /// </summary>
+        private bool ValidateReaderBehavior(CommandBehavior behavior)
         {
             if (_disposed)
                 throw new ObjectDisposedException(nameof(AdbcCommand));
 
-            bool closeConnection = (behavior & CommandBehavior.CloseConnection) != 0;
             switch (behavior & ~CommandBehavior.CloseConnection)
             {
                 case CommandBehavior.SchemaOnly:   // The schema is not known until a read happens
                 case CommandBehavior.Default:
-                    QueryResult result = this.ExecuteQuery();
-                    return new AdbcDataReader(this, result, this.DecimalBehavior, this.StructBehavior, closeConnection);
+                    break;
 
                 default:
                     throw new InvalidOperationException($"{behavior} is not supported with this provider");
             }
+
+            return (behavior & CommandBehavior.CloseConnection) != 0;
         }
 
         protected override void Dispose(bool disposing)

csharp/src/Client/AdbcConnection.cs

@@ -551,7 +551,7 @@ public override DataTable GetSchema(Adbc.AdbcConnection adbcConnection, string?[
                     State state = new State(result, indices.ToArray(), loaders.ToArray());
                     while (true)
                     {
-                        using (RecordBatch? batch = stream.ReadNextRecordBatchAsync().Result)
+                        using (RecordBatch? batch = stream.ReadNextRecordBatchAsync().AsTask().GetAwaiter().GetResult())
                         {
                             if (batch == null) { return result; }
 

csharp/src/Client/AdbcDataReader.cs

@@ -44,6 +44,8 @@ namespace Apache.Arrow.Adbc.Client
     /// </summary>
     public sealed class AdbcDataReader : DbDataReader, IDbColumnSchemaGenerator
     {
+        private static readonly Task<bool> s_true = Task.FromResult(true);
+
         private readonly AdbcCommand adbcCommand;
         private readonly bool closeConnection;
         private readonly QueryResult adbcQueryResult;
@@ -336,7 +338,30 @@ public override bool Read()
             // old batch — they must see the exception again immediately.
             this.recordBatch?.Dispose();
             this.recordBatch = null;
-            this.recordBatch = ReadNextRecordBatchAsync().Result;
+
+            this.recordBatch = ReadNextRecordBatchAsync().AsTask().GetAwaiter().GetResult();
+
+            return this.recordBatch != null;
+        }
+
+        public override Task<bool> ReadAsync(CancellationToken cancellationToken)
+        {
+            if (this.recordBatch != null && this.currentRowInRecordBatch < this.recordBatch.Length - 1)
+            {
+                this.currentRowInRecordBatch++;
+                return s_true;
+            }
+
+            return FetchNextBatchAsync(cancellationToken);
+        }
+
+        private async Task<bool> FetchNextBatchAsync(CancellationToken cancellationToken)
+        {
+            // Clear the previous batch first: a caller retrying after a mid-stream error
+            // must see the exception again, never stale rows from the old batch.
+            this.recordBatch?.Dispose();
+            this.recordBatch = null;
+            this.recordBatch = await ReadNextRecordBatchAsync(cancellationToken).ConfigureAwait(false);
 
             return this.recordBatch != null;
         }
@@ -389,18 +414,20 @@ public ReadOnlyCollection<AdbcColumn> GetAdbcColumnSchema()
         /// </summary>
         /// <param name="cancellationToken">An optional cancellation token</param>
         /// <returns><see cref="RecordBatch"/> or null</returns>
-        private ValueTask<RecordBatch?> ReadNextRecordBatchAsync(CancellationToken cancellationToken = default)
+        private async ValueTask<RecordBatch?> ReadNextRecordBatchAsync(CancellationToken cancellationToken = default)
         {
             this.currentRowInRecordBatch = 0;
 
-            RecordBatch? recordBatch = this.adbcQueryResult.Stream?.ReadNextRecordBatchAsync(cancellationToken).Result;
+            RecordBatch? recordBatch = this.adbcQueryResult.Stream is not null
+                ? await this.adbcQueryResult.Stream.ReadNextRecordBatchAsync(cancellationToken).ConfigureAwait(false)
+                : null;
 
             if (recordBatch != null)
             {
                 this.TotalBatches += 1;
             }
 
-            return new ValueTask<RecordBatch?>(recordBatch);
+            return recordBatch;
         }
     }
 }