fix: handle drop column on clickhouse and remove from schema in clickpipes - #4609
fix: handle drop column on clickhouse and remove from schema in clickpipes#4609masterashu wants to merge 4 commits into
Conversation
❌ Test FailureAnalysis: The new test Test_Column_Added_Back_After_Destination_Column_Drop timed out because the ClickHouse destination never received the re-added record (stuck at 2 vs expected 3 for the full 200s) in two suite variants, indicating a real bug in this PR's column drop/re-add handling rather than a flaky failure. |
❌ Test FailureAnalysis: The PR's own column-drop feature test (Test_Column_Added_Back_After_Destination_Column_Drop) fails deterministically on both ClickHouse and ClickHouse_Cluster variants with a persistent, non-converging record mismatch (source=3 vs destination=2) that never normalizes — a real regression in the PR's column-drop handling, not a timing flake. |
Code reviewNo issues found. Checked for bugs and CLAUDE.md compliance. |
|
@masterashu could you add a description for the expected behavior for this PR? Thanks! |
| actualColSet[col.Name] = struct{}{} | ||
| } | ||
|
|
||
| currentSchema := gen.tableNameSchemaMapping[gen.TableName] |
There was a problem hiding this comment.
The gen.tableNameSchemaMapping does not represent the freshest schema stored in the catalog, but instead represent the schema that was fetched from catalog earlier at the start of the startNormalize. This introduces a race condition between Sync and Normalize flow:
- normalize fetch schemas from catalog
- sync applies an
add columnDDL and updated the catalog - normalize take this schemas from 1, removes the deleted column, and saves it back to catalog
The catalog is now stuck with a stale schema with a missing new column
Since we only care about column removal in the error handling, instead of passing back the updated schemas, pass back the removed column instead for the NormalizeResponse. Then in the readModifyLoop you can focus on just removing the deleted column from the fresh schema fetched from catalog, instead of replacing the entire schema today.
| c.logger.Warn("[clickhouse] auto-removing destination-dropped columns from schema", | ||
| slog.String("table", gen.TableName), | ||
| slog.Any("removedColumns", removedCols)) |
There was a problem hiding this comment.
iiuc, since we don't check against source db schema, if the user drops the column from ClickHouse today, but does not drop it from source, we will gracefully recover in this scenario as well and continue to replicate without the dropped column; and this column essentially becomes an excluded column, unless the table gets resynced which it would reappear again.
I think this is more of a product decision on whether we should fail loudly here, vs. we should continue. My only concern here with continue is if customer accidentally dropped a column, then previously it would fail the pipe, they would have manually added it back and the pipe would recover. vs with this change manually adding back the column would do nothing, and would require a resync.
If this is intended behavior we should have an user-facing log for it
cc @morsapaes on thoughts here.
There was a problem hiding this comment.
Just realized that if user dropped the column and then add it back, the historical data would be lost anyways. so maybe continuing the pipe is the more graceful option here (and user will just have to resync if it was a mistake)
🔄 Flaky Test DetectedAnalysis: The flow/e2e package hit its 20m go test deadline ( ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: Test_MariaDB_PartialRowEvent failed during container setup with "readInitialHandshake ... err EOF: connection was bad" — a MariaDB testcontainer readiness race unrelated to this ClickHouse column-drop PR, and the same test passed in the other two matrix jobs. ✅ Automatically retrying the workflow |
| logger.Error("failed to persist auto-removed destination columns to catalog", | ||
| slog.Any("error", updateErr)) |
There was a problem hiding this comment.
this should fail loudly and get retried. while it's also reasonable to just let the next iteration pick up the missing column if there is a transient error, if there's any internal errors we wouldn't be notified and it would always fail to persist to catalog.
There was a problem hiding this comment.
also means can remove the updateErr == nil check below
| a.Alerter.LogFlowWarning(ctx, flowJobName, | ||
| fmt.Errorf("destination column(s) were dropped from table %s and automatically removed "+ | ||
| "from the mirror schema; future syncs will omit those columns", name)) | ||
| } |
There was a problem hiding this comment.
currently the warning message doesn't specify columns that are dropped, let's make the log more informative about that, since we now have this information
|
Codex identified a race condition worth mentioning: Today the The edge case that can cause correctness issue is:
Given catalog never removed the column, pg's ADD COLUMN ddl got ignored since no schema delta was detected, so ClickHouse never re-added that column. In this case, once the normalize error finally got unblocked, subsequent normalize will silently perform the "schema correction" introduced in this PR, and then permanently skip replicating the column that was added back. The odds of this happening is pretty small, but not zero. Here's a potential best-effort workaround: #4678. It's still not perfect because if normalize fails and catalog correction fails, catalog correction doesn't get retried. A more bulletproof solution would be to update schema and advance offset together when there are column removals, but I am not sure if this additional code complexity is worthwhile given the unlikely scenario. I am also okay with not trying to handle this edge case at all, if we feel that this is getting too much into the weeds. If user delete a column in CH accidentally and added it back, is probably warrants a resync anyways. if user delete a column in PG and added it back immediately, it's also very unlikely they also manually removed the column from ClickHouse in the same batch. |
|
discussed separately in slack, would be safer to add a source-check to make sure columns are also removed from source table before updating the catalog; or keep erroring otherwise. This also allow us to keep the invariant of replicated columns = all columns - excluded columns, while allowing user to manually drop columns in clickhouse that have already been deleted from source, if they desire. |
2 participants
Problem
When customer removes a column from table in Clickhouse, Normalize fails due to SQL query error mentioning unknown column.
In order to deal with this, we had to change the source schema for the table to remove that column.
This PR aims to make this easier by identifying if the column is deleted and retrying and updating the schema.
We check the errors in Normalization step and match if its occurring from a missing column, if it is we refresh the schema and retry normalization and also update the schema.