diff --git a/src/SqlServer/src/Eventuous.SqlServer/Scripts/1_Schema.sql b/src/SqlServer/src/Eventuous.SqlServer/Scripts/1_Schema.sql index af28256c..6ec0d709 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 @@ -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 diff --git a/src/SqlServer/src/Eventuous.SqlServer/Scripts/2_AppendEvents.sql b/src/SqlServer/src/Eventuous.SqlServer/Scripts/2_AppendEvents.sql index e4d04256..b86f39eb 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 00000000..19b9d31c --- /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); + } + } +} 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 00000000..ec595e80 --- /dev/null +++ b/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] +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(() => duplicate.ExecuteNonQueryAsync()); + await Assert.That(error!.Number).IsEqualTo(2601); + } +}