From fed13695a3e6f2adde54fcfe3343e73868c05d2f Mon Sep 17 00:00:00 2001 From: Curt Hagenlocher Date: Fri, 30 Oct 2026 09:13:20 -0700 Subject: [PATCH 1/5] feat(csharp): implement AdbcErrorFromArrayStream for exported and imported streams Streams exported by CAdbcDriverExporter now keep the exception from a failed read, return an errno that matches its status, and report it through AdbcErrorFromArrayStream. CAdbcDriverImporter calls AdbcErrorFromArrayStream when a stream read fails, so the AdbcException keeps its status, SQLSTATE, vendor code and details. Closes #4982 Generated-by: Claude Opus 5.5 --- .../C/CAdbcDriverExporter.Streams.cs | 353 ++++++++++++++++++ .../C/CAdbcDriverExporter.cs | 31 +- .../C/CAdbcDriverImporter.Streams.cs | 224 +++++++++++ .../C/CAdbcDriverImporter.cs | 4 +- csharp/src/Apache.Arrow.Adbc/C/Delegates.cs | 5 + .../Apache.Arrow.Adbc/C/NativeArrowStructs.cs | 67 ++++ .../FixtureDriver.cs | 26 ++ .../AotFixtureTests.cs | 18 + .../ExportedDriverRoundTrip11Tests.cs | 196 ++++++++++ 9 files changed, 897 insertions(+), 27 deletions(-) create mode 100644 csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.Streams.cs create mode 100644 csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs create mode 100644 csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs 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..f1cea1868d --- /dev/null +++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.Streams.cs @@ -0,0 +1,353 @@ +/* + * 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.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 StreamGetSchemaPtr => &GetStreamSchema; + private static unsafe delegate* unmanaged StreamGetNextPtr => &GetStreamNext; + private static unsafe delegate* unmanaged StreamGetLastErrorPtr => &GetStreamLastError; + private static unsafe delegate* unmanaged StreamReleasePtr => &ReleaseStream; + private static unsafe delegate* unmanaged ErrorFromArrayStreamPtr => &GetErrorFromArrayStream; +#else + private static unsafe IntPtr StreamGetSchemaPtr = NativeDelegate.AsNativePointer(GetStreamSchema); + private static unsafe IntPtr StreamGetNextPtr = NativeDelegate.AsNativePointer(GetStreamNext); + private static unsafe IntPtr StreamGetLastErrorPtr = NativeDelegate.AsNativePointer(GetStreamLastError); + private static unsafe IntPtr StreamReleasePtr = NativeDelegate.AsNativePointer(ReleaseStream); + private static unsafe IntPtr ErrorFromArrayStreamPtr = NativeDelegate.AsNativePointer(GetErrorFromArrayStream); +#endif + + /// + /// Exports a stream produced by the managed driver. The exported stream takes + /// ownership on success; if exporting fails, nobody else will dispose it. + /// + /// + /// When reading the stream fails, the exported stream keeps the exception so that + /// AdbcErrorFromArrayStream can report its status, SQLSTATE, vendor code and + /// details, and returns an errno that corresponds to the status. + /// + private 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(); + + ValueTask pending = stream.Stream.ReadNextRecordBatchAsync(); + RecordBatch? batch = pending.IsCompletedSuccessfully ? pending.Result : pending.AsTask().GetAwaiter().GetResult(); + if (batch != null) + { + CArrowArrayExporter.ExportRecordBatch(batch, cArray); + } + return 0; + } + catch (Exception e) + { + return stream?.SetError(e) ?? Errno.EINVAL; + } + } + +#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; + } + } + + /// + /// The state behind an exported : the managed stream + /// and the error from the last call that failed, if any. + /// + 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; + } + + /// + /// Records as the stream's error and returns the + /// errno that corresponds to its status. + /// + 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; + } + } + } + } + + /// + /// The errno values that an reports, which depend on the + /// C runtime of the platform. + /// + 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; + } + } + + /// + /// Maps a status to an errno the same way the C and Go drivers do. + /// + 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 through the ADBC C API. /// /// - /// When a caller requests ADBC 1.1.0, every 1.1.0 entry point is populated - /// except ErrorFromArrayStream; exported streams report errors through - /// get_last_error instead. Functionality that the managed driver doesn't - /// override surfaces as , 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 + /// , which is the same status a driver + /// manager reports for a missing entry point. /// - public class CAdbcDriverExporter + public partial class CAdbcDriverExporter { /// /// The value of ADBC_ERROR_VENDOR_CODE_PRIVATE_DATA. A caller sets an error's @@ -279,8 +278,7 @@ public unsafe static AdbcStatusCode AdbcDriverInit(int version, CAdbcDriver* nat { 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 @@ private unsafe static void FillError(CAdbcError* error, Exception exception) } } - /// - /// Exports a stream produced by the managed driver. The exported stream takes - /// ownership on success; if exporting fails, nobody else will dispose it. - /// - 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..ca372efd82 --- /dev/null +++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs @@ -0,0 +1,224 @@ +/* + * 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 + { + /// + /// A result stream imported from a driver. When the stream fails, it asks the driver + /// for the ADBC error behind the failure (see AdbcErrorFromArrayStream) and + /// throws it as an . + /// + private sealed unsafe class ImportedArrayStream : IArrowArrayStream + { + private readonly ImportedAdbcDriver? _driver; + private readonly Schema _schema; + private CArrowArrayStream* _cStream; + + /// + /// Imports , which must have been allocated with + /// . On success, the imported stream owns + /// it; on failure, the caller still does. + /// + 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); + } + + _cStream = cStream; + } + + ~ImportedArrayStream() + { + Dispose(); + } + + public Schema Schema => _schema; + + public ValueTask ReadNextRecordBatchAsync(CancellationToken cancellationToken = default) + { + if (_cStream == null) + { + throw new ObjectDisposedException(nameof(ImportedArrayStream)); + } + + if (cancellationToken.IsCancellationRequested) + { + return new ValueTask(Task.FromCanceled(cancellationToken)); + } + + try + { + return new ValueTask(ReadNext()); + } + catch (Exception e) + { + return new ValueTask(Task.FromException(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; + FreeStream(cStream); + } + GC.SuppressFinalize(this); + } + + /// + /// Releases and frees a stream allocated with . + /// + /// + /// Unlike , this can release a stream that + /// exported in the same process on .NET Framework, + /// where a callback created from a managed delegate can only be called through a + /// delegate of the same type. + /// + public static void FreeStream(CArrowArrayStream* cStream) + { + NativeArrayStream* nativeStream = NativeArrayStream.From(cStream); + if (nativeStream->release != default) + { +#if NET5_0_OR_GREATER + nativeStream->release(cStream); +#else + Marshal.GetDelegateForFunctionPointer(nativeStream->release)(cStream); +#endif + nativeStream->release = default; + } + CArrowArrayStream.Free(cStream); + } + + /// + /// Builds the exception for a stream call that returned : + /// the driver's ADBC error if it reports one, and otherwise the stream's last error. + /// + 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(nativeDriver.ErrorFromArrayStream)(cStream, &status); +#endif + // The stream owns the error, so it must not be released here + if (error != null) + { + IReadOnlyList>? 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 = Marshal.GetDelegateForFunctionPointer(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 Marshal.GetDelegateForFunctionPointer(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 Marshal.GetDelegateForFunctionPointer(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 @@ public void ForgetStream() 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 @@ public unsafe void Dispose() } 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..e87aea4c6f --- /dev/null +++ b/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs @@ -0,0 +1,67 @@ +/* + * 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 Apache.Arrow.C; + +namespace Apache.Arrow.Adbc.C +{ + /// + /// The layout of , 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 AdbcErrorFromArrayStream). + /// + [StructLayout(LayoutKind.Sequential)] + internal unsafe struct NativeArrayStream + { +#if NET5_0_OR_GREATER + public delegate* unmanaged get_schema; + public delegate* unmanaged get_next; + public delegate* unmanaged get_last_error; + public delegate* unmanaged 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; + } + + /// + /// The layout of , whose release callback Apache.Arrow doesn't expose. + /// + [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; + } +} 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 @@ public override QueryResult ExecuteQuery() { throw new DetailedException(); } + if (_sqlQuery == "FAIL_STREAM") + { + return new QueryResult(-1, new FailingStream(ResultSchema(), new DetailedException())); + } _executed = true; @@ -254,4 +258,26 @@ public SingleBatchStream(Schema schema, RecordBatch batch) public void Dispose() { _batch?.Dispose(); _batch = null; } } + + /// + /// IArrowArrayStream that fails on its first read, after the result was handed out. + /// + 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 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 @@ public void ErrorDetailsCrossAotBoundary() 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()); + 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(async () => await stream.ReadNextRecordBatchAsync()); + Assert.Equal(AdbcStatusCode.InvalidArgument, ex.Status); + Assert.Equal("42000", ex.SqlState); + KeyValuePair 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..4f24dde16e 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,155 @@ public void ErrorWithoutDetailsReportsVendorCode() Assert.Empty(ex.Details); } + [Fact] + public async Task StreamErrorKeepsStatusAndDetails() + { + using AdbcDriver imported = Load(new Driver11()); + using AdbcDatabase db = imported.Open(new Dictionary()); + 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(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()); + 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(async () => await stream.ReadNextRecordBatchAsync()); + Assert.Equal(AdbcStatusCode.Cancelled, ex.Status); + Assert.Equal(message, ex.Message); + } + + [Fact] + public void StreamSchemaErrorKeepsStatus() + { + using AdbcDriver imported = Load(new Driver11()); + using AdbcDatabase db = imported.Open(new Dictionary()); + using AdbcConnection conn = db.Connect(null); + using AdbcStatement stmt = conn.CreateStatement(); + stmt.SqlQuery = "STREAM_FAIL_SCHEMA"; + + AdbcException ex = Assert.ThrowsAny(() => 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()); + 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(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 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() { @@ -683,6 +834,14 @@ public override QueryResult ExecuteQuery() 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 +917,42 @@ public SingleBatchStream(Schema schema, RecordBatch batch) public void Dispose() { _batch?.Dispose(); _batch = null; } } + + /// + /// Returns one batch and then fails with the given exception, or fails to report + /// its schema if failSchema is set. + /// + 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 ValueTask ReadNextRecordBatchAsync(CancellationToken cancellationToken = default) + { + if (_returnedBatch) + { + throw _exception; + } + + _returnedBatch = true; + Int32Array values = new Int32Array.Builder().Append(1).Append(2).Build(); + return new ValueTask(new RecordBatch(s_schema, new IArrowArray[] { values }, 2)); + } + + public void Dispose() { } + } } } From 2e2cae80e35a2756fd164798e0681b78f0d40d42 Mon Sep 17 00:00:00 2001 From: Curt Hagenlocher Date: Fri, 30 Oct 2026 09:30:19 -0700 Subject: [PATCH 2/5] fix(csharp): keep stream callbacks compatible with Apache.Arrow on .NET Framework On .NET Framework, a callback created from a managed delegate can only be called through a delegate of the same type, so use Apache.Arrow's stream callback types for exported and imported streams. Also start the read on a context-free thread when get_next is called on a thread with a synchronization context. Generated-by: Claude Opus 5.5 --- .../C/CAdbcDriverExporter.Streams.cs | 28 +++- .../C/CAdbcDriverImporter.Streams.cs | 15 +- .../Apache.Arrow.Adbc/C/NativeArrowStructs.cs | 48 ++++++ .../ExportedDriverRoundTrip11Tests.cs | 146 +++++++++++++++++- 4 files changed, 221 insertions(+), 16 deletions(-) diff --git a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.Streams.cs b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.Streams.cs index f1cea1868d..e7a5b83f6b 100644 --- a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.Streams.cs +++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.Streams.cs @@ -17,6 +17,7 @@ using System; using System.Runtime.InteropServices; +using System.Threading; using System.Threading.Tasks; using Apache.Arrow.C; using Apache.Arrow.Ipc; @@ -32,10 +33,10 @@ public partial class CAdbcDriverExporter private static unsafe delegate* unmanaged StreamReleasePtr => &ReleaseStream; private static unsafe delegate* unmanaged ErrorFromArrayStreamPtr => &GetErrorFromArrayStream; #else - private static unsafe IntPtr StreamGetSchemaPtr = NativeDelegate.AsNativePointer(GetStreamSchema); - private static unsafe IntPtr StreamGetNextPtr = NativeDelegate.AsNativePointer(GetStreamNext); - private static unsafe IntPtr StreamGetLastErrorPtr = NativeDelegate.AsNativePointer(GetStreamLastError); - private static unsafe IntPtr StreamReleasePtr = NativeDelegate.AsNativePointer(ReleaseStream); + private static unsafe IntPtr StreamGetSchemaPtr = ArrayStreamCallbacks.Export(GetStreamSchema, ArrayStreamCallbacks.GetSchemaType); + private static unsafe IntPtr StreamGetNextPtr = ArrayStreamCallbacks.Export(GetStreamNext, ArrayStreamCallbacks.GetNextType); + private static unsafe IntPtr StreamGetLastErrorPtr = ArrayStreamCallbacks.Export(GetStreamLastError, ArrayStreamCallbacks.GetLastErrorType); + private static unsafe IntPtr StreamReleasePtr = ArrayStreamCallbacks.Export(ReleaseStream, ArrayStreamCallbacks.ReleaseType); private static unsafe IntPtr ErrorFromArrayStreamPtr = NativeDelegate.AsNativePointer(GetErrorFromArrayStream); #endif @@ -48,7 +49,7 @@ public partial class CAdbcDriverExporter /// AdbcErrorFromArrayStream can report its status, SQLSTATE, vendor code and /// details, and returns an errno that corresponds to the status. /// - private unsafe static void ExportStream(IArrowArrayStream? stream, CArrowArrayStream* cStream) + internal unsafe static void ExportStream(IArrowArrayStream? stream, CArrowArrayStream* cStream) { try { @@ -118,8 +119,7 @@ private unsafe static int GetStreamNext(CArrowArrayStream* cStream, CArrowArray* stream = GetExportedStream(cStream) ?? throw new ArgumentException("not a stream exported by this driver", nameof(cStream)); stream.ClearError(); - ValueTask pending = stream.Stream.ReadNextRecordBatchAsync(); - RecordBatch? batch = pending.IsCompletedSuccessfully ? pending.Result : pending.AsTask().GetAwaiter().GetResult(); + RecordBatch? batch = ReadNextBatch(stream.Stream); if (batch != null) { CArrowArrayExporter.ExportRecordBatch(batch, cArray); @@ -132,6 +132,20 @@ private unsafe static int GetStreamNext(CArrowArrayStream* cStream, CArrowArray* } } + 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 pending = stream.ReadNextRecordBatchAsync(); + return pending.IsCompletedSuccessfully ? pending.Result : pending.AsTask().GetAwaiter().GetResult(); + } + #if NET5_0_OR_GREATER [UnmanagedCallersOnly] #endif diff --git a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs index ca372efd82..ee7e84b565 100644 --- a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs +++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs @@ -140,10 +140,9 @@ public void Dispose() /// Releases and frees a stream allocated with . /// /// - /// Unlike , this can release a stream that - /// exported in the same process on .NET Framework, - /// where a callback created from a managed delegate can only be called through a - /// delegate of the same type. + /// On .NET Framework, this releases the stream through the same callback types as + /// the rest of this class (see ArrayStreamCallbacks), so it can release streams + /// exported in the same process by either Apache.Arrow or . /// public static void FreeStream(CArrowArrayStream* cStream) { @@ -153,7 +152,7 @@ public static void FreeStream(CArrowArrayStream* cStream) #if NET5_0_OR_GREATER nativeStream->release(cStream); #else - Marshal.GetDelegateForFunctionPointer(nativeStream->release)(cStream); + ArrayStreamCallbacks.Import(nativeStream->release, ArrayStreamCallbacks.ReleaseType)(cStream); #endif nativeStream->release = default; } @@ -193,7 +192,7 @@ private Exception StreamError(CArrowArrayStream* cStream, int errno) #if NET5_0_OR_GREATER byte* message = nativeStream->get_last_error(cStream); #else - byte* message = Marshal.GetDelegateForFunctionPointer(nativeStream->get_last_error)(cStream); + byte* message = ArrayStreamCallbacks.Import(nativeStream->get_last_error, ArrayStreamCallbacks.GetLastErrorType)(cStream); #endif return new Exception(message == null ? $"Array stream operation failed with no message. Error code: {errno}" @@ -206,7 +205,7 @@ private static int GetSchema(CArrowArrayStream* cStream, CArrowSchema* cSchema) #if NET5_0_OR_GREATER return nativeStream->get_schema(cStream, cSchema); #else - return Marshal.GetDelegateForFunctionPointer(nativeStream->get_schema)(cStream, cSchema); + return ArrayStreamCallbacks.Import(nativeStream->get_schema, ArrayStreamCallbacks.GetSchemaType)(cStream, cSchema); #endif } @@ -216,7 +215,7 @@ private static int GetNext(CArrowArrayStream* cStream, CArrowArray* cArray) #if NET5_0_OR_GREATER return nativeStream->get_next(cStream, cArray); #else - return Marshal.GetDelegateForFunctionPointer(nativeStream->get_next)(cStream, cArray); + return ArrayStreamCallbacks.Import(nativeStream->get_next, ArrayStreamCallbacks.GetNextType)(cStream, cArray); #endif } } diff --git a/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs b/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs index e87aea4c6f..fcc037e401 100644 --- a/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs +++ b/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs @@ -16,6 +16,7 @@ */ using System; +using System.Reflection; using System.Runtime.InteropServices; using Apache.Arrow.C; @@ -64,4 +65,51 @@ internal unsafe struct NativeArray public static NativeArray* From(CArrowArray* array) => (NativeArray*)array; } + +#if !NET5_0_OR_GREATER + /// + /// Creates and calls callbacks on .NET Framework. + /// + /// + /// On .NET Framework, a function pointer created from a managed delegate can only be called + /// through a delegate of the same type. Apache.Arrow exports and calls stream callbacks + /// through its own internal delegate types, so ADBC uses those types too, to stay compatible + /// with streams that Apache.Arrow exports and with consumers that import streams through it. + /// If they can't be found, ADBC falls back to its own types. + /// + 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"); + + /// + /// Returns a function pointer for the static method behind . + /// + public static IntPtr Export(T callback, Type? arrowType) where T : Delegate + { + Delegate exported = arrowType == null ? callback : Delegate.CreateDelegate(arrowType, callback.Method); + return NativeDelegate.AsNativePointer(exported); + } + + /// + /// Returns a delegate that calls . + /// + public static T Import(IntPtr pointer, Type? arrowType) where T : Delegate + { + if (arrowType == null) + { + return Marshal.GetDelegateForFunctionPointer(pointer); + } + + // For a pointer created from a managed delegate, this returns that delegate + Delegate imported = Marshal.GetDelegateForFunctionPointer(pointer, arrowType); + 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/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs b/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs index 4f24dde16e..4c4c87e77a 100644 --- a/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs +++ b/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs @@ -465,6 +465,112 @@ public unsafe void ExportedStreamReportsErrnoAndError() #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(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)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()); + 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(() => 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() { @@ -940,6 +1046,8 @@ public FailingStream(Exception exception, bool failSchema = false) public Schema Schema => _failSchema ? throw _exception : s_schema; + public bool Disposed { get; private set; } + public ValueTask ReadNextRecordBatchAsync(CancellationToken cancellationToken = default) { if (_returnedBatch) @@ -948,11 +1056,47 @@ public FailingStream(Exception exception, bool failSchema = false) } _returnedBatch = true; + return new ValueTask(Batch()); + } + + public static RecordBatch Batch() + { Int32Array values = new Int32Array.Builder().Append(1).Append(2).Build(); - return new ValueTask(new RecordBatch(s_schema, new IArrowArray[] { values }, 2)); + return new RecordBatch(s_schema, new IArrowArray[] { values }, 2); + } + + public void Dispose() => Disposed = true; + } + + /// + /// Returns one batch after resuming on the caller's synchronization context. + /// + private sealed class ContextResumingStream : IArrowArrayStream + { + private bool _returnedBatch; + + public Schema Schema => FailingStream.Batch().Schema; + + public async ValueTask ReadNextRecordBatchAsync(CancellationToken cancellationToken = default) + { + await Task.Delay(1); + if (_returnedBatch) + { + return null; + } + _returnedBatch = true; + return FailingStream.Batch(); } public void Dispose() { } } + + /// + /// A synchronization context whose thread is always busy, like a blocked UI thread. + /// + private sealed class BlockedContext : SynchronizationContext + { + public override void Post(SendOrPostCallback d, object? state) { } + } } } From ae66fdac2a869a36ecd563937397ed7f85b287f7 Mon Sep 17 00:00:00 2001 From: Curt Hagenlocher Date: Fri, 30 Oct 2026 09:45:27 -0700 Subject: [PATCH 3/5] refactor(csharp): call imported stream callbacks without reflection The non-generic Marshal.GetDelegateForFunctionPointer returns the original delegate for a managed callback without casting it, so the importer can wrap a delegate of any type instead of looking up Apache.Arrow's types. Generated-by: Claude Opus 5.5 --- .../C/CAdbcDriverImporter.Streams.cs | 13 +++++----- .../Apache.Arrow.Adbc/C/NativeArrowStructs.cs | 26 +++++++++---------- 2 files changed, 18 insertions(+), 21 deletions(-) diff --git a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs index ee7e84b565..7d93ebff64 100644 --- a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs +++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs @@ -140,9 +140,8 @@ public void Dispose() /// Releases and frees a stream allocated with . /// /// - /// On .NET Framework, this releases the stream through the same callback types as - /// the rest of this class (see ArrayStreamCallbacks), so it can release streams - /// exported in the same process by either Apache.Arrow or . + /// On .NET Framework, this releases the stream through ArrayStreamCallbacks, so + /// it can release a stream exported in the same process with any delegate type. /// public static void FreeStream(CArrowArrayStream* cStream) { @@ -152,7 +151,7 @@ public static void FreeStream(CArrowArrayStream* cStream) #if NET5_0_OR_GREATER nativeStream->release(cStream); #else - ArrayStreamCallbacks.Import(nativeStream->release, ArrayStreamCallbacks.ReleaseType)(cStream); + ArrayStreamCallbacks.Import(nativeStream->release)(cStream); #endif nativeStream->release = default; } @@ -192,7 +191,7 @@ private Exception StreamError(CArrowArrayStream* cStream, int errno) #if NET5_0_OR_GREATER byte* message = nativeStream->get_last_error(cStream); #else - byte* message = ArrayStreamCallbacks.Import(nativeStream->get_last_error, ArrayStreamCallbacks.GetLastErrorType)(cStream); + byte* message = ArrayStreamCallbacks.Import(nativeStream->get_last_error)(cStream); #endif return new Exception(message == null ? $"Array stream operation failed with no message. Error code: {errno}" @@ -205,7 +204,7 @@ private static int GetSchema(CArrowArrayStream* cStream, CArrowSchema* cSchema) #if NET5_0_OR_GREATER return nativeStream->get_schema(cStream, cSchema); #else - return ArrayStreamCallbacks.Import(nativeStream->get_schema, ArrayStreamCallbacks.GetSchemaType)(cStream, cSchema); + return ArrayStreamCallbacks.Import(nativeStream->get_schema)(cStream, cSchema); #endif } @@ -215,7 +214,7 @@ private static int GetNext(CArrowArrayStream* cStream, CArrowArray* cArray) #if NET5_0_OR_GREATER return nativeStream->get_next(cStream, cArray); #else - return ArrayStreamCallbacks.Import(nativeStream->get_next, ArrayStreamCallbacks.GetNextType)(cStream, cArray); + return ArrayStreamCallbacks.Import(nativeStream->get_next)(cStream, cArray); #endif } } diff --git a/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs b/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs index fcc037e401..f8bafe150c 100644 --- a/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs +++ b/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs @@ -71,11 +71,13 @@ internal unsafe struct NativeArray /// Creates and calls callbacks on .NET Framework. /// /// - /// On .NET Framework, a function pointer created from a managed delegate can only be called - /// through a delegate of the same type. Apache.Arrow exports and calls stream callbacks - /// through its own internal delegate types, so ADBC uses those types too, to stay compatible - /// with streams that Apache.Arrow exports and with consumers that import streams through it. - /// If they can't be found, ADBC falls back to its own types. + /// On .NET Framework, + /// 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 + /// , 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. /// internal static class ArrayStreamCallbacks { @@ -94,17 +96,13 @@ public static IntPtr Export(T callback, Type? arrowType) where T : Delegate } /// - /// Returns a delegate that calls . + /// Returns a delegate that calls , whichever delegate type + /// it was created from. /// - public static T Import(IntPtr pointer, Type? arrowType) where T : Delegate + public static T Import(IntPtr pointer) where T : Delegate { - if (arrowType == null) - { - return Marshal.GetDelegateForFunctionPointer(pointer); - } - - // For a pointer created from a managed delegate, this returns that delegate - Delegate imported = Marshal.GetDelegateForFunctionPointer(pointer, arrowType); + // 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"); } From 7f03e2d021b3a03b60d9b3db34aa32d8eb9e9e90 Mon Sep 17 00:00:00 2001 From: Curt Hagenlocher Date: Fri, 30 Oct 2026 09:46:22 -0700 Subject: [PATCH 4/5] docs(csharp): reference the arrow-dotnet issue for stream callback types Generated-by: Claude Opus 5.5 --- csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs b/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs index f8bafe150c..02d800a6ac 100644 --- a/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs +++ b/csharp/src/Apache.Arrow.Adbc/C/NativeArrowStructs.cs @@ -77,7 +77,8 @@ internal unsafe struct NativeArray /// , 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. + /// 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. /// internal static class ArrayStreamCallbacks { From 87397b98d8eae8cbc330477f0c1cab07369a9192 Mon Sep 17 00:00:00 2001 From: Curt Hagenlocher Date: Fri, 30 Oct 2026 09:51:13 -0700 Subject: [PATCH 5/5] fix(csharp): keep the driver loaded while an imported stream is open An imported stream calls into the driver until it's released, including ErrorFromArrayStream when a read fails, so hold a driver reference for the stream's lifetime as the database, connection and statement do. Generated-by: Claude Opus 5.5 --- .../C/CAdbcDriverImporter.Streams.cs | 13 ++++++-- .../ExportedDriverRoundTrip11Tests.cs | 31 +++++++++++++++++++ 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs index 7d93ebff64..19dbec6192 100644 --- a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs +++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Streams.cs @@ -42,7 +42,8 @@ private sealed unsafe class ImportedArrayStream : IArrowArrayStream /// /// Imports , which must have been allocated with /// . On success, the imported stream owns - /// it; on failure, the caller still does. + /// it and holds a reference to , whose callbacks it calls + /// until it's disposed; on failure, the caller still owns the stream. /// public ImportedArrayStream(ImportedAdbcDriver? driver, CArrowArrayStream* cStream) { @@ -68,6 +69,7 @@ public ImportedArrayStream(ImportedAdbcDriver? driver, CArrowArrayStream* cStrea CArrowSchema.Free(cSchema); } + _driver?.AddReference(); _cStream = cStream; } @@ -131,7 +133,14 @@ public void Dispose() if (cStream != null) { _cStream = null; - FreeStream(cStream); + try + { + FreeStream(cStream); + } + finally + { + _driver?.RemoveReference(); + } } GC.SuppressFinalize(this); } diff --git a/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs b/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs index 4c4c87e77a..62e84997c4 100644 --- a/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs +++ b/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTrip11Tests.cs @@ -371,6 +371,33 @@ public async Task StreamCancellationIsReportedAsCancelled(string query, string m 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())) + 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(async () => await stream.ReadNextRecordBatchAsync()); + Assert.Equal(AdbcStatusCode.Cancelled, ex.Status); + Assert.False(fixture.Disposed); + } + + Assert.True(fixture.Disposed); + } + [Fact] public void StreamSchemaErrorKeepsStatus() { @@ -785,6 +812,10 @@ private sealed class Driver11 : AdbcDriver 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 parameters) { var options = new Dictionary();