adapter: clean up semantics of CatalogItem::desc/desc_opt (#34294)

Now it only returns a `RelationDesc` for types that are actually
queryable relations, as initially intended.

Author: aljoscha

src/adapter/src/catalog/apply.rs

@@ -759,11 +759,6 @@ impl CatalogState {
                     item_type.details.array_id = Some(item_id);
                 }
 
-                // Assert that no built-in types are record types so that we don't
-                // need to bother to build a description. Only record types need
-                // descriptions.
-                let desc = None;
-                assert!(!matches!(typ.details.typ, CatalogType::Record { .. }));
                 let schema_id = self.resolve_system_schema(typ.schema);
 
                 self.insert_item(
@@ -780,7 +775,6 @@ impl CatalogState {
                         create_sql: None,
                         global_id,
                         details: typ.details.clone(),
-                        desc,
                         resolved_ids: ResolvedIds::empty(),
                     }),
                     MZ_SYSTEM_ROLE_ID,

src/adapter/src/catalog/state.rs

@@ -1419,14 +1419,15 @@ impl CatalogState {
                 cluster_id: in_cluster,
             }),
             Plan::CreateType(CreateTypePlan { typ, .. }) => {
-                let desc = match typ.inner.desc(&session_catalog) {
-                    Ok(desc) => desc,
-                    Err(err) => return Err((err.into(), cached_expr)),
-                };
+                // Even if we don't need the `RelationDesc` here, error out
+                // early and eagerly, as a kind of soft assertion that we _can_
+                // build the `RelationDesc` when needed.
+                if let Err(err) = typ.inner.desc(&session_catalog) {
+                    return Err((err.into(), cached_expr));
+                }
                 CatalogItem::Type(Type {
                     create_sql: Some(typ.create_sql),
                     global_id,
-                    desc,
                     details: CatalogTypeDetails {
                         array_id: None,
                         typ: typ.inner,

src/adapter/src/coord/sequencer/inner.rs

@@ -1479,10 +1479,14 @@ impl Coordinator {
     ) -> Result<ExecuteResponse, AdapterError> {
         let id_ts = self.get_catalog_write_ts().await;
         let (item_id, global_id) = self.catalog().allocate_user_id(id_ts).await?;
+        // Validate the type definition (e.g., composite columns) before storing.
+        plan.typ
+            .inner
+            .desc(&self.catalog().for_session(session))
+            .map_err(AdapterError::from)?;
         let typ = Type {
             create_sql: Some(plan.typ.create_sql),
             global_id,
-            desc: plan.typ.inner.desc(&self.catalog().for_session(session))?,
             details: CatalogTypeDetails {
                 array_id: None,
                 typ: plan.typ.inner,

src/catalog/src/memory/objects.rs

@@ -40,9 +40,9 @@ use mz_sql::ast::{
 };
 use mz_sql::catalog::{
     CatalogClusterReplica, CatalogError as SqlCatalogError, CatalogItem as SqlCatalogItem,
-    CatalogItemType as SqlCatalogItemType, CatalogItemType, CatalogSchema, CatalogTypeDetails,
-    DefaultPrivilegeAclItem, DefaultPrivilegeObject, IdReference, RoleAttributes, RoleMembership,
-    RoleVars, SystemObjectType,
+    CatalogItemType as SqlCatalogItemType, CatalogItemType, CatalogSchema, CatalogType,
+    CatalogTypeDetails, DefaultPrivilegeAclItem, DefaultPrivilegeObject, IdReference,
+    RoleAttributes, RoleMembership, RoleVars, SystemObjectType,
 };
 use mz_sql::names::{
     Aug, CommentObjectId, DatabaseId, DependencyIds, FullItemName, QualifiedItemName,
@@ -1486,7 +1486,6 @@ pub struct Type {
     pub global_id: GlobalId,
     #[serde(skip)]
     pub details: CatalogTypeDetails<IdReference>,
-    pub desc: Option<RelationDesc>,
     /// Other catalog objects referenced by this type.
     pub resolved_ids: ResolvedIds,
 }
@@ -1706,6 +1705,14 @@ impl CatalogItem {
         }
     }
 
+    /// Returns the [`RelationDesc`] for items that yield rows, at the requested
+    /// version.
+    ///
+    /// Some item types honor `version` so callers can ask for the schema that
+    /// matches a specific [`GlobalId`] or historical definition. Other relation
+    /// types ignore `version` because they have a single shape. Non-relational
+    /// items ( for example functions, indexes, sinks, secrets, and connections)
+    /// return [`SqlCatalogError::InvalidDependency`].
     pub fn desc(
         &self,
         name: &FullItemName,
@@ -1727,13 +1734,13 @@ impl CatalogItem {
             CatalogItem::MaterializedView(mview) => {
                 Some(Cow::Owned(mview.desc.at_version(version)))
             }
-            CatalogItem::Type(typ) => typ.desc.as_ref().map(Cow::Borrowed),
             CatalogItem::ContinualTask(ct) => Some(Cow::Borrowed(&ct.desc)),
             CatalogItem::Func(_)
             | CatalogItem::Index(_)
             | CatalogItem::Sink(_)
             | CatalogItem::Secret(_)
-            | CatalogItem::Connection(_) => None,
+            | CatalogItem::Connection(_)
+            | CatalogItem::Type(_) => None,
         }
     }
 
@@ -2434,7 +2441,12 @@ impl CatalogEntry {
 
     /// Reports if the item has columns.
     pub fn has_columns(&self) -> bool {
-        self.desc_opt_latest().is_some()
+        match self.item() {
+            CatalogItem::Type(Type { details, .. }) => {
+                matches!(details.typ, CatalogType::Record { .. })
+            }
+            _ => self.desc_opt_latest().is_some(),
+        }
     }
 
     /// Returns the [`mz_sql::func::Func`] associated with this `CatalogEntry`.

src/sql/src/names.rs

@@ -9,6 +9,7 @@
 
 //! Structured name types for SQL objects.
 
+use std::borrow::Cow;
 use std::collections::{BTreeMap, BTreeSet};
 use std::fmt;
 use std::str::FromStr;
@@ -1761,20 +1762,50 @@ impl<'a> Fold<Raw, Aug> for NameResolver<'a> {
                 print_id: _,
             } => {
                 let item = self.catalog.get_item(id).at_version(*version);
-                let desc = match item.desc(full_name) {
-                    Ok(desc) => desc,
-                    Err(e) => {
-                        if self.status.is_ok() {
-                            self.status = Err(e.into());
+                let name = normalize::column_name(column_name.column.clone());
+                let desc = match item.item_type() {
+                    CatalogItemType::Type => {
+                        let details = item
+                            .type_details()
+                            .expect("type items must carry type details");
+                        match details.typ.desc(self.catalog) {
+                            Ok(Some(desc)) => Cow::Owned(desc),
+                            Ok(None) => {
+                                if self.status.is_ok() {
+                                    self.status = Err(PlanError::TypeWithoutColumns {
+                                        type_name: full_name.clone().into(),
+                                    });
+                                }
+                                return ast::ColumnName {
+                                    relation: ResolvedItemName::Error,
+                                    column: ResolvedColumnReference::Error,
+                                };
+                            }
+                            Err(e) => {
+                                if self.status.is_ok() {
+                                    self.status = Err(e);
+                                }
+                                return ast::ColumnName {
+                                    relation: ResolvedItemName::Error,
+                                    column: ResolvedColumnReference::Error,
+                                };
+                            }
                         }
-                        return ast::ColumnName {
-                            relation: ResolvedItemName::Error,
-                            column: ResolvedColumnReference::Error,
-                        };
                     }
+                    _ => match item.desc(full_name) {
+                        Ok(desc) => desc,
+                        Err(e) => {
+                            if self.status.is_ok() {
+                                self.status = Err(e.into());
+                            }
+                            return ast::ColumnName {
+                                relation: ResolvedItemName::Error,
+                                column: ResolvedColumnReference::Error,
+                            };
+                        }
+                    },
                 };
 
-                let name = normalize::column_name(column_name.column.clone());
                 let Some((index, _typ)) = desc.get_by_name(&name) else {
                     if self.status.is_ok() {
                         let similar = desc.iter_similar_names(&name).cloned().collect();

src/sql/src/plan/error.rs

@@ -70,6 +70,9 @@ pub enum PlanError {
         table: Option<PartialItemName>,
         column: ColumnName,
     },
+    TypeWithoutColumns {
+        type_name: PartialItemName,
+    },
     WrongJoinTypeForLateralColumn {
         table: Option<PartialItemName>,
         column: ColumnName,
@@ -531,6 +534,11 @@ impl fmt::Display for PlanError {
                 "column {} must appear in the GROUP BY clause or be used in an aggregate function",
                 ColumnDisplay { table, column },
             ),
+            Self::TypeWithoutColumns { type_name } => write!(
+                f,
+                "type {} does not have addressable columns",
+                type_name.item.quoted(),
+            ),
             Self::WrongJoinTypeForLateralColumn { table, column } => write!(
                 f,
                 "column {} cannot be referenced from this part of the query: \

src/sql/src/pure.rs

@@ -398,10 +398,20 @@ pub(crate) fn purify_create_sink_avro_doc_on_options(
                 // Attach comment for each column in the item, if the user has
                 // not already provided an overriding `DOC ON` option for the
                 // column.
-                if let Ok(desc) = item.desc(&full_name) {
+                let column_descs: Result<Option<RelationDesc>, PlanError> = match item.item_type() {
+                    CatalogItemType::Type => item
+                        .type_details()
+                        .and_then(|details| details.typ.desc(catalog).transpose())
+                        .transpose(),
+                    _ => item
+                        .desc(&full_name)
+                        .map(|desc| Some(desc.into_owned()))
+                        .map_err(Into::into),
+                };
+
+                if let Ok(Some(desc)) = column_descs {
                     for (pos, column_name) in desc.iter_names().enumerate() {
-                        let comment = comments_map.get(&Some(pos + 1));
-                        if let Some(comment_str) = comment {
+                        if let Some(comment_str) = comments_map.get(&Some(pos + 1)) {
                             let doc_on_column_key = AvroDocOn {
                                 identifier: DocOnIdentifier::Column(ColumnName {
                                     relation: full_resolved_name.clone(),

test/sqllogictest/comment.slt

@@ -378,7 +378,7 @@ CREATE TYPE custom_map_type AS MAP (KEY TYPE = text, VALUE TYPE = custom_type)
 statement ok
 COMMENT ON TYPE custom_map_type IS 'custom_map_type_comment';
 
-statement error cannot be depended upon
+statement error does not have addressable columns
 COMMENT ON COLUMN custom_map_type.key IS 'comment_on_key';
 
 query TT rowsort
@@ -394,10 +394,10 @@ CREATE TYPE custom_list_type AS LIST (ELEMENT TYPE = custom_type)
 statement ok
 COMMENT ON TYPE custom_list_type IS 'custom_list_type_comment';
 
-statement error cannot be depended upon
+statement error does not have addressable columns
 COMMENT ON COLUMN custom_list_type.element IS 'comment_on_element';
 
-statement error cannot be depended upon
+statement error does not have addressable columns
 COMMENT ON COLUMN custom_map_type.key IS 'comment_on_key';
 
 statement ok