mz_compute/logging/
reachability.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
10//! Logging dataflows for events generated by timely dataflow.
11
12use std::collections::BTreeMap;
13use std::convert::TryInto;
14use std::rc::Rc;
15use std::time::Duration;
16
17use columnar::Index;
18use mz_compute_client::logging::LoggingConfig;
19use mz_ore::cast::CastFrom;
20use mz_repr::{Datum, Diff, Row, Timestamp};
21use mz_row_spine::RowRowBuilder;
22use mz_timely_util::columnar::batcher;
23use mz_timely_util::columnar::builder::ColumnBuilder;
24use mz_timely_util::columnar::{Col2ValBatcher, Column, columnar_exchange};
25use mz_timely_util::replay::MzReplay;
26use timely::dataflow::Scope;
27use timely::dataflow::channels::pact::{ExchangeCore, Pipeline};
28use timely::dataflow::operators::Operator;
29use timely::dataflow::operators::generic::operator::empty;
30
31use crate::extensions::arrange::MzArrangeCore;
32use crate::logging::initialize::ReachabilityEvent;
33use crate::logging::{EventQueue, LogCollection, LogVariant, TimelyLog, consolidate_and_pack};
34use crate::typedefs::ConsolidateColumnBatcher;
35use crate::typedefs::RowRowSpine;
36
37/// The return type of [`construct`].
38pub(super) struct Return {
39    /// Collections to export.
40    pub collections: BTreeMap<LogVariant, LogCollection>,
41}
42
43/// Constructs the logging dataflow fragment for reachability logs.
44///
45/// Params
46/// * `scope`: The Timely scope hosting the log analysis dataflow.
47/// * `config`: Logging configuration
48/// * `event_queue`: The source to read log events from.
49pub(super) fn construct(
50    scope: Scope<'_, Timestamp>,
51    config: &LoggingConfig,
52    event_queue: EventQueue<Column<(Duration, ReachabilityEvent)>, 3>,
53) -> Return {
54    let collections = scope.scoped("timely reachability logging", move |scope| {
55        let enable_logging = config.enable_logging;
56        let interval_ms = std::cmp::max(1, config.interval.as_millis());
57        type UpdatesKey = (bool, usize, usize, usize, Timestamp);
58
59        type CB = ColumnBuilder<((UpdatesKey, ()), Timestamp, Diff)>;
60        let (logs, token) = if enable_logging {
61            event_queue.links.mz_replay(
62                scope,
63                "reachability logs",
64                config.interval,
65                event_queue.activator,
66            )
67        } else {
68            let token: Rc<dyn std::any::Any> = Rc::new(Box::new(()));
69            (empty(scope), token)
70        };
71        let logs = logs.unary::<CB, _, _, _>(Pipeline, "FlatMapReachability", move |_, _| {
72            move |input, output| {
73                input.for_each_stamp(|time, data| {
74                    output
75                        .session_with_builder(&time)
76                        .give_iterator(data.flat_map(|d| {
77                            d.borrow().into_index_iter().flat_map(
78                                move |(time, (operator_id, massaged))| {
79                                    let time_ms =
80                                        ((time.as_millis() / interval_ms) + 1) * interval_ms;
81                                    let time_ms: Timestamp = time_ms.try_into().expect("must fit");
82                                    massaged.into_iter().map(
83                                        move |(source, port, update_type, ts, diff)| {
84                                            let datum =
85                                                (update_type, operator_id, source, port, ts);
86                                            ((datum, ()), time_ms, diff)
87                                        },
88                                    )
89                                },
90                            )
91                        }));
92                });
93            }
94        });
95
96        // Restrict results by those logs that are meant to be active.
97        let logs_active = [LogVariant::Timely(TimelyLog::Reachability)];
98        let worker_id = scope.index();
99
100        let updates = consolidate_and_pack::<
101            ConsolidateColumnBatcher<UpdatesKey, _, _, _>,
102            ColumnBuilder<_>,
103            _,
104            _,
105            _,
106            _,
107        >(
108            logs,
109            TimelyLog::Reachability,
110            move |data, packer, session| {
111                for ((datum, ()), time, diff) in data.iter() {
112                    let (update_type, operator_id, source, port, ts) = datum;
113                    let update_type = if *update_type { "source" } else { "target" };
114                    let data = packer.pack_slice(&[
115                        Datum::UInt64(u64::cast_from(*operator_id)),
116                        Datum::UInt64(u64::cast_from(worker_id)),
117                        Datum::UInt64(u64::cast_from(*source)),
118                        Datum::UInt64(u64::cast_from(*port)),
119                        Datum::String(update_type),
120                        Datum::from(*ts),
121                    ]);
122                    session.give((data, time, diff));
123                }
124            },
125        );
126
127        let mut result = BTreeMap::new();
128        for variant in logs_active {
129            if config.index_logs.contains_key(&variant) {
130                let exchange = ExchangeCore::<ColumnBuilder<_>, _>::new_core(
131                    columnar_exchange::<Row, Row, Timestamp, Diff>,
132                );
133                let trace = updates
134                    .clone()
135                    .mz_arrange_core::<
136                        _,
137                        Col2ValBatcher<_, _, _, _, batcher::Chunker<_>, RowRowBuilder<_, _>>,
138                        RowRowSpine<_, _>,
139                    >(exchange, &format!("Arrange {variant:?}"))
140                    .trace;
141                let collection = LogCollection {
142                    trace,
143                    token: Rc::clone(&token),
144                };
145                result.insert(variant, collection);
146            }
147        }
148        result
149    });
150
151    Return { collections }
152}
mz_compute/logging/reachability.rs

mz_compute/logging/
reachability.rs
