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 d9cb6a2115 feat(csharp): implement AdbcErrorFromArrayStream for
exported and imported streams (#4983)
d9cb6a2115 is described below
commit d9cb6a2115c57f9c708cf6c394e91254647bafc6
Author: Curt Hagenlocher <[email protected]>
AuthorDate: Sat Oct 31 06:51:35 2026 -0700
feat(csharp): implement AdbcErrorFromArrayStream for exported and imported
streams (#4983)
When a C# driver exported through `CAdbcDriverExporter` fails partway
through reading a result stream, the error's status, vendor code,
SQLSTATE and details are now preserved across the C ABI.
Exported streams no longer go through Arrow's
`CArrowArrayStreamExporter`. They keep the exception from a failed
`get_next`/`get_schema`, return an errno matching the status (e.g.
`ECANCELED` for `Cancelled`), and report the error through
`AdbcErrorFromArrayStream`. On the consumer side, `CAdbcDriverImporter`
calls the driver's `ErrorFromArrayStream` when a read fails and throws
an `AdbcException` with the status and details, falling back to the
previous message-only exception for 1.0 drivers or when the driver
returns null.
Because Apache.Arrow keeps the `ArrowArrayStream` callback fields
internal, this adds internal structs with the same layout.
This PR was prepared with AI assistance (Claude Code).
Closes #4982
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---
.../C/CAdbcDriverExporter.Streams.cs | 367 ++++++++++++++++++++
.../src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.cs | 31 +-
.../C/CAdbcDriverImporter.Streams.cs | 231 +++++++++++++
.../src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.cs | 4 +-
csharp/src/Apache.Arrow.Adbc/C/Delegates.cs | 5 +
.../src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs | 114 +++++++
.../FixtureDriver.cs | 26 ++
.../AotFixtureTests.cs | 18 +
.../ExportedDriverRoundTrip11Tests.cs | 371 +++++++++++++++++++++
9 files changed, 1140 insertions(+), 27 deletions(-)
diff --git a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.Streams.cs
b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.Streams.cs
new file mode 100644
index 0000000000..e7a5b83f6b
--- /dev/null
+++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.Streams.cs
@@ -0,0 +1,367 @@
+/*
+ * 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.InteropServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Apache.Arrow.C;
+using Apache.Arrow.Ipc;
+
+namespace Apache.Arrow.Adbc.C
+{
+ public partial class CAdbcDriverExporter
+ {
+#if NET5_0_OR_GREATER
+ private static unsafe delegate* unmanaged<CArrowArrayStream*,
CArrowSchema*, int> StreamGetSchemaPtr => &GetStreamSchema;
+ private static unsafe delegate* unmanaged<CArrowArrayStream*,
CArrowArray*, int> StreamGetNextPtr => &GetStreamNext;
+ private static unsafe delegate* unmanaged<CArrowArrayStream*, byte*>
StreamGetLastErrorPtr => &GetStreamLastError;
+ private static unsafe delegate* unmanaged<CArrowArrayStream*, void>
StreamReleasePtr => &ReleaseStream;
+ private static unsafe delegate* unmanaged<CArrowArrayStream*,
AdbcStatusCode*, CAdbcError*> ErrorFromArrayStreamPtr =>
&GetErrorFromArrayStream;
+#else
+ private static unsafe IntPtr StreamGetSchemaPtr =
ArrayStreamCallbacks.Export<ArrayStreamGetSchema>(GetStreamSchema,
ArrayStreamCallbacks.GetSchemaType);
+ private static unsafe IntPtr StreamGetNextPtr =
ArrayStreamCallbacks.Export<ArrayStreamGetNext>(GetStreamNext,
ArrayStreamCallbacks.GetNextType);
+ private static unsafe IntPtr StreamGetLastErrorPtr =
ArrayStreamCallbacks.Export<ArrayStreamGetLastError>(GetStreamLastError,
ArrayStreamCallbacks.GetLastErrorType);
+ private static unsafe IntPtr StreamReleasePtr =
ArrayStreamCallbacks.Export<ArrayStreamRelease>(ReleaseStream,
ArrayStreamCallbacks.ReleaseType);
+ private static unsafe IntPtr ErrorFromArrayStreamPtr =
NativeDelegate<ErrorFromArrayStream>.AsNativePointer(GetErrorFromArrayStream);
+#endif
+
+ /// <summary>
+ /// Exports a stream produced by the managed driver. The exported
stream takes
+ /// ownership on success; if exporting fails, nobody else will dispose
it.
+ /// </summary>
+ /// <remarks>
+ /// When reading the stream fails, the exported stream keeps the
exception so that
+ /// <c>AdbcErrorFromArrayStream</c> can report its status, SQLSTATE,
vendor code and
+ /// details, and returns an errno that corresponds to the status.
+ /// </remarks>
+ internal unsafe static void ExportStream(IArrowArrayStream? stream,
CArrowArrayStream* cStream)
+ {
+ try
+ {
+ if (stream == null)
+ {
+ throw new ArgumentNullException(nameof(stream));
+ }
+ if (cStream == null)
+ {
+ throw new ArgumentNullException(nameof(cStream));
+ }
+
+ NativeArrayStream* nativeStream =
NativeArrayStream.From(cStream);
+ nativeStream->private_data =
(void*)GCHandle.ToIntPtr(GCHandle.Alloc(new ExportedStream(stream)));
+ nativeStream->get_schema = StreamGetSchemaPtr;
+ nativeStream->get_next = StreamGetNextPtr;
+ nativeStream->get_last_error = StreamGetLastErrorPtr;
+ nativeStream->release = StreamReleasePtr;
+ }
+ catch
+ {
+ stream?.Dispose();
+ throw;
+ }
+ }
+
+ private unsafe static ExportedStream?
GetExportedStream(CArrowArrayStream* cStream)
+ {
+ NativeArrayStream* nativeStream = NativeArrayStream.From(cStream);
+ if (cStream == null ||
+ (IntPtr)nativeStream->release != (IntPtr)StreamReleasePtr ||
+ nativeStream->private_data == null)
+ {
+ return null;
+ }
+ return
GCHandle.FromIntPtr((IntPtr)nativeStream->private_data).Target as
ExportedStream;
+ }
+
+#if NET5_0_OR_GREATER
+ [UnmanagedCallersOnly]
+#endif
+ private unsafe static int GetStreamSchema(CArrowArrayStream* cStream,
CArrowSchema* cSchema)
+ {
+ ExportedStream? stream = null;
+ try
+ {
+ stream = GetExportedStream(cStream) ?? throw new
ArgumentException("not a stream exported by this driver", nameof(cStream));
+ stream.ClearError();
+ CArrowSchemaExporter.ExportSchema(stream.Stream.Schema,
cSchema);
+ return 0;
+ }
+ catch (Exception e)
+ {
+ return stream?.SetError(e) ?? Errno.EINVAL;
+ }
+ }
+
+#if NET5_0_OR_GREATER
+ [UnmanagedCallersOnly]
+#endif
+ private unsafe static int GetStreamNext(CArrowArrayStream* cStream,
CArrowArray* cArray)
+ {
+ ExportedStream? stream = null;
+ try
+ {
+ NativeArray.From(cArray)->release = default;
+ stream = GetExportedStream(cStream) ?? throw new
ArgumentException("not a stream exported by this driver", nameof(cStream));
+ stream.ClearError();
+
+ RecordBatch? batch = ReadNextBatch(stream.Stream);
+ if (batch != null)
+ {
+ CArrowArrayExporter.ExportRecordBatch(batch, cArray);
+ }
+ return 0;
+ }
+ catch (Exception e)
+ {
+ return stream?.SetError(e) ?? Errno.EINVAL;
+ }
+ }
+
+ private static RecordBatch? ReadNextBatch(IArrowArrayStream stream)
+ {
+ // A managed caller in the same process may call get_next on a
thread with a
+ // synchronization context. Blocking that thread deadlocks a read
that resumes on
+ // the context, so start the read without it. A native caller has
no context.
+ if (SynchronizationContext.Current != null ||
TaskScheduler.Current != TaskScheduler.Default)
+ {
+ return Task.Run(() =>
stream.ReadNextRecordBatchAsync().AsTask()).GetAwaiter().GetResult();
+ }
+
+ ValueTask<RecordBatch?> pending =
stream.ReadNextRecordBatchAsync();
+ return pending.IsCompletedSuccessfully ? pending.Result :
pending.AsTask().GetAwaiter().GetResult();
+ }
+
+#if NET5_0_OR_GREATER
+ [UnmanagedCallersOnly]
+#endif
+ private unsafe static byte* GetStreamLastError(CArrowArrayStream*
cStream)
+ {
+ try
+ {
+ return GetExportedStream(cStream)?.LastErrorMessage;
+ }
+ catch (Exception)
+ {
+ return null;
+ }
+ }
+
+#if NET5_0_OR_GREATER
+ [UnmanagedCallersOnly]
+#endif
+ private unsafe static void ReleaseStream(CArrowArrayStream* cStream)
+ {
+ NativeArrayStream* nativeStream = NativeArrayStream.From(cStream);
+ try
+ {
+ if (nativeStream->private_data != null)
+ {
+ GCHandle gch =
GCHandle.FromIntPtr((IntPtr)nativeStream->private_data);
+ try
+ {
+ ((ExportedStream)gch.Target!).Dispose();
+ }
+ finally
+ {
+ gch.Free();
+ nativeStream->private_data = null;
+ }
+ }
+ }
+ catch (Exception)
+ {
+ // An exception would end the process, and the consumer has no
way to learn of it.
+ }
+ nativeStream->release = default;
+ }
+
+#if NET5_0_OR_GREATER
+ [UnmanagedCallersOnly]
+#endif
+ private unsafe static CAdbcError*
GetErrorFromArrayStream(CArrowArrayStream* cStream, AdbcStatusCode* status)
+ {
+ try
+ {
+ return GetExportedStream(cStream)?.GetError(status);
+ }
+ catch (Exception)
+ {
+ return null;
+ }
+ }
+
+ /// <summary>
+ /// The state behind an exported <see cref="CArrowArrayStream"/>: the
managed stream
+ /// and the error from the last call that failed, if any.
+ /// </summary>
+ private sealed unsafe class ExportedStream : IDisposable
+ {
+ private IArrowArrayStream? _stream;
+ private CAdbcError* _error;
+ private AdbcStatusCode _status;
+
+ public ExportedStream(IArrowArrayStream stream)
+ {
+ _stream = stream;
+ }
+
+ public IArrowArrayStream Stream => _stream ?? throw new
ObjectDisposedException(nameof(ExportedStream));
+
+ public byte* LastErrorMessage => _error != null && _error->release
!= default ? _error->message : null;
+
+ public CAdbcError* GetError(AdbcStatusCode* status)
+ {
+ if (_error == null || _error->release == default)
+ {
+ return null;
+ }
+ if (status != null)
+ {
+ *status = _status;
+ }
+ return _error;
+ }
+
+ /// <summary>
+ /// Records <paramref name="exception"/> as the stream's error and
returns the
+ /// errno that corresponds to its status.
+ /// </summary>
+ public int SetError(Exception exception)
+ {
+ ClearError();
+ _status = CAdbcDriverExporter.SetError(null, exception);
+ if (_status == AdbcStatusCode.Success)
+ {
+ // An errno of 0 would read as the end of the stream
+ _status = AdbcStatusCode.UnknownError;
+ }
+
+ try
+ {
+ if (_error == null)
+ {
+ _error =
(CAdbcError*)Marshal.AllocHGlobal(sizeof(CAdbcError));
+ *_error = default;
+ }
+
+ // AdbcErrorFromArrayStream is part of ADBC 1.1.0, so the
consumer
+ // understands the 1.1.0 error layout.
+ _error->vendor_code = ErrorVendorCodePrivateData;
+ CAdbcDriverExporter.SetError(_error, exception);
+ }
+ catch (Exception)
+ {
+ // Report the errno without a message.
+ }
+ return Errno.FromStatus(_status);
+ }
+
+ public void ClearError()
+ {
+ if (_error != null)
+ {
+ if (_error->release != default)
+ {
+ ReleaseError(_error);
+ }
+ *_error = default;
+ }
+ }
+
+ public void Dispose()
+ {
+ try
+ {
+ _stream?.Dispose();
+ }
+ finally
+ {
+ _stream = null;
+ if (_error != null)
+ {
+ ClearError();
+ Marshal.FreeHGlobal((IntPtr)_error);
+ _error = null;
+ }
+ }
+ }
+ }
+
+ /// <summary>
+ /// The errno values that an <see cref="CArrowArrayStream"/> reports,
which depend on the
+ /// C runtime of the platform.
+ /// </summary>
+ internal static class Errno
+ {
+ public const int ENOENT = 2;
+ public const int EIO = 5;
+ public const int EACCES = 13;
+ public const int EEXIST = 17;
+ public const int EINVAL = 22;
+
+ public static readonly int ENOTSUP;
+ public static readonly int ECANCELED;
+ public static readonly int ETIMEDOUT;
+
+ static Errno()
+ {
+ if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
+ {
+ ENOTSUP = 129;
+ ECANCELED = 105;
+ ETIMEDOUT = 138;
+ }
+ else if (RuntimeInformation.IsOSPlatform(OSPlatform.OSX))
+ {
+ ENOTSUP = 45;
+ ECANCELED = 89;
+ ETIMEDOUT = 60;
+ }
+ else if
(RuntimeInformation.IsOSPlatform(OSPlatform.Create("FREEBSD")))
+ {
+ ENOTSUP = 45;
+ ECANCELED = 85;
+ ETIMEDOUT = 60;
+ }
+ else
+ {
+ // Linux
+ ENOTSUP = 95;
+ ECANCELED = 125;
+ ETIMEDOUT = 110;
+ }
+ }
+
+ /// <summary>
+ /// Maps a status to an errno the same way the C and Go drivers do.
+ /// </summary>
+ public static int FromStatus(AdbcStatusCode status) => status
switch
+ {
+ AdbcStatusCode.Success => 0,
+ AdbcStatusCode.NotImplemented => ENOTSUP,
+ AdbcStatusCode.NotFound => ENOENT,
+ AdbcStatusCode.AlreadyExists => EEXIST,
+ AdbcStatusCode.InvalidArgument or AdbcStatusCode.InvalidState
=> EINVAL,
+ AdbcStatusCode.Cancelled => ECANCELED,
+ AdbcStatusCode.Timeout => ETIMEDOUT,
+ AdbcStatusCode.Unauthenticated or AdbcStatusCode.Unauthorized
=> EACCES,
+ _ => EIO,
+ };
+ }
+ }
+}
diff --git a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.cs
b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.cs
index 90c12346a1..0db85d4719 100644
--- a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.cs
+++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.cs
@@ -30,13 +30,12 @@ namespace Apache.Arrow.Adbc.C
/// Exposes a managed <see cref="AdbcDriver"/> through the ADBC C API.
/// </summary>
/// <remarks>
- /// When a caller requests ADBC 1.1.0, every 1.1.0 entry point is populated
- /// except <c>ErrorFromArrayStream</c>; exported streams report errors
through
- /// <c>get_last_error</c> instead. Functionality that the managed driver
doesn't
- /// override surfaces as <see cref="AdbcStatusCode.NotImplemented"/>,
which is
- /// the same status a driver manager reports for a missing entry point.
+ /// When a caller requests ADBC 1.1.0, every 1.1.0 entry point is
populated.
+ /// Functionality that the managed driver doesn't override surfaces as
+ /// <see cref="AdbcStatusCode.NotImplemented"/>, which is the same status
a driver
+ /// manager reports for a missing entry point.
/// </remarks>
- public class CAdbcDriverExporter
+ public partial class CAdbcDriverExporter
{
/// <summary>
/// The value of <c>ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA</c>. A caller
sets an error's
@@ -279,8 +278,7 @@ namespace Apache.Arrow.Adbc.C
{
nativeDriver->ErrorGetDetailCount = ErrorGetDetailCountPtr;
nativeDriver->ErrorGetDetail = ErrorGetDetailPtr;
- // ErrorFromArrayStream is left null: exported streams report
errors
- // through ArrowArrayStream.get_last_error only.
+ nativeDriver->ErrorFromArrayStream = ErrorFromArrayStreamPtr;
nativeDriver->DatabaseGetOption = DatabaseGetOptionPtr;
nativeDriver->DatabaseGetOptionBytes =
DatabaseGetOptionBytesPtr;
@@ -430,23 +428,6 @@ namespace Apache.Arrow.Adbc.C
}
}
- /// <summary>
- /// Exports a stream produced by the managed driver. The exported
stream takes
- /// ownership on success; if exporting fails, nobody else will dispose
it.
- /// </summary>
- private unsafe static void ExportStream(IArrowArrayStream? stream,
CArrowArrayStream* cStream)
- {
- try
- {
- CArrowArrayStreamExporter.ExportArrayStream(stream!, cStream);
- }
- catch
- {
- stream?.Dispose();
- throw;
- }
- }
-
private static IntPtr FromDisposable(IDisposable d)
{
GCHandle gch = GCHandle.Alloc(d);
diff --git a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs
b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs
new file mode 100644
index 0000000000..19dbec6192
--- /dev/null
+++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs
@@ -0,0 +1,231 @@
+/*
+ * 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.Collections.Generic;
+using System.Runtime.InteropServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Apache.Arrow.Adbc.Extensions;
+using Apache.Arrow.C;
+using Apache.Arrow.Ipc;
+
+namespace Apache.Arrow.Adbc.C
+{
+ public static partial class CAdbcDriverImporter
+ {
+ /// <summary>
+ /// A result stream imported from a driver. When the stream fails, it
asks the driver
+ /// for the ADBC error behind the failure (see
<c>AdbcErrorFromArrayStream</c>) and
+ /// throws it as an <see cref="AdbcException"/>.
+ /// </summary>
+ private sealed unsafe class ImportedArrayStream : IArrowArrayStream
+ {
+ private readonly ImportedAdbcDriver? _driver;
+ private readonly Schema _schema;
+ private CArrowArrayStream* _cStream;
+
+ /// <summary>
+ /// Imports <paramref name="cStream"/>, which must have been
allocated with
+ /// <see cref="CArrowArrayStream.Create"/>. On success, the
imported stream owns
+ /// it and holds a reference to <paramref name="driver"/>, whose
callbacks it calls
+ /// until it's disposed; on failure, the caller still owns the
stream.
+ /// </summary>
+ public ImportedArrayStream(ImportedAdbcDriver? driver,
CArrowArrayStream* cStream)
+ {
+ if (NativeArrayStream.From(cStream)->release == default)
+ {
+ throw new ArgumentException("Tried to import an array
stream that has already been released.", nameof(cStream));
+ }
+
+ _driver = driver;
+
+ CArrowSchema* cSchema = CArrowSchema.Create();
+ try
+ {
+ int errno = GetSchema(cStream, cSchema);
+ if (errno != 0)
+ {
+ throw StreamError(cStream, errno);
+ }
+ _schema = CArrowSchemaImporter.ImportSchema(cSchema);
+ }
+ finally
+ {
+ CArrowSchema.Free(cSchema);
+ }
+
+ _driver?.AddReference();
+ _cStream = cStream;
+ }
+
+ ~ImportedArrayStream()
+ {
+ Dispose();
+ }
+
+ public Schema Schema => _schema;
+
+ public ValueTask<RecordBatch?>
ReadNextRecordBatchAsync(CancellationToken cancellationToken = default)
+ {
+ if (_cStream == null)
+ {
+ throw new
ObjectDisposedException(nameof(ImportedArrayStream));
+ }
+
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return new
ValueTask<RecordBatch?>(Task.FromCanceled<RecordBatch?>(cancellationToken));
+ }
+
+ try
+ {
+ return new ValueTask<RecordBatch?>(ReadNext());
+ }
+ catch (Exception e)
+ {
+ return new
ValueTask<RecordBatch?>(Task.FromException<RecordBatch?>(e));
+ }
+ }
+
+ private RecordBatch? ReadNext()
+ {
+ CArrowArray* cArray = CArrowArray.Create();
+ try
+ {
+ int errno = GetNext(_cStream, cArray);
+ if (errno != 0)
+ {
+ throw StreamError(_cStream, errno);
+ }
+
+ // A released array marks the end of the stream
+ if (NativeArray.From(cArray)->release == IntPtr.Zero)
+ {
+ return null;
+ }
+
+ return CArrowArrayImporter.ImportRecordBatch(cArray,
_schema);
+ }
+ finally
+ {
+ CArrowArray.Free(cArray);
+ }
+ }
+
+ public void Dispose()
+ {
+ CArrowArrayStream* cStream = _cStream;
+ if (cStream != null)
+ {
+ _cStream = null;
+ try
+ {
+ FreeStream(cStream);
+ }
+ finally
+ {
+ _driver?.RemoveReference();
+ }
+ }
+ GC.SuppressFinalize(this);
+ }
+
+ /// <summary>
+ /// Releases and frees a stream allocated with <see
cref="CArrowArrayStream.Create"/>.
+ /// </summary>
+ /// <remarks>
+ /// On .NET Framework, this releases the stream through
<c>ArrayStreamCallbacks</c>, so
+ /// it can release a stream exported in the same process with any
delegate type.
+ /// </remarks>
+ public static void FreeStream(CArrowArrayStream* cStream)
+ {
+ NativeArrayStream* nativeStream =
NativeArrayStream.From(cStream);
+ if (nativeStream->release != default)
+ {
+#if NET5_0_OR_GREATER
+ nativeStream->release(cStream);
+#else
+
ArrayStreamCallbacks.Import<ArrayStreamRelease>(nativeStream->release)(cStream);
+#endif
+ nativeStream->release = default;
+ }
+ CArrowArrayStream.Free(cStream);
+ }
+
+ /// <summary>
+ /// Builds the exception for a stream call that returned <paramref
name="errno"/>:
+ /// the driver's ADBC error if it reports one, and otherwise the
stream's last error.
+ /// </summary>
+ private Exception StreamError(CArrowArrayStream* cStream, int
errno)
+ {
+ if (_driver != null && _driver.DriverVersion >=
AdbcVersion.Version_1_1_0)
+ {
+ ref CAdbcDriver nativeDriver = ref _driver.DriverUnsafe;
+ if (nativeDriver.ErrorFromArrayStream != default)
+ {
+ AdbcStatusCode status = AdbcStatusCode.UnknownError;
+#if NET5_0_OR_GREATER
+ CAdbcError* error =
nativeDriver.ErrorFromArrayStream(cStream, &status);
+#else
+ CAdbcError* error =
Marshal.GetDelegateForFunctionPointer<ErrorFromArrayStream>(nativeDriver.ErrorFromArrayStream)(cStream,
&status);
+#endif
+ // The stream owns the error, so it must not be
released here
+ if (error != null)
+ {
+ IReadOnlyList<KeyValuePair<string, byte[]>>?
details = ReadErrorDetails(_driver, error);
+ return new ImportedAdbcException(
+ ref *error,
+ status == AdbcStatusCode.Success ?
AdbcStatusCode.UnknownError : status,
+ details);
+ }
+ }
+ }
+
+ NativeArrayStream* nativeStream =
NativeArrayStream.From(cStream);
+#if NET5_0_OR_GREATER
+ byte* message = nativeStream->get_last_error(cStream);
+#else
+ byte* message =
ArrayStreamCallbacks.Import<ArrayStreamGetLastError>(nativeStream->get_last_error)(cStream);
+#endif
+ return new Exception(message == null
+ ? $"Array stream operation failed with no message. Error
code: {errno}"
+ : MarshalExtensions.PtrToStringUTF8(message));
+ }
+
+ private static int GetSchema(CArrowArrayStream* cStream,
CArrowSchema* cSchema)
+ {
+ NativeArrayStream* nativeStream =
NativeArrayStream.From(cStream);
+#if NET5_0_OR_GREATER
+ return nativeStream->get_schema(cStream, cSchema);
+#else
+ return
ArrayStreamCallbacks.Import<ArrayStreamGetSchema>(nativeStream->get_schema)(cStream,
cSchema);
+#endif
+ }
+
+ private static int GetNext(CArrowArrayStream* cStream,
CArrowArray* cArray)
+ {
+ NativeArrayStream* nativeStream =
NativeArrayStream.From(cStream);
+#if NET5_0_OR_GREATER
+ return nativeStream->get_next(cStream, cArray);
+#else
+ return
ArrayStreamCallbacks.Import<ArrayStreamGetNext>(nativeStream->get_next)(cStream,
cArray);
+#endif
+ }
+ }
+ }
+}
diff --git a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.cs
b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.cs
index 6d4fc59e7a..1d28f96993 100644
--- a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.cs
+++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.cs
@@ -1776,7 +1776,7 @@ namespace Apache.Arrow.Adbc.C
public IArrowArrayStream ImportStream()
{
Debug.Assert(_arrayStream != null);
- IArrowArrayStream arrayStream =
CArrowArrayStreamImporter.ImportArrayStream(_arrayStream);
+ IArrowArrayStream arrayStream = new
ImportedArrayStream(_driver, _arrayStream);
_arrayStream = null;
return arrayStream;
}
@@ -1992,7 +1992,7 @@ namespace Apache.Arrow.Adbc.C
}
if (_arrayStream != null)
{
- CArrowArrayStream.Free(_arrayStream);
+ ImportedArrayStream.FreeStream(_arrayStream);
_arrayStream = null;
}
}
diff --git a/csharp/src/Apache.Arrow.Adbc/C/Delegates.cs
b/csharp/src/Apache.Arrow.Adbc/C/Delegates.cs
index e2838129b9..2873f7b8c5 100644
--- a/csharp/src/Apache.Arrow.Adbc/C/Delegates.cs
+++ b/csharp/src/Apache.Arrow.Adbc/C/Delegates.cs
@@ -31,6 +31,11 @@ namespace Apache.Arrow.Adbc.C
internal unsafe delegate CAdbcErrorDetail ErrorGetDetail(CAdbcError*
error, int index);
internal unsafe delegate CAdbcError*
ErrorFromArrayStream(CArrowArrayStream* stream, AdbcStatusCode* status);
+ internal unsafe delegate int ArrayStreamGetSchema(CArrowArrayStream*
stream, CArrowSchema* schema);
+ internal unsafe delegate int ArrayStreamGetNext(CArrowArrayStream* stream,
CArrowArray* array);
+ internal unsafe delegate byte* ArrayStreamGetLastError(CArrowArrayStream*
stream);
+ internal unsafe delegate void ArrayStreamRelease(CArrowArrayStream*
stream);
+
internal unsafe delegate AdbcStatusCode DatabaseGetOption(CAdbcDatabase*
database, byte* name, byte* value, nint* length, CAdbcError* error);
internal unsafe delegate AdbcStatusCode
DatabaseGetOptionBytes(CAdbcDatabase* database, byte* name, byte* value, nint*
length, CAdbcError* error);
internal unsafe delegate AdbcStatusCode
DatabaseGetOptionDouble(CAdbcDatabase* database, byte* name, double* value,
CAdbcError* error);
diff --git a/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs
b/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs
new file mode 100644
index 0000000000..02d800a6ac
--- /dev/null
+++ b/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs
@@ -0,0 +1,114 @@
+/*
+ * 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.Reflection;
+using System.Runtime.InteropServices;
+using Apache.Arrow.C;
+
+namespace Apache.Arrow.Adbc.C
+{
+ /// <summary>
+ /// The layout of <see cref="CArrowArrayStream"/>, whose callbacks
Apache.Arrow doesn't
+ /// expose. ADBC exports and imports result streams itself so that it can
attach an
+ /// ADBC error to a stream that fails (see
<c>AdbcErrorFromArrayStream</c>).
+ /// </summary>
+ [StructLayout(LayoutKind.Sequential)]
+ internal unsafe struct NativeArrayStream
+ {
+#if NET5_0_OR_GREATER
+ public delegate* unmanaged<CArrowArrayStream*, CArrowSchema*, int>
get_schema;
+ public delegate* unmanaged<CArrowArrayStream*, CArrowArray*, int>
get_next;
+ public delegate* unmanaged<CArrowArrayStream*, byte*> get_last_error;
+ public delegate* unmanaged<CArrowArrayStream*, void> release;
+#else
+ public IntPtr get_schema;
+ public IntPtr get_next;
+ public IntPtr get_last_error;
+ public IntPtr release;
+#endif
+ public void* private_data;
+
+ public static NativeArrayStream* From(CArrowArrayStream* stream) =>
(NativeArrayStream*)stream;
+ }
+
+ /// <summary>
+ /// The layout of <see cref="CArrowArray"/>, whose release callback
Apache.Arrow doesn't expose.
+ /// </summary>
+ [StructLayout(LayoutKind.Sequential)]
+ internal unsafe struct NativeArray
+ {
+ public long length;
+ public long null_count;
+ public long offset;
+ public long n_buffers;
+ public long n_children;
+ public byte** buffers;
+ public CArrowArray** children;
+ public CArrowArray* dictionary;
+ public IntPtr release;
+ public void* private_data;
+
+ public static NativeArray* From(CArrowArray* array) =>
(NativeArray*)array;
+ }
+
+#if !NET5_0_OR_GREATER
+ /// <summary>
+ /// Creates and calls <see cref="CArrowArrayStream"/> callbacks on .NET
Framework.
+ /// </summary>
+ /// <remarks>
+ /// On .NET Framework, <see
cref="Marshal.GetDelegateForFunctionPointer{TDelegate}(IntPtr)"/>
+ /// returns the original delegate for a function pointer created from a
managed delegate, and
+ /// fails if that delegate has a different type. ADBC calls callbacks
through
+ /// <see cref="Import{T}"/>, which accepts a delegate of any type.
Apache.Arrow calls them
+ /// through its own internal delegate types, so ADBC exports its callbacks
with those types,
+ /// found by reflection, to stay compatible with consumers that import
streams through
+ /// Apache.Arrow. If they can't be found, ADBC exports its own types. Once
Apache.Arrow
+ /// accepts callbacks of any delegate type (apache/arrow-dotnet#491), the
reflection can go.
+ /// </remarks>
+ internal static class ArrayStreamCallbacks
+ {
+ public static readonly Type? GetSchemaType =
ArrowDelegateType("GetSchemaArrayStream");
+ public static readonly Type? GetNextType =
ArrowDelegateType("GetNextArrayStream");
+ public static readonly Type? GetLastErrorType =
ArrowDelegateType("GetLastErrorArrayStream");
+ public static readonly Type? ReleaseType =
ArrowDelegateType("ReleaseArrayStream");
+
+ /// <summary>
+ /// Returns a function pointer for the static method behind <paramref
name="callback"/>.
+ /// </summary>
+ public static IntPtr Export<T>(T callback, Type? arrowType) where T :
Delegate
+ {
+ Delegate exported = arrowType == null ? callback :
Delegate.CreateDelegate(arrowType, callback.Method);
+ return NativeDelegate<Delegate>.AsNativePointer(exported);
+ }
+
+ /// <summary>
+ /// Returns a delegate that calls <paramref name="pointer"/>,
whichever delegate type
+ /// it was created from.
+ /// </summary>
+ public static T Import<T>(IntPtr pointer) where T : Delegate
+ {
+ // Unlike the generic overload, this doesn't cast the delegate it
returns to T
+ Delegate imported = Marshal.GetDelegateForFunctionPointer(pointer,
typeof(T));
+ return imported as T ?? (T)Delegate.CreateDelegate(typeof(T),
imported, "Invoke");
+ }
+
+ private static Type? ArrowDelegateType(string name) =>
+ typeof(CArrowArrayStreamExporter).GetNestedType(name,
BindingFlags.NonPublic);
+ }
+#endif
+}
diff --git
a/csharp/test/AotInterop/Apache.Arrow.Adbc.TestFixture.Native/FixtureDriver.cs
b/csharp/test/AotInterop/Apache.Arrow.Adbc.TestFixture.Native/FixtureDriver.cs
index dcfba33643..362dfb1734 100644
---
a/csharp/test/AotInterop/Apache.Arrow.Adbc.TestFixture.Native/FixtureDriver.cs
+++
b/csharp/test/AotInterop/Apache.Arrow.Adbc.TestFixture.Native/FixtureDriver.cs
@@ -157,6 +157,10 @@ namespace Apache.Arrow.Adbc.TestFixture.Native
{
throw new DetailedException();
}
+ if (_sqlQuery == "FAIL_STREAM")
+ {
+ return new QueryResult(-1, new FailingStream(ResultSchema(),
new DetailedException()));
+ }
_executed = true;
@@ -254,4 +258,26 @@ namespace Apache.Arrow.Adbc.TestFixture.Native
public void Dispose() { _batch?.Dispose(); _batch = null; }
}
+
+ /// <summary>
+ /// IArrowArrayStream that fails on its first read, after the result was
handed out.
+ /// </summary>
+ internal sealed class FailingStream : IArrowArrayStream
+ {
+ private readonly Schema _schema;
+ private readonly Exception _exception;
+
+ public FailingStream(Schema schema, Exception exception)
+ {
+ _schema = schema;
+ _exception = exception;
+ }
+
+ public Schema Schema => _schema;
+
+ public ValueTask<RecordBatch?>
ReadNextRecordBatchAsync(CancellationToken cancellationToken = default)
+ => throw _exception;
+
+ public void Dispose() { }
+ }
}
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 5afc90d17c..5f07ccfd6d 100644
---
a/csharp/test/AotInterop/Apache.Arrow.Adbc.TestFixture.Tests/AotFixtureTests.cs
+++
b/csharp/test/AotInterop/Apache.Arrow.Adbc.TestFixture.Tests/AotFixtureTests.cs
@@ -276,6 +276,24 @@ namespace Apache.Arrow.Adbc.TestFixture.Tests
Assert.Equal(new byte[] { 1, 2, 3 }, detail.Value);
}
+ [SkippableFact]
+ public async Task StreamErrorDetailsCrossAotBoundary()
+ {
+ using AdbcDriver driver = LoadFixture();
+ using AdbcDatabase db = driver.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "FAIL_STREAM";
+
+ using IArrowArrayStream stream = stmt.ExecuteQuery().Stream!;
+ AdbcException ex = await
Assert.ThrowsAnyAsync<AdbcException>(async () => await
stream.ReadNextRecordBatchAsync());
+ Assert.Equal(AdbcStatusCode.InvalidArgument, ex.Status);
+ Assert.Equal("42000", ex.SqlState);
+ KeyValuePair<string, byte[]> detail = Assert.Single(ex.Details);
+ Assert.Equal("fixture.detail", detail.Key);
+ Assert.Equal(new byte[] { 1, 2, 3 }, detail.Value);
+ }
+
[SkippableFact]
public async Task ExecuteQueryAsyncWithTokenRoundTrips()
{
diff --git
a/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs
b/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs
index 62ed7af2c1..62e84997c4 100644
--- a/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs
+++ b/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs
@@ -17,10 +17,12 @@
using System;
using System.Collections.Generic;
+using System.Runtime.InteropServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Apache.Arrow.Adbc.C;
+using Apache.Arrow.C;
using Apache.Arrow.Ipc;
using Apache.Arrow.Types;
using Xunit;
@@ -325,6 +327,288 @@ namespace Apache.Arrow.Adbc.Tests
Assert.Empty(ex.Details);
}
+ [Fact]
+ public async Task StreamErrorKeepsStatusAndDetails()
+ {
+ using AdbcDriver imported = Load(new Driver11());
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "STREAM_FAIL_DETAILED";
+
+ QueryResult result = stmt.ExecuteQuery();
+ using IArrowArrayStream stream = result.Stream!;
+ using (RecordBatch? batch = await
stream.ReadNextRecordBatchAsync())
+ {
+ Assert.Equal(2, batch!.Length);
+ }
+
+ AdbcException ex = await
Assert.ThrowsAnyAsync<AdbcException>(async () => await
stream.ReadNextRecordBatchAsync());
+ Assert.Equal(AdbcStatusCode.InvalidData, ex.Status);
+ Assert.Equal("42000", ex.SqlState);
+ Assert.Contains("query failed", ex.Message);
+ Assert.Equal(2, ex.Details.Count);
+ Assert.Equal("query_id", ex.Details[0].Key);
+ Assert.Equal("q-123",
Encoding.UTF8.GetString(ex.Details[0].Value));
+ }
+
+ [Theory]
+ [InlineData("STREAM_FAIL_CANCELLED", "read cancelled")]
+ [InlineData("STREAM_FAIL_CANCELED", "token cancelled")]
+ public async Task StreamCancellationIsReportedAsCancelled(string
query, string message)
+ {
+ using AdbcDriver imported = Load(new Driver11());
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = query;
+
+ using IArrowArrayStream stream = stmt.ExecuteQuery().Stream!;
+ (await stream.ReadNextRecordBatchAsync())?.Dispose();
+
+ AdbcException ex = await
Assert.ThrowsAnyAsync<AdbcException>(async () => await
stream.ReadNextRecordBatchAsync());
+ Assert.Equal(AdbcStatusCode.Cancelled, ex.Status);
+ Assert.Equal(message, ex.Message);
+ }
+
+ [Fact]
+ public async Task StreamKeepsDriverLoaded()
+ {
+ var fixture = new Driver11();
+ IArrowArrayStream stream;
+ 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 = "STREAM_FAIL_CANCELLED";
+ stream = stmt.ExecuteQuery().Stream!;
+ }
+
+ using (stream)
+ {
+ Assert.False(fixture.Disposed);
+ (await stream.ReadNextRecordBatchAsync())?.Dispose();
+
+ AdbcException ex = await
Assert.ThrowsAnyAsync<AdbcException>(async () => await
stream.ReadNextRecordBatchAsync());
+ Assert.Equal(AdbcStatusCode.Cancelled, ex.Status);
+ Assert.False(fixture.Disposed);
+ }
+
+ Assert.True(fixture.Disposed);
+ }
+
+ [Fact]
+ public void StreamSchemaErrorKeepsStatus()
+ {
+ using AdbcDriver imported = Load(new Driver11());
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "STREAM_FAIL_SCHEMA";
+
+ AdbcException ex = Assert.ThrowsAny<AdbcException>(() =>
stmt.ExecuteQuery());
+ Assert.Equal(AdbcStatusCode.Unauthorized, ex.Status);
+ Assert.Equal("no access", ex.Message);
+ }
+
+ [Fact]
+ public async Task Version1_0DriverStreamErrorHasMessageOnly()
+ {
+ 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 = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "STREAM_FAIL_CANCELLED";
+
+ using IArrowArrayStream stream = stmt.ExecuteQuery().Stream!;
+ (await stream.ReadNextRecordBatchAsync())?.Dispose();
+
+ // Without AdbcErrorFromArrayStream, only the message crosses the
C ABI
+ Exception ex = await Assert.ThrowsAnyAsync<Exception>(async () =>
await stream.ReadNextRecordBatchAsync());
+ Assert.False(ex is AdbcException);
+ Assert.Equal("read cancelled", ex.Message);
+ }
+
+#if NET5_0_OR_GREATER
+ [Fact]
+ public unsafe void ExportedStreamReportsErrnoAndError()
+ {
+ CAdbcDriver driver = default;
+ CAdbcError error = default;
+ CAdbcDatabase database = default;
+ CAdbcConnection connection = default;
+ CAdbcStatement statement = default;
+ Assert.Equal(AdbcStatusCode.Success,
CAdbcDriverExporter.AdbcDriverInit(AdbcVersion.Version_1_1_0, &driver, &error,
new Driver11()));
+ Assert.Equal(AdbcStatusCode.Success, driver.DatabaseNew(&database,
&error));
+ Assert.Equal(AdbcStatusCode.Success,
driver.DatabaseInit(&database, &error));
+ Assert.Equal(AdbcStatusCode.Success,
driver.ConnectionNew(&connection, &error));
+ Assert.Equal(AdbcStatusCode.Success,
driver.ConnectionInit(&connection, &database, &error));
+ Assert.Equal(AdbcStatusCode.Success,
driver.StatementNew(&connection, &statement, &error));
+ fixed (byte* query =
Encoding.UTF8.GetBytes("STREAM_FAIL_CANCELLED\0"))
+ {
+ Assert.Equal(AdbcStatusCode.Success,
driver.StatementSetSqlQuery(&statement, query, &error));
+ }
+
+ CArrowArrayStream* stream = CArrowArrayStream.Create();
+ CArrowArray* array = CArrowArray.Create();
+ try
+ {
+ long rows = 0;
+ Assert.Equal(AdbcStatusCode.Success,
driver.StatementExecuteQuery(&statement, stream, &rows, &error));
+ NativeArrayStream* nativeStream =
NativeArrayStream.From(stream);
+
+ Assert.Equal(0, nativeStream->get_next(stream, array));
+ CArrowArray.CallReleaseFunc(array);
+
+ AdbcStatusCode status = AdbcStatusCode.Success;
+ Assert.True(driver.ErrorFromArrayStream(stream, &status) ==
null);
+
+ Assert.Equal(CAdbcDriverExporter.Errno.ECANCELED,
nativeStream->get_next(stream, array));
+ Assert.Equal("read cancelled",
Marshal.PtrToStringUTF8((IntPtr)nativeStream->get_last_error(stream)));
+
+ CAdbcError* streamError = driver.ErrorFromArrayStream(stream,
&status);
+ Assert.True(streamError != null);
+ Assert.Equal(AdbcStatusCode.Cancelled, status);
+ Assert.Equal("read cancelled",
Marshal.PtrToStringUTF8((IntPtr)streamError->message));
+ }
+ finally
+ {
+ CArrowArray.Free(array);
+ CArrowArrayStream.Free(stream);
+ driver.StatementRelease(&statement, &error);
+ driver.ConnectionRelease(&connection, &error);
+ driver.DatabaseRelease(&database, &error);
+ driver.release(&driver, &error);
+ }
+ }
+
+#endif
+
+ [Fact]
+ public async Task ArrowImporterReadsExportedStream()
+ {
+ IArrowArrayStream imported;
+ unsafe
+ {
+ CArrowArrayStream* cStream = CArrowArrayStream.Create();
+ try
+ {
+ CAdbcDriverExporter.ExportStream(new FailingStream(new
AdbcException("read cancelled", AdbcStatusCode.Cancelled)), cStream);
+ imported =
CArrowArrayStreamImporter.ImportArrayStream(cStream);
+ }
+ finally
+ {
+ CArrowArrayStream.Free(cStream);
+ }
+ }
+
+ using (imported)
+ {
+ Assert.Equal("x",
Assert.Single(imported.Schema.FieldsList).Name);
+ using (RecordBatch? batch = await
imported.ReadNextRecordBatchAsync())
+ {
+ Assert.Equal(2, batch!.Length);
+ }
+ Exception ex = await Assert.ThrowsAnyAsync<Exception>(async ()
=> await imported.ReadNextRecordBatchAsync());
+ Assert.Equal("read cancelled", ex.Message);
+ }
+ }
+
+ [Fact]
+ public void ExportedStreamDoesNotBlockSynchronizationContext()
+ {
+ int rows = -1;
+ Exception? failure = null;
+ Thread thread = new Thread(() =>
+ {
+ SynchronizationContext.SetSynchronizationContext(new
BlockedContext());
+ try
+ {
+ unsafe
+ {
+ CArrowArrayStream* cStream =
CArrowArrayStream.Create();
+ try
+ {
+ CAdbcDriverExporter.ExportStream(new
ContextResumingStream(), cStream);
+ using IArrowArrayStream imported =
CArrowArrayStreamImporter.ImportArrayStream(cStream);
+ using RecordBatch? batch =
imported.ReadNextRecordBatchAsync().AsTask().GetAwaiter().GetResult();
+ rows = batch!.Length;
+ }
+ finally
+ {
+ CArrowArrayStream.Free(cStream);
+ }
+ }
+ }
+ catch (Exception e)
+ {
+ failure = e;
+ }
+ });
+ thread.IsBackground = true;
+ thread.Start();
+
+ Assert.True(thread.Join(TimeSpan.FromSeconds(30)), "reading the
exported stream deadlocked");
+ Assert.Null(failure);
+ Assert.Equal(2, rows);
+ }
+
+ [Fact]
+ public void RejectedBindStreamIsReleased()
+ {
+ AdbcDriverInit adapter = (int version, ref CAdbcDriver
nativeDriver, ref CAdbcError error) =>
+ {
+ AdbcStatusCode status = CallExporter(version, ref
nativeDriver, ref error, new Driver11());
+ unsafe
+ {
+#if NET5_0_OR_GREATER
+ nativeDriver.StatementBindStream =
+ (delegate* unmanaged<CAdbcStatement*,
CArrowArrayStream*, CAdbcError*,
AdbcStatusCode>)Marshal.GetFunctionPointerForDelegate(s_rejectBindStream);
+#else
+ nativeDriver.StatementBindStream =
Marshal.GetFunctionPointerForDelegate(s_rejectBindStream);
+#endif
+ }
+ return status;
+ };
+
+ using AdbcDriver imported = CAdbcDriverImporter.Load(adapter);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+
+ // The driver fails without taking the stream, so the importer
releases it
+ var stream = new FailingStream(new InvalidOperationException());
+ AdbcException ex = Assert.ThrowsAny<AdbcException>(() =>
stmt.BindStream(stream));
+ Assert.Equal(AdbcStatusCode.InvalidState, ex.Status);
+ Assert.True(stream.Disposed);
+ }
+
+#if NET5_0_OR_GREATER
+ private unsafe delegate AdbcStatusCode
StatementBindStream(CAdbcStatement* statement, CArrowArrayStream* stream,
CAdbcError* error);
+#endif
+
+ // On .NET Framework, the importer can only call a managed callback
through the library's own delegate type
+ private static readonly unsafe StatementBindStream s_rejectBindStream
= (statement, stream, error) => AdbcStatusCode.InvalidState;
+
+ [Fact]
+ public void StatusesMapToErrno()
+ {
+ Assert.Equal(CAdbcDriverExporter.Errno.ECANCELED,
CAdbcDriverExporter.Errno.FromStatus(AdbcStatusCode.Cancelled));
+ Assert.Equal(CAdbcDriverExporter.Errno.ETIMEDOUT,
CAdbcDriverExporter.Errno.FromStatus(AdbcStatusCode.Timeout));
+ Assert.Equal(CAdbcDriverExporter.Errno.EINVAL,
CAdbcDriverExporter.Errno.FromStatus(AdbcStatusCode.InvalidArgument));
+ Assert.Equal(CAdbcDriverExporter.Errno.ENOTSUP,
CAdbcDriverExporter.Errno.FromStatus(AdbcStatusCode.NotImplemented));
+ Assert.Equal(CAdbcDriverExporter.Errno.EIO,
CAdbcDriverExporter.Errno.FromStatus(AdbcStatusCode.IOError));
+ Assert.Equal(CAdbcDriverExporter.Errno.EIO,
CAdbcDriverExporter.Errno.FromStatus(AdbcStatusCode.UnknownError));
+ }
+
[Fact]
public void UnknownOptionIsNotFound()
{
@@ -528,6 +812,10 @@ namespace Apache.Arrow.Adbc.Tests
public override int DriverVersion => AdbcVersion.Version_1_1_0;
+ public bool Disposed { get; private set; }
+
+ public override void Dispose() => Disposed = true;
+
public override AdbcDatabase Open(IReadOnlyDictionary<string,
string> parameters)
{
var options = new Dictionary<string, object>();
@@ -683,6 +971,14 @@ namespace Apache.Arrow.Adbc.Tests
throw new PlainException();
case "FAIL_CANCELED":
throw new OperationCanceledException();
+ case "STREAM_FAIL_DETAILED":
+ return new QueryResult(-1, new FailingStream(new
DetailedException()));
+ case "STREAM_FAIL_CANCELLED":
+ return new QueryResult(-1, new FailingStream(new
AdbcException("read cancelled", AdbcStatusCode.Cancelled)));
+ case "STREAM_FAIL_CANCELED":
+ return new QueryResult(-1, new FailingStream(new
OperationCanceledException("token cancelled")));
+ case "STREAM_FAIL_SCHEMA":
+ return new QueryResult(-1, new FailingStream(new
AdbcException("no access", AdbcStatusCode.Unauthorized), failSchema: true));
case "BLOCK":
Started.TrySetResult(true);
if (_cancelled.Wait(TimeSpan.FromSeconds(30)))
@@ -758,5 +1054,80 @@ namespace Apache.Arrow.Adbc.Tests
public void Dispose() { _batch?.Dispose(); _batch = null; }
}
+
+ /// <summary>
+ /// Returns one batch and then fails with the given exception, or
fails to report
+ /// its schema if <c>failSchema</c> is set.
+ /// </summary>
+ private sealed class FailingStream : IArrowArrayStream
+ {
+ private static readonly Schema s_schema = new Schema.Builder()
+ .Field(f =>
f.Name("x").DataType(Int32Type.Default).Nullable(false))
+ .Build();
+
+ private readonly Exception _exception;
+ private readonly bool _failSchema;
+ private bool _returnedBatch;
+
+ public FailingStream(Exception exception, bool failSchema = false)
+ {
+ _exception = exception;
+ _failSchema = failSchema;
+ }
+
+ public Schema Schema => _failSchema ? throw _exception : s_schema;
+
+ public bool Disposed { get; private set; }
+
+ public ValueTask<RecordBatch?>
ReadNextRecordBatchAsync(CancellationToken cancellationToken = default)
+ {
+ if (_returnedBatch)
+ {
+ throw _exception;
+ }
+
+ _returnedBatch = true;
+ return new ValueTask<RecordBatch?>(Batch());
+ }
+
+ public static RecordBatch Batch()
+ {
+ Int32Array values = new
Int32Array.Builder().Append(1).Append(2).Build();
+ return new RecordBatch(s_schema, new IArrowArray[] { values },
2);
+ }
+
+ public void Dispose() => Disposed = true;
+ }
+
+ /// <summary>
+ /// Returns one batch after resuming on the caller's synchronization
context.
+ /// </summary>
+ private sealed class ContextResumingStream : IArrowArrayStream
+ {
+ private bool _returnedBatch;
+
+ public Schema Schema => FailingStream.Batch().Schema;
+
+ public async ValueTask<RecordBatch?>
ReadNextRecordBatchAsync(CancellationToken cancellationToken = default)
+ {
+ await Task.Delay(1);
+ if (_returnedBatch)
+ {
+ return null;
+ }
+ _returnedBatch = true;
+ return FailingStream.Batch();
+ }
+
+ public void Dispose() { }
+ }
+
+ /// <summary>
+ /// A synchronization context whose thread is always busy, like a
blocked UI thread.
+ /// </summary>
+ private sealed class BlockedContext : SynchronizationContext
+ {
+ public override void Post(SendOrPostCallback d, object? state) { }
+ }
}
}
