---
title: "Ingest CCE-LTER ZooDB Holoplankton"
calcofi:
target_name: ingest_cce_lter_zoodb
workflow_type: ingest
dependency:
- ingest_swfsc_ichthyo
output: data/parquet/cce-lter_zoodb/manifest.json
provider: cce-lter
dataset: zoodb
workflow_url: https://calcofi.io/workflows/ingest_cce-lter_zoodb.html
questions_file: metadata/cce-lter/zoodb/questions.csv
dataset_meta:
dataset_name: ZooDB Holoplankton Community
# display trio, read by the release `dataset` table and the
# consumer apps (calcofi4db >= 3.15.0) — see NEWS for why these
# left the apps' own hardcoded maps
dataset_name_short: ZooDB (Holoplankton Community)
category: Zooplankton
color: "#38d9a9"
description: >
Holoplankton abundance and carbon biomass by 33 higher taxa (copepods
to family, plus amphipods, chaetognaths, salps, pyrosomes, pteropods,
ostracods, foraminifera, radiolaria and more) from CalCOFI net tows,
1951-2021, enumerated by microscopy (Ohman Lab / SIO Pelagic Invertebrate
Collection). One row per (sample, taxon, measurement_type); unpooled tows
are station-resolved, pooled samples are per-cruise regional composites.
Raw (uncorrected) abundance per 1000 m3 and per m2, plus biomass mg C/m2.
citation_main: ""
citation_others: "Plankton sample analysis supported by NSF grants to M.D. Ohman and the CCE-LTER site, and by the SIO Pelagic Invertebrate Collection."
link_calcofi_org: "https://calcofi.org/data/marine-ecosystem-data/zooplankton/"
link_data_source: "https://oceaninformatics.ucsd.edu/zoodb/"
link_others: []
license: ""
pi_names: Mark D. Ohman; Bertha Lavaniegos; Linsey Sala
# publishes the consolidated core: the sample/measurement/taxon source shape is
# wrangled in the notebook and projected into sample / obs + the taxa refs.
tables_owned:
- {table: sample, shared: true, note: "core event dimension (tow grain)"}
- {table: obs, shared: true, note: "core occurrence headline (bio, zooplankton abundance)"}
- {table: taxon, shared: true, note: "shared taxa reference"}
- {table: dataset_taxon, shared: true, note: "zoodb taxon_id -> taxon_key crosswalk"}
- {table: measurement_type, shared: true, note: "shared registry across datasets"}
erd:
color: "#d9c2f0"
editor_options:
chunk_output_type: console
---
## Overview
**Source**: [ZooDB](https://oceaninformatics.ucsd.edu/zoodb/) — the SIO Ocean
Informatics zooplankton database (Mark Ohman Lab, SIO Pelagic Invertebrate
Collection). Per-taxon CSVs are scripted from the portal's `save.php` endpoint
(public access) by `libs/download_zoodb.R` and consolidated to
`zoodb_holoplankton.csv` (see the Acquire step below + `by_taxon/_PROVENANCE.md`).
- **Provider**: `cce-lter` (curated by CCE-LTER / Ohman Lab; sibling of
`cce-lter_euphausiids`, which the ZooDB methods note as a *separate* database)
- **Grain**: one row per **(sample × taxon × measurement_type)**. A measurement
row means the taxon was *analyzed* for that sample; `measurement_value = 0`
means analyzed-but-absent. Coverage varies by taxon/cruise — a **missing**
(sample, taxon) is *not analyzed*, so we never impute zeros.
- **Samples** (506): 351 `unpooled` net tows (station-resolved) + 155 `pooled`
per-cruise regional composites (Lavaniegos & Ohman 2007; no per-station
position; a cruise/region may hold >1 composite, e.g. shallow vs deep strata).
- **Taxa** (33): higher taxa — all genera/species and phases/stages combined —
standardized to WoRMS (`taxon_worms.csv`).
- **Measurements**: `zooplankton_abundance` (count/1000 m³),
`zooplankton_abundance_areal` (count/m²), `zooplankton_biomass_carbon`
(mg C/m²). Raw (uncorrected for net type).
This ingest **proceeds with documented provisional decisions** (cruise/ship
resolution, timezone, raw vs corrected, pooled positioning); each is tracked in
the *Questions for Data Providers* section.
```{mermaid}
graph LR
A[zoodb_holoplankton.csv<br/>10,316 sample x taxon rows] --> B[zoodb_sample<br/>503 samples + keys]
A --> C[zoodb_measurement<br/>long: abundance x2 + biomass]
A --> D[zoodb_taxon<br/>33 higher taxa + WoRMS]
C -.sample_id.-> B
C -.taxon_id.-> D
B -.ship/cruise/grid.-> E[(shared refs)]
```
## Setup
```{r}
#| label: setup
#| message: false
devtools::load_all(here::here("../calcofi4db"))
devtools::load_all(here::here("../calcofi4r"))
librarian::shelf(
CalCOFI/calcofi4db, CalCOFI/calcofi4r,
DBI, dplyr, DT, fs, glue, here, janitor, jsonlite, knitr, lubridate, purrr,
readr, sf, stringr, tibble, tidyr, units, quiet = T)
options(readr.show_col_types = F)
options(DT.options = list(scrollX = TRUE))
source(here("libs/ingest.R")) # overwrite, dir_data
cc <- read_calcofi_meta(here("ingest_cce-lter_zoodb.qmd"))
provider <- cc$provider
dataset <- cc$dataset
tables_owned <- cc$tables_owned
dir_label <- glue("{provider}_{dataset}")
dir_parquet <- here(glue("data/parquet/{dir_label}"))
dir_stage <- cc_stage_path("parquet", dir_label, create = TRUE)
db_path <- here(glue("data/wrangling/{dir_label}.duckdb"))
if (overwrite) {
if (file_exists(db_path)) file_delete(db_path)
if (file_exists(paste0(db_path, ".wal"))) file_delete(paste0(db_path, ".wal"))
if (dir_exists(paste0(db_path, ".tmp"))) dir_delete(paste0(db_path, ".tmp"))
}
dir_create(dirname(db_path))
con <- get_duckdb_con(db_path)
load_duckdb_extension(con, "spatial")
meas_type_csv <- here("metadata/measurement_type.csv")
d_meas_type <- read_measurement_type(meas_type_csv)
```
## Acquire Source Data
ZooDB is an interactive portal, not a file download. `libs/download_zoodb.R`
scripts its public login + `save.php` CSV export — the 33 higher taxa × 2
abundance units (`pooled=combined`, raw/uncorrected, all years/months/times/
regions) — and merges the two unit files per taxon (on the source `RowNumber`)
into `zoodb_holoplankton.csv`. It runs once then caches; set `overwrite_all =
TRUE` in `libs/ingest.R` to re-scrape. Exact query parameters are recorded in
`by_taxon/_PROVENANCE.md`.
```{r}
#| label: acquire
source(here("libs/download_zoodb.R"))
zoo_dir <- path_expand(glue("{dir_data}/cce-lter/ZooDB"))
zoo_csv <- download_zoodb(zoo_dir, overwrite = overwrite_all)
```
## Read Source Data
The consolidated long table (one row per taxon × sample) carries `N/A` for the
per-station fields on pooled rows, read as `NA`.
```{r}
#| label: read-source
stopifnot("zoodb_holoplankton.csv not found" = file_exists(zoo_csv))
# archive source (consolidated + per-taxon exports + portal method PDFs) to GCS
sync_to_gcs(
local_dir = zoo_dir,
gcs_prefix = glue("archive/{provider}/{dataset}"),
bucket = "calcofi-files-public",
exclude = c(".DS_Store", "*.tmp", "*.gdoc"))
d_raw <- read_csv(zoo_csv, na = c("", "NA", "N/A"))
cat(glue("Read {format(nrow(d_raw), big.mark=',')} rows, {ncol(d_raw)} columns ",
"from {basename(zoo_csv)}"), "\n")
```
## Clean and Type-Cast
Canonical field names follow `metadata/field_dictionary.csv`. A `sample_key`
uniquely identifies each sample: `U|cruise|towbegin|line|station` (unpooled) or
`P|cruise|date|region` (pooled).
```{r}
#| label: clean
d <- d_raw |>
transmute(
taxon_slug = taxon,
source = str_to_lower(source),
cruise_orig = cruise,
ship_name = ship,
date = as.Date(date),
line = suppressWarnings(as.numeric(line)),
station = suppressWarnings(as.numeric(station)),
region = region,
datetime_start_utc = as_datetime(tow_begin), # Q01: tz assumed local/ship
datetime_end_utc = as_datetime(tow_end),
latitude = suppressWarnings(as.numeric(latitude)),
longitude = suppressWarnings(as.numeric(longitude)),
max_depth_m = suppressWarnings(as.numeric(max_depth_m)),
min_depth_m = suppressWarnings(as.numeric(min_depth_m)),
net_type = net_type,
n_tows_pooled = suppressWarnings(as.integer(n_tows_pooled)),
abundance_per_1000m3 = suppressWarnings(as.numeric(abundance_per_1000m3)),
abundance_per_m2 = suppressWarnings(as.numeric(abundance_per_m2)),
biomass_mgC_per_m2 = suppressWarnings(as.numeric(biomass_mgC_per_m2))) |>
mutate(
site_key = if_else(
is.na(line) | is.na(station), NA_character_,
sprintf("%05.1f %05.1f", line, station)),
# pooled composites can co-occur per cruise/region (e.g. shallow vs deep
# strata) — n_tows_pooled disambiguates the otherwise-identical key
sample_key = if_else(
source == "unpooled",
paste("U", cruise_orig, datetime_start_utc, line, station, sep = "|"),
paste("P", cruise_orig, date, region, n_tows_pooled, sep = "|")))
cat(glue(
"Cleaned {format(nrow(d), big.mark=',')} (sample x taxon) rows; ",
"{n_distinct(d$sample_key)} samples ",
"({n_distinct(d$sample_key[d$source=='unpooled'])} unpooled, ",
"{n_distinct(d$sample_key[d$source=='pooled'])} pooled); ",
"{n_distinct(d$taxon_slug)} taxa"), "\n")
```
## Build Taxon Table (WoRMS)
`zoodb_taxon` is the 33 higher taxa with WoRMS AphiaIDs + classification,
resolved offline and cached in `metadata/cce-lter/zoodb/taxon_worms.csv`
(33/33 matched; 2 WoRMS-unaccepted concepts flagged, Q07).
```{r}
#| label: taxon
d_worms <- read_csv(here("metadata/cce-lter/zoodb/taxon_worms.csv"))
zoodb_taxon <- d_worms |>
arrange(taxon_zoodb) |>
transmute(
taxon_id = row_number(),
taxon_zoodb, taxon_slug,
aphia_id = as.integer(aphia_id),
scientific_name = scientific_name_accepted,
rank, taxon_status = status,
kingdom, phylum, class, order_taxon = order, family) # 'order' is a SQL reserved word
stopifnot("taxon_slug mismatch source vs WoRMS map" =
all(unique(d$taxon_slug) %in% zoodb_taxon$taxon_slug))
dbWriteTable(con, "zoodb_taxon", zoodb_taxon, overwrite = TRUE)
cat(glue("zoodb_taxon: {nrow(zoodb_taxon)} taxa ",
"({sum(!is.na(zoodb_taxon$aphia_id))} WoRMS-matched, ",
"{sum(zoodb_taxon$taxon_status=='unaccepted')} unaccepted [Q07])"), "\n")
```
## Build Sample Table + Resolve Keys
One row per distinct sample. Resolve `ship_key`/`ship_nodc` against the shared
ship registry, derive `cruise_key` as `YYYY-MM-{ship_nodc}` (validated against
`cruise`), and fall back to a unique year-month match for pooled samples (which
have no ship). Q01 (timezone) and Q02 (cruise mapping) are tracked.
```{r}
#| label: sample-keys
# distinct sample-level attributes (constant across taxa within a sample)
d_sample0 <- d |>
distinct(
sample_key, source, cruise_orig, ship_name, date,
datetime_start_utc, datetime_end_utc, line, station, site_key, region,
latitude, longitude, max_depth_m, min_depth_m, net_type, n_tows_pooled)
stopifnot("sample_key not unique over sample attributes" =
n_distinct(d_sample0$sample_key) == nrow(d_sample0))
# shared reference tables from the ichthyo ingest
load_prior_tables(
con, parquet_dir = cc_stage_path("parquet", "swfsc_ichthyo"),
tables = c("ship", "cruise", "grid"), geom_tables = c("grid"), as_view = TRUE)
d_ship <- dbGetQuery(con, "SELECT ship_key, ship_name, ship_nodc FROM ship") |>
mutate(ship_name_norm = str_squish(str_to_upper(ship_name))) |>
distinct(ship_name_norm, .keep_all = TRUE)
# year-month -> unique cruise_key (pooled samples + fallback)
cr_ym <- dbGetQuery(con, "SELECT cruise_key, date_ym FROM cruise") |>
mutate(ym = format(as.Date(date_ym), "%Y%m")) |>
group_by(ym) |>
summarize(cruise_key_ym = if (n() == 1) cruise_key[1] else NA_character_,
.groups = "drop")
valid_ck <- dbGetQuery(con, "SELECT DISTINCT cruise_key FROM cruise")$cruise_key
d_sample <- d_sample0 |>
mutate(ship_name_norm = ship_name |>
str_replace("^R/?V\\.?\\s+", "") |> str_to_upper() |> str_squish()) |>
left_join(d_ship |> select(ship_name_norm, ship_key, ship_nodc),
by = "ship_name_norm") |>
mutate(
ym = format(date, "%Y%m"),
cruise_key_ship = if_else(
is.na(ship_nodc), NA_character_,
as.character(glue("{format(date, '%Y-%m')}-{ship_nodc}"))),
cruise_key_ship = if_else(cruise_key_ship %in% valid_ck,
cruise_key_ship, NA_character_)) |>
left_join(cr_ym, by = "ym") |>
mutate(cruise_key = coalesce(cruise_key_ship, cruise_key_ym)) |>
arrange(source, date, cruise_orig, region, line, station, sample_key) |>
mutate(sample_id = row_number(), .before = 1)
# id <-> key map for the measurement join, then drop the key from the table
sample_map <- d_sample |> select(sample_id, sample_key)
zoodb_sample <- d_sample |>
select(sample_id, source, cruise_orig, cruise_key, ship_name, ship_key,
date, datetime_start_utc, datetime_end_utc, line, station, site_key,
region, latitude, longitude, max_depth_m, min_depth_m, net_type,
n_tows_pooled)
dbWriteTable(con, "zoodb_sample", zoodb_sample, overwrite = TRUE)
n_ship <- sum(!is.na(d_sample$ship_key))
n_cruise <- sum(!is.na(d_sample$cruise_key))
cat(glue(
"zoodb_sample: {nrow(zoodb_sample)} samples; ",
"ship_key {n_ship}/{nrow(d_sample)} ({round(100*n_ship/nrow(d_sample),1)}%), ",
"cruise_key {n_cruise}/{nrow(d_sample)} ({round(100*n_cruise/nrow(d_sample),1)}%)"), "\n")
```
## Add Spatial
Unpooled tows get a point geometry and a CalCOFI `grid_key`; pooled samples have
no position (NULL geometry / grid, Q05).
```{r}
#| label: spatial
add_point_geom(con, "zoodb_sample", lon_col = "longitude", lat_col = "latitude")
grid_stats <- assign_grid_key(con, "zoodb_sample")
grid_stats |> datatable(caption = "Grid assignment (unpooled tows)")
```
## Pivot Measurements to Long Format
The three measured quantities pivot into long form, joined to `sample_id`
(via `sample_key`) and `taxon_id` (via `taxon_slug`). Explicit zeros (analyzed-
but-absent) are retained; only NA / non-finite values are dropped.
```{r}
#| label: measurement
meas_recode <- c(
abundance_per_1000m3 = "zooplankton_abundance",
abundance_per_m2 = "zooplankton_abundance_areal",
biomass_mgC_per_m2 = "zooplankton_biomass_carbon")
zoodb_measurement <- d |>
left_join(sample_map, by = "sample_key") |>
left_join(zoodb_taxon |> select(taxon_id, taxon_slug), by = "taxon_slug") |>
select(sample_id, taxon_id,
abundance_per_1000m3, abundance_per_m2, biomass_mgC_per_m2) |>
pivot_longer(
cols = c(abundance_per_1000m3, abundance_per_m2, biomass_mgC_per_m2),
names_to = "src_col", values_to = "measurement_value") |>
filter(!is.na(measurement_value), is.finite(measurement_value)) |>
mutate(measurement_type = unname(meas_recode[src_col])) |>
arrange(sample_id, taxon_id, measurement_type) |>
transmute(measurement_id = row_number(),
sample_id, taxon_id, measurement_type, measurement_value)
dbWriteTable(con, "zoodb_measurement", zoodb_measurement, overwrite = TRUE)
cat(glue(
"zoodb_measurement: {format(nrow(zoodb_measurement), big.mark=',')} rows ",
"({format(sum(zoodb_measurement$measurement_value==0), big.mark=',')} explicit zeros = ",
"analyzed-but-absent, Q04)"), "\n")
```
## Add Measurement Types
```{r}
#| label: add-measurement-type
zoo_types <- tibble(
measurement_type = c("zooplankton_abundance", "zooplankton_abundance_areal",
"zooplankton_biomass_carbon"),
description = c(
"Zooplankton numerical abundance (volumetric density) per net tow, by higher taxon. Raw (uncorrected for net type).",
"Zooplankton areal abundance (depth-integrated density) per net tow, by higher taxon. Raw (uncorrected).",
"Zooplankton carbon biomass per net tow, by higher taxon. Raw (uncorrected for net type)."),
units = c("count/1000m3", "count/m2", "mgC/m2"),
is_canonical = c(TRUE, NA, TRUE),
`_source_column` = c("abundance_per_1000m3", "abundance_per_m2", "biomass_mgC_per_m2"),
`_source_table` = "zoodb_measurement",
`_source_datasets` = "cce-lter_zoodb",
`_qual_column` = NA_character_, `_prec_column` = NA_character_)
new_types <- zoo_types |> filter(!measurement_type %in% d_meas_type$measurement_type)
if (nrow(new_types) > 0) {
d_meas_type <- bind_rows(d_meas_type, new_types)
write_csv(d_meas_type, meas_type_csv, na = "")
cat(glue("Added measurement type(s): {paste(new_types$measurement_type, collapse=', ')}"), "\n")
} else cat("zooplankton measurement types already registered\n")
dbWriteTable(con, "measurement_type", d_meas_type, overwrite = TRUE)
```
## Load Dataset Metadata
```{r}
#| label: load-dataset-metadata
d_dataset <- ingest_yaml_to_dataset_df(read_ingest_yaml(here()))
dbWriteTable(con, "dataset", d_dataset, overwrite = TRUE)
cat(glue("dataset: {nrow(d_dataset)} dataset(s) registered"), "\n")
```
## Schema Documentation
```{r}
#| label: schema
zoodb_rels <- list(
primary_keys = list(
zoodb_sample = "sample_id",
zoodb_measurement = "measurement_id",
zoodb_taxon = "taxon_id",
measurement_type = "measurement_type"),
foreign_keys = list(
list(table = "zoodb_measurement", column = "sample_id",
ref_table = "zoodb_sample", ref_column = "sample_id"),
list(table = "zoodb_measurement", column = "taxon_id",
ref_table = "zoodb_taxon", ref_column = "taxon_id"),
list(table = "zoodb_measurement", column = "measurement_type",
ref_table = "measurement_type", ref_column = "measurement_type")))
# the SOURCE shape, documenting the wrangling above. The published tables are
# the consolidated core, so relationships.json is written alongside the parquet
# outputs below, from core_relationships().
cc_erd(
con,
tables = c("zoodb_sample", "zoodb_measurement", "zoodb_taxon",
"measurement_type", "dataset"),
rels = zoodb_rels,
colors = list(
lightblue = c("zoodb_sample", "zoodb_measurement"),
lightgreen = "zoodb_taxon",
lightyellow = "measurement_type",
white = "dataset"))
```
## Validate
```{r}
#| label: validate
results <- validate_for_release(con, checks = "all", strict = FALSE)
cat("Validation:", ifelse(results$passed, "PASSED", "FAILED"), "\n")
if (length(results$errors) > 0)
cat("Errors:\n", paste("-", results$errors, collapse = "\n"), "\n")
if (length(results$warnings) > 0)
cat("Warnings:\n", paste("-", results$warnings, collapse = "\n"), "\n")
# NULLs in cross-dataset keys (cruise_key, ship_key, site_key, grid_key) are the
# EXPECTED unmatched / not-applicable remainder (pooled samples have no station;
# cruise/ship mapping is partial — Q02). Not a hard failure.
cov <- dbGetQuery(con,
"SELECT
AVG(CASE WHEN ship_key IS NOT NULL THEN 1 ELSE 0 END) ship,
AVG(CASE WHEN cruise_key IS NOT NULL THEN 1 ELSE 0 END) cruise,
AVG(CASE WHEN grid_key IS NOT NULL THEN 1 ELSE 0 END) grid
FROM zoodb_sample")
cat(glue("Match coverage: ship_key {round(100*cov$ship,1)}%, ",
"cruise_key {round(100*cov$cruise,1)}%, grid_key {round(100*cov$grid,1)}%"), "\n")
# FK integrity: every measurement resolves to a sample and a taxon
orphan <- dbGetQuery(con,
"SELECT
SUM(CASE WHEN s.sample_id IS NULL THEN 1 ELSE 0 END) AS no_sample,
SUM(CASE WHEN t.taxon_id IS NULL THEN 1 ELSE 0 END) AS no_taxon
FROM zoodb_measurement m
LEFT JOIN zoodb_sample s USING (sample_id)
LEFT JOIN zoodb_taxon t USING (taxon_id)")
cat(glue("Orphan measurements: {orphan$no_sample} w/o sample, {orphan$no_taxon} w/o taxon"), "\n")
```
## Data Preview
```{r}
#| label: preview-sample
samp_cols <- dbGetQuery(con,
"SELECT column_name FROM information_schema.columns
WHERE table_name='zoodb_sample' AND data_type NOT LIKE 'GEOMETRY%'")$column_name
dbGetQuery(con, glue("SELECT {paste(samp_cols, collapse=', ')} FROM zoodb_sample LIMIT 100")) |>
datatable(caption = "zoodb_sample — first 100 rows", rownames = FALSE, filter = "top")
```
```{r}
#| label: preview-measurement
dbGetQuery(con,
"SELECT m.measurement_id, m.sample_id, x.taxon_zoodb, m.measurement_type, m.measurement_value
FROM zoodb_measurement m JOIN zoodb_taxon x USING (taxon_id)
ORDER BY m.measurement_id LIMIT 100") |>
datatable(caption = "zoodb_measurement — first 100 rows", rownames = FALSE)
```
```{r}
#| label: preview-taxon
dbGetQuery(con,
"SELECT t.taxon_id, t.taxon_zoodb, t.scientific_name, t.aphia_id, t.rank,
t.phylum, t.class, COUNT(DISTINCT m.sample_id) AS n_samples
FROM zoodb_taxon t LEFT JOIN zoodb_measurement m USING (taxon_id)
GROUP BY ALL ORDER BY t.taxon_id") |>
datatable(caption = "zoodb_taxon — 33 higher taxa (samples analyzed)", rownames = FALSE)
```
## Emit Core Tables
Project this dataset into the shared consolidated core model
(`design_env-bio-consolidation.md`, `RUNBOOK.md` §3b). These core tables **are**
this ingest's output: `release_database.qmd` concatenates the per-dataset shards
rather than re-deriving the core, so there is exactly one projection to keep
correct.
**This projection lives here, in the notebook that owns the dataset**, rather than
in a `switch(dataset_key, …)` arm inside `calcofi4db`. The reusable *shapes* stay in
the package — `sample_arm_self()` for a one-event-table dataset,
`compat_measurement_sql()` for the compat VIEW — so declaring a dataset's projection
is a handful of parameters, not copied SQL.
```{r}
#| label: emit_core
ds_key <- "cce-lter_zoodb"
# filter the crosswalk to THIS dataset -- passing the whole file leaks other
# datasets' taxa into this shard (44 rows instead of its own 33; the retired
# emit_core_tables() wrapper filtered internally) — each shard must carry only its own slice, since
# merge_taxon_shards() unions them at release time.
mt_taxon <- read_csv(here("metadata/measurement_taxon.csv"),
col_types = cols(worms_id = "i", itis_id = "i",
bin_value = "d", .default = "c")) |>
filter(dataset_key == ds_key)
tx_over <- read_csv(here("metadata/taxon_override.csv"), show_col_types = FALSE)
# taxa references: zoodb_taxon carries aphia_id + denormalized lineage, so the
# generic builders pick it up from the connection
# cross-reference: resolve each taxon against BOTH authorities (cached in
# metadata/taxon_xref.csv, so a re-run costs no API calls). This fills the
# `worms_id` COLUMN on itis:-keyed taxa without touching their key — a consumer
# joining on worms_id used to match ZERO rows for every seabird and marine
# mammal — backfills `itis_id` the other way, replaces an id its authority has
# deprecated so the key is always an accepted id, and fetches the real
# `taxonomic_status` with the date it was checked. Must precede the lineage
# fetch, which should ask about the accepted id, not the deprecated one.
ensure_taxon_xref(con, mt_taxon, tx_over,
cache_csv = here("metadata/taxon_xref.csv"))
# lineage: fetch each taxon's WoRMS/ITIS classification (cached in
# metadata/taxon_lineage.csv, so a re-run costs no API calls) and stage it as the
# `taxon` hierarchy build_taxon_reference() reads. Without it a crosswalk- or
# vocabulary-resolved taxon reaches the release with a key and a name and NOTHING
# else — no rank, no parent_taxon_key, no classification — so hierarchy rollups
# ("all Decapoda") silently match nothing and no error is raised anywhere.
ensure_taxon_lineage(con, mt_taxon, tx_over,
cache_csv = here("metadata/taxon_lineage.csv"))
n_taxon <- build_taxon_reference(con, mt_taxon, tx_over)
n_ds_taxon <- build_dataset_taxon(con, mt_taxon, tx_over)
# 1. sample — one row per zoodb_sample tow
append_sample(con, sample_arm_self(
ds_key, "zoodb_sample", "sample_id", "tow",
site_expr = "site_key", depth_min = "min_depth_m", depth_max = "max_depth_m"))
# 2. obs — one row per measurement; taxon via dataset_taxon
append_obs(con, glue("
SELECT 'bio', '{ds_key}', {ns_key(ds_key, 'tow', 'sp.sample_id')},
sp.grid_key, sp.cruise_key, sp.latitude, sp.longitude,
CAST(sp.datetime_start_utc AS TIMESTAMP), sp.min_depth_m, sp.max_depth_m,
dt.taxon_key, NULL::VARCHAR, m.measurement_type, m.measurement_value,
NULL::VARCHAR, NULL::DOUBLE
FROM zoodb_measurement m JOIN zoodb_sample sp USING (sample_id)
LEFT JOIN dataset_taxon dt ON dt.dataset_key = '{ds_key}'
AND dt.ds_taxa_code = CAST(m.taxon_id AS VARCHAR)"))
core <- list(
sample = dbGetQuery(con, "SELECT COUNT(*) FROM sample")[[1]],
obs = dbGetQuery(con, "SELECT COUNT(*) FROM obs")[[1]],
taxon = n_taxon,
dataset_taxon = n_ds_taxon)
cat(glue(
"core projection — sample={core$sample %||% 0} obs={core$obs %||% 0} ",
"taxon={core$taxon %||% 0} dataset_taxon={core$dataset_taxon %||% 0}\n"))
# every measurement must reach obs, and every obs must resolve a
# sampling event; taxon_key coverage is reported (the source vocabulary may
# include entries WoRMS does not resolve)
n_obs <- dbGetQuery(con, "SELECT COUNT(*) FROM obs")[[1]]
n_exp <- dbGetQuery(con, "
SELECT COUNT(*) FROM zoodb_measurement m JOIN zoodb_sample sp USING (sample_id)")[[1]]
stopifnot(
"obs must be one row per measurement" = n_obs == n_exp,
"every obs.sample_key must resolve in sample" =
dbGetQuery(con, "SELECT COUNT(*) FROM obs o LEFT JOIN sample s USING (sample_key)
WHERE s.sample_key IS NULL")[[1]] == 0)
n_tax <- dbGetQuery(con, "SELECT COUNT(*) FROM obs WHERE taxon_key IS NOT NULL")[[1]]
cat(glue("obs parity: {format(n_obs, big.mark = ',')} rows; ",
"{round(100 * n_tax / max(n_obs, 1), 1)}% taxon-resolved"), "\n")
# serve the retired per-dataset table names as VIEWs over the core, so
# in-notebook consumers and ad-hoc queries keep working against the old names
# (exact for every column the core models, lossy for the rest)
invisible(dbExecute(con, "DROP TABLE IF EXISTS zoodb_measurement_src"))
invisible(dbExecute(con, "ALTER TABLE zoodb_measurement RENAME TO zoodb_measurement_src"))
invisible(dbExecute(con, glue(
"CREATE OR REPLACE VIEW zoodb_measurement AS
{compat_measurement_sql(ds_key, 'tow', 'sample_id', 'measurement_id')}")))
cat(glue("compat view zoodb_measurement over obs: ",
"{dbGetQuery(con, 'SELECT COUNT(*) FROM zoodb_measurement')[[1]]} rows"), "\n")
```
## Write Parquet Outputs
```{r}
#| label: write-parquet
dir_create(dir_parquet)
mismatches <- list(
measurement_types = collect_measurement_type_mismatches(
con, here("metadata/measurement_type.csv")),
cruise_keys = collect_cruise_key_mismatches(con, "zoodb_sample"))
tbls_out <- core_output_tables(con, extra = c("measurement_type", "dataset"))
parquet_stats <- write_parquet_outputs(
con = con,
output_dir = dir_parquet,
tables = tbls_out,
sort_by = list(obs = c("grid_key", "measurement_type")),
strip_provenance = FALSE,
mismatches = mismatches)
build_relationships_json(
rels = core_relationships(tbls_out), output_dir = dir_parquet,
provider = provider, dataset = dataset)
parquet_stats |> mutate(file = basename(path)) |> select(-path) |>
datatable(caption = "Parquet export statistics")
```
## Write Metadata
```{r}
#| label: write-metadata
d_tbls_rd <- read_csv(here("metadata/cce-lter/zoodb/tbls_redefine.csv"))
d_flds_rd <- read_csv(here("metadata/cce-lter/zoodb/flds_redefine.csv"))
metadata_path <- build_metadata_json(
con = con,
d_tbls_rd = d_tbls_rd,
d_flds_rd = d_flds_rd,
metadata_derived_csv = c(here("metadata/core_dictionary.csv"),
here("metadata/cce-lter/zoodb/metadata_derived.csv")),
output_dir = dir_parquet,
tables = tbls_out,
set_comments = TRUE,
provider = provider,
dataset = dataset,
workflow_url = cc$workflow_url,
tables_owned = tables_owned)
```
## Upload to GCS
```{r}
#| label: upload-gcs
sync_to_gcs(
local_dir = dir_stage,
sidecar_dir = dir_parquet,
gcs_prefix = glue("ingest/{dir_label}"),
bucket = "calcofi-db")
```
## Questions for Data Providers
Follow-up questions for CCE-LTER (Mark Ohman, Bertha Lavaniegos, Linsey Sala),
**ranked by importance**. This ingest proceeded with the documented provisional
decisions noted above; `high` answers may change cruise/ship matching or the
timezone assumption on the next run. Tracked in
`metadata/cce-lter/zoodb/questions.csv`.
```{r}
#| label: provider-questions
# one validated read + render for every ingest: the vocabulary and the column
# order live in calcofi4db, not in 16 hand-written factor() calls
questions_datatable(
here(cc$questions_file),
caption = "Questions for the CCE-LTER ZooDB data providers (ranked)")
```
## Cleanup
```{r}
#| label: cleanup
close_duckdb(con)
cat(glue("Parquet outputs written to: {dir_parquet}"), "\n")
```
::: {.callout-caution collapse="true"}
## Session Info
```{r session_info}
devtools::session_info()
```
:::
