Partition Support

CHANGELOG.md

@@ -10,6 +10,12 @@ This package uses **lockstep major versioning** with the core NSchema package: `
 
 As a consequence, breaking changes that are specific to this provider (rather than the core API) are signalled by a **minor version bump** rather than a major one, and called out explicitly in this changelog.
 
+## [5.7.0] - 2026-10-07
+
+### Added
+
+- **Publication support.**, Includes column lists, row filters and `TABLES IN SCHEMA` on PostgreSQL 15 and later.
+
 ## [5.6.1] - 2026-08-13
 
 ### Fixed
@@ -162,6 +168,7 @@ First stable release of the PostgreSQL provider for NSchema, tracking the 1.0 re
 - `SqlType.Citext` and `SqlType.Jsonb` Postgres-specific type helpers on `SqlType`.
 - SourceLink and symbol packages (`.snupkg`) published alongside the main package for source-level debugging.
 
+[5.7.0]: https://github.com/nschema-org/NSchema.Postgres/compare/v5.6.1...v5.7.0
 [5.6.1]: https://github.com/nschema-org/NSchema.Postgres/compare/v5.6.0...v5.6.1
 [5.6.0]: https://github.com/nschema-org/NSchema.Postgres/compare/v5.5.0...v5.6.0
 [5.5.0]: https://github.com/nschema-org/NSchema.Postgres/compare/v5.4.0...v5.5.0

Directory.Packages.props

@@ -4,19 +4,19 @@
     <CentralPackageTransitivePinningEnabled>true</CentralPackageTransitivePinningEnabled>
   </PropertyGroup>
   <ItemGroup>
-    <PackageVersion Include="Microsoft.Extensions.TimeProvider.Testing" Version="10.9.0" />
-    <PackageVersion Include="Microsoft.SourceLink.GitHub" Version="10.0.400" />
-    <PackageVersion Include="Microsoft.Testing.Extensions.CodeCoverage" Version="18.10.0" />
-    <PackageVersion Include="Microsoft.Testing.Extensions.TrxReport" Version="2.3.3" />
+    <PackageVersion Include="Microsoft.Extensions.TimeProvider.Testing" Version="10.10.0" />
+    <PackageVersion Include="Microsoft.SourceLink.GitHub" Version="10.0.401" />
+    <PackageVersion Include="Microsoft.Testing.Extensions.CodeCoverage" Version="18.12.0" />
+    <PackageVersion Include="Microsoft.Testing.Extensions.TrxReport" Version="2.5.1" />
     <PackageVersion Include="Npgsql.DependencyInjection" Version="10.0.3" />
-    <PackageVersion Include="NSchema.Core" Version="5.11.0" />
-    <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="18.9.0" />
+    <PackageVersion Include="NSchema.Core" Version="5.14.0" />
+    <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="18.10.1" />
     <PackageVersion Include="Npgsql" Version="10.0.3" />
     <PackageVersion Include="NSubstitute" Version="6.2.0" />
     <PackageVersion Include="Shouldly" Version="4.3.0" />
     <PackageVersion Include="SSH.NET" Version="2026.0.0" />
-    <PackageVersion Include="Testcontainers.PostgreSql" Version="4.14.0" />
-    <PackageVersion Include="Verify.XunitV3" Version="31.28.0" />
-    <PackageVersion Include="xunit.v3" Version="4.0.0" />
+    <PackageVersion Include="Testcontainers.PostgreSql" Version="4.15.0" />
+    <PackageVersion Include="Verify.XunitV3" Version="32.0.1" />
+    <PackageVersion Include="xunit.v3" Version="4.0.1" />
   </ItemGroup>
 </Project>

NSchema.Postgres.slnx

@@ -3,6 +3,7 @@
     <File Path=".editorconfig" />
     <File Path=".github\workflows\cicd.yml" />
     <File Path="Directory.Packages.props" />
+    <File Path="dotnet-tools.json" />
     <File Path="ritten.json" />
   </Folder>
   <Folder Name="/postgres/">

dotnet-tools.json

@@ -3,7 +3,7 @@
   "isRoot": true,
   "tools": {
     "ritten": {
-      "version": "0.6.0",
+      "version": "0.22.0",
       "commands": [
         "ritten"
       ],

src/NSchema.Postgres/Models/PublicationRow.cs

@@ -0,0 +1,10 @@
+namespace NSchema.Postgres.Models;
+
+internal sealed record PublicationRow(
+    string Name,
+    bool AllTables,
+    bool Insert,
+    bool Update,
+    bool Delete,
+    bool Truncate,
+    string? Comment);

src/NSchema.Postgres/Models/PublishedTableRow.cs

@@ -0,0 +1,8 @@
+namespace NSchema.Postgres.Models;
+
+internal sealed record PublishedTableRow(
+    string Publication,
+    string Schema,
+    string Table,
+    string[]? Columns,
+    string? Filter);

src/NSchema.Postgres/Models/TableRow.cs

@@ -1,3 +1,3 @@
 namespace NSchema.Postgres.Models;
 
-internal sealed record TableRow(string Schema, string Name);
+internal sealed record TableRow(string Schema, string Name, char ReplicaIdentity = 'd', string? ReplicaIdentityIndex = null);

src/NSchema.Postgres/NSchema.Postgres.csproj

@@ -31,7 +31,7 @@
     <EmbedUntrackedSources>true</EmbedUntrackedSources>
     <IncludeSymbols>true</IncludeSymbols>
     <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-    <Version>5.6.1</Version>
+    <Version>5.7.0</Version>
     <AssemblyVersion>$(Version.Split('-')[0])</AssemblyVersion>
     <FileVersion>$(Version.Split('-')[0])</FileVersion>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>

src/NSchema.Postgres/Sql/PostgresDatabaseIntrospector.cs

@@ -10,6 +10,7 @@
 using NSchema.Model.Enums;
 using NSchema.Model.Extensions;
 using NSchema.Model.Indexes;
+using NSchema.Model.Publications;
 using NSchema.Model.Routines;
 using NSchema.Model.Schemas;
 using NSchema.Model.Sequences;
@@ -93,6 +94,9 @@ private async ValueTask<Result<Database>> Read(PlanningScope scope, Cancellation
         // through NSchema comes back looking the same and no longer routing a single row the way it used to.
         var partitioning = await QueryPartitioning(conn, schemas, cancellationToken);
 
+        // Publications are database-global, so the schema list does not narrow them.
+        var publications = await QueryPublications(conn, cancellationToken);
+
         var database = Build(
             tables, columns, primaryKeys, foreignKeys, uniqueConstraints, checkConstraints, exclusionConstraints, indexes, triggers,
             schemaComments, tableComments, columnComments, indexComments, constraintComments,
@@ -102,6 +106,7 @@ private async ValueTask<Result<Database>> Read(PlanningScope scope, Cancellation
             functions, functionComments, procedures, procedureComments, aggregates, aggregateComments,
             extensions
         );
+        database.Publications.AddRange(publications);
 
         return partitioning.Count == 0
             ? Result.Success(database)
@@ -146,6 +151,107 @@ ORDER BY 1
 
     // ── Queries ───────────────────────────────────────────────────────────────
 
+    // Column lists, row filters and TABLES IN SCHEMA arrived in Postgres 15; before it, none of their catalog
+    // columns exist.
+    private static async Task<List<Publication>> QueryPublications(NpgsqlConnection conn, CancellationToken ct)
+    {
+        var modern = conn.PostgreSqlVersion.Major >= 15;
+
+        var publications = new List<PublicationRow>();
+        await using (var cmd = conn.CreateCommand())
+        {
+            cmd.CommandText = """
+                SELECT p.pubname, p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate,
+                       obj_description(p.oid, 'pg_publication')
+                FROM pg_publication p
+                ORDER BY p.pubname
+                """;
+            await using var reader = await cmd.ExecuteReaderAsync(ct);
+            while (await reader.ReadAsync(ct))
+            {
+                publications.Add(new PublicationRow(reader.GetString(0), reader.GetBoolean(1), reader.GetBoolean(2),
+                    reader.GetBoolean(3), reader.GetBoolean(4), reader.GetBoolean(5), reader.IsDBNull(6) ? null : reader.GetString(6)));
+            }
+        }
+
+        var tables = new List<PublishedTableRow>();
+        await using (var cmd = conn.CreateCommand())
+        {
+            cmd.CommandText = modern
+                ? """
+                  SELECT p.pubname, n.nspname, c.relname,
+                         (SELECT array_agg(a.attname ORDER BY a.attnum)
+                          FROM pg_attribute a
+                          WHERE a.attrelid = pr.prrelid AND a.attnum = ANY(pr.prattrs::int2[])),
+                         pg_get_expr(pr.prqual, pr.prrelid)
+                  FROM pg_publication_rel pr
+                  JOIN pg_publication p ON p.oid = pr.prpubid
+                  JOIN pg_class c ON c.oid = pr.prrelid
+                  JOIN pg_namespace n ON n.oid = c.relnamespace
+                  ORDER BY p.pubname, n.nspname, c.relname
+                  """
+                : """
+                  SELECT p.pubname, n.nspname, c.relname, NULL::text[], NULL::text
+                  FROM pg_publication_rel pr
+                  JOIN pg_publication p ON p.oid = pr.prpubid
+                  JOIN pg_class c ON c.oid = pr.prrelid
+                  JOIN pg_namespace n ON n.oid = c.relnamespace
+                  ORDER BY p.pubname, n.nspname, c.relname
+                  """;
+            await using var reader = await cmd.ExecuteReaderAsync(ct);
+            while (await reader.ReadAsync(ct))
+            {
+                tables.Add(new PublishedTableRow(reader.GetString(0), reader.GetString(1), reader.GetString(2),
+                    reader.IsDBNull(3) ? null : reader.GetFieldValue<string[]>(3), reader.IsDBNull(4) ? null : reader.GetString(4)));
+            }
+        }
+
+        var schemas = new List<(string Publication, string Schema)>();
+        if (modern)
+        {
+            await using var cmd = conn.CreateCommand();
+            cmd.CommandText = """
+                SELECT p.pubname, n.nspname
+                FROM pg_publication_namespace pn
+                JOIN pg_publication p ON p.oid = pn.pnpubid
+                JOIN pg_namespace n ON n.oid = pn.pnnspid
+                ORDER BY p.pubname, n.nspname
+                """;
+            await using var reader = await cmd.ExecuteReaderAsync(ct);
+            while (await reader.ReadAsync(ct))
+            {
+                schemas.Add((reader.GetString(0), reader.GetString(1)));
+            }
+        }
+
+        return [.. publications.Select(p => new Publication
+        {
+            Name = p.Name,
+            AllTables = p.AllTables,
+            Tables = [.. tables.Where(t => t.Publication == p.Name).Select(t => new PublishedTable(
+                new ObjectAddress(t.Schema, t.Table),
+                t.Columns?.Select(c => new SqlIdentifier(c)).ToList(),
+                t.Filter is { } filter ? new SqlText(filter) : null))],
+            Schemas = [.. schemas.Where(s => s.Publication == p.Name).Select(s => new SqlIdentifier(s.Schema))],
+            Operations = (p.Insert ? PublishedOperations.Insert : PublishedOperations.None)
+                | (p.Update ? PublishedOperations.Update : PublishedOperations.None)
+                | (p.Delete ? PublishedOperations.Delete : PublishedOperations.None)
+                | (p.Truncate ? PublishedOperations.Truncate : PublishedOperations.None),
+            Comment = p.Comment,
+        })];
+    }
+
+    // Postgres's default ('d', the primary key) is the engine's default, which the model leaves unstated. An index
+    // identity whose index has since been dropped identifies nothing, as Postgres treats it.
+    internal static ReplicaIdentity? MapReplicaIdentity(TableRow row) => row.ReplicaIdentity switch
+    {
+        'f' => ReplicaIdentity.Full,
+        'n' => ReplicaIdentity.Nothing,
+        'i' when row.ReplicaIdentityIndex is { } index => ReplicaIdentity.UsingIndex(index),
+        'i' => ReplicaIdentity.Nothing,
+        _ => null,
+    };
+
     private static void AddSchemasParameter(NpgsqlCommand cmd, string[]? schemas)
     {
         var parameter = cmd.Parameters.Add("schemas", NpgsqlDbType.Array | NpgsqlDbType.Text);
@@ -156,22 +262,27 @@ private static async Task<List<TableRow>> QueryTables(NpgsqlConnection conn, str
     {
         var rows = new List<TableRow>();
         await using var cmd = conn.CreateCommand();
+        // The replica identity rides along: relreplident, and the index when it names one.
         cmd.CommandText = """
-            SELECT table_schema, table_name
-            FROM information_schema.tables
-            WHERE table_type = 'BASE TABLE'
-            AND (@schemas::text[] IS NULL OR table_schema = ANY(@schemas))
-            AND table_schema NOT IN ('pg_catalog', 'information_schema')
-            AND table_schema NOT LIKE 'pg\_toast%' ESCAPE '\'
-            AND table_schema NOT LIKE 'pg\_temp%' ESCAPE '\'
-            ORDER BY table_schema, table_name
+            SELECT t.table_schema, t.table_name, c.relreplident, ri.relname
+            FROM information_schema.tables t
+            JOIN pg_namespace n ON n.nspname = t.table_schema
+            JOIN pg_class c ON c.relnamespace = n.oid AND c.relname = t.table_name
+            LEFT JOIN pg_index i ON i.indrelid = c.oid AND i.indisreplident
+            LEFT JOIN pg_class ri ON ri.oid = i.indexrelid
+            WHERE t.table_type = 'BASE TABLE'
+            AND (@schemas::text[] IS NULL OR t.table_schema = ANY(@schemas))
+            AND t.table_schema NOT IN ('pg_catalog', 'information_schema')
+            AND t.table_schema NOT LIKE 'pg\_toast%' ESCAPE '\'
+            AND t.table_schema NOT LIKE 'pg\_temp%' ESCAPE '\'
+            ORDER BY t.table_schema, t.table_name
             """;
         AddSchemasParameter(cmd, schemas);
 
         await using var reader = await cmd.ExecuteReaderAsync(ct);
         while (await reader.ReadAsync(ct))
         {
-            rows.Add(new TableRow(reader.GetString(0), reader.GetString(1)));
+            rows.Add(new TableRow(reader.GetString(0), reader.GetString(1), reader.GetChar(2), reader.IsDBNull(3) ? null : reader.GetString(3)));
         }
 
         return rows;
@@ -1832,6 +1943,7 @@ List<TableGrantRow> allTableGrants
             Indexes = [.. idxs],
             Grants = [.. grants],
             Triggers = [.. triggers],
+            ReplicaIdentity = MapReplicaIdentity(tableRow),
         };
     }
 

src/NSchema.Postgres/Sql/PostgresSqlDialect.Publications.cs

@@ -0,0 +1,84 @@
+using NSchema.Model.Publications;
+using NSchema.Plan.Domain;
+using NSchema.Plan.Domain.Publications;
+
+namespace NSchema.Postgres.Sql;
+
+internal sealed partial class PostgresSqlDialect
+{
+    // ── Publications ──────────────────────────────────────────────────────────
+
+    // Tables first, each run opened once: the only spelling Postgres before 15 accepts, and valid after it.
+    protected override Result<IReadOnlyList<SqlStatement>> CreatePublication(CreatePublication action)
+    {
+        var publication = action.Publication;
+        var targets = new List<string>();
+        if (publication.Tables.Count > 0)
+        {
+            targets.Add("TABLE " + string.Join(", ", publication.Tables.Select(PublishedTable)));
+        }
+        if (publication.Schemas.Count > 0)
+        {
+            targets.Add("TABLES IN SCHEMA " + string.Join(", ", publication.Schemas.Select(Quote)));
+        }
+
+        var sql = $"CREATE PUBLICATION {Quote(publication.Name)}";
+        if (publication.AllTables)
+        {
+            sql += " FOR ALL TABLES";
+        }
+        else if (targets.Count > 0)
+        {
+            sql += " FOR " + string.Join(", ", targets);
+        }
+        if (publication.Operations != PublishedOperations.All)
+        {
+            sql += $" WITH (publish = '{Publish(publication.Operations)}')";
+        }
+
+        return Statement(sql);
+    }
+
+    protected override Result<IReadOnlyList<SqlStatement>> DropPublication(DropPublication action) =>
+        Statement($"DROP PUBLICATION {Quote(action.PublicationName)}");
+
+    protected override Result<IReadOnlyList<SqlStatement>> RenamePublication(RenamePublication action) =>
+        Statement($"ALTER PUBLICATION {Quote(action.OldName)} RENAME TO {Quote(action.NewName)}");
+
+    protected override Result<IReadOnlyList<SqlStatement>> AddPublicationTable(AddPublicationTable action) =>
+        Statement($"ALTER PUBLICATION {Quote(action.PublicationName)} ADD TABLE {PublishedTable(action.Table)}");
+
+    protected override Result<IReadOnlyList<SqlStatement>> DropPublicationTable(DropPublicationTable action) =>
+        Statement($"ALTER PUBLICATION {Quote(action.PublicationName)} DROP TABLE {Qualify(action.Table)}");
+
+    protected override Result<IReadOnlyList<SqlStatement>> AddPublicationSchema(AddPublicationSchema action) =>
+        Statement($"ALTER PUBLICATION {Quote(action.PublicationName)} ADD TABLES IN SCHEMA {Quote(action.Schema)}");
+
+    protected override Result<IReadOnlyList<SqlStatement>> DropPublicationSchema(DropPublicationSchema action) =>
+        Statement($"ALTER PUBLICATION {Quote(action.PublicationName)} DROP TABLES IN SCHEMA {Quote(action.Schema)}");
+
+    protected override Result<IReadOnlyList<SqlStatement>> SetPublicationOperations(SetPublicationOperations action) =>
+        Statement($"ALTER PUBLICATION {Quote(action.PublicationName)} SET (publish = '{Publish(action.NewOperations)}')");
+
+    protected override Result<IReadOnlyList<SqlStatement>> SetPublicationComment(SetPublicationComment action) =>
+        Comment($"PUBLICATION {Quote(action.PublicationName)}", action.NewComment);
+
+    private string PublishedTable(PublishedTable table)
+    {
+        var sql = Qualify(table.Table);
+        if (table.Columns is { } columns)
+        {
+            sql += $" ({ColumnList(columns)})";
+        }
+        if (table.Filter is { } filter)
+        {
+            sql += $" WHERE ({filter.Value})";
+        }
+        return sql;
+    }
+
+    private static string Publish(PublishedOperations operations) => string.Join(", ",
+        new[] { PublishedOperations.Insert, PublishedOperations.Update, PublishedOperations.Delete, PublishedOperations.Truncate }
+            .Where(o => operations.HasFlag(o))
+            .Select(o => o.ToString().ToLowerInvariant()));
+}

src/NSchema.Postgres/Sql/PostgresSqlDialect.Tables.cs

@@ -1,3 +1,4 @@
+using NSchema.Model.Tables;
 using NSchema.Plan.Domain;
 using NSchema.Plan.Domain.Tables;
 
@@ -25,4 +26,14 @@ protected override Result<IReadOnlyList<SqlStatement>> CreateTable(CreateTable a
 
     protected override Result<IReadOnlyList<SqlStatement>> SetTableComment(SetTableComment action) =>
         Comment($"TABLE {Qualify(action.Table)}", action.NewComment);
+
+    // No declared identity is the engine's default, which Postgres spells DEFAULT (the primary key).
+    protected override Result<IReadOnlyList<SqlStatement>> SetReplicaIdentity(SetReplicaIdentity action) =>
+        Statement($"ALTER TABLE {Qualify(action.Table)} REPLICA IDENTITY " + action.NewIdentity switch
+        {
+            null => "DEFAULT",
+            { Kind: ReplicaIdentityKind.Full } => "FULL",
+            { Kind: ReplicaIdentityKind.Index, Index: { } index } => $"USING INDEX {Quote(index)}",
+            _ => "NOTHING",
+        });
 }