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-dotnet.git
The following commit(s) were added to refs/heads/main by this push:
new bb2f7ed fix: Call C Data Interface callbacks through any delegate
type on .NET Framework (#492)
bb2f7ed is described below
commit bb2f7ed9723e8d74f9c6ff1390b462754820f121
Author: Curt Hagenlocher <[email protected]>
AuthorDate: Sat Oct 31 07:33:40 2026 -0700
fix: Call C Data Interface callbacks through any delegate type on .NET
Framework (#492)
## What's Changed
On the net462 and netstandard2.0 builds,
`Marshal.GetDelegateForFunctionPointer<T>` returns the original managed
delegate cast to T when the function pointer was created in-process from
a delegate. If another library exported the callback with its own
delegate type, the cast throws `InvalidCastException`.
Add an internal `NativeDelegate.GetDelegateForFunctionPointer<T>` helper
that uses the non-generic overload and wraps a delegate of another type
by binding to its Invoke method, and use it at all call sites.
Closes #491.
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
src/Apache.Arrow/C/CArrowArray.cs | 2 +-
src/Apache.Arrow/C/CArrowArrayImporter.cs | 2 +-
src/Apache.Arrow/C/CArrowArrayStream.cs | 2 +-
src/Apache.Arrow/C/CArrowArrayStreamImporter.cs | 8 +-
src/Apache.Arrow/C/CArrowDeviceArrayStream.cs | 2 +-
.../C/CArrowDeviceArrayStreamImporter.cs | 8 +-
src/Apache.Arrow/C/CArrowSchema.cs | 2 +-
src/Apache.Arrow/C/CArrowSchemaImporter.cs | 2 +-
src/Apache.Arrow/C/NativeDelegate.cs | 20 +
.../CDataInterfaceForeignDelegateTests.cs | 422 +++++++++++++++++++++
10 files changed, 456 insertions(+), 14 deletions(-)
diff --git a/src/Apache.Arrow/C/CArrowArray.cs
b/src/Apache.Arrow/C/CArrowArray.cs
index 882ca8c..a9ba34f 100644
--- a/src/Apache.Arrow/C/CArrowArray.cs
+++ b/src/Apache.Arrow/C/CArrowArray.cs
@@ -84,7 +84,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
array->release(array);
#else
-
Marshal.GetDelegateForFunctionPointer<CArrowArrayExporter.ReleaseArrowArray>(array->release)(array);
+
NativeDelegate.GetDelegateForFunctionPointer<CArrowArrayExporter.ReleaseArrowArray>(array->release)(array);
#endif
Debug.Assert(array->release == default,
"Calling the CArrowArray release func should have
set it to NULL");
diff --git a/src/Apache.Arrow/C/CArrowArrayImporter.cs
b/src/Apache.Arrow/C/CArrowArrayImporter.cs
index a873595..d23ec0c 100644
--- a/src/Apache.Arrow/C/CArrowArrayImporter.cs
+++ b/src/Apache.Arrow/C/CArrowArrayImporter.cs
@@ -123,7 +123,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
cArray->release(cArray);
#else
-
Marshal.GetDelegateForFunctionPointer<CArrowArrayExporter.ReleaseArrowArray>(cArray->release)(cArray);
+
NativeDelegate.GetDelegateForFunctionPointer<CArrowArrayExporter.ReleaseArrowArray>(cArray->release)(cArray);
#endif
}
}
diff --git a/src/Apache.Arrow/C/CArrowArrayStream.cs
b/src/Apache.Arrow/C/CArrowArrayStream.cs
index a8dbfd4..df089a5 100644
--- a/src/Apache.Arrow/C/CArrowArrayStream.cs
+++ b/src/Apache.Arrow/C/CArrowArrayStream.cs
@@ -110,7 +110,7 @@ namespace Apache.Arrow.C
arrayStream->release(arrayStream);
#else
-
Marshal.GetDelegateForFunctionPointer<CArrowArrayStreamExporter.ReleaseArrayStream>(arrayStream->release)(arrayStream);
+
NativeDelegate.GetDelegateForFunctionPointer<CArrowArrayStreamExporter.ReleaseArrayStream>(arrayStream->release)(arrayStream);
#endif
}
Marshal.FreeHGlobal((IntPtr)arrayStream);
diff --git a/src/Apache.Arrow/C/CArrowArrayStreamImporter.cs
b/src/Apache.Arrow/C/CArrowArrayStreamImporter.cs
index bbe7c67..aafb666 100644
--- a/src/Apache.Arrow/C/CArrowArrayStreamImporter.cs
+++ b/src/Apache.Arrow/C/CArrowArrayStreamImporter.cs
@@ -79,7 +79,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
byte* error = arrayStream->get_last_error(arrayStream);
#else
- byte* error =
Marshal.GetDelegateForFunctionPointer<CArrowArrayStreamExporter.GetLastErrorArrayStream>(arrayStream->get_last_error)(arrayStream);
+ byte* error =
NativeDelegate.GetDelegateForFunctionPointer<CArrowArrayStreamExporter.GetLastErrorArrayStream>(arrayStream->get_last_error)(arrayStream);
#endif
if (error == null)
{
@@ -103,7 +103,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
int errno = cArrayStream->get_schema(cArrayStream, &cSchema);
#else
- int errno =
Marshal.GetDelegateForFunctionPointer<CArrowArrayStreamExporter.GetSchemaArrayStream>(cArrayStream->get_schema)(cArrayStream,
&cSchema);
+ int errno =
NativeDelegate.GetDelegateForFunctionPointer<CArrowArrayStreamExporter.GetSchemaArrayStream>(cArrayStream->get_schema)(cArrayStream,
&cSchema);
#endif
if (errno != 0)
{
@@ -141,7 +141,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
int errno = cArrayStream->get_next(cArrayStream, &cArray);
#else
- int errno =
Marshal.GetDelegateForFunctionPointer<CArrowArrayStreamExporter.GetNextArrayStream>(cArrayStream->get_next)(cArrayStream,
&cArray);
+ int errno =
NativeDelegate.GetDelegateForFunctionPointer<CArrowArrayStreamExporter.GetNextArrayStream>(cArrayStream->get_next)(cArrayStream,
&cArray);
#endif
if (errno != 0)
{
@@ -166,7 +166,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
cArrayStream->release(cArrayStream);
#else
-
Marshal.GetDelegateForFunctionPointer<CArrowArrayStreamExporter.ReleaseArrayStream>(cArrayStream->release)(cArrayStream);
+
NativeDelegate.GetDelegateForFunctionPointer<CArrowArrayStreamExporter.ReleaseArrayStream>(cArrayStream->release)(cArrayStream);
#endif
}
}
diff --git a/src/Apache.Arrow/C/CArrowDeviceArrayStream.cs
b/src/Apache.Arrow/C/CArrowDeviceArrayStream.cs
index 45461aa..1c7774e 100644
--- a/src/Apache.Arrow/C/CArrowDeviceArrayStream.cs
+++ b/src/Apache.Arrow/C/CArrowDeviceArrayStream.cs
@@ -115,7 +115,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
stream->release(stream);
#else
-
Marshal.GetDelegateForFunctionPointer<CArrowDeviceArrayStreamExporter.ReleaseDeviceArrayStream>(stream->release)(stream);
+
NativeDelegate.GetDelegateForFunctionPointer<CArrowDeviceArrayStreamExporter.ReleaseDeviceArrayStream>(stream->release)(stream);
#endif
}
Marshal.FreeHGlobal((IntPtr)stream);
diff --git a/src/Apache.Arrow/C/CArrowDeviceArrayStreamImporter.cs
b/src/Apache.Arrow/C/CArrowDeviceArrayStreamImporter.cs
index 6536716..e6c4e65 100644
--- a/src/Apache.Arrow/C/CArrowDeviceArrayStreamImporter.cs
+++ b/src/Apache.Arrow/C/CArrowDeviceArrayStreamImporter.cs
@@ -84,7 +84,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
error = stream->get_last_error(stream);
#else
- error =
Marshal.GetDelegateForFunctionPointer<CArrowDeviceArrayStreamExporter.GetLastErrorDeviceArrayStream>(stream->get_last_error)(stream);
+ error =
NativeDelegate.GetDelegateForFunctionPointer<CArrowDeviceArrayStreamExporter.GetLastErrorDeviceArrayStream>(stream->get_last_error)(stream);
#endif
}
@@ -115,7 +115,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
int errno = cDeviceArrayStream->get_schema(cDeviceArrayStream,
&cSchema);
#else
- int errno =
Marshal.GetDelegateForFunctionPointer<CArrowDeviceArrayStreamExporter.GetSchemaDeviceArrayStream>(cDeviceArrayStream->get_schema)(cDeviceArrayStream,
&cSchema);
+ int errno =
NativeDelegate.GetDelegateForFunctionPointer<CArrowDeviceArrayStreamExporter.GetSchemaDeviceArrayStream>(cDeviceArrayStream->get_schema)(cDeviceArrayStream,
&cSchema);
#endif
if (errno != 0)
{
@@ -153,7 +153,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
int errno =
cDeviceArrayStream->get_next(cDeviceArrayStream, &cDeviceArray);
#else
- int errno =
Marshal.GetDelegateForFunctionPointer<CArrowDeviceArrayStreamExporter.GetNextDeviceArrayStream>(cDeviceArrayStream->get_next)(cDeviceArrayStream,
&cDeviceArray);
+ int errno =
NativeDelegate.GetDelegateForFunctionPointer<CArrowDeviceArrayStreamExporter.GetNextDeviceArrayStream>(cDeviceArrayStream->get_next)(cDeviceArrayStream,
&cDeviceArray);
#endif
if (errno != 0)
{
@@ -178,7 +178,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
cDeviceArrayStream->release(cDeviceArrayStream);
#else
-
Marshal.GetDelegateForFunctionPointer<CArrowDeviceArrayStreamExporter.ReleaseDeviceArrayStream>(cDeviceArrayStream->release)(cDeviceArrayStream);
+
NativeDelegate.GetDelegateForFunctionPointer<CArrowDeviceArrayStreamExporter.ReleaseDeviceArrayStream>(cDeviceArrayStream->release)(cDeviceArrayStream);
#endif
}
}
diff --git a/src/Apache.Arrow/C/CArrowSchema.cs
b/src/Apache.Arrow/C/CArrowSchema.cs
index 50c363b..3695e91 100644
--- a/src/Apache.Arrow/C/CArrowSchema.cs
+++ b/src/Apache.Arrow/C/CArrowSchema.cs
@@ -75,7 +75,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
schema->release(schema);
#else
-
Marshal.GetDelegateForFunctionPointer<CArrowSchemaExporter.ReleaseArrowSchema>(schema->release)(schema);
+
NativeDelegate.GetDelegateForFunctionPointer<CArrowSchemaExporter.ReleaseArrowSchema>(schema->release)(schema);
#endif
}
Marshal.FreeHGlobal((IntPtr)schema);
diff --git a/src/Apache.Arrow/C/CArrowSchemaImporter.cs
b/src/Apache.Arrow/C/CArrowSchemaImporter.cs
index c92c0fa..dfa5c08 100644
--- a/src/Apache.Arrow/C/CArrowSchemaImporter.cs
+++ b/src/Apache.Arrow/C/CArrowSchemaImporter.cs
@@ -156,7 +156,7 @@ namespace Apache.Arrow.C
#if NET5_0_OR_GREATER
_cSchema->release(_cSchema);
#else
-
Marshal.GetDelegateForFunctionPointer<CArrowSchemaExporter.ReleaseArrowSchema>(_cSchema->release)(_cSchema);
+
NativeDelegate.GetDelegateForFunctionPointer<CArrowSchemaExporter.ReleaseArrowSchema>(_cSchema->release)(_cSchema);
#endif
}
}
diff --git a/src/Apache.Arrow/C/NativeDelegate.cs
b/src/Apache.Arrow/C/NativeDelegate.cs
index 744979e..550a700 100644
--- a/src/Apache.Arrow/C/NativeDelegate.cs
+++ b/src/Apache.Arrow/C/NativeDelegate.cs
@@ -32,4 +32,24 @@ namespace Apache.Arrow.C
public IntPtr Pointer { get; }
}
+
+#if !NET5_0_OR_GREATER
+ internal static class NativeDelegate
+ {
+ /// <summary>
+ /// Gets a delegate of type <typeparamref name="T"/> for calling a C
Data Interface callback.
+ /// </summary>
+ /// <remarks>
+ /// If the function pointer was created in this process from a managed
delegate,
+ /// <see
cref="Marshal.GetDelegateForFunctionPointer{TDelegate}(IntPtr)"/> returns that
original
+ /// delegate cast to <typeparamref name="T"/>, which throws if another
library exported the callback
+ /// using its own delegate type with the same signature. In that case,
wrap the original delegate.
+ /// </remarks>
+ public static T GetDelegateForFunctionPointer<T>(IntPtr ptr) where T :
Delegate
+ {
+ Delegate d = Marshal.GetDelegateForFunctionPointer(ptr, typeof(T));
+ return d as T ?? (T)Delegate.CreateDelegate(typeof(T), d,
"Invoke");
+ }
+ }
+#endif
}
diff --git a/test/Apache.Arrow.Tests/CDataInterfaceForeignDelegateTests.cs
b/test/Apache.Arrow.Tests/CDataInterfaceForeignDelegateTests.cs
new file mode 100644
index 0000000..f060c6f
--- /dev/null
+++ b/test/Apache.Arrow.Tests/CDataInterfaceForeignDelegateTests.cs
@@ -0,0 +1,422 @@
+// 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.Diagnostics.CodeAnalysis;
+using System.Runtime.InteropServices;
+using System.Threading.Tasks;
+using Apache.Arrow.C;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Types;
+using Xunit;
+
+namespace Apache.Arrow.Tests
+{
+ /// <summary>
+ /// Tests importing C Data Interface structures whose callbacks were
exported in this
+ /// process by another library, using that library's own delegate types
rather than
+ /// Arrow's internal ones.
+ /// </summary>
+ [Experimental("ArrowDeviceDataApi")]
+ public class CDataInterfaceForeignDelegateTests
+ {
+ private unsafe delegate void ForeignReleaseSchema(CArrowSchema*
schema);
+ private unsafe delegate void ForeignReleaseArray(CArrowArray* array);
+ private unsafe delegate int ForeignGetSchema(CArrowArrayStream*
stream, CArrowSchema* schema);
+ private unsafe delegate int ForeignGetNext(CArrowArrayStream* stream,
CArrowArray* array);
+ private unsafe delegate byte* ForeignGetLastError(CArrowArrayStream*
stream);
+ private unsafe delegate void ForeignReleaseStream(CArrowArrayStream*
stream);
+ private unsafe delegate int
ForeignGetDeviceSchema(CArrowDeviceArrayStream* stream, CArrowSchema* schema);
+ private unsafe delegate int
ForeignGetDeviceNext(CArrowDeviceArrayStream* stream, CArrowDeviceArray* array);
+ private unsafe delegate byte*
ForeignGetDeviceLastError(CArrowDeviceArrayStream* stream);
+ private unsafe delegate void
ForeignReleaseDeviceStream(CArrowDeviceArrayStream* stream);
+
+ private const string ErrorMessage = "foreign stream failure";
+ private const int ErrorCode = 5; // EIO
+
+ // The struct fields are function pointers on .NET 5+ and IntPtr
otherwise; both are pointer-sized.
+ private static unsafe void SetCallback(void* field, IntPtr callback)
+ {
+ *(IntPtr*)field = callback;
+ }
+
+ [Fact]
+ public unsafe void ImportSchema()
+ {
+ bool released = false;
+ ForeignReleaseSchema release = schema =>
+ {
+ released = true;
+ schema->release = default;
+ };
+
+ CArrowSchema* cSchema = CArrowSchema.Create();
+ IntPtr format = Marshal.StringToHGlobalAnsi("n");
+ try
+ {
+ cSchema->format = (byte*)format;
+ SetCallback(&cSchema->release,
Marshal.GetFunctionPointerForDelegate(release));
+
+ ArrowType type = CArrowSchemaImporter.ImportType(cSchema);
+
+ Assert.IsType<NullType>(type);
+ Assert.True(released);
+ }
+ finally
+ {
+ CArrowSchema.Free(cSchema);
+ Marshal.FreeHGlobal(format);
+ GC.KeepAlive(release);
+ }
+ }
+
+ [Fact]
+ public unsafe void ImportArray()
+ {
+ bool released = false;
+ ForeignReleaseArray release = array =>
+ {
+ released = true;
+ array->release = default;
+ };
+
+ CArrowArray* cArray = CArrowArray.Create();
+ try
+ {
+ cArray->length = 3;
+ cArray->null_count = 3;
+ SetCallback(&cArray->release,
Marshal.GetFunctionPointerForDelegate(release));
+
+ using (IArrowArray array =
CArrowArrayImporter.ImportArray(cArray, NullType.Default))
+ {
+ Assert.Equal(3, array.Length);
+ }
+
+ Assert.True(released);
+ }
+ finally
+ {
+ CArrowArray.Free(cArray);
+ GC.KeepAlive(release);
+ }
+ }
+
+ [Fact]
+ public unsafe void CallReleaseFuncOnArray()
+ {
+ bool released = false;
+ ForeignReleaseArray release = array =>
+ {
+ released = true;
+ array->release = default;
+ };
+
+ CArrowArray* cArray = CArrowArray.Create();
+ SetCallback(&cArray->release,
Marshal.GetFunctionPointerForDelegate(release));
+ CArrowArray.Free(cArray);
+
+ Assert.True(released);
+ GC.KeepAlive(release);
+ }
+
+ [Fact]
+ public unsafe void FreeSchema()
+ {
+ bool released = false;
+ ForeignReleaseSchema release = schema =>
+ {
+ released = true;
+ schema->release = default;
+ };
+
+ CArrowSchema* cSchema = CArrowSchema.Create();
+ SetCallback(&cSchema->release,
Marshal.GetFunctionPointerForDelegate(release));
+ CArrowSchema.Free(cSchema);
+
+ Assert.True(released);
+ GC.KeepAlive(release);
+ }
+
+ [Fact]
+ public async Task ImportArrayStream()
+ {
+ using (var foreign = new ForeignArrayStream(failGetNext: false))
+ {
+ using (IArrowArrayStream stream = foreign.Import())
+ {
+ Assert.Equal(foreign.Batch.Schema.FieldsList[0].Name,
stream.Schema.FieldsList[0].Name);
+
+ using (RecordBatch batch = await
stream.ReadNextRecordBatchAsync())
+ {
+ Assert.Equal(foreign.Batch.Length, batch.Length);
+ }
+ Assert.Null(await stream.ReadNextRecordBatchAsync());
+ Assert.False(foreign.Released);
+ }
+
+ Assert.True(foreign.Released);
+ }
+ }
+
+ [Fact]
+ public async Task ImportArrayStreamGetsLastError()
+ {
+ using (var foreign = new ForeignArrayStream(failGetNext: true))
+ {
+ using (IArrowArrayStream stream = foreign.Import())
+ {
+ Exception ex = await Assert.ThrowsAsync<Exception>(
+ async () => await stream.ReadNextRecordBatchAsync());
+ Assert.Equal(ErrorMessage, ex.Message);
+ }
+
+ Assert.True(foreign.Released);
+ }
+ }
+
+ [Fact]
+ public void FreeArrayStream()
+ {
+ using (var foreign = new ForeignArrayStream(failGetNext: false))
+ {
+ foreign.Free();
+
+ Assert.True(foreign.Released);
+ }
+ }
+
+ [Fact]
+ public async Task ImportDeviceArrayStream()
+ {
+ using (var foreign = new ForeignDeviceArrayStream(failGetNext:
false))
+ {
+ using (IArrowArrayStream stream = foreign.Import())
+ {
+ Assert.Equal(foreign.Batch.Schema.FieldsList[0].Name,
stream.Schema.FieldsList[0].Name);
+
+ using (RecordBatch batch = await
stream.ReadNextRecordBatchAsync())
+ {
+ Assert.Equal(foreign.Batch.Length, batch.Length);
+ }
+ Assert.Null(await stream.ReadNextRecordBatchAsync());
+ Assert.False(foreign.Released);
+ }
+
+ Assert.True(foreign.Released);
+ }
+ }
+
+ [Fact]
+ public async Task ImportDeviceArrayStreamGetsLastError()
+ {
+ using (var foreign = new ForeignDeviceArrayStream(failGetNext:
true))
+ {
+ using (IArrowArrayStream stream = foreign.Import())
+ {
+ Exception ex = await Assert.ThrowsAsync<Exception>(
+ async () => await stream.ReadNextRecordBatchAsync());
+ Assert.Equal(ErrorMessage, ex.Message);
+ }
+
+ Assert.True(foreign.Released);
+ }
+ }
+
+ [Fact]
+ public void FreeDeviceArrayStream()
+ {
+ using (var foreign = new ForeignDeviceArrayStream(failGetNext:
false))
+ {
+ foreign.Free();
+
+ Assert.True(foreign.Released);
+ }
+ }
+
+ private static RecordBatch CreateBatch()
+ {
+ return new RecordBatch.Builder()
+ .Append("ints", false, col => col.Int32(builder =>
builder.AppendRange(new[] { 1, 2, 3 })))
+ .Build();
+ }
+
+ /// <summary>
+ /// A single-batch stream implemented with this test's own delegate
types.
+ /// </summary>
+ private sealed unsafe class ForeignArrayStream : IDisposable
+ {
+ private readonly bool _failGetNext;
+ private readonly IntPtr _error;
+ private readonly ForeignGetSchema _getSchema;
+ private readonly ForeignGetNext _getNext;
+ private readonly ForeignGetLastError _getLastError;
+ private readonly ForeignReleaseStream _release;
+ private CArrowArrayStream* _stream;
+ private bool _batchReturned;
+
+ public ForeignArrayStream(bool failGetNext)
+ {
+ _failGetNext = failGetNext;
+ _error = Marshal.StringToHGlobalAnsi(ErrorMessage);
+ Batch = CreateBatch();
+ _getSchema = (stream, schema) =>
+ {
+ CArrowSchemaExporter.ExportSchema(Batch.Schema, schema);
+ return 0;
+ };
+ _getNext = (stream, array) =>
+ {
+ if (_failGetNext)
+ {
+ return ErrorCode;
+ }
+ if (_batchReturned)
+ {
+ *array = default;
+ }
+ else
+ {
+ CArrowArrayExporter.ExportRecordBatch(Batch, array);
+ _batchReturned = true;
+ }
+ return 0;
+ };
+ _getLastError = stream => (byte*)_error;
+ _release = stream =>
+ {
+ Released = true;
+ stream->release = default;
+ };
+
+ _stream = CArrowArrayStream.Create();
+ Export(_stream);
+ }
+
+ public RecordBatch Batch { get; }
+
+ public bool Released { get; private set; }
+
+ public IArrowArrayStream Import() =>
CArrowArrayStreamImporter.ImportArrayStream(_stream);
+
+ public void Free()
+ {
+ CArrowArrayStream.Free(_stream);
+ _stream = null;
+ }
+
+ private void Export(CArrowArrayStream* stream)
+ {
+ SetCallback(&stream->get_schema,
Marshal.GetFunctionPointerForDelegate(_getSchema));
+ SetCallback(&stream->get_next,
Marshal.GetFunctionPointerForDelegate(_getNext));
+ SetCallback(&stream->get_last_error,
Marshal.GetFunctionPointerForDelegate(_getLastError));
+ SetCallback(&stream->release,
Marshal.GetFunctionPointerForDelegate(_release));
+ }
+
+ public void Dispose()
+ {
+ if (_stream != null)
+ {
+ Free();
+ }
+ Batch.Dispose();
+ Marshal.FreeHGlobal(_error);
+ }
+ }
+
+ /// <summary>
+ /// A single-batch device stream implemented with this test's own
delegate types.
+ /// </summary>
+ private sealed unsafe class ForeignDeviceArrayStream : IDisposable
+ {
+ private readonly bool _failGetNext;
+ private readonly IntPtr _error;
+ private readonly ForeignGetDeviceSchema _getSchema;
+ private readonly ForeignGetDeviceNext _getNext;
+ private readonly ForeignGetDeviceLastError _getLastError;
+ private readonly ForeignReleaseDeviceStream _release;
+ private CArrowDeviceArrayStream* _stream;
+ private bool _batchReturned;
+
+ public ForeignDeviceArrayStream(bool failGetNext)
+ {
+ _failGetNext = failGetNext;
+ _error = Marshal.StringToHGlobalAnsi(ErrorMessage);
+ Batch = CreateBatch();
+ _getSchema = (stream, schema) =>
+ {
+ CArrowSchemaExporter.ExportSchema(Batch.Schema, schema);
+ return 0;
+ };
+ _getNext = (stream, array) =>
+ {
+ if (_failGetNext)
+ {
+ return ErrorCode;
+ }
+ if (_batchReturned)
+ {
+ *array = default;
+ }
+ else
+ {
+ CArrowDeviceArrayExporter.ExportRecordBatch(Batch,
array);
+ _batchReturned = true;
+ }
+ return 0;
+ };
+ _getLastError = stream => (byte*)_error;
+ _release = stream =>
+ {
+ Released = true;
+ stream->release = default;
+ };
+
+ _stream = CArrowDeviceArrayStream.Create();
+ Export(_stream);
+ }
+
+ public RecordBatch Batch { get; }
+
+ public bool Released { get; private set; }
+
+ public IArrowArrayStream Import() =>
CArrowDeviceArrayStreamImporter.ImportDeviceArrayStream(_stream);
+
+ public void Free()
+ {
+ CArrowDeviceArrayStream.Free(_stream);
+ _stream = null;
+ }
+
+ private void Export(CArrowDeviceArrayStream* stream)
+ {
+ stream->device_type = ArrowDeviceType.Cpu;
+ SetCallback(&stream->get_schema,
Marshal.GetFunctionPointerForDelegate(_getSchema));
+ SetCallback(&stream->get_next,
Marshal.GetFunctionPointerForDelegate(_getNext));
+ SetCallback(&stream->get_last_error,
Marshal.GetFunctionPointerForDelegate(_getLastError));
+ SetCallback(&stream->release,
Marshal.GetFunctionPointerForDelegate(_release));
+ }
+
+ public void Dispose()
+ {
+ if (_stream != null)
+ {
+ Free();
+ }
+ Batch.Dispose();
+ Marshal.FreeHGlobal(_error);
+ }
+ }
+ }
+}
