Line data Source code
1 : //
2 : // cloudsync_changes_sqlite.c
3 : // cloudsync
4 : //
5 : // Created by Marco Bambini on 23/09/24.
6 : //
7 :
8 : #include <stdio.h>
9 : #include <string.h>
10 :
11 : #include "cloudsync_changes_sqlite.h"
12 : #include "../utils.h"
13 : #include "../dbutils.h"
14 :
15 : #ifndef SQLITE_CORE
16 : SQLITE_EXTENSION_INIT3
17 : #endif
18 :
19 : typedef struct cloudsync_changes_vtab {
20 : sqlite3_vtab base; // base class, must be first
21 : sqlite3 *db;
22 : cloudsync_context *data;
23 : } cloudsync_changes_vtab;
24 :
25 : typedef struct cloudsync_changes_cursor {
26 : sqlite3_vtab_cursor base; // base class, must be first
27 : cloudsync_changes_vtab *vtab;
28 : sqlite3_stmt *vm; // prepared statement
29 : } cloudsync_changes_cursor;
30 :
31 : char *cloudsync_changes_columns[] = {"tbl", "pk", "col_name", "col_value", "col_version", "db_version", "site_id", "cl", "seq"};
32 : #define COLNAME_FROM_INDEX(i) cloudsync_changes_columns[i]
33 : #define COL_TBL_INDEX 0
34 : #define COL_PK_INDEX 1
35 : #define COL_NAME_INDEX 2
36 : #define COL_VALUE_INDEX 3
37 : #define COL_VERSION_INDEX 4
38 : #define COL_DBVERSION_INDEX 5
39 : #define COL_SITEID_INDEX 6
40 : #define COL_CL_INDEX 7
41 : #define COL_SEQ_INDEX 8
42 :
43 : #if CLOUDSYNC_UNITTEST
44 : bool force_vtab_filter_abort = false;
45 : #define CHECK_VFILTERTEST_ABORT() if (force_vtab_filter_abort) rc = SQLITE_ERROR
46 : #else
47 : #define CHECK_VFILTERTEST_ABORT()
48 : #endif
49 :
50 : // MARK: -
51 :
52 3 : int vtab_set_error (sqlite3_vtab *vtab, const char *format, ...) {
53 : va_list arg;
54 3 : va_start (arg, format);
55 3 : char *err = sqlite3_vmprintf(format, arg);
56 3 : va_end (arg);
57 :
58 3 : if (vtab->zErrMsg) sqlite3_free(vtab->zErrMsg);
59 3 : vtab->zErrMsg = err;
60 3 : return SQLITE_ERROR;
61 : }
62 :
63 684 : const char *vtab_opname_from_value (int value) {
64 684 : switch (value) {
65 591 : case SQLITE_INDEX_CONSTRAINT_EQ: return "=";
66 39 : case SQLITE_INDEX_CONSTRAINT_GT: return ">";
67 28 : case SQLITE_INDEX_CONSTRAINT_LE: return "<=";
68 1 : case SQLITE_INDEX_CONSTRAINT_LT: return "<";
69 11 : case SQLITE_INDEX_CONSTRAINT_GE: return ">=";
70 1 : case SQLITE_INDEX_CONSTRAINT_LIKE: return "LIKE";
71 1 : case SQLITE_INDEX_CONSTRAINT_GLOB: return "GLOB";
72 7 : case SQLITE_INDEX_CONSTRAINT_NE: return "!=";
73 1 : case SQLITE_INDEX_CONSTRAINT_ISNOT: return "IS NOT";
74 1 : case SQLITE_INDEX_CONSTRAINT_ISNOTNULL: return "IS NOT NULL";
75 1 : case SQLITE_INDEX_CONSTRAINT_ISNULL: return "IS NULL";
76 1 : case SQLITE_INDEX_CONSTRAINT_IS: return "IS";
77 :
78 : // The REGEXP operator is a special syntax for the regexp() user function.
79 : // No regexp() user function is defined by default and so use of the REGEXP operator will normally result in an error message.
80 : // If an application-defined SQL function named "regexp" is added at run-time, then the "X REGEXP Y" operator will be implemented as
81 : // a call to "regexp(Y,X)".
82 : // case SQLITE_INDEX_CONSTRAINT_REGEXP: return "REGEX";
83 :
84 : // MATCH is only valid for virtual FTS tables
85 : // The MATCH operator is a special syntax for the match() application-defined function.
86 : // The default match() function implementation raises an exception and is not really useful for anything.
87 : // But extensions can override the match() function with more helpful logic.
88 : // case SQLITE_INDEX_CONSTRAINT_MATCH: return "MATCH";
89 : }
90 1 : return NULL;
91 684 : }
92 :
93 148 : int vtab_colname_is_legal (const char *name) {
94 148 : int count = sizeof(cloudsync_changes_columns) / sizeof (char *);
95 :
96 1083 : for (int i=0; i<count; ++i) {
97 1082 : if (strcasecmp(cloudsync_changes_columns[i], name) == 0) return 1;
98 935 : }
99 1 : return 0;
100 148 : }
101 :
102 811 : char *vtab_build_changes_sql (cloudsync_context *data, const char *idxs) {
103 : DEBUG_VTAB("build_changes_sql");
104 :
105 : /*
106 : * This SQLite query dynamically generates and executes a consolidated query
107 : * to fetch changes from all tables related to cloud synchronization.
108 : *
109 : * It works in the following steps:
110 : *
111 : * 1. `table_names` CTE: Retrieves all table names in the database that end
112 : * with '_cloudsync' and extracts the base table name (without the '_cloudsync' suffix).
113 : *
114 : * 2. `changes_query` CTE: Constructs individual SELECT statements for each
115 : * cloud sync table, fetching data about changes in columns:
116 : * - `pk`: Primary key of the table.
117 : * - `col_name`: Name of the changed column.
118 : * - `col_version`: Version of the changed column.
119 : * - `db_version`: Database version when the change was recorded.
120 : * - `site_id`: Site identifier associated with the change.
121 : * - `cl`: Coalesced version (either `t2.col_version` or 1 if `t2.col_version` is NULL).
122 : * - `seq`: Sequence number of the change.
123 : * Each query is constructed by joining the table with `cloudsync_site_id`
124 : * for resolving the site ID and performing a LEFT JOIN with itself to
125 : * identify columns with NULL values in `t2.col_name`.
126 : *
127 : * 3. `union_query` CTE: Combines all the constructed SELECT statements
128 : * from `changes_query` into a single query using `UNION ALL`.
129 : *
130 : * 4. `final_query` CTE: Wraps the combined UNION ALL query into another
131 : * SELECT statement, preparing for additional filtering.
132 : *
133 : * 5. Final SELECT: Adds a WHERE clause to filter records with `db_version`
134 : * greater than a specified value (using a placeholder `?`), and orders
135 : * the results by `db_version` and `seq`.
136 : *
137 : * The overall result is a consolidated view of changes across all
138 : * cloud sync tables, filtered and ordered based on the `db_version` and
139 : * `seq` fields.
140 : */
141 :
142 811 : const char *query =
143 : "WITH table_names AS ( "
144 : " SELECT format('%q',SUBSTR(tbl_name, 1, LENGTH(tbl_name) - 10)) AS table_name_literal, format('%w',SUBSTR(tbl_name, 1, LENGTH(tbl_name) - 10)) AS table_name_identifier, format('%w',tbl_name) AS table_meta "
145 : " FROM sqlite_master "
146 : " WHERE type = 'table' AND tbl_name LIKE '%_cloudsync' "
147 : "), "
148 : "changes_query AS ( "
149 : " SELECT "
150 : " 'SELECT "
151 : " ''' || \"table_name_literal\" || ''' AS tbl, "
152 : " t1.pk AS pk, "
153 : " t1.col_name AS col_name, "
154 : " cloudsync_col_value(''' || \"table_name_literal\" || ''', t1.col_name, t1.pk) AS col_value, "
155 : " t1.col_version AS col_version, "
156 : " t1.db_version AS db_version, "
157 : " site_tbl.site_id AS site_id, "
158 : " t1.seq AS seq, "
159 : " COALESCE(t2.col_version, 1) AS cl "
160 : " FROM \"' || \"table_meta\" || '\" AS t1 "
161 : " LEFT JOIN cloudsync_site_id AS site_tbl ON t1.site_id = site_tbl.rowid "
162 : " LEFT JOIN \"' || \"table_meta\" || '\" AS t2 ON t1.pk = t2.pk AND t2.col_name = ''" CLOUDSYNC_TOMBSTONE_VALUE "'' "
163 : " WHERE col_value IS NOT ''" CLOUDSYNC_RLS_RESTRICTED_VALUE "''' "
164 : " AS query_string FROM table_names "
165 : "), "
166 : "union_query AS ( "
167 : " SELECT GROUP_CONCAT(query_string, ' UNION ALL ') AS union_query FROM changes_query "
168 : "), "
169 : "final_query AS ( "
170 : " SELECT "
171 : " 'SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq FROM (' || union_query || ')' "
172 : " AS final_string FROM union_query "
173 : ") "
174 : "SELECT final_string || ' ";
175 811 : const char *final_query = ";' FROM final_query;";
176 :
177 : static size_t query_len = 0;
178 : static size_t final_query_len = 0;
179 811 : if (query_len == 0) query_len = strlen(query);
180 811 : if (final_query_len == 0) final_query_len = strlen(final_query);
181 :
182 811 : size_t idx_len = strlen(idxs);
183 811 : size_t blen = query_len + idx_len + final_query_len + 128;
184 811 : char *sql = (char *)cloudsync_memory_alloc((sqlite3_uint64)blen);
185 811 : if (!sql) return NULL;
186 :
187 : // build final sql statement taking in account the dynamic idxs string provided by the user
188 811 : memcpy(sql, query, query_len);
189 811 : memcpy(sql + (query_len), idxs, idx_len);
190 811 : memcpy(sql + (query_len + idx_len), final_query, final_query_len+1);
191 :
192 811 : char *value = NULL;
193 811 : int rc = database_select_text(data, sql, &value);
194 811 : cloudsync_memory_free(sql);
195 :
196 811 : return (rc == DBRES_OK) ? value : NULL;
197 811 : }
198 :
199 : // MARK: -
200 :
201 217 : int cloudsync_changesvtab_connect (sqlite3 *db, void *aux, int argc, const char *const *argv, sqlite3_vtab **vtab, char **err) {
202 : DEBUG_VTAB("cloudsync_changesvtab_connect");
203 :
204 217 : int rc = sqlite3_declare_vtab(db, "CREATE TABLE x (tbl TEXT NOT NULL, pk BLOB NOT NULL, col_name TEXT NOT NULL,"
205 : "col_value TEXT, col_version INTEGER NOT NULL, db_version INTEGER NOT NULL,"
206 : "site_id BLOB NOT NULL, cl INTEGER NOT NULL, seq INTEGER NOT NULL);");
207 217 : if (rc == SQLITE_OK) {
208 : // memory internally managed by SQLite, so I cannot use memory_alloc here
209 217 : cloudsync_changes_vtab *vnew = sqlite3_malloc64(sizeof(cloudsync_changes_vtab));
210 217 : if (vnew == NULL) return SQLITE_NOMEM;
211 :
212 217 : memset(vnew, 0, sizeof(cloudsync_changes_vtab));
213 217 : vnew->db = db;
214 217 : vnew->data = aux;
215 :
216 217 : *vtab = (sqlite3_vtab *)vnew;
217 217 : }
218 :
219 217 : return rc;
220 217 : }
221 :
222 217 : int cloudsync_changesvtab_disconnect (sqlite3_vtab *vtab) {
223 : DEBUG_VTAB("cloudsync_changesvtab_disconnect");
224 :
225 217 : cloudsync_changes_vtab *p = (cloudsync_changes_vtab *)vtab;
226 217 : sqlite3_free(p);
227 217 : return SQLITE_OK;
228 : }
229 :
230 811 : int cloudsync_changesvtab_open (sqlite3_vtab *vtab, sqlite3_vtab_cursor **pcursor) {
231 : DEBUG_VTAB("cloudsync_changesvtab_open");
232 :
233 811 : cloudsync_changes_cursor *cursor = cloudsync_memory_alloc(sizeof(cloudsync_changes_cursor));
234 811 : if (cursor == NULL) return SQLITE_NOMEM;
235 :
236 811 : memset(cursor, 0, sizeof(cloudsync_changes_cursor));
237 811 : cursor->vtab = (cloudsync_changes_vtab *)vtab;
238 :
239 811 : *pcursor = (sqlite3_vtab_cursor *)cursor;
240 811 : return SQLITE_OK;
241 811 : }
242 :
243 811 : int cloudsync_changesvtab_close (sqlite3_vtab_cursor *cursor) {
244 : DEBUG_VTAB("cloudsync_changesvtab_close");
245 :
246 811 : cloudsync_changes_cursor *c = (cloudsync_changes_cursor *)cursor;
247 811 : if (c->vm) {
248 27 : sqlite3_finalize(c->vm);
249 27 : c->vm = NULL;
250 27 : }
251 :
252 811 : cloudsync_memory_free(cursor);
253 811 : return SQLITE_OK;
254 : }
255 :
256 846 : int cloudsync_changesvtab_best_index (sqlite3_vtab *vtab, sqlite3_index_info *idxinfo) {
257 : DEBUG_VTAB("cloudsync_changesvtab_best_index");
258 :
259 : // the goal here is to build a WHERE clause and gives an estimate
260 : // of the cost of that clause
261 :
262 : // we'll incrementally build the clause and we avoid the realloc
263 : // of memory, so perform a quick loop to estimate memory usage
264 :
265 : // count the number of contrainst and order by clauses
266 846 : int count1 = idxinfo->nConstraint;
267 846 : int count2 = idxinfo->nOrderBy;
268 :
269 : // col_version is the longest column name (11)
270 : // +1 for the space
271 : // IS NOT NULL is the longest constraint value (11)
272 : // +1 for the space
273 : // +1 for the ? character
274 : // +5 for space AND space
275 : // +512 for the extra space and for the WHERE and ORDER BY literals
276 :
277 : // memory internally manager by SQLite, so I cannot use memory_alloc here
278 846 : size_t slen = (count1 * (11 + 1 + 11 + 1 + 5)) + (count2 * 11 + 1 + 5) + 512;
279 846 : char *s = (char *)sqlite3_malloc64((sqlite3_uint64)slen);
280 846 : if (!s) return SQLITE_NOMEM;
281 846 : size_t sindex= 0;
282 :
283 846 : int idxnum = 0;
284 846 : int arg_index = 1;
285 846 : int orderconsumed = 1;
286 :
287 : // is there a WHERE clause ?
288 846 : if (count1 > 0) sindex += snprintf(s+sindex, slen-sindex, "WHERE ");
289 :
290 : // check constraints
291 1529 : for (int i=0; i < count1; ++i) {
292 : // analyze only usable constraints
293 683 : struct sqlite3_index_constraint *constraint = &idxinfo->aConstraint[i];
294 683 : if (constraint->usable == false) continue;
295 :
296 683 : int idx = constraint->iColumn;
297 683 : uint8_t op = constraint->op;
298 :
299 683 : const char *colname = (idx >= 0 && idx < CLOUDSYNC_CHANGES_NCOLS) ? COLNAME_FROM_INDEX(idx) : "rowid";
300 683 : const char *opname = vtab_opname_from_value(op);
301 683 : if (!opname) continue;
302 :
303 : // build next constraint
304 683 : if (i > 0) sindex += snprintf(s+sindex, slen-sindex, " AND ");
305 :
306 : // handle special case where value is not needed
307 683 : if ((op == SQLITE_INDEX_CONSTRAINT_ISNULL) || (op == SQLITE_INDEX_CONSTRAINT_ISNOTNULL)) {
308 2 : sindex += snprintf(s+sindex, slen-sindex, "%s %s", colname, opname);
309 2 : idxinfo->aConstraintUsage[i].argvIndex = 0;
310 2 : } else {
311 681 : sindex += snprintf(s+sindex, slen-sindex, "%s %s ?", colname, opname);
312 681 : idxinfo->aConstraintUsage[i].argvIndex = arg_index++;
313 : }
314 683 : idxinfo->aConstraintUsage[i].omit = 1;
315 :
316 : //a bitmask (idxnum) is built up based on which constraints are applied
317 683 : if (idx == COL_DBVERSION_INDEX) idxnum |= 2; // set bit 1
318 599 : else if (idx == COL_SITEID_INDEX) idxnum |= 4; // set bit 2
319 683 : }
320 :
321 : // is there an ORDER BY clause ?
322 : // if not use a default one
323 846 : if (count2 > 0) sindex += snprintf(s+sindex, slen-sindex, " ORDER BY ");
324 764 : else sindex += snprintf(s+sindex, slen-sindex, " ORDER BY db_version, seq ASC");
325 :
326 992 : for (int i=0; i < count2; ++i) {
327 146 : struct sqlite3_index_orderby *orderby = &idxinfo->aOrderBy[i];
328 :
329 : // build next constraint
330 146 : if (i > 0) sindex += snprintf(s+sindex, slen-sindex, ", ");
331 :
332 146 : int idx = orderby->iColumn;
333 146 : const char *colname = (idx >= 0 && idx < CLOUDSYNC_CHANGES_NCOLS) ? COLNAME_FROM_INDEX(idx) : "rowid";
334 146 : if (!vtab_colname_is_legal(colname)) orderconsumed = 0;
335 :
336 146 : sindex += snprintf(s+sindex, slen-sindex, "%s %s", colname, orderby->desc ? " DESC" : " ASC");
337 146 : }
338 :
339 846 : idxinfo->idxNum = idxnum;
340 846 : idxinfo->idxStr = s;
341 846 : idxinfo->needToFreeIdxStr = 1;
342 846 : idxinfo->orderByConsumed = orderconsumed;
343 :
344 : // the goal of the xBestIndex function is to help SQLite's query planner decide on the most efficient way
345 : // to execute a query on the virtual table. It does so by evaluating which constraints (filters) can be applied
346 : // and providing an estimate of the cost and number of rows that the query will return.
347 :
348 :
349 : /*
350 :
351 : By commenting the following code we assume that the developer is using an SQLite library
352 : more recent than 3.8.2 released on 2013-12-06
353 :
354 : int version = sqlite3_libversion_number();
355 : if (version >= 3008002) {
356 : // field sqlite3_int64 estimatedRows is available (estimated number of rows returned)
357 : }
358 :
359 : if (version >= 3009000) {
360 : // field int idxFlags is available (mask of SQLITE_INDEX_SCAN)
361 : }
362 :
363 : if (version >= 3010000) {
364 : // field sqlite3_uint64 colUsed is available (input: mask of columns used by statement)
365 : }
366 :
367 : */
368 :
369 : // perform estimated cost and row count based on the constraints
370 846 : if ((idxnum & 6) == 6) {
371 : // both DbVrsn and SiteId constraints are present
372 : // query is expected to be highly selective, returning only one row, with a very low execution cost
373 28 : idxinfo->estimatedCost = 1.0;
374 28 : idxinfo->estimatedRows = 1;
375 846 : } else if ((idxnum & 2) == 2) {
376 : // only DbVrsn constraint is present
377 : // query is expected to return more rows (10) and take more time (cost of 10.0) than in the previous case
378 43 : idxinfo->estimatedCost = 10.0;
379 43 : idxinfo->estimatedRows = 10;
380 818 : } else if ((idxnum & 4) == 4) {
381 : // only SiteId constraint is present
382 : // query is expected to be very inefficient, returning a large number of rows and taking a long time to execute
383 552 : idxinfo->estimatedCost = (double)INT32_MAX;
384 552 : idxinfo->estimatedRows = (sqlite3_int64)INT32_MAX;
385 552 : } else {
386 : // no constraints are present
387 : // worst-case scenario, where the query returns all rows from the virtual table
388 223 : idxinfo->estimatedCost = (double)INT64_MAX;
389 223 : idxinfo->estimatedRows = (sqlite3_int64)INT64_MAX;
390 : }
391 :
392 846 : return SQLITE_OK;
393 846 : }
394 :
395 :
396 811 : int cloudsync_changesvtab_filter (sqlite3_vtab_cursor *cursor, int idxn, const char *idxs, int argc, sqlite3_value **argv) {
397 : DEBUG_VTAB("cloudsync_changesvtab_filter");
398 :
399 811 : cloudsync_changes_cursor *c = (cloudsync_changes_cursor *)cursor;
400 811 : cloudsync_context *data = c->vtab->data;
401 811 : sqlite3 *db = c->vtab->db;
402 811 : char *sql = vtab_build_changes_sql(data, idxs);
403 811 : if (sql == NULL) {
404 : // vtab_build_changes_sql returns NULL when no *_cloudsync meta-tables
405 : // exist (cloudsync_init was never called, or the last configured table
406 : // was cleaned up): the inner GROUP_CONCAT produces a NULL row and the
407 : // outer SELECT yields a NULL final string. Distinguish this from a
408 : // genuine OOM by checking whether cloudsync is configured, so the user
409 : // gets an actionable message instead of "out of memory".
410 2 : if (!cloudsync_config_exists(data) || dbutils_table_settings_count_tables(data) == 0) {
411 1 : return vtab_set_error((sqlite3_vtab *)c->vtab,
412 : "cloudsync has no tables configured for sync. Call "
413 : "SELECT cloudsync_init('<table_name>') to enable sync on a "
414 : "table before querying cloudsync_changes.");
415 : }
416 : // settings rows exist but no *_cloudsync tables do (e.g. a dump/restore
417 : // that skipped the meta tables): still not an OOM
418 1 : sqlite3_stmt *vm = NULL;
419 1 : bool meta_missing = false;
420 1 : if (sqlite3_prepare_v2(db, "SELECT 1 FROM sqlite_master WHERE type='table' AND tbl_name LIKE '%_cloudsync' LIMIT 1;", -1, &vm, NULL) == SQLITE_OK) {
421 1 : meta_missing = (sqlite3_step(vm) != SQLITE_ROW);
422 1 : }
423 1 : if (vm) sqlite3_finalize(vm);
424 1 : if (meta_missing) {
425 1 : return vtab_set_error((sqlite3_vtab *)c->vtab,
426 : "cloudsync settings reference tables whose sync metadata is "
427 : "missing (no *_cloudsync tables found). Re-run "
428 : "SELECT cloudsync_init('<table_name>') for each configured "
429 : "table to rebuild it.");
430 : }
431 0 : return SQLITE_NOMEM;
432 : }
433 :
434 : // the xFilter method may be called multiple times on the same sqlite3_vtab_cursor*
435 809 : if (c->vm) sqlite3_finalize(c->vm);
436 809 : c->vm = NULL;
437 :
438 809 : int rc = sqlite3_prepare_v2(db, sql, -1, &c->vm, NULL);
439 809 : cloudsync_memory_free(sql);
440 809 : if (rc != SQLITE_OK) goto abort_filter;
441 :
442 1439 : for (int i=0; i<argc; ++i) {
443 630 : rc = sqlite3_bind_value(c->vm, i+1, argv[i]);
444 630 : if (rc != SQLITE_OK) goto abort_filter;
445 630 : }
446 :
447 809 : rc = sqlite3_step(c->vm);
448 809 : CHECK_VFILTERTEST_ABORT();
449 :
450 809 : if (rc == SQLITE_DONE) {
451 12 : sqlite3_finalize(c->vm);
452 12 : c->vm = NULL;
453 809 : } else if (rc != SQLITE_ROW) {
454 1 : goto abort_filter;
455 : }
456 :
457 808 : return SQLITE_OK;
458 :
459 : abort_filter:
460 : // error condition
461 : DEBUG_VTAB("cloudsync_changesvtab_filter: %s\n", sqlite3_errmsg(db));
462 1 : if (c->vm) {
463 1 : sqlite3_finalize(c->vm);
464 1 : c->vm = NULL;
465 1 : }
466 1 : return rc;
467 811 : }
468 :
469 57901 : int cloudsync_changesvtab_next (sqlite3_vtab_cursor *cursor) {
470 : DEBUG_VTAB("cloudsync_changesvtab_next");
471 :
472 57901 : cloudsync_changes_cursor *c = (cloudsync_changes_cursor *)cursor;
473 57901 : int rc = sqlite3_step(c->vm);
474 :
475 57901 : if (rc == SQLITE_DONE) {
476 769 : sqlite3_finalize(c->vm);
477 769 : c->vm = NULL;
478 769 : rc = SQLITE_OK;
479 57901 : } else if (rc == SQLITE_ROW) {
480 57132 : rc = SQLITE_OK;
481 57132 : }
482 :
483 57901 : if (rc != SQLITE_OK) DEBUG_VTAB("cloudsync_changesvtab_next: %s\n", sqlite3_errmsg(c->vtab->db));
484 57901 : return rc;
485 : }
486 :
487 58709 : int cloudsync_changesvtab_eof (sqlite3_vtab_cursor *cursor) {
488 : DEBUG_VTAB("cloudsync_changesvtab_eof");
489 :
490 : // we must return false (zero) if the specified cursor currently points to a valid row of data, or true (non-zero) otherwise
491 58709 : cloudsync_changes_cursor *c = (cloudsync_changes_cursor *)cursor;
492 58709 : return (c->vm) ? 0 : 1;
493 : }
494 :
495 493853 : int cloudsync_changesvtab_column (sqlite3_vtab_cursor *cursor, sqlite3_context *ctx, int col) {
496 : DEBUG_VTAB("cloudsync_changesvtab_column %d\n", col);
497 :
498 493853 : cloudsync_changes_cursor *c = (cloudsync_changes_cursor *)cursor;
499 493853 : sqlite3_value *value = sqlite3_column_value(c->vm, col);
500 493853 : sqlite3_result_value(ctx, value);
501 :
502 493853 : return SQLITE_OK;
503 : }
504 :
505 88 : int cloudsync_changesvtab_rowid (sqlite3_vtab_cursor *cursor, sqlite3_int64 *rowid) {
506 : DEBUG_VTAB("cloudsync_changesvtab_rowid");
507 :
508 88 : cloudsync_changes_cursor *c = (cloudsync_changes_cursor *)cursor;
509 88 : sqlite3_int64 seq = sqlite3_column_int64(c->vm, COL_SEQ_INDEX);
510 88 : sqlite3_int64 db_version = sqlite3_column_int64(c->vm, COL_DBVERSION_INDEX);
511 :
512 : // for an explanation see https://github.com/sqliteai/sqlite-sync/blob/main/docs/RowID.md
513 88 : *rowid = (db_version << 30) | seq;
514 88 : return SQLITE_OK;
515 : }
516 :
517 3150 : int cloudsync_changesvtab_insert_gos (sqlite3_vtab *vtab, cloudsync_context *data, cloudsync_table_context *table, const char *insert_pk, int insert_pk_len, const char *insert_name, sqlite3_value *insert_value, sqlite3_int64 insert_col_version, sqlite3_int64 insert_db_version, const char *insert_site_id, int insert_site_id_len, sqlite3_int64 insert_seq, int64_t *rowid) {
518 : DEBUG_VTAB("cloudsync_changesvtab_insert_gos");
519 :
520 : // Grow-Only Set (GOS) Algorithm: Only insertions are allowed, deletions and updates are prevented from a trigger.
521 3150 : int rc = merge_insert_col(data, table, insert_pk, insert_pk_len, insert_name, insert_value, (int64_t)insert_col_version, (int64_t)insert_db_version, insert_site_id, insert_site_id_len, (int64_t)insert_seq, rowid);
522 :
523 3150 : if (rc != SQLITE_OK) {
524 0 : vtab_set_error(vtab, "%s", cloudsync_errmsg(data));
525 0 : }
526 :
527 3150 : return rc;
528 : }
529 :
530 51495 : int cloudsync_changesvtab_insert (sqlite3_vtab *vtab, int argc, sqlite3_value **argv, sqlite3_int64 *rowid) {
531 : DEBUG_VTAB("cloudsync_changesvtab_insert");
532 :
533 : // this function performs the merging logic for an insert in a cloud-synchronized table. It handles
534 : // different scenarios including conflicts, causal lengths, delete operations, and resurrecting rows
535 : // based on the incoming data (from remote nodes or clients) and the local database state
536 :
537 : // this function handles different CRDT algorithms (GOS, DWS, AWS, and CLS).
538 : // the merging strategy is determined based on the table->algo value.
539 :
540 : // meta table declaration:
541 : // tbl TEXT NOT NULL, pk BLOB NOT NULL, col_name TEXT NOT NULL,"
542 : // "col_value ANY, col_version INTEGER NOT NULL, db_version INTEGER NOT NULL,"
543 : // "site_id BLOB NOT NULL, cl INTEGER NOT NULL, seq INTEGER NOT NULL
544 :
545 : // meta information to retrieve from arguments:
546 : // argv[0] -> table name (TEXT)
547 : // argv[1] -> primary key (BLOB)
548 : // argv[2] -> column name (TEXT or NULL if sentinel)
549 : // argv[3] -> column value (ANY)
550 : // argv[4] -> column version (INTEGER)
551 : // argv[5] -> database version (INTEGER)
552 : // argv[6] -> site ID (BLOB, identifies the origin of the update)
553 : // argv[7] -> causal length (INTEGER, tracks the order of operations)
554 : // argv[8] -> sequence number (INTEGER, unique per operation)
555 :
556 : // extract table name
557 51495 : const char *insert_tbl = (const char *)sqlite3_value_text(argv[0]);
558 :
559 : // lookup table
560 51495 : cloudsync_context *data = (cloudsync_context *)(((cloudsync_changes_vtab *)vtab)->data);
561 51495 : cloudsync_table_context *table = table_lookup(data, insert_tbl);
562 51495 : if (!table) return vtab_set_error(vtab, "Unable to find table %s,", insert_tbl);
563 :
564 : // extract the remaining fields from the input values
565 51495 : const char *insert_pk = (const char *)sqlite3_value_blob(argv[1]);
566 51495 : int insert_pk_len = sqlite3_value_bytes(argv[1]);
567 51495 : const char *insert_name = (sqlite3_value_type(argv[2]) == SQLITE_NULL) ? CLOUDSYNC_TOMBSTONE_VALUE : (const char *)sqlite3_value_text(argv[2]);
568 51495 : sqlite3_value *insert_value = argv[3];
569 51495 : int64_t insert_col_version = (int64_t)sqlite3_value_int(argv[4]);
570 51495 : int64_t insert_db_version = (int64_t)sqlite3_value_int(argv[5]);
571 51495 : const char *insert_site_id = (const char *)sqlite3_value_blob(argv[6]);
572 51495 : int insert_site_id_len = sqlite3_value_bytes(argv[6]);
573 51495 : int64_t insert_cl = (int64_t)sqlite3_value_int(argv[7]);
574 51495 : int64_t insert_seq = (int64_t)sqlite3_value_int(argv[8]);
575 :
576 : // perform different logic for each different table algorithm
577 51495 : if (table_algo_isgos(table)) return cloudsync_changesvtab_insert_gos(vtab, data, table, insert_pk, insert_pk_len, insert_name, insert_value, insert_col_version, insert_db_version, insert_site_id, insert_site_id_len, insert_seq, (int64_t *)rowid);
578 :
579 48345 : int rc = merge_insert (data, table, insert_pk, insert_pk_len, insert_cl, insert_name, insert_value, insert_col_version, insert_db_version, insert_site_id, insert_site_id_len, insert_seq, (int64_t *)rowid);
580 48345 : if (rc != SQLITE_OK) {
581 0 : return vtab_set_error(vtab, "%s", cloudsync_errmsg(data));
582 : }
583 :
584 48345 : return SQLITE_OK;
585 51495 : }
586 :
587 51496 : int cloudsync_changesvtab_update (sqlite3_vtab *vtab, int argc, sqlite3_value **argv, sqlite3_int64 *rowid) {
588 : DEBUG_VTAB("cloudsync_changesvtab_update");
589 :
590 : // only INSERT statements are allowed
591 51496 : bool is_insert = (argc > 1 && sqlite3_value_type(argv[0]) == SQLITE_NULL);
592 51496 : if (!is_insert) {
593 1 : vtab_set_error(vtab, "Only INSERT and SELECT statements are allowed against the cloudsync_changes table");
594 1 : return SQLITE_MISUSE;
595 : }
596 :
597 : // argv[0] is set only in case of DELETE statement (it contains the rowid of a row in the virtual table to be deleted)
598 : // argv[1] is the rowid of a new row to be inserted into the virtual table (always NULL in our case)
599 : // so reduce the number of meaningful arguments by 2
600 51495 : return cloudsync_changesvtab_insert(vtab, argc-2, &argv[2], rowid);
601 51496 : }
602 :
603 : // MARK: -
604 :
605 253 : int cloudsync_vtab_register_changes (sqlite3 *db, cloudsync_context *xdata) {
606 : static sqlite3_module cloudsync_changes_module = {
607 : /* iVersion */ 0,
608 : /* xCreate */ 0, // Eponymous only virtual table
609 : /* xConnect */ cloudsync_changesvtab_connect,
610 : /* xBestIndex */ cloudsync_changesvtab_best_index,
611 : /* xDisconnect */ cloudsync_changesvtab_disconnect,
612 : /* xDestroy */ 0,
613 : /* xOpen */ cloudsync_changesvtab_open,
614 : /* xClose */ cloudsync_changesvtab_close,
615 : /* xFilter */ cloudsync_changesvtab_filter,
616 : /* xNext */ cloudsync_changesvtab_next,
617 : /* xEof */ cloudsync_changesvtab_eof,
618 : /* xColumn */ cloudsync_changesvtab_column,
619 : /* xRowid */ cloudsync_changesvtab_rowid,
620 : /* xUpdate */ cloudsync_changesvtab_update,
621 : /* xBegin */ 0,
622 : /* xSync */ 0,
623 : /* xCommit */ 0,
624 : /* xRollback */ 0,
625 : /* xFindMethod */ 0,
626 : /* xRename */ 0,
627 : /* xSavepoint */ 0,
628 : /* xRelease */ 0,
629 : /* xRollbackTo */ 0,
630 : /* xShadowName */ 0,
631 : /* xIntegrity */ 0
632 : };
633 :
634 253 : return sqlite3_create_module(db, "cloudsync_changes", &cloudsync_changes_module, (void *)xdata);
635 : }
|
