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 3 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.
| // removes any columns from the schema that no longer exist there, and returns a rebuilt normalize query | ||
| // together with the filtered schema. This is used to recover from ErrNoSuchColumnInTable when the user | ||
| // has manually dropped a column from the destination after it was dropped at the source. | ||
| func (c *ClickHouseConnector) refreshSchemaAndRebuildQuery( |
There was a problem hiding this comment.
the e2e tests are great; let's also add unit tests for refreshSchemaAndRebuildQuery
| // If a destination column is missing (user dropped it), try to auto-recover | ||
| // by refreshing the schema and retrying the INSERT without that column. | ||
| var chEx *clickhouse.Exception | ||
| if errors.As(err, &chEx) && ch_proto.Error(chEx.Code) == ch_proto.ErrNoSuchColumnInTable { |
There was a problem hiding this comment.
nit: errors.AsType[*clickhouse.Exception](err)
| 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)
| ); updateErr != nil { | ||
| logger.Error("failed to persist auto-refreshed table schemas to catalog", | ||
| slog.Any("error", updateErr)) | ||
| } | ||
| for name, schema := range res.UpdatedSchemaMapping { | ||
| tableNameSchemaMapping[name] = schema | ||
| a.Alerter.LogFlowWarning(ctx, config.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.
If ReadModifyWriteTableSchemasToCatalog fails above, this would still add a user-facing warning that destination columns were dropped. should only surface log if update succeeds
| @@ -524,6 +531,44 @@ func (c *ClickHouseConnector) NormalizeRecords( | |||
| slog.Int("parallelWorker", i)) | |||
|
|
|||
| if err := c.execWithConnection(errCtx, chConn, q.query); err != nil { | |||
There was a problem hiding this comment.
nit: can factor out exec into a helper to improve readability, currently the error handling with continue is a bit hard to follow; and there are some duplicated loggings introduced that wouldn't be needed. e.g.
func (c *ClickHouseConnector) execNormalizeQuery(
ctx context.Context,
logger log.Logger,
conn clickhouse.Conn,
q queryInfo,
) (*protos.TableSchema, error) {
err := c.execWithConnection(ctx, conn, q.query)
if err == nil {
return nil, nil
}
chEx, ok := errors.AsType[*clickhouse.Exception](err)
if !ok || chEx == nil || ch_proto.Error(chEx.Code) != ch_proto.ErrNoSuchColumnInTable {
return nil, err
}
logger.Warn("[clickhouse] missing column in destination, attempting schema refresh and retry",
slog.Any("error", err))
retryQuery, refreshedSchema, refreshErr := c.refreshSchemaAndRebuildQuery(ctx, q.generator)
if refreshErr != nil {
logger.Error("[clickhouse] failed to refresh schema after missing column error",
slog.Any("error", refreshErr))
return nil, err
}
if retryErr := c.execWithConnection(ctx, conn, retryQuery); retryErr != nil {
logger.Error("[clickhouse] retry after schema refresh failed",
slog.String("query", retryQuery), slog.Any("error", retryErr))
return nil, err
}
logger.Warn("[clickhouse] successfully retried normalize after dropping missing destination columns")
return refreshedSchema, nil
}
| } | ||
|
|
||
| a.recordDeliveryLag(ctx, config.FlowJobName, res.StartBatchID, res.EndBatchID) | ||
| // If any destination columns were auto-removed during normalize (user dropped them), persist |
There was a problem hiding this comment.
nit: better implemented as a helper as this method is getting quite long
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.