CalCOFI.io CalCOFI.io Workflows

Ingest SIO Cetacean eDNA Detections

Overview

Source: data/edna_data/edna.csv in CalCOFI/marmam-app, read from a pinned commit. One row per NOAA CalCOFI Genomics Program (NCOG) seawater sample screened for cetaceans: 136 rows over 12 cruises (2014-02 to 2016-11), mostly 10 m. Seven species columns (Gg, Tt, Mn, Dc, Lb, Lo, Dd) hold an x where the species’ DNA was detected. eDNA_Process says whether the sample was sequenced (71) or did not amplify (65, “not sequenced PCR negative”).

The app’s edna-processed.csv is not read (a re-typed copy). This is not calpoly_whale-edna (the Zenodo ASV tables behind a whale-density model).

  • Provider: sio (tentative; the samples are NCOG’s, the screening appears to be the Whale Acoustics Lab’s; Q03)
  • Grain: one water sample (sample_label, the NCOG name YYYYMM_LLL.L_SSS.S_depth)
  • Occurrence: edna_presence 1/0 for each of the seven species on every sequenced sample: 1 = the species’ DNA was detected, 0 = it was not detected in a sample that amplified and was sequenced (a non-detection, not a confirmed absence). A PCR-negative sample (“not sequenced PCR negative”) did not amplify, so it says nothing about any species: it is kept as a sample row with no obs until the provider answers Q04 (Ben, 2026-10-01)
  • Time: the table gives none. NCOG filters its eDNA from the CTD-rosette Niskin bottles, so each sample takes the time and cruise of its calcofi_bottle cast: the one cast at the same station on the cruise of the label’s month, found with match_by_site_datetime(), then the Niskin bottle nearest the sample’s depth with match_nearest_by_depth() (Q06)

graph LR
  R[edna.csv<br/>one row per sample, 7 species columns] --> S[edna_sample]
  R --> D[edna_detection<br/>sample x species, 0/1]
  D -.sample_label.-> S
  B[calcofi_bottle<br/>cast + Niskin bottle] -.site + cruise month, depth.-> S

Setup

devtools::load_all(here::here("../calcofi4db"))
devtools::load_all(here::here("../calcofi4r"))
librarian::shelf(
  CalCOFI/calcofi4db, CalCOFI/calcofi4r,
  DBI, dplyr, DT, fs, glue, here, jsonlite, knitr,
  lubridate, purrr, readr, sf, stringr, tibble, tidyr, quiet = T)
options(readr.show_col_types = F)
options(DT.options = list(scrollX = TRUE))
source(here("libs/ingest.R"))

cc           <- read_calcofi_meta(here("ingest_sio_cetacean-edna.qmd"))
provider     <- cc$provider
dataset      <- cc$dataset
tables_owned <- cc$tables_owned
ds_key       <- as.character(glue("{provider}_{dataset}"))
dir_label    <- ds_key
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"))
dir_meta     <- here(glue("metadata/{provider}/{dataset}"))
dir_dl       <- here(glue("data/cache/{dir_label}")); dir_create(dir_dl)
publish_to_gcs <- Sys.getenv("CALCOFI_SKIP_GCS") == ""

if (overwrite) {
  if (file_exists(db_path))                 file_delete(db_path)
  if (file_exists(paste0(db_path, ".wal"))) file_delete(paste0(db_path, ".wal"))
}
dir_create(dirname(db_path))
con <- get_duckdb_con(db_path)
load_duckdb_extension(con, "spatial")

Read Source Data

# CalCOFI/marmam-app at a pinned commit (see ingest_sio_cetacean-sightings.qmd)
src_sha  <- "7bc0e18d7cb0ac7ba58d7bc4d47e57c50bd9bc80"
src_url  <- glue("https://raw.githubusercontent.com/CalCOFI/marmam-app/{src_sha}/data/edna_data/edna.csv")
src_file <- file.path(dir_dl, "edna.csv")
downloaded <- overwrite_all || !file_exists(src_file)
if (downloaded) download.file(src_url, src_file, quiet = TRUE, mode = "wb")
src_stamp <- stamp_source_access(files = src_file) |>
  mutate(source = as.character(src_url), method = if (downloaded) "download" else "file_mtime")

if (publish_to_gcs) {
  sync_to_gcs(local_dir = dir_dl, gcs_prefix = glue("archive/{provider}/{dataset}"),
              bucket = "calcofi-files-public")
} else {
  cat("CALCOFI_SKIP_GCS set -- source NOT archived to gs://calcofi-files-public\n")
}
rsync /Users/bbest/Github/CalCOFI/.worktrees/workflows-pr117/data/cache/sio_cetacean-edna -> gs://calcofi-files-public/archive/sio/cetacean-edna
Sync complete: 1 copied, 0 removed (rsync)
# A tibble: 1 × 4
  file    action    size reason        
  <chr>   <chr>    <dbl> <chr>         
1 <rsync> uploaded    NA parallel rsync
d_raw <- read_csv(src_file, col_types = cols(.default = "c"))
sp_cols <- c("Gg", "Tt", "Mn", "Dc", "Lb", "Lo", "Dd")
stopifnot(
  "row count changed at this commit" = nrow(d_raw) == 136L,
  "the seven species columns are present" = all(sp_cols %in% names(d_raw)),
  "a species cell is 'x' or blank" = all(unlist(d_raw[sp_cols]) %in% c("x", NA)))
cat(glue("edna.csv: {nrow(d_raw)} rows, {sum(unlist(d_raw[sp_cols]) %in% 'x')} detections"), "\n")
edna.csv: 136 rows, 15 detections

Build Sample + Detection Tables

Three samples are listed twice, once sequenced and once PCR-negative. The sequenced row is kept (Q05): it is the one that went further through the pipeline, and two of the three carry a detection.

num <- function(x) suppressWarnings(as.numeric(x))

dupes <- d_raw |> group_by(Sample) |> filter(n() > 1) |> ungroup()
stopifnot(
  "exactly three samples are listed twice" = n_distinct(dupes$Sample) == 3L && nrow(dupes) == 6L,
  "each duplicate pair is one sequenced + one PCR-negative row" =
    all(table(dupes$eDNA_Process) == 3L))
d_dedup <- d_raw |>
  arrange(Sample, eDNA_Process != "sequenced") |>
  distinct(Sample, .keep_all = TRUE)

d_sample <- d_dedup |>
  transmute(
    sample_label  = Sample,
    edna_process  = eDNA_Process,
    illumina_name = Illumina_name,
    station_name,
    season        = str_trim(season),
    # line/station from the NCOG sample name (YYYYMM_LLL.L_SSS.S_depth), not the
    # line/station columns: on 7 rows those disagree with the name, and the
    # row's own latitude/longitude sit on the name's station (asserted below)
    line_src      = num(line),
    station_src   = num(station),
    line          = num(substr(Sample, 8, 12)),
    station       = num(substr(Sample, 14, 18)),
    site_key      = normalize_site_key(sprintf("%.1f %.1f", line, station)),
    depth_m       = num(depth),
    latitude      = num(latitude),
    longitude     = num(longitude),
    cruise_label  = cruise,
    # no sampling date in the source: the 15th of the cruise month (CCYYMM) is
    # the anchor the bottle-cast match searches around (Q06)
    nominal_date  = make_date(2000L + as.integer(substr(cruise, 3, 4)),
                              as.integer(substr(cruise, 5, 6)), 15L),
    # the match key: station + the cruise's designated year-month, which is
    # how a cruise_key begins (YYYY-MM-NODC)
    site_ym       = paste(site_key, format(nominal_date, "%Y-%m")))
xy_name <- cc_calcofi_to_lonlat(d_sample$line, d_sample$station)
n_ls_fix <- sum(d_sample$line != d_sample$line_src | d_sample$station != d_sample$station_src)
stopifnot(
  "every sample name is YYYYMM_LLL.L_SSS.S_depth" =
    all(grepl("^\\d{6}_\\d{3}\\.\\d_\\d{3}\\.\\d_\\d+(\\.\\d+)?$", d_sample$sample_label)),
  "the name's station is where the row's own position is (within 0.02 deg)" =
    all(abs(xy_name$latitude - d_sample$latitude) < 0.02 &
        abs(xy_name$longitude - d_sample$longitude) < 0.02),
  "the name's depth is the depth column" =
    all(as.numeric(sub(".*_", "", d_sample$sample_label)) == d_sample$depth_m),
  "sample_label is unique"   = !anyDuplicated(d_sample$sample_label),
  "every sample has a station, depth and position" =
    !anyNA(d_sample$site_key) && !anyNA(d_sample$depth_m) &&
    !anyNA(d_sample$latitude) && !anyNA(d_sample$longitude),
  "cruise_label parses to a date" = !anyNA(d_sample$nominal_date),
  "the label's YYYYMM matches the sample name's" =
    all(substr(d_sample$sample_label, 3, 6) == substr(d_sample$cruise_label, 3, 6)))
dbWriteTable(con, "edna_sample", d_sample, overwrite = TRUE)

d_detection <- d_dedup |>
  select(sample_label = Sample, all_of(sp_cols)) |>
  pivot_longer(-sample_label, names_to = "species_code", values_to = "cell") |>
  transmute(sample_label, species_code, detected = as.integer(cell %in% "x"))
dbWriteTable(con, "edna_detection", d_detection, overwrite = TRUE)

cat(glue("line/station taken from the sample name; the line/station columns disagree on ",
         "{n_ls_fix} rows (Q10)"), "\n")
line/station taken from the sample name; the line/station columns disagree on 7 rows (Q10) 
cat(glue("edna_sample: {nrow(d_sample)} samples ",
         "({sum(d_sample$edna_process == 'sequenced')} sequenced); ",
         "edna_detection: {nrow(d_detection)} rows, {sum(d_detection$detected)} detections"), "\n")
edna_sample: 133 samples (71 sequenced); edna_detection: 931 rows, 15 detections

Resolve cruise_key, time + Spatial

NCOG eDNA is filtered from the CTD-rosette Niskin bottles, so a sample’s time and cruise are those of its bottle cast, not of the net tow at the same station. match_by_site_datetime() finds, for each sample, the calcofi_bottle cast at the same site_key on the cruise of the label’s month (key site_ym; exactly one cast per sample at this commit, asserted) nearest the 15th of that month, and match_nearest_by_depth() then finds the Niskin bottle on that cast nearest the sample’s depth (within 5 m). The sample takes the cast’s cruise_key and time. Every one of the 133 samples has its cast, including the 16 on CC1611, a Sally Ride cruise with no net tows (Q06).

The matched bottle is kept in the source table as evidence (it also confirms the 515 m and 170 m depths, Q09). It is not set as a cross-dataset parent_sample_key; that is a modelling decision still open.

bottle_parquet <- cc_stage_path("parquet", "calcofi_bottle", "sample.parquet")
dbExecute(con, glue(
  "CREATE OR REPLACE TABLE bottle_cast AS
   SELECT sample_key AS cast_sample_key, site_key, cruise_key, datetime,
          site_key || ' ' || substr(cruise_key, 1, 7) AS site_ym
   FROM read_parquet('{bottle_parquet}')
   WHERE sample_type = 'cast' AND site_key IS NOT NULL
     AND cruise_key IS NOT NULL AND datetime IS NOT NULL"))
[1] 35595
dbExecute(con, glue(
  "CREATE OR REPLACE TABLE bottle_niskin AS
   SELECT sample_key AS bottle_sample_key, parent_sample_key AS cast_sample_key,
          depth_max_m AS depth_m
   FROM read_parquet('{bottle_parquet}')
   WHERE sample_type = 'bottle' AND depth_max_m IS NOT NULL
     AND parent_sample_key IN (
       SELECT cast_sample_key FROM bottle_cast
       WHERE site_ym IN (SELECT site_ym FROM edna_sample))"))
[1] 2716
# match_by_site_datetime() takes the nearest cast by date and LIMIT 1, so a
# station occupied twice on one cruise would be a coin toss: assert there is none
n_multi_cast <- dbGetQuery(con, "
  SELECT COUNT(*) FROM (
    SELECT site_ym FROM bottle_cast WHERE site_ym IN (SELECT site_ym FROM edna_sample)
    GROUP BY 1 HAVING COUNT(*) > 1)")[[1]]
stopifnot("one bottle cast per station x cruise month among the eDNA samples" = n_multi_cast == 0)

# 45 days either side of the 15th: a cruise can begin late in the month before
# its designated month (CC1402's cast at 90.0 37.0 is 2014-01-29)
cast_match <- match_by_site_datetime(
  con, "edna_sample", "bottle_cast",
  fk_col = "cast_sample_key", ref_pk = "cast_sample_key",
  key_col = "site_ym", datetime_col = "nominal_date", ref_datetime_col = "datetime",
  window_days = 45)
match_by_site_datetime: 133/133 edna_sample rows matched to bottle_cast (100%) within 45 d
niskin_match <- match_nearest_by_depth(
  con, "edna_sample", "bottle_niskin",
  fk_col = "bottle_sample_key", ref_pk = "bottle_sample_key",
  parent_fk = "cast_sample_key", axis_col = "depth_m", tolerance = 5)
match_nearest_by_depth: 133/133 edna_sample rows matched to bottle_niskin (100%) within 5 depth_m
dbExecute(con, "ALTER TABLE edna_sample ADD COLUMN IF NOT EXISTS cruise_key VARCHAR")
[1] 0
dbExecute(con, "ALTER TABLE edna_sample ADD COLUMN IF NOT EXISTS datetime_start_utc TIMESTAMP")
[1] 0
dbExecute(con, "
  UPDATE edna_sample e SET cruise_key = c.cruise_key, datetime_start_utc = c.datetime
  FROM bottle_cast c WHERE c.cast_sample_key = e.cast_sample_key")
[1] 133
# the shared crosswalk (owned by ingest_sio_cetacean-sightings.qmd) is re-validated
# here rather than trusted: each key must be the label's year-month + its ship's
# NODC code, and where an eDNA label is in it (CC1611), the bottle casts must
# place the samples on that same cruise
d_ship  <- dbGetQuery(con, glue(
  "SELECT ship_key, ship_nodc FROM read_parquet('{cc_stage_path('parquet','swfsc_ichthyo','ship.parquet')}')"))
d_xwalk <- read_csv(here("metadata/sio/cetacean-sightings/cruise_label_crosswalk.csv"),
                    col_types = cols(.default = "c")) |>
  left_join(d_ship, by = "ship_key") |>
  mutate(minted = paste0("20", substr(cruise_label, 3, 4), "-", substr(cruise_label, 5, 6), "-", ship_nodc))
d_label_ck <- dbGetQuery(con, "
  SELECT cruise_label, COUNT(DISTINCT cruise_key) AS n_keys, ANY_VALUE(cruise_key) AS cruise_key,
         COUNT(*) AS n_samples
  FROM edna_sample GROUP BY 1 ORDER BY 1")
d_xw_edna <- d_label_ck |> inner_join(d_xwalk |> select(cruise_label, xwalk_key = cruise_key), by = "cruise_label")
stopifnot(
  "every sample has its bottle cast (a new sample without one must be looked at)" =
    cast_match$matched == cast_match$total,
  "every sample has a Niskin bottle within 5 m of its depth on that cast" =
    niskin_match$matched == nrow(d_sample),
  "every sample has a cruise_key and a time" =
    dbGetQuery(con, "SELECT COUNT(*) FROM edna_sample
                     WHERE cruise_key IS NULL OR datetime_start_utc IS NULL")[[1]] == 0,
  "a cruise label's samples sit on one cruise_key" = all(d_label_ck$n_keys == 1),
  "the cruise_key's year-month is the label's" =
    all(substr(d_label_ck$cruise_key, 1, 7) ==
          paste0("20", substr(d_label_ck$cruise_label, 3, 4), "-", substr(d_label_ck$cruise_label, 5, 6))),
  "every crosswalk ship is in the ship reference" = !anyNA(d_xwalk$ship_nodc),
  "each crosswalk cruise_key is the label's year-month + its ship's NODC code" =
    all(d_xwalk$minted == d_xwalk$cruise_key),
  "where the crosswalk names an eDNA label, the bottle casts agree with it" =
    all(d_xw_edna$cruise_key == d_xw_edna$xwalk_key))

d_depth_gap <- dbGetQuery(con, "
  SELECT e.sample_label, e.depth_m, n.depth_m AS bottle_depth_m,
         ABS(e.depth_m - n.depth_m) AS depth_diff_m
  FROM edna_sample e JOIN bottle_niskin n USING (bottle_sample_key)
  WHERE e.depth_m <> n.depth_m ORDER BY depth_diff_m DESC")
cat(glue("bottle cast matched for {cast_match$matched}/{cast_match$total} samples; ",
         "Niskin bottle within 5 m for {niskin_match$matched} ",
         "({nrow(d_depth_gap)} not at the exact depth, max {max(c(0, d_depth_gap$depth_diff_m))} m); ",
         "{nrow(d_label_ck)} cruise labels, {nrow(d_xw_edna)} of them in the shared crosswalk"), "\n")
bottle cast matched for 133/133 samples; Niskin bottle within 5 m for 133 (12 not at the exact depth, max 4 m); 12 cruise labels, 1 of them in the shared crosswalk 
d_label_ck |> datatable(caption = "Cruise label -> cruise_key, from the bottle casts", rownames = FALSE)
# spatial last: nothing above has to UPDATE a table that carries geometry
load_prior_tables(con, parquet_dir = cc_stage_path("parquet", "swfsc_ichthyo"),
                  tables = c("grid"), geom_tables = c("grid"), as_view = TRUE)
Loaded grid: 218 rows (VIEW) (GEOMETRY)
# A tibble: 1 × 3
  table  rows has_geom
  <chr> <dbl> <lgl>   
1 grid    218 TRUE    
add_point_geom(con, "edna_sample", lon_col = "longitude", lat_col = "latitude")
Added geom column to edna_sample table
assign_grid_key(con, "edna_sample")
Assigned grid_key to edna_sample: in_grid = 133
   status   n
1 in_grid 133

Load Dataset Metadata + Schema

this_provider <- provider; this_dataset <- dataset
d_dataset <- ingest_yaml_to_dataset_df(read_ingest_yaml(here())) |>
  filter(provider == this_provider, dataset == this_dataset)
stopifnot(nrow(d_dataset) == 1)
dbWriteTable(con, "dataset", as.data.frame(d_dataset), overwrite = TRUE)

d_meas_type <- read_measurement_type(here("metadata/measurement_type.csv"))
stopifnot("edna_presence is registered" = "edna_presence" %in% d_meas_type$measurement_type)
dbWriteTable(con, "measurement_type", as.data.frame(d_meas_type), overwrite = TRUE)

d_qual <- read_csv(here("metadata/measurement_qual.csv"), col_types = cols(.default = "c")) |>
  filter(code_set == "cetacean-edna")
stopifnot("every eDNA_Process value is a registered code" =
            all(d_sample$edna_process %in% d_qual$qual_code))

src_rels <- list(
  primary_keys = list(edna_sample = "sample_label"),
  foreign_keys = list(
    list(table = "edna_detection", column = "sample_label",
         ref_table = "edna_sample", ref_column = "sample_label")))
cc_erd(con, tables = c("edna_sample", "edna_detection", "dataset"), rels = src_rels,
       colors = list(lightblue = c("edna_sample", "edna_detection"), white = "dataset"))

Questions for Data Providers

questions_datatable(
  here(cc$questions_file),
  caption = "Questions for the cetacean eDNA data providers (ranked)")

Emit Core Tables

source core transform
edna_sample sample (water) every sample, sequenced or not; key sio_cetacean-edna:water:<sample_label>; depth_min_m = depth_max_m = depth_m; time and cruise of the matched bottle cast
edna_detection.detected obs edna_presence 1/0 for each of the seven species on every sequenced sample; measurement_qual = edna_process verbatim (always sequenced)
PCR-negative samples – held (Q04): a sample row, no obs; the 0s in the source are not published

Not published (kept in the archived source): illumina_name, station_name, season, the app’s detections / SpeciesName / SubOrder convenience columns, the matched bottle (bottle_sample_key, evidence only), and the species columns of the PCR-negative samples.

Taxon vocabulary

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)

# the same codes as the visual survey; their names come from its vocabulary
d_vocab <- read_csv(here("metadata/sio/cetacean-sightings/species_codes.csv"),
                    col_types = cols(.default = "c")) |>
  filter(species_code %in% sp_cols) |>
  transmute(ds_taxa_code = species_code, ds_scientific_name = scientific_name,
            ds_common_name = common_name)
stopifnot("all seven codes named" = setequal(d_vocab$ds_taxa_code, sp_cols))
n_staged <- append_dataset_taxon(con, ds_key, d_vocab)
ensure_taxon_xref(con, mt_taxon, tx_over, cache_csv = here("metadata/taxon_xref.csv"))
taxon xref: 0 TSN + 2 AphiaID + 5 name requested; 0 to fetch, 7 cached
taxon xref: 0 TSN + 5 AphiaID + 0 name requested; 0 to fetch, 5 cached
_taxon_xref: 12 cross-reference rows (12 with worms_id, 6 with itis_id, 0 re-keyed)
ensure_taxon_lineage(con, mt_taxon, tx_over, cache_csv = here("metadata/taxon_lineage.csv"))
taxon lineage: 7 WoRMS + 0 ITIS requested; 0 to fetch, 7 cached
taxon lineage: 0 Aves taxa keyed on ITIS, 7 taxa on WoRMS
taxon xref: 0 TSN + 27 AphiaID + 0 name requested; 0 to fetch, 27 cached
taxon: 27 lineage rows for 7 taxa (0 unresolved)
n_taxon    <- build_taxon_reference(con, mt_taxon, tx_over)
n_ds_taxon <- resolve_dataset_taxon(con, mt_taxon, tx_over)
taxon overrides ('sio_cetacean-edna'): 2 registry row(s) matched 2 vocabulary row(s): 2 applied, 0 skipped
n_tx_group <- build_taxon_group(con, read_taxon_group_rules(here("metadata/taxon_group.csv")))
chk <- check_dataset_taxon(con, ds_key, codes = sp_cols)
check_dataset_taxon('sio_cetacean-edna'): 7 taxa, 7 keyed by an authority, 0 local (0 allowed); 0 finding(s)
dt <- dbGetQuery(con, "SELECT ds_taxa_code, taxon_key FROM dataset_taxon WHERE dataset_key = ?",
                 params = list(ds_key))
key_of <- function(code) dt$taxon_key[dt$ds_taxa_code == code]
stopifnot(
  "every code keys a taxon"             = !anyNA(dt$taxon_key) && nrow(dt) == 7L,
  "Dc keys Delphinus capensis (Q08)"    = identical(key_of("Dc"), "worms:383816"),
  "Dc and Dd are distinct taxa (Q08)"   = !identical(key_of("Dc"), key_of("Dd")),
  "Lo keys Sagmatias obliquidens (Q08)" = identical(key_of("Lo"), "worms:1571909"))
cat(glue("vocabulary: {n_staged} codes; taxon {n_taxon}, dataset_taxon {n_ds_taxon}"), "\n")
vocabulary: 7 codes; taxon 27, dataset_taxon 7

Sample, obs

append_sample(con, sample_arm_self(
  ds_key, "edna_sample", "sample_label", "water",
  site_expr = "site_key", depth_min = "depth_m", depth_max = "depth_m"))
Loaded extension: spatial
# a PCR-negative sample did not amplify: its 0s are not non-detections, so it
# publishes no obs until the provider answers Q04 (Ben, 2026-10-01). Its sample
# row stays: the water was collected and screened, the result is what is held.
append_obs(con, glue("
  SELECT 'bio', '{ds_key}', {ns_key(ds_key, 'water', 's.sample_label')},
         s.grid_key, s.cruise_key, s.latitude, s.longitude,
         CAST(s.datetime_start_utc AS TIMESTAMP), s.depth_m, s.depth_m,
         dt.taxon_key, NULL::VARCHAR, 'edna_presence',
         CAST(d.detected AS DOUBLE), s.edna_process, NULL::DOUBLE
  FROM edna_detection d JOIN edna_sample s USING (sample_label)
  JOIN dataset_taxon dt ON dt.dataset_key = '{ds_key}' AND dt.ds_taxa_code = d.species_code
  WHERE s.edna_process = 'sequenced'"))
Loaded extension: h3
q1 <- function(sql) dbGetQuery(con, sql)[[1]]
n_seq <- sum(d_sample$edna_process == "sequenced")
n_pcr_neg_detect <- d_detection |>
  semi_join(d_sample |> filter(edna_process != "sequenced"), by = "sample_label") |>
  pull(detected) |> sum()
core <- list(
  sample            = q1("SELECT COUNT(*) FROM sample"),
  sample_sequenced  = n_seq,
  sample_held       = nrow(d_sample) - n_seq,
  obs               = q1("SELECT COUNT(*) FROM obs"),
  obs_present       = q1("SELECT COUNT(*) FROM obs WHERE measurement_value = 1"))
cat(glue("core projection -- ", paste(names(core), unlist(core), sep = "=", collapse = " ")), "\n")
core projection -- sample=133 sample_sequenced=71 sample_held=62 obs=497 obs_present=15 
stopifnot(
  "one sample per eDNA sample, PCR-negative ones included" = core$sample == nrow(d_sample),
  "seven obs per sequenced sample"        = core$obs == 7L * n_seq,
  "one obs per sample x taxon (Q08)"      =
    q1("SELECT COUNT(*) FROM (SELECT sample_key, taxon_key FROM obs GROUP BY ALL HAVING COUNT(*) > 1)") == 0,
  "the source has no detection on a PCR-negative sample" = n_pcr_neg_detect == 0,
  "presences = detections in the source"  = core$obs_present == sum(d_detection$detected),
  "a PCR-negative sample publishes no obs (held, Q04)" =
    q1("SELECT COUNT(*) FROM obs WHERE measurement_qual IS DISTINCT FROM 'sequenced'") == 0,
  "every obs.sample_key resolves in sample" =
    q1("SELECT COUNT(*) FROM obs o LEFT JOIN sample s USING (sample_key) WHERE s.sample_key IS NULL") == 0,
  "every obs.taxon_key resolves in taxon" =
    q1("SELECT COUNT(*) FROM obs o LEFT JOIN taxon t USING (taxon_key) WHERE t.taxon_key IS NULL") == 0)

Declared NULLs

results <- validate_for_release(con, checks = "all", strict = FALSE)
nullable <- tribble(
  ~table,   ~column,             ~n,                ~reason,
  "sample", "parent_sample_key", nrow(d_sample),    "single-level dataset: every sample is a root (the matched bottle is not set as a cross-dataset parent)",
  "sample", "grid_key",          q1("SELECT COUNT(*) FROM sample WHERE grid_key IS NULL"),
    "station outside the CalCOFI grid cells")
stopifnot(
  "no sample lacks a time (every one has its bottle cast)" =
    q1("SELECT COUNT(*) FROM sample WHERE datetime IS NULL") == 0,
  "no sample lacks a cruise_key" =
    q1("SELECT COUNT(*) FROM sample WHERE cruise_key IS NULL") == 0,
  "every sample is a root" =
    q1("SELECT COUNT(*) FROM sample WHERE parent_sample_key IS NULL") == nrow(d_sample),
  "the grid_key NULLs are counted, not assumed (a ratchet: none at this commit)" =
    nullable$n[nullable$column == "grid_key"] == 0)
cat("validate_for_release:", ifelse(results$passed, "PASSED", "reported findings -- reconciled against the declared NULLs below"), "\n")
validate_for_release: reported findings -- reconciled against the declared NULLs below 
nullable |> datatable(caption = "Declared NULLs (count and reason)", rownames = FALSE)

Measurement bounds

b <- check_measurement_bounds(con, "obs", mt = d_meas_type, dataset_key = ds_key,
                              depth_col = "depth_max_m")
bounds_datatable(b)
stopifnot("edna_presence is declared and respected" = all(b$status == "ok"))

Preview

dbGetQuery(con, "
  SELECT dt.ds_taxa_code, t.scientific_name,
         COUNT(*) FILTER (WHERE o.measurement_qual = 'sequenced') AS n_sequenced,
         SUM(o.measurement_value) AS n_detected
  FROM obs o JOIN taxon t USING (taxon_key)
  JOIN dataset_taxon dt ON dt.taxon_key = o.taxon_key AND dt.dataset_key = o.dataset_key
  GROUP BY ALL ORDER BY n_detected DESC") |>
  datatable(caption = "Detections by species", rownames = FALSE)
core_tbls <- core_output_tables(con)
cc_erd(con, tables = core_tbls, rels = core_relationships(core_tbls))

Write Outputs + Upload

dir_create(dir_parquet)
tbls_out <- core_output_tables(con, extra = c("measurement_type", "dataset"))
write_parquet_outputs(
  con = con, output_dir = dir_parquet, tables = tbls_out,
  sort_by = list(obs = c("grid_key", "measurement_type")),
  strip_provenance = FALSE)
Reused sample: 133 rows unchanged (no re-write/upload)
Reused obs: 497 rows unchanged (no re-write/upload)
Reused taxon: 27 rows unchanged (no re-write/upload)
Reused dataset_taxon: 7 rows unchanged (no re-write/upload)
Reused taxon_group: 7 rows unchanged (no re-write/upload)
Reused measurement_type: 220 rows unchanged (no re-write/upload)
Reused dataset: 1 rows unchanged (no re-write/upload)
Wrote manifest to /Users/bbest/Github/CalCOFI/.worktrees/workflows-pr117/data/parquet/sio_cetacean-edna/manifest.json
# A tibble: 7 × 5
  table             rows file_size path                              partitioned
  <chr>            <dbl>     <dbl> <chr>                             <lgl>      
1 sample             133      7604 /Users/bbest/_big/calcofi/parque… FALSE      
2 obs                497      8975 /Users/bbest/_big/calcofi/parque… FALSE      
3 taxon               27      5008 /Users/bbest/_big/calcofi/parque… FALSE      
4 dataset_taxon        7      1727 /Users/bbest/_big/calcofi/parque… FALSE      
5 taxon_group          7      1058 /Users/bbest/_big/calcofi/parque… FALSE      
6 measurement_type   220     16838 /Users/bbest/_big/calcofi/parque… FALSE      
7 dataset              1      5072 /Users/bbest/_big/calcofi/parque… FALSE      
build_relationships_json(
  rels = core_relationships(tbls_out), output_dir = dir_parquet,
  provider = provider, dataset = dataset)
Wrote relationships.json: 7 PKs, 8 FKs
[1] "/Users/bbest/Github/CalCOFI/.worktrees/workflows-pr117/data/parquet/sio_cetacean-edna/relationships.json"
d_tbls_rd <- read_csv(file.path(dir_meta, "tbls_redefine.csv"))
d_flds_rd <- read_csv(file.path(dir_meta, "flds_redefine.csv"))
build_metadata_json(
  con = con, d_tbls_rd = d_tbls_rd, d_flds_rd = d_flds_rd,
  metadata_derived_csv = c(here("metadata/core_dictionary.csv"),
                           file.path(dir_meta, "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,
  sources = src_stamp)
Set DuckDB COMMENT ON for tables and columns
Wrote metadata.json: 7 tables, 106 columns
metadata.json documentation gaps (7 tables, 106 columns) — these render blank in cc_describe_table() / cc_db_catalog():
tables with no description_md: 2    measurement_type, dataset
columns with no description_md: 43    dataset.provider, dataset.dataset, dataset.dataset_name, dataset.dataset_name_short, dataset.category, dataset.color, dataset.description, dataset.citation_main, dataset.citation_others, dataset.link_calcofi_org, dataset.link_data_source, dataset.link_others (+31 more)
measurement columns with no units: 34    dataset.dataset_name_short, dataset.category, dataset.color, dataset.citation_main, dataset.citation_others, dataset.link_calcofi_org, dataset.link_data_source, dataset.link_others, dataset.tables, dataset.coverage_temporal, dataset.coverage_spatial, dataset.license (+22 more)
  backfill via metadata/{provider}/{dataset}/flds_redefine.csv, then re-run
[1] "/Users/bbest/Github/CalCOFI/.worktrees/workflows-pr117/data/parquet/sio_cetacean-edna/metadata.json"
if (publish_to_gcs) {
  sync_to_gcs(local_dir = dir_stage, sidecar_dir = dir_parquet,
              gcs_prefix = glue("ingest/{dir_label}"), bucket = "calcofi-db")
} else {
  cat("CALCOFI_SKIP_GCS set -- parquet kept local, NOT synced to gs://calcofi-db\n")
}
rsync /Users/bbest/_big/calcofi/parquet/sio_cetacean-edna -> gs://calcofi-db/ingest/sio_cetacean-edna
Copied 3 sidecar(s) from /Users/bbest/Github/CalCOFI/.worktrees/workflows-pr117/data/parquet/sio_cetacean-edna
Sync complete: 10 copied, 0 removed (rsync)
# A tibble: 10 × 4
   file    action    size reason        
   <chr>   <chr>    <dbl> <chr>         
 1 <rsync> uploaded    NA parallel rsync
 2 <rsync> uploaded    NA parallel rsync
 3 <rsync> uploaded    NA parallel rsync
 4 <rsync> uploaded    NA parallel rsync
 5 <rsync> uploaded    NA parallel rsync
 6 <rsync> uploaded    NA parallel rsync
 7 <rsync> uploaded    NA parallel rsync
 8 <rsync> uploaded    NA parallel rsync
 9 <rsync> uploaded    NA parallel rsync
10 <rsync> uploaded    NA parallel rsync
close_duckdb(con)
cat(glue("Parquet outputs written to: {dir_parquet}"), "\n")
Parquet outputs written to: /Users/bbest/Github/CalCOFI/.worktrees/workflows-pr117/data/parquet/sio_cetacean-edna