fix(sql): chunk the key-value IN list at the dialect's bind ceiling (Driver.Sql-002, Driver.Sql.Dialect-002)

ISqlDialect gains MaxBindParameters — the per-command ceiling the seam was
missing while modelling everything else that differs per backend: 2098 usable
of SQL Server's 2100 (sp_executesql spends two), 1000 for Oracle's IN-list
limit (ORA-01795, tighter than its 64k bind count), 65535 for the Postgres
Bind message.

SqlGroupPlanner.AddKeyValuePlans now emits several plans for one group when
its distinct keys exceed that ceiling, instead of one statement the provider
refuses after the connection opened — which left every tag in an EAV group
publishing BadCommunicationError on every poll, permanently, and reported the
database endpoint as down. Chunks share the group key; members follow their
own key so a shared keyValue never splits. SqlPollReader is unchanged: it runs
one command per plan and slices each plan's own Members back, proven by a
fixture read at a ceiling of 1.

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/ColonMarkerDialect.cs

@@ -65,6 +65,9 @@ public sealed class ColonMarkerDialect : ISqlDialect
     /// <inheritdoc/>
     public DbProviderFactory Factory => Inner.Factory;
 
+    /// <inheritdoc/>
+    public int MaxBindParameters => Inner.MaxBindParameters;
+
     /// <inheritdoc/>
     public string LivenessSql => Inner.LivenessSql;
 

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/OracleDialectTests.cs

@@ -190,6 +190,20 @@ public class OracleDialectTests
         Should.Throw<ArgumentException>(() => d.QuoteIdentifier(new string('б', 65)));
     }
 
+    // ---- bind-parameter ceiling ----
+
+    /// <summary>
+    ///     1000 is the <c>IN</c>-expression-list limit (<c>ORA-01795</c>), NOT Oracle's much larger 64k bind
+    ///     count — the seam has to report the limit the planner's <c>IN (:p_k0, …)</c> list actually hits, or
+    ///     it would chunk to a size the server still refuses.
+    /// </summary>
+    [Fact]
+    public void MaxBindParameters_isTheInListLimit_notTheLargerBindVariableCount()
+    {
+        new OracleDialect().MaxBindParameters.ShouldBe(1000);
+        new OracleDialect().MaxBindParameters.ShouldBeLessThan(new PostgresDialect().MaxBindParameters);
+    }
+
     // ---- liveness + row limit ----
 
     [Fact]

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/PostgresDialectTests.cs

@@ -219,6 +219,15 @@ public class PostgresDialectTests
             .ShouldBe("SELECT \"a\" LIMIT 1");
     }
 
+    /// <summary>
+    ///     65535 — the <c>Bind</c> message's 16-bit parameter count, a protocol ceiling rather than a server
+    ///     setting. Pinned so the planner's chunking has a real number here instead of a special case for
+    ///     "this dialect has no meaningful limit".
+    /// </summary>
+    [Fact]
+    public void MaxBindParameters_isTheWireProtocolsSixteenBitCount()
+        => new PostgresDialect().MaxBindParameters.ShouldBe(65535);
+
     [Fact]
     public void Dialect_identifiesItselfAsPostgres_andExposesTheAbstractFactory()
     {

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlCatalogDefaultSchemaTests.cs

@@ -105,6 +105,8 @@ public sealed class SqlCatalogDefaultSchemaTests
 
         public DbProviderFactory Factory => Inner.Factory;
 
+        public int MaxBindParameters => Inner.MaxBindParameters;
+
         public string LivenessSql => Inner.LivenessSql;
 
         public string SingleRowLimitPrefix => Inner.SingleRowLimitPrefix;

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlCatalogGateDriverTests.cs

@@ -337,6 +337,8 @@ public sealed class SqlCatalogGateDriverTests
 
         public System.Data.Common.DbProviderFactory Factory => Inner.Factory;
 
+        public int MaxBindParameters => Inner.MaxBindParameters;
+
         public string LivenessSql => Inner.LivenessSql;
 
         public string SingleRowLimitPrefix => Inner.SingleRowLimitPrefix;

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlGroupPlannerChunkingTests.cs

@@ -0,0 +1,318 @@
+using System.Data.Common;
+using System.Globalization;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests;
+
+/// <summary>
+///     Pins the bind-parameter ceiling (Driver.Sql-002 / Driver.Sql.Dialect-002): a key-value group holding
+///     more distinct <c>keyValue</c>s than the dialect's <see cref="ISqlDialect.MaxBindParameters"/> is split
+///     into several plans over the <b>same</b> group key, rather than emitting one statement the provider
+///     rejects.
+///     <para><b>Why this is a correctness test and not a tuning one.</b> Over the ceiling the provider refuses
+///     the statement <em>after</em> the connection opened, so every tag in the group publishes
+///     <c>BadCommunicationError</c> on every poll, permanently — and a key-value (EAV) staging table is
+///     precisely the shape that attracts thousands of tags on one table, which the group-key rule folds into
+///     one group.</para>
+///     <para>A single group emitting several plans needs <b>no reader change</b>: <c>SqlPollReader</c> runs
+///     one command per plan and slices each plan's own <c>Members</c> back, so the last test here drives the
+///     whole read path over a real database to prove exactly that.</para>
+/// </summary>
+public sealed class SqlGroupPlannerChunkingTests
+{
+    private static SqlTagDefinition KvTag(string name, string keyValue, string table = "dbo.TagValues")
+        => new(name, SqlTagModel.KeyValue, table,
+            KeyColumn: "tag_name", KeyValue: keyValue, ValueColumn: "num_value");
+
+    // ---- the shipped ceilings ----
+
+    /// <summary>
+    ///     Oracle's real ceiling, on the real dialect: an <c>IN</c> expression list may hold 1000 entries, so
+    ///     1001 distinct keys cannot be one statement. This is the shape the finding describes — no test-only
+    ///     dialect involved.
+    /// </summary>
+    [Fact]
+    public void AGroupBeyondTheOracleInListLimit_isSplit_ratherThanEmittedAsOneRejectedStatement()
+    {
+        var dialect = new OracleDialect();
+        var tags = Enumerable.Range(0, 1001)
+            .Select(i => KvTag("T" + i.ToString(CultureInfo.InvariantCulture),
+                "Key" + i.ToString(CultureInfo.InvariantCulture)))
+            .ToArray();
+
+        var plans = SqlGroupPlanner.Plan(tags, dialect);
+
+        plans.Count.ShouldBe(2);
+        plans.ShouldAllBe(p => p.Parameters.Count <= 1000);
+        plans.Sum(p => p.Parameters.Count).ShouldBe(1001);
+    }
+
+    /// <summary>
+    ///     The same for SQL Server, whose 2100-parameter command cap is the other limit the finding names.
+    /// </summary>
+    [Fact]
+    public void AGroupBeyondTheSqlServerParameterCap_isSplit()
+    {
+        var dialect = new SqlServerDialect();
+        var tags = Enumerable.Range(0, dialect.MaxBindParameters + 1)
+            .Select(i => KvTag("T" + i.ToString(CultureInfo.InvariantCulture),
+                "Key" + i.ToString(CultureInfo.InvariantCulture)))
+            .ToArray();
+
+        var plans = SqlGroupPlanner.Plan(tags, dialect);
+
+        plans.Count.ShouldBe(2);
+        plans.ShouldAllBe(p => p.Parameters.Count <= dialect.MaxBindParameters);
+    }
+
+    // ---- how a split group is shaped ----
+
+    [Fact]
+    public void EachChunkBindsAtMostTheCeiling_andTheChunksCoverEveryKeyExactlyOnce()
+    {
+        var tags = Enumerable.Range(0, 7)
+            .Select(i => KvTag("T" + i.ToString(CultureInfo.InvariantCulture),
+                "Key" + i.ToString(CultureInfo.InvariantCulture)))
+            .ToArray();
+
+        var plans = SqlGroupPlanner.Plan(tags, new BindCeilingDialect(3));
+
+        plans.Count.ShouldBe(3);
+        plans.Select(p => p.Parameters.Count).ShouldBe(new[] { 3, 3, 1 });
+        plans.SelectMany(p => p.Parameters)
+            .ShouldBe(Enumerable.Range(0, 7).Select(i => (object?)("Key" + i.ToString(CultureInfo.InvariantCulture))));
+    }
+
+    /// <summary>
+    ///     Every member must land in the chunk that binds <em>its own</em> key — a member whose key is bound
+    ///     by a different plan reads a result set that cannot contain its row and publishes
+    ///     <c>BadNoData</c> forever, which is the quiet way to get this wrong.
+    /// </summary>
+    [Fact]
+    public void EveryMemberLandsInTheChunkThatBindsItsOwnKey()
+    {
+        var tags = Enumerable.Range(0, 7)
+            .Select(i => KvTag("T" + i.ToString(CultureInfo.InvariantCulture),
+                "Key" + i.ToString(CultureInfo.InvariantCulture)))
+            .ToArray();
+
+        var plans = SqlGroupPlanner.Plan(tags, new BindCeilingDialect(3));
+
+        foreach (var plan in plans)
+            foreach (var member in plan.Members)
+                plan.Parameters.ShouldContain(member.KeyValue);
+
+        // …and no member is dropped or duplicated across the split.
+        plans.SelectMany(p => p.Members).Select(m => m.Name)
+            .ShouldBe(tags.Select(t => t.Name), ignoreOrder: true);
+        plans.Sum(p => p.Members.Count).ShouldBe(tags.Length);
+    }
+
+    /// <summary>
+    ///     Two tags on one <c>keyValue</c> bind one parameter and must stay in one chunk — splitting them
+    ///     would bind the same key twice and cost a whole extra round-trip for a row already fetched.
+    /// </summary>
+    [Fact]
+    public void MembersSharingAKeyValue_stayInTheSameChunk()
+    {
+        var plans = SqlGroupPlanner.Plan(
+        [
+            KvTag("A", "K1"), KvTag("B", "K2"), KvTag("C", "K1"), KvTag("D", "K3"),
+        ], new BindCeilingDialect(2));
+
+        plans.Count.ShouldBe(2);
+        plans[0].Parameters.ShouldBe(new object[] { "K1", "K2" });
+        plans[0].Members.Select(m => m.Name).ShouldBe(new[] { "A", "B", "C" });
+        plans[1].Parameters.ShouldBe(new object[] { "K3" });
+        plans[1].Members.Select(m => m.Name).ShouldBe(new[] { "D" });
+    }
+
+    /// <summary>
+    ///     Each chunk is its own command, so its markers restart at <c>k0</c> — carrying the global index
+    ///     forward would emit <c>IN (@k3, @k4, @k5)</c> against a three-parameter command, which is legal but
+    ///     makes a plan's SQL unreadable next to its own parameter list.
+    /// </summary>
+    [Fact]
+    public void EachChunkNumbersItsMarkersFromZero_andTheSqlMatchesItsOwnParameters()
+    {
+        var tags = Enumerable.Range(0, 5)
+            .Select(i => KvTag("T" + i.ToString(CultureInfo.InvariantCulture),
+                "Key" + i.ToString(CultureInfo.InvariantCulture)))
+            .ToArray();
+
+        var plans = SqlGroupPlanner.Plan(tags, new BindCeilingDialect(2));
+
+        foreach (var plan in plans)
+        {
+            plan.ParameterNames.ShouldBe(
+                Enumerable.Range(0, plan.Parameters.Count)
+                    .Select(i => "@k" + i.ToString(CultureInfo.InvariantCulture)));
+            plan.SqlText.ShouldEndWith("IN (" + string.Join(", ", plan.ParameterNames) + ")");
+        }
+    }
+
+    /// <summary>All chunks of one group carry that group's key — the split is an execution detail, not a regrouping.</summary>
+    [Fact]
+    public void EveryChunkOfOneGroupCarriesTheSameGroupKey()
+    {
+        var tags = Enumerable.Range(0, 5)
+            .Select(i => KvTag("T" + i.ToString(CultureInfo.InvariantCulture),
+                "Key" + i.ToString(CultureInfo.InvariantCulture)))
+            .ToArray();
+
+        var plans = SqlGroupPlanner.Plan(tags, new BindCeilingDialect(2));
+
+        plans.Count.ShouldBeGreaterThan(1);
+        plans.Select(p => p.GroupKey).Distinct().Count().ShouldBe(1);
+    }
+
+    /// <summary>
+    ///     The falsifiability control. Chunking must not fragment an ordinary group: a planner that split
+    ///     eagerly (or off a stale count) would pass every test above and quietly turn one round-trip per
+    ///     source into N.
+    /// </summary>
+    [Theory]
+    [InlineData(1)]
+    [InlineData(2)]
+    [InlineData(3)]
+    public void AGroupAtOrUnderTheCeiling_stillEmitsExactlyOnePlan(int distinctKeys)
+    {
+        var tags = Enumerable.Range(0, distinctKeys)
+            .Select(i => KvTag("T" + i.ToString(CultureInfo.InvariantCulture),
+                "Key" + i.ToString(CultureInfo.InvariantCulture)))
+            .ToArray();
+
+        var plans = SqlGroupPlanner.Plan(tags, new BindCeilingDialect(3));
+
+        plans.Count.ShouldBe(1);
+        plans[0].Members.Count.ShouldBe(distinctKeys);
+    }
+
+    /// <summary>
+    ///     Two groups that each need splitting stay separate, and the chunks of one never mix with the other's
+    ///     — the split happens inside a group, after grouping.
+    /// </summary>
+    [Fact]
+    public void TwoOversizedGroups_areSplitIndependently()
+    {
+        var tags = Enumerable.Range(0, 3).Select(i => KvTag("A" + i, "K" + i, "dbo.First"))
+            .Concat(Enumerable.Range(0, 3).Select(i => KvTag("B" + i, "K" + i, "dbo.Second")))
+            .ToArray();
+
+        var plans = SqlGroupPlanner.Plan(tags, new BindCeilingDialect(2));
+
+        plans.Count.ShouldBe(4);
+        // Group order still follows the input order of each group's first member (dbo.First, then dbo.Second),
+        // and a group's own chunks stay adjacent and in key order.
+        plans.Select(p => p.Members[0].Table).ShouldBe(["dbo.First", "dbo.First", "dbo.Second", "dbo.Second"]);
+    }
+
+    /// <summary>A wide-row plan binds at most one value, so a ceiling of 1 must not disturb it.</summary>
+    [Fact]
+    public void AWideRowPlan_isUnaffectedByTheCeiling()
+    {
+        var tag = new SqlTagDefinition("A", SqlTagModel.WideRow, "dbo.LatestStatus",
+            ColumnName: "oven_temp", RowSelectorColumn: "station_id", RowSelectorValue: "7");
+
+        var plans = SqlGroupPlanner.Plan([tag], new BindCeilingDialect(1));
+
+        plans.Count.ShouldBe(1);
+        plans[0].Parameters.Count.ShouldBe(1);
+    }
+
+    /// <summary>
+    ///     A dialect reporting a ceiling below 1 cannot be chunked to — it must fail loudly rather than loop
+    ///     forever or emit an empty <c>IN ()</c> list, which is a syntax error on every backend.
+    /// </summary>
+    [Theory]
+    [InlineData(0)]
+    [InlineData(-1)]
+    public void ADialectReportingANonPositiveCeiling_isRejected(int ceiling)
+        => Should.Throw<ArgumentException>(
+            () => SqlGroupPlanner.Plan([KvTag("A", "K1")], new BindCeilingDialect(ceiling)));
+
+    // ---- the whole read path, over a real database ----
+
+    /// <summary>
+    ///     Drives <see cref="SqlPollReader"/> over the SQLite fixture with a ceiling of 1, so three keys become
+    ///     three plans, and asserts every tag still reads its own value. This is what proves several plans per
+    ///     group need <b>no reader change</b> — the claim the chunking design rests on.
+    /// </summary>
+    [Fact]
+    public async Task ThePollReader_readsEveryTagCorrectly_whenOneGroupIsSplitAcrossSeveralPlans()
+    {
+        using var fixture = new SqlitePollFixture();
+        SqlTagDefinition[] tags =
+        [
+            Fixture("Speed", SqlitePollFixture.PresentKey),
+            Fixture("Pressure", SqlitePollFixture.SecondPresentKey),
+            Fixture("Missing", SqlitePollFixture.AbsentKey),
+        ];
+
+        var reader = new SqlPollReader(
+            fixture.Factory, fixture.ConnectionString, new BindCeilingDialect(1),
+            commandTimeout: TimeSpan.FromSeconds(10), operationTimeout: TimeSpan.FromSeconds(15),
+            maxConcurrentGroups: 4, nullIsBad: false,
+            resolve: reference => tags.FirstOrDefault(t => t.Name == reference));
+
+        var snapshots = await reader.ReadAsync(["Speed", "Pressure", "Missing"], CancellationToken.None);
+
+        snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
+        snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
+        snapshots[1].Value.ShouldBe(SqlitePollFixture.SecondPresentValue);
+        snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.Good);
+        snapshots[2].StatusCode.ShouldBe(SqlStatusCodes.BadNoData);
+    }
+
+    private static SqlTagDefinition Fixture(string name, string keyValue)
+        => new(name, SqlTagModel.KeyValue, SqlitePollFixture.KeyValueTable,
+            KeyColumn: SqlitePollFixture.KeyColumn, KeyValue: keyValue,
+            ValueColumn: SqlitePollFixture.ValueColumn,
+            TimestampColumn: SqlitePollFixture.TimestampColumn);
+
+    /// <summary>
+    ///     A <b>test-only</b> dialect that reports a caller-chosen <see cref="ISqlDialect.MaxBindParameters"/>
+    ///     and delegates everything else to <see cref="SqliteDialect"/> — so a chunked plan can be asserted on
+    ///     at a readable size <em>and</em> actually executed against the fixture database.
+    ///     <para>A decorator rather than a subclass, for the reason the other fakes here give:
+    ///     <see cref="SqliteDialect"/> is sealed, and <c>new</c>-hiding a member would leave interface dispatch
+    ///     calling the original, so the substitution would silently not happen.</para>
+    /// </summary>
+    private sealed class BindCeilingDialect(int maxBindParameters) : ISqlDialect
+    {
+        private static readonly SqliteDialect Inner = new();
+
+        public int MaxBindParameters { get; } = maxBindParameters;
+
+        public SqlProvider Provider => Inner.Provider;
+
+        public DbProviderFactory Factory => Inner.Factory;
+
+        public string LivenessSql => Inner.LivenessSql;
+
+        public string SingleRowLimitPrefix => Inner.SingleRowLimitPrefix;
+
+        public string SingleRowLimitSuffix => Inner.SingleRowLimitSuffix;
+
+        public string ListSchemasSql => Inner.ListSchemasSql;
+
+        public string DefaultSchemaSql => Inner.DefaultSchemaSql;
+
+        public string ListTablesSql => Inner.ListTablesSql;
+
+        public string ListColumnsSql => Inner.ListColumnsSql;
+
+        public string ParameterMarker(string baseName) => Inner.ParameterMarker(baseName);
+
+        public string ParameterBindName(string baseName) => Inner.ParameterBindName(baseName);
+
+        public void ConfigureCommand(DbCommand command) => Inner.ConfigureCommand(command);
+
+        public string QuoteIdentifier(string ident) => Inner.QuoteIdentifier(ident);
+
+        public DriverDataType MapColumnType(string sqlDataType) => Inner.MapColumnType(sqlDataType);
+    }
+}

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlServerDialectTests.cs

@@ -84,6 +84,18 @@ public class SqlServerDialectTests
             .ShouldBe("SELECT TOP 1 [a]");
     }
 
+    /// <summary>
+    ///     2098, not 2100: <c>Microsoft.Data.SqlClient</c> routes a parameterized command through
+    ///     <c>sp_executesql</c>, whose statement text and parameter-declaration arguments consume two of the
+    ///     server's 2100. A ceiling of 2100 would be rejected by a caller that filled it exactly.
+    /// </summary>
+    [Fact]
+    public void MaxBindParameters_leavesRoomForSpExecutesqlsOwnTwoParameters()
+    {
+        new SqlServerDialect().MaxBindParameters.ShouldBe(2098);
+        new SqlServerDialect().MaxBindParameters.ShouldBeLessThan(2100);
+    }
+
     // ---- extra guards on the injection boundary (beyond the plan's golden set) ----
 
     [Fact]

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqliteDialect.cs

@@ -64,6 +64,13 @@ public sealed class SqliteDialect : ISqlDialect
         // Intentionally empty: see the summary.
     }
 
+    /// <summary>
+    ///     999 — <c>SQLITE_MAX_VARIABLE_NUMBER</c>'s historical default, and still the safe floor: SQLite
+    ///     raised it to 32766 in 3.32, but the value is a compile-time setting of whatever native library the
+    ///     bundle happens to carry, so the lower number is the one that always executes.
+    /// </summary>
+    public int MaxBindParameters => 999;
+
     /// <inheritdoc/>
     public string LivenessSql => "SELECT 1";