---
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: "#ffa94d"
description: >
Ichthyoplankton (fish eggs and larvae) collected by bongo and manta net
tows on CalCOFI cruises, processed by SWFSC.
citation_main: "NOAA Fisheries SWFSC. CalCOFI Ichthyoplankton Database."
# /data/biology/ichthyoplankton/ 404s ("Page not found – CalCOFI"); the page
# lives under marine-ecosystem-data and is titled "Fish Eggs & Larvae"
link_calcofi_org: https://calcofi.org/data/marine-ecosystem-data/fish-eggs-larvae/
# 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"}
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.
### 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")
```
## 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")
```
## 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
)
# 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 applies data corrections and validates referential integrity.
### Data Corrections
Apply known data corrections identified by data managers.
```{r}
#| label: apply_corrections
# apply data corrections
apply_data_corrections(con, verbose = TRUE)
```
### 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
)
}
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"
)
# tables consolidated into lookup
deprecated_lookup <- c("tow_type")
# all deprecated tables
deprecated_tables <- c(deprecated_ichthyo, deprecated_lookup)
# 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: 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 <- build_dataset_taxon(con, mt_taxon, tx_over)
n_pruned <- prune_taxon_shard(con, ds_key)
# 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.
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
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"))
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, NULL::DOUBLE AS depth_max_m,
t.tow_type_key AS tow_type
FROM tow t JOIN site s USING (site_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, NULL::DOUBLE AS depth_max_m,
t.tow_type_key AS tow_type
FROM net n JOIN tow t USING (tow_uuid) JOIN site s USING (site_uuid)"))
# 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.
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(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"))
# 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`.
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,
i.tally, 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')"))
# 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" =
core$obs ==
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]],
"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)
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")
# 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")
)
# write parquet files with manifest
# core shards + the shared reference tables this ingest owns for the whole
# database (grid/cruise/ship/lookup). 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"))
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
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()
```
:::
