apexbase 1.25.0

High-performance HTAP embedded database with Rust core
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

"""Architecture contracts for cross-layer cache invalidation."""

from apexbase import ApexClient


def _open_clients(tmp_path):
    writer = ApexClient(str(tmp_path))
    writer.create_table(
        "cache_contract",
        {"name": "string", "score": "int", "category": "string"},
    )
    writer.use_table("cache_contract")
    writer.store(
        [
            {"name": "alpha", "score": 10, "category": "x"},
            {"name": "beta", "score": 20, "category": "y"},
        ]
    )
    writer.flush()

    reader = ApexClient(str(tmp_path))
    reader.use_table("cache_contract")
    return writer, reader


def _warm(reader, sql):
    first = reader.execute(sql).to_dict()
    assert reader.execute(sql).to_dict() == first
    return first


def test_direct_write_invalidates_python_result_cache_across_clients(tmp_path):
    writer, reader = _open_clients(tmp_path)
    sql = "SELECT * FROM cache_contract WHERE score > 15 LIMIT 100"

    try:
        assert [row["name"] for row in _warm(reader, sql)] == ["beta"]
        assert reader._simple_sql_cache[sql][0] == "numeric_range_limit"
        assert reader._numeric_range_rows_cache

        assert writer.replace(
            1, {"name": "alpha", "score": 30, "category": "x"}
        )
        assert [row["name"] for row in reader.execute(sql).to_dict()] == [
            "alpha",
            "beta",
        ]
    finally:
        reader.close()
        writer.close()


def test_batch_write_advances_table_epoch_once(tmp_path):
    writer, reader = _open_clients(tmp_path)

    try:
        before = writer._storage._table_epoch()
        writer.store(
            [
                {"name": "gamma", "score": 30, "category": "x"},
                {"name": "delta", "score": 40, "category": "z"},
            ]
        )
        after = writer._storage._table_epoch()

        assert after == before + 1
        assert reader.execute(
            "SELECT name FROM cache_contract WHERE score >= 30 ORDER BY score"
        ).to_dict() == [{"name": "gamma"}, {"name": "delta"}]
    finally:
        reader.close()
        writer.close()


def test_transaction_commit_invalidates_all_warmed_read_caches(tmp_path):
    writer, reader = _open_clients(tmp_path)
    group_sql = (
        "SELECT category, COUNT(*) AS n FROM cache_contract "
        "GROUP BY category ORDER BY category"
    )
    point_sql = "SELECT name, score FROM cache_contract WHERE _id = 2"

    try:
        assert _warm(reader, group_sql) == [
            {"category": "x", "n": 1},
            {"category": "y", "n": 1},
        ]
        assert _warm(reader, point_sql) == [{"name": "beta", "score": 20}]
        assert _warm(reader, "SELECT COUNT(*) FROM cache_contract") == [
            {"COUNT(*)": 2}
        ]

        writer.execute("BEGIN")
        writer.execute(
            "INSERT INTO cache_contract (name, score, category) "
            "VALUES ('gamma', 30, 'x')"
        )
        writer.execute("UPDATE cache_contract SET score = 25 WHERE _id = 2")
        writer.execute("COMMIT")

        assert reader.execute(group_sql).to_dict() == [
            {"category": "x", "n": 2},
            {"category": "y", "n": 1},
        ]
        assert reader.execute(point_sql).to_dict() == [
            {"name": "beta", "score": 25}
        ]
        assert reader.execute("SELECT COUNT(*) FROM cache_contract").to_dict() == [
            {"COUNT(*)": 3}
        ]
    finally:
        reader.close()
        writer.close()


def test_schema_rewrite_invalidates_cached_backend_schema(tmp_path):
    writer, reader = _open_clients(tmp_path)

    try:
        assert reader.list_fields() == ["name", "score", "category"]
        _warm(reader, "SELECT name, score FROM cache_contract")

        writer.execute("ALTER TABLE cache_contract ADD COLUMN note STRING")

        assert reader.list_fields() == ["name", "score", "category", "note"]
        assert reader.execute("SELECT name, note FROM cache_contract").to_dict() == [
            {"name": "alpha", "note": ""},
            {"name": "beta", "note": ""},
        ]
    finally:
        reader.close()
        writer.close()