Skip to content

graflo.hq.sync.group

Group folded rows by owner, and give every owner vertex its sync hash.

Runs after the fold, on the staging directory. For each owned vertex type it writes owners/<slot>.jsonl.zst: one row per folded owner vertex, in key order, with the owner's document and every edge it owns. The owner's sync hash is set on that document and on the folded vertex row, so a bulk load writes it and a sync run compares it.

Attributes

OWNERS = 'owners' module-attribute

RUNS = 'runs.json' module-attribute

Classes

OwnerRun

Bases: BaseModel

One owned vertex type's grouped rows.

Source code in graflo/hq/sync/group.py 
28
29
30
31
32
33
34
35
36
37
class OwnerRun(BaseModel):
    """One owned vertex type's grouped rows."""

    owner: str
    slot: str
    """The vertex run's slot; also names the owners file."""
    hash_field: str
    edges: list[RunCounts]
    """The edge runs the owner type owns."""
    owners: int

Attributes

edges instance-attribute

The edge runs the owner type owns.

hash_field instance-attribute
owner instance-attribute
owners instance-attribute
slot instance-attribute

The vertex run's slot; also names the owners file.

Functions:

group_owners(staging_dir, counts, hash_fields)

Group every owned type of hash_fields and set its owners' hashes.

Parameters:

Name Type Description Default
staging_dir Path

A folded staging directory whose edge rows carry okey.

 required
counts FoldCounts

Its fold counts.

 required
hash_fields Mapping[str, str]

Owned vertex type -> its set_by: sync field.

 required
Source code in graflo/hq/sync/group.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def group_owners(
    staging_dir: Path, counts: FoldCounts, hash_fields: Mapping[str, str]
) -> list[OwnerRun]:
    """Group every owned type of *hash_fields* and set its owners' hashes.

    Args:
        staging_dir: A folded staging directory whose edge rows carry ``okey``.
        counts: Its fold counts.
        hash_fields: Owned vertex type -> its ``set_by: sync`` field.
    """
    root = Path(staging_dir)
    out: list[OwnerRun] = []
    for owner, hash_field in sorted(hash_fields.items()):
        vertex = counts.vertex(owner)
        if vertex is None:
            continue
        edges = [
            r for r in counts.runs if r.type.kind == "edge" and r.type.owner == owner
        ]
        raw = owners_path(root, vertex.slot).with_suffix(".raw")
        _group(root, vertex, edges, raw)
        path = owners_path(root, vertex.slot)
        n = write_rows(path, _hashed(read_rows(raw), hash_field))
        raw.unlink()
        _rehash_vertices(folded_path(root, vertex.slot), path, hash_field)
        out.append(
            OwnerRun(
                owner=owner,
                slot=vertex.slot,
                hash_field=hash_field,
                edges=edges,
                owners=n,
            )
        )
    (root / OWNERS).mkdir(parents=True, exist_ok=True)
    (root / OWNERS / RUNS).write_text(
        json.dumps([run.model_dump(mode="json") for run in out], indent=2)
    )
    return out

grouped_owners(staging_dir)

The owner runs :func:group_owners wrote, for a resumed run.

Source code in graflo/hq/sync/group.py 
85
86
87
88
89
90
def grouped_owners(staging_dir: Path) -> list[OwnerRun]:
    """The owner runs :func:`group_owners` wrote, for a resumed run."""
    path = Path(staging_dir) / OWNERS / RUNS
    if not path.exists():
        return []
    return [OwnerRun.model_validate(run) for run in json.loads(path.read_text())]

owners_path(staging_dir, slot)

Source code in graflo/hq/sync/group.py 
40
41
def owners_path(staging_dir: Path, slot: str) -> Path:
    return Path(staging_dir) / OWNERS / f"{slot}.jsonl.zst"