Load the CTD cast archive into PostgreSQL (`ctd` schema)

originals verbatim → ctd.file / ctd.scan; flags and products live beside them

Published

2026-08-19

1 Overview

This notebook puts the entire calcofi.org CTD db-CSV archive — every final and preliminary cast file, every one of its 82 columns, verbatim — into the CTD team’s multi-user PostgreSQL database (calcofi, schema ctd; DDL in server/postgis/init/40_ctd.sql). It is the originals layer of the QA/QC design: nothing here is cleaned, deduplicated or thinned. -99 sentinels, NaNs and superseded preliminary files all go in, each row traceable to (archive, path, line). Problems become rows in ctd.flag; the release ingest (ingest_calcofi_ctd-cast.qmd) keeps doing the cleaning for the public release.

Three stages, each resumable and idempotent on the file’s sha256:

discover + build parquet (laptop, where the archives are extracted under cc_stage_path("ctd-cast", "unzip")): one typed parquet per cast file;
upload the parquet to gs://calcofi-db/pg/ctd_scan_raw/ (public bucket — this is calcofi.org’s public data), so the load can run anywhere;
load into ctd.file + ctd.scan with DuckDB’s postgres extension, then ctd.refresh_derived() (best-stage flags, ctd.cast, the v_scan_qc / v_scan_clean views) and the data dictionary (ctd.scan_column from metadata/measurement_type.csv).

Run the load on the server for speed (host=postgis, no tunnel) — see the last section — or from a laptop through cc_pg_tunnel(); both use the same functions in libs/pg_ctd.R.

2 Setup

Code

librarian::shelf(
  calcofi/calcofi4db, calcofi/calcofi4r,
  DBI, digest, dplyr, DT, duckdb, fs, glue, here, RPostgres, stringr, tibble, tidyr,
  quiet = TRUE)
source(here("libs/pg_ctd.R"))

dir_ext   <- calcofi4db::cc_stage_path("ctd-cast", "unzip")     # archives extracted by the ingest
dir_pq    <- calcofi4db::cc_stage_path("pg", "ctd_scan_raw")    # typed parquet, one per file
gcs_pq    <- "gs://calcofi-db/pg/ctd_scan_raw"                  # public mirror of dir_pq
s3_pq     <- "s3://calcofi-db/pg/ctd_scan_raw"                  # same, as DuckDB reads it (anonymous)
gcs_zips  <- "gs://calcofi-files-public/_sync/calcofi/ctd-cast/download"  # where the archives live off-box
pg_user   <- if (nzchar(Sys.getenv("PGUSER"))) Sys.getenv("PGUSER") else "bebest"
pg_dsn    <- glue("dbname=calcofi host={params$pg_host} port={params$pg_port} user={pg_user}")

dir_create(dir_pq)
cat(glue("archives: {dir_ext}\nparquet:  {dir_pq}\npg:       {pg_dsn}\n"))

archives: /Users/bbest/_big/calcofi/ctd-cast/unzip
parquet:  /Users/bbest/_big/calcofi/pg/ctd_scan_raw
pg:       dbname=calcofi host=localhost port=15432 user=bebest

3 1. Discover the db-CSV cast files

Same rules as the ingest’s file discovery (final > preliminary-with-bottle > preliminary-without-bottle, direction from the last character of the file stem) — but no per-cruise “best stage” filter here: everything is loaded; ctd.file.is_best_stage records which file wins per study.

Code

files <- pg_ctd_discover_files(dir_ext, compute_sha = TRUE)
cat(glue("{nrow(files)} files, {round(sum(files$n_bytes)/1e9, 2)} GB, {n_distinct(files$study)} studies\n"))

409 files, 4.58 GB, 142 studies

Code

files |> count(data_stage, cast_dir) |> datatable(caption = "files by stage × direction", options = list(dom = "t"))

Code

# study (9709NH) -> release cruise_key (1997-09-32NM) via the release cruise table
cruise_keys <- pg_ctd_cruise_keys(unique(files$study))
unmapped <- names(cruise_keys)[is.na(cruise_keys)]
cat(glue("{sum(!is.na(cruise_keys))} of {length(cruise_keys)} studies map to a release cruise_key",
         if (length(unmapped)) glue("; unmapped: {paste(unmapped, collapse = ', ')}") else ""), "\n")

142 of 142 studies map to a release cruise_key

4 2. Build typed parquet (one per file)

Every column is kept. Types: text for the identifiers, timestamp for the two date-times (20-Sep-1997 10:43:30 in most files; 01/29/2015 05:55 / 9/9/2026 9:05 in seven cruises — all parsed), smallint for the *Q flag columns, double for everything else. Blanks → NULL. A non-blank cell that still cannot be typed — bottle comments such as DIC or POSSIBLY 85;LEAKY BTM VLV sitting in OxBuM (1210NH), concatenations like 15.86-99.0 (9803SP), NaN in a flag column — goes verbatim into a sibling _issues parquet and then into ctd.scan_issue, keyed to the exact (file, row, column); the ctd.scan cell is NULL. Nothing from the files is dropped; the report below is what the QC team should look at first.

Code

files <- pg_ctd_build_parquet(files, dir_pq, overwrite = FALSE)

  50/409 files, 1,216,435 rows, 53 issue cells so far
  100/409 files, 2,318,059 rows, 139 issue cells so far
  150/409 files, 3,867,817 rows, 147 issue cells so far
  200/409 files, 5,334,166 rows, 530 issue cells so far
  250/409 files, 6,796,947 rows, 583 issue cells so far
  300/409 files, 8,098,653 rows, 1,622 issue cells so far
  350/409 files, 9,589,977 rows, 1,630 issue cells so far
  400/409 files, 10,713,989 rows, 2,558 issue cells so far
  409/409 files, 10,812,360 rows, 2,558 issue cells so far

Code

cat(glue("{format(sum(files$n_rows), big.mark = ',')} scan rows across {nrow(files)} files; parquet {round(sum(file_size(files$parquet))/1e9, 2)} GB\n"))

10,812,360 scan rows across 409 files; parquet 1.16 GB

Code

cf <- attr(files, "cast_failures")
if (is.null(cf) || nrow(cf) == 0) {
  cat("untypable cells: none — every non-blank value typed cleanly\n")
} else {
  cat(glue("{format(sum(cf$n_failed), big.mark = ',')} untypable cells in {n_distinct(cf$path)} files → ctd.scan_issue\n"))
  cf |> group_by(column_name) |> summarise(files = n(), cells = sum(n_failed), .groups = "drop") |> arrange(desc(cells)) |>
    datatable(caption = "untypable source cells by column (kept verbatim in ctd.scan_issue; NULL in ctd.scan)", options = list(dom = "t"))
}

2,558 untypable cells in 47 files → ctd.scan_issue

5 3. Upload the parquet to GCS

Code

cmd <- glue("gcloud storage rsync -r --quiet '{dir_pq}' '{gcs_pq}'")
cat(cmd, "\n"); status <- system(cmd)

gcloud storage rsync -r --quiet '/Users/bbest/_big/calcofi/pg/ctd_scan_raw' 'gs://calcofi-db/pg/ctd_scan_raw'

Code

stopifnot(status == 0)
cat("uploaded\n")

uploaded

6 4. Load into PostgreSQL

Through the tunnel from a laptop (params$pg_host = localhost, pg_port = the local end) or on the server (pg_host = postgis, pg_port = 5432). Files already in ctd.file (by sha256) are skipped; scans are inserted per study batch so an interrupted run resumes.

Code

if (params$pg_host == "localhost") calcofi4r::cc_pg_tunnel(local_port = params$pg_port)
n <- pg_ctd_load(files, parquet_root = dir_pq, pg_dsn = pg_dsn, gcs_uri_root = gcs_zips, cruise_keys = cruise_keys)

ctd.file: 409 already registered, 0 new
ctd.scan: 0 files pending in 0 studies

Code

res <- pg_ctd_finish(pg_dsn, measurement_type_csv = here("metadata/measurement_type.csv"))
print(res)

  files    scans issue_cells casts best_files scan_size
1   409 10812360        2558 24928        306   5842 MB

Code

if (params$pg_host == "localhost") calcofi4r::cc_pg_tunnel_close(local_port = params$pg_port)

7 What landed

Code

if (params$pg_host == "localhost") calcofi4r::cc_pg_tunnel(local_port = params$pg_port)
con <- calcofi4r::cc_pg_connect(port = params$pg_port, host = params$pg_host)
dbGetQuery(con, "
  SELECT data_stage, cast_dir, count(*) AS files, sum(n_rows) AS scans, count(*) FILTER (WHERE is_best_stage) AS best
  FROM ctd.file GROUP BY 1,2 ORDER BY 1,2") |> datatable(caption = "ctd.file by stage × direction", options = list(dom = "t"))

Code

dbGetQuery(con, "
  SELECT min(datetime_utc)::date AS first_cast, max(datetime_utc)::date AS last_cast, count(*) AS casts,
         count(DISTINCT study) AS studies, count(*) FILTER (WHERE geom IS NULL) AS casts_without_position
  FROM ctd.cast WHERE is_best_stage") |> datatable(caption = "ctd.cast (best stage)", options = list(dom = "t"))

Code

dbGetQuery(con, "
  SELECT column_name, count(*) AS cells, count(DISTINCT file_id) AS files, min(raw_value) AS example
  FROM ctd.scan_issue GROUP BY 1 ORDER BY 2 DESC") |> datatable(caption = "ctd.scan_issue — untypable source cells, verbatim", options = list(dom = "t"))

Code

dbGetQuery(con, "
  SELECT column_name, source_header, kind, qual_column, measurement_type, units
  FROM ctd.scan_column ORDER BY ordinal") |> datatable(caption = "ctd.scan_column — the data dictionary")

Code

dbDisconnect(con)
if (params$pg_host == "localhost") calcofi4r::cc_pg_tunnel_close(local_port = params$pg_port)

8 Running the load on the server

# on the server, inside the rstudio container (has R + duckdb + RPostgres; reaches postgis by name)
docker exec -u bebest -w /home/bebest rstudio Rscript -e '
  source("/share/github/CalCOFI/workflows/libs/pg_ctd.R")
  files <- readRDS("/share/data/ctd/exports/pg_ctd_files.rds")   # written by this notebook (see below)
  pg_ctd_load(files, parquet_root = "s3://calcofi-db/pg/ctd_scan_raw",
              pg_dsn = "dbname=calcofi host=postgis port=5432 user=bebest",
              gcs_uri_root = "gs://calcofi-files-public/_sync/calcofi/ctd-cast/download")
  pg_ctd_finish("dbname=calcofi host=postgis port=5432 user=bebest")'

Code

# the file manifest (paths, stages, sha256, n_rows, cruise_key) for a server-side load or audit
files_out <- files |> mutate(cruise_key = unname(cruise_keys[study])) |> select(-path_abs, -parquet)
dir_create(here("data/pg"))
readr::write_csv(files_out, here("data/pg/ctd_scan_raw_files.csv"), na = "")
cat(glue("manifest: data/pg/ctd_scan_raw_files.csv ({nrow(files_out)} files)\n"))

manifest: data/pg/ctd_scan_raw_files.csv (409 files)