mz_adapter/catalog/
builtin_table_updates.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
10mod notice;
11
12use bytesize::ByteSize;
13use ipnet::IpNet;
14use mz_audit_log::VersionedStorageUsage;
15use mz_catalog::SYSTEM_CONN_ID;
16use mz_catalog::builtin::{
17    BuiltinTable, MZ_AGGREGATES, MZ_ARRAY_TYPES, MZ_BASE_TYPES, MZ_CLUSTER_REPLICA_SIZE_INTERNAL,
18    MZ_CLUSTER_REPLICA_SIZES, MZ_COLUMNS, MZ_EGRESS_IPS, MZ_FUNCTIONS, MZ_INDEX_COLUMNS,
19    MZ_LICENSE_KEYS, MZ_LIST_TYPES, MZ_MAP_TYPES, MZ_OPERATORS, MZ_PSEUDO_TYPES, MZ_ROLE_AUTH,
20    MZ_SESSIONS, MZ_STORAGE_USAGE_BY_SHARD, MZ_SUBSCRIPTIONS, MZ_TYPE_PG_METADATA, MZ_TYPES,
21    MZ_WEBHOOKS_SOURCES,
22};
23use mz_catalog::memory::error::Error;
24use mz_catalog::memory::objects::{
25    CatalogItem, DataSourceDesc, Func, Index, Table, TableDataSource, Type,
26};
27use mz_expr::MirScalarExpr;
28use mz_license_keys::ValidatedLicenseKey;
29use mz_orchestrator::{CpuLimit, DiskLimit, MemoryLimit};
30use mz_ore::cast::CastFrom;
31use mz_ore::collections::CollectionExt;
32use mz_persist_client::batch::ProtoBatch;
33use mz_repr::adt::array::ArrayDimension;
34use mz_repr::adt::mz_acl_item::PrivilegeMap;
35use mz_repr::role_id::RoleId;
36use mz_repr::{
37    CatalogItemId, Datum, Diff, GlobalId, ReprColumnType, Row, RowPacker, SqlScalarType,
38};
39use mz_sql::ast::{CreateIndexStatement, Statement};
40use mz_sql::catalog::{CatalogType, TypeCategory};
41use mz_sql::func::FuncImplCatalogDetails;
42use mz_sql::names::SchemaSpecifier;
43use mz_sql_parser::ast::display::AstDisplay;
44use mz_storage_client::client::TableData;
45use smallvec::smallvec;
46use uuid::Uuid;
47
48// DO NOT add any more imports from `crate` outside of `crate::catalog`.
49use crate::active_compute_sink::ActiveSubscribe;
50use crate::catalog::CatalogState;
51use crate::coord::ConnMeta;
52
53/// An update to a built-in table.
54#[derive(Debug, Clone)]
55pub struct BuiltinTableUpdate<T = CatalogItemId> {
56    /// The reference of the table to update.
57    pub id: T,
58    /// The data to put into the table.
59    pub data: TableData,
60}
61
62impl<T> BuiltinTableUpdate<T> {
63    /// Create a [`BuiltinTableUpdate`] from a [`Row`].
64    pub fn row(id: T, row: Row, diff: Diff) -> BuiltinTableUpdate<T> {
65        BuiltinTableUpdate {
66            id,
67            data: TableData::Rows(vec![(row, diff)]),
68        }
69    }
70
71    pub fn batch(id: T, batch: ProtoBatch) -> BuiltinTableUpdate<T> {
72        BuiltinTableUpdate {
73            id,
74            data: TableData::Batches(smallvec![batch]),
75        }
76    }
77}
78
79impl CatalogState {
80    pub fn resolve_builtin_table_updates(
81        &self,
82        builtin_table_update: Vec<BuiltinTableUpdate<&'static BuiltinTable>>,
83    ) -> Vec<BuiltinTableUpdate<CatalogItemId>> {
84        builtin_table_update
85            .into_iter()
86            .map(|builtin_table_update| self.resolve_builtin_table_update(builtin_table_update))
87            .collect()
88    }
89
90    pub fn resolve_builtin_table_update(
91        &self,
92        BuiltinTableUpdate { id, data }: BuiltinTableUpdate<&'static BuiltinTable>,
93    ) -> BuiltinTableUpdate<CatalogItemId> {
94        let id = self.resolve_builtin_table(id);
95        BuiltinTableUpdate { id, data }
96    }
97
98    pub(super) fn pack_role_auth_update(
99        &self,
100        id: RoleId,
101        diff: Diff,
102    ) -> BuiltinTableUpdate<&'static BuiltinTable> {
103        let role_auth = self.get_role_auth(&id);
104        let role = self.get_role(&id);
105        BuiltinTableUpdate::row(
106            &*MZ_ROLE_AUTH,
107            Row::pack_slice(&[
108                Datum::String(&role_auth.role_id.to_string()),
109                Datum::UInt32(role.oid),
110                match &role_auth.password_hash {
111                    Some(hash) => Datum::String(hash),
112                    None => Datum::Null,
113                },
114                Datum::TimestampTz(
115                    mz_ore::now::to_datetime(role_auth.updated_at)
116                        .try_into()
117                        .expect("must fit"),
118                ),
119            ]),
120            diff,
121        )
122    }
123
124    pub(super) fn pack_item_update(
125        &self,
126        id: CatalogItemId,
127        diff: Diff,
128    ) -> Vec<BuiltinTableUpdate<&'static BuiltinTable>> {
129        let entry = self.get_entry(&id);
130        let oid = entry.oid();
131        let conn_id = entry.item().conn_id().unwrap_or(&SYSTEM_CONN_ID);
132        let schema_id = &self
133            .get_schema(
134                &entry.name().qualifiers.database_spec,
135                &entry.name().qualifiers.schema_spec,
136                conn_id,
137            )
138            .id;
139        let name = &entry.name().item;
140        let owner_id = entry.owner_id();
141        let privileges_row = self.pack_privilege_array_row(entry.privileges());
142        let privileges = privileges_row.unpack_first();
143        let mut updates = match entry.item() {
144            CatalogItem::Index(index) => self.pack_index_update(id, index, diff),
145            CatalogItem::Source(source) => {
146                match &source.data_source {
147                    DataSourceDesc::Webhook { .. } => {
148                        vec![self.pack_webhook_source_update(id, diff)]
149                    }
150                    // Old-syntax subsource metadata (mz_postgres/mysql/sql_server_source_tables)
151                    // is now derived from create_sql by materialized views over
152                    // mz_catalog_raw, so ingestion exports need no special packing.
153                    DataSourceDesc::Ingestion { .. }
154                    | DataSourceDesc::OldSyntaxIngestion { .. }
155                    | DataSourceDesc::IngestionExport { .. }
156                    | DataSourceDesc::Introspection(_)
157                    | DataSourceDesc::Progress
158                    | DataSourceDesc::Catalog => vec![],
159                }
160            }
161            // mz_sinks, mz_kafka_sinks and mz_iceberg_sinks read create_sql
162            // out of mz_catalog_raw, so there is nothing to pack here.
163            CatalogItem::Sink(_) => vec![],
164            CatalogItem::Type(ty) => {
165                self.pack_type_update(id, oid, schema_id, name, owner_id, privileges, ty, diff)
166            }
167            CatalogItem::Func(func) => {
168                self.pack_func_update(id, schema_id, name, owner_id, func, diff)
169            }
170            // Tables, views, materialized views, and metric sinks are exposed
171            // through materialized views derived from `mz_catalog_raw`, and
172            // logs and secrets never had builtin-table rows, so none pack a row
173            // here.
174            CatalogItem::Table(_)
175            | CatalogItem::View(_)
176            | CatalogItem::MaterializedView(_)
177            | CatalogItem::Log(_)
178            | CatalogItem::Secret(_)
179            | CatalogItem::MetricSink(_) => vec![],
180            // Connection details (mz_kafka_connections, mz_ssh_tunnel_connections,
181            // mz_aws_connections, mz_aws_privatelink_connections) are now derived
182            // from the persisted create_sql by materialized views over
183            // mz_catalog_raw, so connections need no special packing here.
184            CatalogItem::Connection(_) => vec![],
185        };
186
187        // Always report the latest for an objects columns.
188        if let Some(desc) = entry.relation_desc_latest() {
189            let defaults = match entry.item() {
190                CatalogItem::Table(Table {
191                    data_source: TableDataSource::TableWrites { defaults },
192                    ..
193                }) => Some(defaults),
194                _ => None,
195            };
196            for (i, (column_name, column_type)) in desc.iter().enumerate() {
197                let default: Option<String> = defaults.map(|d| d[i].to_ast_string_stable());
198                let default: Datum = default
199                    .as_ref()
200                    .map(|d| Datum::String(d))
201                    .unwrap_or(Datum::Null);
202                let pgtype = mz_pgrepr::Type::from(&column_type.scalar_type);
203                let (type_name, type_oid) = match &column_type.scalar_type {
204                    SqlScalarType::List {
205                        custom_id: Some(custom_id),
206                        ..
207                    }
208                    | SqlScalarType::Map {
209                        custom_id: Some(custom_id),
210                        ..
211                    }
212                    | SqlScalarType::Record {
213                        custom_id: Some(custom_id),
214                        ..
215                    } => {
216                        let entry = self.get_entry(custom_id);
217                        // NOTE(benesch): the `mz_columns.type text` field is
218                        // wrong. Types do not have a name that can be
219                        // represented as a single textual field. There can be
220                        // multiple types with the same name in different
221                        // schemas and databases. We should eventually deprecate
222                        // the `type` field in favor of a new `type_id` field
223                        // that can be joined against `mz_types`.
224                        //
225                        // For now, in the interest of pragmatism, we just use
226                        // the type's item name, and accept that there may be
227                        // ambiguity if the same type name is used in multiple
228                        // schemas. The ambiguity is mitigated by the OID, which
229                        // can be joined against `mz_types.oid` to resolve the
230                        // ambiguity.
231                        let name = &*entry.name().item;
232                        let oid = entry.oid();
233                        (name, oid)
234                    }
235                    _ => (pgtype.name(), pgtype.oid()),
236                };
237                updates.push(BuiltinTableUpdate::row(
238                    &*MZ_COLUMNS,
239                    Row::pack_slice(&[
240                        Datum::String(&id.to_string()),
241                        Datum::String(column_name),
242                        Datum::UInt64(u64::cast_from(i + 1)),
243                        Datum::from(column_type.nullable),
244                        Datum::String(type_name),
245                        default,
246                        Datum::UInt32(type_oid),
247                        Datum::Int32(pgtype.typmod()),
248                    ]),
249                    diff,
250                ));
251            }
252        }
253
254        updates
255    }
256
257    fn pack_index_update(
258        &self,
259        id: CatalogItemId,
260        index: &Index,
261        diff: Diff,
262    ) -> Vec<BuiltinTableUpdate<&'static BuiltinTable>> {
263        let mut updates = vec![];
264
265        let create_stmt = mz_sql::parse::parse(&index.create_sql)
266            .unwrap_or_else(|e| {
267                panic!(
268                    "create_sql cannot be invalid: `{}` --- error: `{}`",
269                    index.create_sql, e
270                )
271            })
272            .into_element()
273            .ast;
274
275        let key_sqls = match &create_stmt {
276            Statement::CreateIndex(CreateIndexStatement { key_parts, .. }) => key_parts
277                .as_ref()
278                .expect("key_parts is filled in during planning"),
279            _ => unreachable!(),
280        };
281
282        let on_entry = self.get_entry_by_global_id(&index.on);
283        let on_desc = on_entry
284            .relation_desc()
285            .expect("can only create indexes on items with a valid description");
286        let repr_col_types: Vec<ReprColumnType> = on_desc
287            .typ()
288            .column_types
289            .iter()
290            .map(ReprColumnType::from)
291            .collect();
292        for (i, key) in index.keys.iter().enumerate() {
293            let nullable = key.typ(&repr_col_types).nullable;
294            let seq_in_index = u64::cast_from(i + 1);
295            let key_sql = key_sqls
296                .get(i)
297                .expect("missing sql information for index key")
298                .to_ast_string_simple();
299            let (field_number, expression) = match key {
300                MirScalarExpr::Column(col, _) => {
301                    (Datum::UInt64(u64::cast_from(*col + 1)), Datum::Null)
302                }
303                _ => (Datum::Null, Datum::String(&key_sql)),
304            };
305            updates.push(BuiltinTableUpdate::row(
306                &*MZ_INDEX_COLUMNS,
307                Row::pack_slice(&[
308                    Datum::String(&id.to_string()),
309                    Datum::UInt64(seq_in_index),
310                    field_number,
311                    expression,
312                    Datum::from(nullable),
313                ]),
314                diff,
315            ));
316        }
317
318        updates
319    }
320
321    fn pack_type_update(
322        &self,
323        id: CatalogItemId,
324        oid: u32,
325        schema_id: &SchemaSpecifier,
326        name: &str,
327        owner_id: &RoleId,
328        privileges: Datum,
329        typ: &Type,
330        diff: Diff,
331    ) -> Vec<BuiltinTableUpdate<&'static BuiltinTable>> {
332        let mut out = vec![];
333
334        let redacted = typ.create_sql.as_ref().map(|create_sql| {
335            mz_sql::parse::parse(create_sql)
336                .unwrap_or_else(|_| panic!("create_sql cannot be invalid: {}", create_sql))
337                .into_element()
338                .ast
339                .to_ast_string_redacted()
340        });
341
342        out.push(BuiltinTableUpdate::row(
343            &*MZ_TYPES,
344            Row::pack_slice(&[
345                Datum::String(&id.to_string()),
346                Datum::UInt32(oid),
347                Datum::String(&schema_id.to_string()),
348                Datum::String(name),
349                Datum::String(&TypeCategory::from_catalog_type(&typ.details.typ).to_string()),
350                Datum::String(&owner_id.to_string()),
351                privileges,
352                if let Some(create_sql) = &typ.create_sql {
353                    Datum::String(create_sql)
354                } else {
355                    Datum::Null
356                },
357                if let Some(redacted) = &redacted {
358                    Datum::String(redacted)
359                } else {
360                    Datum::Null
361                },
362            ]),
363            diff,
364        ));
365
366        let mut row = Row::default();
367        let mut packer = row.packer();
368
369        fn append_modifier(packer: &mut RowPacker<'_>, mods: &[i64]) {
370            if mods.is_empty() {
371                packer.push(Datum::Null);
372            } else {
373                packer.push_list(mods.iter().map(|m| Datum::Int64(*m)));
374            }
375        }
376
377        let index_id = match &typ.details.typ {
378            CatalogType::Array {
379                element_reference: element_id,
380            } => {
381                packer.push(Datum::String(&id.to_string()));
382                packer.push(Datum::String(&element_id.to_string()));
383                &MZ_ARRAY_TYPES
384            }
385            CatalogType::List {
386                element_reference: element_id,
387                element_modifiers,
388            } => {
389                packer.push(Datum::String(&id.to_string()));
390                packer.push(Datum::String(&element_id.to_string()));
391                append_modifier(&mut packer, element_modifiers);
392                &MZ_LIST_TYPES
393            }
394            CatalogType::Map {
395                key_reference: key_id,
396                value_reference: value_id,
397                key_modifiers,
398                value_modifiers,
399            } => {
400                packer.push(Datum::String(&id.to_string()));
401                packer.push(Datum::String(&key_id.to_string()));
402                packer.push(Datum::String(&value_id.to_string()));
403                append_modifier(&mut packer, key_modifiers);
404                append_modifier(&mut packer, value_modifiers);
405                &MZ_MAP_TYPES
406            }
407            CatalogType::Pseudo => {
408                packer.push(Datum::String(&id.to_string()));
409                &MZ_PSEUDO_TYPES
410            }
411            _ => {
412                packer.push(Datum::String(&id.to_string()));
413                &MZ_BASE_TYPES
414            }
415        };
416        out.push(BuiltinTableUpdate::row(index_id, row, diff));
417
418        if let Some(pg_metadata) = &typ.details.pg_metadata {
419            // Resolved the same way as a column's wire type, so this matches the
420            // `type_len` pgwire sends in `RowDescription`. Pseudo-types and arrays
421            // of them (`_record`) have no scalar type, so their length is unknown.
422            let typlen =
423                match mz_sql::plan::scalar_type_from_catalog(&self.for_system_session(), id, &[]) {
424                    Ok(scalar_type) => Datum::Int16(mz_pgrepr::Type::from(&scalar_type).typlen()),
425                    Err(_) => Datum::Null,
426                };
427            out.push(BuiltinTableUpdate::row(
428                &*MZ_TYPE_PG_METADATA,
429                Row::pack_slice(&[
430                    Datum::String(&id.to_string()),
431                    Datum::UInt32(pg_metadata.typinput_oid),
432                    Datum::UInt32(pg_metadata.typreceive_oid),
433                    Datum::UInt32(pg_metadata.typsend_oid),
434                    typlen,
435                ]),
436                diff,
437            ));
438        }
439
440        out
441    }
442
443    fn pack_func_update(
444        &self,
445        id: CatalogItemId,
446        schema_id: &SchemaSpecifier,
447        name: &str,
448        owner_id: &RoleId,
449        func: &Func,
450        diff: Diff,
451    ) -> Vec<BuiltinTableUpdate<&'static BuiltinTable>> {
452        let mut updates = vec![];
453        for func_impl_details in func.inner.func_impls() {
454            let arg_type_ids = func_impl_details
455                .arg_typs
456                .iter()
457                .map(|typ| self.get_system_type(typ).id().to_string())
458                .collect::<Vec<_>>();
459
460            let mut row = Row::default();
461            row.packer()
462                .try_push_array(
463                    &[ArrayDimension {
464                        lower_bound: 1,
465                        length: arg_type_ids.len(),
466                    }],
467                    arg_type_ids.iter().map(|id| Datum::String(id)),
468                )
469                .expect(
470                    "arg_type_ids is 1 dimensional, and its length is used for the array length",
471                );
472            let arg_type_ids = row.unpack_first();
473
474            updates.push(BuiltinTableUpdate::row(
475                &*MZ_FUNCTIONS,
476                Row::pack_slice(&[
477                    Datum::String(&id.to_string()),
478                    Datum::UInt32(func_impl_details.oid),
479                    Datum::String(&schema_id.to_string()),
480                    Datum::String(name),
481                    arg_type_ids,
482                    Datum::from(
483                        func_impl_details
484                            .variadic_typ
485                            .map(|typ| self.get_system_type(typ).id().to_string())
486                            .as_deref(),
487                    ),
488                    Datum::from(
489                        func_impl_details
490                            .return_typ
491                            .map(|typ| self.get_system_type(typ).id().to_string())
492                            .as_deref(),
493                    ),
494                    func_impl_details.return_is_set.into(),
495                    Datum::String(&owner_id.to_string()),
496                ]),
497                diff,
498            ));
499
500            if let mz_sql::func::Func::Aggregate(_) = func.inner {
501                updates.push(BuiltinTableUpdate::row(
502                    &*MZ_AGGREGATES,
503                    Row::pack_slice(&[
504                        Datum::UInt32(func_impl_details.oid),
505                        // TODO(database-issues#1064): Support ordered-set aggregate functions.
506                        Datum::String("n"),
507                        Datum::Int16(0),
508                    ]),
509                    diff,
510                ));
511            }
512        }
513        updates
514    }
515
516    pub fn pack_op_update(
517        &self,
518        operator: &str,
519        func_impl_details: FuncImplCatalogDetails,
520        diff: Diff,
521    ) -> BuiltinTableUpdate<&'static BuiltinTable> {
522        let arg_type_ids = func_impl_details
523            .arg_typs
524            .iter()
525            .map(|typ| self.get_system_type(typ).id().to_string())
526            .collect::<Vec<_>>();
527
528        let mut row = Row::default();
529        row.packer()
530            .try_push_array(
531                &[ArrayDimension {
532                    lower_bound: 1,
533                    length: arg_type_ids.len(),
534                }],
535                arg_type_ids.iter().map(|id| Datum::String(id)),
536            )
537            .expect("arg_type_ids is 1 dimensional, and its length is used for the array length");
538        let arg_type_ids = row.unpack_first();
539
540        BuiltinTableUpdate::row(
541            &*MZ_OPERATORS,
542            Row::pack_slice(&[
543                Datum::UInt32(func_impl_details.oid),
544                Datum::String(operator),
545                arg_type_ids,
546                Datum::from(
547                    func_impl_details
548                        .return_typ
549                        .map(|typ| self.get_system_type(typ).id().to_string())
550                        .as_deref(),
551                ),
552            ]),
553            diff,
554        )
555    }
556
557    pub fn pack_storage_usage_update(
558        &self,
559        VersionedStorageUsage::V1(event): VersionedStorageUsage,
560        diff: Diff,
561    ) -> BuiltinTableUpdate<&'static BuiltinTable> {
562        let id = &MZ_STORAGE_USAGE_BY_SHARD;
563        let row = Row::pack_slice(&[
564            Datum::UInt64(event.id),
565            Datum::from(event.shard_id.as_deref()),
566            Datum::UInt64(event.size_bytes),
567            Datum::TimestampTz(
568                mz_ore::now::to_datetime(event.collection_timestamp)
569                    .try_into()
570                    .expect("must fit"),
571            ),
572        ]);
573        BuiltinTableUpdate::row(id, row, diff)
574    }
575
576    pub fn pack_egress_ip_update(
577        &self,
578        ip: &IpNet,
579    ) -> Result<BuiltinTableUpdate<&'static BuiltinTable>, Error> {
580        let id = &MZ_EGRESS_IPS;
581        let addr = ip.network();
582        let row = Row::pack_slice(&[
583            Datum::String(&addr.to_string()),
584            Datum::Int32(ip.prefix_len().into()),
585            Datum::String(&format!("{}/{}", addr, ip.prefix_len())),
586        ]);
587        Ok(BuiltinTableUpdate::row(id, row, Diff::ONE))
588    }
589
590    pub fn pack_license_key_update(
591        &self,
592        license_key: &ValidatedLicenseKey,
593    ) -> Result<BuiltinTableUpdate<&'static BuiltinTable>, Error> {
594        let id = &MZ_LICENSE_KEYS;
595        let row = Row::pack_slice(&[
596            Datum::String(&license_key.id),
597            Datum::String(&license_key.organization),
598            Datum::String(&license_key.environment_id),
599            Datum::TimestampTz(
600                mz_ore::now::to_datetime(license_key.expiration * 1000)
601                    .try_into()
602                    .expect("must fit"),
603            ),
604            Datum::TimestampTz(
605                mz_ore::now::to_datetime(license_key.not_before * 1000)
606                    .try_into()
607                    .expect("must fit"),
608            ),
609        ]);
610        Ok(BuiltinTableUpdate::row(id, row, Diff::ONE))
611    }
612
613    pub fn pack_all_replica_size_updates(&self) -> Vec<BuiltinTableUpdate<&'static BuiltinTable>> {
614        let mut updates = Vec::new();
615        for (size, alloc) in &self.cluster_replica_sizes.0 {
616            // Just invent something when the limits are `None`, which only happens in non-prod
617            // environments (tests, process orchestrator, etc.)
618            let DiskLimit(ByteSize(disk_bytes)) =
619                (alloc.disk_limit).unwrap_or(DiskLimit::ARBITRARY);
620
621            // The disk column of mz_clusters / mz_cluster_replicas MVs needs
622            // `swap_enabled` and `disk_bytes`; expose them through a parallel
623            // internal table. Unlike the public sizes table below, we write
624            // here unconditionally — `cluster_replica_size_has_disk` previously
625            // indexed the in-memory map without checking `disabled`, so a
626            // managed cluster pinned to a disabled size still resolved its
627            // `disk` column from the real allocation. Writing disabled rows
628            // here preserves that behavior.
629            let internal_row = Row::pack_slice(&[
630                size.as_str().into(),
631                Datum::from(alloc.swap_enabled),
632                disk_bytes.into(),
633            ]);
634            updates.push(BuiltinTableUpdate::row(
635                &*MZ_CLUSTER_REPLICA_SIZE_INTERNAL,
636                internal_row,
637                Diff::ONE,
638            ));
639
640            if alloc.disabled {
641                continue;
642            }
643
644            let cpu_limit = alloc.cpu_limit.unwrap_or(CpuLimit::MAX);
645            let MemoryLimit(ByteSize(memory_bytes)) =
646                (alloc.memory_limit).unwrap_or(MemoryLimit::MAX);
647
648            let row = Row::pack_slice(&[
649                size.as_str().into(),
650                u64::cast_from(alloc.scale).into(),
651                u64::cast_from(alloc.workers).into(),
652                cpu_limit.as_nanocpus().into(),
653                memory_bytes.into(),
654                disk_bytes.into(),
655                (alloc.credits_per_hour).into(),
656            ]);
657
658            updates.push(BuiltinTableUpdate::row(
659                &*MZ_CLUSTER_REPLICA_SIZES,
660                row,
661                Diff::ONE,
662            ));
663        }
664
665        updates
666    }
667
668    pub fn pack_subscribe_update(
669        &self,
670        id: GlobalId,
671        subscribe: &ActiveSubscribe,
672        session_uuid: Uuid,
673        diff: Diff,
674    ) -> BuiltinTableUpdate<&'static BuiltinTable> {
675        let mut row = Row::default();
676        let mut packer = row.packer();
677        packer.push(Datum::String(&id.to_string()));
678        packer.push(Datum::Uuid(session_uuid));
679        packer.push(Datum::String(&subscribe.cluster_id.to_string()));
680
681        let start_dt = mz_ore::now::to_datetime(subscribe.start_time);
682        packer.push(Datum::TimestampTz(start_dt.try_into().expect("must fit")));
683
684        let depends_on: Vec<_> = subscribe
685            .depends_on
686            .iter()
687            .map(|id| id.to_string())
688            .collect();
689        packer.push_list(depends_on.iter().map(|s| Datum::String(s)));
690
691        BuiltinTableUpdate::row(&*MZ_SUBSCRIPTIONS, row, diff)
692    }
693
694    pub fn pack_session_update(
695        &self,
696        conn: &ConnMeta,
697        diff: Diff,
698    ) -> BuiltinTableUpdate<&'static BuiltinTable> {
699        let connect_dt = mz_ore::now::to_datetime(conn.connected_at());
700        BuiltinTableUpdate::row(
701            &*MZ_SESSIONS,
702            Row::pack_slice(&[
703                Datum::Uuid(conn.uuid()),
704                Datum::UInt32(conn.conn_id().unhandled()),
705                Datum::String(&conn.authenticated_role_id().to_string()),
706                Datum::from(conn.client_ip().map(|ip| ip.to_string()).as_deref()),
707                Datum::TimestampTz(connect_dt.try_into().expect("must fit")),
708            ]),
709            diff,
710        )
711    }
712
713    fn pack_privilege_array_row(&self, privileges: &PrivilegeMap) -> Row {
714        let mut row = Row::default();
715        let flat_privileges: Vec<_> = privileges.all_values_owned().collect();
716        row.packer()
717            .try_push_array(
718                &[ArrayDimension {
719                    lower_bound: 1,
720                    length: flat_privileges.len(),
721                }],
722                flat_privileges
723                    .into_iter()
724                    .map(|mz_acl_item| Datum::MzAclItem(mz_acl_item.clone())),
725            )
726            .expect("privileges is 1 dimensional, and its length is used for the array length");
727        row
728    }
729
730    pub fn pack_webhook_source_update(
731        &self,
732        item_id: CatalogItemId,
733        diff: Diff,
734    ) -> BuiltinTableUpdate<&'static BuiltinTable> {
735        let url = self
736            .try_get_webhook_url(&item_id)
737            .expect("webhook source should exist");
738        let url = url.to_string();
739        let name = &self.get_entry(&item_id).name().item;
740        let id_str = item_id.to_string();
741
742        BuiltinTableUpdate::row(
743            &*MZ_WEBHOOKS_SOURCES,
744            Row::pack_slice(&[
745                Datum::String(&id_str),
746                Datum::String(name),
747                Datum::String(&url),
748            ]),
749            diff,
750        )
751    }
752}
mz_adapter/catalog/builtin_table_updates.rs

mz_adapter/catalog/
builtin_table_updates.rs
