Materialize Workflows: Periodically send copy state for workflows which are heavily filtered

[rohit-nayak-ps]

Description

We have an edge case where, if a materialize workflow is set up such that only very few rows match the filter, it is possible, during the copy phase, that no row is sent for an entire copyPhaseDuration either because there is no match or we never reach the packet size. Then the workflow will restart on the target without the copy state being updated and hence the workflow will never make progress.

This PR sends a sync response when a particular number of rows are filtered out without any response being sent.

Currently, the threshold is set to 10000 rows. It is probably a bit low and ideally it should be sent with time periodicity rather than row count periodicity, but since the copyPhaseDuration is set by the target, the source has no idea of what the duration is. A flag would make this configurable, but we are trying to get away a profusion of flags and enforce sensible defaults where possible.

We also add a per-workflow stat VReplicationFilteredRowsCopyStateSyncCount to keep track of the number of such sync responses sent.

Related Issue(s)

https://github.com/vitessio/vitess/issues/12937

Checklist

• [ ] "Backport to:" labels have been added if this change should be back-ported
• [X] Tests were added or are not required
• [ ] Did the new or modified tests pass consistently locally and on the CI
• [ ] Documentation was added or is not required