---
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"
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")
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")
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)
# get PK column for dedup
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
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")
# --- 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.
n_common <- apply_taxon_common(con_wdl, here("metadata/taxon_common.csv"))
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)))
# `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.
```{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)
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()."))
```
## 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"))
```
## 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)
# 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
# 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 (YYYY-MM-NODC)
if ("cruise" %in% tbls) {
bad_ck <- dbGetQuery(
con_wdl,
"SELECT cruise_key FROM cruise
WHERE cruise_key IS NOT NULL
AND NOT regexp_matches(cruise_key, '^\\d{4}-\\d{2}-.+$')"
)
if (nrow(bad_ck) > 0) {
warning(glue("cruise_key format violations: {nrow(bad_ck)} rows"))
} else {
message("cruise_key: all match YYYY-MM-NODC format")
}
}
# Warning message: cruise_key format violations: 1 rows
# cruise_key: 2019-07-
# 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
# 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")
dbExecute(con_wdl, "DROP VIEW IF EXISTS cruise")
dbExecute(con_wdl, "DROP TABLE IF EXISTS 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}"))
}
)
```
## Show Combined Schema
```{r}
# dir_frozen used later; define early so ERD can reference it
dir_frozen <- here(glue("data/releases/{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("data/releases/{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.
```{r}
#| label: freeze_release
dir_frozen <- here(glue("data/releases/{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("releases", 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
export_parquet(con_wdl, "cruise",
file.path(dir_frozen_pq, "cruise.parquet"), compression = "zstd")
message("Exported cruise.parquet")
# export merged tables (e.g., ship with _new additions)
for (tbl in setdiff(derived_tables, "cruise")) {
export_parquet(con_wdl, tbl,
file.path(dir_frozen_pq, paste0(tbl, ".parquet")), compression = "zstd")
message(glue("Exported {tbl}.parquet (merged)"))
}
# --- consolidated core tables (derived here in Phase 2) ---------------------
# single-file exports (sample carries geom as GeoParquet; the long tables +
# measurement_type are plain); obs / obs_ctd_full are Hive-partitioned + sorted
# for compression + predicate pushdown (see design "Parquet partitioning").
core_single <- intersect(
unique(c("sample", "obs_attribute", "sample_measurement", "measurement_type",
"taxon", "dataset_taxon", "taxon_group", # unified taxa refs, rebuilt here
# `dataset` is rebuilt here too — from the ingest YAML, filtered to
# in_release, keyed, and with coverage measured rather than asserted.
# It must be exported locally for the same reason as the taxa refs:
# the gcs_prefix = NA below only helps if there is a local file to
# upload instead. See the note there for what shipped without it.
"dataset",
# every CRS-normalized table (grid, spatial, …): the normalization only
# reaches the release if the LOCAL copy is what gets uploaded
crs_local_tables)),
dbListTables(con_wdl))
for (tbl in core_single)
export_parquet(con_wdl, tbl,
file.path(dir_frozen_pq, paste0(tbl, ".parquet")), compression = "zstd")
# The trailing `datetime` is a tiebreak, not decoration: without it the first
# three columns leave large tie groups whose rows land in arbitrary order,
# scattering latitude/longitude/datetime and defeating delta encoding. Measured
# on one partition: 27.55 -> 20.20 MB (CTD), 23.22 -> 16.95 MB (mets).
# The supplemental ingests sort by this same key, so the shard the release reads
# is already clustered the way it wants — keep the two in step.
core_sort <- "grid_key NULLS LAST, depth_min_m NULLS LAST, measurement_type, datetime"
if ("obs" %in% dbListTables(con_wdl)) {
dbExecute(con_wdl, glue(
"COPY (SELECT * FROM obs ORDER BY dataset_key, {core_sort})
TO '{file.path(dir_frozen_pq, 'obs')}'
(FORMAT PARQUET, COMPRESSION 'zstd', PARTITION_BY (dataset_key), OVERWRITE_OR_IGNORE)"))
# also a single-file obs.parquet: browser DuckDB-WASM (db-query/match.js) and
# plain-HTTPS consumers can't glob the Hive-partitioned obs/ dir over GCS.
export_parquet(con_wdl, glue("SELECT * FROM obs ORDER BY dataset_key, {core_sort}"),
file.path(dir_frozen_pq, "obs.parquet"), compression = "zstd")
}
# Supplementals are exported and uploaded from here, like every other released
# table. A GCS server-side copy from the ingest bucket was measured and rejected:
# it removes ~1.5 GB from the upload, but only stays correct while the ingest
# shard, the assembled core and the published release agree — and nothing tracks
# that they do. A stale source would publish stale data behind a green release.
# Pushing from one place keeps the release self-consistent; see the `core_sort`
# note above for the change that actually shrank this leg.
for (st in intersect(supp_tbls, dbListTables(con_wdl)))
dbExecute(con_wdl, glue(
"COPY (SELECT * FROM {st} ORDER BY cruise_key, {core_sort})
TO '{file.path(dir_frozen_pq, st)}'
(FORMAT PARQUET, COMPRESSION 'zstd', PARTITION_BY (cruise_key), OVERWRITE_OR_IGNORE)"))
message(glue("Exported core tables: {paste(core_single, collapse=', ')}, obs (partitioned)"))
# 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)),
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"))
```
## Release Notes
```{r}
#| label: release_notes
#| results: asis
# build release notes
tables_list <- paste0(
"- ",
freeze_stats$table,
" (",
format(freeze_stats$rows, big.mark = ","),
" rows)"
)
release_notes <- paste0(
"# CalCOFI Database Release ",
release_version,
"\n\n",
"**Release Date**: ",
Sys.Date(),
"\n\n",
"## Tables Included\n\n",
paste(tables_list, collapse = "\n"),
"\n\n",
"## Total\n\n",
"- **Tables**: ",
nrow(freeze_stats),
"\n",
"- **Total Rows**: ",
format(sum(freeze_stats$rows, na.rm = TRUE), big.mark = ","),
"\n\n",
"## Data Sources\n\n",
"- `ingest_swfsc_ichthyo.qmd` - Ichthyo tables (cruise, ship, site, tow, net, species, ichthyo, grid, segment, lookup, taxon, taxa_rank)\n",
"- `ingest_calcofi_bottle.qmd` - Bottle/cast tables (casts, bottle, bottle_measurement, cast_condition, measurement_type)\n",
"- `ingest_calcofi_ctd-cast.qmd` - CTD tables (ctd_cast, ctd_thin, ctd_summary, measurement_type; full ctd_measurement available as supplemental)\n",
"- `ingest_calcofi_dic.qmd` - DIC/alkalinity tables (dic_sample, dic_measurement, dic_summary, dataset)\n\n",
"## Cross-Dataset Integration\n\n",
"- **Ship matching**: Reconciled ship codes between bottle casts and swfsc ship reference\n",
"- **Cruise bridge**: Derived cruise_key (YYYY-MM-NODC) for bottle casts via ship matching + datetime\n",
"- **Taxonomy**: Standardized species with WoRMS AphiaID, ITIS TSN, GBIF backbone key\n",
"- **Taxon hierarchy**: Built taxon + taxa_rank tables from WoRMS/ITIS classification\n\n",
"## Access\n\n",
"Parquet files can be queried directly from GCS:\n\n",
"```r\n",
"library(duckdb)\n",
"con <- dbConnect(duckdb())\n",
"dbExecute(con, 'INSTALL httpfs; LOAD httpfs;')\n",
"dbGetQuery(con, \"\n",
" SELECT * FROM read_parquet(\n",
" 'https://storage.googleapis.com/calcofi-db/ducklake/releases/",
release_version,
"/parquet/ichthyo.parquet')\n",
" LIMIT 10\")\n",
"```\n\n",
"Or use calcofi4r:\n\n",
"```r\n",
"library(calcofi4r)\n",
"con <- cc_get_db(version = '",
release_version,
"')\n",
"```\n"
)
writeLines(release_notes, file.path(dir_frozen, "RELEASE_NOTES.md"))
message(glue(
"Release notes written to {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("ducklake/releases/{release_version}")
gcloud <- find_gcloud()
# 1. GCS server-side copy for ingest tables (auto-discovered from registry)
copy_rows <- freeze_stats |> filter(!is.na(gcs_prefix))
message(glue("Copying {nrow(copy_rows)} tables from ingest/ to releases/ on GCS..."))
for (i in seq_len(nrow(copy_rows))) {
tbl <- copy_rows$table[i]
pfx <- copy_rows$gcs_prefix[i]
part <- copy_rows$partitioned[i]
if (part) {
# partitioned: copy directory
src <- glue("gs://{gcs_bucket}/{pfx}/{tbl}")
dst <- glue("gs://{gcs_bucket}/{gcs_release}/parquet/{tbl}")
res <- system2(gcloud, c("storage", "cp", "-r",
paste0(src, "/*"), dst), stdout = TRUE, stderr = TRUE)
} else {
src <- glue("gs://{gcs_bucket}/{pfx}/{tbl}.parquet")
dst <- glue("gs://{gcs_bucket}/{gcs_release}/parquet/{tbl}.parquet")
res <- system2(gcloud, c("storage", "cp",
src, dst), stdout = TRUE, stderr = TRUE)
}
rc <- attr(res, "status") %||% 0L
if (rc != 0) {
stop(glue("GCS copy failed for {tbl}: {src} -> {dst}\n",
" exit code {rc}: {paste(res, collapse = '\n')}"))
}
message(glue(" {tbl}: copied from {pfx}"))
}
# 2. upload derived tables from local parquet — single files (cruise,
# measurement_type, sample, obs_attribute, sample_measurement) AND partitioned dirs
# (obs, obs_ctd_full: Hive-partitioned, uploaded recursively).
derived_local <- list.files(dir_frozen_pq, pattern = "[.]parquet$",
full.names = TRUE)
for (pq in derived_local) {
tbl <- tools::file_path_sans_ext(basename(pq))
gcs_path <- glue("gs://{gcs_bucket}/{gcs_release}/parquet/{tbl}.parquet")
put_gcs_file(pq, gcs_path)
message(glue(" {tbl}: uploaded (derived)"))
}
# partitioned derived dirs (obs, obs_ctd_full)
derived_dirs <- list.dirs(dir_frozen_pq, recursive = FALSE)
for (d in derived_dirs) {
tbl <- basename(d)
dst <- glue("gs://{gcs_bucket}/{gcs_release}/parquet/{tbl}")
# rsync, not cp: a partitioned upload at a few MiB/s takes a long time, and
# `cp` restarts it from zero every time. A single transient failure near the
# end therefore cost the whole upload (observed on obs_ctd_full: 88 of 96
# partitions landed, then the run died and a retry would have re-sent all 88).
# rsync skips what already matches, so a retry costs only what is missing.
res <- system2(gcloud, c("storage", "rsync", "-r", d, dst),
stdout = TRUE, stderr = TRUE)
if ((attr(res, "status") %||% 0L) != 0)
stop(glue("GCS rsync failed for derived partitioned {tbl}: ",
"{paste(utils::tail(res, 20), collapse='\n')}"))
message(glue(" {tbl}: synced (derived, partitioned)"))
}
# 2b. PRUNE objects that are no longer part of this release.
# 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. That happened: renaming
# _spatial -> spatial left BOTH _spatial.parquet and spatial.parquet under
# v2026.08.02, and the stale 26 MB copy had to be removed by hand.
#
# Prune against what was actually exported/copied, not against a hand-maintained
# list, since that is the drift this is guarding against in the first place.
gcs_expected <- c(
tools::file_path_sans_ext(basename(list.files(dir_frozen_pq, pattern = "[.]parquet$"))),
basename(list.dirs(dir_frozen_pq, recursive = FALSE)),
copy_rows$table)
gcs_have <- system2(gcloud, c("storage", "ls",
glue("gs://{gcs_bucket}/{gcs_release}/parquet/")), stdout = TRUE, stderr = FALSE)
gcs_have_tbl <- sub("[.]parquet$", "", basename(sub("/$", "", gcs_have)))
orphans <- setdiff(gcs_have_tbl[nzchar(gcs_have_tbl)], gcs_expected)
if (length(orphans)) {
for (o in orphans) {
uri <- grep(glue("/{o}(\\.parquet)?/?$"), gcs_have, value = TRUE)[1]
if (is.na(uri)) next
system2(gcloud, c("storage", "rm", "-r", uri), stdout = TRUE, stderr = TRUE)
message(glue(" pruned orphan no longer in this release: {o}"))
}
} 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)
# sum bytes of the uploaded parquet tree on GCS
du_out <- system2(
gcloud,
c("storage", "du", "--summarize",
glue("gs://{gcs_bucket}/{gcs_release}/parquet/")),
stdout = TRUE, stderr = TRUE)
total_bytes <- suppressWarnings(
as.numeric(sub("\\s.*$", "", trimws(du_out[1]))))
if (is.na(total_bytes)) {
warning(glue("Could not parse gcloud storage du output: {paste(du_out, collapse='; ')}"))
total_bytes <- 0
}
catalog <- list(
version = release_version,
release_date = as.character(Sys.Date()),
total_rows = sum(tables_df$rows, na.rm = TRUE),
total_size = total_bytes,
tables = tables_df)
# GUARD: the catalog must describe everything that was actually published.
# Uploads are driven by the FILESYSTEM (every .parquet file and partitioned dir in
# dir_frozen_pq), while the catalog is assembled from hand-maintained lists — so
# the two can drift silently, and did: `taxon` was exported and uploaded but never
# catalogued, leaving it invisible to cc_get_db(). Compare the two directly rather
# than trusting the lists to stay in step.
exported_tbls <- c(
tools::file_path_sans_ext(basename(
list.files(dir_frozen_pq, pattern = "[.]parquet$"))),
basename(list.dirs(dir_frozen_pq, recursive = FALSE)))
uncatalogued <- setdiff(unique(exported_tbls), tables_df$name)
if (length(uncatalogued))
stop(glue(
"catalog.json would omit {length(uncatalogued)} published table(s): ",
"{paste(uncatalogued, collapse = ', ')}. They are uploaded from ",
"dir_frozen_pq but absent from freeze_stats — add them to `core_spec` ",
"(or `derived_tables`), or they ship invisible to cc_get_db()."))
catalog_path <- file.path(dir_frozen, "catalog.json")
jsonlite::write_json(catalog, catalog_path, auto_unbox = TRUE, pretty = TRUE)
put_gcs_file(catalog_path,
glue("gs://{gcs_bucket}/{gcs_release}/catalog.json"))
# 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"))
# 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")
}
# 4. update versions.json (latest.txt promotion is deferred to test_release.qmd)
# discover all releases from GCS and rebuild versions.json
gcs_ls <- system2(gcloud, c("storage", "ls",
glue("gs://{gcs_bucket}/ducklake/releases/")),
stdout = TRUE, stderr = TRUE)
release_vers <- regmatches(gcs_ls,
regexpr("v[0-9]{4}[.][0-9]{2}[.]*[0-9]*", gcs_ls))
https_base <- glue("https://storage.googleapis.com/{gcs_bucket}/ducklake/releases")
all_versions <- purrr::compact(lapply(release_vers, function(v) {
tryCatch({
cat_data <- jsonlite::fromJSON(glue("{https_base}/{v}/catalog.json"))
list(
version = cat_data$version,
release_date = cat_data$release_date %||% NA_character_,
tables = if (is.data.frame(cat_data$tables)) nrow(cat_data$tables)
else length(cat_data$tables),
total_rows = as.integer(cat_data$total_rows %||% 0),
size_mb = round((cat_data$total_size %||% 0) / 1024 / 1024, 1))
}, error = function(e) NULL)
}))
all_versions <- all_versions[order(
sapply(all_versions, `[[`, "version"), decreasing = TRUE)]
versions_local <- tempfile(fileext = ".json")
jsonlite::write_json(list(versions = all_versions), versions_local,
auto_unbox = TRUE, pretty = TRUE)
put_gcs_file(versions_local,
glue("gs://{gcs_bucket}/ducklake/releases/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}/ducklake/releases/versions.json"),
glue("gs://{gcs_bucket}/{gcs_release}/catalog.json"),
glue("gs://{gcs_bucket}/{gcs_release}/metadata.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"))
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("data/releases/_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()
```
:::
