Row-count sweep

count_by_folder walks the table metadata in a folder and records the current row count of every table across the requested lakehouse layers. It is an inventory and trending function: a lightweight, read-only sweep that runs as the final stage of a pipeline and makes zero changes to the hot path.

Overview

The sweep exists to answer "how many rows are in each table right now, per run" without touching the loading logic. It is the b job in a pipeline — the observational tail that runs after all loads (the a jobs) have completed:

• It only ever reads — it counts rows, it never optimizes, vacuums, or writes data. Nothing about it can affect a load.
• It runs last, so a slow or failing sweep cannot delay or break the tables it is measuring. Counting is off the critical path.
• It emits one record per (run, table, layer), each carrying the run identifiers so downstream analytics can join sweeps to loads and to each other.

Because it is metadata-driven, adding a table to a folder automatically brings it into the sweep — no per-table wiring.

How it works

For every table resolved from the folder, and for every requested layer, the function:

1. Resolves the full table name for the layer (e.g. Bronze.dbo.customers).
2. Reads the row count with spark.table(name).count().
3. Opens a dedicated log segment and records the count together with the run identifiers, then emits an MSG_MAINT_ROWCOUNT log line.

Counts run concurrently (max_workers, default 10). A table that cannot be resolved or read is logged and skipped — it never aborts the sweep — so one bad table does not cost you the counts for all the others.

Run identifiers and the downstream join

The sweep reads the logging run identifiers (run_id, activity_id, parent_run_id) from the logging config globals that set_run_id / init_logging populate. If logging has not been initialized yet, the sweep bootstraps it (log_source="Maintenance", log_object="count_by_folder"), so it is safe to call standalone.

Because every record carries the run_id, the sweep output is joinable on run_id downstream. That is what turns per-run snapshots into trends: join a run's sweep to that same run's load telemetry to see, per table, what was loaded versus what is now on disk.

Full-load reconciliation

For a full load, the sweep count is a direct reconciliation check against the load: the swept row count should equal the number of rows the load read from source, i.e.

sweep_count == source_in

A mismatch on a full-load table is a signal — rows were dropped, duplicated, or filtered somewhere between source and the materialized table — and the join on run_id is what lets you spot it per run.

Usage

from easyfabric.maintenance import count_by_folder

counts = count_by_folder(
    "Files/Objects/Sales",
    layers=["Bronze", "Silver", "Gold"],
)

# counts == [
#   {"layer": "Bronze", "table": "Bronze.dbo.customers", "row_count": 1042},
#   {"layer": "Silver", "table": "Silver.dbo.customers", "row_count": 1042},
#   ...
# ]

Run it as the last cell of the orchestration notebook, after all loads have finished, so the counts reflect the fully materialized state of the run.

Parameters

Parameter	Type	Default	Description
table_folder	str	Required	Folder containing the YAML table metadata to sweep.
config_manager	ConfigManager	None	Config manager instance; initialized automatically when omitted.
except_folders	list[str]	[]	Sub-folders to exclude from the sweep.
except_files	list[str]	[]	Individual metadata files to exclude.
layers	list[str]	Required	Layers to count, any of ["Bronze", "Silver", "Gold"] (case-insensitive).
max_workers	int	10	Maximum concurrent counts.

Returns: list[dict], one entry per successfully counted table with layer, table, and row_count.

Raises: ValueError if layers is empty or contains a value other than Bronze, Silver, or Gold.

Cost

The sweep is deliberately cheap and can be left on:

• Metadata-only when Delta statistics are enabled. count() on a Delta table is answered from the transaction-log statistics, so the count is served from metadata rather than a full data scan.
• N counts per run — one per (table × layer), and no more. The work scales linearly with the number of tables, runs concurrently, and sits off the critical path at the end of the pipeline, so it adds observability without taxing the loads.