diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index 8304565a96..117fb8c855 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -418,6 +418,22 @@ export interface paths { patch?: never; trace?: never; }; + "/api/control/locations/{location_id}/targets/{target}/nodes": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get: operations["listPersonCatalogNodes"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/control/locations/{location_id}/targets/{target}/nodes/{node_key}/sql": { parameters: { query?: never; @@ -956,6 +972,31 @@ export interface components { /** Format: uuid */ run_id: string; }; + /** @description A fingerprint row's build identity; the `sqb` run that wrote it, not a scheduler run. */ + CatalogNodeBuild: { + /** Format: date-time */ + observed_at: string; + run_id: string; + version_hash: string; + }; + /** @description A current snapshot node's declared metadata and latest build, without SQL, source or config. */ + CatalogNodeRecord: { + column_count?: number | null; + /** @description Keys of the nodes this node reads. */ + dependencies: string[]; + description?: string | null; + /** @description The snapshot node kind, such as `model`, `seed`, `source` or `audit`. */ + kind: string; + last_build?: null | components["schemas"]["CatalogNodeBuild"]; + materialization?: string | null; + name: string; + node_key: string; + path?: string | null; + relation?: null | components["schemas"]["CatalogRelation"]; + /** Format: uuid */ + snapshot_id: string; + tags: string[]; + }; CatalogObservation: { /** Format: uuid */ event_id: string; @@ -967,6 +1008,11 @@ export interface components { next_after?: null | components["schemas"]["WarehouseMirrorCursor"]; records: components["schemas"]["CatalogObservation"][]; }; + CatalogRelation: { + database?: string | null; + name: string; + schema: string; + }; /** @description SSE invalidation topics; `all` means refetch the authorized REST resources. */ ChangeNotice: { topic: string; @@ -3080,6 +3126,44 @@ export interface operations { }; }; }; + listPersonCatalogNodes: { + parameters: { + query?: never; + header?: never; + path: { + location_id: string; + target: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Every node of the current snapshot; empty before the first publication */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["CatalogNodeRecord"][]; + }; + }; + /** @description Session rejected */ + 401: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + 403: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ApiErrorBody"]; + }; + }; + }; + }; readPersonNodeSql: { parameters: { query?: never; diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs index be80c63098..f40bb3d87d 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs @@ -101,6 +101,10 @@ pub(crate) fn person_routes() -> Router { "/api/control/locations/{location_id}/targets/{target}/runs/{run_id}/nodes", get(crate::api::_helpers::resources::person_reads::run_nodes), ) + .route( + "/api/control/locations/{location_id}/targets/{target}/nodes", + get(crate::api::_helpers::resources::person_reads::nodes), + ) .route( "/api/control/locations/{location_id}/targets/{target}/jobs", get(crate::api::_helpers::resources::person_reads::jobs), diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs index 3a7d8f6d34..79fbb670ad 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs @@ -1,4 +1,4 @@ -//! Person reads of run nodes, jobs, triggers and ticks; refusal precedes every lookup, so unknown objects look forbidden. +//! Person reads of run nodes, catalog nodes, jobs, triggers and ticks; refusal precedes every lookup, so unknown objects look forbidden. use axum::extract::{Path, Query, State}; use axum::http::StatusCode; @@ -16,6 +16,8 @@ use crate::runs::models::RunNodeRecord; use crate::scheduler::main::list_person_ticks::list_person_ticks; use crate::scheduler::models::TickRecord; use crate::store::models::ControlStore; +use crate::warehouse_metadata::main::list_person_catalog_nodes::list_person_catalog_nodes; +use crate::warehouse_metadata::models::CatalogNodeRecord; #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/runs/{run_id}/nodes", operation_id = "listPersonRunNodes", params(("location_id" = Uuid, Path), ("target" = String, Path), ("run_id" = Uuid, Path)), @@ -40,6 +42,28 @@ pub(crate) async fn run_nodes( .ok_or_else(|| not_found("run not found")) } +#[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/nodes", operation_id = "listPersonCatalogNodes", + params(("location_id" = Uuid, Path), ("target" = String, Path)), + responses((status = 200, body = Vec, description = "Every node of the current snapshot; empty before the first publication"), + (status = 401, description = "Session rejected"), + (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))] +pub(crate) async fn nodes( + (State(store), Extension(context)): (State, Extension), + Extension(default_role): Extension, + Path((location_id, target)): Path<(Uuid, String)>, + Extension(AuthenticatedPerson(principal, _)): Extension, +) -> Result>, ApiError> { + Ok(Json( + list_person_catalog_nodes( + &store, + (&context, principal), + (location_id, &target), + default_role, + ) + .await?, + )) +} + #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/jobs", operation_id = "listPersonJobs", params(("location_id" = Uuid, Path), ("target" = String, Path)), responses((status = 200, body = Vec), (status = 401, description = "Session rejected"), diff --git a/crates/sqlbuild-orchestration/src/api/models.rs b/crates/sqlbuild-orchestration/src/api/models.rs index d52b2bac13..530893c084 100644 --- a/crates/sqlbuild-orchestration/src/api/models.rs +++ b/crates/sqlbuild-orchestration/src/api/models.rs @@ -21,6 +21,7 @@ use crate::api::_helpers::resources::{ person_operations::change_trigger, person_operations::recent_runs, person_reads::run_nodes, + person_reads::nodes, person_reads::jobs, person_reads::triggers, person_reads::ticks, diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/catalog_nodes.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/catalog_nodes.rs new file mode 100644 index 0000000000..a4b23af9ff --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/catalog_nodes.rs @@ -0,0 +1,80 @@ +//! Catalog listing rows from snapshot node definitions and each node's latest build. + +use serde_json::Value; +use uuid::Uuid; + +use crate::sensors::constants::SOURCE_NODE_KIND; +use crate::snapshots::models::SnapshotNode; +use crate::warehouse_metadata::models::{CatalogNodeBuild, CatalogNodeRecord, CatalogRelation}; + +/// A declared value from the definition itself, its schema entry or its config, in that order. +fn declared<'a>(definition: &'a Value, key: &str) -> Option<&'a Value> { + [ + definition.get(key), + definition.get("schema").and_then(|entry| entry.get(key)), + definition.get("config").and_then(|config| config.get(key)), + ] + .into_iter() + .flatten() + .find(|value| !value.is_null()) +} + +fn texts(value: Option<&Value>) -> Vec { + match value { + Some(Value::String(text)) => vec![text.clone()], + Some(Value::Array(items)) => items + .iter() + .filter_map(|item| item.as_str().map(str::to_owned)) + .collect(), + _ => Vec::new(), + } +} + +fn text(value: Option<&Value>) -> Option { + value.and_then(Value::as_str).map(str::to_owned) +} + +fn relation(node: &SnapshotNode) -> Option { + let definition = &node.definition; + let (relation, name_key) = match definition.get("destination") { + Some(destination) if !destination.is_null() => (destination, "name"), + _ if node.kind == SOURCE_NODE_KIND => (definition, "table"), + _ => return None, + }; + Some(CatalogRelation { + database: text(relation.get("database")), + schema: text(relation.get("schema"))?, + name: text(relation.get(name_key))?, + }) +} + +pub(crate) fn catalog_node( + (node_key, node): (&str, &SnapshotNode), + snapshot_id: Uuid, + last_build: Option, +) -> CatalogNodeRecord { + let definition = &node.definition; + CatalogNodeRecord { + node_key: node_key.into(), + kind: node.kind.clone(), + name: text(definition.get("name")).unwrap_or_else(|| { + node_key + .split_once(':') + .map_or(node_key, |(_, name)| name) + .into() + }), + path: text(definition.get("path")), + materialization: text(definition.get("materialization")), + description: text(declared(definition, "description")), + tags: texts(declared(definition, "tags")), + dependencies: texts(definition.get("deps")), + relation: relation(node), + column_count: match definition.get("columns") { + Some(Value::Array(columns)) => Some(columns.len()), + Some(Value::Object(columns)) => Some(columns.len()), + _ => None, + }, + last_build, + snapshot_id, + } +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/fingerprint_reads.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/fingerprint_reads.rs index dd3bd789e8..8b77655a04 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/fingerprint_reads.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/fingerprint_reads.rs @@ -1,4 +1,4 @@ -//! One indexed exact-node/destination read, shared by evidence and definition comparison. +//! Indexed exact-node/destination reads, shared by evidence, definition comparison and the catalog. use uuid::Uuid; @@ -7,50 +7,109 @@ use crate::store::main::{decode_row::decode_row, query::query}; use crate::store::models::{ControlTransaction, Parameter}; use crate::warehouse_metadata::models::{FingerprintObservation, FingerprintQuery}; +/// Nodes read per statement; well under both stores' bound-parameter limits. +const NODES_PER_STATEMENT: usize = 100; + +type LatestBuild = Option<(FingerprintObservation, String)>; + pub(crate) async fn latest_fingerprint( transaction: ControlTransaction, (location_id, target, options): (Uuid, &str, &FingerprintQuery), -) -> Result<(ControlTransaction, Option<(FingerprintObservation, String)>), ControlError> { - if options.node_type.trim().is_empty() || options.node_name.trim().is_empty() { +) -> Result<(ControlTransaction, LatestBuild), ControlError> { + let (transaction, mut builds) = latest_fingerprints( + transaction, + (location_id, target, std::slice::from_ref(options)), + ) + .await?; + Ok((transaction, builds.pop().flatten())) +} + +/// Each query's newest build, in order; NULL-namespace shapes share exact-seek statements. +pub(crate) async fn latest_fingerprints( + mut transaction: ControlTransaction, + (location_id, target, queries): (Uuid, &str, &[FingerprintQuery]), +) -> Result<(ControlTransaction, Vec), ControlError> { + if queries + .iter() + .any(|options| options.node_type.trim().is_empty() || options.node_name.trim().is_empty()) + { return Err(ControlError::Invalid( "fingerprint reads require an exact node".into(), )); } - let mut parameters = vec![ - Parameter::Id(location_id), - Parameter::Text(target.into()), - Parameter::Text(options.node_type.clone()), - Parameter::Text(options.node_name.clone()), - ]; - let mut predicates: Vec = Vec::new(); - for (column, value) in [ - ("target_database", &options.database), - ("target_schema", &options.schema), - ("target_name", &options.name), - ] { - predicates.push(if let Some(value) = value { - parameters.push(Parameter::Text(value.clone())); - format!("{column}=${}", parameters.len()) - } else { - format!("{column} IS NULL") - }); + let mut builds: Vec = vec![None; queries.len()]; + let namespace = |options: &FingerprintQuery| { + [ + options.database.is_some(), + options.schema.is_some(), + options.name.is_some(), + ] + }; + let mut shapes: Vec<[bool; 3]> = queries.iter().map(namespace).collect(); + shapes.sort_unstable(); + shapes.dedup(); + for shape in shapes { + let positions: Vec = (0..queries.len()) + .filter(|position| namespace(&queries[*position]) == shape) + .collect(); + for chunk in positions.chunks(NODES_PER_STATEMENT) { + let mut parameters = vec![Parameter::Id(location_id), Parameter::Text(target.into())]; + let mut rows: Vec = Vec::with_capacity(chunk.len()); + for position in chunk { + let options = &queries[*position]; + let first = parameters.len() + 1; + parameters.push(Parameter::Integer(i64::try_from(*position).map_err( + |_| ControlError::Invalid("too many fingerprint reads".into()), + )?)); + parameters.push(Parameter::Text(options.node_type.clone())); + parameters.push(Parameter::Text(options.node_name.clone())); + for value in [&options.database, &options.schema, &options.name] + .into_iter() + .flatten() + { + parameters.push(Parameter::Text(value.clone())); + } + rows.push(format!( + "({})", + (first..=parameters.len()) + .map(|index| format!("${index}")) + .collect::>() + .join(",") + )); + } + let mut column = 3; + let predicates: Vec = ["target_database", "target_schema", "target_name"] + .into_iter() + .zip(shape) + .map(|(name, present)| { + if present { + column += 1; + format!("g.{name}=w.column{column}") + } else { + format!("g.{name} IS NULL") + } + }) + .collect(); + let found; + (transaction, found) = query(transaction, + &format!("SELECT w.column1 AS position,f.event_id,f.observed_at,f.source_event_id,f.record_json ->> 'run_id' AS run_id, \ + f.record_json ->> 'definition_hash' AS definition_hash,f.record_json ->> 'version_hash' AS version_hash, \ + f.record_json ->> 'schema_fingerprint' AS schema_fingerprint,f.record_json ->> 'metadata_json' AS metadata_json \ + FROM (VALUES {}) AS w JOIN sqb_warehouse_fingerprints f ON f.event_id=(SELECT g.event_id FROM sqb_warehouse_fingerprints g \ + WHERE g.location_id=$1 AND g.target=$2 AND g.node_type=w.column2 AND g.node_name=w.column3 AND {} \ + ORDER BY g.target_database,g.target_schema,g.target_name,g.observed_at DESC,g.event_id DESC LIMIT 1)", + rows.join(","), predicates.join(" AND ")), parameters).await?; + for mut row in found { + let position: usize = + serde_json::from_value(row.remove("position").unwrap_or_default()) + .map_err(|error| ControlError::Invalid(error.to_string()))?; + let metadata = row + .remove("metadata_json") + .and_then(|value| value.as_str().map(str::to_owned)) + .unwrap_or_default(); + builds[position] = Some((decode_row(row)?, metadata)); + } + } } - let (transaction, rows) = query(transaction, - &format!("SELECT event_id,observed_at,source_event_id,record_json ->> 'run_id' AS run_id, \ - record_json ->> 'definition_hash' AS definition_hash,record_json ->> 'version_hash' AS version_hash, \ - record_json ->> 'schema_fingerprint' AS schema_fingerprint,record_json ->> 'metadata_json' AS metadata_json FROM sqb_warehouse_fingerprints \ - WHERE location_id=$1 AND target=$2 AND node_type=$3 AND node_name=$4 AND {} \ - ORDER BY target_database,target_schema,target_name,observed_at DESC,event_id DESC LIMIT 1", predicates.join(" AND ")), parameters).await?; - let evidence = rows - .into_iter() - .next() - .map(|mut row| { - let metadata = row - .remove("metadata_json") - .and_then(|value| value.as_str().map(str::to_owned)) - .unwrap_or_default(); - Ok::<_, ControlError>((decode_row(row)?, metadata)) - }) - .transpose()?; - Ok((transaction, evidence)) + Ok((transaction, builds)) } diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/mod.rs index 35e0fdc180..bc28809f41 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/mod.rs @@ -1,4 +1,5 @@ pub(crate) mod catalog; +pub(crate) mod catalog_nodes; pub(crate) mod events; pub(crate) mod fingerprint_reads; pub(crate) mod fingerprints; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/list_person_catalog_nodes.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/list_person_catalog_nodes.rs new file mode 100644 index 0000000000..1803a07f26 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/list_person_catalog_nodes.rs @@ -0,0 +1,61 @@ +//! A person lists the current nodes with view on all of them; builds are batched indexed seeks. + +use uuid::Uuid; + +use crate::authorization::main::authorize_listing::authorize_listing; +use crate::credentials::models::{AuditContext, DefaultRole}; +use crate::snapshots::main::read_current_record::read_current_record; +use crate::snapshots::main::read_snapshot_nodes::read_snapshot_nodes; +use crate::store::errors::ControlError; +use crate::store::main::{begin::begin, commit::commit}; +use crate::store::models::{ControlStore, TransactionMode}; +use crate::warehouse_metadata::_helpers::catalog_nodes::catalog_node; +use crate::warehouse_metadata::_helpers::fingerprint_reads::latest_fingerprints; +use crate::warehouse_metadata::_helpers::node_status::destination; +use crate::warehouse_metadata::models::{CatalogNodeBuild, CatalogNodeRecord, FingerprintQuery}; + +pub async fn list_person_catalog_nodes( + store: &ControlStore, + credentials: (&AuditContext, Uuid), + (location_id, target): (Uuid, &str), + default_role: DefaultRole, +) -> Result, ControlError> { + let (transaction, _) = authorize_listing( + begin(store, TransactionMode::Write).await?, + credentials, + (location_id, target), + default_role, + ) + .await?; + let (transaction, snapshot) = read_current_record(transaction, (location_id, target)).await?; + let Some(snapshot) = snapshot else { + commit(transaction).await?; + return Ok(Vec::new()); + }; + let (transaction, nodes) = read_snapshot_nodes(transaction, snapshot.snapshot_id, None).await?; + let mut positions: Vec = Vec::new(); + let mut queries: Vec = Vec::new(); + for (position, (_, node)) in nodes.iter().enumerate() { + if let Some(query) = destination(node) { + positions.push(position); + queries.push(query); + } + } + let (transaction, found) = + latest_fingerprints(transaction, (location_id, target, &queries)).await?; + let mut builds: Vec> = vec![None; nodes.len()]; + for (position, build) in positions.iter().zip(found) { + builds[*position] = build.map(|(record, _)| CatalogNodeBuild { + observed_at: record.observed_at, + run_id: record.run_id, + version_hash: record.version_hash, + }); + } + let records: Vec = nodes + .iter() + .zip(builds) + .map(|((key, node), build)| catalog_node((key, node), snapshot.snapshot_id, build)) + .collect(); + commit(transaction).await?; + Ok(records) +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs index 4ebb52f84f..f93dd8430f 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs @@ -7,6 +7,7 @@ pub mod ingest_migrations; pub mod ingest_old_name_views; pub mod ingest_source_freshness; pub mod ingest_statistics; +pub mod list_person_catalog_nodes; pub(crate) mod prune_statistics; pub mod read_person_catalog; pub mod read_person_fingerprint; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs index 61e87d6a6c..cf6e2589c2 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs @@ -609,3 +609,39 @@ impl Default for StatisticsRetention { } } } + +/// A current snapshot node's declared metadata and latest build, without SQL, source or config. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct CatalogNodeRecord { + pub node_key: String, + /// The snapshot node kind, such as `model`, `seed`, `source` or `audit`. + pub kind: String, + pub name: String, + pub path: Option, + pub materialization: Option, + pub description: Option, + pub tags: Vec, + /// Keys of the nodes this node reads. + pub dependencies: Vec, + /// The warehouse relation the node builds or reads. + pub relation: Option, + pub column_count: Option, + /// The newest fingerprint recorded for the node at its current relation, from any build. + pub last_build: Option, + pub snapshot_id: Uuid, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct CatalogRelation { + pub database: Option, + pub schema: String, + pub name: String, +} + +/// A fingerprint row's build identity; the `sqb` run that wrote it, not a scheduler run. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct CatalogNodeBuild { + pub observed_at: DateTime, + pub run_id: String, + pub version_hash: String, +} diff --git a/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs b/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs index 0c8b545af4..be16d1b217 100644 --- a/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs @@ -6,12 +6,15 @@ use uuid::Uuid; use crate::credentials::models::{AuditContext, DefaultRole}; use crate::runs::tests::helpers::snapshot; use crate::snapshots::main::publish_snapshot::publish_snapshot; +use crate::snapshots::models::SnapshotInput; use crate::snapshots::models::SnapshotNode; use crate::store::errors::ControlError; -use crate::store::models::ControlStore; +use crate::store::main::{begin::begin, commit::commit, query::query}; +use crate::store::models::{ControlStore, Parameter, TransactionMode}; use crate::warehouse_metadata::main::{ ingest_catalog::ingest_catalog, ingest_fingerprints::ingest_fingerprints, ingest_source_freshness::ingest_source_freshness, + list_person_catalog_nodes::list_person_catalog_nodes, read_person_node_status::read_person_node_status, }; use crate::warehouse_metadata::models::{MetadataBatch, NodeStatusQuery}; @@ -197,6 +200,50 @@ pub(crate) async fn node_status_reads( ); assert_eq!(changed_status["source_data"], after["source_data"]); assert!(!changed_status.to_string().contains("metadata_json")); + let listed: Vec = list_person_catalog_nodes( + store, + credentials, + (location, "development"), + DefaultRole::None, + ) + .await + .expect("catalog nodes") + .into_iter() + .map(|node| json!(node)) + .collect(); + let listed_node = |key: &str| { + listed + .iter() + .find(|node| node["node_key"] == key) + .cloned() + .unwrap_or_else(|| panic!("{key} listed")) + }; + let orders = listed_node("model:orders"); + assert_eq!( + orders["last_build"]["observed_at"], + json!(changed.records[0]["ts"]), + "the newest build of the model wins" + ); + assert_eq!(orders["last_build"]["run_id"], batch.records[0]["run_id"]); + assert_eq!( + orders["relation"], + json!({"database": null, "schema": "main", "name": "orders"}) + ); + assert_eq!(orders["dependencies"], json!(["source:raw_orders"])); + assert_eq!( + listed_node("udf:orders")["last_build"]["run_id"], + "orders-function-run", + "a same-named function keeps its own build" + ); + assert!( + listed + .iter() + .filter(|node| node["relation"].is_null()) + .all(|node| node["last_build"].is_null()) + ); + assert!(!json!(listed).to_string().contains("metadata_json")); + listing_ignores_build_history(store, credentials, (location, &listed)).await; + listing_reads_wide_targets_in_batches(store, credentials, (location, input.clone())).await; assert!(matches!( read_person_node_status( store, @@ -225,3 +272,203 @@ pub(crate) async fn node_status_reads( .is_none() ); } + +/// Older builds copied 200,000 times over must not change the listing or make it scan them. +async fn listing_ignores_build_history( + store: &ControlStore, + credentials: (&AuditContext, Uuid), + (location, listed): (Uuid, &[Value]), +) { + let transaction = begin(store, TransactionMode::Write).await.expect("begin"); + let (mut transaction, rows) = query( + transaction, + "SELECT event_id FROM sqb_warehouse_fingerprints WHERE location_id=$1 AND target='development'", + vec![Parameter::Id(location)], + ) + .await + .expect("current builds"); + let templates: Vec = rows + .iter() + .map(|row| serde_json::from_value(row["event_id"].clone()).expect("event id")) + .collect(); + let copies = 200_000 / templates.len(); + let oldest = chrono::DateTime::parse_from_rfc3339("2025-01-01T00:00:00Z") + .expect("history start") + .to_utc(); + for template in &templates { + for first in (0..copies).step_by(1_000) { + let mut parameters = vec![Parameter::Id(*template)]; + let mut values: Vec = Vec::new(); + for copy in first..copies.min(first + 1_000) { + values.push(format!( + "(${},${},${})", + parameters.len() + 1, + parameters.len() + 2, + parameters.len() + 3 + )); + parameters.push(Parameter::Id(Uuid::now_v7())); + parameters.push(Parameter::Text(format!("history-{template}-{copy}"))); + parameters.push(Parameter::Time( + oldest + chrono::Duration::seconds(i64::try_from(copy).expect("seconds")), + )); + } + (transaction, _) = query( + transaction, + &format!( + "INSERT INTO sqb_warehouse_fingerprints (event_id,source_event_id,observed_at,location_id,target, \ + definition_text_hash,record_json,received_at,state_database,state_schema,node_type,node_name, \ + target_database,target_schema,target_name) \ + SELECT history.column1,history.column2,history.column3,location_id,target,definition_text_hash, \ + record_json,received_at,state_database,state_schema,node_type,node_name,target_database, \ + target_schema,target_name \ + FROM (VALUES {}) AS history, sqb_warehouse_fingerprints WHERE event_id=$1", + values.join(",") + ), + parameters, + ) + .await + .expect("older build history"); + } + } + commit(transaction).await.expect("commit history"); + let mut timings: Vec = Vec::new(); + for _ in 0..3 { + let started = std::time::Instant::now(); + let relisted: Vec = list_person_catalog_nodes( + store, + credentials, + (location, "development"), + DefaultRole::None, + ) + .await + .expect("catalog nodes over a long history") + .into_iter() + .map(|node| json!(node)) + .collect(); + timings.push(started.elapsed()); + assert_eq!(relisted, listed, "older builds never replace the newest"); + } + timings.sort(); + assert!( + timings[1] < std::time::Duration::from_millis(100), + "listing {} nodes over {} builds took {:?}", + listed.len(), + templates.len() * (copies + 1), + timings[1] + ); +} + +/// Three thousand built nodes, half in a named database, are read in batches and keep their own builds. +async fn listing_reads_wide_targets_in_batches( + store: &ControlStore, + credentials: (&AuditContext, Uuid), + (location, mut input): (Uuid, SnapshotInput), +) { + let names: Vec = (0..3_000).map(|index| format!("wide_{index:04}")).collect(); + let database = |index: usize| (index % 2 == 1).then(|| "analytics_db".to_owned()); + let built_at = |index: usize| { + chrono::DateTime::parse_from_rfc3339("2026-02-01T00:00:00Z") + .expect("build start") + .to_utc() + + chrono::Duration::seconds(i64::try_from(index).expect("seconds")) + }; + for (index, name) in names.iter().enumerate() { + input.nodes.insert(format!("model:{name}"), SnapshotNode { + kind: "model".into(), + definition: json!({"name":name,"destination":{"database":database(index),"schema":"main","name":name}}), + raw_sql: None, + compiled_sql: None, + python_source: None, + }); + } + input.generation += 1; + publish_snapshot(store, &input) + .await + .expect("wide snapshot"); + let transaction = begin(store, TransactionMode::Write).await.expect("begin"); + let (mut transaction, rows) = query( + transaction, + "SELECT event_id,record_json ->> 'run_id' AS run_id FROM sqb_warehouse_fingerprints \ + WHERE location_id=$1 AND target='development' AND node_type='model' AND node_name='orders' \ + AND target_database IS NULL AND target_schema='main' AND target_name='orders' LIMIT 1", + vec![Parameter::Id(location)], + ) + .await + .expect("template build"); + let template: Uuid = serde_json::from_value(rows[0]["event_id"].clone()).expect("event id"); + for (chunk_index, chunk) in names.chunks(300).enumerate() { + let mut parameters = vec![Parameter::Id(template)]; + let mut values: Vec = Vec::new(); + for (offset, name) in chunk.iter().enumerate() { + let index = chunk_index * 300 + offset; + values.push(format!( + "(${},${},${},${},${})", + parameters.len() + 1, + parameters.len() + 2, + parameters.len() + 3, + parameters.len() + 4, + parameters.len() + 5 + )); + parameters.push(Parameter::Id(Uuid::now_v7())); + parameters.push(Parameter::Text(format!("wide-{name}"))); + parameters.push(Parameter::Text(name.clone())); + parameters.push(Parameter::Time(built_at(index))); + parameters.push(Parameter::OptionalText(database(index))); + } + (transaction, _) = query( + transaction, + &format!( + "INSERT INTO sqb_warehouse_fingerprints (event_id,source_event_id,node_name,target_name,location_id,target, \ + observed_at,target_database,definition_text_hash,record_json,received_at,state_database,state_schema, \ + node_type,target_schema) \ + SELECT wide.column1,wide.column2,wide.column3,wide.column3,location_id,target,wide.column4,wide.column5, \ + definition_text_hash,record_json,received_at,state_database,state_schema,node_type,target_schema \ + FROM (VALUES {}) AS wide, sqb_warehouse_fingerprints WHERE event_id=$1", + values.join(",") + ), + parameters, + ) + .await + .expect("wide builds"); + } + commit(transaction).await.expect("commit builds"); + let mut timings: Vec = Vec::new(); + for _ in 0..3 { + let started = std::time::Instant::now(); + let listed = list_person_catalog_nodes( + store, + credentials, + (location, "development"), + DefaultRole::None, + ) + .await + .expect("wide catalog"); + timings.push(started.elapsed()); + let builds: std::collections::BTreeMap<&str, _> = listed + .iter() + .map(|node| (node.node_key.as_str(), node.last_build.as_ref())) + .collect(); + let built = names + .iter() + .enumerate() + .filter(|(index, name)| { + builds + .get(format!("model:{name}").as_str()) + .copied() + .flatten() + .is_some_and(|build| { + build.observed_at == built_at(*index) + && json!(build.run_id) == rows[0]["run_id"] + }) + }) + .count(); + assert_eq!(built, names.len(), "every wide node keeps its own build"); + } + timings.sort(); + assert!( + timings[1] < std::time::Duration::from_millis(450), + "listing {} nodes took {:?}", + names.len(), + timings[1] + ); +} diff --git a/crates/sqlbuild-orchestration/tests/api/test_denials.rs b/crates/sqlbuild-orchestration/tests/api/test_denials.rs index 439f159c7a..d2cb06ea01 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_denials.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_denials.rs @@ -20,8 +20,9 @@ use crate::store::tests::store_helpers::{close_test_store, open_test_store, run} use crate::store::tests::store_test_types::Backend; /// Location-scoped reads and the action each needs. -const READS: [(&str, &str); 11] = [ +const READS: [(&str, &str); 12] = [ ("runs", "view"), + ("nodes", "view"), ("jobs", "view"), ("triggers", "view"), ("changes", "view"), diff --git a/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs b/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs index 9753574753..f2d7d6a691 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs @@ -285,6 +285,7 @@ fn given_browser_session_roles_when_work_is_requested_then_scope_revocation_and_ assert!(!page.to_string().contains("select 1 as id")); let known_reads = [ format!("{metadata}/{}/nodes", seed_run.run_id), + format!("/api/control/locations/{location}/targets/development/nodes"), format!("/api/control/locations/{location}/targets/development/jobs"), format!("/api/control/locations/{location}/targets/development/triggers"), format!( diff --git a/ui/src/lib/api/http-catalog.test.ts b/ui/src/lib/api/http-catalog.test.ts new file mode 100644 index 0000000000..6c5452fd76 --- /dev/null +++ b/ui/src/lib/api/http-catalog.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it } from 'vitest'; +import type { components } from '$control/schema.d.ts'; +import { toCatalogNodes } from './http-catalog'; + +type W = components['schemas']; + +function node(key: string, extra: Partial = {}): W['CatalogNodeRecord'] { + const [kind, name] = key.split(':'); + return { node_key: key, kind, name, tags: [], dependencies: [], snapshot_id: '01a1b000-0000-7000-8000-000000000001', ...extra }; +} + +describe('catalog node mapping', () => { + it('derives the catalog kind from the model materialization and drops nodes the catalog does not list', () => { + const nodes = toCatalogNodes([ + node('model:orders_view', { materialization: 'view' }), + node('model:order_events', { materialization: 'incremental' }), + node('model:order_partitions', { materialization: 'partition_tracked' }), + node('seed:countries', { materialization: 'table' }), + node('source:raw_orders'), + node('audit:not_null:model:orders_view:order_id', { dependencies: ['model:orders_view'] }), + node('macro:cents') + ]); + expect(nodes.map((n) => [n.key, n.kind, n.materialization])).toEqual([ + ['model:orders_view', 'view', 'view'], + ['model:order_events', 'incremental', 'incremental'], + ['model:order_partitions', 'table', 'partition_tracked'], + ['seed:countries', 'seed', null], + ['source:raw_orders', 'source', null] + ]); + }); + + it('keeps only upstream nodes that are listed, so graphs never point at a hidden node', () => { + const [orders] = toCatalogNodes([ + node('model:orders', { dependencies: ['source:raw_orders', 'macro:cents', 'model:unpublished'] }), + node('source:raw_orders'), + node('macro:cents') + ]); + expect(orders.upstream).toEqual(['source:raw_orders']); + }); + + it('reports the fingerprint as a sqb build and leaves unreported build state unknown, not empty', () => { + const [orders] = toCatalogNodes([ + node('model:orders', { + materialization: 'table', + description: 'One row per order', + tags: ['marts'], + column_count: 4, + relation: { database: null, schema: 'main', name: 'orders' }, + last_build: { observed_at: '2026-11-07T06:00:00Z', run_id: 'sqb-run-1', version_hash: 'v1' } + }) + ]); + expect(orders.lastBuild).toEqual({ fingerprintRunId: 'sqb-run-1', runId: null, jobKey: null, builtAt: '2026-11-07T06:00:00Z', fingerprint: 'v1' }); + expect([orders.description, orders.tags, orders.columnCount, orders.relation]).toEqual(['One row per order', ['marts'], 4, { database: null, schema: 'main', name: 'orders' }]); + expect([orders.stale, orders.checks, orders.lastOutcome, orders.group, orders.owner]).toEqual([null, null, null, null, []]); + }); +}); diff --git a/ui/src/lib/api/http-catalog.ts b/ui/src/lib/api/http-catalog.ts new file mode 100644 index 0000000000..bba6472397 --- /dev/null +++ b/ui/src/lib/api/http-catalog.ts @@ -0,0 +1,50 @@ +import type { components } from '$control/schema.d.ts'; +import type { CatalogNode } from './types/build'; +import type { NodeKind } from './types/operate'; + +type Wire = components['schemas']; + +/** The catalog shows models, seeds and sources; audits, functions, macros and hooks are not catalog nodes. */ +function kindOf(wire: Wire['CatalogNodeRecord']): NodeKind | null { + if (wire.kind === 'source' || wire.kind === 'seed') return wire.kind; + if (wire.kind !== 'model') return null; + if (wire.materialization === 'view') return 'view'; + return wire.materialization?.startsWith('incremental') ? 'incremental' : 'table'; +} + +/** + * One node of the current snapshot. The server reports declared metadata and the newest + * fingerprint, not run outcomes, staleness, check counts, owners or groups, so those stay null + * or empty and the pages hide them. The fingerprint's run is a `sqb` run, not a scheduler run. + */ +export function toCatalogNode(wire: Wire['CatalogNodeRecord']): CatalogNode | null { + const kind = kindOf(wire); + if (!kind) return null; + return { + key: wire.node_key, + kind, + materialization: kind === 'source' || kind === 'seed' ? null : (wire.materialization ?? null), + group: null, + tags: wire.tags, + owner: [], + description: wire.description ?? '', + path: wire.path ?? '', + upstream: wire.dependencies, + relation: wire.relation ? { database: wire.relation.database ?? null, schema: wire.relation.schema, name: wire.relation.name } : null, + columnCount: wire.column_count ?? 0, + lastBuild: wire.last_build + ? { fingerprintRunId: wire.last_build.run_id, runId: null, jobKey: null, builtAt: wire.last_build.observed_at, fingerprint: wire.last_build.version_hash } + : null, + lastOutcome: null, + stale: null, + sourceFreshness: null, + checks: null + }; +} + +/** Catalog nodes in listing order, keeping only dependencies that are themselves listed. */ +export function toCatalogNodes(wire: Wire['CatalogNodeRecord'][]): CatalogNode[] { + const nodes = wire.flatMap((node) => toCatalogNode(node) ?? []); + const listed = new Set(nodes.map((node) => node.key)); + return nodes.map((node) => ({ ...node, upstream: node.upstream.filter((key) => listed.has(key)) })); +} diff --git a/ui/src/lib/api/http-client.ts b/ui/src/lib/api/http-client.ts index c1b9455b47..70a1df67b9 100644 --- a/ui/src/lib/api/http-client.ts +++ b/ui/src/lib/api/http-client.ts @@ -2,11 +2,12 @@ import createFetch from 'openapi-fetch'; import type { components, paths } from '$control/schema.d.ts'; import { ApiError, type ApiClient } from './client'; import { checkDetailOf, toCheck, toCheckEvidence, toCheckSql } from './http-checks'; +import { toCatalogNodes } from './http-catalog'; import { jobDetailOf, toJob, toTick, toTrigger, triggerDetailOf } from './http-jobs'; import { RUN_WINDOW, runDetailOf, runRowPage, runsMatching, runTabCountsOf, toRunNode, toRuns } from './http-runs'; import type { ChangeSubscription, ChangedChannel, ChangedSignal, ConnectionState } from './signals'; import type { AuthMode, Bootstrap, Capability, Location, Scope } from './types'; -import type { Check, CheckDetail, CheckEvidence, CheckSql } from './types/build'; +import type { CatalogNode, Check, CheckDetail, CheckEvidence, CheckSql } from './types/build'; import type { Job, JobDetail, RerunMode, Run, RunDetail, RunListQuery, RunRowPage, RunRowQuery, RunTabCounts, Trigger, TriggerDetail, TriggerState } from './types/operate'; type Wire = components['schemas']; @@ -34,7 +35,8 @@ const SUPPORTED: ReadonlySet = new Set([ 'listChecks', 'getCheck', 'getCheckSql', - 'getCheckEvidence' + 'getCheckEvidence', + 'listCatalogNodes' ]); /** A UI location is one server location and target; its id carries both. */ @@ -235,6 +237,13 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt if (!response.ok) throw new ApiError(errorText(error, response), response.status); return requireRun(scope, body.run_id); }, + async listCatalogNodes(scope: Scope): Promise { + const { locationId, target } = parseLocationKey(scope.locationId); + const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/nodes', { params: { path: { location_id: locationId, target } } }); + if (response.status === 401) onUnauthorized(); + if (!data) throw new ApiError(errorText(error, response), response.status); + return toCatalogNodes(data); + }, listChecks: async (scope: Scope): Promise => (await checkResults(scope)).map(toCheck), getCheck: async (scope: Scope, id: string): Promise => checkDetailOf(await checkResults(scope, id)), async getCheckSql(scope: Scope, checkId: string): Promise { diff --git a/ui/src/lib/api/mock/build.ts b/ui/src/lib/api/mock/build.ts index 5745def597..ba93c80a50 100644 --- a/ui/src/lib/api/mock/build.ts +++ b/ui/src/lib/api/mock/build.ts @@ -230,8 +230,8 @@ export class MockBuild { const thisWeek = this.#deploys.filter((d) => d.publishedAt >= weekAgo).flatMap((d) => d.nodes); const isModelChange = (key: string) => MODEL_KINDS.has(snapshotNode(key)?.kind ?? 'table'); - const changed = nodes.filter((n) => n.stale.some((s) => s.reason !== 'behind')).map((n) => n.key); - const behind = nodes.filter((n) => n.stale.some((s) => s.reason === 'behind')).map((n) => n.key); + const changed = nodes.filter((n) => n.stale?.some((s) => s.reason !== 'behind')).map((n) => n.key); + const behind = nodes.filter((n) => n.stale?.some((s) => s.reason === 'behind')).map((n) => n.key); const staleSources = nodes.filter((n) => n.sourceFreshness && n.sourceFreshness.status !== 'pass'); const attention: AttentionItem[] = []; @@ -258,13 +258,15 @@ export class MockBuild { const groups = new Map(); for (const n of nodes) { if (n.kind === 'source') continue; - const g = groups.get(n.group) ?? { group: n.group, total: 0, current: 0, changed: 0, behind: 0, failed: 0 }; + const group = n.group ?? ''; + const g = groups.get(group) ?? { group, total: 0, current: 0, changed: 0, behind: 0, failed: 0 }; g.total += 1; if (n.lastOutcome === 'failed') g.failed += 1; - if (n.stale.some((s) => s.reason !== 'behind')) g.changed += 1; - if (n.stale.some((s) => s.reason === 'behind')) g.behind += 1; - if (!n.stale.length && n.lastOutcome !== 'failed' && n.lastBuild) g.current += 1; - groups.set(n.group, g); + const stale = n.stale ?? []; + if (stale.some((s) => s.reason !== 'behind')) g.changed += 1; + if (stale.some((s) => s.reason === 'behind')) g.behind += 1; + if (!stale.length && n.lastOutcome !== 'failed' && n.lastBuild) g.current += 1; + groups.set(group, g); } const latest = runs.find((r) => r.endedAt); return this.#ctx.reply({ @@ -395,7 +397,7 @@ export class MockBuild { return state.lastBuild ? Date.parse(state.lastBuild.builtAt) : -Infinity; } - #catalogNode(state: NodeState, checks: Check[]): CatalogNode { + #catalogNode(state: NodeState, checks: Check[]): CatalogNode & Pick { const { sn, lastBuild, lastOutcome, stale } = state; const spec = NODE_SPECS[sn.key]; return { diff --git a/ui/src/lib/api/types/build.ts b/ui/src/lib/api/types/build.ts index 839f9b4aff..6bc9353ffb 100644 --- a/ui/src/lib/api/types/build.ts +++ b/ui/src/lib/api/types/build.ts @@ -56,7 +56,8 @@ export type CatalogNode = { kind: NodeKind; /** e.g. `table`, `view`, `incremental · merge`, `incremental · microbatch`; null for sources and tasks. */ materialization: string | null; - group: string; + /** Null when the server does not report groups. */ + group: string | null; tags: string[]; /** Declared `owner`, or the project default. */ owner: string[]; @@ -69,10 +70,15 @@ export type CatalogNode = { lastBuild: NodeBuild | null; /** The latest build attempt's outcome, from run nodes and audit results. */ lastOutcome: NodeOutcome | null; - stale: StaleReason[]; + /** + * Null when the server does not report build state; outcome and staleness are then unknown, + * not absent, and pages hide them instead of showing "Not built". + */ + stale: StaleReason[] | null; /** Sources only. */ sourceFreshness: SourceFreshness | null; - checks: CheckCounts; + /** Null when the server does not report check counts. */ + checks: CheckCounts | null; }; // ---- node detail ---- @@ -153,7 +159,10 @@ export type NodeHistory = { export type NodeJob = { key: string; kind: 'job' | 'group'; triggerIds: string[] }; -export type CatalogNodeDetail = CatalogNode & { +/** The detail read reports build state, so its staleness and check counts are never unknown. */ +export type CatalogNodeDetail = Omit & { + stale: StaleReason[]; + checks: CheckCounts; downstream: string[]; /** Outcomes of the most recent builds, oldest first, up to 40 (`recent_outcomes`). */ recentOutcomes: NodeOutcome[]; diff --git a/ui/src/lib/lineage/catalog-graph.ts b/ui/src/lib/lineage/catalog-graph.ts index 5f76580332..aa96ee033f 100644 --- a/ui/src/lib/lineage/catalog-graph.ts +++ b/ui/src/lib/lineage/catalog-graph.ts @@ -26,7 +26,7 @@ export function cardStale(reasons: StaleReason[]): CardStale | undefined { export function checksBadge(node: CatalogNode): { checks?: string; checksColor?: string } { const c = node.checks; - if (!c.total) return {}; + if (!c?.total) return {}; if (c.failing) return { checks: `✕ ${c.failing}`, checksColor: 'var(--sb-error)' }; if (c.warning) return { checks: `! ${c.warning}`, checksColor: 'var(--sb-warning)' }; return c.passing === c.total ? { checks: `✓ ${c.passing}` } : { checks: `${c.passing}/${c.total}` }; @@ -39,7 +39,7 @@ export function cardData(node: CatalogNode): Record { group: node.group, nodeKind: node.kind, status: cardStatus(node), - stale: cardStale(node.stale), + stale: cardStale(node.stale ?? []), icon: nodeIconKey(node), lastBuiltAt: node.kind === 'source' ? (node.sourceFreshness?.loadedAt ?? undefined) : node.lastBuild?.builtAt, description: node.description || undefined, diff --git a/ui/src/lib/nav.test.ts b/ui/src/lib/nav.test.ts index a163f01bda..8cdc115678 100644 --- a/ui/src/lib/nav.test.ts +++ b/ui/src/lib/nav.test.ts @@ -16,6 +16,10 @@ describe('visibleNav', () => { ]); }); + it('keeps lineage hidden until node detail is served, even when the catalog list is', () => { + expect(labels(serving('listCatalogNodes'), ['view'])).toEqual([['Build', ['Catalog']]]); + }); + it('hides admin pages from people without an admin capability, even when the server serves them', () => { expect(labels(serving('listAccounts', 'listRoles', 'listAuditEntries'), ['view', 'job.run'])).toEqual([]); }); diff --git a/ui/src/lib/nav.ts b/ui/src/lib/nav.ts index 789191fae9..269b0c967e 100644 --- a/ui/src/lib/nav.ts +++ b/ui/src/lib/nav.ts @@ -26,7 +26,7 @@ export const NAV: readonly NavGroup[] = [ { section: 'Build', items: [ - { label: 'Lineage', href: '/lineage', icon: GitBranchIcon, loads: 'listCatalogNodes' }, + { label: 'Lineage', href: '/lineage', icon: GitBranchIcon, loads: 'getCatalogNode' }, { label: 'Catalog', href: '/catalog', icon: DatabaseIcon, loads: 'listCatalogNodes', badge: { value: (c) => c.catalogNodes, tone: 'muted' } }, { label: 'Checks', href: '/checks', icon: CheckCircleIcon, loads: 'listChecks', badge: { value: (c) => c.failingChecks, tone: 'warning' } }, { label: 'Changes', href: '/changes', icon: GitCompareIcon, loads: 'listSnapshotChanges' } diff --git a/ui/src/routes/catalog/+page.svelte b/ui/src/routes/catalog/+page.svelte index f2f5dba0bc..afe9457a83 100644 --- a/ui/src/routes/catalog/+page.svelte +++ b/ui/src/routes/catalog/+page.svelte @@ -33,26 +33,34 @@ const matchesStatus = (n: CatalogNode, f: StatusFilter | '') => !f || - (f === 'stale' && n.stale.length > 0) || - (f === 'changed' && n.stale.some((s) => s.reason !== 'behind')) || - (f === 'behind' && n.stale.some((s) => s.reason === 'behind')) || + (f === 'stale' && !!n.stale?.length) || + (f === 'changed' && !!n.stale?.some((s) => s.reason !== 'behind')) || + (f === 'behind' && !!n.stale?.some((s) => s.reason === 'behind')) || (f === 'failed' && n.lastOutcome === 'failed') || (f === 'warn' && n.lastOutcome === 'warn'); + const kinds = $derived(NODE_KINDS.filter((k) => all.some((n) => n.kind === k))); + const owners = $derived([...new Set(all.flatMap((n) => n.owner))].sort()); + const groups = $derived([...new Set(all.flatMap((n) => (n.group === null ? [] : [n.group])))]); + /** Outcome and staleness come together; a server that reports neither leaves them unknown. */ + const showBuildState = $derived(all.some((n) => n.stale !== null)); + const showChecks = $derived(all.some((n) => n.checks !== null)); + const linksNodes = $derived(app.client.supports('getCatalogNode')); + const canRun = $derived(app.client.supports('launchAdhocRun')); + const filtered = $derived( all.filter((n) => { if (kindFilter !== 'all' && n.kind !== kindFilter) return false; - if (ownerFilter && !n.owner.includes(ownerFilter)) return false; - if (groupFilter && n.group !== groupFilter) return false; - if (!matchesStatus(n, statusFilter)) return false; + // A filter whose control is hidden, for example from a shared link, never narrows the list. + if (owners.length && ownerFilter && !n.owner.includes(ownerFilter)) return false; + if (groups.length && groupFilter && n.group !== groupFilter) return false; + if (showBuildState && !matchesStatus(n, statusFilter)) return false; const q = query.trim().toLowerCase(); return !q || [n.key, n.description, kindLine(n), ...n.tags, ...n.owner].some((s) => s.toLowerCase().includes(q)); }) ); - const kinds = $derived(NODE_KINDS.filter((k) => all.some((n) => n.kind === k))); - const owners = $derived([...new Set(all.flatMap((n) => n.owner))].sort()); - const groups = $derived([...new Set(all.map((n) => n.group))]); + const columns = $derived(3 + (owners.length ? 1 : 0) + (showBuildState ? 2 : 0) + (showChecks ? 1 : 0)); const counts = $derived({ failed: all.filter((n) => n.lastOutcome === 'failed').length, warn: all.filter((n) => n.lastOutcome === 'warn').length, @@ -74,29 +82,31 @@ - (runOpen = true)}>▷ Execute{narrowed ? ` ${filteredRunnable.length}` : ''} + {#if canRun} (runOpen = true)}>▷ Execute{narrowed ? ` ${filteredRunnable.length}` : ''}{/if} - +{#if canRun}{/if} {#snippet filters()} ({ value: k, label: nodeKindPlural[k], count: all.filter((n) => n.kind === k).length }))]} /> - ({ ...o, count: all.filter((n) => matchesStatus(n, o.value)).length }))} /> - ({ value: o, label: o, count: all.filter((n) => n.owner.includes(o)).length }))} /> - ({ value: g, label: g, count: all.filter((n) => n.group === g).length }))} /> + {#if showBuildState} ({ ...o, count: all.filter((n) => matchesStatus(n, o.value)).length }))} />{/if} + {#if owners.length} ({ value: o, label: o, count: all.filter((n) => n.owner.includes(o)).length }))} />{/if} + {#if groups.length} ({ value: g, label: g, count: all.filter((n) => n.group === g).length }))} />{/if} {/snippet} @@ -106,31 +116,33 @@ Name Kind - Owner - Last outcome - Stale - Checks + {#if owners.length}Owner{/if} + {#if showBuildState}Last outcomeStale{/if} + {#if showChecks}Checks{/if} Last built {#each filtered as n (n.key)} {@const Glyph = nodeIcon[nodeIconKey(n)] ?? nodeIcon.table} - goto(`/catalog/${n.key}`)}> + linksNodes && goto(`/catalog/${n.key}`)}>
- e.stopPropagation()}>{n.key} + {#if linksNodes} e.stopPropagation()}>{n.key}{:else}{n.key}{/if}
{n.description}
- {kindLine(n)}
{n.group}
+ {kindLine(n)}{#if n.group}
{n.group}
{/if} + {#if owners.length} {#if n.owner.length}
{#each n.owner as o (o)}{o}{/each}
{:else}—{/if} + {/if} + {#if showBuildState} {#if n.kind === 'source'} {#if n.sourceFreshness} @@ -139,20 +151,23 @@ {:else}Not built{/if} - - - {n.checks.total ? `${n.checks.passing} / ${n.checks.total}` : '—'} + + {/if} + {#if showChecks} + + {n.checks?.total ? `${n.checks.passing} / ${n.checks.total}` : '—'} + {/if} {#if n.kind === 'source'} {:else if n.lastBuild} - + {:else}Never{/if} {:else} - + {/each} diff --git a/ui/src/routes/catalog/[name]/utils.ts b/ui/src/routes/catalog/[name]/utils.ts index 9eac2101aa..c4831567fc 100644 --- a/ui/src/routes/catalog/[name]/utils.ts +++ b/ui/src/routes/catalog/[name]/utils.ts @@ -1,13 +1,13 @@ // Pure helpers for the catalog node page. -import type { CatalogNode } from '$lib/api'; +import type { CatalogNodeDetail } from '$lib/api'; export type Tab = 'overview' | 'columns' | 'sql' | 'drift' | 'lineage' | 'collineage' | 'runs' | 'checks' | 'warehouse' | 'history'; export type DriftKind = 'direct' | 'compiled' | 'upstream'; /** Drift against what was last built in this target, from the node's stale reasons. */ -export function driftOf(node: Pick): DriftKind | null { +export function driftOf(node: Pick): DriftKind | null { for (const r of node.stale) if (r.reason === 'changed') return r.change === 'compiled' ? 'compiled' : 'direct'; return node.stale.some((r) => r.reason === 'changed_upstream') ? 'upstream' : null; } @@ -19,7 +19,7 @@ export const driftLabel: Record = { }; /** The SQL viewer's note when only an upstream definition changed. */ -export function upstreamNote(node: Pick): string | undefined { +export function upstreamNote(node: Pick): string | undefined { const r = node.stale.find((s) => s.reason === 'changed_upstream'); return r && r.reason === 'changed_upstream' ? `${r.upstream.join(', ')} changed since this node’s last build.` : undefined; } diff --git a/ui/src/routes/lineage/+page.svelte b/ui/src/routes/lineage/+page.svelte index 9db2c4d92e..83f904fb3f 100644 --- a/ui/src/routes/lineage/+page.svelte +++ b/ui/src/routes/lineage/+page.svelte @@ -266,15 +266,15 @@ return matchesFilterStatus(c, statusFilter); } function matchesFilterStatus(c: CatalogNode, f: typeof statusFilter): boolean { - if (f === 'stale') return c.stale.length > 0; - if (f === 'changed') return c.stale.some((r) => r.reason !== 'behind'); - if (f === 'behind') return c.stale.some((r) => r.reason === 'behind'); + if (f === 'stale') return (c.stale?.length ?? 0) > 0; + if (f === 'changed') return !!c.stale?.some((r) => r.reason !== 'behind'); + if (f === 'behind') return !!c.stale?.some((r) => r.reason === 'behind'); if (f === 'failed') return c.lastOutcome === 'failed'; return true; } const catalogNodes = $derived(catalog.data ?? []); const kindOptions = $derived(NODE_KINDS.filter((k) => catalogNodes.some((n) => n.kind === k)).map((k) => ({ value: k, label: nodeKindPlural[k], count: catalogNodes.filter((n) => n.kind === k).length }))); - const groupOptions = $derived([...new Set(catalogNodes.map((n) => n.group))].map((g) => ({ value: g, label: g, count: catalogNodes.filter((n) => n.group === g).length }))); + const groupOptions = $derived([...new Set(catalogNodes.flatMap((n) => (n.group === null ? [] : [n.group])))].map((g) => ({ value: g, label: g, count: catalogNodes.filter((n) => n.group === g).length }))); const ownerOptions = $derived([...new Set(catalogNodes.flatMap((n) => n.owner))].sort().map((o) => ({ value: o, label: o, count: catalogNodes.filter((n) => n.owner.includes(o)).length }))); const statusOptions = [ { value: 'stale' as const, label: 'Stale' },