Land $name params, ORDER BY/LIMIT subqueries, LSM shutdown durability (native layer deferred)

.github/workflows/ci.yml

@@ -42,14 +42,17 @@ jobs:
           restore-keys: |
             nuget-${{ runner.os }}-
 
+      - name: Verify vendored ANTLR tool JAR
+        run: test -f build/antlr/antlr4-4.13.1-complete.jar
+
       - name: Restore
         run: dotnet restore OutWit.slnx
 
       - name: Build
         run: dotnet build OutWit.slnx --configuration Release --no-restore
 
       - name: Test
-        run: dotnet test OutWit.slnx --configuration Release --no-build --verbosity normal --logger "trx;LogFileName=test-results.trx" --results-directory TestResults
+        run: dotnet test OutWit.slnx --configuration Release --no-build --verbosity normal --filter "Category!=Performance" --logger "trx;LogFileName=test-results.trx" --results-directory TestResults
 
       - name: Upload Test Results
         if: always()

Docs/WitSQL.md

@@ -918,6 +918,7 @@ WitSQL supports named and positional parameters:
 -- Named parameters
 SELECT * FROM Users WHERE Id = @UserId;
 SELECT * FROM Users WHERE Name = :name;
+SELECT * FROM Users WHERE MigrationId = $id;
 
 -- Positional parameters  
 SELECT * FROM Users WHERE Id = ?;

Sources/Core/OutWit.Database.Core.Tests/LSM/LsmParallelWriterTests.cs

@@ -348,6 +348,117 @@ public void DisposedWriterThrowsTest()
 
     #endregion
 
+    #region Shutdown Durability Tests
+
+    // These lock in the drain-before-cancel contract: every buffer submitted to the
+    // merge channel must be durably written through to the store on shutdown, and
+    // awaited writes must complete successfully rather than be faulted/dropped by a
+    // premature cancellation. A large maxPendingBuffers guarantees every submit is
+    // accepted (never silently rejected by a full bounded channel), so the assertions
+    // are deterministic: anything missing means the shutdown path lost queued work.
+
+    [Test]
+    public void DisposeDrainsAllQueuedBuffersTest()
+    {
+        var dir = Path.Combine(m_testDir, "dispose_drain_sync");
+        var options = new LsmOptions { EnableWal = false, MemTableSizeLimit = 1024 * 1024 };
+
+        using var store = new StoreLsm(dir, options);
+
+        const int count = 200;
+        var writer = new LsmParallelWriter(store, maxPendingBuffers: count * 2);
+
+        // Queue one buffer per key; do NOT wait for the background merge to catch up.
+        for (int i = 0; i < count; i++)
+        {
+            writer.Put(ToBytes($"k{i:D5}"), ToBytes($"v{i:D5}"));
+            writer.FlushCurrentBuffer();
+        }
+
+        // Dispose must drain the whole queue before returning.
+        writer.Dispose();
+
+        for (int i = 0; i < count; i++)
+            Assert.That(store.Get(ToBytes($"k{i:D5}")), Is.EqualTo(ToBytes($"v{i:D5}")), $"Missing k{i:D5} after Dispose");
+    }
+
+    [Test]
+    public async Task DisposeAsyncDrainsAllQueuedBuffersTest()
+    {
+        var dir = Path.Combine(m_testDir, "dispose_drain_async");
+        var options = new LsmOptions { EnableWal = false, MemTableSizeLimit = 1024 * 1024 };
+
+        using var store = new StoreLsm(dir, options);
+
+        const int count = 200;
+        var writer = new LsmParallelWriter(store, maxPendingBuffers: count * 2);
+
+        for (int i = 0; i < count; i++)
+        {
+            writer.Put(ToBytes($"k{i:D5}"), ToBytes($"v{i:D5}"));
+            writer.FlushCurrentBuffer();
+        }
+
+        await writer.DisposeAsync();
+
+        for (int i = 0; i < count; i++)
+            Assert.That(store.Get(ToBytes($"k{i:D5}")), Is.EqualTo(ToBytes($"v{i:D5}")), $"Missing k{i:D5} after DisposeAsync");
+    }
+
+    [Test]
+    public async Task DisposeCompletesAwaitedWritesSuccessfullyTest()
+    {
+        var dir = Path.Combine(m_testDir, "dispose_awaited");
+        var options = new LsmOptions { EnableWal = false, MemTableSizeLimit = 1024 * 1024 };
+
+        using var store = new StoreLsm(dir, options);
+
+        const int count = 50;
+        var writer = new LsmParallelWriter(store, maxPendingBuffers: count * 2);
+
+        // Issue awaited writes but capture the tasks WITHOUT awaiting them yet, so they
+        // are still in flight when shutdown begins. WriteAsync completes synchronously
+        // on the un-full channel, so all buffers + completions are enqueued before the
+        // state machines yield at 'await completion.Task'.
+        var flushTasks = new List<Task>();
+        for (int i = 0; i < count; i++)
+        {
+            writer.Put(ToBytes($"k{i:D5}"), ToBytes($"v{i:D5}"));
+            flushTasks.Add(writer.FlushCurrentBufferAsync());
+        }
+
+        await writer.DisposeAsync();
+
+        // Every awaited write must resolve successfully - not faulted or cancelled.
+        await Task.WhenAll(flushTasks);
+        Assert.That(flushTasks.All(t => t.IsCompletedSuccessfully), Is.True, "An awaited write did not complete successfully across shutdown");
+
+        for (int i = 0; i < count; i++)
+            Assert.That(store.Get(ToBytes($"k{i:D5}")), Is.EqualTo(ToBytes($"v{i:D5}")), $"Missing k{i:D5} after awaited write + Dispose");
+    }
+
+    [Test]
+    public void DisposeCompletesPromptlyWithoutHangingTest()
+    {
+        var dir = Path.Combine(m_testDir, "dispose_no_hang");
+        var options = new LsmOptions { EnableWal = false, MemTableSizeLimit = 1024 * 1024 };
+
+        using var store = new StoreLsm(dir, options);
+
+        var writer = new LsmParallelWriter(store, maxPendingBuffers: 500);
+        for (int i = 0; i < 100; i++)
+        {
+            writer.Put(ToBytes($"k{i:D5}"), ToBytes($"v{i:D5}"));
+            writer.FlushCurrentBuffer();
+        }
+
+        // The drain join is bounded; Dispose must return well within it.
+        var disposeTask = Task.Run(() => writer.Dispose());
+        Assert.That(disposeTask.Wait(TimeSpan.FromSeconds(10)), Is.True, "Dispose did not complete within the bounded drain window");
+    }
+
+    #endregion
+
     #region Integration Tests
 
     [Test]

Sources/Core/OutWit.Database.Core/Indexes/IndexMetadataJsonContext.cs

@@ -0,0 +1,7 @@
+using System.Text.Json.Serialization;
+
+namespace OutWit.Database.Core.Indexes;
+
+[JsonSerializable(typeof(IndexMetadata))]
+[JsonSerializable(typeof(List<string>))]
+internal sealed partial class IndexMetadataJsonContext : JsonSerializerContext;

Sources/Core/OutWit.Database.Core/Indexes/IndexMetadataStore.cs

@@ -56,7 +56,7 @@ public void SaveIndex(string name, bool isUnique)
 
             var metadata = new IndexMetadata { Name = name, IsUnique = isUnique };
             var key = CreateKey(name);
-            var value = JsonSerializer.SerializeToUtf8Bytes(metadata);
+            var value = JsonSerializer.SerializeToUtf8Bytes(metadata, IndexMetadataJsonContext.Default.IndexMetadata);
 
             m_store.Put(key, value);
 
@@ -84,7 +84,7 @@ public void SaveIndex(string name, bool isUnique)
             if (value == null)
                 return null;
 
-            return JsonSerializer.Deserialize<IndexMetadata>(value);
+            return JsonSerializer.Deserialize(value, IndexMetadataJsonContext.Default.IndexMetadata);
         }
 
         /// <summary>
@@ -150,7 +150,7 @@ public async ValueTask SaveIndexAsync(string name, bool isUnique, CancellationTo
 
             var metadata = new IndexMetadata { Name = name, IsUnique = isUnique };
             var key = CreateKey(name);
-            var value = JsonSerializer.SerializeToUtf8Bytes(metadata);
+            var value = JsonSerializer.SerializeToUtf8Bytes(metadata, IndexMetadataJsonContext.Default.IndexMetadata);
 
             await m_store.PutAsync(key, value, cancellationToken).ConfigureAwait(false);
 
@@ -177,7 +177,7 @@ public async ValueTask SaveIndexAsync(string name, bool isUnique, CancellationTo
             if (value == null)
                 return null;
 
-            return JsonSerializer.Deserialize<IndexMetadata>(value);
+            return JsonSerializer.Deserialize(value, IndexMetadataJsonContext.Default.IndexMetadata);
         }
 
         /// <summary>
@@ -244,7 +244,7 @@ private List<string> LoadCatalog()
 
             try
             {
-                return JsonSerializer.Deserialize<List<string>>(value) ?? [];
+                return JsonSerializer.Deserialize(value, IndexMetadataJsonContext.Default.ListString) ?? [];
             }
             catch
             {
@@ -254,7 +254,7 @@ private List<string> LoadCatalog()
 
         private void SaveCatalog(List<string> catalog)
         {
-            var value = JsonSerializer.SerializeToUtf8Bytes(catalog);
+            var value = JsonSerializer.SerializeToUtf8Bytes(catalog, IndexMetadataJsonContext.Default.ListString);
             m_store.Put(CATALOG_KEY, value);
         }
 
@@ -271,7 +271,7 @@ private async ValueTask<List<string>> LoadCatalogAsync(CancellationToken cancell
 
             try
             {
-                return JsonSerializer.Deserialize<List<string>>(value) ?? [];
+                return JsonSerializer.Deserialize(value, IndexMetadataJsonContext.Default.ListString) ?? [];
             }
             catch
             {
@@ -281,7 +281,7 @@ private async ValueTask<List<string>> LoadCatalogAsync(CancellationToken cancell
 
         private async ValueTask SaveCatalogAsync(List<string> catalog, CancellationToken cancellationToken = default)
         {
-            var value = JsonSerializer.SerializeToUtf8Bytes(catalog);
+            var value = JsonSerializer.SerializeToUtf8Bytes(catalog, IndexMetadataJsonContext.Default.ListString);
             await m_store.PutAsync(CATALOG_KEY, value, cancellationToken).ConfigureAwait(false);
         }
 

Sources/Core/OutWit.Database.Core/LSM/LsmMemTableFlusher.cs

@@ -270,10 +270,9 @@ public void Dispose()
         m_disposed = true;
 
         m_flushChannel.Writer.Complete();
-        m_cts.Cancel();
-
         Task.WaitAll(m_flushTasks, TimeSpan.FromSeconds(10));
 
+        m_cts.Cancel();
         m_cts.Dispose();
     }
 
@@ -287,10 +286,9 @@ public async ValueTask DisposeAsync()
         m_disposed = true;
 
         m_flushChannel.Writer.Complete();
-        await m_cts.CancelAsync();
-
         await Task.WhenAll(m_flushTasks).WaitAsync(TimeSpan.FromSeconds(10));
 
+        await m_cts.CancelAsync();
         m_cts.Dispose();
     }
 

Sources/Core/OutWit.Database.Core/LSM/LsmParallelCompactor.cs

@@ -272,10 +272,9 @@ public void Dispose()
         m_disposed = true;
 
         m_jobChannel.Writer.Complete();
-        m_cts.Cancel();
-
         Task.WaitAll(m_workerTasks, TimeSpan.FromSeconds(30));
 
+        m_cts.Cancel();
         m_cts.Dispose();
     }
 
@@ -289,10 +288,9 @@ public async ValueTask DisposeAsync()
         m_disposed = true;
 
         m_jobChannel.Writer.Complete();
-        await m_cts.CancelAsync();
-
         await Task.WhenAll(m_workerTasks).WaitAsync(TimeSpan.FromSeconds(30));
 
+        await m_cts.CancelAsync();
         m_cts.Dispose();
     }