Complete V3 native async operation migration (#16)

* Add native async partition planning

* Generalize native async JSON operations

* Harden native async operation lifecycle

* Add async native read stream setup

* Add async native query stream setup

* Add async native CDF stream setup

* Add async native partition stream setup

* Add async native insert setup

* Add async native merge setup

* Add async native schema setup

* Clean up managed native sync bindings

---------

Co-authored-by: STARGAZER <stargazer@XingdeMacBook-Air.local>

Author: Cappu7ino

docs/ai/limitations.md

@@ -39,7 +39,7 @@ This file is intentionally direct. Use it to prevent incorrect SDK integrations
 
 - The public APIs are asynchronous, but backend execution does not imply unlimited parallelism.
 - Flight clients own a gRPC channel per `DeltaTableServiceClient` instance.
-- V3 native calls cross a synchronous FFI boundary and use a native engine handle.
+- V3 native uses a native engine handle; main operations use callback-notified async operation handles, while batch pulls and imported write sources still follow Arrow C Stream ownership rules.
 - Avoid issuing multiple concurrent native operations through the same client unless the caller has validated backend behavior for that scenario.
 - Partitioned reads are the preferred pattern for independent parallel reads on V3.
 

docs/ai/overview.md

@@ -58,7 +58,7 @@ Most SDK operations are asynchronous and stream Arrow `RecordBatch` values. Help
 - Treat `IAsyncEnumerable<RecordBatch>` and `IArrowArrayStream` results as streaming resources.
 - Avoid buffering full tables unless data volume is known to be small.
 - Use per-request storage options for SAS/OneLake authentication instead of global mutable state.
-- For V3 native operations, assume a synchronous FFI boundary inside the backend call path even when the public API is asynchronous.
+- For V3 native operations, schema reads, partition planning, JSON-result table operations, table/query/CDF/partition stream setup, and insert/merge setup use callback-notified native async operation handles. Batch consumption and write source consumption still use Arrow C Stream ownership rules.
 
 ## Major Tradeoffs
 

docs/architecture/cdf-and-partitioned-reads.md

@@ -30,6 +30,11 @@ Constraints:
 - The table must have CDF enabled.
 - V1/V2 Flight wrappers throw for CDF APIs.
 
+Runtime behavior:
+
+- V3 CDF stream setup uses the native async operation handle path, matching table and SQL stream setup.
+- Batch consumption remains Arrow C Stream based; consumers should dispose stream and reader resources promptly.
+
 ## CDF Result Shape
 
 CDF rows include regular table columns plus metadata such as:
@@ -64,6 +69,12 @@ Descriptor fields:
 - total partition count
 - file count
 
+Runtime behavior:
+
+- V3 partition planning returns JSON through the native async operation handle path.
+- V3 partition stream setup also uses native async operation handles before returning an Arrow C Stream.
+- Partition stream batch consumption remains Arrow C Stream based.
+
 ## Partition Token Rules
 
 - Tokens are opaque.

docs/architecture/execution-model.md

@@ -66,11 +66,14 @@ Properties:
 - Uses Arrow C Data/C Stream interfaces for schemas and batches.
 - Uses direct Arrow C Stream pulls by default, with bounded Rust-side prefetch available through `DeltaTableServiceClientOptions.EnableNativeReadPrefetch`.
 - Owns a native engine handle through a `SafeHandle` wrapper.
+- Uses a native async operation `SafeHandle` for V3 schema reads, partition planning, JSON-result table operations, read stream setup, and insert/merge setup.
 - Shares one process-wide Tokio runtime across native engine handles.
 - Avoids the Flight service boundary but requires native runtime assets.
 
 Each `NativeRustBackend` still owns its native engine handle and per-engine error state. The shared runtime reduces thread and stack reservation overhead when multiple V3 clients are created in the same process. Native merge work is scheduled onto Tokio worker threads so deep delta-rs/DataFusion merge execution does not run on the .NET caller stack.
 
+Schema reads, `GetReadPartitionsAsync`, table creation, protocol upgrade, SQL DML operations, table/query/CDF/partition stream setup, and insert/merge setup start callback-notified native operations on the shared Tokio runtime. The managed backend awaits a `TaskCompletionSource`, takes the schema, JSON result, or Arrow stream once after native completion is signaled, and cancels the native operation if the managed cancellation token is signaled. The public API and result models are unchanged.
+
 When read-stream prefetch is enabled, production is bounded in two ways: each exported stream has a small native queue, and active backend batch production is capped process-wide. These limits provide backpressure for high-concurrency readers without changing the public `IAsyncEnumerable<RecordBatch>` or `IArrowArrayStream` shapes. The default read path remains direct batch pulling because local benchmarks showed prefetch overhead can dominate small/local reads.
 
 ## Streaming First

docs/architecture/native-interop.md

@@ -10,6 +10,7 @@ Managed V3 execution flows through:
 
 - [../../src/DeltaLakeSharp.Client/Internal/NativeRustBackend.cs](../../src/DeltaLakeSharp.Client/Internal/NativeRustBackend.cs)
 - [../../src/DeltaLakeSharp.Client/Internal/Native/NativeEngineHandle.cs](../../src/DeltaLakeSharp.Client/Internal/Native/NativeEngineHandle.cs)
+- [../../src/DeltaLakeSharp.Client/Internal/Native/NativeAsyncOperationHandle.cs](../../src/DeltaLakeSharp.Client/Internal/Native/NativeAsyncOperationHandle.cs)
 - [../../src/DeltaLakeSharp.Client/Internal/Native/NativeMethods.net8.cs](../../src/DeltaLakeSharp.Client/Internal/Native/NativeMethods.net8.cs)
 - [../../src/DeltaLakeSharp.Client/Internal/Native/NativeMethods.net472.cs](../../src/DeltaLakeSharp.Client/Internal/Native/NativeMethods.net472.cs)
 
@@ -43,9 +44,10 @@ Native merge work runs on Tokio worker threads instead of polling the whole merg
 | Data | Representation | Ownership Rule |
 | --- | --- | --- |
 | command metadata | JSON string | Managed code builds command payload; Rust parses it. |
-| schema | Arrow C Data schema | Managed code imports schema and frees temporary native structures. |
+| schema | Arrow C Data schema | Managed code imports schema and frees temporary native structures; async schema reads take the schema result exactly once. |
 | read batches | Arrow C Stream | Imported managed stream owns the release callback; Rust can use bounded prefetch behind the stream when enabled. |
-| write batches | Arrow C Stream | Managed stream is exported to Rust for operation duration. |
+| write batches | Arrow C Stream | Managed stream is exported to Rust for operation duration; async insert and merge keep the exported stream and native storage alive until completion notification. |
+| one-shot async operation | native operation pointer | Managed code awaits a `TaskCompletionSource`, takes the owned result string or Arrow stream once after native completion notification, and destroys the operation handle. |
 | string results | native string pointer | Managed code frees returned native strings. |
 
 ## Native Library Discovery
@@ -69,7 +71,13 @@ Common failure modes:
 
 ## Concurrency Expectations
 
-The public API is asynchronous, but V3 crosses a synchronous FFI boundary for native calls. Do not assume unlimited parallelism through a single client instance. For parallel reads, prefer V3 partition planning and independent partition consumption.
+The public API is asynchronous, and the main V3 native operations use callback-notified native operation handles. Do not assume unlimited parallelism through a single client instance. For parallel reads, prefer V3 partition planning and independent partition consumption.
+
+Schema reads, partition planning, table creation, protocol upgrade, SQL DML operations, table/query/CDF/partition stream setup, and insert/merge setup use native async operation handles with completion notification. Managed code starts the relevant `*_async_with_callback` export, awaits a `TaskCompletionSource`, takes the result string with `dts_async_operation_take_result`, the schema with `dts_async_operation_take_schema`, or the stream with `dts_async_operation_take_stream` after the native callback fires, and releases the handle through `dts_async_operation_destroy`. Cancellation requests call `dts_async_operation_cancel` before managed code surfaces `OperationCanceledException`. This keeps the public API shapes unchanged while moving those one-shot operations onto the shared Tokio runtime instead of blocking the managed caller thread for the whole native operation.
+
+For async insert and merge, Rust imports the caller-provided Arrow C Stream before spawning the write task. Managed code keeps both the exported `IArrowArrayStream` adapter and the `CArrowArrayStream` storage alive until native completion is signaled, then disposes and frees them. Cancellation waits for the aborted native task to drop the imported reader before notifying managed code, which prevents the native writer from reading through released managed stream state while still avoiding a blocking write FFI call.
+
+Synchronous Rust C ABI exports are retained for native ABI compatibility, direct Rust unit coverage, and diagnostics. Managed SDK production paths should prefer the callback exports for operations with meaningful native work.
 
 By default, V3 read streams pull each batch through the Arrow C Stream callback and synchronously bridge to the async DataFusion stream. `DeltaTableServiceClientOptions.EnableNativeReadPrefetch` enables an experimental prefetch mode that places a small Rust-owned bounded queue behind the exported Arrow C Stream. In that mode, a Tokio producer task advances the DataFusion stream and sends ready batch results into the queue, while the Arrow C Stream pull side drains queued batches. The queue is bounded per stream, and native read production is guarded by a process-wide active-production limit so full per-stream queues do not monopolize global read capacity.
 

src/DeltaLakeSharp.Client/Internal/Native/NativeAsyncOperationHandle.cs

@@ -0,0 +1,36 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System;
+using System.Runtime.InteropServices;
+
+namespace DeltaLakeSharp.Client.Internal.Native
+{
+    internal sealed class NativeAsyncOperationHandle : SafeHandle
+    {
+        public NativeAsyncOperationHandle()
+            : base(IntPtr.Zero, ownsHandle: true)
+        {
+        }
+
+        public override bool IsInvalid => handle == IntPtr.Zero;
+
+        public static NativeAsyncOperationHandle FromIntPtr(IntPtr ptr)
+        {
+            if (ptr == IntPtr.Zero)
+            {
+                throw new InvalidOperationException("Failed to create native async operation.");
+            }
+
+            var handle = new NativeAsyncOperationHandle();
+            handle.SetHandle(ptr);
+            return handle;
+        }
+
+        protected override bool ReleaseHandle()
+        {
+            NativeMethods.AsyncOperationDestroy(handle);
+            return true;
+        }
+    }
+}

src/DeltaLakeSharp.Client/Internal/Native/NativeMethods.cs

@@ -12,6 +12,9 @@ internal static partial class NativeMethods
         internal const string LibraryName = "delta_table_service_native";
         private static IntPtr _loadedHandle;
 
+        [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
+        internal delegate void NativeAsyncOperationCompletedCallback(IntPtr operation, IntPtr userData);
+
         private static string[] GetCandidateLibraryPaths()
         {
             string fileName = GetPlatformLibraryFileName();