graflo.hq.sync.run

Write a grouped sync run: ensure every vertex, then rewrite the changed owners.

The input is staged, folded and grouped exactly as a bulk load's (see :mod:graflo.hq.bulk.run and :mod:graflo.hq.sync.group), so a sync of a release writes what a bulk load of it would. Vertices come first, key-only for the owned types, so an owned edge finds both endpoints whichever chunk writes it. Every chunk is recorded in the run ledger; a stopped run resumes.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

SyncKeyOnlyError

Bases: ValueError

A vertex type the run does not own carries more than its key.

Source code in graflo/hq/sync/run.py

class SyncKeyOnlyError(ValueError):
    """A vertex type the run does not own carries more than its key."""

SyncReport

Bases: BaseModel

Owners rewritten and owners skipped, per owned vertex type.

Source code in graflo/hq/sync/run.py

class SyncReport(BaseModel):
    """Owners rewritten and owners skipped, per owned vertex type."""

    written: dict[str, int] = {}
    skipped: dict[str, int] = {}

Attributes

skipped = {} class-attribute instance-attribute

written = {} class-attribute instance-attribute

Functions:

check_key_only(schema, staging_dir, counts, owned)

Refuse a folded vertex of a type the run does not own that has properties.

Such a type is only ever created by a sync run, never updated, so a property on it would go stale. Identity and reserved fields are its key and its stamp, and may be set.

Raises:

Type	Description
SyncKeyOnlyError	Naming each type and the fields found.

Source code in graflo/hq/sync/run.py

def check_key_only(
    schema: Schema, staging_dir: Path, counts: FoldCounts, owned: set[str]
) -> None:
    """Refuse a folded vertex of a type the run does not own that has properties.

    Such a type is only ever created by a sync run, never updated, so a
    property on it would go stale. Identity and reserved fields are its key and
    its stamp, and may be set.

    Raises:
        SyncKeyOnlyError: Naming each type and the fields found.
    """
    vc = schema.core_schema.vertex_config
    found: dict[str, set[str]] = {}
    for run in counts.runs:
        name = str(run.type.name)
        if run.type.kind != "vertex" or name in owned or not run.folded:
            continue
        allowed = set(vc.identity_fields(name)) | {
            f.name for f in vc.properties(name) if f.reserved
        }
        for row in read_rows(folded_path(staging_dir, run.slot)):
            extra = {k for k, v in row["doc"].items() if v is not None} - allowed
            if extra:
                found.setdefault(name, set()).update(extra)
    if found:
        listed = "; ".join(f"{k}: {sorted(v)}" for k, v in sorted(found.items()))
        raise SyncKeyOnlyError(
            "a sync run writes only the key of a type it does not own, but these "
            f"carry properties: {listed}"
        )

write_sync(schema, conn_conf, staging_dir, ledger, cfg, owner_runs, owned)

Ensure the folded vertices, then rewrite every owner whose hash changed.

Parameters:

Name	Type	Description	Default
schema	Schema	The schema, logical names, initialized for the target flavor.	required
conn_conf	DBConfig	The target; it must declare supports_sync.	required
staging_dir	Path	A folded and grouped staging directory.	required
ledger	RunLedger	Its ledger.	required
cfg	BulkConfig	Chunk size and parallelism.	required
owner_runs	list[OwnerRun]	The grouped owner types.	required
owned	Mapping[EdgeId, OwnedEdge]	How each owned edge is owned, by declared edge id.	required

Raises:

Type	Description
ValueError	The target has no sync writer.

Source code in graflo/hq/sync/run.py

def write_sync(
    schema: Schema,
    conn_conf: DBConfig,
    staging_dir: Path,
    ledger: RunLedger,
    cfg: BulkConfig,
    owner_runs: list[OwnerRun],
    owned: Mapping[EdgeId, OwnedEdge],
) -> SyncReport:
    """Ensure the folded vertices, then rewrite every owner whose hash changed.

    Args:
        schema: The schema, logical names, initialized for the target flavor.
        conn_conf: The target; it must declare ``supports_sync``.
        staging_dir: A folded and grouped staging directory.
        ledger: Its ledger.
        cfg: Chunk size and parallelism.
        owner_runs: The grouped owner types.
        owned: How each owned edge is owned, by declared edge id.

    Raises:
        ValueError: The target has no sync writer.
    """
    ConnectionManager.require(conn_conf, ConnectionCapability.SYNC)
    counts = FoldCounts.load(staging_dir)
    ctx = WriteContext.for_flavor(schema, conn_conf.connection_type)
    target_schema = physical_schema(schema, conn_conf.connection_type)
    owner_types = {run.owner for run in owner_runs}
    edge_types = {
        run.owner: _edge_types(schema, ctx, run.owner, owned) for run in owner_runs
    }
    with ConnectionManager(connection_config=conn_conf) as db:
        db.sync_begin(
            target_schema,
            [t for types in edge_types.values() for _, t in types],
        )
    ledger.mark(Phase.BEGUN)
    report = SyncReport()
    lock = threading.Lock()

    def owners(chunk: SyncOwnerChunk, owner: str) -> None:
        done = conns.get().sync_write_owners(target_schema, chunk)
        with lock:
            report.written[owner] = report.written.get(owner, 0) + done.written
            report.skipped[owner] = report.skipped.get(owner, 0) + done.skipped

    with _Connections(conn_conf) as conns:
        _write_chunks(
            (
                c
                for run in counts.runs
                if run.type.kind == "vertex"
                for c in _ensure_chunks(ctx, staging_dir, run, owner_types, cfg)
            ),
            lambda chunk: conns.get().sync_ensure_vertices(target_schema, chunk),
            ledger,
            cfg.parallelism,
        )
        _write_chunks(
            (
                c
                for run in owner_runs
                for c in _owner_chunks(
                    schema, ctx, staging_dir, run, edge_types[run.owner], cfg
                )
            ),
            lambda chunk: owners(chunk, chunk.chunk_id.split(":")[1]),
            ledger,
            cfg.parallelism,
        )
    ledger.mark(Phase.FINISHED)
    for owner in sorted(report.written):
        logger.info(
            "Synced %s: %s owners rewritten, %s unchanged",
            owner,
            report.written[owner],
            report.skipped.get(owner, 0),
        )
    return report