## ----setup, include=FALSE-----------------------------------------------------
knitr::opts_chunk$set(
  collapse = TRUE,
  comment = "#>",
  message = FALSE,
  warning = FALSE
)

## ----load---------------------------------------------------------------------
library(marcxmlr)
library(dplyr)

xml_file <- system.file(
  "extdata", "gpo-cgp-part00-sample.xml",
  package = "marcxmlr", mustWork = TRUE
)
marc <- read_marcxml(xml_file)

n_records <- n_distinct(marc$record_id)
field_instances <- marc |>
  filter(field_type == "datafield") |>
  distinct(record_id, field_order, tag)

overview <- data.frame(
  Measure = c("Bibliographic records", "Canonical rows", "Data-field instances"),
  Value = c(n_records, nrow(marc), nrow(field_instances))
)
knitr::kable(overview, format.args = list(big.mark = ","))

## ----coverage-----------------------------------------------------------------
tags_of_interest <- data.frame(
  tag = c("245", "650", "651", "655", "710", "856"),
  meaning = c(
    "Title statement", "Topical subject", "Geographic subject",
    "Genre/form term", "Corporate added entry", "Electronic location/access"
  )
)

coverage <- field_instances |>
  filter(tag %in% tags_of_interest$tag) |>
  distinct(record_id, tag) |>
  count(tag, name = "records") |>
  right_join(tags_of_interest, by = "tag") |>
  mutate(
    records = coalesce(records, 0L),
    percent = 100 * records / n_records
  ) |>
  select(tag, meaning, records, percent)

knitr::kable(
  coverage, digits = 1,
  caption = "Record-level field coverage (denominator: all 1,000 records)."
)

## ----subject-counts-----------------------------------------------------------
subject_counts <- field_instances |>
  filter(tag == "650") |>
  count(record_id, name = "n_650") |>
  right_join(
    data.frame(record_id = unique(marc$record_id)),
    by = "record_id"
  ) |>
  mutate(n_650 = coalesce(n_650, 0L)) |>
  arrange(record_id)

subject_summary <- data.frame(
  Measure = c(
    "Records with no 650", "Mean fields per record",
    "Median fields per record", "90th percentile", "Maximum"
  ),
  Value = c(
    sum(subject_counts$n_650 == 0L), mean(subject_counts$n_650),
    median(subject_counts$n_650),
    as.numeric(quantile(subject_counts$n_650, 0.90)),
    max(subject_counts$n_650)
  )
)
subject_summary$Value <- c(
  sprintf("%.0f", subject_summary$Value[1]),
  sprintf("%.2f", subject_summary$Value[2]),
  sprintf("%.0f", subject_summary$Value[3:5])
)
knitr::kable(subject_summary)

## ----subject-count-figure, fig.width=6.3, fig.height=3.5, fig.cap="Topical subject fields per record. The last category combines records with five or more 650 fields."----
subjects_binned <- factor(
  pmin(subject_counts$n_650, 5L),
  levels = 0:5,
  labels = c("0", "1", "2", "3", "4", "5+")
)
barplot(
  table(subjects_binned),
  xlab = "Number of 650 fields in a record",
  ylab = "Number of records",
  col = "grey70", border = "grey35"
)

## ----field-level-profiles-----------------------------------------------------
field_profiles <- marc |>
  filter(field_type == "datafield") |>
  group_by(record_id, field_order, tag, ind2) |>
  summarise(
    has_0 = any(
      subfield_code == "0" & !is.na(value) & nzchar(trimws(value)),
      na.rm = TRUE
    ),
    n_subdivisions = sum(subfield_code %in% c("v", "x", "y", "z")),
    .groups = "drop"
  )

selected_tags <- c("100", "110", "650", "651", "655", "710")
authority_by_tag <- field_profiles |>
  filter(tag %in% selected_tags) |>
  group_by(tag) |>
  summarise(
    field_instances = n(),
    with_0 = sum(has_0),
    percent_with_0 = 100 * mean(has_0),
    .groups = "drop"
  ) |>
  arrange(tag)

knitr::kable(
  authority_by_tag, digits = 1,
  caption = "Presence of nonempty $0 by MARC field tag (field-instance denominator)."
)

## ----subdivision-profile------------------------------------------------------
topical_fields <- field_profiles |>
  filter(tag == "650") |>
  mutate(
    subdivision_group = factor(
      case_when(
        n_subdivisions == 0L ~ "None",
        n_subdivisions == 1L ~ "One",
        TRUE ~ "Two or more"
      ),
      levels = c("None", "One", "Two or more")
    )
  )

subdivision_profile <- topical_fields |>
  group_by(subdivision_group) |>
  summarise(
    field_instances = n(),
    with_0 = sum(has_0),
    percent_with_0 = 100 * mean(has_0),
    .groups = "drop"
  )
knitr::kable(
  subdivision_profile, digits = 1,
  caption = "Topical 650 fields by number of subdivisions and presence of $0."
)

## ----subdivision-figure, fig.width=6.2, fig.height=3.5, fig.cap="Authority identifier presence by subdivision complexity, among 650 field instances."----
barplot(
  subdivision_profile$percent_with_0,
  names.arg = subdivision_profile$subdivision_group,
  ylim = c(0, 100),
  ylab = "650 field instances with nonempty $0 (%)",
  xlab = "Number of subject subdivisions",
  col = "grey70", border = "grey35"
)

## ----vocabulary-check---------------------------------------------------------
vocabulary_profile <- topical_fields |>
  group_by(ind2) |>
  summarise(
    field_instances = n(),
    with_0 = sum(has_0),
    percent_with_0 = 100 * mean(has_0),
    .groups = "drop"
  ) |>
  arrange(desc(field_instances))
knitr::kable(
  vocabulary_profile, digits = 1,
  caption = "Topical 650 fields by second indicator (subject-heading system)."
)

lcsh_profile <- topical_fields |>
  filter(ind2 == "0") |>
  group_by(subdivision_group) |>
  summarise(
    field_instances = n(),
    with_0 = sum(has_0),
    percent_with_0 = 100 * mean(has_0),
    .groups = "drop"
  )
knitr::kable(
  lcsh_profile, digits = 1,
  caption = "Subdivision association restricted to 650 fields with second indicator 0."
)

## ----inspection-list----------------------------------------------------------
control_numbers <- marc |>
  filter(field_type == "controlfield", tag == "001") |>
  transmute(record_id, control_001 = value)

titles <- marc |>
  filter(tag == "245", subfield_code == "a") |>
  group_by(record_id) |>
  summarise(title = first(value), .groups = "drop")

# Retain each subdivision's original value for human interpretation.
subject_text <- marc |>
  filter(tag == "650", subfield_code %in% c("a", "v", "x", "y", "z")) |>
  arrange(record_id, field_order, subfield_order) |>
  group_by(record_id, field_order) |>
  summarise(heading = paste(value, collapse = " -- "), .groups = "drop")

inspection <- topical_fields |>
  filter(n_subdivisions >= 2L, !has_0) |>
  arrange(desc(n_subdivisions), record_id, field_order) |>
  slice_head(n = 8L) |>
  left_join(control_numbers, by = "record_id") |>
  left_join(titles, by = "record_id") |>
  left_join(subject_text, by = c("record_id", "field_order")) |>
  select(control_001, title, heading, n_subdivisions)

knitr::kable(
  inspection, caption = "Illustrative 650 fields for inspection (not a list of errors)."
)

## ----occurrence-specific-records----------------------------------------------
candidate_ids <- topical_fields |>
  filter(n_subdivisions >= 2L, !has_0) |>
  distinct(record_id)

candidate_records <- marc |>
  semi_join(candidate_ids, by = "record_id")

c(records = n_distinct(candidate_records$record_id),
  canonical_rows = nrow(candidate_records))

## ----record-query-------------------------------------------------------------
q <- marc_has("650", subfield = "0") & !marc_has("856")

selected_table <- marc_filter_records(marc, q)

query_summary <- data.frame(
  Measure = c("Selected records", "Canonical rows retained", "650 field instances retained"),
  Value = c(
    n_distinct(selected_table$record_id),
    nrow(selected_table),
    selected_table |>
      filter(tag == "650") |>
      distinct(record_id, field_order) |>
      nrow()
  )
)
knitr::kable(query_summary)

## ----query-during-ingestion---------------------------------------------------
selected_xml <- tryCatch(
  read_marcxml(xml_file, filter = q),
  error = function(e) {
    if (grepl("requires Expat support at package build time",
              conditionMessage(e), fixed = TRUE)) return(NULL)
    stop(e)
  }
)

if (!is.null(selected_xml)) {
  # Compare bibliographic control numbers, not numeric record_id values:
  # native prefiltering renumbers selected records consecutively.
  control_numbers <- function(x) {
    x |>
      filter(field_type == "controlfield", tag == "001") |>
      pull(value) |>
      sort()
  }
  stopifnot(identical(control_numbers(selected_xml),
                      control_numbers(selected_table)))
  knitr::kable(data.frame(
    Method = c("Before materialisation", "After materialisation"),
    Records = c(n_distinct(selected_xml$record_id),
                n_distinct(selected_table$record_id)),
    Rows = c(nrow(selected_xml), nrow(selected_table))
  ))
} else {
  cat("Native XML prefiltering unavailable: this build lacks Expat support.\n")
}

## ----full-catalogue, eval=FALSE-----------------------------------------------
# input_dir <- "/path/to/benchmark-inputs"
# source_files <- sort(Sys.glob(file.path(
#   input_dir, "cataloging-records-all-cgp-XML-*.xml"
# )))
# 
# marcxml_to_parquet(
#   source_files,
#   output_dir = "gpo-canonical-parquet",
#   batch_records = 5000L,
#   workers = 4L
# )
# 
# ds <- arrow::open_dataset("gpo-canonical-parquet")
# 
# # Field-instance counts, not canonical-row counts.
# by_vocabulary <- ds |>
#   filter(tag == "650") |>
#   distinct(record_id, field_order, ind2) |>
#   count(ind2, name = "field_instances") |>
#   collect()
# 
# with_id <- ds |>
#   filter(tag == "650", subfield_code == "0",
#          !is.na(value), value != "") |>
#   distinct(record_id, field_order, ind2) |>
#   count(ind2, name = "with_0") |>
#   collect()
# 
# left_join(by_vocabulary, with_id, by = "ind2")

## ----full-catalogue-query, eval=FALSE-----------------------------------------
# # Select during ingestion; do not materialise the full catalogue.
# marcxml_to_parquet(
#   source_files,
#   output_dir = "gpo-selected-parquet",
#   filter = q,
#   batch_records = 5000L,
#   workers = 4L
# )
# 
# # Or select complete records from the unfiltered Dataset already on disk.
# selected_ds <- marc_filter_records(ds, q)
# # selected_ds remains lazy; use dplyr::collect() only if it fits in memory.