From 91685df89d6a1645f361065d3a45306fe03777c3 Mon Sep 17 00:00:00 2001 From: Yegor Lukash Date: Mon, 3 Aug 2026 15:12:28 +0300 Subject: [PATCH 1/3] Support tables in any schema via fully-qualified resource names --- AGENTS.md | 2 +- CHANGELOG.md | 11 ++ docs/examples/config.toml | 2 +- src/benchmarks/match_streams_bench.zig | 5 +- src/benchmarks/partition_key_bench.zig | 2 - src/benchmarks/serializer_bench.zig | 1 - src/config/config.zig | 122 ++++++++++++++++++++- src/domain/change_event.zig | 9 +- src/e2e/cdc_test.zig | 14 +-- src/main.zig | 13 ++- src/processor/processor.zig | 8 +- src/processor/routing_integration_test.zig | 44 +++++--- src/serialization/json.zig | 8 -- src/source/postgres/converter.zig | 14 +-- src/testing/test_helpers.zig | 28 ++--- 15 files changed, 200 insertions(+), 83 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index ae6920d..8b90ab3 100644 --- a/AGENTS.md +++ b/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 diff --git a/CHANGELOG.md b/CHANGELOG.md index a0d22ea..fc2bc57 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/examples/config.toml b/docs/examples/config.toml index 71fe2cf..6fa4e03 100644 --- a/docs/examples/config.toml +++ b/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] diff --git a/src/benchmarks/match_streams_bench.zig b/src/benchmarks/match_streams_bench.zig index d043a16..ddc7601 100644 --- a/src/benchmarks/match_streams_bench.zig +++ b/src/benchmarks/match_streams_bench.zig @@ -14,13 +14,12 @@ const CountingAllocator = bench_helpers.CountingAllocator; const iterations = 100000; -// An INSERT on public.; matchStreams only reads op/schema/resource, so -// data is left unset by ChangeEvent.init. +// An INSERT on ; 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, }); diff --git a/src/benchmarks/partition_key_bench.zig b/src/benchmarks/partition_key_bench.zig index 05f160e..24f1855 100644 --- a/src/benchmarks/partition_key_bench.zig +++ b/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, }; diff --git a/src/benchmarks/serializer_bench.zig b/src/benchmarks/serializer_bench.zig index 2d4a5e0..4dbc8c2 100644 --- a/src/benchmarks/serializer_bench.zig +++ b/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, }; diff --git a/src/config/config.zig b/src/config/config.zig index 49b6556..85c4de1 100644 --- a/src/config/config.zig +++ b/src/config/config.zig @@ -92,10 +92,36 @@ pub const SinkConfig = struct { }; pub const StreamSource = struct { - resource: []const u8, // table/collection/index + // Fully-qualified resource: table/collection/index as "schema.table". A bare + // name is accepted and normalized to the public schema by qualifyResources. + resource: []const u8, operations: []const []const u8, // ["insert", "update", "delete"] + + /// A resource split into namespace and object name. For Postgres this is + /// schema + table. + pub const Qualified = struct { + schema: []const u8, + name: []const u8, + }; + + /// Split `resource` on its '.' into schema and table, for the source-side + /// validator that queries the two separately. qualifyResources runs first, so + /// a dot is present; a bare name still falls back to the public schema. + pub fn qualifiedResource(self: StreamSource) Qualified { + if (std.mem.indexOfScalar(u8, self.resource, '.')) |dot| { + return .{ .schema = self.resource[0..dot], .name = self.resource[dot + 1 ..] }; + } + return .{ .schema = "public", .name = self.resource }; + } }; +/// Fully-qualify a resource name as `schema.table`, defaulting a bare name (no +/// '.') to the `public` schema. Caller owns the result. +pub fn qualifyResourceName(allocator: std.mem.Allocator, resource: []const u8) ![]const u8 { + if (std.mem.indexOfScalar(u8, resource, '.') != null) return allocator.dupe(u8, resource); + return std.fmt.allocPrint(allocator, "public.{s}", .{resource}); +} + pub const StreamFlow = struct { format: []const u8, }; @@ -253,6 +279,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 +349,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 +459,19 @@ pub const Config = struct { } try validateStreams(allocator, self.streams); } + + /// Rewrite each stream's `resource` into its fully-qualified `schema.table` + /// form (a bare name defaults to the public schema), so the processor matches + /// it against the source's qualified resource with a plain compare. Call once + /// after validate(). Mutates the stream slice in place, so it must be the + /// mutable arena memory from the TOML parser; pass that arena's allocator + /// (`parsed.arena.allocator()`) so the rewritten names share the config's + /// lifetime. + pub fn qualifyResources(self: Config, allocator: std.mem.Allocator) !void { + for (@constCast(self.streams)) |*stream| { + stream.source.resource = try qualifyResourceName(allocator, stream.source.resource); + } + } }; const testing = std.testing; @@ -515,6 +570,36 @@ test "Stream.hasDeleteOperation reflects the configured operations" { try testing.expect(with_delete.hasDeleteOperation()); } +test "qualifiedResource splits schema.table and defaults to public" { + const bare: StreamSource = .{ .resource = "users", .operations = &.{} }; + const q1 = bare.qualifiedResource(); + try testing.expectEqualStrings("public", q1.schema); + try testing.expectEqualStrings("users", q1.name); + + const qualified: StreamSource = .{ .resource = "app.users", .operations = &.{} }; + const q2 = qualified.qualifiedResource(); + try testing.expectEqualStrings("app", q2.schema); + try testing.expectEqualStrings("users", q2.name); +} + +test "qualifyResourceName defaults a bare name to public and keeps a qualified one" { + const bare = try qualifyResourceName(testing.allocator, "users"); + defer testing.allocator.free(bare); + try testing.expectEqualStrings("public.users", bare); + + const qualified = try qualifyResourceName(testing.allocator, "app.users"); + defer testing.allocator.free(qualified); + try testing.expectEqualStrings("app.users", qualified); +} + +test "qualifyResources rewrites bare stream resources in place" { + var parsed = try Config.loadFromTomlString(testing.allocator, valid_config_toml); + defer parsed.deinit(); + + try parsed.value.qualifyResources(parsed.arena.allocator()); + try testing.expectEqualStrings("public.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]); @@ -834,6 +919,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 = &.{.{ diff --git a/src/domain/change_event.zig b/src/domain/change_event.zig index 77803df..a2d60d1 100644 --- a/src/domain/change_event.zig +++ b/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, }; diff --git a/src/e2e/cdc_test.zig b/src/e2e/cdc_test.zig index 014b49d..3cf6ccf 100644 --- a/src/e2e/cdc_test.zig +++ b/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); } } diff --git a/src/main.zig b/src/main.zig index a4d7457..a9ce93e 100644 --- a/src/main.zig +++ b/src/main.zig @@ -90,6 +90,9 @@ fn run(init: std.process.Init) !void { const config = parsed.value; try config.validate(allocator); + // Normalize bare resource names to schema.table so the processor matches them + // against the source's qualified names. Names live in the Parsed arena. + try config.qualifyResources(parsed.arena.allocator()); const conninfo = try config.loadPostgresConninfo(allocator, init.environ_map); defer allocator.free(conninfo); @@ -286,22 +289,24 @@ fn validatePostgres(allocator: std.mem.Allocator, cfg: Config, conninfo: []const try validator.checkWalLevel(); for (cfg.streams) |stream| { - validator.checkTableExists("public", stream.source.resource) catch |err| { + const target = stream.source.qualifiedResource(); + + validator.checkTableExists(target.schema, target.name) catch |err| { printStatus("ERROR: Table validation failed for '{s}': {}\n", .{ stream.source.resource, err }); return err; }; // A missing routing key column would silently collapse partitioning. const routing_key = stream.sink.routing_key; - validator.checkColumnExists("public", stream.source.resource, routing_key) catch |err| { + validator.checkColumnExists(target.schema, target.name, routing_key) catch |err| { printStatus("ERROR: Routing key validation failed for '{s}' (column '{s}'): {}\n", .{ stream.source.resource, routing_key, err }); return err; }; // Only a stream that tracks DELETE needs REPLICA IDENTITY FULL, so the - // deleted row carries all columns. Schema is fixed to "public" for now. + // deleted row carries all columns. if (stream.hasDeleteOperation()) { - validator.checkReplicaIdentity("public", stream.source.resource) catch |err| { + validator.checkReplicaIdentity(target.schema, target.name) catch |err| { printStatus("ERROR: Replica identity validation failed for '{s}': {}\n", .{ stream.source.resource, err }); return err; }; diff --git a/src/processor/processor.zig b/src/processor/processor.zig index 2f6ed16..cd90598 100644 --- a/src/processor/processor.zig +++ b/src/processor/processor.zig @@ -31,12 +31,10 @@ fn tallyEvent(list: *std.ArrayList(EventCount), allocator: std.mem.Allocator, st pub fn matchStreams(allocator: std.mem.Allocator, streams: []const Stream, change: ChangeEvent) !std.ArrayList(Stream) { var matched = std.ArrayList(Stream).empty; - // Streams target the public schema (startup validation enforces it), so a - // change from any other schema must not match even when the table name - // collides. Routing tables from other schemas is #50. - if (!std.mem.eql(u8, change.meta.schema, "public")) return matched; - for (streams) |stream| { + // Resource is a fully-qualified name on both sides (config normalized to + // schema.table, the source tags each change likewise), so identity is a + // plain string compare and stays source-agnostic. if (!std.mem.eql(u8, stream.source.resource, change.meta.resource)) { continue; } diff --git a/src/processor/routing_integration_test.zig b/src/processor/routing_integration_test.zig index f265a93..4fcf473 100644 --- a/src/processor/routing_integration_test.zig +++ b/src/processor/routing_integration_test.zig @@ -38,7 +38,7 @@ fn execSQL(conn: *c.PGconn, sql: [:0]const u8) !void { } } -test "matchStreams: a change from another schema is not routed to a public stream" { +test "matchStreams: each change is routed to the stream matching its schema" { const allocator = testing.allocator; var prng = std.Random.DefaultPrng.init(@intCast(test_helpers.nowMicros(std.testing.io))); @@ -72,9 +72,9 @@ test "matchStreams: a change from another schema is not routed to a public strea execSQL(setup_conn, drop_public) catch {}; } - // public.: what the stream targets. A same-named table in another - // schema is also published, so its change reaches the decoder and must be - // dropped by matchStreams, not routed as if it were public.
. + // A same-named table exists in two schemas, both published. Their changes + // reach the decoder tagged with their real schema, so matchStreams must send + // each to the stream targeting that schema, not collapse them by table name. const create_public = try test_helpers.formatSqlZ(allocator, "CREATE TABLE {s} (id SERIAL PRIMARY KEY, name TEXT)", .{table_name}); defer allocator.free(create_public); try execSQL(setup_conn, create_public); @@ -121,28 +121,40 @@ test "matchStreams: a change from another schema is not routed to a public strea mut_batch.deinit(); } - const stream = try test_helpers.createTestStreamConfig(allocator, table_name, "unused_topic"); - defer allocator.free(stream.name); - const streams = [_]Stream{stream}; + // One stream per schema: the public one uses a bare resource (defaults to + // public), the other qualifies it as "schema.table". + const stream_public = try test_helpers.createTestStreamConfig(allocator, table_name, "unused_public"); + defer allocator.free(stream_public.name); + defer allocator.free(stream_public.source.resource); - // Both inserts arrive tagged with their real schema; only the public one - // matches the stream, the other-schema change matches nothing. + const other_resource = try std.fmt.allocPrint(allocator, "{s}.{s}", .{ other_schema, table_name }); + defer allocator.free(other_resource); + const stream_other = try test_helpers.createTestStreamConfig(allocator, other_resource, "unused_other"); + defer allocator.free(stream_other.name); + defer allocator.free(stream_other.source.resource); + + const streams = [_]Stream{ stream_public, stream_other }; + + // Each insert matches exactly the stream whose fully-qualified resource equals + // the change's, so the public row lands on the public stream and the + // other-schema row on the other stream, despite the shared table name. var public_matched: usize = 0; - var other_dropped: usize = 0; + var other_matched: usize = 0; for (batch.changes) |change| { var matched = try matchStreams(allocator, &streams, change); defer matched.deinit(allocator); - if (std.mem.eql(u8, change.meta.schema, "public")) { - try testing.expectEqual(@as(usize, 1), matched.items.len); + try testing.expectEqual(@as(usize, 1), matched.items.len); + try testing.expectEqualStrings(change.meta.resource, matched.items[0].source.resource); + + if (std.mem.eql(u8, change.meta.resource, stream_public.source.resource)) { public_matched += 1; } else { - try testing.expectEqualStrings(other_schema, change.meta.schema); - try testing.expectEqual(@as(usize, 0), matched.items.len); - other_dropped += 1; + try testing.expectEqualStrings(stream_other.source.resource, change.meta.resource); + other_matched += 1; } } try testing.expectEqual(@as(usize, 1), public_matched); - try testing.expectEqual(@as(usize, 1), other_dropped); + try testing.expectEqual(@as(usize, 1), other_matched); } diff --git a/src/serialization/json.zig b/src/serialization/json.zig index 6e1bec4..f721ef4 100644 --- a/src/serialization/json.zig +++ b/src/serialization/json.zig @@ -32,8 +32,6 @@ pub const JsonSerializer = struct { try encodeString(event.meta.source, writer); try writer.writeAll(",\"resource\":"); try encodeString(event.meta.resource, writer); - try writer.writeAll(",\"schema\":"); - try encodeString(event.meta.schema, writer); try writer.writeAll(",\"timestamp\":"); try writer.print("{d}", .{event.meta.timestamp}); @@ -149,7 +147,6 @@ test "JsonSerializer serialize INSERT event" { 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, }; @@ -194,7 +191,6 @@ test "JsonSerializer serialize UPDATE event" { 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, }; @@ -240,7 +236,6 @@ test "JsonSerializer rejects non-finite float" { 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, }; @@ -272,7 +267,6 @@ test "JsonSerializer validate JSON output is parseable" { 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, }; @@ -317,7 +311,6 @@ test "JsonSerializer string escaping" { 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, }; @@ -359,7 +352,6 @@ test "JsonSerializer escapes control characters and quoted field names" { 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, }; diff --git a/src/source/postgres/converter.zig b/src/source/postgres/converter.zig index f4b87a8..e482b7b 100644 --- a/src/source/postgres/converter.zig +++ b/src/source/postgres/converter.zig @@ -88,8 +88,9 @@ pub const Converter = struct { fn buildMetadata(self: *Self, allocator: std.mem.Allocator, rel_info: anytype, lsn: u64) !Metadata { return .{ .source = try allocator.dupe(u8, "postgres"), - .resource = try allocator.dupe(u8, rel_info.relation_name), - .schema = try allocator.dupe(u8, rel_info.namespace), + // Fully-qualified name (schema.table); the downstream stream match is a + // plain string compare, so the schema is part of the identity here. + .resource = try std.fmt.allocPrint(allocator, "{s}.{s}", .{ rel_info.namespace, rel_info.relation_name }), // Commit time of the surrounding transaction in Unix seconds, shifted // from the Postgres epoch. 0 if no BEGIN was seen (protocol-impossible // for DML on a healthy stream). @@ -353,8 +354,7 @@ test "convert INSERT: basic message to ChangeEvent" { // Verify: metadata try testing.expectEqualStrings("postgres", event.meta.source); - try testing.expectEqualStrings("users", event.meta.resource); - try testing.expectEqualStrings("public", event.meta.schema); + try testing.expectEqualStrings("public.users", event.meta.resource); try testing.expectEqual(@as(i64, 1700000000), event.meta.timestamp); try testing.expectEqualStrings("1/3259A308", event.meta.lsn.?); @@ -443,8 +443,7 @@ test "convert UPDATE: message with old and new tuples" { // Verify: metadata (no BEGIN seen in this test -> timestamp stays 0) try testing.expectEqualStrings("postgres", event.meta.source); - try testing.expectEqualStrings("users", event.meta.resource); - try testing.expectEqualStrings("public", event.meta.schema); + try testing.expectEqualStrings("public.users", event.meta.resource); try testing.expectEqual(@as(i64, 0), event.meta.timestamp); try testing.expectEqualStrings("0/0", event.meta.lsn.?); @@ -524,8 +523,7 @@ test "convert DELETE: message to ChangeEvent" { // Verify: metadata try testing.expectEqualStrings("postgres", event.meta.source); - try testing.expectEqualStrings("users", event.meta.resource); - try testing.expectEqualStrings("public", event.meta.schema); + try testing.expectEqualStrings("public.users", event.meta.resource); // Verify: delete_data present try testing.expect(event.data == .delete); diff --git a/src/testing/test_helpers.zig b/src/testing/test_helpers.zig index d7526c3..fa50d55 100644 --- a/src/testing/test_helpers.zig +++ b/src/testing/test_helpers.zig @@ -82,29 +82,23 @@ pub fn createTestTable(conn: *c.PGconn, allocator: std.mem.Allocator, table_name _ = c.PQexec(conn, replica_sql_z.ptr); } -/// Create test stream configuration +/// Create test stream configuration. `table_name` may be bare or schema-qualified; +/// the resource is normalized the same way the config loader does, so it matches +/// the source's qualified `meta.resource`. Caller frees `name` and `source.resource`. pub fn createTestStreamConfig(allocator: std.mem.Allocator, table_name: []const u8, topic_name: []const u8) !Stream { - const source = StreamSource{ - .resource = table_name, - .operations = &[_][]const u8{ "insert", "update", "delete" }, - }; - - const flow = StreamFlow{ - .format = "json", - }; - - const sink = StreamSink{ - .destination = topic_name, - .routing_key = "id", - }; + const resource = try config_module.qualifyResourceName(allocator, table_name); + errdefer allocator.free(resource); const name = try allocator.dupe(u8, table_name); return Stream{ .name = name, - .source = source, - .flow = flow, - .sink = sink, + .source = StreamSource{ + .resource = resource, + .operations = &[_][]const u8{ "insert", "update", "delete" }, + }, + .flow = StreamFlow{ .format = "json" }, + .sink = StreamSink{ .destination = topic_name, .routing_key = "id" }, }; } From 087a03a2ceaadbe404f4de15870165c9161c14f5 Mon Sep 17 00:00:00 2001 From: Yegor Lukash Date: Mon, 3 Aug 2026 17:02:22 +0300 Subject: [PATCH 2/3] Build the qualified resource with one concat instead of allocPrint --- src/source/postgres/converter.zig | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/source/postgres/converter.zig b/src/source/postgres/converter.zig index e482b7b..135fa1a 100644 --- a/src/source/postgres/converter.zig +++ b/src/source/postgres/converter.zig @@ -89,8 +89,9 @@ pub const Converter = struct { return .{ .source = try allocator.dupe(u8, "postgres"), // Fully-qualified name (schema.table); the downstream stream match is a - // plain string compare, so the schema is part of the identity here. - .resource = try std.fmt.allocPrint(allocator, "{s}.{s}", .{ rel_info.namespace, rel_info.relation_name }), + // plain string compare, so the schema is part of the identity. concat is + // one allocation; allocPrint's growing writer would take several. + .resource = try std.mem.concat(allocator, u8, &[_][]const u8{ rel_info.namespace, ".", rel_info.relation_name }), // Commit time of the surrounding transaction in Unix seconds, shifted // from the Postgres epoch. 0 if no BEGIN was seen (protocol-impossible // for DML on a healthy stream). From 8de89c743953a131da7f24c091647ad88f81832d Mon Sep 17 00:00:00 2001 From: Yegor Lukash Date: Mon, 3 Aug 2026 19:12:29 +0300 Subject: [PATCH 3/3] Treat the resource as one opaque name: normalize on load, validate via to_regclass --- src/config/config.zig | 129 ++++++++++++------------- src/main.zig | 17 ++-- src/source/postgres/validator.zig | 45 +++++---- src/source/postgres/validator_test.zig | 20 ++-- src/testing/test_helpers.zig | 7 +- 5 files changed, 108 insertions(+), 110 deletions(-) diff --git a/src/config/config.zig b/src/config/config.zig index 85c4de1..198ad39 100644 --- a/src/config/config.zig +++ b/src/config/config.zig @@ -92,36 +92,13 @@ pub const SinkConfig = struct { }; pub const StreamSource = struct { - // Fully-qualified resource: table/collection/index as "schema.table". A bare - // name is accepted and normalized to the public schema by qualifyResources. + // 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"] - - /// A resource split into namespace and object name. For Postgres this is - /// schema + table. - pub const Qualified = struct { - schema: []const u8, - name: []const u8, - }; - - /// Split `resource` on its '.' into schema and table, for the source-side - /// validator that queries the two separately. qualifyResources runs first, so - /// a dot is present; a bare name still falls back to the public schema. - pub fn qualifiedResource(self: StreamSource) Qualified { - if (std.mem.indexOfScalar(u8, self.resource, '.')) |dot| { - return .{ .schema = self.resource[0..dot], .name = self.resource[dot + 1 ..] }; - } - return .{ .schema = "public", .name = self.resource }; - } }; -/// Fully-qualify a resource name as `schema.table`, defaulting a bare name (no -/// '.') to the `public` schema. Caller owns the result. -pub fn qualifyResourceName(allocator: std.mem.Allocator, resource: []const u8) ![]const u8 { - if (std.mem.indexOfScalar(u8, resource, '.') != null) return allocator.dupe(u8, resource); - return std.fmt.allocPrint(allocator, "public.{s}", .{resource}); -} - pub const StreamFlow = struct { format: []const u8, }; @@ -173,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. @@ -460,16 +443,17 @@ pub const Config = struct { try validateStreams(allocator, self.streams); } - /// Rewrite each stream's `resource` into its fully-qualified `schema.table` - /// form (a bare name defaults to the public schema), so the processor matches - /// it against the source's qualified resource with a plain compare. Call once - /// after validate(). Mutates the stream slice in place, so it must be the - /// mutable arena memory from the TOML parser; pass that arena's allocator - /// (`parsed.arena.allocator()`) so the rewritten names share the config's - /// lifetime. - pub fn qualifyResources(self: Config, allocator: std.mem.Allocator) !void { - for (@constCast(self.streams)) |*stream| { - stream.source.resource = try qualifyResourceName(allocator, stream.source.resource); + // 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}); + } } } }; @@ -570,36 +554,49 @@ test "Stream.hasDeleteOperation reflects the configured operations" { try testing.expect(with_delete.hasDeleteOperation()); } -test "qualifiedResource splits schema.table and defaults to public" { - const bare: StreamSource = .{ .resource = "users", .operations = &.{} }; - const q1 = bare.qualifiedResource(); - try testing.expectEqualStrings("public", q1.schema); - try testing.expectEqualStrings("users", q1.name); - - const qualified: StreamSource = .{ .resource = "app.users", .operations = &.{} }; - const q2 = qualified.qualifiedResource(); - try testing.expectEqualStrings("app", q2.schema); - try testing.expectEqualStrings("users", q2.name); -} - -test "qualifyResourceName defaults a bare name to public and keeps a qualified one" { - const bare = try qualifyResourceName(testing.allocator, "users"); - defer testing.allocator.free(bare); - try testing.expectEqualStrings("public.users", bare); - - const qualified = try qualifyResourceName(testing.allocator, "app.users"); - defer testing.allocator.free(qualified); - try testing.expectEqualStrings("app.users", qualified); -} - -test "qualifyResources rewrites bare stream resources in place" { +test "load normalizes a bare resource to the public schema" { var parsed = try Config.loadFromTomlString(testing.allocator, valid_config_toml); defer parsed.deinit(); - - try parsed.value.qualifyResources(parsed.arena.allocator()); 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]); @@ -756,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]); @@ -862,7 +859,7 @@ 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); @@ -870,7 +867,7 @@ test "parse multiple streams" { 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); diff --git a/src/main.zig b/src/main.zig index a9ce93e..01e8302 100644 --- a/src/main.zig +++ b/src/main.zig @@ -90,9 +90,6 @@ fn run(init: std.process.Init) !void { const config = parsed.value; try config.validate(allocator); - // Normalize bare resource names to schema.table so the processor matches them - // against the source's qualified names. Names live in the Parsed arena. - try config.qualifyResources(parsed.arena.allocator()); const conninfo = try config.loadPostgresConninfo(allocator, init.environ_map); defer allocator.free(conninfo); @@ -289,25 +286,25 @@ fn validatePostgres(allocator: std.mem.Allocator, cfg: Config, conninfo: []const try validator.checkWalLevel(); for (cfg.streams) |stream| { - const target = stream.source.qualifiedResource(); + const resource = stream.source.resource; - validator.checkTableExists(target.schema, target.name) catch |err| { - printStatus("ERROR: Table validation failed for '{s}': {}\n", .{ stream.source.resource, err }); + validator.checkTableExists(resource) catch |err| { + printStatus("ERROR: Table validation failed for '{s}': {}\n", .{ resource, err }); return err; }; // A missing routing key column would silently collapse partitioning. const routing_key = stream.sink.routing_key; - validator.checkColumnExists(target.schema, target.name, routing_key) catch |err| { - printStatus("ERROR: Routing key validation failed for '{s}' (column '{s}'): {}\n", .{ stream.source.resource, routing_key, err }); + validator.checkColumnExists(resource, routing_key) catch |err| { + printStatus("ERROR: Routing key validation failed for '{s}' (column '{s}'): {}\n", .{ resource, routing_key, err }); return err; }; // Only a stream that tracks DELETE needs REPLICA IDENTITY FULL, so the // deleted row carries all columns. if (stream.hasDeleteOperation()) { - validator.checkReplicaIdentity(target.schema, target.name) catch |err| { - printStatus("ERROR: Replica identity validation failed for '{s}': {}\n", .{ stream.source.resource, err }); + validator.checkReplicaIdentity(resource) catch |err| { + printStatus("ERROR: Replica identity validation failed for '{s}': {}\n", .{ resource, err }); return err; }; } diff --git a/src/source/postgres/validator.zig b/src/source/postgres/validator.zig index 45c4599..d174e33 100644 --- a/src/source/postgres/validator.zig +++ b/src/source/postgres/validator.zig @@ -108,45 +108,44 @@ pub const PostgresValidator = struct { print("PostgreSQL validation: wal_level = '{s}' ✓\n", .{wal_level_str}); } - pub fn checkTableExists(self: *Self, schema: []const u8, table_name: []const u8) ValidationError!void { - const query = std.fmt.allocPrintSentinel(self.allocator, "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_schema = '{s}' AND table_name = '{s}');", .{ schema, table_name }, 0) catch return ValidationError.OutOfMemory; + pub fn checkTableExists(self: *Self, resource: []const u8) ValidationError!void { + // to_regclass resolves the whole `schema.table` name (a bare name via + // search_path) and returns NULL when it does not exist, so the resource + // stays one opaque string here. + const query = std.fmt.allocPrintSentinel(self.allocator, "SELECT to_regclass('{s}') IS NOT NULL;", .{resource}, 0) catch return ValidationError.OutOfMemory; defer self.allocator.free(query); const result = try self.executeQuery(query.ptr); defer c.PQclear(result); - const exists = c.PQgetvalue(result, 0, 0); - const exists_str = std.mem.span(exists); - - if (!std.mem.eql(u8, exists_str, "t")) { - std.log.warn("PostgreSQL validation: Table '{s}.{s}' does not exist", .{ schema, table_name }); - std.log.warn("Fix: Create the table or check the table name in configuration", .{}); + const exists = std.mem.span(c.PQgetvalue(result, 0, 0)); + if (!std.mem.eql(u8, exists, "t")) { + std.log.warn("PostgreSQL validation: Table '{s}' does not exist", .{resource}); + std.log.warn("Fix: create the table or check the resource name in configuration", .{}); return ValidationError.TableNotFound; } - print("PostgreSQL validation: Table '{s}.{s}' exists ✓\n", .{ schema, table_name }); + print("PostgreSQL validation: Table '{s}' exists ✓\n", .{resource}); } /// Check that a column exists on a table. Used for the stream's routing key: /// a typo (or the default `id` on a table without one) would otherwise route /// every change to the same partition, unnoticed. - pub fn checkColumnExists(self: *Self, schema: []const u8, table_name: []const u8, column_name: []const u8) ValidationError!void { - const query = std.fmt.allocPrintSentinel(self.allocator, "SELECT EXISTS (SELECT FROM information_schema.columns WHERE table_schema = '{s}' AND table_name = '{s}' AND column_name = '{s}');", .{ schema, table_name, column_name }, 0) catch return ValidationError.OutOfMemory; + pub fn checkColumnExists(self: *Self, resource: []const u8, column_name: []const u8) ValidationError!void { + const query = std.fmt.allocPrintSentinel(self.allocator, "SELECT EXISTS (SELECT FROM pg_attribute WHERE attrelid = to_regclass('{s}') AND attname = '{s}' AND attnum > 0 AND NOT attisdropped);", .{ resource, column_name }, 0) catch return ValidationError.OutOfMemory; defer self.allocator.free(query); const result = try self.executeQuery(query.ptr); defer c.PQclear(result); - const exists = c.PQgetvalue(result, 0, 0); - const exists_str = std.mem.span(exists); - - if (!std.mem.eql(u8, exists_str, "t")) { - std.log.warn("PostgreSQL validation: Column '{s}' does not exist on table '{s}.{s}'", .{ column_name, schema, table_name }); + const exists = std.mem.span(c.PQgetvalue(result, 0, 0)); + if (!std.mem.eql(u8, exists, "t")) { + std.log.warn("PostgreSQL validation: Column '{s}' does not exist on table '{s}'", .{ column_name, resource }); std.log.warn("Fix: set stream.sink.routing_key to an existing column", .{}); return ValidationError.ColumnNotFound; } - print("PostgreSQL validation: Column '{s}.{s}.{s}' exists ✓\n", .{ schema, table_name, column_name }); + print("PostgreSQL validation: Column '{s}.{s}' exists ✓\n", .{ resource, column_name }); } /// Require REPLICA IDENTITY FULL on a table whose stream tracks DELETE, so the @@ -154,8 +153,8 @@ pub const PostgresValidator = struct { /// drops the non-key columns from the DELETE old row, breaking the documented /// format. Call only for delete-tracking streams: FULL is irrelevant otherwise /// and only inflates UPDATE WAL. - pub fn checkReplicaIdentity(self: *Self, schema: []const u8, table_name: []const u8) ValidationError!void { - const query = try std.fmt.allocPrintSentinel(self.allocator, "SELECT c.relreplident FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = '{s}' AND c.relname = '{s}';", .{ schema, table_name }, 0); + pub fn checkReplicaIdentity(self: *Self, resource: []const u8) ValidationError!void { + const query = try std.fmt.allocPrintSentinel(self.allocator, "SELECT relreplident FROM pg_class WHERE oid = to_regclass('{s}');", .{resource}, 0); defer self.allocator.free(query); const result = try self.executeQuery(query.ptr); @@ -164,19 +163,19 @@ pub const PostgresValidator = struct { // checkTableExists runs first, so an empty result only happens on a race // (the table was dropped between the two queries). if (c.PQntuples(result) == 0) { - std.log.warn("PostgreSQL validation: Table '{s}.{s}' not found while checking replica identity", .{ schema, table_name }); + std.log.warn("PostgreSQL validation: Table '{s}' not found while checking replica identity", .{resource}); return ValidationError.TableNotFound; } const identity = std.mem.span(c.PQgetvalue(result, 0, 0)); if (identity.len == 0 or identity[0] != 'f') { - std.log.warn("PostgreSQL validation: Table '{s}.{s}' has REPLICA IDENTITY {s}, but this stream tracks DELETE and needs the full old row", .{ schema, table_name, replicaIdentityName(identity) }); - std.log.warn("Fix: ALTER TABLE {s}.{s} REPLICA IDENTITY FULL", .{ schema, table_name }); + std.log.warn("PostgreSQL validation: Table '{s}' has REPLICA IDENTITY {s}, but this stream tracks DELETE and needs the full old row", .{ resource, replicaIdentityName(identity) }); + std.log.warn("Fix: ALTER TABLE {s} REPLICA IDENTITY FULL", .{resource}); return ValidationError.InvalidReplicaIdentity; } - print("PostgreSQL validation: Table '{s}.{s}' REPLICA IDENTITY FULL ✓\n", .{ schema, table_name }); + print("PostgreSQL validation: Table '{s}' REPLICA IDENTITY FULL ✓\n", .{resource}); } }; diff --git a/src/source/postgres/validator_test.zig b/src/source/postgres/validator_test.zig index 554cc15..24bdd8c 100644 --- a/src/source/postgres/validator_test.zig +++ b/src/source/postgres/validator_test.zig @@ -51,7 +51,7 @@ test "PostgresValidator: table existence check requires PostgreSQL" { const conn_str = "host=localhost port=5432 dbname=outboxx_test user=postgres password=password"; try validator.connect(conn_str); - try validator.checkTableExists("public", "users"); + try validator.checkTableExists("public.users"); std.log.info("PostgreSQL validation: Table existence check passed", .{}); } @@ -64,7 +64,7 @@ test "PostgresValidator: table not found should error" { try validator.connect(conn_str); - const result = validator.checkTableExists("public", "nonexistent_table_xyz"); + const result = validator.checkTableExists("public.nonexistent_table_xyz"); try testing.expectError(error.TableNotFound, result); } @@ -77,7 +77,7 @@ test "PostgresValidator: invalid schema should error" { try validator.connect(conn_str); - const result = validator.checkTableExists("nonexistent_schema_xyz", "users"); + const result = validator.checkTableExists("nonexistent_schema_xyz.users"); try testing.expectError(error.TableNotFound, result); } @@ -89,7 +89,7 @@ test "PostgresValidator: routing key column exists" { const conn_str = "host=localhost port=5432 dbname=outboxx_test user=postgres password=password"; try validator.connect(conn_str); - try validator.checkColumnExists("public", "users", "id"); + try validator.checkColumnExists("public.users", "id"); } test "PostgresValidator: missing routing key column should error" { @@ -101,7 +101,7 @@ test "PostgresValidator: missing routing key column should error" { try validator.connect(conn_str); - const result = validator.checkColumnExists("public", "users", "nonexistent_column_xyz"); + const result = validator.checkColumnExists("public.users", "nonexistent_column_xyz"); try testing.expectError(error.ColumnNotFound, result); } @@ -114,7 +114,7 @@ test "PostgresValidator: replica identity FULL passes" { try validator.connect(conn_str); // users is set to REPLICA IDENTITY FULL by the dev init script. - try validator.checkReplicaIdentity("public", "users"); + try validator.checkReplicaIdentity("public.users"); } test "PostgresValidator: replica identity not FULL should error" { @@ -127,7 +127,7 @@ test "PostgresValidator: replica identity not FULL should error" { try validator.connect(conn_str); // system_logs keeps the default replica identity (the init script never // alters it), so a delete-tracking stream on it must be rejected. - const result = validator.checkReplicaIdentity("public", "system_logs"); + const result = validator.checkReplicaIdentity("public.system_logs"); try testing.expectError(error.InvalidReplicaIdentity, result); } @@ -161,13 +161,13 @@ test "PostgresValidator: methods fail when not connected" { const wal_result = validator.checkWalLevel(); try testing.expectError(error.ConnectionFailed, wal_result); - const table_result = validator.checkTableExists("public", "users"); + const table_result = validator.checkTableExists("public.users"); try testing.expectError(error.ConnectionFailed, table_result); - const column_result = validator.checkColumnExists("public", "users", "id"); + const column_result = validator.checkColumnExists("public.users", "id"); try testing.expectError(error.ConnectionFailed, column_result); - const identity_result = validator.checkReplicaIdentity("public", "users"); + const identity_result = validator.checkReplicaIdentity("public.users"); try testing.expectError(error.ConnectionFailed, identity_result); } diff --git a/src/testing/test_helpers.zig b/src/testing/test_helpers.zig index fa50d55..0b33249 100644 --- a/src/testing/test_helpers.zig +++ b/src/testing/test_helpers.zig @@ -86,7 +86,12 @@ pub fn createTestTable(conn: *c.PGconn, allocator: std.mem.Allocator, table_name /// the resource is normalized the same way the config loader does, so it matches /// the source's qualified `meta.resource`. Caller frees `name` and `source.resource`. pub fn createTestStreamConfig(allocator: std.mem.Allocator, table_name: []const u8, topic_name: []const u8) !Stream { - const resource = try config_module.qualifyResourceName(allocator, table_name); + // Mirror the config loader: a bare name resolves to the public schema, so the + // stream resource matches the source's qualified meta.resource. + const resource = if (std.mem.indexOfScalar(u8, table_name, '.') != null) + try allocator.dupe(u8, table_name) + else + try std.fmt.allocPrint(allocator, "public.{s}", .{table_name}); errdefer allocator.free(resource); const name = try allocator.dupe(u8, table_name);