Fix SQLServer 2016 tag table compatibility bug

src/Akka.Persistence.Sql.Tests.Common/Containers/SqlServer2016Container.cs

@@ -0,0 +1,108 @@
+// -----------------------------------------------------------------------
+//  <copyright file="SqlServer2016Container.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using System;
+using System.Collections.Generic;
+using System.Data.Common;
+using System.Threading.Tasks;
+using Akka.Util;
+using Docker.DotNet.Models;
+using Microsoft.Data.SqlClient;
+
+namespace Akka.Persistence.Sql.Tests.Common.Containers
+{
+    /// <summary>
+    /// Runs a SQL Server 2022 container but sets the database compatibility level to 130
+    /// (SQL Server 2016) so that LinqToDB auto-detects it as SQL Server 2016.
+    /// This means STRING_AGG is not available, forcing the client-side tag aggregation fallback.
+    /// Uses the generic "SqlServer" provider name to trigger LinqToDB auto-detection.
+    /// </summary>
+    public sealed class SqlServer2016Container : DockerContainer
+    {
+        private const string User = "sa";
+
+        private const string Password = "Password12!";
+
+        private readonly DbConnectionStringBuilder _connectionStringBuilder;
+
+        public SqlServer2016Container() : base("mcr.microsoft.com/mssql/server", "2022-latest", $"mssql2016-{Guid.NewGuid():N}")
+            => _connectionStringBuilder = new DbConnectionStringBuilder
+            {
+                ["Server"] = $"localhost,{Port}",
+                ["User Id"] = User,
+                ["Password"] = Password,
+                ["TrustServerCertificate"] = "true",
+            };
+
+        public override string ConnectionString => _connectionStringBuilder.ToString();
+
+        // Use generic "SqlServer" so LinqToDB auto-detects version from compatibility level
+        public override string ProviderName => LinqToDB.ProviderName.SqlServer;
+
+        private int Port { get; } = ThreadLocalRandom.Current.Next(9000, 10000);
+
+        protected override string ReadyMarker => "Recovery is complete.";
+
+        protected override void GenerateDatabaseName()
+        {
+            base.GenerateDatabaseName();
+
+            _connectionStringBuilder["Database"] = DatabaseName;
+        }
+
+        protected override void ConfigureContainer(CreateContainerParameters parameters)
+        {
+            parameters.ExposedPorts = new Dictionary<string, EmptyStruct>
+            {
+                ["1433/tcp"] = new(),
+            };
+
+            parameters.HostConfig = new HostConfig
+            {
+                PortBindings = new Dictionary<string, IList<PortBinding>>
+                {
+                    ["1433/tcp"] = new List<PortBinding> { new() { HostPort = $"{Port}" } },
+                },
+            };
+
+            parameters.Env = new[]
+            {
+                "ACCEPT_EULA=Y",
+                $"MSSQL_SA_PASSWORD={Password}",
+                "MSSQL_PID=Express",
+            };
+        }
+
+        public override async Task InitializeDbAsync()
+        {
+            _connectionStringBuilder["Database"] = "master";
+
+            await using var connection = new SqlConnection(ConnectionString);
+            await connection.OpenAsync();
+
+            GenerateDatabaseName();
+
+            await using var createCommand = new SqlCommand
+            {
+                CommandText = $"CREATE DATABASE [{DatabaseName}]",
+                Connection = connection,
+            };
+            await createCommand.ExecuteNonQueryAsync();
+
+            // Set compatibility level to 130 (SQL Server 2016)
+            // This causes LinqToDB to auto-detect the version as SQL Server 2016,
+            // which does not support STRING_AGG
+            await using var compatCommand = new SqlCommand
+            {
+                CommandText = $"ALTER DATABASE [{DatabaseName}] SET COMPATIBILITY_LEVEL = 130",
+                Connection = connection,
+            };
+            await compatCommand.ExecuteNonQueryAsync();
+
+            await connection.CloseAsync();
+        }
+    }
+}

src/Akka.Persistence.Sql.Tests/Query/SqlServer2016/TagTable/SqlServer2016AllEventsSpec.cs

@@ -0,0 +1,28 @@
+// -----------------------------------------------------------------------
+//  <copyright file="SqlServer2016AllEventsSpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using Akka.Persistence.Sql.Config;
+using Akka.Persistence.Sql.Tests.Common.Containers;
+using Akka.Persistence.Sql.Tests.Common.Query;
+using Akka.Persistence.Sql.Tests.SqlServer;
+using Xunit;
+using Xunit.Abstractions;
+#if !DEBUG
+using Akka.Persistence.Sql.Tests.Common.Internal.Xunit;
+#endif
+
+namespace Akka.Persistence.Sql.Tests.Query.SqlServer2016.TagTable
+{
+#if !DEBUG
+    [SkipWindows]
+#endif
+    [Collection(nameof(SqlServer2016PersistenceSpec))]
+    public class SqlServer2016AllEventsSpec : BaseAllEventsSpec<SqlServer2016Container>
+    {
+        public SqlServer2016AllEventsSpec(ITestOutputHelper output, SqlServer2016Container fixture)
+            : base(TagMode.TagTable, output, nameof(SqlServer2016AllEventsSpec), fixture) { }
+    }
+}

src/Akka.Persistence.Sql.Tests/Query/SqlServer2016/TagTable/SqlServer2016CurrentAllEventsSpec.cs

@@ -0,0 +1,28 @@
+// -----------------------------------------------------------------------
+//  <copyright file="SqlServer2016CurrentAllEventsSpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using Akka.Persistence.Sql.Config;
+using Akka.Persistence.Sql.Tests.Common.Containers;
+using Akka.Persistence.Sql.Tests.Common.Query;
+using Akka.Persistence.Sql.Tests.SqlServer;
+using Xunit;
+using Xunit.Abstractions;
+#if !DEBUG
+using Akka.Persistence.Sql.Tests.Common.Internal.Xunit;
+#endif
+
+namespace Akka.Persistence.Sql.Tests.Query.SqlServer2016.TagTable
+{
+#if !DEBUG
+    [SkipWindows]
+#endif
+    [Collection(nameof(SqlServer2016PersistenceSpec))]
+    public class SqlServer2016CurrentAllEventsSpec : BaseCurrentAllEventsSpec<SqlServer2016Container>
+    {
+        public SqlServer2016CurrentAllEventsSpec(ITestOutputHelper output, SqlServer2016Container fixture)
+            : base(TagMode.TagTable, output, nameof(SqlServer2016CurrentAllEventsSpec), fixture) { }
+    }
+}

src/Akka.Persistence.Sql.Tests/Query/SqlServer2016/TagTable/SqlServer2016CurrentEventsByTagSpec.cs

@@ -0,0 +1,28 @@
+// -----------------------------------------------------------------------
+//  <copyright file="SqlServer2016CurrentEventsByTagSpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using Akka.Persistence.Sql.Config;
+using Akka.Persistence.Sql.Tests.Common.Containers;
+using Akka.Persistence.Sql.Tests.Common.Query;
+using Akka.Persistence.Sql.Tests.SqlServer;
+using Xunit;
+using Xunit.Abstractions;
+#if !DEBUG
+using Akka.Persistence.Sql.Tests.Common.Internal.Xunit;
+#endif
+
+namespace Akka.Persistence.Sql.Tests.Query.SqlServer2016.TagTable
+{
+#if !DEBUG
+    [SkipWindows]
+#endif
+    [Collection(nameof(SqlServer2016PersistenceSpec))]
+    public class SqlServer2016CurrentEventsByTagSpec : BaseCurrentEventsByTagSpec<SqlServer2016Container>
+    {
+        public SqlServer2016CurrentEventsByTagSpec(ITestOutputHelper output, SqlServer2016Container fixture)
+            : base(TagMode.TagTable, output, nameof(SqlServer2016CurrentEventsByTagSpec), fixture) { }
+    }
+}

src/Akka.Persistence.Sql.Tests/Query/SqlServer2016/TagTable/SqlServer2016EventsByPersistenceIdSpec.cs

@@ -0,0 +1,28 @@
+// -----------------------------------------------------------------------
+//  <copyright file="SqlServer2016EventsByPersistenceIdSpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using Akka.Persistence.Sql.Config;
+using Akka.Persistence.Sql.Tests.Common.Containers;
+using Akka.Persistence.Sql.Tests.Common.Query;
+using Akka.Persistence.Sql.Tests.SqlServer;
+using Xunit;
+using Xunit.Abstractions;
+#if !DEBUG
+using Akka.Persistence.Sql.Tests.Common.Internal.Xunit;
+#endif
+
+namespace Akka.Persistence.Sql.Tests.Query.SqlServer2016.TagTable
+{
+#if !DEBUG
+    [SkipWindows]
+#endif
+    [Collection(nameof(SqlServer2016PersistenceSpec))]
+    public class SqlServer2016EventsByPersistenceIdSpec : BaseEventsByPersistenceIdSpec<SqlServer2016Container>
+    {
+        public SqlServer2016EventsByPersistenceIdSpec(ITestOutputHelper output, SqlServer2016Container fixture)
+            : base(TagMode.TagTable, output, nameof(SqlServer2016EventsByPersistenceIdSpec), fixture) { }
+    }
+}

src/Akka.Persistence.Sql.Tests/Query/SqlServer2016/TagTable/SqlServer2016EventsByTagSpec.cs

@@ -0,0 +1,84 @@
+// -----------------------------------------------------------------------
+//  <copyright file="SqlServer2016EventsByTagSpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Akka.Persistence.Sql.Config;
+using Akka.Persistence.Sql.Db;
+using Akka.Persistence.Sql.Journal.Types;
+using Akka.Persistence.Sql.Tests.Common.Containers;
+using Akka.Persistence.Sql.Tests.Common.Query;
+using Akka.Persistence.Sql.Tests.SqlServer;
+using FluentAssertions;
+using LinqToDB;
+using LinqToDB.Data;
+using LinqToDB.DataProvider.SqlServer;
+using Xunit;
+using Xunit.Abstractions;
+#if !DEBUG
+using Akka.Persistence.Sql.Tests.Common.Internal.Xunit;
+#endif
+
+namespace Akka.Persistence.Sql.Tests.Query.SqlServer2016.TagTable
+{
+#if !DEBUG
+    [SkipWindows]
+#endif
+    [Collection(nameof(SqlServer2016PersistenceSpec))]
+    public class SqlServer2016EventsByTagSpec : BaseEventsByTagSpec<SqlServer2016Container>
+    {
+        private readonly SqlServer2016Container _fixture;
+
+        public SqlServer2016EventsByTagSpec(ITestOutputHelper output, SqlServer2016Container fixture)
+            : base(TagMode.TagTable, output, nameof(SqlServer2016EventsByTagSpec), fixture)
+        {
+            _fixture = fixture;
+        }
+
+        [Fact(DisplayName = "StringAggregate should throw on SQL Server 2016 compatibility level")]
+        public async Task StringAggregateShouldThrowOnSqlServer2016()
+        {
+            // Arrange: connect to the database using the generic "SqlServer" provider
+            // LinqToDB will auto-detect the compatibility level as 130 (SQL Server 2016)
+            var dataProvider = SqlServerTools.GetDataProvider(
+                SqlServerVersion.AutoDetect,
+                SqlServerProvider.MicrosoftDataSqlClient,
+                _fixture.ConnectionString);
+            await using var db = new DataConnection(dataProvider, _fixture.ConnectionString);
+
+            // Verify LinqToDB detected the version as SQL Server 2016
+            var sqlServerProvider = db.DataProvider as SqlServerDataProvider;
+            sqlServerProvider.Should().NotBeNull();
+            sqlServerProvider!.Version.Should().Be(SqlServerVersion.v2016,
+                "LinqToDB should auto-detect compatibility level 130 as SQL Server 2016");
+
+            // Verify our SupportsStringAggregate property returns false
+            var akkaConnection = new AkkaDataConnection(LinqToDB.ProviderName.SqlServer, db);
+            akkaConnection.SupportsStringAggregate.Should().BeFalse(
+                "SQL Server 2016 does not support STRING_AGG");
+
+            // Act & Assert: StringAggregate should throw when converted to SQL
+            var tagTable = db.GetTable<JournalTagRow>();
+            var journalTable = db.GetTable<JournalRow>();
+
+            // This is the exact query pattern used in AddTagDataFromTagTableWithStringAggregateAsync
+            var act = () => journalTable
+                .Select(x => new
+                {
+                    row = x,
+                    tags = tagTable
+                        .Where(r => r.OrderingId == x.Ordering)
+                        .StringAggregate(";", r => r.TagValue)
+                        .ToValue(),
+                })
+                .ToList();
+
+            act.Should().Throw<Exception>(
+                "STRING_AGG is not supported on SQL Server 2016 (compatibility level 130)");
+        }
+    }
+}

src/Akka.Persistence.Sql.Tests/Query/SqlServer2016/TagTable/SqlServer2016PersistenceIdsSpec.cs

@@ -0,0 +1,28 @@
+// -----------------------------------------------------------------------
+//  <copyright file="SqlServer2016PersistenceIdsSpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using Akka.Persistence.Sql.Config;
+using Akka.Persistence.Sql.Tests.Common.Containers;
+using Akka.Persistence.Sql.Tests.Common.Query;
+using Akka.Persistence.Sql.Tests.SqlServer;
+using Xunit;
+using Xunit.Abstractions;
+#if !DEBUG
+using Akka.Persistence.Sql.Tests.Common.Internal.Xunit;
+#endif
+
+namespace Akka.Persistence.Sql.Tests.Query.SqlServer2016.TagTable
+{
+#if !DEBUG
+    [SkipWindows]
+#endif
+    [Collection(nameof(SqlServer2016PersistenceSpec))]
+    public class SqlServer2016PersistenceIdsSpec : BasePersistenceIdsSpec<SqlServer2016Container>
+    {
+        public SqlServer2016PersistenceIdsSpec(ITestOutputHelper output, SqlServer2016Container fixture)
+            : base(TagMode.TagTable, output, nameof(SqlServer2016PersistenceIdsSpec), fixture) { }
+    }
+}

src/Akka.Persistence.Sql.Tests/Query/SqlServer2016/TagTable/SqlServer2016QueryThrottleSpecs.cs

@@ -0,0 +1,22 @@
+// -----------------------------------------------------------------------
+//  <copyright file="SqlServer2016QueryThrottleSpecs.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using Akka.Persistence.Sql.Config;
+using Akka.Persistence.Sql.Tests.Common.Containers;
+using Akka.Persistence.Sql.Tests.SqlServer;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Akka.Persistence.Sql.Tests.Query.SqlServer2016.TagTable;
+
+[Collection(nameof(SqlServer2016PersistenceSpec))]
+public class SqlServer2016QueryThrottleSpecs : QueryThrottleSpecsBase<SqlServer2016Container>
+{
+    public SqlServer2016QueryThrottleSpecs(ITestOutputHelper output, SqlServer2016Container fixture)
+        : base(TagMode.TagTable, output, nameof(SqlServer2016QueryThrottleSpecs), fixture)
+    {
+    }
+}