Skip to content

Commit 1f0b7fc

Browse files
rkistnerrkistner
committed
[WIP] test restructure for improving sync_local performance.
1 parent 96a1de4 commit 1f0b7fc

9 files changed

Lines changed: 332 additions & 30 deletions

File tree

‎crates/core/src/crud_vtab.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -239,7 +239,7 @@ impl SimpleCrudTransactionMode {
239239
prepare_lazy(&mut self.set_updated_rows, || {
240240
// language=SQLite
241241
db.prepare_v3(
242-
"INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?, ?)",
242+
"INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id, bucket) VALUES(?, ?, 0)",
243243
0,
244244
)
245245
})

‎crates/core/src/fix_data.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,8 +38,8 @@ pub fn apply_v035_fix(db: *mut sqlite::sqlite3) -> Result<i64, PowerSyncError> {
3838
// language=SQLite
3939
let statement = db.prepare_v2(&format!(
4040
"
41-
INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id)
42-
SELECT ?1, id FROM {}
41+
INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id, bucket)
42+
SELECT ?1, id, 0 FROM {}
4343
WHERE NOT EXISTS (
4444
SELECT 1 FROM ps_oplog
4545
WHERE row_type = ?1 AND row_id = {}.id

‎crates/core/src/migrations.rs‎

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ use crate::fix_data::apply_v035_fix;
1313
use crate::schema::inspection::ExistingView;
1414
use crate::sync::BucketPriority;
1515

16-
pub const LATEST_VERSION: i32 = 12;
16+
pub const LATEST_VERSION: i32 = 13;
1717

1818
pub fn powersync_migrate(
1919
ctx: *mut sqlite::context,
@@ -424,5 +424,29 @@ json_object('sql', 'DELETE FROM ps_migration WHERE id >= 12')
424424
local_db.exec_safe(stmt).into_db_result(local_db)?;
425425
}
426426

427+
if current_version < 13 && target_version >= 13 {
428+
let stmt = "\
429+
ALTER TABLE ps_updated_rows RENAME TO ps_updated_rows_old;
430+
CREATE TABLE ps_updated_rows(
431+
row_type TEXT,
432+
row_id TEXT,
433+
bucket INTEGER NOT NULL,
434+
PRIMARY KEY(row_type, row_id, bucket)) STRICT, WITHOUT ROWID;
435+
INSERT INTO ps_updated_rows(row_type, row_id, bucket)
436+
SELECT row_type, row_id, 0 FROM ps_updated_rows_old;
437+
INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id, bucket)
438+
SELECT row_type, row_id, bucket FROM ps_oplog;
439+
DROP TABLE ps_updated_rows_old;
440+
INSERT INTO ps_migration(id, down_migrations) VALUES(13, json_array(
441+
json_object('sql', 'ALTER TABLE ps_updated_rows RENAME TO ps_updated_rows_13'),
442+
json_object('sql', 'CREATE TABLE ps_updated_rows(\n row_type TEXT,\n row_id TEXT,\n PRIMARY KEY(row_type, row_id)) STRICT, WITHOUT ROWID'),
443+
json_object('sql', 'INSERT INTO ps_updated_rows(row_type, row_id)\n SELECT row_type, row_id FROM ps_updated_rows_13\n GROUP BY row_type, row_id'),
444+
json_object('sql', 'DROP TABLE ps_updated_rows_13'),
445+
json_object('sql', 'DELETE FROM ps_migration WHERE id >= 13')
446+
));
447+
";
448+
local_db.exec_safe(stmt).into_db_result(local_db)?;
449+
}
450+
427451
Ok(())
428452
}

‎crates/core/src/operations.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,8 +51,8 @@ pub fn delete_bucket(db: *mut sqlite::sqlite3, name: &str) -> Result<(), ResultC
5151
// language=SQLite
5252
let updated_statement = db.prepare_v2(
5353
"\
54-
INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id)
55-
SELECT row_type, row_id
54+
INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id, bucket)
55+
SELECT row_type, row_id, 0
5656
FROM ps_oplog
5757
WHERE bucket = ?1",
5858
)?;

‎crates/core/src/sync/operations.rs‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -52,8 +52,9 @@ INSERT INTO ps_oplog(bucket, op_id, key, row_type, row_id, data, hash) VALUES (?
5252

5353
let updated_row_statement = db.prepare_v2(
5454
"\
55-
INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
55+
INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id, bucket) VALUES(?1, ?2, ?3)",
5656
)?;
57+
updated_row_statement.bind_int64(3, bucket_id)?;
5758

5859
let mut last_op: Option<i64> = None;
5960
let mut add_checksum = Checksum::zero();
@@ -150,6 +151,12 @@ INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
150151
insert_statement.bind_int(7, checksum.bitcast_i32())?;
151152
insert_statement.exec()?;
152153

154+
if let (Some(object_type), Some(object_id)) = (object_type, object_id) {
155+
updated_row_statement.bind_text(1, object_type, sqlite::Destructor::STATIC)?;
156+
updated_row_statement.bind_text(2, object_id, sqlite::Destructor::STATIC)?;
157+
updated_row_statement.exec()?;
158+
}
159+
153160
op_checksum += checksum;
154161
} else if op == OpType::MOVE {
155162
add_checksum += checksum;
@@ -158,8 +165,8 @@ INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
158165
// language=SQLite
159166
let clear_statement1 = db
160167
.prepare_v2(
161-
"INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id)
162-
SELECT row_type, row_id
168+
"INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id, bucket)
169+
SELECT row_type, row_id, ?1
163170
FROM ps_oplog
164171
WHERE bucket = ?1",
165172
)

‎crates/core/src/sync_local.rs‎

Lines changed: 20 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -299,10 +299,7 @@ impl<'a> SyncOperation<'a> {
299299
.prepare_v2(
300300
"\
301301
WITH updated_rows AS (
302-
SELECT b.row_type, b.row_id FROM ps_buckets AS buckets
303-
CROSS JOIN ps_oplog AS b ON b.bucket = buckets.id
304-
AND (b.op_id > buckets.last_applied_op)
305-
UNION ALL SELECT row_type, row_id FROM ps_updated_rows
302+
SELECT row_type, row_id FROM ps_updated_rows
306303
)
307304
308305
SELECT
@@ -325,7 +322,7 @@ SELECT
325322
.db
326323
.prepare_v2(
327324
"\
328-
-- 1. Filter oplog by the ops added but not applied yet (oplog b).
325+
-- 1. Filter updated rows by the buckets included in this partial checkpoint.
329326
-- We do not do any DISTINCT operation here, since that introduces a temp b-tree.
330327
-- We filter out duplicates using the GROUP BY below.
331328
WITH
@@ -334,10 +331,8 @@ WITH
334331
OR name IN (SELECT value FROM json_each(json_extract(?1, '$.buckets')))
335332
),
336333
updated_rows AS (
337-
SELECT b.row_type, b.row_id FROM ps_buckets AS buckets
338-
CROSS JOIN ps_oplog AS b ON b.bucket = buckets.id
339-
AND (b.op_id > buckets.last_applied_op)
340-
WHERE buckets.id IN (SELECT id FROM involved_buckets)
334+
SELECT row_type, row_id FROM ps_updated_rows
335+
WHERE bucket IN (SELECT id FROM involved_buckets)
341336
)
342337
343338
-- 2. Find *all* current ops over different buckets for those objects (oplog r).
@@ -406,7 +401,22 @@ SELECT
406401
.into_db_result(self.db)?;
407402
BucketPriority::SENTINEL
408403
}
409-
Some(partial) => partial.priority,
404+
Some(partial) => {
405+
let stmt = self
406+
.db
407+
.prepare_v2(
408+
"DELETE FROM ps_updated_rows
409+
WHERE bucket IN (
410+
SELECT id FROM ps_buckets
411+
WHERE ?1 IS NULL
412+
OR name IN (SELECT value FROM json_each(json_extract(?1, '$.buckets')))
413+
)",
414+
)
415+
.into_db_result(self.db)?;
416+
stmt.bind_text(1, partial.args, Destructor::STATIC)?;
417+
stmt.exec()?;
418+
partial.priority
419+
}
410420
}
411421
.into();
412422

‎dart/test/sync_local_performance_test.dart‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,11 @@ SELECT
8282
8383
FROM generate_bucket_rows;
8484
85+
INSERT OR IGNORE INTO ps_updated_rows (bucket, row_type, row_id)
86+
SELECT b.bucket, b.row_type, b.row_id FROM ps_buckets AS buckets
87+
CROSS JOIN ps_oplog AS b ON b.bucket = buckets.id
88+
AND (b.op_id > buckets.last_applied_op);
89+
8590
COMMIT;
8691
''');
8792
// Enable this to see stats for initial data generation
@@ -120,6 +125,71 @@ COMMIT;
120125
// The tests below are for comparing different queries, not run as part of the
121126
// standard test suite.
122127

128+
test('sync_local new new query', () {
129+
var timer = Stopwatch()..start();
130+
final q = '''
131+
-- 1. Filter by the ops added but not applied yet
132+
WITH updated_rows AS (
133+
SELECT row_type, row_id FROM ps_updated_rows
134+
)
135+
136+
-- 2. Find *all* current ops over different buckets for those objects (oplog r).
137+
SELECT
138+
b.row_type,
139+
b.row_id,
140+
(
141+
-- 3. For each unique row, select the data from the latest oplog entry.
142+
-- The max(r.op_id) clause is used to select the latest oplog entry.
143+
-- The iif is to avoid the max(r.op_id) column ending up in the results.
144+
SELECT iif(max(r.op_id), r.data, null)
145+
FROM ps_oplog r
146+
WHERE r.row_type = b.row_type
147+
AND r.row_id = b.row_id
148+
149+
) as data
150+
FROM updated_rows b
151+
-- Group for (2)
152+
GROUP BY b.row_type, b.row_id;
153+
''';
154+
db.select(q);
155+
print('${timer.elapsed.inMilliseconds}ms ${vfs.stats()}');
156+
}, skip: skip);
157+
158+
test('sync_local new old query', () {
159+
// Same as "new query", but ignoring the data in ps_updated_rows since it's unfair to that test.
160+
var timer = Stopwatch()..start();
161+
final q = '''
162+
-- 1. Filter oplog by the ops added but not applied yet (oplog b).
163+
-- We do not do any DISTINCT operation here, since that introduces a temp b-tree.
164+
-- We filter out duplicates using the GROUP BY below.
165+
WITH updated_rows AS (
166+
SELECT b.row_type, b.row_id FROM ps_buckets AS buckets
167+
CROSS JOIN ps_oplog AS b ON b.bucket = buckets.id
168+
AND (b.op_id > buckets.last_applied_op)
169+
)
170+
171+
-- 2. Find *all* current ops over different buckets for those objects (oplog r).
172+
SELECT
173+
b.row_type,
174+
b.row_id,
175+
(
176+
-- 3. For each unique row, select the data from the latest oplog entry.
177+
-- The max(r.op_id) clause is used to select the latest oplog entry.
178+
-- The iif is to avoid the max(r.op_id) column ending up in the results.
179+
SELECT iif(max(r.op_id), r.data, null)
180+
FROM ps_oplog r
181+
WHERE r.row_type = b.row_type
182+
AND r.row_id = b.row_id
183+
184+
) as data
185+
FROM updated_rows b
186+
-- Group for (2)
187+
GROUP BY b.row_type, b.row_id;
188+
''';
189+
db.select(q);
190+
print('${timer.elapsed.inMilliseconds}ms ${vfs.stats()}');
191+
}, skip: skip);
192+
123193
test('sync_local new query', () {
124194
// This is the query we're using now.
125195
// This query only uses a single TEMP B-TREE for the GROUP BY operation,

‎dart/test/utils/fix_035_fixtures.dart‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,9 +25,9 @@ const dataMigrated = '''
2525
(1, 1, 'todos', 't1', '', '{}', 100),
2626
(1, 2, 'todos', 't2', '', '{}', 20),
2727
(2, 3, 'lists', 'l1', '', '{}', 3)
28-
;INSERT INTO ps_updated_rows(row_type, row_id) VALUES
29-
('lists', 'l3'),
30-
('todos', 't3')
28+
;INSERT INTO ps_updated_rows(row_type, row_id, bucket) VALUES
29+
('lists', 'l3', 0),
30+
('todos', 't3', 0)
3131
;INSERT INTO ps_data__lists(id, data) VALUES
3232
('l1', '{}'),
3333
('l3', '{}')

0 commit comments

Comments
 (0)