Function test_sqlite_write_read_round_trip

python/pathway/tests/test_sqlite.py:237–288

End-to-end round-trip: write a table covering every Pathway type to SQLite via ``pw.io.sqlite.write``, read it back through ``pw.io.sqlite.read``, and verify that both paths produce identical jsonlines — i.e. the writer/reader pair round-trips every type without loss. Exercised in bo

(
    tmp_path: pathlib.Path, serialization_tester, output_table_type
)

@needs_multiprocessing_fork
@pytest.mark.parametrize("output_table_type", ["stream_of_changes", "snapshot"])
def test_sqlite_write_read_round_trip(
    tmp_path: pathlib.Path, serialization_tester, output_table_type
):
    """End-to-end round-trip: write a table covering every Pathway type to
    SQLite via ``pw.io.sqlite.write``, read it back through
    ``pw.io.sqlite.read``, and verify that both paths produce identical
    jsonlines — i.e. the writer/reader pair round-trips every type
    without loss. Exercised in both ``stream_of_changes`` and
    ``snapshot`` output modes.
    """
    database_path = tmp_path / "variety.db"
    direct_path = tmp_path / "direct.jsonl"
    roundtrip_path = tmp_path / "roundtrip.jsonl"

    # Ground truth: write the original table straight to jsonlines.
    table, known_rows = serialization_tester.create_variety_table(with_optionals=False)
    schema = table.schema
    pw.io.jsonlines.write(table, direct_path)
    run_all()
    G.clear()

    # Writer leg: same table, written through the SQLite output connector.
    table, _ = serialization_tester.create_variety_table(with_optionals=False)
    write_kwargs: dict = {"init_mode": "create_if_not_exists"}
    if output_table_type == "snapshot":
        write_kwargs["output_table_type"] = "snapshot"
        write_kwargs["primary_key"] = [table.pkey]
    pw.io.sqlite.write(table, database_path, "variety", **write_kwargs)
    run_all()
    G.clear()

    # Reader leg: read the SQLite table back and re-emit jsonlines.
    read_table = pw.io.sqlite.read(
        database_path, "variety", schema, autocommit_duration_ms=1
    )
    pw.io.jsonlines.write(read_table, roundtrip_path)
    wait_result_with_checker(
        FileLinesNumberChecker(roundtrip_path, len(known_rows)), 30
    )

    def load(path: pathlib.Path) -> list[dict]:
        with open(path) as f:
            rows = [json.loads(line) for line in f]
        # The time/diff metadata differs between runs; compare only the
        # data columns.
        for row in rows:
            row.pop("time", None)
            row.pop("diff", None)
        rows.sort(key=lambda r: r["pkey"])
        return rows

    assert load(direct_path) == load(roundtrip_path)


def test_sqlite_snapshot_write(tmp_path: pathlib.Path):