From 6a19279619fdbe6cf0bbd61c7ab4f3402ebb25e5 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Sat, 7 Nov 2026 11:45:26 +0000 Subject: [PATCH 1/4] feat(ui): list the live catalog from the control API Add a person read of the current snapshot's nodes for a location target, with each node's declared metadata and its newest recorded build, and wire the Catalog page to it. The page hides the build-state, checks, owner and group parts the server does not report, links nodes only when node detail is available, and offers Execute only when ad hoc runs are supported. --- clients/typescript/src/schema.d.ts | 84 ++++++++++ .../authentication/password_sessions.rs | 4 + .../api/_helpers/resources/person_reads.rs | 26 ++- .../sqlbuild-orchestration/src/api/models.rs | 1 + .../_helpers/catalog_nodes.rs | 155 ++++++++++++++++++ .../src/warehouse_metadata/_helpers/mod.rs | 1 + .../main/list_person_catalog_nodes.rs | 40 +++++ .../src/warehouse_metadata/main/mod.rs | 1 + .../src/warehouse_metadata/models.rs | 38 ++++- .../tests/api/node_status/helpers.rs | 43 +++++ .../tests/api/test_denials.rs | 3 +- .../tests/api/test_person_sessions.rs | 1 + ui/src/lib/api/http-catalog.test.ts | 56 +++++++ ui/src/lib/api/http-catalog.ts | 50 ++++++ ui/src/lib/api/http-client.ts | 13 +- ui/src/lib/api/mock/build.ts | 18 +- ui/src/lib/api/types/build.ts | 17 +- ui/src/lib/lineage/catalog-graph.ts | 4 +- ui/src/lib/nav.test.ts | 4 + ui/src/lib/nav.ts | 2 +- ui/src/routes/catalog/+page.svelte | 65 +++++--- ui/src/routes/catalog/[name]/utils.ts | 6 +- ui/src/routes/lineage/+page.svelte | 8 +- 23 files changed, 587 insertions(+), 53 deletions(-) create mode 100644 crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/catalog_nodes.rs create mode 100644 crates/sqlbuild-orchestration/src/warehouse_metadata/main/list_person_catalog_nodes.rs create mode 100644 ui/src/lib/api/http-catalog.test.ts create mode 100644 ui/src/lib/api/http-catalog.ts 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..c5689e0849 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/catalog_nodes.rs @@ -0,0 +1,155 @@ +//! Catalog listing rows from snapshot node definitions and one latest-build read per target. + +use std::collections::BTreeMap; + +use chrono::{DateTime, Utc}; +use serde::Deserialize; +use serde_json::Value; +use uuid::Uuid; + +use crate::sensors::constants::SOURCE_NODE_KIND; +use crate::snapshots::models::SnapshotNode; +use crate::store::errors::ControlError; +use crate::store::main::query_records::query_records; +use crate::store::models::{ControlTransaction, Parameter}; +use crate::warehouse_metadata::_helpers::node_status::destination; +use crate::warehouse_metadata::models::{ + CatalogNodeBuild, CatalogNodeRecord, CatalogRelation, FingerprintQuery, +}; + +#[derive(Deserialize)] +struct BuildRow { + node_type: String, + node_name: String, + target_database: Option, + target_schema: Option, + target_name: Option, + observed_at: DateTime, + event_id: Uuid, + run_id: String, + version_hash: String, +} + +/// The newest fingerprint of every node and relation in the target; ties go to the larger event ID. +pub(crate) async fn latest_builds( + transaction: ControlTransaction, + (location_id, target): (Uuid, &str), +) -> Result< + ( + ControlTransaction, + BTreeMap, + ), + ControlError, +> { + let (transaction, rows): (_, Vec) = query_records(transaction, + "SELECT a.node_type,a.node_name,a.target_database,a.target_schema,a.target_name,a.observed_at,a.event_id, \ + a.record_json ->> 'run_id' AS run_id,a.record_json ->> 'version_hash' AS version_hash \ + FROM sqb_warehouse_fingerprints a JOIN (SELECT node_type,node_name,COALESCE(target_database,'') AS d, \ + COALESCE(target_schema,'') AS s,COALESCE(target_name,'') AS n,MAX(observed_at) AS latest \ + FROM sqb_warehouse_fingerprints WHERE location_id=$1 AND target=$2 AND node_type IS NOT NULL AND node_name IS NOT NULL \ + GROUP BY node_type,node_name,COALESCE(target_database,''),COALESCE(target_schema,''),COALESCE(target_name,'')) l \ + ON a.node_type=l.node_type AND a.node_name=l.node_name AND COALESCE(a.target_database,'')=l.d \ + AND COALESCE(a.target_schema,'')=l.s AND COALESCE(a.target_name,'')=l.n AND a.observed_at=l.latest \ + WHERE a.location_id=$1 AND a.target=$2", + vec![Parameter::Id(location_id), Parameter::Text(target.into())]).await?; + let mut builds: BTreeMap = BTreeMap::new(); + for row in rows { + let key = FingerprintQuery { + node_type: row.node_type, + node_name: row.node_name, + database: row.target_database, + schema: row.target_schema, + name: row.target_name, + }; + if builds + .get(&key) + .is_some_and(|(event_id, _)| *event_id > row.event_id) + { + continue; + } + let build = CatalogNodeBuild { + observed_at: row.observed_at, + run_id: row.run_id, + version_hash: row.version_hash, + }; + builds.insert(key, (row.event_id, build)); + } + let builds: BTreeMap = builds + .into_iter() + .map(|(key, (_, build))| (key, build)) + .collect(); + Ok((transaction, builds)) +} + +/// 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, + builds: &BTreeMap, +) -> CatalogNodeRecord { + let definition = &node.definition; + let last_build = destination(node).and_then(|query| builds.get(&query).cloned()); + 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/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..31d4dc3863 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/list_person_catalog_nodes.rs @@ -0,0 +1,40 @@ +//! A person lists the current snapshot's nodes only with view on every one of them, like jobs. + +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, latest_builds}; +use crate::warehouse_metadata::models::CatalogNodeRecord; + +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 (transaction, builds) = latest_builds(transaction, (location_id, target)).await?; + commit(transaction).await?; + Ok(nodes + .iter() + .map(|(key, node)| catalog_node((key, node), snapshot.snapshot_id, &builds)) + .collect()) +} 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..11bb9f6e5a 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs @@ -455,7 +455,7 @@ pub struct CatalogPage { } /// Exact node and physical destination, not a selector; omitted namespace values match SQL NULL. -#[derive(Clone, Debug, Deserialize)] +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Deserialize)] #[serde(deny_unknown_fields)] pub struct FingerprintQuery { pub node_type: String, @@ -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..d07275fe50 100644 --- a/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs @@ -12,6 +12,7 @@ use crate::store::models::ControlStore; 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 +198,48 @@ 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")); assert!(matches!( read_person_node_status( store, 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..f2d8193514 100644 --- a/ui/src/routes/catalog/+page.svelte +++ b/ui/src/routes/catalog/+page.svelte @@ -33,9 +33,9 @@ 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'); @@ -52,7 +52,13 @@ 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 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 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 +80,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 +114,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 +149,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' }, From 93b474ccbe788478143ee24825471a19bbace859 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Sat, 7 Nov 2026 12:17:04 +0000 Subject: [PATCH 2/4] fix(ui): read each catalog node's build with an indexed seek The listing grouped the target's whole fingerprint history on every load, so its cost grew with history while it held the write transaction. Read each listed node's newest build with the exact-destination lookup node status uses, and bound the listing over 200,000 older builds with a time limit. A catalog filter whose control is hidden no longer narrows the list, so a shared status or group link cannot empty it. --- .../_helpers/catalog_nodes.rs | 81 +---------------- .../main/list_person_catalog_nodes.rs | 32 +++++-- .../src/warehouse_metadata/models.rs | 2 +- .../tests/api/node_status/helpers.rs | 89 ++++++++++++++++++- ui/src/routes/catalog/+page.svelte | 24 ++--- 5 files changed, 128 insertions(+), 100 deletions(-) diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/catalog_nodes.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/catalog_nodes.rs index c5689e0849..a4b23af9ff 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/catalog_nodes.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/catalog_nodes.rs @@ -1,85 +1,11 @@ -//! Catalog listing rows from snapshot node definitions and one latest-build read per target. +//! Catalog listing rows from snapshot node definitions and each node's latest build. -use std::collections::BTreeMap; - -use chrono::{DateTime, Utc}; -use serde::Deserialize; use serde_json::Value; use uuid::Uuid; use crate::sensors::constants::SOURCE_NODE_KIND; use crate::snapshots::models::SnapshotNode; -use crate::store::errors::ControlError; -use crate::store::main::query_records::query_records; -use crate::store::models::{ControlTransaction, Parameter}; -use crate::warehouse_metadata::_helpers::node_status::destination; -use crate::warehouse_metadata::models::{ - CatalogNodeBuild, CatalogNodeRecord, CatalogRelation, FingerprintQuery, -}; - -#[derive(Deserialize)] -struct BuildRow { - node_type: String, - node_name: String, - target_database: Option, - target_schema: Option, - target_name: Option, - observed_at: DateTime, - event_id: Uuid, - run_id: String, - version_hash: String, -} - -/// The newest fingerprint of every node and relation in the target; ties go to the larger event ID. -pub(crate) async fn latest_builds( - transaction: ControlTransaction, - (location_id, target): (Uuid, &str), -) -> Result< - ( - ControlTransaction, - BTreeMap, - ), - ControlError, -> { - let (transaction, rows): (_, Vec) = query_records(transaction, - "SELECT a.node_type,a.node_name,a.target_database,a.target_schema,a.target_name,a.observed_at,a.event_id, \ - a.record_json ->> 'run_id' AS run_id,a.record_json ->> 'version_hash' AS version_hash \ - FROM sqb_warehouse_fingerprints a JOIN (SELECT node_type,node_name,COALESCE(target_database,'') AS d, \ - COALESCE(target_schema,'') AS s,COALESCE(target_name,'') AS n,MAX(observed_at) AS latest \ - FROM sqb_warehouse_fingerprints WHERE location_id=$1 AND target=$2 AND node_type IS NOT NULL AND node_name IS NOT NULL \ - GROUP BY node_type,node_name,COALESCE(target_database,''),COALESCE(target_schema,''),COALESCE(target_name,'')) l \ - ON a.node_type=l.node_type AND a.node_name=l.node_name AND COALESCE(a.target_database,'')=l.d \ - AND COALESCE(a.target_schema,'')=l.s AND COALESCE(a.target_name,'')=l.n AND a.observed_at=l.latest \ - WHERE a.location_id=$1 AND a.target=$2", - vec![Parameter::Id(location_id), Parameter::Text(target.into())]).await?; - let mut builds: BTreeMap = BTreeMap::new(); - for row in rows { - let key = FingerprintQuery { - node_type: row.node_type, - node_name: row.node_name, - database: row.target_database, - schema: row.target_schema, - name: row.target_name, - }; - if builds - .get(&key) - .is_some_and(|(event_id, _)| *event_id > row.event_id) - { - continue; - } - let build = CatalogNodeBuild { - observed_at: row.observed_at, - run_id: row.run_id, - version_hash: row.version_hash, - }; - builds.insert(key, (row.event_id, build)); - } - let builds: BTreeMap = builds - .into_iter() - .map(|(key, (_, build))| (key, build)) - .collect(); - Ok((transaction, builds)) -} +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> { @@ -125,10 +51,9 @@ fn relation(node: &SnapshotNode) -> Option { pub(crate) fn catalog_node( (node_key, node): (&str, &SnapshotNode), snapshot_id: Uuid, - builds: &BTreeMap, + last_build: Option, ) -> CatalogNodeRecord { let definition = &node.definition; - let last_build = destination(node).and_then(|query| builds.get(&query).cloned()); CatalogNodeRecord { node_key: node_key.into(), kind: node.kind.clone(), 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 index 31d4dc3863..381cdeafa4 100644 --- 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 @@ -1,4 +1,4 @@ -//! A person lists the current snapshot's nodes only with view on every one of them, like jobs. +//! A person lists the current nodes with view on all of them; each build is one indexed seek. use uuid::Uuid; @@ -9,8 +9,10 @@ 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, latest_builds}; -use crate::warehouse_metadata::models::CatalogNodeRecord; +use crate::warehouse_metadata::_helpers::catalog_nodes::catalog_node; +use crate::warehouse_metadata::_helpers::fingerprint_reads::latest_fingerprint; +use crate::warehouse_metadata::_helpers::node_status::destination; +use crate::warehouse_metadata::models::{CatalogNodeBuild, CatalogNodeRecord}; pub async fn list_person_catalog_nodes( store: &ControlStore, @@ -30,11 +32,23 @@ pub async fn list_person_catalog_nodes( commit(transaction).await?; return Ok(Vec::new()); }; - let (transaction, nodes) = read_snapshot_nodes(transaction, snapshot.snapshot_id, None).await?; - let (transaction, builds) = latest_builds(transaction, (location_id, target)).await?; + let (mut transaction, nodes) = + read_snapshot_nodes(transaction, snapshot.snapshot_id, None).await?; + let mut records: Vec = Vec::with_capacity(nodes.len()); + for (key, node) in &nodes { + let mut last_build: Option = None; + if let Some(query) = destination(node) { + let (next, build) = + latest_fingerprint(transaction, (location_id, target, &query)).await?; + transaction = next; + last_build = build.map(|(record, _)| CatalogNodeBuild { + observed_at: record.observed_at, + run_id: record.run_id, + version_hash: record.version_hash, + }); + } + records.push(catalog_node((key, node), snapshot.snapshot_id, last_build)); + } commit(transaction).await?; - Ok(nodes - .iter() - .map(|(key, node)| catalog_node((key, node), snapshot.snapshot_id, &builds)) - .collect()) + Ok(records) } diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs index 11bb9f6e5a..cf6e2589c2 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs @@ -455,7 +455,7 @@ pub struct CatalogPage { } /// Exact node and physical destination, not a selector; omitted namespace values match SQL NULL. -#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Deserialize)] +#[derive(Clone, Debug, Deserialize)] #[serde(deny_unknown_fields)] pub struct FingerprintQuery { pub node_type: String, diff --git a/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs b/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs index d07275fe50..88e518ad71 100644 --- a/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs @@ -8,7 +8,8 @@ use crate::runs::tests::helpers::snapshot; use crate::snapshots::main::publish_snapshot::publish_snapshot; 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, @@ -240,6 +241,7 @@ pub(crate) async fn node_status_reads( .all(|node| node["last_build"].is_null()) ); assert!(!json!(listed).to_string().contains("metadata_json")); + listing_ignores_build_history(store, credentials, (location, &listed)).await; assert!(matches!( read_person_node_status( store, @@ -268,3 +270,88 @@ 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] + ); +} diff --git a/ui/src/routes/catalog/+page.svelte b/ui/src/routes/catalog/+page.svelte index f2d8193514..afe9457a83 100644 --- a/ui/src/routes/catalog/+page.svelte +++ b/ui/src/routes/catalog/+page.svelte @@ -39,17 +39,6 @@ (f === 'failed' && n.lastOutcome === 'failed') || (f === 'warn' && n.lastOutcome === 'warn'); - 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; - 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.flatMap((n) => (n.group === null ? [] : [n.group])))]); @@ -58,6 +47,19 @@ 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; + // 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 columns = $derived(3 + (owners.length ? 1 : 0) + (showBuildState ? 2 : 0) + (showChecks ? 1 : 0)); const counts = $derived({ failed: all.filter((n) => n.lastOutcome === 'failed').length, From bf0062770301aeeeb2ff00a1495585f431e6e9d2 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Sat, 7 Nov 2026 12:43:19 +0000 Subject: [PATCH 3/4] perf(ui): batch the catalog's latest-build seeks One store round trip per node made a 3,000-node listing take about 0.6 s on PostgreSQL. Read the newest build of up to 100 nodes per statement, each still an exact seek on the node index, and bound a 3,000-node listing with a time limit on both stores. --- .../_helpers/fingerprint_reads.rs | 137 +++++++++++++----- .../main/list_person_catalog_nodes.rs | 41 +++--- .../tests/api/node_status/helpers.rs | 97 +++++++++++++ 3 files changed, 219 insertions(+), 56 deletions(-) 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/main/list_person_catalog_nodes.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/list_person_catalog_nodes.rs index 381cdeafa4..1803a07f26 100644 --- 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 @@ -1,4 +1,4 @@ -//! A person lists the current nodes with view on all of them; each build is one indexed seek. +//! A person lists the current nodes with view on all of them; builds are batched indexed seeks. use uuid::Uuid; @@ -10,9 +10,9 @@ 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_fingerprint; +use crate::warehouse_metadata::_helpers::fingerprint_reads::latest_fingerprints; use crate::warehouse_metadata::_helpers::node_status::destination; -use crate::warehouse_metadata::models::{CatalogNodeBuild, CatalogNodeRecord}; +use crate::warehouse_metadata::models::{CatalogNodeBuild, CatalogNodeRecord, FingerprintQuery}; pub async fn list_person_catalog_nodes( store: &ControlStore, @@ -32,23 +32,30 @@ pub async fn list_person_catalog_nodes( commit(transaction).await?; return Ok(Vec::new()); }; - let (mut transaction, nodes) = - read_snapshot_nodes(transaction, snapshot.snapshot_id, None).await?; - let mut records: Vec = Vec::with_capacity(nodes.len()); - for (key, node) in &nodes { - let mut last_build: Option = None; + 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) { - let (next, build) = - latest_fingerprint(transaction, (location_id, target, &query)).await?; - transaction = next; - last_build = build.map(|(record, _)| CatalogNodeBuild { - observed_at: record.observed_at, - run_id: record.run_id, - version_hash: record.version_hash, - }); + positions.push(position); + queries.push(query); } - records.push(catalog_node((key, node), snapshot.snapshot_id, last_build)); } + 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/tests/api/node_status/helpers.rs b/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs index 88e518ad71..0ab43bc73d 100644 --- a/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs @@ -6,6 +6,7 @@ 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::main::{begin::begin, commit::commit, query::query}; @@ -242,6 +243,7 @@ pub(crate) async fn node_status_reads( ); 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, @@ -355,3 +357,98 @@ async fn listing_ignores_build_history( timings[1] ); } + +/// Three thousand built nodes are read in batches, not one store round trip per node. +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(); + for name in &names { + input.nodes.insert(format!("model:{name}"), SnapshotNode { + kind: "model".into(), + definition: json!({"name":name,"destination":{"database":null,"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 in names.chunks(300) { + let mut parameters = vec![Parameter::Id(template)]; + let mut values: Vec = Vec::new(); + for name in chunk { + values.push(format!( + "(${},${},${})", + parameters.len() + 1, + parameters.len() + 2, + parameters.len() + 3 + )); + parameters.push(Parameter::Id(Uuid::now_v7())); + parameters.push(Parameter::Text(format!("wide-{name}"))); + parameters.push(Parameter::Text(name.clone())); + } + (transaction, _) = query( + transaction, + &format!( + "INSERT INTO sqb_warehouse_fingerprints (event_id,source_event_id,node_name,target_name,location_id,target, \ + observed_at,definition_text_hash,record_json,received_at,state_database,state_schema,node_type, \ + target_database,target_schema) \ + SELECT wide.column1,wide.column2,wide.column3,wide.column3,location_id,target,observed_at, \ + definition_text_hash,record_json,received_at,state_database,state_schema,node_type, \ + target_database,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 built = listed + .iter() + .filter(|node| node.node_key.starts_with("model:wide_")) + .filter(|node| { + node.last_build + .as_ref() + .is_some_and(|build| 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] + ); +} From 647bd6918c88177a1d61f9f411082d2c8bddf3eb Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Sat, 7 Nov 2026 12:57:39 +0000 Subject: [PATCH 4/4] test(ui): give each wide catalog node its own build and database shape Every wide node copied one build, so the guard could not tell whether a node received its own row. Each node now has a distinct build time and alternate nodes live in a named database, and the guard checks each node's own build. --- .../tests/api/node_status/helpers.rs | 56 +++++++++++++------ 1 file changed, 38 insertions(+), 18 deletions(-) diff --git a/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs b/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs index 0ab43bc73d..be16d1b217 100644 --- a/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/node_status/helpers.rs @@ -358,17 +358,24 @@ async fn listing_ignores_build_history( ); } -/// Three thousand built nodes are read in batches, not one store round trip per node. +/// 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(); - for name in &names { + 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":null,"schema":"main","name":name}}), + definition: json!({"name":name,"destination":{"database":database(index),"schema":"main","name":name}}), raw_sql: None, compiled_sql: None, python_source: None, @@ -389,29 +396,33 @@ async fn listing_reads_wide_targets_in_batches( .await .expect("template build"); let template: Uuid = serde_json::from_value(rows[0]["event_id"].clone()).expect("event id"); - for chunk in names.chunks(300) { + for (chunk_index, chunk) in names.chunks(300).enumerate() { let mut parameters = vec![Parameter::Id(template)]; let mut values: Vec = Vec::new(); - for name in chunk { + 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() + 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,definition_text_hash,record_json,received_at,state_database,state_schema,node_type, \ - target_database,target_schema) \ - SELECT wide.column1,wide.column2,wide.column3,wide.column3,location_id,target,observed_at, \ - definition_text_hash,record_json,received_at,state_database,state_schema,node_type, \ - target_database,target_schema \ + 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(",") ), @@ -433,13 +444,22 @@ async fn listing_reads_wide_targets_in_batches( .await .expect("wide catalog"); timings.push(started.elapsed()); - let built = listed + 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() - .filter(|node| node.node_key.starts_with("model:wide_")) - .filter(|node| { - node.last_build - .as_ref() - .is_some_and(|build| json!(build.run_id) == rows[0]["run_id"]) + .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");