Add cdc iterator #22
Add cdc iterator #22
Conversation
alarbada
requested review from
raulb,
hariso,
lovromazgon,
lyuboxa and
maha-hajja
| return nil, fmt.Errorf("failed to get start position: %w", err) | ||
| } | ||
|
|
||
| iterator.canalTomb.Go(func() error { |
There was a problem hiding this comment.
We use tomb to manage a single goroutine, it feels like this could be simplified.
| c.canal.Close() | ||
| return fmt.Errorf("context cancelled from runCanal: %w", ctx.Err()) |
There was a problem hiding this comment.
Wouldn't it be nice if we waited for canal to stop and give us back an error or nil? This seems racy.
| case data, ok := <-c.data: | ||
| if !ok { // closed | ||
| if err := c.canalTomb.Err(); err != nil { | ||
| return sdk.Record{}, fmt.Errorf("canal exited unexpectedly: %w", err) |
There was a problem hiding this comment.
Is this really always unexpected? Can't there be a graceful stop of the canal, e.g. when the context gets cancelled? This also feels racy, based on which goroutine first detects the closing context we might get different errors returned from this function.
| for i, col := range columns { | ||
| val := rows[i] | ||
| if s, ok := val.(string); ok { | ||
| // I don't know why exactly, but "github.com/go-mysql-org/go-mysql/canal" |
There was a problem hiding this comment.
Hmm, strange. Ideally we would check the table columns and only execute this on columns that we know have a timestamp. Otherwise, we might be parsing dates when that's not intended.
| return payload | ||
| } | ||
|
|
||
| func tryParseCanalStrDate(s string) string { |
There was a problem hiding this comment.
Wait, so we parse the time and then format it? What's the purpose of this? 🧐
| } | ||
|
|
||
| func (c *cdcIterator) buildRecord(e *canal.RowsEvent) (sdk.Record, error) { | ||
| pos, err := c.canal.GetMasterPos() |
There was a problem hiding this comment.
The position handling here feels sketchy. Are we sure this is correct? How do we know that the name in this position corresponds to the one that e.Header.LogPos points to? 🤔
Description
Adds cdc iterator, with tests. Stacked from #21