Skip to main content

mz_catalog/builtin/
mz_introspection.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//! Built-in catalog items for the `mz_introspection` schema.
11
12use std::collections::BTreeMap;
13use std::sync::LazyLock;
14
15use mz_compute_client::logging::{ComputeLog, DifferentialLog, LogVariant, TimelyLog};
16use mz_pgrepr::oid;
17use mz_repr::adt::numeric::NumericMaxScale;
18use mz_repr::namespaces::MZ_INTROSPECTION_SCHEMA;
19use mz_repr::{RelationDesc, SemanticType, SqlScalarType};
20
21use super::{
22    BuiltinLog, BuiltinView, Cardinality, LinkProperties, Ontology, OntologyLink, PUBLIC_SELECT,
23};
24
25pub static MZ_DATAFLOW_OPERATORS_PER_WORKER: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
26    name: "mz_dataflow_operators_per_worker",
27    schema: MZ_INTROSPECTION_SCHEMA,
28    oid: oid::LOG_MZ_DATAFLOW_OPERATORS_PER_WORKER_OID,
29    variant: LogVariant::Timely(TimelyLog::Operates),
30    access: vec![PUBLIC_SELECT],
31    ontology: Some(Ontology {
32        entity_name: "dataflow_operator_per_worker",
33        description: "Timely dataflow operators present on a specific worker.",
34        links: &const { [] },
35        column_semantic_types: &[],
36    }),
37});
38
39pub static MZ_DATAFLOW_ADDRESSES_PER_WORKER: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
40    name: "mz_dataflow_addresses_per_worker",
41    schema: MZ_INTROSPECTION_SCHEMA,
42    oid: oid::LOG_MZ_DATAFLOW_ADDRESSES_PER_WORKER_OID,
43    variant: LogVariant::Timely(TimelyLog::Addresses),
44    access: vec![PUBLIC_SELECT],
45    ontology: Some(Ontology {
46        entity_name: "dataflow_address_per_worker",
47        description: "Scope address of each Timely operator per worker.",
48        links: &const {
49            [OntologyLink {
50                name: "address_of",
51                target: "dataflow_operator_per_worker",
52                properties: LinkProperties::fk_composite(
53                    "id",
54                    "id",
55                    Cardinality::ManyToOne,
56                    &[("worker_id", "worker_id")],
57                ),
58            }]
59        },
60        column_semantic_types: &[],
61    }),
62});
63
64pub static MZ_DATAFLOW_CHANNELS_PER_WORKER: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
65    name: "mz_dataflow_channels_per_worker",
66    schema: MZ_INTROSPECTION_SCHEMA,
67    oid: oid::LOG_MZ_DATAFLOW_CHANNELS_PER_WORKER_OID,
68    variant: LogVariant::Timely(TimelyLog::Channels),
69    access: vec![PUBLIC_SELECT],
70    ontology: Some(Ontology {
71        entity_name: "dataflow_channel_per_worker",
72        description: "Timely dataflow communication channels per worker.",
73        links: &const {
74            [
75                OntologyLink {
76                    name: "source_operator",
77                    target: "dataflow_operator_per_worker",
78                    properties: LinkProperties::fk_composite(
79                        "from_index",
80                        "id",
81                        Cardinality::ManyToOne,
82                        &[("worker_id", "worker_id")],
83                    ),
84                },
85                OntologyLink {
86                    name: "target_operator",
87                    target: "dataflow_operator_per_worker",
88                    properties: LinkProperties::fk_composite(
89                        "to_index",
90                        "id",
91                        Cardinality::ManyToOne,
92                        &[("worker_id", "worker_id")],
93                    ),
94                },
95            ]
96        },
97        column_semantic_types: &[],
98    }),
99});
100
101pub static MZ_SCHEDULING_ELAPSED_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
102    name: "mz_scheduling_elapsed_raw",
103    schema: MZ_INTROSPECTION_SCHEMA,
104    oid: oid::LOG_MZ_SCHEDULING_ELAPSED_RAW_OID,
105    variant: LogVariant::Timely(TimelyLog::Elapsed),
106    access: vec![PUBLIC_SELECT],
107    ontology: None,
108});
109
110pub static MZ_COMPUTE_OPERATOR_DURATIONS_HISTOGRAM_RAW: LazyLock<BuiltinLog> =
111    LazyLock::new(|| BuiltinLog {
112        name: "mz_compute_operator_durations_histogram_raw",
113        schema: MZ_INTROSPECTION_SCHEMA,
114        oid: oid::LOG_MZ_COMPUTE_OPERATOR_DURATIONS_HISTOGRAM_RAW_OID,
115        variant: LogVariant::Timely(TimelyLog::Histogram),
116        access: vec![PUBLIC_SELECT],
117        ontology: None,
118    });
119
120pub static MZ_SCHEDULING_PARKS_HISTOGRAM_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
121    name: "mz_scheduling_parks_histogram_raw",
122    schema: MZ_INTROSPECTION_SCHEMA,
123    oid: oid::LOG_MZ_SCHEDULING_PARKS_HISTOGRAM_RAW_OID,
124    variant: LogVariant::Timely(TimelyLog::Parks),
125    access: vec![PUBLIC_SELECT],
126    ontology: None,
127});
128
129pub static MZ_ARRANGEMENT_RECORDS_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
130    name: "mz_arrangement_records_raw",
131    schema: MZ_INTROSPECTION_SCHEMA,
132    oid: oid::LOG_MZ_ARRANGEMENT_RECORDS_RAW_OID,
133    variant: LogVariant::Differential(DifferentialLog::ArrangementRecords),
134    access: vec![PUBLIC_SELECT],
135    ontology: None,
136});
137
138pub static MZ_ARRANGEMENT_BATCHES_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
139    name: "mz_arrangement_batches_raw",
140    schema: MZ_INTROSPECTION_SCHEMA,
141    oid: oid::LOG_MZ_ARRANGEMENT_BATCHES_RAW_OID,
142    variant: LogVariant::Differential(DifferentialLog::ArrangementBatches),
143    access: vec![PUBLIC_SELECT],
144    ontology: None,
145});
146
147pub static MZ_ARRANGEMENT_SHARING_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
148    name: "mz_arrangement_sharing_raw",
149    schema: MZ_INTROSPECTION_SCHEMA,
150    oid: oid::LOG_MZ_ARRANGEMENT_SHARING_RAW_OID,
151    variant: LogVariant::Differential(DifferentialLog::Sharing),
152    access: vec![PUBLIC_SELECT],
153    ontology: None,
154});
155
156pub static MZ_ARRANGEMENT_BATCHER_RECORDS_RAW: LazyLock<BuiltinLog> =
157    LazyLock::new(|| BuiltinLog {
158        name: "mz_arrangement_batcher_records_raw",
159        schema: MZ_INTROSPECTION_SCHEMA,
160        oid: oid::LOG_MZ_ARRANGEMENT_BATCHER_RECORDS_RAW_OID,
161        variant: LogVariant::Differential(DifferentialLog::BatcherRecords),
162        access: vec![PUBLIC_SELECT],
163        ontology: None,
164    });
165
166pub static MZ_ARRANGEMENT_BATCHER_SIZE_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
167    name: "mz_arrangement_batcher_size_raw",
168    schema: MZ_INTROSPECTION_SCHEMA,
169    oid: oid::LOG_MZ_ARRANGEMENT_BATCHER_SIZE_RAW_OID,
170    variant: LogVariant::Differential(DifferentialLog::BatcherSize),
171    access: vec![PUBLIC_SELECT],
172    ontology: None,
173});
174
175pub static MZ_ARRANGEMENT_BATCHER_CAPACITY_RAW: LazyLock<BuiltinLog> =
176    LazyLock::new(|| BuiltinLog {
177        name: "mz_arrangement_batcher_capacity_raw",
178        schema: MZ_INTROSPECTION_SCHEMA,
179        oid: oid::LOG_MZ_ARRANGEMENT_BATCHER_CAPACITY_RAW_OID,
180        variant: LogVariant::Differential(DifferentialLog::BatcherCapacity),
181        access: vec![PUBLIC_SELECT],
182        ontology: None,
183    });
184
185pub static MZ_ARRANGEMENT_BATCHER_ALLOCATIONS_RAW: LazyLock<BuiltinLog> =
186    LazyLock::new(|| BuiltinLog {
187        name: "mz_arrangement_batcher_allocations_raw",
188        schema: MZ_INTROSPECTION_SCHEMA,
189        oid: oid::LOG_MZ_ARRANGEMENT_BATCHER_ALLOCATIONS_RAW_OID,
190        variant: LogVariant::Differential(DifferentialLog::BatcherAllocations),
191        access: vec![PUBLIC_SELECT],
192        ontology: None,
193    });
194
195pub static MZ_COMPUTE_EXPORTS_PER_WORKER: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
196    name: "mz_compute_exports_per_worker",
197    schema: MZ_INTROSPECTION_SCHEMA,
198    oid: oid::LOG_MZ_COMPUTE_EXPORTS_PER_WORKER_OID,
199    variant: LogVariant::Compute(ComputeLog::DataflowCurrent),
200    access: vec![PUBLIC_SELECT],
201    ontology: Some(Ontology {
202        entity_name: "compute_export_per_worker",
203        description: "Active compute exports (dataflows) present per worker.",
204        links: &const { [] },
205        column_semantic_types: &[("export_id", SemanticType::GlobalId)],
206    }),
207});
208
209pub static MZ_COMPUTE_DATAFLOW_GLOBAL_IDS_PER_WORKER: LazyLock<BuiltinLog> =
210    LazyLock::new(|| BuiltinLog {
211        name: "mz_compute_dataflow_global_ids_per_worker",
212        schema: MZ_INTROSPECTION_SCHEMA,
213        oid: oid::LOG_MZ_COMPUTE_DATAFLOW_GLOBAL_IDS_PER_WORKER_OID,
214        variant: LogVariant::Compute(ComputeLog::DataflowGlobal),
215        access: vec![PUBLIC_SELECT],
216        ontology: Some(Ontology {
217            entity_name: "dataflow_global_id_per_worker",
218            description: "Mapping from internal dataflow IDs to GlobalIds per worker.",
219            links: &const {
220                [OntologyLink {
221                    name: "global_id_of",
222                    target: "compute_export_per_worker",
223                    properties: LinkProperties::MapsTo {
224                        source_column: "global_id",
225                        target_column: "export_id",
226                        via: None,
227                        from_type: Some(SemanticType::GlobalId),
228                        to_type: Some(SemanticType::GlobalId),
229                        note: None,
230                    },
231                }]
232            },
233            column_semantic_types: &[("global_id", SemanticType::GlobalId)],
234        }),
235    });
236
237pub static MZ_CLUSTER_PROMETHEUS_METRICS: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
238    name: "mz_cluster_prometheus_metrics",
239    schema: MZ_INTROSPECTION_SCHEMA,
240    oid: oid::LOG_MZ_CLUSTER_PROMETHEUS_METRICS_OID,
241    variant: LogVariant::Compute(ComputeLog::PrometheusMetrics),
242    access: vec![PUBLIC_SELECT],
243    ontology: Some(Ontology {
244        entity_name: "cluster_prometheus_metric",
245        description: "Prometheus metrics gathered from the cluster's metrics registry.",
246        links: &const { [] },
247        column_semantic_types: &[],
248    }),
249});
250
251pub static MZ_COMPUTE_FRONTIERS_PER_WORKER: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
252    name: "mz_compute_frontiers_per_worker",
253    schema: MZ_INTROSPECTION_SCHEMA,
254    oid: oid::LOG_MZ_COMPUTE_FRONTIERS_PER_WORKER_OID,
255    variant: LogVariant::Compute(ComputeLog::FrontierCurrent),
256    access: vec![PUBLIC_SELECT],
257    ontology: Some(Ontology {
258        entity_name: "compute_frontier_per_worker",
259        description: "Per-worker output frontier timestamps for each compute export.",
260        links: &const {
261            [OntologyLink {
262                name: "frontier_of",
263                target: "compute_export_per_worker",
264                properties: LinkProperties::measures_composite(
265                    "export_id",
266                    "export_id",
267                    "time",
268                    &[("worker_id", "worker_id")],
269                ),
270            }]
271        },
272        column_semantic_types: &[
273            ("export_id", SemanticType::GlobalId),
274            ("time", SemanticType::MzTimestamp),
275        ],
276    }),
277});
278
279pub static MZ_COMPUTE_IMPORT_FRONTIERS_PER_WORKER: LazyLock<BuiltinLog> =
280    LazyLock::new(|| BuiltinLog {
281        name: "mz_compute_import_frontiers_per_worker",
282        schema: MZ_INTROSPECTION_SCHEMA,
283        oid: oid::LOG_MZ_COMPUTE_IMPORT_FRONTIERS_PER_WORKER_OID,
284        variant: LogVariant::Compute(ComputeLog::ImportFrontierCurrent),
285        access: vec![PUBLIC_SELECT],
286        ontology: Some(Ontology {
287            entity_name: "compute_import_frontier_per_worker",
288            description: "Per-worker input frontier timestamps for each compute dataflow.",
289            links: &const {
290                [OntologyLink {
291                    name: "import_frontier_of",
292                    target: "compute_export_per_worker",
293                    properties: LinkProperties::fk_composite(
294                        "export_id",
295                        "export_id",
296                        Cardinality::ManyToOne,
297                        &[("worker_id", "worker_id")],
298                    ),
299                }]
300            },
301            column_semantic_types: &[
302                ("export_id", SemanticType::GlobalId),
303                ("import_id", SemanticType::GlobalId),
304                ("time", SemanticType::MzTimestamp),
305            ],
306        }),
307    });
308
309pub static MZ_COMPUTE_ERROR_COUNTS_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
310    name: "mz_compute_error_counts_raw",
311    schema: MZ_INTROSPECTION_SCHEMA,
312    oid: oid::LOG_MZ_COMPUTE_ERROR_COUNTS_RAW_OID,
313    variant: LogVariant::Compute(ComputeLog::ErrorCount),
314    access: vec![PUBLIC_SELECT],
315    ontology: None,
316});
317
318pub static MZ_COMPUTE_HYDRATION_TIMES_PER_WORKER: LazyLock<BuiltinLog> =
319    LazyLock::new(|| BuiltinLog {
320        name: "mz_compute_hydration_times_per_worker",
321        schema: MZ_INTROSPECTION_SCHEMA,
322        oid: oid::LOG_MZ_COMPUTE_HYDRATION_TIMES_PER_WORKER_OID,
323        variant: LogVariant::Compute(ComputeLog::HydrationTime),
324        access: vec![PUBLIC_SELECT],
325        ontology: Some(Ontology {
326            entity_name: "hydration_time_per_worker",
327            description: "Hydration duration and lifecycle timestamps for each compute export \
328                          per worker.",
329            links: &const {
330                [OntologyLink {
331                    name: "hydration_time_of",
332                    target: "compute_export_per_worker",
333                    properties: LinkProperties::measures_composite(
334                        "export_id",
335                        "export_id",
336                        "time_ns",
337                        &[("worker_id", "worker_id")],
338                    ),
339                }]
340            },
341            column_semantic_types: &[("export_id", SemanticType::GlobalId)],
342        }),
343    });
344
345pub static MZ_COMPUTE_OPERATOR_HYDRATION_STATUSES_PER_WORKER: LazyLock<BuiltinLog> =
346    LazyLock::new(|| BuiltinLog {
347        name: "mz_compute_operator_hydration_statuses_per_worker",
348        schema: MZ_INTROSPECTION_SCHEMA,
349        oid: oid::LOG_MZ_COMPUTE_OPERATOR_HYDRATION_STATUSES_PER_WORKER_OID,
350        variant: LogVariant::Compute(ComputeLog::OperatorHydrationStatus),
351        access: vec![PUBLIC_SELECT],
352        ontology: Some(Ontology {
353            entity_name: "hydration_status_per_worker",
354            description: "Hydration state for each LIR node per worker.",
355            links: &const {
356                [OntologyLink {
357                    name: "hydration_of",
358                    target: "compute_export_per_worker",
359                    properties: LinkProperties::fk_composite(
360                        "export_id",
361                        "export_id",
362                        Cardinality::ManyToOne,
363                        &[("worker_id", "worker_id")],
364                    ),
365                }]
366            },
367            column_semantic_types: &[("export_id", SemanticType::GlobalId)],
368        }),
369    });
370
371pub static MZ_ACTIVE_PEEKS_PER_WORKER: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
372    name: "mz_active_peeks_per_worker",
373    schema: MZ_INTROSPECTION_SCHEMA,
374    oid: oid::LOG_MZ_ACTIVE_PEEKS_PER_WORKER_OID,
375    variant: LogVariant::Compute(ComputeLog::PeekCurrent),
376    access: vec![PUBLIC_SELECT],
377    ontology: Some(Ontology {
378        entity_name: "active_peek_per_worker",
379        description: "In-flight peek requests currently executing per worker.",
380        links: &const { [] },
381        column_semantic_types: &[
382            ("object_id", SemanticType::GlobalId),
383            ("time", SemanticType::MzTimestamp),
384        ],
385    }),
386});
387
388pub static MZ_COMPUTE_LIR_MAPPING_PER_WORKER: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
389    name: "mz_compute_lir_mapping_per_worker",
390    schema: MZ_INTROSPECTION_SCHEMA,
391    oid: oid::LOG_MZ_COMPUTE_LIR_MAPPING_PER_WORKER_OID,
392    variant: LogVariant::Compute(ComputeLog::LirMapping),
393    access: vec![PUBLIC_SELECT],
394    ontology: Some(Ontology {
395        entity_name: "lir_mapping_per_worker",
396        description: "Mapping from LIR node IDs to dataflow operator address ranges per worker.",
397        links: &const {
398            [OntologyLink {
399                name: "export_of",
400                target: "compute_export_per_worker",
401                properties: LinkProperties::fk_composite(
402                    "global_id",
403                    "export_id",
404                    Cardinality::ManyToOne,
405                    &[("worker_id", "worker_id")],
406                ),
407            }]
408        },
409        column_semantic_types: &[("global_id", SemanticType::GlobalId)],
410    }),
411});
412
413pub static MZ_PEEK_DURATIONS_HISTOGRAM_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
414    name: "mz_peek_durations_histogram_raw",
415    schema: MZ_INTROSPECTION_SCHEMA,
416    oid: oid::LOG_MZ_PEEK_DURATIONS_HISTOGRAM_RAW_OID,
417    variant: LogVariant::Compute(ComputeLog::PeekDuration),
418    access: vec![PUBLIC_SELECT],
419    ontology: None,
420});
421
422pub static MZ_ARRANGEMENT_HEAP_SIZE_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
423    name: "mz_arrangement_heap_size_raw",
424    schema: MZ_INTROSPECTION_SCHEMA,
425    oid: oid::LOG_MZ_ARRANGEMENT_HEAP_SIZE_RAW_OID,
426    variant: LogVariant::Compute(ComputeLog::ArrangementHeapSize),
427    access: vec![PUBLIC_SELECT],
428    ontology: None,
429});
430
431pub static MZ_ARRANGEMENT_HEAP_CAPACITY_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
432    name: "mz_arrangement_heap_capacity_raw",
433    schema: MZ_INTROSPECTION_SCHEMA,
434    oid: oid::LOG_MZ_ARRANGEMENT_HEAP_CAPACITY_RAW_OID,
435    variant: LogVariant::Compute(ComputeLog::ArrangementHeapCapacity),
436    access: vec![PUBLIC_SELECT],
437    ontology: None,
438});
439
440pub static MZ_ARRANGEMENT_HEAP_ALLOCATIONS_RAW: LazyLock<BuiltinLog> =
441    LazyLock::new(|| BuiltinLog {
442        name: "mz_arrangement_heap_allocations_raw",
443        schema: MZ_INTROSPECTION_SCHEMA,
444        oid: oid::LOG_MZ_ARRANGEMENT_HEAP_ALLOCATIONS_RAW_OID,
445        variant: LogVariant::Compute(ComputeLog::ArrangementHeapAllocations),
446        access: vec![PUBLIC_SELECT],
447        ontology: None,
448    });
449
450pub static MZ_MESSAGE_BATCH_COUNTS_RECEIVED_RAW: LazyLock<BuiltinLog> =
451    LazyLock::new(|| BuiltinLog {
452        name: "mz_message_batch_counts_received_raw",
453        schema: MZ_INTROSPECTION_SCHEMA,
454        oid: oid::LOG_MZ_MESSAGE_BATCH_COUNTS_RECEIVED_RAW_OID,
455        variant: LogVariant::Timely(TimelyLog::BatchesReceived),
456        access: vec![PUBLIC_SELECT],
457        ontology: None,
458    });
459
460pub static MZ_MESSAGE_BATCH_COUNTS_SENT_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
461    name: "mz_message_batch_counts_sent_raw",
462    schema: MZ_INTROSPECTION_SCHEMA,
463    oid: oid::LOG_MZ_MESSAGE_BATCH_COUNTS_SENT_RAW_OID,
464    variant: LogVariant::Timely(TimelyLog::BatchesSent),
465    access: vec![PUBLIC_SELECT],
466    ontology: None,
467});
468
469pub static MZ_MESSAGE_COUNTS_RECEIVED_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
470    name: "mz_message_counts_received_raw",
471    schema: MZ_INTROSPECTION_SCHEMA,
472    oid: oid::LOG_MZ_MESSAGE_COUNTS_RECEIVED_RAW_OID,
473    variant: LogVariant::Timely(TimelyLog::MessagesReceived),
474    access: vec![PUBLIC_SELECT],
475    ontology: None,
476});
477
478pub static MZ_MESSAGE_COUNTS_SENT_RAW: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
479    name: "mz_message_counts_sent_raw",
480    schema: MZ_INTROSPECTION_SCHEMA,
481    oid: oid::LOG_MZ_MESSAGE_COUNTS_SENT_RAW_OID,
482    variant: LogVariant::Timely(TimelyLog::MessagesSent),
483    access: vec![PUBLIC_SELECT],
484    ontology: None,
485});
486
487pub static MZ_DATAFLOW_OPERATOR_REACHABILITY_RAW: LazyLock<BuiltinLog> =
488    LazyLock::new(|| BuiltinLog {
489        name: "mz_dataflow_operator_reachability_raw",
490        schema: MZ_INTROSPECTION_SCHEMA,
491        oid: oid::LOG_MZ_DATAFLOW_OPERATOR_REACHABILITY_RAW_OID,
492        variant: LogVariant::Timely(TimelyLog::Reachability),
493        access: vec![PUBLIC_SELECT],
494        ontology: None,
495    });
496
497pub static MZ_DATAFLOWS_PER_WORKER: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
498    name: "mz_dataflows_per_worker",
499    schema: MZ_INTROSPECTION_SCHEMA,
500    oid: oid::VIEW_MZ_DATAFLOWS_PER_WORKER_OID,
501    desc: RelationDesc::builder()
502        .with_column("id", SqlScalarType::UInt64.nullable(true))
503        .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
504        .with_column("name", SqlScalarType::String.nullable(false))
505        .finish(),
506    column_comments: BTreeMap::new(),
507    sql: "SELECT
508    addrs.address[1] AS id,
509    ops.worker_id,
510    ops.name
511FROM
512    mz_introspection.mz_dataflow_addresses_per_worker addrs,
513    mz_introspection.mz_dataflow_operators_per_worker ops
514WHERE
515    addrs.id = ops.id AND
516    addrs.worker_id = ops.worker_id AND
517    mz_catalog.list_length(addrs.address) = 1",
518    access: vec![PUBLIC_SELECT],
519    ontology: None,
520});
521
522pub static MZ_DATAFLOWS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
523    name: "mz_dataflows",
524    schema: MZ_INTROSPECTION_SCHEMA,
525    oid: oid::VIEW_MZ_DATAFLOWS_OID,
526    desc: RelationDesc::builder()
527        .with_column("id", SqlScalarType::UInt64.nullable(true))
528        .with_column("name", SqlScalarType::String.nullable(false))
529        .finish(),
530    column_comments: BTreeMap::from_iter([
531        ("id", "The ID of the dataflow."),
532        ("name", "The internal name of the dataflow."),
533    ]),
534    sql: "
535SELECT id, name
536FROM mz_introspection.mz_dataflows_per_worker
537WHERE worker_id = 0::uint8",
538    access: vec![PUBLIC_SELECT],
539    ontology: Some(Ontology {
540        entity_name: "dataflow",
541        description: "Dataflow instances",
542        links: &const { [] },
543        column_semantic_types: &[],
544    }),
545});
546
547pub static MZ_DATAFLOW_ADDRESSES: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
548    name: "mz_dataflow_addresses",
549    schema: MZ_INTROSPECTION_SCHEMA,
550    oid: oid::VIEW_MZ_DATAFLOW_ADDRESSES_OID,
551    desc: RelationDesc::builder()
552        .with_column("id", SqlScalarType::UInt64.nullable(false))
553        .with_column(
554            "address",
555            SqlScalarType::List {
556                element_type: Box::new(SqlScalarType::UInt64),
557                custom_id: None,
558            }
559            .nullable(false),
560        )
561        .with_key(vec![0])
562        .finish(),
563    column_comments: BTreeMap::from_iter([
564        (
565            "id",
566            "The ID of the channel or operator. Corresponds to `mz_dataflow_channels.id` or `mz_dataflow_operators.id`.",
567        ),
568        (
569            "address",
570            "A list of scope-local indexes indicating the path from the root to this channel or operator.",
571        ),
572    ]),
573    sql: "
574SELECT id, address
575FROM mz_introspection.mz_dataflow_addresses_per_worker
576WHERE worker_id = 0::uint8",
577    access: vec![PUBLIC_SELECT],
578    ontology: Some(Ontology {
579        entity_name: "dataflow_address",
580        description: "Address (scope path) of dataflow operators",
581        links: &const {
582            [OntologyLink {
583                name: "address_of_operator",
584                target: "dataflow_operator",
585                properties: LinkProperties::fk("id", "id", Cardinality::OneToOne),
586            }]
587        },
588        column_semantic_types: &[],
589    }),
590});
591
592pub static MZ_DATAFLOW_CHANNELS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
593    name: "mz_dataflow_channels",
594    schema: MZ_INTROSPECTION_SCHEMA,
595    oid: oid::VIEW_MZ_DATAFLOW_CHANNELS_OID,
596    desc: RelationDesc::builder()
597        .with_column("id", SqlScalarType::UInt64.nullable(false))
598        .with_column("from_index", SqlScalarType::UInt64.nullable(false))
599        .with_column("from_port", SqlScalarType::UInt64.nullable(false))
600        .with_column("to_index", SqlScalarType::UInt64.nullable(false))
601        .with_column("to_port", SqlScalarType::UInt64.nullable(false))
602        .with_column("type", SqlScalarType::String.nullable(false))
603        .with_key(vec![0])
604        .finish(),
605    column_comments: BTreeMap::from_iter([
606        ("id", "The ID of the channel."),
607        (
608            "from_index",
609            "The scope-local index of the source operator. Corresponds to `mz_dataflow_addresses.address`.",
610        ),
611        ("from_port", "The source operator's output port."),
612        (
613            "to_index",
614            "The scope-local index of the target operator. Corresponds to `mz_dataflow_addresses.address`.",
615        ),
616        ("to_port", "The target operator's input port."),
617        ("type", "The container type of the channel."),
618    ]),
619    sql: "
620SELECT id, from_index, from_port, to_index, to_port, type
621FROM mz_introspection.mz_dataflow_channels_per_worker
622WHERE worker_id = 0::uint8",
623    access: vec![PUBLIC_SELECT],
624    ontology: Some(Ontology {
625        entity_name: "dataflow_channel",
626        description: "Communication channels between operators",
627        links: &const {
628            [OntologyLink {
629                name: "channel_in_dataflow",
630                target: "dataflow",
631                properties: LinkProperties::MapsTo {
632                    source_column: "from_index",
633                    target_column: "id",
634                    via: Some("mz_introspection.mz_dataflow_operator_dataflows"),
635                    from_type: None,
636                    to_type: None,
637                    note: Some(
638                        "Channels do not have a direct dataflow_id. Use mz_dataflow_addresses to find the parent scope, then correlate with mz_dataflow_operator_dataflows.",
639                    ),
640                },
641            }]
642        },
643        column_semantic_types: &[],
644    }),
645});
646
647pub static MZ_DATAFLOW_OPERATORS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
648    name: "mz_dataflow_operators",
649    schema: MZ_INTROSPECTION_SCHEMA,
650    oid: oid::VIEW_MZ_DATAFLOW_OPERATORS_OID,
651    desc: RelationDesc::builder()
652        .with_column("id", SqlScalarType::UInt64.nullable(false))
653        .with_column("name", SqlScalarType::String.nullable(false))
654        .with_key(vec![0])
655        .finish(),
656    column_comments: BTreeMap::from_iter([
657        ("id", "The ID of the operator."),
658        ("name", "The internal name of the operator."),
659    ]),
660    sql: "
661SELECT id, name
662FROM mz_introspection.mz_dataflow_operators_per_worker
663WHERE worker_id = 0::uint8",
664    access: vec![PUBLIC_SELECT],
665    ontology: Some(Ontology {
666        entity_name: "dataflow_operator",
667        description: "Operators within dataflows",
668        links: &const { [] },
669        column_semantic_types: &[],
670    }),
671});
672
673pub static MZ_DATAFLOW_GLOBAL_IDS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
674    name: "mz_dataflow_global_ids",
675    schema: MZ_INTROSPECTION_SCHEMA,
676    oid: oid::VIEW_MZ_DATAFLOW_GLOBAL_IDS_OID,
677    desc: RelationDesc::builder()
678        .with_column("id", SqlScalarType::UInt64.nullable(false))
679        .with_column("global_id", SqlScalarType::String.nullable(false))
680        .with_key(vec![0, 1])
681        .finish(),
682    column_comments: BTreeMap::from_iter([
683        ("id", "The dataflow ID."),
684        ("global_id", "A global ID associated with that dataflow."),
685    ]),
686    sql: "
687SELECT id, global_id
688FROM mz_introspection.mz_compute_dataflow_global_ids_per_worker
689WHERE worker_id = 0::uint8",
690    access: vec![PUBLIC_SELECT],
691    ontology: None,
692});
693
694pub static MZ_MAPPABLE_OBJECTS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
695    name: "mz_mappable_objects",
696    schema: MZ_INTROSPECTION_SCHEMA,
697    oid: oid::VIEW_MZ_MAPPABLE_OBJECTS_OID,
698    desc: RelationDesc::builder()
699        .with_column("name", SqlScalarType::String.nullable(false))
700        .with_column("global_id", SqlScalarType::String.nullable(false))
701        .finish(),
702    column_comments: BTreeMap::from_iter([
703        (
704            "name",
705            "The name of the object. This name is unquoted, and you might need to call `quote_ident` if you want to reference the name shown here.",
706        ),
707        ("global_id", "The global ID of the object."),
708    ]),
709    sql: "
710SELECT COALESCE(md.name || '.', '') || ms.name || '.' || mo.name AS name, mgi.global_id AS global_id
711FROM      mz_catalog.mz_objects mo
712          JOIN mz_introspection.mz_compute_exports mce ON (mo.id = mce.export_id)
713          JOIN mz_catalog.mz_schemas ms ON (mo.schema_id = ms.id)
714          JOIN mz_introspection.mz_dataflow_global_ids mgi ON (mce.dataflow_id = mgi.id)
715     LEFT JOIN mz_catalog.mz_databases md ON (ms.database_id = md.id);",
716    access: vec![PUBLIC_SELECT],
717    ontology: Some(Ontology {
718        entity_name: "mappable_object",
719        description: "Objects that can be mapped to dataflow operators",
720        links: &const { [] },
721        column_semantic_types: &[("global_id", SemanticType::GlobalId)],
722    }),
723});
724
725pub static MZ_LIR_MAPPING: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
726    name: "mz_lir_mapping",
727    schema: MZ_INTROSPECTION_SCHEMA,
728    oid: oid::VIEW_MZ_LIR_MAPPING_OID,
729    desc: RelationDesc::builder()
730        .with_column("global_id", SqlScalarType::String.nullable(false))
731        .with_column("lir_id", SqlScalarType::UInt64.nullable(false))
732        .with_column("operator", SqlScalarType::String.nullable(false))
733        .with_column("parent_lir_id", SqlScalarType::UInt64.nullable(true))
734        .with_column("nesting", SqlScalarType::UInt16.nullable(false))
735        .with_column("operator_id_start", SqlScalarType::UInt64.nullable(false))
736        .with_column("operator_id_end", SqlScalarType::UInt64.nullable(false))
737        .with_key(vec![0, 1])
738        .finish(),
739    column_comments: BTreeMap::from_iter([
740        ("global_id", "The global ID."),
741        ("lir_id", "The LIR node ID."),
742        (
743            "operator",
744            "The LIR operator, in the format `OperatorName INPUTS [OPTIONS]`.",
745        ),
746        (
747            "parent_lir_id",
748            "The parent of this LIR node. May be `NULL`.",
749        ),
750        ("nesting", "The nesting level of this LIR node."),
751        (
752            "operator_id_start",
753            "The first dataflow operator ID implementing this LIR operator (inclusive).",
754        ),
755        (
756            "operator_id_end",
757            "The first dataflow operator ID _after_ this LIR operator (exclusive).",
758        ),
759    ]),
760    sql: "
761SELECT global_id, lir_id, operator, parent_lir_id, nesting, operator_id_start, operator_id_end
762FROM mz_introspection.mz_compute_lir_mapping_per_worker
763WHERE worker_id = 0::uint8",
764    access: vec![PUBLIC_SELECT],
765    ontology: Some(Ontology {
766        entity_name: "lir_mapping",
767        description: "LIR (low-level IR) to dataflow operator mapping",
768        links: &const { [] },
769        column_semantic_types: &[("global_id", SemanticType::GlobalId)],
770    }),
771});
772
773pub static MZ_DATAFLOW_OPERATOR_DATAFLOWS_PER_WORKER: LazyLock<BuiltinView> =
774    LazyLock::new(|| BuiltinView {
775        name: "mz_dataflow_operator_dataflows_per_worker",
776        schema: MZ_INTROSPECTION_SCHEMA,
777        oid: oid::VIEW_MZ_DATAFLOW_OPERATOR_DATAFLOWS_PER_WORKER_OID,
778        desc: RelationDesc::builder()
779            .with_column("id", SqlScalarType::UInt64.nullable(false))
780            .with_column("name", SqlScalarType::String.nullable(false))
781            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
782            .with_column("dataflow_id", SqlScalarType::UInt64.nullable(false))
783            .with_column("dataflow_name", SqlScalarType::String.nullable(false))
784            .finish(),
785        column_comments: BTreeMap::new(),
786        sql: "SELECT
787    ops.id,
788    ops.name,
789    ops.worker_id,
790    dfs.id as dataflow_id,
791    dfs.name as dataflow_name
792FROM
793    mz_introspection.mz_dataflow_operators_per_worker ops,
794    mz_introspection.mz_dataflow_addresses_per_worker addrs,
795    mz_introspection.mz_dataflows_per_worker dfs
796WHERE
797    ops.id = addrs.id AND
798    ops.worker_id = addrs.worker_id AND
799    dfs.id = addrs.address[1] AND
800    dfs.worker_id = addrs.worker_id",
801        access: vec![PUBLIC_SELECT],
802        ontology: None,
803    });
804
805pub static MZ_DATAFLOW_OPERATOR_DATAFLOWS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
806    name: "mz_dataflow_operator_dataflows",
807    schema: MZ_INTROSPECTION_SCHEMA,
808    oid: oid::VIEW_MZ_DATAFLOW_OPERATOR_DATAFLOWS_OID,
809    desc: RelationDesc::builder()
810        .with_column("id", SqlScalarType::UInt64.nullable(false))
811        .with_column("name", SqlScalarType::String.nullable(false))
812        .with_column("dataflow_id", SqlScalarType::UInt64.nullable(false))
813        .with_column("dataflow_name", SqlScalarType::String.nullable(false))
814        .finish(),
815    column_comments: BTreeMap::from_iter([
816        (
817            "id",
818            "The ID of the operator. Corresponds to `mz_dataflow_operators.id`.",
819        ),
820        ("name", "The internal name of the operator."),
821        (
822            "dataflow_id",
823            "The ID of the dataflow hosting the operator. Corresponds to `mz_dataflows.id`.",
824        ),
825        (
826            "dataflow_name",
827            "The internal name of the dataflow hosting the operator.",
828        ),
829    ]),
830    sql: "
831SELECT id, name, dataflow_id, dataflow_name
832FROM mz_introspection.mz_dataflow_operator_dataflows_per_worker
833WHERE worker_id = 0::uint8",
834    access: vec![PUBLIC_SELECT],
835    ontology: Some(Ontology {
836        entity_name: "dataflow_operator_dataflow",
837        description: "Mapping of operators to their parent dataflow",
838        links: &const {
839            [OntologyLink {
840                name: "operator_in_dataflow",
841                target: "dataflow",
842                properties: LinkProperties::fk("dataflow_id", "id", Cardinality::ManyToOne),
843            }]
844        },
845        column_semantic_types: &[],
846    }),
847});
848
849pub static MZ_COMPUTE_EXPORTS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
850    name: "mz_compute_exports",
851    schema: MZ_INTROSPECTION_SCHEMA,
852    oid: oid::VIEW_MZ_COMPUTE_EXPORTS_OID,
853    desc: RelationDesc::builder()
854        .with_column("export_id", SqlScalarType::String.nullable(false))
855        .with_column("dataflow_id", SqlScalarType::UInt64.nullable(false))
856        .with_key(vec![0])
857        .finish(),
858    column_comments: BTreeMap::from_iter([
859        (
860            "export_id",
861            "The ID of the index, materialized view, or subscription exported by the dataflow. Corresponds to `mz_catalog.mz_indexes.id`, `mz_catalog.mz_materialized_views.id`, or `mz_internal.mz_subscriptions.id`.",
862        ),
863        (
864            "dataflow_id",
865            "The ID of the dataflow. Corresponds to `mz_dataflows.id`.",
866        ),
867    ]),
868    sql: "
869SELECT export_id, dataflow_id
870FROM mz_introspection.mz_compute_exports_per_worker
871WHERE worker_id = 0::uint8",
872    access: vec![PUBLIC_SELECT],
873    ontology: Some(Ontology {
874        entity_name: "compute_export",
875        description: "Compute exports (maintained collections)",
876        links: &const {
877            [
878                OntologyLink {
879                    name: "export_of",
880                    target: "object",
881                    properties: LinkProperties::fk_mapped(
882                        "export_id",
883                        "id",
884                        Cardinality::ManyToOne,
885                        mz_repr::SemanticType::GlobalId,
886                        "mz_internal.mz_object_global_ids",
887                    ),
888                },
889                OntologyLink {
890                    name: "introspection_uses_global_id",
891                    target: "object_global_id",
892                    properties: LinkProperties::MapsTo {
893                        source_column: "export_id",
894                        target_column: "global_id",
895                        via: None,
896                        from_type: None,
897                        to_type: None,
898                        note: Some(
899                            "mz_introspection tables use GlobalId. To join with mz_catalog tables (which use CatalogItemId), go through mz_internal.mz_object_global_ids.",
900                        ),
901                    },
902                },
903            ]
904        },
905        column_semantic_types: &[("export_id", SemanticType::GlobalId)],
906    }),
907});
908
909pub static MZ_COMPUTE_FRONTIERS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
910    name: "mz_compute_frontiers",
911    schema: MZ_INTROSPECTION_SCHEMA,
912    oid: oid::VIEW_MZ_COMPUTE_FRONTIERS_OID,
913    desc: RelationDesc::builder()
914        .with_column("export_id", SqlScalarType::String.nullable(false))
915        .with_column("time", SqlScalarType::MzTimestamp.nullable(false))
916        .with_key(vec![0])
917        .finish(),
918    column_comments: BTreeMap::from_iter([
919        (
920            "export_id",
921            "The ID of the dataflow export. Corresponds to `mz_compute_exports.export_id`.",
922        ),
923        (
924            "time",
925            "The next timestamp at which the dataflow output may change.",
926        ),
927    ]),
928    sql: "SELECT
929    export_id, pg_catalog.min(time) AS time
930FROM mz_introspection.mz_compute_frontiers_per_worker
931GROUP BY export_id",
932    access: vec![PUBLIC_SELECT],
933    ontology: Some(Ontology {
934        entity_name: "compute_frontier",
935        description: "Per-replica compute frontiers",
936        links: &const {
937            [OntologyLink {
938                name: "compute_frontier_of",
939                target: "object",
940                properties: LinkProperties::fk_mapped(
941                    "export_id",
942                    "id",
943                    Cardinality::ManyToOne,
944                    mz_repr::SemanticType::GlobalId,
945                    "mz_internal.mz_object_global_ids",
946                ),
947            }]
948        },
949        column_semantic_types: &const {
950            [
951                ("export_id", SemanticType::GlobalId),
952                ("time", SemanticType::MzTimestamp),
953            ]
954        },
955    }),
956});
957
958pub static MZ_DATAFLOW_CHANNEL_OPERATORS_PER_WORKER: LazyLock<BuiltinView> =
959    LazyLock::new(|| BuiltinView {
960        name: "mz_dataflow_channel_operators_per_worker",
961        schema: MZ_INTROSPECTION_SCHEMA,
962        oid: oid::VIEW_MZ_DATAFLOW_CHANNEL_OPERATORS_PER_WORKER_OID,
963        desc: RelationDesc::builder()
964            .with_column("id", SqlScalarType::UInt64.nullable(false))
965            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
966            .with_column("from_operator_id", SqlScalarType::UInt64.nullable(true))
967            .with_column(
968                "from_operator_address",
969                SqlScalarType::List {
970                    element_type: Box::new(SqlScalarType::UInt64),
971                    custom_id: None,
972                }
973                .nullable(false),
974            )
975            .with_column("to_operator_id", SqlScalarType::UInt64.nullable(true))
976            .with_column(
977                "to_operator_address",
978                SqlScalarType::List {
979                    element_type: Box::new(SqlScalarType::UInt64),
980                    custom_id: None,
981                }
982                .nullable(false),
983            )
984            .with_column("type", SqlScalarType::String.nullable(false))
985            .finish(),
986        column_comments: BTreeMap::new(),
987        sql: "
988WITH
989channel_addresses(id, worker_id, address, from_index, to_index, type) AS (
990     SELECT id, worker_id, address, from_index, to_index, type
991     FROM mz_introspection.mz_dataflow_channels_per_worker mdc
992     INNER JOIN mz_introspection.mz_dataflow_addresses_per_worker mda
993     USING (id, worker_id)
994),
995channel_operator_addresses(id, worker_id, from_address, to_address, type) AS (
996     SELECT id, worker_id,
997            address || from_index AS from_address,
998            address || to_index AS to_address,
999            type
1000     FROM channel_addresses
1001),
1002operator_addresses(id, worker_id, address) AS (
1003     SELECT id, worker_id, address
1004     FROM mz_introspection.mz_dataflow_addresses_per_worker mda
1005     INNER JOIN mz_introspection.mz_dataflow_operators_per_worker mdo
1006     USING (id, worker_id)
1007)
1008SELECT coa.id,
1009       coa.worker_id,
1010       from_ops.id AS from_operator_id,
1011       coa.from_address AS from_operator_address,
1012       to_ops.id AS to_operator_id,
1013       coa.to_address AS to_operator_address,
1014       coa.type
1015FROM channel_operator_addresses coa
1016     LEFT OUTER JOIN operator_addresses from_ops
1017          ON coa.from_address = from_ops.address AND
1018             coa.worker_id = from_ops.worker_id
1019     LEFT OUTER JOIN operator_addresses to_ops
1020          ON coa.to_address = to_ops.address AND
1021             coa.worker_id = to_ops.worker_id
1022",
1023        access: vec![PUBLIC_SELECT],
1024        ontology: None,
1025    });
1026
1027pub static MZ_DATAFLOW_CHANNEL_OPERATORS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
1028    name: "mz_dataflow_channel_operators",
1029    schema: MZ_INTROSPECTION_SCHEMA,
1030    oid: oid::VIEW_MZ_DATAFLOW_CHANNEL_OPERATORS_OID,
1031    desc: RelationDesc::builder()
1032        .with_column("id", SqlScalarType::UInt64.nullable(false))
1033        .with_column("from_operator_id", SqlScalarType::UInt64.nullable(true))
1034        .with_column(
1035            "from_operator_address",
1036            SqlScalarType::List {
1037                element_type: Box::new(SqlScalarType::UInt64),
1038                custom_id: None,
1039            }
1040            .nullable(false),
1041        )
1042        .with_column("to_operator_id", SqlScalarType::UInt64.nullable(true))
1043        .with_column(
1044            "to_operator_address",
1045            SqlScalarType::List {
1046                element_type: Box::new(SqlScalarType::UInt64),
1047                custom_id: None,
1048            }
1049            .nullable(false),
1050        )
1051        .with_column("type", SqlScalarType::String.nullable(false))
1052        .finish(),
1053    column_comments: BTreeMap::from_iter([
1054        (
1055            "id",
1056            "The ID of the channel. Corresponds to `mz_dataflow_channels.id`.",
1057        ),
1058        (
1059            "from_operator_id",
1060            "The ID of the source of the channel. Corresponds to `mz_dataflow_operators.id`.",
1061        ),
1062        (
1063            "from_operator_address",
1064            "The address of the source of the channel. Corresponds to `mz_dataflow_addresses.address`.",
1065        ),
1066        (
1067            "to_operator_id",
1068            "The ID of the target of the channel. Corresponds to `mz_dataflow_operators.id`.",
1069        ),
1070        (
1071            "to_operator_address",
1072            "The address of the target of the channel. Corresponds to `mz_dataflow_addresses.address`.",
1073        ),
1074        ("type", "The container type of the channel."),
1075    ]),
1076    sql: "
1077SELECT id, from_operator_id, from_operator_address, to_operator_id, to_operator_address, type
1078FROM mz_introspection.mz_dataflow_channel_operators_per_worker
1079WHERE worker_id = 0::uint8",
1080    access: vec![PUBLIC_SELECT],
1081    ontology: None,
1082});
1083
1084pub static MZ_COMPUTE_IMPORT_FRONTIERS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
1085    name: "mz_compute_import_frontiers",
1086    schema: MZ_INTROSPECTION_SCHEMA,
1087    oid: oid::VIEW_MZ_COMPUTE_IMPORT_FRONTIERS_OID,
1088    desc: RelationDesc::builder()
1089        .with_column("export_id", SqlScalarType::String.nullable(false))
1090        .with_column("import_id", SqlScalarType::String.nullable(false))
1091        .with_column("time", SqlScalarType::MzTimestamp.nullable(false))
1092        .with_key(vec![0, 1])
1093        .finish(),
1094    column_comments: BTreeMap::from_iter([
1095        (
1096            "export_id",
1097            "The ID of the dataflow export. Corresponds to `mz_compute_exports.export_id`.",
1098        ),
1099        (
1100            "import_id",
1101            "The ID of the dataflow import. Corresponds to `mz_catalog.mz_sources.id` or `mz_catalog.mz_tables.id` or `mz_compute_exports.export_id`.",
1102        ),
1103        (
1104            "time",
1105            "The next timestamp at which the dataflow input may change.",
1106        ),
1107    ]),
1108    sql: "SELECT
1109    export_id, import_id, pg_catalog.min(time) AS time
1110FROM mz_introspection.mz_compute_import_frontiers_per_worker
1111GROUP BY export_id, import_id",
1112    access: vec![PUBLIC_SELECT],
1113    ontology: Some(Ontology {
1114        entity_name: "compute_import_frontier",
1115        description: "Import frontiers for compute dependencies",
1116        links: &const {
1117            [OntologyLink {
1118                name: "compute_import_frontier_of",
1119                target: "object",
1120                properties: LinkProperties::fk_mapped(
1121                    "export_id",
1122                    "id",
1123                    Cardinality::ManyToOne,
1124                    mz_repr::SemanticType::GlobalId,
1125                    "mz_internal.mz_object_global_ids",
1126                ),
1127            }]
1128        },
1129        column_semantic_types: &const {
1130            [
1131                ("export_id", SemanticType::GlobalId),
1132                ("import_id", SemanticType::GlobalId),
1133                ("time", SemanticType::MzTimestamp),
1134            ]
1135        },
1136    }),
1137});
1138
1139pub static MZ_RECORDS_PER_DATAFLOW_OPERATOR_PER_WORKER: LazyLock<BuiltinView> =
1140    LazyLock::new(|| BuiltinView {
1141        name: "mz_records_per_dataflow_operator_per_worker",
1142        schema: MZ_INTROSPECTION_SCHEMA,
1143        oid: oid::VIEW_MZ_RECORDS_PER_DATAFLOW_OPERATOR_PER_WORKER_OID,
1144        desc: RelationDesc::builder()
1145            .with_column("id", SqlScalarType::UInt64.nullable(false))
1146            .with_column("name", SqlScalarType::String.nullable(false))
1147            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
1148            .with_column("dataflow_id", SqlScalarType::UInt64.nullable(false))
1149            .with_column("records", SqlScalarType::Int64.nullable(true))
1150            .with_column("batches", SqlScalarType::Int64.nullable(true))
1151            .with_column("size", SqlScalarType::Int64.nullable(true))
1152            .with_column("capacity", SqlScalarType::Int64.nullable(true))
1153            .with_column("allocations", SqlScalarType::Int64.nullable(true))
1154            .finish(),
1155        column_comments: BTreeMap::new(),
1156        sql: "
1157SELECT
1158    dod.id,
1159    dod.name,
1160    dod.worker_id,
1161    dod.dataflow_id,
1162    ar_size.records AS records,
1163    ar_size.batches AS batches,
1164    ar_size.size AS size,
1165    ar_size.capacity AS capacity,
1166    ar_size.allocations AS allocations
1167FROM
1168    mz_introspection.mz_dataflow_operator_dataflows_per_worker dod
1169    LEFT OUTER JOIN mz_introspection.mz_arrangement_sizes_per_worker ar_size ON
1170        dod.id = ar_size.operator_id AND
1171        dod.worker_id = ar_size.worker_id",
1172        access: vec![PUBLIC_SELECT],
1173        ontology: None,
1174    });
1175
1176pub static MZ_RECORDS_PER_DATAFLOW_OPERATOR: LazyLock<BuiltinView> =
1177    LazyLock::new(|| BuiltinView {
1178        name: "mz_records_per_dataflow_operator",
1179        schema: MZ_INTROSPECTION_SCHEMA,
1180        oid: oid::VIEW_MZ_RECORDS_PER_DATAFLOW_OPERATOR_OID,
1181        desc: RelationDesc::builder()
1182            .with_column("id", SqlScalarType::UInt64.nullable(false))
1183            .with_column("name", SqlScalarType::String.nullable(false))
1184            .with_column("dataflow_id", SqlScalarType::UInt64.nullable(false))
1185            .with_column("records", SqlScalarType::Int64.nullable(true))
1186            .with_column("batches", SqlScalarType::Int64.nullable(true))
1187            .with_column("size", SqlScalarType::Int64.nullable(true))
1188            .with_column("capacity", SqlScalarType::Int64.nullable(true))
1189            .with_column("allocations", SqlScalarType::Int64.nullable(true))
1190            .with_key(vec![0, 1, 2])
1191            .finish(),
1192        column_comments: BTreeMap::from_iter([
1193            (
1194                "id",
1195                "The ID of the operator. Corresponds to `mz_dataflow_operators.id`.",
1196            ),
1197            ("name", "The internal name of the operator."),
1198            (
1199                "dataflow_id",
1200                "The ID of the dataflow. Corresponds to `mz_dataflows.id`.",
1201            ),
1202            ("records", "The number of records in the operator."),
1203            ("batches", "The number of batches in the dataflow."),
1204            ("size", "The utilized size in bytes of the arrangement."),
1205            (
1206                "capacity",
1207                "The capacity in bytes of the arrangement. Can be larger than the size.",
1208            ),
1209            (
1210                "allocations",
1211                "The number of separate memory allocations backing the arrangement.",
1212            ),
1213        ]),
1214        sql: "
1215SELECT
1216    id,
1217    name,
1218    dataflow_id,
1219    SUM(records)::int8 AS records,
1220    SUM(batches)::int8 AS batches,
1221    SUM(size)::int8 AS size,
1222    SUM(capacity)::int8 AS capacity,
1223    SUM(allocations)::int8 AS allocations
1224FROM mz_introspection.mz_records_per_dataflow_operator_per_worker
1225GROUP BY id, name, dataflow_id",
1226        access: vec![PUBLIC_SELECT],
1227        ontology: None,
1228    });
1229
1230pub static MZ_RECORDS_PER_DATAFLOW_PER_WORKER: LazyLock<BuiltinView> =
1231    LazyLock::new(|| BuiltinView {
1232        name: "mz_records_per_dataflow_per_worker",
1233        schema: MZ_INTROSPECTION_SCHEMA,
1234        oid: oid::VIEW_MZ_RECORDS_PER_DATAFLOW_PER_WORKER_OID,
1235        desc: RelationDesc::builder()
1236            .with_column("id", SqlScalarType::UInt64.nullable(false))
1237            .with_column("name", SqlScalarType::String.nullable(false))
1238            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
1239            .with_column("records", SqlScalarType::Int64.nullable(true))
1240            .with_column("batches", SqlScalarType::Int64.nullable(true))
1241            .with_column("size", SqlScalarType::Int64.nullable(true))
1242            .with_column("capacity", SqlScalarType::Int64.nullable(true))
1243            .with_column("allocations", SqlScalarType::Int64.nullable(true))
1244            .with_key(vec![0, 1, 2])
1245            .finish(),
1246        column_comments: BTreeMap::new(),
1247        sql: "
1248SELECT
1249    rdo.dataflow_id as id,
1250    dfs.name,
1251    rdo.worker_id,
1252    SUM(rdo.records)::int8 as records,
1253    SUM(rdo.batches)::int8 as batches,
1254    SUM(rdo.size)::int8 as size,
1255    SUM(rdo.capacity)::int8 as capacity,
1256    SUM(rdo.allocations)::int8 as allocations
1257FROM
1258    mz_introspection.mz_records_per_dataflow_operator_per_worker rdo,
1259    mz_introspection.mz_dataflows_per_worker dfs
1260WHERE
1261    rdo.dataflow_id = dfs.id AND
1262    rdo.worker_id = dfs.worker_id
1263GROUP BY
1264    rdo.dataflow_id,
1265    dfs.name,
1266    rdo.worker_id",
1267        access: vec![PUBLIC_SELECT],
1268        ontology: None,
1269    });
1270
1271pub static MZ_RECORDS_PER_DATAFLOW: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
1272    name: "mz_records_per_dataflow",
1273    schema: MZ_INTROSPECTION_SCHEMA,
1274    oid: oid::VIEW_MZ_RECORDS_PER_DATAFLOW_OID,
1275    desc: RelationDesc::builder()
1276        .with_column("id", SqlScalarType::UInt64.nullable(false))
1277        .with_column("name", SqlScalarType::String.nullable(false))
1278        .with_column("records", SqlScalarType::Int64.nullable(true))
1279        .with_column("batches", SqlScalarType::Int64.nullable(true))
1280        .with_column("size", SqlScalarType::Int64.nullable(true))
1281        .with_column("capacity", SqlScalarType::Int64.nullable(true))
1282        .with_column("allocations", SqlScalarType::Int64.nullable(true))
1283        .with_key(vec![0, 1])
1284        .finish(),
1285    column_comments: BTreeMap::from_iter([
1286        (
1287            "id",
1288            "The ID of the dataflow. Corresponds to `mz_dataflows.id`.",
1289        ),
1290        ("name", "The internal name of the dataflow."),
1291        ("records", "The number of records in the dataflow."),
1292        ("batches", "The number of batches in the dataflow."),
1293        ("size", "The utilized size in bytes of the arrangements."),
1294        (
1295            "capacity",
1296            "The capacity in bytes of the arrangements. Can be larger than the size.",
1297        ),
1298        (
1299            "allocations",
1300            "The number of separate memory allocations backing the arrangements.",
1301        ),
1302    ]),
1303    sql: "
1304SELECT
1305    id,
1306    name,
1307    SUM(records)::int8 as records,
1308    SUM(batches)::int8 as batches,
1309    SUM(size)::int8 as size,
1310    SUM(capacity)::int8 as capacity,
1311    SUM(allocations)::int8 as allocations
1312FROM
1313    mz_introspection.mz_records_per_dataflow_per_worker
1314GROUP BY
1315    id,
1316    name",
1317    access: vec![PUBLIC_SELECT],
1318    ontology: Some(Ontology {
1319        entity_name: "records_per_dataflow",
1320        description: "Record counts aggregated per dataflow",
1321        links: &const {
1322            [OntologyLink {
1323                name: "details_of",
1324                target: "dataflow",
1325                properties: LinkProperties::fk("id", "id", Cardinality::OneToOne),
1326            }]
1327        },
1328        column_semantic_types: &[],
1329    }),
1330});
1331
1332pub static MZ_PEEK_DURATIONS_HISTOGRAM_PER_WORKER: LazyLock<BuiltinView> =
1333    LazyLock::new(|| BuiltinView {
1334        name: "mz_peek_durations_histogram_per_worker",
1335        schema: MZ_INTROSPECTION_SCHEMA,
1336        oid: oid::VIEW_MZ_PEEK_DURATIONS_HISTOGRAM_PER_WORKER_OID,
1337        desc: RelationDesc::builder()
1338            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
1339            .with_column("type", SqlScalarType::String.nullable(false))
1340            .with_column("duration_ns", SqlScalarType::UInt64.nullable(false))
1341            .with_column("count", SqlScalarType::Int64.nullable(false))
1342            .with_key(vec![0, 1, 2])
1343            .finish(),
1344        column_comments: BTreeMap::new(),
1345        sql: "SELECT
1346    worker_id, type, duration_ns, pg_catalog.count(*) AS count
1347FROM
1348    mz_introspection.mz_peek_durations_histogram_raw
1349GROUP BY
1350    worker_id, type, duration_ns",
1351        access: vec![PUBLIC_SELECT],
1352        ontology: None,
1353    });
1354
1355pub static MZ_PEEK_DURATIONS_HISTOGRAM: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
1356    name: "mz_peek_durations_histogram",
1357    schema: MZ_INTROSPECTION_SCHEMA,
1358    oid: oid::VIEW_MZ_PEEK_DURATIONS_HISTOGRAM_OID,
1359    desc: RelationDesc::builder()
1360        .with_column("type", SqlScalarType::String.nullable(false))
1361        .with_column("duration_ns", SqlScalarType::UInt64.nullable(false))
1362        .with_column(
1363            "count",
1364            SqlScalarType::Numeric {
1365                max_scale: Some(NumericMaxScale::ZERO),
1366            }
1367            .nullable(false),
1368        )
1369        .with_key(vec![0, 1])
1370        .finish(),
1371    column_comments: BTreeMap::from_iter([
1372        ("type", "The peek variant: `index` or `persist`."),
1373        (
1374            "duration_ns",
1375            "The upper bound of the bucket in nanoseconds.",
1376        ),
1377        (
1378            "count",
1379            "The (noncumulative) count of peeks in this bucket.",
1380        ),
1381    ]),
1382    sql: "
1383SELECT
1384    type, duration_ns,
1385    pg_catalog.sum(count) AS count
1386FROM mz_introspection.mz_peek_durations_histogram_per_worker
1387GROUP BY type, duration_ns",
1388    access: vec![PUBLIC_SELECT],
1389    ontology: Some(Ontology {
1390        entity_name: "peek_duration",
1391        description: "Histogram of SELECT query durations",
1392        links: &const { [] },
1393        column_semantic_types: &[],
1394    }),
1395});
1396
1397pub static MZ_SCHEDULING_ELAPSED_PER_WORKER: LazyLock<BuiltinView> =
1398    LazyLock::new(|| BuiltinView {
1399        name: "mz_scheduling_elapsed_per_worker",
1400        schema: MZ_INTROSPECTION_SCHEMA,
1401        oid: oid::VIEW_MZ_SCHEDULING_ELAPSED_PER_WORKER_OID,
1402        desc: RelationDesc::builder()
1403            .with_column("id", SqlScalarType::UInt64.nullable(false))
1404            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
1405            .with_column("elapsed_ns", SqlScalarType::Int64.nullable(false))
1406            .with_key(vec![0, 1])
1407            .finish(),
1408        column_comments: BTreeMap::new(),
1409        sql: "SELECT
1410    id, worker_id, pg_catalog.count(*) AS elapsed_ns
1411FROM
1412    mz_introspection.mz_scheduling_elapsed_raw
1413GROUP BY
1414    id, worker_id",
1415        access: vec![PUBLIC_SELECT],
1416        ontology: None,
1417    });
1418
1419pub static MZ_SCHEDULING_ELAPSED: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
1420    name: "mz_scheduling_elapsed",
1421    schema: MZ_INTROSPECTION_SCHEMA,
1422    oid: oid::VIEW_MZ_SCHEDULING_ELAPSED_OID,
1423    desc: RelationDesc::builder()
1424        .with_column("id", SqlScalarType::UInt64.nullable(false))
1425        .with_column(
1426            "elapsed_ns",
1427            SqlScalarType::Numeric {
1428                max_scale: Some(NumericMaxScale::ZERO),
1429            }
1430            .nullable(false),
1431        )
1432        .with_key(vec![0])
1433        .finish(),
1434    column_comments: BTreeMap::from_iter([
1435        (
1436            "id",
1437            "The ID of the operator. Corresponds to `mz_dataflow_operators.id`.",
1438        ),
1439        (
1440            "elapsed_ns",
1441            "The total elapsed time spent in the operator in nanoseconds.",
1442        ),
1443    ]),
1444    sql: "
1445SELECT
1446    id,
1447    pg_catalog.sum(elapsed_ns) AS elapsed_ns
1448FROM mz_introspection.mz_scheduling_elapsed_per_worker
1449GROUP BY id",
1450    access: vec![PUBLIC_SELECT],
1451    ontology: Some(Ontology {
1452        entity_name: "scheduling_elapsed",
1453        description: "CPU time spent per operator",
1454        links: &const {
1455            [OntologyLink {
1456                name: "elapsed_for_operator",
1457                target: "dataflow_operator",
1458                properties: LinkProperties::measures("id", "id", "cpu_time_ns"),
1459            }]
1460        },
1461        column_semantic_types: &[],
1462    }),
1463});
1464
1465pub static MZ_COMPUTE_OPERATOR_DURATIONS_HISTOGRAM_PER_WORKER: LazyLock<BuiltinView> =
1466    LazyLock::new(|| BuiltinView {
1467        name: "mz_compute_operator_durations_histogram_per_worker",
1468        schema: MZ_INTROSPECTION_SCHEMA,
1469        oid: oid::VIEW_MZ_COMPUTE_OPERATOR_DURATIONS_HISTOGRAM_PER_WORKER_OID,
1470        desc: RelationDesc::builder()
1471            .with_column("id", SqlScalarType::UInt64.nullable(false))
1472            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
1473            .with_column("duration_ns", SqlScalarType::UInt64.nullable(false))
1474            .with_column("count", SqlScalarType::Int64.nullable(false))
1475            .with_key(vec![0, 1, 2])
1476            .finish(),
1477        column_comments: BTreeMap::new(),
1478        sql: "SELECT
1479    id, worker_id, duration_ns, pg_catalog.count(*) AS count
1480FROM
1481    mz_introspection.mz_compute_operator_durations_histogram_raw
1482GROUP BY
1483    id, worker_id, duration_ns",
1484        access: vec![PUBLIC_SELECT],
1485        ontology: None,
1486    });
1487
1488pub static MZ_COMPUTE_OPERATOR_DURATIONS_HISTOGRAM: LazyLock<BuiltinView> =
1489    LazyLock::new(|| BuiltinView {
1490        name: "mz_compute_operator_durations_histogram",
1491        schema: MZ_INTROSPECTION_SCHEMA,
1492        oid: oid::VIEW_MZ_COMPUTE_OPERATOR_DURATIONS_HISTOGRAM_OID,
1493        desc: RelationDesc::builder()
1494            .with_column("id", SqlScalarType::UInt64.nullable(false))
1495            .with_column("duration_ns", SqlScalarType::UInt64.nullable(false))
1496            .with_column(
1497                "count",
1498                SqlScalarType::Numeric {
1499                    max_scale: Some(NumericMaxScale::ZERO),
1500                }
1501                .nullable(false),
1502            )
1503            .with_key(vec![0, 1])
1504            .finish(),
1505        column_comments: BTreeMap::from_iter([
1506            (
1507                "id",
1508                "The ID of the operator. Corresponds to `mz_dataflow_operators.id`.",
1509            ),
1510            (
1511                "duration_ns",
1512                "The upper bound of the duration bucket in nanoseconds.",
1513            ),
1514            (
1515                "count",
1516                "The (noncumulative) count of invocations in the bucket.",
1517            ),
1518        ]),
1519        sql: "
1520SELECT
1521    id,
1522    duration_ns,
1523    pg_catalog.sum(count) AS count
1524FROM mz_introspection.mz_compute_operator_durations_histogram_per_worker
1525GROUP BY id, duration_ns",
1526        access: vec![PUBLIC_SELECT],
1527        ontology: None,
1528    });
1529
1530pub static MZ_SCHEDULING_PARKS_HISTOGRAM_PER_WORKER: LazyLock<BuiltinView> =
1531    LazyLock::new(|| BuiltinView {
1532        name: "mz_scheduling_parks_histogram_per_worker",
1533        schema: MZ_INTROSPECTION_SCHEMA,
1534        oid: oid::VIEW_MZ_SCHEDULING_PARKS_HISTOGRAM_PER_WORKER_OID,
1535        desc: RelationDesc::builder()
1536            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
1537            .with_column("slept_for_ns", SqlScalarType::UInt64.nullable(false))
1538            .with_column("requested_ns", SqlScalarType::UInt64.nullable(false))
1539            .with_column("count", SqlScalarType::Int64.nullable(false))
1540            .with_key(vec![0, 1, 2])
1541            .finish(),
1542        column_comments: BTreeMap::new(),
1543        sql: "SELECT
1544    worker_id, slept_for_ns, requested_ns, pg_catalog.count(*) AS count
1545FROM
1546    mz_introspection.mz_scheduling_parks_histogram_raw
1547GROUP BY
1548    worker_id, slept_for_ns, requested_ns",
1549        access: vec![PUBLIC_SELECT],
1550        ontology: None,
1551    });
1552
1553pub static MZ_SCHEDULING_PARKS_HISTOGRAM: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
1554    name: "mz_scheduling_parks_histogram",
1555    schema: MZ_INTROSPECTION_SCHEMA,
1556    oid: oid::VIEW_MZ_SCHEDULING_PARKS_HISTOGRAM_OID,
1557    desc: RelationDesc::builder()
1558        .with_column("slept_for_ns", SqlScalarType::UInt64.nullable(false))
1559        .with_column("requested_ns", SqlScalarType::UInt64.nullable(false))
1560        .with_column(
1561            "count",
1562            SqlScalarType::Numeric {
1563                max_scale: Some(NumericMaxScale::ZERO),
1564            }
1565            .nullable(false),
1566        )
1567        .with_key(vec![0, 1])
1568        .finish(),
1569    column_comments: BTreeMap::from_iter([
1570        (
1571            "slept_for_ns",
1572            "The actual length of the park event in nanoseconds.",
1573        ),
1574        (
1575            "requested_ns",
1576            "The requested length of the park event in nanoseconds.",
1577        ),
1578        (
1579            "count",
1580            "The (noncumulative) count of park events in this bucket.",
1581        ),
1582    ]),
1583    sql: "
1584SELECT
1585    slept_for_ns,
1586    requested_ns,
1587    pg_catalog.sum(count) AS count
1588FROM mz_introspection.mz_scheduling_parks_histogram_per_worker
1589GROUP BY slept_for_ns, requested_ns",
1590    access: vec![PUBLIC_SELECT],
1591    ontology: Some(Ontology {
1592        entity_name: "scheduling_parks",
1593        description: "Histogram of operator park durations",
1594        links: &const { [] },
1595        column_semantic_types: &[],
1596    }),
1597});
1598
1599pub static MZ_COMPUTE_ERROR_COUNTS_PER_WORKER: LazyLock<BuiltinView> =
1600    LazyLock::new(|| BuiltinView {
1601        name: "mz_compute_error_counts_per_worker",
1602        schema: MZ_INTROSPECTION_SCHEMA,
1603        oid: oid::VIEW_MZ_COMPUTE_ERROR_COUNTS_PER_WORKER_OID,
1604        desc: RelationDesc::builder()
1605            .with_column("export_id", SqlScalarType::String.nullable(false))
1606            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
1607            .with_column("count", SqlScalarType::Int64.nullable(false))
1608            .with_key(vec![0, 1, 2])
1609            .finish(),
1610        column_comments: BTreeMap::new(),
1611        sql: "
1612WITH MUTUALLY RECURSIVE
1613    -- Indexes that reuse existing indexes rather than maintaining separate dataflows.
1614    -- For these we don't log error counts separately, so we need to forward the error counts from
1615    -- their dependencies instead.
1616    index_reuses(reuse_id text, index_id text) AS (
1617        SELECT d.object_id, d.dependency_id
1618        FROM mz_internal.mz_compute_dependencies d
1619        JOIN mz_introspection.mz_compute_exports e ON (e.export_id = d.object_id)
1620        WHERE NOT EXISTS (
1621            SELECT 1 FROM mz_introspection.mz_dataflows
1622            WHERE id = e.dataflow_id
1623        )
1624    ),
1625    -- Error counts that were directly logged on compute exports.
1626    direct_errors(export_id text, worker_id uint8, count int8) AS (
1627        SELECT export_id, worker_id, count
1628        FROM mz_introspection.mz_compute_error_counts_raw
1629    ),
1630    -- Error counts propagated to index reused.
1631    all_errors(export_id text, worker_id uint8, count int8) AS (
1632        SELECT * FROM direct_errors
1633        UNION
1634        SELECT r.reuse_id, e.worker_id, e.count
1635        FROM all_errors e
1636        JOIN index_reuses r ON (r.index_id = e.export_id)
1637    )
1638SELECT * FROM all_errors",
1639        access: vec![PUBLIC_SELECT],
1640        ontology: Some(Ontology {
1641            entity_name: "compute_error_per_worker",
1642            description: "Error counts per compute collection per worker.",
1643            links: &const {
1644                [OntologyLink {
1645                    name: "errors_in",
1646                    target: "compute_export_per_worker",
1647                    properties: LinkProperties::measures_composite(
1648                        "export_id",
1649                        "export_id",
1650                        "count",
1651                        &[("worker_id", "worker_id")],
1652                    ),
1653                }]
1654            },
1655            column_semantic_types: &[("export_id", SemanticType::GlobalId)],
1656        }),
1657    });
1658
1659pub static MZ_COMPUTE_ERROR_COUNTS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
1660    name: "mz_compute_error_counts",
1661    schema: MZ_INTROSPECTION_SCHEMA,
1662    oid: oid::VIEW_MZ_COMPUTE_ERROR_COUNTS_OID,
1663    desc: RelationDesc::builder()
1664        .with_column("export_id", SqlScalarType::String.nullable(false))
1665        .with_column(
1666            "count",
1667            SqlScalarType::Numeric {
1668                max_scale: Some(NumericMaxScale::ZERO),
1669            }
1670            .nullable(false),
1671        )
1672        .with_key(vec![0])
1673        .finish(),
1674    column_comments: BTreeMap::from_iter([
1675        (
1676            "export_id",
1677            "The ID of the dataflow export. Corresponds to `mz_compute_exports.export_id`.",
1678        ),
1679        (
1680            "count",
1681            "The count of errors present in this dataflow export.",
1682        ),
1683    ]),
1684    sql: "
1685SELECT
1686    export_id,
1687    pg_catalog.sum(count) AS count
1688FROM mz_introspection.mz_compute_error_counts_per_worker
1689GROUP BY export_id
1690HAVING pg_catalog.sum(count) != 0",
1691    access: vec![PUBLIC_SELECT],
1692    ontology: Some(Ontology {
1693        entity_name: "compute_error_count",
1694        description: "Error counts per compute collection",
1695        links: &const {
1696            [OntologyLink {
1697                name: "errors_in",
1698                target: "compute_export",
1699                properties: LinkProperties::fk("export_id", "export_id", Cardinality::OneToOne),
1700            }]
1701        },
1702        column_semantic_types: &[("export_id", SemanticType::GlobalId)],
1703    }),
1704});
1705
1706pub static MZ_MESSAGE_COUNTS_PER_WORKER: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
1707    name: "mz_message_counts_per_worker",
1708    schema: MZ_INTROSPECTION_SCHEMA,
1709    oid: oid::VIEW_MZ_MESSAGE_COUNTS_PER_WORKER_OID,
1710    desc: RelationDesc::builder()
1711        .with_column("channel_id", SqlScalarType::UInt64.nullable(false))
1712        .with_column("from_worker_id", SqlScalarType::UInt64.nullable(false))
1713        .with_column("to_worker_id", SqlScalarType::UInt64.nullable(false))
1714        .with_column("sent", SqlScalarType::Int64.nullable(false))
1715        .with_column("received", SqlScalarType::Int64.nullable(false))
1716        .with_column("batch_sent", SqlScalarType::Int64.nullable(false))
1717        .with_column("batch_received", SqlScalarType::Int64.nullable(false))
1718        .with_key(vec![0, 1, 2])
1719        .finish(),
1720    column_comments: BTreeMap::new(),
1721    sql: "
1722WITH batch_sent_cte AS (
1723    SELECT
1724        channel_id,
1725        from_worker_id,
1726        to_worker_id,
1727        pg_catalog.count(*) AS sent
1728    FROM
1729        mz_introspection.mz_message_batch_counts_sent_raw
1730    GROUP BY
1731        channel_id, from_worker_id, to_worker_id
1732),
1733batch_received_cte AS (
1734    SELECT
1735        channel_id,
1736        from_worker_id,
1737        to_worker_id,
1738        pg_catalog.count(*) AS received
1739    FROM
1740        mz_introspection.mz_message_batch_counts_received_raw
1741    GROUP BY
1742        channel_id, from_worker_id, to_worker_id
1743),
1744sent_cte AS (
1745    SELECT
1746        channel_id,
1747        from_worker_id,
1748        to_worker_id,
1749        pg_catalog.count(*) AS sent
1750    FROM
1751        mz_introspection.mz_message_counts_sent_raw
1752    GROUP BY
1753        channel_id, from_worker_id, to_worker_id
1754),
1755received_cte AS (
1756    SELECT
1757        channel_id,
1758        from_worker_id,
1759        to_worker_id,
1760        pg_catalog.count(*) AS received
1761    FROM
1762        mz_introspection.mz_message_counts_received_raw
1763    GROUP BY
1764        channel_id, from_worker_id, to_worker_id
1765)
1766SELECT
1767    sent_cte.channel_id,
1768    sent_cte.from_worker_id,
1769    sent_cte.to_worker_id,
1770    sent_cte.sent,
1771    received_cte.received,
1772    batch_sent_cte.sent AS batch_sent,
1773    batch_received_cte.received AS batch_received
1774FROM sent_cte
1775JOIN received_cte USING (channel_id, from_worker_id, to_worker_id)
1776JOIN batch_sent_cte USING (channel_id, from_worker_id, to_worker_id)
1777JOIN batch_received_cte USING (channel_id, from_worker_id, to_worker_id)",
1778    access: vec![PUBLIC_SELECT],
1779    ontology: None,
1780});
1781
1782pub static MZ_MESSAGE_COUNTS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
1783    name: "mz_message_counts",
1784    schema: MZ_INTROSPECTION_SCHEMA,
1785    oid: oid::VIEW_MZ_MESSAGE_COUNTS_OID,
1786    desc: RelationDesc::builder()
1787        .with_column("channel_id", SqlScalarType::UInt64.nullable(false))
1788        .with_column(
1789            "sent",
1790            SqlScalarType::Numeric {
1791                max_scale: Some(NumericMaxScale::ZERO),
1792            }
1793            .nullable(false),
1794        )
1795        .with_column(
1796            "received",
1797            SqlScalarType::Numeric {
1798                max_scale: Some(NumericMaxScale::ZERO),
1799            }
1800            .nullable(false),
1801        )
1802        .with_column(
1803            "batch_sent",
1804            SqlScalarType::Numeric {
1805                max_scale: Some(NumericMaxScale::ZERO),
1806            }
1807            .nullable(false),
1808        )
1809        .with_column(
1810            "batch_received",
1811            SqlScalarType::Numeric {
1812                max_scale: Some(NumericMaxScale::ZERO),
1813            }
1814            .nullable(false),
1815        )
1816        .with_key(vec![0])
1817        .finish(),
1818    column_comments: BTreeMap::from_iter([
1819        (
1820            "channel_id",
1821            "The ID of the channel. Corresponds to `mz_dataflow_channels.id`.",
1822        ),
1823        ("sent", "The number of messages sent."),
1824        ("received", "The number of messages received."),
1825        ("batch_sent", "The number of batches sent."),
1826        ("batch_received", "The number of batches received."),
1827    ]),
1828    sql: "
1829SELECT
1830    channel_id,
1831    pg_catalog.sum(sent) AS sent,
1832    pg_catalog.sum(received) AS received,
1833    pg_catalog.sum(batch_sent) AS batch_sent,
1834    pg_catalog.sum(batch_received) AS batch_received
1835FROM mz_introspection.mz_message_counts_per_worker
1836GROUP BY channel_id",
1837    access: vec![PUBLIC_SELECT],
1838    ontology: Some(Ontology {
1839        entity_name: "message_count",
1840        description: "Inter-worker message counts",
1841        links: &const {
1842            [OntologyLink {
1843                name: "counts_for",
1844                target: "dataflow_channel",
1845                properties: LinkProperties::fk("channel_id", "id", Cardinality::OneToOne),
1846            }]
1847        },
1848        column_semantic_types: &[],
1849    }),
1850});
1851
1852pub static MZ_ACTIVE_PEEKS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
1853    name: "mz_active_peeks",
1854    schema: MZ_INTROSPECTION_SCHEMA,
1855    oid: oid::VIEW_MZ_ACTIVE_PEEKS_OID,
1856    desc: RelationDesc::builder()
1857        .with_column("id", SqlScalarType::Uuid.nullable(false))
1858        .with_column("object_id", SqlScalarType::String.nullable(false))
1859        .with_column("type", SqlScalarType::String.nullable(false))
1860        .with_column("time", SqlScalarType::MzTimestamp.nullable(false))
1861        .with_key(vec![0])
1862        .finish(),
1863    column_comments: BTreeMap::from_iter([
1864        ("id", "The ID of the peek request."),
1865        (
1866            "object_id",
1867            "The ID of the collection the peek is targeting. Corresponds to `mz_catalog.mz_indexes.id`, `mz_catalog.mz_materialized_views.id`, `mz_catalog.mz_sources.id`, or `mz_catalog.mz_tables.id`.",
1868        ),
1869        (
1870            "type",
1871            "The type of the corresponding peek: `index` if targeting an index or temporary dataflow; `persist` for a source, materialized view, or table.",
1872        ),
1873        ("time", "The timestamp the peek has requested."),
1874    ]),
1875    sql: "
1876SELECT id, object_id, type, time
1877FROM mz_introspection.mz_active_peeks_per_worker
1878WHERE worker_id = 0::uint8",
1879    access: vec![PUBLIC_SELECT],
1880    ontology: Some(Ontology {
1881        entity_name: "active_peek",
1882        description: "Currently executing SELECT queries",
1883        links: &const { [] },
1884        column_semantic_types: &const {
1885            [
1886                ("object_id", SemanticType::GlobalId),
1887                ("time", SemanticType::MzTimestamp),
1888            ]
1889        },
1890    }),
1891});
1892
1893pub static MZ_DATAFLOW_OPERATOR_REACHABILITY_PER_WORKER: LazyLock<BuiltinView> =
1894    LazyLock::new(|| BuiltinView {
1895        name: "mz_dataflow_operator_reachability_per_worker",
1896        schema: MZ_INTROSPECTION_SCHEMA,
1897        oid: oid::VIEW_MZ_DATAFLOW_OPERATOR_REACHABILITY_PER_WORKER_OID,
1898        desc: RelationDesc::builder()
1899            .with_column("id", SqlScalarType::UInt64.nullable(false))
1900            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
1901            .with_column("port", SqlScalarType::UInt64.nullable(false))
1902            .with_column("update_type", SqlScalarType::String.nullable(false))
1903            .with_column("time", SqlScalarType::MzTimestamp.nullable(true))
1904            .with_column("count", SqlScalarType::Int64.nullable(false))
1905            .with_key(vec![0, 1, 2, 3, 4])
1906            .finish(),
1907        column_comments: BTreeMap::new(),
1908        sql: "SELECT
1909    addr2.id,
1910    reachability.worker_id,
1911    port,
1912    update_type,
1913    time,
1914    pg_catalog.count(*) as count
1915FROM
1916    mz_introspection.mz_dataflow_operator_reachability_raw reachability,
1917    mz_introspection.mz_dataflow_addresses_per_worker addr1,
1918    mz_introspection.mz_dataflow_addresses_per_worker addr2
1919WHERE
1920    addr2.address =
1921    CASE
1922        WHEN source = 0 THEN addr1.address
1923        ELSE addr1.address || reachability.source
1924    END
1925    AND addr1.id = reachability.id
1926    AND addr1.worker_id = reachability.worker_id
1927    AND addr2.worker_id = reachability.worker_id
1928GROUP BY addr2.id, reachability.worker_id, port, update_type, time",
1929        access: vec![PUBLIC_SELECT],
1930        ontology: None,
1931    });
1932
1933pub static MZ_DATAFLOW_OPERATOR_REACHABILITY: LazyLock<BuiltinView> =
1934    LazyLock::new(|| BuiltinView {
1935        name: "mz_dataflow_operator_reachability",
1936        schema: MZ_INTROSPECTION_SCHEMA,
1937        oid: oid::VIEW_MZ_DATAFLOW_OPERATOR_REACHABILITY_OID,
1938        desc: RelationDesc::builder()
1939            .with_column("id", SqlScalarType::UInt64.nullable(false))
1940            .with_column("port", SqlScalarType::UInt64.nullable(false))
1941            .with_column("update_type", SqlScalarType::String.nullable(false))
1942            .with_column("time", SqlScalarType::MzTimestamp.nullable(true))
1943            .with_column(
1944                "count",
1945                SqlScalarType::Numeric {
1946                    max_scale: Some(NumericMaxScale::ZERO),
1947                }
1948                .nullable(false),
1949            )
1950            .with_key(vec![0, 1, 2, 3])
1951            .finish(),
1952        column_comments: BTreeMap::new(),
1953        sql: "
1954SELECT
1955    id,
1956    port,
1957    update_type,
1958    time,
1959    pg_catalog.sum(count) as count
1960FROM mz_introspection.mz_dataflow_operator_reachability_per_worker
1961GROUP BY id, port, update_type, time",
1962        access: vec![PUBLIC_SELECT],
1963        ontology: None,
1964    });
1965
1966pub static MZ_ARRANGEMENT_SIZES_PER_WORKER: LazyLock<BuiltinView> = LazyLock::new(|| {
1967    BuiltinView {
1968        name: "mz_arrangement_sizes_per_worker",
1969        schema: MZ_INTROSPECTION_SCHEMA,
1970        oid: oid::VIEW_MZ_ARRANGEMENT_SIZES_PER_WORKER_OID,
1971        desc: RelationDesc::builder()
1972            .with_column("operator_id", SqlScalarType::UInt64.nullable(false))
1973            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
1974            .with_column("records", SqlScalarType::Int64.nullable(true))
1975            .with_column("batches", SqlScalarType::Int64.nullable(true))
1976            .with_column("size", SqlScalarType::Int64.nullable(true))
1977            .with_column("capacity", SqlScalarType::Int64.nullable(true))
1978            .with_column("allocations", SqlScalarType::Int64.nullable(true))
1979            .finish(),
1980        column_comments: BTreeMap::new(),
1981        sql: "
1982WITH operators_per_worker_cte AS (
1983    SELECT
1984        id AS operator_id,
1985        worker_id
1986    FROM
1987        mz_introspection.mz_dataflow_operators_per_worker
1988),
1989batches_cte AS (
1990    SELECT
1991        operator_id,
1992        worker_id,
1993        COUNT(*) AS batches
1994    FROM
1995        mz_introspection.mz_arrangement_batches_raw
1996    GROUP BY
1997        operator_id, worker_id
1998),
1999records_cte AS (
2000    SELECT
2001        operator_id,
2002        worker_id,
2003        COUNT(*) AS records
2004    FROM
2005        mz_introspection.mz_arrangement_records_raw
2006    GROUP BY
2007        operator_id, worker_id
2008),
2009heap_size_cte AS (
2010    SELECT
2011        operator_id,
2012        worker_id,
2013        COUNT(*) AS size
2014    FROM
2015        mz_introspection.mz_arrangement_heap_size_raw
2016    GROUP BY
2017        operator_id, worker_id
2018),
2019heap_capacity_cte AS (
2020    SELECT
2021        operator_id,
2022        worker_id,
2023        COUNT(*) AS capacity
2024    FROM
2025        mz_introspection.mz_arrangement_heap_capacity_raw
2026    GROUP BY
2027        operator_id, worker_id
2028),
2029heap_allocations_cte AS (
2030    SELECT
2031        operator_id,
2032        worker_id,
2033        COUNT(*) AS allocations
2034    FROM
2035        mz_introspection.mz_arrangement_heap_allocations_raw
2036    GROUP BY
2037        operator_id, worker_id
2038),
2039batcher_records_cte AS (
2040    SELECT
2041        operator_id,
2042        worker_id,
2043        COUNT(*) AS records
2044    FROM
2045        mz_introspection.mz_arrangement_batcher_records_raw
2046    GROUP BY
2047        operator_id, worker_id
2048),
2049batcher_size_cte AS (
2050    SELECT
2051        operator_id,
2052        worker_id,
2053        COUNT(*) AS size
2054    FROM
2055        mz_introspection.mz_arrangement_batcher_size_raw
2056    GROUP BY
2057        operator_id, worker_id
2058),
2059batcher_capacity_cte AS (
2060    SELECT
2061        operator_id,
2062        worker_id,
2063        COUNT(*) AS capacity
2064    FROM
2065        mz_introspection.mz_arrangement_batcher_capacity_raw
2066    GROUP BY
2067        operator_id, worker_id
2068),
2069batcher_allocations_cte AS (
2070    SELECT
2071        operator_id,
2072        worker_id,
2073        COUNT(*) AS allocations
2074    FROM
2075        mz_introspection.mz_arrangement_batcher_allocations_raw
2076    GROUP BY
2077        operator_id, worker_id
2078),
2079combined AS (
2080    SELECT
2081        opw.operator_id,
2082        opw.worker_id,
2083        CASE
2084            WHEN records_cte.records IS NULL AND batcher_records_cte.records IS NULL THEN NULL
2085            ELSE COALESCE(records_cte.records, 0) + COALESCE(batcher_records_cte.records, 0)
2086        END AS records,
2087        batches_cte.batches AS batches,
2088        CASE
2089            WHEN heap_size_cte.size IS NULL AND batcher_size_cte.size IS NULL THEN NULL
2090            ELSE COALESCE(heap_size_cte.size, 0) + COALESCE(batcher_size_cte.size, 0)
2091        END AS size,
2092        CASE
2093            WHEN heap_capacity_cte.capacity IS NULL AND batcher_capacity_cte.capacity IS NULL THEN NULL
2094            ELSE COALESCE(heap_capacity_cte.capacity, 0) + COALESCE(batcher_capacity_cte.capacity, 0)
2095        END AS capacity,
2096        CASE
2097            WHEN heap_allocations_cte.allocations IS NULL AND batcher_allocations_cte.allocations IS NULL THEN NULL
2098            ELSE COALESCE(heap_allocations_cte.allocations, 0) + COALESCE(batcher_allocations_cte.allocations, 0)
2099        END AS allocations
2100    FROM
2101                    operators_per_worker_cte opw
2102    LEFT OUTER JOIN batches_cte USING (operator_id, worker_id)
2103    LEFT OUTER JOIN records_cte USING (operator_id, worker_id)
2104    LEFT OUTER JOIN heap_size_cte USING (operator_id, worker_id)
2105    LEFT OUTER JOIN heap_capacity_cte USING (operator_id, worker_id)
2106    LEFT OUTER JOIN heap_allocations_cte USING (operator_id, worker_id)
2107    LEFT OUTER JOIN batcher_records_cte USING (operator_id, worker_id)
2108    LEFT OUTER JOIN batcher_size_cte USING (operator_id, worker_id)
2109    LEFT OUTER JOIN batcher_capacity_cte USING (operator_id, worker_id)
2110    LEFT OUTER JOIN batcher_allocations_cte USING (operator_id, worker_id)
2111)
2112SELECT
2113    operator_id, worker_id, records, batches, size, capacity, allocations
2114FROM combined
2115WHERE
2116       records     IS NOT NULL
2117    OR batches     IS NOT NULL
2118    OR size        IS NOT NULL
2119    OR capacity    IS NOT NULL
2120    OR allocations IS NOT NULL
2121",
2122        access: vec![PUBLIC_SELECT],
2123        ontology: None,
2124    }
2125});
2126
2127pub static MZ_ARRANGEMENT_SIZES: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
2128    name: "mz_arrangement_sizes",
2129    schema: MZ_INTROSPECTION_SCHEMA,
2130    oid: oid::VIEW_MZ_ARRANGEMENT_SIZES_OID,
2131    desc: RelationDesc::builder()
2132        .with_column("operator_id", SqlScalarType::UInt64.nullable(false))
2133        .with_column("records", SqlScalarType::Int64.nullable(true))
2134        .with_column("batches", SqlScalarType::Int64.nullable(true))
2135        .with_column("size", SqlScalarType::Int64.nullable(true))
2136        .with_column("capacity", SqlScalarType::Int64.nullable(true))
2137        .with_column("allocations", SqlScalarType::Int64.nullable(true))
2138        .with_key(vec![0])
2139        .finish(),
2140    column_comments: BTreeMap::from_iter([
2141        (
2142            "operator_id",
2143            "The ID of the operator that created the arrangement. Corresponds to `mz_dataflow_operators.id`.",
2144        ),
2145        ("records", "The number of records in the arrangement."),
2146        ("batches", "The number of batches in the arrangement."),
2147        ("size", "The utilized size in bytes of the arrangement."),
2148        (
2149            "capacity",
2150            "The capacity in bytes of the arrangement. Can be larger than the size.",
2151        ),
2152        (
2153            "allocations",
2154            "The number of separate memory allocations backing the arrangement.",
2155        ),
2156    ]),
2157    sql: "
2158SELECT
2159    operator_id,
2160    SUM(records)::int8 AS records,
2161    SUM(batches)::int8 AS batches,
2162    SUM(size)::int8 AS size,
2163    SUM(capacity)::int8 AS capacity,
2164    SUM(allocations)::int8 AS allocations
2165FROM mz_introspection.mz_arrangement_sizes_per_worker
2166GROUP BY operator_id",
2167    access: vec![PUBLIC_SELECT],
2168    ontology: Some(Ontology {
2169        entity_name: "arrangement_size",
2170        description: "Aggregated arrangement sizes (records, batches, bytes)",
2171        links: &const {
2172            [OntologyLink {
2173                name: "arrangement_of_operator",
2174                target: "dataflow_operator",
2175                properties: LinkProperties::Measures {
2176                    source_column: "operator_id",
2177                    target_column: "id",
2178                    metric: "arrangement_size",
2179                    source_id_type: None,
2180                    requires_mapping: None,
2181                    note: Some(
2182                        "Both IDs are local uint64 operator IDs within a dataflow, not GlobalIds.",
2183                    ),
2184                    extra_key_columns: None,
2185                },
2186            }]
2187        },
2188        column_semantic_types: &[],
2189    }),
2190});
2191
2192pub static MZ_ARRANGEMENT_SHARING_PER_WORKER: LazyLock<BuiltinView> =
2193    LazyLock::new(|| BuiltinView {
2194        name: "mz_arrangement_sharing_per_worker",
2195        schema: MZ_INTROSPECTION_SCHEMA,
2196        oid: oid::VIEW_MZ_ARRANGEMENT_SHARING_PER_WORKER_OID,
2197        desc: RelationDesc::builder()
2198            .with_column("operator_id", SqlScalarType::UInt64.nullable(false))
2199            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
2200            .with_column("count", SqlScalarType::Int64.nullable(false))
2201            .with_key(vec![0, 1])
2202            .finish(),
2203        column_comments: BTreeMap::new(),
2204        sql: "
2205SELECT
2206    operator_id,
2207    worker_id,
2208    pg_catalog.count(*) AS count
2209FROM mz_introspection.mz_arrangement_sharing_raw
2210GROUP BY operator_id, worker_id",
2211        access: vec![PUBLIC_SELECT],
2212        ontology: None,
2213    });
2214
2215pub static MZ_ARRANGEMENT_SHARING: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
2216    name: "mz_arrangement_sharing",
2217    schema: MZ_INTROSPECTION_SCHEMA,
2218    oid: oid::VIEW_MZ_ARRANGEMENT_SHARING_OID,
2219    desc: RelationDesc::builder()
2220        .with_column("operator_id", SqlScalarType::UInt64.nullable(false))
2221        .with_column("count", SqlScalarType::Int64.nullable(false))
2222        .with_key(vec![0])
2223        .finish(),
2224    column_comments: BTreeMap::from_iter([
2225        (
2226            "operator_id",
2227            "The ID of the operator that created the arrangement. Corresponds to `mz_dataflow_operators.id`.",
2228        ),
2229        (
2230            "count",
2231            "The number of operators that share the arrangement.",
2232        ),
2233    ]),
2234    sql: "
2235SELECT operator_id, count
2236FROM mz_introspection.mz_arrangement_sharing_per_worker
2237WHERE worker_id = 0::uint8",
2238    access: vec![PUBLIC_SELECT],
2239    ontology: Some(Ontology {
2240        entity_name: "arrangement_sharing",
2241        description: "Arrangement sharing between operators",
2242        links: &const {
2243            [OntologyLink {
2244                name: "shared_by",
2245                target: "dataflow_operator",
2246                properties: LinkProperties::fk("operator_id", "id", Cardinality::OneToOne),
2247            }]
2248        },
2249        column_semantic_types: &[],
2250    }),
2251});
2252
2253pub static MZ_DATAFLOW_OPERATOR_PARENTS_PER_WORKER: LazyLock<BuiltinView> =
2254    LazyLock::new(|| BuiltinView {
2255        name: "mz_dataflow_operator_parents_per_worker",
2256        schema: MZ_INTROSPECTION_SCHEMA,
2257        oid: oid::VIEW_MZ_DATAFLOW_OPERATOR_PARENTS_PER_WORKER_OID,
2258        desc: RelationDesc::builder()
2259            .with_column("id", SqlScalarType::UInt64.nullable(false))
2260            .with_column("parent_id", SqlScalarType::UInt64.nullable(false))
2261            .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
2262            .finish(),
2263        column_comments: BTreeMap::new(),
2264        sql: "
2265WITH operator_addrs AS(
2266    SELECT
2267        id, address, worker_id
2268    FROM mz_introspection.mz_dataflow_addresses_per_worker
2269        INNER JOIN mz_introspection.mz_dataflow_operators_per_worker
2270            USING (id, worker_id)
2271),
2272parent_addrs AS (
2273    SELECT
2274        id,
2275        address[1:list_length(address) - 1] AS parent_address,
2276        worker_id
2277    FROM operator_addrs
2278)
2279SELECT pa.id, oa.id AS parent_id, pa.worker_id
2280FROM parent_addrs AS pa
2281    INNER JOIN operator_addrs AS oa
2282        ON pa.parent_address = oa.address
2283        AND pa.worker_id = oa.worker_id",
2284        access: vec![PUBLIC_SELECT],
2285        ontology: None,
2286    });
2287
2288pub static MZ_DATAFLOW_OPERATOR_PARENTS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
2289    name: "mz_dataflow_operator_parents",
2290    schema: MZ_INTROSPECTION_SCHEMA,
2291    oid: oid::VIEW_MZ_DATAFLOW_OPERATOR_PARENTS_OID,
2292    desc: RelationDesc::builder()
2293        .with_column("id", SqlScalarType::UInt64.nullable(false))
2294        .with_column("parent_id", SqlScalarType::UInt64.nullable(false))
2295        .finish(),
2296    column_comments: BTreeMap::from_iter([
2297        (
2298            "id",
2299            "The ID of the operator. Corresponds to `mz_dataflow_operators.id`.",
2300        ),
2301        (
2302            "parent_id",
2303            "The ID of the operator's parent operator. Corresponds to `mz_dataflow_operators.id`.",
2304        ),
2305    ]),
2306    sql: "
2307SELECT id, parent_id
2308FROM mz_introspection.mz_dataflow_operator_parents_per_worker
2309WHERE worker_id = 0::uint8",
2310    access: vec![PUBLIC_SELECT],
2311    ontology: None,
2312});
2313
2314pub static MZ_DATAFLOW_ARRANGEMENT_SIZES: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
2315    name: "mz_dataflow_arrangement_sizes",
2316    schema: MZ_INTROSPECTION_SCHEMA,
2317    oid: oid::VIEW_MZ_DATAFLOW_ARRANGEMENT_SIZES_OID,
2318    desc: RelationDesc::builder()
2319        .with_column("id", SqlScalarType::UInt64.nullable(false))
2320        .with_column("name", SqlScalarType::String.nullable(false))
2321        .with_column("records", SqlScalarType::Int64.nullable(true))
2322        .with_column("batches", SqlScalarType::Int64.nullable(true))
2323        .with_column("size", SqlScalarType::Int64.nullable(true))
2324        .with_column("capacity", SqlScalarType::Int64.nullable(true))
2325        .with_column("allocations", SqlScalarType::Int64.nullable(true))
2326        .with_key(vec![0, 1])
2327        .finish(),
2328    column_comments: BTreeMap::from_iter([
2329        (
2330            "id",
2331            "The ID of the [dataflow]. Corresponds to `mz_dataflows.id`.",
2332        ),
2333        ("name", "The name of the [dataflow]."),
2334        (
2335            "records",
2336            "The number of records in all arrangements in the dataflow.",
2337        ),
2338        (
2339            "batches",
2340            "The number of batches in all arrangements in the dataflow.",
2341        ),
2342        ("size", "The utilized size in bytes of the arrangements."),
2343        (
2344            "capacity",
2345            "The capacity in bytes of the arrangements. Can be larger than the size.",
2346        ),
2347        (
2348            "allocations",
2349            "The number of separate memory allocations backing the arrangements.",
2350        ),
2351    ]),
2352    sql: "
2353SELECT
2354    mdod.dataflow_id AS id,
2355    mdod.dataflow_name AS name,
2356    SUM(mas.records)::int8 AS records,
2357    SUM(mas.batches)::int8 AS batches,
2358    SUM(mas.size)::int8 AS size,
2359    SUM(mas.capacity)::int8 AS capacity,
2360    SUM(mas.allocations)::int8 AS allocations
2361FROM mz_introspection.mz_dataflow_operator_dataflows AS mdod
2362LEFT JOIN mz_introspection.mz_arrangement_sizes AS mas
2363    ON mdod.id = mas.operator_id
2364GROUP BY mdod.dataflow_id, mdod.dataflow_name",
2365    access: vec![PUBLIC_SELECT],
2366    ontology: None,
2367});
2368
2369pub static MZ_EXPECTED_GROUP_SIZE_ADVICE: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
2370    name: "mz_expected_group_size_advice",
2371    schema: MZ_INTROSPECTION_SCHEMA,
2372    oid: oid::VIEW_MZ_EXPECTED_GROUP_SIZE_ADVICE_OID,
2373    desc: RelationDesc::builder()
2374        .with_column("dataflow_id", SqlScalarType::UInt64.nullable(false))
2375        .with_column("dataflow_name", SqlScalarType::String.nullable(false))
2376        .with_column("region_id", SqlScalarType::UInt64.nullable(false))
2377        .with_column("region_name", SqlScalarType::String.nullable(false))
2378        .with_column("levels", SqlScalarType::Int64.nullable(false))
2379        .with_column("to_cut", SqlScalarType::Int64.nullable(false))
2380        .with_column(
2381            "savings",
2382            SqlScalarType::Numeric {
2383                max_scale: Some(NumericMaxScale::ZERO),
2384            }
2385            .nullable(true),
2386        )
2387        .with_column("hint", SqlScalarType::Float64.nullable(false))
2388        .finish(),
2389    column_comments: BTreeMap::from_iter([
2390        (
2391            "dataflow_id",
2392            "The ID of the [dataflow]. Corresponds to `mz_dataflows.id`.",
2393        ),
2394        (
2395            "dataflow_name",
2396            "The internal name of the dataflow hosting the min/max aggregation or Top K.",
2397        ),
2398        (
2399            "region_id",
2400            "The ID of the root operator scope. Corresponds to `mz_dataflow_operators.id`.",
2401        ),
2402        (
2403            "region_name",
2404            "The internal name of the root operator scope for the min/max aggregation or Top K.",
2405        ),
2406        (
2407            "levels",
2408            "The number of levels in the hierarchical scheme implemented by the region.",
2409        ),
2410        (
2411            "to_cut",
2412            "The number of levels that can be eliminated (cut) from the region's hierarchy.",
2413        ),
2414        (
2415            "savings",
2416            "A conservative estimate of the amount of memory in bytes to be saved by applying the hint.",
2417        ),
2418        (
2419            "hint",
2420            "The hint value that will eliminate `to_cut` levels from the region's hierarchy.",
2421        ),
2422    ]),
2423    sql: "
2424        -- The mz_expected_group_size_advice view provides tuning suggestions for the GROUP SIZE
2425        -- query hints. This tuning hint is effective for min/max/top-k patterns, where a stack
2426        -- of arrangements must be built. For each dataflow and region corresponding to one
2427        -- such pattern, we look for how many levels can be eliminated without hitting a level
2428        -- that actually substantially filters the input. The advice is constructed so that
2429        -- setting the hint for the affected region will eliminate these redundant levels of
2430        -- the hierarchical rendering.
2431        --
2432        -- A number of helper CTEs are used for the view definition. The first one, operators,
2433        -- looks for operator names that comprise arrangements of inputs to each level of a
2434        -- min/max/top-k hierarchy.
2435        WITH operators AS (
2436            SELECT
2437                dod.dataflow_id,
2438                dor.id AS region_id,
2439                dod.id,
2440                ars.records,
2441                ars.size
2442            FROM
2443                mz_introspection.mz_dataflow_operator_dataflows dod
2444                JOIN mz_introspection.mz_dataflow_addresses doa
2445                    ON dod.id = doa.id
2446                JOIN mz_introspection.mz_dataflow_addresses dra
2447                    ON dra.address = doa.address[:list_length(doa.address) - 1]
2448                JOIN mz_introspection.mz_dataflow_operators dor
2449                    ON dor.id = dra.id
2450                JOIN mz_introspection.mz_arrangement_sizes ars
2451                    ON ars.operator_id = dod.id
2452            WHERE
2453                dod.name = 'Arranged TopK input'
2454                OR dod.name = 'Arranged MinsMaxesHierarchical input'
2455                OR dod.name = 'Arrange ReduceMinsMaxes'
2456            ),
2457        -- The second CTE, levels, simply computes the heights of the min/max/top-k hierarchies
2458        -- identified in operators above.
2459        levels AS (
2460            SELECT o.dataflow_id, o.region_id, COUNT(*) AS levels
2461            FROM operators o
2462            GROUP BY o.dataflow_id, o.region_id
2463        ),
2464        -- The third CTE, pivot, determines for each min/max/top-k hierarchy, the first input
2465        -- operator. This operator is crucially important, as it records the number of records
2466        -- that was given as input to the gadget as a whole.
2467        pivot AS (
2468            SELECT
2469                o1.dataflow_id,
2470                o1.region_id,
2471                o1.id,
2472                o1.records
2473            FROM operators o1
2474            WHERE
2475                o1.id = (
2476                    SELECT MIN(o2.id)
2477                    FROM operators o2
2478                    WHERE
2479                        o2.dataflow_id = o1.dataflow_id
2480                        AND o2.region_id = o1.region_id
2481                    OPTIONS (AGGREGATE INPUT GROUP SIZE = 8)
2482                )
2483        ),
2484        -- The fourth CTE, candidates, will look for operators where the number of records
2485        -- maintained is not significantly different from the number at the pivot (excluding
2486        -- the pivot itself). These are the candidates for being cut from the dataflow region
2487        -- by adjusting the hint. The query includes a constant, heuristically tuned on TPC-H
2488        -- load generator data, to give some room for small deviations in number of records.
2489        -- The intuition for allowing for this deviation is that we are looking for a strongly
2490        -- reducing point in the hierarchy. To see why one such operator ought to exist in an
2491        -- untuned hierarchy, consider that at each level, we use hashing to distribute rows
2492        -- among groups where the min/max/top-k computation is (partially) applied. If the
2493        -- hierarchy has too many levels, the first-level (pivot) groups will be such that many
2494        -- groups might be empty or contain only one row. Each subsequent level will have a number
2495        -- of groups that is reduced exponentially. So at some point, we will find the level where
2496        -- we actually start having a few rows per group. That's where we will see the row counts
2497        -- significantly drop off.
2498        candidates AS (
2499            SELECT
2500                o.dataflow_id,
2501                o.region_id,
2502                o.id,
2503                o.records,
2504                o.size
2505            FROM
2506                operators o
2507                JOIN pivot p
2508                    ON o.dataflow_id = p.dataflow_id
2509                        AND o.region_id = p.region_id
2510                        AND o.id <> p.id
2511            WHERE o.records >= p.records * (1 - 0.15)
2512        ),
2513        -- The fifth CTE, cuts, computes for each relevant dataflow region, the number of
2514        -- candidate levels that should be cut. We only return here dataflow regions where at
2515        -- least one level must be cut. Note that once we hit a point where the hierarchy starts
2516        -- to have a filtering effect, i.e., after the last candidate, it is dangerous to suggest
2517        -- cutting the height of the hierarchy further. This is because we will have way less
2518        -- groups in the next level, so there should be even further reduction happening or there
2519        -- is some substantial skew in the data. But if the latter is the case, then we should not
2520        -- tune the GROUP SIZE hints down anyway to avoid hurting latency upon updates directed
2521        -- at these unusually large groups. In addition to selecting the levels to cut, we also
2522        -- compute a conservative estimate of the memory savings in bytes that will result from
2523        -- cutting these levels from the hierarchy. The estimate is based on the sizes of the
2524        -- input arrangements for each level to be cut. These arrangements should dominate the
2525        -- size of each level that can be cut, since the reduction gadget internal to the level
2526        -- does not remove much data at these levels.
2527        cuts AS (
2528            SELECT c.dataflow_id, c.region_id, COUNT(*) AS to_cut, SUM(c.size) AS savings
2529            FROM candidates c
2530            GROUP BY c.dataflow_id, c.region_id
2531            HAVING COUNT(*) > 0
2532        )
2533        -- Finally, we compute the hint suggestion for each dataflow region based on the number of
2534        -- levels and the number of candidates to be cut. The hint is computed taking into account
2535        -- the fan-in used in rendering for the hash partitioning and reduction of the groups,
2536        -- currently equal to 16.
2537        SELECT
2538            dod.dataflow_id,
2539            dod.dataflow_name,
2540            dod.id AS region_id,
2541            dod.name AS region_name,
2542            l.levels,
2543            c.to_cut,
2544            c.savings,
2545            pow(16, l.levels - c.to_cut) - 1 AS hint
2546        FROM cuts c
2547            JOIN levels l
2548                ON c.dataflow_id = l.dataflow_id AND c.region_id = l.region_id
2549            JOIN mz_introspection.mz_dataflow_operator_dataflows dod
2550                ON dod.dataflow_id = c.dataflow_id AND dod.id = c.region_id",
2551    access: vec![PUBLIC_SELECT],
2552    ontology: Some(Ontology {
2553        entity_name: "group_size_advice",
2554        description: "Advice on expected group sizes for reduce operators",
2555        links: &const { [] },
2556        column_semantic_types: &[],
2557    }),
2558});