fix(postgres): validate publication contains all selected streams before CDC#971
Conversation
|
|
video ExplanationIssue752_Fix_Ishanya_sharma.mp4 |
|
@ishanyawhocares Assigning @vikaxsh to review your PR |
|
Hello @vikaxsh, its been a week. just wanted to request the review to know if there is any changes required to the code, or is it ready to be merged? |
|
@ishanyawhocares Sorry for the delay in review. You will definitely have a review soon |
| if publication == "" || len(streams) == 0 { | ||
| return nil | ||
| } | ||
| pubTables, err := fetchPublicationTables(ctx, conn, publication) |
There was a problem hiding this comment.
Let's collapse this into two functions: one thin DB layer validatePublicationContainsStreams and one pure matcher checkStreamsInPublication
There was a problem hiding this comment.
The split is already as:
-
validatePublicationContainsStreams: the DB layer entry point called from
PreCDC (assigns the actual query to a small helper, fetchPublicationTables). -
checkStreamsInPublication: Its a pure matcher, no DB dependency.
Let me know if you'd rather have fetchPublicationTables folded back into
validatePublicationContainsStreams directly that is having only 2 functions. I would change that
|
|
||
| func (p *Postgres) PreCDC(ctx context.Context, streams []types.StreamInterface) error { | ||
| // Validate all selected streams exist in the publication before touching the slot. | ||
| // Fixes: https://github.com/datazip-inc/olake/issues/752 |
| // Build a normalised lookup set: "schema.table" → present | ||
| pubSet := make(map[string]struct{}, len(pubTables)) | ||
| for _, r := range pubTables { | ||
| key := strings.ToLower(r.Schema) + "." + strings.ToLower(r.Table) |
There was a problem hiding this comment.
why are you lowercasing table names?
There was a problem hiding this comment.
It was initially to avoid a case mismatch, but as streams.json is
generated by discover() off information_schema, table names in there are
already lowercase by default.
so this mismatch will never happen, but just kept it as a small defensive check, if someone edits streams.json, afterwards and types a table name with a different casing than the actual publication.
Happy to change it as well, let me know if you'd like that.
There was a problem hiding this comment.
we should keep it case case-sensitive
There was a problem hiding this comment.
current code might fail on case-sensitive table name (like myTable and myTABLE)
hash-data
requested a deployment
to
integration_tests
GitHub Actions
Waiting
hash-data
requested a deployment
to
integration_tests
GitHub Actions
Waiting
|
Hello @vikaxsh , I have updated the rest of the code as per the requirements, can you please confirm on the fetchPublicationTables review so I can update that as well. |
…lidate-publication-streams
|
hello @vikaxsh, just checking in, can you please go through this, so I can do final commits, and ready to merge. |
ishanyawhocares
requested a deployment
to
integration_tests
GitHub Actions
Waiting
ishanyawhocares
requested a deployment
to
integration_tests
GitHub Actions
Waiting
| ) | ||
| } | ||
|
|
||
| // Build a normalised lookup set: "schema.table" -> present |
There was a problem hiding this comment.
we have generic types.Set[T] implementation, we can use here as well
| // Build a normalised lookup set: "schema.table" → present | ||
| pubSet := make(map[string]struct{}, len(pubTables)) | ||
| for _, r := range pubTables { | ||
| key := strings.ToLower(r.Schema) + "." + strings.ToLower(r.Table) |
There was a problem hiding this comment.
we should keep it case case-sensitive
| // Build a normalised lookup set: "schema.table" → present | ||
| pubSet := make(map[string]struct{}, len(pubTables)) | ||
| for _, r := range pubTables { | ||
| key := strings.ToLower(r.Schema) + "." + strings.ToLower(r.Table) |
There was a problem hiding this comment.
current code might fail on case-sensitive table name (like myTable and myTABLE)
5 participants
Closes #752
Problem
When a table is selected in
streams.jsonbut not added to the Postgrespublication slot, OLake either fails deep in WAL processing with a cryptic
error, or silently produces no CDC data for that table — giving the user
no indication of what's wrong.
Solution
Added an explicit pre-flight check in
PreCDCthat queriespg_publication_tablesand compares the result against the selected streams.If any table is missing, the sync fails immediately with a clear, actionable
error: publication validation failed: the following tables are selected in streams.json
but are NOT tracked by publication "my_pub":
public.payments
Run: ALTER PUBLICATION my_pub ADD TABLE public.payments
Changes
drivers/postgres/internal/cdc.go: addedvalidatePublicationContainsStreams,checkStreamsInPublication,fetchPublicationTablesfunctions; called fromPreCDCdrivers/postgres/internal/cdc_publication_test.go: 10 table-driven unit testscovering happy path, missing tables, empty publication, case-insensitivity, etc.
Ishanya Sharma
BITS Pilani, Goa campus
cc @nayanj98