release: 2.5.3 — fix composite UNIQUE indexes

Executor's index-write / unique-check / index-delete paths all read
idx.Columns[0] only. Multi-column UNIQUE indexes silently degraded
to "unique on first column", falsely rejecting any second row with
the same first-column value regardless of the rest of the tuple.

Found via a Postgres → CosmoKvD dry-run on a 200k-row mail database:
UNIQUE (mailbox_id, uid) on the messages table rejected the second
row with mailbox_id=1 even though Postgres had a million distinct
(mailbox_id, uid) tuples with no duplicates.

Fix encodes ALL indexed columns as a tuple (each value carries its
own FoundationDB-style 00 00 terminator so the sequence is byte-
comparable in the same order as SQL row comparison). NULL handling
follows SQL semantics: any NULL anywhere in the tuple disables the
uniqueness check, since NULL != NULL even for compound keys. Single-
column WHERE prefix scans against composite indexes keep working
because the new encoding starts with enc(first_column).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: vkuttyp

Directory.Build.props

@@ -7,6 +7,6 @@
     <Authors>vkuttyp</Authors>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
     <RepositoryUrl>https://github.com/vkuttyp/CosmoSQLClient-Dotnet</RepositoryUrl>
-    <Version>2.5.2</Version>
+    <Version>2.5.3</Version>
   </PropertyGroup>
 </Project>

src/CosmoSQLClient.CosmoKv/Execution/Executor.cs

@@ -172,17 +172,33 @@ private async Task<int> ExecuteCreateIndex(CreateIndexStatement s, CancellationT
         await _catalog.CreateIndexAsync(index, ct);
 
         // Backfill existing rows so the new index covers data already present.
-        // Single-column-only in Phase 4; the multi-col schema field is reserved.
-        int colIdx = table.IndexOf(index.Columns[0]);
+        var colIdxs = ResolveIndexColumns(table, index);
         await foreach (var (row, rowId) in ScanRowsWithIdAsync(table, ct))
         {
-            var enc = IndexKeyCodec.Encode(row[colIdx]);
+            var enc = IndexKeyCodec.Encode(ProjectIndexedValues(row, colIdxs));
             var key = KvKeys.IndexEntry(table.Name, index.Name, enc, rowId);
             await _kv.SetAsync(key, Array.Empty<byte>(), ct);
         }
         return 0;
     }
 
+    /// <summary>Map each column name in <paramref name="idx"/> to its position in the table.</summary>
+    private static int[] ResolveIndexColumns(TableSchema table, IndexSchema idx)
+    {
+        var result = new int[idx.Columns.Count];
+        for (int i = 0; i < idx.Columns.Count; i++)
+            result[i] = table.IndexOf(idx.Columns[i]);
+        return result;
+    }
+
+    /// <summary>Project a row's indexed columns into a SqlValue tuple.</summary>
+    private static SqlValue[] ProjectIndexedValues(SqlValue[] row, int[] colIdxs)
+    {
+        var vals = new SqlValue[colIdxs.Length];
+        for (int i = 0; i < colIdxs.Length; i++) vals[i] = row[colIdxs[i]];
+        return vals;
+    }
+
     // ── DROP ────────────────────────────────────────────────────────────────
 
     private async Task<int> ExecuteDropTable(DropTableStatement s, CancellationToken ct)
@@ -399,31 +415,36 @@ private async Task WriteIndexEntriesAsync(
     {
         foreach (var idx in _catalog.IndexesFor(table.Name))
         {
-            int colIdx = table.IndexOf(idx.Columns[0]);
-            var val = row[colIdx];
+            var colIdxs = ResolveIndexColumns(table, idx);
+            var vals = ProjectIndexedValues(row, colIdxs);
             // UNIQUE check before writing the new entry. NULL is treated
             // per SQL semantics: NULL <> NULL, so multiple NULL rows in a
-            // UNIQUE column are allowed.
-            if (idx.IsUnique && !val.IsNull)
+            // UNIQUE column are allowed. For composite indexes, the row is
+            // considered unique-violating only if EVERY indexed column has
+            // a non-NULL value (any NULL anywhere preserves SQL's NULL-is-
+            // not-comparable semantics).
+            if (idx.IsUnique && !vals.Any(v => v.IsNull))
             {
-                await EnforceUniqueAsync(table, idx, val, excludeRowId: rowId, ct);
+                await EnforceUniqueAsync(table, idx, vals, excludeRowId: rowId, ct);
             }
-            var enc = IndexKeyCodec.Encode(val);
+            var enc = IndexKeyCodec.Encode(vals);
             var key = KvKeys.IndexEntry(table.Name, idx.Name, enc, rowId);
             await _kv.SetAsync(key, Array.Empty<byte>(), ct);
         }
     }
 
     /// <summary>
     /// Scan the unique index for any existing entry with the same encoded
-    /// value as <paramref name="val"/>, excluding the row that owns the
+    /// tuple as <paramref name="vals"/>, excluding the row that owns the
     /// pending write (so UPDATE on the same key isn't a self-conflict).
     /// Throws <see cref="UniqueConstraintViolationException"/> on match.
+    /// Tuple-aware: composite UNIQUE indexes match all columns, not just
+    /// the first.
     /// </summary>
     private async Task EnforceUniqueAsync(
-        TableSchema table, IndexSchema idx, SqlValue val, long excludeRowId, CancellationToken ct)
+        TableSchema table, IndexSchema idx, SqlValue[] vals, long excludeRowId, CancellationToken ct)
     {
-        var encWithTerm = IndexKeyCodec.Encode(val);
+        var encWithTerm = IndexKeyCodec.Encode(vals);
         var prefix = KvKeys.IndexPrefixBytes(idx.TableName, idx.Name);
         var fullPrefix = new byte[prefix.Length + encWithTerm.Length];
         Buffer.BlockCopy(prefix,      0, fullPrefix, 0,             prefix.Length);
@@ -434,8 +455,11 @@ private async Task EnforceUniqueAsync(
         {
             long otherRowId = KvKeys.ExtractRowId(item.Key);
             if (otherRowId == excludeRowId) continue;
+            var formatted = vals.Length == 1
+                ? vals[0].ToString() ?? "NULL"
+                : "(" + string.Join(", ", vals.Select(v => v.ToString() ?? "NULL")) + ")";
             throw new UniqueConstraintViolationException(
-                $"Duplicate value for UNIQUE column on index '{idx.Name}': {val}.");
+                $"Duplicate value for UNIQUE column on index '{idx.Name}': {formatted}.");
         }
     }
 
@@ -444,8 +468,8 @@ private async Task DeleteIndexEntriesAsync(
     {
         foreach (var idx in _catalog.IndexesFor(table.Name))
         {
-            int colIdx = table.IndexOf(idx.Columns[0]);
-            var enc = IndexKeyCodec.Encode(row[colIdx]);
+            var colIdxs = ResolveIndexColumns(table, idx);
+            var enc = IndexKeyCodec.Encode(ProjectIndexedValues(row, colIdxs));
             var key = KvKeys.IndexEntry(table.Name, idx.Name, enc, rowId);
             await _kv.DeleteAsync(key, ct);
         }

src/CosmoSQLClient.CosmoKv/Storage/IndexKeyCodec.cs

@@ -51,6 +51,25 @@ public static byte[] Encode(SqlValue v)
         return ms.ToArray();
     }
 
+    /// <summary>
+    /// Encode a tuple of values, in order, with terminators between them.
+    /// Used for composite indexes — `(a, b, c)` encodes as `enc(a) enc(b)
+    /// enc(c)`, where each `enc()` already includes its own terminator, so
+    /// the whole tuple stays byte-comparable in the same lexicographic
+    /// order as SQL row comparison `(a, b, c) < (a', b', c')`.
+    /// </summary>
+    public static byte[] Encode(IReadOnlyList<SqlValue> vals)
+    {
+        if (vals.Count == 1) return Encode(vals[0]);
+        var ms = new MemoryStream(capacity: 16 * vals.Count);
+        foreach (var v in vals)
+        {
+            byte[] part = Encode(v);
+            ms.Write(part, 0, part.Length);
+        }
+        return ms.ToArray();
+    }
+
     /// <summary>
     /// Encode without the terminator suffix — used for range-scan seek
     /// targets where the boundary should be exclusive of the value's

tests/CosmoSQLClient.CosmoKv.Tests/Phase14CompositeUniqueTests.cs

@@ -0,0 +1,113 @@
+using CosmoSQLClient.Core;
+using CosmoSQLClient.CosmoKv;
+
+namespace CosmoSQLClient.CosmoKv.Tests;
+
+/// <summary>
+/// v2.5.1 — composite UNIQUE index correctness regression. Pre-fix, the
+/// executor encoded only the first column of a multi-column index for
+/// both writes and the uniqueness check, so a UNIQUE (a, b) silently
+/// degraded to UNIQUE (a). A second row with same a but different b
+/// was falsely rejected. Found via the marivil → cosmokvd dry-run on
+/// the messages table's UNIQUE (mailbox_id, uid).
+/// </summary>
+public class Phase14CompositeUniqueTests : IAsyncLifetime
+{
+    private readonly string _dir = Path.Combine(
+        Path.GetTempPath(),
+        "cosmosql-cosmokv-p14-" + Guid.NewGuid().ToString("N"));
+    private CosmoKvConnection? _conn;
+
+    public async Task InitializeAsync()
+    {
+        _conn = await CosmoKvConnection.OpenAsync(
+            new CosmoKvConfiguration { DataSource = _dir });
+        await _conn.ExecuteAsync("""
+            CREATE TABLE Messages (
+                Id BIGINT IDENTITY PRIMARY KEY,
+                MailboxId BIGINT NOT NULL,
+                Uid BIGINT NOT NULL,
+                Subject NVARCHAR(64) NOT NULL)
+            """);
+        await _conn.ExecuteAsync(
+            "CREATE UNIQUE INDEX ix_msg_mbox_uid ON Messages(MailboxId, Uid)");
+    }
+
+    public async Task DisposeAsync()
+    {
+        if (_conn is not null) await _conn.DisposeAsync();
+        try { if (Directory.Exists(_dir)) Directory.Delete(_dir, recursive: true); } catch { }
+    }
+
+    [Fact]
+    public async Task SameFirstColumn_DifferentSecond_AreNotDuplicates()
+    {
+        // Both rows have MailboxId=1; only Uid differs. Pre-fix this
+        // threw "duplicate value for UNIQUE column on index 'ix_msg_mbox_uid': 1".
+        await _conn!.ExecuteAsync(
+            "INSERT INTO Messages (MailboxId, Uid, Subject) VALUES (1, 1, 'a')");
+        await _conn.ExecuteAsync(
+            "INSERT INTO Messages (MailboxId, Uid, Subject) VALUES (1, 2, 'b')");
+        await _conn.ExecuteAsync(
+            "INSERT INTO Messages (MailboxId, Uid, Subject) VALUES (1, 3, 'c')");
+
+        var rows = await _conn.QueryAsync(
+            "SELECT COUNT(*) AS n FROM Messages WHERE MailboxId = @m",
+            new[] { SqlParameter.Named("@m", SqlValue.From(1L)) });
+        Assert.Equal(3L, rows[0]["n"].AsInt());
+    }
+
+    [Fact]
+    public async Task ExactDuplicateTuple_IsRejected()
+    {
+        await _conn!.ExecuteAsync(
+            "INSERT INTO Messages (MailboxId, Uid, Subject) VALUES (1, 1, 'first')");
+        var ex = await Assert.ThrowsAsync<UniqueConstraintViolationException>(() =>
+            _conn.ExecuteAsync(
+                "INSERT INTO Messages (MailboxId, Uid, Subject) VALUES (1, 1, 'dup')"));
+        Assert.Contains("ix_msg_mbox_uid", ex.Message);
+    }
+
+    [Fact]
+    public async Task SingleColumnPrefixScan_StillFindsCompositeEntries()
+    {
+        // The planner uses idx.Columns[0] for a single-col WHERE; the
+        // tuple-encoded entry must still be in [enc(first), nextlex(enc(first))).
+        // 6 rows across 3 different MailboxId values.
+        for (long m = 1; m <= 3; m++)
+            for (long u = 1; u <= 2; u++)
+                await _conn!.ExecuteAsync(
+                    "INSERT INTO Messages (MailboxId, Uid, Subject) VALUES (@m, @u, 's')",
+                    new[]
+                    {
+                        SqlParameter.Named("@m", SqlValue.From(m)),
+                        SqlParameter.Named("@u", SqlValue.From(u)),
+                    });
+
+        var rows = await _conn!.QueryAsync(
+            "SELECT COUNT(*) AS n FROM Messages WHERE MailboxId = @m",
+            new[] { SqlParameter.Named("@m", SqlValue.From(2L)) });
+        Assert.Equal(2L, rows[0]["n"].AsInt());
+    }
+
+    [Fact]
+    public async Task NullInComposite_DoesNotEnforceUniqueness()
+    {
+        // SQL: NULL <> NULL even when comparing tuples. Two rows with
+        // any NULL in the indexed columns must both be allowed.
+        await _conn!.ExecuteAsync("""
+            CREATE TABLE Tags (
+                Id BIGINT IDENTITY PRIMARY KEY,
+                A BIGINT NULL,
+                B NVARCHAR(64) NULL)
+            """);
+        await _conn.ExecuteAsync("CREATE UNIQUE INDEX ix_tags_a_b ON Tags(A, B)");
+        await _conn.ExecuteAsync("INSERT INTO Tags (A, B) VALUES (NULL, 'x')");
+        await _conn.ExecuteAsync("INSERT INTO Tags (A, B) VALUES (NULL, 'x')");
+        await _conn.ExecuteAsync("INSERT INTO Tags (A, B) VALUES (1, NULL)");
+        await _conn.ExecuteAsync("INSERT INTO Tags (A, B) VALUES (1, NULL)");
+
+        var rows = await _conn.QueryAsync("SELECT COUNT(*) AS n FROM Tags");
+        Assert.Equal(4L, rows[0]["n"].AsInt());
+    }
+}