From 1fdb6bf2818aeb2d260aedcd9ab4e329190a88cd Mon Sep 17 00:00:00 2001 From: Neal Mummau Date: Tue, 13 Oct 2026 22:40:52 -0400 Subject: [PATCH 1/3] perf(sqlserver): cover stream name lookups with stream version Replace the UQ_StreamName constraint with a unique index that includes Version, avoiding key lookups during appends and stream reads. Upgrade existing schemas during initialization and add regression tests for the upgrade, repeated initialization, and preserved data and uniqueness. Release note: This is a non-breaking schema optimization. Applications using InitializeDatabase=true receive the index upgrade on startup. Applications managing their schema separately must replace the UQ_StreamName constraint with a unique nonclustered index on StreamName that includes Version, retaining the UQ_StreamName name. Existing schemas remain compatible but do not receive the optimization until upgraded. Fixes #617 --- .../Eventuous.SqlServer/Scripts/1_Schema.sql | 15 +++- .../Store/SchemaTests.cs | 69 +++++++++++++++++++ 2 files changed, 82 insertions(+), 2 deletions(-) create mode 100644 src/SqlServer/test/Eventuous.Tests.SqlServer/Store/SchemaTests.cs diff --git a/src/SqlServer/src/Eventuous.SqlServer/Scripts/1_Schema.sql b/src/SqlServer/src/Eventuous.SqlServer/Scripts/1_Schema.sql index af28256c3..d27b6fad9 100644 --- a/src/SqlServer/src/Eventuous.SqlServer/Scripts/1_Schema.sql +++ b/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 diff --git a/src/SqlServer/test/Eventuous.Tests.SqlServer/Store/SchemaTests.cs b/src/SqlServer/test/Eventuous.Tests.SqlServer/Store/SchemaTests.cs new file mode 100644 index 000000000..7af42e4c8 --- /dev/null +++ b/src/SqlServer/test/Eventuous.Tests.SqlServer/Store/SchemaTests.cs @@ -0,0 +1,69 @@ +// 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] +public class SchemaTests(StoreFixture fixture) { + [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(() => duplicate.ExecuteNonQueryAsync()); + await Assert.That(error!.Number).IsEqualTo(2601); + } +} From 0b76b20032b5ddeb21c4955e2b6c3a64b1e547fd Mon Sep 17 00:00:00 2001 From: Neal Mummau Date: Tue, 13 Oct 2026 22:53:20 -0400 Subject: [PATCH 2/3] perf(sqlserver): remove redundant stream index Remove IDX_EventsStream, since UQ_StreamIdAndStreamPosition already supports filtering by StreamId. Drop the index during schema initialization for existing databases. Add regression tests for fresh schemas, upgrades, repeated initialization, and event payload preservation. Applications without InitializeDatabase enabled can apply: DROP INDEX IF EXISTS IDX_EventsStream ON eventuous.Messages; Replace eventuous with the configured schema name. Existing schemas remain compatible without this optimization. Fixes #618 --- .../Eventuous.SqlServer/Scripts/1_Schema.sql | 7 ++- .../Store/SchemaTests.cs | 47 +++++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/src/SqlServer/src/Eventuous.SqlServer/Scripts/1_Schema.sql b/src/SqlServer/src/Eventuous.SqlServer/Scripts/1_Schema.sql index d27b6fad9..6ec0d7094 100644 --- a/src/SqlServer/src/Eventuous.SqlServer/Scripts/1_Schema.sql +++ b/src/SqlServer/src/Eventuous.SqlServer/Scripts/1_Schema.sql @@ -45,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 diff --git a/src/SqlServer/test/Eventuous.Tests.SqlServer/Store/SchemaTests.cs b/src/SqlServer/test/Eventuous.Tests.SqlServer/Store/SchemaTests.cs index 7af42e4c8..ec595e809 100644 --- a/src/SqlServer/test/Eventuous.Tests.SqlServer/Store/SchemaTests.cs +++ b/src/SqlServer/test/Eventuous.Tests.SqlServer/Store/SchemaTests.cs @@ -8,6 +8,53 @@ namespace Eventuous.Tests.SqlServer.Store; [ClassDataSource] 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)] From 0a8916f17e6ca72d12926d68825bc30a70181135 Mon Sep 17 00:00:00 2001 From: Neal Mummau Date: Tue, 13 Oct 2026 23:03:59 -0400 Subject: [PATCH 3/3] perf(sqlserver): simplify append position tracking Use SCOPE_IDENTITY() to capture the last inserted GlobalPosition, removing the table variable, OUTPUT INTO, and MAX query. Add regression tests for empty, single-event, and multi-event batches on new and existing streams, including reused connections. Applications with InitializeDatabase disabled must deploy the updated append_events procedure to receive this optimization. Validated with all 46 SQL Server store tests, including concurrency tests. Fixes #619 --- .../Scripts/2_AppendEvents.sql | 15 +---- .../Store/AppendPositionTests.cs | 64 +++++++++++++++++++ 2 files changed, 66 insertions(+), 13 deletions(-) create mode 100644 src/SqlServer/test/Eventuous.Tests.SqlServer/Store/AppendPositionTests.cs diff --git a/src/SqlServer/src/Eventuous.SqlServer/Scripts/2_AppendEvents.sql b/src/SqlServer/src/Eventuous.SqlServer/Scripts/2_AppendEvents.sql index e4d042562..b86f39eb6 100644 --- a/src/SqlServer/src/Eventuous.SqlServer/Scripts/2_AppendEvents.sql +++ b/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; diff --git a/src/SqlServer/test/Eventuous.Tests.SqlServer/Store/AppendPositionTests.cs b/src/SqlServer/test/Eventuous.Tests.SqlServer/Store/AppendPositionTests.cs new file mode 100644 index 000000000..19b9d31ca --- /dev/null +++ b/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] +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); + } + } +}