---
title: "Release CalCOFI Database"
calcofi:
target_name: release_database
workflow_type: release
dependency:
- auto
# a small file ONLY this notebook writes — never the data/releases
# directory, which test_release.qmd also writes into (see the `cleanup`
# chunk); that shared ownership made this target permanently outdated.
output: data/releases/_release_stamp.json
# cross-dataset foreign keys (relationships spanning ingests) are authored in
# metadata/relationships_cross.csv; intra-dataset FKs live in each ingest's
# relationships.json. both are merged below into the release relationships.json.
# ERD color overrides for common/cross-cutting tables (neutral). dataset
# colors themselves come from each ingest's calcofi.erd.color.
erd_overrides:
dataset: "#e8e8e8"
measurement_type: "#e8e8e8"
cruise: "#e8e8e8"
sample: "#e8e8e8"
obs: "#e8e8e8"
obs_attribute: "#e8e8e8"
sample_measurement: "#e8e8e8"
obs_ctd_full: "#e8e8e8"
obs_mets_full: "#e8e8e8"
spatial: "#cfe8ea"
spatial_attribute: "#cfe8ea"
# browser-shaped objects (Explorer plan D4; chunk browser_objects)
sample_spatial: "#cfe8ea"
sample_root: "#e8e8e8"
obs_bio: "#e8e8e8"
obs_env: "#e8e8e8"
# the one baseline every anomaly subtracts (chunk browser_objects; calcofi4db >= 3.26.0)
climatology: "#cfe8ea"
execute:
echo: true
message: true
warning: true
editor_options:
chunk_output_type: console
format:
html:
code-fold: true
editor:
markdown:
wrap: 72
---
## Overview {.unnumbered}
**Goal**: Create a frozen (immutable) release of the CalCOFI integrated
database by assembling all ingest parquet outputs. This is the "caboose"
notebook that always runs last, after all ingest notebooks complete.
**Upstream notebooks** are auto-discovered from `calcofi:` YAML
frontmatter in each `.qmd`. All workflows with `workflow_type: ingest`
or `spatial` feed into this release notebook via `dependency: [auto]`
in `_targets.R`.
```{r}
#| label: gen_fig_workflow
#| results: asis
#| echo: false
#| message: false
#| warning: false
# echo must stay false: this chunk emits a ```{mermaid} cell, and echoing R
# source that contains literal ``` fences would corrupt the document structure.
librarian::shelf(targets, quiet = TRUE)
# auto-discover the pipeline graph straight from _targets.R, so newly added
# workflows show up without editing this diagram. callr_function = NULL keeps it
# in-process (safe when this notebook is rendered via tar_make); outdated = FALSE
# skips the up-to-date check (no node status needed here).
invisible(suppressMessages(capture.output(
net <- tar_network(targets_only = TRUE, outdated = FALSE, callr_function = NULL))))
nodes <- net$vertices$name
edges <- net$edges
# group each node by name so the diagram colors input / ingest / release / test
node_grp <- ifelse(nodes == "release_database", "rel",
ifelse(nodes == "test_release", "test",
ifelse(nodes == "corrections_csv", "input", "ingest")))
# emit Mermaid source; rendered client-side by mermaid.js (the project-level
# `mermaid-format: png` is disabled — it routed this through headless Chrome,
# which hung intermittently; see _quarto.yml). release_database (this notebook)
# is highlighted.
mmd <- c(
"flowchart LR",
vapply(nodes, function(n) sprintf(' %s["%s"]', n, n), ""),
apply(edges, 1, function(r) sprintf(" %s --> %s", r[["from"]], r[["to"]])),
" classDef input fill:#eeeeee,stroke:#999999,color:#333333;",
" classDef ingest fill:#e3f2fd,stroke:#1565c0,color:#0d3c61;",
" classDef rel fill:#ef6c00,stroke:#b35100,color:#ffffff,font-weight:bold;",
" classDef test fill:#e8f4e8,stroke:#2e7d32,color:#1b5e20;")
for (grp in c("input", "ingest", "rel", "test")) {
members <- nodes[node_grp == grp]
if (length(members))
mmd <- c(mmd, sprintf(" class %s %s;", paste(members, collapse = ","), grp))
}
cap <- paste(
"Pipeline dependency graph, auto-discovered from `_targets.R`: every workflow",
"in this folder is a node and edges are dependencies. `release_database` (this",
"notebook, orange) is the caboose — it runs last, after all ingests, to assemble",
"the frozen release. Click to zoom.")
cat("```{mermaid}\n")
cat("%%| label: fig-workflow\n")
cat('%%| fig-cap: "', cap, '"\n', sep = "")
cat(mmd, sep = "\n")
cat("\n```\n")
```
## Setup
```{r}
#| label: setup
devtools::load_all(here::here("../calcofi4db"))
devtools::load_all(here::here("../calcofi4r"))
# cleanup_gcs_obsolete(dry_run = F)
librarian::shelf(
CalCOFI / calcofi4db,
CalCOFI / calcofi4r,
DBI,
dplyr,
DT,
fs,
glue,
here,
jsonlite,
purrr,
tibble,
quiet = T
)
options(DT.options = list(scrollX = TRUE))
# release version
release_version <- format(Sys.Date(), "v%Y.%m.%d")
# where the release goes and how its objects are laid out. Both are env-driven so
# a STAGING run (CALCOFI_RELEASE_PREFIX=ducklake-staging/releases) exercises the
# whole freeze/upload path against a scratch prefix before the real one changes;
# see .claude/plans/2026-08-25 … content-addressed release tables.md, tests T1-T7.
release_prefix <- Sys.getenv("CALCOFI_RELEASE_PREFIX", "ducklake/releases")
tables_prefix <- Sys.getenv("CALCOFI_TABLES_PREFIX", "ducklake/tables")
# canonical (content-addressed tables/{table}/{hash}/ + a compat copy) is the design and
# what v2026.08.25 shipped; v2026.09.04 came out compat because the default was, so its
# objects live only under releases/v2026.09.04/parquet/ and the next canonical release
# re-uploads them once. Set CALCOFI_RELEASE_LAYOUT=compat only to reproduce that.
release_layout <- match.arg(Sys.getenv("CALCOFI_RELEASE_LAYOUT", "canonical"),
c("compat", "canonical"))
staging <- release_prefix != "ducklake/releases"
# a STAGING run must not write the SHARED content store either: canonical objects are
# keyed by row signature, a row-identical re-export is not byte-identical for every
# table, and an upload into ducklake/tables/ silently replaced the promoted release's
# bytes on 2026-09-04 (173 objects; the real run then server-side copied those bytes
# while its catalog described the local files). Stage the store under its own prefix.
if (staging && tables_prefix == calcofi4db::CC_TABLES_PREFIX)
stop(glue("a staging run (CALCOFI_RELEASE_PREFIX={release_prefix}) must set ",
"CALCOFI_TABLES_PREFIX to a staging prefix (e.g. ducklake-staging/tables); ",
"it must never write the shared {tables_prefix}/ store"))
# a STAGING run must not touch the real release's local record: its sidecars and
# parquet land under *-staging roots keyed by the same date-derived version
sidecar_root <- if (staging) "data/releases-staging" else "data/releases"
stage_root <- if (staging) "releases-staging" else "releases"
if (staging)
message(glue("STAGING release: prefix {release_prefix}, layout {release_layout}, sidecars {sidecar_root}"))
message(glue("Release version: {release_version}"))
# --- refuse to re-cut the version consumers are currently reading -------------
#
# The version is the DATE, so two runs on one day reuse the tag — and the second
# overwrites `gs://…/releases/{version}/` in place. That is how v2026.08.10 was
# republished on 2026-08-11 with data that then FAILED test_release: promotion
# was correctly withheld, but promotion was never needed, because `latest.txt`
# already pointed at the path being overwritten. Consumers reading `latest` got
# unverified data without a single byte of `latest.txt` changing.
#
# The gate everyone relies on ("a failing release is not promoted") silently does
# not hold when the version does not change. So: if this run would overwrite the
# currently-promoted release, stop. Re-cutting a version nobody is reading is
# fine and stays unguarded.
#
# Override deliberately with CALCOFI_ALLOW_REPUBLISH=TRUE when the intent really
# is to replace a promoted release in place (and accept that consumers see the
# new bytes before any test has passed).
# Read the pointer through the authenticated API, NOT
# https://storage.googleapis.com/.../latest.txt — that URL is CDN-cached, and
# this guard consumed it for months. On 2026-08-14 the cache made it wrong in
# both directions within an hour: it false-fired on a re-cut after a rollback
# (harmless), and — the direction that matters — for an hour after any promotion
# the cache still shows the PREVIOUS version, so this comparison concludes
# `latest.txt` points elsewhere and lets a run overwrite the release consumers
# are actively reading. A guard that fails open for an hour after every
# promotion is worse than no guard, because it reads as protection.
promoted <- calcofi4db::read_promoted_release(bucket = "calcofi-db", prefix = release_prefix)
if (!is.na(promoted) && identical(promoted, release_version) &&
!isTRUE(as.logical(Sys.getenv("CALCOFI_ALLOW_REPUBLISH", "FALSE"))))
stop(glue(
"release {release_version} is the version `latest.txt` currently points at, ",
"so cutting it again would overwrite what consumers are reading — before any ",
"test has run against the new bytes.\n",
" Wait for the date to roll over, or set CALCOFI_ALLOW_REPUBLISH=TRUE if ",
"replacing the promoted release in place is genuinely what you want."))
```
## Assemble from Ingest Outputs
Create VIEWs on local parquet files from each ingest (zero-copy).
For tables appearing in multiple ingests, use the canonical (first) source.
```{r}
#| label: assemble_working
con_wdl <- get_duckdb_con(":memory:")
load_duckdb_extension(con_wdl, "spatial")
# auto-discover table registry from all ingest manifests. An ingest that declares
# `in_release: false` in its calcofi: YAML block is skipped everywhere below: it
# still runs in the pipeline and writes its own data/parquet/{dataset}/ outputs,
# but nothing of it reaches the frozen release. That is how a dataset under
# review (currently cdfw_dungeness-crab) is staged without leaking into a release.
ds_excluded <- release_excluded_datasets(here())
if (length(ds_excluded))
message(glue("Held out of this release (in_release: false): ",
"{paste(ds_excluded, collapse = ', ')}"))
# keep only the data/parquet/* dirs that belong in the release — used by the
# relationships.json / metadata.json / manifest.json globs further down
in_release_dirs <- function(paths)
paths[!basename(dirname(paths)) %in% ds_excluded]
registry <- build_release_table_registry(here())
# The consolidated core is now emitted per-dataset: every ingest writes its own
# `sample`/`obs`/… shard. The registry marks the FIRST ingest supplying a table
# name as canonical, which is correct for a genuinely shared reference (`grid`,
# `cruise`) but would silently keep ONE dataset's `obs` and drop the other 14.
# So the core is excluded here and assembled by union below (assemble_core()).
core_shard_tables <- c(
"sample", "obs", "obs_attribute", "sample_measurement", "obs_ctd_full", "obs_mets_full",
"taxon", "dataset_taxon", "taxon_group")
# use only canonical, non-supplemental tables
reg_canon <- registry |>
filter(canonical, !supplemental, !table %in% core_shard_tables)
message(glue(
"{nrow(reg_canon)} canonical tables from ",
"{length(unique(reg_canon$ingest))} ingests"))
# --- authoritative dataset metadata + ERD coloring from ingest YAML ----
# table -> provider_dataset(s) owned, from each ingest's calcofi.tables_owned
ingest_yaml <- read_ingest_yaml(here(), in_release_only = TRUE)
table_dataset <- list()
add_owner <- function(tbl, pd) {
if (is.null(tbl)) return(invisible())
table_dataset[[tbl]] <<- unique(c(table_dataset[[tbl]], pd))
}
for (key in names(ingest_yaml)) {
cc <- ingest_yaml[[key]]
for (e in cc$tables_owned %||% list()) add_owner(e$table, key)
for (ad in cc$additional_datasets %||% list()) {
pd2 <- paste0(ad$provider, "_", ad$dataset)
for (e in ad$tables_owned %||% list()) add_owner(e$table, pd2)
}
}
# one color per dataset (from calcofi.erd.color)
dataset_colors <- lapply(ingest_yaml, function(cc) cc$erd$color)
# release-level config: neutral ERD overrides for common tables
rel_cfg <- read_calcofi_meta(here("release_database.qmd"))
release_overrides <- rel_cfg$erd_overrides
# cross-dataset foreign keys (relationships spanning ingests) are authored in a
# reviewable CSV; intra-dataset FKs live in each ingest's relationships.json.
cross_fks_df <- readr::read_csv(
here("metadata/relationships_cross.csv"), show_col_types = FALSE)
cross_fks <- lapply(seq_len(nrow(cross_fks_df)), function(i)
as.list(cross_fks_df[i, c("table", "column", "ref_table", "ref_column")]))
# stroke-based color map consumed by every cc_erd() call below
color_map <- cc_erd_color_map(
table_dataset = table_dataset,
dataset_colors = dataset_colors,
overrides = release_overrides,
neutral = "#dcdcdc")
# create VIEWs on local parquet for each canonical table
# _new delta tables handled separately for merging
#
# A table carrying geometry MUST be listed here, and the cost of omitting it is
# silent: `load_prior_tables()` only converts the parquet's WKB BLOB back to
# GEOMETRY for the tables named, so an omitted one arrives as a BLOB, the CRS
# normalization below (which selects on `data_type LIKE 'GEOMETRY%'`) never sees
# it, it never joins `crs_local_tables`, and it is therefore GCS-copied straight
# from the ingest bucket with whatever tag the ingest happened to mint. Nothing
# fails; a consumer's ST_Intersects against `sample.geom` does, later.
# `region` gained a POLYGON when the phytoplankton pooling regions stopped being
# provisional centroids (workflows#76).
all_geom_tables <- c("grid", "site", "segment", "casts", "ctd_cast", "spatial",
"region")
main_tables <- reg_canon |> filter(!grepl("_new$", table))
new_tables <- registry |> filter(grepl("_new$", table))
load_stats <- purrr::map_dfr(
split(main_tables, seq_len(nrow(main_tables))),
function(row) {
load_prior_tables(
con = con_wdl,
parquet_dir = row$parquet_dir,
tables = row$table,
geom_tables = all_geom_tables,
as_view = TRUE
)
})
# merge {table}_new additions into their base tables
# driven by calcofi.modifies in YAML frontmatter
if (nrow(new_tables) > 0) {
# group _new tables by their base table
base_names <- unique(sub("_new$", "", new_tables$table))
for (base_tbl in base_names) {
delta_rows <- new_tables |> filter(table == paste0(base_tbl, "_new"))
# replace VIEW with TABLE for this base table (so we can INSERT)
base_src <- main_tables |> filter(table == base_tbl)
if (nrow(base_src) > 0) {
dbExecute(con_wdl, glue("DROP VIEW IF EXISTS {base_tbl}"))
load_prior_tables(
con = con_wdl, parquet_dir = base_src$parquet_dir[1],
tables = base_tbl, geom_tables = all_geom_tables)
# PK column for dedup — the DECLARED PK from core_relationships(), not
# "first column by ordinal position". For `cruise` that ordinal-first
# column is cruise_uuid, which is NULL on every delta row (no ingest
# mints one), so `WHERE cruise_uuid NOT IN (...)` evaluates NULL and no
# row would ever insert — silently, since the INSERT still "succeeds"
# with 0 rows added. No behaviour change today (every current _new
# table's first column already IS its PK, e.g. ship_new -> ship_key);
# this is the prerequisite for a future cruise_new delta (WS-B D4).
declared_pk <- core_relationships(base_tbl)$primary_keys[[base_tbl]]
if (is.null(declared_pk)) {
pk_col <- dbGetQuery(con_wdl, glue(
"SELECT column_name FROM information_schema.columns
WHERE table_name = '{base_tbl}'
ORDER BY ordinal_position LIMIT 1"))$column_name
warning(glue(
"release_database.qmd: no declared PK for '{base_tbl}' in ",
"core_relationships() — falling back to its first column ",
"('{pk_col}') by ordinal position for _new dedup."), call. = FALSE)
} else {
pk_col <- declared_pk
}
for (j in seq_len(nrow(delta_rows))) {
dr <- delta_rows[j, ]
pq_path <- file.path(dr$parquet_dir, paste0(base_tbl, "_new.parquet"))
if (file.exists(pq_path)) {
dbExecute(con_wdl, glue(
"INSERT INTO {base_tbl}
SELECT * FROM read_parquet('{pq_path}')
WHERE {pk_col} NOT IN (SELECT {pk_col} FROM {base_tbl})"))
n_new <- dbGetQuery(con_wdl, glue(
"SELECT COUNT(*) AS n FROM read_parquet('{pq_path}')"))$n
message(glue("Merged {n_new} {base_tbl} addition(s) from {dr$ingest}"))
}
}
}
}
}
load_stats |>
datatable(caption = "Assembled tables (VIEWs on local parquet)")
```
## Dataset Reference
The `dataset` reference table, keyed by `dataset_key = provider_dataset`.
The Phase-1 `v_obs_env` / `v_obs_bio` / `v_obs` VIEWs that used to be built here
are gone. They projected each dataset's per-dataset measurement tables into a
common shape to prove the consolidation target non-destructively, *without*
re-running the ingests. That job is done: every ingest now emits its slice of
`obs` directly, so the views' source tables (`bottle_measurement`, `casts`,
`ctd_measurement`, …) no longer exist and the real `obs` table assembled below
supersedes them. They were release-local — nothing outside this notebook read
them.
```{r}
#| label: dataset_ref
# dataset reference: dataset_key = provider_dataset, built from the ingest YAML
# rather than metadata/dataset.csv. The YAML is authoritative (it deprecates the
# CSV) and, more to the point, it cannot go stale: it is derived from the same
# `calcofi:` blocks that define the pipeline, so every ingest is present by
# construction. The CSV had drifted — it was missing calcofi_mets,
# cce-lter_picoplankton-bacteria and sio_mesopelagic-fish, which orphaned
# 533,571 obs rows against the obs.dataset_key foreign key.
d_dataset <- ingest_yaml_to_dataset_df(ingest_yaml) |>
mutate(dataset_key = paste0(provider, "_", dataset), .before = 1)
dbExecute(con_wdl, "DROP VIEW IF EXISTS dataset")
dbWriteTable(con_wdl, "dataset", as.data.frame(d_dataset), overwrite = TRUE)
dbGetQuery(con_wdl, "SELECT dataset_key, dataset_name FROM dataset ORDER BY 1") |>
datatable(caption = "dataset reference")
```
## Consolidated Core Tables
The **core** tables every consumer reads, replacing the ~40 per-dataset triples:
keyed by a namespaced `sample_key` (`dataset_key:sample_type:id`) and stamped
with a computed H3 `hex_id`. See `design_env-bio-consolidation.md`.
This step **concatenates, it does not derive.** Each ingest projects itself into
the core in its own notebook ("Emit Core Tables") — the single authoritative
projection, owned by the notebook that owns the dataset — and writes its slice as
parquet. `assemble_core()` unions those
shards, renumbers the surrogate ids globally (every ingest numbers from 1 within
its own shard) and merges the `taxon` slices by source priority. Deriving the
core here as well is what let the two projections drift apart, so that
duplication is gone.
```{r}
#| label: core_tables
# measurement_type: authoritative from the metadata CSV (adds abundance, count,
# body_length, and the event-level effort types), replacing any per-ingest VIEW
# so the FK parity check below sees the current vocabulary.
dbExecute(con_wdl, "DROP VIEW IF EXISTS measurement_type")
# Read the registry through calcofi4db::read_measurement_type() rather than
# DuckDB's read_csv_auto. This used to be a direct read_csv_auto, and that is how
# the release shipped literal "NA" strings: an ingest wrote the registry with
# readr's default `na = "NA"`, which is invisible to read_csv() but NOT to
# read_csv_auto, whose default nullstr is the empty string only. 161 rows of
# `_qual_column` and 192 of `_prec_column` were affected, plus `is_canonical`.
# The helper reads strictly (na = "") and ERRORS on sentinel strings, so a
# corrupted registry now fails the release instead of being published by it.
d_meas_type_reg <- read_measurement_type(here("metadata/measurement_type.csv"))
dbWriteTable(con_wdl, "_measurement_type_reg", as.data.frame(d_meas_type_reg),
overwrite = TRUE)
# derive provider/dataset from _source_datasets (first source) so the schema site
# + query app ("browse measurement types") keep their provider/dataset columns.
dbExecute(con_wdl,
"CREATE OR REPLACE TABLE measurement_type AS
SELECT *,
split_part(split_part(_source_datasets, ';', 1), '_', 1) AS provider,
regexp_replace(split_part(_source_datasets, ';', 1), '^[^_]*_', '') AS dataset
FROM _measurement_type_reg")
dbExecute(con_wdl, "DROP TABLE _measurement_type_reg")
# r_* types are the bottle's pre-QC, interpolated-to-standard-depth series
# (Rasmus Swalethorp, 2026-09-01, WS-G: "we should not use already interpolated
# data points from the bottle database" for any further interpolation). A
# `variable` crosswalk entry is exactly that reuse — it tells a consumer this
# type is comparable to, and poolable with, another dataset's canonical series —
# so no r_* type may ever carry one. is_canonical is set FALSE on all six at the
# registry (metadata/measurement_type.csv); this is the second, structural gate.
d_r_star_variable <- dbGetQuery(con_wdl,
"SELECT measurement_type, variable FROM measurement_type
WHERE measurement_type LIKE 'r\\_%' ESCAPE '\\' AND variable IS NOT NULL")
stopifnot(
"an r_* (pre-QC, interpolated) measurement_type carries a `variable` crosswalk entry - it must not, see WS-G 2026-09-03" =
nrow(d_r_star_variable) == 0)
# --- assemble the core from the per-dataset shards --------------------------
# Each ingest emits its own slice from its own notebook, which is the single
# authoritative projection (calcofi4db holds only the generic shapes). This step only concatenates: it UNIONs the
# shards, renumbers the surrogate ids globally (each ingest numbers from 1 within
# its own shard), merges the `taxon` slices with source priority, and asserts
# `sample_key` is globally unique. Nothing is re-derived here — that duplication
# is exactly what let the release and the ingests drift apart.
# Supplemental full-resolution tables are DISCOVERED from the ingests' YAML, not
# hardcoded — obs_ctd_full was the only one until calcofi_mets added obs_mets_full,
# and a hardcoded name silently drops a new one from the release while the ingest
# keeps writing it. BUILD_OBS_CTD_FULL=FALSE still skips them all for a fast run.
build_supplemental <- as.logical(Sys.getenv("BUILD_OBS_CTD_FULL", "TRUE"))
supp_tbls <- supplemental_core_tables(here(), build_supplemental)
if (length(supp_tbls))
message(glue("supplemental tables: {paste(supp_tbls, collapse = ', ')}"))
core_n <- assemble_core(con_wdl, root = here(), supplemental = supp_tbls)
# Vernacular names, applied ONCE to the merged `taxon` rather than in each of the
# 10 taxa-emitting ingests. `common_name` only ever came from a dataset's own
# vocabulary, so every taxon resolved through measurement_taxon.csv /
# taxon_override.csv arrived with none — 57% of them at v2026.08.14, including
# worms:440388 Metacarcinus magister, whose missing "Dungeness crab" in
# db-viz-hex surfaced this.
#
# Central for the same reason `dataset` and the observed coverage columns are:
# the shards are MERGED here, not rebuilt, so one application cannot drift across
# ten of them. A dataset's own common name always wins — it is what the provider
# publishes. A taxon whose WoRMS vernaculars are ambiguous stays NULL until a
# human picks one in the registry; see metadata/taxon_common.csv.
#
# A group label is never a common name (calcofi4db >= 3.33.0): the registry's
# `dataset_taxon` rule values ("diatom, centric", "other", …) and the label of
# any dataset-local key ("undefined (code not in source definitions; Q05)",
# zooscan "nauplii") are refused at rank 4 — 24 taxa carried one as their
# common_name at v2026.08.25. The group's own name lives in taxon_group.
tg_rules <- read_taxon_group_rules(here("metadata/taxon_group.csv"))
n_common <- apply_taxon_common(con_wdl, here("metadata/taxon_common.csv"),
group_rules = tg_rules)
tibble(
table = names(core_n),
rows = unlist(core_n)) |>
datatable(caption = "Core tables assembled from per-dataset ingest shards")
```
```{r}
#| label: core_by_dataset
dbGetQuery(con_wdl,
"SELECT dataset_key, count(*) n_obs, count(DISTINCT sample_key) n_samples,
count(DISTINCT hex_id) n_hex
FROM obs GROUP BY 1 ORDER BY 1") |>
datatable(caption = "obs: consolidated observations by dataset")
```
### Observed Coverage
Each dataset's temporal and spatial extent, **measured from the assembled core
rather than asserted**. These overwrite the `dataset` table's
`coverage_temporal` / `coverage_spatial`, which used to carry a hand-written
string from each ingest's `calcofi.dataset_meta` YAML.
Those strings could not help going stale — authored once, with the data growing
underneath them. At `v2026.08.06` seven of fifteen were wrong: `cce-lter_zoodb`
claimed coverage through 2021-05 when its data ends 2015-04, `calcofi_phyllosoma`
stopped a year short of its own rows, and three said "present" while in fact
stalling in 2019, 2022 and 2023. The YAML keys are now gone; the only ones left
are where the data genuinely cannot answer (see `coverage_fallback` below), and
each carries a comment saying so.
```{r}
#| label: dataset_coverage
# measured, not asserted. observed_coverage() filters coordinates with
# isfinite() rather than IS NOT NULL: NaN survives a nullity test and min()/max()
# propagate it, so one poisoned row would blow a dataset's whole bbox out to NaN
# with every check still passing.
d_cov <- observed_coverage(con_wdl)
# fall back to a declared static value ONLY where the data cannot answer.
# calcofi_phytoplankton is region-pooled: it carries real coordinates but no
# datetime at all, so it measures spatially and not temporally. Held-out
# datasets (in_release: false) never reach the core, so they never appear here.
d_dataset_cov <- d_dataset |>
left_join(d_cov, by = "dataset_key") |>
mutate(
coverage_temporal = coalesce(coverage_temporal_observed, coverage_temporal),
coverage_spatial = coalesce(coverage_spatial_observed, coverage_spatial)) |>
select(all_of(names(d_dataset)))
# source_accessed: measured, never asserted (calcofi4db >= 3.30.0, R/citation.R).
# When an ingest stamped its own reads (stamp_source_access() -> metadata.json
# sources[]) that stamp wins (method download / file_mtime); otherwise the last
# commit of its manifest.json sidecar stands in (method sidecar_commit) — the
# ingest rewrites that file every run, so its history is when the source was
# last read, and no ingest has to re-run to say so. Both land on `dataset`.
d_src <- resolve_source_accessed(here("data/parquet", d_dataset$dataset_key))
d_dataset_cov <- d_dataset_cov |>
left_join(select(d_src, dataset_key, source_accessed, source_accessed_method),
by = "dataset_key")
src_methods <- table(d_src$source_accessed_method)
cat(sprintf("source_accessed: %d of %d datasets (%s)\n",
sum(!is.na(d_src$source_accessed)), nrow(d_src),
paste(names(src_methods), src_methods, sep = " = ", collapse = ", ")))
# the attribution contract, enforced here as well as by the index build:
# check_dataset_citation() — a citation with a year and a locator, a license from
# metadata/license.csv (`custom` with a URL), a bare DOI that resolves, and the
# source's own authority (EDI / NCEI / ERDDAP / DataCite, cached in
# metadata/{provider}/{dataset}/citation_authority.json) compared for drift.
# Error findings stop the release unless an open/proposed questions.csv row on
# related_table = dataset covers the field; drift only warns. The network half
# follows CALCOFI_SKIP_LINK_CHECK like the link check does.
d_cit <- check_dataset_citation(
ingest_yaml, network = !nzchar(Sys.getenv("CALCOFI_SKIP_LINK_CHECK")),
cache_dir = here("metadata"))
assert_dataset_citation(d_cit)
d_cit |>
filter(finding != "ok") |>
select(dataset_key, finding, level, exempt, question, authority, checked, detail) |>
datatable(caption = "citation check: findings (error rows are exempt only while a provider question is open/proposed)")
# `dataset` is written as a TABLE at [dataset_table] above, so the drop has to
# match that type. DuckDB's `DROP VIEW IF EXISTS` does NOT no-op on a type
# mismatch — it raises "Existing object dataset is of type Table, trying to drop
# type View" — so the unconditional DROP VIEW here failed every release run.
# Ask the catalog rather than assume, since a compat VIEW of the same name is a
# legitimate state for this connection to be in.
ds_type <- dbGetQuery(con_wdl, "
SELECT table_type FROM information_schema.tables
WHERE table_name = 'dataset'")$table_type
if (length(ds_type))
dbExecute(con_wdl, if (identical(ds_type[1], "VIEW"))
"DROP VIEW IF EXISTS dataset" else "DROP TABLE IF EXISTS dataset")
dbWriteTable(con_wdl, "dataset", as.data.frame(d_dataset_cov), overwrite = TRUE)
# report which half of which dataset fell back, so a silent gap cannot hide as a
# confidently-rendered string on the schema site
coverage_fallback <- d_dataset |>
left_join(d_cov, by = "dataset_key") |>
filter(is.na(coverage_temporal_observed) | is.na(coverage_spatial_observed)) |>
transmute(dataset_key,
temporal = if_else(is.na(coverage_temporal_observed),
paste("asserted:", coverage_temporal), "measured"),
spatial = if_else(is.na(coverage_spatial_observed),
paste("asserted:", coverage_spatial), "measured"))
cat(glue(
"coverage measured for {sum(!is.na(d_cov$coverage_temporal_observed))} datasets ",
"temporally, {sum(!is.na(d_cov$coverage_spatial_observed))} spatially; ",
"{nrow(coverage_fallback)} fell back to an asserted value\n"))
d_cov |>
select(dataset_key, coverage_temporal_observed, coverage_spatial_observed) |>
datatable(caption = "observed coverage, measured from obs + sample")
```
### Core Table Parity Checks
Hard assertions on the assembled core. The old checks compared it against the
per-dataset tables (`net`, `casts`, `ctd_cast`, …), which the ingests no longer
publish — and which was only meaningful while the core was re-derived here.
Now that each ingest emits its own slice, the checks that matter are
**conservation** (no shard silently dropped by the union), **global key
uniqueness** after renumbering, and **referential integrity**. A break fails the
render. See `design_env-bio-consolidation.md` Verification.
```{r}
#| label: core_parity
q <- function(sql) dbGetQuery(con_wdl, sql)$n
# (A) shard conservation — the assembled core must contain exactly the rows the
# ingests emitted. This replaces the old per-dataset assertions (which compared
# against `net`/`casts`/`ctd_cast`, tables the ingests no longer publish) and is
# a stronger check: it catches a shard silently dropped by the union, which the
# canonical-first registry would otherwise do without complaint.
shard_total <- function(tbl) {
paths <- core_shard_paths(tbl, root = here())
if (!length(paths)) return(0)
sum(vapply(paths, function(p) {
rd <- if (grepl("\\*\\*", p))
glue("read_parquet('{p}', hive_partitioning = true, union_by_name = true)") else
glue("read_parquet('{p}', union_by_name = true)")
as.numeric(dbGetQuery(con_wdl, glue("SELECT COUNT(*) AS n FROM {rd}"))$n)
}, numeric(1)))
}
core_tbls <- intersect(
c("sample", "obs", "obs_attribute", "sample_measurement"), dbListTables(con_wdl))
conservation <- tibble(
table = core_tbls,
shards = vapply(core_tbls, shard_total, numeric(1)),
assembled = vapply(core_tbls, function(t)
as.numeric(q(glue("SELECT COUNT(*) AS n FROM {t}"))), numeric(1))) |>
mutate(ok = shards == assembled)
print(as.data.frame(conservation))
stopifnot("every ingest shard must survive the union" = all(conservation$ok))
# (B) surrogate ids must be globally unique after renumbering ----------------
id_dups <- c(
"obs.obs_id" = q("SELECT COUNT(*) n FROM (SELECT obs_id FROM obs GROUP BY 1 HAVING COUNT(*) > 1)"),
"obs_attribute.obs_attribute_id" = if ("obs_attribute" %in% dbListTables(con_wdl))
q("SELECT COUNT(*) n FROM (SELECT obs_attribute_id FROM obs_attribute GROUP BY 1 HAVING COUNT(*) > 1)") else 0,
"sample.sample_key" = q("SELECT COUNT(*) n FROM (SELECT sample_key FROM sample GROUP BY 1 HAVING COUNT(*) > 1)"))
if (any(id_dups > 0)) print(id_dups[id_dups > 0])
stopifnot("core surrogate keys must be globally unique" = all(id_dups == 0))
# (C) FK validity — every core row resolves against its reference, INCLUDING the
# unified taxon (obs/obs_attribute/dataset_taxon all key into taxon.taxon_key) ---
fk_bad <- c(
"obs.dataset_key" = q("SELECT COUNT(*) n FROM obs WHERE dataset_key NOT IN (SELECT dataset_key FROM dataset)"),
"obs.sample_key" = q("SELECT COUNT(*) n FROM obs WHERE sample_key NOT IN (SELECT sample_key FROM sample)"),
"obs.grid_key" = q("SELECT COUNT(*) n FROM obs WHERE grid_key IS NOT NULL AND grid_key NOT IN (SELECT grid_key FROM grid)"),
"obs.measurement_type" = q("SELECT COUNT(*) n FROM obs WHERE measurement_type NOT IN (SELECT measurement_type FROM measurement_type)"),
"obs.taxon_key" = q("SELECT COUNT(*) n FROM obs WHERE taxon_key IS NOT NULL AND taxon_key NOT IN (SELECT taxon_key FROM taxon)"),
"sample.parent_sample_key" = q("SELECT COUNT(*) n FROM sample WHERE parent_sample_key IS NOT NULL AND parent_sample_key NOT IN (SELECT sample_key FROM sample)"),
"obs_attribute.sample_key" = if ("obs_attribute" %in% dbListTables(con_wdl))
q("SELECT COUNT(*) n FROM obs_attribute WHERE sample_key NOT IN (SELECT sample_key FROM sample)") else 0,
"obs_attribute.taxon_key" = if ("obs_attribute" %in% dbListTables(con_wdl))
q("SELECT COUNT(*) n FROM obs_attribute WHERE taxon_key IS NOT NULL AND taxon_key NOT IN (SELECT taxon_key FROM taxon)") else 0,
"dataset_taxon.taxon_key" = q("SELECT COUNT(*) n FROM dataset_taxon WHERE taxon_key NOT IN (SELECT taxon_key FROM taxon)"),
"sample_measurement.sample_key" = q("SELECT COUNT(*) n FROM sample_measurement WHERE sample_key NOT IN (SELECT sample_key FROM sample)"),
# the measurement vocabulary must cover EVERY grain, not just obs: promoting
# bottom_depth_m into sample_measurement added a type that was not registered,
# and with only obs.measurement_type asserted it went unnoticed.
"sample_measurement.measurement_type" = q("SELECT COUNT(*) n FROM sample_measurement WHERE measurement_type NOT IN (SELECT measurement_type FROM measurement_type)"),
"obs_attribute.measurement_type" = if ("obs_attribute" %in% dbListTables(con_wdl))
q("SELECT COUNT(*) n FROM obs_attribute WHERE measurement_type NOT IN (SELECT measurement_type FROM measurement_type)") else 0)
if (any(fk_bad > 0)) print(fk_bad[fk_bad > 0])
stopifnot("core FK validity" = all(fk_bad == 0))
# (D) the DIC -> bottle dedup: DIC observations sharing a physical Niskin must
# point at the bottle's event, not mint a second one
n_dic_shared <- q("SELECT COUNT(*) n FROM obs
WHERE dataset_key = 'calcofi_dic'
AND sample_key LIKE 'calcofi_bottle:bottle:%'")
message(glue("DIC observations sharing a bottle event: {format(n_dic_shared, big.mark = ',')}"))
# (E) obs_attribute vs its headline — reported (sources are not always internally
# consistent, so this is a signal, not an assertion)
if ("obs_attribute" %in% dbListTables(con_wdl)) {
attr_check <- dbGetQuery(con_wdl, "
WITH f AS (SELECT sample_key, taxon_key, life_stage, SUM(count) s FROM obs_attribute
WHERE measurement_type='stage' GROUP BY 1,2,3),
o AS (SELECT sample_key, taxon_key, life_stage, SUM(measurement_value) a FROM obs
WHERE measurement_type='abundance' GROUP BY 1,2,3)
SELECT count(*) n_occ, count(*) FILTER (WHERE f.s > o.a) n_stage_gt_headline
FROM f JOIN o USING (sample_key, taxon_key, life_stage)")
message(glue("obs_attribute stage vs abundance: {attr_check$n_stage_gt_headline}/{attr_check$n_occ} occurrences exceed headline (source quirk)"))
}
message("Core parity checks passed.")
```
### Taxon Authority Coverage
A taxon that reaches the release without an authority id is invisible to any
consumer that filters or joins on one, and until v2026.08.05 nothing said so:
all 128 Farallon taxa and 64,956 observations were unreachable through
`db-viz-hex::get_sp()`'s `worms_id` join while every check here passed.
`check_taxon_ids()` **fails the release** on a dataset-local `taxon_key` that is
not declared below. The allowlist is deliberately one key at a time — these are
non-taxonomic operational classes the source records as data, not lookup
failures — so a *new* unresolved taxon can never hide among the known ones.
Two more things are checked here because this is the one place every dataset is
present at once. `check_taxon_registries()` **fails the release** on a
`taxon_override.csv` or `taxon_group.csv` row naming a `dataset_key` nothing
supplies — each ingest reads the whole registry while loading only its own
vocabulary, so a typo there is invisible until now, and a registry row that
matches nothing is how a missing id hides. And `report_taxon_overrides()` shows,
per override row, how many vocabulary rows it matched, keyed, and **skipped**:
since calcofi4db 3.33.0 an override never replaces an id the source supplied
unless it names the row by the dataset's own code, which is what stopped six
`taxa`-matched phytoplankton group rows from collapsing 287 species onto 6 class
keys. A skip is the rule working; it is reported so it is never silent.
```{r}
#| label: taxon_authority_coverage
# non-taxonomic classes: real categories in the source that no authority can key.
# Anything NOT on this list that resolves to a dataset-local key fails the render.
TAXON_LOCAL_ALLOW <- c(
# ZooScan operational classes (Q03 in cce-lter/zooscan/questions.csv)
"cce-lter_zooscan:13", # eggs
"cce-lter_zooscan:15", # multiples (several organisms in one vignette)
"cce-lter_zooscan:16", # nauplii (crustacean naupliar stage, not a taxon)
"cce-lter_zooscan:18", # others
# phytoplankton: two different things, kept apart on purpose (Q05 in
# calcofi/phytoplankton/questions.csv). Nine codes are absent from the source
# Definitions sheet altogether, so there is no name to resolve; 232 is present
# and named (*Danasphaera indica*) but has no WoRMS record, fuzzy included.
#
# This list was 14 and is now 10. The other four (40, 231, 337, 597) were
# never unnameable — they carry real names in the Definitions sheet and only
# fell through because their `taxa` is "other", which no functional-group
# override row matches. They now resolve to accepted genera via
# `species_code` rows in metadata/taxon_override.csv. An allowlist is for taxa
# no authority CAN key, not for ones our own join missed.
paste0("calcofi_phytoplankton:",
c("4", "59", "218", "229", "232", "300", "454", "532", "540", "596")))
tx_cover <- check_taxon_ids(con_wdl, allow = TAXON_LOCAL_ALLOW, halt = TRUE)
# the registries, validated where every dataset IS present (taxon plan D5): a
# row naming a dataset_key that no shard and no measurement_taxon row supplies
# fails the release. `tg_rules` was read in `core_tables` for apply_taxon_common().
mt_taxon_all <- readr::read_csv(
here("metadata/measurement_taxon.csv"),
col_types = readr::cols(worms_id = "i", itis_id = "i", bin_value = "d", .default = "c"))
tx_over_all <- readr::read_csv(here("metadata/taxon_override.csv"), show_col_types = FALSE)
check_taxon_registries(con_wdl, overrides = tx_over_all, group_rules = tg_rules,
measurement_taxon = mt_taxon_all, halt = TRUE)
# what each override row did: matched / applied / skipped (the source's own id
# kept). `source_json_known = FALSE` means the shard predates ds_source_json, so
# the skip count is unknowable here and the ingest's own message is the fact.
tx_over_rpt <- report_taxon_overrides(con_wdl, tx_over_all)
tx_cover |>
datatable(caption = paste(
"Taxon authority coverage by dataset.",
"`n_local_key` counts taxa no authority resolved (all allowlisted, or this",
"chunk would have failed); `n_no_worms` is reported, not gated — WoRMS",
"legitimately lacks a few taxa, and an itis:-keyed bird is correctly keyed",
"either way. `n_no_rank_order` should be 0: it was every ITIS-keyed taxon",
"until the rank vocabulary moved out of a single ingest's connection into",
"calcofi4db::taxa_rank_reference()."))
```
```{r}
#| label: taxon_override_report
tx_over_rpt |>
dplyr::group_by(dataset_key) |>
dplyr::summarise(
override_rows = dplyr::n(), n_matched = sum(n_matched),
n_applied = sum(n_applied), n_skipped = sum(n_skipped),
source_json_known = all(source_json_known), .groups = "drop") |>
datatable(caption = paste(
"taxon_override.csv per dataset: vocabulary rows each registry row matched,",
"keyed (`n_applied`) and skipped because the source supplied its own id",
"(`n_skipped`; NA where `source_json_known` is FALSE). Per-row detail,",
"with the skipped codes, is in `tx_over_rpt`."))
```
## Declared Measurement Bounds
The backstop for the per-dataset bounds check that each ingest runs (see
`check_measurement_bounds()` in every `ingest_*.qmd`). The ingest is where a
finding is *actionable* — the source is open, the provider can be asked, and a
question lands in `questions.csv`. This chunk exists so that a notebook which
skipped the check, or a `measurement_type.csv` bound edited after an ingest last
ran, cannot ship an impossible value to consumers.
The two halves are gated differently, on purpose:
* **`out_of_range` fails the release.** A bound was agreed and the data breaks it.
There is no reading of that which should reach a consumer.
* **`undeclared` is reported, not gated.** 73 of 98 (dataset, type) pairs and 67%
of `obs` rows had no bound at v2026.08.07, so gating on it would block every
release rather than fix anything. `BOUNDS_UNDECLARED_MAX` ratchets: it is the
count on the day this landed, and it may only ever go **down**. A new
undeclared type therefore fails the release even though the backlog does not.
```{r}
#| label: bounds_coverage
# ratchet, not a target. Lower it whenever bounds are declared; never raise it.
# Raising it to make a release pass is how the backlog became 73 in the first
# place — take the finding to the ingest and declare the bound there.
#
# 73 -> 30 for `obs` at v2026.08.08: every type whose live values already
# satisfied a defensible bound got one, reusing the vocabulary the registry had
# already agreed for the same quantity under its other name (btl_temperature
# -2..40, salinity_* 0..45, oxygen umol/kg 0..700, ...). What remains is where
# the defensible bound is VIOLATED by published data — i.e. the findings — each
# a `proposed` question in its dataset's questions.csv. Declaring those before
# the provider answers would delete real observations.
#
# The count covers `obs` AND the supplemental tables from v2026.08.08, so it
# jumped when they were first checked rather than because anything regressed.
# Set from the measured value on the day; it may only ever go DOWN.
BOUNDS_UNDECLARED_MAX <- 77L
d_bounds <- purrr::map_dfr(
dbGetQuery(con_wdl, "SELECT DISTINCT dataset_key FROM obs ORDER BY 1")$dataset_key,
\(dk) check_measurement_bounds(con_wdl, "obs", dataset_key = dk) |>
mutate(table = "obs", dataset_key = dk, .before = 1))
# The SUPPLEMENTAL tables are published and were not checked here until
# v2026.08.08. That gap was not theoretical: v2026.08.07's `obs_ctd_full` shipped
# 5,963 `ph` values below the declared floor (to -2.98) that the CTD ingest had
# already removed from its own output — the released bytes did not match the
# staged ones, and every check in this notebook looked only at `obs`, so nothing
# anywhere disagreed. Checking `obs` alone certifies a third of the release.
#
# Cost is not a reason to skip them: 216M rows of obs_ctd_full check in ~20s,
# because the work is a GROUP BY per type over a column DuckDB reads lazily.
d_bounds <- bind_rows(d_bounds, purrr::map_dfr(
intersect(supp_tbls, dbListTables(con_wdl)),
\(tb) check_measurement_bounds(con_wdl, tb) |>
mutate(table = tb, dataset_key = paste0("(supplemental) ", tb), .before = 1)))
n_oob <- sum(d_bounds$status == "out_of_range")
n_und <- sum(d_bounds$status == "undeclared")
cat(glue("bounds: {n_oob} out-of-range type(s), {n_und} undeclared ",
"(ratchet {BOUNDS_UNDECLARED_MAX}), ",
"{sum(d_bounds$status == 'ok')} ok, across ",
"{n_distinct(d_bounds$table)} table(s) and ",
"{format(sum(d_bounds$n_total), big.mark = ',')} values\n"))
d_bounds |>
filter(status != "ok") |>
select(table, dataset_key, measurement_type, status, n_total, n_bad, pct_bad,
v_min, v_max, valid_min, valid_max) |>
datatable(caption = paste(
"Measured values against metadata/measurement_type.csv, across `obs` AND the",
"supplemental full-resolution tables.",
"`out_of_range` fails this render; `undeclared` is the coverage backlog —",
"nothing was checked for those types, so their absence from the",
"out_of_range list means nothing."))
# `stopifnot` rather than a warning: a warning here is indistinguishable from the
# ~40 benign ones a full release render emits, which is how the CTD values shipped
stopifnot(
"values outside declared valid_min/valid_max — fix at the ingest, not here" =
n_oob == 0,
"a measurement type lost its bounds; declare it in the owning ingest" =
n_und <= BOUNDS_UNDECLARED_MAX)
if (n_und < BOUNDS_UNDECLARED_MAX)
cat(glue("\nBOUNDS_UNDECLARED_MAX can be tightened to {n_und}.\n"))
```
### Depth coverage
```{r}
#| label: depth_coverage
# A depth is a COORDINATE, and until v2026.08.14 nothing bounded one: that
# release shipped a CTD cast with scans at 14,671 m over a 101 m seafloor. The
# `pressure` value it was derived from (17,964 dbar) was deleted by its declared
# bound; the depth was not, because drop_out_of_bounds() cannot see a coordinate
# column. Two checks with two consequences:
# - the ABSOLUTE ceiling, CC_DEPTH_MAX_M (6,500 m; nothing in the region is
# deeper, and it is the ceiling already declared on `pressure`): an error,
# fails this render;
# - the SEAFLOOR at the position, +10 m over the deepest GEBCO cell within one
# cell of it: also impossible in principle, but the bottom we know is a ~460 m
# cell and 1949-1975 positions are rounded to the minute, so on a slope or a
# canyon a real cast reads deeper than the cell. Measured at v2026.08.14: 695
# of 412,640 root samples, all but that CTD cast within 1.2 km. Those are
# position-precision findings for the owning ingests — reported and ratcheted
# here, never deleted.
# `seafloor_depth_m` (bilinear GEBCO 2025, positive down, land 0, NA outside the
# raster) is stamped on `sample` so every consumer can make the same comparison.
DEPTH_SEAFLOOR_OVER_MAX <- 694L # ratchet: measured at v2026.08.25 (695 at v2026.08.14 incl. the CTD test cast); only ever DOWN
gebco_tif <- path.expand(Sys.getenv(
"CALCOFI_GEBCO_TIF",
"~/_big/gebco_2025_sub_ice_topo_geotiff/gebco_2025_sub_ice_n90.0_s0.0_w-180.0_e-90.0.tif"))
if (!file.exists(gebco_tif)) {
# D29 (2026-08-31): no local tile is no longer fatal — the same grid is published as a
# streamable COG, and GDAL's /vsicurl/ range reads fetch only the blocks the positions
# touch. The laptop keeps its local file (faster, offline); any other machine just runs.
gebco_tif <- paste0("/vsicurl/https://storage.googleapis.com/calcofi-db/",
"bathymetry/gebco_2025_sub_ice_n90_w180_e90_cog.tif")
cat(glue("CALCOFI_GEBCO_TIF names no local file - streaming {gebco_tif}\n"))
}
d_depth <- check_depth_bounds(
con_wdl, tbls = c("sample", "obs", intersect(supp_tbls, dbListTables(con_wdl))))
cat(glue("depth bounds: {sum(d_depth$status != 'ok')} of {nrow(d_depth)} ",
"(table, dataset, column) groups outside 0..{CC_DEPTH_MAX_M} m or NaN\n"))
d_depth |>
filter(status != "ok") |>
datatable(caption = glue(
"Depth coordinates outside 0..{CC_DEPTH_MAX_M} m, or NaN — fails this render"))
sf_sample <- sample_seafloor(con_wdl, gebco_tif)
add_sample_seafloor(con_wdl, gebco_tif, seafloor = sf_sample)
# station_uuid (calcofi4db >= 3.32.0, WS-B / Ed Weber's ask): the SWFSC station
# occupation (ichthyo `site`) each event belongs to — ichthyo's own rows carry
# their own site; bottle/CTD/PIC/crab roots are matched on cruise + station +
# occupation order, or a unique occupation within 24h. Called AFTER
# add_sample_seafloor() (both rebuild `sample` — DuckDB cannot UPDATE a table
# with a CRS-tagged geom column — so this one's own row-count/PK assertions are
# the last thing to touch the table before check_core_pk_unique() below).
d_station <- match_station_occupation(con_wdl)
d_station |>
datatable(caption = "station_uuid match method, by dataset (root samples only)")
# every NULL seafloor is one of four things, and only three may ship (D29): a sample
# with no/NaN coordinates or one genuinely off the GEBCO tile is the owning ingest's
# question; a positioned sample INSIDE the tile that still reads NULL can only be a
# regression in the sampling itself, and fails the release.
d_sf_null <- check_seafloor_nulls(con_wdl, source_bbox = c(-180, 0, -90, 90))
d_sf_null |>
datatable(caption = "Samples with NULL seafloor_depth_m, by cause (inside_tile_null fails the release)")
stopifnot(
"seafloor_depth_m is NULL for a positioned sample inside the GEBCO tile - the sampling regressed" =
attr(d_sf_null, "n_inside_null") == 0)
d_over <- check_depth_vs_seafloor(con_wdl, sf_sample, tolerance_m = 10)
attr(d_over, "summary") |>
datatable(caption = paste(
"Root samples deeper than the seafloor (+10 m over the deepest GEBCO cell",
"within one cell of the position), by dataset. `n_unknown` = outside the raster."))
d_over |>
head(200) |>
datatable(caption = "Worst 200 — fix position or depth at the owning ingest")
stopifnot(
"a depth coordinate is NaN, negative or beyond CC_DEPTH_MAX_M — fix at the ingest" =
all(d_depth$status == "ok"),
"more samples below the seafloor than the ratchet allows — fix at the ingest, never raise it" =
nrow(d_over) <= DEPTH_SEAFLOOR_OVER_MAX)
if (nrow(d_over) < DEPTH_SEAFLOOR_OVER_MAX)
cat(glue("\nDEPTH_SEAFLOOR_OVER_MAX can be tightened to {nrow(d_over)}.\n"))
```
### CTD accepted QC flags pending
`ingest_calcofi_ctd-cast.qmd`'s `apply_accepted_flags` chunk applies every
*accepted* row of the CTD team's PostgreSQL QA/QC ledger
(`gs://calcofi-db/qc/ctd/flag_accepted.parquet`, a nightly snapshot — see
`CLAUDE.md` § *The CTD team's PostgreSQL database*) and records how many it
applied in that ingest's `metadata.json` (`n_flags_applied`). This chunk
compares the snapshot's current count to that stamp: a gap means flags were
accepted by the team **after** the CTD ingest last rendered, so this release
still ships the pre-acceptance `measurement_qual`. **Warn only** — the CTD
ingest is 128 minutes and this round deliberately does not re-run it (§
*Avoiding the CTD ingest*, 2026-09-03 pre-release plan).
```{r}
#| label: qc_flags_pending
flag_snapshot_url <- "https://storage.googleapis.com/calcofi-db/qc/ctd/flag_accepted.parquet"
n_flags_snapshot <- tryCatch({
tmp <- tempfile(fileext = ".parquet")
utils::download.file(flag_snapshot_url, tmp, mode = "wb", quiet = TRUE)
n <- dbGetQuery(con_wdl, glue("SELECT COUNT(*) AS n FROM read_parquet('{tmp}')"))$n
unlink(tmp)
n
}, error = function(e) {
warning("could not reach flag_accepted.parquet: ", conditionMessage(e))
NA_integer_
})
ctd_meta_path <- here("data/parquet/calcofi_ctd-cast/metadata.json")
n_flags_applied <- if (file.exists(ctd_meta_path)) {
m <- jsonlite::read_json(ctd_meta_path)
if (!is.null(m$n_flags_applied)) as.integer(m$n_flags_applied) else 0L
} else 0L
n_flags_pending <- if (is.na(n_flags_snapshot)) NA_integer_ else
max(0L, n_flags_snapshot - n_flags_applied)
if (is.na(n_flags_pending)) {
cat("n_flags_pending: could not be measured (snapshot unreachable) — not fatal, warn only\n")
} else {
cat(glue(
"CTD accepted QC flags: {n_flags_snapshot} in snapshot · ",
"{n_flags_applied} reflected in this release's calcofi_ctd-cast shard · ",
"{n_flags_pending} pending (not yet applied)\n"))
if (n_flags_pending > 0)
warning(glue(
"{n_flags_pending} accepted CTD QC flag(s) are not yet reflected in this release — ",
"the CTD ingest has not rendered since they were accepted. Not fatal (warn only)."))
}
```
## Cruise Coverage — Samples With No Observations
The one shape of loss that every other check here is blind to, by construction.
PK/FK validation runs **child → parent**: it asks whether each `obs` row has a
parent in `sample`. A cruise whose observations vanish entirely leaves its casts
behind as parents with no children, which violates nothing. The bounds check above
inspects `obs`, which such a cruise has left. So `v2026.08.08` published 10
`calcofi_ctd-cast` cruises holding all 1,186 of their casts and none of their
874,000 observations, and every check in this notebook passed.
The grain is the **cruise**, not the sample: a CTD `sample` row is one physical
cast *per direction* while `obs` keeps one direction, so ~half of that dataset's
cast rows legitimately carry no observations. A dataset that emits no observations
at all is exempt — `sio_pic-zooplankton` is a net-tow registry whose biovolumes are
pending from the provider, so its 587 `sample`-only cruises are its designed state,
not 587 failures.
`ORPHAN_CRUISES_MAX` ratchets exactly like `BOUNDS_UNDECLARED_MAX`: it is the
per-dataset backlog measured on the day this landed, and it may only ever go
**down**. `calcofi_ctd-cast` is deliberately **not** in it — its correct value is
zero and the ingest now asserts that, so a release cut before the CTD ingest is
re-run fails here rather than republishing the loss.
```{r}
#| label: cruise_coverage
# ratchet, not a target. Each of these is an open question about a dataset that
# has cruises with no observations; none has been shown to be legitimate. Lower
# an entry whenever one is resolved at its ingest; never raise one.
ORPHAN_CRUISES_MAX <- c(
# every orphan is a position-less or mis-positioned event whose measurements
# exist but are held out of `obs` by the `grid_key IS NOT NULL` filter in each
# ingest's core projection. Investigated 2026-08-10; the causes differ:
# cce-lter_zoodb 156 tows with NULL datetime/latitude/grid_key
# swfsc_ichthyo 15 cruises have no tows in the source at all (nothing
# lost); the other 5 hold 1,977 ichthyo rows behind
# ungridded sites AND a NULL measurement_type
# calcofi_mets 1207OS publishes no lat/lon (mets_16, answered "skip the
# spatial join") — ungridded by design, not by defect. Was 5
# until the Longitude_W sign repair landed the other four.
# swfsc_cufes 1,475 samples with ZERO rows in cufes_measurement
# cce-lter_euph. 4 tows with ZERO rows in euphausiids_measurement
# The last two are not losses: the provider recorded the event and no counts.
# cdfw_dungeness-crab is deliberately ABSENT: its 14 orphan cruises are an
# inventory grain, not a loss, and that is expressed by EFFORT_ONLY_TYPES below
# rather than by an allowance. An allowance of 14 would also hide the next 14
# real losses in that dataset; the exemption hides none, because its observing
# sample types stay held to zero.
"cce-lter_zoodb" = 41L,
"swfsc_ichthyo" = 20L,
"swfsc_cufes" = 3L,
"calcofi_mets" = 1L,
"cce-lter_euphausiids" = 1L,
# farallon on ERDDAP (2026-09-04): CAC_FI_SBAS_obs carries NO observations for
# two cruises whose transects it does serve — 2021-01-33UD (490 transects) and
# 2022-10-33UD (260) — where DataZoo's export had 625 rows for the 2021 cruise.
# Taken as served (Q11, high, asked 2026-09-04); the transects are real effort
# and the Explorer keeps samples with no observation row out of every
# denominator. Ben, 2026-09-04: allow by name. Drops to 0 when Q11 is answered.
"farallon_bird-mammal" = 2L)
# NOT `d_cov` — that name is live from the `dataset_coverage` chunk above and is
# read again in `upload_frozen` (d_cov$coverage_temporal_observed). Reusing it
# here silently replaced that data frame and killed a 50-minute release run at
# the very last chunk, after the freeze and most of the upload had completed.
# Sample types that record EFFORT or INVENTORY rather than an analyzed event, so
# a cruise made only of them is not a finding. cdfw_dungeness-crab's 2,011 `tow`
# rows are a 60-year sorting log of which archived jars exist — only 216 were
# ever examined — while its 310 `subsample` rows are the lab-examined aliquots
# and every one yields obs. Keyed by dataset because `tow` IS an observing type
# for the net-tow ingests.
EFFORT_ONLY_TYPES <- c("cdfw_dungeness-crab" = "tow")
d_cruise_cov <- check_cruise_coverage(
con_wdl, max_orphan_cruises = ORPHAN_CRUISES_MAX,
effort_only_types = EFFORT_ONLY_TYPES)
d_cruise_cov |>
datatable(caption = paste(
"Cruises carrying samples with no observations, per dataset.",
"`emits_obs = FALSE` marks a registry-only dataset, exempt by design.",
"Anything above its ratchet fails this render."))
tighten <- d_cruise_cov$dataset_key[
d_cruise_cov$dataset_key %in% names(ORPHAN_CRUISES_MAX) &
d_cruise_cov$cruises_no_obs < ORPHAN_CRUISES_MAX[d_cruise_cov$dataset_key]]
if (length(tighten))
cat(glue(
"ORPHAN_CRUISES_MAX can be tightened: ",
"{paste(sprintf('%s -> %d', tighten,
d_cruise_cov$cruises_no_obs[match(tighten, d_cruise_cov$dataset_key)]),
collapse = ', ')}\n"))
```
## Ungridded Observations — Released, and Asked About
From v2026.08.11 `obs` carries observations that resolve no CalCOFI grid cell.
Every ingest used to filter `WHERE grid_key IS NOT NULL` in its core projection
while the `sample` arm did not, so an off-grid event kept its sample row and lost
every observation under it — which is how four `calcofi_mets` cruises reached
v2026.08.08 as 11,762 underway samples with zero observations.
The exclusion also contradicted this pipeline's own reasoning: `obs_mets_full`
was already gated on *a position* rather than on `grid_key` because "a ship on
transit is legitimately outside the CalCOFI station grid", and
`calcofi_phytoplankton` is region-pooled and has emitted ungridded `obs` from the
start.
Not dropping them puts the burden here instead: an ungridded observation is an
off-grid position, a coarser spatial notion, or **a coordinate error**, and
nothing in the pipeline can tell those apart. The sign-flipped `Longitude_W` that
put five CalCOFI cruises in the Taiwan Strait was invisible precisely *because*
being off-grid removed the rows silently. So this reports rather than gates, and
each dataset with a non-zero share owes a `questions.csv` entry — the `finding`
column is written to be pasted straight into one.
```{r}
#| label: ungridded_obs
d_ungridded <- check_ungridded_obs(con_wdl)
d_ungridded |>
select(dataset_key, n_obs, n_ungridded, pct_ungridded, n_no_position) |>
datatable(caption = paste(
"Observations resolving no CalCOFI grid cell, per dataset.",
"`n_no_position` is the subset carrying no latitude/longitude at all —",
"the distinction a provider needs in order to answer.")) |>
formatCurrency(c("n_obs", "n_ungridded", "n_no_position"),
currency = "", digits = 0, mark = ",")
d_ungridded |>
filter(!is.na(finding)) |>
select(dataset_key, finding) |>
datatable(caption = paste(
"Paste each finding into that dataset's questions.csv as `context`,",
"with status `open` until the provider says which of the three it is."))
```
## Scan Manifests for Mismatches
```{r}
#| label: scan_mismatches
# scan all ingest manifests for unresolved mismatches
all_manifests <- in_release_dirs(list.files(
"data/parquet", "manifest.json",
recursive = TRUE, full.names = TRUE))
all_mismatches <- purrr::compact(lapply(all_manifests, function(mf) {
m <- jsonlite::read_json(mf)
if (is.null(m$mismatches)) return(NULL)
dataset <- basename(dirname(mf))
purrr::imap_dfr(m$mismatches, function(items, category) {
if (length(items) == 0) return(NULL)
purrr::map_dfr(items, function(x) {
# replace NULL values with NA so as_tibble works
x[vapply(x, is.null, logical(1))] <- NA
as_tibble(x)
}) |>
mutate(dataset = dataset, category = category, .before = 1)
})
}))
if (length(all_mismatches) > 0) {
d_mismatches <- bind_rows(all_mismatches)
message(glue("{nrow(d_mismatches)} unresolved mismatch(es) across manifests"))
d_mismatches |>
datatable(caption = "Unresolved mismatches (from manifest.json)")
} else {
message("No unresolved mismatches found across manifests")
}
```
## Validate
Cross-dataset validation to ensure data integrity before freezing.
```{r}
#| label: validate
# core primary keys: HARD gate. v2026.08.25 shipped `sample` with 4,855 keys twice
# (a seafloor-stamp join bug) while this chunk only warned on ship/cruise; every
# count or join keyed on sample over-counted those rows 2x until the next release.
pk_res <- check_core_pk_unique(con_wdl, c(core_tbls, "taxon", "dataset_taxon", "cruise",
"ship", "grid", "dataset", "measurement_type"))
cat(glue("core primary keys unique: {paste(pk_res$table, collapse = ', ')}\n"))
# grid_key integrity: casts.grid_key should all be in grid.grid_key
tbls <- DBI::dbListTables(con_wdl)
if (all(c("casts", "grid") %in% tbls)) {
# use information_schema to check columns (avoids GEOMETRY type issues)
casts_cols_wdl <- dbGetQuery(
con_wdl,
"SELECT column_name FROM information_schema.columns
WHERE table_name = 'casts'"
)$column_name
grid_cols_wdl <- dbGetQuery(
con_wdl,
"SELECT column_name FROM information_schema.columns
WHERE table_name = 'grid'"
)$column_name
if ("grid_key" %in% casts_cols_wdl && "grid_key" %in% grid_cols_wdl) {
grid_orphans <- dbGetQuery(
con_wdl,
"SELECT COUNT(*) AS n FROM casts c
WHERE c.grid_key IS NOT NULL
AND c.grid_key NOT IN (SELECT grid_key FROM grid)"
)$n
message(glue("Grid key orphans in casts: {grid_orphans}"))
# Grid key orphans in casts: 0
}
}
# ship PK uniqueness
if ("ship" %in% tbls) {
ship_dups <- dbGetQuery(
con_wdl,
"SELECT ship_key, COUNT(*) AS n FROM ship
GROUP BY ship_key HAVING COUNT(*) > 1"
)
if (nrow(ship_dups) > 0) {
warning(glue("Duplicate ship_key values: {nrow(ship_dups)}"))
} else {
message("ship_key: all unique")
}
}
# ship_key: all unique
# cruise PK uniqueness
if ("cruise" %in% tbls) {
cruise_dups <- dbGetQuery(
con_wdl,
"SELECT cruise_key, COUNT(*) AS n FROM cruise
GROUP BY cruise_key HAVING COUNT(*) > 1"
)
if (nrow(cruise_dups) > 0) {
warning(glue("Duplicate cruise_key values: {nrow(cruise_dups)}"))
} else {
message("cruise_key: all unique")
}
}
# cruise_key: all unique
# cruise bridge coverage
if ("casts" %in% tbls) {
bridge_stats <- dbGetQuery(
con_wdl,
"SELECT
COUNT(*) AS total_casts,
SUM(CASE WHEN ship_key IS NOT NULL THEN 1 ELSE 0 END) AS with_ship_key,
SUM(CASE WHEN cruise_key IS NOT NULL THEN 1 ELSE 0 END) AS with_cruise_key
FROM casts"
)
bridge_stats |> datatable(caption = "Cruise bridge coverage")
}
# cruise_key format validation: this used to be a warn-only regexp check here
# ("cruise_key format violations: 1 rows" / cruise_key: 2019-07-, forever, on
# every render). It is superseded by calcofi4db::check_cruise_key_integrity()
# (WS-B, the `cruise_key_integrity` chunk below, run once `cruise` is complete
# and enriched) — a HARD gate covering format, date_ym/NODC agreement, the FK
# to `cruise`, cruise_uuid hygiene and date-span containment, not a warning
# repeated release after release.
# site_key format validation (NNN.N NNN.N)
for (tbl_name in intersect(c("site", "casts", "ctd_cast"), tbls)) {
tbl_cols <- dbGetQuery(
con_wdl,
glue(
"SELECT column_name FROM information_schema.columns
WHERE table_name = '{tbl_name}'"
)
)$column_name
if ("site_key" %in% tbl_cols) {
bad_sk <- dbGetQuery(
con_wdl,
glue(
"SELECT COUNT(*) AS n FROM {tbl_name}
WHERE site_key IS NOT NULL
AND NOT regexp_matches(site_key, '^\\d{{3}}\\.\\d \\d{{3}}\\.\\d$')"
)
)$n
if (bad_sk > 0) {
warning(glue("site_key format violations in {tbl_name}: {bad_sk} rows"))
} else {
message(glue("site_key in {tbl_name}: all match NNN.N NNN.N format"))
}
}
}
# site_key in casts: all match NNN.N NNN.N format
# Warning message: site_key format violations in site: 982 rows
# complete the `cruise` reference BEFORE enrichment (WS-B design memo D4):
# `cruise` is the SWFSC ichthyo export's station-occupation cruise list, not a
# designation registry — 152 cruise_keys that bottle/CTD/METS/picoplankton key
# events to (measured at v2026.08.25: 153,306 sample rows, 3.8M obs rows) name
# no row in it, and nothing failed. This adds one `cruise` row per such key
# (cruise_key_method = 'derived') so the FK holds by construction and stamps
# cruise_key_method = 'swfsc' + cruise_key_datasets on every row, existing and
# new. Must run before the enrichment below so the added rows get
# year/month/ship_name/ship_nodc/per-dataset counts the same way every other
# cruise does (via `cr.*` in that SELECT).
if (all(c("cruise", "ship", "sample") %in% tbls)) {
cruise_added <- complete_cruise_reference(con_wdl)
cat(glue("cruise reference completed: {nrow(cruise_added)} derived row(s) added\n"))
if (nrow(cruise_added) > 0)
cruise_added |> datatable(caption = "Derived cruise rows added (cruise_key_method = 'derived')")
tbls <- DBI::dbListTables(con_wdl) # `cruise` may have been materialized from a VIEW
}
# enrich the `cruise` reference in place with per-cruise x dataset event counts
# from the consolidated obs/sample (this is the former `cruise_summary`, folded
# into `cruise` so there is a single cruise table). count(DISTINCT root_sample_key)
# = distinct sampling-event roots (station occupations for net tows; casts for
# bottle/CTD/DIC). LEFT JOINs keep every reference cruise (no cruise dropped, so
# cruise_key FKs stay valid) and all its columns (cr.*). Add a FILTER column to
# extend to new datasets.
if (all(c("cruise", "ship") %in% tbls)) {
dbExecute(con_wdl, "CREATE OR REPLACE TEMP TABLE cruise_ref AS SELECT * FROM cruise")
# drop by the object's actual type: complete_cruise_reference() above turns the
# `cruise` VIEW into a TABLE, and DuckDB's `DROP VIEW IF EXISTS` errors on a
# table ("Existing object cruise is of type Table") rather than no-op — the
# unconditional pair here halted the 2026-09-04 staging run 27 minutes in.
cruise_kind <- dbGetQuery(con_wdl,
"SELECT table_type FROM information_schema.tables WHERE table_name = 'cruise'")$table_type
if (length(cruise_kind))
dbExecute(con_wdl, if (grepl("VIEW", cruise_kind[1], ignore.case = TRUE))
"DROP VIEW cruise" else "DROP TABLE cruise")
dbExecute(
con_wdl,
"CREATE TABLE cruise AS
WITH ev AS (
SELECT o.cruise_key, o.dataset_key, COUNT(DISTINCT s.root_sample_key) AS n_events
FROM obs o JOIN sample s ON o.sample_key = s.sample_key
WHERE o.cruise_key IS NOT NULL
GROUP BY 1, 2),
piv AS (
SELECT cruise_key,
COALESCE(MAX(n_events) FILTER (WHERE dataset_key = 'swfsc_ichthyo'), 0) AS ichthyo,
COALESCE(MAX(n_events) FILTER (WHERE dataset_key = 'calcofi_bottle'), 0) AS bottle,
COALESCE(MAX(n_events) FILTER (WHERE dataset_key = 'calcofi_ctd-cast'), 0) AS ctd_cast,
COALESCE(MAX(n_events) FILTER (WHERE dataset_key = 'calcofi_dic'), 0) AS dic
FROM ev GROUP BY 1)
SELECT cr.*,
EXTRACT(YEAR FROM cr.date_ym)::INTEGER AS year,
EXTRACT(MONTH FROM cr.date_ym)::INTEGER AS month,
sh.ship_name, sh.ship_nodc,
COALESCE(piv.ichthyo, 0) AS ichthyo,
COALESCE(piv.bottle, 0) AS bottle,
COALESCE(piv.ctd_cast, 0) AS ctd_cast,
COALESCE(piv.dic, 0) AS dic
FROM cruise_ref cr
LEFT JOIN ship sh ON cr.ship_key = sh.ship_key
LEFT JOIN piv USING (cruise_key)
ORDER BY year DESC, month DESC")
n_cs <- dbGetQuery(con_wdl, "SELECT COUNT(*) AS n FROM cruise")$n
message(glue("Enriched cruise table: {n_cs} rows"))
}
tbl(con_wdl, "cruise") |>
collect() |>
datatable(caption = "cruise (enriched with per-dataset event counts)")
# run standard release validation (wrapped in tryCatch for GEOMETRY compat)
tryCatch(
{
validation <- validate_for_release(con_wdl)
if (validation$passed) {
message("Release validation passed!")
} else {
cat("Validation FAILED:\n")
cat(paste("-", validation$errors, collapse = "\n"))
}
},
error = function(e) {
message(glue("validate_for_release skipped: {e$message}"))
}
)
```
## Cruise Key Integrity
`calcofi4db::check_cruise_key_integrity()` (>= 3.32.0, WS-B / Ed Weber's ask) is the hard
gate over `cruise_key` as it is actually used, not as it is assumed to behave: format,
`date_ym`/NODC agreement, the FK from `sample`/`obs` into the (now-completed) `cruise`
reference, `cruise_uuid` hygiene (unique for `'swfsc'` rows, NULL for `'derived'` ones),
every event's date within its cruise's span, the ichthyo notebook's own
`cruise_uuid`/`cruise_key` agreement (read from its manifest — the release's `sample` table
carries no `cruise_uuid` column, so that check can only run inside the notebook that still
has it) and three ratchets (span overlaps, derived-row count, NULL-`cruise_key` backlog per
dataset). See CLAUDE.md § "Provider UUIDs are columns, and the cruise key is checked
against the cruise" and the WS-B design memo (`.claude/plans_todo/2026-09-03 WS-B …md`) for
the numbers behind each check.
```{r}
#| label: cruise_key_integrity
# the ichthyo notebook's own cruise_uuid <-> cruise_key check can only run
# there, while `site.cruise_uuid` still exists (dropped by its compat VIEW
# rebuild before parquet is written) — read its result out of the manifest.
ich_manifest_path <- here("data/parquet/swfsc_ichthyo/manifest.json")
manifest_ichthyo_n <- if (file.exists(ich_manifest_path)) {
m <- jsonlite::read_json(ich_manifest_path)
cu <- m$mismatches$cruise_uuid
if (length(cu) >= 1 && !is.null(cu[[1]]$n_mismatch)) as.integer(cu[[1]]$n_mismatch) else NA_integer_
} else NA_integer_
cat(glue("ichthyo manifest cruise_uuid mismatch count: ",
"{if (is.na(manifest_ichthyo_n)) 'NOT FOUND (re-render ingest_swfsc_ichthyo.qmd)' else manifest_ichthyo_n}\n"))
# seven calcofi_ctd-cast casts whose header TIMESTAMP is 187-948 days off
# (1997/2012 dates inside 1999/2013 archives) — CTD keys every cast from its
# ARCHIVE NAME, so cruise_key is right by construction; the timestamp is the
# defect (ctd-cast Q32). Named exceptions, not a raised tolerance — an eighth,
# unlisted violator still fails. Not fixed this round: the CTD ingest is not
# being re-run (umbrella § "Avoiding the CTD ingest").
ctd_span_exceptions <- c(
"calcofi_ctd-cast:cast:9908_067d", "calcofi_ctd-cast:cast:9908_067u",
"calcofi_ctd-cast:cast:9908_069d", "calcofi_ctd-cast:cast:9908_069u",
"calcofi_ctd-cast:cast:9908_070d", "calcofi_ctd-cast:cast:9908_070u",
"calcofi_ctd-cast:cast:1307_021u")
d_cki <- check_cruise_key_integrity(
con_wdl,
tolerance_days = 31L,
known_outside_span = ctd_span_exceptions,
manifest_ichthyo = manifest_ichthyo_n,
ratchets = list(
span_overlaps_max = 2L, # measured 3 pairs at v2026.08.25, 1 within transit tolerance
derived_max = 152L, # 152 cruises no SWFSC site row names; only ever down
key_null_max = c( # root samples with NULL cruise_key, per dataset; only ever down
# measured by the WS-B spike on v2026.08.25 (dic: unmatched Niskins, Q07;
# pic: pre-1951 tows; bottle: 49 casts)
calcofi_dic = 3255L,
"sio_pic-zooplankton" = 5087L,
calcofi_bottle = 49L,
# crab: was 1639 over the 2,011 sorting-log tows; the core now holds the
# 216 examined ones only (WS-C), 97 of whose log entries resolve no cruise
"cdfw_dungeness-crab" = 97L,
# the backlog the gate measured the first time it ran end-to-end
# (2026-09-04 staging, first attempt through this chunk): datasets the
# spike never counted, each an open question for its ingest to file —
# underway/transect events with no cruise designation (cufes, farallon:
# CAC2022_8 has no Aug-2022 ichthyo cruise), net tows whose year-month
# is ambiguous between two cruises of one ship (zoodb, zooscan,
# euphausiids, phyllosoma), picoplankton's undesignated casts, and the
# region-pooled phytoplankton. Baselines, not targets.
calcofi_phyllosoma = 44L,
calcofi_phytoplankton = 168L,
"cce-lter_euphausiids" = 420L,
"cce-lter_picoplankton-bacteria" = 7216L,
"cce-lter_zoodb" = 111L,
"cce-lter_zooscan" = 421L,
"farallon_bird-mammal" = 1228L,
swfsc_cufes = 5053L)),
halt = TRUE)
d_cki |>
datatable(caption = paste(
"cruise_key integrity — one row per check/ratchet.",
"mode = fail stops the release on n > 0; mode = ratchet stops it on n > allowance."))
```
## Show Combined Schema
```{r}
# dir_frozen used later; define early so ERD can reference it
dir_frozen <- here(glue("{sidecar_root}/{release_version}"))
dir.create(dir_frozen, recursive = TRUE, showWarnings = FALSE)
# --- retire the per-dataset event/measurement/summary + per-dataset taxon tables
# the consolidated core (obs/sample/obs_attribute/sample_measurement) + unified
# taxon/dataset_taxon/taxon_group replace them; all were materialized upstream, so
# drop them now. `core_keep` = the DEFAULT published set (in the ERD + catalog);
# `supplemental_keep` = ancillary full tables that are hosted + tagged to the
# release but excluded from the ERD and default table list (opt-in deep dives).
core_keep <- c(
"obs", "sample", "obs_attribute", "sample_measurement", # core facts
"grid", "cruise", "ship", "measurement_type", "dataset", "region", # refs
"taxon", "dataset_taxon", "taxon_group", "lookup", # taxa + lookups
"spatial", "spatial_attribute")
supplemental_keep <- supp_tbls # hosted, hidden by default (from the ingests)
# drop by the object's actual type (DROP VIEW on a TABLE — or vice versa — errors
# even with IF EXISTS; some are parquet VIEWs, some are temp tables like cruise_ref)
retire_objs <- DBI::dbGetQuery(con_wdl,
"SELECT table_name, table_type FROM information_schema.tables")
retire_objs <- retire_objs[!retire_objs$table_name %in% c(core_keep, supplemental_keep) &
retire_objs$table_name != "_measurement_taxon", , drop = FALSE]
for (i in seq_len(nrow(retire_objs))) {
kind <- if (grepl("VIEW", retire_objs$table_type[i], ignore.case = TRUE)) "VIEW" else "TABLE"
DBI::dbExecute(con_wdl, glue('DROP {kind} IF EXISTS "{retire_objs$table_name[i]}"'))
}
retire_tbls <- retire_objs$table_name
message(glue("retired {length(retire_tbls)} per-dataset tables: ",
"{paste(head(retire_tbls, 8), collapse=', ')}…"))
erd <- cc_erd(con_wdl, layout = "elk")
plot(erd)
erd <- cc_erd(con_wdl, colors = color_map)
plot(erd)
```
```{r}
#| label: combined_schema
#| fig-width: 12
#| fig-height: 10
# exclude internal tables and the SUPPLEMENTAL tables (obs_ctd_full) so the ERD
# stays the default core schema. Use the CURRENT tables (the retire step above
# dropped the per-dataset tables) so the ERD + FK checks are core-only.
schema_tbls <- setdiff(
DBI::dbListTables(con_wdl),
c("_meta", "_sp_update", "casts_derived", "ctd_cast_derived",
"_measurement_taxon", supplemental_keep))
# merge per-dataset relationships.json files (auto-discovered — every ingest
# writes data/parquet/{provider}_{dataset}/relationships.json, so new datasets
# are picked up without editing this list)
rels_paths <- in_release_dirs(
Sys.glob(here("data/parquet/*/relationships.json")))
dir_frozen <- here(glue("{sidecar_root}/{release_version}"))
dir.create(dir_frozen, recursive = TRUE, showWarnings = FALSE)
rels_merged_path <- file.path(dir_frozen, "relationships.json")
if (length(rels_paths) > 0) {
merge_relationships_json(rels_paths, rels_merged_path)
# append cross-dataset FKs authored in metadata/relationships_cross.csv
rels_merged <- jsonlite::fromJSON(
rels_merged_path, simplifyVector = FALSE)
rels_merged$foreign_keys <- c(
rels_merged$foreign_keys, cross_fks)
jsonlite::write_json(
rels_merged, rels_merged_path,
auto_unbox = TRUE, pretty = TRUE, null = "null")
# emit a flat, reviewable view of every relationship (intra + cross) alongside
# relationships.json / erd.mmd, so the cross-dataset graph is legible as a table
g <- function(x, k) { v <- x[[k]]; if (is.null(v)) NA_character_ else as.character(v) }
fk_df <- do.call(rbind, lapply(rels_merged$foreign_keys, function(fk)
data.frame(
from_table = g(fk, "table"), from_column = g(fk, "column"),
to_table = g(fk, "ref_table"), to_column = g(fk, "ref_column"),
stringsAsFactors = FALSE)))
cross_keys <- paste(cross_fks_df$table, cross_fks_df$column,
cross_fks_df$ref_table, cross_fks_df$ref_column)
fk_df$scope <- ifelse(
paste(fk_df$from_table, fk_df$from_column,
fk_df$to_table, fk_df$to_column) %in% cross_keys, "cross", "intra")
readr::write_csv(fk_df, file.path(dir_frozen, "relationships_all.csv"))
# validate: every cross-FK target column exists in the assembled release schema
schema_cols <- unlist(lapply(schema_tbls, function(t)
paste(t, DBI::dbListFields(con_wdl, t))))
cross_targets <- paste(cross_fks_df$ref_table, cross_fks_df$ref_column)
missing_targets <- cross_fks_df[!(cross_targets %in% schema_cols), , drop = FALSE]
if (nrow(missing_targets) > 0) {
warning(glue(
"cross-FK target(s) missing from release schema: ",
"{paste(missing_targets$ref_table, missing_targets$ref_column, collapse = ', ')}"))
} else {
message(glue(
"cross-FK check: all {nrow(cross_fks_df)} cross-dataset targets present; ",
"wrote {nrow(fk_df)} relationships to relationships_all.csv"))
}
}
# render dataset-colored ERD (stroke outlines; cc_erd handles GEOMETRY natively)
cc_erd(
con_wdl,
tables = schema_tbls,
rels_path = rels_merged_path,
colors = color_map)
```
## Normalize geometry CRS
Every geometry column in the release is tagged **EPSG:4326**, here, at the last
point before the freeze — so the guarantee holds regardless of what any
individual ingest produced.
It did not hold before. v2026.08.02 shipped `spatial.geom` as `EPSG:4326` but
`sample.geom` and both `grid` geometries as `OGC:CRS84`, because they are minted
differently: `ST_Read()` over GeoJSON tags one thing and `ST_Point(lon, lat)`
tags another. The two label the *same* coordinates — WGS 84 lon/lat — but
**DuckDB refuses `ST_Intersects` across differing CRS tags**, so a spatial join
between `sample` and `spatial` simply errored, which is how this was found.
`ST_SetCRS` relabels without transforming, which is what is wanted: nothing here
is being reprojected, the tags are being made to agree. (Pedantically EPSG:4326
declares lat/lon axis order while all of this is lon/lat, so `OGC:CRS84` is the
more literal label — but EPSG:4326 is the conventional one, is what
`calcofi4r::cc_tbl()` assigns to consumers, and is what the ingests already
document.)
Doing it at release time rather than only at source means a fix does not require
re-running all 16 ingests, and a future ingest that mints geometry some third way
cannot reintroduce the mismatch.
```{r}
#| label: normalize_crs
# Non-finite coordinates first: NaN is not NULL, so it survives IS NOT NULL, and
# ST_Point(NaN, NaN) is a real non-NULL GEOMETRY that survives `geom IS NOT NULL`
# too. Its presence does not merely add junk rows — it makes ST_Intersects return
# a DIFFERENT NUMBER OF MATCHES at different thread counts, dropping valid
# unrelated pairs, so every spatial join over v2026.08.02 silently under-counted
# by a different amount on every machine. calcofi4db 3.4.2 stops them being
# minted; normalizing here as well means the release is clean without re-running
# all 16 ingests for shards written before that.
n_nonfinite <- DBI::dbGetQuery(con_wdl, "
SELECT COUNT(*) AS n FROM sample
WHERE isnan(latitude) OR isnan(longitude) OR isinf(latitude) OR isinf(longitude)")$n
if (n_nonfinite > 0) {
DBI::dbExecute(con_wdl, "
CREATE OR REPLACE TABLE sample AS
SELECT * REPLACE (
CASE WHEN isnan(latitude) OR isinf(latitude) THEN NULL ELSE latitude END AS latitude,
CASE WHEN isnan(longitude) OR isinf(longitude) THEN NULL ELSE longitude END AS longitude,
CASE WHEN isnan(latitude) OR isinf(latitude) OR isnan(longitude) OR isinf(longitude)
THEN NULL ELSE geom END AS geom)
FROM sample")
cat(glue("normalized {n_nonfinite} non-finite coordinate(s) in sample to NULL ",
"(and dropped their geometry)"), "\n")
} else {
cat("no non-finite coordinates in sample\n")
}
geom_cols <- DBI::dbGetQuery(con_wdl, "
SELECT c.table_name, c.column_name, t.table_type
FROM information_schema.columns c
JOIN information_schema.tables t
ON t.table_schema = c.table_schema AND t.table_name = c.table_name
WHERE c.table_schema = 'main' AND c.data_type LIKE 'GEOMETRY%'
ORDER BY c.table_name, c.column_name")
# One pass per TABLE, not per column: `grid` carries both `geom` and `geom_ctr`,
# and rewriting the table once per column would do the work twice.
#
# Several of these are VIEWs over the ingest parquet (load_prior_tables(as_view =
# TRUE)), and DuckDB refuses `CREATE OR REPLACE TABLE` over a view — so
# materialize to a temp name, drop the original as whatever type it actually is,
# then rename. That also converts the view into a real table, which is required
# anyway: a view would just re-read the un-normalized parquet underneath.
for (tb in unique(geom_cols$table_name)) {
cols <- geom_cols[geom_cols$table_name == tb, ]
repl <- paste(sprintf("ST_SetCRS(%s, 'EPSG:4326') AS %s",
cols$column_name, cols$column_name), collapse = ", ")
tmp <- paste0("_crsnorm_", tb)
DBI::dbExecute(con_wdl, glue(
"CREATE OR REPLACE TABLE {tmp} AS SELECT * REPLACE ({repl}) FROM {tb}"))
kind <- if (grepl("VIEW", cols$table_type[1], ignore.case = TRUE)) "VIEW" else "TABLE"
DBI::dbExecute(con_wdl, glue('DROP {kind} IF EXISTS "{tb}"'))
DBI::dbExecute(con_wdl, glue('ALTER TABLE {tmp} RENAME TO "{tb}"'))
cat(glue(" {tb}: {nrow(cols)} geometry column(s) -> EPSG:4326 (was a {kind})"), "\n")
}
cat(glue("normalized {nrow(geom_cols)} geometry column(s) across ",
"{length(unique(geom_cols$table_name))} table(s)"), "\n")
# assert it took — a silent partial normalization would put us straight back to
# a join that errors only for some pairs
still_off <- DBI::dbGetQuery(con_wdl, "
SELECT table_name || '.' || column_name AS col
FROM information_schema.columns
WHERE table_schema = 'main' AND data_type LIKE 'GEOMETRY%'
AND data_type NOT LIKE '%EPSG%'")
stopifnot("every geometry column must be tagged EPSG:4326" = nrow(still_off) == 0)
cat("all geometry columns tagged EPSG:4326\n")
# THE ASSERTION ABOVE IS NOT ENOUGH ON ITS OWN. Most tables are uploaded by a GCS
# server-side copy straight from the ingest bucket, never passing through this
# connection — so normalizing here would leave the connection clean, the assertion
# green, and the PUBLISHED grid.parquet still tagged OGC:CRS84. Force every table
# whose geometry was touched to be exported locally and uploaded from there.
crs_local_tables <- unique(geom_cols$table_name)
cat(glue("these will be uploaded from the LOCAL export, not GCS-copied: ",
"{paste(crs_local_tables, collapse=', ')}"), "\n")
```
## Create Frozen Release
Strip provenance columns and export clean parquet files for public
access. See
[Frozen DuckLake](https://ducklake.select/2025/10/24/frozen-ducklake/)
pattern.
### Release narrative (RELEASES.md)
```{r}
#| label: browser_objects
# --- browser-shaped release objects (CalCOFI Explorer plan D4; calcofi4db >= 3.24.0) ----------------
# One object per lens, fetched whole by the browser and aggregated there: `obs_bio` (the bio realm,
# ~22 MB), `obs_env/measurement_type=*` (one variable = one object), `sample_root` (the dense integer
# join key + the cruise tracks) and `sample_spatial` (exact polygon membership per root sample,
# computed ONCE here, chunked per layer, instead of per app on the 16 GB server), plus
# `coverage.json` for the first paint before DuckDB-WASM wakes up. The quality predicate and the
# density expression are calcofi4r's — cc_qual_ok_sql(), cc_density_sql() — so there is one copy of
# each (D8 rule 2); the picker defaults (rule 4) are cc_default_stage()/cc_default_denominator().
# Built after normalize_crs so non-finite coordinates are already NULL.
qual_sql <- calcofi4r::cc_qual_ok_sql("o")
dens_sql <- calcofi4r::cc_density_sql()
n_root <- build_sample_root(con_wdl)
n_bio <- build_obs_slim(con_wdl, "bio", qual_ok_sql = qual_sql, density_sql = dens_sql)
n_env <- build_obs_slim(con_wdl, "env", qual_ok_sql = qual_sql, density_sql = dens_sql)
# --- D-S1 (pre-release plan 2026-09-03; calcofi4db >= 3.31.0): the pair IS the observation store ----
# obs_bio / obs_env now carry sample_key, measurement_prec and hex_id too, so each is a strict superset
# of obs under a name mapping (realm = the table, value = measurement_value), and `obs` becomes a VIEW
# the catalog carries (`views.obs`, calcofi4db::obs_view_sql(), written by build_release_catalog() from
# release_views() because both tables ship) while its own objects still ship this once — the catalog
# marks the table `deprecated`, `replaced_by: [obs_bio, obs_env]`, `removed_in: next`. cc_get_db() (R and
# Python) and db-query's `__TBL:obs__` serve `obs` through the view from this release on, so the gate
# is that the pair reproduces obs: per (realm, dataset_key) the row count, the distinct obs_ids and an
# order-independent signature of every non-depth column must match, and no non-NULL depth may change.
# The one documented difference is the depth FALLBACK (a bio row with no depth in obs carries its
# tow's span: 482,250 ichthyo rows at v2026.08.28), reported per dataset as n_depth_filled.
obs_parity <- check_obs_pair_parity(con_wdl)
datatable(obs_parity, caption = "obs_bio + obs_env reproduce obs (check_obs_pair_parity: counts, ids, signature; depth filled / changed)")
stopifnot(all(obs_parity$ok), sum(obs_parity$n_pair) == n_bio + n_env,
dbGetQuery(con_wdl, glue("SELECT count(*) AS n FROM ({substitute_view_tables(obs_view_sql())})"))$n ==
dbGetQuery(con_wdl, "SELECT count(*) AS n FROM obs")$n)
ss <- build_sample_spatial(con_wdl)
# the ONE seasonal baseline every anomaly is a departure from (calcofi4db >= 3.26.0): a plain mean per dataset x
# station x calendar month x 10 m depth_bin x measurement type over 1993-2013, >= 3 cruises per cell, the window
# stamped on every row. ctd-transects, the Explorer's Sections lens and calcofi4r::cc_climatology() subtract this
# table — each used to compute its own, and by 2026-08-31 the three had drifted (all-months pooling; 5 m bins over
# the thinned 10 m series; one arbitrary cast per grid cell), so one July 2026 section read +1.4 degC in one
# product and ~0 in another. Partitioned by measurement_type like obs_env, so a browser fetches one variable.
n_clim <- build_climatology(con_wdl, qual_ok_sql = qual_sql)
cat(glue("sample_root {n_root} · obs_bio {n_bio} · obs_env {n_env} · climatology {n_clim} rows\n"))
datatable(dbGetQuery(con_wdl, "
SELECT dataset_key, count(*) AS n_cells, count(DISTINCT measurement_type) AS n_types, count(DISTINCT grid_key) AS n_stations,
min(n_cruises) AS min_cruises, round(avg(n_cruises), 1) AS mean_cruises, any_value(clim_yr_min) AS yr_min, any_value(clim_yr_max) AS yr_max
FROM climatology GROUP BY 1 ORDER BY 1"), caption = "climatology — cells per dataset (each ≥ 3 cruises; the window is on every row)")
stopifnot(n_clim > 0, nrow(dbGetQuery(con_wdl, "SELECT DISTINCT clim_yr_min, clim_yr_max FROM climatology")) == 1)
datatable(ss, caption = "sample_spatial — exact per-root-sample membership, one layer at a time (n_polys = 0: a line or point layer, skipped)")
# what the picker shows for the default taxon (Pacific sardine): dataset × stage × effort class
datatable(dbGetQuery(con_wdl, "
SELECT dataset_key, life_stage, effort_class, tow_type, count(*) AS n,
count(density_per_10m2) AS n_10m2, count(density_per_1000m3) AS n_1000m3
FROM obs_bio WHERE taxon_key = 'worms:217452' GROUP BY ALL ORDER BY 1, 2, 3, 4"),
caption = "obs_bio · worms:217452 (Pacific sardine) · the D8 rule-2 columns")
# depth and hex coverage of the bio rows per dataset — a dataset with n_depth = 0 carries no tow
# span anywhere in `sample` (swfsc_ichthyo Q08, cce-lter_euphausiids): an ingest gap, not a cut gap
datatable(dbGetQuery(con_wdl, "
SELECT dataset_key, count(*) AS n, count(depth_max_m) AS n_depth, count(hex7) AS n_hex,
count(*) FILTER (WHERE qual_ok) AS n_qual_ok, count(DISTINCT effort_class) AS n_classes
FROM obs_bio GROUP BY 1 ORDER BY 1"), caption = "obs_bio — depth, hex and quality coverage per dataset")
cov <- build_coverage(con_wdl, release_version)
coverage_path <- file.path(dir_frozen, "coverage.json")
jsonlite::write_json(cov, coverage_path, auto_unbox = TRUE, digits = NA)
# the per-station year x month card, its own sidecar so the first paint stays small
coverage_stations_path <- file.path(dir_frozen, "coverage_stations.json")
jsonlite::write_json(build_coverage_stations(con_wdl, release_version), coverage_stations_path, auto_unbox = TRUE, digits = NA)
# the two GeoJSON sidecars the map draws without a spatial extension: grid cells + centroids, and the
# polygon layers (simplified for display; membership is exact and lives in sample_spatial)
grid_geojson_path <- file.path(dir_frozen, "grid.geojson")
spatial_geojson_path <- file.path(dir_frozen, "spatial.geojson")
spatial_layers_path <- file.path(dir_frozen, "spatial_layers.json")
unlink(c(grid_geojson_path, spatial_geojson_path, spatial_layers_path))
dbExecute(con_wdl, glue("COPY (SELECT grid_key, station, line, shore, pattern, zone, ST_X(geom_ctr) AS lon_ctr, ST_Y(geom_ctr) AS lat_ctr,
ST_GeomFromWKB(ST_AsWKB(geom)) AS geom FROM grid)
TO '{grid_geojson_path}' WITH (FORMAT GDAL, DRIVER 'GeoJSON')"))
dbExecute(con_wdl, glue("COPY (SELECT spatial_key, layer, name, ST_SimplifyPreserveTopology(ST_GeomFromWKB(ST_AsWKB(geom)), 0.002) AS geom
FROM spatial WHERE ST_GeometryType(geom) IN ('POLYGON', 'MULTIPOLYGON'))
TO '{spatial_geojson_path}' WITH (FORMAT GDAL, DRIVER 'GeoJSON')"))
# the boundary-layer registry + what only the release knows about each layer (counts, bbox, names,
# sample_spatial memberships) — the explorer's Layers card reads this, never the CSV (D23, calcofi4db >= 3.28.0).
# `built` = the ingest_spatial manifest's mtime: the PMTiles archives are rebuilt outside releases, and the skew
# must at least be visible (plan 2026-08-31, Risks).
jsonlite::write_json(
build_spatial_layers(
con_wdl, here("metadata/spatial_layers.csv"), release_version,
pmtiles_base = "https://storage.googleapis.com/calcofi-files-public/_spatial/",
built = format(file.mtime(here("data/parquet/spatial/manifest.json")), "%Y-%m-%d")),
spatial_layers_path, auto_unbox = TRUE, digits = NA, null = "null")
cat(glue("coverage.json {round(file.size(coverage_path) / 1e3)} KB · coverage_stations.json {round(file.size(coverage_stations_path) / 1e3)} KB · ",
"grid.geojson {round(file.size(grid_geojson_path) / 1e3)} KB · spatial.geojson {round(file.size(spatial_geojson_path) / 1e6, 1)} MB · ",
"spatial_layers.json {round(file.size(spatial_layers_path) / 1e3)} KB\n"))
# sample_spatial is a core relationship table (ERD + default table list); since D-S1 obs_bio and
# obs_env are CORE too — the observation tables every consumer reads, in the ERD and cc_get_db()'s
# default set (obs is the deprecated table + the view over them; see the parity gate above). Only
# sample_root stays supplemental — hosted and catalogued, hidden from cc_get_db() by default.
core_keep <- c(core_keep, "sample_spatial", "climatology", # climatology: default table, in the ERD (FKs to grid, dataset, measurement_type)
"obs_bio", "obs_env")
supplemental_keep <- c(supplemental_keep, "sample_root")
```
```{r}
#| label: release_notes_narrative
# RELEASES.md is the database's NEWS file (CLAUDE.md § "RELEASES.md is not
# optional"). A release with no section for itself is refused HERE, before the
# freeze, for the same reason the packages refuse a version bump without a NEWS
# entry: the notes are written by the change that made them necessary, not
# reconstructed afterwards. A non-empty `# Unreleased` becomes this version's
# section; an empty one with no existing section stops the render.
releases_md_path <- here("RELEASES.md")
releases_md <- readLines(releases_md_path, warn = FALSE)
if (!staging) {
releases_md <- promote_unreleased(releases_md, release_version, Sys.Date())
writeLines(releases_md, releases_md_path)
} else {
# a staging run renders its notes by the same rule (so an empty `# Unreleased` still stops it),
# but promotes in memory only — RELEASES.md on disk is never touched by a staging run
releases_md <- promote_unreleased(releases_md, release_version, Sys.Date())
message("staging run: RELEASES.md left untouched (# Unreleased promoted in memory only)")
}
notes_section <- release_notes_section(releases_md, release_version)
stopifnot("RELEASES.md has no section for this release" = !is.null(notes_section))
cat(glue("RELEASES.md: # {notes_section$heading}\n"))
```
```{r}
#| label: freeze_release
dir_frozen <- here(glue("{sidecar_root}/{release_version}"))
# BULK PARQUET STAGES OUTSIDE THE REPO, sidecars stay in it — the same split
# 146da92 applied to ingest outputs, which stopped one step short of the release.
# Each frozen release drops 1.6-11 GB of parquet, and `data/.gitignore` had to
# name `releases/*/parquet/` explicitly to keep 41 GB of it out of history: a
# guard against something that should not have been in the working tree at all.
# The sidecars beside it (catalog.json, metadata.json, relationships*.json/csv,
# RELEASE_NOTES.md, erd.mmd, test_results.json) are small, diffable and the
# provenance record, so they remain tracked under data/releases/{version}/.
#
# Nothing downstream needs editing: the GCS upload and the orphan prune both
# discover tables with list.files()/list.dirs() over this directory rather than
# from a hand-maintained list.
dir_frozen_pq <- cc_stage_path(stage_root, release_version, "parquet",
create = TRUE)
dir.create(dir_frozen, recursive = TRUE, showWarnings = FALSE)
message(glue("Creating frozen release: {release_version}"))
message(glue(" sidecars -> {dir_frozen}"))
message(glue(" parquet -> {dir_frozen_pq}"))
# `cruise` is enriched (derived) in this notebook — export locally
# all other tables are GCS-copied from ingest/ (including provenance columns)
derived_tables <- "cruise"
if (nrow(new_tables) > 0) {
# tables with _new additions need local merge + export
merged_base <- unique(sub("_new$", "", new_tables$table))
derived_tables <- c(derived_tables, merged_base)
}
# export only derived/merged tables to local parquet
# Exports happen once freeze_stats is final (end of this chunk), through ONE
# writer — export_release_parquet() — so every released table is byte-stable
# and content-hashed. The per-table export calls that used to sit here, and the
# GCS copies of lookup/spatial_attribute from the ingest bucket, are gone: a
# release is written from the assembled database and nowhere else.
# build freeze stats from registry (auto-discovered)
# exclude _new delta tables (intermediate) and supplemental
freeze_stats <- reg_canon |>
filter(!supplemental, !grepl("_new$", table)) |>
select(table, rows, partitioned, gcs_prefix)
# merged tables (from _new additions) → mark as derived (gcs_prefix = NA → upload from local)
if (nrow(new_tables) > 0) {
merged_base <- unique(sub("_new$", "", new_tables$table))
freeze_stats <- freeze_stats |>
mutate(gcs_prefix = if_else(table %in% merged_base, NA_character_, gcs_prefix))
}
# add derived tables (cruise, etc.)
for (dt in derived_tables) {
if (!dt %in% freeze_stats$table) {
n <- dbGetQuery(con_wdl, glue("SELECT COUNT(*) AS n FROM {dt}"))$n
freeze_stats <- freeze_stats |>
bind_rows(tibble(
table = dt, rows = n, partitioned = FALSE, gcs_prefix = NA_character_))
}
}
# `measurement_type` (rebuilt from the authoritative CSV), `cruise` (enriched
# with per-dataset event counts), and the unified taxon refs (taxon/dataset_taxon/
# taxon_group, rebuilt from the per-dataset taxon tables) are derived + exported
# locally here, so upload the local copy rather than GCS-copying the stale ingest
# parquet (esp. the old ichthyo `taxon` hierarchy the new `taxon` replaces).
#
# `dataset` belongs here for exactly the same reason, and did not have it. Every
# ingest writes its own full 16-row `dataset` shard, so `build_release_table_registry()`
# hands this table a `gcs_prefix` and the release server-side-copied one arbitrary
# ingest's copy — discarding the one built at [dataset_table] / [dataset_coverage]
# above. Three things were wrong in every release through v2026.08.11 as a result,
# none of them visible to any check here, because everything downstream of
# `con_wdl` saw the correct table:
# * no `dataset_key` column at all, so the namespaced key that `obs.dataset_key`
# is supposed to join to did not exist in the published reference table;
# * 16 rows, not 15 — `cdfw_dungeness-crab` is `in_release: false` because
# permission to publish is unsettled, and `read_ingest_yaml(in_release_only =
# TRUE)` correctly drops it, but the ingest shards predate that filter;
# * `coverage_temporal`/`coverage_spatial` as *asserted* in the YAML rather than
# the values `observed_coverage()` measures from the assembled core, which is
# the whole point of the `dataset_coverage` chunk.
freeze_stats <- freeze_stats |>
mutate(gcs_prefix = if_else(
table %in% c("dataset", "measurement_type", "cruise", "taxon", "dataset_taxon",
"taxon_group", crs_local_tables),
NA_character_, gcs_prefix))
# add consolidated core + supplemental tables (gcs_prefix = NA → upload from local)
#
# `taxon` MUST be here. It is exported by `core_single` above and uploaded by the
# filesystem sweep below, but the catalog is built from `freeze_stats` — so when it
# was missing from this list it shipped as a published-but-uncatalogued table:
# present in `metadata.json` (17 tables) and absent from `catalog.json` (16), which
# is the file `cc_get_db()` reads. No ingest manifest declares it either, because
# the unified taxon reference is rebuilt centrally here. The `catalog_covers_export`
# assertion after the catalog is built now makes this drift impossible to ship.
core_spec <- dplyr::bind_rows(
tibble(
table = c("sample", "obs", "obs_attribute", "sample_measurement",
"taxon", "dataset_taxon", "taxon_group"),
partitioned = c(FALSE, TRUE, FALSE, FALSE,
FALSE, FALSE, FALSE)),
# browser-shaped objects built in [browser_objects]; obs_env and climatology are one object per variable
tibble(table = c("sample_spatial", "sample_root", "obs_bio", "obs_env", "climatology"),
partitioned = c(FALSE, FALSE, FALSE, TRUE, TRUE)),
tibble(table = supp_tbls, partitioned = TRUE))
core_spec <- core_spec |>
filter(table %in% dbListTables(con_wdl), !table %in% freeze_stats$table) |>
mutate(
rows = vapply(table, function(t)
as.numeric(dbGetQuery(con_wdl, glue("SELECT COUNT(*) AS n FROM {t}"))$n), numeric(1)),
gcs_prefix = NA_character_)
freeze_stats <- bind_rows(freeze_stats, core_spec)
# keep only the DEFAULT core + shared refs + unified taxa (`core_keep`) plus the
# SUPPLEMENTAL tables (`obs_ctd_full`); flag the latter so the catalog/metadata
# mark them and cc_get_db()/db-schema hide them by default.
freeze_stats <- freeze_stats |>
filter(table %in% c(core_keep, supplemental_keep)) |>
mutate(supplemental = table %in% supplemental_keep)
# refresh row counts from the ASSEMBLED DB — reg_canon carries stale ingest-manifest
# counts for the rebuilt/derived tables (measurement_type from CSV, the unified
# `taxon` that replaces the old hierarchy, cruise, …), so the catalog/metadata
# would otherwise show wrong `rows`. The frozen parquet is already correct.
.present <- intersect(freeze_stats$table, DBI::dbListTables(con_wdl))
.rows_now <- setNames(
vapply(.present, function(t)
as.numeric(dbGetQuery(con_wdl, glue("SELECT COUNT(*) AS n FROM \"{t}\""))$n), numeric(1)),
.present)
freeze_stats <- freeze_stats |>
mutate(rows = ifelse(table %in% .present, .rows_now[table], rows))
freeze_stats |>
datatable(caption = glue("Frozen release {release_version} — {nrow(freeze_stats)} core+ref tables"))
# --- deterministic export of every released table ---------------------------
# ORDER BY a unique key + a single writer thread + pinned writer options: the
# same rows always give the same bytes (measured 2026-08-25: a total ORDER BY
# alone was NOT byte-stable at default threads). Provenance columns are
# stripped: `_ingested_at` changed on every ingest and would have made every
# table look changed to the content hash below.
sort_keys <- release_sort_keys()
missing_key <- setdiff(freeze_stats$table, names(sort_keys))
if (length(missing_key))
stop(glue("no sort key registered for released table(s): ",
"{paste(missing_key, collapse = ', ')} — add to calcofi4db::release_sort_keys()"))
missing_tbl <- setdiff(freeze_stats$table, dbListTables(con_wdl))
if (length(missing_tbl))
stop(glue("released table(s) not in the assembled database: {paste(missing_tbl, collapse = ', ')}"))
if (dir.exists(dir_frozen_pq)) unlink(dir_frozen_pq, recursive = TRUE)
dir.create(dir_frozen_pq, recursive = TRUE)
export_files <- list()
for (tbl in freeze_stats$table) {
sk <- sort_keys[[tbl]]
part <- if (isTRUE(freeze_stats$partitioned[freeze_stats$table == tbl])) sk$partition_by else NULL
t0 <- Sys.time()
if (!is.null(part)) {
export_files[[tbl]] <- export_release_parquet(
con_wdl, tbl, file.path(dir_frozen_pq, tbl), sk$order_by, partition_by = part)
# obs is ALSO published as one file: browser DuckDB-WASM (db-query) and other
# plain-HTTPS consumers cannot expand a Hive glob over GCS
if (tbl == "obs")
export_files[["obs.parquet"]] <- export_release_parquet(
con_wdl, tbl, file.path(dir_frozen_pq, "obs.parquet"), sk$order_by)
} else {
export_files[[tbl]] <- export_release_parquet(
con_wdl, tbl, file.path(dir_frozen_pq, paste0(tbl, ".parquet")), sk$order_by)
}
message(glue(" exported {tbl}: {nrow(export_files[[tbl]])} file(s), ",
"{round(sum(export_files[[tbl]]$bytes) / 1e6, 1)} MB, ",
"{round(as.numeric(difftime(Sys.time(), t0, units = 'secs')))} s"))
}
# --- the previous release's catalog is the reuse ledger ---------------------
prev_version <- calcofi4db::read_promoted_release(bucket = "calcofi-db", prefix = release_prefix)
prev_catalog <- NULL
if (!is.na(prev_version) && prev_version != release_version) {
prev_catalog <- tryCatch(
jsonlite::fromJSON(glue(
"https://storage.googleapis.com/calcofi-db/{release_prefix}/{prev_version}/catalog.json"),
simplifyVector = TRUE),
error = function(e) { message(glue("no previous catalog readable ({conditionMessage(e)})")); NULL })
}
message(glue("previous release for reuse: {if (is.null(prev_catalog)) 'none' else prev_version}"))
# one row per parquet object: bytes, sha256, content_hash, since
release_objs <- dplyr::bind_rows(lapply(freeze_stats$table, function(tbl) {
sk <- sort_keys[[tbl]]
part <- if (isTRUE(freeze_stats$partitioned[freeze_stats$table == tbl])) sk$partition_by else NULL
o <- release_objects(con_wdl, tbl, dir_frozen_pq, export_files[[tbl]], release_version,
partition_by = part, prev_catalog = prev_catalog)
if (tbl == "obs")
o <- dplyr::bind_rows(o, release_objects(con_wdl, "obs", dir_frozen_pq,
export_files[["obs.parquet"]], release_version,
prev_catalog = prev_catalog))
o
}))
release_objs |>
dplyr::count(table, since, wt = bytes, name = "bytes") |>
dplyr::mutate(mb = round(bytes / 1e6, 1)) |>
datatable(caption = "Release objects by table and the release that first carried their content (`since`)")
```
## Release Notes
```{r}
#| label: release_notes
#| results: asis
# narrative from RELEASES.md + a generated appendix. catalog.json is not written
# until upload_frozen, so the appendix is built from freeze_stats here and the
# notes are re-published by test_release.qmd with the real catalog (total size)
# and the validation result once the version is promoted.
pkg_versions <- c(
calcofi4db = as.character(packageVersion("calcofi4db")),
calcofi4r = as.character(packageVersion("calcofi4r")))
py_toml <- here("../calcofi4py/pyproject.toml")
if (file.exists(py_toml)) {
v_py <- sub('^version = "(.*)"$', "\\1",
grep('^version = ', readLines(py_toml, warn = FALSE), value = TRUE)[1])
if (!is.na(v_py) && nzchar(v_py)) pkg_versions <- c(pkg_versions, calcofi4py = v_py)
}
catalog_pre <- add_release_citation(list(
version = release_version,
release_date = as.character(Sys.Date()),
total_size = NULL,
tables = freeze_stats |>
mutate(supplemental = dplyr::coalesce(supplemental, FALSE)) |>
select(name = table, rows, partitioned, supplemental)))
# `releases_md` is the promoted text from [release_notes_narrative] — on disk for a real release,
# in memory only for a staging run (whose date has no section on disk). The datasets block
# feeds the appendix's "How to cite" (each dataset's citation_main · license) before
# metadata.json exists; test_release.qmd re-publishes from the real sidecars.
release_notes <- render_release_notes(
release_version, releases_md,
catalog = catalog_pre,
metadata = list(datasets = d_dataset_cov),
test_results = NULL,
pkg_versions = pkg_versions,
promoted = FALSE)
writeLines(release_notes, file.path(dir_frozen, "RELEASE_NOTES.md"))
cat(release_notes)
```
## Upload Frozen Release to GCS
```{r}
#| label: upload_frozen
gcs_bucket <- "calcofi-db"
gcs_release <- glue("{release_prefix}/{release_version}")
gcloud <- find_gcloud()
# 1. GCS server-side copy for ingest tables (auto-discovered from registry)
# 1. plan: upload only what changed; reuse the rest ----
# compat layout: an object whose content_hash matches the previous release is
# GCS server-side COPIED from there (no upload bytes). canonical layout: it
# already exists under tables/ and the catalog just points at it; a compat
# copy under this release's parquet/ keeps every legacy URL working.
plan <- freeze_plan(release_objs, prev_catalog, release_version,
layout = release_layout, release_prefix = release_prefix)
if (release_layout == "canonical" && tables_prefix != CC_TABLES_PREFIX) {
plan$path <- sub(paste0("^", CC_TABLES_PREFIX, "/"), paste0(tables_prefix, "/"), plan$path)
# `exists` was decided against the PREVIOUS catalog's paths, and a staging catalog can name
# objects the staging store never held (the 2026-09-04 staging run catalogued production
# ducklake/tables/ paths before the tables-prefix guard existed; measured 2026-09-05: the
# measurement_type object was `exists` and the compat copy 404'd). An object is reused only if
# it is actually in the store this run writes to; otherwise it is uploaded like a new one.
have <- system2(gcloud, c("storage", "ls", "-r", glue("gs://{gcs_bucket}/{tables_prefix}/**")),
stdout = TRUE, stderr = FALSE)
have <- sub(glue("^gs://{gcs_bucket}/"), "", have[grepl("[.]parquet$", have)])
missing <- plan$action == "exists" & !plan$path %in% have
if (any(missing)) {
message(glue(" {sum(missing)} object(s) marked `exists` are absent from {tables_prefix}/ — uploading them"))
plan$action[missing] <- "upload"
plan$bytes[missing] <- plan$bytes_local[missing]
plan$sha256[missing] <- plan$sha256_local[missing]
}
}
plan |>
dplyr::count(action, wt = bytes, name = "bytes") |>
dplyr::mutate(n = as.vector(table(plan$action)[action]), mb = round(bytes / 1e6, 1)) |>
datatable(caption = glue("Freeze plan for {release_version} ({release_layout} layout)"))
upload_release_objects(plan, dir_frozen_pq, gcs_bucket, compat = TRUE)
# 2. PRUNE objects under this release's parquet/ that are not in the plan.
# Uploads add and overwrite but never delete, so re-cutting a version after a
# table is renamed leaves the old object behind — outside the catalog, untracked,
# and still readable by anyone who knows the old path (it happened: _spatial and
# spatial both sat under v2026.08.02). Prune against the plan, not a hand list.
gcs_have <- system2(gcloud, c("storage", "ls", "-r",
glue("gs://{gcs_bucket}/{gcs_release}/parquet/**")), stdout = TRUE, stderr = FALSE)
gcs_have <- sub(glue("^gs://{gcs_bucket}/"), "", gcs_have[grepl("[.]parquet$", gcs_have)])
orphans <- setdiff(gcs_have, plan$compat_path)
if (length(orphans)) {
for (o in orphans) system2(gcloud, c("storage", "rm", glue("gs://{gcs_bucket}/{o}")),
stdout = TRUE, stderr = TRUE)
message(glue(" pruned {length(orphans)} orphan object(s) no longer in this release"))
} else message(" no orphaned objects to prune")
# 3. build and upload catalog.json (needed by cc_get_db()) ----
tables_df <- freeze_stats |>
mutate(supplemental = dplyr::coalesce(supplemental, FALSE)) |>
select(name = table, rows, partitioned, supplemental)
# GUARD: the catalog must describe everything that was actually published, and
# nothing else — `taxon` once shipped uncatalogued and was invisible to
# cc_get_db(). The plan is derived from the exported files, the catalog from
# freeze_stats; they must name the same tables.
if (!setequal(unique(plan$table), tables_df$name))
stop(glue("catalog/plan table mismatch: only in plan = ",
"{paste(setdiff(unique(plan$table), tables_df$name), collapse = ', ')}; ",
"only in catalog = {paste(setdiff(tables_df$name, unique(plan$table)), collapse = ', ')}"))
catalog <- build_release_catalog(release_version, tables_df, plan, layout = release_layout)
# the release cites itself: `citation` (decided wording, three partners) + the
# Zenodo `concept_doi`; the version `doi` arrives after WS-F's tag + GitHub release
# mints it, and publish_release_notes() writes it into this catalog then
# (objects untouched). Until then the citation resolves to db-schema for the version.
catalog <- add_release_citation(catalog)
total_bytes <- catalog$total_size
catalog_path <- file.path(dir_frozen, "catalog.json")
jsonlite::write_json(catalog, catalog_path, auto_unbox = TRUE, pretty = TRUE, digits = NA)
put_gcs_file(catalog_path,
glue("gs://{gcs_bucket}/{gcs_release}/catalog.json"))
# the coverage cube behind the explorer's first paint (built in [browser_objects])
for (f in c(coverage_path, coverage_stations_path, grid_geojson_path, spatial_geojson_path, spatial_layers_path))
put_gcs_file(f, glue("gs://{gcs_bucket}/{gcs_release}/{basename(f)}"))
# upload RELEASE_NOTES.md
notes_path <- file.path(dir_frozen, "RELEASE_NOTES.md")
if (file.exists(notes_path))
put_gcs_file(notes_path,
glue("gs://{gcs_bucket}/{gcs_release}/RELEASE_NOTES.md"))
# the running changelog every version's notes are cut from
put_gcs_file(releases_md_path,
glue("gs://{gcs_bucket}/{release_prefix}/RELEASES.md"))
# upload relationships.json
rels_json <- file.path(dir_frozen, "relationships.json")
if (file.exists(rels_json))
put_gcs_file(rels_json,
glue("gs://{gcs_bucket}/{gcs_release}/relationships.json"))
# build and upload metadata.json (table/column descriptions + units).
# auto-discover every ingest's metadata.json (same as rels_paths) so newly added
# datasets' tables/columns are merged in — not just a hardcoded set.
meta_paths <- in_release_dirs(Sys.glob(here("data/parquet/*/metadata.json")))
meta_json_path <- file.path(dir_frozen, "metadata.json")
if (length(meta_paths) > 0) {
# data-derived one-to-many measurement_type -> dataset(s) map. build the
# table -> owning-dataset(s) lookup from the ingest YAML's tables_owned (the
# `measurement_type` lookup table is excluded — its measurement_type column is
# the vocabulary, not measured rows), then scan each measurement table in the
# assembled DB for the types it actually reports.
table_datasets <- list()
for (key in names(ingest_yaml)) {
cc <- ingest_yaml[[key]]
for (e in cc$tables_owned %||% list())
table_datasets[[e$table]] <- union(table_datasets[[e$table]], key)
for (ad in cc$additional_datasets %||% list()) {
k2 <- paste0(ad$provider, "_", ad$dataset)
for (e in ad$tables_owned %||% list())
table_datasets[[e$table]] <- union(table_datasets[[e$table]], k2)
}
}
table_datasets[["measurement_type"]] <- NULL
meas_ds <- derive_measurement_type_datasets(con_wdl, table_datasets)
cat(glue("derived dataset membership for {length(meas_ds)} measurement types\n"))
merge_metadata_json(
paths = meta_paths,
output_path = meta_json_path,
release_version = release_version,
release_tables_csv = here("metadata/release_tables.csv"),
release_columns_csv = here("metadata/release_columns.csv"),
measurement_type_csv = here("metadata/measurement_type.csv"),
ingest_yaml = ingest_yaml,
table_rows = setNames(freeze_stats$rows, freeze_stats$table),
measurement_datasets = meas_ds)
# enrich columns with data_type from the working DuckDB (so the schema
# site can render types without spinning up DuckDB-WASM)
schema_cols <- DBI::dbGetQuery(con_wdl, "
SELECT table_name, column_name, data_type
FROM information_schema.columns
WHERE table_schema = 'main'")
meta <- jsonlite::read_json(meta_json_path, simplifyVector = FALSE)
n_typed <- 0L
for (i in seq_len(nrow(schema_cols))) {
key <- paste0(schema_cols$table_name[i], ".", schema_cols$column_name[i])
if (key %in% names(meta$columns)) {
meta$columns[[key]]$data_type <- schema_cols$data_type[i]
} else {
meta$columns[[key]] <- list(data_type = schema_cols$data_type[i])
}
n_typed <- n_typed + 1L
}
# --- contributions + observed temporal for the derived core tables ---------
# obs/sample/obs_attribute/sample_measurement are materialized here (no per-ingest
# metadata.json), so compute each dataset's row share directly from the data.
# Unlike the measurement_type vocabulary lookup, every core row belongs to
# exactly one dataset (dataset_key) -> clean, non-over-attributed stacks.
ds_workflow <- vapply(ingest_yaml,
function(cc) cc$workflow_url %||% NA_character_, character(1))
for (tbl in intersect(c("obs", "sample", "obs_attribute", "sample_measurement"),
dbListTables(con_wdl))) {
by_ds <- DBI::dbGetQuery(con_wdl, glue(
"SELECT dataset_key, COUNT(*) AS n FROM {tbl} GROUP BY 1 ORDER BY n DESC"))
total <- sum(by_ds$n)
meta$contributions[[tbl]] <- list(
total_rows = total,
over_attributed = FALSE,
by_dataset = lapply(seq_len(nrow(by_ds)), function(i) list(
provider_dataset = by_ds$dataset_key[i],
rows = by_ds$n[i],
pct = round(by_ds$n[i] / total * 100, 1),
workflow = unname(ds_workflow[by_ds$dataset_key[i]]))))
}
# observed temporal AND spatial extent per dataset, measured in the
# dataset_coverage chunk above from the real data (obs + sample). These drive
# the calcofi.io/workflows cards and the schema site; `coverage_bbox` carries
# the same box numerically so a map consumer does not have to parse the label.
for (i in seq_len(nrow(d_cov))) {
k <- d_cov$dataset_key[i]
if (!k %in% names(meta$datasets)) next
if (!is.na(d_cov$coverage_temporal_observed[i]))
meta$datasets[[k]]$coverage_temporal_observed <-
d_cov$coverage_temporal_observed[i]
if (!is.na(d_cov$coverage_spatial_observed[i])) {
meta$datasets[[k]]$coverage_spatial_observed <-
d_cov$coverage_spatial_observed[i]
meta$datasets[[k]]$coverage_bbox <- list(
lat_min = d_cov$lat_min[i], lat_max = d_cov$lat_max[i],
lon_min = d_cov$lon_min[i], lon_max = d_cov$lon_max[i])
}
}
# retire per-dataset tables from metadata.json too (core_keep + supplemental
# from the retire step) so the schema site's Tables/ERD/contributions show only
# the core+refs+taxa (obs_ctd_full is kept but flagged supplemental below).
keep_meta <- c(core_keep, supplemental_keep)
meta$tables <- meta$tables[names(meta$tables) %in% keep_meta]
meta$contributions <- meta$contributions[names(meta$contributions) %in% keep_meta]
meta$columns <- meta$columns[
vapply(names(meta$columns), function(k) sub("[.].*$", "", k) %in% keep_meta, logical(1))]
# mark the supplemental tables in metadata.json so db-schema can badge + hide
for (t in intersect(supplemental_keep, names(meta$tables)))
meta$tables[[t]]$supplemental <- TRUE
jsonlite::write_json(meta, meta_json_path,
auto_unbox = TRUE, pretty = TRUE, null = "null")
message(glue("metadata.json enriched with data_type for {n_typed} columns"))
put_gcs_file(meta_json_path,
glue("gs://{gcs_bucket}/{gcs_release}/metadata.json"))
# erd.mmd sidecar: Mermaid ER diagram driven by relationships.json
rels_for_erd <- file.path(dir_frozen, "relationships.json")
if (file.exists(rels_for_erd)) {
erd <- cc_erd(
con = con_wdl,
rels_path = rels_for_erd,
colors = color_map,
view_type = "all")
erd_path <- file.path(dir_frozen, "erd.mmd")
writeLines(unclass(erd), erd_path)
# validate the Mermaid parses before publishing — a malformed erd.mmd
# (e.g. erDiagram styling unsupported by an older mermaid) would break the
# schema site, which renders it client-side with mermaid. Validate with
# mermaid-cli (mmdc); KEEP schema/_config.yml `mermaid_version` >= this
# mmdc's bundled mermaid so the site accepts what passes here.
mmdc <- Sys.which("mmdc")
if (nzchar(mmdc)) {
erd_svg_check <- tempfile(fileext = ".svg")
erd_val <- suppressWarnings(system2(
mmdc, c("-i", erd_path, "-o", erd_svg_check),
stdout = TRUE, stderr = TRUE))
if (!identical(attr(erd_val, "status"), NULL) &&
!identical(attr(erd_val, "status"), 0L)) {
stop(glue(
"erd.mmd failed Mermaid validation; not uploading.\n",
"{paste(erd_val, collapse = '\n')}"))
}
message("erd.mmd passed Mermaid validation (mmdc)")
} else {
warning("mmdc not found; skipping Mermaid validation of erd.mmd")
}
put_gcs_file(erd_path,
glue("gs://{gcs_bucket}/{gcs_release}/erd.mmd"))
message(glue("erd.mmd uploaded ({length(attr(erd, 'tables'))} tables)"))
} else {
warning("relationships.json missing; skipping erd.mmd sidecar")
}
} else {
warning("No per-ingest metadata.json files found; skipping release metadata.json")
}
# 3b. datasets.json — the dataset catalog record (plan 2026-09-05 § D-1; calcofi4db >= 4.1.0) ----
# One record per dataset_key joining metadata.json, coverage.json (built in [browser_objects]) and
# the catalog.json just built with the registries — metadata/{category,provider,license,
# dataset_status,distribution,portal}.csv and the descriptive sidecars
# metadata/{provider}/{dataset}/dataset_meta.yml — and with what the live services answer NOW
# (erddap.calcofi.io's allDatasets, the netCDF manifests, the earlier releases' metadata.json for
# since_version). It is built HERE rather than beside coverage.json because it points at the
# catalog's content-addressed objects, which exist only after the freeze plan. check_dataset_catalog()
# is a release gate like assert_dataset_citation(): a record without a name, a registered category and
# provider, a description, a bbox or a download fails; a missing citation is exempt only while a
# provider question covers it; every listed URL must answer (one-byte ranged GET, behind
# CALCOFI_SKIP_LINK_CHECK, like the index's link check). Nothing on calcofi.io/datasets is authored
# by hand: the landing page fetches this file (Phase 1). holdings.csv is regenerated from the sidecars.
if (file.exists(meta_json_path)) {
cat_registries <- read_catalog_registries(here("metadata"))
net_ok <- !nzchar(Sys.getenv("CALCOFI_SKIP_LINK_CHECK"))
erddap_now <- if (net_ok) fetch_erddap_datasets() else NULL
netcdf_now <- if (net_ok) fetch_netcdf_manifests(c(names(ingest_yaml), paste0(names(ingest_yaml), "_full"))) else list()
# since_version is a fact about the PROMOTED history, so it is always derived from the
# production prefix — a staging run's own versions.json holds three throwaway versions and
# would date every dataset to the oldest of them (measured 2026-09-05: ichthyo "since v2026.08.25")
prod_https <- glue("https://storage.googleapis.com/{gcs_bucket}/ducklake/releases")
prod_versions <- tryCatch(jsonlite::fromJSON(glue("{prod_https}/versions.json"), simplifyVector = FALSE),
error = function(e) NULL)
since <- if (!is.null(prod_versions) && net_ok) dataset_since_versions(prod_versions, base = prod_https) else character()
# source_accessed is measured in [dataset_coverage]; keyed by dataset_key here
src_acc <- setNames(format(d_dataset_cov$source_accessed), paste0(d_dataset_cov$provider, "_", d_dataset_cov$dataset))
bathy <- tryCatch(jsonlite::fromJSON("https://storage.googleapis.com/calcofi-db/bathymetry/gebco_2025.json",
simplifyVector = FALSE), error = function(e) NULL)
datasets_rec <- build_dataset_catalog(
meta_json_path, coverage_path, catalog, cat_registries, version = release_version,
erddap = erddap_now, netcdf = netcdf_now, since = since, source_accessed = src_acc,
spatial_layers = spatial_layers_path, bathymetry = bathy, release_prefix = release_prefix)
# a dataset in no earlier release first appears in this one
for (i in seq_along(datasets_rec$datasets))
if (is.null(datasets_rec$datasets[[i]]$since_version)) datasets_rec$datasets[[i]]$since_version <- release_version
d_catalog_chk <- check_dataset_catalog(datasets_rec, cat_registries, network = net_ok)
assert_dataset_catalog(d_catalog_chk)
d_catalog_chk |>
filter(finding != "ok") |>
datatable(caption = "dataset catalog check: findings (error rows are exempt only while a provider question is open/proposed)")
datasets_paths <- write_dataset_catalog(datasets_rec, dir_frozen)
validate_dataset_catalog(datasets_paths[1])
for (f in datasets_paths)
put_gcs_file(f, glue("gs://{gcs_bucket}/{gcs_release}/{substring(f, nchar(dir_frozen) + 2)}"))
write_holdings_csv(cat_registries, here("metadata/holdings.csv"))
cat(glue("datasets.json: {length(datasets_rec$datasets)} datasets · {length(datasets_rec$holdings)} holdings · ",
"{length(datasets_rec$reference)} reference rows · {round(file.size(datasets_paths[1]) / 1e3)} KB\n"))
# 3c. eml/{dataset_key}.xml — one EML 2.2 document per dataset (plan § D-8, Decision 13;
# calcofi4db >= 4.2.0) ----
# The record is the only source: title, abstract, creators, contact, licence, keywords, the
# measured coverage, the taxa coverage.json resolved, gear.csv's dwc_samplingProtocol sentences
# and a dataTable per released table with the attributeList metadata.json's columns{} describes.
# publish_to-obis, publish_to-edi, ERDDAP's globals and the dataset page's JSON-LD all read THIS
# document, so none of them is typed twice. check_eml() is a release gate like
# check_dataset_catalog(): eml_validate() against EML 2.2's local XSDs (no network) plus the
# required-element checklist — an element the record cannot supply is a finding, exempt only
# while an open/proposed questions.csv row on the dataset names the field, never a made-up value.
eml_docs <- build_eml_catalog(
datasets_rec, sidecars = cat_registries, meta = meta_json_path, coverage = coverage_path,
gear = read_gear_registry(here("metadata/gear.csv")))
eml_paths <- write_eml_files(eml_docs, dir_frozen)
d_eml_chk <- check_eml_catalog(eml_docs, eml_paths, datasets_rec)
assert_eml(d_eml_chk)
d_eml_chk |>
filter(finding != "ok") |>
datatable(caption = "EML check: findings (error rows are exempt only while a provider question is open/proposed)")
for (f in eml_paths)
put_gcs_file(f, glue("gs://{gcs_bucket}/{gcs_release}/eml/{basename(f)}"))
cat(glue("eml/: {length(eml_paths)} documents · all valid EML 2.2 · ",
"{round(sum(file.size(eml_paths)) / 1e3)} KB\n"))
} else {
stop("datasets.json needs metadata.json; no per-ingest metadata.json was merged")
}
# 3d. stac — the static STAC catalog (plan 2026-09-05 § D-5.3; calcofi4db >= 4.3.0) ----
# STAC is a MACHINE surface, never the UI the catalog depends on: the pages read datasets.json
# directly. build_stac() is a pure function of the record just written, metadata.json (the column
# descriptions) and spatial_layers.json — a root catalog, one collection per PUBLIC dataset with an
# item per release (parquet objects with table:columns / file:size / file:checksum, the CF netCDF,
# the ERDDAP pages, the ISO 19115 record), and one collection per spatial layer with its PMTiles.
# It lives beside the releases rather than inside one (gs://calcofi-db/stac/) because a STAC root is
# one stable URL that gains an item per release; stac-browser at calcofi.io/stac/ reads it.
# A staging run writes stac-staging/ — the browser's root is the production catalog and a rehearsal
# must not rewrite it.
stac_prefix <- if (grepl("staging", release_prefix)) "stac-staging" else "stac"
stac_base <- glue("https://storage.googleapis.com/{gcs_bucket}/{stac_prefix}")
dir_stac <- file.path(dir_frozen, "stac")
unlink(dir_stac, recursive = TRUE)
stac_paths <- build_stac(
record = datasets_rec,
catalog = catalog,
spatial_layers = spatial_layers_path,
dir = dir_stac,
base_url = stac_base,
metadata = meta_json_path)
stac_chk <- check_stac(dir_stac, network = net_ok)
assert_stac(stac_chk)
stac_chk |>
filter(finding != "ok") |>
datatable(caption = "STAC check: findings (structural always; stac-validator when installed)")
# the whole prefix is regenerated every release, so an unmatched object is stale by construction
stac_res <- system2(gcloud, c(
"storage", "rsync", "-r", "--delete-unmatched-destination-objects",
dir_stac, glue("gs://{gcs_bucket}/{stac_prefix}")), stdout = TRUE, stderr = TRUE)
if (!identical(attr(stac_res, "status") %||% 0L, 0L))
stop(glue("STAC upload failed: {paste(stac_res, collapse='; ')}"))
system2(gcloud, c("storage", "objects", "update", "--cache-control=no-cache",
glue("gs://{gcs_bucket}/{stac_prefix}/catalog.json")), stdout = TRUE, stderr = TRUE)
cat(glue("STAC: {length(stac_paths)} documents -> gs://{gcs_bucket}/{stac_prefix}/ ",
"({sum(stac_chk$finding == 'ok')} ok, {sum(stac_chk$level == 'warn')} warn)\n"))
# 4. update versions.json (latest.txt promotion is deferred to test_release.qmd)
# rebuilt from every catalog.json under the prefix by calcofi4db::build_versions_json(),
# which also stamps the archive-policy fields: `consolidated` from
# metadata/release_policy.yml and `retired` from a thinned version's retired.json
# (scripts/thin_releases.R). This release's own record comes from the catalog
# just built, so a stale cached copy on GCS cannot win.
release_policy <- yaml::read_yaml(here("metadata/release_policy.yml"))
all_versions <- build_versions_json(
gcs_bucket, release_prefix, consolidated = release_policy$consolidated,
current = list(
version = release_version,
release_date = catalog$release_date,
tables = length(catalog$tables),
total_rows = as.numeric(catalog$total_rows %||% 0),
size_mb = round((catalog$total_size %||% 0) / 1024 / 1024, 1)))
# the register lives on the bucket AND as a tracked sidecar (data/releases/versions.json,
# or the staging root): the tracked copy sat at v2026.02 for seven months because only
# the tempfile was uploaded
versions_local <- here(sidecar_root, "versions.json")
jsonlite::write_json(list(versions = all_versions), versions_local,
auto_unbox = TRUE, pretty = TRUE)
put_gcs_file(versions_local,
glue("gs://{gcs_bucket}/{release_prefix}/versions.json"))
# the schema site (calcofi.io/db-schema) fetches these JSON/mmd sidecars at runtime
# and they are OVERWRITTEN in place when a release is re-run (e.g. to fix a bug).
# GCS defaults to `cache-control: public, max-age=3600`, so a corrected re-upload
# stays masked by browser/CDN caches for up to an hour. Tag the mutable sidecars
# `no-cache` (revalidate-always; cheap 304s) so a re-render is visible immediately.
sidecar_urls <- c(
glue("gs://{gcs_bucket}/{release_prefix}/versions.json"),
glue("gs://{gcs_bucket}/{gcs_release}/catalog.json"),
glue("gs://{gcs_bucket}/{gcs_release}/metadata.json"),
glue("gs://{gcs_bucket}/{gcs_release}/datasets.json"),
# the EML folder: every publisher (OBIS, EDI, ERDDAP) and the dataset page read these
glue("gs://{gcs_bucket}/{gcs_release}/eml/*.xml"),
glue("gs://{gcs_bucket}/{gcs_release}/coverage.json"),
glue("gs://{gcs_bucket}/{gcs_release}/coverage_stations.json"),
glue("gs://{gcs_bucket}/{gcs_release}/grid.geojson"),
glue("gs://{gcs_bucket}/{gcs_release}/spatial.geojson"),
glue("gs://{gcs_bucket}/{gcs_release}/spatial_layers.json"),
glue("gs://{gcs_bucket}/{gcs_release}/relationships.json"),
glue("gs://{gcs_bucket}/{gcs_release}/erd.mmd"),
glue("gs://{gcs_bucket}/{gcs_release}/RELEASE_NOTES.md"),
glue("gs://{gcs_bucket}/{release_prefix}/RELEASES.md"))
cc_res <- system2(gcloud,
c("storage", "objects", "update", "--cache-control=no-cache", sidecar_urls),
stdout = TRUE, stderr = TRUE)
if (!identical(attr(cc_res, "status") %||% 0L, 0L))
warning(glue("could not set no-cache on sidecars: {paste(cc_res, collapse='; ')}"))
message("runtime sidecars tagged cache-control: no-cache")
# NOTE: latest.txt is NOT updated here. Promotion is gated on the
# query-test pass in test_release.qmd, which writes latest.txt only
# when every pre-baked query in CalCOFI/db-query/_queries succeeds.
message(glue(
"Release {release_version} uploaded ({length(all_versions)} versions tracked); ",
"latest.txt promotion deferred to test_release.qmd"))
```
## Cleanup
```{r}
#| label: cleanup
# --- the target's declared output -------------------------------------------
# `output:` used to be the `data/releases` DIRECTORY. Two things were wrong with
# that, and together they made this target permanently outdated:
#
# 1. It is not exclusively ours. `test_release.qmd` — a target DOWNSTREAM of
# this one — writes `data/releases/{version}/test_results.json` into it. So
# the directory hash targets recorded when this target finished no longer
# matched a few minutes later, and `release_database` reported itself
# outdated the instant the pipeline completed. Verified on v2026.08.08: our
# files landed 16:46-17:06, test_results.json at 17:08:47. Every subsequent
# `tar_make()` on this target or anything downstream of it re-ran a ~40 min
# freeze + multi-GB re-upload of an already-promoted release.
# 2. It accumulates every release ever cut, so the hash also covered ~136
# historical sidecars. Pruning an old local release would invalidate the
# current one, and the hash grows without bound.
#
# So the output is now a small file that ONLY this chunk writes. It is
# deterministic — version plus a digest of the frozen catalog, no wall clock —
# so a re-run over unchanged inputs reproduces it byte-for-byte and leaves
# `test_release` skipped rather than cascading.
stamp_path <- here(glue("{sidecar_root}/_release_stamp.json"))
jsonlite::write_json(
list(
release_version = release_version,
n_tables = nrow(freeze_stats),
n_rows = sum(freeze_stats$rows, na.rm = TRUE),
# the catalog is written by freeze_release() and summarises every table in
# the release, so its digest changes exactly when the release content does.
# tools::md5sum() is base R — no dependency to add to the shelf() call above.
catalog_md5 = unname(tools::md5sum(
file.path(dir_frozen, "catalog.json")))),
stamp_path, auto_unbox = TRUE, pretty = TRUE)
cat(glue("release stamp: {stamp_path}\n"))
# close in-memory DuckDB connection
close_duckdb(con_wdl)
message("Assembly DuckDB connection closed")
# summary
message(glue("\n=== Summary ==="))
message(glue("Frozen release: {release_version} created at {dir_frozen}"))
message(glue("Tables: {nrow(freeze_stats)}"))
message(glue("Total rows: {format(sum(freeze_stats$rows, na.rm = TRUE), big.mark = ',')}"))
```
::: {.callout-caution collapse="true"}
## Session Info
```{r session_info}
devtools::session_info()
```
:::
