Support tables outside the public schema

AGENTS.md

@@ -80,7 +80,7 @@ Keyed by the stream's routing key (default `id`); UPDATE emits only the new row.
 
 ```json
 {"op":"INSERT","data":{"id":1,"name":"Alice"},
- "meta":{"source":"postgres","resource":"users","schema":"public","timestamp":1700000000,"lsn":"1/3259A308"}}
+ "meta":{"source":"postgres","resource":"public.users","timestamp":1700000000,"lsn":"1/3259A308"}}
 ```
 
 `timestamp` is the transaction's commit time (Unix seconds), stable across

CHANGELOG.md

@@ -4,6 +4,17 @@ All notable changes to Outboxx are documented here.
 
 ## Unreleased
 
+### Added
+
+- Streams can target tables in any schema. Set `resource = "schema.table"` in a
+  stream's source; a bare name still defaults to the `public` schema.
+
+### Changed
+
+- `meta.resource` is the fully-qualified name (`public.users`, was `users`) and
+  `meta.schema` is removed. A consumer that needs the schema splits
+  `meta.resource` on its first `.`.
+
 ## 0.3.0 - 2026-07-18
 
 First GA release: the at-least-once guarantee is now enforced end to end and

docs/examples/config.toml

@@ -52,7 +52,7 @@ port = 9464          # conventional OpenTelemetry Prometheus exporter port
 name = "users-stream"
 
 [streams.source]
-resource = "users"                           # table name (public schema)
+resource = "users"                           # table; use "schema.table" for a non-public schema (default: public)
 operations = ["insert", "update", "delete"]  # subset of insert | update | delete
 
 [streams.flow]

src/benchmarks/match_streams_bench.zig

@@ -14,13 +14,12 @@ const CountingAllocator = bench_helpers.CountingAllocator;
 
 const iterations = 100000;
 
-// An INSERT on public.<resource>; matchStreams only reads op/schema/resource, so
-// data is left unset by ChangeEvent.init.
+// An INSERT on <resource>; matchStreams only reads op and resource, so data is
+// left unset by ChangeEvent.init.
 fn insertChange(resource: []const u8) ChangeEvent {
     return ChangeEvent.init(.INSERT, .{
         .source = "postgres",
         .resource = resource,
-        .schema = "public",
         .timestamp = 0,
         .lsn = null,
     });

src/benchmarks/partition_key_bench.zig

@@ -24,7 +24,6 @@ fn createEventWithIntegerKey(allocator: std.mem.Allocator) !ChangeEvent {
     const metadata = Metadata{
         .source = try allocator.dupe(u8, "postgres"),
         .resource = try allocator.dupe(u8, "users"),
-        .schema = try allocator.dupe(u8, "public"),
         .timestamp = 1234567890,
         .lsn = null,
     };
@@ -45,7 +44,6 @@ fn createEventWithStringKey(allocator: std.mem.Allocator) !ChangeEvent {
     const metadata = Metadata{
         .source = try allocator.dupe(u8, "postgres"),
         .resource = try allocator.dupe(u8, "orders"),
-        .schema = try allocator.dupe(u8, "public"),
         .timestamp = 1234567890,
         .lsn = null,
     };

src/benchmarks/serializer_bench.zig

@@ -24,7 +24,6 @@ fn buildChangeEvent(allocator: std.mem.Allocator) !ChangeEvent {
     const metadata = Metadata{
         .source = try allocator.dupe(u8, "postgres"),
         .resource = try allocator.dupe(u8, "users"),
-        .schema = try allocator.dupe(u8, "public"),
         .timestamp = 1234567890,
         .lsn = null,
     };

src/config/config.zig

@@ -92,7 +92,10 @@ pub const SinkConfig = struct {
 };
 
 pub const StreamSource = struct {
-    resource: []const u8, // table/collection/index
+    // Fully-qualified resource name (Postgres: schema.table). A bare name is
+    // accepted and normalized to the public schema when the config is loaded, so
+    // everything downstream treats the resource as one opaque string.
+    resource: []const u8,
     operations: []const []const u8, // ["insert", "update", "delete"]
 };
 
@@ -147,17 +150,23 @@ pub const Config = struct {
     pub fn loadFromTomlFile(io: std.Io, allocator: std.mem.Allocator, file_path: []const u8) !toml.Parsed(Config) {
         var parser = toml.Parser(Config).init(allocator);
         defer parser.deinit();
-        return parser.parseFile(io, file_path) catch |err| {
+        var parsed = parser.parseFile(io, file_path) catch |err| {
             std.log.warn("Failed to parse config file '{s}': {}", .{ file_path, err });
             return err;
         };
+        errdefer parsed.deinit();
+        try normalizeResources(&parsed);
+        return parsed;
     }
 
     /// Parse a config string; caller owns and must deinit the returned result.
     pub fn loadFromTomlString(allocator: std.mem.Allocator, content: []const u8) !toml.Parsed(Config) {
         var parser = toml.Parser(Config).init(allocator);
         defer parser.deinit();
-        return parser.parseString(content);
+        var parsed = try parser.parseString(content);
+        errdefer parsed.deinit();
+        try normalizeResources(&parsed);
+        return parsed;
     }
 
     /// Read the Kafka SASL password from the environment; caller owns the result.
@@ -253,6 +262,22 @@ pub const Config = struct {
         }
     }
 
+    // A table reference, optionally schema-qualified as "schema.table". Each part
+    // is a Postgres identifier; more than one dot is rejected, since without
+    // quoted identifiers a dot can only separate schema from table.
+    fn validateResource(value: []const u8, field_name: []const u8) !void {
+        const dot = std.mem.indexOfScalar(u8, value, '.') orelse
+            return validatePostgresIdentifier(value, field_name);
+
+        if (std.mem.indexOfScalarPos(u8, value, dot + 1, '.') != null) {
+            std.log.warn("Invalid {s}: '{s}' has more than one '.'; expected \"schema.table\"", .{ field_name, value });
+            return error.InvalidIdentifierFormat;
+        }
+
+        try validatePostgresIdentifier(value[0..dot], field_name);
+        try validatePostgresIdentifier(value[dot + 1 ..], field_name);
+    }
+
     // Kafka topic charset: a-z, A-Z, 0-9, '.', '_', '-'.
     fn validateKafkaTopicName(value: []const u8, field_name: []const u8) !void {
         try validateStringLength(value, ValidationLimits.MAX_KAFKA_TOPIC_LEN, field_name);
@@ -307,7 +332,7 @@ pub const Config = struct {
         try validateStreamName(stream.name, "stream.name");
 
         // Source validation
-        try validatePostgresIdentifier(stream.source.resource, "stream.source.resource");
+        try validateResource(stream.source.resource, "stream.source.resource");
         try validateOperations(allocator, stream.source.operations);
 
         // Flow validation
@@ -417,6 +442,20 @@ pub const Config = struct {
         }
         try validateStreams(allocator, self.streams);
     }
+
+    // Part of the load contract: rewrite each bare stream resource to a
+    // fully-qualified `schema.table`, defaulting to the public schema, so the
+    // processor and the source-side validator treat the resource as one opaque
+    // name. Runs right after parsing; names are (re)allocated in the Parsed arena
+    // and live as long as the config.
+    fn normalizeResources(parsed: *toml.Parsed(Config)) !void {
+        const arena = parsed.arena.allocator();
+        for (@constCast(parsed.value.streams)) |*stream| {
+            if (std.mem.indexOfScalar(u8, stream.source.resource, '.') == null) {
+                stream.source.resource = try std.fmt.allocPrint(arena, "public.{s}", .{stream.source.resource});
+            }
+        }
+    }
 };
 
 const testing = std.testing;
@@ -515,6 +554,49 @@ test "Stream.hasDeleteOperation reflects the configured operations" {
     try testing.expect(with_delete.hasDeleteOperation());
 }
 
+test "load normalizes a bare resource to the public schema" {
+    var parsed = try Config.loadFromTomlString(testing.allocator, valid_config_toml);
+    defer parsed.deinit();
+    try testing.expectEqualStrings("public.users", parsed.value.streams[0].source.resource);
+}
+
+test "load leaves a schema-qualified resource unchanged" {
+    const toml_content =
+        \\[metadata]
+        \\version = "v0"
+        \\
+        \\[source]
+        \\type = "postgres"
+        \\
+        \\[source.postgres]
+        \\connection_env = "PG_URL"
+        \\slot_name = "slot"
+        \\publication_name = "pub"
+        \\
+        \\[sink]
+        \\type = "kafka"
+        \\
+        \\[sink.kafka]
+        \\brokers = ["kafka1:9092"]
+        \\
+        \\[[streams]]
+        \\name = "app-users"
+        \\
+        \\[streams.source]
+        \\resource = "app.users"
+        \\operations = ["insert"]
+        \\
+        \\[streams.flow]
+        \\format = "json"
+        \\
+        \\[streams.sink]
+        \\destination = "outboxx.app_users"
+    ;
+    var parsed = try Config.loadFromTomlString(testing.allocator, toml_content);
+    defer parsed.deinit();
+    try testing.expectEqualStrings("app.users", parsed.value.streams[0].source.resource);
+}
+
 test "supported adapter types are implemented" {
     try testing.expectEqual(@as(usize, 1), SupportedValues.SOURCE_TYPES.len);
     try testing.expectEqualStrings("postgres", SupportedValues.SOURCE_TYPES[0]);
@@ -671,7 +753,7 @@ test "parse stream with inline comments and optional routing_key" {
     try testing.expect(cfg.streams.len == 1);
     const stream = cfg.streams[0];
     try testing.expectEqualStrings("users-stream", stream.name);
-    try testing.expectEqualStrings("users", stream.source.resource);
+    try testing.expectEqualStrings("public.users", stream.source.resource);
     try testing.expect(stream.source.operations.len == 2);
     try testing.expectEqualStrings("insert", stream.source.operations[0]);
     try testing.expectEqualStrings("update", stream.source.operations[1]);
@@ -777,15 +859,15 @@ test "parse multiple streams" {
 
     const stream1 = cfg.streams[0];
     try testing.expectEqualStrings("users-stream", stream1.name);
-    try testing.expectEqualStrings("users", stream1.source.resource);
+    try testing.expectEqualStrings("public.users", stream1.source.resource);
     try testing.expect(stream1.source.operations.len == 2);
     try testing.expectEqualStrings("json", stream1.flow.format);
     try testing.expectEqualStrings("outboxx.users", stream1.sink.destination);
     try testing.expectEqualStrings("id", stream1.sink.routing_key);
 
     const stream2 = cfg.streams[1];
     try testing.expectEqualStrings("orders-stream", stream2.name);
-    try testing.expectEqualStrings("orders", stream2.source.resource);
+    try testing.expectEqualStrings("public.orders", stream2.source.resource);
     try testing.expect(stream2.source.operations.len == 3);
     try testing.expectEqualStrings("delete", stream2.source.operations[2]);
     try testing.expectEqualStrings("outboxx.orders", stream2.sink.destination);
@@ -834,6 +916,39 @@ test "Config validation - unsupported version" {
     try testing.expectError(error.UnsupportedConfigVersion, cfg.validate(testing.allocator));
 }
 
+test "Config validation - schema-qualified resource passes" {
+    var cfg = createTestDefault();
+    cfg.streams = &.{.{
+        .name = "test_stream",
+        .source = .{ .resource = "app.users", .operations = &.{"insert"} },
+        .flow = .{ .format = "json" },
+        .sink = .{ .destination = "test_topic", .routing_key = "id" },
+    }};
+    try cfg.validate(testing.allocator);
+}
+
+test "Config validation - resource with more than one dot fails" {
+    var cfg = createTestDefault();
+    cfg.streams = &.{.{
+        .name = "test_stream",
+        .source = .{ .resource = "db.app.users", .operations = &.{"insert"} },
+        .flow = .{ .format = "json" },
+        .sink = .{ .destination = "test_topic", .routing_key = "id" },
+    }};
+    try testing.expectError(error.InvalidIdentifierFormat, cfg.validate(testing.allocator));
+}
+
+test "Config validation - resource with empty schema fails" {
+    var cfg = createTestDefault();
+    cfg.streams = &.{.{
+        .name = "test_stream",
+        .source = .{ .resource = ".users", .operations = &.{"insert"} },
+        .flow = .{ .format = "json" },
+        .sink = .{ .destination = "test_topic", .routing_key = "id" },
+    }};
+    try testing.expectError(error.EmptyString, cfg.validate(testing.allocator));
+}
+
 test "Config validation - unsupported format should fail" {
     var cfg = createTestDefault();
     cfg.streams = &.{.{

src/domain/change_event.zig

@@ -105,8 +105,7 @@ pub const DataSection = union(enum) {
 /// Source metadata for a change event.
 pub const Metadata = struct {
     source: []const u8, // "postgres"
-    resource: []const u8, // table/collection name
-    schema: []const u8, // database schema
+    resource: []const u8, // fully-qualified resource name (Postgres: schema.table)
     timestamp: i64, // commit time of the change's transaction (Unix seconds)
     lsn: ?[]const u8, // WAL position of the record (source-specific; dedup key)
 };
@@ -143,7 +142,6 @@ pub const ChangeEvent = struct {
         // Free metadata strings (owned by ChangeEvent)
         allocator.free(self.meta.source);
         allocator.free(self.meta.resource);
-        allocator.free(self.meta.schema);
         if (self.meta.lsn) |lsn| allocator.free(lsn);
 
         // Free data rows
@@ -225,7 +223,6 @@ test "ChangeEvent creation and memory management" {
     const metadata = Metadata{
         .source = try allocator.dupe(u8, "postgres"),
         .resource = try allocator.dupe(u8, "users"),
-        .schema = try allocator.dupe(u8, "public"),
         .timestamp = 1234567890,
         .lsn = null,
     };
@@ -288,7 +285,6 @@ test "UPDATE event with old and new data" {
     const metadata = Metadata{
         .source = try allocator.dupe(u8, "postgres"),
         .resource = try allocator.dupe(u8, "users"),
-        .schema = try allocator.dupe(u8, "public"),
         .timestamp = 1234567890,
         .lsn = null,
     };
@@ -338,7 +334,6 @@ test "getPartitionKeyValue extracts field values" {
         const metadata = Metadata{
             .source = try allocator.dupe(u8, "postgres"),
             .resource = try allocator.dupe(u8, "users"),
-            .schema = try allocator.dupe(u8, "public"),
             .timestamp = 1234567890,
             .lsn = null,
         };
@@ -381,7 +376,6 @@ test "getPartitionKeyValue extracts field values" {
         const metadata = Metadata{
             .source = try allocator.dupe(u8, "postgres"),
             .resource = try allocator.dupe(u8, "users"),
-            .schema = try allocator.dupe(u8, "public"),
             .timestamp = 1234567890,
             .lsn = null,
         };
@@ -422,7 +416,6 @@ test "partitionKeyInt formats i64 boundaries and returns null for other types" {
     const metadata = Metadata{
         .source = try allocator.dupe(u8, "postgres"),
         .resource = try allocator.dupe(u8, "users"),
-        .schema = try allocator.dupe(u8, "public"),
         .timestamp = 1234567890,
         .lsn = null,
     };

src/e2e/cdc_test.zig

@@ -60,6 +60,7 @@ test "E2E: INSERT operation - full pipeline verification" {
     // Create stream configuration
     const stream_config = try test_helpers.createTestStreamConfig(allocator, table_name, topic_name);
     defer allocator.free(stream_config.name);
+    defer allocator.free(stream_config.source.resource);
 
     // Create source
     // NOTE: source will be deinit'd by processor.deinit() - no need for defer here
@@ -136,8 +137,7 @@ test "E2E: INSERT operation - full pipeline verification" {
         try test_helpers.assertJsonField(msg, "op", "INSERT");
 
         // Verify metadata
-        try test_helpers.assertJsonField(msg, "meta.resource", table_name);
-        try test_helpers.assertJsonField(msg, "meta.schema", "public");
+        try test_helpers.assertJsonField(msg, "meta.resource", stream_config.source.resource);
         try test_helpers.assertJsonField(msg, "meta.source", "postgres");
         try test_helpers.assertJsonHasField(msg, "meta.timestamp");
 
@@ -200,6 +200,7 @@ test "E2E: UPDATE operation - full pipeline verification" {
     // Create stream configuration
     const stream_config = try test_helpers.createTestStreamConfig(allocator, table_name, topic_name);
     defer allocator.free(stream_config.name);
+    defer allocator.free(stream_config.source.resource);
 
     // Create source
     // NOTE: source will be deinit'd by processor.deinit() - no need for defer here
@@ -278,7 +279,7 @@ test "E2E: UPDATE operation - full pipeline verification" {
     try test_helpers.assertJsonField(messages[0], "data.name", "Alice");
 
     try test_helpers.assertJsonField(messages[1], "op", "UPDATE");
-    try test_helpers.assertJsonField(messages[1], "meta.resource", table_name);
+    try test_helpers.assertJsonField(messages[1], "meta.resource", stream_config.source.resource);
     try test_helpers.assertJsonField(messages[1], "data.name", "Alice Updated");
 
     const data_obj_1 = messages[1].value.object.get("data").?.object;
@@ -333,6 +334,7 @@ test "E2E: DELETE operation - full pipeline verification" {
     // Create stream configuration
     const stream_config = try test_helpers.createTestStreamConfig(allocator, table_name, topic_name);
     defer allocator.free(stream_config.name);
+    defer allocator.free(stream_config.source.resource);
 
     // Create source
     // NOTE: source will be deinit'd by processor.deinit() - no need for defer here
@@ -420,12 +422,10 @@ test "E2E: DELETE operation - full pipeline verification" {
 
         if (std.mem.eql(u8, op, "INSERT")) {
             insert_count += 1;
-            try test_helpers.assertJsonField(msg, "meta.resource", table_name);
-            try test_helpers.assertJsonField(msg, "meta.schema", "public");
+            try test_helpers.assertJsonField(msg, "meta.resource", stream_config.source.resource);
         } else if (std.mem.eql(u8, op, "DELETE")) {
             delete_count += 1;
-            try test_helpers.assertJsonField(msg, "meta.resource", table_name);
-            try test_helpers.assertJsonField(msg, "meta.schema", "public");
+            try test_helpers.assertJsonField(msg, "meta.resource", stream_config.source.resource);
         }
     }