From 0c93f25e3a4be7a6a8159f6a0baa9b243fb95fe7 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sat, 18 Jul 2026 21:26:54 -0500 Subject: [PATCH 1/2] chore: bump Weasel family 9.17.0 -> 9.18.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Weasel 9.18.0 ships weasel#362 (PR weasel#374) — ManagedTenantPartitions parity with the PostgreSQL ManagedListPartitions: TenantDropBehavior (RetainData/DeleteData purge-before-merge), AllowOrdinalSharing + explicit-ordinal add overloads (tenant bucketing under the 15,000-partition ceiling), MigrateAllTablesAsync new-table back-fill, and TenantPartitionAddResult / TablePartitionStatus batch status reporting. The prerequisite for polecat#335. Co-Authored-By: Claude Fable 5 --- Directory.Packages.props | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index 84ebd9e..3185b2e 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -60,10 +60,16 @@ databases and retries transient connection failures, with SqlServerMigrator now overriding ReleaseConnectionPoolAsync (SqlConnection.ClearPool) and IsTransientConnectionFailure over the SQL Server/Azure SQL resource-limit error - set. Polecat picks this up through the migrator base class — no code change. --> - - - + set. Polecat picks this up through the migrator base class — no code change. + Weasel 9.18.0: weasel#362/#374 — ManagedTenantPartitions parity with the + PostgreSQL ManagedListPartitions: TenantDropBehavior (RetainData/DeleteData), + AllowOrdinalSharing + explicit-ordinal add overloads (tenant bucketing), + MigrateAllTablesAsync new-table back-fill, and TenantPartitionAddResult / + TablePartitionStatus batch status reporting. Consumed by Polecat #335 + (per-tenant partitioning parity for documents + streams). --> + + + From 25835c7714d6aeb779f23542db54b4ae81c51a49 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sat, 18 Jul 2026 21:27:10 -0500 Subject: [PATCH 2/2] feat(#335): per-tenant managed partitioning parity with Marten (documents + streams) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Brings Polecat's managed per-tenant partitioning to full Marten equivalence on Weasel 9.18.0: - pc_streams now partitions alongside pc_events under UseTenantPartitionedEvents (mt_streams parity): tenant_ordinal joins the PK after (tenant_id, id) so readers keep their prefix seek, the stream INSERT stamps the planner-resolved ordinal, and the version UPDATE partition-eliminates on it. - Conjoined document tables can be tenant-partitioned: the DocumentTable "single-tenant only" restriction is lifted for the managed strategy — tables gain a tenant_ordinal PK column driven by the one pc_tenant_partitions registry per database. Every write path (MERGE upsert/insert/overwrite, update-only, version-checked bulk) resolves the ordinal SERVER-SIDE by joining the registry into the MERGE source, so parameter slots are unchanged and cross-process ordinal state can never mis-route rows; the ON clause carries tenant_ordinal for partition elimination. - Policy APIs: Policies.AllDocumentsAreMultiTenantedWithPartitioning() (forces conjoined tenancy) and Policies.PartitionMultiTenantedDocumentsUsingPolecatManagement(), with a per-type ForDocument(p => p.DisableTenantPartitioning) escape hatch; the daemon's DeadLetterEvent document is always excluded (Marten parity). - Runtime tenant onboarding: store.Advanced.AddPolecatManagedTenantsAsync (batch add with TablePartitionStatus[] reporting + explicit-ordinal bucketing overload) and RemovePolecatManagedTenantsAsync with TenantDropBehavior — DeleteData purges the tenants' rows before MERGE RANGE (weasel#362 item 1), fully-released ordinals drop their pc_events_sequence_{ordinal}, and in-process caches are evicted so a removed tenant can be re-onboarded. - Tenants are provisioned lazily on first write everywhere: the session flush pipeline, the daemon's projection batch, and both bulk-insert surfaces resolve ordinals through the new shared TenantPartitionOrdinalRegistry (TenantEventSequenceRegistry now composes it, so events/streams/documents share one tenant -> ordinal cache). Table creation hydrates the manager first so tables created after tenants exist bake the full boundary set. No reserved default-tenant partition is needed: Polecat has no global projections, and integer ordinals sidestep Marten's suffix-identifier constraint entirely. 15 new integration tests (streams schema/placement/update path; document schema shape, ordinal stamping, ForTenant, bulk, opt-out, config guards, add/remove with both drop behaviors, shared-registry coherence across pc_events/pc_streams/documents); docs updated. Closes #335. Co-Authored-By: Claude Fable 5 --- docs/documents/partitioning.md | 67 ++- docs/events/multitenancy.md | 19 +- .../tenant_partitioned_streams_tests.cs | 139 ++++++ .../Harness/TenantPartitioningCollection.cs | 25 +- src/Polecat.Tests/Harness/TestSchema.cs | 84 ++++ .../tenant_partitioned_documents_tests.cs | 397 ++++++++++++++++++ src/Polecat/AdvancedOperations.cs | 187 +++++++++ .../Events/Daemon/PolecatProjectionBatch.cs | 13 + src/Polecat/Events/EventGraph.cs | 47 ++- .../Events/Schema/EventStoreFeatureSchema.cs | 4 +- src/Polecat/Events/Schema/StreamsTable.cs | 12 + .../Schema/TenantEventSequenceRegistry.cs | 33 +- .../Schema/TenantPartitionOrdinalRegistry.cs | 64 +++ .../Storage/SqlServerEventStoreDialect.cs | 43 +- .../Internal/DocumentProviderRegistry.cs | 15 +- src/Polecat/Internal/DocumentSessionBase.cs | 14 + src/Polecat/Internal/DocumentTableEnsurer.cs | 16 + .../Operations/ClosedShapeOperationAdapter.cs | 7 + .../ClosedShape/PolecatDocumentStorage.cs | 22 +- src/Polecat/Storage/DocumentMapping.cs | 13 + src/Polecat/Storage/DocumentTable.cs | 29 +- ...lServerDocumentStorageDescriptorBuilder.cs | 44 +- src/Polecat/StoreOptions.cs | 5 +- src/Polecat/StorePolicies.cs | 73 ++++ 24 files changed, 1325 insertions(+), 47 deletions(-) create mode 100644 src/Polecat.Tests/Events/tenant_partitioned_streams_tests.cs create mode 100644 src/Polecat.Tests/Harness/TestSchema.cs create mode 100644 src/Polecat.Tests/Storage/tenant_partitioned_documents_tests.cs create mode 100644 src/Polecat/Events/Schema/TenantPartitionOrdinalRegistry.cs diff --git a/docs/documents/partitioning.md b/docs/documents/partitioning.md index 232290e..b7f8053 100644 --- a/docs/documents/partitioning.md +++ b/docs/documents/partitioning.md @@ -59,8 +59,71 @@ column/type is reported as a rebuild rather than performed silently. ## Limitations -- Supported for **single-tenant** document tables only; combining partitioning with conjoined - multi-tenancy throws at start-up. +- Supported for **single-tenant** document tables only; combining member RANGE partitioning with + conjoined multi-tenancy throws at start-up. Conjoined document tables can instead use the managed + per-tenant partitioning below. - Dropping aged partitions for retention (`SWITCH`/`MERGE RANGE`) and externally-managed partition rolling are not yet wired into the document API — for now, manage those out of band, or keep pruning with a predicate delete. + +## Managed per-tenant partitioning (#335) + +Conjoined multi-tenanted document tables can be physically partitioned **per tenant** through the +store's shared managed tenant partitioning — the SQL Server counterpart of Marten's +`AllDocumentsAreMultiTenantedWithPartitioning` + `PartitionMultiTenantedDocumentsUsingMartenManagement`: + +```csharp +var store = DocumentStore.For(opts => +{ + opts.Connection("..."); + + // Make every document conjoined multi-tenanted AND tenant-partitioned: + opts.Policies.AllDocumentsAreMultiTenantedWithPartitioning(); + + // — or, when the store is already conjoined, enable just the partitioning: + // opts.Events.TenancyStyle = TenancyStyle.Conjoined; + // opts.Policies.PartitionMultiTenantedDocumentsUsingPolecatManagement(); + + // Per-type escape hatch (the [SingleTenanted]/DisablePartitioningIfAny analogue): + opts.Policies.ForDocument(p => p.DisableTenantPartitioning = true); +}); +``` + +Every document table then carries a `tenant_ordinal int` primary-key column and is `RANGE RIGHT` +partitioned on it, driven by the **one `pc_tenant_partitions` registry per database** — the same +registry, ordinals, and physical layout as +[per-tenant event partitioning](/events/multitenancy#per-tenant-event-partitioning), so a store using +both keeps a single coherent tenant → ordinal map across `pc_events`, `pc_streams`, and every +document table. The ordinal is resolved server-side from the registry on every write (upsert, insert, +update, bulk insert, and projection writes), so cross-process ordinal drift cannot mis-route rows. + +Tenants are onboarded **lazily on first write** (matching the event-append behavior), or explicitly +with per-table status reporting: + +```csharp +// Onboard tenants up front — returns Weasel TablePartitionStatus[] per managed table: +var statuses = await store.Advanced.AddPolecatManagedTenantsAsync(ct, "tenant-a", "tenant-b"); + +// Tenant bucketing (Weasel 9.18.0): map many small tenants onto one shared partition ordinal. +// Requires ManagedTenantPartitions.AllowOrdinalSharing: +await store.Advanced.AddPolecatManagedTenantsAsync( + new Dictionary { ["small-1"] = 1, ["small-2"] = 1 }, ct); + +// Remove a tenant. SQL Server's MERGE RANGE alone would retain the rows — +// TenantDropBehavior.DeleteData physically purges the tenant's rows from every managed +// table first (PostgreSQL managed-drop parity): +await store.Advanced.RemovePolecatManagedTenantsAsync( + ["tenant-b"], TenantDropBehavior.DeleteData, ct); +``` + +Notes: + +- Requires `TenancyStyle.Conjoined` (asserted at store construction) and cannot be combined with + member `PartitionByRange` on the same document type — a SQL Server table supports only one + partition scheme. +- The registry and partition function/scheme are database-global objects: one tenant-partitioned + store per database. +- `AddPolecatManagedTenantsAsync` splits the tables of document types **registered** with the store + (`opts.Schema.For()` or prior use); a table created later bakes the full boundary set at + creation. +- The daemon's dead-letter document (`pc_doc_deadletterevent`) is always excluded, mirroring Marten. diff --git a/docs/events/multitenancy.md b/docs/events/multitenancy.md index 756b506..e53d140 100644 --- a/docs/events/multitenancy.md +++ b/docs/events/multitenancy.md @@ -100,9 +100,11 @@ When enabled, Polecat: `pc_events_sequence_{ordinal}` object (created on demand) via `NEXT VALUE FOR`, rather than a single global `IDENTITY`. `seq_id` is therefore unique only *within* a tenant, so the `pc_events` primary key becomes composite `(tenant_ordinal, seq_id)`. -* **Physically partitions `pc_events` by tenant** — the table is `RANGE RIGHT` partitioned on the - tenant `ordinal`, and a new partition is split in as each tenant registers. A tenant's events live - in their own physical partition, so per-tenant scans and rebuilds touch only that partition. +* **Physically partitions `pc_events` and `pc_streams` by tenant** — both tables are `RANGE RIGHT` + partitioned on the tenant `ordinal`, and a new partition is split in as each tenant registers. A + tenant's events and stream rows live in their own physical partitions, so per-tenant scans and + rebuilds touch only those partitions (Marten parity: `mt_streams` rides `mt_events`' tenant + partitioning). ```cs // Each tenant's seq_id starts at 1 and advances independently @@ -145,7 +147,12 @@ path byte-for-byte. The flag **requires** `TenancyStyle.Conjoined` (there is not otherwise) and is currently incompatible with `UseArchivedStreamPartitioning` — a SQL Server table supports only one partition scheme; both raise an error at store construction. -Physical partitioning applies to `pc_events` (the table that drives the bounded per-tenant scan); -`pc_streams` is accessed by point lookup and is left unpartitioned. The partition function/scheme are -database-global objects, so a single database should host one tenant-partitioned event store. +Physical partitioning applies to `pc_events` and (since #335) `pc_streams`. The partition +function/scheme are database-global objects, so a single database should host one tenant-partitioned +event store. + +Document tables can join the same managed per-tenant partitioning — including runtime tenant +onboarding/removal via `store.Advanced.AddPolecatManagedTenantsAsync` / +`RemovePolecatManagedTenantsAsync` — see +[Document multi-tenancy partitioning](/documents/partitioning#managed-per-tenant-partitioning-335). ::: diff --git a/src/Polecat.Tests/Events/tenant_partitioned_streams_tests.cs b/src/Polecat.Tests/Events/tenant_partitioned_streams_tests.cs new file mode 100644 index 0000000..60721b0 --- /dev/null +++ b/src/Polecat.Tests/Events/tenant_partitioned_streams_tests.cs @@ -0,0 +1,139 @@ +using JasperFx; +using Microsoft.Data.SqlClient; +using Polecat.Tests.Harness; +using Polecat.TestUtils; + +namespace Polecat.Tests.Events; + +/// +/// Covers #335 scope 2 — pc_streams is partitioned alongside pc_events under +/// UseTenantPartitionedEvents (Marten parity: mt_streams rides mt_events' tenant partitioning): +/// schema shape, physical partition placement of stream rows, and the stream-version update path +/// against the partitioned table. +/// +[Collection("tenant-partitioning")] +public class tenant_partitioned_streams_tests : IAsyncLifetime +{ + private const string Schema = "pt_streams"; + + public async Task InitializeAsync() + { + await TestSchema.DropSchemaTablesAsync(Schema); + await PartitionTestCleanup.DropEventsPartitionObjectsAsync(); + await TestSchema.DropSequencesAsync(Schema); + } + + public Task DisposeAsync() => Task.CompletedTask; + + private static DocumentStore CreateStore() + { + return DocumentStore.For(opts => + { + opts.ConnectionString = ConnectionSource.ConnectionString; + opts.DatabaseSchemaName = Schema; + opts.AutoCreateSchemaObjects = AutoCreate.All; + opts.UseNativeJsonType = ConnectionSource.SupportsNativeJson; + opts.Events.TenancyStyle = TenancyStyle.Conjoined; + opts.EventGraph.UseTenantPartitionedEvents = true; + }); + } + + [Fact] + public async Task streams_table_is_partitioned_alongside_events() + { + using var store = CreateStore(); + await store.Database.ApplyAllConfiguredChangesToDatabaseAsync(); + + // pc_streams carries the tenant_ordinal partition column and sits on its own + // partition scheme, exactly like pc_events. + (await TestSchema.ColumnExistsAsync(Schema, "pc_streams", "tenant_ordinal")).ShouldBeTrue(); + (await TableIsOnPartitionSchemeAsync("pc_streams", "ps_pc_streams_tenant_ordinal")).ShouldBeTrue(); + (await TableIsOnPartitionSchemeAsync("pc_events", "ps_pc_events_tenant_ordinal")).ShouldBeTrue(); + } + + [Fact] + public async Task stream_rows_land_in_their_tenant_partition() + { + using var store = CreateStore(); + await store.Database.ApplyAllConfiguredChangesToDatabaseAsync(); + + await using (var red = store.LightweightSession(new SessionOptions { TenantId = "Red" })) + { + red.Events.StartStream(Guid.NewGuid(), new QuestStarted("Red"), new MonsterSlain("a", 1)); + await red.SaveChangesAsync(); + } + + await using (var blue = store.LightweightSession(new SessionOptions { TenantId = "Blue" })) + { + blue.Events.StartStream(Guid.NewGuid(), new QuestStarted("Blue")); + await blue.SaveChangesAsync(); + } + + // Stream rows carry their tenant's registry ordinal and land in distinct physical + // partitions of pc_streams. + var rows = await TestSchema.QueryAsync($""" + SELECT $PARTITION.pf_pc_streams_tenant_ordinal(st.tenant_ordinal) AS p, st.tenant_id, + st.tenant_ordinal, tp.ordinal + FROM [{Schema}].[pc_streams] st + JOIN [{Schema}].[pc_tenant_partitions] tp ON tp.tenant_id = st.tenant_id + ORDER BY st.tenant_id + """); + + rows.Count.ShouldBe(2); + rows.Select(r => r[0]).Distinct().Count().ShouldBe(2); // distinct physical partitions + foreach (var row in rows) + { + ((int)row[2]).ShouldBe((int)row[3]); // row ordinal == the tenant's registry ordinal + } + } + + [Fact] + public async Task appending_to_an_existing_stream_updates_the_partitioned_stream_row() + { + using var store = CreateStore(); + await store.Database.ApplyAllConfiguredChangesToDatabaseAsync(); + + var stream = Guid.NewGuid(); + + await using (var session = store.LightweightSession(new SessionOptions { TenantId = "Red" })) + { + session.Events.StartStream(stream, new QuestStarted("Quest")); + await session.SaveChangesAsync(); + } + + // Second append hits the UPDATE path against the partitioned pc_streams (with the + // tenant_ordinal partition-elimination predicate). + await using (var session = store.LightweightSession(new SessionOptions { TenantId = "Red" })) + { + session.Events.Append(stream, new MembersJoined(1, "Town", ["Hero"]), new MonsterSlain("b", 2)); + await session.SaveChangesAsync(); + } + + await using var query = store.QuerySession(new SessionOptions { TenantId = "Red" }); + var state = await query.Events.FetchStreamStateAsync(stream); + state.ShouldNotBeNull(); + state.Version.ShouldBe(3); + + (await query.Events.FetchStreamAsync(stream)).Count.ShouldBe(3); + } + + private static async Task TableIsOnPartitionSchemeAsync(string table, string scheme) + { + await using var conn = new SqlConnection(ConnectionSource.ConnectionString); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = """ + SELECT COUNT(*) + FROM sys.tables t + JOIN sys.schemas s ON t.schema_id = s.schema_id + JOIN sys.indexes i ON i.object_id = t.object_id AND i.index_id IN (0, 1) + JOIN sys.data_spaces ds ON i.data_space_id = ds.data_space_id + WHERE s.name = @schema AND t.name = @table AND ds.name = @scheme + """; + cmd.Parameters.AddWithValue("@schema", Schema); + cmd.Parameters.AddWithValue("@table", table); + cmd.Parameters.AddWithValue("@scheme", scheme); + var count = (int)(await cmd.ExecuteScalarAsync())!; + return count == 1; + } +} diff --git a/src/Polecat.Tests/Harness/TenantPartitioningCollection.cs b/src/Polecat.Tests/Harness/TenantPartitioningCollection.cs index b15e755..a85f098 100644 --- a/src/Polecat.Tests/Harness/TenantPartitioningCollection.cs +++ b/src/Polecat.Tests/Harness/TenantPartitioningCollection.cs @@ -14,8 +14,9 @@ namespace Polecat.Tests.Harness; public class TenantPartitioningCollection; /// -/// Drops the database-global partition function/scheme that the managed tenant partitioning leaves -/// behind for a tenant-partitioned pc_events — they outlive a schema/table drop and must be +/// Drops the database-global partition functions/schemes that the managed tenant partitioning +/// leaves behind for tenant-partitioned tables (pc_events, pc_streams since #335, +/// and any tenant-partitioned pc_doc_*) — they outlive a schema/table drop and must be /// removed explicitly so each test starts clean. /// public static class PartitionTestCleanup @@ -26,21 +27,27 @@ public static async Task DropEventsPartitionObjectsAsync() await conn.OpenAsync(); await using var cmd = conn.CreateCommand(); cmd.CommandText = """ - -- The scheme is shared across every test schema's pc_events, so drop ALL tables sitting on - -- it (in any schema) before the scheme/function can be removed. + -- The schemes are shared across every test schema's tables, so drop ALL tables sitting + -- on any pc_* tenant-ordinal scheme (in any schema) before the schemes/functions can be + -- removed. DECLARE @sql nvarchar(max) = N''; SELECT @sql = @sql + 'DROP TABLE [' + s.name + '].[' + t.name + '];' FROM sys.tables t JOIN sys.schemas s ON t.schema_id = s.schema_id JOIN sys.indexes i ON i.object_id = t.object_id AND i.index_id IN (0, 1) JOIN sys.data_spaces ds ON i.data_space_id = ds.data_space_id - WHERE ds.name = 'ps_pc_events_tenant_ordinal'; + WHERE ds.name LIKE 'ps[_]pc[_]%[_]tenant[_]ordinal'; IF @sql <> N'' EXEC sp_executesql @sql; - IF EXISTS (SELECT 1 FROM sys.partition_schemes WHERE name = 'ps_pc_events_tenant_ordinal') - DROP PARTITION SCHEME ps_pc_events_tenant_ordinal; - IF EXISTS (SELECT 1 FROM sys.partition_functions WHERE name = 'pf_pc_events_tenant_ordinal') - DROP PARTITION FUNCTION pf_pc_events_tenant_ordinal; + SET @sql = N''; + SELECT @sql = @sql + 'DROP PARTITION SCHEME [' + name + '];' + FROM sys.partition_schemes WHERE name LIKE 'ps[_]pc[_]%[_]tenant[_]ordinal'; + IF @sql <> N'' EXEC sp_executesql @sql; + + SET @sql = N''; + SELECT @sql = @sql + 'DROP PARTITION FUNCTION [' + name + '];' + FROM sys.partition_functions WHERE name LIKE 'pf[_]pc[_]%[_]tenant[_]ordinal'; + IF @sql <> N'' EXEC sp_executesql @sql; """; await cmd.ExecuteNonQueryAsync(); } diff --git a/src/Polecat.Tests/Harness/TestSchema.cs b/src/Polecat.Tests/Harness/TestSchema.cs new file mode 100644 index 0000000..b807141 --- /dev/null +++ b/src/Polecat.Tests/Harness/TestSchema.cs @@ -0,0 +1,84 @@ +using Microsoft.Data.SqlClient; +using Polecat.TestUtils; + +namespace Polecat.Tests.Harness; + +/// +/// Shared raw-SQL schema helpers for the tenant-partitioning test suites (#335): drop a test +/// schema's tables (FKs first), drop its sequences, and small introspection/query utilities. +/// +public static class TestSchema +{ + public static async Task DropSchemaTablesAsync(string schema) + { + await using var conn = new SqlConnection(ConnectionSource.ConnectionString); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = """ + DECLARE @sql nvarchar(max) = N''; + SELECT @sql = @sql + 'ALTER TABLE [' + s.name + '].[' + t.name + '] DROP CONSTRAINT [' + fk.name + '];' + FROM sys.foreign_keys fk + JOIN sys.tables t ON fk.parent_object_id = t.object_id + JOIN sys.schemas s ON t.schema_id = s.schema_id + WHERE s.name = @schema; + + SELECT @sql = @sql + 'DROP TABLE [' + s.name + '].[' + t.name + '];' + FROM sys.tables t + JOIN sys.schemas s ON t.schema_id = s.schema_id + WHERE s.name = @schema; + EXEC sp_executesql @sql; + """; + cmd.Parameters.AddWithValue("@schema", schema); + await cmd.ExecuteNonQueryAsync(); + } + + public static async Task DropSequencesAsync(string schema) + { + await using var conn = new SqlConnection(ConnectionSource.ConnectionString); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = """ + DECLARE @sql nvarchar(max) = N''; + SELECT @sql = @sql + 'DROP SEQUENCE [' + s.name + '].[' + sq.name + '];' + FROM sys.sequences sq + JOIN sys.schemas s ON sq.schema_id = s.schema_id + WHERE s.name = @schema; + EXEC sp_executesql @sql; + """; + cmd.Parameters.AddWithValue("@schema", schema); + await cmd.ExecuteNonQueryAsync(); + } + + public static async Task ColumnExistsAsync(string schema, string table, string column) + { + await using var conn = new SqlConnection(ConnectionSource.ConnectionString); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = """ + SELECT COUNT(*) FROM sys.columns c + WHERE c.object_id = OBJECT_ID(@table) AND c.name = @column + """; + cmd.Parameters.AddWithValue("@table", $"[{schema}].[{table}]"); + cmd.Parameters.AddWithValue("@column", column); + var count = (int)(await cmd.ExecuteScalarAsync())!; + return count == 1; + } + + public static async Task> QueryAsync(string sql) + { + await using var conn = new SqlConnection(ConnectionSource.ConnectionString); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = sql; + var rows = new List(); + await using var reader = await cmd.ExecuteReaderAsync(); + while (await reader.ReadAsync()) + { + var values = new object[reader.FieldCount]; + reader.GetValues(values); + rows.Add(values); + } + + return rows; + } +} diff --git a/src/Polecat.Tests/Storage/tenant_partitioned_documents_tests.cs b/src/Polecat.Tests/Storage/tenant_partitioned_documents_tests.cs new file mode 100644 index 0000000..d3806f3 --- /dev/null +++ b/src/Polecat.Tests/Storage/tenant_partitioned_documents_tests.cs @@ -0,0 +1,397 @@ +using JasperFx; +using Polecat.Linq; +using Polecat.Tests.Harness; +using Polecat.TestUtils; +using Weasel.Core; +using Weasel.SqlServer.Tables.Partitioning; + +namespace Polecat.Tests.Storage; + +public class PartitionedTenantDoc +{ + public Guid Id { get; set; } + public string Name { get; set; } = string.Empty; +} + +public class OptedOutTenantDoc +{ + public Guid Id { get; set; } + public string Name { get; set; } = string.Empty; +} + +/// +/// Covers #335 scopes 1 + 3 + 4 — managed per-tenant partitioning of conjoined document tables: +/// the AllDocumentsAreMultiTenantedWithPartitioning / PartitionMultiTenantedDocumentsUsingPolecatManagement +/// policies, the tenant_ordinal schema shape + server-side ordinal resolution on every write path +/// (upsert, insert, update, bulk), per-type opt-out, config guards, and the +/// AddPolecatManagedTenantsAsync / RemovePolecatManagedTenantsAsync runtime onboarding APIs with +/// TenantDropBehavior semantics. +/// +[Collection("tenant-partitioning")] +public class tenant_partitioned_documents_tests : IAsyncLifetime +{ + private const string Schema = "pt_docs"; + + public async Task InitializeAsync() + { + await TestSchema.DropSchemaTablesAsync(Schema); + await PartitionTestCleanup.DropEventsPartitionObjectsAsync(); + await TestSchema.DropSequencesAsync(Schema); + } + + public Task DisposeAsync() => Task.CompletedTask; + + private static DocumentStore CreateStore(Action? configure = null) + { + return DocumentStore.For(opts => + { + opts.ConnectionString = ConnectionSource.ConnectionString; + opts.DatabaseSchemaName = Schema; + opts.AutoCreateSchemaObjects = AutoCreate.All; + opts.UseNativeJsonType = ConnectionSource.SupportsNativeJson; + opts.Policies.AllDocumentsAreMultiTenantedWithPartitioning(); + configure?.Invoke(opts); + }); + } + + [Fact] + public void all_documents_policy_forces_conjoined_tenancy() + { + using var store = CreateStore(); + store.Options.Events.TenancyStyle.ShouldBe(TenancyStyle.Conjoined); + } + + [Fact] + public void management_only_policy_requires_conjoined_tenancy() + { + var ex = Should.Throw(() => DocumentStore.For(opts => + { + opts.ConnectionString = ConnectionSource.ConnectionString; + // TenancyStyle left at the default (Single) + opts.Policies.PartitionMultiTenantedDocumentsUsingPolecatManagement(); + })); + + ex.Message.ShouldContain("Conjoined"); + } + + [Fact] + public async Task document_table_is_partitioned_and_writes_carry_the_tenant_ordinal() + { + using var store = CreateStore(); + + await using (var red = store.LightweightSession(new SessionOptions { TenantId = "Red" })) + { + red.Store(new PartitionedTenantDoc { Id = Guid.NewGuid(), Name = "red-1" }); + red.Store(new PartitionedTenantDoc { Id = Guid.NewGuid(), Name = "red-2" }); + await red.SaveChangesAsync(); + } + + await using (var blue = store.LightweightSession(new SessionOptions { TenantId = "Blue" })) + { + blue.Store(new PartitionedTenantDoc { Id = Guid.NewGuid(), Name = "blue-1" }); + await blue.SaveChangesAsync(); + } + + // The table carries tenant_ordinal, and every row's ordinal matches its tenant's + // registry assignment — SQL Server's $PARTITION proves physical placement (RANGE RIGHT: + // ordinal N lands in partition N + 1). + (await TestSchema.ColumnExistsAsync(Schema, "pc_doc_partitionedtenantdoc", "tenant_ordinal")) + .ShouldBeTrue(); + + var rows = await TestSchema.QueryAsync($""" + SELECT d.tenant_id, d.tenant_ordinal, tp.ordinal, + $PARTITION.pf_pc_doc_partitionedtenantdoc_tenant_ordinal(d.tenant_ordinal) AS p, + COUNT(*) AS c + FROM [{Schema}].[pc_doc_partitionedtenantdoc] d + JOIN [{Schema}].[pc_tenant_partitions] tp ON tp.tenant_id = d.tenant_id + GROUP BY d.tenant_id, d.tenant_ordinal, tp.ordinal, + $PARTITION.pf_pc_doc_partitionedtenantdoc_tenant_ordinal(d.tenant_ordinal) + ORDER BY d.tenant_id + """); + + rows.Count.ShouldBe(2); + rows.Select(r => r[3]).Distinct().Count().ShouldBe(2); // distinct physical partitions + foreach (var row in rows) + { + ((int)row[1]).ShouldBe((int)row[2]); // row ordinal == registry ordinal + } + + rows.Single(r => (string)r[0] == "Red")[4].ShouldBe(2); + rows.Single(r => (string)r[0] == "Blue")[4].ShouldBe(1); + } + + [Fact] + public async Task load_update_and_delete_round_trip_per_tenant() + { + using var store = CreateStore(); + var redId = Guid.NewGuid(); + var blueId = Guid.NewGuid(); + + await using (var red = store.LightweightSession(new SessionOptions { TenantId = "Red" })) + { + red.Store(new PartitionedTenantDoc { Id = redId, Name = "red" }); + await red.SaveChangesAsync(); + } + + await using (var blue = store.LightweightSession(new SessionOptions { TenantId = "Blue" })) + { + blue.Store(new PartitionedTenantDoc { Id = blueId, Name = "blue" }); + await blue.SaveChangesAsync(); + } + + // Cross-tenant isolation on loads. + await using (var redQuery = store.QuerySession(new SessionOptions { TenantId = "Red" })) + { + (await redQuery.LoadAsync(redId))!.Name.ShouldBe("red"); + (await redQuery.LoadAsync(blueId)).ShouldBeNull(); + } + + // Update through the MERGE matched branch (partition-eliminated by tenant_ordinal). + await using (var red = store.LightweightSession(new SessionOptions { TenantId = "Red" })) + { + red.Store(new PartitionedTenantDoc { Id = redId, Name = "red-updated" }); + await red.SaveChangesAsync(); + } + + // session.Update (the update-only SQL path). + await using (var red = store.LightweightSession(new SessionOptions { TenantId = "Red" })) + { + red.Update(new PartitionedTenantDoc { Id = redId, Name = "red-updated-again" }); + await red.SaveChangesAsync(); + } + + // session.Insert (the insert-only MERGE path). + var redSecondId = Guid.NewGuid(); + await using (var red = store.LightweightSession(new SessionOptions { TenantId = "Red" })) + { + red.Insert(new PartitionedTenantDoc { Id = redSecondId, Name = "red-inserted" }); + await red.SaveChangesAsync(); + } + + await using (var redQuery = store.QuerySession(new SessionOptions { TenantId = "Red" })) + { + (await redQuery.LoadAsync(redId))!.Name.ShouldBe("red-updated-again"); + (await redQuery.LoadAsync(redSecondId))!.Name.ShouldBe("red-inserted"); + } + + // Delete only touches the owning tenant. + await using (var red = store.LightweightSession(new SessionOptions { TenantId = "Red" })) + { + red.Delete(redId); + await red.SaveChangesAsync(); + } + + await using (var redQuery = store.QuerySession(new SessionOptions { TenantId = "Red" })) + { + (await redQuery.LoadAsync(redId)).ShouldBeNull(); + } + + await using (var blueQuery = store.QuerySession(new SessionOptions { TenantId = "Blue" })) + { + (await blueQuery.LoadAsync(blueId))!.Name.ShouldBe("blue"); + } + } + + [Fact] + public async Task for_tenant_writes_resolve_the_override_tenants_ordinal() + { + using var store = CreateStore(); + var greenId = Guid.NewGuid(); + + // A session for one tenant writing via ForTenant for another must provision + stamp the + // OVERRIDE tenant's ordinal. + await using (var session = store.LightweightSession(new SessionOptions { TenantId = "Red" })) + { + session.ForTenant("Green").Store(new PartitionedTenantDoc { Id = greenId, Name = "green" }); + await session.SaveChangesAsync(); + } + + var rows = await TestSchema.QueryAsync($""" + SELECT d.tenant_ordinal, tp.ordinal + FROM [{Schema}].[pc_doc_partitionedtenantdoc] d + JOIN [{Schema}].[pc_tenant_partitions] tp ON tp.tenant_id = 'Green' + WHERE d.tenant_id = 'Green' + """); + rows.Count.ShouldBe(1); + ((int)rows[0][0]).ShouldBe((int)rows[0][1]); + + await using var query = store.QuerySession(new SessionOptions { TenantId = "Green" }); + (await query.LoadAsync(greenId))!.Name.ShouldBe("green"); + } + + [Fact] + public async Task bulk_insert_provisions_and_stamps_the_tenant_ordinal() + { + using var store = CreateStore(); + + var docs = Enumerable.Range(1, 5) + .Select(i => new PartitionedTenantDoc { Id = Guid.NewGuid(), Name = $"bulk-{i}" }) + .ToArray(); + await store.Advanced.BulkInsertAsync(docs, BulkInsertMode.InsertsOnly, 2, "Bulky"); + + var rows = await TestSchema.QueryAsync($""" + SELECT COUNT(*) + FROM [{Schema}].[pc_doc_partitionedtenantdoc] d + JOIN [{Schema}].[pc_tenant_partitions] tp ON tp.tenant_id = d.tenant_id + WHERE d.tenant_id = 'Bulky' AND d.tenant_ordinal = tp.ordinal + """); + ((int)rows[0][0]).ShouldBe(5); + } + + [Fact] + public async Task per_type_opt_out_keeps_a_plain_conjoined_table() + { + using var store = CreateStore(opts => + opts.Policies.ForDocument(p => p.DisableTenantPartitioning = true)); + + await using (var session = store.LightweightSession(new SessionOptions { TenantId = "Red" })) + { + session.Store(new OptedOutTenantDoc { Id = Guid.NewGuid(), Name = "plain" }); + await session.SaveChangesAsync(); + } + + (await TestSchema.ColumnExistsAsync(Schema, "pc_doc_optedouttenantdoc", "tenant_ordinal")) + .ShouldBeFalse(); + } + + [Fact] + public void combining_with_range_partitioning_throws() + { + var ex = Should.Throw(() => CreateStore(opts => + opts.Schema.For() + .PartitionByRange(x => x.Id, Guid.NewGuid()))); + + ex.Message.ShouldContain("one partition scheme"); + } + + [Fact] + public async Task add_managed_tenants_reports_per_table_statuses_and_splits_registered_tables() + { + using var store = CreateStore(opts => opts.Schema.For()); + + // Materialize the registered document table (plus the shared registry) up front. + await store.Database.ApplyAllConfiguredChangesToDatabaseAsync(); + + var statuses = await store.Advanced.AddPolecatManagedTenantsAsync( + CancellationToken.None, "t1", "t2"); + + statuses.ShouldNotBeEmpty(); + statuses.ShouldContain(s => + s.Identifier.QualifiedName == $"{Schema}.pc_doc_partitionedtenantdoc" + && s.Status == PartitionMigrationStatus.Complete); + + // Idempotent — a second add of the same tenants emits no failures and keeps ordinals. + var again = await store.Advanced.AddPolecatManagedTenantsAsync(CancellationToken.None, "t1"); + again.ShouldAllBe(s => s.Status == PartitionMigrationStatus.Complete); + + var registry = await TestSchema.QueryAsync( + $"SELECT tenant_id, ordinal FROM [{Schema}].[pc_tenant_partitions] ORDER BY ordinal"); + registry.Count.ShouldBe(2); + + // Writes for an onboarded tenant flow straight through (cached or registry-resolved). + await using var session = store.LightweightSession(new SessionOptions { TenantId = "t1" }); + session.Store(new PartitionedTenantDoc { Id = Guid.NewGuid(), Name = "onboarded" }); + await session.SaveChangesAsync(); + } + + [Fact] + public async Task remove_managed_tenants_with_delete_data_purges_the_tenants_rows() + { + using var store = CreateStore(); + + await using (var session = store.LightweightSession(new SessionOptions { TenantId = "Doomed" })) + { + session.Store(new PartitionedTenantDoc { Id = Guid.NewGuid(), Name = "doomed-1" }); + session.Store(new PartitionedTenantDoc { Id = Guid.NewGuid(), Name = "doomed-2" }); + await session.SaveChangesAsync(); + } + + await using (var session = store.LightweightSession(new SessionOptions { TenantId = "Kept" })) + { + session.Store(new PartitionedTenantDoc { Id = Guid.NewGuid(), Name = "kept" }); + await session.SaveChangesAsync(); + } + + await store.Advanced.RemovePolecatManagedTenantsAsync( + ["Doomed"], TenantDropBehavior.DeleteData); + + // Doomed's rows are physically gone, Kept's remain; the registry row is dropped. + var rows = await TestSchema.QueryAsync( + $"SELECT tenant_id, COUNT(*) FROM [{Schema}].[pc_doc_partitionedtenantdoc] GROUP BY tenant_id"); + rows.Count.ShouldBe(1); + ((string)rows[0][0]).ShouldBe("Kept"); + + var registry = await TestSchema.QueryAsync( + $"SELECT tenant_id FROM [{Schema}].[pc_tenant_partitions]"); + registry.Count.ShouldBe(1); + ((string)registry[0][0]).ShouldBe("Kept"); + + // A removed tenant can be re-onboarded (fresh ordinal) and written again — the + // in-process caches were evicted. + await using (var session = store.LightweightSession(new SessionOptions { TenantId = "Doomed" })) + { + session.Store(new PartitionedTenantDoc { Id = Guid.NewGuid(), Name = "reborn" }); + await session.SaveChangesAsync(); + } + + await using var query = store.QuerySession(new SessionOptions { TenantId = "Doomed" }); + var reborn = await query.Query().ToListAsync(); + reborn.Count.ShouldBe(1); + reborn[0].Name.ShouldBe("reborn"); + } + + [Fact] + public async Task remove_managed_tenants_with_retain_data_keeps_the_rows() + { + using var store = CreateStore(); + + await using (var session = store.LightweightSession(new SessionOptions { TenantId = "Merged" })) + { + session.Store(new PartitionedTenantDoc { Id = Guid.NewGuid(), Name = "merged" }); + await session.SaveChangesAsync(); + } + + await store.Advanced.RemovePolecatManagedTenantsAsync(["Merged"]); + + // Historical merge-only semantics: the boundary is gone but the rows survive. + var rows = await TestSchema.QueryAsync( + $"SELECT COUNT(*) FROM [{Schema}].[pc_doc_partitionedtenantdoc] WHERE tenant_id = 'Merged'"); + ((int)rows[0][0]).ShouldBe(1); + + var registry = await TestSchema.QueryAsync( + $"SELECT COUNT(*) FROM [{Schema}].[pc_tenant_partitions] WHERE tenant_id = 'Merged'"); + ((int)registry[0][0]).ShouldBe(0); + } + + [Fact] + public async Task events_streams_and_documents_share_one_registry_and_ordinal() + { + using var store = CreateStore(opts => + { + opts.Events.TenancyStyle = TenancyStyle.Conjoined; + opts.EventGraph.UseTenantPartitionedEvents = true; + }); + + await using (var session = store.LightweightSession(new SessionOptions { TenantId = "Shared" })) + { + session.Events.StartStream(Guid.NewGuid(), new QuestStarted("Shared Quest")); + session.Store(new PartitionedTenantDoc { Id = Guid.NewGuid(), Name = "shared" }); + await session.SaveChangesAsync(); + } + + // One registry row; pc_events, pc_streams, and the document table all stamp the SAME ordinal. + var rows = await TestSchema.QueryAsync($""" + SELECT tp.ordinal, + (SELECT DISTINCT tenant_ordinal FROM [{Schema}].[pc_events] WHERE tenant_id = 'Shared'), + (SELECT DISTINCT tenant_ordinal FROM [{Schema}].[pc_streams] WHERE tenant_id = 'Shared'), + (SELECT DISTINCT tenant_ordinal FROM [{Schema}].[pc_doc_partitionedtenantdoc] WHERE tenant_id = 'Shared') + FROM [{Schema}].[pc_tenant_partitions] tp WHERE tp.tenant_id = 'Shared' + """); + + rows.Count.ShouldBe(1); + var ordinal = (int)rows[0][0]; + ((int)rows[0][1]).ShouldBe(ordinal); + ((int)rows[0][2]).ShouldBe(ordinal); + ((int)rows[0][3]).ShouldBe(ordinal); + } +} diff --git a/src/Polecat/AdvancedOperations.cs b/src/Polecat/AdvancedOperations.cs index 32c8645..b337dbb 100644 --- a/src/Polecat/AdvancedOperations.cs +++ b/src/Polecat/AdvancedOperations.cs @@ -9,8 +9,10 @@ using Polecat.Projections.Flattened; using Polecat.Schema.Identity.Sequences; using Polecat.Storage; +using Microsoft.Extensions.Logging.Abstractions; using Weasel.Core; using Weasel.SqlServer; +using Weasel.SqlServer.Tables.Partitioning; namespace Polecat; @@ -113,6 +115,13 @@ public async Task BulkInsertAsync(IReadOnlyCollection documents, BulkInser var ensurer = _store.ResolveTableEnsurer(tenantId); await ensurer.EnsureTableAsync(provider, token); + // #335: tenant-partitioned documents — provision the tenant's partition ordinal before the + // bulk MERGEs resolve it server-side from the registry (cached no-op for known tenants). + if (mapping.TenantPartitioned) + { + await _store.Events.TenantOrdinals.ResolveAsync(tenantId, token); + } + var storage = _store.Options.Providers.ClosedShapeGraph.StorageFor().QueryOnly; // Pre-process: assign ids + sync ITenanted before the operations serialize each document @@ -249,6 +258,13 @@ public async Task BulkInsertWithVersionAsync( var ensurer = _store.ResolveTableEnsurer(tenantId); await ensurer.EnsureTableAsync(provider, token); + // #335: tenant-partitioned documents — provision the tenant's partition ordinal before the + // version-checked MERGEs resolve it server-side from the registry. + if (mapping.TenantPartitioned) + { + await _store.Events.TenantOrdinals.ResolveAsync(tenantId, token); + } + // #273 doc-side convergence: source the version-checked MERGE's full column set from the // closed-shape descriptor's write binders (doc_type, guid_version, partition, soft-delete) // instead of the bespoke hardcoded id/data/version/dotnet_type[/tenant_id] subset. @@ -656,4 +672,175 @@ public Task EventProjectionScenario(Action configuration, Ca configuration(scenario); return scenario.Execute(ct); } + + // ---- runtime tenant onboarding under managed per-tenant partitioning (#335) ---- + + /// + /// Explicitly onboard tenants under the store's managed per-tenant partitioning — the SQL + /// Server counterpart of Marten's AddMartenManagedTenantsAsync. Each tenant is + /// registered in the pc_tenant_partitions registry, allocated a compact partition + /// ordinal, and every managed table (pc_events / pc_streams under + /// Events.UseTenantPartitionedEvents, plus all tenant-partitioned document tables) is + /// SPLIT for the new ordinals. Idempotent — already-registered tenants keep their ordinal and + /// emit no DDL. Under Events.UseTenantPartitionedEvents the per-tenant + /// pc_events_sequence_{ordinal} objects are provisioned as well. + /// + /// Only document tables of document types already registered with the store (schema config or + /// prior use) participate in the split; a table created later bakes the full boundary set at + /// creation. Tenants are also onboarded lazily on first write — this API exists for explicit + /// provisioning with per-table status reporting (partial failures surface per table, Marten + /// batch-add parity). + /// + /// + /// Per-table migration statuses for every table wired to the managed strategy. + public async Task AddPolecatManagedTenantsAsync( + CancellationToken token, params string[] tenantIds) + { + var events = AssertManagedTenantPartitioning(); + var manager = events.TenantPartitionManager; + + var result = await manager.AddPartitionsToAllTables( + NullLogger.Instance, _store.Database, tenantIds, token) + .ConfigureAwait(false); + + await EnsureTenantSequencesAsync(events, result.Ordinals, token).ConfigureAwait(false); + + return result.Tables; + } + + /// + /// Onboard tenants with explicitly assigned partition ordinals — the tenant bucketing seam + /// (Weasel 9.18.0 / weasel#362). With + /// ManagedTenantPartitions.AllowOrdinalSharing enabled, multiple tenant ids may map to + /// one ordinal so small tenants share a physical partition (the mitigation for SQL Server's + /// 15,000-partition ceiling). Re-registering a tenant with its current ordinal is a no-op; + /// re-registering with a different ordinal throws, because existing rows keep the old ordinal. + /// + /// Per-table migration statuses for every table wired to the managed strategy. + public async Task AddPolecatManagedTenantsAsync( + IReadOnlyDictionary tenantIdToOrdinal, CancellationToken token = default) + { + var events = AssertManagedTenantPartitioning(); + var manager = events.TenantPartitionManager; + + var result = await manager.AddPartitionsToAllTables( + NullLogger.Instance, _store.Database, tenantIdToOrdinal, token) + .ConfigureAwait(false); + + await EnsureTenantSequencesAsync(events, result.Ordinals, token).ConfigureAwait(false); + + return result.Tables; + } + + /// + /// Remove tenants from the store's managed per-tenant partitioning, retaining their rows + /// ( — the tenants' data merges into the + /// neighboring partition). The counterpart of Marten's RemoveMartenManagedTenantsAsync; + /// use the overload for data-removing semantics. + /// + public Task RemovePolecatManagedTenantsAsync(string[] tenantIds, CancellationToken token = default) + => RemovePolecatManagedTenantsAsync(tenantIds, TenantDropBehavior.RetainData, token); + + /// + /// Remove tenants from the store's managed per-tenant partitioning. SQL Server's + /// MERGE RANGE only removes the partition boundary — pass + /// to physically delete the tenants' rows from + /// every managed table before the merge (PostgreSQL managed-drop parity, Weasel 9.18.0 / + /// weasel#362). An ordinal still shared with other tenants is never merged or purged. Fully + /// released ordinals also drop their per-tenant pc_events_sequence_{ordinal} under + /// Events.UseTenantPartitionedEvents, and the removed tenants are evicted from the + /// in-process caches so a later write re-provisions instead of stamping a stale ordinal. + /// + public async Task RemovePolecatManagedTenantsAsync( + string[] tenantIds, TenantDropBehavior behavior, CancellationToken token = default) + { + var events = AssertManagedTenantPartitioning(); + var manager = events.TenantPartitionManager; + + // Hydrate + capture the ordinals of the tenants being dropped BEFORE the drop clears the + // registry rows, so fully-released ordinals can be identified afterwards. + await manager.InitializeAsync(_store.Database, token).ConfigureAwait(false); + var captured = tenantIds + .Where(t => !string.IsNullOrEmpty(t) && manager.Ordinals.ContainsKey(t)) + .ToDictionary(t => t, t => manager.Ordinals[t], StringComparer.Ordinal); + + await manager.DropPartitionFromAllTables( + NullLogger.Instance, _store.Database, tenantIds, behavior, token) + .ConfigureAwait(false); + + // An ordinal is released only when no remaining tenant references it (ordinal sharing). + var released = captured.Values + .Distinct() + .Where(o => !manager.Ordinals.Values.Contains(o)) + .ToArray(); + + if (events.UseTenantPartitionedEvents && released.Length > 0) + { + await DropTenantSequencesAsync(events, released, token).ConfigureAwait(false); + } + + foreach (var tenantId in captured.Keys) + { + events.TenantOrdinals.Evict(tenantId); + events.TenantSequences.Evict(tenantId); + } + } + + private Events.EventGraph AssertManagedTenantPartitioning() + { + var events = _store.Events; + if (!events.AnyTenantPartitioning) + { + throw new InvalidOperationException( + "Managed per-tenant partitioning is not enabled for this store. Enable " + + "Events.UseTenantPartitionedEvents and/or " + + "StoreOptions.Policies.PartitionMultiTenantedDocumentsUsingPolecatManagement() first."); + } + + return events; + } + + private static Task EnsureTenantSequencesAsync( + Events.EventGraph events, IReadOnlyDictionary ordinals, CancellationToken token) + { + return events.UseTenantPartitionedEvents && ordinals.Count > 0 + ? events.TenantSequences.EnsureSequencesForOrdinalsAsync(ordinals.Values, token) + : Task.CompletedTask; + } + + /// + /// Drop the per-tenant event sequences of fully-released ordinals (the #335 analogue of + /// Marten's orphan-sequence cleanup on tenant removal). + /// + private async Task DropTenantSequencesAsync(Events.EventGraph events, int[] ordinals, CancellationToken token) + { + var schema = events.DatabaseSchemaName; + var connStr = _store.Options.ConnectionString; + + await _resilience.ExecuteAsync(static async (state, ct) => + { + var (connectionString, schemaName, released) = state; + await using var conn = new SqlConnection(connectionString); + await conn.OpenAsync(ct).ConfigureAwait(false); + + foreach (var ordinal in released) + { + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $""" + IF EXISTS ( + SELECT 1 FROM sys.sequences sq + JOIN sys.schemas sc ON sq.schema_id = sc.schema_id + WHERE sq.name = @seq AND sc.name = @schema) + BEGIN + DECLARE @sql nvarchar(max) = + N'DROP SEQUENCE ' + QUOTENAME(@schema) + N'.' + QUOTENAME(@seq); + EXEC sp_executesql @sql; + END + """; + cmd.Parameters.AddWithValue("@seq", $"pc_events_sequence_{ordinal}"); + cmd.Parameters.AddWithValue("@schema", schemaName); + await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + } + }, (connStr, schema, ordinals), token).ConfigureAwait(false); + } } diff --git a/src/Polecat/Events/Daemon/PolecatProjectionBatch.cs b/src/Polecat/Events/Daemon/PolecatProjectionBatch.cs index dc5bcdb..7dfc377 100644 --- a/src/Polecat/Events/Daemon/PolecatProjectionBatch.cs +++ b/src/Polecat/Events/Daemon/PolecatProjectionBatch.cs @@ -108,6 +108,19 @@ public async Task ExecuteAsync(CancellationToken token) } } + // #335: tenant-partitioned documents — the daemon batch executes session operations + // directly (no DocumentSessionBase.SaveChangesAsync), so provision the partition ordinal + // for every tenant this batch writes under before the transaction opens. The projection + // write SQL resolves tenant_ordinal server-side from the registry these provisions + // populate; known tenants are cached no-ops. + if (_store.Options.Policies.DocumentTenantPartitioningEnabled) + { + foreach (var adapter in allOps.OfType()) + { + await _store.Events.TenantOrdinals.ResolveAsync(adapter.SessionTenantId, token); + } + } + // Add progress operations while (_progressOps.TryDequeue(out var progressOp)) { diff --git a/src/Polecat/Events/EventGraph.cs b/src/Polecat/Events/EventGraph.cs index 55debba..b675408 100644 --- a/src/Polecat/Events/EventGraph.cs +++ b/src/Polecat/Events/EventGraph.cs @@ -137,10 +137,11 @@ public TenancyStyle TenancyStyle /// /// The single Weasel.SqlServer managed per-tenant partition strategy (polecat#171) — maps each /// tenant to a compact integer ordinal, owns the pc_tenant_partitions registry, and - /// physically partitions pc_events by that ordinal (RANGE RIGHT). One instance is shared - /// between the events-table DDL () and the runtime partition split - /// () so Weasel can match the table to this strategy - /// by reference. Only built when is enabled. + /// physically partitions every managed table by that ordinal (RANGE RIGHT): pc_events + /// and pc_streams under , plus the + /// tenant-partitioned document tables (#335). One instance is shared between the table DDL and + /// the runtime partition split so Weasel can match each table to this strategy by reference — + /// one registry per database. /// internal ManagedTenantPartitions TenantPartitionManager => _tenantPartitions ??= new ManagedTenantPartitions( @@ -149,8 +150,18 @@ public TenancyStyle TenancyStyle column: "tenant_ordinal"); private PolecatDatabase? _tenantPartitionDatabase; + private TenantPartitionOrdinalRegistry? _tenantOrdinals; private TenantEventSequenceRegistry? _tenantSequences; + /// + /// True when anything in the store partitions by tenant — the event store tables + /// () or the document tables + /// (, #335). + /// Gates the shared pc_tenant_partitions registry feature schema. + /// + internal bool AnyTenantPartitioning => + UseTenantPartitionedEvents || _options.Policies.DocumentTenantPartitioningEnabled; + /// /// Wire the owning database so per-tenant provisioning can SPLIT physical partitions at runtime. /// Set by during construction. @@ -158,16 +169,25 @@ public TenancyStyle TenancyStyle internal void AttachTenantPartitionDatabase(PolecatDatabase database) => _tenantPartitionDatabase = database; + /// + /// The store's single tenant → partition-ordinal registry (#335), shared by the append + /// planner, the stream-row SQL, the document write pipeline, and the runtime tenant + /// onboarding APIs. Conjoined tenancy is a precondition for any tenant partitioning, so all + /// tenants share the one configured connection/database. + /// + internal TenantPartitionOrdinalRegistry TenantOrdinals => + _tenantOrdinals ??= new TenantPartitionOrdinalRegistry( + TenantPartitionManager, + _tenantPartitionDatabase ?? throw new InvalidOperationException( + "The owning database has not been attached for per-tenant partitioning.")); + /// /// Resolves (and lazily provisions) each tenant's ordinal, physical partition, and per-tenant - /// event sequence when is enabled. Conjoined tenancy - /// is a precondition, so all tenants share the one configured connection/database. + /// event sequence when is enabled. /// internal TenantEventSequenceRegistry TenantSequences => _tenantSequences ??= new TenantEventSequenceRegistry( - TenantPartitionManager, - _tenantPartitionDatabase ?? throw new InvalidOperationException( - "The owning database has not been attached for per-tenant partitioning."), + TenantOrdinals, _options.ConnectionString, DatabaseSchemaName, _options.ResiliencePipeline); /// @@ -178,6 +198,15 @@ internal void AttachTenantPartitionDatabase(PolecatDatabase database) /// internal void AssertTenantPartitioningValidity() { + if (_options.Policies.DocumentTenantPartitioningEnabled && TenancyStyle != TenancyStyle.Conjoined) + { + throw new InvalidOperationException( + "Tenant-partitioned documents (StoreOptions.Policies.AllDocumentsAreMultiTenantedWithPartitioning / " + + "PartitionMultiTenantedDocumentsUsingPolecatManagement) require Events.TenancyStyle = " + + "TenancyStyle.Conjoined — there is nothing to partition by when every document lives in " + + "the default tenant."); + } + if (!UseTenantPartitionedEvents) return; if (TenancyStyle != TenancyStyle.Conjoined) diff --git a/src/Polecat/Events/Schema/EventStoreFeatureSchema.cs b/src/Polecat/Events/Schema/EventStoreFeatureSchema.cs index 12ed997..a16ddb9 100644 --- a/src/Polecat/Events/Schema/EventStoreFeatureSchema.cs +++ b/src/Polecat/Events/Schema/EventStoreFeatureSchema.cs @@ -39,7 +39,9 @@ protected override IEnumerable schemaObjects() // ordinal) is owned by the Weasel.SqlServer ManagedTenantPartitions strategy; the physical // partition function/scheme are emitted as part of the partitioned pc_events DDL above, and the // per-tenant pc_events_sequence_{ordinal} objects are created on demand at first append. - if (_events.UseTenantPartitionedEvents) + // #335: tenant-partitioned documents share the same one-registry-per-database, so the registry + // also materializes when only the document-side policy is on. + if (_events.AnyTenantPartitioning) { foreach (var schemaObject in ((Weasel.Core.Migrations.IFeatureSchema)_events.TenantPartitionManager).Objects) diff --git a/src/Polecat/Events/Schema/StreamsTable.cs b/src/Polecat/Events/Schema/StreamsTable.cs index dcb15f8..66a7468 100644 --- a/src/Polecat/Events/Schema/StreamsTable.cs +++ b/src/Polecat/Events/Schema/StreamsTable.cs @@ -1,5 +1,6 @@ using Weasel.SqlServer; using Weasel.SqlServer.Tables; +using Weasel.SqlServer.Tables.Partitioning; namespace Polecat.Events.Schema; @@ -24,6 +25,17 @@ public StreamsTable(EventGraph events) AddColumn("id", idType).AsPrimaryKey().NotNull(); + // #335: partition pc_streams alongside pc_events under per-tenant partitioning (Marten + // parity — mt_streams rides mt_events' tenant partitioning). SQL Server requires the + // partition column in the clustered index, so tenant_ordinal joins the primary key AFTER + // (tenant_id, id) — existing readers keep their (tenant_id, id) prefix seek. The ordinal is + // stamped by the append path's stream-row SQL from the planner-resolved tenant cache. + if (events.UseTenantPartitionedEvents) + { + AddColumn(events.TenantPartitionManager.Column, "int").NotNull().AsPrimaryKey(); + this.PartitionByManagedTenants(events.TenantPartitionManager); + } + AddColumn("type", "varchar(250)").AllowNulls(); AddColumn("version", "bigint").NotNull().DefaultValue(0); diff --git a/src/Polecat/Events/Schema/TenantEventSequenceRegistry.cs b/src/Polecat/Events/Schema/TenantEventSequenceRegistry.cs index 5a41c58..89ae6e7 100644 --- a/src/Polecat/Events/Schema/TenantEventSequenceRegistry.cs +++ b/src/Polecat/Events/Schema/TenantEventSequenceRegistry.cs @@ -25,19 +25,17 @@ namespace Polecat.Events.Schema; /// internal sealed class TenantEventSequenceRegistry { - private readonly ManagedTenantPartitions _partitions; - private readonly PolecatDatabase _database; + private readonly TenantPartitionOrdinalRegistry _ordinals; private readonly string _connectionString; private readonly string _schemaName; private readonly ResiliencePipeline _resilience; private readonly ConcurrentDictionary _cache = new(StringComparer.Ordinal); - public TenantEventSequenceRegistry(ManagedTenantPartitions partitions, PolecatDatabase database, + public TenantEventSequenceRegistry(TenantPartitionOrdinalRegistry ordinals, string connectionString, string schemaName, ResiliencePipeline resilience) { - _partitions = partitions; - _database = database; + _ordinals = ordinals; _connectionString = connectionString; _schemaName = schemaName; _resilience = resilience; @@ -51,9 +49,10 @@ public async ValueTask ResolveAsync(string tenantId, Cancellation { if (_cache.TryGetValue(tenantId, out var cached)) return cached; - // Allocate the tenant ordinal + SPLIT the physical pc_events partition (idempotent). - var ordinal = await _partitions.AddPartitionToAllTables(_database, tenantId, token) - .ConfigureAwait(false); + // Allocate the tenant ordinal + SPLIT every managed-partitioned table (idempotent), + // through the store's shared ordinal registry (#335) so document writes and the + // stream-row SQL see the same tenant -> ordinal cache. + var ordinal = await _ordinals.ResolveAsync(tenantId, token).ConfigureAwait(false); // Ensure the tenant's per-tenant event sequence exists. await EnsureSequenceAsync(ordinal, token).ConfigureAwait(false); @@ -62,6 +61,24 @@ public async ValueTask ResolveAsync(string tenantId, Cancellation return _cache.GetOrAdd(tenantId, storage); } + /// + /// Evict a removed tenant so a later append re-provisions instead of reusing a stale + /// ordinal/sequence pairing (#335 runtime tenant removal). + /// + public void Evict(string tenantId) => _cache.TryRemove(tenantId, out _); + + /// + /// Imperative sequence provisioning for + /// — creates pc_events_sequence_{ordinal} for each ordinal (idempotent). + /// + public async Task EnsureSequencesForOrdinalsAsync(IEnumerable ordinals, CancellationToken token) + { + foreach (var ordinal in ordinals.Distinct()) + { + await EnsureSequenceAsync(ordinal, token).ConfigureAwait(false); + } + } + private async Task EnsureSequenceAsync(int ordinal, CancellationToken token) { await _resilience.ExecuteAsync(static async (state, ct) => diff --git a/src/Polecat/Events/Schema/TenantPartitionOrdinalRegistry.cs b/src/Polecat/Events/Schema/TenantPartitionOrdinalRegistry.cs new file mode 100644 index 0000000..99bf238 --- /dev/null +++ b/src/Polecat/Events/Schema/TenantPartitionOrdinalRegistry.cs @@ -0,0 +1,64 @@ +using System.Collections.Concurrent; +using Polecat.Storage; +using Weasel.SqlServer.Tables.Partitioning; + +namespace Polecat.Events.Schema; + +/// +/// Resolves (and lazily provisions) each tenant's compact partition ordinal against the store's +/// shared strategy (#335). This is the single tenant → +/// ordinal cache shared by every managed-tenant-partitioned table in the store: pc_events +/// and pc_streams under , and the +/// tenant-partitioned document tables under +/// . +/// +/// is idempotent — a known tenant returns its cached ordinal with no +/// database traffic; an unknown tenant is registered in pc_tenant_partitions and every +/// table wired to the strategy is SPLIT for the new ordinal (all via Weasel's +/// ). +/// is the synchronous read used by SQL-building closures that run +/// strictly after a resolve (the append planner, flush-time binders). +/// +/// +internal sealed class TenantPartitionOrdinalRegistry +{ + private readonly ManagedTenantPartitions _partitions; + private readonly PolecatDatabase _database; + + private readonly ConcurrentDictionary _cache = new(StringComparer.Ordinal); + + public TenantPartitionOrdinalRegistry(ManagedTenantPartitions partitions, PolecatDatabase database) + { + _partitions = partitions; + _database = database; + } + + /// + /// The tenant's partition ordinal, provisioning the registry row + physical partitions on + /// first use (idempotent — an already-registered tenant just hydrates the cache). + /// + public async ValueTask ResolveAsync(string tenantId, CancellationToken token) + { + if (_cache.TryGetValue(tenantId, out var cached)) return cached; + + var ordinal = await _partitions.AddPartitionToAllTables(_database, tenantId, token) + .ConfigureAwait(false); + + return _cache.GetOrAdd(tenantId, ordinal); + } + + /// + /// Synchronous cache read for SQL-building code that runs after the planner/flush pipeline + /// has already resolved the tenant. Returns false when the tenant has not been resolved in + /// this process yet. + /// + public bool TryGetOrdinal(string tenantId, out int ordinal) => _cache.TryGetValue(tenantId, out ordinal); + + /// + /// Drop a tenant from the in-memory cache after + /// + /// removes it from the registry, so a later write for the same tenant re-provisions instead of + /// stamping a stale ordinal. + /// + public void Evict(string tenantId) => _cache.TryRemove(tenantId, out _); +} diff --git a/src/Polecat/Events/Storage/SqlServerEventStoreDialect.cs b/src/Polecat/Events/Storage/SqlServerEventStoreDialect.cs index 391a4a4..5d05bee 100644 --- a/src/Polecat/Events/Storage/SqlServerEventStoreDialect.cs +++ b/src/Polecat/Events/Storage/SqlServerEventStoreDialect.cs @@ -114,8 +114,14 @@ private static IStorageDialect ResolveStorageDialect(bool isGuid) private static Action BuildInsertStreamCommandConfigurer( EventGraph graph, bool isGuid, IStorageDialect dialect) { + // #335: under per-tenant partitioning pc_streams carries the tenant_ordinal partition + // column, stamped from the planner-resolved tenant cache (the append planner always + // resolves the tenant's ordinal before any stream operation executes). + var partitioned = graph.UseTenantPartitionedEvents; var prefix = $"insert into {graph.StreamsTableName} " + - "(id, type, version, timestamp, created, tenant_id) values ("; + (partitioned + ? "(id, type, version, timestamp, created, tenant_id, tenant_ordinal) values (" + : "(id, type, version, timestamp, created, tenant_id) values ("); return (builder, stream) => { @@ -136,10 +142,36 @@ private static IStorageDialect ResolveStorageDialect(bool isGuid) var tenantParam = builder.AppendParameter(stream.TenantId); dialect.SetParameterType(tenantParam, StorageColumnType.String); + if (partitioned) + { + builder.Append(", "); + var ordinalParam = builder.AppendParameter(ResolveCachedOrdinal(graph, stream)); + dialect.SetParameterType(ordinalParam, StorageColumnType.Int); + } + builder.Append(")"); }; } + /// + /// The stream tenant's partition ordinal from the store's shared tenant cache (#335). The + /// append planner resolves (and provisions) every stream's tenant before its operations + /// execute, so this synchronous read cannot miss on the append path — a miss means a new + /// code path is building stream-row SQL without resolving the tenant first. + /// + private static int ResolveCachedOrdinal(EventGraph graph, StreamAction stream) + { + if (!graph.TenantOrdinals.TryGetOrdinal(stream.TenantId, out var ordinal)) + { + throw new InvalidOperationException( + $"Per-tenant partitioning is enabled but tenant '{stream.TenantId}' has no resolved " + + "partition ordinal in this process. Stream-row SQL must run after the append planner " + + "(or AdvancedOperations.AddPolecatManagedTenantsAsync) has resolved the tenant."); + } + + return ordinal; + } + /// /// Closure for update {schema}.pc_streams set version = @version, timestamp = /// SYSDATETIMEOFFSET() where id = @id and version = @expected [and tenant_id = @tenant]. @@ -173,6 +205,15 @@ private static IStorageDialect ResolveStorageDialect(bool isGuid) var tenantParam = builder.AppendParameter(stream.TenantId); dialect.SetParameterType(tenantParam, StorageColumnType.String); } + + // #335: partition-eliminate the version bump under per-tenant partitioning — + // tenant_ordinal is in the clustered key of the partitioned pc_streams. + if (graph.UseTenantPartitionedEvents) + { + builder.Append(" and tenant_ordinal = "); + var ordinalParam = builder.AppendParameter(ResolveCachedOrdinal(graph, stream)); + dialect.SetParameterType(ordinalParam, StorageColumnType.Int); + } }; } diff --git a/src/Polecat/Internal/DocumentProviderRegistry.cs b/src/Polecat/Internal/DocumentProviderRegistry.cs index 1a5b372..8ca14df 100644 --- a/src/Polecat/Internal/DocumentProviderRegistry.cs +++ b/src/Polecat/Internal/DocumentProviderRegistry.cs @@ -163,6 +163,14 @@ public void ConfigurePartitionedDocuments() var exprType = expr.GetType(); if (!exprType.IsGenericType) continue; + // #335: under the tenant-partitioned-documents policy, EVERY registered document type is + // partitioned — materialize its provider so the table joins the managed set (created by + // schema migration, SPLIT by AddPolecatManagedTenantsAsync) at activation time. + if (_options.Policies.DocumentTenantPartitioningEnabled) + { + GetProvider(exprType.GetGenericArguments()[0]); + } + var partitioningField = exprType.GetField("Partitioning", BindingFlags.NonPublic | BindingFlags.Instance); if (partitioningField?.GetValue(expr) is not DocumentPartitioning) continue; @@ -171,8 +179,11 @@ public void ConfigurePartitionedDocuments() if (_options.Events.TenancyStyle == TenancyStyle.Conjoined) { throw new NotSupportedException( - "RANGE partitioning of document tables is currently supported for single-tenant tables " + - $"only, but '{docType.Name}' uses conjoined tenancy."); + "RANGE partitioning of document tables on a caller-chosen member is supported for " + + $"single-tenant tables only, but '{docType.Name}' uses conjoined tenancy — a SQL " + + "Server table supports only one partition scheme. Conjoined tables can instead be " + + "partitioned per tenant via " + + "StoreOptions.Policies.PartitionMultiTenantedDocumentsUsingPolecatManagement() (#335)."); } // Materialize the provider so DocumentFeatureSchema yields its (partitioned) table. diff --git a/src/Polecat/Internal/DocumentSessionBase.cs b/src/Polecat/Internal/DocumentSessionBase.cs index 9fe6e02..d07fdf9 100644 --- a/src/Polecat/Internal/DocumentSessionBase.cs +++ b/src/Polecat/Internal/DocumentSessionBase.cs @@ -437,6 +437,20 @@ private async Task SaveChangesInternalAsync(CancellationToken token) await _tableEnsurer.EnsureEventStoreSchemaAsync(token); } + // #335: tenant-partitioned documents — resolve (and lazily provision, mirroring the event + // append path) the partition ordinal for every tenant this flush writes under, BEFORE the + // data transaction opens (provisioning runs DDL on its own connection). The document write + // SQL then resolves tenant_ordinal server-side from the pc_tenant_partitions registry these + // provisions populate; an already-known tenant is a cached no-op. + if (Options.Policies.DocumentTenantPartitioningEnabled && _workTracker.Operations.Count > 0) + { + await _eventGraph.TenantOrdinals.ResolveAsync(TenantId, token); + foreach (var adapter in _workTracker.Operations.OfType()) + { + await _eventGraph.TenantOrdinals.ResolveAsync(adapter.SessionTenantId, token); + } + } + await _transactional.BeginTransactionAsync(token); using var tx = _transactional.Transaction!; try diff --git a/src/Polecat/Internal/DocumentTableEnsurer.cs b/src/Polecat/Internal/DocumentTableEnsurer.cs index 22d0895..f3335fb 100644 --- a/src/Polecat/Internal/DocumentTableEnsurer.cs +++ b/src/Polecat/Internal/DocumentTableEnsurer.cs @@ -92,6 +92,14 @@ public async Task EnsureTableAsync(DocumentProvider provider, CancellationToken // read GetFieldValue). Convert it to the inner primitive type in place — drop // PK, ALTER COLUMN, re-add PK — before Weasel diffs the table, so the diff stays clean. await ConvertStrongTypedIdColumnIfNeededAsync(conn, provider.Mapping, table, token); + + // #335: a tenant-partitioned document table created after tenants already exist must + // bake the full boundary set into its CREATE — hydrate the shared partition manager's + // tenant map from pc_tenant_partitions first (cached; a no-op after the first call). + if (provider.Mapping.TenantPartitioned) + { + await _options.EventGraph.TenantPartitionManager.InitializeAsync(conn, token); + } var autoCreate = provider.Mapping.Partitioning is { ExternallyManaged: true } ? AutoCreate.CreateOnly : AutoCreate.CreateOrUpdate; @@ -303,6 +311,14 @@ public async Task EnsureEventStoreSchemaAsync(CancellationToken token) .Select(p => p.NaturalKeyDefinition!) .ToList(); + // #335: under per-tenant partitioning, hydrate the shared partition manager's tenant map + // first so pc_events/pc_streams CREATEd against an existing registry bake the full + // boundary set (cached; a no-op after the first call). + if (_options.EventGraph.AnyTenantPartitioning) + { + await _options.EventGraph.TenantPartitionManager.InitializeAsync(conn, token); + } + var eventSchema = new Events.Schema.EventStoreFeatureSchema(_options.EventGraph, naturalKeys); var migration = await SchemaMigration.DetermineAsync(conn, token, eventSchema.Objects); await migrator.ApplyAllAsync(conn, migration, AutoCreate.CreateOrUpdate, ct: token); diff --git a/src/Polecat/Internal/Operations/ClosedShapeOperationAdapter.cs b/src/Polecat/Internal/Operations/ClosedShapeOperationAdapter.cs index 54b7cac..27e9e5a 100644 --- a/src/Polecat/Internal/Operations/ClosedShapeOperationAdapter.cs +++ b/src/Polecat/Internal/Operations/ClosedShapeOperationAdapter.cs @@ -31,6 +31,13 @@ public ClosedShapeOperationAdapter(Weasel.Storage.IStorageOperation inner, IStor public object? DocumentId { get; } public Type DocumentType => _inner.DocumentType; + /// + /// The tenant this operation will write under (the session's tenant, or the ForTenant + /// override wrapped in a TenantScopedStorageSession). Used by the flush pipeline to + /// provision tenant partition ordinals before operations execute (#335). + /// + internal string SessionTenantId => _session.TenantId; + public OperationRole Role() => _inner.Role(); public void ConfigureCommand(Weasel.SqlServer.ICommandBuilder builder) diff --git a/src/Polecat/Storage/ClosedShape/PolecatDocumentStorage.cs b/src/Polecat/Storage/ClosedShape/PolecatDocumentStorage.cs index 78dfa34..675860d 100644 --- a/src/Polecat/Storage/ClosedShape/PolecatDocumentStorage.cs +++ b/src/Polecat/Storage/ClosedShape/PolecatDocumentStorage.cs @@ -427,7 +427,21 @@ private string BulkVersionCheckedSql() var usingValues = string.Join(", ", usingCols.Select(_ => "?")); var sourceCols = string.Join(", ", usingCols); + + // Managed tenant partitioning (#335): resolve tenant_ordinal server-side by joining the + // pc_tenant_partitions registry into the MERGE source — parameter slots unchanged. + var usingClause = _mapping.TenantPartitioned + ? $"(SELECT {string.Join(", ", usingCols.Select(c => $"v.{c}"))}, tp.ordinal AS tenant_ordinal " + + $"FROM (VALUES ({usingValues})) AS v ({sourceCols}) " + + $"LEFT JOIN {_mapping.StoreOptions.EventGraph.TenantPartitionsTableName} tp " + + "ON tp.tenant_id = v.tenant_id) AS s" + : $"(VALUES ({usingValues})) AS s ({sourceCols})"; + var on = conjoined ? "t.id = s.id AND t.tenant_id = s.tenant_id" : "t.id = s.id"; + if (_mapping.TenantPartitioned) + { + on += " AND t.tenant_ordinal = s.tenant_ordinal"; + } // UPDATE (matched AND version == expected): data, version+1, client cols, server literals. var setList = new List { "data = s.data", "version = t.version + 1" }; @@ -455,9 +469,15 @@ private string BulkVersionCheckedSql() insertVals.Add("s.tenant_id"); } + if (_mapping.TenantPartitioned) + { + insertCols.Add("tenant_ordinal"); + insertVals.Add("s.tenant_ordinal"); + } + _bulkVersionCheckedSql = $"MERGE {table} WITH (HOLDLOCK) AS t " + - $"USING (VALUES ({usingValues})) AS s ({sourceCols}) ON {on} " + + $"USING {usingClause} ON {on} " + $"WHEN MATCHED AND t.version = s.expected_version THEN UPDATE SET {string.Join(", ", setList)} " + $"WHEN NOT MATCHED THEN INSERT ({string.Join(", ", insertCols)}) VALUES ({string.Join(", ", insertVals)}) " + $"OUTPUT inserted.id;"; diff --git a/src/Polecat/Storage/DocumentMapping.cs b/src/Polecat/Storage/DocumentMapping.cs index ad6b692..07cc77b 100644 --- a/src/Polecat/Storage/DocumentMapping.cs +++ b/src/Polecat/Storage/DocumentMapping.cs @@ -251,6 +251,19 @@ public DocumentMapping(Type documentType, StoreOptions options) /// True when a promoted partition column must be written on every upsert. public bool HasPartitionColumn => Partitioning is { RequiresDuplicatedColumn: true }; + /// + /// True when this document's table is physically partitioned per tenant through the store's + /// shared managed tenant partitioning (#335): conjoined tenancy + the store-wide policy + /// () with + /// no per-type opt-out. The table gains a tenant_ordinal int primary-key column whose + /// value is resolved server-side from pc_tenant_partitions on every write. + /// + public bool TenantPartitioned => + TenancyStyle == TenancyStyle.Conjoined && StoreOptions.Policies.IsTenantPartitioned(DocumentType); + + /// The partition column every managed-tenant-partitioned table carries (#335). + public const string TenantOrdinalColumn = "tenant_ordinal"; + /// SQL fragment appended to an INSERT column list for the partition column (or empty). public string PartitionInsertColumns => HasPartitionColumn ? $", {Partitioning!.ColumnName}" : string.Empty; diff --git a/src/Polecat/Storage/DocumentTable.cs b/src/Polecat/Storage/DocumentTable.cs index 758316d..06887d9 100644 --- a/src/Polecat/Storage/DocumentTable.cs +++ b/src/Polecat/Storage/DocumentTable.cs @@ -3,6 +3,7 @@ using Weasel.Core; using Weasel.SqlServer; using Weasel.SqlServer.Tables; +using Weasel.SqlServer.Tables.Partitioning; namespace Polecat.Storage; @@ -86,15 +87,37 @@ public DocumentTable(DocumentMapping mapping) // additive CreateDeltaAsync override (#267) — it is defaulted, so INSERTs that omit it still // succeed — so this is purely additive with no destructive migration. + // Managed per-tenant partitioning (#335): the conjoined table is physically partitioned by + // the store's shared tenant-ordinal strategy (one pc_tenant_partitions registry per + // database). SQL Server requires the partition column in the clustered index, so + // tenant_ordinal joins the primary key after (tenant_id, id) — reads still prefix-seek on + // (tenant_id, id); every write resolves the ordinal server-side from the registry. + if (mapping.TenantPartitioned) + { + if (mapping.Partitioning is not null) + { + throw new NotSupportedException( + $"Document '{mapping.DocumentType.Name}' cannot combine PartitionByRange with the " + + "store's managed tenant partitioning — a SQL Server table supports only one " + + "partition scheme. Opt the type out via " + + "Policies.ForDocument(p => p.DisableTenantPartitioning = true) to keep the " + + "custom RANGE partitioning."); + } + + AddColumn(DocumentMapping.TenantOrdinalColumn, "int").NotNull().AsPrimaryKey(); + this.PartitionByManagedTenants(mapping.StoreOptions.EventGraph.TenantPartitionManager); + } // Declarative SQL Server RANGE partitioning (#211). The partition column must be part of the // table's unique (clustered) index, so a promoted member joins the primary key. - if (mapping.Partitioning is { } partitioning) + else if (mapping.Partitioning is { } partitioning) { if (mapping.TenancyStyle == TenancyStyle.Conjoined) { throw new NotSupportedException( - "RANGE partitioning of document tables is currently supported for single-tenant tables " + - $"only, but '{mapping.DocumentType.Name}' uses conjoined tenancy."); + "RANGE partitioning of document tables on a caller-chosen member is supported for " + + $"single-tenant tables only, but '{mapping.DocumentType.Name}' uses conjoined " + + "tenancy. Conjoined tables can instead be partitioned per tenant via " + + "StoreOptions.Policies.PartitionMultiTenantedDocumentsUsingPolecatManagement() (#335)."); } if (partitioning.RequiresDuplicatedColumn) diff --git a/src/Polecat/Storage/SqlServerDocumentStorageDescriptorBuilder.cs b/src/Polecat/Storage/SqlServerDocumentStorageDescriptorBuilder.cs index d4dc989..5adcf00 100644 --- a/src/Polecat/Storage/SqlServerDocumentStorageDescriptorBuilder.cs +++ b/src/Polecat/Storage/SqlServerDocumentStorageDescriptorBuilder.cs @@ -297,6 +297,18 @@ private static string BuildMergeSql( var usingValues = string.Join(", ", usingColumns.Select(_ => "?")); var sourceColumns = string.Join(", ", usingColumns); + // Managed tenant partitioning (#335): the tenant's partition ordinal is resolved + // server-side by joining the pc_tenant_partitions registry into the MERGE source — the + // VALUES tuple (and so the positional parameter slots) is unchanged. An unregistered + // tenant yields NULL and fails the tenant_ordinal NOT NULL constraint loudly; the session + // flush pipeline auto-provisions tenants before executing operations. + var usingClause = mapping.TenantPartitioned + ? $"(SELECT {string.Join(", ", usingColumns.Select(c => $"v.{c}"))}, tp.ordinal AS tenant_ordinal " + + $"FROM (VALUES ({usingValues})) AS v ({sourceColumns}) " + + $"LEFT JOIN {mapping.StoreOptions.EventGraph.TenantPartitionsTableName} tp " + + "ON tp.tenant_id = v.tenant_id) AS s" + : $"(VALUES ({usingValues})) AS s ({sourceColumns})"; + var onClause = isConjoined ? "t.id = s.id AND t.tenant_id = s.tenant_id" : "t.id = s.id"; @@ -306,6 +318,12 @@ private static string BuildMergeSql( onClause += $" AND t.{partitionColumn} = s.{partitionColumn}"; } + if (mapping.TenantPartitioned) + { + // tenant_ordinal is in the PK; the predicate keeps MERGE partition-eliminated. + onClause += " AND t.tenant_ordinal = s.tenant_ordinal"; + } + // INSERT branch. Off/Optimistic: version literal 1 + created_at. Numeric: the version // value is the revision CASE over the source columns (auto -> initial revision 1). var insertColumns = new List(); @@ -334,6 +352,12 @@ private static string BuildMergeSql( insertColumns.Add("created_at"); insertValues.Add(ServerTimestamp); + if (mapping.TenantPartitioned) + { + insertColumns.Add("tenant_ordinal"); + insertValues.Add("s.tenant_ordinal"); + } + foreach (var binder in binders.Where(b => b.IsServerSide)) { insertColumns.Add(binder.ColumnName); @@ -346,7 +370,7 @@ private static string BuildMergeSql( if (insertOnly) { return $"MERGE {table} WITH (HOLDLOCK) AS t " + - $"USING (VALUES ({usingValues})) AS s ({sourceColumns}) ON {onClause} " + + $"USING {usingClause} ON {onClause} " + $"{insertClause} " + $"OUTPUT {OutputColumn(mode)};"; } @@ -388,7 +412,7 @@ private static string BuildMergeSql( } return $"MERGE {table} WITH (HOLDLOCK) AS t " + - $"USING (VALUES ({usingValues})) AS s ({sourceColumns}) ON {onClause} " + + $"USING {usingClause} ON {onClause} " + $"WHEN MATCHED{guard} THEN UPDATE SET {string.Join(", ", updateAssignments)} " + $"{insertClause} " + $"OUTPUT {OutputColumn(mode)};"; @@ -445,6 +469,15 @@ private static string BuildUpdateSql( var usingValues = string.Join(", ", usingColumns.Select(_ => "?")); var sourceColumns = string.Join(", ", usingColumns); + // Managed tenant partitioning (#335): same server-side ordinal join as BuildMergeSql, + // keyed on the update ops' tenant_id_pk source column. + var usingClause = mapping.TenantPartitioned + ? $"(SELECT {string.Join(", ", usingColumns.Select(c => $"v.{c}"))}, tp.ordinal AS tenant_ordinal " + + $"FROM (VALUES ({usingValues})) AS v ({sourceColumns}) " + + $"LEFT JOIN {mapping.StoreOptions.EventGraph.TenantPartitionsTableName} tp " + + "ON tp.tenant_id = v.tenant_id_pk) AS s" + : $"(VALUES ({usingValues})) AS s ({sourceColumns})"; + var onClause = "t.id = s.id"; if (isConjoined) { @@ -456,6 +489,11 @@ private static string BuildUpdateSql( onClause += $" AND t.{partitionColumn} = s.{partitionColumn}_pk"; } + if (mapping.TenantPartitioned) + { + onClause += " AND t.tenant_ordinal = s.tenant_ordinal"; + } + var updateAssignments = new List { "data = s.data" }; updateAssignments.Add(numeric ? "version = CASE WHEN s.rev0 = 0 THEN t.version + 1 ELSE s.rev1 + 1 END" @@ -474,7 +512,7 @@ private static string BuildUpdateSql( }; return $"MERGE {table} WITH (HOLDLOCK) AS t " + - $"USING (VALUES ({usingValues})) AS s ({sourceColumns}) ON {onClause} " + + $"USING {usingClause} ON {onClause} " + $"WHEN MATCHED{guard} THEN UPDATE SET {string.Join(", ", updateAssignments)} " + $"OUTPUT {OutputColumn(mode)};"; } diff --git a/src/Polecat/StoreOptions.cs b/src/Polecat/StoreOptions.cs index 45df25f..e38be0e 100644 --- a/src/Polecat/StoreOptions.cs +++ b/src/Polecat/StoreOptions.cs @@ -32,6 +32,7 @@ public class StoreOptions public StoreOptions() { + Policies = new StorePolicies(this); EventGraph = new EventGraph(this); Events.EventGraph = EventGraph; Projections = new PolecatProjectionOptions(EventGraph); @@ -144,9 +145,9 @@ internal void ReadJasperFxOptions(JasperFxOptions? options) public SchemaConfiguration Schema { get; } = new(); /// - /// Document storage policies (e.g., soft deletes). + /// Document storage policies (e.g., soft deletes, tenant-partitioned documents). /// - public StorePolicies Policies { get; } = new(); + public StorePolicies Policies { get; } /// /// Global session listeners applied to all sessions. diff --git a/src/Polecat/StorePolicies.cs b/src/Polecat/StorePolicies.cs index 6343db3..2f51077 100644 --- a/src/Polecat/StorePolicies.cs +++ b/src/Polecat/StorePolicies.cs @@ -7,8 +7,15 @@ namespace Polecat; /// public class StorePolicies { + private readonly StoreOptions _parent; private bool _allDocumentsSoftDeleted; private readonly HashSet _softDeletedTypes = new(); + private readonly HashSet _tenantPartitioningDisabledTypes = new(); + + internal StorePolicies(StoreOptions parent) + { + _parent = parent; + } /// /// Enable soft deletes for all document types. @@ -18,6 +25,40 @@ public void AllDocumentsSoftDeleted() _allDocumentsSoftDeleted = true; } + /// + /// Make every document conjoined multi-tenanted AND physically partition every document table + /// per tenant through the store's shared managed tenant partitioning (#335 — the SQL Server + /// counterpart of Marten's AllDocumentsAreMultiTenantedWithPartitioning + + /// PartitionMultiTenantedDocumentsUsingMartenManagement). Sets + /// Events.TenancyStyle = TenancyStyle.Conjoined (document tenancy is store-wide in + /// Polecat) and adds a tenant_ordinal column + RANGE RIGHT partitioning to every + /// document table, driven by the one pc_tenant_partitions registry per database. + /// + /// Tenants are onboarded lazily on first write, or explicitly (with per-table status + /// reporting and ordinal bucketing) through + /// . + /// + /// + public void AllDocumentsAreMultiTenantedWithPartitioning() + { + _parent.Events.TenancyStyle = TenancyStyle.Conjoined; + DocumentTenantPartitioningEnabled = true; + } + + /// + /// Physically partition every conjoined document table per tenant through the store's shared + /// managed tenant partitioning (#335), without changing the tenancy style — the store must + /// already be configured with Events.TenancyStyle = TenancyStyle.Conjoined (asserted at + /// store construction). The SQL Server counterpart of Marten's + /// PartitionMultiTenantedDocumentsUsingMartenManagement: SQL Server has no per-table + /// schema argument because the ordinal registry (pc_tenant_partitions) always lives in + /// the event store schema — one registry per database. + /// + public void PartitionMultiTenantedDocumentsUsingPolecatManagement() + { + DocumentTenantPartitioningEnabled = true; + } + /// /// Enable soft deletes for a specific document type. /// @@ -29,12 +70,36 @@ public void ForDocument(Action configure) { _softDeletedTypes.Add(typeof(T)); } + + if (policy.DisableTenantPartitioning) + { + _tenantPartitioningDisabledTypes.Add(typeof(T)); + } } internal bool IsSoftDeleted(Type documentType) { return _allDocumentsSoftDeleted || _softDeletedTypes.Contains(documentType); } + + /// + /// True when the tenant-partitioned-documents policy is active for the store (#335). + /// + internal bool DocumentTenantPartitioningEnabled { get; private set; } + + /// + /// Whether this document type's table is managed-tenant-partitioned: the store-wide policy is + /// on and the type has not opted out via + /// ForDocument<T>(p => p.DisableTenantPartitioning = true). The daemon's + /// dead-letter document is always excluded (Marten parity — its writes must never depend on + /// tenant onboarding, or a failing projection could dead-letter into a second failure). + /// + internal bool IsTenantPartitioned(Type documentType) + { + return DocumentTenantPartitioningEnabled + && documentType != typeof(JasperFx.Events.Daemon.DeadLetterEvent) + && !_tenantPartitioningDisabledTypes.Contains(documentType); + } } /// @@ -46,4 +111,12 @@ public class DocumentPolicy /// Enable soft deletes for this document type. /// public bool SoftDeleted { get; set; } + + /// + /// Opt this document type out of the store-wide managed tenant partitioning policy (#335) — + /// its table stays a plain conjoined table (tenant_id in the primary key, no physical + /// partitions). The Polecat analogue of Marten's [SingleTenanted] / + /// DisablePartitioningIfAny escape hatches. + /// + public bool DisableTenantPartitioning { get; set; } }