Local ESGF Store

Source: R/store.R

EsgStore manages a local DuckDB manifest and a fixed directory layout for query result snapshots, dictionaries, source files, downloaded NetCDF files, Parquet regional extracts, and generated outputs.

Author

Hongyuan Jia

Active bindings

path

Store directory.

manifest

DuckDB manifest path.

is_open

Whether the manifest connection is open.

Methods

Public methods

Method new()

Create or open a local store.

Usage

EsgStore$new(path = NULL, create = TRUE, overwrite = FALSE)

Arguments

path

Store directory. Default: store_dir().

create

If TRUE, create the store directory when it does not exist. Default: TRUE.

overwrite

If TRUE, remove an existing store directory before creating a new store. Default: FALSE.

Returns

An EsgStore object.

Method close()

Close the DuckDB connection.

Usage

EsgStore$close()

Returns

The store object itself, invisibly.

Method get_meta()

Return a store metadata value.

Usage

EsgStore$get_meta(key, default = NULL)

Arguments

key

Metadata key.

default

Value returned when key is not set.

Returns

A single string, or default.

Method set_meta()

Set a store metadata value.

Usage

EsgStore$set_meta(key, value)

Arguments

key

Metadata key.

value

Metadata value. NULL is stored as NA.

Returns

The store object, invisibly.

Method download_layout()

Return the store download layout policy.

Usage

EsgStore$download_layout()

Returns

A named list describing how store downloads are placed under downloads/.

Method set_download_layout()

Configure how store-managed ESGF downloads are placed under downloads/.

Usage

EsgStore$set_download_layout(
  layout = c("flat", "dataset", "drs", "template"),
  template = NULL,
  include_version = TRUE,
  collision = c("error", "checksum", "suffix"),
  missing = c("fallback", "error")
)

Arguments

layout

Download layout. "flat" stores files directly under downloads/; "dataset" groups by dataset; "drs" uses a CMIP6-style DRS path; "template" uses template.

template

Optional subdirectory template for layout = "template", using placeholders such as {source_id}.

include_version

Whether DRS paths include the ESGF version. Default: TRUE.

collision

How to handle different logical files that map to the same local path. Default: "error".

missing

How to handle missing layout fields. Default: "fallback".

Returns

The store object, invisibly.

Method register_artifact()

Register a file artifact in the store manifest.

Usage

EsgStore$register_artifact(
  kind,
  path,
  role = NULL,
  project = NULL,
  status = "available",
  checksum = NULL,
  checksum_type = "sha256",
  size = NULL,
  query_id = NULL,
  file_key = NULL,
  dict_id = NULL,
  source_url = NULL,
  source_repo = NULL,
  source_tag = NULL,
  source_commit = NULL,
  metadata = list()
)

Arguments

kind

Artifact kind.

path

Artifact path. Absolute paths must be inside the store root.

role

Artifact role. If NULL, a role is inferred from kind.

project

Optional ESGF project.

status

Artifact status. Default: "available".

checksum

Expected checksum. If NULL and path exists, it is calculated with checksum_type.

checksum_type

Checksum algorithm. Default: "sha256".

size

Artifact size in bytes. If NULL and path exists, it is read from the file.

query_id, file_key, dict_id

Optional manifest links.

source_url, source_repo, source_tag, source_commit

Optional source provenance.

metadata

Optional metadata list encoded as JSON.

Returns

The artifact ID.

Method artifact_path()

Return an artifact path from the manifest.

Usage

EsgStore$artifact_path(artifact_id)

Arguments

artifact_id

Artifact ID.

Returns

Absolute artifact path.

Method validate()

Validate registered artifact files against the manifest.

Usage

EsgStore$validate()

Returns

A data.table with validation results.

Method add_query()

Add an ESGF query to the long-lived store query registry.

Usage

EsgStore$add_query(query, label = NULL, track = FALSE)

Arguments

query

An EsgQuery object.

label

Optional label.

track

Whether to mark the query as tracked. Default: FALSE.

Returns

The stable query ID.

Method track_query()

Mark a stored ESGF query as tracked.

Usage

EsgStore$track_query(query_id)

Arguments

query_id

Query ID returned by $add_query().

Returns

The store object, invisibly.

Method untrack_query()

Mark a stored ESGF query as untracked.

Usage

EsgStore$untrack_query(query_id)

Arguments

query_id

Query ID returned by $add_query().

Returns

The store object, invisibly.

Method tag_query()

Add tags to a stored ESGF query.

Usage

EsgStore$tag_query(query_id, tag, replace = FALSE)

Arguments

query_id

Query ID returned by $add_query().

tag

Character vector of tags.

replace

Whether to replace existing tags for the query.

Returns

A data.table of tags for the query.

Method untag_query()

Remove tags from a stored ESGF query.

Usage

EsgStore$untag_query(query_id, tag = NULL)

Arguments

query_id

Query ID returned by $add_query().

tag

Optional tags. If NULL, all tags are removed.

Returns

A data.table of remaining tags for the query.

Method query_tags()

List stored ESGF query tags.

Usage

EsgStore$query_tags(query_id = NULL)

Arguments

query_id

Optional query ID filter.

Returns

A data.table of query tags.

Method require_query()

Record that one stored query depends on another stored query.

Usage

EsgStore$require_query(query_id, parent_query_id)

Arguments

query_id

Child query ID.

parent_query_id

Required parent query ID.

Returns

A data.table of query dependency edges.

Method unrequire_query()

Remove query dependency edges.

Usage

EsgStore$unrequire_query(query_id, parent_query_id = NULL)

Arguments

query_id

Child query ID.

parent_query_id

Optional parent query ID. If NULL, all parents for query_id are removed.

Returns

A data.table of remaining dependency edges for the query.

Method query_graph()

List stored query dependency edges.

Usage

EsgStore$query_graph(
  query_id = NULL,
  direction = c("children", "parents", "both"),
  recursive = TRUE
)

Arguments

query_id

Optional query ID anchor.

direction

Which edge direction to return for an anchor.

recursive

Whether to include transitive edges.

Returns

A data.table of dependency edges.

Method queries()

List stored ESGF queries.

Usage

EsgStore$queries(tracked = NULL)

Arguments

tracked

Optional tracked-state filter.

Returns

A data.table of stored query records.

Method query_files()

List files linked to a stored ESGF query.

Usage

EsgStore$query_files(query_id, status = NULL)

Arguments

query_id

Query ID returned by $add_query().

status

Optional query-file status filter.

Returns

A data.table of linked file records.

Method preview_update_queries()

Preview tracked ESGF query updates without changing the store.

Usage

EsgStore$preview_update_queries(
  query_id = NULL,
  tracked = TRUE,
  tag = NULL,
  children = FALSE,
  detail = FALSE,
  all = TRUE,
  limit = FALSE,
  fields = "*",
  ...
)

Arguments

query_id

Optional query ID. If NULL, tracked queries are previewed by default.

tracked

Tracked-state filter used when query_id is NULL.

tag

Optional query tag filter used when query_id is NULL.

children

Whether to include dependency children of selected queries.

detail

Whether to return per-file changes together with the summary. Default: FALSE.

all, limit, fields

Arguments passed to EsgQuery$collect().

...

Additional File query filters passed to EsgQuery$collect().

Returns

A data.table summary, or a list with summary and changes when detail = TRUE.

Method update_queries()

Refresh stored ESGF queries and link their current File records.

Usage

EsgStore$update_queries(
  query_id = NULL,
  tracked = TRUE,
  tag = NULL,
  children = FALSE,
  enqueue = FALSE,
  downloader = NULL,
  replica = "auto",
  session_label = NULL,
  service = "HTTPServer",
  probe = TRUE,
  probe_concurrency = NULL,
  probe_cache_seconds = 3600L,
  strategy = c("fastest", "first", "stable"),
  all = TRUE,
  limit = FALSE,
  fields = "*",
  ...
)

Arguments

query_id

Optional query ID. If NULL, tracked queries are updated by default.

tracked

Tracked-state filter used when query_id is NULL.

tag

Optional query tag filter used when query_id is NULL.

children

Whether to include dependency children of selected queries.

enqueue

Whether to enqueue current files after updating. Default: FALSE.

downloader

Optional Downloader used when enqueue = TRUE.

replica

Replica policy passed to $download_plan() when enqueuing.

session_label

Optional download session label.

service

ESGF URL service used for the download plan.

probe

Whether to probe candidate URLs before ranking.

probe_concurrency

Maximum concurrent URL probes when probe = TRUE. Default comes from the downloader worker count when enqueue = TRUE.

probe_cache_seconds

Seconds to reuse fresh data-node probe history before probing a URL again. Default: 3600.

strategy

Candidate ranking strategy.

all, limit, fields

Arguments passed to EsgQuery$collect().

...

Additional File query filters passed to EsgQuery$collect().

Returns

A data.table of query-file links touched by the update.

Method download_preflight()

Preview a tracked query download without changing the store.

Usage

EsgStore$download_preflight(
  query_id,
  downloader = NULL,
  replica = "auto",
  service = "HTTPServer",
  probe = TRUE,
  probe_concurrency = NULL,
  probe_cache_seconds = 3600L,
  strategy = c("fastest", "first", "stable"),
  all = TRUE,
  limit = FALSE,
  fields = "*",
  ...
)

Arguments

query_id

Query ID returned by $add_query().

downloader

Optional Downloader used only for node history, network policy, and cooldown policy.

replica

Replica policy passed to $download_plan().

service, probe, strategy

Download plan arguments.

probe_concurrency

Maximum concurrent URL probes when probe = TRUE. Default comes from downloader when supplied.

probe_cache_seconds

Seconds to reuse fresh data-node probe history before probing a URL again. Default: 3600.

all, limit, fields

Arguments passed to EsgQuery$collect().

...

Additional File query filters passed to EsgQuery$collect().

Returns

A list with summary, changes, files, and candidates.

Method download_query()

Refresh, enqueue, and optionally run downloads for a stored ESGF query.

Usage

EsgStore$download_query(
  query_id,
  downloader = NULL,
  replica = "auto",
  dry_run = FALSE,
  run = TRUE,
  background = FALSE,
  mode = c("process", "daemon"),
  session_label = NULL,
  service = "HTTPServer",
  probe = TRUE,
  probe_concurrency = NULL,
  probe_cache_seconds = 3600L,
  strategy = c("fastest", "first", "stable"),
  progress = TRUE,
  overwrite = FALSE,
  resume = TRUE,
  all = TRUE,
  limit = FALSE,
  fields = "*",
  ...
)

Arguments

query_id

Query ID returned by $add_query().

downloader

Optional Downloader. Default: $downloader().

replica

Replica policy passed to $download_plan().

dry_run

Whether to return a download preflight without changing the store, enqueueing, or downloading. Default: FALSE.

run

Whether to run the queued session immediately. Default: TRUE.

background

Whether to run the queued session in the background. Default: FALSE.

mode

Background execution mode. "process" starts a detached Rscript; "daemon" submits the job to a running downloader daemon.

session_label

Optional download session label.

service, probe, strategy

Download plan arguments.

probe_concurrency

Maximum concurrent URL probes when probe = TRUE. Default comes from the downloader worker count.

probe_cache_seconds

Seconds to reuse fresh data-node probe history before probing a URL again. Default: 3600.

progress, overwrite, resume

Run arguments.

all, limit, fields

Arguments passed to EsgQuery$collect().

...

Additional File query filters passed to EsgQuery$collect().

Returns

The created downloader session ID, NA_character_ when there is no pending file to download, or a one-row background job record when run = TRUE and background = TRUE.

Method download_status()

Return downloader tasks linked to stored query files.

Usage

EsgStore$download_status(query_id = NULL, session_id = NULL, downloader = NULL)

Arguments

query_id

Optional stored query ID.

session_id

Optional downloader session ID.

downloader

Optional Downloader. Default: $downloader().

Returns

A data.table of downloader task rows.

Method query_status()

Summarise tracked ESGF query file and download status.

Usage

EsgStore$query_status(query_id = NULL, downloader = NULL)

Arguments

query_id

Optional stored query ID vector. If NULL, all stored ESGF queries are summarised.

downloader

Optional Downloader. Default: $downloader().

Returns

A data.table with one row per stored query.

Method query_updates()

List tracked query update runs.

Usage

EsgStore$query_updates(query_id = NULL, latest = FALSE)

Arguments

query_id

Optional stored query ID filter.

latest

Whether to return only the latest update per query.

Returns

A data.table of update run summaries.

Method query_changes()

List per-file changes recorded by tracked query updates.

Usage

EsgStore$query_changes(update_id = NULL, query_id = NULL, change_type = NULL)

Arguments

update_id

Optional update run ID filter.

query_id

Optional stored query ID filter.

change_type

Optional change type filter.

Returns

A data.table of per-file query update changes.

Method workflow_status()

Summarise query, download, local, and extraction status together.

Usage

EsgStore$workflow_status(query_id = NULL, downloader = NULL)

Arguments

query_id

Optional stored query ID filter.

downloader

Optional Downloader. Default: $downloader().

Returns

A data.table with one row per stored query.

Method workflow_report()

Return a compact ESGF query workflow health report.

Usage

EsgStore$workflow_report(query_id = NULL, downloader = NULL)

Arguments

query_id

Optional stored query ID filter.

downloader

Optional Downloader. Default: $downloader().

Returns

A list with summary, updates, changes, downloads, and nodes.

Method remove_query()

Remove stored ESGF queries and optionally delete orphaned local files.

Usage

EsgStore$remove_query(query_id, delete = c("none", "orphaned"))

Arguments

query_id

Stored query ID vector.

delete

Whether to leave local files untouched ("none") or delete files orphaned by the removal ("orphaned").

Returns

A data.table describing removed queries.

Method remove_files()

Remove ESGF file records and optionally delete local artifacts.

Usage

EsgStore$remove_files(file_key, delete_local = FALSE, force = FALSE)

Arguments

file_key

File key vector.

delete_local

Whether to delete local NetCDF files. Default: FALSE.

force

Whether to remove files still linked to queries. Default: FALSE.

Returns

A data.table describing removed file records.

Method prune_orphans()

Report or remove file records no longer linked to any query.

Usage

EsgStore$prune_orphans(delete_local = FALSE)

Arguments

delete_local

Whether to delete local NetCDF files and remove orphaned registry records. Default: FALSE.

Returns

A data.table of orphaned file records.

Method storage_report()

Summarise store download storage, registered local assets, temporary files, and cleanup candidates.

Usage

EsgStore$storage_report(detail = FALSE)

Arguments

detail

Whether to return detailed file tables. Default: FALSE.

Returns

A summary data.table, or a list when detail = TRUE.

Method validate_files()

Validate store-managed NetCDF downloads against the manifest.

Usage

EsgStore$validate_files(query_id = NULL, checksum = FALSE, layout = TRUE)

Arguments

query_id

Optional stored query IDs to validate. When NULL, all known downloaded ESGF files are checked.

checksum

Whether to compute file checksums. Default: FALSE.

layout

Whether to compare registered files with the current download layout policy. Default: TRUE.

Returns

A list with summary, files, artifacts, untracked, and actions data.tables. The method is read-only.

Method repair_files()

Repair safe store download inconsistencies reported by $validate_files().

Usage

EsgStore$repair_files(actions = NULL, dry_run = TRUE)

Arguments

actions

Optional action table from $validate_files()$actions. When NULL, actions are generated from $validate_files().

dry_run

Whether to only report planned repairs. Default: TRUE.

Returns

A data.table describing attempted repairs.

Method cleanup_downloads()

Report or remove download cleanup candidates.

Usage

EsgStore$cleanup_downloads(
  scope = c("tmp", "orphan_records", "untracked_files", "missing_records"),
  dry_run = TRUE,
  older_than = NULL
)

Arguments

scope

Cleanup scopes. Supported values are "tmp", "orphan_records", "untracked_files", and "missing_records".

dry_run

Whether to only report cleanup candidates. Default: TRUE.

older_than

Optional age filter for file scopes. A numeric value is interpreted as seconds before now; a POSIXct value is used as an absolute mtime cutoff.

Returns

A data.table describing cleanup candidates or removals.

Method retry_downloads()

Requeue retryable downloader tasks linked to stored query files.

Usage

EsgStore$retry_downloads(
  query_id = NULL,
  session_id = NULL,
  downloader = NULL,
  status = c("error", "cancelled"),
  run = TRUE,
  ...
)

Arguments

query_id

Optional stored query ID.

session_id

Optional downloader session ID.

downloader

Optional Downloader. Default: $downloader().

status

Retryable statuses. Default: c("error", "cancelled").

run

Whether to run requeued tasks immediately. Default: TRUE.

...

Additional arguments passed to Downloader$run().

Returns

A data.table of matching task rows after retry handling.

Method add_files()

Add File or Aggregation query results to the local file catalog.

Usage

EsgStore$add_files(files, label = NULL)

Arguments

files

An EsgResultFile or EsgResultAggregation object.

label

Optional label for this query run.

Returns

The created or updated query ID.

Method downloader()

Return a Downloader bound to this store.

Usage

EsgStore$downloader(...)

Arguments

...

Additional arguments passed to Downloader$new().

Returns

A Downloader object.

Method download_files()

Enqueue and optionally download ESGF file records through the store downloader.

Usage

EsgStore$download_files(
  files = NULL,
  query_id = NULL,
  replica = "auto",
  downloader = NULL,
  run = TRUE,
  background = FALSE,
  mode = c("process", "daemon"),
  session_label = NULL,
  service = "HTTPServer",
  probe = TRUE,
  probe_concurrency = NULL,
  probe_cache_seconds = 3600L,
  strategy = c("fastest", "first", "stable"),
  progress = TRUE,
  overwrite = FALSE,
  resume = TRUE,
  ...
)

Arguments

files

Optional EsgResultFile or EsgResultAggregation object. If supplied, it is cataloged before the download plan is created.

query_id

Optional file collection query IDs to enqueue when files is NULL. If NULL, all cataloged files missing local paths are considered.

replica

Replica policy passed to $download_plan().

downloader

Optional Downloader. Default: $downloader().

run

Whether to run the queued session immediately. Default: TRUE.

background

Whether to run the queued session in the background. Default: FALSE.

mode

Background execution mode. "process" starts a detached Rscript; "daemon" submits the job to a running downloader daemon.

session_label

Optional download session label.

service

ESGF URL service to download from. Default: "HTTPServer".

probe

Whether to lightly probe URLs before ranking them.

probe_concurrency

Maximum concurrent URL probes when probe = TRUE. Default comes from the downloader worker count.

probe_cache_seconds

Seconds to reuse fresh data-node probe history before probing a URL again. Default: 3600.

strategy

Candidate ranking strategy.

progress

Whether to show per-file download progress.

overwrite

Whether to overwrite existing final files.

resume

Whether to resume interrupted .part files.

...

Additional arguments passed to $download_plan() and Downloader$run().

Returns

The created downloader session ID, or a one-row background job record when run = TRUE and background = TRUE.

Method sync_downloads()

Register completed downloader tasks as local store artifacts.

Usage

EsgStore$sync_downloads(downloader = NULL)

Arguments

downloader

Optional Downloader. Default: $downloader().

Returns

A data.table of completed tasks.

Method plan_region()

Plan regional extraction jobs from cataloged files.

Usage

EsgStore$plan_region(
  query_id,
  lon,
  lat,
  time,
  site_id = "site-1",
  variable_id = NULL,
  filters = list(),
  method = "nearest"
)

Arguments

query_id

Query ID returned by $add_files().

lon, lat

Target longitude and latitude.

time

Length-2 time range.

site_id

Site identifier. Default: "site-1".

variable_id

Optional variable IDs. If NULL, all cataloged variables in the query are used.

filters

Named list of exact-match file catalog filters.

method

Grid extraction method. One of "nearest", "idw", "bilinear", or "mean". Default: "nearest".

Returns

A data.table of extraction plan rows.

EsgStore$extract()

Execute pending or failed regional extraction plans.

Independent remote reads run concurrently, with at most getOption("epwshiftr.mirai_workers", 4L) source tasks per call. Set options(epwshiftr.mirai_workers = 1L) for serial reads. The same limit applies to shared batch acquisition; separate R sessions have separate limits. The calling process owns manifest and Parquet writes. With shared caching enabled, pending plans for one physical file reuse its connection. With caching disabled, workers return one plan at a time to bound memory. Complete ordinary extraction caches are consumed directly without starting workers. Speedup depends on the number of independent tasks and the remote service.

Usage

EsgStore$extract(
  plan_id = NULL,
  status = c("pending", "failed"),
  fallback = c("auto", "error"),
  overwrite = FALSE,
  resume = TRUE,
  reporter = NULL
)

Arguments

plan_id

Optional plan IDs to run.

status

Plan statuses to run when plan_id is NULL. Default: c("pending", "failed").

fallback

What to do when OPeNDAP open, metadata, or read access fails. "auto" downloads through HTTPServer and retries once when possible; "error" marks the plan failed without downloading. Default: "auto".

overwrite

If TRUE, overwrite existing Parquet outputs. Default: FALSE.

resume

Whether to reuse complete existing extraction outputs. Default: TRUE.

reporter

Optional workflow reporter used by task-level runs.

Returns

A data.table of processed extraction plan rows.

Method query()

Run a DuckDB SQL query against the extraction manifest.

Usage

EsgStore$query(sql)

Arguments

sql

SQL query.

Returns

A data.table.

Method summarise()

Summarise extracted Parquet outputs by manifest columns.

Usage

EsgStore$summarise(
  by = c("source_id", "experiment_id", "variant_label", "frequency", "variable_id",
    "site_id", "year")
)

Arguments

by

Character vector of grouping columns. Default groups by source, experiment, variant, frequency, variable, site and year.

Returns

A data.table.

Method coverage()

Check extraction coverage for planned jobs.

Usage

EsgStore$coverage(plan_id = NULL)

Arguments

plan_id

Optional plan IDs to check.

Returns

A data.table with one row per plan.

Method assert_complete()

Assert that selected extraction plans are complete.

Usage

EsgStore$assert_complete(plan_id = NULL)

Arguments

plan_id

Optional plan IDs to check.

Returns

The store object itself, invisibly.

Method clone()

The objects of this class are cloneable with this method.

Usage

EsgStore$clone(deep = FALSE)

Arguments

deep

Whether to make a deep clone.