diff --git a/Cargo.lock b/Cargo.lock index fc7d69ec0b..58443e066c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4290,6 +4290,8 @@ dependencies = [ "serde", "serde_json", "sqlbuild-core", + "sqlbuild-cursor", + "sqlbuild-model-config", "sqlbuild-recall", "sqlbuild-state", "sqlx-core", diff --git a/Cargo.toml b/Cargo.toml index d2927259d1..bb3bb388e2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,8 +42,8 @@ order = [ "sqlbuild-config", "sqlbuild-host", "sqlbuild-recall", - "sqlbuild-orchestration", "sqlbuild-model-config", + "sqlbuild-orchestration", "sqlbuild-discovery", "sqlbuild-render", "sqlbuild-scopes", diff --git a/benchmarks/duckshop/edge_cases/orchestration/microbatch_coverage.json b/benchmarks/duckshop/edge_cases/orchestration/microbatch_coverage.json new file mode 100644 index 0000000000..7b2af242fa --- /dev/null +++ b/benchmarks/duckshop/edge_cases/orchestration/microbatch_coverage.json @@ -0,0 +1,368 @@ +{ + "microbatches": [ + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-old-generation", + "created_at": "2026-01-01T00:30:00Z", + "observed_at": "2026-01-01T00:30:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T10:00:00+00:00", + "partition_end": "2026-01-01T11:00:00+00:00", + "execution_run_id": "run-0", + "model_version_hash": "orders-v1", + "physical_generation_id": "orders-generation-0" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-other-scope", + "created_at": "2026-01-01T00:35:00Z", + "observed_at": "2026-01-01T00:35:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T12:00:00+00:00", + "partition_end": "2026-01-01T13:00:00+00:00", + "execution_run_id": "run-0", + "model_version_hash": "orders-v1", + "scope_key": "orders:main.orders_previous" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-older-requirement", + "created_at": "2026-01-01T00:45:00Z", + "observed_at": "2026-01-01T00:45:00Z", + "record_type": "replay_requirement", + "partition_start": null, + "partition_end": null, + "completion_type": null, + "execution_run_id": "run-replay", + "model_version_hash": "orders-v9", + "replay_requirement_id": "orders-older-requirement", + "required_model_version_hash": "orders-v9", + "previous_model_version_hash": "orders-v8", + "replay_policy": "full" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-1", + "created_at": "2026-01-01T01:30:00Z", + "observed_at": "2026-01-01T01:30:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01 00:00:00+00", + "partition_end": "2026-01-01T09:00:00+08:00", + "execution_run_id": "run-1", + "model_version_hash": "orders-v1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-2-older", + "created_at": "2026-01-01T01:00:00Z", + "observed_at": "2026-01-01T01:00:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T01:00:00+00:00", + "partition_end": "2026-01-01T02:00:00+00:00", + "execution_run_id": "run-old", + "model_version_hash": "orders-v1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-2", + "created_at": "2026-01-01T02:30:00Z", + "observed_at": "2026-01-01T02:30:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T09:00:00+08:00", + "partition_end": "2026-01-01T02:00:00+00:00", + "execution_run_id": "run-2", + "model_version_hash": "orders-v1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-4", + "created_at": "2026-01-01T04:30:00Z", + "observed_at": "2026-01-01T04:30:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T03:00:00+00:00", + "partition_end": "2026-01-01T04:00:00+00:00", + "execution_run_id": "run-3", + "model_version_hash": "orders-v2" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-5", + "created_at": "2026-01-01T05:30:00Z", + "observed_at": "2026-01-01T05:30:00Z", + "record_type": "synthetic_completion", + "partition_start": "2026-01-01T04:00:00+00:00", + "partition_end": "2026-01-01T05:00:00+00:00", + "execution_run_id": "", + "fingerprint_status": "unknown", + "model_version_hash": null, + "definition_hash": null, + "synthetic_reason": "Retained rows", + "unaccounted_policy": "synthesize", + "observed_row_count": 12, + "completion_type": "initial" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-4-rerun", + "created_at": "2026-01-01T05:45:00Z", + "observed_at": "2026-01-01T05:45:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T03:00:00+00:00", + "partition_end": "2026-01-01T04:00:00+00:00", + "execution_run_id": "run-5", + "model_version_hash": "orders-v2" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-7", + "created_at": "2026-01-01T05:50:00Z", + "observed_at": "2026-01-01T05:50:00Z", + "record_type": "synthetic_completion", + "partition_start": "2026-01-01T06:00:00+00:00", + "partition_end": "2026-01-01T07:00:00+00:00", + "execution_run_id": "", + "fingerprint_status": "unknown", + "model_version_hash": null, + "definition_hash": null, + "synthetic_reason": "Retained rows", + "unaccounted_policy": "synthesize", + "observed_row_count": 12, + "completion_type": "initial" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-requirement", + "created_at": "2026-01-01T06:30:00Z", + "observed_at": "2026-01-01T06:30:00Z", + "record_type": "replay_requirement", + "partition_start": null, + "partition_end": null, + "completion_type": null, + "execution_run_id": "run-replay", + "model_version_hash": "orders-v2", + "replay_requirement_id": "orders-requirement", + "required_model_version_hash": "orders-v2", + "previous_model_version_hash": "orders-v1", + "replay_policy": "full" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "customers-batch-1", + "created_at": "2026-01-01T01:30:00Z", + "observed_at": "2026-01-01T01:30:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T00:00:00+00:00", + "partition_end": "2026-01-01T01:00:00+00:00", + "execution_run_id": "run-1", + "model_version_hash": "customers-v1", + "model_name": "customers", + "scope_key": "customers:main.customers", + "target_name": "customers" + }, + { + "cursor_column": "event_id", + "cursor_type": "integer", + "batch_size": "1", + "run_start": "2", + "run_end": "11", + "event_id": "events-batch-2-older", + "created_at": "2026-01-01T00:30:00Z", + "observed_at": "2026-01-01T00:30:00Z", + "record_type": "partition_completion", + "partition_start": "2", + "partition_end": "5", + "execution_run_id": "run-old", + "model_version_hash": "events-v1", + "model_name": "events", + "scope_key": "events:main.events", + "target_name": "events", + "physical_generation_id": "events-generation-1" + }, + { + "cursor_column": "event_id", + "cursor_type": "integer", + "batch_size": "1", + "run_start": "2", + "run_end": "11", + "event_id": "events-batch-2", + "created_at": "2026-01-01T01:00:00Z", + "observed_at": "2026-01-01T01:00:00Z", + "record_type": "partition_completion", + "partition_start": "2", + "partition_end": "9", + "execution_run_id": "run-a", + "model_version_hash": "events-v1", + "model_name": "events", + "scope_key": "events:main.events", + "target_name": "events", + "physical_generation_id": "events-generation-1" + }, + { + "cursor_column": "event_id", + "cursor_type": "integer", + "batch_size": "1", + "run_start": "2", + "run_end": "11", + "event_id": "events-batch-9", + "created_at": "2026-01-01T02:00:00Z", + "observed_at": "2026-01-01T02:00:00Z", + "record_type": "partition_completion", + "partition_start": "9", + "partition_end": "10", + "execution_run_id": "run-b", + "model_version_hash": "events-v1", + "model_name": "events", + "scope_key": "events:main.events", + "target_name": "events", + "physical_generation_id": "events-generation-1" + }, + { + "cursor_column": "event_id", + "cursor_type": "integer", + "batch_size": "1", + "run_start": "2", + "run_end": "11", + "event_id": "events-batch-10", + "created_at": "2026-01-01T03:00:00Z", + "observed_at": "2026-01-01T03:00:00Z", + "record_type": "partition_completion", + "partition_start": "10", + "partition_end": "11", + "execution_run_id": "run-c", + "model_version_hash": "events-v1", + "model_name": "events", + "scope_key": "events:main.events", + "target_name": "events", + "physical_generation_id": "events-generation-1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01T00:00:00+00:00", + "run_end": "2026-01-01T02:00:00+00:00", + "event_id": "payments-batch-0", + "created_at": "2026-01-01T00:30:00Z", + "observed_at": "2026-01-01T00:30:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T00:00:00+00:00", + "partition_end": "2026-01-01T01:00:00+00:00", + "execution_run_id": "run-p1", + "model_version_hash": "payments-v1", + "model_name": "payments", + "scope_key": "payments:main.payments", + "target_name": "payments", + "physical_generation_id": "payments-generation-1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01T00:00:00+00:00", + "run_end": "2026-01-01T02:00:00+00:00", + "event_id": "payments-batch-1", + "created_at": "2026-01-01T00:40:00Z", + "observed_at": "2026-01-01T00:40:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T01:00:00+00:00", + "partition_end": "2026-01-01T02:00:00+00:00", + "execution_run_id": "run-p2", + "model_version_hash": "payments-v1", + "model_name": "payments", + "scope_key": "payments:main.payments", + "target_name": "payments", + "physical_generation_id": "payments-generation-1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01T00:00:00+00:00", + "run_end": "2026-01-01T02:00:00+00:00", + "event_id": "payments-requirement", + "created_at": "2026-01-01T01:00:00Z", + "observed_at": "2026-01-01T01:00:00Z", + "record_type": "replay_requirement", + "partition_start": null, + "partition_end": null, + "completion_type": null, + "execution_run_id": "run-replay", + "model_version_hash": "payments-v2", + "replay_requirement_id": "payments-requirement", + "required_model_version_hash": "payments-v2", + "previous_model_version_hash": "payments-v1", + "replay_policy": "full", + "model_name": "payments", + "scope_key": "payments:main.payments", + "target_name": "payments", + "physical_generation_id": "payments-generation-1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01T00:00:00+00:00", + "run_end": "2026-01-01T02:00:00+00:00", + "event_id": "payments-batch-0-v3", + "created_at": "2026-01-01T02:00:00Z", + "observed_at": "2026-01-01T02:00:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T00:00:00+00:00", + "partition_end": "2026-01-01T01:00:00+00:00", + "execution_run_id": "run-p3", + "model_version_hash": "payments-v3", + "model_name": "payments", + "scope_key": "payments:main.payments", + "target_name": "payments", + "physical_generation_id": "payments-generation-1" + } + ] +} diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index ae6e30de48..de824cbd63 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -338,6 +338,22 @@ export interface paths { patch?: never; trace?: never; }; + "/api/control/locations/{location_id}/targets/{target}/microbatches/{model}": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get: operations["readPersonMicrobatchCoverage"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/control/locations/{location_id}/targets/{target}/node-status": { parameters: { query?: never; @@ -824,6 +840,11 @@ export interface components { /** Format: uuid */ request_id: string; }; + /** + * @description How one recorded stretch of a microbatch model's cursor stands. + * @enum {string} + */ + BatchState: "complete" | "gap" | "replay_required"; /** @enum {string} */ BuiltinRole: "viewer" | "operator" | "developer" | "admin"; /** @description Browser cancellation cannot supply an actor, claim or lifecycle transition. */ @@ -934,6 +955,14 @@ export interface components { /** Format: uuid */ request_id: string; }; + /** @description One stretch of the cursor, as raw bound texts. */ + CoverageBatch: { + end: string; + /** @description The run that executed the latest completion here; absent for gaps and external completions. */ + run_id?: string | null; + start: string; + state: components["schemas"]["BatchState"]; + }; /** @description The current successful snapshot, read with a consistent generation and change set. */ CurrentSnapshot: { changes: components["schemas"]["SnapshotChange"][]; @@ -1162,6 +1191,25 @@ export interface components { /** Format: uuid */ scan_id: string; }; + /** @description Recorded coverage of one model's current relation; tracked only for concurrent microbatch models. */ + MicrobatchCoverage: { + batch_size: string; + /** @description The newest batches, oldest first. */ + batches: components["schemas"]["CoverageBatch"][]; + /** @description Whether older batches were left out of `batches`. */ + batches_truncated: boolean; + /** @description End of the completed range that starts at `earliest` without a hole. */ + contiguous_frontier?: string | null; + cursor_column: string; + cursor_type: string; + /** @description Start of the earliest completed partition. */ + earliest: string; + /** @description Holes between completed partitions, oldest first. */ + known_gaps: components["schemas"]["CoverageBatch"][]; + model: string; + physical_generation_id: string; + target_name: string; + }; MutationResponse: { applied: boolean; }; @@ -2682,6 +2730,48 @@ export interface operations { }; }; }; + readPersonMicrobatchCoverage: { + parameters: { + query?: { + /** @description Maximum newest batches; defaults to 200, capped at 1000 */ + limit?: number; + }; + header?: never; + path: { + location_id: string; + target: string; + model: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Null when the model has no tracked coverage */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": null | components["schemas"]["MicrobatchCoverage"]; + }; + }; + /** @description Session rejected */ + 401: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + 403: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ApiErrorBody"]; + }; + }; + }; + }; readPersonNodeStatus: { parameters: { query: { diff --git a/crates/sqlbuild-orchestration/Cargo.toml b/crates/sqlbuild-orchestration/Cargo.toml index a2cbd1322e..f680f7a1ea 100644 --- a/crates/sqlbuild-orchestration/Cargo.toml +++ b/crates/sqlbuild-orchestration/Cargo.toml @@ -22,6 +22,8 @@ rusqlite.workspace = true serde.workspace = true serde_json.workspace = true sqlbuild-core.workspace = true +sqlbuild-cursor.workspace = true +sqlbuild-model-config.workspace = true sqlbuild-recall.workspace = true sqlbuild-state.workspace = true sqlx-core.workspace = true 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 0a955d6235..0e14f0579e 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs @@ -129,6 +129,10 @@ pub(crate) fn person_routes() -> Router { "/api/control/locations/{location_id}/targets/{target}/maintenance", get(crate::api::_helpers::resources::person_maintenance::read), ) + .route( + "/api/control/locations/{location_id}/targets/{target}/microbatches/{model}", + get(crate::api::_helpers::resources::person_maintenance::coverage), + ) .layer(DefaultBodyLimit::max(PASSWORD_BODY_BYTES)) } diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs index 1c704c0b52..2793482bce 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs @@ -1,4 +1,4 @@ -//! Maintenance history reads use live person grants over the whole target. +//! Maintenance history and microbatch coverage reads use live person grants over the whole target. use axum::extract::{Path, Query, State}; use axum::{Extension, Json}; @@ -8,7 +8,10 @@ use crate::api::errors::ApiError; use crate::credentials::models::{AuditContext, AuthenticatedPerson, DefaultRole}; use crate::store::models::ControlStore; use crate::warehouse_history::main::read_person_maintenance::read_person_maintenance; -use crate::warehouse_history::models::{MaintenanceHistory, MaintenanceQuery}; +use crate::warehouse_history::main::read_person_microbatch_coverage::read_person_microbatch_coverage; +use crate::warehouse_history::models::{ + CoverageQuery, MaintenanceHistory, MaintenanceQuery, MicrobatchCoverage, +}; #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/maintenance", operation_id = "readPersonMaintenance", params(("location_id" = Uuid, Path), ("target" = String, Path), @@ -32,3 +35,29 @@ pub(crate) async fn read( .await?, )) } + +#[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/microbatches/{model}", operation_id = "readPersonMicrobatchCoverage", + params(("location_id" = Uuid, Path), ("target" = String, Path), ("model" = String, Path), + ("limit" = Option, Query, description = "Maximum newest batches; defaults to 200, capped at 1000")), + responses((status = 200, body = Option, description = "Null when the model has no tracked coverage"), + (status = 401, description = "Session rejected"), + (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))] +pub(crate) async fn coverage( + (State(store), Extension(context)): (State, Extension), + Extension(default_role): Extension, + (Path((location_id, target, model)), Query(options)): ( + Path<(Uuid, String, String)>, + Query, + ), + Extension(AuthenticatedPerson(principal, _)): Extension, +) -> Result>, ApiError> { + Ok(Json( + read_person_microbatch_coverage( + &store, + (&context, principal), + ((location_id, &target), (&model, &options)), + default_role, + ) + .await?, + )) +} diff --git a/crates/sqlbuild-orchestration/src/api/models.rs b/crates/sqlbuild-orchestration/src/api/models.rs index 0174bb33c6..0b13d5a793 100644 --- a/crates/sqlbuild-orchestration/src/api/models.rs +++ b/crates/sqlbuild-orchestration/src/api/models.rs @@ -29,6 +29,7 @@ use crate::api::_helpers::resources::{ person_checks::sql, person_checks::evidence, person_maintenance::read, + person_maintenance::coverage, person_location_work::request, person_location_work::result, person_node_sql::read, diff --git a/crates/sqlbuild-orchestration/src/store/schema.sql b/crates/sqlbuild-orchestration/src/store/schema.sql index 537f4ccdd4..a7e4973c6d 100644 --- a/crates/sqlbuild-orchestration/src/store/schema.sql +++ b/crates/sqlbuild-orchestration/src/store/schema.sql @@ -464,6 +464,7 @@ CREATE TABLE sqb_warehouse_microbatches ( scan_id {uuid} ); CREATE INDEX sqb_microbatches_namespace ON sqb_warehouse_microbatches (location_id,target,state_database,state_schema,observed_at,event_id); +CREATE INDEX sqb_microbatches_model ON sqb_warehouse_microbatches (location_id,target,model_name); CREATE INDEX sqb_freshness_source_latest ON sqb_warehouse_freshness ( location_id, target, (record_json ->> 'source_name'), (record_json ->> 'target_database'), (record_json ->> 'target_schema'), diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/coverage.rs b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/coverage.rs new file mode 100644 index 0000000000..83ca28dac9 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/coverage.rs @@ -0,0 +1,331 @@ +//! Microbatch coverage projected from the mirrored completion and replay-requirement facts. + +use std::cmp::Ordering; +use std::collections::HashMap; + +use chrono::{DateTime, Datelike, NaiveDateTime, Timelike}; +use sqlbuild_cursor::algebra::main::normalize_timestamp_text::normalize_timestamp_text; +use sqlbuild_cursor::algebra::main::parse_integer_text::parse_integer_text; +use sqlbuild_cursor::algebra::models::{CivilDate, LocalDateTime, SortKey}; +use sqlbuild_model_config::model_validation::main::parse_iso_timestamp::parse_iso_timestamp; +use sqlbuild_state::microbatches::main::project_coverage::project_coverage; +use sqlbuild_state::microbatches::models::{ + BoundInterval, CompletionFact, CoverageProjection, CursorBound, TextInterval, +}; +use uuid::Uuid; + +use crate::store::errors::ControlError; +use crate::store::models::{ControlTransaction, Parameter}; +use crate::warehouse_history::_helpers::maintenance::records; +use crate::warehouse_history::models::{BatchState, CoverageBatch, MicrobatchCoverage}; +use crate::warehouse_metadata::models::WarehouseMicrobatchRecord; + +const ORDINARY_COMPLETION: &str = "partition_completion"; +const SYNTHETIC_COMPLETION: &str = "synthetic_completion"; +const REPLAY_REQUIREMENT: &str = "replay_requirement"; +const UNKNOWN_FINGERPRINT: &str = "unknown"; +const INTEGER_CURSOR: &str = "integer"; +/// The `datetime.fromisoformat` rules cursor bounds are validated with. +const ISOFORMAT_PYTHON_MINOR: u8 = 12; + +/// The key a cursor bound sorts by: the integer, or the wall time of a naive value and the UTC instant of an aware one. +fn sort_key(text: &str, cursor_type: &str) -> Option { + if cursor_type == INTEGER_CURSOR { + return parse_integer_text(text).map(SortKey::Integer); + } + let (micros, _) = parse_iso_timestamp(&normalize_timestamp_text(text), ISOFORMAT_PYTHON_MINOR)?; + let wall: NaiveDateTime = DateTime::from_timestamp_micros(micros)?.naive_utc(); + Some(SortKey::Moment(LocalDateTime { + date: CivilDate { + year: i64::from(wall.year()), + month: wall.month(), + day: wall.day(), + }, + hour: wall.hour(), + minute: wall.minute(), + second: wall.second(), + microsecond: wall.nanosecond() / 1000, + })) +} + +fn order(left: &CursorBound, right: &CursorBound) -> Ordering { + match (&left.key, &right.key) { + (SortKey::Integer(left), SortKey::Integer(right)) => left.cmp(right), + (SortKey::Moment(left), SortKey::Moment(right)) => left.cmp(right), + _ => Ordering::Equal, + } +} + +/// Bound texts of one cursor type, each parsed once. +struct Bounds<'a> { + cursor_type: &'a str, + keys: HashMap, +} + +impl Bounds<'_> { + fn bound(&mut self, text: &str) -> Result { + if let Some(key) = self.keys.get(text) { + return Ok(CursorBound { + text: text.into(), + key: *key, + }); + } + let key: SortKey = sort_key(text, self.cursor_type).ok_or_else(|| { + ControlError::Schema(format!( + "stored microbatch bound is not a {} cursor value", + self.cursor_type + )) + })?; + self.keys.insert(text.into(), key); + Ok(CursorBound { + text: text.into(), + key, + }) + } + + fn interval(&mut self, interval: &TextInterval) -> Result { + Ok(BoundInterval { + start: self.bound(&interval.start)?, + end: self.bound(&interval.end)?, + }) + } +} + +fn newest<'a>( + records: impl Iterator, +) -> Option<&'a WarehouseMicrobatchRecord> { + records.max_by(|left, right| { + (left.created_at, &left.event_id).cmp(&(right.created_at, &right.event_id)) + }) +} + +/// Completion facts of the current relation and the records they came from, in the same order. +fn completions<'a>( + scope: &[&'a WarehouseMicrobatchRecord], + bounds: &mut Bounds<'_>, +) -> Result<(Vec, Vec<&'a WarehouseMicrobatchRecord>), ControlError> { + let mut facts: Vec = Vec::new(); + let mut sources: Vec<&WarehouseMicrobatchRecord> = Vec::new(); + for record in scope { + let (Some(start), Some(end)) = (&record.partition_start, &record.partition_end) else { + continue; + }; + if ![ORDINARY_COMPLETION, SYNTHETIC_COMPLETION].contains(&record.record_type.as_str()) { + continue; + } + facts.push(CompletionFact { + interval: BoundInterval { + start: bounds.bound(start)?, + end: bounds.bound(end)?, + }, + ordinary: record.record_type == ORDINARY_COMPLETION, + unknown_fingerprint: record.fingerprint_status == UNKNOWN_FINGERPRINT, + created: (record.created_at - DateTime::UNIX_EPOCH).as_seconds_f64(), + event_id: record.event_id.clone(), + }); + sources.push(record); + } + Ok((facts, sources)) +} + +/// The newest replay requirement's range and required version, unless a newer completion at a third version ended it, as the executor decides. +fn replay_range( + scope: &[&WarehouseMicrobatchRecord], + bounds: &mut Bounds<'_>, +) -> Result)>, ControlError> { + let Some(requirement) = newest( + scope + .iter() + .copied() + .filter(|record| record.record_type == REPLAY_REQUIREMENT), + ) else { + return Ok(None); + }; + let superseded: bool = scope.iter().any(|record| { + record.record_type == ORDINARY_COMPLETION + && (record.created_at, &record.event_id) + > (requirement.created_at, &requirement.event_id) + && record.model_version_hash.as_ref().is_some_and(|version| { + Some(version) != requirement.required_model_version_hash.as_ref() + }) + }); + if superseded { + return Ok(None); + } + Ok(Some(( + bounds.interval(&TextInterval { + start: requirement.run_start.clone(), + end: requirement.run_end.clone(), + })?, + requirement.required_model_version_hash.clone(), + ))) +} + +fn overlaps(left: &BoundInterval, right: &BoundInterval) -> bool { + order(&left.start, &right.end).is_lt() && order(&right.start, &left.end).is_lt() +} + +/// One stretch of the cursor and the completion that accounts for it, if any. +struct Piece { + interval: BoundInterval, + state: BatchState, + completion: Option, +} + +/// Adjacent stretches of one completion, or of one hole, read as one batch; empty stretches are dropped. +fn merged(mut pieces: Vec) -> Vec { + pieces.sort_by(|left, right| order(&left.interval.start, &right.interval.start)); + let mut kept: Vec = Vec::with_capacity(pieces.len()); + for piece in pieces { + if order(&piece.interval.start, &piece.interval.end).is_ge() { + continue; + } + match kept.last_mut() { + Some(last) + if last.completion == piece.completion + && last.state == piece.state + && order(&last.interval.end, &piece.interval.start).is_eq() => + { + last.interval.end = piece.interval.end; + } + _ => kept.push(piece), + } + } + kept +} + +fn batch(piece: &Piece, sources: &[&WarehouseMicrobatchRecord]) -> CoverageBatch { + CoverageBatch { + start: piece.interval.start.text.clone(), + end: piece.interval.end.text.clone(), + state: piece.state, + run_id: piece + .completion + .map(|completion| sources[completion].execution_run_id.clone()) + .filter(|run| !run.is_empty()), + } +} + +/// Completed segments, spelled as their completion recorded them where they share its bounds, and the holes between them, oldest first; and the holes alone. +fn batches( + projection: &CoverageProjection, + sources: &[&WarehouseMicrobatchRecord], + (bounds, replay): (&mut Bounds<'_>, Option<&(BoundInterval, Option)>), +) -> Result<(Vec, Vec), ControlError> { + let mut completed: Vec = Vec::new(); + for segment in &projection.segments { + let mut interval: BoundInterval = bounds.interval(&segment.interval)?; + let source: &WarehouseMicrobatchRecord = sources[segment.completion]; + for (recorded, bound) in [ + (&source.partition_start, &mut interval.start), + (&source.partition_end, &mut interval.end), + ] { + let Some(recorded) = recorded else { continue }; + let recorded: CursorBound = bounds.bound(recorded)?; + if order(&recorded, bound).is_eq() { + *bound = recorded; + } + } + let stale: bool = replay.is_some_and(|(range, required)| { + overlaps(&interval, range) + && source.fingerprint_status != UNKNOWN_FINGERPRINT + && source.model_version_hash != *required + }); + completed.push(Piece { + interval, + state: if stale { + BatchState::ReplayRequired + } else { + BatchState::Complete + }, + completion: Some(segment.completion), + }); + } + let mut holes: Vec = Vec::new(); + for hole in projection + .known_missing + .iter() + .chain(&projection.unaccounted) + { + holes.push(Piece { + interval: bounds.interval(hole)?, + state: BatchState::Gap, + completion: None, + }); + } + let holes: Vec = merged(holes); + let gaps: Vec = holes.iter().map(|piece| batch(piece, sources)).collect(); + let mut every: Vec = merged(completed); + every.extend(holes); + every.sort_by(|left, right| order(&left.interval.start, &right.interval.start)); + Ok(( + every.iter().map(|piece| batch(piece, sources)).collect(), + gaps, + )) +} + +/// Coverage of the model's newest recorded relation and generation, or `None` when none is tracked. +pub(crate) async fn microbatch_coverage( + transaction: ControlTransaction, + (location_id, target): (Uuid, &str), + (model, limit): (&str, u32), +) -> Result<(ControlTransaction, Option), ControlError> { + let (transaction, history): (ControlTransaction, Vec) = records(transaction, + "SELECT record_json AS record FROM sqb_warehouse_microbatches WHERE location_id=$1 AND target=$2 AND model_name=$3", + vec![Parameter::Id(location_id), Parameter::Text(target.into()), Parameter::Text(model.into())]).await?; + let Some(current) = newest(history.iter()) else { + return Ok((transaction, None)); + }; + let scope: Vec<&WarehouseMicrobatchRecord> = history + .iter() + .filter(|record| { + record.scope_key == current.scope_key + && record.physical_generation_id == current.physical_generation_id + }) + .collect(); + let mut bounds = Bounds { + cursor_type: ¤t.cursor_type, + keys: HashMap::new(), + }; + let (facts, sources) = completions(&scope, &mut bounds)?; + let (Some(earliest), Some(latest)) = ( + facts + .iter() + .map(|fact| &fact.interval.start) + .min_by(|left, right| order(left, right)), + facts + .iter() + .map(|fact| &fact.interval.end) + .max_by(|left, right| order(left, right)), + ) else { + return Ok((transaction, None)); + }; + let projection: CoverageProjection = project_coverage( + &facts, + &[BoundInterval { + start: earliest.clone(), + end: latest.clone(), + }], + ); + let replay: Option<(BoundInterval, Option)> = replay_range(&scope, &mut bounds)?; + let (mut recent, known_gaps) = batches(&projection, &sources, (&mut bounds, replay.as_ref()))?; + let keep: usize = usize::try_from(limit.clamp(1, 1000)).unwrap_or(usize::MAX); + let dropped: usize = recent.len().saturating_sub(keep); + recent.drain(..dropped); + Ok(( + transaction, + Some(MicrobatchCoverage { + model: model.into(), + target_name: current.target_name.clone(), + physical_generation_id: current.physical_generation_id.clone(), + cursor_column: current.cursor_column.clone(), + cursor_type: current.cursor_type.clone(), + batch_size: current.batch_size.clone(), + earliest: earliest.text.clone(), + contiguous_frontier: projection.contiguous_frontier.clone(), + known_gaps, + batches: recent, + batches_truncated: dropped > 0, + }), + )) +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs index 279f7a16dc..6c52ab9094 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs @@ -17,7 +17,8 @@ struct Recorded { record: T, } -async fn records( +/// The `record` column of every row `sql` selects. +pub(crate) async fn records( transaction: ControlTransaction, sql: &str, parameters: Vec, diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs index e0260321e5..51c3c3a651 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs @@ -1,2 +1,3 @@ pub(crate) mod checks; +pub(crate) mod coverage; pub(crate) mod maintenance; diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs index 5a314e91ca..55736dc22c 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs @@ -3,3 +3,4 @@ pub mod read_person_check_evidence; pub mod read_person_check_sql; pub mod read_person_checks; pub mod read_person_maintenance; +pub mod read_person_microbatch_coverage; diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs new file mode 100644 index 0000000000..a89c679521 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs @@ -0,0 +1,34 @@ +//! Live view permission over the whole target and one model's recorded microbatch coverage. + +use uuid::Uuid; + +use crate::authorization::main::authorize_listing::authorize_listing; +use crate::credentials::models::{AuditContext, DefaultRole}; +use crate::store::errors::ControlError; +use crate::store::main::{begin::begin, commit::commit}; +use crate::store::models::{ControlStore, TransactionMode}; +use crate::warehouse_history::_helpers::coverage::microbatch_coverage; +use crate::warehouse_history::models::{CoverageQuery, MicrobatchCoverage}; + +pub async fn read_person_microbatch_coverage( + store: &ControlStore, + credentials: (&AuditContext, Uuid), + ((location_id, target), (model, options)): ((Uuid, &str), (&str, &CoverageQuery)), + default_role: DefaultRole, +) -> Result, ControlError> { + let (transaction, _) = authorize_listing( + begin(store, TransactionMode::Write).await?, + credentials, + (location_id, target), + default_role, + ) + .await?; + let (transaction, coverage) = microbatch_coverage( + transaction, + (location_id, target), + (model, options.limit.unwrap_or(200)), + ) + .await?; + commit(transaction).await?; + Ok(coverage) +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs index 7488923fa9..e87cb64e62 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs @@ -1,4 +1,4 @@ -//! Person reads of warehouse-owned history mirrored into the control store: checks and maintenance. +//! Person reads of warehouse-owned history mirrored into the control store: checks, maintenance and microbatch coverage. pub(crate) mod _helpers; pub mod main; diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/models.rs b/crates/sqlbuild-orchestration/src/warehouse_history/models.rs index cca287650b..57c26430cb 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_history/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/models.rs @@ -112,3 +112,51 @@ pub struct MaintenanceQuery { pub node: Option, pub limit: Option, } + +/// How one recorded stretch of a microbatch model's cursor stands. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum BatchState { + Complete, + /// A hole between completed partitions. + Gap, + /// Completed at another version than the newest replay requirement covering it. + ReplayRequired, +} + +/// One stretch of the cursor, as raw bound texts. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct CoverageBatch { + pub start: String, + pub end: String, + pub state: BatchState, + /// The run that executed the latest completion here; absent for gaps and external completions. + pub run_id: Option, +} + +/// Recorded coverage of one model's current relation; tracked only for concurrent microbatch models. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct MicrobatchCoverage { + pub model: String, + pub target_name: String, + pub physical_generation_id: String, + pub cursor_column: String, + pub cursor_type: String, + pub batch_size: String, + /// Start of the earliest completed partition. + pub earliest: String, + /// End of the completed range that starts at `earliest` without a hole. + pub contiguous_frontier: Option, + /// Holes between completed partitions, oldest first. + pub known_gaps: Vec, + /// The newest batches, oldest first. + pub batches: Vec, + /// Whether older batches were left out of `batches`. + pub batches_truncated: bool, +} + +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CoverageQuery { + pub limit: Option, +} diff --git a/crates/sqlbuild-orchestration/tests/api/test_contracts.rs b/crates/sqlbuild-orchestration/tests/api/test_contracts.rs index 25527b6ad0..4d798f30a5 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_contracts.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_contracts.rs @@ -50,6 +50,8 @@ mod test_location_health; mod test_location_work; #[path = "test_maintenance.rs"] mod test_maintenance; +#[path = "test_microbatch_coverage.rs"] +mod test_microbatch_coverage; #[path = "test_oidc.rs"] mod test_oidc; #[path = "test_oidc_groups.rs"] diff --git a/crates/sqlbuild-orchestration/tests/api/test_microbatch_coverage.rs b/crates/sqlbuild-orchestration/tests/api/test_microbatch_coverage.rs new file mode 100644 index 0000000000..79bffe3bbc --- /dev/null +++ b/crates/sqlbuild-orchestration/tests/api/test_microbatch_coverage.rs @@ -0,0 +1,169 @@ +//! Microbatch coverage through the public API: completed segments, holes and stale versions. + +use reqwest::StatusCode; +use serde_json::{Value, json}; + +use crate::api::tests::helpers::stop_api; +use crate::api::tests::test_types::WarehouseMetadataTestCase; +use crate::api::tests::warehouse_history_helpers::{ + batch, coverage_batch, fixture, get, login, records, seed_target, start_password_api, +}; +use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; +use crate::store::tests::store_test_types::Backend; +use crate::warehouse_metadata::main::ingest_microbatches::ingest_microbatches; + +const HOUR_0: &str = "2026-01-01 00:00:00+00"; +const HOUR_1: &str = "2026-01-01T09:00:00+08:00"; +const HOUR_2: &str = "2026-01-01T02:00:00+00:00"; +const HOUR_3: &str = "2026-01-01T03:00:00+00:00"; +const HOUR_4: &str = "2026-01-01T04:00:00+00:00"; +const HOUR_5: &str = "2026-01-01T05:00:00+00:00"; +const HOUR_6: &str = "2026-01-01T06:00:00+00:00"; +const HOUR_7: &str = "2026-01-01T07:00:00+00:00"; + +#[test] +fn given_microbatch_history_when_people_read_coverage_then_holes_and_stale_versions_are_marked() { + let test_cases = [ + WarehouseMetadataTestCase { + description: "SQLite microbatch coverage", + backend: Backend::Sqlite, + expected_observations: 20, + }, + WarehouseMetadataTestCase { + description: "PostgreSQL microbatch coverage", + backend: Backend::Postgres, + expected_observations: 20, + }, + ]; + for test_case in test_cases { + run(async { + let fixture_store = open_test_store(test_case.backend).await.expect("store"); + let store = &fixture_store.store; + let (location, session_id) = seed_target(store).await; + let history = records( + "microbatch_history.json", + &fixture("microbatch_coverage.json")["microbatches"], + ); + assert_eq!(history.len(), test_case.expected_observations); + ingest_microbatches( + store, + (location, "development"), + &batch(session_id, "microbatches", history), + ) + .await + .expect("real microbatch ingestion"); + + let (api, client) = start_password_api(store).await; + let viewer = login(&client, &api.url, "orders-viewer").await; + let path = + format!("/api/control/locations/{location}/targets/development/microbatches"); + let read = |model: &str| format!("{path}/{model}"); + + let (status, orders) = get(&client, (&api.url, &viewer), &read("orders")).await; + assert_eq!(status, StatusCode::OK, "{}", test_case.description); + let early_gap = coverage_batch(HOUR_2, HOUR_3, "gap", Value::Null); + let late_gap = coverage_batch(HOUR_5, HOUR_6, "gap", Value::Null); + assert_eq!( + orders, + json!({ + "model": "orders", + "target_name": "orders_\u{1f9fe}", + "physical_generation_id": "orders-generation-1", + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "earliest": HOUR_0, + "contiguous_frontier": HOUR_2, + "known_gaps": [early_gap, late_gap], + "batches": [ + coverage_batch(HOUR_0, HOUR_1, "replay_required", json!("run-1")), + coverage_batch(HOUR_1, HOUR_2, "replay_required", json!("run-2")), + early_gap, + coverage_batch(HOUR_3, HOUR_4, "complete", json!("run-5")), + coverage_batch(HOUR_4, HOUR_5, "complete", Value::Null), + late_gap, + coverage_batch(HOUR_6, HOUR_7, "complete", Value::Null), + ], + "batches_truncated": false, + }), + "bounds sort by instant; the newest completion and requirement win; other generations, scopes and models never appear" + ); + + let (_, newest) = get( + &client, + (&api.url, &viewer), + &format!("{}?limit=2", read("orders")), + ) + .await; + assert_eq!( + (&newest["batches"], &newest["batches_truncated"]), + ( + &json!([ + late_gap, + coverage_batch(HOUR_6, HOUR_7, "complete", Value::Null) + ]), + &json!(true) + ) + ); + assert_eq!( + newest["known_gaps"], + json!([early_gap, late_gap]), + "gaps are never truncated" + ); + + let (_, events) = get(&client, (&api.url, &viewer), &read("events")).await; + assert_eq!( + ( + &events["earliest"], + &events["contiguous_frontier"], + &events["batches"] + ), + ( + &json!("2"), + &json!("11"), + &json!([ + coverage_batch("2", "9", "complete", json!("run-a")), + coverage_batch("9", "10", "complete", json!("run-b")), + coverage_batch("10", "11", "complete", json!("run-c")), + ]) + ), + "integer bounds sort by value, not text" + ); + + let (_, payments) = get(&client, (&api.url, &viewer), &read("payments")).await; + assert_eq!( + payments["batches"], + json!([ + coverage_batch( + "2026-01-01T00:00:00+00:00", + "2026-01-01T01:00:00+00:00", + "complete", + json!("run-p3") + ), + coverage_batch( + "2026-01-01T01:00:00+00:00", + "2026-01-01T02:00:00+00:00", + "complete", + json!("run-p2") + ), + ]), + "a completion at a third version ends the requirement, as the executor decides" + ); + + let (status, untracked) = get(&client, (&api.url, &viewer), &read("inventory")).await; + assert_eq!((status, untracked), (StatusCode::OK, Value::Null)); + assert_eq!( + get( + &client, + (&api.url, &viewer), + &read("orders").replace("development", "preview") + ) + .await + .0, + StatusCode::FORBIDDEN + ); + stop_api(api).await; + close_test_store(fixture_store).await; + }); + } +} diff --git a/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs b/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs index b7679b1b7d..abe809c055 100644 --- a/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs @@ -212,3 +212,8 @@ pub(crate) fn ids(history: &Value, kind: &str) -> Vec { .map(|record| record["event_id"].as_str().expect("event").to_owned()) .collect() } + +/// One microbatch coverage batch as the API returns it. +pub(crate) fn coverage_batch(start: &str, end: &str, state: &str, run_id: Value) -> Value { + json!({"start": start, "end": end, "state": state, "run_id": run_id}) +} diff --git a/fensu.toml b/fensu.toml index b36a40292e..54ddd88f0a 100644 --- a/fensu.toml +++ b/fensu.toml @@ -1371,7 +1371,7 @@ paths = ["crates/sqlbuild-orchestration/src/api/_helpers/resources/administrator reason = "Each function is one utoipa-annotated axum route: OpenAPI needs one annotated handler per operation, and each body only checks the caller's scope and calls one store operation. Scope checks and store work live in their shared owners." [[dupes.allowlist]] -paths = ["crates/sqlbuild-orchestration/src/authorization/main/read_security_audit.rs", "crates/sqlbuild-orchestration/src/credentials/_helpers/active_session.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_triggers.rs", "crates/sqlbuild-orchestration/src/platform/main/read_location_work.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs", "crates/sqlbuild-orchestration/src/runs/main/list_runs.rs", "crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs", "crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_ticks.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/attempts.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/facts.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_current_record.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_documents.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_nodes.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs"] +paths = ["crates/sqlbuild-orchestration/src/authorization/main/read_security_audit.rs", "crates/sqlbuild-orchestration/src/credentials/_helpers/active_session.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_triggers.rs", "crates/sqlbuild-orchestration/src/platform/main/read_location_work.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs", "crates/sqlbuild-orchestration/src/runs/main/list_runs.rs", "crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs", "crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_ticks.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/attempts.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/facts.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_current_record.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_documents.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_nodes.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs"] reason = "Each is one control-store read with its own SQL and record type. Transactions, person authorization and decoding go through the shared begin_authorized, query_records, read_records and query_bodies helpers; what remains alike is the call shape." [[dupes.allowlist]]