28102732334

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

Build # 28102732334

Build Type

Pull #334

github

Committed by

web-flow

Commit Message

Merge 0037674ea into 38ddceb66

Pull Request Pull Request #334: Add Apache Arrow result streaming to DuckDBCommand

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%)

6 existing lines in 2 files now uncovered.

2856 of 3134 relevant lines covered (91.13%)

421088.62 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 / 28102732334

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