@spec count(GeoGenius.Context.t(), Ecto.UUID.t()) :: non_neg_integer()

The number of rows currently staged for a run.

@spec create(GeoGenius.Context.t(), Ecto.UUID.t(), Ecto.UUID.t()) :: String.t()

Creates a run's staging table, if it does not already exist, returning its name.

@spec drop(GeoGenius.Context.t(), Ecto.UUID.t()) :: :ok

Drops a terminal or orphaned run's staging table, if it exists.

@spec drop(GeoGenius.Context.t(), Ecto.UUID.t(), Ecto.UUID.t()) :: :ok

Drops the current executor's active staging table, if it exists.

@spec insert(
  GeoGenius.Context.t(),
  Ecto.UUID.t(),
  Ecto.UUID.t(),
  [GeoGenius.Staging.Row.t()],
  keyword()
) :: non_neg_integer()

Bulk-inserts rows, returning the number of rows written.

Issues one round trip regardless of batch size: the three fields unzip into parallel arrays and cross as a single unnest insert rather than as one statement per row. payload binds as a list of maps, never as pre-encoded JSON text -- a string would decode back as a jsonb string value holding an escaped object rather than the object itself, and no constraint catches that because a JSON string is valid jsonb.

Returns 0 without touching the database for an empty list, but still validates run_id first, so a malformed run id fails the same way it would for a non-empty batch rather than being masked by the empty-list shortcut.

opts reaches Repo.query/3. One batch of a national artifact is a single multi-megabyte unnest insert, which is exactly the shape DBConnection's fifteen-second default cuts short, so a caller raises :timeout here rather than on its whole Repo. The insert always forces checkout_retries: 0: after a disconnect, only the pipeline can safely decide whether to start a new import attempt rather than repeating an ambiguously completed mutation.

@spec leaked(GeoGenius.Context.t()) :: [{Ecto.UUID.t(), String.t()}]

Every staging table no run still needs.

A table is leaked when its run has reached a terminal status, and also when no import_run row carries its id at all. The second case is the one that cannot be recovered any other way: an earlier version of this function joined import_run to pg_class, so a table whose run row was gone had nothing to join to, never appeared here, and mix geo_genius.sweep_staging could never drop it. The schema down migration refuses to drop a non-empty schema, so an unreclaimable table eventually blocks an uninstall outright.

The scan therefore starts from pg_class and left-joins import_run. Candidate names must match staging_ followed by 32 hex characters, and the run id is read back out of the name, so a table with no run row still yields the id GeoGenius.Catalog.drop_staging/2 needs. That id is then round-tripped: a row survives only when staging_table_name(id) reproduces the table's own name exactly, which keeps PostgreSQL the authority on how a staging table is named and means nothing whose name this library did not derive is ever offered up for dropping.

Returns {run_id, table_name} pairs ordered by table name.

@spec reset(GeoGenius.Context.t(), Ecto.UUID.t(), Ecto.UUID.t()) :: String.t()

Empties a run's staging table, creating it if it is not there, and returns its name.

create/3 alone is CREATE UNLOGGED TABLE IF NOT EXISTS, which is what a staging pass wants on the second call and not what it wants at the start of execution. A table can survive an abrupt worker or VM death, or be created manually before the pipeline starts. Neither case authorizes those rows as input to a future executor.

So a pass starts from a table that is empty by construction. The table is dropped rather than truncated, which also resets the identity sequence the keyset pagination in stream/3 reads through, and leaves nothing of the previous attempt's shape behind for this one to inherit.

Nothing durable is lost by emptying it: staging is reproducible scratch space, not attempt evidence or a recovery checkpoint. A corrected retry has a new run and stages from its own artifact snapshot. What makes that retry cheap is the artifact cache, which skips the network.

@spec stream(GeoGenius.Context.t(), Ecto.UUID.t(), keyword()) :: Enumerable.t()

Streams a run's staged rows in insertion order.

Reads through keyset pagination -- WHERE id > $1 ORDER BY id LIMIT $2 -- rather than Ecto.Adapters.SQL.stream/4, which needs an enclosing transaction. Normalization writes hundreds of thousands of rows through separate statements after each row is read, and holding one transaction open for that whole phase would pin a connection and a snapshot for as long as the slowest phase in the pipeline runs. Each page carries the last row's id forward as the next page's lower bound rather than an offset, since OFFSET re-scans and a concurrent insert would shift it.

:batch_size defaults to 1000 and must be a positive integer -- a batch size of zero would page with LIMIT 0, which always returns no rows and would silently stream nothing rather than fail.

@spec table_name(Ecto.UUID.t()) :: String.t()

The staging table's name for a run.

Derived from the run's uuid, so the identifier is [0-9a-f_]+ by construction rather than by escaping. staging_table_name/1 in SQL applies the same rule; the test asserts the two agree rather than trusting that they do.