[clickhouse] apply TableSchemaDeltas in normalize #2148
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
Would be great if we could get in #2114 as part of landing this |
serprex
reviewed
Dec 9, 2024
| return nil, err | ||
| } | ||
|
|
||
| if err = c.replayTableSchemaDeltas(ctx, schemaDeltas); err != nil { |
There was a problem hiding this comment.
Suggested change
| if err = c.replayTableSchemaDeltas(ctx, schemaDeltas); err != nil { | |
| if err := c.replayTableSchemaDeltas(ctx, schemaDeltas); err != nil { |
serprex
reviewed
Dec 9, 2024
flow/connectors/clickhouse/cdc.go
Outdated
| @@ -122,8 +119,16 @@ func (c *ClickHouseConnector) SyncRecords(ctx context.Context, req *model.SyncRe | |||
| return res, nil | |||
| } | |||
|
|
|||
| // For CH specifically, we don't want to replay during sync, but rather during the normalization phase. | |||
| // This a no-op for CH, private variant to be called during normalize. | |||
| func (c *ClickHouseConnector) ReplayTableSchemaDeltas(ctx context.Context, flowJobName string, | |||
There was a problem hiding this comment.
should move this method to a separate connector interface, there's a few connectors that skip this already
serprex
reviewed
Dec 9, 2024
| @@ -172,6 +173,15 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon | |||
| startTime := time.Now() | |||
| errGroup, errCtx := errgroup.WithContext(ctx) | |||
| errGroup.Go(func() error { | |||
| pgMetadata := connmetadata.NewPostgresMetadataFromCatalog(logger, a.CatalogPool) | |||
There was a problem hiding this comment.
Further down we have
syncBatchID, err := dstConn.GetLastSyncBatchID(errCtx, flowName)
Could lift that up to where we get lastOffset. Then use same value between two goroutines. Would make this logic work for connectors (pg:pg) which don't use catalog for metadata too
heavycrystal
force-pushed
the
ch-apply-schemadeltas-in-normalize
branch
from
ed726eb to
2c645b1
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
TableSchemaDeltas are already stored in catalog for audit
instead of applying in SyncFlow like other connectors, apply them during normalize
as part of the philosophy of not connecting to CH during SyncFlow