diff --git a/Directory.Packages.props b/Directory.Packages.props index bc2eead..89a5098 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -50,9 +50,11 @@ Func — SOURCE BREAKING, which is why the bump and the call-site change land together — plus reachable NaturalKeyBuilder / NaturalKeyFor() and loud failure on an unbindable [NaturalKeySource] (polecat#369). - Also carries jasperfx#568/#572 daemon race fixes, picked up for free. --> - - + Also carries jasperfx#568/#572 daemon race fixes, picked up for free. + JasperFx 2.36.3: the floor Weasel 9.21.0 depends on — bumped in lockstep with the + Weasel bump below so the matrix stays coherent rather than resolving transitively. --> + + - + @@ -124,16 +126,25 @@ 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). --> - - - + (per-tenant partitioning parity for documents + streams). + Weasel 9.21.0: weasel#401 — Weasel.SqlServer.Tables.Partitioning.ManagedRangePartitions + plus Weasel.Core.Partitioning's RollingWindowPolicy / PartitionPeriod: rolling + time-window RANGE partitions whose CreateDelta is purely additive (a window that has + rolled forward is never mistaken for drift) with NEXT USED + SPLIT RANGE roll-forward + and partition TRUNCATE + MERGE RANGE retention. Consumed by Polecat #386 + (PartitionOn(...).ByRollingRange(...)). Also rides along from the intervening line: + weasel#391 (managed tenant bucketing actually shares one partition — #335's ordinal + sharing) and weasel#399 (TableColumn.MatchesForDelta compares through the virtual + Equals, so a subclassed column no longer churns the delta). --> + + + - + diff --git a/docs/documents/partitioning.md b/docs/documents/partitioning.md index b7f8053..c662699 100644 --- a/docs/documents/partitioning.md +++ b/docs/documents/partitioning.md @@ -1,13 +1,20 @@ # Table Partitioning -Polecat can declaratively **RANGE-partition a document table** on a member you choose — the SQL Server -companion to Marten's `PartitionOn`. The classic use is a time-series retention table partitioned by -month, so that old data can eventually be pruned by dropping a partition instead of issuing a large -`DELETE`. +Polecat can **RANGE-partition a document table** on a member you choose — the SQL Server companion to +Marten's `PartitionOn`. The classic use is a time-series retention table partitioned by month, so that old +data is reclaimed by retiring a partition instead of issuing a large `DELETE`. Three strategies are +available: + +| Strategy | Who owns the partitions | +| -------------------------------------------------------------- | ----------------------------------------------------------------------------- | +| `ByRange(...)` / `PartitionByRange(...)` | you declare a fixed boundary list; Polecat rolls additions forward in place | +| [`ByRollingRange(...)`](#rolling-time-windows) | Polecat, from a rolling-window policy — provisioning *and* retention | +| [`ByExternallyManagedRange(...)`](#externally-managed-range-partitions) | something outside Polecat, which Polecat then never touches | ::: tip This is built on SQL Server partition **functions** and **schemes** rather than the child-table model -PostgreSQL/Marten uses, so the migration story differs. It requires `Weasel.SqlServer` 9.3.0 or later. +PostgreSQL/Marten uses, so the migration story differs. Declarative range partitioning requires +`Weasel.SqlServer` 9.3.0 or later; rolling time windows require 9.21.0 or later. ::: ## Partitioning by a date member @@ -57,14 +64,142 @@ opts.Schema.For() Schema migration adds the new partition with no data movement. Removing a boundary or changing the column/type is reported as a rebuild rather than performed silently. +## Rolling time windows + +Declaring every boundary up front only works while the set of partitions is *fixed*. Real time-series +storage needs it to **move**: provision next month, retire last year. Rather than hand-writing that DDL on +a schedule forever, describe the window and let Polecat own it: + +```csharp +var store = DocumentStore.For(opts => +{ + opts.Connection(connectionString); + + // Keep 12 months of history, provision 3 months ahead. Polecat splits in the partitions at the + // leading edge and retires the aged ones at the trailing edge — no application-authored DDL. + opts.Schema.For() + .PartitionOn(x => x.BucketEnd) + .ByRollingRange(PartitionPeriod.Month, periodsAhead: 3, periodsBehind: 12); +}); +``` + +`PartitionPeriod` (from `Weasel.Core.Partitioning`) supports `Hour`, `Day`, `Week`, `Month`, and `Year`, and +the whole window is computed in UTC. The partition member must be a `DateTime` or `DateTimeOffset` — a rolling window is a function of the +clock, so anything else is rejected at *configuration* time with a message that names the member, rather +than surfacing as an opaque partition-function error during the first migration. + +There are also overloads taking a `RollingWindowPolicy` directly, or a pre-built `ManagedRangePartitions`. +Pass the *same* manager instance to several document types to roll all of their tables forward in one pass: + +```csharp +using Weasel.SqlServer.Tables.Partitioning; // ManagedRangePartitions +using Weasel.Core.Partitioning; // RollingWindowPolicy, PartitionPeriod + +var manager = new ManagedRangePartitions( + RollingWindowPolicy.Monthly(periodsAhead: 3, periodsBehind: 12), + column: "bucket_end", sqlDataType: "datetimeoffset"); + +opts.Schema.For().PartitionOn(x => x.BucketEnd).ByRollingRange(manager); +opts.Schema.For().PartitionOn(x => x.BucketEnd).ByRollingRange(manager); +``` + +The manager is also where you set `Filegroup` if the partitions should not go to `PRIMARY`, and it takes a +`TimeProvider` so the window can be rolled forward deterministically in tests instead of waiting on the +calendar. + +### How the two halves are driven + +| | Driven by | Why | +| -------------------------------- | -------------------------------- | ------------------------------------------------------------------------------------------------------------------------- | +| **Provision** the leading edge | ordinary schema migration | with a rolling-window manager attached the partition delta is purely additive, so a rolled-forward window is a `SPLIT` | +| **Retire** the trailing edge | startup pass + `Advanced.*` | migration never removes data, so the retention half has to be driven separately | + +The window is a pure function of the policy and the clock, which is what makes this safe: a window that has +rolled forward differs from the database by exactly one new boundary at the leading edge and one aged +boundary at the trailing edge. A boundary the declaration no longer names is a period that has **aged out** — +the normal steady state of a rolling window, not drift — so `CreateDelta` reports `Additive` or `None`, never +`Rebuild`. (A column or type change still rebuilds, as it must.) + +Polecat runs the maintenance pass — roll forward, then retire everything below the retention floor — at +startup, alongside the schema changes it already applies: + +```csharp +builder.Services.AddPolecat(opts => +{ + // ... the ByRollingRange() configuration above +}).ApplyAllDatabaseChangesOnStartup(); +``` + +Applying changes on startup is how a host says "Polecat owns this schema", and retiring a partition is +emphatically a schema change — so the pass is gated on the same opt-in as the migration itself. + +### Retention is a partition operation, not a `DELETE` + +Retiring a period is `TRUNCATE TABLE ... WITH (PARTITIONS (n))` followed by +`ALTER PARTITION FUNCTION ... MERGE RANGE`, **in that order**. `MERGE RANGE` on a partition that still holds +rows does not reclaim anything — it *moves* those rows into the neighbouring partition, which is the +opposite of the point. Truncating first deallocates the partition's pages in O(1), and the merge that +follows is then metadata-only against an empty partition. If the truncate fails, the boundary is +deliberately left in place. + +Only boundaries the policy itself would have produced are ever retired, so a hand-added boundary — or one +left over from a different period size — is left strictly alone. + +::: warning +Retiring a period removes its rows. That is the point — it is what makes reclaim O(1) instead of a mass +`DELETE` — but choose `periodsBehind` to match the retention policy you actually want. +::: + +If a process is long-lived enough to outrun the number of periods you provision ahead — an hourly window +especially — run the pass yourself on whatever cadence the period size demands: + +```csharp +// Roll every rolling-window table forward to its current window and retire the partitions that have +// aged past their retention floor. Idempotent, and safe to run from several nodes at once. +await store.Advanced.ApplyRollingPartitionsAsync(token); + +// ...or run just one half +await store.Advanced.RollPartitionsForwardAsync(token); // additive only, never removes data +await store.Advanced.DropAgedRollingPartitionsAsync(token); // retention only +``` + +Each returns Weasel `TablePartitionStatus[]` — one entry per managed table, so a partial failure surfaces +per table rather than taking the whole pass down. + +::: tip +A SQL Server RANGE function always spans `(-infinity, +infinity)`, so the outermost partitions absorb any +row written outside the provisioned window. Unlike PostgreSQL there is no "no partition of relation" error +to guard against and no `DEFAULT` overflow partition to declare. +::: + +## Externally-managed range partitions + +If something genuinely outside Polecat owns the partitions, use the externally-managed variant. Polecat +creates the partition function, scheme, and table once with the supplied initial boundaries and then never +reconciles the partitioning again, so runtime `SPLIT`/`SWITCH`/`MERGE` from elsewhere survives a later +schema apply: + +```csharp +opts.Schema.For() + .PartitionOn(x => x.BucketEnd) + .ByExternallyManagedRange( + new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero), + new DateTimeOffset(2026, 2, 1, 0, 0, 0, TimeSpan.Zero)); +``` + +For an ordinary time-series retention table, prefer [rolling time windows](#rolling-time-windows). Opting +out of Weasel means opting out of its **ordering and dependency management**, not just its DDL generation — +a hand-rolled rebuild that re-creates a partitioned table can easily do so before a helper object its +indexes depend on exists, and the resulting failure looks like a broken index rather than what it is. + ## Limitations - 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. +- One partition scheme per table is a SQL Server constraint, so a document type cannot combine member + RANGE partitioning (declared, rolling, or externally managed) with the store's managed per-tenant + partitioning. ## Managed per-tenant partitioning (#335) diff --git a/src/Polecat.Tests/Storage/rolling_range_partitioning_tests.cs b/src/Polecat.Tests/Storage/rolling_range_partitioning_tests.cs new file mode 100644 index 0000000..85c24f0 --- /dev/null +++ b/src/Polecat.Tests/Storage/rolling_range_partitioning_tests.cs @@ -0,0 +1,525 @@ +using Microsoft.Data.SqlClient; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Polecat.Linq; +using Polecat.Tests.Harness; +using Shouldly; +using Weasel.Core.Partitioning; +using Weasel.SqlServer.Tables.Partitioning; + +namespace Polecat.Tests.Storage; + +/// +/// #386: rolling time-window RANGE partitions for time-series document tables. The window is a pure +/// function of the policy and the clock, so every test here drives a +/// rather than waiting on the calendar. +/// +[Collection("integration")] +public class rolling_range_partitioning_tests : IntegrationContext +{ + private const string Schema = "doc_rolling_partitioning"; + + // Mid-month on purpose: nothing here may depend on "now" landing on a period boundary. + private static readonly DateTimeOffset July = new(2026, 7, 15, 9, 30, 0, TimeSpan.Zero); + + public rolling_range_partitioning_tests(DefaultStoreFixture fixture) : base(fixture) + { + } + + // ---- window shape ------------------------------------------------------------------------- + + [Fact] + public async Task creates_the_whole_declared_window_on_first_migration() + { + const string table = "pc_doc_rollingmetricssample"; + await ResetAsync(table); + + await StoreOptions(opts => + { + opts.DatabaseSchemaName = Schema; + opts.Schema.For() + .PartitionOn(x => x.BucketEnd) + .ByRollingRange(PartitionPeriod.Month, periodsAhead: 1, periodsBehind: 1, + new MutableTimeProvider(July)); + }); + + (await ScalarAsync($"SELECT COUNT(*) FROM sys.partition_functions WHERE name = 'pf_{table}_bucket_end'")) + .ShouldBe(1); + (await ScalarAsync($"SELECT COUNT(*) FROM sys.partition_schemes WHERE name = 'ps_{table}_bucket_end'")) + .ShouldBe(1); + + // June, July, August — the three periods of a (1 ahead, 1 behind) monthly window — plus the + // exclusive end of the newest provisioned period, which keeps the top partition empty so every + // later SPLIT stays metadata-only. + (await BoundaryCountAsync(table)).ShouldBe(4); + (await PartitionCountAsync(table)).ShouldBe(5); + + // The promoted partition column joins the primary key, as SQL Server requires. + (await ScalarAsync( + $""" + SELECT COUNT(*) FROM sys.index_columns ic + JOIN sys.indexes i ON i.object_id = ic.object_id AND i.index_id = ic.index_id + JOIN sys.columns c ON c.object_id = ic.object_id AND c.column_id = ic.column_id + WHERE i.is_primary_key = 1 AND c.name = 'bucket_end' + AND ic.object_id = OBJECT_ID('[{Schema}].[{table}]') + """)).ShouldBe(1); + } + + [Fact] + public async Task documents_land_in_the_partition_for_their_period() + { + const string table = "pc_doc_rollingmetricssample"; + await ResetAsync(table); + + await StoreOptions(opts => + { + opts.DatabaseSchemaName = Schema; + opts.Schema.For() + .PartitionOn(x => x.BucketEnd) + .ByRollingRange(PartitionPeriod.Month, periodsAhead: 1, periodsBehind: 1, + new MutableTimeProvider(July)); + }); + + var june = new RollingMetricsSample { Id = Guid.NewGuid(), BucketEnd = July.AddMonths(-1), Value = 1 }; + var july = new RollingMetricsSample { Id = Guid.NewGuid(), BucketEnd = July, Value = 2 }; + var august = new RollingMetricsSample { Id = Guid.NewGuid(), BucketEnd = July.AddMonths(1), Value = 3 }; + + theSession.Store(june, july, august); + await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); + + var loaded = await theSession.LoadAsync(july.Id, TestContext.Current.CancellationToken); + loaded.ShouldNotBeNull(); + loaded!.BucketEnd.ShouldBe(july.BucketEnd); + + // Three periods, three physical partitions. + (await ScalarAsync( + $""" + SELECT COUNT(DISTINCT $PARTITION.pf_{table}_bucket_end(bucket_end)) + FROM [{Schema}].[{table}] + """)).ShouldBe(3); + } + + [Fact] + public async Task a_row_outside_the_provisioned_window_is_stored_rather_than_rejected() + { + const string table = "pc_doc_overflowmetricssample"; + await ResetAsync(table); + + await StoreOptions(opts => + { + opts.DatabaseSchemaName = Schema; + opts.Schema.For() + .PartitionOn(x => x.BucketEnd) + .ByRollingRange(PartitionPeriod.Month, periodsAhead: 1, periodsBehind: 1, + new MutableTimeProvider(July)); + }); + + // A SQL Server RANGE function always spans (-infinity, +infinity), so the outermost partitions + // absorb anything outside the window — there is no PostgreSQL-style "no partition of relation" + // rejection to guard against, and no DEFAULT partition to declare. + var ancient = new OverflowMetricsSample + { + Id = Guid.NewGuid(), BucketEnd = July.AddYears(-5), Value = 1 + }; + + theSession.Store(ancient); + await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); + + (await theSession.LoadAsync(ancient.Id, TestContext.Current.CancellationToken)) + .ShouldNotBeNull(); + } + + // ---- roll-forward ------------------------------------------------------------------------- + + [Fact] + public async Task rolling_the_window_forward_is_additive_and_never_a_rebuild() + { + const string table = "pc_doc_rolledwindowsample"; + await ResetAsync(table); + + var clock = new MutableTimeProvider(July); + await StoreOptions(opts => + { + opts.DatabaseSchemaName = Schema; + opts.Schema.For() + .PartitionOn(x => x.BucketEnd) + .ByRollingRange(PartitionPeriod.Month, periodsAhead: 1, periodsBehind: 1, clock); + }); + + var doc = new RolledWindowSample { Id = Guid.NewGuid(), BucketEnd = July, Value = 7 }; + theSession.Store(doc); + await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); + + (await BoundaryCountAsync(table)).ShouldBe(4); + + // One month later the declared window is [Jul, Aug, Sep] against a database holding + // [Jun, Jul, Aug]: the June boundary the declaration no longer names is an aged period, not + // drift, so ordinary schema migration only ever SPLITs the new leading edge in. + clock.UtcNow = July.AddMonths(1); + await theDatabase.ApplyAllConfiguredChangesToDatabaseAsync(ct: TestContext.Current.CancellationToken); + + (await BoundaryCountAsync(table)).ShouldBe(5); + (await PartitionCountAsync(table)).ShouldBe(6); + + // A rebuild would have taken the table with it. The row is the proof it was a SPLIT. + var reloaded = await theSession.LoadAsync(doc.Id, TestContext.Current.CancellationToken); + reloaded.ShouldNotBeNull(); + reloaded!.Value.ShouldBe(7); + } + + [Fact] + public async Task one_shared_manager_rolls_every_table_wired_to_it_forward() + { + await ResetAsync("pc_doc_sharedwindowa"); + await ResetAsync("pc_doc_sharedwindowb"); + + var clock = new MutableTimeProvider(July); + var manager = new ManagedRangePartitions( + RollingWindowPolicy.Monthly(periodsAhead: 1, periodsBehind: 1), + column: "bucket_end", sqlDataType: "datetimeoffset", clock); + + await StoreOptions(opts => + { + opts.DatabaseSchemaName = Schema; + opts.Schema.For().PartitionOn(x => x.BucketEnd).ByRollingRange(manager); + opts.Schema.For().PartitionOn(x => x.BucketEnd).ByRollingRange(manager); + }); + + (await BoundaryCountAsync("pc_doc_sharedwindowa")).ShouldBe(4); + (await BoundaryCountAsync("pc_doc_sharedwindowb")).ShouldBe(4); + + // One pass over one manager, both tables rolled forward. + clock.UtcNow = July.AddMonths(1); + var statuses = await theStore.Advanced.RollPartitionsForwardAsync(TestContext.Current.CancellationToken); + + statuses.ShouldAllBe(x => x.Status == PartitionMigrationStatus.Complete); + (await BoundaryCountAsync("pc_doc_sharedwindowa")).ShouldBe(5); + (await BoundaryCountAsync("pc_doc_sharedwindowb")).ShouldBe(5); + } + + // ---- retention ---------------------------------------------------------------------------- + + [Fact] + public async Task the_additive_and_retention_halves_can_be_run_separately() + { + const string table = "pc_doc_retainedwindowsample"; + await ResetAsync(table); + + var clock = new MutableTimeProvider(July); + await StoreOptions(opts => + { + opts.DatabaseSchemaName = Schema; + opts.Schema.For() + .PartitionOn(x => x.BucketEnd) + .ByRollingRange(PartitionPeriod.Month, periodsAhead: 1, periodsBehind: 1, clock); + }); + + var june = new RetainedWindowSample { Id = Guid.NewGuid(), BucketEnd = July.AddMonths(-1), Value = 1 }; + var july = new RetainedWindowSample { Id = Guid.NewGuid(), BucketEnd = July, Value = 2 }; + var august = new RetainedWindowSample { Id = Guid.NewGuid(), BucketEnd = July.AddMonths(1), Value = 3 }; + + theSession.Store(june, july, august); + await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); + + // Two months on, the retention floor is August: June and July have aged out. + clock.UtcNow = July.AddMonths(2); + + // The additive half provisions October and November and touches nothing else — in particular it + // makes no retention decision, so all three rows are still there. + await theStore.Advanced.RollPartitionsForwardAsync(TestContext.Current.CancellationToken); + (await BoundaryCountAsync(table)).ShouldBe(6); + await CountShouldBeAsync(3); + + // The retention half truncates the aged partitions and merges their boundaries away, leaving the + // window exactly as declared. + await theStore.Advanced.DropAgedRollingPartitionsAsync(TestContext.Current.CancellationToken); + (await BoundaryCountAsync(table)).ShouldBe(4); + (await PartitionCountAsync(table)).ShouldBe(5); + + await CountShouldBeAsync(1); + await using var query = theStore.QuerySession(); + var survivor = await query.Query() + .SingleAsync(TestContext.Current.CancellationToken); + survivor.Id.ShouldBe(august.Id); + } + + [Fact] + public async Task the_apply_pass_is_idempotent() + { + const string table = "pc_doc_idempotentwindowsample"; + await ResetAsync(table); + + var clock = new MutableTimeProvider(July); + await StoreOptions(opts => + { + opts.DatabaseSchemaName = Schema; + opts.Schema.For() + .PartitionOn(x => x.BucketEnd) + .ByRollingRange(PartitionPeriod.Month, periodsAhead: 1, periodsBehind: 1, clock); + }); + + theSession.Store(new IdempotentWindowSample + { + Id = Guid.NewGuid(), BucketEnd = July.AddMonths(1), Value = 1 + }); + await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); + + clock.UtcNow = July.AddMonths(2); + + for (var i = 0; i < 3; i++) + { + var statuses = await theStore.Advanced + .ApplyRollingPartitionsAsync(TestContext.Current.CancellationToken); + + statuses.ShouldNotBeEmpty(); + statuses.ShouldAllBe(x => x.Status == PartitionMigrationStatus.Complete); + + (await BoundaryCountAsync(table)).ShouldBe(4); + (await PartitionCountAsync(table)).ShouldBe(5); + } + + // The August row is inside the retained window and survives every pass. + await CountShouldBeAsync(1); + } + + [Fact] + public async Task concurrent_apply_passes_do_not_throw_and_converge_on_the_window() + { + const string table = "pc_doc_concurrentwindowsample"; + await ResetAsync(table); + + var clock = new MutableTimeProvider(July); + await StoreOptions(opts => + { + opts.DatabaseSchemaName = Schema; + opts.Schema.For() + .PartitionOn(x => x.BucketEnd) + .ByRollingRange(PartitionPeriod.Month, periodsAhead: 1, periodsBehind: 1, clock); + }); + + clock.UtcNow = July.AddMonths(2); + + // Several nodes starting at once. A losing SPLIT/MERGE is reported per table rather than thrown, + // and the window still ends up exactly as declared. + await Task.WhenAll(Enumerable.Range(0, 3).Select(_ => + theStore.Advanced.ApplyRollingPartitionsAsync(TestContext.Current.CancellationToken))); + + (await BoundaryCountAsync(table)).ShouldBe(4); + (await PartitionCountAsync(table)).ShouldBe(5); + } + + // ---- host startup ------------------------------------------------------------------------ + + [Fact] + public async Task host_startup_rolls_forward_and_retires_under_ApplyAllDatabaseChangesOnStartup() + { + const string table = "pc_doc_startupwindowsample"; + await ResetAsync(table); + + // Provision the window as it stood in July, with a row in a period that will later age out. + var aged = Guid.NewGuid(); + await StoreOptions(opts => + { + opts.DatabaseSchemaName = Schema; + opts.Schema.For() + .PartitionOn(x => x.BucketEnd) + .ByRollingRange(PartitionPeriod.Month, periodsAhead: 1, periodsBehind: 1, + new MutableTimeProvider(July)); + }); + + theSession.Store(new StartupWindowSample { Id = aged, BucketEnd = July.AddMonths(-1), Value = 1 }); + theSession.Store(new StartupWindowSample + { + Id = Guid.NewGuid(), BucketEnd = July.AddMonths(1), Value = 2 + }); + await theSession.SaveChangesAsync(TestContext.Current.CancellationToken); + + // Two months later the same application starts again. Nothing but the startup opt-in drives the + // roll-forward and the retirement — no application-authored DDL anywhere. + var services = new ServiceCollection(); + services.AddPolecat(opts => + { + opts.ConnectionString = ConnectionSource.ConnectionString; + opts.DatabaseSchemaName = Schema; + opts.UseNativeJsonType = ConnectionSource.SupportsNativeJson; + opts.Schema.For() + .PartitionOn(x => x.BucketEnd) + .ByRollingRange(PartitionPeriod.Month, periodsAhead: 1, periodsBehind: 1, + new MutableTimeProvider(July.AddMonths(2))); + }).ApplyAllDatabaseChangesOnStartup(); + + await using var provider = services.BuildServiceProvider(); + foreach (var hostedService in provider.GetServices()) + { + await hostedService.StartAsync(TestContext.Current.CancellationToken); + } + + (await BoundaryCountAsync(table)).ShouldBe(4); + (await PartitionCountAsync(table)).ShouldBe(5); + + var store = provider.GetRequiredService(); + await using var query = store.QuerySession(); + var remaining = await query.Query().ToListAsync(TestContext.Current.CancellationToken); + remaining.Count.ShouldBe(1); + remaining[0].Id.ShouldNotBe(aged); + } + + // ---- configuration-time guards ----------------------------------------------------------- + + [Fact] + public void rejects_a_non_temporal_partition_key() + { + var options = new StoreOptions(); + + var ex = Should.Throw(() => + options.Schema.For() + .PartitionOn(x => x.Sequence) + .ByRollingRange(PartitionPeriod.Month, periodsAhead: 1, periodsBehind: 1)); + + ex.Message.ShouldContain("DateTime or DateTimeOffset"); + ex.Message.ShouldContain("sequence"); + } + + [Fact] + public void rejects_a_shared_manager_whose_column_does_not_match_the_member() + { + var options = new StoreOptions(); + var manager = new ManagedRangePartitions( + RollingWindowPolicy.Monthly(periodsAhead: 1, periodsBehind: 1), column: "occurred_at"); + + var ex = Should.Throw(() => + options.Schema.For().PartitionOn(x => x.BucketEnd).ByRollingRange(manager)); + + ex.Message.ShouldContain("occurred_at"); + ex.Message.ShouldContain("bucket_end"); + } + + // ---- helpers ----------------------------------------------------------------------------- + + private async Task CountShouldBeAsync(int expected) where T : class + { + await using var query = theStore.QuerySession(); + (await query.Query().CountAsync(TestContext.Current.CancellationToken)).ShouldBe(expected); + } + + private async Task ScalarAsync(string sql) + { + await using var conn = await OpenConnectionAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = sql; + return (int)(await cmd.ExecuteScalarAsync(TestContext.Current.CancellationToken))!; + } + + private Task BoundaryCountAsync(string table) => ScalarAsync( + $""" + SELECT COUNT(*) FROM sys.partition_range_values prv + JOIN sys.partition_functions pf ON pf.function_id = prv.function_id + WHERE pf.name = 'pf_{table}_bucket_end' + """); + + private Task PartitionCountAsync(string table) => ScalarAsync( + $""" + SELECT COUNT(*) FROM sys.partitions p + JOIN sys.objects o ON p.object_id = o.object_id + JOIN sys.schemas s ON o.schema_id = s.schema_id + WHERE s.name = '{Schema}' AND o.name = '{table}' AND p.index_id IN (0, 1) + """); + + /// + /// Drop the table and its database-scoped partition function/scheme so each test starts from + /// nothing regardless of what a prior run left behind. + /// + private async Task ResetAsync(string table) + { + await using var conn = await OpenConnectionAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $""" + IF OBJECT_ID('[{Schema}].[{table}]','U') IS NOT NULL DROP TABLE [{Schema}].[{table}]; + IF EXISTS (SELECT 1 FROM sys.partition_schemes WHERE name='ps_{table}_bucket_end') DROP PARTITION SCHEME [ps_{table}_bucket_end]; + IF EXISTS (SELECT 1 FROM sys.partition_functions WHERE name='pf_{table}_bucket_end') DROP PARTITION FUNCTION [pf_{table}_bucket_end]; + """; + await cmd.ExecuteNonQueryAsync(TestContext.Current.CancellationToken); + } +} + +/// +/// The only thing asks of a clock is "what time is it", so a +/// three-line provider beats taking on Microsoft.Extensions.TimeProvider.Testing for it. +/// +internal sealed class MutableTimeProvider(DateTimeOffset now) : TimeProvider +{ + public DateTimeOffset UtcNow { get; set; } = now; + + public override DateTimeOffset GetUtcNow() => UtcNow; +} + +// One document type per test: a SQL Server partition function and scheme are database-scoped objects +// named from the table, so sharing a type across tests would make them contend for the same objects. +public class RollingMetricsSample +{ + public Guid Id { get; set; } + public DateTimeOffset BucketEnd { get; set; } + public double Value { get; set; } +} + +public class OverflowMetricsSample +{ + public Guid Id { get; set; } + public DateTimeOffset BucketEnd { get; set; } + public double Value { get; set; } +} + +public class RolledWindowSample +{ + public Guid Id { get; set; } + public DateTimeOffset BucketEnd { get; set; } + public double Value { get; set; } +} + +public class RetainedWindowSample +{ + public Guid Id { get; set; } + public DateTimeOffset BucketEnd { get; set; } + public double Value { get; set; } +} + +public class IdempotentWindowSample +{ + public Guid Id { get; set; } + public DateTimeOffset BucketEnd { get; set; } + public double Value { get; set; } +} + +public class ConcurrentWindowSample +{ + public Guid Id { get; set; } + public DateTimeOffset BucketEnd { get; set; } + public double Value { get; set; } +} + +public class StartupWindowSample +{ + public Guid Id { get; set; } + public DateTimeOffset BucketEnd { get; set; } + public double Value { get; set; } +} + +public class SharedWindowA +{ + public Guid Id { get; set; } + public DateTimeOffset BucketEnd { get; set; } + public double Value { get; set; } +} + +public class SharedWindowB +{ + public Guid Id { get; set; } + public DateTimeOffset BucketEnd { get; set; } + public double Value { get; set; } +} + +public class NonTemporalSample +{ + public Guid Id { get; set; } + public int Sequence { get; set; } +} diff --git a/src/Polecat/AdvancedOperations.cs b/src/Polecat/AdvancedOperations.cs index b337dbb..8697a1d 100644 --- a/src/Polecat/AdvancedOperations.cs +++ b/src/Polecat/AdvancedOperations.cs @@ -786,6 +786,53 @@ await manager.DropPartitionFromAllTables( } } + // ---- rolling time-window range partitions (#386) ---- + + /// + /// Roll every document table configured with + /// PartitionOn(x => x.Timestamp).ByRollingRange(...) forward to its current window and + /// retire the periods that have aged past the policy's retention floor. Idempotent, and safe to + /// run on every startup. + /// + /// Polecat already runs this at startup when the host opted into + /// ApplyAllDatabaseChangesOnStartup(). Call it yourself on whatever cadence the period + /// size demands (an hourly window needs a far tighter cadence than a monthly one) if the process + /// is long-lived enough to outrun the number of periods provisioned ahead. + /// + /// + /// Retiring a period destroys the rows in it. That is the point: it is what makes time-based + /// retention a partition TRUNCATE — an O(1) page deallocation — instead of a mass + /// DELETE. Use for the purely additive half. + /// + /// + /// Per-table statuses for every table wired to a rolling-window strategy. + /// + public Task ApplyRollingPartitionsAsync(CancellationToken token = default) + => RollingPartitions.ApplyAsync(RollingPartitionDatabases(), NullLogger.Instance, + rollForward: true, dropAged: true, token); + + /// + /// The purely additive half of : SPLIT in the boundaries + /// of the current window that the partition function does not carry yet. Nothing is truncated or + /// merged, so this is safe to run without making a retention decision. + /// + public Task RollPartitionsForwardAsync(CancellationToken token = default) + => RollingPartitions.ApplyAsync(RollingPartitionDatabases(), NullLogger.Instance, + rollForward: true, dropAged: false, token); + + /// + /// The retention half of : truncate and then + /// MERGE RANGE away every rolling-window period older than its policy's retention floor. + /// Only boundaries the policy itself would have produced are considered, so a hand-added boundary + /// — or one left over from a different period size — is left strictly alone. + /// + public Task DropAgedRollingPartitionsAsync(CancellationToken token = default) + => RollingPartitions.ApplyAsync(RollingPartitionDatabases(), NullLogger.Instance, + rollForward: false, dropAged: true, token); + + private IEnumerable RollingPartitionDatabases() => + _store.Options.Tenancy?.AllDatabases() ?? [_store.Database]; + private Events.EventGraph AssertManagedTenantPartitioning() { var events = _store.Events; diff --git a/src/Polecat/DocumentStore.DocumentStoreUsage.cs b/src/Polecat/DocumentStore.DocumentStoreUsage.cs index 8a0731c..c78449a 100644 --- a/src/Polecat/DocumentStore.DocumentStoreUsage.cs +++ b/src/Polecat/DocumentStore.DocumentStoreUsage.cs @@ -161,7 +161,7 @@ private DocumentMappingDescriptor BuildMappingDescriptor( UseNumericRevisions = mapping.UseNumericRevisions, SubClassCount = mapping.SubClasses.Count, SubClasses = mapping.SubClasses.Select(x => TypeDescriptor.For(x.DocumentType)).ToArray(), - PartitioningStrategy = mapping.Partitioning == null ? null : "Range", + PartitioningStrategy = PartitioningStrategyName(mapping.Partitioning), Partitioning = BuildPartitioning(mapping.Partitioning), Ddl = ddl, }; @@ -179,13 +179,27 @@ private DocumentMappingDescriptor BuildMappingDescriptor( return null; } - var names = partitioning.Boundaries - .Select(b => Convert.ToString(b, CultureInfo.InvariantCulture) ?? string.Empty) - .ToArray(); + var strategy = PartitioningStrategyName(partitioning)!; - return new PartitioningDescriptor { Strategy = "Range", PartitionNames = names }; + // #386: a rolling window has no declared boundary list — the window is a function of the policy + // and the clock, so report the boundaries it expects to exist right now. + var names = partitioning.RollingWindow is { } rollingWindow + ? rollingWindow.Boundaries() + : partitioning.Boundaries + .Select(b => Convert.ToString(b, CultureInfo.InvariantCulture) ?? string.Empty) + .ToArray(); + + return new PartitioningDescriptor { Strategy = strategy, PartitionNames = names }; } + private static string? PartitioningStrategyName(Storage.DocumentPartitioning? partitioning) => + partitioning switch + { + null => null, + { RollingWindow: not null } => "RollingRange", + _ => "Range" + }; + private static string WriteSchemaCreationDdl( Storage.DocumentMapping mapping, SqlServerMigrator migrator) diff --git a/src/Polecat/Internal/PolecatActivator.cs b/src/Polecat/Internal/PolecatActivator.cs index c49e06d..24ef07a 100644 --- a/src/Polecat/Internal/PolecatActivator.cs +++ b/src/Polecat/Internal/PolecatActivator.cs @@ -34,6 +34,16 @@ public async Task StartAsync(CancellationToken cancellationToken) { var documentStore = (DocumentStore)_store; await documentStore.Database.ApplyAllConfiguredChangesToDatabaseAsync(ct: cancellationToken); + + // #386: roll every configured rolling-window RANGE partition forward and retire the aged + // ones. The migration above already provisions the leading edge — with a rolling-window + // manager attached the delta is additive, a SPLIT rather than a rebuild — but migration + // never removes data, so the retention half has to be driven separately. Gated on the same + // opt-in as the migration itself: applying changes on startup is how a host says "Polecat + // owns this schema", and retiring a partition is emphatically a schema change. + var databases = _store.Options.Tenancy?.AllDatabases() ?? [documentStore.Database]; + await Storage.RollingPartitions.ApplyAsync(databases, _logger, rollForward: true, dropAged: true, + cancellationToken); } // Run initial data seeders after schema migration diff --git a/src/Polecat/Storage/DocumentMappingExpression.cs b/src/Polecat/Storage/DocumentMappingExpression.cs index 315bd79..1e41520 100644 --- a/src/Polecat/Storage/DocumentMappingExpression.cs +++ b/src/Polecat/Storage/DocumentMappingExpression.cs @@ -1,5 +1,7 @@ using System.Diagnostics.CodeAnalysis; using System.Linq.Expressions; +using Weasel.Core.Partitioning; +using Weasel.SqlServer.Tables.Partitioning; namespace Polecat.Storage; @@ -184,11 +186,17 @@ public DocumentMappingExpression PartitionByRange( /// /// #255: begin a fluent declaration of RANGE partitioning on a member, mirroring Marten's - /// PartitionOn(x => x.Member). Follow with - /// (Polecat manages + rolls the boundaries) or - /// (Polecat provisions the - /// partitioned table once, then leaves the partitions to be managed externally — the - /// time-series-retention pattern). + /// PartitionOn(x => x.Member). Follow with: + /// + /// — a fixed set of boundaries + /// Polecat owns and rolls forward in place as you add to the list; + /// + /// — a rolling time window Polecat provisions ahead of and retires behind the clock, which is + /// the supported way to run a time-series table (#386); + /// — Polecat + /// provisions the partitioned table once and never touches the partitions again, for when + /// something genuinely outside Polecat owns them. + /// /// public PartitioningExpression PartitionOn(Expression> member) { @@ -199,9 +207,23 @@ public PartitioningExpression PartitionOn(Expression(Expression> member, TValue[] boundaries, bool externallyManaged) { - var idMemberName = DocumentMapping.FindIdProperty(typeof(T))?.Name ?? "Id"; - Partitioning = DocumentPartitioning.For(member, boundaries, idMemberName, externallyManaged); + Partitioning = DocumentPartitioning.For(member, boundaries, IdMemberName(), externallyManaged); + } + + /// + /// #386: internal hook used by to set a + /// rolling-time-window descriptor. Returns the manager that owns the window. + /// + internal ManagedRangePartitions SetRollingWindow(Expression> member, + RollingWindowPolicy policy, TimeProvider? timeProvider, ManagedRangePartitions? prebuilt) + { + Partitioning = DocumentPartitioning.ForRollingWindow(member, IdMemberName(), policy, timeProvider, + prebuilt); + + return Partitioning.RollingWindow!; } + + private static string IdMemberName() => DocumentMapping.FindIdProperty(typeof(T))?.Name ?? "Id"; } /// @@ -230,10 +252,77 @@ public DocumentMappingExpression ByRange(params TValue[] boundaries) } /// - /// #255: externally-managed RANGE partitioning. Polecat creates the partition function/scheme + /// #386: RANGE-partition the table over a rolling time window that Polecat itself owns — + /// it provisions the periods at the leading edge and retires the aged ones at the trailing edge + /// on the same schedule it applies every other schema change. This is the supported way to run a + /// time-series document table: retention becomes a partition TRUNCATE + MERGE RANGE + /// (an O(1) page deallocation, not a mass DELETE) without giving up Weasel's schema + /// ordering and dependency management the way does. + /// + /// The partitioned member must be a DateTime or DateTimeOffset, and the whole + /// window is computed in UTC. Polecat promotes the member into a real column and adds it to + /// the primary key, as SQL Server requires of a partitioned table's unique index. + /// + /// + /// The size of a single partition — hour, day, week, month or year. + /// + /// How many periods beyond the current one to provision. At least one is strongly recommended so + /// rows written at the very end of a period always have a partition waiting for them. + /// + /// + /// How many completed periods to retain. Periods older than this are retired by the retention + /// pass, which destroys their rows by design. + /// + /// + /// Clock used to resolve "now". Defaults to ; supply a fake to + /// roll the window forward deterministically in tests. + /// + /// + /// The manager that owns the window, so Filegroup can be set or the same instance shared + /// with another document type. + /// + /// + public ManagedRangePartitions ByRollingRange(PartitionPeriod period, int periodsAhead, int periodsBehind, + TimeProvider? timeProvider = null) + => ByRollingRange(new RollingWindowPolicy(period, periodsAhead, periodsBehind), timeProvider); + + /// + /// #386: RANGE-partition the table over the rolling time window described by + /// . See + /// . + /// + public ManagedRangePartitions ByRollingRange(RollingWindowPolicy policy, TimeProvider? timeProvider = null) + { + ArgumentNullException.ThrowIfNull(policy); + + return _parent.SetRollingWindow(_member, policy, timeProvider, prebuilt: null); + } + + /// + /// #386: RANGE-partition the table over a rolling time window owned by a pre-built + /// . Pass the same manager instance to several + /// document types to roll every one of their tables forward in a single pass. The manager's + /// column and SQL data type must match what the partition member resolves to. + /// + public ManagedRangePartitions ByRollingRange(ManagedRangePartitions partitions) + { + ArgumentNullException.ThrowIfNull(partitions); + + return _parent.SetRollingWindow(_member, partitions.Policy, timeProvider: null, partitions); + } + + /// + /// #255: externally-managed RANGE partitioning: Polecat creates the partition function/scheme /// and table once (with the supplied ) and then never - /// reconciles the partitioning, so the app/DBA can SPLIT new partitions and SWITCH/DROP old - /// ones at runtime for time-series retention without a later schema apply clobbering them. + /// reconciles the partitioning, so whatever owns the partitions can SPLIT new ones and + /// SWITCH/DROP old ones at runtime without a later schema apply clobbering them. + /// + /// Reach for this only when something genuinely outside Polecat owns the partitions. For an + /// ordinary time-series retention table prefer + /// , which keeps the whole + /// lifecycle inside Weasel's schema model instead of leaving the application to hand-write + /// NEXT USED/SPLIT/MERGE DDL on a schedule forever. + /// /// public DocumentMappingExpression ByExternallyManagedRange(params TValue[] initialBoundaries) { diff --git a/src/Polecat/Storage/DocumentPartitioning.cs b/src/Polecat/Storage/DocumentPartitioning.cs index 11b67b2..fee76c5 100644 --- a/src/Polecat/Storage/DocumentPartitioning.cs +++ b/src/Polecat/Storage/DocumentPartitioning.cs @@ -1,6 +1,8 @@ using System.Linq.Expressions; using System.Reflection; using System.Text; +using Weasel.Core.Partitioning; +using Weasel.SqlServer.Tables.Partitioning; namespace Polecat.Storage; @@ -16,7 +18,8 @@ internal sealed class DocumentPartitioning private readonly Func? _getter; private DocumentPartitioning(string columnName, string sqlDataType, bool partitionOnId, - IReadOnlyList boundaries, Func? getter, bool externallyManaged) + IReadOnlyList boundaries, Func? getter, bool externallyManaged, + ManagedRangePartitions? rollingWindow = null) { ColumnName = columnName; SqlDataType = sqlDataType; @@ -24,6 +27,7 @@ private DocumentPartitioning(string columnName, string sqlDataType, bool partiti Boundaries = boundaries; _getter = getter; ExternallyManaged = externallyManaged; + RollingWindow = rollingWindow; } /// The partition column name. id when is true. @@ -54,6 +58,15 @@ private DocumentPartitioning(string columnName, string sqlDataType, bool partiti /// public bool ExternallyManaged { get; } + /// + /// #386: when set, the partition boundaries are a pure function of a + /// and the clock rather than a + /// caller-supplied list, and Weasel owns every statement that moves the window: NEXT USED + + /// SPLIT RANGE at the leading edge, partition TRUNCATE + MERGE RANGE at the trailing one. + /// is empty in this mode — ask the manager for the current window. + /// + public ManagedRangePartitions? RollingWindow { get; } + /// Extract the partition value from a document instance for the write path. public object GetValue(object document) { @@ -78,23 +91,101 @@ public static DocumentPartitioning For( IReadOnlyList boundaries, string idMemberName, bool externallyManaged = false) + { + var boxed = boundaries.Select(b => (object)b!).ToArray(); + var (column, sqlType, partitionOnId, getter) = Resolve(member, idMemberName); + + return new DocumentPartitioning(column, sqlType, partitionOnId, boxed, getter, externallyManaged); + } + + /// + /// #386: resolve a member expression into a descriptor whose boundaries are owned by a rolling + /// time window rather than declared up front. Pass to share one + /// manager across several document types so a single pass rolls all of their tables forward; + /// otherwise a manager is built for the resolved column. + /// + public static DocumentPartitioning ForRollingWindow( + Expression> member, + string idMemberName, + RollingWindowPolicy policy, + TimeProvider? timeProvider, + ManagedRangePartitions? prebuilt) + { + var (column, sqlType, partitionOnId, getter) = Resolve(member, idMemberName); + AssertTemporalPartitionKey(policy, typeof(TValue), column); + + ManagedRangePartitions manager; + if (prebuilt == null) + { + manager = new ManagedRangePartitions(policy, column, sqlType, timeProvider); + } + else + { + // A shared manager carries the column and function-parameter type with it, so it only fits + // a document type whose partition member resolves to exactly that pair. Silently going with + // the manager's column would partition the table on a column Polecat never promoted. + AssertManagerMatchesMember(prebuilt, column, sqlType); + manager = prebuilt; + } + + return new DocumentPartitioning(column, sqlType, partitionOnId, [], getter, + externallyManaged: false, manager); + } + + private static (string Column, string SqlDataType, bool PartitionOnId, Func? Getter) + Resolve(Expression> member, string idMemberName) { var memberInfo = ResolveMember(member); var sqlType = ToSqlServerType(typeof(TValue)); - var boxed = boundaries.Select(b => (object)b!).ToArray(); if (string.Equals(memberInfo.Name, idMemberName, StringComparison.Ordinal)) { - return new DocumentPartitioning("id", sqlType, partitionOnId: true, boxed, getter: null, - externallyManaged); + return ("id", sqlType, true, null); } - var column = ToSnakeCase(memberInfo.Name); var compiled = member.Compile(); - Func getter = doc => compiled((T)doc); - return new DocumentPartitioning(column, sqlType, partitionOnId: false, boxed, getter, - externallyManaged); + return (ToSnakeCase(memberInfo.Name), sqlType, false, doc => compiled((T)doc)); + } + + /// + /// A rolling window is a function of the clock, so the partition key has to actually be a point + /// in time. Failing here — at configuration — turns what would otherwise surface as an opaque + /// SQL Server partition-function type error during the first migration into a message that names + /// the member. + /// + private static void AssertTemporalPartitionKey(RollingWindowPolicy policy, Type valueType, string column) + { + var memberType = Nullable.GetUnderlyingType(valueType) ?? valueType; + + if (memberType != typeof(DateTimeOffset) && memberType != typeof(DateTime)) + { + throw new InvalidOperationException( + $"A rolling range partition ({policy}) has to be keyed on a DateTime or DateTimeOffset " + + $"member, but '{column}' of {typeof(T).Name} is {memberType.Name}. Use " + + "PartitionOn(...).ByRange(...) for a non-temporal partition key."); + } + } + + private static void AssertManagerMatchesMember(ManagedRangePartitions manager, string column, + string sqlDataType) + { + if (!string.Equals(manager.Column, column, StringComparison.OrdinalIgnoreCase)) + { + throw new InvalidOperationException( + $"The supplied ManagedRangePartitions partitions on column '{manager.Column}', but the " + + $"partition member of {typeof(T).Name} resolves to column '{column}'. A shared " + + "rolling-window manager can only be used by document types whose partition member maps " + + "to the same column name."); + } + + if (!string.Equals(manager.SqlDataType, sqlDataType, StringComparison.OrdinalIgnoreCase)) + { + throw new InvalidOperationException( + $"The supplied ManagedRangePartitions declares its partition function over " + + $"'{manager.SqlDataType}', but the partition member of {typeof(T).Name} maps to " + + $"'{sqlDataType}'."); + } } private static MemberInfo ResolveMember(Expression> member) diff --git a/src/Polecat/Storage/DocumentTable.cs b/src/Polecat/Storage/DocumentTable.cs index 06887d9..ef5972a 100644 --- a/src/Polecat/Storage/DocumentTable.cs +++ b/src/Polecat/Storage/DocumentTable.cs @@ -125,10 +125,21 @@ public DocumentTable(DocumentMapping mapping) AddColumn(partitioning.ColumnName, partitioning.SqlDataType).AsPrimaryKey().NotNull(); } - var range = PartitionByRange(partitioning.ColumnName, partitioning.SqlDataType); - foreach (var boundary in partitioning.Boundaries) + // #386: a rolling time window derives its boundaries from the policy and the clock, so the + // manager IS the partitioning strategy — the same instance every DocumentTable rebuild hands + // to Weasel, which is how ManagedRangePartitions.ResolveManagedTables (reference identity) + // finds the tables it owns during the roll-forward/retention pass. + if (partitioning.RollingWindow is { } rollingWindow) { - range.AddBoundary(boundary); + this.PartitionByRollingWindow(rollingWindow); + } + else + { + var range = PartitionByRange(partitioning.ColumnName, partitioning.SqlDataType); + foreach (var boundary in partitioning.Boundaries) + { + range.AddBoundary(boundary); + } } } } diff --git a/src/Polecat/Storage/RollingPartitions.cs b/src/Polecat/Storage/RollingPartitions.cs new file mode 100644 index 0000000..a872153 --- /dev/null +++ b/src/Polecat/Storage/RollingPartitions.cs @@ -0,0 +1,82 @@ +using Microsoft.Extensions.Logging; +using Weasel.SqlServer.Tables; +using Weasel.SqlServer.Tables.Partitioning; + +namespace Polecat.Storage; + +/// +/// #386: drives the rolling time-window RANGE partitions configured through +/// +/// (built on weasel#401). +/// +/// Provisioning at the leading edge is already covered by ordinary schema migration: with a +/// attached, CreateDelta is purely additive, so a +/// window that has rolled forward diffs as "SPLIT in the new boundary" rather than as a rebuild +/// of the partition function and every table sitting on it. What migration deliberately does NOT +/// do is remove data, so retiring the aged periods at the trailing edge has to be driven +/// explicitly — that is what this type is for, and it is why the maintenance pass runs alongside +/// the startup schema application rather than inside it. +/// +/// +/// The managers are discovered from each database's own schema objects rather than tracked on +/// , so this stays correct for a document type that grows a rolling +/// window later, and re-running it is always idempotent. +/// +/// +internal static class RollingPartitions +{ + /// + /// Every distinct rolling-window manager attached to a table of this database. Reference identity + /// is the key, exactly as matches tables + /// back to their manager — so one manager shared across several document types rolls all of their + /// tables forward in a single pass. + /// + public static IReadOnlyList ManagersFor(PolecatDatabase database) + { + var managers = new List(); + + foreach (var table in database.AllObjects().OfType()) + { + if (table.SqlServerPartitioning is ManagedRangePartitions manager + && !managers.Any(x => ReferenceEquals(x, manager))) + { + managers.Add(manager); + } + } + + return managers; + } + + /// + /// Run the maintenance pass over every database. + /// + /// SPLIT in the boundaries of the current window that do not exist yet. + /// + /// Retire the periods that have fallen below the policy's retention floor. This removes data by + /// design — the partition TRUNCATE is what reclaims the storage in O(1). + /// + public static async Task ApplyAsync(IEnumerable databases, + ILogger logger, bool rollForward, bool dropAged, CancellationToken token) + { + var statuses = new List(); + + foreach (var database in databases) + { + foreach (var manager in ManagersFor(database)) + { + var results = (rollForward, dropAged) switch + { + (true, true) => await manager.ApplyAsync(database, logger, token).ConfigureAwait(false), + (true, false) => await manager.RollForwardAsync(database, logger, token).ConfigureAwait(false), + (false, true) => await manager.DropAgedPartitionsAsync(database, logger, token) + .ConfigureAwait(false), + _ => [] + }; + + statuses.AddRange(results); + } + } + + return statuses.ToArray(); + } +}