---
title: "Ingest NOAA CalCOFI Database"
calcofi:
target_name: ingest_swfsc_ichthyo
workflow_type: ingest
dependency: []
output: data/parquet/swfsc_ichthyo/manifest.json
provider: swfsc
dataset: ichthyo
workflow_url: https://calcofi.io/workflows/ingest_swfsc_ichthyo.html
questions_file: metadata/swfsc/ichthyo/questions.csv
dataset_meta:
dataset_name: SWFSC Ichthyoplankton
# 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: Ichthyoplankton
category: "Fish Eggs & Larvae"
color: "#d3115d"
# publishes the consolidated core plus the shared reference tables this ingest
# owns for the whole database. The site/tow/net/ichthyo/species source shape is
# wrangled in the notebook and projected into sample / obs / obs_attribute /
# sample_measurement + the taxa refs; it survives as compat VIEWs over the core.
tables_owned:
- {table: sample, shared: true, note: "core event dimension (site -> tow -> net adjacency list)"}
- {table: obs, shared: true, note: "core occurrence headline (bio, larval abundance)"}
- {table: obs_attribute, shared: true, note: "larval size + stage distributions"}
- {table: sample_measurement, shared: true, note: "net effort (volume filtered, haul factor, ...)"}
- {table: taxon, shared: true, note: "shared taxa reference (WoRMS/ITIS lineage)"}
- {table: dataset_taxon, shared: true, note: "species_id -> taxon_key crosswalk"}
- {table: cruise}
- {table: ship, shared: true, note: "ship registry; also modified by bottle/ctd ingests"}
- {table: lookup}
- {table: grid, shared: true, note: "spatial grid referenced across datasets"}
- {table: fish, note: "fish grown past the larval stage, per net and species code (SWFSC Fish)"}
erd:
color: "#cdebc6"
---
## Overview
**Goal**: Generate the database from source files with workflow scripts
to make updating easier and provenance fully transparent. This allows us
to:
- Rename tables and column names, control data types and use Unicode
encoding for a consistent database ingestion strategy, per Database
naming conventions in [Database – CalCOFI.io
Docs](https://calcofi.io/docs/db.html).
- Differentiate between raw and derived or updated tables and columns.
For instance, the taxonomy for any given species can change over
time, such as lumping or splitting of a given taxa, and by taxonomic
authority (e.g., WoRMS, ITIS or GBIF). These taxonomic identifiers
and the full taxonomic hierarchy should get regularly updated
regardless of source observational data, and can either be updated
in the table directly or joined one-to-one with a seperate table in
a materialized view (so as not to slow down queries with a regular
view).
This workflow processes NOAA CalCOFI database CSV files and outputs
parquet files. The workflow:
1. Reads CSV files from source directory (with GCS archive sync)
2. Loads into local wrangling DuckDB with transformations
3. Restructures primary keys (natural keys + sequential IDs)
4. Creates lookup table and consolidates ichthyo tables
5. Validates data quality and flags issues
6. Exports to parquet files (for later integration into Working DuckLake)
```{mermaid}
%%| label: overview
%%| fig-cap: "Overview diagram of CSV ingestion process into the database."
%%| file: diagrams/ingest_noaa-calcofi-db_overview.mmd
```
See also [5.3 Ingest datasets with documentation – Database – CalCOFI.io Docs](https://calcofi.io/docs/db.html#ingest-datasets-with-documentation) for generic overview of ingestion process.
```{r}
#| label: setup
# devtools::install_local(here::here("../calcofi4db"), force = T)
devtools::load_all(here::here("../calcofi4db"))
devtools::load_all(here::here("../calcofi4r"))
# options(error=NULL)
librarian::shelf(
CalCOFI / calcofi4db,
CalCOFI / calcofi4r,
DBI,
dplyr,
DT,
fs,
glue,
gargle,
googledrive,
here,
htmltools,
janitor,
jsonlite,
knitr,
listviewer,
litedown,
lubridate,
purrr,
readr,
rlang,
sf,
stringr,
tibble,
tidyr,
units,
uuid,
webshot2,
quiet = T
)
options(readr.show_col_types = F)
options(DT.options = list(scrollX = TRUE))
# common ingest settings (overwrite, dir_data)
source(here("libs/ingest.R"))
# define paths
# provider/dataset/metadata read from this file's authoritative YAML block
cc <- read_calcofi_meta(here("ingest_swfsc_ichthyo.qmd"))
provider <- cc$provider
dataset <- cc$dataset
dataset_name <- cc$dataset_meta$dataset_name
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)
}
# clear stale WAL/tmp from an interrupted run (avoids "WAL checkpoint
# iteration does not match" errors when reopening)
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"))
# keep dir_parquet so write_parquet_outputs can content-hash dedup against
# the prior run (only changed partitions re-written/uploaded)
}
dir_create(dirname(db_path))
con <- get_duckdb_con(db_path)
load_duckdb_extension(con, "spatial")
# load data using calcofi4db package
# - reads from local Google Drive mount
# - syncs to GCS archive if files changed (creates new timestamped archive)
# - tracks GCS archive path for provenance
d <- read_csv_files(
provider = provider,
dataset = dataset,
dir_data = dir_data,
sync_archive = TRUE,
metadata_dir = here("metadata")
) # workflows/metadata/{provider}/{dataset}/
# show source files summary
message(glue("Loaded {nrow(d$source_files)} tables from {d$paths$dir_csv}"))
message(glue("Total rows: {sum(d$source_files$nrow)}"))
```
## Check for any mismatched tables and fields
```{r}
#| label: data_integrity_checkpoint
#| output: asis
# check data integrity - detects mismatches and controls chunk evaluation
integrity <- check_data_integrity(
d = d,
dataset_name = dataset_name,
halt_on_fail = TRUE
)
# render the pass/fail message
render_integrity_message(integrity)
```
## Show Source Files
```{r}
#| label: source-files
show_source_files(d)
```
## Show CSV Tables and Fields to Ingest
```{r}
#| label: tbls_in
d$d_csv$tables |>
datatable(caption = "Tables to ingest.")
```
```{r}
#| label: flds_in
d$d_csv$fields |>
datatable(caption = "Fields to ingest.")
```
## Show tables and fields redefined
```{r}
#| label: tbls_rd
show_tables_redefine(d)
```
```{r}
#| label: flds_rd
show_fields_redefine(d)
```
## Load Tables into Database
```{r}
#| label: load_tbls_to_db
# use ingest_dataset() which handles:
# - transform_data() for applying redefinitions
# - provenance tracking via gcs_path from read_csv_files()
# - automatic uuid column detection
# - ingest_to_working() for each table
tbl_stats <- ingest_dataset(
con = con,
d = d,
mode = if (overwrite) "replace" else "append",
verbose = TRUE
)
tbl_stats |>
datatable(rownames = FALSE, filter = "top")
```
## Establish Primary Keys
**UUID-first approach**: Source tables (site, tow, net) retain their `*_uuid` columns as
primary unique identifiers. These UUIDs are minted at sea and remain stable throughout
the data lifecycle — even when rows are removed and re-included during QA/QC. Sequential
integer IDs would lose this stability because re-sorting or row additions change the
assignment. Only `cruise` uses a natural key (`cruise_key`) because it has few rows with
easily identifiable attributes (ship + year-month). The `ichthyo` table uses a deterministic
UUID v5 hashed from its composite natural key. Other derived tables without source UUIDs
(lookup, segment) still use sequential integer IDs for convenience.
### Data Corrections
Apply known data corrections identified by data managers — **before**
`create_cruise_key()` derives any key, not after. Correction 1 patches a blank
`ship.ship_nodc` (BOLD HORIZON has none in the source `shiplookup`); until
2026-09 this chunk ran 280 lines *after* `create_cruise_key()`, so the July
2019 Bold Horizon cruise was already minted as `cruise_key = "2019-07-"` (an
empty NODC segment — DuckDB's `CONCAT()` treats NULL as `''`) by the time the
correction landed, and the key was never re-derived (WS-B / question
`swfsc_ichthyo_14`). `create_cruise_key()` now also refuses to mint a
malformed key outright (`calcofi4db` ≥ 3.32.0), so a reordering mistake here
fails loudly instead of shipping a bad key again. Correction 2 (missing invert
species rows) is order-independent — it only appends to `species`.
```{r}
#| label: apply_corrections
# apply data corrections BEFORE create_cruise_key() — see prose above
apply_data_corrections(con, verbose = TRUE)
```
### Cruises SWFSC moved back to staging (Q09)
SWFSC's Ed Weber (2026-09-25, `swfsc_ichthyo_09`): cruises **198202JD** and **198212JD**
(`1982-02-31JD`, `1982-12-31JD`) are partially sorted and "have been moved back to the staging
schema", so they are **not part of the source**; every other tow without a row is a true zero and
stays in the denominator. The 2026-09-26 export (`Cruise.csv` 694 rows) no longer contains them. The
v2026.10.01 release was built from the earlier export and still carried them (1,128 sites, 1,112
tows, 1,145 nets, 1,100 `obs` rows, all `abundance`, of which 1,055 on 1982-02 and 45 on
1982-12). This chunk is the explicit exclusion: it stops the render if an export that still holds
either cruise is read, rather than letting a partially sorted cruise be counted as zero catch.
```{r}
#| label: exclude_staging_cruises
# Q09: no row of these cruises may survive into the ingest. They are removed at
# the source (SWFSC), so the check is on what was LOADED, before any key is minted.
q09_cruises <- tibble::tribble(
~ship_key, ~date_ym, ~cruise,
"JD", "1982-02-01", "198202JD",
"JD", "1982-12-01", "198212JD")
q09_in <- paste0("('", q09_cruises$ship_key, "', '", q09_cruises$date_ym, "')", collapse = ", ")
n_q09 <- dbGetQuery(con, glue("
SELECT COUNT(*) FROM cruise
WHERE (ship_key, CAST(date_ym AS VARCHAR)) IN ({q09_in})"))[[1]]
stopifnot(
"cruises 198202JD / 198212JD are in the source export: SWFSC moved them back to staging (swfsc_ichthyo_09, 2026-09-25); re-sync the 2026-09-26 or later export" =
n_q09 == 0)
cat(glue("Q09: {nrow(q09_cruises)} staging cruises ({paste(q09_cruises$cruise, collapse = ', ')}) absent from the source export"), "\n")
```
### Create cruise_key (natural key)
The cruise_key is a natural key in format YYYY-MM-NODC (4-digit year + 2-digit month + NODC ship code).
```{r}
#| label: create_cruise_key
# create cruise_key as natural primary key (YYYY-MM-NODC format)
create_cruise_key(
con,
cruise_tbl = "cruise",
ship_tbl = "ship",
date_col = "date_ym"
)
# verify uniqueness
cruise_keys <- tbl(con, "cruise") |> pull(cruise_key)
if (any(duplicated(cruise_keys))) {
dups <- cruise_keys[duplicated(cruise_keys)] |> unique() |> head(5)
stop(glue(
"cruise_key must be unique — found {sum(duplicated(cruise_keys))} ",
"duplicates, e.g.: {paste(dups, collapse = ', ')}. ",
"Try deleting {db_path} and re-rendering."
))
}
# show sample cruise keys
tbl(con, "cruise") |>
select(cruise_uuid, cruise_key, ship_key, date_ym) |>
head(10) |>
collect() |>
datatable(caption = "Sample cruise_key values (YYYY-MM-NODC format)")
```
### Propagate cruise_key to child tables
Propagate the natural cruise_key to the site table for convenience in queries and sorting.
The structural foreign key (site.cruise_uuid → cruise.cruise_uuid) comes from the source data.
```{r}
#| label: propagate_cruise_key
# propagate cruise_key from cruise to site (via cruise_uuid)
propagate_natural_key(
con = con,
child_tbl = "site",
parent_tbl = "cruise",
key_col = "cruise_key",
join_col = "cruise_uuid"
)
# verify cruise_key is now in site
tbl(con, "site") |>
select(site_uuid, cruise_uuid, cruise_key, order_occ) |>
head(10) |>
collect() |>
datatable(caption = "Sample site rows with cruise_key")
```
### Cruise date spans
Every downstream ingest keys its events to a cruise. Until calcofi4db 3.20.0
that was done by the event's calendar month, and a CalCOFI cruise routinely
straddles one (5508BD ran 7 Aug – 25 Sep 1955; 184 of the 664 bottle cruises span
two months) — the shorn-off casts landed on the *next* cruise of the same ship
without any FK ever failing (v2026.08.14: 664 bottle cruises released as 799
keys). `calcofi4db::resolve_cruise_key()` now matches an event to the cruise whose
observed **date span** contains it, and this is where the span comes from: each
cruise's own tows. It rides on the `cruise` shard, so nothing downstream needs a
new input.
```{r}
#| label: cruise_date_span
d_span <- add_cruise_date_span(
con,
event_sql = "SELECT s.cruise_key, t.datetime_start_utc AS datetime
FROM tow t JOIN site s USING (site_uuid)")
# an overlap between two cruises of one ship would make span matching
# ambiguous; it is a reference-data error, not a tolerance to widen
stopifnot("cruise date spans of one ship overlap" = !any(d_span$overlaps))
cat(glue(
"{sum(!is.na(d_span$date_min))}/{nrow(d_span)} cruises have a span; ",
"{sum(d_span$spills_month, na.rm = TRUE)} extend outside their designated ",
"month — expected, and exactly why the span exists\n"))
d_span |>
filter(spills_month) |>
select(cruise_key, date_ym, date_min, date_max, n_events) |>
datatable(caption = "Cruises whose tows extend outside their designated month")
```
## Create Lookup Table
Create unified lookup table from vocabularies for egg stages, larva stages, and tow types.
```{r}
#| label: create_lookup_table
# egg stage vocabulary (Moser & Ahlstrom, 1985)
egg_stage_vocab <- tibble(
stage_int = 1:11,
stage_description = c(
"egg, stage 1 of 11 (Moser & Ahlstrom, 1985)",
"egg, stage 2 of 11 (Moser & Ahlstrom, 1985)",
"egg, stage 3 of 11 (Moser & Ahlstrom, 1985)",
"egg, stage 4 of 11 (Moser & Ahlstrom, 1985)",
"egg, stage 5 of 11 (Moser & Ahlstrom, 1985)",
"egg, stage 6 of 11 (Moser & Ahlstrom, 1985)",
"egg, stage 7 of 11 (Moser & Ahlstrom, 1985)",
"egg, stage 8 of 11 (Moser & Ahlstrom, 1985)",
"egg, stage 9 of 11 (Moser & Ahlstrom, 1985)",
"egg, stage 10 of 11 (Moser & Ahlstrom, 1985)",
"egg, stage 11 of 11 (Moser & Ahlstrom, 1985)"
)
)
# larva stage vocabulary
larva_stage_vocab <- tibble(
stage_int = 1:5,
stage_txt = c("YOLK", "PREF", "FLEX", "POST", "TRNS"),
stage_description = c(
"larva, yolk sac",
"larva, preflexion",
"larva, flexion",
"larva, postflexion",
"larva, transformation"
)
)
# tow type vocabulary (from tow_type table)
tow_type_vocab <- tbl(con, "tow_type") |>
collect() |>
mutate(
lookup_num = row_number(),
lookup_chr = tow_type_key,
description = description
) |>
select(lookup_num, lookup_chr, description)
# create unified lookup table
create_lookup_table(
con = con,
egg_stage_vocab = egg_stage_vocab,
larva_stage_vocab = larva_stage_vocab,
tow_type_vocab = tow_type_vocab
)
# show lookup table
tbl(con, "lookup") |>
collect() |>
datatable(caption = "Lookup table with vocabularies")
```
## Egg stages 12-15 are screened out (swfsc_ichthyo_02)
The egg stage scale this ingest publishes is Moser & Ahlstrom (1985), stages 1-11. SWFSC
(Ed Weber, 2026-09-25): "Stages 12-15 have been used for Dover Sole and sometimes Rex Sole but I
do not have a reference. You or I could screen them out if wanted." Ben Best (2026-10-01): "we can
filter out the observations with egg stages 12-15". So those egg-stage rows leave the `stage`
distribution here, before consolidation, and are written to
`data/flagged/egg_stage_12_15_screened.csv`. The eggs themselves are still counted: the egg
abundance (`Eggs.csv`) is a separate total and is not touched.
```{r}
#| label: screen_egg_stages
d_egg_1215 <- dbGetQuery(con, "
SELECT e.net_uuid, e.species_id, s.scientific_name, e.stage, e.tally
FROM egg_stage e LEFT JOIN species s USING (species_id)
WHERE e.stage > 11 ORDER BY e.species_id, e.stage, e.net_uuid")
dir_create(here("data/flagged"))
write_csv(d_egg_1215, here("data/flagged/egg_stage_12_15_screened.csv"), na = "")
n_before <- dbGetQuery(con, "SELECT COUNT(*) FROM egg_stage")[[1]]
invisible(dbExecute(con, "DELETE FROM egg_stage WHERE stage > 11"))
stopifnot(
"egg stages must be 1-11 (Moser & Ahlstrom 1985) after screening" =
dbGetQuery(con, "SELECT COUNT(*) FROM egg_stage WHERE stage NOT BETWEEN 1 AND 11")[[1]] == 0,
"screening removed exactly the flagged rows" =
dbGetQuery(con, "SELECT COUNT(*) FROM egg_stage")[[1]] == n_before - nrow(d_egg_1215))
cat(glue("egg stages 12-15 screened out: {format(nrow(d_egg_1215), big.mark = ',')} rows, ",
"{format(sum(d_egg_1215$tally), big.mark = ',')} eggs, on ",
"{paste(unique(d_egg_1215$scientific_name), collapse = ', ')}\n"))
d_egg_1215 |>
count(species_id, scientific_name, stage, wt = tally, name = "eggs") |>
datatable(caption = "Egg-stage rows screened out (stages 12-15), eggs by species and stage")
```
## Larval size and stage from `LarvaeMeasured`
The provider's 2026-09-26 export replaced `larvaesize` and `larvaestage` with one table,
`LarvaeMeasured` (`larva_measured` here): a row per group of measured larvae with a length, a
stage or both, keyed to its `Larvae` count row by `CountId`, which carries the net and the species
(Ed Weber, 2026-09-25, `swfsc_ichthyo_04`: "you might have to aggregate, for example, if there are
fish with multiple lengths but the same stage"). This chunk rebuilds the two distributions the
rest of the notebook consumes at their old grain: `larva_size` sums `tally` by
(net, species, length), `larva_stage` by (net, species, stage). The joint length x stage the new
table carries is not used (it was never in the source before).
A measured row whose `CountId` names no `Larvae` row has no net and no species, so it cannot be
placed; it is written to `data/flagged/orphan_larva_measured.csv` and left out.
```{r}
#| label: larva_measured
# every larva count row is one (net, species): the join below must not fan out
stopifnot(
"Larvae.CountId must be unique" =
dbGetQuery(con, "SELECT COUNT(*) - COUNT(DISTINCT count_uuid) FROM larva")[[1]] == 0,
"Larvae (net, species) must be unique: one count row per taxon per net" =
dbGetQuery(con, "SELECT COUNT(*) FROM (SELECT net_uuid, species_id FROM larva
GROUP BY ALL HAVING COUNT(*) > 1)")[[1]] == 0)
# measured rows that reach no count row: no net, no species
d_lm_orphan <- dbGetQuery(con, "
SELECT m.larva_measured_uuid, m.count_uuid, m.length_mm, m.stage, m.tally
FROM larva_measured m ANTI JOIN larva l USING (count_uuid)")
dir_create(here("data/flagged"))
write_csv(d_lm_orphan, here("data/flagged/orphan_larva_measured.csv"), na = "")
for (sql in c(
"CREATE OR REPLACE TABLE larva_size AS
SELECT l.net_uuid, l.species_id, m.length_mm, CAST(SUM(m.tally) AS INTEGER) AS tally
FROM larva_measured m JOIN larva l USING (count_uuid)
WHERE m.length_mm IS NOT NULL
GROUP BY ALL",
"CREATE OR REPLACE TABLE larva_stage AS
SELECT l.net_uuid, l.species_id, m.stage, CAST(SUM(m.tally) AS INTEGER) AS tally
FROM larva_measured m JOIN larva l USING (count_uuid)
WHERE m.stage IS NOT NULL
GROUP BY ALL"))
invisible(dbExecute(con, sql))
# the two distributions account for every placeable measured row: nothing is
# dropped by the aggregation except rows with neither a length nor a stage
n_lm <- dbGetQuery(con, "
SELECT COUNT(*) AS n,
COUNT(*) FILTER (WHERE length_mm IS NOT NULL) AS n_len,
COUNT(*) FILTER (WHERE stage IS NOT NULL) AS n_stage,
COUNT(*) FILTER (WHERE length_mm IS NULL AND stage IS NULL) AS n_neither,
SUM(tally) FILTER (WHERE length_mm IS NOT NULL) AS tally_len,
SUM(tally) FILTER (WHERE stage IS NOT NULL) AS tally_stage
FROM larva_measured m SEMI JOIN larva l USING (count_uuid)")
stopifnot(
"larva_size must carry every placeable measured tally with a length" =
dbGetQuery(con, "SELECT SUM(tally) FROM larva_size")[[1]] == n_lm$tally_len,
"larva_stage must carry every placeable measured tally with a stage" =
dbGetQuery(con, "SELECT SUM(tally) FROM larva_stage")[[1]] == n_lm$tally_stage,
"every larval stage must be in the YOLK/PREF/FLEX/POST/TRNS vocabulary" =
all(dbGetQuery(con, "SELECT DISTINCT stage FROM larva_stage")$stage %in%
larva_stage_vocab$stage_txt))
cat(glue(
"larva_measured: {format(nrow(d_lm_orphan) + n_lm$n, big.mark = ',')} rows; ",
"{format(nrow(d_lm_orphan), big.mark = ',')} orphan (CountId in no Larvae row, ",
"tally {format(sum(d_lm_orphan$tally), big.mark = ',')}, flagged); ",
"{format(n_lm$n_neither, big.mark = ',')} with neither length nor stage\n",
"larva_size: {format(dbGetQuery(con, 'SELECT COUNT(*) FROM larva_size')[[1]], big.mark = ',')} ",
"(net, species, length) rows\n",
"larva_stage: {format(dbGetQuery(con, 'SELECT COUNT(*) FROM larva_stage')[[1]], big.mark = ',')} ",
"(net, species, stage) rows\n"))
```
## Consolidate to Tidy Ichthyo Table
Consolidate 5 ichthyoplankton tables (egg, eggstage, larva, larvastage, larvasize) into
a single tidy table.
```{mermaid}
%%| label: consolidate_ichthyo_diagram
%%| fig-cap: "Consolidate 5 ichthyoplankton tables into a single tidy ichthyo table with lookup."
erDiagram
egg {
uuid net_uuid FK
int species_id FK
int tally
}
egg_stage {
uuid net_uuid FK
int species_id FK
int stage
int tally
}
larva {
uuid net_uuid FK
int species_id FK
int tally
}
larva_stage {
uuid net_uuid FK
int species_id FK
str stage
int tally
}
larva_size {
uuid net_uuid FK
int species_id FK
dbl length_mm
int tally
}
ichthyo {
uuid ichthyo_uuid PK
uuid net_uuid FK
int species_id FK
str life_stage
str measurement_type
dbl measurement_value
int tally
}
lookup {
int lookup_id PK
str category
int lookup_num
str lookup_chr
str description
}
egg ||--|{ ichthyo : "life_stage = egg"
egg_stage ||--|{ ichthyo : "life_stage = egg, type = stage"
larva ||--|{ ichthyo : "life_stage = larva"
larva_stage ||--|{ ichthyo : "life_stage = larva, type = stage"
larva_size ||--|{ ichthyo : "life_stage = larva, type = size"
ichthyo }o--|| lookup : "measurement_value"
```
```{r}
#| label: consolidate_ichthyo
message("Consolidating ichthyoplankton tables...")
# consolidate all ichthyo tables (keeps net_uuid as FK to net table)
consolidate_ichthyo_tables(
con = con,
output_tbl = "ichthyo",
larva_stage_vocab = larva_stage_vocab
)
# consolidate_ichthyo_tables() skips a source table that does not exist; after the
# 2026-09-26 export dropped larvaesize/larvaestage, that would have shipped no
# size or stage rows without an error. Every one of the six parts must be present.
d_parts <- dbGetQuery(con, "
SELECT DISTINCT life_stage || '/' || COALESCE(measurement_type, '') AS part FROM ichthyo")
stopifnot(
"ichthyo must hold all six parts (egg, egg stage, larva, larva stage, larva size, invert)" =
setequal(d_parts$part,
c("egg/", "egg/stage", "larva/", "larva/stage", "larva/size", "invert/")))
# assign ichthyo_uuid — deterministic UUID v5 from composite natural key
assign_deterministic_uuids(
con = con,
table_name = "ichthyo",
id_col = "ichthyo_uuid",
key_cols = c(
"net_uuid",
"species_id",
"life_stage",
"measurement_type",
"measurement_value"
)
)
# show sample rows
tbl(con, "ichthyo") |>
head(20) |>
collect() |>
datatable(caption = "Sample ichthyo table rows (tidy format)")
```
```{r}
#| label: ichthyo_summary
# summarize ichthyo table
ichthyo_summary <- tbl(con, "ichthyo") |>
group_by(life_stage, measurement_type) |>
summarize(
n_rows = n(),
n_species = n_distinct(species_id),
total_tally = sum(tally, na.rm = TRUE),
.groups = "drop"
) |>
collect()
ichthyo_summary |>
arrange(life_stage, measurement_type) |>
datatable(
caption = HTML(mark(
"The `ichthyo` table summary by life_stage and measurement_type"
))
) |>
formatCurrency(
columns = c("n_rows", "n_species", "total_tally"),
currency = "",
digits = 0
)
```
## Data Quality Improvements
This section validates referential integrity. Data corrections (§ "Establish
Primary Keys" above) now run *before* `create_cruise_key()`, not here — see
that section for why.
### Validate Referential Integrity
Run validation checks and flag invalid rows for review.
```{r}
#| label: validate_integrity
# ensure flagged directory exists
dir_flagged <- here("data/flagged")
if (!dir.exists(dir_flagged)) {
dir.create(dir_flagged, recursive = TRUE)
}
# validate egg stages (must be 1-11)
invalid_egg_stages <- validate_egg_stages(con, "egg_stage", "stage")
invalid_egg_stages_csv <- file.path(dir_flagged, "invalid_egg_stages.csv")
invalid_egg_stages_desc <- "Egg stage values NOT 1 to 11 (ie, not in Moser & Ahlstrom 1985 vocab)"
if (nrow(invalid_egg_stages) > 0) {
flag_invalid_rows(
invalid_rows = invalid_egg_stages,
output_path = invalid_egg_stages_csv,
description = invalid_egg_stages_desc
)
} else if (file.exists(invalid_egg_stages_csv)) {
# stages 12-15 are screened out above (screen_egg_stages); a sidecar from an
# earlier run would claim flagged rows that no longer exist
file.remove(invalid_egg_stages_csv)
}
show_flagged_file(
invalid_egg_stages,
invalid_egg_stages_csv,
invalid_egg_stages_desc
)
# define validation checks
validations <- list(
list(
type = "fk",
data_tbl = "ichthyo",
col = "species_id",
ref_tbl = "species",
ref_col = "species_id",
output_file = "orphan_species.csv",
description = "Species IDs not found in species table"
),
list(
type = "fk",
data_tbl = "ichthyo",
col = "net_uuid",
ref_tbl = "net",
ref_col = "net_uuid",
output_file = "orphan_nets.csv",
description = "Net UUIDs not found in net table"
)
)
# run validations
validation_results <- validate_dataset(
con = con,
validations = validations,
output_dir = dir_flagged
)
# show validation summary with GitHub links
show_validation_results(validation_results)
```
```{r}
#| label: delete_flagged
# optionally delete flagged rows (dry run first)
if (validation_results$total_flagged > 0) {
message(glue("Found {validation_results$total_flagged} invalid rows"))
# dry run to see what would be deleted
delete_stats <- delete_flagged_rows(
con = con,
validation_results = validation_results,
dry_run = TRUE
)
delete_stats |> datatable(caption = "Rows to be deleted (dry run)")
# uncomment to actually delete:
delete_flagged_rows(con, validation_results, dry_run = FALSE)
}
```
### Drop Deprecated Tables
The source tables have been consolidated into `ichthyo` (tidy format) and `lookup`
(vocabularies). Drop these before creating the schema diagram.
Note: `*_uuid` columns are retained as primary unique identifiers (see rationale
in "Establish Primary Keys" above).
```{r}
#| label: drop_deprecated
# tables consolidated into ichthyo
deprecated_ichthyo <- c(
"egg",
"egg_stage",
"larva",
"larva_stage",
"larva_size",
"larva_measured"
)
# tables consolidated into lookup
deprecated_lookup <- c("tow_type")
# tables of the 2026-09-26 export that are loaded (so the integrity check sees
# every source column) but NOT ingested; each needs a design decision first:
# - cufes: CUFES is its own ingest (ingest_swfsc_cufes.qmd, from this export's Cufes.csv); switching
# its source to this export is a change to that notebook
# - species_itis / species_worms: the provider's ITIS/WoRMS names and lineage
# (swfsc_ichthyo_05/13); emit_core reads SpeciesWormsLookup from the source to
# check the keys, so the wrangling copies are not needed
# - preservative: a lookup nothing references in this export
# `fish` (grown-out larvae, Ben 2026-10-04) is kept: published as its own table in
# emit_core.
not_ingested <- c("cufes", "species_itis", "species_worms", "preservative")
# all deprecated tables
deprecated_tables <- c(deprecated_ichthyo, deprecated_lookup, not_ingested)
# ShipLookup.ShipIces -> ship.ship_ices, beside ship_nodc (swfsc_ichthyo_14; Ben,
# 2026-10-01: "carry the ICES code ... as its own column on the ship and cruise
# tables, beside the NODC code"). cruise_key stays YYYY-MM-NODC: ship_nodc is
# untouched and still keys every cruise. The release copies it onto `cruise`.
# Every SWFSC ship carries one; where it differs from the NODC code it is reported.
d_ices <- dbGetQuery(con, "
SELECT ship_key, ship_name, ship_nodc, ship_ices FROM ship
WHERE ship_ices IS DISTINCT FROM ship_nodc ORDER BY 1")
stopifnot(
"every SWFSC ship carries an ICES code (swfsc_ichthyo_14)" =
dbGetQuery(con, "SELECT COUNT(*) FROM ship WHERE ship_ices IS NULL")[[1]] == 0)
d_ices |>
datatable(caption = "Ships whose ICES code differs from the NODC code (cruise_key keeps NODC)")
# drop each deprecated table
for (tbl in deprecated_tables) {
if (tbl %in% DBI::dbListTables(con)) {
DBI::dbExecute(con, glue("DROP TABLE {tbl}"))
message(glue("Dropped deprecated table: {tbl}"))
}
}
message(glue(
"\nRemaining tables: {paste(sort(DBI::dbListTables(con)), collapse = ', ')}"
))
```
## Standardize Taxonomy
Update species table with WoRMS/ITIS/GBIF identifiers using local lookups
against `spp.duckdb` (MarineSensitivity species DB). Falls back to WoRMS API
only for species not found locally. Then build taxonomy hierarchy via recursive
CTEs.
```{r}
#| label: standardize_species
# MarineSensitivity species DB for local taxonomy lookups
spp_db_path <- Sys.getenv(
"SPP_DB_PATH",
unset = "/Users/bbest/_big/msens/derived/spp.duckdb"
)
sp_results <- standardize_species_local(
con = con,
spp_db_path = spp_db_path,
overwrite = overwrite
)
sp_results |>
datatable(caption = "Species standardization results")
```
```{r}
#| label: build_taxon
taxon_rows <- build_taxon_hierarchy(
con = con,
spp_db_path = spp_db_path,
overwrite = overwrite
)
# show taxon stats
if (nrow(taxon_rows) > 0) {
taxon_rows |>
count(authority, taxonRank) |>
arrange(authority, taxonRank) |>
datatable(caption = "Taxon hierarchy by authority and rank")
}
# show taxa_rank lookup
dbReadTable(con, "taxa_rank") |>
datatable(caption = "Taxa rank ordering")
```
### Taxonomy Statistics
```{r}
#| label: taxonomy_stats
n_species <- dbGetQuery(con, "SELECT COUNT(*) AS n FROM species")$n
n_worms <- dbGetQuery(
con,
"SELECT COUNT(*) AS n FROM species WHERE worms_id IS NOT NULL"
)$n
n_itis <- dbGetQuery(
con,
"SELECT COUNT(*) AS n FROM species WHERE itis_id IS NOT NULL"
)$n
n_gbif <- dbGetQuery(
con,
"SELECT COUNT(*) AS n FROM species WHERE gbif_id IS NOT NULL"
)$n
n_taxon <- dbGetQuery(con, "SELECT COUNT(*) AS n FROM taxon")$n
tibble(
metric = c(
"total species",
"with worms_id",
"with itis_id",
"with gbif_id",
"taxon hierarchy rows"
),
count = c(n_species, n_worms, n_itis, n_gbif, n_taxon)
) |>
datatable(caption = "Taxonomy standardization summary")
```
## Schema Documentation
```{r}
#| label: erd_tbls
tbls <- dbListTables(con) |> sort()
cc_erd(con, tables = tbls)
```
### Primary Key Strategy
**UUID-first**: Source tables use UUIDs minted at sea as primary identifiers.
Only `cruise` uses a natural key because it has few, easily identifiable rows.
Derived tables (lookup, segment) use sequential integer IDs since they
have no source UUID. The `ichthyo` table uses a deterministic UUID v5 hashed
from its composite natural key.
| Table | Primary Key | Type |
|-------|-------------|------|
| `cruise` | `cruise_key` | Natural key (YYYY-MM-NODC); also retains `cruise_uuid` from source |
| `ship` | `ship_key` | Natural key (2-letter) |
| `site` | `site_uuid` | Source UUID (minted at sea) |
| `tow` | `tow_uuid` | Source UUID (minted at sea) |
| `net` | `net_uuid` | Source UUID (minted at sea) |
| `species` | `species_id` | Natural from source |
| `ichthyo` | `ichthyo_uuid` | Deterministic UUID v5 (from net_uuid, species_id, life_stage, measurement_type, measurement_value) |
| `lookup` | `lookup_id` | Sequential (derived table, sorted by lookup_type, lookup_num) |
| `taxon` | `(authority, taxonID)` | Compound natural key (authority + ID within authority); ERD uses `taxonID` only |
| `taxa_rank` | `taxonRank` | Natural key (rank name string) |
### Foreign Key Relationships
DuckDB doesn't support `ALTER TABLE ADD FOREIGN KEY`. We define relationships as lists for visualization and `relationships.json` export.
**Foreign Key Relationships:**
```
ship.ship_key (PK)
↓
cruise.cruise_key (PK) ←── cruise.ship_key (FK)
↓ cruise.cruise_uuid (unique, source)
site.site_uuid (PK) ←── site.cruise_uuid (FK → cruise.cruise_uuid)
↓ site.cruise_key (denormalized, for queries)
tow.tow_uuid (PK) ←── tow.site_uuid (FK → site.site_uuid)
↓ tow.tow_type_key → lookup (lookup_type='tow_type')
net.net_uuid (PK) ←── net.tow_uuid (FK → tow.tow_uuid)
↓
ichthyo.ichthyo_uuid (PK) ←── ichthyo.net_uuid (FK → net.net_uuid)
ichthyo.species_id (FK) → species.species_id
taxon.(authority, taxonID) (PK) ←── taxon.taxonRank (FK) → taxa_rank.taxonRank
taxa_rank.taxonRank (PK)
species.worms_id ··> taxon.taxonID (logical, WHERE authority='WoRMS')
species.itis_id ··> taxon.taxonID (logical, WHERE authority='ITIS')
```
```{r}
#| label: erd_fk
# define PK/FK relationships for visualization and relationships.json
# uses UUID PKs for source tables, sequential IDs for derived tables
ichthyo_rels <- list(
primary_keys = list(
cruise = "cruise_key",
ship = "ship_key",
site = "site_uuid",
tow = "tow_uuid",
net = "net_uuid",
species = "species_id",
ichthyo = "ichthyo_uuid",
lookup = "lookup_id",
taxon = "taxonID",
taxa_rank = "taxonRank",
grid = "grid_key",
segment = "segment_id"
),
foreign_keys = list(
list(
table = "ichthyo",
column = "net_uuid",
ref_table = "net",
ref_column = "net_uuid"
),
list(
table = "ichthyo",
column = "species_id",
ref_table = "species",
ref_column = "species_id"
),
list(
table = "net",
column = "tow_uuid",
ref_table = "tow",
ref_column = "tow_uuid"
),
list(
table = "tow",
column = "site_uuid",
ref_table = "site",
ref_column = "site_uuid"
),
list(
table = "site",
column = "cruise_key",
ref_table = "cruise",
ref_column = "cruise_key"
),
list(
table = "cruise",
column = "ship_key",
ref_table = "ship",
ref_column = "ship_key"
),
list(
table = "taxon",
column = "taxonRank",
ref_table = "taxa_rank",
ref_column = "taxonRank"
),
list(
table = "segment",
column = "cruise_key",
ref_table = "cruise",
ref_column = "cruise_key"
),
list(
table = "segment",
column = "site_uuid_beg",
ref_table = "site",
ref_column = "site_uuid"
),
list(
table = "invert",
column = "net_uuid",
ref_table = "net",
ref_column = "net_uuid"
),
list(
table = "invert",
column = "species_id",
ref_table = "species",
ref_column = "species_id"
)
)
)
cc_erd(con, rels = ichthyo_rels)
```
## Add Spatial
### Add `site.geom`
```{r}
#| label: mk_site_pts
# add geometry column using DuckDB spatial
# note: DuckDB spatial doesn't track SRID metadata (unlike PostGIS)
# all geometries assumed WGS84 (EPSG:4326) by convention
add_point_geom(con, "site", lon_col = "longitude", lat_col = "latitude")
```
### Fix calcofi4r `grid`
Problems with
[calcofi4r::`cc_grid`](https://calcofi.io/calcofi4r/articles/calcofi4r.html):
- uses old station (line, position) vs newer site (line, station)
- `sta_lin`, `sta_pos`: integer, so drops necessary decimal that is found in `site_key`
- `sta_lin == 90, sta_pos == 120` repeats for:
- `sta_pattern` == 'historical' (`sta_dpos` == 20); and
- `sta_pattern` == 'standard' (`sta_dpos` == 10)
```{r}
#| label: mk_grid_v2
librarian::shelf(
calcofi4r,
mapview,
quiet = T
)
cc_grid_v2 <- calcofi4r::cc_grid |>
# handle bundled data that may still have sta_key (renamed to site_key)
rename(any_of(c(site_key = "sta_key"))) |>
select(
site_key,
shore = sta_shore,
pattern = sta_pattern,
spacing = sta_dpos
) |>
separate_wider_delim(
site_key,
",",
names = c("line", "station"),
cols_remove = F
) |>
mutate(
line = as.double(line),
station = as.double(station),
grid_key = ifelse(
pattern == "historical",
glue("st{station}-ln{line}_hist"),
glue("st{station}-ln{line}")
),
zone = glue("{shore}-{pattern}")
) |>
relocate(grid_key, station) |>
st_as_sf() |>
mutate(
area_km2 = st_area(geom) |>
set_units(km^2) |>
as.numeric()
)
cc_grid_ctrs_v2 <- calcofi4r::cc_grid_ctrs |>
rename(any_of(c(site_key = "sta_key"))) |>
select(site_key, pattern = sta_pattern) |>
left_join(
cc_grid_v2 |>
st_drop_geometry(),
by = c("site_key", "pattern")
) |>
select(-site_key) |>
relocate(grid_key)
cc_grid_v2 <- cc_grid_v2 |>
select(-site_key)
cc_grid_v2 |>
st_drop_geometry() |>
datatable()
mapview(cc_grid_v2, zcol = "zone") +
mapview(cc_grid_ctrs_v2, cex = 1)
```
```{r}
#| label: grid_to_db
grid <- cc_grid_v2 |>
as.data.frame() |>
left_join(
cc_grid_ctrs_v2 |>
as.data.frame() |>
select(grid_key, geom_ctr = geom),
by = "grid_key"
) |>
st_as_sf(sf_column_name = "geom")
# convert sf geometry to WKB for DuckDB
grid_df <- grid |>
mutate(
geom_wkb = sf::st_as_binary(geom, hex = TRUE),
geom_ctr_wkb = sf::st_as_binary(geom_ctr, hex = TRUE)
) |>
sf::st_drop_geometry() |>
select(-geom_ctr)
# write to DuckDB
dbWriteTable(con, "grid", grid_df, overwrite = TRUE)
# convert WKB to native GEOMETRY (requires storage_compatibility_version = 'latest')
dbExecute(con, "ALTER TABLE grid ADD COLUMN IF NOT EXISTS geom GEOMETRY")
dbExecute(con, "UPDATE grid SET geom = ST_GeomFromHEXWKB(geom_wkb)")
dbExecute(con, "ALTER TABLE grid DROP COLUMN geom_wkb")
dbExecute(con, "ALTER TABLE grid ADD COLUMN IF NOT EXISTS geom_ctr GEOMETRY")
dbExecute(con, "UPDATE grid SET geom_ctr = ST_GeomFromHEXWKB(geom_ctr_wkb)")
dbExecute(con, "ALTER TABLE grid DROP COLUMN geom_ctr_wkb")
message("Grid table created with geometry columns")
```
### Update `site.grid_key`
```{r}
#| label: update_site_from_grid
grid_stats <- assign_grid_key(con, "site")
grid_stats |> datatable()
# add standardized site_key (NNN.N NNN.N format)
standardize_site_key(con, "site", "line", "station")
```
### Add `segment`: line segments between consecutive sites
```{r}
#| label: mk_segment
# use SQL to avoid GEOMETRY column type issue with tbl()
segment <- tbl(
con,
sql(
"SELECT cruise_key, order_occ, site_uuid, longitude AS lon, latitude AS lat
FROM site"
)
) |>
left_join(
tbl(con, "tow") |>
select(site_uuid, datetime_start_utc),
by = "site_uuid"
) |>
group_by(
cruise_key,
order_occ,
site_uuid,
lon,
lat
) |>
summarize(
time_beg = min(datetime_start_utc, na.rm = T),
time_end = max(datetime_start_utc, na.rm = T),
.groups = "drop"
) |>
collect()
segment <- segment |>
arrange(cruise_key, order_occ, time_beg) |>
group_by(cruise_key) |>
mutate(
site_uuid_beg = lag(site_uuid),
lon_beg = lag(lon),
lat_beg = lag(lat),
time_beg = lag(time_beg)
) |>
ungroup() |>
filter(!is.na(lon_beg), !is.na(lat_beg)) |>
mutate(
m = pmap(
list(lon_beg, lat_beg, lon, lat),
\(x1, y1, x2, y2) {
matrix(c(x1, y1, x2, y2), nrow = 2, byrow = T)
}
),
geom = map(m, st_linestring)
) |>
select(
cruise_key,
site_uuid_beg,
site_uuid_end = site_uuid,
lon_beg,
lat_beg,
lon_end = lon,
lat_end = lat,
time_beg,
time_end,
geom
) |>
st_as_sf(
sf_column_name = "geom",
crs = 4326
) |>
mutate(
time_hr = as.numeric(difftime(time_end, time_beg, units = "hours")),
length_km = st_length(geom) |>
set_units(km) |>
as.numeric(),
km_per_hr = length_km / time_hr
)
# convert to WKB and write to DuckDB
segment_df <- segment |>
mutate(geom_wkb = sf::st_as_binary(geom, hex = TRUE)) |>
sf::st_drop_geometry()
dbWriteTable(con, "segment", segment_df, overwrite = TRUE)
dbExecute(con, "ALTER TABLE segment ADD COLUMN IF NOT EXISTS geom GEOMETRY")
dbExecute(con, "UPDATE segment SET geom = ST_GeomFromHEXWKB(geom_wkb)")
dbExecute(con, "ALTER TABLE segment DROP COLUMN geom_wkb")
# assign segment_id sorted by time_beg
assign_sequential_ids(
con = con,
table_name = "segment",
id_col = "segment_id",
sort_cols = c("time_beg")
)
message("Segment table created")
```
```{r}
#| label: view_segment
# slowish, so use cached figure
map_segment_png <- here(glue("figures/{provider}_{dataset}_segment_map.png"))
if (!file_exists(map_segment_png)) {
# exclude native GEOMETRY column (unsupported by duckdb R driver);
# use ST_AsText to convert to WKT for sf
seg_cols <- dbGetQuery(
con,
"SELECT column_name FROM information_schema.columns
WHERE table_name = 'segment' AND data_type != 'GEOMETRY'"
)$column_name
seg_sql <- paste(
"SELECT",
paste(seg_cols, collapse = ", "),
", ST_AsText(geom) AS geom_wkt FROM segment"
)
segment_sf <- dbGetQuery(con, seg_sql) |>
st_as_sf(wkt = "geom_wkt", crs = 4326) |>
select(-geom_wkt) |>
mutate(year = year(time_beg))
m <- mapView(segment_sf, zcol = "year")
mapshot2(m, file = map_segment_png)
}
htmltools::img(
src = map_segment_png |> str_replace(here(), "."),
width = "600px"
)
```
## Report
```{r}
#| label: show_latest
# cc_erd handles GEOMETRY columns natively (unlike dm_from_con)
cc_erd(con, rels = ichthyo_rels)
```
```{r}
#| label: effort_stats
# use sql() to avoid GEOMETRY column type issue with tbl()
d_eff <- tbl(
con,
sql(
"SELECT segment_id, cruise_key, time_beg, time_hr, length_km FROM segment"
)
) |>
mutate(
year = year(time_beg)
) |>
group_by(year) |>
summarize(
time_hr = sum(time_hr, na.rm = T),
length_km = sum(length_km, na.rm = T)
) |>
collect()
total_hours <- sum(d_eff$time_hr, na.rm = T)
total_km <- sum(d_eff$length_km, na.rm = T)
fmt <- function(x, ...) format(x, big.mark = ",", ...)
message(glue(
"Total effort: {fmt(round(total_hours))} hours ({fmt(round(total_hours/24))} days, {fmt(round(total_hours/24/365, 1))} years)"
))
message(glue("Total distance: {fmt(round(total_km))} km"))
```
## Load Dataset Metadata
```{r}
#| label: load-dataset-metadata
# dataset registry built from authoritative ingest_*.qmd YAML (was dataset.csv)
d_dataset <- ingest_yaml_to_dataset_df(read_ingest_yaml(here()))
dbWriteTable(con, "dataset", d_dataset, overwrite = TRUE)
message(glue("dataset: {nrow(d_dataset)} datasets registered"))
```
## Questions for Data Providers
Open questions tracked in `metadata/swfsc/ichthyo/questions.csv`, surfaced here so
they travel with the workflow rather than living in someone's inbox. The blocker
is quantified by this notebook's own flagged-data sidecars: `orphan_species.csv`
excludes **316,316 specimens** across 17 unresolvable `species_id` values, so
every abundance total in the release is low by that amount until they resolve.
```{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 SWFSC ichthyoplankton data providers (ranked)")
```
## Validate Local Database
Validate data quality in the local wrangling database before exporting to parquet.
The parquet outputs from this workflow can later be used to update the Working DuckLake.
```{r}
#| label: validate
# validate data quality
validation <- validate_for_release(con)
if (validation$passed) {
message("Validation passed!")
if (nrow(validation$checks) > 0) {
validation$checks |>
filter(status != "pass") |>
datatable(caption = "Validation Warnings")
}
} else {
cat("Validation FAILED:\n")
cat(paste("-", validation$errors, collapse = "\n"))
}
```
## Enforce Column Types
Force integer/smallint types on columns that R's `numeric` mapped to `DOUBLE`
during `dbWriteTable()`. Uses `flds_redefine.csv` (`type_new`) as the source of
truth for source-table columns, plus explicit overrides for derived-table columns.
```{r}
#| label: enforce_types
type_changes <- enforce_column_types(
con = con,
d_flds_rd = d$d_flds_rd,
type_overrides = list(
ichthyo.ichthyo_uuid = "UUID",
ichthyo.net_uuid = "UUID",
ichthyo.species_id = "SMALLINT",
ichthyo.tally = "INTEGER",
lookup.lookup_id = "INTEGER",
lookup.lookup_num = "INTEGER",
segment.segment_id = "INTEGER",
segment.site_uuid_beg = "UUID",
segment.site_uuid_end = "UUID",
species.worms_id = "INTEGER",
species.itis_id = "INTEGER",
species.gbif_id = "INTEGER",
taxon.taxonID = "INTEGER",
taxon.acceptedNameUsageID = "INTEGER",
taxon.parentNameUsageID = "INTEGER",
taxa_rank.rank_order = "SMALLINT"
),
tables = dbListTables(con),
verbose = TRUE
)
if (nrow(type_changes) > 0) {
type_changes |>
datatable(caption = "Column type changes applied")
}
```
## Data Preview
Preview first and last rows of each table before writing parquet outputs.
```{r}
#| label: preview_tables
#| results: asis
preview_tables(
con,
c(
"cruise",
"ship",
"site",
"tow",
"net",
"species",
"taxon",
"taxa_rank",
"ichthyo",
"lookup",
"grid",
"segment"
)
)
```
## Emit Core Tables
Project this dataset into the shared consolidated core model
(`design_env-bio-consolidation.md`). 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.
Ichthyo is the deepest hierarchy in the model — `site` -> `tow` -> `net` — which
`sample` carries as an adjacency list (`parent_sample_key` / `root_sample_key`),
with `site_key`, `order_occ`, `grid_key` and `cruise_key` inherited down from the
site and the net gear code on `tow_type`. Net effort (volume filtered, standard
haul factor, …) becomes `sample_measurement`; the larval `size`/`stage`
distributions become `obs_attribute` under the abundance headline in `obs`.
```{r}
#| label: emit_core
ds_key <- "swfsc_ichthyo"
# filter the crosswalk to THIS dataset -- an unfiltered read leaks other datasets'
# taxa into this shard (the retired emit_core_tables() wrapper filtered internally)
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)
# This projection lives here, in the notebook that owns the dataset, not in a
# switch(dataset_key, ...) arm inside calcofi4db. The reusable SHAPES stay in the
# package (compat_event_sql / ns_key / prune_taxon_shard), so this is a declaration.
#
# taxa: the vocabulary is DECLARED here and RESOLVED by the package (taxon plan
# D1, calcofi4db >= 3.29.0). append_dataset_taxon() stages one `dataset_taxon`
# row per CalCOFI species code with `taxon_key` empty and the ids the SOURCE
# supplied (worms_id / itis_id / gbif_id) carried along as `ds_source_json`;
# resolve_dataset_taxon() fills the key from the authorities. ds_prefix is
# "calcofi", not the dataset_key: `species` is the shared CalCOFI species list
# (ichthyo + the folded invert export), so its ds_taxon_key is `calcofi:<id>`.
# There is no ichthyo arm inside calcofi4db any more — a column renamed or
# dropped here now fails at this call rather than becoming NA at release.
#
# AphiaID (swfsc_ichthyo_05/_13; Ed Weber 2026-09-25 "aggregate by AphiaID", Ben
# Best 2026-10-01 "aggregate taxa by AphiaID ... Bill's taxonomy stands"): the ids
# declared are the SOURCE's own Species.AphiaId / Species.Tsn, verbatim -- not the
# species table's, which standardize_species_local() above re-points to the
# accepted id from spp.duckdb (so ds_source_json said "source" about an id the
# source never gave). ensure_taxon_xref() then keys each code by the ACCEPTED id
# of its AphiaID (a key must be accepted; the source id stays in ds_source_json),
# and the source name stays verbatim as ds_scientific_name, the display name.
# Codes that share an AphiaID share a taxon_key; obs / obs_attribute below sum them.
d_src_species <- d$d_csv$data$data[[which(d$d_csv$data$tbl == "Species")]] |>
transmute(ds_taxa_code = as.character(SppCode),
worms_src = as.integer(AphiaId), itis_src = as.integer(Tsn))
d_vocab <- dbGetQuery(con, "
SELECT CAST(species_id AS VARCHAR) AS ds_taxa_code,
scientific_name AS ds_scientific_name,
common_name AS ds_common_name,
worms_id, itis_id, gbif_id
FROM species ORDER BY species_id") |>
left_join(d_src_species, by = "ds_taxa_code") |>
mutate(worms_id = coalesce(worms_src, as.integer(worms_id)),
itis_id = coalesce(itis_src, as.integer(itis_id))) |>
select(-worms_src, -itis_src)
stopifnot(
"every source species code is in the vocabulary" =
all(d_src_species$ds_taxa_code %in% d_vocab$ds_taxa_code),
"every source species code carries its own AphiaID" = !anyNA(d_src_species$worms_src))
n_staged <- append_dataset_taxon(con, ds_key, d_vocab, ds_prefix = "calcofi")
# build_taxon_reference() reads the WoRMS lineage `taxon` table built above
# (build_taxon_hierarchy()) as the authority for rank/parent/classification, then
# OVERWRITES `taxon` with the unified shape. prune_taxon_shard() trims it back to
# the transitive parent closure of this dataset's vocabulary — the hierarchy is
# broader than the taxa these observations reach, and ancestors must survive
# because descendant expansion walks parent_taxon_key.
# 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 <- resolve_dataset_taxon(con, mt_taxon, tx_over)
n_pruned <- prune_taxon_shard(con, ds_key)
# the ingest asserts its own crosswalk (D6): every species code the observations
# use is in the vocabulary, every staged row keys an authority id, every Aves
# taxon keys itis:. Halts the render on any finding.
obs_codes <- dbGetQuery(con,
"SELECT DISTINCT CAST(species_id AS VARCHAR) AS c FROM ichthyo
WHERE species_id IS NOT NULL")$c
chk_taxon <- check_dataset_taxon(con, ds_key, codes = obs_codes)
cat(glue("vocabulary: {n_staged} CalCOFI species codes staged; ",
"taxon {n_taxon} (pruned to {n_pruned$taxon}), ",
"dataset_taxon {n_ds_taxon}"), "\n")
# the provider's SpeciesWormsLookup ("their accepted taxon names", not updated in a
# year) against the key each code resolved to: a code keyed to another id than its
# source AphiaID (WoRMS has superseded it), or a lookup name WoRMS no longer uses,
# is listed in data/flagged/species_worms_lookup_check.csv for the provider
d_wl <- d$d_csv$data$data[[which(d$d_csv$data$tbl == "SpeciesWormsLookup")]] |>
transmute(worms_src = as.integer(AphiaId), lookup_name = Taxon, lookup_rank = taxon_rank)
d_taxkey <- dbGetQuery(con, glue("
SELECT d.ds_taxa_code, d.ds_scientific_name, d.taxon_key,
t.scientific_name AS key_name, t.rank AS key_rank
FROM dataset_taxon d LEFT JOIN taxon t USING (taxon_key)
WHERE d.dataset_key = '{ds_key}'")) |>
left_join(d_src_species, by = "ds_taxa_code") |>
left_join(d_wl, by = "worms_src") |>
mutate(rekeyed = taxon_key != paste0("worms:", worms_src),
lookup_stale = !is.na(lookup_name) & lookup_name != key_name)
d_taxkey_flag <- d_taxkey |>
filter(rekeyed | lookup_stale | is.na(lookup_name)) |>
select(ds_taxa_code, ds_scientific_name, worms_src, lookup_name, lookup_rank,
taxon_key, key_name, key_rank, rekeyed, lookup_stale) |>
arrange(as.integer(ds_taxa_code))
write_csv(d_taxkey_flag, here("data/flagged/species_worms_lookup_check.csv"), na = "")
d_shared <- d_taxkey |> group_by(taxon_key, key_name) |>
summarize(n_codes = n(), codes = paste(sort(as.integer(ds_taxa_code)), collapse = ", "),
source_names = paste(unique(ds_scientific_name), collapse = "; "), .groups = "drop") |>
filter(n_codes > 1) |> arrange(desc(n_codes), taxon_key)
stopifnot(
"every code's source AphiaID is in SpeciesWormsLookup" = all(!is.na(d_taxkey$lookup_name)))
cat(glue("AphiaID: {nrow(d_taxkey)} codes; {sum(d_taxkey$rekeyed)} keyed to the accepted id ",
"of a superseded source AphiaID; {sum(d_taxkey$lookup_stale)} lookup names differ from ",
"WoRMS's for the key; {nrow(d_shared)} taxon_keys shared by ",
"{sum(d_shared$n_codes)} codes (their counts are summed below)\n"))
d_shared |>
datatable(caption = "Species codes that share one taxon_key (one AphiaID): observations summed")
d_taxkey_flag |>
datatable(caption = "Codes keyed to another id than their source AphiaID, or whose SpeciesWormsLookup name WoRMS no longer uses")
# sample — three chained levels. `site` has no datetime of its own, so it takes
# the earliest tow time; `tow` and `net` inherit site_key/order_occ/grid_key/
# cruise_key from the site, and both carry the net gear code as tow_type.
# 16th column data_stage (ichthyo does not distinguish one, so NULL); 17th
# source_uuid — the provider's own site_uuid/tow_uuid/net_uuid, verbatim
# (calcofi4db >= 3.32.0, WS-B/Ed Weber's ask). Every other ingest's arm stays
# 15-column and gets source_uuid = NULL for free (append_sample()'s arity
# dispatch), so this is additive.
append_sample(con, glue("
SELECT {ns_key(ds_key, 'site', 's.site_uuid')} AS sample_key, 'site' AS sample_type,
NULL::VARCHAR AS parent_sample_key,
{ns_key(ds_key, 'site', 's.site_uuid')} AS root_sample_key,
'{ds_key}' AS dataset_key,
s.grid_key, s.site_key, s.cruise_key, CAST(s.order_occ AS INTEGER) AS order_occ,
s.latitude, s.longitude,
CAST(td.dt AS TIMESTAMP) AS datetime,
NULL::DOUBLE AS depth_min_m, NULL::DOUBLE AS depth_max_m,
NULL::VARCHAR AS tow_type,
NULL::VARCHAR AS data_stage,
s.site_uuid AS source_uuid
FROM site s
LEFT JOIN (SELECT site_uuid, min(datetime_start_utc) AS dt FROM tow GROUP BY 1) td
ON td.site_uuid = s.site_uuid"))
# depth_max_m of a tow and of its nets: Net.NetDepth, the MAXIMUM POSSIBLE depth of
# the net in m (nominal, not a measured depth; not wire-out) -- swfsc_ichthyo_08,
# and Ben's reading 2026-10-04 (swfsc_ichthyo_19, proposed to Ed). Where the sea
# floor is shallower the net did not reach it: 210 of 59,976 stations of the
# 2026-09-26 export put it more than 10 m below the deepest GEBCO cell around them.
# It is published as is (never clipped), and the release reports those stations in
# their own line of the seafloor check (check_depth_vs_seafloor(nominal_datasets =)).
# A net column but a property of the tow: the tow takes it only where every net of
# the tow agrees; a tow with no NetDepth (the surface Manta tows) keeps NULL.
d_netdepth <- dbGetQuery(con, "
SELECT t.tow_type_key,
COUNT(*) AS n_tow,
COUNT(*) FILTER (WHERE nd.n_depth = 1) AS n_tow_depth,
COUNT(*) FILTER (WHERE nd.n_depth > 1) AS n_tow_disagree,
MIN(nd.depth_max_m) AS depth_min,
MAX(nd.depth_max_m) AS depth_max
FROM tow t
LEFT JOIN (SELECT tow_uuid, COUNT(DISTINCT net_depth_m) AS n_depth,
MAX(net_depth_m) AS depth_max_m
FROM net GROUP BY 1) nd USING (tow_uuid)
GROUP BY 1 ORDER BY 2 DESC")
stopifnot(
"every net of a tow must report one NetDepth (swfsc_ichthyo_08)" =
sum(d_netdepth$n_tow_disagree) == 0,
"NetDepth must be a positive depth in m (never NaN, never <= 0)" =
dbGetQuery(con, "SELECT COUNT(*) FROM net WHERE net_depth_m IS NOT NULL
AND (isnan(net_depth_m) OR net_depth_m <= 0)")[[1]] == 0)
d_netdepth |>
datatable(caption = "Net.NetDepth (maximum possible net depth) -> sample.depth_max_m of tows and nets, by tow type")
append_sample(con, glue("
SELECT {ns_key(ds_key, 'tow', 't.tow_uuid')} AS sample_key, 'tow' AS sample_type,
{ns_key(ds_key, 'site', 't.site_uuid')} AS parent_sample_key,
{ns_key(ds_key, 'site', 't.site_uuid')} AS root_sample_key,
'{ds_key}' AS dataset_key, s.grid_key, s.site_key, s.cruise_key,
CAST(s.order_occ AS INTEGER) AS order_occ, s.latitude, s.longitude,
CAST(t.datetime_start_utc AS TIMESTAMP) AS datetime,
0::DOUBLE AS depth_min_m, CAST(nd.depth_max_m AS DOUBLE) AS depth_max_m,
t.tow_type_key AS tow_type,
NULL::VARCHAR AS data_stage,
t.tow_uuid AS source_uuid
FROM tow t JOIN site s USING (site_uuid)
LEFT JOIN (SELECT tow_uuid, MAX(net_depth_m) AS depth_max_m
FROM net GROUP BY 1) nd USING (tow_uuid)"))
append_sample(con, glue("
SELECT {ns_key(ds_key, 'net', 'n.net_uuid')} AS sample_key, 'net' AS sample_type,
{ns_key(ds_key, 'tow', 'n.tow_uuid')} AS parent_sample_key,
{ns_key(ds_key, 'site', 't.site_uuid')} AS root_sample_key,
'{ds_key}' AS dataset_key, s.grid_key, s.site_key, s.cruise_key,
CAST(s.order_occ AS INTEGER) AS order_occ, s.latitude, s.longitude,
CAST(t.datetime_start_utc AS TIMESTAMP) AS datetime,
0::DOUBLE AS depth_min_m, CAST(n.net_depth_m AS DOUBLE) AS depth_max_m,
t.tow_type_key AS tow_type,
NULL::VARCHAR AS data_stage,
n.net_uuid AS source_uuid
FROM net n JOIN tow t USING (tow_uuid) JOIN site s USING (site_uuid)"))
# the NetDepth projection landed on exactly the tows and nets that carry one
d_depth <- dbGetQuery(con, "
SELECT sample_type, COUNT(*) AS n, COUNT(depth_max_m) AS n_depth_max
FROM sample GROUP BY 1 ORDER BY 1")
stopifnot(
"every tow with a NetDepth carries it as depth_max_m" =
d_depth$n_depth_max[d_depth$sample_type == "tow"] == sum(d_netdepth$n_tow_depth),
"every net with a NetDepth carries it as depth_max_m" =
d_depth$n_depth_max[d_depth$sample_type == "net"] ==
dbGetQuery(con, "SELECT COUNT(net_depth_m) FROM net")[[1]],
"a site has no depth of its own" =
d_depth$n_depth_max[d_depth$sample_type == "site"] == 0)
cat(glue("depth_max_m from NetDepth: ",
"{paste(sprintf('%s %s of %s', d_depth$sample_type, format(d_depth$n_depth_max, big.mark=','), format(d_depth$n, big.mark=',')), collapse = '; ')}"), "\n")
# obs — the abundance headline: BASE rows only (measurement_type IS NULL in the
# source long table). taxon_key resolves through dataset_taxon on species_id —
# the global worms:/itis: key, not the dataset-local species_id. Aggregated by
# AphiaID (swfsc_ichthyo_04/_05/_13): codes that share a taxon_key in one net are
# SUMMED into one row, so (sample_key, taxon_key, life_stage, measurement_type) is
# unique -- e.g. 683 Sebastes + 3023 S. crocotulus (both AphiaID 126175), and 788
# S. leptorhynchus + 792 S. californiensis (both 275218). Each code keeps its own
# dataset_taxon row and verbatim name; only the counts merge.
append_obs(con, glue("
SELECT 'bio', '{ds_key}', {ns_key(ds_key, 'net', 'i.net_uuid')},
s.grid_key, s.cruise_key, s.latitude, s.longitude,
CAST(t.datetime_start_utc AS TIMESTAMP), NULL::DOUBLE, NULL::DOUBLE,
dt.taxon_key, i.life_stage,
'abundance', CAST(SUM(i.tally) AS DOUBLE), NULL::VARCHAR, NULL::DOUBLE
FROM ichthyo i JOIN net n USING (net_uuid) JOIN tow t USING (tow_uuid)
JOIN site s USING (site_uuid)
LEFT JOIN dataset_taxon dt ON dt.dataset_key = '{ds_key}'
AND dt.ds_taxa_code = CAST(i.species_id AS VARCHAR)
WHERE i.measurement_type IS NULL
GROUP BY i.net_uuid, s.grid_key, s.cruise_key, s.latitude, s.longitude,
t.datetime_start_utc, dt.taxon_key, i.life_stage"))
# obs_attribute — the size and stage distributions UNDER that headline. `size`
# takes the canonical name `body_length`; a `stage` bin also carries the human
# label from `lookup`. Summed by AphiaID exactly as obs (and, for larvae, already
# summed over the LarvaeMeasured rows of one length or one stage).
append_obs_attribute(con, glue("
SELECT '{ds_key}', {ns_key(ds_key, 'net', 'i.net_uuid')},
dt.taxon_key, i.life_stage,
CASE i.measurement_type WHEN 'size' THEN 'body_length' ELSE i.measurement_type END,
i.measurement_value,
CASE WHEN i.measurement_type = 'stage' THEN lk.description ELSE NULL END,
CAST(SUM(i.tally) AS INTEGER), NULL::VARCHAR
FROM ichthyo i
LEFT JOIN dataset_taxon dt ON dt.dataset_key = '{ds_key}'
AND dt.ds_taxa_code = CAST(i.species_id AS VARCHAR)
LEFT JOIN lookup lk ON lk.lookup_type = i.life_stage || '_stage'
AND lk.lookup_num = CAST(i.measurement_value AS INTEGER)
WHERE i.measurement_type IN ('stage','size')
GROUP BY i.net_uuid, dt.taxon_key, i.life_stage, i.measurement_type,
i.measurement_value, lk.description"))
# the published keys are unique (Ed Weber 2026-09-25: "you will have to aggregate");
# v2026.10.01 had 3 duplicated obs keys, the 2026-09-26 export 25 before this sum
n_obs_dup <- dbGetQuery(con, "
SELECT COUNT(*) FROM (SELECT sample_key, taxon_key, life_stage, measurement_type
FROM obs GROUP BY ALL HAVING COUNT(*) > 1)")[[1]]
n_attr_dup <- dbGetQuery(con, "
SELECT COUNT(*) FROM (SELECT sample_key, taxon_key, life_stage, measurement_type, bin_value
FROM obs_attribute GROUP BY ALL HAVING COUNT(*) > 1)")[[1]]
n_rows_src <- dbGetQuery(con, "
SELECT COUNT(*) FROM ichthyo i JOIN net n USING (net_uuid) JOIN tow t USING (tow_uuid)
JOIN site s USING (site_uuid) WHERE i.measurement_type IS NULL")[[1]]
stopifnot(
"obs (sample_key, taxon_key, life_stage, measurement_type) must be unique" = n_obs_dup == 0,
"obs_attribute (sample_key, taxon_key, life_stage, measurement_type, bin_value) must be unique" =
n_attr_dup == 0,
"every obs row has a taxon_key (else unresolved codes would be summed together)" =
dbGetQuery(con, "SELECT COUNT(*) FROM obs WHERE taxon_key IS NULL")[[1]] == 0,
"summing by AphiaID keeps every individual" =
dbGetQuery(con, "SELECT SUM(measurement_value) FROM obs")[[1]] ==
dbGetQuery(con, "SELECT SUM(i.tally) FROM ichthyo i JOIN net n USING (net_uuid)
JOIN tow t USING (tow_uuid) JOIN site s USING (site_uuid)
WHERE i.measurement_type IS NULL")[[1]])
n_obs_now <- dbGetQuery(con, "SELECT COUNT(*) FROM obs")[[1]]
cat(glue("AphiaID aggregation: {format(n_rows_src, big.mark = ',')} source count rows -> ",
"{format(n_obs_now, big.mark = ',')} obs rows ({n_rows_src - n_obs_now} summed into a ",
"row of the same net and AphiaID)\n"))
# fish — SWFSC's Fish table (Ed Weber 2026-09-26: "the Fish (i.e., grown out of
# larvae size) table"; Ben 2026-10-04: wanted, in lower snake_case). Counts per net
# and species code of fish past the larval stage. Published as its own table at
# the source grain -- (sample_key of the net, species_id), with the taxon_key the
# code resolves to -- and NOT projected into obs: no life_stage in the vocabulary
# says "grown out of larvae" exactly (juvenile, or juvenile and adult?), and 2,923
# of its net x species pairs also have a Larvae count, so whether a fish is also in
# that count must be known before an abundance could be summed (swfsc_ichthyo_20).
d_fish_orphan <- dbGetQuery(con, "
SELECT COUNT(*) FILTER (WHERE n.net_uuid IS NULL) AS no_net,
COUNT(*) FILTER (WHERE dt.taxon_key IS NULL) AS no_taxon
FROM fish f LEFT JOIN net n USING (net_uuid)
LEFT JOIN dataset_taxon dt ON dt.dataset_key = 'swfsc_ichthyo'
AND dt.ds_taxa_code = CAST(f.species_id AS VARCHAR)")
stopifnot(
"every fish row names a released net" = d_fish_orphan$no_net == 0,
"every fish species code resolves to a taxon_key" = d_fish_orphan$no_taxon == 0,
"fish (net, species code) is unique" =
dbGetQuery(con, "SELECT COUNT(*) FROM (SELECT net_uuid, species_id FROM fish
GROUP BY ALL HAVING COUNT(*) > 1)")[[1]] == 0,
"a fish tally is a positive count" =
dbGetQuery(con, "SELECT COUNT(*) FROM fish WHERE tally IS NULL OR tally <= 0")[[1]] == 0)
invisible(dbExecute(con, glue("
CREATE OR REPLACE TABLE fish_pub AS
SELECT {ns_key(ds_key, 'net', 'f.net_uuid')} AS sample_key,
CAST(f.species_id AS INTEGER) AS species_id, dt.taxon_key,
CAST(f.tally AS INTEGER) AS tally
FROM fish f
LEFT JOIN dataset_taxon dt ON dt.dataset_key = '{ds_key}'
AND dt.ds_taxa_code = CAST(f.species_id AS VARCHAR)
ORDER BY sample_key, species_id")))
invisible(dbExecute(con, "DROP TABLE fish"))
invisible(dbExecute(con, "ALTER TABLE fish_pub RENAME TO fish"))
d_fish <- dbGetQuery(con, "
SELECT COUNT(*) AS n, COUNT(DISTINCT sample_key) AS nets, COUNT(DISTINCT species_id) AS codes,
SUM(tally) AS fish FROM fish")
cat(glue("fish: {format(d_fish$n, big.mark = ',')} rows, {format(d_fish$fish, big.mark = ',')} fish ",
"on {format(d_fish$nets, big.mark = ',')} nets, {d_fish$codes} species codes\n"))
# net effort, as SWFSC gives it (swfsc_ichthyo_03; Ed Weber 2026-09-26: "you should
# trust us on the standard haul factor instead of calculating based on angles for
# every 10-m of wire"). Nothing here recomputes it: std_haul_factor is Net.Shf,
# volume_sampled Net.Vol, prop_sorted Net.PropSorted, verbatim. "They cannot be
# missing": a net without one stops the render (a question back to SWFSC, never a
# silent drop). A proportion sorted outside (0, 1] is reported, not changed.
d_effort <- dbGetQuery(con, "
SELECT COUNT(*) AS n_net,
COUNT(*) FILTER (WHERE volume_sampled IS NULL) AS no_volume,
COUNT(*) FILTER (WHERE standard_haul_factor IS NULL) AS no_shf,
COUNT(*) FILTER (WHERE prop_sorted IS NULL) AS no_prop_sorted,
COUNT(*) FILTER (WHERE volume_sampled <= 0 OR standard_haul_factor <= 0) AS not_positive,
COUNT(*) FILTER (WHERE prop_sorted <= 0 OR prop_sorted > 1) AS prop_sorted_out
FROM net")
stopifnot(
"net effort cannot be missing (swfsc_ichthyo_03): volume, standard haul factor, proportion sorted" =
d_effort$no_volume + d_effort$no_shf + d_effort$no_prop_sorted == 0,
"volume sampled and standard haul factor are positive" = d_effort$not_positive == 0)
d_prop_out <- dbGetQuery(con, "
SELECT n.net_uuid, t.tow_type_key, s.cruise_key, s.site_key, n.prop_sorted,
n.volume_sampled, n.standard_haul_factor
FROM net n JOIN tow t USING (tow_uuid) JOIN site s USING (site_uuid)
WHERE n.prop_sorted <= 0 OR n.prop_sorted > 1")
# true zeros (swfsc_ichthyo_09; Ed Weber: "The others are true zeros and have valid
# positive zooplankton volumes"). obs stays positive-only by design: a tow with no
# egg, larva or invertebrate row is a ZERO for every taxon, implied by its `sample`
# rows and their effort, which every tow and net keeps (the denominator). Only the
# two staging cruises were removed (exclude_staging_cruises); nothing marks a tow
# "unsorted". Zooplankton volume is measured on the oblique tows (C1, CB) only.
d_zero <- dbGetQuery(con, "
WITH caught AS (SELECT DISTINCT n.tow_uuid FROM ichthyo i JOIN net n USING (net_uuid)),
z AS (
SELECT t.tow_uuid, t.tow_type_key, COUNT(n.net_uuid) AS n_net,
MIN(n.volume_sampled) AS vol_min, MAX(n.totalplankton) AS tp_max
FROM tow t LEFT JOIN net n USING (tow_uuid)
WHERE t.tow_uuid NOT IN (SELECT tow_uuid FROM caught)
GROUP BY 1, 2)
SELECT tow_type_key, COUNT(*) AS n_zero_tow,
COUNT(*) FILTER (WHERE n_net = 0) AS no_net,
COUNT(*) FILTER (WHERE n_net > 0 AND vol_min > 0) AS volume_positive,
COUNT(*) FILTER (WHERE n_net > 0 AND tp_max > 0) AS zooplankton_positive,
COUNT(*) FILTER (WHERE n_net > 0 AND tp_max IS NULL) AS zooplankton_missing
FROM z GROUP BY 1 ORDER BY 1")
stopifnot(
"every zero-catch tow that has a net has a positive volume sampled (true zero, in the denominator)" =
all(d_zero$volume_positive == d_zero$n_zero_tow - d_zero$no_net))
d_zero |>
datatable(caption = "Zero-catch tows (no egg, larva or invertebrate row): true zeros, by tow type")
cat(glue(
"true zeros: {format(sum(d_zero$n_zero_tow), big.mark = ',')} of ",
"{format(dbGetQuery(con, 'SELECT COUNT(*) FROM tow')[[1]], big.mark = ',')} tows caught nothing; ",
"all {format(sum(d_zero$volume_positive), big.mark = ',')} with a net have a positive volume; ",
"{sum(d_zero$no_net)} tow(s) have no net at all (no effort: not in any density denominator); ",
"oblique (C1/CB) zero tows with no zooplankton volume: ",
"{sum(d_zero$zooplankton_missing[d_zero$tow_type_key %in% c('C1', 'CB')])}\n",
"net effort: {format(d_effort$n_net, big.mark = ',')} nets, none missing; ",
"{nrow(d_prop_out)} with prop_sorted outside (0, 1]\n"))
if (nrow(d_prop_out)) d_prop_out |> datatable(caption = "Nets whose proportion sorted is outside (0, 1] (reported to SWFSC)")
# sample_measurement — the five net-level effort quantities, long-formatted
append_sample_measurement(con, glue("
SELECT {ns_key(ds_key, 'net', 'net_uuid')}, '{ds_key}',
mt, mv, NULL::VARCHAR
FROM (
SELECT net_uuid, 'volume_sampled' mt, volume_sampled mv FROM net UNION ALL
SELECT net_uuid, 'std_haul_factor', standard_haul_factor FROM net UNION ALL
SELECT net_uuid, 'prop_sorted', prop_sorted FROM net UNION ALL
SELECT net_uuid, 'small_plankton_biomass', smallplankton FROM net UNION ALL
SELECT net_uuid, 'total_plankton_biomass', totalplankton FROM net)
WHERE mv IS NOT NULL"))
core <- list(
sample = dbGetQuery(con, "SELECT COUNT(*) FROM sample")[[1]],
obs = dbGetQuery(con, "SELECT COUNT(*) FROM obs")[[1]],
obs_attribute = dbGetQuery(con, "SELECT COUNT(*) FROM obs_attribute")[[1]],
sample_measurement = dbGetQuery(con, "SELECT COUNT(*) FROM sample_measurement")[[1]],
taxon = n_pruned$taxon,
dataset_taxon = n_pruned$dataset_taxon)
cat(glue(
"core projection — sample={core$sample %||% 0} obs={core$obs %||% 0} ",
"obs_attribute={core$obs_attribute %||% 0} ",
"sample_measurement={core$sample_measurement %||% 0} ",
"taxon={core$taxon %||% 0} dataset_taxon={core$dataset_taxon %||% 0}\n"))
cat(glue("taxa shard: {n_taxon} taxon rows built, {core$taxon} kept after pruning ",
"to this dataset's vocabulary + its lineage ancestors"), "\n")
# the three event levels must each survive at their own grain, and the
# abundance headline must not absorb the size/stage rows
n_site <- dbGetQuery(con, "SELECT COUNT(*) FROM sample WHERE sample_type='site'")[[1]]
n_tow <- dbGetQuery(con, "SELECT COUNT(*) FROM sample WHERE sample_type='tow'")[[1]]
n_net <- dbGetQuery(con, "SELECT COUNT(*) FROM sample WHERE sample_type='net'")[[1]]
d_attr <- dbGetQuery(con, "
SELECT measurement_type, COUNT(*) n FROM obs_attribute GROUP BY 1 ORDER BY 1")
stopifnot(
"sample must hold one row per site" = n_site == dbGetQuery(con, "SELECT COUNT(DISTINCT site_uuid) FROM site")[[1]],
"sample must hold one row per tow" = n_tow == dbGetQuery(con, "SELECT COUNT(DISTINCT tow_uuid) FROM tow")[[1]],
"sample must hold one row per net" = n_net == dbGetQuery(con, "SELECT COUNT(DISTINCT net_uuid) FROM net")[[1]],
"sample_key must be globally unique" =
dbGetQuery(con, "SELECT COUNT(*) FROM (SELECT sample_key FROM sample
GROUP BY 1 HAVING COUNT(*) > 1)")[[1]] == 0,
"obs must carry only the abundance headline (one row per net x AphiaID x life stage)" =
core$obs ==
dbGetQuery(con, "SELECT COUNT(*) FROM (
SELECT DISTINCT i.net_uuid, dt.taxon_key, i.life_stage
FROM ichthyo i JOIN net n USING (net_uuid)
JOIN tow t USING (tow_uuid) JOIN site s USING (site_uuid)
LEFT JOIN dataset_taxon dt ON dt.dataset_key = 'swfsc_ichthyo'
AND dt.ds_taxa_code = CAST(i.species_id AS VARCHAR)
WHERE i.measurement_type IS NULL)")[[1]],
"obs must carry no size/stage rows" =
dbGetQuery(con, "SELECT COUNT(*) FROM obs
WHERE measurement_type <> 'abundance'")[[1]] == 0,
"obs_attribute must hold exactly body_length + stage" =
setequal(d_attr$measurement_type, c("body_length", "stage")),
"the net -> tow -> site chain must resolve" =
dbGetQuery(con, "SELECT COUNT(*) FROM sample c
LEFT JOIN sample p ON c.parent_sample_key = p.sample_key
WHERE c.parent_sample_key IS NOT NULL AND p.sample_key IS NULL")[[1]] == 0,
"every sample_measurement.sample_key must resolve in sample" =
dbGetQuery(con, "SELECT COUNT(*) FROM sample_measurement m
LEFT JOIN sample s USING (sample_key)
WHERE s.sample_key IS NULL")[[1]] == 0,
"every obs.taxon_key must RESOLVE in taxon (not merely be non-NULL)" =
dbGetQuery(con, "SELECT COUNT(*) FROM obs o LEFT JOIN taxon t USING (taxon_key)
WHERE o.taxon_key IS NOT NULL AND t.taxon_key IS NULL")[[1]] == 0,
"every taxon.parent_taxon_key must resolve (the lineage chain)" =
dbGetQuery(con, "SELECT COUNT(*) FROM taxon c LEFT JOIN taxon p
ON c.parent_taxon_key = p.taxon_key
WHERE c.parent_taxon_key IS NOT NULL AND p.taxon_key IS NULL")[[1]] == 0,
# source_uuid (calcofi4db >= 3.32.0, WS-B): every site/tow/net row must carry
# the provider's own UUID verbatim — a gap here would mean the append_sample()
# arm's 17th column silently stopped projecting it
"every site/tow/net sample row must carry source_uuid" =
dbGetQuery(con, "SELECT COUNT(*) FROM sample
WHERE sample_type IN ('site','tow','net') AND source_uuid IS NULL")[[1]] == 0)
cat(glue("sample: {format(n_site, big.mark=',')} sites / {format(n_tow, big.mark=',')} tows / ",
"{format(n_net, big.mark=',')} nets; obs_attribute ",
"{paste(sprintf('%s=%s', d_attr$measurement_type, format(d_attr$n, big.mark=',')), collapse=', ')}"), "\n")
# UUID check (calcofi4db >= 3.32.0, WS-B / design memo D2): site.cruise_uuid is
# the ONLY place a real cruise_uuid<->cruise_key link survives past this chunk —
# the compat block just below replaces `site` with a VIEW carrying only
# cruise_key (not cruise_uuid), and the release's `sample` table has no
# cruise_uuid column at all (a 1.3M-row denormalization to serve one check
# would be worse than the check). So this assertion runs HERE, once, while both
# columns still coexist, and its result travels in manifest.json
# (`mismatches$cruise_uuid`, written below) for check_cruise_key_integrity()
# (calcofi4db::check_cruise_key_integrity(), release_database.qmd) to gate on.
n_cruise_uuid_mismatch <- dbGetQuery(con, "
SELECT COUNT(*) AS n FROM site s JOIN cruise c ON c.cruise_uuid = s.cruise_uuid
WHERE s.cruise_key IS DISTINCT FROM c.cruise_key OR s.cruise_key IS NULL")[[1]]
stopifnot(
"every ichthyo site's cruise_key must agree with the cruise row its cruise_uuid names" =
n_cruise_uuid_mismatch == 0)
cat(glue("cruise_uuid <-> cruise_key check: {n_cruise_uuid_mismatch} mismatch(es)\n"))
# serve the retired per-dataset names as VIEWs over the core: the source id comes
# back out of the namespaced sample_key, the containment FK out of
# parent_sample_key, and the net effort columns by pivoting sample_measurement
# back out of long form. Exact for every column the core models, lossy for the
# rest (net.side, tow.tow_number, the legacy site columns) — which the release
# drops anyway. The real tables are DROPped: they are 213k/459k rows and are no
# longer written to parquet.
compat <- list(
site = compat_event_sql(ds_key, "site", "site_uuid", NULL,
c(order_occ = "order_occ", longitude = "longitude", latitude = "latitude",
cruise_key = "cruise_key", geom = "geom", grid_key = "grid_key",
site_key = "site_key")),
tow = compat_event_sql(ds_key, "tow", "tow_uuid", "site_uuid",
c(tow_type_key = "tow_type", datetime_start_utc = "datetime")),
net = compat_event_sql(ds_key, "net", "net_uuid", "tow_uuid", character(),
c(standard_haul_factor = "std_haul_factor", volume_sampled = "volume_sampled",
prop_sorted = "prop_sorted", smallplankton = "small_plankton_biomass",
totalplankton = "total_plankton_biomass")))
for (nm in names(compat)) {
# DROP has to know which it is: DuckDB refuses "DROP TABLE" on a view (and
# vice versa), and a re-run inside one session finds the view already there
t <- dbGetQuery(con, glue(
"SELECT table_type FROM information_schema.tables WHERE table_name = '{nm}'"))
if (nrow(t)) {
kind <- if (grepl("VIEW", t$table_type[1], ignore.case = TRUE)) "VIEW" else "TABLE"
invisible(dbExecute(con, glue('DROP {kind} IF EXISTS "{nm}"')))
}
invisible(dbExecute(con, glue("CREATE OR REPLACE VIEW {nm} AS {compat[[nm]]}")))
}
n_compat <- sapply(names(compat), function(nm)
dbGetQuery(con, glue("SELECT COUNT(*) FROM {nm}"))[[1]])
stopifnot(
"compat views must reconstruct the source grains exactly" =
all(n_compat[c("site", "tow", "net")] == c(n_site, n_tow, n_net)))
cat(glue("compat views over core: ",
"{paste(sprintf('%s=%s', names(n_compat), format(n_compat, big.mark=',')), collapse=', ')}"), "\n")
```
## Write Parquet Outputs
Export tables to parquet files for downstream use.
```{r}
#| label: write_parquet
# collect mismatches for manifest
mismatches <- list(
ships = collect_ship_mismatches(con, "cruise"),
cruise_keys = collect_cruise_key_mismatches(con, "cruise"),
# calcofi4db >= 3.32.0 (WS-B / design memo D2): site.cruise_uuid <-> cruise.cruise_key
# agreement, computed in the emit_core chunk while site still carried
# cruise_uuid (the compat VIEW below drops it). release_database.qmd's
# check_cruise_key_integrity() reads this and fails the release if it is
# missing or non-zero.
cruise_uuid = tibble(n_mismatch = n_cruise_uuid_mismatch)
)
# write parquet files with manifest
# core shards + the shared reference tables this ingest owns for the whole
# database (grid/cruise/ship/lookup), and the dataset's own `fish` table. The
# per-dataset site/tow/net/ichthyo tables are no longer written — they are VIEWs
# over the core now.
tbls_out <- core_output_tables(
con, extra = c("grid", "cruise", "ship", "lookup", "dataset", "fish"))
parquet_stats <- write_parquet_outputs(
con = con,
output_dir = dir_parquet,
tables = tbls_out,
sort_by = list(
obs = c("grid_key", "measurement_type"),
sample = "hilbert:longitude,latitude"),
strip_provenance = FALSE,
mismatches = mismatches
)
parquet_stats |>
mutate(file = basename(path)) |>
select(-path) |>
datatable(caption = "Parquet export statistics")
```
## Write Metadata
Build `metadata.json` sidecar file documenting all tables and columns in parquet outputs. DuckDB `COMMENT ON` does not propagate to parquet, so this provides the metadata externally.
```{r}
#| label: write_metadata
metadata_path <- build_metadata_json(
con = con,
d_tbls_rd = d$d_tbls_rd,
d_flds_rd = d$d_flds_rd,
metadata_derived_csv = c(
here("metadata/core_dictionary.csv"),
here("metadata/swfsc/ichthyo/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
)
# write relationships.json sidecar with PKs/FKs
# `fish` (this dataset's own table) is declared in core_relationships() (calcofi4db
# dev): key (sample_key, species_id), FKs to the net in `sample` and to `taxon`
stopifnot("calcofi4db must declare fish's key (core_relationships)" =
identical(core_relationships("fish")$primary_keys$fish, c("sample_key", "species_id")))
build_relationships_json(
rels = core_relationships(tbls_out),
output_dir = dir_parquet,
provider = provider,
dataset = dataset
)
# show metadata summary
metadata <- jsonlite::fromJSON(metadata_path)
tibble(
table = names(metadata$tables),
n_cols = map_int(
names(metadata$tables),
~ sum(grepl(glue("^{.x}\\."), names(metadata$columns)))
),
name_long = map_chr(metadata$tables, ~ .x$name_long)
) |>
datatable(caption = "Table metadata summary")
```
```{r}
#| label: show_metadata_json
listviewer::jsonedit(
jsonlite::fromJSON(metadata_path, simplifyVector = FALSE),
mode = "view")
```
```{r}
#| label: show_relationships_json
listviewer::jsonedit(
jsonlite::fromJSON(
file.path(dir_parquet, "relationships.json"),
simplifyVector = FALSE),
mode = "view")
```
## Upload to GCS Archive
Upload parquet files, manifest, and metadata sidecar to `gs://calcofi-db/ingest/{provider}_{dataset}/`.
```{r}
#| label: upload_gcs
gcs_ingest_prefix <- glue("ingest/{dir_label}")
gcs_bucket <- "calcofi-db"
# sync to GCS — only uploads new or changed files
sync_results <- sync_to_gcs(
local_dir = dir_stage,
sidecar_dir = dir_parquet,
gcs_prefix = gcs_ingest_prefix,
bucket = gcs_bucket
)
```
## Cleanup
```{r}
#| label: cleanup
# close local wrangling database connection
close_duckdb(con)
message("Local wrangling database connection closed")
# note: parquet outputs are in data/parquet/swfsc_ichthyo/
# these can be used to update the Working DuckLake in a separate workflow
message(glue("Parquet outputs written to: {dir_parquet}"))
message(glue("GCS outputs at: gs://{gcs_bucket}/{gcs_ingest_prefix}/"))
```
## TODO
- [ ] Review documentation for more comment descriptions at [Data \> Data Formats \| CalCOFI.org](https://calcofi.com/index.php?option=com_content&view=category&id=73&Itemid=993)
::: {.callout-caution collapse="true"}
## Session Info
```{r session_info}
devtools::session_info()
```
:::
