perf(sqlserver): optimize stream lookups and append writes

src/SqlServer/src/Eventuous.SqlServer/Scripts/1_Schema.sql

@@ -11,11 +11,22 @@ IF OBJECT_ID('__schema__.Streams', 'U') IS NULL
             StreamName NVARCHAR(850)      NOT NULL,
             [Version]  INT DEFAULT (-1)   NOT NULL,
             CONSTRAINT PK_Streams PRIMARY KEY CLUSTERED (StreamId) WITH (OPTIMIZE_FOR_SEQUENTIAL_KEY = ON),
-            CONSTRAINT UQ_StreamName UNIQUE NONCLUSTERED (StreamName),
-            CONSTRAINT CK_VersionGteNegativeOne CHECK ([Version] >= -1)
+            CONSTRAINT CK_VersionGteNegativeOne CHECK ([Version] >= -1),
+            INDEX UQ_StreamName UNIQUE NONCLUSTERED (StreamName) INCLUDE ([Version])
         );
     END
 
+-- Upgrade existing databases. Schema.CreateSchema runs this in a transaction, keeping uniqueness protected
+-- while the constraint is replaced. Preserve the name used by check_stream's duplicate-key handling.
+IF EXISTS (
+    SELECT 1 FROM sys.key_constraints
+    WHERE parent_object_id = OBJECT_ID('__schema__.Streams') AND name = N'UQ_StreamName' AND type = 'UQ'
+)
+    BEGIN
+        ALTER TABLE __schema__.Streams DROP CONSTRAINT UQ_StreamName;
+        CREATE UNIQUE NONCLUSTERED INDEX UQ_StreamName ON __schema__.Streams (StreamName) INCLUDE ([Version]);
+    END
+
 IF OBJECT_ID('__schema__.Messages', 'U') IS NULL
     BEGIN
         CREATE TABLE __schema__.Messages
@@ -34,11 +45,14 @@ IF OBJECT_ID('__schema__.Messages', 'U') IS NULL
             CONSTRAINT UQ_StreamIdAndMessageId UNIQUE NONCLUSTERED (StreamId, MessageId),
             CONSTRAINT CK_StreamPositionGteZero CHECK (StreamPosition >= 0),
             CONSTRAINT CK_JsonDataIsJson CHECK (ISJSON(JsonData) = 1),
-            CONSTRAINT CK_JsonMetadataIsJson CHECK (ISJSON(JsonMetadata) = 1),
-            INDEX IDX_EventsStream (StreamId)
+            CONSTRAINT CK_JsonMetadataIsJson CHECK (ISJSON(JsonMetadata) = 1)
         );
     END
 
+-- UQ_StreamIdAndStreamPosition already supports seeks by StreamId. Remove the redundant index from existing databases.
+-- INDEX IDX_EventsStream (StreamId)
+DROP INDEX IF EXISTS IDX_EventsStream ON __schema__.Messages;
+
 IF OBJECT_ID('__schema__.Checkpoints', 'U') IS NULL
     BEGIN
         CREATE TABLE __schema__.Checkpoints

src/SqlServer/src/Eventuous.SqlServer/Scripts/2_AppendEvents.sql

@@ -17,11 +17,6 @@ BEGIN
         @count_messages INT,
         @new_version INT;
 
-    -- capture inserted rows to compute final position
-    DECLARE @inserted TABLE (
-        GlobalPosition BIGINT
-    );
-
     SELECT @count_messages = COUNT(1) FROM @messages;
 
     EXEC __schema__.check_stream
@@ -47,8 +42,6 @@ BEGIN
             JsonMetadata,
             Created
         )
-        OUTPUT inserted.GlobalPosition
-        INTO @inserted (GlobalPosition)
         SELECT
             message_id,
             message_type,
@@ -59,6 +52,8 @@ BEGIN
             ISNULL(@created, SYSUTCDATETIME())
         FROM @messages;
 
+        SET @position = SCOPE_IDENTITY();
+
         UPDATE s
         SET [Version] = @new_version
         FROM __schema__.Streams s
@@ -99,12 +94,6 @@ BEGIN
         END;
     END CATCH;
 
-    -- final GlobalPosition value to return
-    SELECT @position = (
-        SELECT MAX(GlobalPosition)
-        FROM @inserted
-    );
-
     SELECT
         @new_version current_version,
         @position position;

src/SqlServer/test/Eventuous.Tests.SqlServer/Store/AppendPositionTests.cs

@@ -0,0 +1,64 @@
+// Copyright (C) Eventuous HQ OÜ. All rights reserved
+// Licensed under the Apache License, Version 2.0.
+
+using Microsoft.Data.SqlClient;
+
+namespace Eventuous.Tests.SqlServer.Store;
+
+[ClassDataSource<StoreFixture>]
+public class AppendPositionTests(StoreFixture fixture) {
+    [Test]
+    [Arguments(0)]
+    [Arguments(1)]
+    [Arguments(10)]
+    public async Task ShouldReturnBatchPositionForNewAndExistingStreams(int count) {
+        await using var connection = new SqlConnection(fixture.Container.GetConnectionString());
+        await connection.OpenAsync();
+        await using var transaction = (SqlTransaction)await connection.BeginTransactionAsync();
+
+        // Reuse the session after another append: empty batches must not return an earlier identity.
+        await AppendAndCheck("other-stream", 3, 2);
+        await AppendAndCheck("other-stream", 0, 2);
+        await AppendAndCheck("test-stream", count, count - 1);
+        await AppendAndCheck("test-stream", count, count * 2 - 1);
+        await transaction.CommitAsync();
+
+        async Task AppendAndCheck(string stream, int batchSize, int expectedVersion) {
+            await using var command = new SqlCommand(
+                $"""
+                 DECLARE @messages {fixture.SchemaName}.StreamMessage;
+                 DECLARE @remaining INT = @count;
+                 WHILE @remaining > 0
+                 BEGIN
+                     INSERT INTO @messages (message_id, message_type, json_data, json_metadata)
+                     VALUES (NEWID(), N'TestEvent', @json, @json);
+                     SET @remaining -= 1;
+                 END;
+
+                 EXEC {fixture.SchemaName}.append_events
+                     @stream_name = @name, @expected_version = -2, @created = NULL, @messages = @messages;
+
+                 SELECT MAX(m.GlobalPosition)
+                 FROM {fixture.SchemaName}.Messages m
+                 JOIN {fixture.SchemaName}.Streams s ON s.StreamId = m.StreamId
+                 WHERE s.StreamName = @name;
+                 """,
+                connection,
+                transaction
+            );
+            command.Parameters.AddWithValue("@count", batchSize);
+            command.Parameters.AddWithValue("@name", stream);
+            command.Parameters.AddWithValue("@json", "{}");
+
+            await using var reader = await command.ExecuteReaderAsync();
+            await Assert.That(await reader.ReadAsync()).IsTrue();
+            await Assert.That(reader.GetInt32(0)).IsEqualTo(expectedVersion);
+            long? position = reader.IsDBNull(1) ? null : reader.GetInt64(1);
+
+            await Assert.That(await reader.NextResultAsync()).IsTrue();
+            await Assert.That(await reader.ReadAsync()).IsTrue();
+            long? expectedPosition = batchSize == 0 ? null : reader.GetInt64(0);
+            await Assert.That(position).IsEqualTo(expectedPosition);
+        }
+    }
+}

src/SqlServer/test/Eventuous.Tests.SqlServer/Store/SchemaTests.cs

@@ -0,0 +1,116 @@
+// Copyright (C) Eventuous HQ OÜ. All rights reserved
+// Licensed under the Apache License, Version 2.0.
+
+using Eventuous.SqlServer;
+using Microsoft.Data.SqlClient;
+
+namespace Eventuous.Tests.SqlServer.Store;
+
+[ClassDataSource<StoreFixture>]
+public class SchemaTests(StoreFixture fixture) {
+    [Test]
+    [Arguments(false)]
+    [Arguments(true)]
+    public async Task ShouldOmitRedundantStreamIndexAfterRepeatedInitialization(bool legacyIndex) {
+        var schema = new Schema(fixture.SchemaName);
+        await using var connection = new SqlConnection(fixture.Container.GetConnectionString());
+        await connection.OpenAsync();
+
+        if (legacyIndex) {
+            await using var legacy = new SqlCommand($"CREATE INDEX IDX_EventsStream ON {schema.SchemaName}.Messages (StreamId);", connection);
+            await legacy.ExecuteNonQueryAsync();
+        }
+
+        // Exercise the upgrade on a populated table, including an off-row JSON payload.
+        await using var seed = new SqlCommand(
+            $"""
+             INSERT INTO {schema.SchemaName}.Streams (StreamName, [Version]) VALUES (N'preserved-events', 0);
+             INSERT INTO {schema.SchemaName}.Messages
+                 (MessageId, MessageType, StreamId, StreamPosition, JsonData, JsonMetadata, Created)
+             VALUES
+                 (NEWID(), N'TestEvent', SCOPE_IDENTITY(), 0, @data, @metadata, SYSUTCDATETIME());
+             """,
+            connection
+        );
+        var payload = "{\"value\":\"" + new string('x', 10000) + "\"}";
+        seed.Parameters.AddWithValue("@data", payload);
+        seed.Parameters.AddWithValue("@metadata", "{}");
+        await seed.ExecuteNonQueryAsync();
+
+        await AssertIndexAbsent(!legacyIndex);
+        await schema.CreateSchema(fixture.Container.GetConnectionString(), null, default);
+        await AssertIndexAbsent(true);
+        await schema.CreateSchema(fixture.Container.GetConnectionString(), null, default);
+        await AssertIndexAbsent(true);
+
+        await using var preserved = new SqlCommand($"SELECT JsonData FROM {schema.SchemaName}.Messages;", connection);
+        await Assert.That((string)(await preserved.ExecuteScalarAsync())!).IsEqualTo(payload);
+
+        async Task AssertIndexAbsent(bool absent) {
+            await using var index = new SqlCommand(
+                $"SELECT COUNT(*) FROM sys.indexes WHERE object_id = OBJECT_ID('{schema.SchemaName}.Messages') AND name = N'IDX_EventsStream';",
+                connection
+            );
+            await Assert.That((int)(await index.ExecuteScalarAsync())!).IsEqualTo(absent ? 0 : 1);
+        }
+    }
+
+    [Test]
+    [Arguments(false)]
+    [Arguments(true)]
+    public async Task ShouldCoverStreamLookupAfterRepeatedInitialization(bool legacyConstraint) {
+        var schema = new Schema(fixture.SchemaName);
+        await using var connection = new SqlConnection(fixture.Container.GetConnectionString());
+        await connection.OpenAsync();
+
+        if (legacyConstraint) {
+            await using var legacy = new SqlCommand(
+                $"""
+                 DROP INDEX UQ_StreamName ON {schema.SchemaName}.Streams;
+                 ALTER TABLE {schema.SchemaName}.Streams ADD CONSTRAINT UQ_StreamName UNIQUE NONCLUSTERED (StreamName);
+                 """,
+                connection
+            );
+            await legacy.ExecuteNonQueryAsync();
+        }
+
+        await using var seed = new SqlCommand(
+            $"INSERT INTO {schema.SchemaName}.Streams (StreamName, [Version]) OUTPUT INSERTED.StreamId VALUES (N'preserved-stream', 42);",
+            connection
+        );
+        var streamId = (int)(await seed.ExecuteScalarAsync())!;
+
+        // The first call upgrades the legacy constraint; subsequent calls must leave the covering index intact.
+        await schema.CreateSchema(fixture.Container.GetConnectionString(), null, default);
+        await schema.CreateSchema(fixture.Container.GetConnectionString(), null, default);
+
+        await using var index = new SqlCommand(
+            $"""
+             SELECT COUNT(*)
+             FROM sys.indexes i
+             JOIN sys.index_columns ic ON ic.object_id = i.object_id AND ic.index_id = i.index_id
+             JOIN sys.columns c ON c.object_id = ic.object_id AND c.column_id = ic.column_id
+             WHERE i.object_id = OBJECT_ID('{schema.SchemaName}.Streams')
+               AND i.name = N'UQ_StreamName' AND i.is_unique = 1 AND i.is_unique_constraint = 0 AND i.type = 2
+               AND ((c.name = N'StreamName' AND ic.key_ordinal = 1)
+                 OR (c.name = N'Version' AND ic.is_included_column = 1));
+             """,
+            connection
+        );
+        await Assert.That((int)(await index.ExecuteScalarAsync())!).IsEqualTo(2);
+
+        await using var preserved = new SqlCommand(
+            $"SELECT [Version] FROM {schema.SchemaName}.Streams WHERE StreamId = @id AND StreamName = N'preserved-stream';",
+            connection
+        );
+        preserved.Parameters.AddWithValue("@id", streamId);
+        await Assert.That((int)(await preserved.ExecuteScalarAsync())!).IsEqualTo(42);
+
+        await using var duplicate = new SqlCommand(
+            $"INSERT INTO {schema.SchemaName}.Streams (StreamName) VALUES (N'preserved-stream');",
+            connection
+        );
+        var error = await Assert.ThrowsAsync<SqlException>(() => duplicate.ExecuteNonQueryAsync());
+        await Assert.That(error!.Number).IsEqualTo(2601);
+    }
+}