Skip to main content

selection

Read-time selection of a datasource's files from the pod-wide inventory.

file_metadata is a datasource-agnostic inventory keyed on file_path: one row per physical file, so a file shared by several datasources in a pod is stored — and SHA-256 hashed — once.

task_hash survives on the row, but only as provenance: store_file_metadata upserts by file_path, so the column holds whichever datasource indexed the file most recently. It therefore cannot be used to select a datasource's files. For a file present in two datasources, the later walk takes ownership of the row and the earlier datasource's listing silently loses that file — and a file missing from $file_metadata.cache is never evaluated, so it produces no eligibility row at all. So selection lives with the datasource and is computed at read time from its own root path and accepted extensions. scan_metadata already worked this way; this module is the single implementation, shared by that runtime and by the DAG's $file_metadata context provider, so the two cannot drift.

Datasources with no filesystem root (SQL/DB-backed sources, which write inventory rows carrying row_number rather than a path under a walk root) have nothing to select by, and no way to collide with another datasource's files either. For those datasource_root returns None and the caller keeps the task_hash scope.

Module​

Functions​

accepted_extensions​

def accepted_extensions(datasource: BaseSource) ‑> set[str] | None:

Return datasource's accepted file extensions (lowercased), or None.

Read from the datasource's FileSystemFilter (datasource.filter). Keeps selection from handing a consumer files the datasource would never load (e.g. a stray .txt sharing the root). None means no extension restriction, so every row under the root is considered.

Probed rather than typed, for the same reason as datasource_root: a source reaching here always has a filter in practice, and a missing one reads as "no restriction" — which keeps files in rather than dropping them.

Arguments

  • datasource: The datasource being selected for.

Returns Lowercased extensions, or None for no restriction.

datasource_root​

def datasource_root(datasource: BaseSource) ‑> Path | None:

Return datasource's normalized selection root, or None if it has none.

datasource.path is what file_metadata indexed, so every file the datasource owns is that path or sits under it. It is not necessarily a directory: a single-file source (CSVSource) points at the file itself, and selects accounts for that.

Inventory rows store paths normalized by fs_utils.normalize_path, so the root goes through the same function and the containment test compares like with like — path_is_under is purely lexical and will not rescue a mismatch. Normalized strictly, for the same reason the indexing walk is: a non-strict resolve() answers an unreachable share with the unresolved spelling and no error, and a root that no longer lexically matches the stored rows silently selects nothing at all. Failing loudly beats handing a consumer an empty listing that looks like "this datasource has no files".

This is the predicate callers use to decide whether path selection applies at all, so path is probed rather than assumed: a None answer is what routes a source with no filesystem root to the task_hash scope. Probing beats an isinstance check here because it fails toward doing the selection — see FileMetadataContext._select_rows.

Arguments

  • datasource: Any datasource.

Returns The normalized root, or None when the datasource has nothing on the filesystem to select by: a SQL source (no path), or a path that is not a filesystem location at all, such as CSVSource's http(s) URL form.

Raises

  • OSError: When path is a filesystem path that cannot be resolved — unreachable share, permission denied, or simply not there — even after normalize_path's retries. Deliberately not downgraded to None: see below.

select_for_datasource​

def select_for_datasource(    records: Iterable[FileMetadataRecord],    datasource: FileSystemIterableSource,    root: Path,) ‑> Iterator[FileMetadataRecord]:

Filter a pod-wide inventory down to datasource's own files.

Takes the root as an argument rather than deriving it: a caller has to look it up anyway to know whether path selection applies at all (datasource_root returns None for a SQL source, which keeps the task_hash scope), so re-deriving it here would resolve the path twice and turn a case the caller already handled into an exception it has to know not to trigger.

Arguments

  • records: The whole inventory, e.g. from file_metadata.v1.store.get_file_metadata_inventory. Streamed, so a large inventory is not materialised.
  • datasource: The datasource being selected for, read for its extension filter.
  • root: Its resolved walk root, from datasource_root.

selects​

def selects(record: FileMetadataRecord, root: Path, extensions: set[str] | None) ‑> bool:

Return whether record is one of the datasource's files.

Arguments

  • record: An inventory row.
  • root: The datasource's normalized selection root, from datasource_root.
  • extensions: The datasource's accepted extensions, from accepted_extensions; None for no restriction.

Returns True when the row is root itself or sits under it, and passes the extension filter.