28103508122

Committed 24 Jun 2026 01:51PM UTC coverage: 89.358% (-0.1%) from 89.483%

Build # 28103508122

Build Type

push

github

Committed by

web-flow

Commit Message

Add Apache Arrow result streaming to DuckDBCommand (#334)

* Add Apache Arrow result streaming to DuckDBCommand

Adds a managed Apache Arrow read surface built on DuckDB's current Arrow C Data
Interface (duckdb_to_arrow_schema / duckdb_data_chunk_to_arrow), addressing #26.

- Bindings: duckdb_result_get_arrow_options, duckdb_to_arrow_schema,
  duckdb_data_chunk_to_arrow, error-data helpers, and a DuckDBArrowOptions safe handle.
- Data: DuckDBArrowArrayStream (IArrowArrayStream) that builds the schema once and
  converts each data chunk into an Arrow RecordBatch via Apache.Arrow's C importers.
- DuckDBCommand.ExecuteArrowStream() and ExecuteArrowBatchesAsync() public APIs.
- Apache.Arrow dependency added to DuckDB.NET.Data only.
- Tests covering schema, scalar values, nulls, multi-chunk streaming, and the stream API.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Honor UseStreamingMode in Arrow APIs and fix result stream cleanup

Pass UseStreamingMode through ExecuteArrowStream and fetch chunks via the
streaming or materialized path accordingly. Dispose Arrow options and close
the result if schema building fails in the stream constructor. Add tests for
streaming mode, cancellation, and dispose semantics.

* Refactor Arrow error handling to typed DuckDBErrorData handle

- Add DuckDBErrorData SafeHandle wrapping duckdb_error_data
- Move error-data P/Invokes to NativeMethods.ErrorData
- Retype duckdb_to_arrow_schema/data_chunk_to_arrow returns to DuckDBErrorData
- Add reusable ThrowOnError extension throwing DuckDBException
- Make DuckDBArrowArrayStream internal

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Coverage Stats

1359 of 1583 branches covered (85.85%)

Branch coverage included in aggregate %.

84 of 94 new or added lines in 4 files covered. (89.36%)

2856 of 3134 relevant lines covered (91.13%)

421574.19 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.23

/DuckDB.NET.Data/Arrow/DuckDBArrowArrayStream.cs

using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
using Apache.Arrow;
using Apache.Arrow.C;
using Apache.Arrow.Ipc;
using DuckDB.NET.Native;

namespace DuckDB.NET.Data.Arrow;

/// <summary>
/// Streams the rows of a DuckDB query result as Apache Arrow <see cref="RecordBatch"/> values using
/// DuckDB's Arrow C Data Interface (<c>duckdb_to_arrow_schema</c> / <c>duckdb_data_chunk_to_arrow</c>).
/// Each DuckDB data chunk is converted into one Arrow record batch and imported with no row-by-row marshaling.
/// </summary>
internal sealed class DuckDBArrowArrayStream : IArrowArrayStream
{
    private DuckDBResult result;
    private readonly DuckDBArrowOptions arrowOptions;
    private readonly bool streaming;
    private bool disposed;

    public Schema Schema { get; }

    internal DuckDBArrowArrayStream(DuckDBResult result)
    {
        this.result = result;

        arrowOptions = NativeMethods.Arrow.DuckDBResultGetArrowOptions(ref this.result);
        if (arrowOptions.IsInvalid)
        {
            this.result.Close();
            throw new InvalidOperationException("Failed to obtain Arrow options from the DuckDB result.");
        }

        streaming = NativeMethods.Types.DuckDBResultIsStreaming(this.result) > 0;

        try
        {
            Schema = BuildSchema();
        }
        catch
        {
            arrowOptions.Dispose();
            this.result.Close();
            throw;
        }
    }

    private unsafe Schema BuildSchema()
    {
        var columnCount = NativeMethods.Query.DuckDBColumnCount(ref result);

        var logicalTypes = new DuckDBLogicalType[columnCount];
        var typeHandles = new IntPtr[columnCount];
        var namePointers = new IntPtr[columnCount];

        try
        {
            for (var index = 0UL; index < columnCount; index++)
            {
                var logicalType = NativeMethods.Query.DuckDBColumnLogicalType(ref result, (long)index);
                logicalTypes[index] = logicalType;
                typeHandles[index] = logicalType.DangerousGetHandle();

                var name = NativeMethods.Query.DuckDBColumnName(ref result, (long)index);
                namePointers[index] = Marshal.StringToCoTaskMemUTF8(name);
            }

            var cSchema = CArrowSchema.Create();

            try
            {
                fixed (IntPtr* typesPointer = typeHandles)
                fixed (IntPtr* namesPointer = namePointers)
                {
                    var error = NativeMethods.Arrow.DuckDBToArrowSchema(arrowOptions, (IntPtr)typesPointer, (IntPtr)namesPointer, columnCount, (IntPtr)cSchema);
                    error.ThrowOnError("Failed to convert the DuckDB result schema to an Arrow schema.");
                }

                return CArrowSchemaImporter.ImportSchema(cSchema);
            }
            finally
            {
                CArrowSchema.Free(cSchema);
            }
        }
        finally
        {
            foreach (var pointer in namePointers)
            {
                if (pointer != IntPtr.Zero)
                {
                    Marshal.FreeCoTaskMem(pointer);
                }
            }

            foreach (var logicalType in logicalTypes)
            {
                logicalType?.Dispose();
            }
        }
    }

    public ValueTask<RecordBatch?> ReadNextRecordBatchAsync(CancellationToken cancellationToken = default)
    {
        ObjectDisposedException.ThrowIf(disposed, this);

        if (cancellationToken.IsCancellationRequested)
        {
            return new ValueTask<RecordBatch?>(Task.FromCanceled<RecordBatch?>(cancellationToken));
        }

        var chunk = streaming
            ? NativeMethods.StreamingResult.DuckDBStreamFetchChunk(result)
            : NativeMethods.Query.DuckDBFetchChunk(result);

        if (chunk.IsInvalid)
        {
            chunk.Dispose();
            return new ValueTask<RecordBatch?>((RecordBatch?)null);
        }

        try
        {
            return new ValueTask<RecordBatch?>(ConvertChunk(chunk));
        }
        finally
        {
            chunk.Dispose();
        }
    }

    private unsafe RecordBatch ConvertChunk(DuckDBDataChunk chunk)
    {
        var cArray = CArrowArray.Create();

        try
        {
            var error = NativeMethods.Arrow.DuckDBDataChunkToArrow(arrowOptions, chunk, (IntPtr)cArray);
            error.ThrowOnError("Failed to convert a DuckDB data chunk to an Arrow array.");

            return CArrowArrayImporter.ImportRecordBatch(cArray, Schema);
        }
        finally
        {
            CArrowArray.Free(cArray);
        }
    }

    public void Dispose()
    {
        if (disposed)
        {
            return;
        }

        disposed = true;

        arrowOptions.Dispose();
        result.Close();
    }
}

1	using System.Runtime.InteropServices;
2	using System.Threading;
3	using System.Threading.Tasks;
4	using Apache.Arrow;
5	using Apache.Arrow.C;
6	using Apache.Arrow.Ipc;
7	using DuckDB.NET.Native;
8
9	namespace DuckDB.NET.Data.Arrow;
10
11	/// <summary>
12	/// Streams the rows of a DuckDB query result as Apache Arrow <see cref="RecordBatch"/> values using
13	/// DuckDB's Arrow C Data Interface (<c>duckdb_to_arrow_schema</c> / <c>duckdb_data_chunk_to_arrow</c>).
14	/// Each DuckDB data chunk is converted into one Arrow record batch and imported with no row-by-row marshaling.
15	/// </summary>
16	internal sealed class DuckDBArrowArrayStream : IArrowArrayStream
17	{
18	private DuckDBResult result;
19	private readonly DuckDBArrowOptions arrowOptions;
20	private readonly bool streaming;
21	private bool disposed;
22
23	public Schema Schema { get; }	42✔
24
25	internal DuckDBArrowArrayStream(DuckDBResult result)	33✔
26	{
27	this.result = result;	33✔
28
29	arrowOptions = NativeMethods.Arrow.DuckDBResultGetArrowOptions(ref this.result);	33✔
30	if (arrowOptions.IsInvalid)	33!
31	{
NEW 32	this.result.Close();	×
NEW 33	throw new InvalidOperationException("Failed to obtain Arrow options from the DuckDB result.");	×
34	}
35
36	streaming = NativeMethods.Types.DuckDBResultIsStreaming(this.result) > 0;	33✔
37
38	try
39	{
40	Schema = BuildSchema();	33✔
41	}	33✔
NEW 42	catch	×
43	{
NEW 44	arrowOptions.Dispose();	×
NEW 45	this.result.Close();	×
NEW 46	throw;	×
47	}
48	}	33✔
49
50	private unsafe Schema BuildSchema()
51	{
52	var columnCount = NativeMethods.Query.DuckDBColumnCount(ref result);	33✔
53
54	var logicalTypes = new DuckDBLogicalType[columnCount];	33✔
55	var typeHandles = new IntPtr[columnCount];	33✔
56	var namePointers = new IntPtr[columnCount];	33✔
57
58	try
59	{
60	for (var index = 0UL; index < columnCount; index++)	162✔
61	{
62	var logicalType = NativeMethods.Query.DuckDBColumnLogicalType(ref result, (long)index);	48✔
63	logicalTypes[index] = logicalType;	48✔
64	typeHandles[index] = logicalType.DangerousGetHandle();	48✔
65
66	var name = NativeMethods.Query.DuckDBColumnName(ref result, (long)index);	48✔
67	namePointers[index] = Marshal.StringToCoTaskMemUTF8(name);	48✔
68	}
69
70	var cSchema = CArrowSchema.Create();	33✔
71
72	try
73	{
74	fixed (IntPtr* typesPointer = typeHandles)	33!
75	fixed (IntPtr* namesPointer = namePointers)	33!
76	{
77	var error = NativeMethods.Arrow.DuckDBToArrowSchema(arrowOptions, (IntPtr)typesPointer, (IntPtr)namesPointer, columnCount, (IntPtr)cSchema);	33✔
78	error.ThrowOnError("Failed to convert the DuckDB result schema to an Arrow schema.");	33✔
79	}
80
81	return CArrowSchemaImporter.ImportSchema(cSchema);	33✔
82	}
83	finally
84	{
85	CArrowSchema.Free(cSchema);	33✔
86	}	33✔
87	}
88	finally
89	{
90	foreach (var pointer in namePointers)	162✔
91	{
92	if (pointer != IntPtr.Zero)	48✔
93	{
94	Marshal.FreeCoTaskMem(pointer);	48✔
95	}
96	}
97
98	foreach (var logicalType in logicalTypes)	162✔
99	{
100	logicalType?.Dispose();	48!
101	}
102	}	33✔
103	}	33✔
104
105	public ValueTask<RecordBatch?> ReadNextRecordBatchAsync(CancellationToken cancellationToken = default)
106	{
107	ObjectDisposedException.ThrowIf(disposed, this);	60✔
108
109	if (cancellationToken.IsCancellationRequested)	57✔
110	{
111	return new ValueTask<RecordBatch?>(Task.FromCanceled<RecordBatch?>(cancellationToken));	3✔
112	}
113
114	var chunk = streaming	54✔
115	? NativeMethods.StreamingResult.DuckDBStreamFetchChunk(result)	54✔
116	: NativeMethods.Query.DuckDBFetchChunk(result);	54✔
117
118	if (chunk.IsInvalid)	54✔
119	{
120	chunk.Dispose();	21✔
121	return new ValueTask<RecordBatch?>((RecordBatch?)null);	21✔
122	}
123
124	try
125	{
126	return new ValueTask<RecordBatch?>(ConvertChunk(chunk));	33✔
127	}
128	finally
129	{
130	chunk.Dispose();	33✔
131	}	33✔
132	}	33✔
133
134	private unsafe RecordBatch ConvertChunk(DuckDBDataChunk chunk)
135	{
136	var cArray = CArrowArray.Create();	33✔
137
138	try
139	{
140	var error = NativeMethods.Arrow.DuckDBDataChunkToArrow(arrowOptions, chunk, (IntPtr)cArray);	33✔
141	error.ThrowOnError("Failed to convert a DuckDB data chunk to an Arrow array.");	33✔
142
143	return CArrowArrayImporter.ImportRecordBatch(cArray, Schema);	33✔
144	}
145	finally
146	{
147	CArrowArray.Free(cArray);	33✔
148	}	33✔
149	}	33✔
150
151	public void Dispose()
152	{
153	if (disposed)	36✔
154	{
155	return;	3✔
156	}
157
158	disposed = true;	33✔
159
160	arrowOptions.Dispose();	33✔
161	result.Close();	33✔
162	}	33✔
163	}

Giorgi / DuckDB.NET / 28103508122

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous