Plain Tables

The Wallaby.Providers.Tables package drives capture from tables you describe with plain C# types, with no ORM in between. You register a POCO per table, and Wallaby works out the table, columns and key from System.ComponentModel.DataAnnotations attributes, fluent overrides or a naming convention. Each change then materializes into an instance of your type. Transforms receive an NpgsqlDataSource, so enrichment queries can use Dapper, raw Npgsql, or anything else that takes a connection.

Use it when the application isn't on EF Core or Marten, or for the odd table that lives outside your model. It combines with the other providers on one slot.

Install

dotnet add package Wallaby.Providers.Tables

Register

Chain UseTables(...) and register every type you want captured. Only registered types are handled, so a class that also belongs to an EF Core model is never claimed twice.

You also need to supply a connection string, either with UseConnectionString(...) or any other options-pattern mechanism such as configuration binding. Multi-host connection strings are supported, but Wallaby only connects to the primary.

using System.ComponentModel.DataAnnotations.Schema;
using Wallaby.Abstractions;
using Wallaby.DependencyInjection;
using Wallaby.Providers.Tables;

public enum OrderStatus { Pending, Paid, Shipped }

[Table("orders", Schema = "sales")]
public sealed record Order(int Id, string CustomerRef, decimal Total, OrderStatus Status, DateTimeOffset CreatedAt);

[Table("order_lines", Schema = "sales")]
public sealed record OrderLine(int OrderId, int LineNo, string Sku, int Quantity);

builder.Services.AddWallaby(cdc =>
{
    cdc.UseTables(tables =>
       {
           tables.UseSnakeCase();                                   // CustomerRef -> customer_ref
           tables.Add<Order>();
           tables.Add<OrderLine>().HasKey(l => l.OrderId, l => l.LineNo);
       })
       .UseConnectionString(conn)

       // Example sink configuration
       .AddMeilisearchSink("meili", m => { /* ... */ })
       .WithMappings(sink => sink
            .Map<Order>()
            .ToDestination("orders")
            .UsingTransform(async (db, changes, ct) =>
            {
                await using var connection = await db.OpenConnectionAsync(ct);
                // Dapper, raw Npgsql, anything that takes a connection.
                var docs = new Dictionary<DocumentKey, WallabyDocument?>();
                foreach (var change in changes)
                    docs[change.Key] = new WallabyDocument { ["customer"] = change.Entity!.CustomerRef };
                return docs;
            }));
});

Registration errors (a missing key, an unsupported property type, an ambiguous constructor) throw from UseTables itself, before the host starts.

Mapping rules

Each registered type maps its public instance properties that have a getter to columns. Where more than one rule applies, fluent configuration wins over attributes, and attributes win over convention:

Aspect	Fluent	Attribute	Convention
Table	ToTable("orders", "sales")	[Table("orders", Schema = "sales")]	The type name, in the default schema (public, or DefaultSchema(...))
Column	Column(o => o.CustomerRef, "cust")	[Column("cust")]	The property name
Key	HasKey(o => o.A, o => o.B), in argument order	[Key] members, ordered by [Column(Order = n)] then declaration	A property named Id or {Type}Id
Skip	Ignore(o => o.Scratch)	[NotMapped]	A property with no getter

UseSnakeCase() converts unannotated names to snake_case using the same rules as EFCore.NamingConventions (OrderLine to order_line, HTTPStatus to http_status). Names from attributes and fluent calls are always used verbatim. Names are matched against the relation exactly as Postgres reports them, so a quoted mixed-case identifier needs the same casing.

Types

A property maps to a single column when its type is a scalar that the pgoutput decoder produces or value coercion can convert to:

• the numeric types, string, char, bool and Guid
• DateTime, DateTimeOffset, DateOnly, TimeOnly and TimeSpan
• byte[], IPAddress, PhysicalAddress and BitArray
• enums (from text or number)
• nullable versions of all of these, and single-dimension arrays of them

A property of any other type (a nested class, a collection, JsonElement) fails registration, and the error tells you to mark it [NotMapped] or Ignore(...) it. JSON columns can only be captured into a string property in this version.

Key properties are further limited to the types a backfill cursor can persist: numbers, strings, Guid, dates and times, and byte[]. An enum key is rejected at registration.

Records and constructors

A type needs either a public parameterless constructor, or exactly one public constructor whose parameters all match public properties by name (case-insensitive). Positional records satisfy the second rule:

public sealed record Order(int Id, string CustomerRef, decimal Total);

Constructor parameters receive the row values, and a parameter for an ignored or [NotMapped] property receives its default. Any remaining properties with a setter (init included) are assigned afterwards. A get-only property is still captured into ChangeEvent.Record, but never assigned.

A column that's absent from the change (because of a narrowed selection, or a delete under REPLICA IDENTITY DEFAULT) leaves its member at the default value.

Transforms

Transforms receive the provider's NpgsqlDataSource. Open a pooled connection for each batch and let it return to the pool when the transform ends. There are three UsingTransform overloads: one taking a standalone IWallabyTablesTransform<T> instance, a container-resolved UsingTransform<TEntity, TTransform>(), and an inline lambda.

If an NpgsqlDataSource is registered in the container (AddNpgsqlDataSource, or your own singleton), transforms get that one. Otherwise Wallaby builds one from its own connection string, which also carries any password provider. To choose explicitly, for example a read replica, pass a factory to UseTables(sp => ..., tables => ...).

Class-based transforms

For anything with dependencies, implement IWallabyTablesTransform<TEntity> as a class. It's resolved from the container:

public sealed class OrderSearchTransform : IWallabyTablesTransform<Order>
{
    public async Task<IReadOnlyDictionary<DocumentKey, WallabyDocument?>> TransformAsync(
        NpgsqlDataSource dataSource, IReadOnlyList<ChangeEvent<Order>> changes, CancellationToken ct)
    {
        await using var connection = await dataSource.OpenConnectionAsync(ct);
        var customerRefs = changes.Select(c => c.Entity!.CustomerRef).Distinct().ToArray();
        var names = (await connection.QueryAsync<(string Ref, string Name)>(           // Dapper
                "SELECT ref, name FROM crm.customers WHERE ref = ANY(@refs)", new { refs = customerRefs }))
            .ToDictionary(c => c.Ref, c => c.Name);

        var docs = new Dictionary<DocumentKey, WallabyDocument?>(changes.Count);
        foreach (var c in changes)
            docs[c.Key] = new WallabyDocument
            {
                ["total"] = c.Entity!.Total,
                ["customer"] = names.GetValueOrDefault(c.Entity!.CustomerRef),
            };
        return docs;
    }
}

// register:
sink.Map<Order>()
    .ToDestination("orders")
    .UsingTransform<Order, OrderSearchTransform>();

TIP

The registration itself can also live outside AddWallaby as a mapping class: sink.Apply<OrderSearchMapping>().

Declaring consumed columns

By default Wallaby captures every mapped property. A mapping can narrow that with Consumes(...) or ConsumesAllExcept(...). The entity's captured columns are the union across all its mappings, plus the key, and the table is published with a column list so unselected columns never leave the server:

sink.Map<Order>()
    .Consumes(o => o.CustomerRef, o => o.Total)      // key columns are always kept
    .UsingTransform(...);

sink.Map<Document>()
    .ConsumesAllExcept(d => d.Body)                  // drop a TOAST-prone column no transform reads
    .UsingTransform(...);

Unselected properties keep their default value on the materialized entity and are absent from ChangeEvent.Record. A key property can't be excluded.

Deletes and the old tuple

A delete's ChangeEvent.Entity is built from the old tuple, which holds only the key columns under REPLICA IDENTITY DEFAULT and the whole row under REPLICA IDENTITY FULL. Likewise, ChangeEvent.Changes on an update lists the previous values of whichever columns the old tuple carries.

So mappings that derive a delete's identity or routing from the entity (KeyedBy(...), or an entity-scoped destination) need REPLICA IDENTITY FULL. Self-config reports a missing identity at startup.

Replica identity

Wallaby never alters your tables. Set the identity in your own migrations where a table needs it:

ALTER TABLE sales.orders REPLICA IDENTITY FULL;

Without it, an update that leaves a large (TOASTed) column untouched omits that value from the change. Wallaby then heals the change by re-reading the row and logs a warning for each one, or fails the change if ReselectUnavailableValues is disabled. If no transform reads the large column, ConsumesAllExcept(...) on it avoids both.

Backfills

Backfills read rows by key in keyset order and materialize them through the same path as live changes, so ChangeEvent.Metadata.IsBackfill is the only difference a transform sees. Composite keys page in the declared key order.

NativeAOT

Wallaby.Providers.Tables is trim- and NativeAOT-compatible (IsAotCompatible). Add<T>() carries the annotations the trimmer needs to keep your type's public properties and constructors, and no serializer is involved.

Limitations (v1)

• JSON columns are captured as text only: a jsonb column can only be captured into a string property.
• DependsOn(...) is not supported: a POCO has no navigations to resolve. You can capture the related table as its own type, but its changes won't re-emit the parent's document.
• ScopedByTenant() is not supported: scope by a property instead, with ScopedBy(e => e.TenantId).
• No catalog discovery: every type is registered explicitly. Wallaby doesn't read pg_catalog to find tables or keys.