apexbase 1.35.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

"""Non-transactional secondary-index failure contract (C3.2).

A non-transactional write persists its row before it maintains secondary
indexes. When that index maintenance fails, the error must surface, reads
must fall back to the authoritative scan, and the client must not replay
the mutation through its Arrow IPC fallback (a replay would apply the row
twice, because the first attempt already wrote it).
"""

import os
import tempfile

import pytest

from apexbase import ApexClient


def _block_index_sidecar(index_path):
    """Make persisting the index sidecar fail on every platform.

    A read-only file only fails the save on POSIX; occupying the sidecar
    path with a directory is rejected by the writer everywhere.
    """
    os.remove(index_path)
    os.mkdir(index_path)


def _unblock_index_sidecar(index_path):
    os.rmdir(index_path)


def test_failed_insert_is_not_replayed_and_falls_back_to_scan():
    with tempfile.TemporaryDirectory() as d:
        c = ApexClient(d, _auto_manage=False, durability="safe")
        c.create_table("t", {"value": "int"})
        c.use_table("t")
        c.store([{"value": 1}, {"value": 2}])
        c.flush()
        c.execute("CREATE INDEX idx_value ON t (value) USING HASH")

        index_path = os.path.join(d, "indexes", "t_idx_value.hashidx")
        stale_path = os.path.join(d, "t.apex.index.stale")
        assert os.path.exists(index_path)
        assert not os.path.exists(stale_path)

        # A real write failure: the index file cannot be persisted.
        _block_index_sidecar(index_path)
        try:
            with pytest.raises(RuntimeError):
                c.execute("INSERT INTO t (value) VALUES (999)")
        finally:
            _unblock_index_sidecar(index_path)

        # The row was written exactly once. A replayed INSERT would make it 4.
        assert c.execute("SELECT COUNT(*) FROM t").scalar() == 3
        # The failed index must not be used; the scan finds the row.
        assert c.execute("SELECT value FROM t WHERE value = 999").scalar() == 999
        assert os.path.exists(stale_path)
        c.close()

        # The durable marker survives a reopen and keeps reads authoritative.
        c2 = ApexClient(d, _auto_manage=False, durability="safe")
        c2.use_table("t")
        assert c2.execute("SELECT COUNT(*) FROM t").scalar() == 3
        c2.execute("REINDEX t")
        assert not os.path.exists(stale_path)
        assert c2.execute("SELECT value FROM t WHERE value = 999").scalar() == 999
        c2.close()


def test_stale_marker_is_reaped_with_dropped_table():
    with tempfile.TemporaryDirectory() as d:
        c = ApexClient(d, _auto_manage=False, durability="safe")
        c.create_table("t", {"value": "int"})
        c.use_table("t")
        c.store([{"value": 1}])
        c.flush()

        stale_path = os.path.join(d, "t.apex.index.stale")
        with open(stale_path, "w") as handle:
            handle.write("index maintenance incomplete\n")
        c.execute("DROP TABLE t")
        c.close()

        # A same-name recreation must not inherit the old read fallback.
        c2 = ApexClient(d, _auto_manage=False, durability="safe")
        c2.create_table("t", {"value": "int"})
        c2.use_table("t")
        assert not os.path.exists(stale_path)
        c2.close()