ddcp 0.2.4

Distributed decentralized database-to-database copy
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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275

from crsql_correctness import connect, close, min_db_v
from pprint import pprint


def sync_left_to_right(l, r):
    changes = l.execute("SELECT * FROM crsql_changes")
    for change in changes:
        r.execute(
            "INSERT INTO crsql_changes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", change)
    r.commit()


def test_increments_by_one_in_tx():
    c = connect(":memory:")
    c.execute("create table foo (id primary key, a)")
    c.execute("select crsql_as_crr('foo')")
    c.commit()

    c.execute("INSERT INTO foo VALUES (1, 2)")
    c.execute("INSERT INTO foo VALUES (2, 3)")
    c.commit()

    rows = c.execute("SELECT __crsql_seq FROM foo__crsql_clock").fetchall()
    assert (rows == [(0,), (1,)])


def test_resets_on_every_tx():
    c = connect(":memory:")
    c.execute("create table foo (id primary key, a)")
    c.execute("select crsql_as_crr('foo')")
    c.commit()

    c.execute("INSERT INTO foo VALUES (1, 2)")
    c.execute("INSERT INTO foo VALUES (2, 3)")
    c.commit()

    rows = c.execute("SELECT __crsql_seq FROM foo__crsql_clock").fetchall()
    assert (rows == [(0,), (1,)])

    c.execute("INSERT INTO foo VALUES (3, 4)")
    c.execute("INSERT INTO foo VALUES (5, 6)")
    c.commit()

    rows = c.execute("SELECT __crsql_seq FROM foo__crsql_clock").fetchall()
    assert (rows == [(0,), (1,), (0,), (1,)])


def test_preserved_on_merge():
    # the order we insert changes into crsql_changes should be the order they come back out when using seq
    c = connect(":memory:")
    c.execute("create table foo (id primary key, a)")
    c.execute("select crsql_as_crr('foo')")
    c.commit()

    c2 = connect(":memory:")
    c2.execute("create table foo (id primary key, a)")
    c2.execute("select crsql_as_crr('foo')")
    c2.commit()

    c.execute("INSERT INTO foo VALUES (1, 2)")
    c.execute("INSERT INTO foo VALUES (2, 3)")
    c.execute("INSERT INTO foo VALUES (3, 4)")
    c.execute("INSERT INTO foo VALUES (5, 6)")
    c.execute("UPDATE foo SET a = 1 WHERE id = 1")
    c.execute("DELETE FROM foo WHERE id = 2")
    c.commit()

    sync_left_to_right(c, c2)
    c_rows = c.execute("SELECT *, seq FROM crsql_changes").fetchall()
    c2_rows = c.execute("SELECT *, seq FROM crsql_changes").fetchall()

    assert (c_rows == c2_rows)


def test_incr_by_one():
    # insert
    # update
    # delete
    c = connect(":memory:")
    c.execute("create table foo (a primary key, b, c, d)")
    c.execute("select crsql_as_crr('foo')")
    c.commit()

    c.execute("INSERT INTO foo VALUES (1, 2, 3, 4)")
    c.execute("INSERT INTO foo VALUES (2, 2, 3, 4)")
    c.execute("INSERT INTO foo VALUES (3, 2, 3, 4)")
    c.commit()

    rows = c.execute(
        "SELECT seq FROM crsql_changes WHERE db_version = 1").fetchall()

    assert (rows == [(0,), (1,), (2,), (3,), (4,), (5,), (6,), (7,), (8,)])

    c.execute("UPDATE foo SET c = 'c', d = 'd' WHERE a = 1")
    c.execute("UPDATE foo SET c = 'c', d = 'd' WHERE a = 2")
    c.execute("UPDATE foo SET c = 'c', d = 'd' WHERE a = 3")
    c.commit()

    rows = c.execute(
        "SELECT seq FROM crsql_changes WHERE db_version = 2").fetchall()
    assert (rows == [(0,), (1,), (2,), (3,), (4,), (5,)])

    c.execute("UPDATE foo SET b = 'b' WHERE a = 1")
    c.execute("UPDATE foo SET b = 'b' WHERE a = 2")
    c.execute("UPDATE foo SET b = 'b' WHERE a = 3")
    c.commit()
    rows = c.execute(
        "SELECT seq FROM crsql_changes WHERE db_version = 3").fetchall()
    assert (rows == [(0,), (1,), (2,)])

    c.execute("DELETE FROM foo")
    c.commit()

    rows = c.execute("SELECT seq FROM crsql_changes").fetchall()
    assert (rows == [(0,), (1,), (2,)])

    c.execute("create table bar (a primary key, b);")
    c.execute("select crsql_as_crr('bar')")
    c.commit()

    c.execute("INSERT INTO bar VALUES (1, 2)")
    c.execute("INSERT INTO bar VALUES (3, 4)")
    c.commit()

    c.execute("UPDATE bar SET b = 'b' WHERE a = 1")
    c.execute("UPDATE bar SET b = 'b' WHERE a = 3")
    c.commit()

    rows = c.execute(
        "SELECT __crsql_db_version, __crsql_seq FROM bar__crsql_clock ORDER BY __crsql_db_version ASC").fetchall()
    assert (rows == [(6, 0), (6, 1)])

    # test update of pk vals with col vals

    c.execute("CREATE TABLE baz (a primary key)")
    c.execute("SELECT crsql_as_crr('baz')")
    c.commit()
    c.execute("INSERT INTO baz VALUES (1)")
    c.execute("INSERT INTO baz VALUES (2)")
    c.commit()

    rows = c.execute("SELECT __crsql_seq FROM baz__crsql_clock").fetchall()
    assert (rows == [(0,), (1,)])

    c.execute("UPDATE baz SET a = 11 WHERE a = 1")
    c.execute("UPDATE baz SET a = 22 WHERE a = 2")
    c.commit()
    rows = c.execute(
        "SELECT __crsql_seq FROM baz__crsql_clock ORDER BY __crsql_db_version, __crsql_seq ASC").fetchall()
    assert (rows == [(0,), (1,), (2,), (3,)])

    # c.execute("DELETE FROM baz")

    # pprint(c.execute(
    #     "SELECT *, seq FROM crsql_changes WHERE [table] = 'baz'").fetchall())


# Test seq when resurrecting.
# When a row is deleted and then resurrected through a future insertion with the same primary key
# we have special logic for handling the seq value.
# This tests to ensure seq values are still assigned as expected.
def test_seq_when_re_inserting():
    c = connect(":memory:")
    c.execute("create table foo (a primary key, b)")
    c.execute("select crsql_as_crr('foo')")
    c.commit()

    c.execute("INSERT INTO foo VALUES (1, 2)")
    c.commit()
    seq = c.execute("SELECT seq FROM crsql_changes").fetchone()[0]
    assert (seq == 0)

    c.execute("DELETE FROM foo")
    c.commit()
    seq = c.execute("SELECT seq FROM crsql_changes").fetchone()[0]
    assert (seq == 0)

    # re-insert a row with the same primary key as the deleted one
    c.execute("INSERT INTO foo VALUES (1, 2)")
    c.commit()
    seqs = c.execute("SELECT seq FROM crsql_changes").fetchall()
    # Seqs are appropriately assigned
    # 0 for resurrect
    # 1 for setting b to 2
    assert (seqs == [(0,), (1,)])


# Similar to the above test (test_seq_when_re_inserting) but also performs some updates
# against the row that was re-inserted
def test_seq_when_updating_after_reinsert():
    c = connect(":memory:")
    c.execute("create table foo (a primary key, b)")
    c.execute("select crsql_as_crr('foo')")
    c.commit()

    c.execute("INSERT INTO foo VALUES (1, 2)")
    c.commit()

    c.execute("DELETE FROM foo")
    c.commit()

    # re-insert a row with the same primary key as the deleted one
    c.execute("INSERT INTO foo VALUES (1, 2)")
    c.execute("UPDATE foo SET b = 3 WHERE a = 1")
    seqs = c.execute("SELECT seq FROM crsql_changes").fetchall()
    c.commit()
    # 0 for resurrect
    # 2 for set b = 3 given 1 was used for b = 2 on the insert
    assert (seqs == [(0,), (2,)])


# Same as above tests but re-insertion is done via a merge with another node
# rather than a local insert.
def test_seq_when_resinserting_from_merge():
    c = connect(":memory:")
    c.execute("create table foo (a primary key, b)")
    c.execute("select crsql_as_crr('foo')")
    c.commit()

    c.execute("INSERT INTO foo VALUES (1, 2)")
    c.commit()
    c.execute("DELETE FROM foo")
    c.commit()

    c.execute("INSERT INTO foo VALUES (1, 2)")
    c.commit()

    c2 = connect(":memory:")
    c2.execute("create table foo (a primary key, b)")
    c2.execute("select crsql_as_crr('foo')")
    c2.commit()

    c2.execute("INSERT INTO foo VALUES (1, 2)")
    c2.commit()
    c2.execute("DELETE FROM foo")
    c2.commit()

    # merge the resurrect over
    sync_left_to_right(c, c2)

    seqs = c2.execute("SELECT seq FROM crsql_changes").fetchall()
    # Seqs are appropriately assigned
    # 0 for resurrect
    # 1 for setting b to 2
    assert (seqs == [(0,), (1,)])


# db is locked when doing concurrent writes so the following case cannot happen in SQLite
# def test_preserved_conc_transactions():
#     filename = "./test_preserved_conc_transactions.db"
#     try:
#         os.remove(filename)
#         os.remove("{}-wal".format(filename))
#         os.remove("{}-shm".format(filename))
#     except:
#         None

#     c1 = connect(filename)
#     c1.execute("PRAGMA journal_mode = WAL")
#     c1.execute("PRAGMA synchronous = NORMAL")
#     c2 = connect(filename)

#     c1.execute("create table foo (id primary key, a)")
#     c1.execute("select crsql_as_crr('foo')")
#     c1.commit()

#     c1.execute("INSERT INTO foo VALUES (1, 2)")
#     c2.execute("INSERT INTO foo VALUES (11, 1)")
#     c1.execute("INSERT INTO foo VALUES (2, 3)")
#     c2.execute("INSERT INTO foo VALUES (12, 2)")
#     c1.commit()
#     c2.commit()

#     rows = c1.execute("SELECT __crsql_seq FROM foo__crsql_clock").fetchall()
#     pprint(rows)