diff --git a/misc/python/materialize/checks/all_checks/builtin_version_pin.py b/misc/python/materialize/checks/all_checks/builtin_version_pin.py index a1cacca83b2e2..081bf28175519 100644 --- a/misc/python/materialize/checks/all_checks/builtin_version_pin.py +++ b/misc/python/materialize/checks/all_checks/builtin_version_pin.py @@ -36,7 +36,7 @@ _STABLE_BUILTINS: dict[str, MzVersion | None] = { "mz_array_types": None, "mz_audit_events": MzVersion.parse_mz("v26.33.0"), - "mz_aws_privatelink_connections": None, + "mz_aws_privatelink_connections": MzVersion.parse_mz("v26.37.0"), "mz_base_types": None, "mz_cluster_replica_sizes": None, "mz_cluster_replicas": MzVersion.parse_mz("v26.30.0"), @@ -45,17 +45,17 @@ "mz_default_privileges": MzVersion.parse_mz("v26.30.0"), "mz_egress_ips": None, "mz_functions": None, - "mz_iceberg_sinks": None, + "mz_iceberg_sinks": MzVersion.parse_mz("v26.38.0"), "mz_index_columns": None, - "mz_kafka_connections": None, - "mz_kafka_sinks": None, + "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_operators": None, "mz_pseudo_types": None, - "mz_sinks": None, - "mz_ssh_tunnel_connections": None, + "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": None, "mz_types": None, diff --git a/src/adapter/src/catalog/builtin_table_updates.rs b/src/adapter/src/catalog/builtin_table_updates.rs index 24ea3082ccf43..709cc682ae7f5 100644 --- a/src/adapter/src/catalog/builtin_table_updates.rs +++ b/src/adapter/src/catalog/builtin_table_updates.rs @@ -17,16 +17,16 @@ 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_HISTORY_RETENTION_STRATEGIES, MZ_ICEBERG_SINKS, MZ_INDEX_COLUMNS, MZ_KAFKA_SINKS, - MZ_LICENSE_KEYS, MZ_LIST_TYPES, MZ_MAP_TYPES, MZ_MATERIALIZED_VIEW_REFRESH_STRATEGIES, - MZ_OBJECT_DEPENDENCIES, MZ_OBJECT_GLOBAL_IDS, MZ_OPERATORS, MZ_PSEUDO_TYPES, MZ_REPLACEMENTS, - MZ_ROLE_AUTH, MZ_SESSIONS, MZ_SINKS, MZ_SOURCE_REFERENCES, MZ_STORAGE_USAGE_BY_SHARD, - MZ_SUBSCRIPTIONS, MZ_TABLES, MZ_TYPE_PG_METADATA, MZ_TYPES, MZ_VIEWS, MZ_WEBHOOKS_SOURCES, + MZ_HISTORY_RETENTION_STRATEGIES, MZ_INDEX_COLUMNS, MZ_LICENSE_KEYS, MZ_LIST_TYPES, + MZ_MAP_TYPES, MZ_MATERIALIZED_VIEW_REFRESH_STRATEGIES, MZ_OBJECT_DEPENDENCIES, + MZ_OBJECT_GLOBAL_IDS, MZ_OPERATORS, MZ_PSEUDO_TYPES, MZ_REPLACEMENTS, MZ_ROLE_AUTH, + MZ_SESSIONS, MZ_SOURCE_REFERENCES, MZ_STORAGE_USAGE_BY_SHARD, MZ_SUBSCRIPTIONS, MZ_TABLES, + MZ_TYPE_PG_METADATA, MZ_TYPES, MZ_VIEWS, MZ_WEBHOOKS_SOURCES, }; use mz_catalog::durable::SourceReferences; use mz_catalog::memory::error::Error; use mz_catalog::memory::objects::{ - CatalogEntry, CatalogItem, DataSourceDesc, Func, Index, MaterializedView, Sink, Table, + CatalogEntry, CatalogItem, DataSourceDesc, Func, Index, MaterializedView, Table, TableDataSource, Type, View, }; use mz_expr::MirScalarExpr; @@ -50,7 +50,6 @@ use mz_sql::func::FuncImplCatalogDetails; use mz_sql::names::SchemaSpecifier; use mz_sql_parser::ast::display::AstDisplay; use mz_storage_client::client::TableData; -use mz_storage_types::sinks::{IcebergSinkConnection, KafkaSinkConnection, StorageSinkConnection}; use smallvec::smallvec; // DO NOT add any more imports from `crate` outside of `crate::catalog`. @@ -192,9 +191,9 @@ impl CatalogState { CatalogItem::MaterializedView(mview) => { self.pack_materialized_view_update(id, mview, diff) } - CatalogItem::Sink(sink) => { - self.pack_sink_update(id, oid, schema_id, name, owner_id, sink, diff) - } + // 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) } @@ -514,87 +513,6 @@ impl CatalogState { updates } - fn pack_sink_update( - &self, - id: CatalogItemId, - oid: u32, - schema_id: &SchemaSpecifier, - name: &str, - owner_id: &RoleId, - sink: &Sink, - diff: Diff, - ) -> Vec> { - let mut updates = vec![]; - match &sink.connection { - StorageSinkConnection::Kafka(KafkaSinkConnection { - topic: topic_name, .. - }) => { - updates.push(BuiltinTableUpdate::row( - &*MZ_KAFKA_SINKS, - Row::pack_slice(&[ - Datum::String(&id.to_string()), - Datum::String(topic_name.as_str()), - ]), - diff, - )); - } - StorageSinkConnection::Iceberg(IcebergSinkConnection { - namespace, table, .. - }) => { - updates.push(BuiltinTableUpdate::row( - &*MZ_ICEBERG_SINKS, - Row::pack_slice(&[ - Datum::String(&id.to_string()), - Datum::String(namespace.as_str()), - Datum::String(table.as_str()), - ]), - diff, - )); - } - }; - - let create_stmt = mz_sql::parse::parse(&sink.create_sql) - .unwrap_or_else(|_| panic!("create_sql cannot be invalid: {}", sink.create_sql)) - .into_element() - .ast; - - let envelope = sink.envelope(); - - // The combined format string is used for the deprecated `format` column. - let combined_format = sink.combined_format(); - let (key_format, value_format) = match sink.formats() { - Some((key_format, value_format)) => (key_format, Some(value_format)), - None => (None, None), - }; - - updates.push(BuiltinTableUpdate::row( - &*MZ_SINKS, - Row::pack_slice(&[ - Datum::String(&id.to_string()), - Datum::UInt32(oid), - Datum::String(&schema_id.to_string()), - Datum::String(name), - Datum::String(sink.connection.name()), - Datum::from(sink.connection_id().map(|id| id.to_string()).as_deref()), - // size column now deprecated w/o linked clusters - Datum::Null, - Datum::from(envelope), - // FIXME: These key/value formats are kinda leaky! Should probably live in - // the kafka sink table. - Datum::from(combined_format.as_ref().map(|f| f.as_ref())), - Datum::from(key_format), - Datum::from(value_format), - Datum::String(&sink.cluster_id.to_string()), - Datum::String(&owner_id.to_string()), - Datum::String(&sink.create_sql), - Datum::String(&create_stmt.to_ast_string_redacted()), - ]), - diff, - )); - - updates - } - fn pack_index_update( &self, id: CatalogItemId, diff --git a/src/adapter/src/catalog/open/builtin_schema_migration.rs b/src/adapter/src/catalog/open/builtin_schema_migration.rs index 58b4ec995a88a..6d89904ba0724 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs @@ -101,12 +101,6 @@ static MIGRATIONS: LazyLock> = LazyLock::new(|| { MZ_INTERNAL_SCHEMA, "mz_cluster_replica_metrics_history", ), - MigrationStep::replacement( - "0.160.0", - CatalogItemType::Table, - MZ_CATALOG_SCHEMA, - "mz_sinks", - ), MigrationStep::replacement( "26.18.0-dev.0", CatalogItemType::MaterializedView, @@ -355,6 +349,29 @@ static MIGRATIONS: LazyLock> = LazyLock::new(|| { MZ_CATALOG_SCHEMA, "mz_aws_privatelink_connections", ), + // The three sink tables became materialized views, which moves their + // fingerprints. The old `0.160.0` `mz_sinks` step had to go at the same + // time, because it names the `Table` description and that no longer + // resolves. Nothing is lost: anything that needed the old step upgrades + // from further back than this one, so this one covers it too. + MigrationStep::replacement( + "26.38.0-dev.0", + CatalogItemType::MaterializedView, + MZ_CATALOG_SCHEMA, + "mz_sinks", + ), + MigrationStep::replacement( + "26.38.0-dev.0", + CatalogItemType::MaterializedView, + MZ_CATALOG_SCHEMA, + "mz_kafka_sinks", + ), + MigrationStep::replacement( + "26.38.0-dev.0", + CatalogItemType::MaterializedView, + MZ_CATALOG_SCHEMA, + "mz_iceberg_sinks", + ), ] }); diff --git a/src/adapter/src/catalog/open/builtin_schema_migration_tests.rs b/src/adapter/src/catalog/open/builtin_schema_migration_tests.rs index eeacfea1c459f..787b887e8e6e1 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration_tests.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration_tests.rs @@ -402,3 +402,18 @@ fn init_persist(sim: &mut turmoil::Sim) -> PersistLocation { consensus_uri: "turmoil://consensus:7000".parse().unwrap(), } } + +/// A step naming a builtin that no longer exists, or that changed +/// `CatalogItemType`, panics `validate_migration_steps` at catalog open. That +/// needs a catalog already on disk to reproduce, so the default CI suite never +/// sees it and only the upgrade nightly does. Much cheaper to catch here. +#[mz_ore::test] +fn test_migration_steps_resolve_to_builtins() { + for step in MIGRATIONS.iter() { + assert!( + BUILTIN_LOOKUP.contains_key(&step.object), + "migration step for non-existent builtin: {:?}", + step.object + ); + } +} diff --git a/src/catalog/src/builtin.rs b/src/catalog/src/builtin.rs index 3dfb9e2ae8774..a29b875cf999f 100644 --- a/src/catalog/src/builtin.rs +++ b/src/catalog/src/builtin.rs @@ -1107,11 +1107,11 @@ pub static BUILTINS_STATIC: LazyLock>> = LazyLock::ne Builtin::Log(&MZ_COMPUTE_ERROR_COUNTS_RAW), Builtin::Log(&MZ_COMPUTE_HYDRATION_TIMES_PER_WORKER), Builtin::Log(&MZ_COMPUTE_OPERATOR_HYDRATION_STATUSES_PER_WORKER), - Builtin::Table(&MZ_KAFKA_SINKS), + Builtin::MaterializedView(&MZ_KAFKA_SINKS), Builtin::MaterializedView(&MZ_KAFKA_CONNECTIONS), Builtin::MaterializedView(&MZ_KAFKA_SOURCES), Builtin::Table(&MZ_OBJECT_DEPENDENCIES), - Builtin::Table(&MZ_ICEBERG_SINKS), + Builtin::MaterializedView(&MZ_ICEBERG_SINKS), Builtin::MaterializedView(&MZ_DATABASES), Builtin::MaterializedView(&MZ_SCHEMAS), Builtin::Table(&MZ_COLUMNS), @@ -1125,7 +1125,7 @@ pub static BUILTINS_STATIC: LazyLock>> = LazyLock::ne Builtin::MaterializedView(&MZ_MYSQL_SOURCE_TABLES), Builtin::MaterializedView(&MZ_SQL_SERVER_SOURCE_TABLES), Builtin::MaterializedView(&MZ_KAFKA_SOURCE_TABLES), - Builtin::Table(&MZ_SINKS), + Builtin::MaterializedView(&MZ_SINKS), Builtin::Table(&MZ_VIEWS), Builtin::Table(&MZ_TYPES), Builtin::Table(&MZ_TYPE_PG_METADATA), diff --git a/src/catalog/src/builtin/mz_catalog.rs b/src/catalog/src/builtin/mz_catalog.rs index c2fb6d2a75be2..719501805d533 100644 --- a/src/catalog/src/builtin/mz_catalog.rs +++ b/src/catalog/src/builtin/mz_catalog.rs @@ -198,69 +198,111 @@ pub const TYPE_MZ_ACL_ITEM_ARRAY: BuiltinType = BuiltinType { }, }; -pub static MZ_ICEBERG_SINKS: LazyLock = LazyLock::new(|| BuiltinTable { - name: "mz_iceberg_sinks", - schema: MZ_CATALOG_SCHEMA, - oid: oid::TABLE_MZ_ICEBERG_SINKS_OID, - desc: RelationDesc::builder() - .with_column("id", SqlScalarType::String.nullable(false)) - .with_column("namespace", SqlScalarType::String.nullable(false)) - .with_column("table", SqlScalarType::String.nullable(false)) - .finish(), - column_comments: BTreeMap::from_iter([ - ("id", "The ID of the sink."), - ( - "namespace", - "The namespace of the Iceberg table into which the sink is writing.", - ), - ("table", "The Iceberg table into which the sink is writing."), - ]), - is_retained_metrics_object: false, - access: vec![PUBLIC_SELECT], - ontology: Some(Ontology { - entity_name: "iceberg_sink", - description: "Iceberg-specific sink configuration (namespace, table)", - links: &const { - [OntologyLink { - name: "details_of", - target: "sink", - properties: LinkProperties::fk("id", "id", Cardinality::OneToOne), - }] - }, - column_semantic_types: &[("id", SemanticType::CatalogItemId)], - }), +pub static MZ_ICEBERG_SINKS: LazyLock = LazyLock::new(|| { + BuiltinMaterializedView { + name: "mz_iceberg_sinks", + schema: MZ_CATALOG_SCHEMA, + oid: oid::MV_MZ_ICEBERG_SINKS_OID, + desc: RelationDesc::builder() + .with_column("id", SqlScalarType::String.nullable(false)) + .with_column("namespace", SqlScalarType::String.nullable(false)) + .with_column("table", SqlScalarType::String.nullable(false)) + .finish(), + column_comments: BTreeMap::from_iter([ + ("id", "The ID of the sink."), + ( + "namespace", + "The namespace of the Iceberg table into which the sink is writing.", + ), + ("table", "The Iceberg table into which the sink is writing."), + ]), + // Planning requires `NAMESPACE` and `TABLE`, so both are always there. + sql: " +IN CLUSTER mz_catalog_server +WITH ( + ASSERT NOT NULL id, + ASSERT NOT NULL namespace, + ASSERT NOT NULL \"table\" +) AS +SELECT + mz_internal.parse_catalog_id(data->'key'->'gid') AS id, + parsed->>'namespace' AS namespace, + parsed->>'table' AS \"table\" +FROM + mz_internal.mz_catalog_raw + CROSS JOIN LATERAL ( + SELECT mz_internal.parse_catalog_create_sql(data->'value'->'definition'->'V1'->>'create_sql') + ) AS l(parsed) +WHERE + data->>'kind' = 'Item' AND + parsed->>'sink_type' = 'iceberg'", + is_retained_metrics_object: false, + access: vec![PUBLIC_SELECT], + ontology: Some(Ontology { + entity_name: "iceberg_sink", + description: "Iceberg-specific sink configuration (namespace, table)", + links: &const { + [OntologyLink { + name: "details_of", + target: "sink", + properties: LinkProperties::fk("id", "id", Cardinality::OneToOne), + }] + }, + column_semantic_types: &[("id", SemanticType::CatalogItemId)], + }), + } }); -pub static MZ_KAFKA_SINKS: LazyLock = LazyLock::new(|| BuiltinTable { - name: "mz_kafka_sinks", - schema: MZ_CATALOG_SCHEMA, - oid: oid::TABLE_MZ_KAFKA_SINKS_OID, - desc: RelationDesc::builder() - .with_column("id", SqlScalarType::String.nullable(false)) - .with_column("topic", SqlScalarType::String.nullable(false)) - .with_key(vec![0]) - .finish(), - column_comments: BTreeMap::from_iter([ - ("id", "The ID of the sink."), - ( - "topic", - "The name of the Kafka topic into which the sink is writing.", - ), - ]), - is_retained_metrics_object: false, - access: vec![PUBLIC_SELECT], - ontology: Some(Ontology { - entity_name: "kafka_sink", - description: "Kafka-specific sink configuration (topic)", - links: &const { - [OntologyLink { - name: "details_of", - target: "sink", - properties: LinkProperties::fk("id", "id", Cardinality::OneToOne), - }] - }, - column_semantic_types: &[("id", SemanticType::CatalogItemId)], - }), +pub static MZ_KAFKA_SINKS: LazyLock = LazyLock::new(|| { + BuiltinMaterializedView { + name: "mz_kafka_sinks", + schema: MZ_CATALOG_SCHEMA, + oid: oid::MV_MZ_KAFKA_SINKS_OID, + desc: RelationDesc::builder() + .with_column("id", SqlScalarType::String.nullable(false)) + .with_column("topic", SqlScalarType::String.nullable(false)) + .with_key(vec![0]) + .finish(), + column_comments: BTreeMap::from_iter([ + ("id", "The ID of the sink."), + ( + "topic", + "The name of the Kafka topic into which the sink is writing.", + ), + ]), + // Planning requires `TOPIC`, so it is always there. + sql: " +IN CLUSTER mz_catalog_server +WITH ( + ASSERT NOT NULL id, + ASSERT NOT NULL topic +) AS +SELECT + mz_internal.parse_catalog_id(data->'key'->'gid') AS id, + parsed->>'topic' AS topic +FROM + mz_internal.mz_catalog_raw + CROSS JOIN LATERAL ( + SELECT mz_internal.parse_catalog_create_sql(data->'value'->'definition'->'V1'->>'create_sql') + ) AS l(parsed) +WHERE + data->>'kind' = 'Item' AND + parsed->>'sink_type' = 'kafka'", + is_retained_metrics_object: false, + access: vec![PUBLIC_SELECT], + ontology: Some(Ontology { + entity_name: "kafka_sink", + description: "Kafka-specific sink configuration (topic)", + links: &const { + [OntologyLink { + name: "details_of", + target: "sink", + properties: LinkProperties::fk("id", "id", Cardinality::OneToOne), + }] + }, + column_semantic_types: &[("id", SemanticType::CatalogItemId)], + }), + } }); // Reads Item rows from `mz_catalog_raw`, pulls the broker addresses and any // explicit progress topic out of the persisted `create_sql` via @@ -1224,11 +1266,15 @@ WHERE }); // mz_sources is generated dynamically in BUILTINS_STATIC via builtin::make_mz_sources() // with builtin source/log entries inlined as VALUES. See builtin/builtin.rs. -pub static MZ_SINKS: LazyLock = LazyLock::new(|| { - BuiltinTable { +/// Sink metadata, all of it derived from the persisted `create_sql`. The +/// `CreateSink` arm of `parse_catalog_create_sql` does the parsing, including +/// the deprecated `format` column and its avro/json-only collapse. `size` is +/// deprecated and always NULL. +pub static MZ_SINKS: LazyLock = LazyLock::new(|| { + BuiltinMaterializedView { name: "mz_sinks", schema: MZ_CATALOG_SCHEMA, - oid: oid::TABLE_MZ_SINKS_OID, + oid: oid::MV_MZ_SINKS_OID, desc: RelationDesc::builder() .with_column("id", SqlScalarType::String.nullable(false)) .with_column("oid", SqlScalarType::Oid.nullable(false)) @@ -1294,6 +1340,43 @@ pub static MZ_SINKS: LazyLock = LazyLock::new(|| { "The redacted `CREATE` SQL statement for the sink.", ), ]), + sql: " +IN CLUSTER mz_catalog_server +WITH ( + ASSERT NOT NULL id, + ASSERT NOT NULL oid, + ASSERT NOT NULL schema_id, + ASSERT NOT NULL name, + ASSERT NOT NULL type, + ASSERT NOT NULL cluster_id, + ASSERT NOT NULL owner_id, + ASSERT NOT NULL create_sql, + ASSERT NOT NULL redacted_create_sql +) AS +SELECT + mz_internal.parse_catalog_id(data->'key'->'gid') AS id, + (data->'value'->>'oid')::oid AS oid, + mz_internal.parse_catalog_id(data->'value'->'schema_id') AS schema_id, + data->'value'->>'name' AS name, + parsed->>'sink_type' AS type, + parsed->>'connection_id' AS connection_id, + NULL::text AS size, + parsed->>'envelope_type' AS envelope_type, + parsed->>'format' AS format, + parsed->>'key_format' AS key_format, + parsed->>'value_format' AS value_format, + parsed->>'cluster_id' AS cluster_id, + mz_internal.parse_catalog_id(data->'value'->'owner_id') AS owner_id, + data->'value'->'definition'->'V1'->>'create_sql' AS create_sql, + mz_internal.redact_sql(data->'value'->'definition'->'V1'->>'create_sql') AS redacted_create_sql +FROM + mz_internal.mz_catalog_raw + CROSS JOIN LATERAL ( + SELECT mz_internal.parse_catalog_create_sql(data->'value'->'definition'->'V1'->>'create_sql') + ) AS l(parsed) +WHERE + data->>'kind' = 'Item' AND + parsed->>'type' = 'sink'", is_retained_metrics_object: true, access: vec![PUBLIC_SELECT], ontology: Some(Ontology { diff --git a/src/catalog/src/memory/objects.rs b/src/catalog/src/memory/objects.rs index c422cd79e0148..dd10215d8d8a1 100644 --- a/src/catalog/src/memory/objects.rs +++ b/src/catalog/src/memory/objects.rs @@ -1344,6 +1344,10 @@ impl Sink { } /// Envelope of the sink. + /// + /// NOTE: the `mz_sinks` view works this out from `create_sql` instead, in + /// the `CreateSink` arm of `parse_catalog_create_sql`. Kafka gets it from + /// `ENVELOPE` there, iceberg from `MODE`. Change both or they drift. pub fn envelope(&self) -> Option<&str> { match &self.envelope { SinkEnvelope::Debezium => Some("debezium"), @@ -1356,6 +1360,9 @@ impl Sink { /// if the key-format is none or the key & value formats are /// both the same (either avro or json), we return the value format name, /// otherwise we return a composite name. + /// + /// NOTE: `parse_catalog_create_sql` redoes this collapse for the `mz_sinks` + /// `format` column. Change both or they drift. pub fn combined_format(&self) -> Option> { match &self.connection { StorageSinkConnection::Kafka(connection) => Some(connection.format.get_format_name()), @@ -1364,6 +1371,10 @@ impl Sink { } /// Output distinct key_format and value_format of the sink. + /// + /// NOTE: also derived from `create_sql` for `mz_sinks`. Watch out that a key + /// format exists only when the sink has a `KEY`, so the SQL side has to work + /// that out from the statement for itself. pub fn formats(&self) -> Option<(Option<&str>, &str)> { match &self.connection { StorageSinkConnection::Kafka(connection) => { diff --git a/src/expr/src/scalar/func/impls/jsonb.rs b/src/expr/src/scalar/func/impls/jsonb.rs index 6929ea134192f..f7e29a38173f0 100644 --- a/src/expr/src/scalar/func/impls/jsonb.rs +++ b/src/expr/src/scalar/func/impls/jsonb.rs @@ -19,9 +19,10 @@ use mz_repr::{ArrayRustType, Datum, Row, RowPacker, SqlColumnType, SqlScalarType use mz_sql_parser::ast::display::AstDisplay; use mz_sql_parser::ast::{ AstInfo, AvroSchema, ConnectionOption, ConnectionOptionName, CreateConnectionType, - CreateSubsourceOptionName, Format, FormatSpecifier, KafkaSourceConfigOptionName, - PgConfigOptionName, ProtobufSchema, RawClusterName, RawItemName, SourceEnvelope, - SourceErrorPolicy, UnresolvedItemName, Value, WithOptionValue, + CreateSinkConnection, CreateSubsourceOptionName, Format, FormatSpecifier, + IcebergSinkConfigOptionName, IcebergSinkMode, KafkaSinkConfigOptionName, + KafkaSourceConfigOptionName, PgConfigOptionName, ProtobufSchema, RawClusterName, RawItemName, + SinkEnvelope, SourceEnvelope, SourceErrorPolicy, UnresolvedItemName, Value, WithOptionValue, }; use prost::Message as _; use serde::{Deserialize, Serialize}; @@ -576,7 +577,129 @@ fn parse_catalog_create_sql<'a>(a: &'a str) -> Result { "subsource" } - CreateSink(_) => "sink", + // Everything the mz_sinks, mz_kafka_sinks and mz_iceberg_sinks + // views read. The Rust side of each value lives in `Sink` and + // `StorageSinkConnection`, so those and this have to move together. + // + // NOTE: we bail below if a sink has no resolved `IN CLUSTER`, no + // `TOPIC` on kafka, or no `NAMESPACE`/`TABLE` on iceberg. Planning + // guarantees all four. But if one ever slipped through it would + // take down every view built on this function, not just the sink + // ones. + CreateSink(stmt) => { + let Some(in_cluster) = stmt.in_cluster else { + return Err("missing IN CLUSTER".into()); + }; + info.insert("cluster_id", json!(get_cluster_id(in_cluster)?)); + + match stmt.connection { + CreateSinkConnection::Kafka { + connection, + options, + key: sink_key, + .. + } => { + info.insert("sink_type", json!("kafka")); + info.insert("connection_id", json!(get_item_id(connection)?)); + + let topic = options + .into_iter() + .find(|o| o.name == KafkaSinkConfigOptionName::Topic) + .and_then(|o| o.value.as_ref().and_then(option_string)) + .ok_or("kafka sink missing TOPIC")?; + info.insert("topic", json!(topic)); + + if let Some(envelope) = stmt.envelope { + let envelope_type = match envelope { + SinkEnvelope::Upsert => "upsert", + SinkEnvelope::Debezium => "debezium", + }; + info.insert("envelope_type", json!(envelope_type)); + } + + if let Some(format_spec) = stmt.format { + // A key format only survives if the sink has a + // `KEY`. Without one `kafka_sink_builder` throws + // away the key half of a key/value spec, and does + // not copy a bare spec over to the key either. + let (key_format, value_format) = + match (&format_spec, sink_key.is_some()) { + (FormatSpecifier::Bare(fmt), false) => (None, format_name(fmt)), + (FormatSpecifier::Bare(fmt), true) => { + (Some(format_name(fmt)), format_name(fmt)) + } + (FormatSpecifier::KeyValue { value, .. }, false) => { + (None, format_name(value)) + } + (FormatSpecifier::KeyValue { key, value }, true) => { + (Some(format_name(key)), format_name(value)) + } + }; + if let Some(key_format) = key_format { + info.insert("key_format", json!(key_format)); + } + info.insert("value_format", json!(value_format)); + + // The deprecated combined `format`. Only avro/avro + // and json/json collapse to a single name. + // Everything else, text/text and bytes/bytes + // included, gets the composite form. + let combined = match key_format { + None => value_format.to_string(), + Some(key_format) + if key_format == value_format + && matches!(value_format, "avro" | "json") => + { + value_format.to_string() + } + Some(key_format) => { + format!("key-{key_format}-value-{value_format}") + } + }; + info.insert("format", json!(combined)); + } + } + CreateSinkConnection::Iceberg { + catalog_connection, + options, + .. + } => { + info.insert("sink_type", json!("iceberg")); + // The catalog connection, not the optional AWS one. + info.insert("connection_id", json!(get_item_id(catalog_connection)?)); + + let mut namespace = None; + let mut table = None; + for option in options { + match option.name { + IcebergSinkConfigOptionName::Namespace => { + namespace = option.value.as_ref().and_then(option_string) + } + IcebergSinkConfigOptionName::Table => { + table = option.value.as_ref().and_then(option_string) + } + } + } + info.insert( + "namespace", + json!(namespace.ok_or("iceberg sink missing NAMESPACE")?), + ); + info.insert("table", json!(table.ok_or("iceberg sink missing TABLE")?)); + + // Iceberg spells the envelope `MODE`, and has no format + // columns at all. + if let Some(mode) = stmt.mode { + let envelope_type = match mode { + IcebergSinkMode::Upsert => "upsert", + IcebergSinkMode::Append => "append", + }; + info.insert("envelope_type", json!(envelope_type)); + } + } + } + + "sink" + } CreateIndex(stmt) => { let Some(in_cluster) = stmt.in_cluster else { return Err("missing IN CLUSTER".into()); @@ -1697,4 +1820,180 @@ mod tests { let out = super::parse_catalog_create_sql(sql).expect("ok"); assert_eq!(as_serde(out).get("envelope_type"), None); } + + // --- parse_catalog_create_sql, CreateSink arm ---------------------------- + + const AVRO_FORMAT: &str = "FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION \ + [u12 AS \"materialize\".\"public\".\"csr_conn\"]"; + + /// A persisted kafka-sink `create_sql`: resolved names, and a `TOPIC` that + /// planning guarantees. + fn kafka_sink_sql(key: Option<&str>, format: &str, envelope: &str) -> String { + let key_clause = key.map(|k| format!(" KEY ({k})")).unwrap_or_default(); + format!( + "CREATE SINK \"materialize\".\"public\".\"snk\" \ + IN CLUSTER [u42] \ + FROM [u1 AS \"materialize\".\"public\".\"t\"] \ + INTO KAFKA CONNECTION [u10 AS \"materialize\".\"public\".\"k_conn\"] \ + (TOPIC 'sink-topic'){key_clause} {format} ENVELOPE {envelope}" + ) + } + + fn iceberg_sink_sql(mode: &str) -> String { + format!( + "CREATE SINK \"materialize\".\"public\".\"ice\" \ + IN CLUSTER [u42] \ + FROM [u1 AS \"materialize\".\"public\".\"t\"] \ + INTO ICEBERG CATALOG CONNECTION [u20 AS \"materialize\".\"public\".\"cat_conn\"] \ + (NAMESPACE 'ns', TABLE 'tbl') \ + USING AWS CONNECTION [u21 AS \"materialize\".\"public\".\"aws_conn\"] \ + MODE {mode}" + ) + } + + #[mz_ore::test] + fn sink_kafka_bare_format_without_key() { + let sql = kafka_sink_sql(None, "FORMAT JSON", "DEBEZIUM"); + let out = super::parse_catalog_create_sql(&sql).expect("ok"); + assert_eq!( + as_serde(out), + json!({ + "type": "sink", + "sink_type": "kafka", + "cluster_id": "u42", + "connection_id": "u10", + "topic": "sink-topic", + "envelope_type": "debezium", + "format": "json", + "value_format": "json", + }), + ); + } + + #[mz_ore::test] + fn sink_kafka_bare_format_with_key_derives_key_format() { + // A bare format applies to the key too once the sink has a KEY, which + // is what makes the deprecated `format` column collapse to `avro`. + let sql = kafka_sink_sql(Some("a"), AVRO_FORMAT, "UPSERT"); + let out = super::parse_catalog_create_sql(&sql).expect("ok"); + assert_eq!( + as_serde(out), + json!({ + "type": "sink", + "sink_type": "kafka", + "cluster_id": "u42", + "connection_id": "u10", + "topic": "sink-topic", + "envelope_type": "upsert", + "format": "avro", + "key_format": "avro", + "value_format": "avro", + }), + ); + } + + #[mz_ore::test] + fn sink_kafka_bare_text_format_with_key_does_not_collapse() { + // Only avro/avro and json/json collapse, so a keyed text sink reports + // the composite form even though both halves are `text`. + let sql = kafka_sink_sql(Some("a"), "FORMAT TEXT", "UPSERT"); + let out = super::parse_catalog_create_sql(&sql).expect("ok"); + let out = as_serde(out); + assert_eq!(out["format"], json!("key-text-value-text")); + assert_eq!(out["key_format"], json!("text")); + assert_eq!(out["value_format"], json!("text")); + } + + #[mz_ore::test] + fn sink_kafka_key_value_json_collapses() { + let sql = kafka_sink_sql(Some("a"), "KEY FORMAT JSON VALUE FORMAT JSON", "UPSERT"); + let out = as_serde(super::parse_catalog_create_sql(&sql).expect("ok")); + assert_eq!(out["format"], json!("json")); + assert_eq!(out["key_format"], json!("json")); + assert_eq!(out["value_format"], json!("json")); + } + + #[mz_ore::test] + fn sink_kafka_key_value_mixed_is_composite() { + let sql = kafka_sink_sql(Some("a"), "KEY FORMAT TEXT VALUE FORMAT BYTES", "UPSERT"); + let out = as_serde(super::parse_catalog_create_sql(&sql).expect("ok")); + assert_eq!(out["format"], json!("key-text-value-bytes")); + assert_eq!(out["key_format"], json!("text")); + assert_eq!(out["value_format"], json!("bytes")); + } + + #[mz_ore::test] + fn sink_kafka_key_format_without_key_is_dropped() { + // `kafka_sink_builder` ignores the key half of the format spec when the + // sink has no KEY, so neither `key_format` nor the composite `format` + // may reflect it. + let sql = kafka_sink_sql(None, "KEY FORMAT JSON VALUE FORMAT TEXT", "DEBEZIUM"); + let out = as_serde(super::parse_catalog_create_sql(&sql).expect("ok")); + assert_eq!(out["format"], json!("text")); + assert_eq!(out["key_format"], serde_json::Value::Null); + assert_eq!(out["value_format"], json!("text")); + } + + #[mz_ore::test] + fn sink_kafka_missing_topic_errors() { + let sql = "CREATE SINK \"materialize\".\"public\".\"snk\" \ + IN CLUSTER [u42] \ + FROM [u1 AS \"materialize\".\"public\".\"t\"] \ + INTO KAFKA CONNECTION [u10 AS \"materialize\".\"public\".\"k_conn\"] \ + FORMAT JSON ENVELOPE DEBEZIUM"; + let err = super::parse_catalog_create_sql(sql).unwrap_err(); + assert!( + matches!(err, EvalError::InvalidCatalogJson(msg) if msg.contains("missing TOPIC")), + "wrong error variant/message" + ); + } + + #[mz_ore::test] + fn sink_iceberg_upsert_mode() { + let sql = iceberg_sink_sql("UPSERT"); + let out = super::parse_catalog_create_sql(&sql).expect("ok"); + assert_eq!( + as_serde(out), + json!({ + "type": "sink", + "sink_type": "iceberg", + "cluster_id": "u42", + // The catalog connection, never the AWS connection (u21). + "connection_id": "u20", + "namespace": "ns", + "table": "tbl", + "envelope_type": "upsert", + }), + ); + } + + #[mz_ore::test] + fn sink_iceberg_append_mode() { + let sql = iceberg_sink_sql("APPEND"); + let out = as_serde(super::parse_catalog_create_sql(&sql).expect("ok")); + // `append` is reachable only through an iceberg sink's MODE. + assert_eq!(out["envelope_type"], json!("append")); + } + + #[mz_ore::test] + fn sink_iceberg_missing_table_errors() { + let sql = "CREATE SINK \"materialize\".\"public\".\"ice\" \ + IN CLUSTER [u42] \ + FROM [u1 AS \"materialize\".\"public\".\"t\"] \ + INTO ICEBERG CATALOG CONNECTION [u20 AS \"materialize\".\"public\".\"cat_conn\"] \ + (NAMESPACE 'ns') MODE UPSERT"; + let err = super::parse_catalog_create_sql(sql).unwrap_err(); + assert!( + matches!(err, EvalError::InvalidCatalogJson(msg) if msg.contains("missing TABLE")), + "wrong error variant/message" + ); + } + + #[mz_ore::test] + #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux` + fn sink_arm_leaves_other_item_types_alone() { + let sql = "CREATE VIEW \"materialize\".\"public\".\"v\" AS SELECT 1"; + let out = super::parse_catalog_create_sql(sql).expect("ok"); + assert_eq!(as_serde(out), json!({ "type": "view" })); + } } diff --git a/src/pgrepr-consts/src/oid.rs b/src/pgrepr-consts/src/oid.rs index 9d1c0672e5086..7a4716922990f 100644 --- a/src/pgrepr-consts/src/oid.rs +++ b/src/pgrepr-consts/src/oid.rs @@ -416,7 +416,7 @@ pub const LOG_MZ_MESSAGE_BATCH_COUNTS_SENT_RAW_OID: u32 = 16690; pub const LOG_MZ_MESSAGE_COUNTS_RECEIVED_RAW_OID: u32 = 16691; pub const LOG_MZ_MESSAGE_COUNTS_SENT_RAW_OID: u32 = 16692; pub const LOG_MZ_DATAFLOW_OPERATOR_REACHABILITY_RAW_OID: u32 = 16693; -pub const TABLE_MZ_KAFKA_SINKS_OID: u32 = 16694; +pub const MV_MZ_KAFKA_SINKS_OID: u32 = 16694; pub const MV_MZ_KAFKA_CONNECTIONS_OID: u32 = 16695; pub const MV_MZ_KAFKA_SOURCES_OID: u32 = 16696; pub const MV_MZ_POSTGRES_SOURCES_OID: u32 = 16697; @@ -433,7 +433,7 @@ pub const TABLE_MZ_TABLES_OID: u32 = 16707; pub const MV_MZ_CONNECTIONS_OID: u32 = 16708; pub const MV_MZ_SSH_TUNNEL_CONNECTIONS_OID: u32 = 16709; pub const MV_MZ_SOURCES_OID: u32 = 16710; -pub const TABLE_MZ_SINKS_OID: u32 = 16711; +pub const MV_MZ_SINKS_OID: u32 = 16711; pub const TABLE_MZ_VIEWS_OID: u32 = 16712; pub const MV_MZ_MATERIALIZED_VIEWS_OID: u32 = 16713; pub const TABLE_MZ_TYPES_OID: u32 = 16714; @@ -770,7 +770,7 @@ pub const VIEW_MZ_WALLCLOCK_GLOBAL_LAG_HISTOGRAM_OID: u32 = 17056; pub const MV_MZ_SQL_SERVER_SOURCE_TABLES_OID: u32 = 17057; pub const TABLE_MZ_LICENSE_KEYS_OID: u32 = 17058; pub const TABLE_MZ_ROLE_AUTH_OID: u32 = 17059; -pub const TABLE_MZ_ICEBERG_SINKS_OID: u32 = 17060; +pub const MV_MZ_ICEBERG_SINKS_OID: u32 = 17060; pub const VIEW_MZ_OBJECT_GLOBAL_IDS_OID: u32 = 17061; pub const TABLE_MZ_REPLACEMENTS_OID: u32 = 17062; pub const VIEW_PG_AUTHID_CORE_OID: u32 = 17063; diff --git a/src/storage-types/src/sinks.rs b/src/storage-types/src/sinks.rs index 99226b3483944..694fc42f8c0f0 100644 --- a/src/storage-types/src/sinks.rs +++ b/src/storage-types/src/sinks.rs @@ -186,6 +186,9 @@ impl IntoInlineConnection impl StorageSinkConnection { /// returns an option to not constrain ourselves in the future + /// + /// NOTE: `mz_sinks` digs this out of `create_sql` instead. Note the iceberg + /// case reports the catalog connection, not the optional AWS one. pub fn connection_id(&self) -> Option { use StorageSinkConnection::*; match self { @@ -198,6 +201,9 @@ impl StorageSinkConnection { } /// Returns the name of the sink connection. + /// + /// NOTE: `mz_sinks.type` comes from `create_sql`, not from here, so the two + /// sets of strings have to stay identical. pub fn name(&self) -> &'static str { use StorageSinkConnection::*; match self { @@ -485,6 +491,8 @@ impl KafkaSinkFormatType { } impl KafkaSinkFormat { + /// NOTE: the `mz_sinks` `format` column reimplements this in SQL, in + /// `parse_catalog_create_sql`. Change both or they drift. pub fn get_format_name<'a>(&'a self) -> Cow<'a, str> { // For legacy reasons, if the key-format is none or the key & value formats are // both the same (either avro or json), we return the value format name, diff --git a/test/iceberg/mode-append.td b/test/iceberg/mode-append.td index 3def611add9d1..ae18b97312ab5 100644 --- a/test/iceberg/mode-append.td +++ b/test/iceberg/mode-append.td @@ -40,6 +40,21 @@ MODE APPEND WITH (COMMIT INTERVAL '1s'); +# MODE APPEND is the only way to get envelope_type = append, and the catalog +# relations read it back out of create_sql. +> SELECT type, envelope_type, format, key_format, value_format, size, + connection_id = (SELECT id FROM mz_connections WHERE name = 'append_polaris') + FROM mz_sinks WHERE name = 'append_demo' +iceberg append true + +> SELECT namespace, "table" FROM mz_iceberg_sinks + WHERE id = (SELECT id FROM mz_sinks WHERE name = 'append_demo') +default_namespace append_demo_table + +> SELECT count(*) FROM mz_kafka_sinks + WHERE id = (SELECT id FROM mz_sinks WHERE name = 'append_demo') +0 + # Iceberg sinks commit data asynchronously; wait for at least one commit interval to elapse # before querying the table. $ sleep-is-probably-flaky-i-have-justified-my-need-with-a-comment duration=5s diff --git a/test/sqllogictest/catalog_server_explain.slt b/test/sqllogictest/catalog_server_explain.slt index 300281a804373..ce4e27e26a3c0 100644 --- a/test/sqllogictest/catalog_server_explain.slt +++ b/test/sqllogictest/catalog_server_explain.slt @@ -4756,6 +4756,21 @@ Target cluster: mz_catalog_server EOF +query T multiline +EXPLAIN MATERIALIZED VIEW "mz_catalog"."mz_iceberg_sinks"; +---- +mz_catalog.mz_iceberg_sinks: + →Read mz_internal.mz_catalog_raw + +Source mz_internal.mz_catalog_raw + project=(#2..=#4) + filter=(("Item" = (#0{data} ->> "kind")) AND ("iceberg" = (#1 ->> "sink_type"))) + map=(parse_catalog_create_sql(((((#0{data} -> "value") -> "definition") -> "V1") ->> "create_sql")), parse_catalog_id(((#0{data} -> "key") -> "gid")), (#1{parsed} ->> "namespace"), (#1{parsed} ->> "table")) + +Target cluster: mz_catalog_server + +EOF + query T multiline EXPLAIN MATERIALIZED VIEW "mz_catalog"."mz_indexes"; ---- @@ -4929,6 +4944,21 @@ Target cluster: mz_catalog_server EOF +query T multiline +EXPLAIN MATERIALIZED VIEW "mz_catalog"."mz_kafka_sinks"; +---- +mz_catalog.mz_kafka_sinks: + →Read mz_internal.mz_catalog_raw + +Source mz_internal.mz_catalog_raw + project=(#2, #3) + filter=(("Item" = (#0{data} ->> "kind")) AND ("kafka" = (#1 ->> "sink_type"))) + map=(parse_catalog_create_sql(((((#0{data} -> "value") -> "definition") -> "V1") ->> "create_sql")), parse_catalog_id(((#0{data} -> "key") -> "gid")), (#1{parsed} ->> "topic")) + +Target cluster: mz_catalog_server + +EOF + query T multiline EXPLAIN MATERIALIZED VIEW "mz_catalog"."mz_kafka_sources"; ---- @@ -4981,7 +5011,7 @@ mz_catalog.mz_materialized_views: Project: #4, #0, #5, #1, #6, #2, #9, #8, #3, #7 Map: {=r/s1, s1=r/s1}, "s1" →Arrange (#1{schema_name}, #2{name}) - →Constant (37 rows) + →Constant (40 rows) →Arrange (empty key) (#0{schema_name}) (#0{schema_name}, #1{name}) →Fused with Child Map/Filter/Project Project: #4, #3, #5 @@ -5149,6 +5179,21 @@ Target cluster: mz_catalog_server EOF +query T multiline +EXPLAIN MATERIALIZED VIEW "mz_catalog"."mz_sinks"; +---- +mz_catalog.mz_sinks: + →Read mz_internal.mz_catalog_raw + +Source mz_internal.mz_catalog_raw + project=(#4..=#9, #17, #10..=#15, #2, #16) + filter=(("Item" = (#0{data} ->> "kind")) AND ("sink" = (#3 ->> "type"))) + map=((#0{data} -> "value"), (((#1 -> "definition") -> "V1") ->> "create_sql"), parse_catalog_create_sql(#2), parse_catalog_id(((#0{data} -> "key") -> "gid")), text_to_oid((#1 ->> "oid")), parse_catalog_id((#1 -> "schema_id")), (#1 ->> "name"), (#3{parsed} ->> "sink_type"), (#3{parsed} ->> "connection_id"), (#3{parsed} ->> "envelope_type"), (#3{parsed} ->> "format"), (#3{parsed} ->> "key_format"), (#3{parsed} ->> "value_format"), (#3{parsed} ->> "cluster_id"), parse_catalog_id((#1 -> "owner_id")), redact_sql(#2), null) + +Target cluster: mz_catalog_server + +EOF + query T multiline EXPLAIN MATERIALIZED VIEW "mz_catalog"."mz_sources"; ---- @@ -7946,7 +7991,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_builtin_materialized_views"; ---- Explained Query (fast path): - →Constant (37 rows) + →Constant (40 rows) Target cluster: mz_catalog_server diff --git a/test/sqllogictest/information_schema_tables.slt b/test/sqllogictest/information_schema_tables.slt index 8ea3e4fdb71a7..3f1d1a0dca450 100644 --- a/test/sqllogictest/information_schema_tables.slt +++ b/test/sqllogictest/information_schema_tables.slt @@ -150,7 +150,7 @@ BASE TABLE materialize mz_catalog mz_iceberg_sinks -BASE TABLE +MATERIALIZED VIEW materialize mz_catalog mz_index_columns @@ -166,7 +166,7 @@ MATERIALIZED VIEW materialize mz_catalog mz_kafka_sinks -BASE TABLE +MATERIALIZED VIEW materialize mz_catalog mz_kafka_sources @@ -230,7 +230,7 @@ MATERIALIZED VIEW materialize mz_catalog mz_sinks -BASE TABLE +MATERIALIZED VIEW materialize mz_catalog mz_sources diff --git a/test/sqllogictest/mz_sinks.slt b/test/sqllogictest/mz_sinks.slt new file mode 100644 index 0000000000000..71ffe8123bcaa --- /dev/null +++ b/test/sqllogictest/mz_sinks.slt @@ -0,0 +1,212 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +# Behavioural tests for mz_catalog.mz_sinks, mz_catalog.mz_kafka_sinks +# and mz_catalog.mz_iceberg_sinks. All three are materialized views over +# mz_internal.mz_catalog_raw and get every sink detail out of the +# persisted create_sql, in the CreateSink arm of parse_catalog_create_sql +# (src/expr/src/scalar/func/impls/jsonb.rs). +# +# Note we can't create a sink here at all: purification always talks to +# the broker or the iceberg catalog, and there is no VALIDATE = false to +# skip it. So the views stay empty and what we pin is the column shape, +# the item type, the non-sink exclusion, and the JSON keys the view SQL +# expects back from the helper. Real sinks are covered in +# test/testdrive/mz-sinks.td, the kafka-*-sinks.td files and +# test/iceberg/mode-append.td. + +mode cockroach + +reset-server + +# --auto-index-selects wraps the parse_catalog_create_sql calls below in +# indexed views, and that helper counts as an unstable dependency. +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET unsafe_enable_unstable_dependencies = true +---- +COMPLETE 0 + +# --- Schema ------------------------------------------------------------------- + +# 15 columns, canonical order, types and nullability. `size` is deprecated +# and always NULL, but it still has to be here. +query TTT +SELECT c.name, c.type, c.nullable::text +FROM mz_catalog.mz_columns c +JOIN mz_catalog.mz_objects o ON c.id = o.id +JOIN mz_catalog.mz_schemas s ON o.schema_id = s.id +WHERE o.name = 'mz_sinks' AND s.name = 'mz_catalog' +ORDER BY c.position +---- +id text false +oid oid false +schema_id text false +name text false +type text false +connection_id text true +size text true +envelope_type text true +format text true +key_format text true +value_format text true +cluster_id text false +owner_id text false +create_sql text false +redacted_create_sql text false + +query TTT +SELECT c.name, c.type, c.nullable::text +FROM mz_catalog.mz_columns c +JOIN mz_catalog.mz_objects o ON c.id = o.id +JOIN mz_catalog.mz_schemas s ON o.schema_id = s.id +WHERE o.name = 'mz_kafka_sinks' AND s.name = 'mz_catalog' +ORDER BY c.position +---- +id text false +topic text false + +query TTT +SELECT c.name, c.type, c.nullable::text +FROM mz_catalog.mz_columns c +JOIN mz_catalog.mz_objects o ON c.id = o.id +JOIN mz_catalog.mz_schemas s ON o.schema_id = s.id +WHERE o.name = 'mz_iceberg_sinks' AND s.name = 'mz_catalog' +ORDER BY c.position +---- +id text false +namespace text false +table text false + +# All three are materialized views, not tables. +query TT +SELECT o.name, o.type +FROM mz_catalog.mz_objects o +JOIN mz_catalog.mz_schemas s ON o.schema_id = s.id +WHERE s.name = 'mz_catalog' + AND o.name IN ('mz_sinks', 'mz_kafka_sinks', 'mz_iceberg_sinks') +ORDER BY o.name +---- +mz_iceberg_sinks materialized-view +mz_kafka_sinks materialized-view +mz_sinks materialized-view + +# The retained-metrics index over mz_sinks survives the conversion. +query I +SELECT count(*) FROM mz_catalog.mz_indexes WHERE name = 'mz_sinks_ind' +---- +1 + +# --- Invariants --------------------------------------------------------------- + +# id and oid are both declared keys. +query I +SELECT count(*) - count(DISTINCT id) + count(*) - count(DISTINCT oid) +FROM mz_catalog.mz_sinks +---- +0 + +# `size` is deprecated and hardcoded NULL. +query I +SELECT count(*) FROM mz_catalog.mz_sinks WHERE size IS NOT NULL +---- +0 + +# The detail views partition mz_sinks by type, and every detail row joins back. +query I +SELECT + (SELECT count(*) FROM mz_catalog.mz_kafka_sinks) + - (SELECT count(*) FROM mz_catalog.mz_sinks WHERE type = 'kafka') ++ (SELECT count(*) FROM mz_catalog.mz_iceberg_sinks) + - (SELECT count(*) FROM mz_catalog.mz_sinks WHERE type = 'iceberg') +---- +0 + +query I +SELECT count(*) FROM mz_catalog.mz_kafka_sinks ks +LEFT JOIN mz_catalog.mz_sinks s ON ks.id = s.id +WHERE s.id IS NULL +---- +0 + +# --- Non-sink items are excluded ---------------------------------------------- + +statement ok +CREATE TABLE t (a int) + +statement ok +CREATE VIEW v AS SELECT * FROM t + +statement ok +CREATE MATERIALIZED VIEW mv AS SELECT * FROM t + +statement ok +CREATE SOURCE lg_src IN CLUSTER quickstart FROM LOAD GENERATOR COUNTER + +# No sinks are possible here, so all three stay empty even with other +# item types around. +query III +SELECT + (SELECT count(*) FROM mz_catalog.mz_sinks), + (SELECT count(*) FROM mz_catalog.mz_kafka_sinks), + (SELECT count(*) FROM mz_catalog.mz_iceberg_sinks) +---- +0 0 0 + +statement ok +DROP SOURCE lg_src + +# --- The JSON keys the view SQL reads ----------------------------------------- + +# The views read `parsed->>''`, so renaming a key on the helper side +# NULLs out a column instead of failing anything. These two calls pin the +# key names, against create_sql in the shape we actually persist. + +query TTTTTTTTT +SELECT + parsed->>'type', + parsed->>'sink_type', + parsed->>'cluster_id', + parsed->>'connection_id', + parsed->>'topic', + parsed->>'envelope_type', + parsed->>'format', + parsed->>'key_format', + parsed->>'value_format' +FROM ( + SELECT mz_internal.parse_catalog_create_sql( + 'CREATE SINK "materialize"."public"."snk" IN CLUSTER [u5] ' + || 'FROM [u1 AS "materialize"."public"."t"] ' + || 'INTO KAFKA CONNECTION [u2 AS "materialize"."public"."kc"] (TOPIC ''topic-x'') ' + || 'KEY (a) FORMAT JSON ENVELOPE UPSERT' + ) AS parsed +) +---- +sink kafka u5 u2 topic-x upsert json json json + +query TTTTTTTTT +SELECT + parsed->>'type', + parsed->>'sink_type', + parsed->>'cluster_id', + parsed->>'connection_id', + parsed->>'namespace', + parsed->>'table', + parsed->>'envelope_type', + parsed->>'format', + parsed->>'value_format' +FROM ( + SELECT mz_internal.parse_catalog_create_sql( + 'CREATE SINK "materialize"."public"."ice" IN CLUSTER [u5] ' + || 'FROM [u1 AS "materialize"."public"."t"] ' + || 'INTO ICEBERG CATALOG CONNECTION [u3 AS "materialize"."public"."cc"] ' + || '(NAMESPACE ''ns'', TABLE ''tbl'') MODE APPEND' + ) AS parsed +) +---- +sink iceberg u5 u3 ns tbl append NULL NULL diff --git a/test/testdrive-old-kafka-src-syntax/tables.td b/test/testdrive-old-kafka-src-syntax/tables.td index f90b6707dd0ae..cbbd8bb183b48 100644 --- a/test/testdrive-old-kafka-src-syntax/tables.td +++ b/test/testdrive-old-kafka-src-syntax/tables.td @@ -257,8 +257,11 @@ name type cluster comment ! INSERT INTO view VALUES (1) contains:cannot insert into view 'materialize.public.view' -! INSERT INTO mz_kafka_sinks VALUES ('bad', 'bad') -contains:cannot insert into system table 'mz_catalog.mz_kafka_sinks' +# Pick a target that will stay a system table: the builtin schema migration +# validator refuses to convert mz_storage_usage_by_shard, so this keeps +# testing the system-table path rather than some other error. +! INSERT INTO mz_internal.mz_storage_usage_by_shard VALUES (1, 'bad', 1, now()) +contains:cannot insert into system table 'mz_internal.mz_storage_usage_by_shard' > CREATE TABLE j (time TIMESTAMP NOT NULL); diff --git a/test/testdrive/catalog.td b/test/testdrive/catalog.td index eef5713d4fc51..b49f9ea805fee 100644 --- a/test/testdrive/catalog.td +++ b/test/testdrive/catalog.td @@ -516,15 +516,12 @@ mz_cluster_replica_sizes "" mz_columns "" mz_egress_ips "" mz_functions "" -mz_iceberg_sinks "" mz_index_columns "" -mz_kafka_sinks "" mz_list_types "" mz_map_types "" mz_operators "" mz_pseudo_types "" mz_role_auth "" -mz_sinks "" mz_tables "" mz_types "" mz_views "" @@ -549,14 +546,17 @@ mz_cluster_replicas mz_catalog_server "" mz_connections mz_catalog_server "" mz_databases mz_catalog_server "" mz_default_privileges mz_catalog_server "" +mz_iceberg_sinks mz_catalog_server "" mz_indexes mz_catalog_server "" mz_kafka_connections mz_catalog_server "" +mz_kafka_sinks mz_catalog_server "" mz_kafka_sources mz_catalog_server "" mz_role_members mz_catalog_server "" mz_role_parameters mz_catalog_server "" mz_roles mz_catalog_server "" mz_schemas mz_catalog_server "" mz_secrets mz_catalog_server "" +mz_sinks mz_catalog_server "" mz_sources mz_catalog_server "" mz_ssh_tunnel_connections mz_catalog_server "" mz_system_privileges mz_catalog_server "" @@ -830,7 +830,7 @@ test_table "" # `SHOW TABLES` and `mz_tables` should agree. > SELECT COUNT(*) FROM mz_tables WHERE id LIKE 's%' -34 +31 # There is one entry in mz_indexes for each field_number/expression of the index. > SELECT COUNT(id) FROM mz_indexes WHERE id LIKE 's%' diff --git a/test/testdrive/mz-sinks.td b/test/testdrive/mz-sinks.td index 84309b3a42851..db723ed449439 100644 --- a/test/testdrive/mz-sinks.td +++ b/test/testdrive/mz-sinks.td @@ -40,6 +40,19 @@ $ set-arg-default single-replica-cluster=quickstart FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn ENVELOPE UPSERT; +# Only avro/avro and json/json collapse to a shared name, so a keyed sink +# with a bare TEXT format has to report the composite form. +> CREATE MATERIALIZED VIEW mz_sinks_names AS + SELECT name FROM mz_sinks_table GROUP BY name; + +> CREATE SINK mz_sinks_text + IN CLUSTER ${arg.single-replica-cluster} + FROM mz_sinks_names + INTO KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-mz-sinks-text-${testdrive.seed}') + KEY (name) + FORMAT TEXT + ENVELOPE UPSERT; + > SELECT envelope_type FROM mz_sinks WHERE name = 'mz_sinks_debezium' debezium @@ -48,3 +61,24 @@ upsert > SELECT format FROM mz_sinks WHERE name = 'mz_sinks_upsert' avro + +# Once the sink has a KEY, a bare FORMAT applies to the key as well. +> SELECT key_format, value_format FROM mz_sinks WHERE name = 'mz_sinks_upsert' +avro avro + +> SELECT format, key_format, value_format FROM mz_sinks WHERE name = 'mz_sinks_text' +key-text-value-text text text + +# The rest of the columns, plus the kafka detail relation. +> SELECT type, size, connection_id = (SELECT id FROM mz_connections WHERE name = 'kafka_conn'), + cluster_id = (SELECT id FROM mz_clusters WHERE name = '${arg.single-replica-cluster}') + FROM mz_sinks WHERE name = 'mz_sinks_upsert' +kafka true true + +> SELECT topic FROM mz_kafka_sinks + WHERE id = (SELECT id FROM mz_sinks WHERE name = 'mz_sinks_upsert') +testdrive-mz-upsert-debezium-${testdrive.seed} + +> SELECT count(*) FROM mz_iceberg_sinks + WHERE id IN (SELECT id FROM mz_sinks WHERE name LIKE 'mz_sinks_%') +0 diff --git a/test/testdrive/tables.td b/test/testdrive/tables.td index 82081855a7833..4cc53a8bee3d6 100644 --- a/test/testdrive/tables.td +++ b/test/testdrive/tables.td @@ -266,8 +266,11 @@ name type cluster comment ! INSERT INTO view VALUES (1) contains:cannot insert into view 'materialize.public.view' -! INSERT INTO mz_kafka_sinks VALUES ('bad', 'bad') -contains:cannot insert into system table 'mz_catalog.mz_kafka_sinks' +# Pick a target that will stay a system table: the builtin schema migration +# validator refuses to convert mz_storage_usage_by_shard, so this keeps +# testing the system-table path rather than some other error. +! INSERT INTO mz_internal.mz_storage_usage_by_shard VALUES (1, 'bad', 1, now()) +contains:cannot insert into system table 'mz_internal.mz_storage_usage_by_shard' > CREATE TABLE j (time TIMESTAMP NOT NULL);