flushDeltaBuffer flushes the accumulated changes in the delta buffer
(state *replicationState, conn *stdsql.Conn, tx *stdsql.Tx, reason delta.FlushReason)
| 1025 | |
| 1026 | // flushDeltaBuffer flushes the accumulated changes in the delta buffer |
| 1027 | func (r *LogicalReplicator) flushDeltaBuffer(state *replicationState, conn *stdsql.Conn, tx *stdsql.Tx, reason delta.FlushReason) error { |
| 1028 | defer func() { |
| 1029 | state.deltaBufSize = 0 |
| 1030 | }() |
| 1031 | |
| 1032 | _, err := state.deltas.Flush(state.replicaCtx, conn, tx, reason) |
| 1033 | return err |
| 1034 | } |
| 1035 | |
| 1036 | func (r *LogicalReplicator) append(state *replicationState, relationID uint32, tuple []*pglogrepl.TupleDataColumn, actionType, eventType binlog.RowEventType, onlyKeys bool) error { |
| 1037 | rel, ok := state.relations[relationID] |
no test coverage detected