Function test_sqlite_snapshot_write

python/pathway/tests/test_sqlite.py:291–339

Snapshot-mode writer maintains the current state of the Pathway table: insertions UPSERT on the primary key, deletions remove the matching row. After the pipeline finishes, the SQLite table contains exactly the logically-live rows — earlier values for an updated key do not leak, and

(tmp_path: pathlib.Path)

def test_sqlite_snapshot_write(tmp_path: pathlib.Path):
    """Snapshot-mode writer maintains the current state of the Pathway
    table: insertions UPSERT on the primary key, deletions remove the
    matching row. After the pipeline finishes, the SQLite table
    contains exactly the logically-live rows — earlier values for an
    updated key do not leak, and retracted-only rows are gone.
    """
    database_path = tmp_path / "birds.db"

    def payload(key: int, genus: str, epithet: str) -> bytes:
        return json.dumps({"key": key, "genus": genus, "epithet": epithet}).encode()

    class TestSubject(pw.io.python.ConnectorSubject):
        def run(self):
            # Initial insertions.
            self._add(api.ref_scalar(1), payload(1, "upupa", "epops"))
            self._add(api.ref_scalar(2), payload(2, "bubo", "scandiacus"))
            self._add(api.ref_scalar(3), payload(3, "corvus", "corax"))
            # Update row 1: retract the old version and insert a new one.
            # Snapshot mode should UPSERT on primary_key=key.
            self._remove(api.ref_scalar(1), payload(1, "upupa", "epops"))
            self._add(api.ref_scalar(1), payload(1, "upupa", "marginata"))
            # Delete row 3 outright.
            self._remove(api.ref_scalar(3), payload(3, "corvus", "corax"))

    class InputSchema(pw.Schema):
        key: int
        genus: str
        epithet: str

    table = pw.io.python.read(TestSubject(), schema=InputSchema)
    pw.io.sqlite.write(
        table,
        database_path,
        "birds",
        output_table_type="snapshot",
        primary_key=[table.key],
        init_mode="create_if_not_exists",
    )
    run_all()

    connection = sqlite3.connect(database_path)
    rows = list(
        connection.execute("SELECT key, genus, epithet FROM birds ORDER BY key")
    )
    assert rows == [
        (1, "upupa", "marginata"),
        (2, "bubo", "scandiacus"),
    ]


@pytest.mark.parametrize("init_mode", ["default", "create_if_not_exists", "replace"])