This is an automated email from the ASF dual-hosted git repository.
CurtHagenlocher pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-adbc.git
The following commit(s) were added to refs/heads/main by this push:
new 2be84b1a5 feat(csharp): add async and cancellation support to the base
classes (#4895)
2be84b1a5 is described below
commit 2be84b1a5cd0319af107081a6e0233484a38588c
Author: Curt Hagenlocher <[email protected]>
AuthorDate: Thu Oct 15 12:23:23 2026 -0700
feat(csharp): add async and cancellation support to the base classes (#4895)
## What's changed
This adds asynchronous methods that take a `CancellationToken` to the C#
base classes (`AdbcDriver`, `AdbcDatabase`, `AdbcConnection` and
`AdbcStatement`). The base classes also now implement
`IAsyncDisposable`.
This is the third of several PRs implementing ADBC 1.1. #4884 fixed bugs
in the C importer and exporter, and #4885 added the 1.1 APIs to the base
classes. The Client changes and wider DuckDB tests come next.
### Backwards compatibility
Every new member is virtual and has a default implementation, so
existing drivers compile and behave as before:
- **Default async methods** run the synchronous method on the thread
pool. When the token is cancelled, they call the object's `Cancel`
method, so a driver that supports cancellation stops work without
overriding anything.
- **Cancellation is reported only when it took effect.** An
`OperationCanceledException` is thrown when the token was cancelled
and either `Cancel` succeeded or the driver reported
`AdbcStatusCode.Cancelled`. Otherwise the driver's own error is
rethrown, so a driver that can't cancel doesn't have its failures
hidden.
- **Existing async overrides are honored.** The existing
`ExecuteQueryAsync()` and `ExecuteUpdateAsync()` keep their
signatures. The new `ExecuteQueryAsync(CancellationToken)` and
`ExecuteUpdateAsync(CancellationToken)` call them by default, adding
cancellation through `Cancel`. That way a driver that already
overrides the parameterless versions keeps its asynchronous
implementation.
- **Existing calls bind the same way.** No new overload changes which
method an existing call resolves to.
### New APIs
- **`AdbcDriver`:** `OpenAsync`.
- **`AdbcDatabase`:** `ConnectAsync` and `ConnectWithTypedOptionsAsync`.
- **`AdbcConnection`:**
- `GetInfoAsync`, `GetObjectsAsync`, `GetTableSchemaAsync` and
`GetTableTypesAsync`
- `GetStatisticsAsync` and `GetStatisticsNamesAsync`
- `ReadPartitionAsync`
- `CommitAsync` and `RollbackAsync`
- `SetOptionAsync` and `GetOptionAsync`
- **`AdbcStatement`:**
- `ExecuteQueryAsync(CancellationToken)` and
`ExecuteUpdateAsync(CancellationToken)`
- `ExecuteSchemaAsync` and `ExecutePartitionedAsync`
- `GetParameterSchemaAsync` and `PrepareAsync`
- **`DisposeAsync`** on the database, connection and statement.
There are no C interop changes. Imported drivers get the default async
methods, which call the synchronous ones. Those already include the 1.0
fallbacks from #4885.
## Testing
- **`BaseClassDefaultsTests`:** covers the defaults with drivers that
implement only the synchronous members:
- tokens cancelled before and during a call
- drivers whose `Cancel` succeeds or fails
- existing `ExecuteQueryAsync()`/`ExecuteUpdateAsync()` overrides
- `DisposeAsync`
- **`ExportedDriverRoundTrip11Tests`:**
- a `CancellationToken` cancels a statement running in a driver that
was exported and imported again
- async option round trips
- async calls on an imported 1.0 driver report the same statuses as
the synchronous calls
- **AoT interop tests:** `ExecuteQueryAsync` with a token round trips
through the native AoT fixture.
- **Results:**
- The full solution builds without warnings, and
`dotnet format --verify-no-changes` is clean.
- The core test project passes on net10.0, net8.0 and net472.
- The AoT interop tests pass against a fixture published from this
branch.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
csharp/src/Apache.Arrow.Adbc/AdbcAsync.cs | 156 +++++
csharp/src/Apache.Arrow.Adbc/AdbcConnection.cs | 175 +++++-
csharp/src/Apache.Arrow.Adbc/AdbcDatabase.cs | 53 +-
csharp/src/Apache.Arrow.Adbc/AdbcDriver.cs | 17 +
csharp/src/Apache.Arrow.Adbc/AdbcStatement.cs | 187 +++++-
.../AotFixtureTests.cs | 15 +
.../BaseClassDefaultsTests.cs | 697 ++++++++++++++++++++-
.../Apache.Arrow.Adbc.Tests/Client/ClientTests.cs | 2 +-
.../ExportedDriverRoundTrip11Tests.cs | 77 ++-
.../Apache.Arrow.Adbc.Tests/ImportedDuckDbTests.cs | 28 +-
10 files changed, 1369 insertions(+), 38 deletions(-)
diff --git a/csharp/src/Apache.Arrow.Adbc/AdbcAsync.cs
b/csharp/src/Apache.Arrow.Adbc/AdbcAsync.cs
new file mode 100644
index 000000000..8457a5f42
--- /dev/null
+++ b/csharp/src/Apache.Arrow.Adbc/AdbcAsync.cs
@@ -0,0 +1,156 @@
+/*
+ * 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.Runtime.ExceptionServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Apache.Arrow.Adbc
+{
+ /// <summary>
+ /// Helpers for the default implementations of the asynchronous methods on
+ /// the ADBC base classes. Those defaults run the synchronous method on the
+ /// thread pool and translate a <see cref="CancellationToken"/> into a call
+ /// to the object's <c>Cancel</c> method.
+ /// </summary>
+ internal static class AdbcAsync
+ {
+ private static readonly Action<object?> s_cancel = TryCancel;
+
+ public static Task<T> Run<T>(Func<T> operation, Action? cancel,
CancellationToken cancellationToken)
+ {
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return Task.FromCanceled<T>(cancellationToken);
+ }
+
+ return Task.Run(() =>
+ {
+ CancelState state = new CancelState(cancel);
+ Exception failure;
+ using (Register(state, cancellationToken))
+ {
+ try
+ {
+ return operation();
+ }
+ catch (Exception e) when (!(e is
OperationCanceledException))
+ {
+ failure = e;
+ }
+ }
+
+ // Disposing the registration waits for an in-progress Cancel
to finish
+ throw Translate(failure, state, cancellationToken);
+ }, cancellationToken);
+ }
+
+ public static Task Run(Action operation, Action? cancel,
CancellationToken cancellationToken)
+ {
+ return Run<bool>(() => { operation(); return true; }, cancel,
cancellationToken);
+ }
+
+ /// <summary>
+ /// Attaches cancellation to a task produced by a driver override that
+ /// doesn't itself accept a <see cref="CancellationToken"/>. The token
is
+ /// registered before <paramref name="start"/> is called, so an
override
+ /// that blocks before returning its task can still be cancelled.
+ /// </summary>
+ public static async Task<T> WithCancellation<T>(Func<Task<T>> start,
Action? cancel, CancellationToken cancellationToken)
+ {
+ CancelState state = new CancelState(cancel);
+ Exception failure;
+ using (Register(state, cancellationToken))
+ {
+ try
+ {
+ Task<T> task = start();
+ if (!task.IsCompleted)
+ {
+ // Cancel may complete the task, which would otherwise
resume this method
+ // inline on the cancelling thread. Disposing the
registration from inside
+ // its own callback doesn't wait for Cancel to return,
so Translate could
+ // misjudge whether Cancel succeeded. Resuming on the
thread pool instead
+ // makes the disposal wait for the callback to finish.
+ await task.ContinueWith(_ => { },
CancellationToken.None, TaskContinuationOptions.None,
TaskScheduler.Default).ConfigureAwait(false);
+ }
+ return await task.ConfigureAwait(false);
+ }
+ catch (Exception e) when (!(e is OperationCanceledException))
+ {
+ failure = e;
+ }
+ }
+
+ throw Translate(failure, state, cancellationToken);
+ }
+
+ private static CancellationTokenRegistration Register(CancelState
state, CancellationToken cancellationToken)
+ {
+ if (state.Cancel == null || !cancellationToken.CanBeCanceled)
+ {
+ return default;
+ }
+ return cancellationToken.Register(s_cancel, state);
+ }
+
+ /// <summary>
+ /// Reports a failure as cancellation only if cancellation plausibly
caused it: the
+ /// token was cancelled and either the driver accepted the call to
Cancel or the
+ /// failure itself says the operation was cancelled. Otherwise, a
driver that can't
+ /// cancel would have its own errors hidden behind <see
cref="OperationCanceledException"/>.
+ /// </summary>
+ private static Exception Translate(Exception failure, CancelState
state, CancellationToken cancellationToken)
+ {
+ bool cancelled = cancellationToken.IsCancellationRequested &&
+ (state.Delivered || (failure is AdbcException adbc &&
adbc.Status == AdbcStatusCode.Cancelled));
+ if (cancelled)
+ {
+ return new OperationCanceledException(failure.Message,
failure, cancellationToken);
+ }
+
+ ExceptionDispatchInfo.Capture(failure).Throw();
+ return failure; // not reached
+ }
+
+ private static void TryCancel(object? state)
+ {
+ CancelState cancelState = (CancelState)state!;
+ try
+ {
+ cancelState.Cancel!();
+ cancelState.Delivered = true;
+ }
+ catch (Exception)
+ {
+ // Cancellation is best-effort. A driver that doesn't support
+ // cancellation simply runs the operation to completion.
+ }
+ }
+
+ private sealed class CancelState
+ {
+ public CancelState(Action? cancel) { Cancel = cancel; }
+
+ public Action? Cancel { get; }
+
+ // Set when Cancel returned without throwing
+ public volatile bool Delivered;
+ }
+ }
+}
diff --git a/csharp/src/Apache.Arrow.Adbc/AdbcConnection.cs
b/csharp/src/Apache.Arrow.Adbc/AdbcConnection.cs
index 785d508b0..2ef04b5e4 100644
--- a/csharp/src/Apache.Arrow.Adbc/AdbcConnection.cs
+++ b/csharp/src/Apache.Arrow.Adbc/AdbcConnection.cs
@@ -17,6 +17,8 @@
using System;
using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
using Apache.Arrow.Ipc;
namespace Apache.Arrow.Adbc
@@ -25,7 +27,12 @@ namespace Apache.Arrow.Adbc
/// Provides methods for query execution, managing prepared statements,
/// using transactions, and so on.
/// </summary>
- public abstract class AdbcConnection : IDisposable
+ /// <remarks>
+ /// Each asynchronous method has a default implementation that runs the
+ /// corresponding synchronous method on the thread pool and responds to its
+ /// <see cref="CancellationToken"/> by calling <see cref="Cancel"/>.
+ /// </remarks>
+ public abstract class AdbcConnection : IDisposable, IAsyncDisposable
{
private bool _autoCommit = true;
private bool _readOnly = false;
@@ -39,6 +46,15 @@ namespace Apache.Arrow.Adbc
throw AdbcException.NotImplemented("Connection does not support
transactions");
}
+ /// <summary>
+ /// Asynchronously commit the pending transaction.
+ /// </summary>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ public virtual Task CommitAsync(CancellationToken cancellationToken =
default)
+ {
+ return AdbcAsync.Run(Commit, Cancel, cancellationToken);
+ }
+
/// <summary>
/// Create a new statement that can be executed.
/// </summary>
@@ -71,6 +87,18 @@ namespace Apache.Arrow.Adbc
{
}
+ /// <summary>
+ /// Asynchronously releases the resources used by the connection.
+ /// </summary>
+ /// <remarks>
+ /// The default implementation calls <see cref="Dispose"/>.
+ /// </remarks>
+ public virtual ValueTask DisposeAsync()
+ {
+ Dispose();
+ return default;
+ }
+
/// <summary>
/// Get metadata about the driver/database.
/// </summary>
@@ -90,6 +118,22 @@ namespace Apache.Arrow.Adbc
throw AdbcException.NotImplemented("Connection does not support
GetInfo");
}
+ /// <summary>
+ /// Asynchronously get metadata about the driver/database.
+ /// </summary>
+ /// <param name="codes">
+ /// The metadata items to fetch. An empty list requests all of the
items the
+ /// driver supports.
+ /// </param>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ /// <returns>
+ /// Metadata about the driver and/or database
+ /// </returns>
+ public virtual Task<IArrowArrayStream>
GetInfoAsync(IReadOnlyList<AdbcInfoCode> codes, CancellationToken
cancellationToken = default)
+ {
+ return AdbcAsync.Run(() => GetInfo(codes), Cancel,
cancellationToken);
+ }
+
/// <summary>
/// Get a hierarchical view of all catalogs, database schemas, tables,
/// and columns.
@@ -131,6 +175,29 @@ namespace Apache.Arrow.Adbc
IReadOnlyList<string>? tableTypes,
string? columnNamePattern);
+ /// <summary>
+ /// Asynchronously get a hierarchical view of all catalogs, database
schemas,
+ /// tables, and columns.
+ /// </summary>
+ /// <remarks>
+ /// See <see cref="GetObjects"/> for a description of the parameters.
+ /// </remarks>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ public virtual Task<IArrowArrayStream> GetObjectsAsync(
+ GetObjectsDepth depth,
+ string? catalogPattern,
+ string? dbSchemaPattern,
+ string? tableNamePattern,
+ IReadOnlyList<string>? tableTypes,
+ string? columnNamePattern,
+ CancellationToken cancellationToken = default)
+ {
+ return AdbcAsync.Run(
+ () => GetObjects(depth, catalogPattern, dbSchemaPattern,
tableNamePattern, tableTypes, columnNamePattern),
+ Cancel,
+ cancellationToken);
+ }
+
public enum GetObjectsDepth
{
/// <summary>
@@ -169,11 +236,38 @@ namespace Apache.Arrow.Adbc
/// </param>
public abstract Schema GetTableSchema(string? catalog, string?
dbSchema, string tableName);
+ /// <summary>
+ /// Asynchronously get the Arrow schema of a database table.
+ /// </summary>
+ /// <param name="catalog">
+ /// The catalog of the table (or null).
+ /// </param>
+ /// <param name="dbSchema">
+ /// The database schema of the table (or null).
+ /// </param>
+ /// <param name="tableName">
+ /// The table name.
+ /// </param>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ public virtual Task<Schema> GetTableSchemaAsync(string? catalog,
string? dbSchema, string tableName, CancellationToken cancellationToken =
default)
+ {
+ return AdbcAsync.Run(() => GetTableSchema(catalog, dbSchema,
tableName), Cancel, cancellationToken);
+ }
+
/// <summary>
/// Get a list of table types supported by the database.
/// </summary>
public abstract IArrowArrayStream GetTableTypes();
+ /// <summary>
+ /// Asynchronously get a list of table types supported by the database.
+ /// </summary>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ public virtual Task<IArrowArrayStream>
GetTableTypesAsync(CancellationToken cancellationToken = default)
+ {
+ return AdbcAsync.Run(GetTableTypes, Cancel, cancellationToken);
+ }
+
/// <summary>
/// Options may be set before AdbcConnectionInit. Some drivers may
/// support setting options after initialization as well.
@@ -185,6 +279,24 @@ namespace Apache.Arrow.Adbc
throw AdbcException.NotImplemented("Connection does not support
setting options");
}
+ /// <summary>
+ /// Asynchronously sets a string-valued option.
+ /// </summary>
+ /// <param name="key">Option name</param>
+ /// <param name="value">Option value</param>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ /// <remarks>
+ /// Some options, such as <see
cref="AdbcOptions.Connection.Autocommit"/> or
+ /// <see cref="AdbcOptions.Connection.CurrentCatalog"/>, may require a
round
+ /// trip to the server. Drivers for which that's the case should
override this
+ /// method. The default implementation runs <see
cref="SetOption(string, string)"/>
+ /// on the thread pool.
+ /// </remarks>
+ public virtual Task SetOptionAsync(string key, string value,
CancellationToken cancellationToken = default)
+ {
+ return AdbcAsync.Run(() => SetOption(key, value), Cancel,
cancellationToken);
+ }
+
/// <summary>
/// Sets an integer-valued option.
/// </summary>
@@ -275,6 +387,23 @@ namespace Apache.Arrow.Adbc
return value ?? throw AdbcOptionValue.NotFound(key);
}
+ /// <summary>
+ /// Asynchronously gets the value of a string-valued option.
+ /// </summary>
+ /// <param name="key">Option name</param>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ /// <returns>The option value.</returns>
+ /// <remarks>
+ /// Some options, such as <see
cref="AdbcOptions.Connection.CurrentDbSchema"/>,
+ /// may require a round trip to the server. Drivers for which that's
the case
+ /// should override this method. The default implementation runs
+ /// <see cref="GetOption(string)"/> on the thread pool.
+ /// </remarks>
+ public virtual Task<string> GetOptionAsync(string key,
CancellationToken cancellationToken = default)
+ {
+ return AdbcAsync.Run(() => GetOption(key), Cancel,
cancellationToken);
+ }
+
/// <summary>
/// Gets the value of an integer-valued option.
/// </summary>
@@ -325,6 +454,18 @@ namespace Apache.Arrow.Adbc
throw AdbcException.NotImplemented("Connection does not support
partitions");
}
+ /// <summary>
+ /// Asynchronously create a result set from a serialized
PartitionDescriptor.
+ /// </summary>
+ /// <param name="partition">
+ /// The partition descriptor.
+ /// </param>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ public virtual Task<IArrowArrayStream>
ReadPartitionAsync(PartitionDescriptor partition, CancellationToken
cancellationToken = default)
+ {
+ return AdbcAsync.Run(() => ReadPartition(partition), Cancel,
cancellationToken);
+ }
+
/// <summary>
/// Rollback the pending transaction.
/// </summary>
@@ -333,6 +474,15 @@ namespace Apache.Arrow.Adbc
throw AdbcException.NotImplemented("Connection does not support
transactions");
}
+ /// <summary>
+ /// Asynchronously rollback the pending transaction.
+ /// </summary>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ public virtual Task RollbackAsync(CancellationToken cancellationToken
= default)
+ {
+ return AdbcAsync.Run(Rollback, Cancel, cancellationToken);
+ }
+
/// <summary>
/// Gets or sets the autocommit state.
/// </summary>
@@ -409,6 +559,15 @@ namespace Apache.Arrow.Adbc
throw AdbcException.NotImplemented("Connection does not support
statistics");
}
+ /// <summary>
+ /// Asynchronously gets the names of statistics specific to this
driver.
+ /// </summary>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ public virtual Task<IArrowArrayStream>
GetStatisticsNamesAsync(CancellationToken cancellationToken = default)
+ {
+ return AdbcAsync.Run(GetStatisticsNames, Cancel,
cancellationToken);
+ }
+
/// <summary>
/// Gets statistics about the data distribution of table(s)
/// </summary>
@@ -421,5 +580,19 @@ namespace Apache.Arrow.Adbc
{
throw AdbcException.NotImplemented("Connection does not support
statistics");
}
+
+ /// <summary>
+ /// Asynchronously gets statistics about the data distribution of
table(s)
+ /// </summary>
+ /// <param name="catalogPattern">The catalog or null. May be a search
pattern.</param>
+ /// <param name="schemaPattern">The schema or null. May be a search
pattern.</param>
+ /// <param name="tableNamePattern">The table name or null. May be a
search pattern.</param>
+ /// <param name="approximate">If false, consumer desires exact
statistics regardless of cost</param>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ /// <returns>A result describing the statistics for the
table(s)</returns>
+ public virtual Task<IArrowArrayStream> GetStatisticsAsync(string?
catalogPattern, string? schemaPattern, string tableNamePattern, bool
approximate, CancellationToken cancellationToken = default)
+ {
+ return AdbcAsync.Run(() => GetStatistics(catalogPattern,
schemaPattern, tableNamePattern, approximate), Cancel, cancellationToken);
+ }
}
}
diff --git a/csharp/src/Apache.Arrow.Adbc/AdbcDatabase.cs
b/csharp/src/Apache.Arrow.Adbc/AdbcDatabase.cs
index d5f11b0a7..46b71df42 100644
--- a/csharp/src/Apache.Arrow.Adbc/AdbcDatabase.cs
+++ b/csharp/src/Apache.Arrow.Adbc/AdbcDatabase.cs
@@ -17,6 +17,8 @@
using System;
using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
namespace Apache.Arrow.Adbc
{
@@ -27,7 +29,7 @@ namespace Apache.Arrow.Adbc
/// For example, in-memory databases can place ownership of the actual
/// database in this object.
/// </summary>
- public abstract class AdbcDatabase : IDisposable
+ public abstract class AdbcDatabase : IDisposable, IAsyncDisposable
{
/// <summary>
/// Options are generally set before opening a database. Some drivers
may
@@ -227,6 +229,55 @@ namespace Apache.Arrow.Adbc
return connection;
}
+ /// <summary>
+ /// Asynchronously create a new connection to the database using
options that
+ /// aren't all strings.
+ /// </summary>
+ /// <param name="options">
+ /// Additional options to use when connecting. See
+ /// <see cref="ConnectWithTypedOptions(IReadOnlyDictionary{string,
object})"/>.
+ /// </param>
+ /// <param name="cancellationToken">
+ /// A token used to cancel the operation before it starts. Once
started, the operation
+ /// runs to completion, because databases and drivers have no way to
interrupt it.
+ /// </param>
+ /// <remarks>
+ /// Added in ADBC 1.1.0. The default implementation runs
+ /// <see cref="ConnectWithTypedOptions(IReadOnlyDictionary{string,
object})"/>
+ /// on the thread pool.
+ /// </remarks>
+ public virtual Task<AdbcConnection>
ConnectWithTypedOptionsAsync(IReadOnlyDictionary<string, object>? options,
CancellationToken cancellationToken = default)
+ {
+ return AdbcAsync.Run(() => ConnectWithTypedOptions(options), null,
cancellationToken);
+ }
+
+ /// <summary>
+ /// Asynchronously create a new connection to the database.
+ /// </summary>
+ /// <param name="options">
+ /// Additional options to use when connecting.
+ /// </param>
+ /// <param name="cancellationToken">
+ /// A token used to cancel the operation before it starts. Once
started, the operation
+ /// runs to completion, because databases and drivers have no way to
interrupt it.
+ /// </param>
+ public virtual Task<AdbcConnection>
ConnectAsync(IReadOnlyDictionary<string, string>? options, CancellationToken
cancellationToken = default)
+ {
+ return AdbcAsync.Run(() => Connect(options), null,
cancellationToken);
+ }
+
public virtual void Dispose() { }
+
+ /// <summary>
+ /// Asynchronously releases the resources used by the database.
+ /// </summary>
+ /// <remarks>
+ /// The default implementation calls <see cref="Dispose"/>.
+ /// </remarks>
+ public virtual ValueTask DisposeAsync()
+ {
+ Dispose();
+ return default;
+ }
}
}
diff --git a/csharp/src/Apache.Arrow.Adbc/AdbcDriver.cs
b/csharp/src/Apache.Arrow.Adbc/AdbcDriver.cs
index f07314787..188b64e08 100644
--- a/csharp/src/Apache.Arrow.Adbc/AdbcDriver.cs
+++ b/csharp/src/Apache.Arrow.Adbc/AdbcDriver.cs
@@ -17,6 +17,8 @@
using System;
using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
namespace Apache.Arrow.Adbc
{
@@ -88,6 +90,21 @@ namespace Apache.Arrow.Adbc
return database;
}
+ /// <summary>
+ /// Asynchronously open a database via this driver.
+ /// </summary>
+ /// <param name="parameters">
+ /// Driver-specific parameters.
+ /// </param>
+ /// <param name="cancellationToken">
+ /// A token used to cancel the operation before it starts. Once
started, the operation
+ /// runs to completion, because databases and drivers have no way to
interrupt it.
+ /// </param>
+ public virtual Task<AdbcDatabase>
OpenAsync(IReadOnlyDictionary<string, string> parameters, CancellationToken
cancellationToken = default)
+ {
+ return AdbcAsync.Run(() => Open(parameters), null,
cancellationToken);
+ }
+
public virtual void Dispose()
{
}
diff --git a/csharp/src/Apache.Arrow.Adbc/AdbcStatement.cs
b/csharp/src/Apache.Arrow.Adbc/AdbcStatement.cs
index 09866e3fb..0b8fac04e 100644
--- a/csharp/src/Apache.Arrow.Adbc/AdbcStatement.cs
+++ b/csharp/src/Apache.Arrow.Adbc/AdbcStatement.cs
@@ -16,6 +16,7 @@
*/
using System;
+using System.Threading;
using System.Threading.Tasks;
using Apache.Arrow.Ipc;
@@ -26,7 +27,12 @@ namespace Apache.Arrow.Adbc
/// may be used multiple times and can be reconfigured (e.g. they can
/// be reused to execute multiple different queries).
/// </summary>
- public abstract class AdbcStatement : IDisposable
+ /// <remarks>
+ /// Each asynchronous method has a default implementation that runs the
+ /// corresponding synchronous method on the thread pool and responds to its
+ /// <see cref="CancellationToken"/> by calling <see cref="Cancel"/>.
+ /// </remarks>
+ public abstract class AdbcStatement : IDisposable, IAsyncDisposable
{
public AdbcStatement()
{
@@ -83,9 +89,52 @@ namespace Apache.Arrow.Adbc
/// A <see cref="ValueTuple"/> where the first item is the number
/// of records and the second is the <see cref="IArrowArrayStream"/>.
/// </returns>
- public virtual async ValueTask<QueryResult> ExecuteQueryAsync()
+ /// <remarks>
+ /// <para>New code should call <see
cref="ExecuteQueryAsync(CancellationToken)"/>.</para>
+ /// <para>The default implementation runs <see cref="ExecuteQuery"/>
on the thread pool.
+ /// It doesn't call <see
cref="ExecuteQueryAsync(CancellationToken)"/>, because that
+ /// overload's default implementation calls this one. A driver that
overrides
+ /// <see cref="ExecuteQueryAsync(CancellationToken)"/> should also
override this
+ /// overload to call it with <see cref="CancellationToken.None"/>, so
that callers of
+ /// either overload get the driver's asynchronous
implementation.</para>
+ /// </remarks>
+ public virtual ValueTask<QueryResult> ExecuteQueryAsync()
{
- return await Task.Run(() => ExecuteQuery());
+ // When called from ExecuteQueryAsync(CancellationToken), that
overload registers Cancel
+ return new ValueTask<QueryResult>(AdbcAsync.Run(ExecuteQuery,
null, TakeForwardedToken()));
+ }
+
+ /// <summary>
+ /// Executes the statement and returns a tuple containing the number
+ /// of records and the <see cref="IArrowArrayStream"/>.
+ /// </summary>
+ /// <param name="cancellationToken">
+ /// A token used to cancel the operation. Cancellation calls <see
cref="Cancel"/>.
+ /// </param>
+ /// <returns>
+ /// A <see cref="ValueTuple"/> where the first item is the number
+ /// of records and the second is the <see cref="IArrowArrayStream"/>.
+ /// </returns>
+ /// <remarks>
+ /// The default implementation calls <see
cref="ExecuteQueryAsync()"/>, so that drivers
+ /// written before this overload existed keep their asynchronous
behavior, and calls
+ /// <see cref="Cancel"/> if the token is cancelled.
+ /// </remarks>
+ public virtual ValueTask<QueryResult>
ExecuteQueryAsync(CancellationToken cancellationToken)
+ {
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return new
ValueTask<QueryResult>(Task.FromCanceled<QueryResult>(cancellationToken));
+ }
+
+ if (!cancellationToken.CanBeCanceled)
+ {
+ return ExecuteQueryAsync();
+ }
+ return new ValueTask<QueryResult>(AdbcAsync.WithCancellation(
+ () => Forward(cancellationToken, () =>
ExecuteQueryAsync()).AsTask(),
+ Cancel,
+ cancellationToken));
}
/// <summary>
@@ -98,6 +147,17 @@ namespace Apache.Arrow.Adbc
throw AdbcException.NotImplemented("Statement does not support
ExecuteSchema");
}
+ /// <summary>
+ /// Asynchronously analyzes the statement and returns the schema of
the result set
+ /// that would be expected if the statement were to be executed.
+ /// </summary>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ /// <returns>An Arrow <see cref="Schema"/> describing the result
set.</returns>
+ public virtual Task<Schema> ExecuteSchemaAsync(CancellationToken
cancellationToken = default)
+ {
+ return AdbcAsync.Run(ExecuteSchema, Cancel, cancellationToken);
+ }
+
/// <summary>
/// Executes an update command and returns the number of
/// records effected.
@@ -109,10 +169,84 @@ namespace Apache.Arrow.Adbc
/// Executes an update command and returns the number of
/// records effected.
/// </summary>
- /// <exception cref="NotImplementedException"></exception>
- public virtual async Task<UpdateResult> ExecuteUpdateAsync()
+ /// <remarks>
+ /// <para>New code should call <see
cref="ExecuteUpdateAsync(CancellationToken)"/>.</para>
+ /// <para>The default implementation runs <see cref="ExecuteUpdate"/>
on the thread pool.
+ /// It doesn't call <see
cref="ExecuteUpdateAsync(CancellationToken)"/>, because that
+ /// overload's default implementation calls this one. A driver that
overrides
+ /// <see cref="ExecuteUpdateAsync(CancellationToken)"/> should also
override this
+ /// overload to call it with <see cref="CancellationToken.None"/>, so
that callers of
+ /// either overload get the driver's asynchronous
implementation.</para>
+ /// </remarks>
+ public virtual Task<UpdateResult> ExecuteUpdateAsync()
{
- return await Task.Run(() => ExecuteUpdate());
+ // When called from ExecuteUpdateAsync(CancellationToken), that
overload registers Cancel
+ return AdbcAsync.Run(ExecuteUpdate, null, TakeForwardedToken());
+ }
+
+ /// <summary>
+ /// Executes an update command and returns the number of
+ /// records effected.
+ /// </summary>
+ /// <param name="cancellationToken">
+ /// A token used to cancel the operation. Cancellation calls <see
cref="Cancel"/>.
+ /// </param>
+ /// <remarks>
+ /// The default implementation calls <see
cref="ExecuteUpdateAsync()"/>, so that drivers
+ /// written before this overload existed keep their asynchronous
behavior, and calls
+ /// <see cref="Cancel"/> if the token is cancelled.
+ /// </remarks>
+ public virtual Task<UpdateResult> ExecuteUpdateAsync(CancellationToken
cancellationToken)
+ {
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return Task.FromCanceled<UpdateResult>(cancellationToken);
+ }
+
+ if (!cancellationToken.CanBeCanceled)
+ {
+ return ExecuteUpdateAsync();
+ }
+ return AdbcAsync.WithCancellation(
+ () => Forward(cancellationToken, () => ExecuteUpdateAsync()),
+ Cancel,
+ cancellationToken);
+ }
+
+ // The token overloads forward to the parameterless ones so that
legacy overrides of those
+ // still run. When the parameterless overload isn't overridden, its
default implementation
+ // picks up the token here, on the same thread, so that queued work
which is cancelled
+ // before it starts never runs. The token can't be passed as an
argument without changing
+ // the parameterless overload's signature.
+ [ThreadStatic] private static AdbcStatement? t_forwardingStatement;
+ [ThreadStatic] private static CancellationToken t_forwardedToken;
+
+ private T Forward<T>(CancellationToken cancellationToken, Func<T> call)
+ {
+ t_forwardingStatement = this;
+ t_forwardedToken = cancellationToken;
+ try
+ {
+ return call();
+ }
+ finally
+ {
+ t_forwardingStatement = null;
+ t_forwardedToken = default;
+ }
+ }
+
+ private CancellationToken TakeForwardedToken()
+ {
+ if (!ReferenceEquals(t_forwardingStatement, this))
+ {
+ return CancellationToken.None;
+ }
+
+ CancellationToken token = t_forwardedToken;
+ t_forwardingStatement = null;
+ t_forwardedToken = default;
+ return token;
}
/// <summary>
@@ -124,6 +258,16 @@ namespace Apache.Arrow.Adbc
throw AdbcException.NotImplemented("Statement does not support
ExecutePartitioned");
}
+ /// <summary>
+ /// Asynchronously execute a result set-generating query and get a
list of
+ /// partitions of the result set.
+ /// </summary>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ public virtual Task<PartitionedResult>
ExecutePartitionedAsync(CancellationToken cancellationToken = default)
+ {
+ return AdbcAsync.Run(ExecutePartitioned, Cancel,
cancellationToken);
+ }
+
/// <summary>
/// Get the schema for bound parameters.
/// </summary>
@@ -132,6 +276,15 @@ namespace Apache.Arrow.Adbc
throw AdbcException.NotImplemented("Statement does not support
GetParameterSchema");
}
+ /// <summary>
+ /// Asynchronously get the schema for bound parameters.
+ /// </summary>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ public virtual Task<Schema> GetParameterSchemaAsync(CancellationToken
cancellationToken = default)
+ {
+ return AdbcAsync.Run(GetParameterSchema, Cancel,
cancellationToken);
+ }
+
/// <summary>
/// Turn this statement into a prepared statement to be
/// executed multiple times.
@@ -141,6 +294,16 @@ namespace Apache.Arrow.Adbc
throw AdbcException.NotImplemented("Statement does not support
Prepare");
}
+ /// <summary>
+ /// Asynchronously turn this statement into a prepared statement to be
+ /// executed multiple times.
+ /// </summary>
+ /// <param name="cancellationToken">A token used to cancel the
operation.</param>
+ public virtual Task PrepareAsync(CancellationToken cancellationToken =
default)
+ {
+ return AdbcAsync.Run(Prepare, Cancel, cancellationToken);
+ }
+
/// <summary>
/// Set a string option on a statement.
/// </summary>
@@ -249,6 +412,18 @@ namespace Apache.Arrow.Adbc
{
}
+ /// <summary>
+ /// Asynchronously releases the resources used by the statement.
+ /// </summary>
+ /// <remarks>
+ /// The default implementation calls <see cref="Dispose"/>.
+ /// </remarks>
+ public virtual ValueTask DisposeAsync()
+ {
+ Dispose();
+ return default;
+ }
+
/// <summary>
/// Attempts to cancel an in-progress operation on a connection.
/// </summary>
diff --git
a/csharp/test/AotInterop/Apache.Arrow.Adbc.TestFixture.Tests/AotFixtureTests.cs
b/csharp/test/AotInterop/Apache.Arrow.Adbc.TestFixture.Tests/AotFixtureTests.cs
index 3a98bc5f6..5afc90d17 100644
---
a/csharp/test/AotInterop/Apache.Arrow.Adbc.TestFixture.Tests/AotFixtureTests.cs
+++
b/csharp/test/AotInterop/Apache.Arrow.Adbc.TestFixture.Tests/AotFixtureTests.cs
@@ -275,5 +275,20 @@ namespace Apache.Arrow.Adbc.TestFixture.Tests
Assert.Equal("fixture.detail", detail.Key);
Assert.Equal(new byte[] { 1, 2, 3 }, detail.Value);
}
+
+ [SkippableFact]
+ public async Task ExecuteQueryAsyncWithTokenRoundTrips()
+ {
+ using AdbcDriver driver = LoadFixture();
+ using AdbcDatabase db = driver.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = await db.ConnectAsync(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "SELECT 1";
+
+ using var cts = new System.Threading.CancellationTokenSource();
+ QueryResult result = await stmt.ExecuteQueryAsync(cts.Token);
+ using IArrowArrayStream stream = result.Stream!;
+ Assert.Equal("key", stream.Schema.FieldsList[0].Name);
+ }
}
}
diff --git a/csharp/test/Apache.Arrow.Adbc.Tests/BaseClassDefaultsTests.cs
b/csharp/test/Apache.Arrow.Adbc.Tests/BaseClassDefaultsTests.cs
index dcf3e88a9..d944751ce 100644
--- a/csharp/test/Apache.Arrow.Adbc.Tests/BaseClassDefaultsTests.cs
+++ b/csharp/test/Apache.Arrow.Adbc.Tests/BaseClassDefaultsTests.cs
@@ -17,23 +17,237 @@
using System;
using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
using Apache.Arrow.Ipc;
using Xunit;
namespace Apache.Arrow.Adbc.Tests
{
/// <summary>
- /// Tests the default implementations of the ADBC 1.1.0 members of the ADBC
- /// base classes.
+ /// Tests the default implementations of the ADBC 1.1.0 and asynchronous
+ /// members of the ADBC base classes.
/// </summary>
public class BaseClassDefaultsTests
{
+ [Fact]
+ public async Task ExecuteQueryAsyncDefaultsToExecuteQuery()
+ {
+ var stmt = new SyncOnlyStatement();
+
+ await stmt.ExecuteQueryAsync();
+ await stmt.ExecuteQueryAsync(CancellationToken.None);
+ await stmt.ExecuteUpdateAsync();
+ await stmt.ExecuteUpdateAsync(CancellationToken.None);
+
+ Assert.Equal(2, stmt.QueryCount);
+ Assert.Equal(2, stmt.UpdateCount);
+ }
+
+ [Fact]
+ public async Task TokenOverloadUsesLegacyAsyncOverride()
+ {
+ var stmt = new LegacyAsyncStatement();
+ using var cts = new CancellationTokenSource();
+
+ QueryResult result = await stmt.ExecuteQueryAsync(cts.Token);
+ UpdateResult update = await stmt.ExecuteUpdateAsync(cts.Token);
+
+ Assert.Equal(LegacyAsyncStatement.QueryRows, result.RowCount);
+ Assert.Equal(LegacyAsyncStatement.UpdateRows, update.AffectedRows);
+ Assert.Equal(0, stmt.SyncCount);
+ }
+
+ [Fact]
+ public async Task LegacyOverrideCallingBaseAfterAwaitDoesNotRecurse()
+ {
+ var stmt = new LegacyCallsBaseStatement();
+
+ Task<QueryResult> task =
stmt.ExecuteQueryAsync(CancellationToken.None).AsTask();
+ Assert.Same(task, await Task.WhenAny(task,
Task.Delay(TimeSpan.FromSeconds(30))));
+ QueryResult result = await task;
+
+ Assert.Equal(1, stmt.OverrideCount);
+ Assert.Equal(1, stmt.SyncCount);
+ Assert.Equal(7, result.RowCount);
+ }
+
+ [Fact]
+ public async Task TokenOverloadUsesNewAsyncOverride()
+ {
+ var stmt = new NewAsyncStatement();
+
+ QueryResult result = await
stmt.ExecuteQueryAsync(CancellationToken.None);
+ UpdateResult update = await
stmt.ExecuteUpdateAsync(CancellationToken.None);
+
+ Assert.Equal(NewAsyncStatement.QueryRows, result.RowCount);
+ Assert.Equal(NewAsyncStatement.UpdateRows, update.AffectedRows);
+ Assert.Equal(0, stmt.SyncCount);
+ }
+
+ [Fact]
+ public async Task
ParameterlessOverloadDoesNotForwardToTheTokenOverload()
+ {
+ // Forwarding goes only from the token overload to the
parameterless one, so a
+ // driver that overrides only the token overload isn't reached
this way.
+ var stmt = new NewAsyncStatement();
+
+ await stmt.ExecuteQueryAsync();
+ await stmt.ExecuteUpdateAsync();
+
+ Assert.Equal(2, stmt.SyncCount);
+ }
+
+ [Fact]
+ public async Task RecommendedNewDriverIsReachedThroughBothOverloads()
+ {
+ var stmt = new RecommendedAsyncStatement();
+
+ Assert.Equal(NewAsyncStatement.QueryRows, (await
stmt.ExecuteQueryAsync()).RowCount);
+ Assert.Equal(NewAsyncStatement.QueryRows, (await
stmt.ExecuteQueryAsync(CancellationToken.None)).RowCount);
+ Assert.Equal(NewAsyncStatement.UpdateRows, (await
stmt.ExecuteUpdateAsync()).AffectedRows);
+ Assert.Equal(NewAsyncStatement.UpdateRows, (await
stmt.ExecuteUpdateAsync(CancellationToken.None)).AffectedRows);
+ Assert.Equal(0, stmt.SyncCount);
+ }
+
+ [Fact]
+ public async Task LegacyOverrideCallingBaseRunsOnceWhenCalledDirectly()
+ {
+ var stmt = new LegacyCallsBaseStatement();
+
+ QueryResult result = await stmt.ExecuteQueryAsync();
+
+ Assert.Equal(1, stmt.OverrideCount);
+ Assert.Equal(1, stmt.SyncCount);
+ Assert.Equal(7, result.RowCount);
+ }
+
+ [Fact]
+ public async Task NewOverrideCallingParameterlessBaseDoesNotRecurse()
+ {
+ var stmt = new NewCallsParameterlessBaseStatement();
+
+ Task<QueryResult> task =
stmt.ExecuteQueryAsync(CancellationToken.None).AsTask();
+ Assert.Same(task, await Task.WhenAny(task,
Task.Delay(TimeSpan.FromSeconds(30))));
+
+ Assert.Equal(1, stmt.OverrideCount);
+ Assert.Equal(1, stmt.SyncCount);
+ }
+
+ [Fact]
+ public async Task CancellationCallsCancelAndThrowsOperationCanceled()
+ {
+ var stmt = new BlockingStatement();
+ using var cts = new CancellationTokenSource();
+
+ ValueTask<QueryResult> pending = stmt.ExecuteQueryAsync(cts.Token);
+ await WaitForStartAsync(stmt.Started.Task);
+ cts.Cancel();
+
+ OperationCanceledException ex = await
Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await pending);
+ Assert.Equal(cts.Token, ex.CancellationToken);
+ Assert.IsType<AdbcException>(ex.InnerException);
+ Assert.True(stmt.CancelCalled);
+ }
+
+ [Fact]
+ public async Task
UpdateCancellationCallsCancelAndThrowsOperationCanceled()
+ {
+ var stmt = new BlockingStatement();
+ using var cts = new CancellationTokenSource();
+
+ Task<UpdateResult> pending = stmt.ExecuteUpdateAsync(cts.Token);
+ await WaitForStartAsync(stmt.Started.Task);
+ cts.Cancel();
+
+ OperationCanceledException ex = await
Assert.ThrowsAnyAsync<OperationCanceledException>(() => pending);
+ Assert.Equal(cts.Token, ex.CancellationToken);
+ Assert.True(stmt.CancelCalled);
+ }
+
+ [Fact]
+ public async Task CancellationReachesALegacyAsyncOverride()
+ {
+ var stmt = new LegacyBlockingStatement();
+ using var cts = new CancellationTokenSource();
+
+ // The token overload forwards to the legacy override and calls
Cancel on cancellation
+ ValueTask<QueryResult> pending = stmt.ExecuteQueryAsync(cts.Token);
+ await WaitForStartAsync(stmt.Started.Task);
+ cts.Cancel();
+
+ OperationCanceledException ex = await
Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await pending);
+ Assert.Equal(cts.Token, ex.CancellationToken);
+ Assert.IsType<AdbcException>(ex.InnerException);
+ Assert.True(stmt.CancelCalled);
+ }
+
+ [Fact]
+ public async Task PreCancelledTokenDoesNotExecute()
+ {
+ var stmt = new SyncOnlyStatement();
+ using var cts = new CancellationTokenSource();
+ cts.Cancel();
+
+ await Assert.ThrowsAnyAsync<OperationCanceledException>(async ()
=> await stmt.ExecuteQueryAsync(cts.Token));
+ await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
stmt.ExecuteSchemaAsync(cts.Token));
+
+ Assert.Equal(0, stmt.QueryCount);
+ }
+
+ [Fact]
+ public async Task ConnectionAsyncCancellationCallsConnectionCancel()
+ {
+ var conn = new BlockingConnection();
+ using var cts = new CancellationTokenSource();
+
+ Task<IArrowArrayStream> pending =
conn.GetTableTypesAsync(cts.Token);
+ await WaitForStartAsync(conn.Started.Task);
+ cts.Cancel();
+
+ await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
pending);
+ Assert.True(conn.CancelCalled);
+ }
+
+ [Fact]
+ public async Task CancellationIsBestEffortWhenCancelIsNotImplemented()
+ {
+ // SyncOnlyStatement doesn't override Cancel, so once started the
operation runs to completion
+ var stmt = new SyncOnlyStatement { Delay =
TimeSpan.FromMilliseconds(200) };
+ using var cts = new CancellationTokenSource();
+
+ ValueTask<QueryResult> pending = stmt.ExecuteQueryAsync(cts.Token);
+ await WaitForStartAsync(stmt.Started.Task);
+ cts.Cancel();
+ QueryResult result = await pending;
+
+ Assert.Equal(1, stmt.QueryCount);
+ Assert.Equal(0, result.RowCount);
+ }
+
+ [Fact]
+ public async Task
CancellationDoesNotHideErrorsFromDriversThatCannotCancel()
+ {
+ // FailingStatement doesn't override Cancel, so the default throws
NotImplemented.
+ // The failure is the driver's own and must not be reported as
cancellation.
+ var stmt = new FailingStatement();
+ using var cts = new CancellationTokenSource();
+
+ ValueTask<QueryResult> pending = stmt.ExecuteQueryAsync(cts.Token);
+ await WaitForStartAsync(stmt.Started.Task);
+ cts.Cancel();
+ stmt.Fail.Set();
+
+ AdbcException ex = await Assert.ThrowsAsync<AdbcException>(async
() => await pending);
+ Assert.Equal(AdbcStatusCode.InvalidArgument, ex.Status);
+ }
+
[Fact]
public void DoubleOptionsRoundTripThroughStrings()
{
// "R" formatting doesn't round-trip this value on .NET Framework
const double value = 0.84551240822557006;
- var stmt = new OptionsStatement();
+ var stmt = new SyncOnlyStatement();
stmt.SetOptionDouble("d", value);
Assert.Equal(value, stmt.GetOptionDouble("d"));
@@ -42,7 +256,7 @@ namespace Apache.Arrow.Adbc.Tests
[Fact]
public void TypedOptionsDefaultToStringOptions()
{
- var stmt = new OptionsStatement();
+ var stmt = new SyncOnlyStatement();
stmt.SetOptionInt64("i", -12);
stmt.SetOptionDouble("d", 0.1);
@@ -61,7 +275,7 @@ namespace Apache.Arrow.Adbc.Tests
[Fact]
public void ConnectionGetOptionAnswersStandardOptions()
{
- var conn = new RecordingConnection();
+ var conn = new BlockingConnection();
Assert.Equal(AdbcOptions.Enabled,
conn.GetOption(AdbcOptions.Connection.Autocommit));
Assert.Equal(AdbcOptions.Disabled,
conn.GetOption(AdbcOptions.Connection.ReadOnly));
@@ -70,10 +284,49 @@ namespace Apache.Arrow.Adbc.Tests
Assert.Equal(AdbcStatusCode.NotFound,
Assert.Throws<AdbcException>(() => conn.GetOption("unknown")).Status);
}
+ [Fact]
+ public async Task ConnectionOptionAsyncDefaultsToSyncOptions()
+ {
+ var conn = new BlockingConnection();
+
+ await conn.SetOptionAsync("key", "value");
+ Assert.Equal("value", conn.Options["key"]);
+
+ Assert.Equal(AdbcOptions.Enabled, await
conn.GetOptionAsync(AdbcOptions.Connection.Autocommit));
+ AdbcException e = await Assert.ThrowsAsync<AdbcException>(() =>
conn.GetOptionAsync("unknown"));
+ Assert.Equal(AdbcStatusCode.NotFound, e.Status);
+ }
+
+ [Fact]
+ public async Task ConnectionOptionAsyncHonorsAPreCancelledToken()
+ {
+ var conn = new BlockingConnection();
+ using var cts = new CancellationTokenSource();
+ cts.Cancel();
+
+ await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
conn.SetOptionAsync("key", "value", cts.Token));
+ await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
conn.GetOptionAsync(AdbcOptions.Connection.Autocommit, cts.Token));
+ Assert.Empty(conn.Options);
+ }
+
+ [Fact]
+ public async Task
ConnectionSetOptionAsyncCancellationCallsConnectionCancel()
+ {
+ var conn = new BlockingConnection();
+ using var cts = new CancellationTokenSource();
+
+ Task pending = conn.SetOptionAsync(BlockingConnection.BlockingKey,
"value", cts.Token);
+ await WaitForStartAsync(conn.Started.Task);
+ cts.Cancel();
+
+ await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
pending);
+ Assert.True(conn.CancelCalled);
+ }
+
[Fact]
public void CurrentCatalogSetterDefaultsToSetOption()
{
- var conn = new RecordingConnection();
+ var conn = new BlockingConnection();
conn.CurrentCatalog = "cat";
conn.CurrentDbSchema = "schema";
@@ -98,7 +351,7 @@ namespace Apache.Arrow.Adbc.Tests
});
Assert.Equal(new Dictionary<string, string> { { "s", "text" }, {
"b", "false" }, { "l", "4" }, { "d", "2.5" } }, db.LastStringOptions);
- var binary = ((RecordingConnection)conn).BinaryOptions;
+ var binary = ((BlockingConnection)conn).BinaryOptions;
Assert.Equal(new byte[] { 1, 2 }, binary["bytes"]);
Assert.Equal(new byte[] { 3 }, binary["memory"]);
@@ -106,6 +359,25 @@ namespace Apache.Arrow.Adbc.Tests
Assert.Null(db.LastStringOptions);
}
+ [Fact]
+ public async Task
ConnectWithTypedOptionsAsyncDefaultsToConnectWithTypedOptions()
+ {
+ var db = new ConnectingDatabase();
+
+ using AdbcConnection conn = await
db.ConnectWithTypedOptionsAsync(new Dictionary<string, object>
+ {
+ { "l", 4L },
+ { "bytes", new byte[] { 1 } },
+ });
+
+ Assert.Equal(new Dictionary<string, string> { { "l", "4" } },
db.LastStringOptions);
+ Assert.Equal(new byte[] { 1 },
((BlockingConnection)conn).BinaryOptions["bytes"]);
+
+ using var cts = new CancellationTokenSource();
+ cts.Cancel();
+ await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
db.ConnectWithTypedOptionsAsync(null, cts.Token));
+ }
+
[Fact]
public void
ConnectWithTypedOptionsDisposesTheConnectionIfABinaryOptionFails()
{
@@ -113,7 +385,7 @@ namespace Apache.Arrow.Adbc.Tests
Assert.Throws<AdbcException>(() => db.ConnectWithTypedOptions(new
Dictionary<string, object>
{
- { RecordingConnection.RejectedKey, new byte[] { 1 } },
+ { BlockingConnection.RejectedKey, new byte[] { 1 } },
}));
Assert.True(db.LastConnection!.Disposed);
}
@@ -155,6 +427,75 @@ namespace Apache.Arrow.Adbc.Tests
Assert.Equal("false", driver.Parameters["b"]);
}
+ [Fact]
+ public async Task
CancelThatCompletesALegacyTaskInlineIsReportedAsCancellation()
+ {
+ // Cancel faults the legacy task synchronously, which used to
resume the
+ // cancellation helper inline before Cancel had returned
+ var stmt = new InlineFaultingLegacyStatement();
+ using var cts = new CancellationTokenSource();
+
+ ValueTask<QueryResult> pending = stmt.ExecuteQueryAsync(cts.Token);
+ // Cancel from a thread without a SynchronizationContext, where
the legacy task's
+ // continuations may run inline (xUnit's own context would prevent
that)
+ await Task.Run(() => cts.Cancel());
+
+ OperationCanceledException ex = await
Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await pending);
+ Assert.IsType<TimeoutException>(ex.InnerException);
+ }
+
+ [Fact]
+ public async Task
SynchronouslyBlockingLegacyQueryOverrideCanBeCancelled()
+ {
+ var stmt = new SynchronousLegacyStatement();
+ using var cts = new CancellationTokenSource();
+
+ // The override blocks before returning its task, so the call has
to run elsewhere
+ Task<QueryResult> pending = Task.Run(() =>
stmt.ExecuteQueryAsync(cts.Token).AsTask());
+ await WaitForStartAsync(stmt.Started.Task);
+ cts.Cancel();
+
+ await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
pending);
+ Assert.True(stmt.CancelCalled);
+ }
+
+ [Fact]
+ public async Task
SynchronouslyBlockingLegacyUpdateOverrideCanBeCancelled()
+ {
+ var stmt = new SynchronousLegacyStatement();
+ using var cts = new CancellationTokenSource();
+
+ Task<UpdateResult> pending = Task.Run(() =>
stmt.ExecuteUpdateAsync(cts.Token));
+ await WaitForStartAsync(stmt.Started.Task);
+ cts.Cancel();
+
+ await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
pending);
+ Assert.True(stmt.CancelCalled);
+ }
+
+ [Fact]
+ public async Task QueryCancelledBeforeTheDefaultWorkStartsDoesNotRun()
+ {
+ using var queryCts = new CancellationTokenSource();
+ var queryStmt = new CancelThenCallBaseStatement(queryCts);
+ await Assert.ThrowsAnyAsync<OperationCanceledException>(async ()
=> await queryStmt.ExecuteQueryAsync(queryCts.Token));
+ Assert.Equal(0, queryStmt.QueryCount);
+
+ using var updateCts = new CancellationTokenSource();
+ var updateStmt = new CancelThenCallBaseStatement(updateCts);
+ await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
updateStmt.ExecuteUpdateAsync(updateCts.Token));
+ Assert.Equal(0, updateStmt.UpdateCount);
+ }
+
+ [Fact]
+ public async Task DisposeAsyncCallsDispose()
+ {
+ var stmt = new SyncOnlyStatement();
+ await stmt.DisposeAsync();
+ Assert.True(stmt.Disposed);
+ }
+
+
[Fact]
public void DriverOpenWithObjectsAppliesBinaryValuesAfterOpening()
{
@@ -180,21 +521,325 @@ namespace Apache.Arrow.Adbc.Tests
Assert.True(driver.LastDatabase!.Disposed);
}
- private sealed class OptionsStatement : AdbcStatement
+ // Waits without blocking a thread-pool thread, which the operation
under test needs
+ private static async Task WaitForStartAsync(Task started)
{
+ Assert.Same(started, await Task.WhenAny(started,
Task.Delay(TimeSpan.FromSeconds(30))));
+ }
+
+ private sealed class SyncOnlyStatement : AdbcStatement
+ {
+ public int QueryCount;
+ public int UpdateCount;
+ public bool Disposed;
+ public TimeSpan Delay;
+ public TaskCompletionSource<bool> Started { get; } = new
TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
public Dictionary<string, string> Options { get; } = new
Dictionary<string, string>();
public override void SetOption(string key, string value) =>
Options[key] = value;
public override string GetOption(string key) =>
Options.TryGetValue(key, out string? v) ? v : base.GetOption(key);
- public override QueryResult ExecuteQuery() => throw new
NotImplementedException();
- public override UpdateResult ExecuteUpdate() => throw new
NotImplementedException();
+ public override QueryResult ExecuteQuery()
+ {
+ Started.TrySetResult(true);
+ Thread.Sleep(Delay);
+ Interlocked.Increment(ref QueryCount);
+ return new QueryResult(0, null!);
+ }
+
+ public override UpdateResult ExecuteUpdate()
+ {
+ Interlocked.Increment(ref UpdateCount);
+ return new UpdateResult(0);
+ }
+
+ public override Schema ExecuteSchema() => throw new
InvalidOperationException("should not be called");
+
+ public override void Dispose() => Disposed = true;
+ }
+
+ private sealed class FailingStatement : AdbcStatement
+ {
+ public TaskCompletionSource<bool> Started { get; } = new
TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ public ManualResetEventSlim Fail { get; } = new
ManualResetEventSlim();
+
+ public override QueryResult ExecuteQuery()
+ {
+ Started.TrySetResult(true);
+ Fail.Wait(TimeSpan.FromSeconds(30));
+ throw new AdbcException("bad query",
AdbcStatusCode.InvalidArgument);
+ }
+
+ public override UpdateResult ExecuteUpdate() => throw new
NotSupportedException();
+ }
+
+ /// <summary>
+ /// A driver written before the CancellationToken overloads existed.
+ /// </summary>
+ private sealed class LegacyAsyncStatement : AdbcStatement
+ {
+ public const long QueryRows = 11;
+ public const long UpdateRows = 12;
+ public int SyncCount;
+
+ public override async ValueTask<QueryResult> ExecuteQueryAsync()
+ {
+ await Task.Yield();
+ return new QueryResult(QueryRows, null!);
+ }
+
+ public override async Task<UpdateResult> ExecuteUpdateAsync()
+ {
+ await Task.Yield();
+ return new UpdateResult(UpdateRows);
+ }
+
+ public override QueryResult ExecuteQuery() { SyncCount++; return
new QueryResult(0, null!); }
+ public override UpdateResult ExecuteUpdate() { SyncCount++; return
new UpdateResult(0); }
+ }
+
+ /// <summary>
+ /// A legacy override that does asynchronous work before delegating to
the base class.
+ /// </summary>
+ private sealed class LegacyCallsBaseStatement : AdbcStatement
+ {
+ public int OverrideCount;
+ public int SyncCount;
+
+ public override async ValueTask<QueryResult> ExecuteQueryAsync()
+ {
+ Interlocked.Increment(ref OverrideCount);
+ await Task.Delay(1);
+ return await base.ExecuteQueryAsync();
+ }
+
+ public override QueryResult ExecuteQuery()
+ {
+ Interlocked.Increment(ref SyncCount);
+ return new QueryResult(7, null!);
+ }
+
+ public override UpdateResult ExecuteUpdate() => new
UpdateResult(0);
+ }
+
+ /// <summary>
+ /// A driver written against the CancellationToken overloads.
+ /// </summary>
+ private sealed class NewAsyncStatement : AdbcStatement
+ {
+ public const long QueryRows = 21;
+ public const long UpdateRows = 22;
+ public int SyncCount;
+
+ public override async ValueTask<QueryResult>
ExecuteQueryAsync(CancellationToken cancellationToken)
+ {
+ await Task.Yield();
+ return new QueryResult(QueryRows, null!);
+ }
+
+ public override async Task<UpdateResult>
ExecuteUpdateAsync(CancellationToken cancellationToken)
+ {
+ await Task.Yield();
+ return new UpdateResult(UpdateRows);
+ }
+
+ public override QueryResult ExecuteQuery() { SyncCount++; return
new QueryResult(0, null!); }
+ public override UpdateResult ExecuteUpdate() { SyncCount++; return
new UpdateResult(0); }
+ }
+
+ /// <summary>
+ /// A driver following the recommended pattern: it implements the
token overloads
+ /// and forwards the parameterless ones to them.
+ /// </summary>
+ private sealed class RecommendedAsyncStatement : AdbcStatement
+ {
+ public int SyncCount;
+
+ public override ValueTask<QueryResult> ExecuteQueryAsync() =>
ExecuteQueryAsync(CancellationToken.None);
+
+ public override async ValueTask<QueryResult>
ExecuteQueryAsync(CancellationToken cancellationToken)
+ {
+ await Task.Yield();
+ return new QueryResult(NewAsyncStatement.QueryRows, null!);
+ }
+
+ public override Task<UpdateResult> ExecuteUpdateAsync() =>
ExecuteUpdateAsync(CancellationToken.None);
+
+ public override async Task<UpdateResult>
ExecuteUpdateAsync(CancellationToken cancellationToken)
+ {
+ await Task.Yield();
+ return new UpdateResult(NewAsyncStatement.UpdateRows);
+ }
+
+ public override QueryResult ExecuteQuery() { SyncCount++; return
new QueryResult(0, null!); }
+ public override UpdateResult ExecuteUpdate() { SyncCount++; return
new UpdateResult(0); }
+ }
+
+ /// <summary>
+ /// A token-overload override that falls back to the parameterless
base implementation.
+ /// </summary>
+ private sealed class NewCallsParameterlessBaseStatement : AdbcStatement
+ {
+ public int OverrideCount;
+ public int SyncCount;
+
+ public override async ValueTask<QueryResult>
ExecuteQueryAsync(CancellationToken cancellationToken)
+ {
+ Interlocked.Increment(ref OverrideCount);
+ return await base.ExecuteQueryAsync();
+ }
+
+ public override QueryResult ExecuteQuery()
+ {
+ Interlocked.Increment(ref SyncCount);
+ return new QueryResult(0, null!);
+ }
+
+ public override UpdateResult ExecuteUpdate() => new
UpdateResult(0);
+ }
+
+ /// <summary>
+ /// A legacy driver whose Cancel faults its pending task
synchronously, so any
+ /// continuation that isn't forced onto another thread runs inside
Cancel.
+ /// </summary>
+ private sealed class InlineFaultingLegacyStatement : AdbcStatement
+ {
+ private readonly TaskCompletionSource<QueryResult> _result = new
TaskCompletionSource<QueryResult>();
+
+ public override void Cancel() => _result.TrySetException(new
TimeoutException("interrupted"));
+
+ public override ValueTask<QueryResult> ExecuteQueryAsync() => new
ValueTask<QueryResult>(_result.Task);
+
+ public override QueryResult ExecuteQuery() => throw new
NotSupportedException();
+ public override UpdateResult ExecuteUpdate() => throw new
NotSupportedException();
}
- private sealed class RecordingConnection : AdbcConnection
+ /// <summary>
+ /// A legacy driver whose async overrides run the blocking operation
before
+ /// returning a task.
+ /// </summary>
+ private sealed class SynchronousLegacyStatement : AdbcStatement
{
+ private readonly ManualResetEventSlim _cancel = new
ManualResetEventSlim();
+ public TaskCompletionSource<bool> Started { get; } = new
TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ public bool CancelCalled => _cancel.IsSet;
+
+ public override void Cancel() => _cancel.Set();
+
+ public override ValueTask<QueryResult> ExecuteQueryAsync() => new
ValueTask<QueryResult>(ExecuteQuery());
+ public override Task<UpdateResult> ExecuteUpdateAsync() =>
Task.FromResult(ExecuteUpdate());
+
+ public override QueryResult ExecuteQuery()
+ {
+ Block();
+ return new QueryResult(0, null!);
+ }
+
+ public override UpdateResult ExecuteUpdate()
+ {
+ Block();
+ return new UpdateResult(0);
+ }
+
+ private void Block()
+ {
+ Started.TrySetResult(true);
+ _cancel.Wait(TimeSpan.FromSeconds(30));
+ throw new AdbcException("cancelled", AdbcStatusCode.Cancelled);
+ }
+ }
+
+ /// <summary>
+ /// A legacy override that calls the base implementation after the
token has been
+ /// cancelled, standing in for cancellation that arrives while the
work is queued.
+ /// </summary>
+ private sealed class CancelThenCallBaseStatement : AdbcStatement
+ {
+ private readonly CancellationTokenSource _cts;
+ public int QueryCount;
+ public int UpdateCount;
+
+ public CancelThenCallBaseStatement(CancellationTokenSource cts) {
_cts = cts; }
+
+ public override ValueTask<QueryResult> ExecuteQueryAsync()
+ {
+ _cts.Cancel();
+ return base.ExecuteQueryAsync();
+ }
+
+ public override Task<UpdateResult> ExecuteUpdateAsync()
+ {
+ _cts.Cancel();
+ return base.ExecuteUpdateAsync();
+ }
+
+ public override QueryResult ExecuteQuery()
+ {
+ Interlocked.Increment(ref QueryCount);
+ return new QueryResult(0, null!);
+ }
+
+ public override UpdateResult ExecuteUpdate()
+ {
+ Interlocked.Increment(ref UpdateCount);
+ return new UpdateResult(0);
+ }
+ }
+
+ private sealed class BlockingStatement : AdbcStatement
+ {
+ private readonly ManualResetEventSlim _cancel = new
ManualResetEventSlim();
+ public TaskCompletionSource<bool> Started { get; } = new
TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ public bool CancelCalled => _cancel.IsSet;
+
+ public override void Cancel() => _cancel.Set();
+
+ public override QueryResult ExecuteQuery()
+ {
+ Started.TrySetResult(true);
+ _cancel.Wait(TimeSpan.FromSeconds(30));
+ throw new AdbcException("cancelled", AdbcStatusCode.Cancelled);
+ }
+
+ public override UpdateResult ExecuteUpdate()
+ {
+ Started.TrySetResult(true);
+ _cancel.Wait(TimeSpan.FromSeconds(30));
+ throw new AdbcException("cancelled", AdbcStatusCode.Cancelled);
+ }
+ }
+
+ /// <summary>
+ /// A driver written before the token overloads existed, with a
genuinely
+ /// asynchronous query that completes when Cancel is called.
+ /// </summary>
+ private sealed class LegacyBlockingStatement : AdbcStatement
+ {
+ private readonly TaskCompletionSource<bool> _cancel = new
TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ public TaskCompletionSource<bool> Started { get; } = new
TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ public bool CancelCalled => _cancel.Task.IsCompleted;
+
+ public override void Cancel() => _cancel.TrySetResult(true);
+
+ public override async ValueTask<QueryResult> ExecuteQueryAsync()
+ {
+ Started.TrySetResult(true);
+ await _cancel.Task.ConfigureAwait(false);
+ throw new AdbcException("cancelled", AdbcStatusCode.Cancelled);
+ }
+
+ public override QueryResult ExecuteQuery() => throw new
NotSupportedException();
+ public override UpdateResult ExecuteUpdate() => throw new
NotSupportedException();
+ }
+
+ private sealed class BlockingConnection : AdbcConnection
+ {
+ private readonly ManualResetEventSlim _cancel = new
ManualResetEventSlim();
+ public TaskCompletionSource<bool> Started { get; } = new
TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ public bool CancelCalled => _cancel.IsSet;
public Dictionary<string, string> Options { get; } = new
Dictionary<string, string>();
+ public const string BlockingKey = "test.blocking";
public const string RejectedKey = "test.rejected";
public Dictionary<string, byte[]> BinaryOptions { get; } = new
Dictionary<string, byte[]>();
@@ -211,9 +856,29 @@ namespace Apache.Arrow.Adbc.Tests
public override void Dispose() => Disposed = true;
- public override void SetOption(string key, string value) =>
Options[key] = value;
+ public override void Cancel() => _cancel.Set();
- public override IArrowArrayStream GetTableTypes() => throw new
NotImplementedException();
+ public override void SetOption(string key, string value)
+ {
+ if (key == BlockingKey)
+ {
+ Block();
+ }
+ Options[key] = value;
+ }
+
+ public override IArrowArrayStream GetTableTypes()
+ {
+ Block();
+ return null!;
+ }
+
+ private void Block()
+ {
+ Started.TrySetResult(true);
+ _cancel.Wait(TimeSpan.FromSeconds(30));
+ throw new AdbcException("cancelled", AdbcStatusCode.Cancelled);
+ }
public override AdbcStatement CreateStatement() => throw new
NotImplementedException();
@@ -229,12 +894,12 @@ namespace Apache.Arrow.Adbc.Tests
private sealed class ConnectingDatabase : AdbcDatabase
{
public IReadOnlyDictionary<string, string>? LastStringOptions {
get; private set; }
- public RecordingConnection? LastConnection { get; private set; }
+ public BlockingConnection? LastConnection { get; private set; }
public override AdbcConnection Connect(IReadOnlyDictionary<string,
string>? options)
{
LastStringOptions = options;
- return LastConnection = new RecordingConnection();
+ return LastConnection = new BlockingConnection();
}
}
diff --git a/csharp/test/Apache.Arrow.Adbc.Tests/Client/ClientTests.cs
b/csharp/test/Apache.Arrow.Adbc.Tests/Client/ClientTests.cs
index 5a49fa764..7c96dd02e 100644
--- a/csharp/test/Apache.Arrow.Adbc.Tests/Client/ClientTests.cs
+++ b/csharp/test/Apache.Arrow.Adbc.Tests/Client/ClientTests.cs
@@ -235,7 +235,7 @@ namespace Apache.Arrow.Adbc.Tests.Client
[Fact]
public void ReadAsyncDoesNotDeadlockOnASynchronizationContext()
{
- const int timeoutMilliseconds = 5_000;
+ const int timeoutMilliseconds = 30_000; // generous for loaded CI
runners; a real deadlock never completes
const int expectedRows = 4;
using ManualResetEventSlim finished = new
ManualResetEventSlim(false);
diff --git
a/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs
b/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs
index 6d8877e2a..62ed7af2c 100644
--- a/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs
+++ b/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs
@@ -216,6 +216,22 @@ namespace Apache.Arrow.Adbc.Tests
Assert.Equal(42, stmt.ExecuteUpdate().AffectedRows);
}
+ [Fact]
+ public async Task ConnectionOptionsAsyncRoundTrip()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+
+ await conn.SetOptionAsync(AdbcOptions.Connection.CurrentCatalog,
"main");
+ Assert.Equal("main", fixture.LastConnection!.CurrentCatalog);
+ Assert.Equal("main", await
conn.GetOptionAsync(AdbcOptions.Connection.CurrentCatalog));
+
+ AdbcException e = await Assert.ThrowsAnyAsync<AdbcException>(() =>
conn.GetOptionAsync("unknown"));
+ Assert.Equal(AdbcStatusCode.NotFound, e.Status);
+ }
+
[Fact]
public void StatisticsAreMarshaled()
{
@@ -346,13 +362,33 @@ namespace Apache.Arrow.Adbc.Tests
stmt.SqlQuery = "BLOCK";
Task<QueryResult> pending = Task.Run(() => stmt.ExecuteQuery());
-
Assert.True(fixture.LastStatement!.Started.Wait(TimeSpan.FromSeconds(10)));
+ await WaitForStartAsync(fixture.LastStatement!.Started.Task);
stmt.Cancel();
AdbcException ex = await Assert.ThrowsAnyAsync<AdbcException>(()
=> pending);
Assert.Equal(AdbcStatusCode.Cancelled, ex.Status);
}
+ [Fact]
+ public async Task CancellationTokenCancelsImportedStatement()
+ {
+ var fixture = new Driver11();
+ using AdbcDriver imported = Load(fixture);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "BLOCK";
+
+ using CancellationTokenSource cts = new CancellationTokenSource();
+ ValueTask<QueryResult> pending = stmt.ExecuteQueryAsync(cts.Token);
+ await WaitForStartAsync(fixture.LastStatement!.Started.Task);
+ cts.Cancel();
+
+ OperationCanceledException ex = await
Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await pending);
+ AdbcException inner =
Assert.IsAssignableFrom<AdbcException>(ex.InnerException);
+ Assert.Equal(AdbcStatusCode.Cancelled, inner.Status);
+ }
+
[Fact]
public void Version1_0DriverFallsBackForTypedSetters()
{
@@ -415,6 +451,35 @@ namespace Apache.Arrow.Adbc.Tests
Assert.True(fixture.LastDatabase!.Disposed);
}
+ [Fact]
+ public async Task Version1_0DriverAsyncCallsMatchSyncFallbacks()
+ {
+ var fixture = new Driver11();
+ AdbcDriverInit adapter = (int version, ref CAdbcDriver
nativeDriver, ref CAdbcError error) =>
+ {
+ if (version != AdbcVersion.Version_1_0_0) { return
AdbcStatusCode.NotImplemented; }
+ return CallExporter(version, ref nativeDriver, ref error,
fixture);
+ };
+
+ using AdbcDriver imported = CAdbcDriverImporter.Load(adapter);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = await
db.ConnectWithTypedOptionsAsync(new Dictionary<string, object> { { "timeout",
5L } });
+ Assert.Equal("5", fixture.LastConnection!.Options["timeout"]);
+
+ await conn.SetOptionAsync("other", "x");
+ Assert.Equal("x", fixture.LastConnection.Options["other"]);
+
+ AdbcException ex = await Assert.ThrowsAnyAsync<AdbcException>(()
=> conn.GetOptionAsync("timeout"));
+ Assert.Equal(AdbcStatusCode.NotFound, ex.Status);
+
+ ex = await Assert.ThrowsAnyAsync<AdbcException>(() =>
conn.GetStatisticsNamesAsync());
+ Assert.Equal(AdbcStatusCode.NotImplemented, ex.Status);
+
+ using AdbcStatement stmt = conn.CreateStatement();
+ ex = await Assert.ThrowsAnyAsync<AdbcException>(() =>
stmt.ExecuteSchemaAsync());
+ Assert.Equal(AdbcStatusCode.NotImplemented, ex.Status);
+ }
+
private static AdbcDriver Load(AdbcDriver driver)
{
AdbcDriver imported = CAdbcDriverImporter.Load(
@@ -438,6 +503,12 @@ namespace Apache.Arrow.Adbc.Tests
/// <summary>
/// Stores options with their types so tests can see which typed
setter was used.
/// </summary>
+ // Waits without blocking a thread-pool thread, which the operation
under test needs
+ private static async Task WaitForStartAsync(Task started)
+ {
+ Assert.Same(started, await Task.WhenAny(started,
Task.Delay(TimeSpan.FromSeconds(30))));
+ }
+
private sealed class OptionBag
{
public Dictionary<string, object> Values { get; } = new
Dictionary<string, object>();
@@ -593,7 +664,7 @@ namespace Apache.Arrow.Adbc.Tests
private readonly OptionBag _options = new OptionBag();
private readonly ManualResetEventSlim _cancelled = new
ManualResetEventSlim();
- public ManualResetEventSlim Started { get; } = new
ManualResetEventSlim();
+ public TaskCompletionSource<bool> Started { get; } = new
TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
public override void SetOption(string key, string value) =>
_options.Values[key] = value;
public override void SetOptionBytes(string key, ReadOnlySpan<byte>
value) => _options.Values[key] = value.ToArray();
@@ -613,7 +684,7 @@ namespace Apache.Arrow.Adbc.Tests
case "FAIL_CANCELED":
throw new OperationCanceledException();
case "BLOCK":
- Started.Set();
+ Started.TrySetResult(true);
if (_cancelled.Wait(TimeSpan.FromSeconds(30)))
{
throw new AdbcException("cancelled",
AdbcStatusCode.Cancelled);
diff --git a/csharp/test/Apache.Arrow.Adbc.Tests/ImportedDuckDbTests.cs
b/csharp/test/Apache.Arrow.Adbc.Tests/ImportedDuckDbTests.cs
index ea8a9562f..2595b88b6 100644
--- a/csharp/test/Apache.Arrow.Adbc.Tests/ImportedDuckDbTests.cs
+++ b/csharp/test/Apache.Arrow.Adbc.Tests/ImportedDuckDbTests.cs
@@ -362,11 +362,7 @@ namespace Apache.Arrow.Adbc.Tests
statement.SqlQuery = "SELECT count(*) FROM range(1000000000000)";
Task<long> query = Task.Run(() =>
GetResultCount(statement.ExecuteQuery()));
- await Task.Delay(500);
- statement.Cancel();
-
- Task completed = await Task.WhenAny(query,
Task.Delay(TimeSpan.FromSeconds(30)));
- Assert.Same(query, completed);
+ await CancelUntilCompleteAsync(statement.Cancel, query);
AdbcException exception = await
Assert.ThrowsAnyAsync<AdbcException>(() => query);
// DuckDB reports interruption as InvalidArgument rather than
Cancelled
Assert.Contains("Interrupted", exception.Message);
@@ -382,16 +378,28 @@ namespace Apache.Arrow.Adbc.Tests
statement.SqlQuery = "SELECT count(*) FROM range(1000000000000)";
Task<long> query = Task.Run(() =>
GetResultCount(statement.ExecuteQuery()));
- await Task.Delay(500);
- connection.Cancel();
-
- Task completed = await Task.WhenAny(query,
Task.Delay(TimeSpan.FromSeconds(30)));
- Assert.Same(query, completed);
+ await CancelUntilCompleteAsync(connection.Cancel, query);
AdbcException exception = await
Assert.ThrowsAnyAsync<AdbcException>(() => query);
// DuckDB reports interruption as InvalidArgument rather than
Cancelled
Assert.Contains("Interrupted", exception.Message);
}
+ /// <summary>
+ /// Calls <paramref name="cancel"/> until <paramref name="query"/>
finishes. DuckDB clears
+ /// an interrupt when a query begins, so a single Cancel that arrives
before the query
+ /// has started (as it can on a busy CI runner) would leave the query
running indefinitely.
+ /// </summary>
+ private static async Task CancelUntilCompleteAsync(Action cancel, Task
query)
+ {
+ DateTime deadline = DateTime.UtcNow + TimeSpan.FromSeconds(30);
+ while (!query.IsCompleted && DateTime.UtcNow < deadline)
+ {
+ cancel();
+ await Task.WhenAny(query, Task.Delay(250));
+ }
+ Assert.True(query.IsCompleted, "The query was not cancelled");
+ }
+
private static long GetResultCount(AdbcStatement statement, string
query)
{
statement.SqlQuery = query;
