Skip to content

WIP: Implement type builtin relations #40068

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
SangJunBak wants to merge 1 commit into from
Closed

WIP: Implement type builtin relations #40068

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions doc/user/content/sql/system-catalog/mz_internal.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1637,6 +1637,7 @@ The `mz_webhook_sources` table contains a row for each webhook source in the sys
<!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_builtin_materialized_views -->
<!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_builtin_sources -->
<!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_builtin_tables -->
<!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_builtin_types -->
<!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_builtin_views -->
<!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_catalog_raw -->
<!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_cluster_replica_size_internal -->
Expand Down
12 changes: 6 additions & 6 deletions misc/python/materialize/checks/all_checks/builtin_version_pin.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,10 +34,10 @@
# dependencies, so conversions there cannot poison customer catalogs.
# mz_role_auth is excluded because only mz_system can read it.
_STABLE_BUILTINS: dict[str, MzVersion | None] = {
"mz_array_types": None,
"mz_array_types": MzVersion.parse_mz("v26.48.0"),
"mz_audit_events": MzVersion.parse_mz("v26.33.0"),
"mz_aws_privatelink_connections": MzVersion.parse_mz("v26.37.0"),
"mz_base_types": None,
"mz_base_types": MzVersion.parse_mz("v26.48.0"),
"mz_cluster_replica_sizes": None,
"mz_cluster_replicas": MzVersion.parse_mz("v26.30.0"),
"mz_clusters": MzVersion.parse_mz("v26.30.0"),
Expand All @@ -50,15 +50,15 @@
"mz_kafka_connections": MzVersion.parse_mz("v26.37.0"),
"mz_kafka_sinks": MzVersion.parse_mz("v26.38.0"),
"mz_kafka_sources": MzVersion.parse_mz("v26.34.0"),
"mz_list_types": None,
"mz_map_types": None,
"mz_list_types": MzVersion.parse_mz("v26.48.0"),
"mz_map_types": MzVersion.parse_mz("v26.48.0"),
"mz_operators": None,
"mz_pseudo_types": None,
"mz_pseudo_types": MzVersion.parse_mz("v26.48.0"),
"mz_sinks": MzVersion.parse_mz("v26.38.0"),
"mz_ssh_tunnel_connections": MzVersion.parse_mz("v26.37.0"),
"mz_system_privileges": MzVersion.parse_mz("v26.30.0"),
"mz_tables": MzVersion.parse_mz("v26.39.0"),
"mz_types": None,
"mz_types": MzVersion.parse_mz("v26.48.0"),
"mz_views": MzVersion.parse_mz("v26.39.0"),
}

Expand Down
170 changes: 10 additions & 160 deletions src/adapter/src/catalog/builtin_table_updates.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,15 +14,13 @@ use ipnet::IpNet;
use mz_audit_log::VersionedStorageUsage;
use mz_catalog::SYSTEM_CONN_ID;
use mz_catalog::builtin::{
BuiltinTable, MZ_AGGREGATES, MZ_ARRAY_TYPES, MZ_BASE_TYPES, MZ_CLUSTER_REPLICA_SIZE_INTERNAL,
MZ_CLUSTER_REPLICA_SIZES, MZ_COLUMNS, MZ_EGRESS_IPS, MZ_FUNCTIONS, MZ_INDEX_COLUMNS,
MZ_LICENSE_KEYS, MZ_LIST_TYPES, MZ_MAP_TYPES, MZ_OPERATORS, MZ_PSEUDO_TYPES, MZ_ROLE_AUTH,
MZ_SESSIONS, MZ_STORAGE_USAGE_BY_SHARD, MZ_SUBSCRIPTIONS, MZ_TYPE_PG_METADATA, MZ_TYPES,
MZ_WEBHOOKS_SOURCES,
BuiltinTable, MZ_AGGREGATES, MZ_CLUSTER_REPLICA_SIZE_INTERNAL, MZ_CLUSTER_REPLICA_SIZES,
MZ_COLUMNS, MZ_EGRESS_IPS, MZ_FUNCTIONS, MZ_INDEX_COLUMNS, MZ_LICENSE_KEYS, MZ_OPERATORS,
MZ_ROLE_AUTH, MZ_SESSIONS, MZ_STORAGE_USAGE_BY_SHARD, MZ_SUBSCRIPTIONS, MZ_WEBHOOKS_SOURCES,
};
use mz_catalog::memory::error::Error;
use mz_catalog::memory::objects::{
CatalogItem, DataSourceDesc, Func, Index, Table, TableDataSource, Type,
CatalogItem, DataSourceDesc, Func, Index, Table, TableDataSource,
};
use mz_expr::MirScalarExpr;
use mz_license_keys::ValidatedLicenseKey;
Expand All @@ -31,13 +29,9 @@ use mz_ore::cast::CastFrom;
use mz_ore::collections::CollectionExt;
use mz_persist_client::batch::ProtoBatch;
use mz_repr::adt::array::ArrayDimension;
use mz_repr::adt::mz_acl_item::PrivilegeMap;
use mz_repr::role_id::RoleId;
use mz_repr::{
CatalogItemId, Datum, Diff, GlobalId, ReprColumnType, Row, RowPacker, SqlScalarType,
};
use mz_repr::{CatalogItemId, Datum, Diff, GlobalId, ReprColumnType, Row, SqlScalarType};
use mz_sql::ast::{CreateIndexStatement, Statement};
use mz_sql::catalog::{CatalogType, TypeCategory};
use mz_sql::func::FuncImplCatalogDetails;
use mz_sql::names::SchemaSpecifier;
use mz_sql_parser::ast::display::AstDisplay;
Expand Down Expand Up @@ -127,7 +121,6 @@ impl CatalogState {
diff: Diff,
) -> Vec<BuiltinTableUpdate<&'static BuiltinTable>> {
let entry = self.get_entry(&id);
let oid = entry.oid();
let conn_id = entry.item().conn_id().unwrap_or(&SYSTEM_CONN_ID);
let schema_id = &self
.get_schema(
Expand All @@ -138,8 +131,6 @@ impl CatalogState {
.id;
let name = &entry.name().item;
let owner_id = entry.owner_id();
let privileges_row = self.pack_privilege_array_row(entry.privileges());
let privileges = privileges_row.unpack_first();
let mut updates = match entry.item() {
CatalogItem::Index(index) => self.pack_index_update(id, index, diff),
CatalogItem::Source(source) => {
Expand All @@ -161,19 +152,17 @@ impl CatalogState {
// mz_sinks, mz_kafka_sinks and mz_iceberg_sinks read create_sql
// out of mz_catalog_raw, so there is nothing to pack here.
CatalogItem::Sink(_) => vec![],
CatalogItem::Type(ty) => {
self.pack_type_update(id, oid, schema_id, name, owner_id, privileges, ty, diff)
}
CatalogItem::Func(func) => {
self.pack_func_update(id, schema_id, name, owner_id, func, diff)
}
// Tables, views, materialized views, and metric sinks are exposed
// through materialized views derived from `mz_catalog_raw`, and
// logs and secrets never had builtin-table rows, so none pack a row
// here.
// Tables, views, materialized views, types, and metric sinks are
// exposed through materialized views derived from `mz_catalog_raw`,
// and logs and secrets never had builtin-table rows, so none pack a
// row here.
CatalogItem::Table(_)
| CatalogItem::View(_)
| CatalogItem::MaterializedView(_)
| CatalogItem::Type(_)
| CatalogItem::Log(_)
| CatalogItem::Secret(_)
| CatalogItem::MetricSink(_) => vec![],
Expand Down Expand Up @@ -318,128 +307,6 @@ impl CatalogState {
updates
}

fn pack_type_update(
&self,
id: CatalogItemId,
oid: u32,
schema_id: &SchemaSpecifier,
name: &str,
owner_id: &RoleId,
privileges: Datum,
typ: &Type,
diff: Diff,
) -> Vec<BuiltinTableUpdate<&'static BuiltinTable>> {
let mut out = vec![];

let redacted = typ.create_sql.as_ref().map(|create_sql| {
mz_sql::parse::parse(create_sql)
.unwrap_or_else(|_| panic!("create_sql cannot be invalid: {}", create_sql))
.into_element()
.ast
.to_ast_string_redacted()
});

out.push(BuiltinTableUpdate::row(
&*MZ_TYPES,
Row::pack_slice(&[
Datum::String(&id.to_string()),
Datum::UInt32(oid),
Datum::String(&schema_id.to_string()),
Datum::String(name),
Datum::String(&TypeCategory::from_catalog_type(&typ.details.typ).to_string()),
Datum::String(&owner_id.to_string()),
privileges,
if let Some(create_sql) = &typ.create_sql {
Datum::String(create_sql)
} else {
Datum::Null
},
if let Some(redacted) = &redacted {
Datum::String(redacted)
} else {
Datum::Null
},
]),
diff,
));

let mut row = Row::default();
let mut packer = row.packer();

fn append_modifier(packer: &mut RowPacker<'_>, mods: &[i64]) {
if mods.is_empty() {
packer.push(Datum::Null);
} else {
packer.push_list(mods.iter().map(|m| Datum::Int64(*m)));
}
}

let index_id = match &typ.details.typ {
CatalogType::Array {
element_reference: element_id,
} => {
packer.push(Datum::String(&id.to_string()));
packer.push(Datum::String(&element_id.to_string()));
&MZ_ARRAY_TYPES
}
CatalogType::List {
element_reference: element_id,
element_modifiers,
} => {
packer.push(Datum::String(&id.to_string()));
packer.push(Datum::String(&element_id.to_string()));
append_modifier(&mut packer, element_modifiers);
&MZ_LIST_TYPES
}
CatalogType::Map {
key_reference: key_id,
value_reference: value_id,
key_modifiers,
value_modifiers,
} => {
packer.push(Datum::String(&id.to_string()));
packer.push(Datum::String(&key_id.to_string()));
packer.push(Datum::String(&value_id.to_string()));
append_modifier(&mut packer, key_modifiers);
append_modifier(&mut packer, value_modifiers);
&MZ_MAP_TYPES
}
CatalogType::Pseudo => {
packer.push(Datum::String(&id.to_string()));
&MZ_PSEUDO_TYPES
}
_ => {
packer.push(Datum::String(&id.to_string()));
&MZ_BASE_TYPES
}
};
out.push(BuiltinTableUpdate::row(index_id, row, diff));

if let Some(pg_metadata) = &typ.details.pg_metadata {
// Resolved the same way as a column's wire type, so this matches the
// `type_len` pgwire sends in `RowDescription`. Pseudo-types and arrays
// of them (`_record`) have no scalar type, so their length is unknown.
let typlen =
match mz_sql::plan::scalar_type_from_catalog(&self.for_system_session(), id, &[]) {
Ok(scalar_type) => Datum::Int16(mz_pgrepr::Type::from(&scalar_type).typlen()),
Err(_) => Datum::Null,
};
out.push(BuiltinTableUpdate::row(
&*MZ_TYPE_PG_METADATA,
Row::pack_slice(&[
Datum::String(&id.to_string()),
Datum::UInt32(pg_metadata.typinput_oid),
Datum::UInt32(pg_metadata.typreceive_oid),
Datum::UInt32(pg_metadata.typsend_oid),
typlen,
]),
diff,
));
}

out
}

fn pack_func_update(
&self,
id: CatalogItemId,
Expand Down Expand Up @@ -710,23 +577,6 @@ impl CatalogState {
)
}

fn pack_privilege_array_row(&self, privileges: &PrivilegeMap) -> Row {
let mut row = Row::default();
let flat_privileges: Vec<_> = privileges.all_values_owned().collect();
row.packer()
.try_push_array(
&[ArrayDimension {
lower_bound: 1,
length: flat_privileges.len(),
}],
flat_privileges
.into_iter()
.map(|mz_acl_item| Datum::MzAclItem(mz_acl_item.clone())),
)
.expect("privileges is 1 dimensional, and its length is used for the array length");
row
}

pub fn pack_webhook_source_update(
&self,
item_id: CatalogItemId,
Expand Down
65 changes: 47 additions & 18 deletions src/adapter/src/catalog/open/builtin_schema_migration.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -325,24 +325,6 @@ static MIGRATIONS: LazyLock<Vec<MigrationStep>> = LazyLock::new(|| {
MZ_INTERNAL_SCHEMA,
"mz_kafka_source_tables",
),
// `mz_type_pg_metadata` gained a trailing `typsend` column. See the NOTE
// above: this version must stay at the workspace's current dev version
// until the change ships.
MigrationStep::replacement(
"26.40.0-dev.0",
CatalogItemType::Table,
MZ_INTERNAL_SCHEMA,
"mz_type_pg_metadata",
),
// `mz_type_pg_metadata` gained a trailing `typlen` column. See the NOTE
// above: this version must stay at the workspace's current dev version
// until the change ships.
MigrationStep::replacement(
"26.47.0-dev.0",
CatalogItemType::Table,
MZ_INTERNAL_SCHEMA,
"mz_type_pg_metadata",
),
// Converting the connection-detail builtin tables to materialized views
// changes their catalog fingerprint, so each needs an explicit
// replacement step.
Expand Down Expand Up @@ -552,6 +534,53 @@ static MIGRATIONS: LazyLock<Vec<MigrationStep>> = LazyLock::new(|| {
MZ_INTERNAL_SCHEMA,
"mz_materialized_view_refresh_strategies",
),
// Converting the type relations from builtin tables to materialized
// views over mz_catalog_raw changes their catalog fingerprint, so each
// needs an explicit replacement step. See the NOTE above: this version
// must stay at the workspace's current dev version until the change
// ships.
MigrationStep::replacement(
"26.48.0-dev.0",
CatalogItemType::MaterializedView,
MZ_CATALOG_SCHEMA,
"mz_types",
),
MigrationStep::replacement(
"26.48.0-dev.0",
CatalogItemType::MaterializedView,
MZ_CATALOG_SCHEMA,
"mz_array_types",
),
MigrationStep::replacement(
"26.48.0-dev.0",
CatalogItemType::MaterializedView,
MZ_CATALOG_SCHEMA,
"mz_base_types",
),
MigrationStep::replacement(
"26.48.0-dev.0",
CatalogItemType::MaterializedView,
MZ_CATALOG_SCHEMA,
"mz_list_types",
),
MigrationStep::replacement(
"26.48.0-dev.0",
CatalogItemType::MaterializedView,
MZ_CATALOG_SCHEMA,
"mz_map_types",
),
MigrationStep::replacement(
"26.48.0-dev.0",
CatalogItemType::MaterializedView,
MZ_CATALOG_SCHEMA,
"mz_pseudo_types",
),
MigrationStep::replacement(
"26.48.0-dev.0",
CatalogItemType::MaterializedView,
MZ_INTERNAL_SCHEMA,
"mz_type_pg_metadata",
),
]
});

Expand Down
Loading
Loading