This is an automated email from the ASF dual-hosted git repository.
CurtHagenlocher pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-adbc.git
The following commit(s) were added to refs/heads/main by this push:
new 2f0d732d2 fix(csharp): fix bugs in the C driver importer and exporter
(#4884)
2f0d732d2 is described below
commit 2f0d732d2c9c800cd1cf125a72eea9644dd77973
Author: Curt Hagenlocher <[email protected]>
AuthorDate: Tue Oct 13 08:27:06 2026 -0700
fix(csharp): fix bugs in the C driver importer and exporter (#4884)
## What's changed
This fixes bugs in the C# driver importer (`CAdbcDriverImporter`) and
exporter (`CAdbcDriverExporter`), plus one in `AdbcOptions`. All of them
are in the code on `main` today. I found them while working on fuller
ADBC 1.1 support in C#, which will follow in a separate PR. None of
these fixes depends on that work, and none is a regression from it.
Each fix is a separate commit with its own explanation, so the PR can be
reviewed one commit at a time.
### Crashes and memory safety
- **`ExecutePartitioned` could crash the process.**
- The importer allocated the partitions struct without zeroing it.
It then called its `release` callback whether or not the driver
had written the struct. Any failure, including a driver that
doesn't implement partitions, called a garbage function pointer.
- On netstandard2.0, `release != null` on an `IntPtr` is always true,
so the callback was also called when it was zero.
- In the exporter, the result schema was never exported. The
partitions struct was also published before its buffers were
allocated.
- **Error messages were freed with the wrong allocator.**
- On .NET 5+, the exporter allocated error messages with CoTaskMem
but freed them with `FreeHGlobal`.
- On netstandard2.0, `MarshalExtensions.StringToCoTaskMemUTF8`
allocated with HGlobal, but the importer frees those strings with
`FreeCoTaskMem`.
- **The exporter's error reporting could crash the caller or itself.**
- It dereferenced a NULL `error`, though the argument is optional.
- It freed an error left in the caller's struct as if this exporter
had created it, instead of calling that error's own `release`.
- It could throw out of an unmanaged entry point, for example from a
driver's `SqlState` override, which ends the process.
### Incorrect behavior
- **Updates failed through the exporter.** A NULL stream passed to
`StatementExecuteQuery` means the caller doesn't want a result set.
The exporter ignored that and exported into the NULL pointer. The
importer's own `ExecuteUpdate` passes NULL, so exporting a managed
driver and importing it again broke updates.
- **Vendor error codes were lost in both directions.** The exporter
always reported 0. The importer read the vendor code only when the
error also had a SQLSTATE.
- **Missing functions failed on .NET Framework.** When a driver leaves
`ConnectionCommit`, `ConnectionRollback`, `ConnectionCancel` or
`StatementCancel` empty, the importer fills in a function that
reports `NotImplemented`. On .NET Framework, calling it threw
`InvalidCastException` instead. The fill-ins were created with a
different delegate type from the one they're called through, and
.NET Framework returns the original delegate for a function pointer
created from a managed delegate.
- **A failed cleanup hid the real error.** When creating a database,
connection or statement failed and the importer's cleanup release
also failed, the caller saw the release error.
- **Managed streams leaked if exporting them failed.**
- **Two ingest modes were unrecognized.**
`AdbcOptions.GetIngestMode(string)` didn't recognize the ADBC 1.1
ingest modes `replace` and `create_append`.
### Docs
- `AdbcConnection.GetInfo` now says that an empty code list requests
all codes. The drivers already treat it that way, and it's how a
NULL list from the C API arrives.
## Testing
- **New tests:** most fixes have a test in
`ExportedDriverRoundTripTests` or the new `AdbcOptionsTests`. I ran
these against `main` without their fixes, and each one fails:
- The update test fails.
- The vendor-code test fails.
- The cleanup-error test fails.
- The partitions test crashes the test host with an access violation.
- The missing-functions test fails with `InvalidCastException` on
net472.
- **Tests that only run on .NET 5 and later:** two tests call the
exporter's function pointers directly. They cover the cleanup-error
fix and the error-reporting fix.
- **No test:** the allocator fix and the stream leak. Making the
export fail would mean writing through a NULL output pointer.
- **Results:**
- The full solution builds without warnings.
- The core test project passes on net10.0, net8.0 and net472.
- The AoT interop tests pass against a fixture published from this
branch.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
csharp/src/Apache.Arrow.Adbc/AdbcConnection.cs | 3 +-
csharp/src/Apache.Arrow.Adbc/AdbcOptions.cs | 3 +
.../src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.cs | 151 +++++++++++---
.../C/CAdbcDriverImporter.Defaults.cs | 8 +-
.../src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.cs | 98 ++++++---
.../Extensions/MarshalExtensions.cs | 2 +-
.../Apache.Arrow.Adbc.Tests/AdbcOptionsTests.cs | 42 ++++
.../ExportedDriverRoundTripTests.cs | 218 ++++++++++++++++++++-
8 files changed, 470 insertions(+), 55 deletions(-)
diff --git a/csharp/src/Apache.Arrow.Adbc/AdbcConnection.cs
b/csharp/src/Apache.Arrow.Adbc/AdbcConnection.cs
index 7f9493c4b..f1a110b09 100644
--- a/csharp/src/Apache.Arrow.Adbc/AdbcConnection.cs
+++ b/csharp/src/Apache.Arrow.Adbc/AdbcConnection.cs
@@ -75,7 +75,8 @@ namespace Apache.Arrow.Adbc
/// Get metadata about the driver/database.
/// </summary>
/// <param name="codes">
- /// The metadata items to fetch.
+ /// The metadata items to fetch. An empty list requests all of the
items the
+ /// driver supports; this is what a NULL list means in the C API.
/// </param>
/// <returns>
/// Metadata about the driver and/or database
diff --git a/csharp/src/Apache.Arrow.Adbc/AdbcOptions.cs
b/csharp/src/Apache.Arrow.Adbc/AdbcOptions.cs
index 8116f6b82..daf525b9f 100644
--- a/csharp/src/Apache.Arrow.Adbc/AdbcOptions.cs
+++ b/csharp/src/Apache.Arrow.Adbc/AdbcOptions.cs
@@ -205,6 +205,7 @@ namespace Apache.Arrow.Adbc
/// Insert data; create the table if it does not exist, or error
if the table exists
/// but the schema does not match the schema of the data to append
/// (<see cref="AdbcStatusCode.AlreadyExists"/>).
+ /// Added as part of API version 1.1.0.
/// </summary>
public const string CreateAppend =
"adbc.ingest.mode.create_append";
}
@@ -276,6 +277,8 @@ namespace Apache.Arrow.Adbc
{
IngestMode.Create => BulkIngestMode.Create,
IngestMode.Append => BulkIngestMode.Append,
+ IngestMode.Replace => BulkIngestMode.Replace,
+ IngestMode.CreateAppend => BulkIngestMode.CreateAppend,
_ => throw new NotSupportedException("unknown ingestion mode"),
};
}
diff --git a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.cs
b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.cs
index 7f47213f4..9ea257a76 100644
--- a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.cs
+++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverExporter.cs
@@ -155,15 +155,79 @@ namespace Apache.Arrow.Adbc.C
internal unsafe static void ReleaseError(CAdbcError* error)
{
- if (error != null && ((IntPtr)error->message) != IntPtr.Zero)
+ if (error == null)
{
- Marshal.FreeHGlobal((IntPtr)error->message);
+ return;
}
+
+ if (error->message != null)
+ {
+ Marshal.FreeCoTaskMem((IntPtr)error->message);
+ error->message = null;
+ }
+
+ error->release = default;
}
private unsafe static AdbcStatusCode SetError(CAdbcError* error,
Exception exception)
{
- ReleaseError(error);
+ AdbcStatusCode status = exception switch
+ {
+ AdbcException adbcException => adbcException.Status,
+ ArgumentException => AdbcStatusCode.InvalidArgument,
+ _ => AdbcStatusCode.UnknownError,
+ };
+
+ // The error is optional
+ if (error == null)
+ {
+ return status;
+ }
+
+ try
+ {
+ FillError(error, exception);
+ }
+ catch (Exception)
+ {
+ // SetError runs in the catch block of an unmanaged entry
point, where an
+ // exception would end the process. Report the status without
a message.
+ }
+
+ return status;
+ }
+
+ /// <summary>
+ /// Exports a stream produced by the managed driver. The exported
stream takes
+ /// ownership on success; if exporting fails, nobody else will dispose
it.
+ /// </summary>
+ private unsafe static void ExportStream(IArrowArrayStream? stream,
CArrowArrayStream* cStream)
+ {
+ try
+ {
+ CArrowArrayStreamExporter.ExportArrayStream(stream!, cStream);
+ }
+ catch
+ {
+ stream?.Dispose();
+ throw;
+ }
+ }
+
+ private unsafe static void FillError(CAdbcError* error, Exception
exception)
+ {
+ // The caller may pass an error that still holds a value, possibly
from another
+ // driver, so release it through its own callback.
+ if (error->release != default)
+ {
+#if NET5_0_OR_GREATER
+ error->release(error);
+#else
+
Marshal.GetDelegateForFunctionPointer<ErrorRelease>(error->release)(error);
+#endif
+ }
+ error->message = null;
+ error->release = default;
error->message =
(byte*)MarshalExtensions.StringToCoTaskMemUTF8(exception.Message);
error->sqlstate0 = (byte)0;
@@ -188,15 +252,14 @@ namespace Apache.Arrow.Adbc.C
}
}
}
- return adbcException.Status;
- }
- if (exception is ArgumentException)
- {
- return AdbcStatusCode.InvalidArgument;
+ // INT32_MIN would tell an ADBC 1.1 caller that private_data
holds error
+ // details, which this exporter doesn't provide
+ if (adbcException.NativeError != int.MinValue)
+ {
+ error->vendor_code = adbcException.NativeError;
+ }
}
-
- return AdbcStatusCode.UnknownError;
}
private static IntPtr FromDisposable(IDisposable d)
@@ -237,6 +300,11 @@ namespace Apache.Arrow.Adbc.C
[UnmanagedCallersOnly]
#endif
private unsafe static void ReleasePartitions(CAdbcPartitions*
partitions)
+ {
+ FreePartitions(partitions);
+ }
+
+ private unsafe static void FreePartitions(CAdbcPartitions* partitions)
{
if (partitions != null)
{
@@ -671,13 +739,25 @@ namespace Apache.Arrow.Adbc.C
{
GCHandle gch =
GCHandle.FromIntPtr((IntPtr)nativeStatement->private_data);
AdbcStatement stub = (AdbcStatement)gch.Target!;
+
+ // A null stream means the caller doesn't expect a result set
+ if (stream == null)
+ {
+ UpdateResult update = stub.ExecuteUpdate();
+ if (rows != null)
+ {
+ *rows = update.AffectedRows;
+ }
+ return AdbcStatusCode.Success;
+ }
+
var result = stub.ExecuteQuery();
if (rows != null)
{
*rows = result.RowCount;
}
- CArrowArrayStreamExporter.ExportArrayStream(result.Stream,
stream);
+ ExportStream(result.Stream, stream);
return AdbcStatusCode.Success;
}
catch (Exception e)
@@ -701,21 +781,42 @@ namespace Apache.Arrow.Adbc.C
*rows = result.AffectedRows;
}
- partitions->release = ReleasePartitionsPtr;
- partitions->num_partitions = result.PartitionDescriptors.Count;
- partitions->partitions =
(byte**)Marshal.AllocHGlobal(IntPtr.Size * result.PartitionDescriptors.Count);
- partitions->partition_lengths =
(nuint*)Marshal.AllocHGlobal(IntPtr.Size * result.PartitionDescriptors.Count);
- for (int i = 0; i < partitions->num_partitions; i++)
+ // Build the result locally and publish it only once it's
complete, so that a
+ // failure partway through never hands the caller a partially
filled struct.
+ int count = result.PartitionDescriptors.Count;
+ CAdbcPartitions built = new CAdbcPartitions { num_partitions =
count };
+ try
{
- ReadOnlySpan<byte> partition =
result.PartitionDescriptors[i].Descriptor;
- partitions->partition_lengths[i] = (nuint)partition.Length;
- partitions->partitions[i] =
(byte*)Marshal.AllocHGlobal(partition.Length);
- fixed (void* descriptor = partition)
+ built.partitions =
(byte**)Marshal.AllocHGlobal(IntPtr.Size * Math.Max(1, count));
+ for (int i = 0; i < count; i++)
+ {
+ built.partitions[i] = null;
+ }
+ built.partition_lengths =
(nuint*)Marshal.AllocHGlobal(IntPtr.Size * Math.Max(1, count));
+ for (int i = 0; i < count; i++)
{
- Buffer.MemoryCopy(descriptor,
partitions->partitions[i], partition.Length, partition.Length);
+ ReadOnlySpan<byte> partition =
result.PartitionDescriptors[i].Descriptor;
+ built.partition_lengths[i] = (nuint)partition.Length;
+ built.partitions[i] =
(byte*)Marshal.AllocHGlobal(Math.Max(1, partition.Length));
+ fixed (void* descriptor = partition)
+ {
+ Buffer.MemoryCopy(descriptor, built.partitions[i],
partition.Length, partition.Length);
+ }
}
+
+ if (schema != null)
+ {
+ CArrowSchemaExporter.ExportSchema(result.Schema,
schema);
+ }
+ }
+ catch
+ {
+ FreePartitions(&built);
+ throw;
}
+ built.release = ReleasePartitionsPtr;
+ *partitions = built;
return AdbcStatusCode.Success;
}
catch (Exception e)
@@ -992,7 +1093,7 @@ namespace Apache.Arrow.Adbc.C
IArrowArrayStream stream = Connection.GetObjects(goDepth,
catalogPattern, dbSchemaPattern, tableNamePattern, tableTypes,
columnNamePattern);
- CArrowArrayStreamExporter.ExportArrayStream(stream, cstream);
+ ExportStream(stream, cstream);
}
public unsafe void GetTableSchema(byte* catalog, byte* db_schema,
byte* table_name, CArrowSchema* cschema)
@@ -1008,7 +1109,7 @@ namespace Apache.Arrow.Adbc.C
public unsafe void GetTableTypes(CArrowArrayStream* cArrayStream)
{
-
CArrowArrayStreamExporter.ExportArrayStream(Connection.GetTableTypes(),
cArrayStream);
+ ExportStream(Connection.GetTableTypes(), cArrayStream);
}
public unsafe void ReadPartition(byte* serializedPartition, int
serialized_length, CArrowArrayStream* stream)
@@ -1019,7 +1120,7 @@ namespace Apache.Arrow.Adbc.C
Buffer.MemoryCopy(serializedPartition, partitionPtr,
serialized_length, serialized_length);
}
-
CArrowArrayStreamExporter.ExportArrayStream(Connection.ReadPartition(new
PartitionDescriptor(partition)), stream);
+ ExportStream(Connection.ReadPartition(new
PartitionDescriptor(partition)), stream);
}
public unsafe void GetInfo(int* info_codes, int info_codes_length,
CArrowArrayStream* stream)
@@ -1031,7 +1132,7 @@ namespace Apache.Arrow.Adbc.C
Buffer.MemoryCopy(info_codes, infoCodesPtr, length,
length);
}
-
CArrowArrayStreamExporter.ExportArrayStream(Connection.GetInfo(infoCodes.ToList()),
stream);
+ ExportStream(Connection.GetInfo(infoCodes.ToList()), stream);
}
public unsafe void InitConnection(ref CAdbcDatabase nativeDatabase)
diff --git a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Defaults.cs
b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Defaults.cs
index 9408d299b..53be2db14 100644
--- a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Defaults.cs
+++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.Defaults.cs
@@ -37,7 +37,7 @@ namespace Apache.Arrow.Adbc.C
}
#if !NET5_0_OR_GREATER
- private static unsafe IntPtr ConnectionCommitDefault =
NativeDelegate<ConnectionCommit>.AsNativePointer(ConnectionCommitDefaultImpl);
+ private static unsafe IntPtr ConnectionCommitDefault =
NativeDelegate<ConnectionFn>.AsNativePointer(ConnectionCommitDefaultImpl);
#else
private static unsafe delegate* unmanaged<CAdbcConnection*,
CAdbcError*, AdbcStatusCode> ConnectionCommitDefault =>
&ConnectionCommitDefaultImpl;
[UnmanagedCallersOnly]
@@ -103,7 +103,7 @@ namespace Apache.Arrow.Adbc.C
}
#if !NET5_0_OR_GREATER
- private static unsafe IntPtr ConnectionRollbackDefault =
NativeDelegate<ConnectionRollback>.AsNativePointer(ConnectionRollbackDefaultImpl);
+ private static unsafe IntPtr ConnectionRollbackDefault =
NativeDelegate<ConnectionFn>.AsNativePointer(ConnectionRollbackDefaultImpl);
#else
private static unsafe delegate* unmanaged<CAdbcConnection*,
CAdbcError*, AdbcStatusCode> ConnectionRollbackDefault =>
&ConnectionRollbackDefaultImpl;
[UnmanagedCallersOnly]
@@ -327,7 +327,7 @@ namespace Apache.Arrow.Adbc.C
}
#if !NET5_0_OR_GREATER
- private static unsafe IntPtr ConnectionCancelDefault =
NativeDelegate<ConnectionCancel>.AsNativePointer(ConnectionCancelDefaultImpl);
+ private static unsafe IntPtr ConnectionCancelDefault =
NativeDelegate<ConnectionFn>.AsNativePointer(ConnectionCancelDefaultImpl);
#else
private static unsafe delegate* unmanaged<CAdbcConnection*,
CAdbcError*, AdbcStatusCode> ConnectionCancelDefault =>
&ConnectionCancelDefaultImpl;
[UnmanagedCallersOnly]
@@ -437,7 +437,7 @@ namespace Apache.Arrow.Adbc.C
}
#if !NET5_0_OR_GREATER
- private static unsafe IntPtr StatementCancelDefault =
NativeDelegate<StatementCancel>.AsNativePointer(StatementCancelDefaultImpl);
+ private static unsafe IntPtr StatementCancelDefault =
NativeDelegate<StatementFn>.AsNativePointer(StatementCancelDefaultImpl);
#else
private static unsafe delegate* unmanaged<CAdbcStatement*,
CAdbcError*, AdbcStatusCode> StatementCancelDefault =>
&StatementCancelDefaultImpl;
[UnmanagedCallersOnly]
diff --git a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.cs
b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.cs
index b250665e1..25710e865 100644
--- a/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.cs
+++ b/csharp/src/Apache.Arrow.Adbc/C/CAdbcDriverImporter.cs
@@ -340,9 +340,16 @@ namespace Apache.Arrow.Adbc.C
{
if (result == null && nativeDatabase.private_data != null)
{
- using (CallHelper caller = new CallHelper())
+ // Don't let a failed release hide the error that got
us here
+ try
+ {
+ using (CallHelper caller = new CallHelper())
+ {
+ caller.Call(Driver.DatabaseRelease, ref
nativeDatabase);
+ }
+ }
+ catch (AdbcException)
{
- caller.Call(Driver.DatabaseRelease, ref
nativeDatabase);
}
}
}
@@ -356,25 +363,48 @@ namespace Apache.Arrow.Adbc.C
if (parameters == null) throw new
ArgumentNullException(nameof(parameters));
CAdbcDatabase nativeDatabase = new CAdbcDatabase();
- using (CallHelper caller = new CallHelper())
+ ImportedAdbcDatabase? result = null;
+ try
{
- caller.Call(Driver.DatabaseNew, ref nativeDatabase);
-
- foreach (KeyValuePair<string, object> pair in parameters)
+ using (CallHelper caller = new CallHelper())
{
- switch (pair.Value)
+ caller.Call(Driver.DatabaseNew, ref nativeDatabase);
+
+ foreach (KeyValuePair<string, object> pair in
parameters)
{
- case int intValue:
- case long longValue:
- // TODO!
- break;
+ switch (pair.Value)
+ {
+ case int intValue:
+ case long longValue:
+ // TODO!
+ break;
+ }
}
+
+ caller.Call(Driver.DatabaseInit, ref nativeDatabase);
}
- caller.Call(Driver.DatabaseInit, ref nativeDatabase);
+ result = new ImportedAdbcDatabase(this, nativeDatabase);
+ }
+ finally
+ {
+ if (result == null && nativeDatabase.private_data != null)
+ {
+ // Don't let a failed release hide the error that got
us here
+ try
+ {
+ using (CallHelper caller = new CallHelper())
+ {
+ caller.Call(Driver.DatabaseRelease, ref
nativeDatabase);
+ }
+ }
+ catch (AdbcException)
+ {
+ }
+ }
}
- return new ImportedAdbcDatabase(this, nativeDatabase);
+ return result;
}
public unsafe override void Dispose()
@@ -453,9 +483,16 @@ namespace Apache.Arrow.Adbc.C
{
if (result == null && nativeConnection.private_data !=
null)
{
- using (CallHelper caller = new CallHelper())
+ // Don't let a failed release hide the error that got
us here
+ try
+ {
+ using (CallHelper caller = new CallHelper())
+ {
+ caller.Call(Driver.ConnectionRelease, ref
nativeConnection);
+ }
+ }
+ catch (AdbcException)
{
- caller.Call(Driver.ConnectionRelease, ref
nativeConnection);
}
}
}
@@ -603,9 +640,16 @@ namespace Apache.Arrow.Adbc.C
{
if (result == null && nativeStatement.private_data != null)
{
- using (CallHelper caller = new CallHelper())
+ // Don't let a failed release hide the error that got
us here
+ try
+ {
+ using (CallHelper caller = new CallHelper())
+ {
+ caller.Call(Driver.StatementRelease, ref
nativeStatement);
+ }
+ }
+ catch (AdbcException)
{
- caller.Call(Driver.StatementRelease, ref
nativeStatement);
}
}
}
@@ -1026,6 +1070,9 @@ namespace Apache.Arrow.Adbc.C
try
{
nativePartitions =
(CAdbcPartitions*)Marshal.AllocHGlobal(sizeof(CAdbcPartitions));
+ // AllocHGlobal doesn't zero memory, and a failing
driver may not write
+ // the struct, so release must start out null.
+ *nativePartitions = default;
caller.TranslateCode(
#if NET5_0_OR_GREATER
Driver.StatementExecutePartitions
@@ -1044,17 +1091,17 @@ namespace Apache.Arrow.Adbc.C
}
finally
{
- if (nativePartitions->release != null)
+ if (nativePartitions != null)
{
+ if (nativePartitions->release != default)
+ {
#if NET5_0_OR_GREATER
- nativePartitions->release(nativePartitions);
+
nativePartitions->release(nativePartitions);
#else
-
Marshal.GetDelegateForFunctionPointer<PartitionsRelease>(nativePartitions->release)(nativePartitions);
+
Marshal.GetDelegateForFunctionPointer<PartitionsRelease>(nativePartitions->release)(nativePartitions);
#endif
- }
+ }
- if (nativePartitions != null)
- {
Marshal.FreeHGlobal((IntPtr)nativePartitions);
}
}
@@ -1180,6 +1227,11 @@ namespace Apache.Arrow.Adbc.C
{
_sqlState =
Encoding.ASCII.GetString(&fixedError->sqlstate0, 5);
}
+ }
+
+ // INT32_MIN marks an ADBC 1.1 error with details rather than
a vendor code
+ if (error.vendor_code != int.MinValue)
+ {
_nativeError = error.vendor_code;
}
}
diff --git a/csharp/src/Apache.Arrow.Adbc/Extensions/MarshalExtensions.cs
b/csharp/src/Apache.Arrow.Adbc/Extensions/MarshalExtensions.cs
index 23cf57d0e..fc2ad876b 100644
--- a/csharp/src/Apache.Arrow.Adbc/Extensions/MarshalExtensions.cs
+++ b/csharp/src/Apache.Arrow.Adbc/Extensions/MarshalExtensions.cs
@@ -59,7 +59,7 @@ namespace Apache.Arrow.Adbc.Extensions
int nb = Encoding.UTF8.GetMaxByteCount(s.Length);
- IntPtr pMem = Marshal.AllocHGlobal(nb + 1);
+ IntPtr pMem = Marshal.AllocCoTaskMem(nb + 1);
int nbWritten;
byte* pbMem = (byte*)pMem;
diff --git a/csharp/test/Apache.Arrow.Adbc.Tests/AdbcOptionsTests.cs
b/csharp/test/Apache.Arrow.Adbc.Tests/AdbcOptionsTests.cs
new file mode 100644
index 000000000..6673e13f3
--- /dev/null
+++ b/csharp/test/Apache.Arrow.Adbc.Tests/AdbcOptionsTests.cs
@@ -0,0 +1,42 @@
+/*
+ * 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 Xunit;
+
+namespace Apache.Arrow.Adbc.Tests
+{
+ public class AdbcOptionsTests
+ {
+ [Theory]
+ [InlineData(BulkIngestMode.Create, "adbc.ingest.mode.create")]
+ [InlineData(BulkIngestMode.Append, "adbc.ingest.mode.append")]
+ [InlineData(BulkIngestMode.Replace, "adbc.ingest.mode.replace")]
+ [InlineData(BulkIngestMode.CreateAppend,
"adbc.ingest.mode.create_append")]
+ public void IngestModeRoundTrips(BulkIngestMode mode, string value)
+ {
+ Assert.Equal(value, AdbcOptions.GetIngestMode(mode));
+ Assert.Equal(mode, AdbcOptions.GetIngestMode(value));
+ }
+
+ [Fact]
+ public void UnknownIngestModeThrows()
+ {
+ Assert.Throws<NotSupportedException>(() =>
AdbcOptions.GetIngestMode("adbc.ingest.mode.unknown"));
+ }
+ }
+}
diff --git
a/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTripTests.cs
b/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTripTests.cs
index c9d759670..43ff07aef 100644
--- a/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTripTests.cs
+++ b/csharp/test/Apache.Arrow.Adbc.Tests/ExportedDriverRoundTripTests.cs
@@ -17,6 +17,7 @@
using System;
using System.Collections.Generic;
+using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
using Apache.Arrow.Adbc.C;
@@ -114,6 +115,109 @@ namespace Apache.Arrow.Adbc.Tests
Assert.Contains("boom", ex.Message);
}
+ [Fact]
+ public void ExecuteUpdateRoundTripsWithoutAResultStream()
+ {
+ var fixture = new FixtureDriver();
+
+ // The importer's ExecuteUpdate passes a null stream to
StatementExecuteQuery
+ using AdbcDriver imported =
CAdbcDriverImporter.Load(CreateAdapter(fixture));
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "UPDATE";
+
+ Assert.Equal(3, stmt.ExecuteUpdate().AffectedRows);
+ Assert.False(fixture.LastStatement!.WasExecuted);
+ }
+
+ [Fact]
+ public void ExecutePartitionedRoundTrips()
+ {
+ var fixture = new FixtureDriver();
+
+ using AdbcDriver imported =
CAdbcDriverImporter.Load(CreateAdapter(fixture));
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "PARTITIONS";
+
+ PartitionedResult result = stmt.ExecutePartitioned();
+
+ Assert.Equal(5, result.AffectedRows);
+ Assert.Equal("answer",
Assert.Single(result.Schema.FieldsList).Name);
+ Assert.Equal(2, result.PartitionDescriptors.Count);
+ Assert.Equal(new byte[] { 1, 2 },
result.PartitionDescriptors[0].Descriptor.ToArray());
+ Assert.Equal(new byte[] { 3 },
result.PartitionDescriptors[1].Descriptor.ToArray());
+ }
+
+ [Fact]
+ public void ExecutePartitionedFailureDoesNotReleaseGarbage()
+ {
+ var fixture = new FixtureDriver();
+
+ using AdbcDriver imported =
CAdbcDriverImporter.Load(CreateAdapter(fixture));
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+
+ // The producer fails without writing the partitions struct
+ stmt.SqlQuery = "SELECT 1";
+ for (int i = 0; i < 20; i++)
+ {
+ Assert.Equal(AdbcStatusCode.NotImplemented,
Assert.ThrowsAny<AdbcException>(() => stmt.ExecutePartitioned()).Status);
+ }
+ }
+
+ [Fact]
+ public void MissingFunctionsReportNotImplemented()
+ {
+ // A driver that leaves these entries empty gets the importer's
defaults, which
+ // must be callable through the same delegate types as a real
implementation.
+ var fixture = new FixtureDriver();
+ AdbcDriverInit adapter = (int version, ref CAdbcDriver
nativeDriver, ref CAdbcError error) =>
+ {
+ AdbcStatusCode status = CallExporter(version, ref
nativeDriver, ref error, fixture);
+ unsafe
+ {
+ nativeDriver.ConnectionCommit = default;
+ nativeDriver.ConnectionRollback = default;
+ nativeDriver.ConnectionCancel = default;
+ nativeDriver.StatementCancel = default;
+ }
+ return status;
+ };
+
+ using AdbcDriver imported = CAdbcDriverImporter.Load(adapter);
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+
+ Assert.Equal(AdbcStatusCode.NotImplemented,
Assert.ThrowsAny<AdbcException>(() => conn.Commit()).Status);
+ Assert.Equal(AdbcStatusCode.NotImplemented,
Assert.ThrowsAny<AdbcException>(() => conn.Rollback()).Status);
+ Assert.Equal(AdbcStatusCode.NotImplemented,
Assert.ThrowsAny<AdbcException>(() => conn.Cancel()).Status);
+ Assert.Equal(AdbcStatusCode.NotImplemented,
Assert.ThrowsAny<AdbcException>(() => stmt.Cancel()).Status);
+ }
+
+ [Theory]
+ [InlineData(null)]
+ [InlineData("42000")]
+ public void VendorCodePropagates(string? sqlState)
+ {
+ var fixture = new FixtureDriver { ThrowOnExecute = new
VendorException(sqlState, 1234) };
+
+ using AdbcDriver imported =
CAdbcDriverImporter.Load(CreateAdapter(fixture));
+ using AdbcDatabase db = imported.Open(new Dictionary<string,
string>());
+ using AdbcConnection conn = db.Connect(null);
+ using AdbcStatement stmt = conn.CreateStatement();
+ stmt.SqlQuery = "SELECT 1";
+
+ AdbcException ex = Assert.ThrowsAny<AdbcException>(() =>
stmt.ExecuteQuery());
+ Assert.Equal(AdbcStatusCode.IOError, ex.Status);
+ Assert.Equal(1234, ex.NativeError);
+ Assert.Equal(sqlState, ex.SqlState);
+ }
+
[Fact]
public void StatementSetOptionIsMarshaledToProducer()
{
@@ -208,6 +312,85 @@ namespace Apache.Arrow.Adbc.Tests
Assert.Equal(AdbcStatusCode.NotImplemented, ex.Status);
}
+#if NET5_0_OR_GREATER
+ private static int s_staleErrorReleases;
+
+ [Fact]
+ public unsafe void
ExporterReleasesAnExistingErrorThroughItsOwnCallback()
+ {
+ var fixture = new FixtureDriver { ThrowOnOpen = new
AdbcException("open failed", AdbcStatusCode.InvalidState) };
+
+ CAdbcDriver driver = default;
+ CAdbcError initError = default;
+ Assert.Equal(AdbcStatusCode.Success,
CAdbcDriverExporter.AdbcDriverInit(AdbcVersion.Version_1_0_0, &driver,
&initError, fixture));
+
+ CAdbcDatabase database = default;
+ CAdbcError error = default;
+ Assert.Equal(AdbcStatusCode.Success, driver.DatabaseNew(&database,
&error));
+
+ // An error left over from an earlier call, owned by someone else
+ error.message = (byte*)Marshal.StringToCoTaskMemUTF8("stale");
+ error.release = &ReleaseStaleError;
+ s_staleErrorReleases = 0;
+
+ Assert.Equal(AdbcStatusCode.InvalidState,
driver.DatabaseInit(&database, &error));
+ Assert.Equal(1, s_staleErrorReleases);
+ Assert.Equal("open failed",
Marshal.PtrToStringUTF8((IntPtr)error.message));
+ error.release(&error);
+
+ // The error is optional
+ Assert.Equal(AdbcStatusCode.InvalidState,
driver.DatabaseInit(&database, null));
+
+ Assert.Equal(AdbcStatusCode.Success,
driver.DatabaseRelease(&database, &error));
+ Assert.Equal(AdbcStatusCode.Success, driver.release(&driver,
&error));
+ }
+
+ private static unsafe delegate* unmanaged<CAdbcDatabase*, CAdbcError*,
AdbcStatusCode> s_realDatabaseRelease;
+ private static int s_databaseReleases;
+
+ [Theory]
+ [InlineData(false)]
+ [InlineData(true)]
+ public unsafe void
FailedOpenReleasesTheDatabaseWithoutHidingTheError(bool objectParameters)
+ {
+ var fixture = new FixtureDriver { ThrowOnOpen = new
AdbcException("open failed", AdbcStatusCode.InvalidState) };
+ AdbcDriverInit adapter = (int version, ref CAdbcDriver
nativeDriver, ref CAdbcError error) =>
+ {
+ AdbcStatusCode status = CallExporter(version, ref
nativeDriver, ref error, fixture);
+ s_realDatabaseRelease = nativeDriver.DatabaseRelease;
+ nativeDriver.DatabaseRelease = &FailingDatabaseRelease;
+ return status;
+ };
+
+ using AdbcDriver imported = CAdbcDriverImporter.Load(adapter);
+ s_databaseReleases = 0;
+
+ AdbcException ex = Assert.ThrowsAny<AdbcException>(() =>
objectParameters
+ ? imported.Open(new Dictionary<string, object>())
+ : imported.Open(new Dictionary<string, string>()));
+ Assert.Equal(AdbcStatusCode.InvalidState, ex.Status);
+ Assert.Contains("open failed", ex.Message);
+ Assert.Equal(1, s_databaseReleases);
+ }
+
+ [UnmanagedCallersOnly]
+ private static unsafe AdbcStatusCode
FailingDatabaseRelease(CAdbcDatabase* database, CAdbcError* error)
+ {
+ s_databaseReleases++;
+ s_realDatabaseRelease(database, error);
+ return AdbcStatusCode.IOError;
+ }
+
+ [UnmanagedCallersOnly]
+ private static unsafe void ReleaseStaleError(CAdbcError* error)
+ {
+ s_staleErrorReleases++;
+ Marshal.FreeCoTaskMem((IntPtr)error->message);
+ error->message = null;
+ error->release = default;
+ }
+#endif
+
private static AdbcStatusCode CallExporter(int version, ref
CAdbcDriver nativeDriver, ref CAdbcError error, AdbcDriver driver)
{
unsafe
@@ -241,9 +424,12 @@ namespace Apache.Arrow.Adbc.Tests
public FixtureConnection? LastConnection { get; private set; }
public FixtureStatement? LastStatement { get; private set; }
public Exception? ThrowOnExecute { get; set; }
+ public Exception? ThrowOnOpen { get; set; }
public override AdbcDatabase Open(IReadOnlyDictionary<string,
string> parameters)
{
+ if (ThrowOnOpen != null) { throw ThrowOnOpen; }
+
var db = new FixtureDatabase(this, parameters);
LastDatabase = db;
return db;
@@ -347,13 +533,43 @@ namespace Apache.Arrow.Adbc.Tests
}
public override UpdateResult ExecuteUpdate()
- => throw AdbcException.NotImplemented("fixture does not
support ExecuteUpdate");
+ => _sqlQuery == "UPDATE" ? new UpdateResult(3) : throw
AdbcException.NotImplemented("fixture does not support ExecuteUpdate");
+
+ public override PartitionedResult ExecutePartitioned()
+ {
+ if (_sqlQuery != "PARTITIONS")
+ {
+ return base.ExecutePartitioned();
+ }
+
+ return new PartitionedResult(BuildResultSchema(), 5, new[]
+ {
+ new PartitionDescriptor(new byte[] { 1, 2 }),
+ new PartitionDescriptor(new byte[] { 3 }),
+ });
+ }
private static Schema BuildResultSchema() => new Schema.Builder()
.Field(f =>
f.Name("answer").DataType(Int32Type.Default).Nullable(false))
.Build();
}
+ private sealed class VendorException : AdbcException
+ {
+ private readonly string? _sqlState;
+ private readonly int _nativeError;
+
+ public VendorException(string? sqlState, int nativeError)
+ : base("vendor failure", AdbcStatusCode.IOError)
+ {
+ _sqlState = sqlState;
+ _nativeError = nativeError;
+ }
+
+ public override string? SqlState => _sqlState;
+ public override int NativeError => _nativeError;
+ }
+
private sealed class SingleBatchStream : IArrowArrayStream
{
private readonly Schema _schema;
