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