davidhcoe commented on code in PR #697: URL: https://github.com/apache/arrow-adbc/pull/697#discussion_r1218101842
########## csharp/src/Apache.Arrow.Adbc/Core/Interop.cs: ########## @@ -0,0 +1,694 @@ +/* + * 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.Buffers; +using System.Collections.Generic; +using System.Diagnostics; +using System.Runtime.InteropServices; +using Apache.Arrow.Adbc.Interop; +using Apache.Arrow.C; +using static System.Net.Mime.MediaTypeNames; + +#if NETSTANDARD +using Apache.Arrow.Adbc.Extensions; +#endif + +namespace Apache.Arrow.Adbc.Core +{ + internal static class AdbcInterop + { + private unsafe static readonly NativeDelegate<ErrorRelease> releaseError = new NativeDelegate<ErrorRelease>(ReleaseError); + private unsafe static readonly NativeDelegate<DriverRelease> releaseDriver = new NativeDelegate<DriverRelease>(ReleaseDriver); + + private unsafe static readonly NativeDelegate<DatabaseFn> databaseInit = new NativeDelegate<DatabaseFn>(InitDatabase); + private unsafe static readonly NativeDelegate<DatabaseFn> databaseRelease = new NativeDelegate<DatabaseFn>(ReleaseDatabase); + private unsafe static readonly NativeDelegate<DatabaseSetOption> databaseSetOption = new NativeDelegate<DatabaseSetOption>(SetDatabaseOption); + + private unsafe static readonly NativeDelegate<ConnectionInit> connectionInit = new NativeDelegate<ConnectionInit>(InitConnection); + private unsafe static readonly NativeDelegate<ConnectionFn> connectionRelease = new NativeDelegate<ConnectionFn>(ReleaseConnection); + private unsafe static readonly NativeDelegate<ConnectionGetInfo> connectionGetInfo = new NativeDelegate<ConnectionGetInfo>(GetConnectionInfo); + private unsafe static readonly NativeDelegate<ConnectionSetOption> connectionSetOption = new NativeDelegate<ConnectionSetOption>(SetConnectionOption); + + private unsafe static readonly NativeDelegate<StatementExecuteQuery> statementExecuteQuery = new NativeDelegate<StatementExecuteQuery>(ExecuteStatementQuery); + private unsafe static readonly NativeDelegate<StatementNew> statementNew = new NativeDelegate<StatementNew>(NewStatement); + private unsafe static readonly NativeDelegate<StatementFn> statementRelease = new NativeDelegate<StatementFn>(ReleaseStatement); + private unsafe static readonly NativeDelegate<StatementSetSqlQuery> statementSetSqlQuery = new NativeDelegate<StatementSetSqlQuery>(SetStatementSqlQuery); + + public unsafe static AdbcStatusCode AdbcDriverInit(int version, NativeAdbcDriver* nativeDriver, NativeAdbcError* error, AdbcDriver driver) + { + DriverStub stub = new DriverStub(driver); + GCHandle handle = GCHandle.Alloc(stub); + nativeDriver->private_data = (void*)GCHandle.ToIntPtr(handle); + nativeDriver->release = (delegate* unmanaged[Stdcall]<NativeAdbcDriver*, NativeAdbcError*, AdbcStatusCode>)releaseDriver.Pointer; + + nativeDriver->DatabaseInit = (delegate* unmanaged[Stdcall]<NativeAdbcDatabase*, NativeAdbcError*, AdbcStatusCode>)databaseInit.Pointer; + nativeDriver->DatabaseNew = (delegate* unmanaged[Stdcall]<NativeAdbcDatabase*, NativeAdbcError*, AdbcStatusCode>)stub.newDatabase.Pointer; + nativeDriver->DatabaseSetOption = (delegate* unmanaged[Stdcall]<NativeAdbcDatabase*, byte*, byte*, NativeAdbcError*, AdbcStatusCode>) databaseSetOption.Pointer; + nativeDriver->DatabaseRelease = (delegate* unmanaged[Stdcall]<NativeAdbcDatabase*, NativeAdbcError*, AdbcStatusCode>)databaseRelease.Pointer; + + nativeDriver->ConnectionCommit = (delegate* unmanaged[Stdcall]<NativeAdbcConnection*, NativeAdbcError*, AdbcStatusCode>)connectionRelease.Pointer; + //nativeDriver->ConnectionGetInfo = (delegate* unmanaged[Stdcall]<NativeAdbcConnection *, int*, int, CArrowArrayStream*, NativeAdbcError*, AdbcStatusCode>)connectionGetInfo.Pointer; + //nativeDriver->ConnectionGetTableSchema = null; + //nativeDriver->ConnectionGetTableTypes = null; + nativeDriver->ConnectionInit = (delegate* unmanaged[Stdcall]<NativeAdbcConnection*, NativeAdbcDatabase*, NativeAdbcError*, AdbcStatusCode>)connectionInit.Pointer; + nativeDriver->ConnectionNew = (delegate* unmanaged[Stdcall]<NativeAdbcConnection*, NativeAdbcError*, AdbcStatusCode>)stub.newConnection.Pointer; + nativeDriver->ConnectionSetOption = (delegate* unmanaged[Stdcall]<NativeAdbcConnection*, byte*, byte*, NativeAdbcError*, AdbcStatusCode>)connectionSetOption.Pointer; + //nativeDriver->ConnectionReadPartition = null; + nativeDriver->ConnectionRelease = (delegate* unmanaged[Stdcall]<NativeAdbcConnection*, NativeAdbcError*, AdbcStatusCode>)connectionRelease.Pointer; + //nativeDriver->ConnectionRollback = null; + + // nativeDriver->StatementBind = (delegate* unmanaged[Stdcall]<NativeAdbcStatement*, CArrowArray*, CArrowSchema*, NativeAdbcError*, AdbcStatusCode>) + nativeDriver->StatementNew = (delegate* unmanaged[Stdcall]<NativeAdbcConnection*, NativeAdbcStatement*, NativeAdbcError*, AdbcStatusCode>)statementNew.Pointer; + nativeDriver->StatementSetSqlQuery = (delegate* unmanaged[Stdcall]<NativeAdbcStatement*, byte*, NativeAdbcError *, AdbcStatusCode >)statementSetSqlQuery.Pointer; + nativeDriver->StatementExecuteQuery = (delegate* unmanaged[Stdcall]<NativeAdbcStatement*, CArrowArrayStream*, long*, NativeAdbcError*, AdbcStatusCode>)statementExecuteQuery.Pointer; + nativeDriver->StatementPrepare = (delegate* unmanaged[Stdcall]<NativeAdbcStatement*, NativeAdbcError*, AdbcStatusCode>)statementRelease.Pointer; + nativeDriver->StatementRelease = (delegate* unmanaged[Stdcall]<NativeAdbcStatement*, NativeAdbcError*, AdbcStatusCode>)statementRelease.Pointer; + + return 0; + } + + private unsafe static void ReleaseError(NativeAdbcError* error) + { + if (error != null && ((IntPtr)error->message) != IntPtr.Zero) + { + Marshal.FreeCoTaskMem((IntPtr)error->message); + } + } + + private unsafe static AdbcStatusCode SetError(NativeAdbcError* error, Exception exception) + { + ReleaseError(error); + + #if NETSTANDARD + error->message = (char*)MarshalExtensions.StringToCoTaskMemUTF8(exception.Message); + #else + error->message = (char*)Marshal.StringToCoTaskMemUTF8(exception.Message); + #endif + + error->sqlstate0 = (char)0; + error->sqlstate1 = (char)0; + error->sqlstate2 = (char)0; + error->sqlstate3 = (char)0; + error->sqlstate4 = (char)0; + error->vendor_code = 0; + error->vendor_code = 0; + error->release = (delegate* unmanaged[Stdcall]<NativeAdbcError*, void>)releaseError.Pointer; + + return AdbcStatusCode.UnknownError; + } + + private sealed class PinnedArray : IDisposable + { + IArrowArray _array; + MemoryHandle[] pinnedHandles; + + public PinnedArray(IArrowArray array) + { + _array = array; + pinnedHandles = new MemoryHandle[GetHandleCount(array.Data)]; + int index = 0; + PinBuffers(array.Data, pinnedHandles, ref index); + Debug.Assert(index == pinnedHandles.Length); + } + + public void Dispose() + { + if (_array != null) + { + _array.Dispose(); + foreach (MemoryHandle handle in pinnedHandles) + { + handle.Dispose(); + } + _array = null; + } + } + + static int GetHandleCount(ArrayData data) + { + int handleCount = data.Buffers.Length; + foreach (ArrayData child in data.Children) + { + handleCount += GetHandleCount(child); + } + if (data.Dictionary != null) + { + handleCount += GetHandleCount(data.Dictionary); + } + return handleCount; + } + + static void PinBuffers(ArrayData data, MemoryHandle[] handles, ref int index) + { + foreach (ArrowBuffer buffer in data.Buffers) + { + handles[index++] = buffer.Memory.Pin(); + } + foreach (ArrayData child in data.Children) + { + PinBuffers(child, handles, ref index); + } + if (data.Dictionary != null) + { + PinBuffers(data.Dictionary, handles, ref index); + } + } + } + + private static IntPtr FromDisposable(IDisposable d) + { + GCHandle gch = GCHandle.Alloc(d); + return GCHandle.ToIntPtr(gch); + } + + private static void Dispose(ref IntPtr p) + { + GCHandle gch = GCHandle.FromIntPtr(p); + ((IDisposable)gch.Target).Dispose(); + gch.Free(); + p = IntPtr.Zero; + } + + private unsafe static AdbcStatusCode ReleaseDriver(NativeAdbcDriver* nativeDriver, NativeAdbcError* error) + { + GCHandle gch = GCHandle.FromIntPtr((IntPtr)nativeDriver->private_data); + DriverStub stub = (DriverStub)gch.Target; + stub.Dispose(); + gch.Free(); + nativeDriver->private_data = null; + return 0; + } + + private unsafe static AdbcStatusCode InitDatabase(NativeAdbcDatabase* nativeDatabase, NativeAdbcError* error) + { + GCHandle gch = GCHandle.FromIntPtr((IntPtr)nativeDatabase->private_data); + DatabaseStub stub = (DatabaseStub)gch.Target; + return stub.Init(ref *error); + } + + private unsafe static AdbcStatusCode ReleaseDatabase(NativeAdbcDatabase* nativeDatabase, NativeAdbcError* error) + { + if (nativeDatabase->private_data == null) + { + return AdbcStatusCode.UnknownError; + } + + GCHandle gch = GCHandle.FromIntPtr((IntPtr)nativeDatabase->private_data); + DatabaseStub stub = (DatabaseStub)gch.Target; + stub.Dispose(); + gch.Free(); + nativeDatabase->private_data = null; + return AdbcStatusCode.Success; + } + + private unsafe static AdbcStatusCode SetConnectionOption(NativeAdbcConnection* nativeConnection, byte* name, byte* value, NativeAdbcError* error) + { + GCHandle gch = GCHandle.FromIntPtr((IntPtr)nativeConnection->private_data); + ConnectionStub stub = (ConnectionStub)gch.Target; + return stub.SetOption(name, value, ref *error); + } + + private unsafe static AdbcStatusCode SetDatabaseOption(NativeAdbcDatabase* nativeDatabase, byte* name, byte* value, NativeAdbcError* error) + { + GCHandle gch = GCHandle.FromIntPtr((IntPtr)nativeDatabase->private_data); + DatabaseStub stub = (DatabaseStub)gch.Target; + + return stub.SetOption(name, value, ref *error); + } + + private unsafe static AdbcStatusCode InitConnection(NativeAdbcConnection* nativeConnection, NativeAdbcDatabase* database, NativeAdbcError* error) + { + GCHandle gch = GCHandle.FromIntPtr((IntPtr)nativeConnection->private_data); + ConnectionStub stub = (ConnectionStub)gch.Target; + return stub.InitConnection(ref *database, ref *error); + } + + private unsafe static AdbcStatusCode ReleaseConnection(NativeAdbcConnection* nativeConnection, NativeAdbcError* error) + { + if (nativeConnection->private_data == null) + { + return AdbcStatusCode.UnknownError; + } + + GCHandle gch = GCHandle.FromIntPtr((IntPtr)nativeConnection->private_data); + ConnectionStub stub = (ConnectionStub)gch.Target; + stub.Dispose(); + gch.Free(); + nativeConnection->private_data = null; + return AdbcStatusCode.Success; + } + + private unsafe static AdbcStatusCode GetConnectionInfo(NativeAdbcConnection* nativeConnection, uint* info_codes, int info_codes_length, CArrowArrayStream* stream, NativeAdbcError* error) + { + GCHandle gch = GCHandle.FromIntPtr((IntPtr)nativeConnection->private_data); + ConnectionStub stub = (ConnectionStub)gch.Target; + return stub.GetConnectionInfo(ref *nativeConnection, *info_codes, info_codes_length, ref *stream, ref *error); + } + + private unsafe static AdbcStatusCode SetStatementSqlQuery(NativeAdbcStatement* nativeStatement, byte* text, NativeAdbcError* error) + { + GCHandle gch = GCHandle.FromIntPtr((IntPtr)nativeStatement->private_data); + AdbcStatement stub = (AdbcStatement)gch.Target; + + #if NETSTANDARD + stub.SqlQuery = MarshalExtensions.PtrToStringUTF8((IntPtr)text); + #else + stub.SqlQuery = Marshal.PtrToStringUTF8((IntPtr)text); + #endif + + return AdbcStatusCode.Success; + } + + private unsafe static AdbcStatusCode ExecuteStatementQuery(NativeAdbcStatement* nativeStatement, CArrowArrayStream* stream, long* rows, NativeAdbcError* error) + { + GCHandle gch = GCHandle.FromIntPtr((IntPtr)nativeStatement->private_data); + AdbcStatement stub = (AdbcStatement)gch.Target; + var result = stub.ExecuteQuery(); + if (rows != null) + { + *rows = result.RowCount; + } + + GCHandle streamHandle = GCHandle.Alloc(result.Stream); + stream->private_data = (void*)GCHandle.ToIntPtr(streamHandle); + + return 0; + } + + private unsafe static AdbcStatusCode NewStatement(NativeAdbcConnection* nativeConnection, NativeAdbcStatement* nativeStatement, NativeAdbcError* error) + { + GCHandle gch = GCHandle.FromIntPtr((IntPtr)nativeConnection->private_data); + ConnectionStub stub = (ConnectionStub)gch.Target; + return stub.NewStatement(ref *nativeStatement, ref *error); + } + + private unsafe static AdbcStatusCode ReleaseStatement(NativeAdbcStatement* nativeStatement, NativeAdbcError* error) + { + if (nativeStatement->private_data == null) + { + return AdbcStatusCode.UnknownError; + } + + GCHandle gch = GCHandle.FromIntPtr((IntPtr)nativeStatement->private_data); + AdbcStatement stub = (AdbcStatement)gch.Target; + stub.Dispose(); + gch.Free(); + nativeStatement->private_data = null; + return AdbcStatusCode.Success; + } + } + + [StructLayout(LayoutKind.Sequential)] + public unsafe struct NativeAdbcDatabase + { + public void* private_data; + public NativeAdbcDriver* private_driver; + + public static NativeAdbcDatabase* Create() + { + var ptr = (NativeAdbcDatabase*)Marshal.AllocHGlobal(sizeof(NativeAdbcDatabase)); + + ptr->private_data = null; + ptr->private_driver = null; + + return ptr; + } + + /// <summary> + /// Free a pointer that was allocated in <see cref="Create"/>. + /// </summary> + /// <remarks> + /// Do not call this on a pointer that was allocated elsewhere. + /// </remarks> + public static void Free(NativeAdbcDatabase* database) + { + Marshal.FreeHGlobal((IntPtr)database); + } + } + + [StructLayout(LayoutKind.Sequential)] + public unsafe struct NativeAdbcConnection + { + public void* private_data; + public NativeAdbcDriver* private_driver; + } + + [StructLayout(LayoutKind.Sequential)] + public unsafe struct NativeAdbcStatement + { + public void* private_data; + public NativeAdbcDriver* private_driver; + } + + [StructLayout(LayoutKind.Sequential)] + public unsafe struct NativeAdbcPartitions + { + /// <summary> + /// The number of partitions. + /// </summary> + public int num_partitions; + + /// <summary> + /// The partitions of the result set, where each entry (up to + /// num_partitions entries) is an opaque identifier that can be + /// passed to AdbcConnectionReadPartition. + /// </summary> + public sbyte** partitions; + + /// <summary> + /// The length of each corresponding entry in partitions. + /// </summary> + public int* partition_lengths; + + /// <summary> + /// Opaque implementation-defined state. + /// This field is NULLPTR iff the connection is unintialized/freed. + /// </summary> + public void* private_data; + + /// <summary> + /// Release the contained partitions. + /// + /// Unlike other structures, this is an embedded callback to make it + /// easier for the driver manager and driver to cooperate. + /// </summary> + public delegate* unmanaged[Stdcall]<NativeAdbcPartitions*, void> release; + } + + [StructLayout(LayoutKind.Sequential)] + public unsafe struct NativeAdbcError + { + /// <summary> + /// The error message. + /// </summary> + public char* message; + + /// <summary> + /// A vendor-specific error code, if applicable. + /// </summary> + public int vendor_code; + + /// <summary> + /// A SQLSTATE error code, if provided, as defined by the + /// SQL:2003 standard. If not set, it should be set to + /// "\0\0\0\0\0". + ///</summary> + public char sqlstate0; + public char sqlstate1; + public char sqlstate2; + public char sqlstate3; + public char sqlstate4; Review Comment: verified -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
