fix: potential data corruption bug and improve message sending behaviour #174
Conversation
imor
force-pushed
the
raminder/etl-148-process_syncing_tables-in-apply-loop-should-be-called-on
branch
from
2f20960 to
3c12160
Compare
…pply-loop-should-be-called-on
…pply-loop-should-be-called-on
imor
changed the title
imor
changed the title
| end_loop |= !hook | ||
| .process_syncing_tables(state.next_status_update.flush_lsn, true) | ||
| .await?; |
There was a problem hiding this comment.
With the process syncing here, the system becomes much harder to reason about, but I am failing to see any alternative that would work well with batching.
There was a problem hiding this comment.
Yes, it has become harder to see the logic, but as you said I can't think of a real alternative either.
| if let Some(table_id) = skip_table { | ||
| end_loop |= !hook.skip_table(table_id).await?; |
There was a problem hiding this comment.
Shouldn't this be outside of the sending of the batch? As far as I know we need to skip the table as soon as we get the event.
There was a problem hiding this comment.
This will be called as soon as the relation event indicates a change in schema because the end_batch.is_some() part in the if time_to_send_batch || state.events_batch.len() >= max_batch_size || end_batch.is_some() condition will be true in that case (end_batch will be set to EndBatch::Exclusive).
This again is important to call only after the batch is acked because if we set the table state to skipped before that we run the risk of missing events.
There was a problem hiding this comment.
Oh yeah, nvm, I missed the last condition.
imor
deleted the
raminder/etl-148-process_syncing_tables-in-apply-loop-should-be-called-on
branch
This PR:
ReplicationMessage<LogicalReplicationMessage>messages into a vec, which were later copied to another vec. This reduces number of allocations. Allocations can be further reduced if we send a slice to the destination instead of a vec, but that's not yet handled.I also wanted to create a stream which encapsulates the whole message filtering logic and produced
LogicalReplicationMessages and then batch the output of this stream, but this became complicated due to stream calling async methods. This will be done in a separate PR later on.