mz_storage/source/mysql/replication/
context.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use std::collections::{BTreeMap, BTreeSet};
11use std::pin::Pin;
12
13use mysql_async::BinlogStream;
14use timely::dataflow::operators::CapabilitySet;
15use timely::progress::{Antichain, Timestamp};
16use tracing::trace;
17
18use mz_mysql_util::Config;
19use mz_storage_types::sources::mysql::{GtidPartition, GtidState};
20
21use crate::metrics::source::mysql::MySqlSourceMetrics;
22use crate::source::RawSourceCreationConfig;
23use crate::source::mysql::{DataOutputs, MySqlTableName, RewindRequest, SourceOutputInfo};
24
25/// A container to hold various context information for the replication process, used when
26/// processing events from the binlog stream.
27pub(super) struct ReplContext<'a> {
28    pub(super) config: &'a RawSourceCreationConfig,
29    pub(super) connection_config: &'a Config,
30    pub(super) stream: Pin<&'a mut futures::stream::Peekable<BinlogStream>>,
31    pub(super) table_info: &'a BTreeMap<MySqlTableName, Vec<SourceOutputInfo>>,
32    pub(super) metrics: &'a MySqlSourceMetrics,
33    pub(super) data_outputs: &'a mut DataOutputs,
34    pub(super) data_cap_sets: &'a mut [CapabilitySet<GtidPartition>],
35    // Owned values:
36    pub(super) rewinds: BTreeMap<usize, RewindRequest>,
37    pub(super) errored_outputs: BTreeSet<usize>,
38    /// Whether ports without a pending rewind advance on their own instead of in lockstep.
39    concurrent_replication: bool,
40    data_upper: Antichain<GtidPartition>,
41    /// Whether `data_upper` moved since the last downgrade of the data capability sets.
42    data_upper_stale: bool,
43}
44
45impl<'a> ReplContext<'a> {
46    pub(super) fn new(
47        config: &'a RawSourceCreationConfig,
48        connection_config: &'a Config,
49        stream: Pin<&'a mut futures::stream::Peekable<BinlogStream>>,
50        table_info: &'a BTreeMap<MySqlTableName, Vec<SourceOutputInfo>>,
51        metrics: &'a MySqlSourceMetrics,
52        data_outputs: &'a mut DataOutputs,
53        data_cap_sets: &'a mut [CapabilitySet<GtidPartition>],
54        rewinds: BTreeMap<usize, RewindRequest>,
55        concurrent_replication: bool,
56    ) -> Self {
57        Self {
58            config,
59            connection_config,
60            stream,
61            table_info,
62            metrics,
63            data_outputs,
64            data_cap_sets,
65            rewinds,
66            errored_outputs: BTreeSet::new(),
67            concurrent_replication,
68            data_upper: Antichain::from_elem(GtidPartition::minimum()),
69            data_upper_stale: false,
70        }
71    }
72
73    /// Records `new_upper` as the data frontier. The data capability sets advance to it at the
74    /// next [`Self::downgrade_data_cap_sets`].
75    pub(super) fn advance_data_upper(&mut self, reason: &str, new_upper: Antichain<GtidPartition>) {
76        if new_upper == self.data_upper {
77            return;
78        }
79        let (id, worker_id) = (self.config.id, self.config.worker_id);
80
81        trace!(%id, "timely-{worker_id} [{reason}] advancing data frontier to {new_upper:?}");
82
83        self.metrics.gtid_txids.set(
84            new_upper
85                .iter()
86                .filter_map(|part| match part.timestamp() {
87                    GtidState::Absent => None,
88                    GtidState::Active(tx_id) => Some(tx_id.get()),
89                })
90                .sum(),
91        );
92        self.data_upper = new_upper;
93        self.data_upper_stale = true;
94    }
95
96    /// Drops the rewind requests that the recorded data frontier has passed, then advances the
97    /// data capability sets to it. A port with a pending rewind stays at the minimum. Without
98    /// concurrent replication every port stays there until no rewind is pending.
99    pub(super) fn downgrade_data_cap_sets(&mut self) {
100        if !std::mem::take(&mut self.data_upper_stale) {
101            return;
102        }
103        let (id, worker_id) = (self.config.id, self.config.worker_id);
104        let new_upper = &self.data_upper;
105
106        self.rewinds.retain(|_, req| {
107            // We need to retain the rewind requests whose snapshot_upper
108            // has at least one timestamp such that new_upper is less than
109            // that timestamp
110            let res = req.snapshot_upper.iter().any(|ts| new_upper.less_than(ts));
111            if !res {
112                trace!(%id, "timely-{worker_id} removing rewind request {req:?}");
113            }
114            res
115        });
116
117        // A pending rewind emits negated snapshot data at the minimum timestamp. Its port has to
118        // keep the minimum capability until the rewind resolves.
119        if self.concurrent_replication {
120            for (output_index, cap_set) in self.data_cap_sets.iter_mut().enumerate() {
121                if !self.rewinds.contains_key(&output_index) {
122                    cap_set.downgrade(new_upper.elements());
123                }
124            }
125        } else if self.rewinds.is_empty() {
126            for cap_set in self.data_cap_sets.iter_mut() {
127                cap_set.downgrade(new_upper.elements());
128            }
129        }
130    }
131}
mz_storage/source/mysql/replication/context.rs

mz_storage/source/mysql/replication/
context.rs
