1
2 #if defined(SQLITE_ENABLE_SESSION) && defined(SQLITE_ENABLE_PREUPDATE_HOOK)
3 #include "sqlite3session.h"
4 #include <assert.h>
5 #include <string.h>
6
7 #ifndef SQLITE_AMALGAMATION
8 # include "sqliteInt.h"
9 # include "vdbeInt.h"
10 #endif
11
12 typedef struct SessionTable SessionTable;
13 typedef struct SessionChange SessionChange;
14 typedef struct SessionBuffer SessionBuffer;
15 typedef struct SessionInput SessionInput;
16
17 /*
18 ** Minimum chunk size used by streaming versions of functions.
19 */
20 #ifndef SESSIONS_STRM_CHUNK_SIZE
21 # ifdef SQLITE_TEST
22 # define SESSIONS_STRM_CHUNK_SIZE 64
23 # else
24 # define SESSIONS_STRM_CHUNK_SIZE 1024
25 # endif
26 #endif
27
28 static int sessions_strm_chunk_size = SESSIONS_STRM_CHUNK_SIZE;
29
30 typedef struct SessionHook SessionHook;
31 struct SessionHook {
32 void *pCtx;
33 int (*xOld)(void*,int,sqlite3_value**);
34 int (*xNew)(void*,int,sqlite3_value**);
35 int (*xCount)(void*);
36 int (*xDepth)(void*);
37 };
38
39 /*
40 ** Session handle structure.
41 */
42 struct sqlite3_session {
43 sqlite3 *db; /* Database handle session is attached to */
44 char *zDb; /* Name of database session is attached to */
45 int bEnable; /* True if currently recording */
46 int bIndirect; /* True if all changes are indirect */
47 int bAutoAttach; /* True to auto-attach tables */
48 int rc; /* Non-zero if an error has occurred */
49 void *pFilterCtx; /* First argument to pass to xTableFilter */
50 int (*xTableFilter)(void *pCtx, const char *zTab);
51 i64 nMalloc; /* Number of bytes of data allocated */
52 sqlite3_value *pZeroBlob; /* Value containing X'' */
53 sqlite3_session *pNext; /* Next session object on same db. */
54 SessionTable *pTable; /* List of attached tables */
55 SessionHook hook; /* APIs to grab new and old data with */
56 };
57
58 /*
59 ** Instances of this structure are used to build strings or binary records.
60 */
61 struct SessionBuffer {
62 u8 *aBuf; /* Pointer to changeset buffer */
63 int nBuf; /* Size of buffer aBuf */
64 int nAlloc; /* Size of allocation containing aBuf */
65 };
66
67 /*
68 ** An object of this type is used internally as an abstraction for
69 ** input data. Input data may be supplied either as a single large buffer
70 ** (e.g. sqlite3changeset_start()) or using a stream function (e.g.
71 ** sqlite3changeset_start_strm()).
72 */
73 struct SessionInput {
74 int bNoDiscard; /* If true, do not discard in InputBuffer() */
75 int iCurrent; /* Offset in aData[] of current change */
76 int iNext; /* Offset in aData[] of next change */
77 u8 *aData; /* Pointer to buffer containing changeset */
78 int nData; /* Number of bytes in aData */
79
80 SessionBuffer buf; /* Current read buffer */
81 int (*xInput)(void*, void*, int*); /* Input stream call (or NULL) */
82 void *pIn; /* First argument to xInput */
83 int bEof; /* Set to true after xInput finished */
84 };
85
86 /*
87 ** Structure for changeset iterators.
88 */
89 struct sqlite3_changeset_iter {
90 SessionInput in; /* Input buffer or stream */
91 SessionBuffer tblhdr; /* Buffer to hold apValue/zTab/abPK/ */
92 int bPatchset; /* True if this is a patchset */
93 int bInvert; /* True to invert changeset */
94 int bSkipEmpty; /* Skip noop UPDATE changes */
95 int rc; /* Iterator error code */
96 sqlite3_stmt *pConflict; /* Points to conflicting row, if any */
97 char *zTab; /* Current table */
98 int nCol; /* Number of columns in zTab */
99 int op; /* Current operation */
100 int bIndirect; /* True if current change was indirect */
101 u8 *abPK; /* Primary key array */
102 sqlite3_value **apValue; /* old.* and new.* values */
103 };
104
105 /*
106 ** Each session object maintains a set of the following structures, one
107 ** for each table the session object is monitoring. The structures are
108 ** stored in a linked list starting at sqlite3_session.pTable.
109 **
110 ** The keys of the SessionTable.aChange[] hash table are all rows that have
111 ** been modified in any way since the session object was attached to the
112 ** table.
113 **
114 ** The data associated with each hash-table entry is a structure containing
115 ** a subset of the initial values that the modified row contained at the
116 ** start of the session. Or no initial values if the row was inserted.
117 */
118 struct SessionTable {
119 SessionTable *pNext;
120 char *zName; /* Local name of table */
121 int nCol; /* Number of columns in table zName */
122 int bStat1; /* True if this is sqlite_stat1 */
123 const char **azCol; /* Column names */
124 u8 *abPK; /* Array of primary key flags */
125 int nEntry; /* Total number of entries in hash table */
126 int nChange; /* Size of apChange[] array */
127 SessionChange **apChange; /* Hash table buckets */
128 };
129
130 /*
131 ** RECORD FORMAT:
132 **
133 ** The following record format is similar to (but not compatible with) that
134 ** used in SQLite database files. This format is used as part of the
135 ** change-set binary format, and so must be architecture independent.
136 **
137 ** Unlike the SQLite database record format, each field is self-contained -
138 ** there is no separation of header and data. Each field begins with a
139 ** single byte describing its type, as follows:
140 **
141 ** 0x00: Undefined value.
142 ** 0x01: Integer value.
143 ** 0x02: Real value.
144 ** 0x03: Text value.
145 ** 0x04: Blob value.
146 ** 0x05: SQL NULL value.
147 **
148 ** Note that the above match the definitions of SQLITE_INTEGER, SQLITE_TEXT
149 ** and so on in sqlite3.h. For undefined and NULL values, the field consists
150 ** only of the single type byte. For other types of values, the type byte
151 ** is followed by:
152 **
153 ** Text values:
154 ** A varint containing the number of bytes in the value (encoded using
155 ** UTF-8). Followed by a buffer containing the UTF-8 representation
156 ** of the text value. There is no nul terminator.
157 **
158 ** Blob values:
159 ** A varint containing the number of bytes in the value, followed by
160 ** a buffer containing the value itself.
161 **
162 ** Integer values:
163 ** An 8-byte big-endian integer value.
164 **
165 ** Real values:
166 ** An 8-byte big-endian IEEE 754-2008 real value.
167 **
168 ** Varint values are encoded in the same way as varints in the SQLite
169 ** record format.
170 **
171 ** CHANGESET FORMAT:
172 **
173 ** A changeset is a collection of DELETE, UPDATE and INSERT operations on
174 ** one or more tables. Operations on a single table are grouped together,
175 ** but may occur in any order (i.e. deletes, updates and inserts are all
176 ** mixed together).
177 **
178 ** Each group of changes begins with a table header:
179 **
180 ** 1 byte: Constant 0x54 (capital 'T')
181 ** Varint: Number of columns in the table.
182 ** nCol bytes: 0x01 for PK columns, 0x00 otherwise.
183 ** N bytes: Unqualified table name (encoded using UTF-8). Nul-terminated.
184 **
185 ** Followed by one or more changes to the table.
186 **
187 ** 1 byte: Either SQLITE_INSERT (0x12), UPDATE (0x17) or DELETE (0x09).
188 ** 1 byte: The "indirect-change" flag.
189 ** old.* record: (delete and update only)
190 ** new.* record: (insert and update only)
191 **
192 ** The "old.*" and "new.*" records, if present, are N field records in the
193 ** format described above under "RECORD FORMAT", where N is the number of
194 ** columns in the table. The i'th field of each record is associated with
195 ** the i'th column of the table, counting from left to right in the order
196 ** in which columns were declared in the CREATE TABLE statement.
197 **
198 ** The new.* record that is part of each INSERT change contains the values
199 ** that make up the new row. Similarly, the old.* record that is part of each
200 ** DELETE change contains the values that made up the row that was deleted
201 ** from the database. In the changeset format, the records that are part
202 ** of INSERT or DELETE changes never contain any undefined (type byte 0x00)
203 ** fields.
204 **
205 ** Within the old.* record associated with an UPDATE change, all fields
206 ** associated with table columns that are not PRIMARY KEY columns and are
207 ** not modified by the UPDATE change are set to "undefined". Other fields
208 ** are set to the values that made up the row before the UPDATE that the
209 ** change records took place. Within the new.* record, fields associated
210 ** with table columns modified by the UPDATE change contain the new
211 ** values. Fields associated with table columns that are not modified
212 ** are set to "undefined".
213 **
214 ** PATCHSET FORMAT:
215 **
216 ** A patchset is also a collection of changes. It is similar to a changeset,
217 ** but leaves undefined those fields that are not useful if no conflict
218 ** resolution is required when applying the changeset.
219 **
220 ** Each group of changes begins with a table header:
221 **
222 ** 1 byte: Constant 0x50 (capital 'P')
223 ** Varint: Number of columns in the table.
224 ** nCol bytes: 0x01 for PK columns, 0x00 otherwise.
225 ** N bytes: Unqualified table name (encoded using UTF-8). Nul-terminated.
226 **
227 ** Followed by one or more changes to the table.
228 **
229 ** 1 byte: Either SQLITE_INSERT (0x12), UPDATE (0x17) or DELETE (0x09).
230 ** 1 byte: The "indirect-change" flag.
231 ** single record: (PK fields for DELETE, PK and modified fields for UPDATE,
232 ** full record for INSERT).
233 **
234 ** As in the changeset format, each field of the single record that is part
235 ** of a patchset change is associated with the correspondingly positioned
236 ** table column, counting from left to right within the CREATE TABLE
237 ** statement.
238 **
239 ** For a DELETE change, all fields within the record except those associated
240 ** with PRIMARY KEY columns are omitted. The PRIMARY KEY fields contain the
241 ** values identifying the row to delete.
242 **
243 ** For an UPDATE change, all fields except those associated with PRIMARY KEY
244 ** columns and columns that are modified by the UPDATE are set to "undefined".
245 ** PRIMARY KEY fields contain the values identifying the table row to update,
246 ** and fields associated with modified columns contain the new column values.
247 **
248 ** The records associated with INSERT changes are in the same format as for
249 ** changesets. It is not possible for a record associated with an INSERT
250 ** change to contain a field set to "undefined".
251 **
252 ** REBASE BLOB FORMAT:
253 **
254 ** A rebase blob may be output by sqlite3changeset_apply_v2() and its
255 ** streaming equivalent for use with the sqlite3_rebaser APIs to rebase
256 ** existing changesets. A rebase blob contains one entry for each conflict
257 ** resolved using either the OMIT or REPLACE strategies within the apply_v2()
258 ** call.
259 **
260 ** The format used for a rebase blob is very similar to that used for
261 ** changesets. All entries related to a single table are grouped together.
262 **
263 ** Each group of entries begins with a table header in changeset format:
264 **
265 ** 1 byte: Constant 0x54 (capital 'T')
266 ** Varint: Number of columns in the table.
267 ** nCol bytes: 0x01 for PK columns, 0x00 otherwise.
268 ** N bytes: Unqualified table name (encoded using UTF-8). Nul-terminated.
269 **
270 ** Followed by one or more entries associated with the table.
271 **
272 ** 1 byte: Either SQLITE_INSERT (0x12), DELETE (0x09).
273 ** 1 byte: Flag. 0x01 for REPLACE, 0x00 for OMIT.
274 ** record: (in the record format defined above).
275 **
276 ** In a rebase blob, the first field is set to SQLITE_INSERT if the change
277 ** that caused the conflict was an INSERT or UPDATE, or to SQLITE_DELETE if
278 ** it was a DELETE. The second field is set to 0x01 if the conflict
279 ** resolution strategy was REPLACE, or 0x00 if it was OMIT.
280 **
281 ** If the change that caused the conflict was a DELETE, then the single
282 ** record is a copy of the old.* record from the original changeset. If it
283 ** was an INSERT, then the single record is a copy of the new.* record. If
284 ** the conflicting change was an UPDATE, then the single record is a copy
285 ** of the new.* record with the PK fields filled in based on the original
286 ** old.* record.
287 */
288
289 /*
290 ** For each row modified during a session, there exists a single instance of
291 ** this structure stored in a SessionTable.aChange[] hash table.
292 */
293 struct SessionChange {
294 int op; /* One of UPDATE, DELETE, INSERT */
295 int bIndirect; /* True if this change is "indirect" */
296 int nRecord; /* Number of bytes in buffer aRecord[] */
297 u8 *aRecord; /* Buffer containing old.* record */
298 SessionChange *pNext; /* For hash-table collisions */
299 };
300
301 /*
302 ** Write a varint with value iVal into the buffer at aBuf. Return the
303 ** number of bytes written.
304 */
sessionVarintPut(u8 * aBuf,int iVal)305 static int sessionVarintPut(u8 *aBuf, int iVal){
306 return putVarint32(aBuf, iVal);
307 }
308
309 /*
310 ** Return the number of bytes required to store value iVal as a varint.
311 */
sessionVarintLen(int iVal)312 static int sessionVarintLen(int iVal){
313 return sqlite3VarintLen(iVal);
314 }
315
316 /*
317 ** Read a varint value from aBuf[] into *piVal. Return the number of
318 ** bytes read.
319 */
sessionVarintGet(u8 * aBuf,int * piVal)320 static int sessionVarintGet(u8 *aBuf, int *piVal){
321 return getVarint32(aBuf, *piVal);
322 }
323
324 /* Load an unaligned and unsigned 32-bit integer */
325 #define SESSION_UINT32(x) (((u32)(x)[0]<<24)|((x)[1]<<16)|((x)[2]<<8)|(x)[3])
326
327 /*
328 ** Read a 64-bit big-endian integer value from buffer aRec[]. Return
329 ** the value read.
330 */
sessionGetI64(u8 * aRec)331 static sqlite3_int64 sessionGetI64(u8 *aRec){
332 u64 x = SESSION_UINT32(aRec);
333 u32 y = SESSION_UINT32(aRec+4);
334 x = (x<<32) + y;
335 return (sqlite3_int64)x;
336 }
337
338 /*
339 ** Write a 64-bit big-endian integer value to the buffer aBuf[].
340 */
sessionPutI64(u8 * aBuf,sqlite3_int64 i)341 static void sessionPutI64(u8 *aBuf, sqlite3_int64 i){
342 aBuf[0] = (i>>56) & 0xFF;
343 aBuf[1] = (i>>48) & 0xFF;
344 aBuf[2] = (i>>40) & 0xFF;
345 aBuf[3] = (i>>32) & 0xFF;
346 aBuf[4] = (i>>24) & 0xFF;
347 aBuf[5] = (i>>16) & 0xFF;
348 aBuf[6] = (i>> 8) & 0xFF;
349 aBuf[7] = (i>> 0) & 0xFF;
350 }
351
352 /*
353 ** This function is used to serialize the contents of value pValue (see
354 ** comment titled "RECORD FORMAT" above).
355 **
356 ** If it is non-NULL, the serialized form of the value is written to
357 ** buffer aBuf. *pnWrite is set to the number of bytes written before
358 ** returning. Or, if aBuf is NULL, the only thing this function does is
359 ** set *pnWrite.
360 **
361 ** If no error occurs, SQLITE_OK is returned. Or, if an OOM error occurs
362 ** within a call to sqlite3_value_text() (may fail if the db is utf-16))
363 ** SQLITE_NOMEM is returned.
364 */
sessionSerializeValue(u8 * aBuf,sqlite3_value * pValue,sqlite3_int64 * pnWrite)365 static int sessionSerializeValue(
366 u8 *aBuf, /* If non-NULL, write serialized value here */
367 sqlite3_value *pValue, /* Value to serialize */
368 sqlite3_int64 *pnWrite /* IN/OUT: Increment by bytes written */
369 ){
370 int nByte; /* Size of serialized value in bytes */
371
372 if( pValue ){
373 int eType; /* Value type (SQLITE_NULL, TEXT etc.) */
374
375 eType = sqlite3_value_type(pValue);
376 if( aBuf ) aBuf[0] = eType;
377
378 switch( eType ){
379 case SQLITE_NULL:
380 nByte = 1;
381 break;
382
383 case SQLITE_INTEGER:
384 case SQLITE_FLOAT:
385 if( aBuf ){
386 /* TODO: SQLite does something special to deal with mixed-endian
387 ** floating point values (e.g. ARM7). This code probably should
388 ** too. */
389 u64 i;
390 if( eType==SQLITE_INTEGER ){
391 i = (u64)sqlite3_value_int64(pValue);
392 }else{
393 double r;
394 assert( sizeof(double)==8 && sizeof(u64)==8 );
395 r = sqlite3_value_double(pValue);
396 memcpy(&i, &r, 8);
397 }
398 sessionPutI64(&aBuf[1], i);
399 }
400 nByte = 9;
401 break;
402
403 default: {
404 u8 *z;
405 int n;
406 int nVarint;
407
408 assert( eType==SQLITE_TEXT || eType==SQLITE_BLOB );
409 if( eType==SQLITE_TEXT ){
410 z = (u8 *)sqlite3_value_text(pValue);
411 }else{
412 z = (u8 *)sqlite3_value_blob(pValue);
413 }
414 n = sqlite3_value_bytes(pValue);
415 if( z==0 && (eType!=SQLITE_BLOB || n>0) ) return SQLITE_NOMEM;
416 nVarint = sessionVarintLen(n);
417
418 if( aBuf ){
419 sessionVarintPut(&aBuf[1], n);
420 if( n ) memcpy(&aBuf[nVarint + 1], z, n);
421 }
422
423 nByte = 1 + nVarint + n;
424 break;
425 }
426 }
427 }else{
428 nByte = 1;
429 if( aBuf ) aBuf[0] = '\0';
430 }
431
432 if( pnWrite ) *pnWrite += nByte;
433 return SQLITE_OK;
434 }
435
436 /*
437 ** Allocate and return a pointer to a buffer nByte bytes in size. If
438 ** pSession is not NULL, increase the sqlite3_session.nMalloc variable
439 ** by the number of bytes allocated.
440 */
sessionMalloc64(sqlite3_session * pSession,i64 nByte)441 static void *sessionMalloc64(sqlite3_session *pSession, i64 nByte){
442 void *pRet = sqlite3_malloc64(nByte);
443 if( pSession ) pSession->nMalloc += sqlite3_msize(pRet);
444 return pRet;
445 }
446
447 /*
448 ** Free buffer pFree, which must have been allocated by an earlier
449 ** call to sessionMalloc64(). If pSession is not NULL, decrease the
450 ** sqlite3_session.nMalloc counter by the number of bytes freed.
451 */
sessionFree(sqlite3_session * pSession,void * pFree)452 static void sessionFree(sqlite3_session *pSession, void *pFree){
453 if( pSession ) pSession->nMalloc -= sqlite3_msize(pFree);
454 sqlite3_free(pFree);
455 }
456
457 /*
458 ** This macro is used to calculate hash key values for data structures. In
459 ** order to use this macro, the entire data structure must be represented
460 ** as a series of unsigned integers. In order to calculate a hash-key value
461 ** for a data structure represented as three such integers, the macro may
462 ** then be used as follows:
463 **
464 ** int hash_key_value;
465 ** hash_key_value = HASH_APPEND(0, <value 1>);
466 ** hash_key_value = HASH_APPEND(hash_key_value, <value 2>);
467 ** hash_key_value = HASH_APPEND(hash_key_value, <value 3>);
468 **
469 ** In practice, the data structures this macro is used for are the primary
470 ** key values of modified rows.
471 */
472 #define HASH_APPEND(hash, add) ((hash) << 3) ^ (hash) ^ (unsigned int)(add)
473
474 /*
475 ** Append the hash of the 64-bit integer passed as the second argument to the
476 ** hash-key value passed as the first. Return the new hash-key value.
477 */
sessionHashAppendI64(unsigned int h,i64 i)478 static unsigned int sessionHashAppendI64(unsigned int h, i64 i){
479 h = HASH_APPEND(h, i & 0xFFFFFFFF);
480 return HASH_APPEND(h, (i>>32)&0xFFFFFFFF);
481 }
482
483 /*
484 ** Append the hash of the blob passed via the second and third arguments to
485 ** the hash-key value passed as the first. Return the new hash-key value.
486 */
sessionHashAppendBlob(unsigned int h,int n,const u8 * z)487 static unsigned int sessionHashAppendBlob(unsigned int h, int n, const u8 *z){
488 int i;
489 for(i=0; i<n; i++) h = HASH_APPEND(h, z[i]);
490 return h;
491 }
492
493 /*
494 ** Append the hash of the data type passed as the second argument to the
495 ** hash-key value passed as the first. Return the new hash-key value.
496 */
sessionHashAppendType(unsigned int h,int eType)497 static unsigned int sessionHashAppendType(unsigned int h, int eType){
498 return HASH_APPEND(h, eType);
499 }
500
501 /*
502 ** This function may only be called from within a pre-update callback.
503 ** It calculates a hash based on the primary key values of the old.* or
504 ** new.* row currently available and, assuming no error occurs, writes it to
505 ** *piHash before returning. If the primary key contains one or more NULL
506 ** values, *pbNullPK is set to true before returning.
507 **
508 ** If an error occurs, an SQLite error code is returned and the final values
509 ** of *piHash asn *pbNullPK are undefined. Otherwise, SQLITE_OK is returned
510 ** and the output variables are set as described above.
511 */
sessionPreupdateHash(sqlite3_session * pSession,SessionTable * pTab,int bNew,int * piHash,int * pbNullPK)512 static int sessionPreupdateHash(
513 sqlite3_session *pSession, /* Session object that owns pTab */
514 SessionTable *pTab, /* Session table handle */
515 int bNew, /* True to hash the new.* PK */
516 int *piHash, /* OUT: Hash value */
517 int *pbNullPK /* OUT: True if there are NULL values in PK */
518 ){
519 unsigned int h = 0; /* Hash value to return */
520 int i; /* Used to iterate through columns */
521
522 assert( *pbNullPK==0 );
523 assert( pTab->nCol==pSession->hook.xCount(pSession->hook.pCtx) );
524 for(i=0; i<pTab->nCol; i++){
525 if( pTab->abPK[i] ){
526 int rc;
527 int eType;
528 sqlite3_value *pVal;
529
530 if( bNew ){
531 rc = pSession->hook.xNew(pSession->hook.pCtx, i, &pVal);
532 }else{
533 rc = pSession->hook.xOld(pSession->hook.pCtx, i, &pVal);
534 }
535 if( rc!=SQLITE_OK ) return rc;
536
537 eType = sqlite3_value_type(pVal);
538 h = sessionHashAppendType(h, eType);
539 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
540 i64 iVal;
541 if( eType==SQLITE_INTEGER ){
542 iVal = sqlite3_value_int64(pVal);
543 }else{
544 double rVal = sqlite3_value_double(pVal);
545 assert( sizeof(iVal)==8 && sizeof(rVal)==8 );
546 memcpy(&iVal, &rVal, 8);
547 }
548 h = sessionHashAppendI64(h, iVal);
549 }else if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
550 const u8 *z;
551 int n;
552 if( eType==SQLITE_TEXT ){
553 z = (const u8 *)sqlite3_value_text(pVal);
554 }else{
555 z = (const u8 *)sqlite3_value_blob(pVal);
556 }
557 n = sqlite3_value_bytes(pVal);
558 if( !z && (eType!=SQLITE_BLOB || n>0) ) return SQLITE_NOMEM;
559 h = sessionHashAppendBlob(h, n, z);
560 }else{
561 assert( eType==SQLITE_NULL );
562 assert( pTab->bStat1==0 || i!=1 );
563 *pbNullPK = 1;
564 }
565 }
566 }
567
568 *piHash = (h % pTab->nChange);
569 return SQLITE_OK;
570 }
571
572 /*
573 ** The buffer that the argument points to contains a serialized SQL value.
574 ** Return the number of bytes of space occupied by the value (including
575 ** the type byte).
576 */
sessionSerialLen(u8 * a)577 static int sessionSerialLen(u8 *a){
578 int e = *a;
579 int n;
580 if( e==0 || e==0xFF ) return 1;
581 if( e==SQLITE_NULL ) return 1;
582 if( e==SQLITE_INTEGER || e==SQLITE_FLOAT ) return 9;
583 return sessionVarintGet(&a[1], &n) + 1 + n;
584 }
585
586 /*
587 ** Based on the primary key values stored in change aRecord, calculate a
588 ** hash key. Assume the has table has nBucket buckets. The hash keys
589 ** calculated by this function are compatible with those calculated by
590 ** sessionPreupdateHash().
591 **
592 ** The bPkOnly argument is non-zero if the record at aRecord[] is from
593 ** a patchset DELETE. In this case the non-PK fields are omitted entirely.
594 */
sessionChangeHash(SessionTable * pTab,int bPkOnly,u8 * aRecord,int nBucket)595 static unsigned int sessionChangeHash(
596 SessionTable *pTab, /* Table handle */
597 int bPkOnly, /* Record consists of PK fields only */
598 u8 *aRecord, /* Change record */
599 int nBucket /* Assume this many buckets in hash table */
600 ){
601 unsigned int h = 0; /* Value to return */
602 int i; /* Used to iterate through columns */
603 u8 *a = aRecord; /* Used to iterate through change record */
604
605 for(i=0; i<pTab->nCol; i++){
606 int eType = *a;
607 int isPK = pTab->abPK[i];
608 if( bPkOnly && isPK==0 ) continue;
609
610 /* It is not possible for eType to be SQLITE_NULL here. The session
611 ** module does not record changes for rows with NULL values stored in
612 ** primary key columns. */
613 assert( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT
614 || eType==SQLITE_TEXT || eType==SQLITE_BLOB
615 || eType==SQLITE_NULL || eType==0
616 );
617 assert( !isPK || (eType!=0 && eType!=SQLITE_NULL) );
618
619 if( isPK ){
620 a++;
621 h = sessionHashAppendType(h, eType);
622 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
623 h = sessionHashAppendI64(h, sessionGetI64(a));
624 a += 8;
625 }else{
626 int n;
627 a += sessionVarintGet(a, &n);
628 h = sessionHashAppendBlob(h, n, a);
629 a += n;
630 }
631 }else{
632 a += sessionSerialLen(a);
633 }
634 }
635 return (h % nBucket);
636 }
637
638 /*
639 ** Arguments aLeft and aRight are pointers to change records for table pTab.
640 ** This function returns true if the two records apply to the same row (i.e.
641 ** have the same values stored in the primary key columns), or false
642 ** otherwise.
643 */
sessionChangeEqual(SessionTable * pTab,int bLeftPkOnly,u8 * aLeft,int bRightPkOnly,u8 * aRight)644 static int sessionChangeEqual(
645 SessionTable *pTab, /* Table used for PK definition */
646 int bLeftPkOnly, /* True if aLeft[] contains PK fields only */
647 u8 *aLeft, /* Change record */
648 int bRightPkOnly, /* True if aRight[] contains PK fields only */
649 u8 *aRight /* Change record */
650 ){
651 u8 *a1 = aLeft; /* Cursor to iterate through aLeft */
652 u8 *a2 = aRight; /* Cursor to iterate through aRight */
653 int iCol; /* Used to iterate through table columns */
654
655 for(iCol=0; iCol<pTab->nCol; iCol++){
656 if( pTab->abPK[iCol] ){
657 int n1 = sessionSerialLen(a1);
658 int n2 = sessionSerialLen(a2);
659
660 if( n1!=n2 || memcmp(a1, a2, n1) ){
661 return 0;
662 }
663 a1 += n1;
664 a2 += n2;
665 }else{
666 if( bLeftPkOnly==0 ) a1 += sessionSerialLen(a1);
667 if( bRightPkOnly==0 ) a2 += sessionSerialLen(a2);
668 }
669 }
670
671 return 1;
672 }
673
674 /*
675 ** Arguments aLeft and aRight both point to buffers containing change
676 ** records with nCol columns. This function "merges" the two records into
677 ** a single records which is written to the buffer at *paOut. *paOut is
678 ** then set to point to one byte after the last byte written before
679 ** returning.
680 **
681 ** The merging of records is done as follows: For each column, if the
682 ** aRight record contains a value for the column, copy the value from
683 ** their. Otherwise, if aLeft contains a value, copy it. If neither
684 ** record contains a value for a given column, then neither does the
685 ** output record.
686 */
sessionMergeRecord(u8 ** paOut,int nCol,u8 * aLeft,u8 * aRight)687 static void sessionMergeRecord(
688 u8 **paOut,
689 int nCol,
690 u8 *aLeft,
691 u8 *aRight
692 ){
693 u8 *a1 = aLeft; /* Cursor used to iterate through aLeft */
694 u8 *a2 = aRight; /* Cursor used to iterate through aRight */
695 u8 *aOut = *paOut; /* Output cursor */
696 int iCol; /* Used to iterate from 0 to nCol */
697
698 for(iCol=0; iCol<nCol; iCol++){
699 int n1 = sessionSerialLen(a1);
700 int n2 = sessionSerialLen(a2);
701 if( *a2 ){
702 memcpy(aOut, a2, n2);
703 aOut += n2;
704 }else{
705 memcpy(aOut, a1, n1);
706 aOut += n1;
707 }
708 a1 += n1;
709 a2 += n2;
710 }
711
712 *paOut = aOut;
713 }
714
715 /*
716 ** This is a helper function used by sessionMergeUpdate().
717 **
718 ** When this function is called, both *paOne and *paTwo point to a value
719 ** within a change record. Before it returns, both have been advanced so
720 ** as to point to the next value in the record.
721 **
722 ** If, when this function is called, *paTwo points to a valid value (i.e.
723 ** *paTwo[0] is not 0x00 - the "no value" placeholder), a copy of the *paTwo
724 ** pointer is returned and *pnVal is set to the number of bytes in the
725 ** serialized value. Otherwise, a copy of *paOne is returned and *pnVal
726 ** set to the number of bytes in the value at *paOne. If *paOne points
727 ** to the "no value" placeholder, *pnVal is set to 1. In other words:
728 **
729 ** if( *paTwo is valid ) return *paTwo;
730 ** return *paOne;
731 **
732 */
sessionMergeValue(u8 ** paOne,u8 ** paTwo,int * pnVal)733 static u8 *sessionMergeValue(
734 u8 **paOne, /* IN/OUT: Left-hand buffer pointer */
735 u8 **paTwo, /* IN/OUT: Right-hand buffer pointer */
736 int *pnVal /* OUT: Bytes in returned value */
737 ){
738 u8 *a1 = *paOne;
739 u8 *a2 = *paTwo;
740 u8 *pRet = 0;
741 int n1;
742
743 assert( a1 );
744 if( a2 ){
745 int n2 = sessionSerialLen(a2);
746 if( *a2 ){
747 *pnVal = n2;
748 pRet = a2;
749 }
750 *paTwo = &a2[n2];
751 }
752
753 n1 = sessionSerialLen(a1);
754 if( pRet==0 ){
755 *pnVal = n1;
756 pRet = a1;
757 }
758 *paOne = &a1[n1];
759
760 return pRet;
761 }
762
763 /*
764 ** This function is used by changeset_concat() to merge two UPDATE changes
765 ** on the same row.
766 */
sessionMergeUpdate(u8 ** paOut,SessionTable * pTab,int bPatchset,u8 * aOldRecord1,u8 * aOldRecord2,u8 * aNewRecord1,u8 * aNewRecord2)767 static int sessionMergeUpdate(
768 u8 **paOut, /* IN/OUT: Pointer to output buffer */
769 SessionTable *pTab, /* Table change pertains to */
770 int bPatchset, /* True if records are patchset records */
771 u8 *aOldRecord1, /* old.* record for first change */
772 u8 *aOldRecord2, /* old.* record for second change */
773 u8 *aNewRecord1, /* new.* record for first change */
774 u8 *aNewRecord2 /* new.* record for second change */
775 ){
776 u8 *aOld1 = aOldRecord1;
777 u8 *aOld2 = aOldRecord2;
778 u8 *aNew1 = aNewRecord1;
779 u8 *aNew2 = aNewRecord2;
780
781 u8 *aOut = *paOut;
782 int i;
783
784 if( bPatchset==0 ){
785 int bRequired = 0;
786
787 assert( aOldRecord1 && aNewRecord1 );
788
789 /* Write the old.* vector first. */
790 for(i=0; i<pTab->nCol; i++){
791 int nOld;
792 u8 *aOld;
793 int nNew;
794 u8 *aNew;
795
796 aOld = sessionMergeValue(&aOld1, &aOld2, &nOld);
797 aNew = sessionMergeValue(&aNew1, &aNew2, &nNew);
798 if( pTab->abPK[i] || nOld!=nNew || memcmp(aOld, aNew, nNew) ){
799 if( pTab->abPK[i]==0 ) bRequired = 1;
800 memcpy(aOut, aOld, nOld);
801 aOut += nOld;
802 }else{
803 *(aOut++) = '\0';
804 }
805 }
806
807 if( !bRequired ) return 0;
808 }
809
810 /* Write the new.* vector */
811 aOld1 = aOldRecord1;
812 aOld2 = aOldRecord2;
813 aNew1 = aNewRecord1;
814 aNew2 = aNewRecord2;
815 for(i=0; i<pTab->nCol; i++){
816 int nOld;
817 u8 *aOld;
818 int nNew;
819 u8 *aNew;
820
821 aOld = sessionMergeValue(&aOld1, &aOld2, &nOld);
822 aNew = sessionMergeValue(&aNew1, &aNew2, &nNew);
823 if( bPatchset==0
824 && (pTab->abPK[i] || (nOld==nNew && 0==memcmp(aOld, aNew, nNew)))
825 ){
826 *(aOut++) = '\0';
827 }else{
828 memcpy(aOut, aNew, nNew);
829 aOut += nNew;
830 }
831 }
832
833 *paOut = aOut;
834 return 1;
835 }
836
837 /*
838 ** This function is only called from within a pre-update-hook callback.
839 ** It determines if the current pre-update-hook change affects the same row
840 ** as the change stored in argument pChange. If so, it returns true. Otherwise
841 ** if the pre-update-hook does not affect the same row as pChange, it returns
842 ** false.
843 */
sessionPreupdateEqual(sqlite3_session * pSession,SessionTable * pTab,SessionChange * pChange,int op)844 static int sessionPreupdateEqual(
845 sqlite3_session *pSession, /* Session object that owns SessionTable */
846 SessionTable *pTab, /* Table associated with change */
847 SessionChange *pChange, /* Change to compare to */
848 int op /* Current pre-update operation */
849 ){
850 int iCol; /* Used to iterate through columns */
851 u8 *a = pChange->aRecord; /* Cursor used to scan change record */
852
853 assert( op==SQLITE_INSERT || op==SQLITE_UPDATE || op==SQLITE_DELETE );
854 for(iCol=0; iCol<pTab->nCol; iCol++){
855 if( !pTab->abPK[iCol] ){
856 a += sessionSerialLen(a);
857 }else{
858 sqlite3_value *pVal; /* Value returned by preupdate_new/old */
859 int rc; /* Error code from preupdate_new/old */
860 int eType = *a++; /* Type of value from change record */
861
862 /* The following calls to preupdate_new() and preupdate_old() can not
863 ** fail. This is because they cache their return values, and by the
864 ** time control flows to here they have already been called once from
865 ** within sessionPreupdateHash(). The first two asserts below verify
866 ** this (that the method has already been called). */
867 if( op==SQLITE_INSERT ){
868 /* assert( db->pPreUpdate->pNewUnpacked || db->pPreUpdate->aNew ); */
869 rc = pSession->hook.xNew(pSession->hook.pCtx, iCol, &pVal);
870 }else{
871 /* assert( db->pPreUpdate->pUnpacked ); */
872 rc = pSession->hook.xOld(pSession->hook.pCtx, iCol, &pVal);
873 }
874 assert( rc==SQLITE_OK );
875 if( sqlite3_value_type(pVal)!=eType ) return 0;
876
877 /* A SessionChange object never has a NULL value in a PK column */
878 assert( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT
879 || eType==SQLITE_BLOB || eType==SQLITE_TEXT
880 );
881
882 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
883 i64 iVal = sessionGetI64(a);
884 a += 8;
885 if( eType==SQLITE_INTEGER ){
886 if( sqlite3_value_int64(pVal)!=iVal ) return 0;
887 }else{
888 double rVal;
889 assert( sizeof(iVal)==8 && sizeof(rVal)==8 );
890 memcpy(&rVal, &iVal, 8);
891 if( sqlite3_value_double(pVal)!=rVal ) return 0;
892 }
893 }else{
894 int n;
895 const u8 *z;
896 a += sessionVarintGet(a, &n);
897 if( sqlite3_value_bytes(pVal)!=n ) return 0;
898 if( eType==SQLITE_TEXT ){
899 z = sqlite3_value_text(pVal);
900 }else{
901 z = sqlite3_value_blob(pVal);
902 }
903 if( n>0 && memcmp(a, z, n) ) return 0;
904 a += n;
905 }
906 }
907 }
908
909 return 1;
910 }
911
912 /*
913 ** If required, grow the hash table used to store changes on table pTab
914 ** (part of the session pSession). If a fatal OOM error occurs, set the
915 ** session object to failed and return SQLITE_ERROR. Otherwise, return
916 ** SQLITE_OK.
917 **
918 ** It is possible that a non-fatal OOM error occurs in this function. In
919 ** that case the hash-table does not grow, but SQLITE_OK is returned anyway.
920 ** Growing the hash table in this case is a performance optimization only,
921 ** it is not required for correct operation.
922 */
sessionGrowHash(sqlite3_session * pSession,int bPatchset,SessionTable * pTab)923 static int sessionGrowHash(
924 sqlite3_session *pSession, /* For memory accounting. May be NULL */
925 int bPatchset,
926 SessionTable *pTab
927 ){
928 if( pTab->nChange==0 || pTab->nEntry>=(pTab->nChange/2) ){
929 int i;
930 SessionChange **apNew;
931 sqlite3_int64 nNew = 2*(sqlite3_int64)(pTab->nChange ? pTab->nChange : 128);
932
933 apNew = (SessionChange**)sessionMalloc64(
934 pSession, sizeof(SessionChange*) * nNew
935 );
936 if( apNew==0 ){
937 if( pTab->nChange==0 ){
938 return SQLITE_ERROR;
939 }
940 return SQLITE_OK;
941 }
942 memset(apNew, 0, sizeof(SessionChange *) * nNew);
943
944 for(i=0; i<pTab->nChange; i++){
945 SessionChange *p;
946 SessionChange *pNext;
947 for(p=pTab->apChange[i]; p; p=pNext){
948 int bPkOnly = (p->op==SQLITE_DELETE && bPatchset);
949 int iHash = sessionChangeHash(pTab, bPkOnly, p->aRecord, nNew);
950 pNext = p->pNext;
951 p->pNext = apNew[iHash];
952 apNew[iHash] = p;
953 }
954 }
955
956 sessionFree(pSession, pTab->apChange);
957 pTab->nChange = nNew;
958 pTab->apChange = apNew;
959 }
960
961 return SQLITE_OK;
962 }
963
964 /*
965 ** This function queries the database for the names of the columns of table
966 ** zThis, in schema zDb.
967 **
968 ** Otherwise, if they are not NULL, variable *pnCol is set to the number
969 ** of columns in the database table and variable *pzTab is set to point to a
970 ** nul-terminated copy of the table name. *pazCol (if not NULL) is set to
971 ** point to an array of pointers to column names. And *pabPK (again, if not
972 ** NULL) is set to point to an array of booleans - true if the corresponding
973 ** column is part of the primary key.
974 **
975 ** For example, if the table is declared as:
976 **
977 ** CREATE TABLE tbl1(w, x, y, z, PRIMARY KEY(w, z));
978 **
979 ** Then the four output variables are populated as follows:
980 **
981 ** *pnCol = 4
982 ** *pzTab = "tbl1"
983 ** *pazCol = {"w", "x", "y", "z"}
984 ** *pabPK = {1, 0, 0, 1}
985 **
986 ** All returned buffers are part of the same single allocation, which must
987 ** be freed using sqlite3_free() by the caller
988 */
sessionTableInfo(sqlite3_session * pSession,sqlite3 * db,const char * zDb,const char * zThis,int * pnCol,const char ** pzTab,const char *** pazCol,u8 ** pabPK)989 static int sessionTableInfo(
990 sqlite3_session *pSession, /* For memory accounting. May be NULL */
991 sqlite3 *db, /* Database connection */
992 const char *zDb, /* Name of attached database (e.g. "main") */
993 const char *zThis, /* Table name */
994 int *pnCol, /* OUT: number of columns */
995 const char **pzTab, /* OUT: Copy of zThis */
996 const char ***pazCol, /* OUT: Array of column names for table */
997 u8 **pabPK /* OUT: Array of booleans - true for PK col */
998 ){
999 char *zPragma;
1000 sqlite3_stmt *pStmt;
1001 int rc;
1002 sqlite3_int64 nByte;
1003 int nDbCol = 0;
1004 int nThis;
1005 int i;
1006 u8 *pAlloc = 0;
1007 char **azCol = 0;
1008 u8 *abPK = 0;
1009
1010 assert( pazCol && pabPK );
1011
1012 nThis = sqlite3Strlen30(zThis);
1013 if( nThis==12 && 0==sqlite3_stricmp("sqlite_stat1", zThis) ){
1014 rc = sqlite3_table_column_metadata(db, zDb, zThis, 0, 0, 0, 0, 0, 0);
1015 if( rc==SQLITE_OK ){
1016 /* For sqlite_stat1, pretend that (tbl,idx) is the PRIMARY KEY. */
1017 zPragma = sqlite3_mprintf(
1018 "SELECT 0, 'tbl', '', 0, '', 1 UNION ALL "
1019 "SELECT 1, 'idx', '', 0, '', 2 UNION ALL "
1020 "SELECT 2, 'stat', '', 0, '', 0"
1021 );
1022 }else if( rc==SQLITE_ERROR ){
1023 zPragma = sqlite3_mprintf("");
1024 }else{
1025 return rc;
1026 }
1027 }else{
1028 zPragma = sqlite3_mprintf("PRAGMA '%q'.table_info('%q')", zDb, zThis);
1029 }
1030 if( !zPragma ) return SQLITE_NOMEM;
1031
1032 rc = sqlite3_prepare_v2(db, zPragma, -1, &pStmt, 0);
1033 sqlite3_free(zPragma);
1034 if( rc!=SQLITE_OK ) return rc;
1035
1036 nByte = nThis + 1;
1037 while( SQLITE_ROW==sqlite3_step(pStmt) ){
1038 nByte += sqlite3_column_bytes(pStmt, 1);
1039 nDbCol++;
1040 }
1041 rc = sqlite3_reset(pStmt);
1042
1043 if( rc==SQLITE_OK ){
1044 nByte += nDbCol * (sizeof(const char *) + sizeof(u8) + 1);
1045 pAlloc = sessionMalloc64(pSession, nByte);
1046 if( pAlloc==0 ){
1047 rc = SQLITE_NOMEM;
1048 }
1049 }
1050 if( rc==SQLITE_OK ){
1051 azCol = (char **)pAlloc;
1052 pAlloc = (u8 *)&azCol[nDbCol];
1053 abPK = (u8 *)pAlloc;
1054 pAlloc = &abPK[nDbCol];
1055 if( pzTab ){
1056 memcpy(pAlloc, zThis, nThis+1);
1057 *pzTab = (char *)pAlloc;
1058 pAlloc += nThis+1;
1059 }
1060
1061 i = 0;
1062 while( SQLITE_ROW==sqlite3_step(pStmt) ){
1063 int nName = sqlite3_column_bytes(pStmt, 1);
1064 const unsigned char *zName = sqlite3_column_text(pStmt, 1);
1065 if( zName==0 ) break;
1066 memcpy(pAlloc, zName, nName+1);
1067 azCol[i] = (char *)pAlloc;
1068 pAlloc += nName+1;
1069 abPK[i] = sqlite3_column_int(pStmt, 5);
1070 i++;
1071 }
1072 rc = sqlite3_reset(pStmt);
1073
1074 }
1075
1076 /* If successful, populate the output variables. Otherwise, zero them and
1077 ** free any allocation made. An error code will be returned in this case.
1078 */
1079 if( rc==SQLITE_OK ){
1080 *pazCol = (const char **)azCol;
1081 *pabPK = abPK;
1082 *pnCol = nDbCol;
1083 }else{
1084 *pazCol = 0;
1085 *pabPK = 0;
1086 *pnCol = 0;
1087 if( pzTab ) *pzTab = 0;
1088 sessionFree(pSession, azCol);
1089 }
1090 sqlite3_finalize(pStmt);
1091 return rc;
1092 }
1093
1094 /*
1095 ** This function is only called from within a pre-update handler for a
1096 ** write to table pTab, part of session pSession. If this is the first
1097 ** write to this table, initalize the SessionTable.nCol, azCol[] and
1098 ** abPK[] arrays accordingly.
1099 **
1100 ** If an error occurs, an error code is stored in sqlite3_session.rc and
1101 ** non-zero returned. Or, if no error occurs but the table has no primary
1102 ** key, sqlite3_session.rc is left set to SQLITE_OK and non-zero returned to
1103 ** indicate that updates on this table should be ignored. SessionTable.abPK
1104 ** is set to NULL in this case.
1105 */
sessionInitTable(sqlite3_session * pSession,SessionTable * pTab)1106 static int sessionInitTable(sqlite3_session *pSession, SessionTable *pTab){
1107 if( pTab->nCol==0 ){
1108 u8 *abPK;
1109 assert( pTab->azCol==0 || pTab->abPK==0 );
1110 pSession->rc = sessionTableInfo(pSession, pSession->db, pSession->zDb,
1111 pTab->zName, &pTab->nCol, 0, &pTab->azCol, &abPK
1112 );
1113 if( pSession->rc==SQLITE_OK ){
1114 int i;
1115 for(i=0; i<pTab->nCol; i++){
1116 if( abPK[i] ){
1117 pTab->abPK = abPK;
1118 break;
1119 }
1120 }
1121 if( 0==sqlite3_stricmp("sqlite_stat1", pTab->zName) ){
1122 pTab->bStat1 = 1;
1123 }
1124 }
1125 }
1126 return (pSession->rc || pTab->abPK==0);
1127 }
1128
1129 /*
1130 ** Versions of the four methods in object SessionHook for use with the
1131 ** sqlite_stat1 table. The purpose of this is to substitute a zero-length
1132 ** blob each time a NULL value is read from the "idx" column of the
1133 ** sqlite_stat1 table.
1134 */
1135 typedef struct SessionStat1Ctx SessionStat1Ctx;
1136 struct SessionStat1Ctx {
1137 SessionHook hook;
1138 sqlite3_session *pSession;
1139 };
sessionStat1Old(void * pCtx,int iCol,sqlite3_value ** ppVal)1140 static int sessionStat1Old(void *pCtx, int iCol, sqlite3_value **ppVal){
1141 SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
1142 sqlite3_value *pVal = 0;
1143 int rc = p->hook.xOld(p->hook.pCtx, iCol, &pVal);
1144 if( rc==SQLITE_OK && iCol==1 && sqlite3_value_type(pVal)==SQLITE_NULL ){
1145 pVal = p->pSession->pZeroBlob;
1146 }
1147 *ppVal = pVal;
1148 return rc;
1149 }
sessionStat1New(void * pCtx,int iCol,sqlite3_value ** ppVal)1150 static int sessionStat1New(void *pCtx, int iCol, sqlite3_value **ppVal){
1151 SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
1152 sqlite3_value *pVal = 0;
1153 int rc = p->hook.xNew(p->hook.pCtx, iCol, &pVal);
1154 if( rc==SQLITE_OK && iCol==1 && sqlite3_value_type(pVal)==SQLITE_NULL ){
1155 pVal = p->pSession->pZeroBlob;
1156 }
1157 *ppVal = pVal;
1158 return rc;
1159 }
sessionStat1Count(void * pCtx)1160 static int sessionStat1Count(void *pCtx){
1161 SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
1162 return p->hook.xCount(p->hook.pCtx);
1163 }
sessionStat1Depth(void * pCtx)1164 static int sessionStat1Depth(void *pCtx){
1165 SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
1166 return p->hook.xDepth(p->hook.pCtx);
1167 }
1168
1169
1170 /*
1171 ** This function is only called from with a pre-update-hook reporting a
1172 ** change on table pTab (attached to session pSession). The type of change
1173 ** (UPDATE, INSERT, DELETE) is specified by the first argument.
1174 **
1175 ** Unless one is already present or an error occurs, an entry is added
1176 ** to the changed-rows hash table associated with table pTab.
1177 */
sessionPreupdateOneChange(int op,sqlite3_session * pSession,SessionTable * pTab)1178 static void sessionPreupdateOneChange(
1179 int op, /* One of SQLITE_UPDATE, INSERT, DELETE */
1180 sqlite3_session *pSession, /* Session object pTab is attached to */
1181 SessionTable *pTab /* Table that change applies to */
1182 ){
1183 int iHash;
1184 int bNull = 0;
1185 int rc = SQLITE_OK;
1186 SessionStat1Ctx stat1 = {{0,0,0,0,0},0};
1187
1188 if( pSession->rc ) return;
1189
1190 /* Load table details if required */
1191 if( sessionInitTable(pSession, pTab) ) return;
1192
1193 /* Check the number of columns in this xPreUpdate call matches the
1194 ** number of columns in the table. */
1195 if( pTab->nCol!=pSession->hook.xCount(pSession->hook.pCtx) ){
1196 pSession->rc = SQLITE_SCHEMA;
1197 return;
1198 }
1199
1200 /* Grow the hash table if required */
1201 if( sessionGrowHash(pSession, 0, pTab) ){
1202 pSession->rc = SQLITE_NOMEM;
1203 return;
1204 }
1205
1206 if( pTab->bStat1 ){
1207 stat1.hook = pSession->hook;
1208 stat1.pSession = pSession;
1209 pSession->hook.pCtx = (void*)&stat1;
1210 pSession->hook.xNew = sessionStat1New;
1211 pSession->hook.xOld = sessionStat1Old;
1212 pSession->hook.xCount = sessionStat1Count;
1213 pSession->hook.xDepth = sessionStat1Depth;
1214 if( pSession->pZeroBlob==0 ){
1215 sqlite3_value *p = sqlite3ValueNew(0);
1216 if( p==0 ){
1217 rc = SQLITE_NOMEM;
1218 goto error_out;
1219 }
1220 sqlite3ValueSetStr(p, 0, "", 0, SQLITE_STATIC);
1221 pSession->pZeroBlob = p;
1222 }
1223 }
1224
1225 /* Calculate the hash-key for this change. If the primary key of the row
1226 ** includes a NULL value, exit early. Such changes are ignored by the
1227 ** session module. */
1228 rc = sessionPreupdateHash(pSession, pTab, op==SQLITE_INSERT, &iHash, &bNull);
1229 if( rc!=SQLITE_OK ) goto error_out;
1230
1231 if( bNull==0 ){
1232 /* Search the hash table for an existing record for this row. */
1233 SessionChange *pC;
1234 for(pC=pTab->apChange[iHash]; pC; pC=pC->pNext){
1235 if( sessionPreupdateEqual(pSession, pTab, pC, op) ) break;
1236 }
1237
1238 if( pC==0 ){
1239 /* Create a new change object containing all the old values (if
1240 ** this is an SQLITE_UPDATE or SQLITE_DELETE), or just the PK
1241 ** values (if this is an INSERT). */
1242 SessionChange *pChange; /* New change object */
1243 sqlite3_int64 nByte; /* Number of bytes to allocate */
1244 int i; /* Used to iterate through columns */
1245
1246 assert( rc==SQLITE_OK );
1247 pTab->nEntry++;
1248
1249 /* Figure out how large an allocation is required */
1250 nByte = sizeof(SessionChange);
1251 for(i=0; i<pTab->nCol; i++){
1252 sqlite3_value *p = 0;
1253 if( op!=SQLITE_INSERT ){
1254 TESTONLY(int trc = ) pSession->hook.xOld(pSession->hook.pCtx, i, &p);
1255 assert( trc==SQLITE_OK );
1256 }else if( pTab->abPK[i] ){
1257 TESTONLY(int trc = ) pSession->hook.xNew(pSession->hook.pCtx, i, &p);
1258 assert( trc==SQLITE_OK );
1259 }
1260
1261 /* This may fail if SQLite value p contains a utf-16 string that must
1262 ** be converted to utf-8 and an OOM error occurs while doing so. */
1263 rc = sessionSerializeValue(0, p, &nByte);
1264 if( rc!=SQLITE_OK ) goto error_out;
1265 }
1266
1267 /* Allocate the change object */
1268 pChange = (SessionChange *)sessionMalloc64(pSession, nByte);
1269 if( !pChange ){
1270 rc = SQLITE_NOMEM;
1271 goto error_out;
1272 }else{
1273 memset(pChange, 0, sizeof(SessionChange));
1274 pChange->aRecord = (u8 *)&pChange[1];
1275 }
1276
1277 /* Populate the change object. None of the preupdate_old(),
1278 ** preupdate_new() or SerializeValue() calls below may fail as all
1279 ** required values and encodings have already been cached in memory.
1280 ** It is not possible for an OOM to occur in this block. */
1281 nByte = 0;
1282 for(i=0; i<pTab->nCol; i++){
1283 sqlite3_value *p = 0;
1284 if( op!=SQLITE_INSERT ){
1285 pSession->hook.xOld(pSession->hook.pCtx, i, &p);
1286 }else if( pTab->abPK[i] ){
1287 pSession->hook.xNew(pSession->hook.pCtx, i, &p);
1288 }
1289 sessionSerializeValue(&pChange->aRecord[nByte], p, &nByte);
1290 }
1291
1292 /* Add the change to the hash-table */
1293 if( pSession->bIndirect || pSession->hook.xDepth(pSession->hook.pCtx) ){
1294 pChange->bIndirect = 1;
1295 }
1296 pChange->nRecord = nByte;
1297 pChange->op = op;
1298 pChange->pNext = pTab->apChange[iHash];
1299 pTab->apChange[iHash] = pChange;
1300
1301 }else if( pC->bIndirect ){
1302 /* If the existing change is considered "indirect", but this current
1303 ** change is "direct", mark the change object as direct. */
1304 if( pSession->hook.xDepth(pSession->hook.pCtx)==0
1305 && pSession->bIndirect==0
1306 ){
1307 pC->bIndirect = 0;
1308 }
1309 }
1310 }
1311
1312 /* If an error has occurred, mark the session object as failed. */
1313 error_out:
1314 if( pTab->bStat1 ){
1315 pSession->hook = stat1.hook;
1316 }
1317 if( rc!=SQLITE_OK ){
1318 pSession->rc = rc;
1319 }
1320 }
1321
sessionFindTable(sqlite3_session * pSession,const char * zName,SessionTable ** ppTab)1322 static int sessionFindTable(
1323 sqlite3_session *pSession,
1324 const char *zName,
1325 SessionTable **ppTab
1326 ){
1327 int rc = SQLITE_OK;
1328 int nName = sqlite3Strlen30(zName);
1329 SessionTable *pRet;
1330
1331 /* Search for an existing table */
1332 for(pRet=pSession->pTable; pRet; pRet=pRet->pNext){
1333 if( 0==sqlite3_strnicmp(pRet->zName, zName, nName+1) ) break;
1334 }
1335
1336 if( pRet==0 && pSession->bAutoAttach ){
1337 /* If there is a table-filter configured, invoke it. If it returns 0,
1338 ** do not automatically add the new table. */
1339 if( pSession->xTableFilter==0
1340 || pSession->xTableFilter(pSession->pFilterCtx, zName)
1341 ){
1342 rc = sqlite3session_attach(pSession, zName);
1343 if( rc==SQLITE_OK ){
1344 for(pRet=pSession->pTable; pRet->pNext; pRet=pRet->pNext);
1345 assert( 0==sqlite3_strnicmp(pRet->zName, zName, nName+1) );
1346 }
1347 }
1348 }
1349
1350 assert( rc==SQLITE_OK || pRet==0 );
1351 *ppTab = pRet;
1352 return rc;
1353 }
1354
1355 /*
1356 ** The 'pre-update' hook registered by this module with SQLite databases.
1357 */
xPreUpdate(void * pCtx,sqlite3 * db,int op,char const * zDb,char const * zName,sqlite3_int64 iKey1,sqlite3_int64 iKey2)1358 static void xPreUpdate(
1359 void *pCtx, /* Copy of third arg to preupdate_hook() */
1360 sqlite3 *db, /* Database handle */
1361 int op, /* SQLITE_UPDATE, DELETE or INSERT */
1362 char const *zDb, /* Database name */
1363 char const *zName, /* Table name */
1364 sqlite3_int64 iKey1, /* Rowid of row about to be deleted/updated */
1365 sqlite3_int64 iKey2 /* New rowid value (for a rowid UPDATE) */
1366 ){
1367 sqlite3_session *pSession;
1368 int nDb = sqlite3Strlen30(zDb);
1369
1370 assert( sqlite3_mutex_held(db->mutex) );
1371
1372 for(pSession=(sqlite3_session *)pCtx; pSession; pSession=pSession->pNext){
1373 SessionTable *pTab;
1374
1375 /* If this session is attached to a different database ("main", "temp"
1376 ** etc.), or if it is not currently enabled, there is nothing to do. Skip
1377 ** to the next session object attached to this database. */
1378 if( pSession->bEnable==0 ) continue;
1379 if( pSession->rc ) continue;
1380 if( sqlite3_strnicmp(zDb, pSession->zDb, nDb+1) ) continue;
1381
1382 pSession->rc = sessionFindTable(pSession, zName, &pTab);
1383 if( pTab ){
1384 assert( pSession->rc==SQLITE_OK );
1385 sessionPreupdateOneChange(op, pSession, pTab);
1386 if( op==SQLITE_UPDATE ){
1387 sessionPreupdateOneChange(SQLITE_INSERT, pSession, pTab);
1388 }
1389 }
1390 }
1391 }
1392
1393 /*
1394 ** The pre-update hook implementations.
1395 */
sessionPreupdateOld(void * pCtx,int iVal,sqlite3_value ** ppVal)1396 static int sessionPreupdateOld(void *pCtx, int iVal, sqlite3_value **ppVal){
1397 return sqlite3_preupdate_old((sqlite3*)pCtx, iVal, ppVal);
1398 }
sessionPreupdateNew(void * pCtx,int iVal,sqlite3_value ** ppVal)1399 static int sessionPreupdateNew(void *pCtx, int iVal, sqlite3_value **ppVal){
1400 return sqlite3_preupdate_new((sqlite3*)pCtx, iVal, ppVal);
1401 }
sessionPreupdateCount(void * pCtx)1402 static int sessionPreupdateCount(void *pCtx){
1403 return sqlite3_preupdate_count((sqlite3*)pCtx);
1404 }
sessionPreupdateDepth(void * pCtx)1405 static int sessionPreupdateDepth(void *pCtx){
1406 return sqlite3_preupdate_depth((sqlite3*)pCtx);
1407 }
1408
1409 /*
1410 ** Install the pre-update hooks on the session object passed as the only
1411 ** argument.
1412 */
sessionPreupdateHooks(sqlite3_session * pSession)1413 static void sessionPreupdateHooks(
1414 sqlite3_session *pSession
1415 ){
1416 pSession->hook.pCtx = (void*)pSession->db;
1417 pSession->hook.xOld = sessionPreupdateOld;
1418 pSession->hook.xNew = sessionPreupdateNew;
1419 pSession->hook.xCount = sessionPreupdateCount;
1420 pSession->hook.xDepth = sessionPreupdateDepth;
1421 }
1422
1423 typedef struct SessionDiffCtx SessionDiffCtx;
1424 struct SessionDiffCtx {
1425 sqlite3_stmt *pStmt;
1426 int nOldOff;
1427 };
1428
1429 /*
1430 ** The diff hook implementations.
1431 */
sessionDiffOld(void * pCtx,int iVal,sqlite3_value ** ppVal)1432 static int sessionDiffOld(void *pCtx, int iVal, sqlite3_value **ppVal){
1433 SessionDiffCtx *p = (SessionDiffCtx*)pCtx;
1434 *ppVal = sqlite3_column_value(p->pStmt, iVal+p->nOldOff);
1435 return SQLITE_OK;
1436 }
sessionDiffNew(void * pCtx,int iVal,sqlite3_value ** ppVal)1437 static int sessionDiffNew(void *pCtx, int iVal, sqlite3_value **ppVal){
1438 SessionDiffCtx *p = (SessionDiffCtx*)pCtx;
1439 *ppVal = sqlite3_column_value(p->pStmt, iVal);
1440 return SQLITE_OK;
1441 }
sessionDiffCount(void * pCtx)1442 static int sessionDiffCount(void *pCtx){
1443 SessionDiffCtx *p = (SessionDiffCtx*)pCtx;
1444 return p->nOldOff ? p->nOldOff : sqlite3_column_count(p->pStmt);
1445 }
sessionDiffDepth(void * pCtx)1446 static int sessionDiffDepth(void *pCtx){
1447 return 0;
1448 }
1449
1450 /*
1451 ** Install the diff hooks on the session object passed as the only
1452 ** argument.
1453 */
sessionDiffHooks(sqlite3_session * pSession,SessionDiffCtx * pDiffCtx)1454 static void sessionDiffHooks(
1455 sqlite3_session *pSession,
1456 SessionDiffCtx *pDiffCtx
1457 ){
1458 pSession->hook.pCtx = (void*)pDiffCtx;
1459 pSession->hook.xOld = sessionDiffOld;
1460 pSession->hook.xNew = sessionDiffNew;
1461 pSession->hook.xCount = sessionDiffCount;
1462 pSession->hook.xDepth = sessionDiffDepth;
1463 }
1464
sessionExprComparePK(int nCol,const char * zDb1,const char * zDb2,const char * zTab,const char ** azCol,u8 * abPK)1465 static char *sessionExprComparePK(
1466 int nCol,
1467 const char *zDb1, const char *zDb2,
1468 const char *zTab,
1469 const char **azCol, u8 *abPK
1470 ){
1471 int i;
1472 const char *zSep = "";
1473 char *zRet = 0;
1474
1475 for(i=0; i<nCol; i++){
1476 if( abPK[i] ){
1477 zRet = sqlite3_mprintf("%z%s\"%w\".\"%w\".\"%w\"=\"%w\".\"%w\".\"%w\"",
1478 zRet, zSep, zDb1, zTab, azCol[i], zDb2, zTab, azCol[i]
1479 );
1480 zSep = " AND ";
1481 if( zRet==0 ) break;
1482 }
1483 }
1484
1485 return zRet;
1486 }
1487
sessionExprCompareOther(int nCol,const char * zDb1,const char * zDb2,const char * zTab,const char ** azCol,u8 * abPK)1488 static char *sessionExprCompareOther(
1489 int nCol,
1490 const char *zDb1, const char *zDb2,
1491 const char *zTab,
1492 const char **azCol, u8 *abPK
1493 ){
1494 int i;
1495 const char *zSep = "";
1496 char *zRet = 0;
1497 int bHave = 0;
1498
1499 for(i=0; i<nCol; i++){
1500 if( abPK[i]==0 ){
1501 bHave = 1;
1502 zRet = sqlite3_mprintf(
1503 "%z%s\"%w\".\"%w\".\"%w\" IS NOT \"%w\".\"%w\".\"%w\"",
1504 zRet, zSep, zDb1, zTab, azCol[i], zDb2, zTab, azCol[i]
1505 );
1506 zSep = " OR ";
1507 if( zRet==0 ) break;
1508 }
1509 }
1510
1511 if( bHave==0 ){
1512 assert( zRet==0 );
1513 zRet = sqlite3_mprintf("0");
1514 }
1515
1516 return zRet;
1517 }
1518
sessionSelectFindNew(int nCol,const char * zDb1,const char * zDb2,const char * zTbl,const char * zExpr)1519 static char *sessionSelectFindNew(
1520 int nCol,
1521 const char *zDb1, /* Pick rows in this db only */
1522 const char *zDb2, /* But not in this one */
1523 const char *zTbl, /* Table name */
1524 const char *zExpr
1525 ){
1526 char *zRet = sqlite3_mprintf(
1527 "SELECT * FROM \"%w\".\"%w\" WHERE NOT EXISTS ("
1528 " SELECT 1 FROM \"%w\".\"%w\" WHERE %s"
1529 ")",
1530 zDb1, zTbl, zDb2, zTbl, zExpr
1531 );
1532 return zRet;
1533 }
1534
sessionDiffFindNew(int op,sqlite3_session * pSession,SessionTable * pTab,const char * zDb1,const char * zDb2,char * zExpr)1535 static int sessionDiffFindNew(
1536 int op,
1537 sqlite3_session *pSession,
1538 SessionTable *pTab,
1539 const char *zDb1,
1540 const char *zDb2,
1541 char *zExpr
1542 ){
1543 int rc = SQLITE_OK;
1544 char *zStmt = sessionSelectFindNew(pTab->nCol, zDb1, zDb2, pTab->zName,zExpr);
1545
1546 if( zStmt==0 ){
1547 rc = SQLITE_NOMEM;
1548 }else{
1549 sqlite3_stmt *pStmt;
1550 rc = sqlite3_prepare(pSession->db, zStmt, -1, &pStmt, 0);
1551 if( rc==SQLITE_OK ){
1552 SessionDiffCtx *pDiffCtx = (SessionDiffCtx*)pSession->hook.pCtx;
1553 pDiffCtx->pStmt = pStmt;
1554 pDiffCtx->nOldOff = 0;
1555 while( SQLITE_ROW==sqlite3_step(pStmt) ){
1556 sessionPreupdateOneChange(op, pSession, pTab);
1557 }
1558 rc = sqlite3_finalize(pStmt);
1559 }
1560 sqlite3_free(zStmt);
1561 }
1562
1563 return rc;
1564 }
1565
sessionDiffFindModified(sqlite3_session * pSession,SessionTable * pTab,const char * zFrom,const char * zExpr)1566 static int sessionDiffFindModified(
1567 sqlite3_session *pSession,
1568 SessionTable *pTab,
1569 const char *zFrom,
1570 const char *zExpr
1571 ){
1572 int rc = SQLITE_OK;
1573
1574 char *zExpr2 = sessionExprCompareOther(pTab->nCol,
1575 pSession->zDb, zFrom, pTab->zName, pTab->azCol, pTab->abPK
1576 );
1577 if( zExpr2==0 ){
1578 rc = SQLITE_NOMEM;
1579 }else{
1580 char *zStmt = sqlite3_mprintf(
1581 "SELECT * FROM \"%w\".\"%w\", \"%w\".\"%w\" WHERE %s AND (%z)",
1582 pSession->zDb, pTab->zName, zFrom, pTab->zName, zExpr, zExpr2
1583 );
1584 if( zStmt==0 ){
1585 rc = SQLITE_NOMEM;
1586 }else{
1587 sqlite3_stmt *pStmt;
1588 rc = sqlite3_prepare(pSession->db, zStmt, -1, &pStmt, 0);
1589
1590 if( rc==SQLITE_OK ){
1591 SessionDiffCtx *pDiffCtx = (SessionDiffCtx*)pSession->hook.pCtx;
1592 pDiffCtx->pStmt = pStmt;
1593 pDiffCtx->nOldOff = pTab->nCol;
1594 while( SQLITE_ROW==sqlite3_step(pStmt) ){
1595 sessionPreupdateOneChange(SQLITE_UPDATE, pSession, pTab);
1596 }
1597 rc = sqlite3_finalize(pStmt);
1598 }
1599 sqlite3_free(zStmt);
1600 }
1601 }
1602
1603 return rc;
1604 }
1605
sqlite3session_diff(sqlite3_session * pSession,const char * zFrom,const char * zTbl,char ** pzErrMsg)1606 int sqlite3session_diff(
1607 sqlite3_session *pSession,
1608 const char *zFrom,
1609 const char *zTbl,
1610 char **pzErrMsg
1611 ){
1612 const char *zDb = pSession->zDb;
1613 int rc = pSession->rc;
1614 SessionDiffCtx d;
1615
1616 memset(&d, 0, sizeof(d));
1617 sessionDiffHooks(pSession, &d);
1618
1619 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
1620 if( pzErrMsg ) *pzErrMsg = 0;
1621 if( rc==SQLITE_OK ){
1622 char *zExpr = 0;
1623 sqlite3 *db = pSession->db;
1624 SessionTable *pTo; /* Table zTbl */
1625
1626 /* Locate and if necessary initialize the target table object */
1627 rc = sessionFindTable(pSession, zTbl, &pTo);
1628 if( pTo==0 ) goto diff_out;
1629 if( sessionInitTable(pSession, pTo) ){
1630 rc = pSession->rc;
1631 goto diff_out;
1632 }
1633
1634 /* Check the table schemas match */
1635 if( rc==SQLITE_OK ){
1636 int bHasPk = 0;
1637 int bMismatch = 0;
1638 int nCol; /* Columns in zFrom.zTbl */
1639 u8 *abPK;
1640 const char **azCol = 0;
1641 rc = sessionTableInfo(0, db, zFrom, zTbl, &nCol, 0, &azCol, &abPK);
1642 if( rc==SQLITE_OK ){
1643 if( pTo->nCol!=nCol ){
1644 bMismatch = 1;
1645 }else{
1646 int i;
1647 for(i=0; i<nCol; i++){
1648 if( pTo->abPK[i]!=abPK[i] ) bMismatch = 1;
1649 if( sqlite3_stricmp(azCol[i], pTo->azCol[i]) ) bMismatch = 1;
1650 if( abPK[i] ) bHasPk = 1;
1651 }
1652 }
1653 }
1654 sqlite3_free((char*)azCol);
1655 if( bMismatch ){
1656 if( pzErrMsg ){
1657 *pzErrMsg = sqlite3_mprintf("table schemas do not match");
1658 }
1659 rc = SQLITE_SCHEMA;
1660 }
1661 if( bHasPk==0 ){
1662 /* Ignore tables with no primary keys */
1663 goto diff_out;
1664 }
1665 }
1666
1667 if( rc==SQLITE_OK ){
1668 zExpr = sessionExprComparePK(pTo->nCol,
1669 zDb, zFrom, pTo->zName, pTo->azCol, pTo->abPK
1670 );
1671 }
1672
1673 /* Find new rows */
1674 if( rc==SQLITE_OK ){
1675 rc = sessionDiffFindNew(SQLITE_INSERT, pSession, pTo, zDb, zFrom, zExpr);
1676 }
1677
1678 /* Find old rows */
1679 if( rc==SQLITE_OK ){
1680 rc = sessionDiffFindNew(SQLITE_DELETE, pSession, pTo, zFrom, zDb, zExpr);
1681 }
1682
1683 /* Find modified rows */
1684 if( rc==SQLITE_OK ){
1685 rc = sessionDiffFindModified(pSession, pTo, zFrom, zExpr);
1686 }
1687
1688 sqlite3_free(zExpr);
1689 }
1690
1691 diff_out:
1692 sessionPreupdateHooks(pSession);
1693 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
1694 return rc;
1695 }
1696
1697 /*
1698 ** Create a session object. This session object will record changes to
1699 ** database zDb attached to connection db.
1700 */
sqlite3session_create(sqlite3 * db,const char * zDb,sqlite3_session ** ppSession)1701 int sqlite3session_create(
1702 sqlite3 *db, /* Database handle */
1703 const char *zDb, /* Name of db (e.g. "main") */
1704 sqlite3_session **ppSession /* OUT: New session object */
1705 ){
1706 sqlite3_session *pNew; /* Newly allocated session object */
1707 sqlite3_session *pOld; /* Session object already attached to db */
1708 int nDb = sqlite3Strlen30(zDb); /* Length of zDb in bytes */
1709
1710 /* Zero the output value in case an error occurs. */
1711 *ppSession = 0;
1712
1713 /* Allocate and populate the new session object. */
1714 pNew = (sqlite3_session *)sqlite3_malloc64(sizeof(sqlite3_session) + nDb + 1);
1715 if( !pNew ) return SQLITE_NOMEM;
1716 memset(pNew, 0, sizeof(sqlite3_session));
1717 pNew->db = db;
1718 pNew->zDb = (char *)&pNew[1];
1719 pNew->bEnable = 1;
1720 memcpy(pNew->zDb, zDb, nDb+1);
1721 sessionPreupdateHooks(pNew);
1722
1723 /* Add the new session object to the linked list of session objects
1724 ** attached to database handle $db. Do this under the cover of the db
1725 ** handle mutex. */
1726 sqlite3_mutex_enter(sqlite3_db_mutex(db));
1727 pOld = (sqlite3_session*)sqlite3_preupdate_hook(db, xPreUpdate, (void*)pNew);
1728 pNew->pNext = pOld;
1729 sqlite3_mutex_leave(sqlite3_db_mutex(db));
1730
1731 *ppSession = pNew;
1732 return SQLITE_OK;
1733 }
1734
1735 /*
1736 ** Free the list of table objects passed as the first argument. The contents
1737 ** of the changed-rows hash tables are also deleted.
1738 */
sessionDeleteTable(sqlite3_session * pSession,SessionTable * pList)1739 static void sessionDeleteTable(sqlite3_session *pSession, SessionTable *pList){
1740 SessionTable *pNext;
1741 SessionTable *pTab;
1742
1743 for(pTab=pList; pTab; pTab=pNext){
1744 int i;
1745 pNext = pTab->pNext;
1746 for(i=0; i<pTab->nChange; i++){
1747 SessionChange *p;
1748 SessionChange *pNextChange;
1749 for(p=pTab->apChange[i]; p; p=pNextChange){
1750 pNextChange = p->pNext;
1751 sessionFree(pSession, p);
1752 }
1753 }
1754 sessionFree(pSession, (char*)pTab->azCol); /* cast works around VC++ bug */
1755 sessionFree(pSession, pTab->apChange);
1756 sessionFree(pSession, pTab);
1757 }
1758 }
1759
1760 /*
1761 ** Delete a session object previously allocated using sqlite3session_create().
1762 */
sqlite3session_delete(sqlite3_session * pSession)1763 void sqlite3session_delete(sqlite3_session *pSession){
1764 sqlite3 *db = pSession->db;
1765 sqlite3_session *pHead;
1766 sqlite3_session **pp;
1767
1768 /* Unlink the session from the linked list of sessions attached to the
1769 ** database handle. Hold the db mutex while doing so. */
1770 sqlite3_mutex_enter(sqlite3_db_mutex(db));
1771 pHead = (sqlite3_session*)sqlite3_preupdate_hook(db, 0, 0);
1772 for(pp=&pHead; ALWAYS((*pp)!=0); pp=&((*pp)->pNext)){
1773 if( (*pp)==pSession ){
1774 *pp = (*pp)->pNext;
1775 if( pHead ) sqlite3_preupdate_hook(db, xPreUpdate, (void*)pHead);
1776 break;
1777 }
1778 }
1779 sqlite3_mutex_leave(sqlite3_db_mutex(db));
1780 sqlite3ValueFree(pSession->pZeroBlob);
1781
1782 /* Delete all attached table objects. And the contents of their
1783 ** associated hash-tables. */
1784 sessionDeleteTable(pSession, pSession->pTable);
1785
1786 /* Assert that all allocations have been freed and then free the
1787 ** session object itself. */
1788 assert( pSession->nMalloc==0 );
1789 sqlite3_free(pSession);
1790 }
1791
1792 /*
1793 ** Set a table filter on a Session Object.
1794 */
sqlite3session_table_filter(sqlite3_session * pSession,int (* xFilter)(void *,const char *),void * pCtx)1795 void sqlite3session_table_filter(
1796 sqlite3_session *pSession,
1797 int(*xFilter)(void*, const char*),
1798 void *pCtx /* First argument passed to xFilter */
1799 ){
1800 pSession->bAutoAttach = 1;
1801 pSession->pFilterCtx = pCtx;
1802 pSession->xTableFilter = xFilter;
1803 }
1804
1805 /*
1806 ** Attach a table to a session. All subsequent changes made to the table
1807 ** while the session object is enabled will be recorded.
1808 **
1809 ** Only tables that have a PRIMARY KEY defined may be attached. It does
1810 ** not matter if the PRIMARY KEY is an "INTEGER PRIMARY KEY" (rowid alias)
1811 ** or not.
1812 */
sqlite3session_attach(sqlite3_session * pSession,const char * zName)1813 int sqlite3session_attach(
1814 sqlite3_session *pSession, /* Session object */
1815 const char *zName /* Table name */
1816 ){
1817 int rc = SQLITE_OK;
1818 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
1819
1820 if( !zName ){
1821 pSession->bAutoAttach = 1;
1822 }else{
1823 SessionTable *pTab; /* New table object (if required) */
1824 int nName; /* Number of bytes in string zName */
1825
1826 /* First search for an existing entry. If one is found, this call is
1827 ** a no-op. Return early. */
1828 nName = sqlite3Strlen30(zName);
1829 for(pTab=pSession->pTable; pTab; pTab=pTab->pNext){
1830 if( 0==sqlite3_strnicmp(pTab->zName, zName, nName+1) ) break;
1831 }
1832
1833 if( !pTab ){
1834 /* Allocate new SessionTable object. */
1835 int nByte = sizeof(SessionTable) + nName + 1;
1836 pTab = (SessionTable*)sessionMalloc64(pSession, nByte);
1837 if( !pTab ){
1838 rc = SQLITE_NOMEM;
1839 }else{
1840 /* Populate the new SessionTable object and link it into the list.
1841 ** The new object must be linked onto the end of the list, not
1842 ** simply added to the start of it in order to ensure that tables
1843 ** appear in the correct order when a changeset or patchset is
1844 ** eventually generated. */
1845 SessionTable **ppTab;
1846 memset(pTab, 0, sizeof(SessionTable));
1847 pTab->zName = (char *)&pTab[1];
1848 memcpy(pTab->zName, zName, nName+1);
1849 for(ppTab=&pSession->pTable; *ppTab; ppTab=&(*ppTab)->pNext);
1850 *ppTab = pTab;
1851 }
1852 }
1853 }
1854
1855 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
1856 return rc;
1857 }
1858
1859 /*
1860 ** Ensure that there is room in the buffer to append nByte bytes of data.
1861 ** If not, use sqlite3_realloc() to grow the buffer so that there is.
1862 **
1863 ** If successful, return zero. Otherwise, if an OOM condition is encountered,
1864 ** set *pRc to SQLITE_NOMEM and return non-zero.
1865 */
sessionBufferGrow(SessionBuffer * p,size_t nByte,int * pRc)1866 static int sessionBufferGrow(SessionBuffer *p, size_t nByte, int *pRc){
1867 if( *pRc==SQLITE_OK && (size_t)(p->nAlloc-p->nBuf)<nByte ){
1868 u8 *aNew;
1869 i64 nNew = p->nAlloc ? p->nAlloc : 128;
1870 do {
1871 nNew = nNew*2;
1872 }while( (size_t)(nNew-p->nBuf)<nByte );
1873
1874 aNew = (u8 *)sqlite3_realloc64(p->aBuf, nNew);
1875 if( 0==aNew ){
1876 *pRc = SQLITE_NOMEM;
1877 }else{
1878 p->aBuf = aNew;
1879 p->nAlloc = nNew;
1880 }
1881 }
1882 return (*pRc!=SQLITE_OK);
1883 }
1884
1885 /*
1886 ** Append the value passed as the second argument to the buffer passed
1887 ** as the first.
1888 **
1889 ** This function is a no-op if *pRc is non-zero when it is called.
1890 ** Otherwise, if an error occurs, *pRc is set to an SQLite error code
1891 ** before returning.
1892 */
sessionAppendValue(SessionBuffer * p,sqlite3_value * pVal,int * pRc)1893 static void sessionAppendValue(SessionBuffer *p, sqlite3_value *pVal, int *pRc){
1894 int rc = *pRc;
1895 if( rc==SQLITE_OK ){
1896 sqlite3_int64 nByte = 0;
1897 rc = sessionSerializeValue(0, pVal, &nByte);
1898 sessionBufferGrow(p, nByte, &rc);
1899 if( rc==SQLITE_OK ){
1900 rc = sessionSerializeValue(&p->aBuf[p->nBuf], pVal, 0);
1901 p->nBuf += nByte;
1902 }else{
1903 *pRc = rc;
1904 }
1905 }
1906 }
1907
1908 /*
1909 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1910 ** called. Otherwise, append a single byte to the buffer.
1911 **
1912 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
1913 ** returning.
1914 */
sessionAppendByte(SessionBuffer * p,u8 v,int * pRc)1915 static void sessionAppendByte(SessionBuffer *p, u8 v, int *pRc){
1916 if( 0==sessionBufferGrow(p, 1, pRc) ){
1917 p->aBuf[p->nBuf++] = v;
1918 }
1919 }
1920
1921 /*
1922 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1923 ** called. Otherwise, append a single varint to the buffer.
1924 **
1925 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
1926 ** returning.
1927 */
sessionAppendVarint(SessionBuffer * p,int v,int * pRc)1928 static void sessionAppendVarint(SessionBuffer *p, int v, int *pRc){
1929 if( 0==sessionBufferGrow(p, 9, pRc) ){
1930 p->nBuf += sessionVarintPut(&p->aBuf[p->nBuf], v);
1931 }
1932 }
1933
1934 /*
1935 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1936 ** called. Otherwise, append a blob of data to the buffer.
1937 **
1938 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
1939 ** returning.
1940 */
sessionAppendBlob(SessionBuffer * p,const u8 * aBlob,int nBlob,int * pRc)1941 static void sessionAppendBlob(
1942 SessionBuffer *p,
1943 const u8 *aBlob,
1944 int nBlob,
1945 int *pRc
1946 ){
1947 if( nBlob>0 && 0==sessionBufferGrow(p, nBlob, pRc) ){
1948 memcpy(&p->aBuf[p->nBuf], aBlob, nBlob);
1949 p->nBuf += nBlob;
1950 }
1951 }
1952
1953 /*
1954 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1955 ** called. Otherwise, append a string to the buffer. All bytes in the string
1956 ** up to (but not including) the nul-terminator are written to the buffer.
1957 **
1958 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
1959 ** returning.
1960 */
sessionAppendStr(SessionBuffer * p,const char * zStr,int * pRc)1961 static void sessionAppendStr(
1962 SessionBuffer *p,
1963 const char *zStr,
1964 int *pRc
1965 ){
1966 int nStr = sqlite3Strlen30(zStr);
1967 if( 0==sessionBufferGrow(p, nStr, pRc) ){
1968 memcpy(&p->aBuf[p->nBuf], zStr, nStr);
1969 p->nBuf += nStr;
1970 }
1971 }
1972
1973 /*
1974 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1975 ** called. Otherwise, append the string representation of integer iVal
1976 ** to the buffer. No nul-terminator is written.
1977 **
1978 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
1979 ** returning.
1980 */
sessionAppendInteger(SessionBuffer * p,int iVal,int * pRc)1981 static void sessionAppendInteger(
1982 SessionBuffer *p, /* Buffer to append to */
1983 int iVal, /* Value to write the string rep. of */
1984 int *pRc /* IN/OUT: Error code */
1985 ){
1986 char aBuf[24];
1987 sqlite3_snprintf(sizeof(aBuf)-1, aBuf, "%d", iVal);
1988 sessionAppendStr(p, aBuf, pRc);
1989 }
1990
1991 /*
1992 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1993 ** called. Otherwise, append the string zStr enclosed in quotes (") and
1994 ** with any embedded quote characters escaped to the buffer. No
1995 ** nul-terminator byte is written.
1996 **
1997 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
1998 ** returning.
1999 */
sessionAppendIdent(SessionBuffer * p,const char * zStr,int * pRc)2000 static void sessionAppendIdent(
2001 SessionBuffer *p, /* Buffer to a append to */
2002 const char *zStr, /* String to quote, escape and append */
2003 int *pRc /* IN/OUT: Error code */
2004 ){
2005 int nStr = sqlite3Strlen30(zStr)*2 + 2 + 1;
2006 if( 0==sessionBufferGrow(p, nStr, pRc) ){
2007 char *zOut = (char *)&p->aBuf[p->nBuf];
2008 const char *zIn = zStr;
2009 *zOut++ = '"';
2010 while( *zIn ){
2011 if( *zIn=='"' ) *zOut++ = '"';
2012 *zOut++ = *(zIn++);
2013 }
2014 *zOut++ = '"';
2015 p->nBuf = (int)((u8 *)zOut - p->aBuf);
2016 }
2017 }
2018
2019 /*
2020 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
2021 ** called. Otherwse, it appends the serialized version of the value stored
2022 ** in column iCol of the row that SQL statement pStmt currently points
2023 ** to to the buffer.
2024 */
sessionAppendCol(SessionBuffer * p,sqlite3_stmt * pStmt,int iCol,int * pRc)2025 static void sessionAppendCol(
2026 SessionBuffer *p, /* Buffer to append to */
2027 sqlite3_stmt *pStmt, /* Handle pointing to row containing value */
2028 int iCol, /* Column to read value from */
2029 int *pRc /* IN/OUT: Error code */
2030 ){
2031 if( *pRc==SQLITE_OK ){
2032 int eType = sqlite3_column_type(pStmt, iCol);
2033 sessionAppendByte(p, (u8)eType, pRc);
2034 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
2035 sqlite3_int64 i;
2036 u8 aBuf[8];
2037 if( eType==SQLITE_INTEGER ){
2038 i = sqlite3_column_int64(pStmt, iCol);
2039 }else{
2040 double r = sqlite3_column_double(pStmt, iCol);
2041 memcpy(&i, &r, 8);
2042 }
2043 sessionPutI64(aBuf, i);
2044 sessionAppendBlob(p, aBuf, 8, pRc);
2045 }
2046 if( eType==SQLITE_BLOB || eType==SQLITE_TEXT ){
2047 u8 *z;
2048 int nByte;
2049 if( eType==SQLITE_BLOB ){
2050 z = (u8 *)sqlite3_column_blob(pStmt, iCol);
2051 }else{
2052 z = (u8 *)sqlite3_column_text(pStmt, iCol);
2053 }
2054 nByte = sqlite3_column_bytes(pStmt, iCol);
2055 if( z || (eType==SQLITE_BLOB && nByte==0) ){
2056 sessionAppendVarint(p, nByte, pRc);
2057 sessionAppendBlob(p, z, nByte, pRc);
2058 }else{
2059 *pRc = SQLITE_NOMEM;
2060 }
2061 }
2062 }
2063 }
2064
2065 /*
2066 **
2067 ** This function appends an update change to the buffer (see the comments
2068 ** under "CHANGESET FORMAT" at the top of the file). An update change
2069 ** consists of:
2070 **
2071 ** 1 byte: SQLITE_UPDATE (0x17)
2072 ** n bytes: old.* record (see RECORD FORMAT)
2073 ** m bytes: new.* record (see RECORD FORMAT)
2074 **
2075 ** The SessionChange object passed as the third argument contains the
2076 ** values that were stored in the row when the session began (the old.*
2077 ** values). The statement handle passed as the second argument points
2078 ** at the current version of the row (the new.* values).
2079 **
2080 ** If all of the old.* values are equal to their corresponding new.* value
2081 ** (i.e. nothing has changed), then no data at all is appended to the buffer.
2082 **
2083 ** Otherwise, the old.* record contains all primary key values and the
2084 ** original values of any fields that have been modified. The new.* record
2085 ** contains the new values of only those fields that have been modified.
2086 */
sessionAppendUpdate(SessionBuffer * pBuf,int bPatchset,sqlite3_stmt * pStmt,SessionChange * p,u8 * abPK)2087 static int sessionAppendUpdate(
2088 SessionBuffer *pBuf, /* Buffer to append to */
2089 int bPatchset, /* True for "patchset", 0 for "changeset" */
2090 sqlite3_stmt *pStmt, /* Statement handle pointing at new row */
2091 SessionChange *p, /* Object containing old values */
2092 u8 *abPK /* Boolean array - true for PK columns */
2093 ){
2094 int rc = SQLITE_OK;
2095 SessionBuffer buf2 = {0,0,0}; /* Buffer to accumulate new.* record in */
2096 int bNoop = 1; /* Set to zero if any values are modified */
2097 int nRewind = pBuf->nBuf; /* Set to zero if any values are modified */
2098 int i; /* Used to iterate through columns */
2099 u8 *pCsr = p->aRecord; /* Used to iterate through old.* values */
2100
2101 sessionAppendByte(pBuf, SQLITE_UPDATE, &rc);
2102 sessionAppendByte(pBuf, p->bIndirect, &rc);
2103 for(i=0; i<sqlite3_column_count(pStmt); i++){
2104 int bChanged = 0;
2105 int nAdvance;
2106 int eType = *pCsr;
2107 switch( eType ){
2108 case SQLITE_NULL:
2109 nAdvance = 1;
2110 if( sqlite3_column_type(pStmt, i)!=SQLITE_NULL ){
2111 bChanged = 1;
2112 }
2113 break;
2114
2115 case SQLITE_FLOAT:
2116 case SQLITE_INTEGER: {
2117 nAdvance = 9;
2118 if( eType==sqlite3_column_type(pStmt, i) ){
2119 sqlite3_int64 iVal = sessionGetI64(&pCsr[1]);
2120 if( eType==SQLITE_INTEGER ){
2121 if( iVal==sqlite3_column_int64(pStmt, i) ) break;
2122 }else{
2123 double dVal;
2124 memcpy(&dVal, &iVal, 8);
2125 if( dVal==sqlite3_column_double(pStmt, i) ) break;
2126 }
2127 }
2128 bChanged = 1;
2129 break;
2130 }
2131
2132 default: {
2133 int n;
2134 int nHdr = 1 + sessionVarintGet(&pCsr[1], &n);
2135 assert( eType==SQLITE_TEXT || eType==SQLITE_BLOB );
2136 nAdvance = nHdr + n;
2137 if( eType==sqlite3_column_type(pStmt, i)
2138 && n==sqlite3_column_bytes(pStmt, i)
2139 && (n==0 || 0==memcmp(&pCsr[nHdr], sqlite3_column_blob(pStmt, i), n))
2140 ){
2141 break;
2142 }
2143 bChanged = 1;
2144 }
2145 }
2146
2147 /* If at least one field has been modified, this is not a no-op. */
2148 if( bChanged ) bNoop = 0;
2149
2150 /* Add a field to the old.* record. This is omitted if this modules is
2151 ** currently generating a patchset. */
2152 if( bPatchset==0 ){
2153 if( bChanged || abPK[i] ){
2154 sessionAppendBlob(pBuf, pCsr, nAdvance, &rc);
2155 }else{
2156 sessionAppendByte(pBuf, 0, &rc);
2157 }
2158 }
2159
2160 /* Add a field to the new.* record. Or the only record if currently
2161 ** generating a patchset. */
2162 if( bChanged || (bPatchset && abPK[i]) ){
2163 sessionAppendCol(&buf2, pStmt, i, &rc);
2164 }else{
2165 sessionAppendByte(&buf2, 0, &rc);
2166 }
2167
2168 pCsr += nAdvance;
2169 }
2170
2171 if( bNoop ){
2172 pBuf->nBuf = nRewind;
2173 }else{
2174 sessionAppendBlob(pBuf, buf2.aBuf, buf2.nBuf, &rc);
2175 }
2176 sqlite3_free(buf2.aBuf);
2177
2178 return rc;
2179 }
2180
2181 /*
2182 ** Append a DELETE change to the buffer passed as the first argument. Use
2183 ** the changeset format if argument bPatchset is zero, or the patchset
2184 ** format otherwise.
2185 */
sessionAppendDelete(SessionBuffer * pBuf,int bPatchset,SessionChange * p,int nCol,u8 * abPK)2186 static int sessionAppendDelete(
2187 SessionBuffer *pBuf, /* Buffer to append to */
2188 int bPatchset, /* True for "patchset", 0 for "changeset" */
2189 SessionChange *p, /* Object containing old values */
2190 int nCol, /* Number of columns in table */
2191 u8 *abPK /* Boolean array - true for PK columns */
2192 ){
2193 int rc = SQLITE_OK;
2194
2195 sessionAppendByte(pBuf, SQLITE_DELETE, &rc);
2196 sessionAppendByte(pBuf, p->bIndirect, &rc);
2197
2198 if( bPatchset==0 ){
2199 sessionAppendBlob(pBuf, p->aRecord, p->nRecord, &rc);
2200 }else{
2201 int i;
2202 u8 *a = p->aRecord;
2203 for(i=0; i<nCol; i++){
2204 u8 *pStart = a;
2205 int eType = *a++;
2206
2207 switch( eType ){
2208 case 0:
2209 case SQLITE_NULL:
2210 assert( abPK[i]==0 );
2211 break;
2212
2213 case SQLITE_FLOAT:
2214 case SQLITE_INTEGER:
2215 a += 8;
2216 break;
2217
2218 default: {
2219 int n;
2220 a += sessionVarintGet(a, &n);
2221 a += n;
2222 break;
2223 }
2224 }
2225 if( abPK[i] ){
2226 sessionAppendBlob(pBuf, pStart, (int)(a-pStart), &rc);
2227 }
2228 }
2229 assert( (a - p->aRecord)==p->nRecord );
2230 }
2231
2232 return rc;
2233 }
2234
2235 /*
2236 ** Formulate and prepare a SELECT statement to retrieve a row from table
2237 ** zTab in database zDb based on its primary key. i.e.
2238 **
2239 ** SELECT * FROM zDb.zTab WHERE pk1 = ? AND pk2 = ? AND ...
2240 */
sessionSelectStmt(sqlite3 * db,const char * zDb,const char * zTab,int nCol,const char ** azCol,u8 * abPK,sqlite3_stmt ** ppStmt)2241 static int sessionSelectStmt(
2242 sqlite3 *db, /* Database handle */
2243 const char *zDb, /* Database name */
2244 const char *zTab, /* Table name */
2245 int nCol, /* Number of columns in table */
2246 const char **azCol, /* Names of table columns */
2247 u8 *abPK, /* PRIMARY KEY array */
2248 sqlite3_stmt **ppStmt /* OUT: Prepared SELECT statement */
2249 ){
2250 int rc = SQLITE_OK;
2251 char *zSql = 0;
2252 int nSql = -1;
2253
2254 if( 0==sqlite3_stricmp("sqlite_stat1", zTab) ){
2255 zSql = sqlite3_mprintf(
2256 "SELECT tbl, ?2, stat FROM %Q.sqlite_stat1 WHERE tbl IS ?1 AND "
2257 "idx IS (CASE WHEN ?2=X'' THEN NULL ELSE ?2 END)", zDb
2258 );
2259 if( zSql==0 ) rc = SQLITE_NOMEM;
2260 }else{
2261 int i;
2262 const char *zSep = "";
2263 SessionBuffer buf = {0, 0, 0};
2264
2265 sessionAppendStr(&buf, "SELECT * FROM ", &rc);
2266 sessionAppendIdent(&buf, zDb, &rc);
2267 sessionAppendStr(&buf, ".", &rc);
2268 sessionAppendIdent(&buf, zTab, &rc);
2269 sessionAppendStr(&buf, " WHERE ", &rc);
2270 for(i=0; i<nCol; i++){
2271 if( abPK[i] ){
2272 sessionAppendStr(&buf, zSep, &rc);
2273 sessionAppendIdent(&buf, azCol[i], &rc);
2274 sessionAppendStr(&buf, " IS ?", &rc);
2275 sessionAppendInteger(&buf, i+1, &rc);
2276 zSep = " AND ";
2277 }
2278 }
2279 zSql = (char*)buf.aBuf;
2280 nSql = buf.nBuf;
2281 }
2282
2283 if( rc==SQLITE_OK ){
2284 rc = sqlite3_prepare_v2(db, zSql, nSql, ppStmt, 0);
2285 }
2286 sqlite3_free(zSql);
2287 return rc;
2288 }
2289
2290 /*
2291 ** Bind the PRIMARY KEY values from the change passed in argument pChange
2292 ** to the SELECT statement passed as the first argument. The SELECT statement
2293 ** is as prepared by function sessionSelectStmt().
2294 **
2295 ** Return SQLITE_OK if all PK values are successfully bound, or an SQLite
2296 ** error code (e.g. SQLITE_NOMEM) otherwise.
2297 */
sessionSelectBind(sqlite3_stmt * pSelect,int nCol,u8 * abPK,SessionChange * pChange)2298 static int sessionSelectBind(
2299 sqlite3_stmt *pSelect, /* SELECT from sessionSelectStmt() */
2300 int nCol, /* Number of columns in table */
2301 u8 *abPK, /* PRIMARY KEY array */
2302 SessionChange *pChange /* Change structure */
2303 ){
2304 int i;
2305 int rc = SQLITE_OK;
2306 u8 *a = pChange->aRecord;
2307
2308 for(i=0; i<nCol && rc==SQLITE_OK; i++){
2309 int eType = *a++;
2310
2311 switch( eType ){
2312 case 0:
2313 case SQLITE_NULL:
2314 assert( abPK[i]==0 );
2315 break;
2316
2317 case SQLITE_INTEGER: {
2318 if( abPK[i] ){
2319 i64 iVal = sessionGetI64(a);
2320 rc = sqlite3_bind_int64(pSelect, i+1, iVal);
2321 }
2322 a += 8;
2323 break;
2324 }
2325
2326 case SQLITE_FLOAT: {
2327 if( abPK[i] ){
2328 double rVal;
2329 i64 iVal = sessionGetI64(a);
2330 memcpy(&rVal, &iVal, 8);
2331 rc = sqlite3_bind_double(pSelect, i+1, rVal);
2332 }
2333 a += 8;
2334 break;
2335 }
2336
2337 case SQLITE_TEXT: {
2338 int n;
2339 a += sessionVarintGet(a, &n);
2340 if( abPK[i] ){
2341 rc = sqlite3_bind_text(pSelect, i+1, (char *)a, n, SQLITE_TRANSIENT);
2342 }
2343 a += n;
2344 break;
2345 }
2346
2347 default: {
2348 int n;
2349 assert( eType==SQLITE_BLOB );
2350 a += sessionVarintGet(a, &n);
2351 if( abPK[i] ){
2352 rc = sqlite3_bind_blob(pSelect, i+1, a, n, SQLITE_TRANSIENT);
2353 }
2354 a += n;
2355 break;
2356 }
2357 }
2358 }
2359
2360 return rc;
2361 }
2362
2363 /*
2364 ** This function is a no-op if *pRc is set to other than SQLITE_OK when it
2365 ** is called. Otherwise, append a serialized table header (part of the binary
2366 ** changeset format) to buffer *pBuf. If an error occurs, set *pRc to an
2367 ** SQLite error code before returning.
2368 */
sessionAppendTableHdr(SessionBuffer * pBuf,int bPatchset,SessionTable * pTab,int * pRc)2369 static void sessionAppendTableHdr(
2370 SessionBuffer *pBuf, /* Append header to this buffer */
2371 int bPatchset, /* Use the patchset format if true */
2372 SessionTable *pTab, /* Table object to append header for */
2373 int *pRc /* IN/OUT: Error code */
2374 ){
2375 /* Write a table header */
2376 sessionAppendByte(pBuf, (bPatchset ? 'P' : 'T'), pRc);
2377 sessionAppendVarint(pBuf, pTab->nCol, pRc);
2378 sessionAppendBlob(pBuf, pTab->abPK, pTab->nCol, pRc);
2379 sessionAppendBlob(pBuf, (u8 *)pTab->zName, (int)strlen(pTab->zName)+1, pRc);
2380 }
2381
2382 /*
2383 ** Generate either a changeset (if argument bPatchset is zero) or a patchset
2384 ** (if it is non-zero) based on the current contents of the session object
2385 ** passed as the first argument.
2386 **
2387 ** If no error occurs, SQLITE_OK is returned and the new changeset/patchset
2388 ** stored in output variables *pnChangeset and *ppChangeset. Or, if an error
2389 ** occurs, an SQLite error code is returned and both output variables set
2390 ** to 0.
2391 */
sessionGenerateChangeset(sqlite3_session * pSession,int bPatchset,int (* xOutput)(void * pOut,const void * pData,int nData),void * pOut,int * pnChangeset,void ** ppChangeset)2392 static int sessionGenerateChangeset(
2393 sqlite3_session *pSession, /* Session object */
2394 int bPatchset, /* True for patchset, false for changeset */
2395 int (*xOutput)(void *pOut, const void *pData, int nData),
2396 void *pOut, /* First argument for xOutput */
2397 int *pnChangeset, /* OUT: Size of buffer at *ppChangeset */
2398 void **ppChangeset /* OUT: Buffer containing changeset */
2399 ){
2400 sqlite3 *db = pSession->db; /* Source database handle */
2401 SessionTable *pTab; /* Used to iterate through attached tables */
2402 SessionBuffer buf = {0,0,0}; /* Buffer in which to accumlate changeset */
2403 int rc; /* Return code */
2404
2405 assert( xOutput==0 || (pnChangeset==0 && ppChangeset==0 ) );
2406
2407 /* Zero the output variables in case an error occurs. If this session
2408 ** object is already in the error state (sqlite3_session.rc != SQLITE_OK),
2409 ** this call will be a no-op. */
2410 if( xOutput==0 ){
2411 *pnChangeset = 0;
2412 *ppChangeset = 0;
2413 }
2414
2415 if( pSession->rc ) return pSession->rc;
2416 rc = sqlite3_exec(pSession->db, "SAVEPOINT changeset", 0, 0, 0);
2417 if( rc!=SQLITE_OK ) return rc;
2418
2419 sqlite3_mutex_enter(sqlite3_db_mutex(db));
2420
2421 for(pTab=pSession->pTable; rc==SQLITE_OK && pTab; pTab=pTab->pNext){
2422 if( pTab->nEntry ){
2423 const char *zName = pTab->zName;
2424 int nCol; /* Number of columns in table */
2425 u8 *abPK; /* Primary key array */
2426 const char **azCol = 0; /* Table columns */
2427 int i; /* Used to iterate through hash buckets */
2428 sqlite3_stmt *pSel = 0; /* SELECT statement to query table pTab */
2429 int nRewind = buf.nBuf; /* Initial size of write buffer */
2430 int nNoop; /* Size of buffer after writing tbl header */
2431
2432 /* Check the table schema is still Ok. */
2433 rc = sessionTableInfo(0, db, pSession->zDb, zName, &nCol, 0,&azCol,&abPK);
2434 if( !rc && (pTab->nCol!=nCol || memcmp(abPK, pTab->abPK, nCol)) ){
2435 rc = SQLITE_SCHEMA;
2436 }
2437
2438 /* Write a table header */
2439 sessionAppendTableHdr(&buf, bPatchset, pTab, &rc);
2440
2441 /* Build and compile a statement to execute: */
2442 if( rc==SQLITE_OK ){
2443 rc = sessionSelectStmt(
2444 db, pSession->zDb, zName, nCol, azCol, abPK, &pSel);
2445 }
2446
2447 nNoop = buf.nBuf;
2448 for(i=0; i<pTab->nChange && rc==SQLITE_OK; i++){
2449 SessionChange *p; /* Used to iterate through changes */
2450
2451 for(p=pTab->apChange[i]; rc==SQLITE_OK && p; p=p->pNext){
2452 rc = sessionSelectBind(pSel, nCol, abPK, p);
2453 if( rc!=SQLITE_OK ) continue;
2454 if( sqlite3_step(pSel)==SQLITE_ROW ){
2455 if( p->op==SQLITE_INSERT ){
2456 int iCol;
2457 sessionAppendByte(&buf, SQLITE_INSERT, &rc);
2458 sessionAppendByte(&buf, p->bIndirect, &rc);
2459 for(iCol=0; iCol<nCol; iCol++){
2460 sessionAppendCol(&buf, pSel, iCol, &rc);
2461 }
2462 }else{
2463 rc = sessionAppendUpdate(&buf, bPatchset, pSel, p, abPK);
2464 }
2465 }else if( p->op!=SQLITE_INSERT ){
2466 rc = sessionAppendDelete(&buf, bPatchset, p, nCol, abPK);
2467 }
2468 if( rc==SQLITE_OK ){
2469 rc = sqlite3_reset(pSel);
2470 }
2471
2472 /* If the buffer is now larger than sessions_strm_chunk_size, pass
2473 ** its contents to the xOutput() callback. */
2474 if( xOutput
2475 && rc==SQLITE_OK
2476 && buf.nBuf>nNoop
2477 && buf.nBuf>sessions_strm_chunk_size
2478 ){
2479 rc = xOutput(pOut, (void*)buf.aBuf, buf.nBuf);
2480 nNoop = -1;
2481 buf.nBuf = 0;
2482 }
2483
2484 }
2485 }
2486
2487 sqlite3_finalize(pSel);
2488 if( buf.nBuf==nNoop ){
2489 buf.nBuf = nRewind;
2490 }
2491 sqlite3_free((char*)azCol); /* cast works around VC++ bug */
2492 }
2493 }
2494
2495 if( rc==SQLITE_OK ){
2496 if( xOutput==0 ){
2497 *pnChangeset = buf.nBuf;
2498 *ppChangeset = buf.aBuf;
2499 buf.aBuf = 0;
2500 }else if( buf.nBuf>0 ){
2501 rc = xOutput(pOut, (void*)buf.aBuf, buf.nBuf);
2502 }
2503 }
2504
2505 sqlite3_free(buf.aBuf);
2506 sqlite3_exec(db, "RELEASE changeset", 0, 0, 0);
2507 sqlite3_mutex_leave(sqlite3_db_mutex(db));
2508 return rc;
2509 }
2510
2511 /*
2512 ** Obtain a changeset object containing all changes recorded by the
2513 ** session object passed as the first argument.
2514 **
2515 ** It is the responsibility of the caller to eventually free the buffer
2516 ** using sqlite3_free().
2517 */
sqlite3session_changeset(sqlite3_session * pSession,int * pnChangeset,void ** ppChangeset)2518 int sqlite3session_changeset(
2519 sqlite3_session *pSession, /* Session object */
2520 int *pnChangeset, /* OUT: Size of buffer at *ppChangeset */
2521 void **ppChangeset /* OUT: Buffer containing changeset */
2522 ){
2523 return sessionGenerateChangeset(pSession, 0, 0, 0, pnChangeset, ppChangeset);
2524 }
2525
2526 /*
2527 ** Streaming version of sqlite3session_changeset().
2528 */
sqlite3session_changeset_strm(sqlite3_session * pSession,int (* xOutput)(void * pOut,const void * pData,int nData),void * pOut)2529 int sqlite3session_changeset_strm(
2530 sqlite3_session *pSession,
2531 int (*xOutput)(void *pOut, const void *pData, int nData),
2532 void *pOut
2533 ){
2534 return sessionGenerateChangeset(pSession, 0, xOutput, pOut, 0, 0);
2535 }
2536
2537 /*
2538 ** Streaming version of sqlite3session_patchset().
2539 */
sqlite3session_patchset_strm(sqlite3_session * pSession,int (* xOutput)(void * pOut,const void * pData,int nData),void * pOut)2540 int sqlite3session_patchset_strm(
2541 sqlite3_session *pSession,
2542 int (*xOutput)(void *pOut, const void *pData, int nData),
2543 void *pOut
2544 ){
2545 return sessionGenerateChangeset(pSession, 1, xOutput, pOut, 0, 0);
2546 }
2547
2548 /*
2549 ** Obtain a patchset object containing all changes recorded by the
2550 ** session object passed as the first argument.
2551 **
2552 ** It is the responsibility of the caller to eventually free the buffer
2553 ** using sqlite3_free().
2554 */
sqlite3session_patchset(sqlite3_session * pSession,int * pnPatchset,void ** ppPatchset)2555 int sqlite3session_patchset(
2556 sqlite3_session *pSession, /* Session object */
2557 int *pnPatchset, /* OUT: Size of buffer at *ppChangeset */
2558 void **ppPatchset /* OUT: Buffer containing changeset */
2559 ){
2560 return sessionGenerateChangeset(pSession, 1, 0, 0, pnPatchset, ppPatchset);
2561 }
2562
2563 /*
2564 ** Enable or disable the session object passed as the first argument.
2565 */
sqlite3session_enable(sqlite3_session * pSession,int bEnable)2566 int sqlite3session_enable(sqlite3_session *pSession, int bEnable){
2567 int ret;
2568 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
2569 if( bEnable>=0 ){
2570 pSession->bEnable = bEnable;
2571 }
2572 ret = pSession->bEnable;
2573 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
2574 return ret;
2575 }
2576
2577 /*
2578 ** Enable or disable the session object passed as the first argument.
2579 */
sqlite3session_indirect(sqlite3_session * pSession,int bIndirect)2580 int sqlite3session_indirect(sqlite3_session *pSession, int bIndirect){
2581 int ret;
2582 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
2583 if( bIndirect>=0 ){
2584 pSession->bIndirect = bIndirect;
2585 }
2586 ret = pSession->bIndirect;
2587 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
2588 return ret;
2589 }
2590
2591 /*
2592 ** Return true if there have been no changes to monitored tables recorded
2593 ** by the session object passed as the only argument.
2594 */
sqlite3session_isempty(sqlite3_session * pSession)2595 int sqlite3session_isempty(sqlite3_session *pSession){
2596 int ret = 0;
2597 SessionTable *pTab;
2598
2599 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
2600 for(pTab=pSession->pTable; pTab && ret==0; pTab=pTab->pNext){
2601 ret = (pTab->nEntry>0);
2602 }
2603 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
2604
2605 return (ret==0);
2606 }
2607
2608 /*
2609 ** Return the amount of heap memory in use.
2610 */
sqlite3session_memory_used(sqlite3_session * pSession)2611 sqlite3_int64 sqlite3session_memory_used(sqlite3_session *pSession){
2612 return pSession->nMalloc;
2613 }
2614
2615 /*
2616 ** Do the work for either sqlite3changeset_start() or start_strm().
2617 */
sessionChangesetStart(sqlite3_changeset_iter ** pp,int (* xInput)(void * pIn,void * pData,int * pnData),void * pIn,int nChangeset,void * pChangeset,int bInvert,int bSkipEmpty)2618 static int sessionChangesetStart(
2619 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2620 int (*xInput)(void *pIn, void *pData, int *pnData),
2621 void *pIn,
2622 int nChangeset, /* Size of buffer pChangeset in bytes */
2623 void *pChangeset, /* Pointer to buffer containing changeset */
2624 int bInvert, /* True to invert changeset */
2625 int bSkipEmpty /* True to skip empty UPDATE changes */
2626 ){
2627 sqlite3_changeset_iter *pRet; /* Iterator to return */
2628 int nByte; /* Number of bytes to allocate for iterator */
2629
2630 assert( xInput==0 || (pChangeset==0 && nChangeset==0) );
2631
2632 /* Zero the output variable in case an error occurs. */
2633 *pp = 0;
2634
2635 /* Allocate and initialize the iterator structure. */
2636 nByte = sizeof(sqlite3_changeset_iter);
2637 pRet = (sqlite3_changeset_iter *)sqlite3_malloc(nByte);
2638 if( !pRet ) return SQLITE_NOMEM;
2639 memset(pRet, 0, sizeof(sqlite3_changeset_iter));
2640 pRet->in.aData = (u8 *)pChangeset;
2641 pRet->in.nData = nChangeset;
2642 pRet->in.xInput = xInput;
2643 pRet->in.pIn = pIn;
2644 pRet->in.bEof = (xInput ? 0 : 1);
2645 pRet->bInvert = bInvert;
2646 pRet->bSkipEmpty = bSkipEmpty;
2647
2648 /* Populate the output variable and return success. */
2649 *pp = pRet;
2650 return SQLITE_OK;
2651 }
2652
2653 /*
2654 ** Create an iterator used to iterate through the contents of a changeset.
2655 */
sqlite3changeset_start(sqlite3_changeset_iter ** pp,int nChangeset,void * pChangeset)2656 int sqlite3changeset_start(
2657 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2658 int nChangeset, /* Size of buffer pChangeset in bytes */
2659 void *pChangeset /* Pointer to buffer containing changeset */
2660 ){
2661 return sessionChangesetStart(pp, 0, 0, nChangeset, pChangeset, 0, 0);
2662 }
sqlite3changeset_start_v2(sqlite3_changeset_iter ** pp,int nChangeset,void * pChangeset,int flags)2663 int sqlite3changeset_start_v2(
2664 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2665 int nChangeset, /* Size of buffer pChangeset in bytes */
2666 void *pChangeset, /* Pointer to buffer containing changeset */
2667 int flags
2668 ){
2669 int bInvert = !!(flags & SQLITE_CHANGESETSTART_INVERT);
2670 return sessionChangesetStart(pp, 0, 0, nChangeset, pChangeset, bInvert, 0);
2671 }
2672
2673 /*
2674 ** Streaming version of sqlite3changeset_start().
2675 */
sqlite3changeset_start_strm(sqlite3_changeset_iter ** pp,int (* xInput)(void * pIn,void * pData,int * pnData),void * pIn)2676 int sqlite3changeset_start_strm(
2677 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2678 int (*xInput)(void *pIn, void *pData, int *pnData),
2679 void *pIn
2680 ){
2681 return sessionChangesetStart(pp, xInput, pIn, 0, 0, 0, 0);
2682 }
sqlite3changeset_start_v2_strm(sqlite3_changeset_iter ** pp,int (* xInput)(void * pIn,void * pData,int * pnData),void * pIn,int flags)2683 int sqlite3changeset_start_v2_strm(
2684 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2685 int (*xInput)(void *pIn, void *pData, int *pnData),
2686 void *pIn,
2687 int flags
2688 ){
2689 int bInvert = !!(flags & SQLITE_CHANGESETSTART_INVERT);
2690 return sessionChangesetStart(pp, xInput, pIn, 0, 0, bInvert, 0);
2691 }
2692
2693 /*
2694 ** If the SessionInput object passed as the only argument is a streaming
2695 ** object and the buffer is full, discard some data to free up space.
2696 */
sessionDiscardData(SessionInput * pIn)2697 static void sessionDiscardData(SessionInput *pIn){
2698 if( pIn->xInput && pIn->iNext>=sessions_strm_chunk_size ){
2699 int nMove = pIn->buf.nBuf - pIn->iNext;
2700 assert( nMove>=0 );
2701 if( nMove>0 ){
2702 memmove(pIn->buf.aBuf, &pIn->buf.aBuf[pIn->iNext], nMove);
2703 }
2704 pIn->buf.nBuf -= pIn->iNext;
2705 pIn->iNext = 0;
2706 pIn->nData = pIn->buf.nBuf;
2707 }
2708 }
2709
2710 /*
2711 ** Ensure that there are at least nByte bytes available in the buffer. Or,
2712 ** if there are not nByte bytes remaining in the input, that all available
2713 ** data is in the buffer.
2714 **
2715 ** Return an SQLite error code if an error occurs, or SQLITE_OK otherwise.
2716 */
sessionInputBuffer(SessionInput * pIn,int nByte)2717 static int sessionInputBuffer(SessionInput *pIn, int nByte){
2718 int rc = SQLITE_OK;
2719 if( pIn->xInput ){
2720 while( !pIn->bEof && (pIn->iNext+nByte)>=pIn->nData && rc==SQLITE_OK ){
2721 int nNew = sessions_strm_chunk_size;
2722
2723 if( pIn->bNoDiscard==0 ) sessionDiscardData(pIn);
2724 if( SQLITE_OK==sessionBufferGrow(&pIn->buf, nNew, &rc) ){
2725 rc = pIn->xInput(pIn->pIn, &pIn->buf.aBuf[pIn->buf.nBuf], &nNew);
2726 if( nNew==0 ){
2727 pIn->bEof = 1;
2728 }else{
2729 pIn->buf.nBuf += nNew;
2730 }
2731 }
2732
2733 pIn->aData = pIn->buf.aBuf;
2734 pIn->nData = pIn->buf.nBuf;
2735 }
2736 }
2737 return rc;
2738 }
2739
2740 /*
2741 ** When this function is called, *ppRec points to the start of a record
2742 ** that contains nCol values. This function advances the pointer *ppRec
2743 ** until it points to the byte immediately following that record.
2744 */
sessionSkipRecord(u8 ** ppRec,int nCol)2745 static void sessionSkipRecord(
2746 u8 **ppRec, /* IN/OUT: Record pointer */
2747 int nCol /* Number of values in record */
2748 ){
2749 u8 *aRec = *ppRec;
2750 int i;
2751 for(i=0; i<nCol; i++){
2752 int eType = *aRec++;
2753 if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
2754 int nByte;
2755 aRec += sessionVarintGet((u8*)aRec, &nByte);
2756 aRec += nByte;
2757 }else if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
2758 aRec += 8;
2759 }
2760 }
2761
2762 *ppRec = aRec;
2763 }
2764
2765 /*
2766 ** This function sets the value of the sqlite3_value object passed as the
2767 ** first argument to a copy of the string or blob held in the aData[]
2768 ** buffer. SQLITE_OK is returned if successful, or SQLITE_NOMEM if an OOM
2769 ** error occurs.
2770 */
sessionValueSetStr(sqlite3_value * pVal,u8 * aData,int nData,u8 enc)2771 static int sessionValueSetStr(
2772 sqlite3_value *pVal, /* Set the value of this object */
2773 u8 *aData, /* Buffer containing string or blob data */
2774 int nData, /* Size of buffer aData[] in bytes */
2775 u8 enc /* String encoding (0 for blobs) */
2776 ){
2777 /* In theory this code could just pass SQLITE_TRANSIENT as the final
2778 ** argument to sqlite3ValueSetStr() and have the copy created
2779 ** automatically. But doing so makes it difficult to detect any OOM
2780 ** error. Hence the code to create the copy externally. */
2781 u8 *aCopy = sqlite3_malloc64((sqlite3_int64)nData+1);
2782 if( aCopy==0 ) return SQLITE_NOMEM;
2783 memcpy(aCopy, aData, nData);
2784 sqlite3ValueSetStr(pVal, nData, (char*)aCopy, enc, sqlite3_free);
2785 return SQLITE_OK;
2786 }
2787
2788 /*
2789 ** Deserialize a single record from a buffer in memory. See "RECORD FORMAT"
2790 ** for details.
2791 **
2792 ** When this function is called, *paChange points to the start of the record
2793 ** to deserialize. Assuming no error occurs, *paChange is set to point to
2794 ** one byte after the end of the same record before this function returns.
2795 ** If the argument abPK is NULL, then the record contains nCol values. Or,
2796 ** if abPK is other than NULL, then the record contains only the PK fields
2797 ** (in other words, it is a patchset DELETE record).
2798 **
2799 ** If successful, each element of the apOut[] array (allocated by the caller)
2800 ** is set to point to an sqlite3_value object containing the value read
2801 ** from the corresponding position in the record. If that value is not
2802 ** included in the record (i.e. because the record is part of an UPDATE change
2803 ** and the field was not modified), the corresponding element of apOut[] is
2804 ** set to NULL.
2805 **
2806 ** It is the responsibility of the caller to free all sqlite_value structures
2807 ** using sqlite3_free().
2808 **
2809 ** If an error occurs, an SQLite error code (e.g. SQLITE_NOMEM) is returned.
2810 ** The apOut[] array may have been partially populated in this case.
2811 */
sessionReadRecord(SessionInput * pIn,int nCol,u8 * abPK,sqlite3_value ** apOut,int * pbEmpty)2812 static int sessionReadRecord(
2813 SessionInput *pIn, /* Input data */
2814 int nCol, /* Number of values in record */
2815 u8 *abPK, /* Array of primary key flags, or NULL */
2816 sqlite3_value **apOut, /* Write values to this array */
2817 int *pbEmpty
2818 ){
2819 int i; /* Used to iterate through columns */
2820 int rc = SQLITE_OK;
2821
2822 assert( pbEmpty==0 || *pbEmpty==0 );
2823 if( pbEmpty ) *pbEmpty = 1;
2824 for(i=0; i<nCol && rc==SQLITE_OK; i++){
2825 int eType = 0; /* Type of value (SQLITE_NULL, TEXT etc.) */
2826 if( abPK && abPK[i]==0 ) continue;
2827 rc = sessionInputBuffer(pIn, 9);
2828 if( rc==SQLITE_OK ){
2829 if( pIn->iNext>=pIn->nData ){
2830 rc = SQLITE_CORRUPT_BKPT;
2831 }else{
2832 eType = pIn->aData[pIn->iNext++];
2833 assert( apOut[i]==0 );
2834 if( eType ){
2835 if( pbEmpty ) *pbEmpty = 0;
2836 apOut[i] = sqlite3ValueNew(0);
2837 if( !apOut[i] ) rc = SQLITE_NOMEM;
2838 }
2839 }
2840 }
2841
2842 if( rc==SQLITE_OK ){
2843 u8 *aVal = &pIn->aData[pIn->iNext];
2844 if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
2845 int nByte;
2846 pIn->iNext += sessionVarintGet(aVal, &nByte);
2847 rc = sessionInputBuffer(pIn, nByte);
2848 if( rc==SQLITE_OK ){
2849 if( nByte<0 || nByte>pIn->nData-pIn->iNext ){
2850 rc = SQLITE_CORRUPT_BKPT;
2851 }else{
2852 u8 enc = (eType==SQLITE_TEXT ? SQLITE_UTF8 : 0);
2853 rc = sessionValueSetStr(apOut[i],&pIn->aData[pIn->iNext],nByte,enc);
2854 pIn->iNext += nByte;
2855 }
2856 }
2857 }
2858 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
2859 sqlite3_int64 v = sessionGetI64(aVal);
2860 if( eType==SQLITE_INTEGER ){
2861 sqlite3VdbeMemSetInt64(apOut[i], v);
2862 }else{
2863 double d;
2864 memcpy(&d, &v, 8);
2865 sqlite3VdbeMemSetDouble(apOut[i], d);
2866 }
2867 pIn->iNext += 8;
2868 }
2869 }
2870 }
2871
2872 return rc;
2873 }
2874
2875 /*
2876 ** The input pointer currently points to the second byte of a table-header.
2877 ** Specifically, to the following:
2878 **
2879 ** + number of columns in table (varint)
2880 ** + array of PK flags (1 byte per column),
2881 ** + table name (nul terminated).
2882 **
2883 ** This function ensures that all of the above is present in the input
2884 ** buffer (i.e. that it can be accessed without any calls to xInput()).
2885 ** If successful, SQLITE_OK is returned. Otherwise, an SQLite error code.
2886 ** The input pointer is not moved.
2887 */
sessionChangesetBufferTblhdr(SessionInput * pIn,int * pnByte)2888 static int sessionChangesetBufferTblhdr(SessionInput *pIn, int *pnByte){
2889 int rc = SQLITE_OK;
2890 int nCol = 0;
2891 int nRead = 0;
2892
2893 rc = sessionInputBuffer(pIn, 9);
2894 if( rc==SQLITE_OK ){
2895 nRead += sessionVarintGet(&pIn->aData[pIn->iNext + nRead], &nCol);
2896 /* The hard upper limit for the number of columns in an SQLite
2897 ** database table is, according to sqliteLimit.h, 32676. So
2898 ** consider any table-header that purports to have more than 65536
2899 ** columns to be corrupt. This is convenient because otherwise,
2900 ** if the (nCol>65536) condition below were omitted, a sufficiently
2901 ** large value for nCol may cause nRead to wrap around and become
2902 ** negative. Leading to a crash. */
2903 if( nCol<0 || nCol>65536 ){
2904 rc = SQLITE_CORRUPT_BKPT;
2905 }else{
2906 rc = sessionInputBuffer(pIn, nRead+nCol+100);
2907 nRead += nCol;
2908 }
2909 }
2910
2911 while( rc==SQLITE_OK ){
2912 while( (pIn->iNext + nRead)<pIn->nData && pIn->aData[pIn->iNext + nRead] ){
2913 nRead++;
2914 }
2915 if( (pIn->iNext + nRead)<pIn->nData ) break;
2916 rc = sessionInputBuffer(pIn, nRead + 100);
2917 }
2918 *pnByte = nRead+1;
2919 return rc;
2920 }
2921
2922 /*
2923 ** The input pointer currently points to the first byte of the first field
2924 ** of a record consisting of nCol columns. This function ensures the entire
2925 ** record is buffered. It does not move the input pointer.
2926 **
2927 ** If successful, SQLITE_OK is returned and *pnByte is set to the size of
2928 ** the record in bytes. Otherwise, an SQLite error code is returned. The
2929 ** final value of *pnByte is undefined in this case.
2930 */
sessionChangesetBufferRecord(SessionInput * pIn,int nCol,int * pnByte)2931 static int sessionChangesetBufferRecord(
2932 SessionInput *pIn, /* Input data */
2933 int nCol, /* Number of columns in record */
2934 int *pnByte /* OUT: Size of record in bytes */
2935 ){
2936 int rc = SQLITE_OK;
2937 int nByte = 0;
2938 int i;
2939 for(i=0; rc==SQLITE_OK && i<nCol; i++){
2940 int eType;
2941 rc = sessionInputBuffer(pIn, nByte + 10);
2942 if( rc==SQLITE_OK ){
2943 eType = pIn->aData[pIn->iNext + nByte++];
2944 if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
2945 int n;
2946 nByte += sessionVarintGet(&pIn->aData[pIn->iNext+nByte], &n);
2947 nByte += n;
2948 rc = sessionInputBuffer(pIn, nByte);
2949 }else if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
2950 nByte += 8;
2951 }
2952 }
2953 }
2954 *pnByte = nByte;
2955 return rc;
2956 }
2957
2958 /*
2959 ** The input pointer currently points to the second byte of a table-header.
2960 ** Specifically, to the following:
2961 **
2962 ** + number of columns in table (varint)
2963 ** + array of PK flags (1 byte per column),
2964 ** + table name (nul terminated).
2965 **
2966 ** This function decodes the table-header and populates the p->nCol,
2967 ** p->zTab and p->abPK[] variables accordingly. The p->apValue[] array is
2968 ** also allocated or resized according to the new value of p->nCol. The
2969 ** input pointer is left pointing to the byte following the table header.
2970 **
2971 ** If successful, SQLITE_OK is returned. Otherwise, an SQLite error code
2972 ** is returned and the final values of the various fields enumerated above
2973 ** are undefined.
2974 */
sessionChangesetReadTblhdr(sqlite3_changeset_iter * p)2975 static int sessionChangesetReadTblhdr(sqlite3_changeset_iter *p){
2976 int rc;
2977 int nCopy;
2978 assert( p->rc==SQLITE_OK );
2979
2980 rc = sessionChangesetBufferTblhdr(&p->in, &nCopy);
2981 if( rc==SQLITE_OK ){
2982 int nByte;
2983 int nVarint;
2984 nVarint = sessionVarintGet(&p->in.aData[p->in.iNext], &p->nCol);
2985 if( p->nCol>0 ){
2986 nCopy -= nVarint;
2987 p->in.iNext += nVarint;
2988 nByte = p->nCol * sizeof(sqlite3_value*) * 2 + nCopy;
2989 p->tblhdr.nBuf = 0;
2990 sessionBufferGrow(&p->tblhdr, nByte, &rc);
2991 }else{
2992 rc = SQLITE_CORRUPT_BKPT;
2993 }
2994 }
2995
2996 if( rc==SQLITE_OK ){
2997 size_t iPK = sizeof(sqlite3_value*)*p->nCol*2;
2998 memset(p->tblhdr.aBuf, 0, iPK);
2999 memcpy(&p->tblhdr.aBuf[iPK], &p->in.aData[p->in.iNext], nCopy);
3000 p->in.iNext += nCopy;
3001 }
3002
3003 p->apValue = (sqlite3_value**)p->tblhdr.aBuf;
3004 if( p->apValue==0 ){
3005 p->abPK = 0;
3006 p->zTab = 0;
3007 }else{
3008 p->abPK = (u8*)&p->apValue[p->nCol*2];
3009 p->zTab = p->abPK ? (char*)&p->abPK[p->nCol] : 0;
3010 }
3011 return (p->rc = rc);
3012 }
3013
3014 /*
3015 ** Advance the changeset iterator to the next change. The differences between
3016 ** this function and sessionChangesetNext() are that
3017 **
3018 ** * If pbEmpty is not NULL and the change is a no-op UPDATE (an UPDATE
3019 ** that modifies no columns), this function sets (*pbEmpty) to 1.
3020 **
3021 ** * If the iterator is configured to skip no-op UPDATEs,
3022 ** sessionChangesetNext() does that. This function does not.
3023 */
sessionChangesetNextOne(sqlite3_changeset_iter * p,u8 ** paRec,int * pnRec,int * pbNew,int * pbEmpty)3024 static int sessionChangesetNextOne(
3025 sqlite3_changeset_iter *p, /* Changeset iterator */
3026 u8 **paRec, /* If non-NULL, store record pointer here */
3027 int *pnRec, /* If non-NULL, store size of record here */
3028 int *pbNew, /* If non-NULL, true if new table */
3029 int *pbEmpty
3030 ){
3031 int i;
3032 u8 op;
3033
3034 assert( (paRec==0 && pnRec==0) || (paRec && pnRec) );
3035 assert( pbEmpty==0 || *pbEmpty==0 );
3036
3037 /* If the iterator is in the error-state, return immediately. */
3038 if( p->rc!=SQLITE_OK ) return p->rc;
3039
3040 /* Free the current contents of p->apValue[], if any. */
3041 if( p->apValue ){
3042 for(i=0; i<p->nCol*2; i++){
3043 sqlite3ValueFree(p->apValue[i]);
3044 }
3045 memset(p->apValue, 0, sizeof(sqlite3_value*)*p->nCol*2);
3046 }
3047
3048 /* Make sure the buffer contains at least 10 bytes of input data, or all
3049 ** remaining data if there are less than 10 bytes available. This is
3050 ** sufficient either for the 'T' or 'P' byte and the varint that follows
3051 ** it, or for the two single byte values otherwise. */
3052 p->rc = sessionInputBuffer(&p->in, 2);
3053 if( p->rc!=SQLITE_OK ) return p->rc;
3054
3055 /* If the iterator is already at the end of the changeset, return DONE. */
3056 if( p->in.iNext>=p->in.nData ){
3057 return SQLITE_DONE;
3058 }
3059
3060 sessionDiscardData(&p->in);
3061 p->in.iCurrent = p->in.iNext;
3062
3063 op = p->in.aData[p->in.iNext++];
3064 while( op=='T' || op=='P' ){
3065 if( pbNew ) *pbNew = 1;
3066 p->bPatchset = (op=='P');
3067 if( sessionChangesetReadTblhdr(p) ) return p->rc;
3068 if( (p->rc = sessionInputBuffer(&p->in, 2)) ) return p->rc;
3069 p->in.iCurrent = p->in.iNext;
3070 if( p->in.iNext>=p->in.nData ) return SQLITE_DONE;
3071 op = p->in.aData[p->in.iNext++];
3072 }
3073
3074 if( p->zTab==0 || (p->bPatchset && p->bInvert) ){
3075 /* The first record in the changeset is not a table header. Must be a
3076 ** corrupt changeset. */
3077 assert( p->in.iNext==1 || p->zTab );
3078 return (p->rc = SQLITE_CORRUPT_BKPT);
3079 }
3080
3081 p->op = op;
3082 p->bIndirect = p->in.aData[p->in.iNext++];
3083 if( p->op!=SQLITE_UPDATE && p->op!=SQLITE_DELETE && p->op!=SQLITE_INSERT ){
3084 return (p->rc = SQLITE_CORRUPT_BKPT);
3085 }
3086
3087 if( paRec ){
3088 int nVal; /* Number of values to buffer */
3089 if( p->bPatchset==0 && op==SQLITE_UPDATE ){
3090 nVal = p->nCol * 2;
3091 }else if( p->bPatchset && op==SQLITE_DELETE ){
3092 nVal = 0;
3093 for(i=0; i<p->nCol; i++) if( p->abPK[i] ) nVal++;
3094 }else{
3095 nVal = p->nCol;
3096 }
3097 p->rc = sessionChangesetBufferRecord(&p->in, nVal, pnRec);
3098 if( p->rc!=SQLITE_OK ) return p->rc;
3099 *paRec = &p->in.aData[p->in.iNext];
3100 p->in.iNext += *pnRec;
3101 }else{
3102 sqlite3_value **apOld = (p->bInvert ? &p->apValue[p->nCol] : p->apValue);
3103 sqlite3_value **apNew = (p->bInvert ? p->apValue : &p->apValue[p->nCol]);
3104
3105 /* If this is an UPDATE or DELETE, read the old.* record. */
3106 if( p->op!=SQLITE_INSERT && (p->bPatchset==0 || p->op==SQLITE_DELETE) ){
3107 u8 *abPK = p->bPatchset ? p->abPK : 0;
3108 p->rc = sessionReadRecord(&p->in, p->nCol, abPK, apOld, 0);
3109 if( p->rc!=SQLITE_OK ) return p->rc;
3110 }
3111
3112 /* If this is an INSERT or UPDATE, read the new.* record. */
3113 if( p->op!=SQLITE_DELETE ){
3114 p->rc = sessionReadRecord(&p->in, p->nCol, 0, apNew, pbEmpty);
3115 if( p->rc!=SQLITE_OK ) return p->rc;
3116 }
3117
3118 if( (p->bPatchset || p->bInvert) && p->op==SQLITE_UPDATE ){
3119 /* If this is an UPDATE that is part of a patchset, then all PK and
3120 ** modified fields are present in the new.* record. The old.* record
3121 ** is currently completely empty. This block shifts the PK fields from
3122 ** new.* to old.*, to accommodate the code that reads these arrays. */
3123 for(i=0; i<p->nCol; i++){
3124 assert( p->bPatchset==0 || p->apValue[i]==0 );
3125 if( p->abPK[i] ){
3126 assert( p->apValue[i]==0 );
3127 p->apValue[i] = p->apValue[i+p->nCol];
3128 if( p->apValue[i]==0 ) return (p->rc = SQLITE_CORRUPT_BKPT);
3129 p->apValue[i+p->nCol] = 0;
3130 }
3131 }
3132 }else if( p->bInvert ){
3133 if( p->op==SQLITE_INSERT ) p->op = SQLITE_DELETE;
3134 else if( p->op==SQLITE_DELETE ) p->op = SQLITE_INSERT;
3135 }
3136 }
3137
3138 return SQLITE_ROW;
3139 }
3140
3141 /*
3142 ** Advance the changeset iterator to the next change.
3143 **
3144 ** If both paRec and pnRec are NULL, then this function works like the public
3145 ** API sqlite3changeset_next(). If SQLITE_ROW is returned, then the
3146 ** sqlite3changeset_new() and old() APIs may be used to query for values.
3147 **
3148 ** Otherwise, if paRec and pnRec are not NULL, then a pointer to the change
3149 ** record is written to *paRec before returning and the number of bytes in
3150 ** the record to *pnRec.
3151 **
3152 ** Either way, this function returns SQLITE_ROW if the iterator is
3153 ** successfully advanced to the next change in the changeset, an SQLite
3154 ** error code if an error occurs, or SQLITE_DONE if there are no further
3155 ** changes in the changeset.
3156 */
sessionChangesetNext(sqlite3_changeset_iter * p,u8 ** paRec,int * pnRec,int * pbNew)3157 static int sessionChangesetNext(
3158 sqlite3_changeset_iter *p, /* Changeset iterator */
3159 u8 **paRec, /* If non-NULL, store record pointer here */
3160 int *pnRec, /* If non-NULL, store size of record here */
3161 int *pbNew /* If non-NULL, true if new table */
3162 ){
3163 int bEmpty;
3164 int rc;
3165 do {
3166 bEmpty = 0;
3167 rc = sessionChangesetNextOne(p, paRec, pnRec, pbNew, &bEmpty);
3168 }while( rc==SQLITE_ROW && p->bSkipEmpty && bEmpty);
3169 return rc;
3170 }
3171
3172 /*
3173 ** Advance an iterator created by sqlite3changeset_start() to the next
3174 ** change in the changeset. This function may return SQLITE_ROW, SQLITE_DONE
3175 ** or SQLITE_CORRUPT.
3176 **
3177 ** This function may not be called on iterators passed to a conflict handler
3178 ** callback by changeset_apply().
3179 */
sqlite3changeset_next(sqlite3_changeset_iter * p)3180 int sqlite3changeset_next(sqlite3_changeset_iter *p){
3181 return sessionChangesetNext(p, 0, 0, 0);
3182 }
3183
3184 /*
3185 ** The following function extracts information on the current change
3186 ** from a changeset iterator. It may only be called after changeset_next()
3187 ** has returned SQLITE_ROW.
3188 */
sqlite3changeset_op(sqlite3_changeset_iter * pIter,const char ** pzTab,int * pnCol,int * pOp,int * pbIndirect)3189 int sqlite3changeset_op(
3190 sqlite3_changeset_iter *pIter, /* Iterator handle */
3191 const char **pzTab, /* OUT: Pointer to table name */
3192 int *pnCol, /* OUT: Number of columns in table */
3193 int *pOp, /* OUT: SQLITE_INSERT, DELETE or UPDATE */
3194 int *pbIndirect /* OUT: True if change is indirect */
3195 ){
3196 *pOp = pIter->op;
3197 *pnCol = pIter->nCol;
3198 *pzTab = pIter->zTab;
3199 if( pbIndirect ) *pbIndirect = pIter->bIndirect;
3200 return SQLITE_OK;
3201 }
3202
3203 /*
3204 ** Return information regarding the PRIMARY KEY and number of columns in
3205 ** the database table affected by the change that pIter currently points
3206 ** to. This function may only be called after changeset_next() returns
3207 ** SQLITE_ROW.
3208 */
sqlite3changeset_pk(sqlite3_changeset_iter * pIter,unsigned char ** pabPK,int * pnCol)3209 int sqlite3changeset_pk(
3210 sqlite3_changeset_iter *pIter, /* Iterator object */
3211 unsigned char **pabPK, /* OUT: Array of boolean - true for PK cols */
3212 int *pnCol /* OUT: Number of entries in output array */
3213 ){
3214 *pabPK = pIter->abPK;
3215 if( pnCol ) *pnCol = pIter->nCol;
3216 return SQLITE_OK;
3217 }
3218
3219 /*
3220 ** This function may only be called while the iterator is pointing to an
3221 ** SQLITE_UPDATE or SQLITE_DELETE change (see sqlite3changeset_op()).
3222 ** Otherwise, SQLITE_MISUSE is returned.
3223 **
3224 ** It sets *ppValue to point to an sqlite3_value structure containing the
3225 ** iVal'th value in the old.* record. Or, if that particular value is not
3226 ** included in the record (because the change is an UPDATE and the field
3227 ** was not modified and is not a PK column), set *ppValue to NULL.
3228 **
3229 ** If value iVal is out-of-range, SQLITE_RANGE is returned and *ppValue is
3230 ** not modified. Otherwise, SQLITE_OK.
3231 */
sqlite3changeset_old(sqlite3_changeset_iter * pIter,int iVal,sqlite3_value ** ppValue)3232 int sqlite3changeset_old(
3233 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3234 int iVal, /* Index of old.* value to retrieve */
3235 sqlite3_value **ppValue /* OUT: Old value (or NULL pointer) */
3236 ){
3237 if( pIter->op!=SQLITE_UPDATE && pIter->op!=SQLITE_DELETE ){
3238 return SQLITE_MISUSE;
3239 }
3240 if( iVal<0 || iVal>=pIter->nCol ){
3241 return SQLITE_RANGE;
3242 }
3243 *ppValue = pIter->apValue[iVal];
3244 return SQLITE_OK;
3245 }
3246
3247 /*
3248 ** This function may only be called while the iterator is pointing to an
3249 ** SQLITE_UPDATE or SQLITE_INSERT change (see sqlite3changeset_op()).
3250 ** Otherwise, SQLITE_MISUSE is returned.
3251 **
3252 ** It sets *ppValue to point to an sqlite3_value structure containing the
3253 ** iVal'th value in the new.* record. Or, if that particular value is not
3254 ** included in the record (because the change is an UPDATE and the field
3255 ** was not modified), set *ppValue to NULL.
3256 **
3257 ** If value iVal is out-of-range, SQLITE_RANGE is returned and *ppValue is
3258 ** not modified. Otherwise, SQLITE_OK.
3259 */
sqlite3changeset_new(sqlite3_changeset_iter * pIter,int iVal,sqlite3_value ** ppValue)3260 int sqlite3changeset_new(
3261 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3262 int iVal, /* Index of new.* value to retrieve */
3263 sqlite3_value **ppValue /* OUT: New value (or NULL pointer) */
3264 ){
3265 if( pIter->op!=SQLITE_UPDATE && pIter->op!=SQLITE_INSERT ){
3266 return SQLITE_MISUSE;
3267 }
3268 if( iVal<0 || iVal>=pIter->nCol ){
3269 return SQLITE_RANGE;
3270 }
3271 *ppValue = pIter->apValue[pIter->nCol+iVal];
3272 return SQLITE_OK;
3273 }
3274
3275 /*
3276 ** The following two macros are used internally. They are similar to the
3277 ** sqlite3changeset_new() and sqlite3changeset_old() functions, except that
3278 ** they omit all error checking and return a pointer to the requested value.
3279 */
3280 #define sessionChangesetNew(pIter, iVal) (pIter)->apValue[(pIter)->nCol+(iVal)]
3281 #define sessionChangesetOld(pIter, iVal) (pIter)->apValue[(iVal)]
3282
3283 /*
3284 ** This function may only be called with a changeset iterator that has been
3285 ** passed to an SQLITE_CHANGESET_DATA or SQLITE_CHANGESET_CONFLICT
3286 ** conflict-handler function. Otherwise, SQLITE_MISUSE is returned.
3287 **
3288 ** If successful, *ppValue is set to point to an sqlite3_value structure
3289 ** containing the iVal'th value of the conflicting record.
3290 **
3291 ** If value iVal is out-of-range or some other error occurs, an SQLite error
3292 ** code is returned. Otherwise, SQLITE_OK.
3293 */
sqlite3changeset_conflict(sqlite3_changeset_iter * pIter,int iVal,sqlite3_value ** ppValue)3294 int sqlite3changeset_conflict(
3295 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3296 int iVal, /* Index of conflict record value to fetch */
3297 sqlite3_value **ppValue /* OUT: Value from conflicting row */
3298 ){
3299 if( !pIter->pConflict ){
3300 return SQLITE_MISUSE;
3301 }
3302 if( iVal<0 || iVal>=pIter->nCol ){
3303 return SQLITE_RANGE;
3304 }
3305 *ppValue = sqlite3_column_value(pIter->pConflict, iVal);
3306 return SQLITE_OK;
3307 }
3308
3309 /*
3310 ** This function may only be called with an iterator passed to an
3311 ** SQLITE_CHANGESET_FOREIGN_KEY conflict handler callback. In this case
3312 ** it sets the output variable to the total number of known foreign key
3313 ** violations in the destination database and returns SQLITE_OK.
3314 **
3315 ** In all other cases this function returns SQLITE_MISUSE.
3316 */
sqlite3changeset_fk_conflicts(sqlite3_changeset_iter * pIter,int * pnOut)3317 int sqlite3changeset_fk_conflicts(
3318 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3319 int *pnOut /* OUT: Number of FK violations */
3320 ){
3321 if( pIter->pConflict || pIter->apValue ){
3322 return SQLITE_MISUSE;
3323 }
3324 *pnOut = pIter->nCol;
3325 return SQLITE_OK;
3326 }
3327
3328
3329 /*
3330 ** Finalize an iterator allocated with sqlite3changeset_start().
3331 **
3332 ** This function may not be called on iterators passed to a conflict handler
3333 ** callback by changeset_apply().
3334 */
sqlite3changeset_finalize(sqlite3_changeset_iter * p)3335 int sqlite3changeset_finalize(sqlite3_changeset_iter *p){
3336 int rc = SQLITE_OK;
3337 if( p ){
3338 int i; /* Used to iterate through p->apValue[] */
3339 rc = p->rc;
3340 if( p->apValue ){
3341 for(i=0; i<p->nCol*2; i++) sqlite3ValueFree(p->apValue[i]);
3342 }
3343 sqlite3_free(p->tblhdr.aBuf);
3344 sqlite3_free(p->in.buf.aBuf);
3345 sqlite3_free(p);
3346 }
3347 return rc;
3348 }
3349
sessionChangesetInvert(SessionInput * pInput,int (* xOutput)(void * pOut,const void * pData,int nData),void * pOut,int * pnInverted,void ** ppInverted)3350 static int sessionChangesetInvert(
3351 SessionInput *pInput, /* Input changeset */
3352 int (*xOutput)(void *pOut, const void *pData, int nData),
3353 void *pOut,
3354 int *pnInverted, /* OUT: Number of bytes in output changeset */
3355 void **ppInverted /* OUT: Inverse of pChangeset */
3356 ){
3357 int rc = SQLITE_OK; /* Return value */
3358 SessionBuffer sOut; /* Output buffer */
3359 int nCol = 0; /* Number of cols in current table */
3360 u8 *abPK = 0; /* PK array for current table */
3361 sqlite3_value **apVal = 0; /* Space for values for UPDATE inversion */
3362 SessionBuffer sPK = {0, 0, 0}; /* PK array for current table */
3363
3364 /* Initialize the output buffer */
3365 memset(&sOut, 0, sizeof(SessionBuffer));
3366
3367 /* Zero the output variables in case an error occurs. */
3368 if( ppInverted ){
3369 *ppInverted = 0;
3370 *pnInverted = 0;
3371 }
3372
3373 while( 1 ){
3374 u8 eType;
3375
3376 /* Test for EOF. */
3377 if( (rc = sessionInputBuffer(pInput, 2)) ) goto finished_invert;
3378 if( pInput->iNext>=pInput->nData ) break;
3379 eType = pInput->aData[pInput->iNext];
3380
3381 switch( eType ){
3382 case 'T': {
3383 /* A 'table' record consists of:
3384 **
3385 ** * A constant 'T' character,
3386 ** * Number of columns in said table (a varint),
3387 ** * An array of nCol bytes (sPK),
3388 ** * A nul-terminated table name.
3389 */
3390 int nByte;
3391 int nVar;
3392 pInput->iNext++;
3393 if( (rc = sessionChangesetBufferTblhdr(pInput, &nByte)) ){
3394 goto finished_invert;
3395 }
3396 nVar = sessionVarintGet(&pInput->aData[pInput->iNext], &nCol);
3397 sPK.nBuf = 0;
3398 sessionAppendBlob(&sPK, &pInput->aData[pInput->iNext+nVar], nCol, &rc);
3399 sessionAppendByte(&sOut, eType, &rc);
3400 sessionAppendBlob(&sOut, &pInput->aData[pInput->iNext], nByte, &rc);
3401 if( rc ) goto finished_invert;
3402
3403 pInput->iNext += nByte;
3404 sqlite3_free(apVal);
3405 apVal = 0;
3406 abPK = sPK.aBuf;
3407 break;
3408 }
3409
3410 case SQLITE_INSERT:
3411 case SQLITE_DELETE: {
3412 int nByte;
3413 int bIndirect = pInput->aData[pInput->iNext+1];
3414 int eType2 = (eType==SQLITE_DELETE ? SQLITE_INSERT : SQLITE_DELETE);
3415 pInput->iNext += 2;
3416 assert( rc==SQLITE_OK );
3417 rc = sessionChangesetBufferRecord(pInput, nCol, &nByte);
3418 sessionAppendByte(&sOut, eType2, &rc);
3419 sessionAppendByte(&sOut, bIndirect, &rc);
3420 sessionAppendBlob(&sOut, &pInput->aData[pInput->iNext], nByte, &rc);
3421 pInput->iNext += nByte;
3422 if( rc ) goto finished_invert;
3423 break;
3424 }
3425
3426 case SQLITE_UPDATE: {
3427 int iCol;
3428
3429 if( 0==apVal ){
3430 apVal = (sqlite3_value **)sqlite3_malloc64(sizeof(apVal[0])*nCol*2);
3431 if( 0==apVal ){
3432 rc = SQLITE_NOMEM;
3433 goto finished_invert;
3434 }
3435 memset(apVal, 0, sizeof(apVal[0])*nCol*2);
3436 }
3437
3438 /* Write the header for the new UPDATE change. Same as the original. */
3439 sessionAppendByte(&sOut, eType, &rc);
3440 sessionAppendByte(&sOut, pInput->aData[pInput->iNext+1], &rc);
3441
3442 /* Read the old.* and new.* records for the update change. */
3443 pInput->iNext += 2;
3444 rc = sessionReadRecord(pInput, nCol, 0, &apVal[0], 0);
3445 if( rc==SQLITE_OK ){
3446 rc = sessionReadRecord(pInput, nCol, 0, &apVal[nCol], 0);
3447 }
3448
3449 /* Write the new old.* record. Consists of the PK columns from the
3450 ** original old.* record, and the other values from the original
3451 ** new.* record. */
3452 for(iCol=0; iCol<nCol; iCol++){
3453 sqlite3_value *pVal = apVal[iCol + (abPK[iCol] ? 0 : nCol)];
3454 sessionAppendValue(&sOut, pVal, &rc);
3455 }
3456
3457 /* Write the new new.* record. Consists of a copy of all values
3458 ** from the original old.* record, except for the PK columns, which
3459 ** are set to "undefined". */
3460 for(iCol=0; iCol<nCol; iCol++){
3461 sqlite3_value *pVal = (abPK[iCol] ? 0 : apVal[iCol]);
3462 sessionAppendValue(&sOut, pVal, &rc);
3463 }
3464
3465 for(iCol=0; iCol<nCol*2; iCol++){
3466 sqlite3ValueFree(apVal[iCol]);
3467 }
3468 memset(apVal, 0, sizeof(apVal[0])*nCol*2);
3469 if( rc!=SQLITE_OK ){
3470 goto finished_invert;
3471 }
3472
3473 break;
3474 }
3475
3476 default:
3477 rc = SQLITE_CORRUPT_BKPT;
3478 goto finished_invert;
3479 }
3480
3481 assert( rc==SQLITE_OK );
3482 if( xOutput && sOut.nBuf>=sessions_strm_chunk_size ){
3483 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf);
3484 sOut.nBuf = 0;
3485 if( rc!=SQLITE_OK ) goto finished_invert;
3486 }
3487 }
3488
3489 assert( rc==SQLITE_OK );
3490 if( pnInverted ){
3491 *pnInverted = sOut.nBuf;
3492 *ppInverted = sOut.aBuf;
3493 sOut.aBuf = 0;
3494 }else if( sOut.nBuf>0 ){
3495 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf);
3496 }
3497
3498 finished_invert:
3499 sqlite3_free(sOut.aBuf);
3500 sqlite3_free(apVal);
3501 sqlite3_free(sPK.aBuf);
3502 return rc;
3503 }
3504
3505
3506 /*
3507 ** Invert a changeset object.
3508 */
sqlite3changeset_invert(int nChangeset,const void * pChangeset,int * pnInverted,void ** ppInverted)3509 int sqlite3changeset_invert(
3510 int nChangeset, /* Number of bytes in input */
3511 const void *pChangeset, /* Input changeset */
3512 int *pnInverted, /* OUT: Number of bytes in output changeset */
3513 void **ppInverted /* OUT: Inverse of pChangeset */
3514 ){
3515 SessionInput sInput;
3516
3517 /* Set up the input stream */
3518 memset(&sInput, 0, sizeof(SessionInput));
3519 sInput.nData = nChangeset;
3520 sInput.aData = (u8*)pChangeset;
3521
3522 return sessionChangesetInvert(&sInput, 0, 0, pnInverted, ppInverted);
3523 }
3524
3525 /*
3526 ** Streaming version of sqlite3changeset_invert().
3527 */
sqlite3changeset_invert_strm(int (* xInput)(void * pIn,void * pData,int * pnData),void * pIn,int (* xOutput)(void * pOut,const void * pData,int nData),void * pOut)3528 int sqlite3changeset_invert_strm(
3529 int (*xInput)(void *pIn, void *pData, int *pnData),
3530 void *pIn,
3531 int (*xOutput)(void *pOut, const void *pData, int nData),
3532 void *pOut
3533 ){
3534 SessionInput sInput;
3535 int rc;
3536
3537 /* Set up the input stream */
3538 memset(&sInput, 0, sizeof(SessionInput));
3539 sInput.xInput = xInput;
3540 sInput.pIn = pIn;
3541
3542 rc = sessionChangesetInvert(&sInput, xOutput, pOut, 0, 0);
3543 sqlite3_free(sInput.buf.aBuf);
3544 return rc;
3545 }
3546
3547
3548 typedef struct SessionUpdate SessionUpdate;
3549 struct SessionUpdate {
3550 sqlite3_stmt *pStmt;
3551 u32 *aMask;
3552 SessionUpdate *pNext;
3553 };
3554
3555 typedef struct SessionApplyCtx SessionApplyCtx;
3556 struct SessionApplyCtx {
3557 sqlite3 *db;
3558 sqlite3_stmt *pDelete; /* DELETE statement */
3559 sqlite3_stmt *pInsert; /* INSERT statement */
3560 sqlite3_stmt *pSelect; /* SELECT statement */
3561 int nCol; /* Size of azCol[] and abPK[] arrays */
3562 const char **azCol; /* Array of column names */
3563 u8 *abPK; /* Boolean array - true if column is in PK */
3564 u32 *aUpdateMask; /* Used by sessionUpdateFind */
3565 SessionUpdate *pUp;
3566 int bStat1; /* True if table is sqlite_stat1 */
3567 int bDeferConstraints; /* True to defer constraints */
3568 int bInvertConstraints; /* Invert when iterating constraints buffer */
3569 SessionBuffer constraints; /* Deferred constraints are stored here */
3570 SessionBuffer rebase; /* Rebase information (if any) here */
3571 u8 bRebaseStarted; /* If table header is already in rebase */
3572 u8 bRebase; /* True to collect rebase information */
3573 };
3574
3575 /* Number of prepared UPDATE statements to cache. */
3576 #define SESSION_UPDATE_CACHE_SZ 12
3577
3578 /*
3579 ** Find a prepared UPDATE statement suitable for the UPDATE step currently
3580 ** being visited by the iterator. The UPDATE is of the form:
3581 **
3582 ** UPDATE tbl SET col = ?, col2 = ? WHERE pk1 IS ? AND pk2 IS ?
3583 */
sessionUpdateFind(sqlite3_changeset_iter * pIter,SessionApplyCtx * p,int bPatchset,sqlite3_stmt ** ppStmt)3584 static int sessionUpdateFind(
3585 sqlite3_changeset_iter *pIter,
3586 SessionApplyCtx *p,
3587 int bPatchset,
3588 sqlite3_stmt **ppStmt
3589 ){
3590 int rc = SQLITE_OK;
3591 SessionUpdate *pUp = 0;
3592 int nCol = pIter->nCol;
3593 int nU32 = (pIter->nCol+33)/32;
3594 int ii;
3595
3596 if( p->aUpdateMask==0 ){
3597 p->aUpdateMask = sqlite3_malloc(nU32*sizeof(u32));
3598 if( p->aUpdateMask==0 ){
3599 rc = SQLITE_NOMEM;
3600 }
3601 }
3602
3603 if( rc==SQLITE_OK ){
3604 memset(p->aUpdateMask, 0, nU32*sizeof(u32));
3605 rc = SQLITE_CORRUPT;
3606 for(ii=0; ii<pIter->nCol; ii++){
3607 if( sessionChangesetNew(pIter, ii) ){
3608 p->aUpdateMask[ii/32] |= (1<<(ii%32));
3609 rc = SQLITE_OK;
3610 }
3611 }
3612 }
3613
3614 if( rc==SQLITE_OK ){
3615 if( bPatchset ) p->aUpdateMask[nCol/32] |= (1<<(nCol%32));
3616
3617 if( p->pUp ){
3618 int nUp = 0;
3619 SessionUpdate **pp = &p->pUp;
3620 while( 1 ){
3621 nUp++;
3622 if( 0==memcmp(p->aUpdateMask, (*pp)->aMask, nU32*sizeof(u32)) ){
3623 pUp = *pp;
3624 *pp = pUp->pNext;
3625 pUp->pNext = p->pUp;
3626 p->pUp = pUp;
3627 break;
3628 }
3629
3630 if( (*pp)->pNext ){
3631 pp = &(*pp)->pNext;
3632 }else{
3633 if( nUp>=SESSION_UPDATE_CACHE_SZ ){
3634 sqlite3_finalize((*pp)->pStmt);
3635 sqlite3_free(*pp);
3636 *pp = 0;
3637 }
3638 break;
3639 }
3640 }
3641 }
3642
3643 if( pUp==0 ){
3644 int nByte = sizeof(SessionUpdate) * nU32*sizeof(u32);
3645 int bStat1 = (sqlite3_stricmp(pIter->zTab, "sqlite_stat1")==0);
3646 pUp = (SessionUpdate*)sqlite3_malloc(nByte);
3647 if( pUp==0 ){
3648 rc = SQLITE_NOMEM;
3649 }else{
3650 const char *zSep = "";
3651 SessionBuffer buf;
3652
3653 memset(&buf, 0, sizeof(buf));
3654 pUp->aMask = (u32*)&pUp[1];
3655 memcpy(pUp->aMask, p->aUpdateMask, nU32*sizeof(u32));
3656
3657 sessionAppendStr(&buf, "UPDATE main.", &rc);
3658 sessionAppendIdent(&buf, pIter->zTab, &rc);
3659 sessionAppendStr(&buf, " SET ", &rc);
3660
3661 /* Create the assignments part of the UPDATE */
3662 for(ii=0; ii<pIter->nCol; ii++){
3663 if( p->abPK[ii]==0 && sessionChangesetNew(pIter, ii) ){
3664 sessionAppendStr(&buf, zSep, &rc);
3665 sessionAppendIdent(&buf, p->azCol[ii], &rc);
3666 sessionAppendStr(&buf, " = ?", &rc);
3667 sessionAppendInteger(&buf, ii*2+1, &rc);
3668 zSep = ", ";
3669 }
3670 }
3671
3672 /* Create the WHERE clause part of the UPDATE */
3673 zSep = "";
3674 sessionAppendStr(&buf, " WHERE ", &rc);
3675 for(ii=0; ii<pIter->nCol; ii++){
3676 if( p->abPK[ii] || (bPatchset==0 && sessionChangesetOld(pIter, ii)) ){
3677 sessionAppendStr(&buf, zSep, &rc);
3678 if( bStat1 && ii==1 ){
3679 assert( sqlite3_stricmp(p->azCol[ii], "idx")==0 );
3680 sessionAppendStr(&buf,
3681 "idx IS CASE "
3682 "WHEN length(?4)=0 AND typeof(?4)='blob' THEN NULL "
3683 "ELSE ?4 END ", &rc
3684 );
3685 }else{
3686 sessionAppendIdent(&buf, p->azCol[ii], &rc);
3687 sessionAppendStr(&buf, " IS ?", &rc);
3688 sessionAppendInteger(&buf, ii*2+2, &rc);
3689 }
3690 zSep = " AND ";
3691 }
3692 }
3693
3694 if( rc==SQLITE_OK ){
3695 char *zSql = (char*)buf.aBuf;
3696 rc = sqlite3_prepare_v2(p->db, zSql, buf.nBuf, &pUp->pStmt, 0);
3697 }
3698
3699 if( rc!=SQLITE_OK ){
3700 sqlite3_free(pUp);
3701 pUp = 0;
3702 }else{
3703 pUp->pNext = p->pUp;
3704 p->pUp = pUp;
3705 }
3706 sqlite3_free(buf.aBuf);
3707 }
3708 }
3709 }
3710
3711 assert( (rc==SQLITE_OK)==(pUp!=0) );
3712 if( pUp ){
3713 *ppStmt = pUp->pStmt;
3714 }else{
3715 *ppStmt = 0;
3716 }
3717 return rc;
3718 }
3719
3720 /*
3721 ** Free all cached UPDATE statements.
3722 */
sessionUpdateFree(SessionApplyCtx * p)3723 static void sessionUpdateFree(SessionApplyCtx *p){
3724 SessionUpdate *pUp;
3725 SessionUpdate *pNext;
3726 for(pUp=p->pUp; pUp; pUp=pNext){
3727 pNext = pUp->pNext;
3728 sqlite3_finalize(pUp->pStmt);
3729 sqlite3_free(pUp);
3730 }
3731 p->pUp = 0;
3732 sqlite3_free(p->aUpdateMask);
3733 p->aUpdateMask = 0;
3734 }
3735
3736 /*
3737 ** Formulate a statement to DELETE a row from database db. Assuming a table
3738 ** structure like this:
3739 **
3740 ** CREATE TABLE x(a, b, c, d, PRIMARY KEY(a, c));
3741 **
3742 ** The DELETE statement looks like this:
3743 **
3744 ** DELETE FROM x WHERE a = :1 AND c = :3 AND (:5 OR b IS :2 AND d IS :4)
3745 **
3746 ** Variable :5 (nCol+1) is a boolean. It should be set to 0 if we require
3747 ** matching b and d values, or 1 otherwise. The second case comes up if the
3748 ** conflict handler is invoked with NOTFOUND and returns CHANGESET_REPLACE.
3749 **
3750 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pDelete is left
3751 ** pointing to the prepared version of the SQL statement.
3752 */
sessionDeleteRow(sqlite3 * db,const char * zTab,SessionApplyCtx * p)3753 static int sessionDeleteRow(
3754 sqlite3 *db, /* Database handle */
3755 const char *zTab, /* Table name */
3756 SessionApplyCtx *p /* Session changeset-apply context */
3757 ){
3758 int i;
3759 const char *zSep = "";
3760 int rc = SQLITE_OK;
3761 SessionBuffer buf = {0, 0, 0};
3762 int nPk = 0;
3763
3764 sessionAppendStr(&buf, "DELETE FROM main.", &rc);
3765 sessionAppendIdent(&buf, zTab, &rc);
3766 sessionAppendStr(&buf, " WHERE ", &rc);
3767
3768 for(i=0; i<p->nCol; i++){
3769 if( p->abPK[i] ){
3770 nPk++;
3771 sessionAppendStr(&buf, zSep, &rc);
3772 sessionAppendIdent(&buf, p->azCol[i], &rc);
3773 sessionAppendStr(&buf, " = ?", &rc);
3774 sessionAppendInteger(&buf, i+1, &rc);
3775 zSep = " AND ";
3776 }
3777 }
3778
3779 if( nPk<p->nCol ){
3780 sessionAppendStr(&buf, " AND (?", &rc);
3781 sessionAppendInteger(&buf, p->nCol+1, &rc);
3782 sessionAppendStr(&buf, " OR ", &rc);
3783
3784 zSep = "";
3785 for(i=0; i<p->nCol; i++){
3786 if( !p->abPK[i] ){
3787 sessionAppendStr(&buf, zSep, &rc);
3788 sessionAppendIdent(&buf, p->azCol[i], &rc);
3789 sessionAppendStr(&buf, " IS ?", &rc);
3790 sessionAppendInteger(&buf, i+1, &rc);
3791 zSep = "AND ";
3792 }
3793 }
3794 sessionAppendStr(&buf, ")", &rc);
3795 }
3796
3797 if( rc==SQLITE_OK ){
3798 rc = sqlite3_prepare_v2(db, (char *)buf.aBuf, buf.nBuf, &p->pDelete, 0);
3799 }
3800 sqlite3_free(buf.aBuf);
3801
3802 return rc;
3803 }
3804
3805 /*
3806 ** Formulate and prepare an SQL statement to query table zTab by primary
3807 ** key. Assuming the following table structure:
3808 **
3809 ** CREATE TABLE x(a, b, c, d, PRIMARY KEY(a, c));
3810 **
3811 ** The SELECT statement looks like this:
3812 **
3813 ** SELECT * FROM x WHERE a = ?1 AND c = ?3
3814 **
3815 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pSelect is left
3816 ** pointing to the prepared version of the SQL statement.
3817 */
sessionSelectRow(sqlite3 * db,const char * zTab,SessionApplyCtx * p)3818 static int sessionSelectRow(
3819 sqlite3 *db, /* Database handle */
3820 const char *zTab, /* Table name */
3821 SessionApplyCtx *p /* Session changeset-apply context */
3822 ){
3823 return sessionSelectStmt(
3824 db, "main", zTab, p->nCol, p->azCol, p->abPK, &p->pSelect);
3825 }
3826
3827 /*
3828 ** Formulate and prepare an INSERT statement to add a record to table zTab.
3829 ** For example:
3830 **
3831 ** INSERT INTO main."zTab" VALUES(?1, ?2, ?3 ...);
3832 **
3833 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pInsert is left
3834 ** pointing to the prepared version of the SQL statement.
3835 */
sessionInsertRow(sqlite3 * db,const char * zTab,SessionApplyCtx * p)3836 static int sessionInsertRow(
3837 sqlite3 *db, /* Database handle */
3838 const char *zTab, /* Table name */
3839 SessionApplyCtx *p /* Session changeset-apply context */
3840 ){
3841 int rc = SQLITE_OK;
3842 int i;
3843 SessionBuffer buf = {0, 0, 0};
3844
3845 sessionAppendStr(&buf, "INSERT INTO main.", &rc);
3846 sessionAppendIdent(&buf, zTab, &rc);
3847 sessionAppendStr(&buf, "(", &rc);
3848 for(i=0; i<p->nCol; i++){
3849 if( i!=0 ) sessionAppendStr(&buf, ", ", &rc);
3850 sessionAppendIdent(&buf, p->azCol[i], &rc);
3851 }
3852
3853 sessionAppendStr(&buf, ") VALUES(?", &rc);
3854 for(i=1; i<p->nCol; i++){
3855 sessionAppendStr(&buf, ", ?", &rc);
3856 }
3857 sessionAppendStr(&buf, ")", &rc);
3858
3859 if( rc==SQLITE_OK ){
3860 rc = sqlite3_prepare_v2(db, (char *)buf.aBuf, buf.nBuf, &p->pInsert, 0);
3861 }
3862 sqlite3_free(buf.aBuf);
3863 return rc;
3864 }
3865
sessionPrepare(sqlite3 * db,sqlite3_stmt ** pp,const char * zSql)3866 static int sessionPrepare(sqlite3 *db, sqlite3_stmt **pp, const char *zSql){
3867 return sqlite3_prepare_v2(db, zSql, -1, pp, 0);
3868 }
3869
3870 /*
3871 ** Prepare statements for applying changes to the sqlite_stat1 table.
3872 ** These are similar to those created by sessionSelectRow(),
3873 ** sessionInsertRow(), sessionUpdateRow() and sessionDeleteRow() for
3874 ** other tables.
3875 */
sessionStat1Sql(sqlite3 * db,SessionApplyCtx * p)3876 static int sessionStat1Sql(sqlite3 *db, SessionApplyCtx *p){
3877 int rc = sessionSelectRow(db, "sqlite_stat1", p);
3878 if( rc==SQLITE_OK ){
3879 rc = sessionPrepare(db, &p->pInsert,
3880 "INSERT INTO main.sqlite_stat1 VALUES(?1, "
3881 "CASE WHEN length(?2)=0 AND typeof(?2)='blob' THEN NULL ELSE ?2 END, "
3882 "?3)"
3883 );
3884 }
3885 if( rc==SQLITE_OK ){
3886 rc = sessionPrepare(db, &p->pDelete,
3887 "DELETE FROM main.sqlite_stat1 WHERE tbl=?1 AND idx IS "
3888 "CASE WHEN length(?2)=0 AND typeof(?2)='blob' THEN NULL ELSE ?2 END "
3889 "AND (?4 OR stat IS ?3)"
3890 );
3891 }
3892 return rc;
3893 }
3894
3895 /*
3896 ** A wrapper around sqlite3_bind_value() that detects an extra problem.
3897 ** See comments in the body of this function for details.
3898 */
sessionBindValue(sqlite3_stmt * pStmt,int i,sqlite3_value * pVal)3899 static int sessionBindValue(
3900 sqlite3_stmt *pStmt, /* Statement to bind value to */
3901 int i, /* Parameter number to bind to */
3902 sqlite3_value *pVal /* Value to bind */
3903 ){
3904 int eType = sqlite3_value_type(pVal);
3905 /* COVERAGE: The (pVal->z==0) branch is never true using current versions
3906 ** of SQLite. If a malloc fails in an sqlite3_value_xxx() function, either
3907 ** the (pVal->z) variable remains as it was or the type of the value is
3908 ** set to SQLITE_NULL. */
3909 if( (eType==SQLITE_TEXT || eType==SQLITE_BLOB) && pVal->z==0 ){
3910 /* This condition occurs when an earlier OOM in a call to
3911 ** sqlite3_value_text() or sqlite3_value_blob() (perhaps from within
3912 ** a conflict-handler) has zeroed the pVal->z pointer. Return NOMEM. */
3913 return SQLITE_NOMEM;
3914 }
3915 return sqlite3_bind_value(pStmt, i, pVal);
3916 }
3917
3918 /*
3919 ** Iterator pIter must point to an SQLITE_INSERT entry. This function
3920 ** transfers new.* values from the current iterator entry to statement
3921 ** pStmt. The table being inserted into has nCol columns.
3922 **
3923 ** New.* value $i from the iterator is bound to variable ($i+1) of
3924 ** statement pStmt. If parameter abPK is NULL, all values from 0 to (nCol-1)
3925 ** are transfered to the statement. Otherwise, if abPK is not NULL, it points
3926 ** to an array nCol elements in size. In this case only those values for
3927 ** which abPK[$i] is true are read from the iterator and bound to the
3928 ** statement.
3929 **
3930 ** An SQLite error code is returned if an error occurs. Otherwise, SQLITE_OK.
3931 */
sessionBindRow(sqlite3_changeset_iter * pIter,int (* xValue)(sqlite3_changeset_iter *,int,sqlite3_value **),int nCol,u8 * abPK,sqlite3_stmt * pStmt)3932 static int sessionBindRow(
3933 sqlite3_changeset_iter *pIter, /* Iterator to read values from */
3934 int(*xValue)(sqlite3_changeset_iter *, int, sqlite3_value **),
3935 int nCol, /* Number of columns */
3936 u8 *abPK, /* If not NULL, bind only if true */
3937 sqlite3_stmt *pStmt /* Bind values to this statement */
3938 ){
3939 int i;
3940 int rc = SQLITE_OK;
3941
3942 /* Neither sqlite3changeset_old or sqlite3changeset_new can fail if the
3943 ** argument iterator points to a suitable entry. Make sure that xValue
3944 ** is one of these to guarantee that it is safe to ignore the return
3945 ** in the code below. */
3946 assert( xValue==sqlite3changeset_old || xValue==sqlite3changeset_new );
3947
3948 for(i=0; rc==SQLITE_OK && i<nCol; i++){
3949 if( !abPK || abPK[i] ){
3950 sqlite3_value *pVal;
3951 (void)xValue(pIter, i, &pVal);
3952 if( pVal==0 ){
3953 /* The value in the changeset was "undefined". This indicates a
3954 ** corrupt changeset blob. */
3955 rc = SQLITE_CORRUPT_BKPT;
3956 }else{
3957 rc = sessionBindValue(pStmt, i+1, pVal);
3958 }
3959 }
3960 }
3961 return rc;
3962 }
3963
3964 /*
3965 ** SQL statement pSelect is as generated by the sessionSelectRow() function.
3966 ** This function binds the primary key values from the change that changeset
3967 ** iterator pIter points to to the SELECT and attempts to seek to the table
3968 ** entry. If a row is found, the SELECT statement left pointing at the row
3969 ** and SQLITE_ROW is returned. Otherwise, if no row is found and no error
3970 ** has occured, the statement is reset and SQLITE_OK is returned. If an
3971 ** error occurs, the statement is reset and an SQLite error code is returned.
3972 **
3973 ** If this function returns SQLITE_ROW, the caller must eventually reset()
3974 ** statement pSelect. If any other value is returned, the statement does
3975 ** not require a reset().
3976 **
3977 ** If the iterator currently points to an INSERT record, bind values from the
3978 ** new.* record to the SELECT statement. Or, if it points to a DELETE or
3979 ** UPDATE, bind values from the old.* record.
3980 */
sessionSeekToRow(sqlite3 * db,sqlite3_changeset_iter * pIter,u8 * abPK,sqlite3_stmt * pSelect)3981 static int sessionSeekToRow(
3982 sqlite3 *db, /* Database handle */
3983 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3984 u8 *abPK, /* Primary key flags array */
3985 sqlite3_stmt *pSelect /* SELECT statement from sessionSelectRow() */
3986 ){
3987 int rc; /* Return code */
3988 int nCol; /* Number of columns in table */
3989 int op; /* Changset operation (SQLITE_UPDATE etc.) */
3990 const char *zDummy; /* Unused */
3991
3992 sqlite3changeset_op(pIter, &zDummy, &nCol, &op, 0);
3993 rc = sessionBindRow(pIter,
3994 op==SQLITE_INSERT ? sqlite3changeset_new : sqlite3changeset_old,
3995 nCol, abPK, pSelect
3996 );
3997
3998 if( rc==SQLITE_OK ){
3999 rc = sqlite3_step(pSelect);
4000 if( rc!=SQLITE_ROW ) rc = sqlite3_reset(pSelect);
4001 }
4002
4003 return rc;
4004 }
4005
4006 /*
4007 ** This function is called from within sqlite3changeset_apply_v2() when
4008 ** a conflict is encountered and resolved using conflict resolution
4009 ** mode eType (either SQLITE_CHANGESET_OMIT or SQLITE_CHANGESET_REPLACE)..
4010 ** It adds a conflict resolution record to the buffer in
4011 ** SessionApplyCtx.rebase, which will eventually be returned to the caller
4012 ** of apply_v2() as the "rebase" buffer.
4013 **
4014 ** Return SQLITE_OK if successful, or an SQLite error code otherwise.
4015 */
sessionRebaseAdd(SessionApplyCtx * p,int eType,sqlite3_changeset_iter * pIter)4016 static int sessionRebaseAdd(
4017 SessionApplyCtx *p, /* Apply context */
4018 int eType, /* Conflict resolution (OMIT or REPLACE) */
4019 sqlite3_changeset_iter *pIter /* Iterator pointing at current change */
4020 ){
4021 int rc = SQLITE_OK;
4022 if( p->bRebase ){
4023 int i;
4024 int eOp = pIter->op;
4025 if( p->bRebaseStarted==0 ){
4026 /* Append a table-header to the rebase buffer */
4027 const char *zTab = pIter->zTab;
4028 sessionAppendByte(&p->rebase, 'T', &rc);
4029 sessionAppendVarint(&p->rebase, p->nCol, &rc);
4030 sessionAppendBlob(&p->rebase, p->abPK, p->nCol, &rc);
4031 sessionAppendBlob(&p->rebase, (u8*)zTab, (int)strlen(zTab)+1, &rc);
4032 p->bRebaseStarted = 1;
4033 }
4034
4035 assert( eType==SQLITE_CHANGESET_REPLACE||eType==SQLITE_CHANGESET_OMIT );
4036 assert( eOp==SQLITE_DELETE || eOp==SQLITE_INSERT || eOp==SQLITE_UPDATE );
4037
4038 sessionAppendByte(&p->rebase,
4039 (eOp==SQLITE_DELETE ? SQLITE_DELETE : SQLITE_INSERT), &rc
4040 );
4041 sessionAppendByte(&p->rebase, (eType==SQLITE_CHANGESET_REPLACE), &rc);
4042 for(i=0; i<p->nCol; i++){
4043 sqlite3_value *pVal = 0;
4044 if( eOp==SQLITE_DELETE || (eOp==SQLITE_UPDATE && p->abPK[i]) ){
4045 sqlite3changeset_old(pIter, i, &pVal);
4046 }else{
4047 sqlite3changeset_new(pIter, i, &pVal);
4048 }
4049 sessionAppendValue(&p->rebase, pVal, &rc);
4050 }
4051 }
4052 return rc;
4053 }
4054
4055 /*
4056 ** Invoke the conflict handler for the change that the changeset iterator
4057 ** currently points to.
4058 **
4059 ** Argument eType must be either CHANGESET_DATA or CHANGESET_CONFLICT.
4060 ** If argument pbReplace is NULL, then the type of conflict handler invoked
4061 ** depends solely on eType, as follows:
4062 **
4063 ** eType value Value passed to xConflict
4064 ** -------------------------------------------------
4065 ** CHANGESET_DATA CHANGESET_NOTFOUND
4066 ** CHANGESET_CONFLICT CHANGESET_CONSTRAINT
4067 **
4068 ** Or, if pbReplace is not NULL, then an attempt is made to find an existing
4069 ** record with the same primary key as the record about to be deleted, updated
4070 ** or inserted. If such a record can be found, it is available to the conflict
4071 ** handler as the "conflicting" record. In this case the type of conflict
4072 ** handler invoked is as follows:
4073 **
4074 ** eType value PK Record found? Value passed to xConflict
4075 ** ----------------------------------------------------------------
4076 ** CHANGESET_DATA Yes CHANGESET_DATA
4077 ** CHANGESET_DATA No CHANGESET_NOTFOUND
4078 ** CHANGESET_CONFLICT Yes CHANGESET_CONFLICT
4079 ** CHANGESET_CONFLICT No CHANGESET_CONSTRAINT
4080 **
4081 ** If pbReplace is not NULL, and a record with a matching PK is found, and
4082 ** the conflict handler function returns SQLITE_CHANGESET_REPLACE, *pbReplace
4083 ** is set to non-zero before returning SQLITE_OK.
4084 **
4085 ** If the conflict handler returns SQLITE_CHANGESET_ABORT, SQLITE_ABORT is
4086 ** returned. Or, if the conflict handler returns an invalid value,
4087 ** SQLITE_MISUSE. If the conflict handler returns SQLITE_CHANGESET_OMIT,
4088 ** this function returns SQLITE_OK.
4089 */
sessionConflictHandler(int eType,SessionApplyCtx * p,sqlite3_changeset_iter * pIter,int (* xConflict)(void *,int,sqlite3_changeset_iter *),void * pCtx,int * pbReplace)4090 static int sessionConflictHandler(
4091 int eType, /* Either CHANGESET_DATA or CONFLICT */
4092 SessionApplyCtx *p, /* changeset_apply() context */
4093 sqlite3_changeset_iter *pIter, /* Changeset iterator */
4094 int(*xConflict)(void *, int, sqlite3_changeset_iter*),
4095 void *pCtx, /* First argument for conflict handler */
4096 int *pbReplace /* OUT: Set to true if PK row is found */
4097 ){
4098 int res = 0; /* Value returned by conflict handler */
4099 int rc;
4100 int nCol;
4101 int op;
4102 const char *zDummy;
4103
4104 sqlite3changeset_op(pIter, &zDummy, &nCol, &op, 0);
4105
4106 assert( eType==SQLITE_CHANGESET_CONFLICT || eType==SQLITE_CHANGESET_DATA );
4107 assert( SQLITE_CHANGESET_CONFLICT+1==SQLITE_CHANGESET_CONSTRAINT );
4108 assert( SQLITE_CHANGESET_DATA+1==SQLITE_CHANGESET_NOTFOUND );
4109
4110 /* Bind the new.* PRIMARY KEY values to the SELECT statement. */
4111 if( pbReplace ){
4112 rc = sessionSeekToRow(p->db, pIter, p->abPK, p->pSelect);
4113 }else{
4114 rc = SQLITE_OK;
4115 }
4116
4117 if( rc==SQLITE_ROW ){
4118 /* There exists another row with the new.* primary key. */
4119 pIter->pConflict = p->pSelect;
4120 res = xConflict(pCtx, eType, pIter);
4121 pIter->pConflict = 0;
4122 rc = sqlite3_reset(p->pSelect);
4123 }else if( rc==SQLITE_OK ){
4124 if( p->bDeferConstraints && eType==SQLITE_CHANGESET_CONFLICT ){
4125 /* Instead of invoking the conflict handler, append the change blob
4126 ** to the SessionApplyCtx.constraints buffer. */
4127 u8 *aBlob = &pIter->in.aData[pIter->in.iCurrent];
4128 int nBlob = pIter->in.iNext - pIter->in.iCurrent;
4129 sessionAppendBlob(&p->constraints, aBlob, nBlob, &rc);
4130 return SQLITE_OK;
4131 }else{
4132 /* No other row with the new.* primary key. */
4133 res = xConflict(pCtx, eType+1, pIter);
4134 if( res==SQLITE_CHANGESET_REPLACE ) rc = SQLITE_MISUSE;
4135 }
4136 }
4137
4138 if( rc==SQLITE_OK ){
4139 switch( res ){
4140 case SQLITE_CHANGESET_REPLACE:
4141 assert( pbReplace );
4142 *pbReplace = 1;
4143 break;
4144
4145 case SQLITE_CHANGESET_OMIT:
4146 break;
4147
4148 case SQLITE_CHANGESET_ABORT:
4149 rc = SQLITE_ABORT;
4150 break;
4151
4152 default:
4153 rc = SQLITE_MISUSE;
4154 break;
4155 }
4156 if( rc==SQLITE_OK ){
4157 rc = sessionRebaseAdd(p, res, pIter);
4158 }
4159 }
4160
4161 return rc;
4162 }
4163
4164 /*
4165 ** Attempt to apply the change that the iterator passed as the first argument
4166 ** currently points to to the database. If a conflict is encountered, invoke
4167 ** the conflict handler callback.
4168 **
4169 ** If argument pbRetry is NULL, then ignore any CHANGESET_DATA conflict. If
4170 ** one is encountered, update or delete the row with the matching primary key
4171 ** instead. Or, if pbRetry is not NULL and a CHANGESET_DATA conflict occurs,
4172 ** invoke the conflict handler. If it returns CHANGESET_REPLACE, set *pbRetry
4173 ** to true before returning. In this case the caller will invoke this function
4174 ** again, this time with pbRetry set to NULL.
4175 **
4176 ** If argument pbReplace is NULL and a CHANGESET_CONFLICT conflict is
4177 ** encountered invoke the conflict handler with CHANGESET_CONSTRAINT instead.
4178 ** Or, if pbReplace is not NULL, invoke it with CHANGESET_CONFLICT. If such
4179 ** an invocation returns SQLITE_CHANGESET_REPLACE, set *pbReplace to true
4180 ** before retrying. In this case the caller attempts to remove the conflicting
4181 ** row before invoking this function again, this time with pbReplace set
4182 ** to NULL.
4183 **
4184 ** If any conflict handler returns SQLITE_CHANGESET_ABORT, this function
4185 ** returns SQLITE_ABORT. Otherwise, if no error occurs, SQLITE_OK is
4186 ** returned.
4187 */
sessionApplyOneOp(sqlite3_changeset_iter * pIter,SessionApplyCtx * p,int (* xConflict)(void *,int,sqlite3_changeset_iter *),void * pCtx,int * pbReplace,int * pbRetry)4188 static int sessionApplyOneOp(
4189 sqlite3_changeset_iter *pIter, /* Changeset iterator */
4190 SessionApplyCtx *p, /* changeset_apply() context */
4191 int(*xConflict)(void *, int, sqlite3_changeset_iter *),
4192 void *pCtx, /* First argument for the conflict handler */
4193 int *pbReplace, /* OUT: True to remove PK row and retry */
4194 int *pbRetry /* OUT: True to retry. */
4195 ){
4196 const char *zDummy;
4197 int op;
4198 int nCol;
4199 int rc = SQLITE_OK;
4200
4201 assert( p->pDelete && p->pInsert && p->pSelect );
4202 assert( p->azCol && p->abPK );
4203 assert( !pbReplace || *pbReplace==0 );
4204
4205 sqlite3changeset_op(pIter, &zDummy, &nCol, &op, 0);
4206
4207 if( op==SQLITE_DELETE ){
4208
4209 /* Bind values to the DELETE statement. If conflict handling is required,
4210 ** bind values for all columns and set bound variable (nCol+1) to true.
4211 ** Or, if conflict handling is not required, bind just the PK column
4212 ** values and, if it exists, set (nCol+1) to false. Conflict handling
4213 ** is not required if:
4214 **
4215 ** * this is a patchset, or
4216 ** * (pbRetry==0), or
4217 ** * all columns of the table are PK columns (in this case there is
4218 ** no (nCol+1) variable to bind to).
4219 */
4220 u8 *abPK = (pIter->bPatchset ? p->abPK : 0);
4221 rc = sessionBindRow(pIter, sqlite3changeset_old, nCol, abPK, p->pDelete);
4222 if( rc==SQLITE_OK && sqlite3_bind_parameter_count(p->pDelete)>nCol ){
4223 rc = sqlite3_bind_int(p->pDelete, nCol+1, (pbRetry==0 || abPK));
4224 }
4225 if( rc!=SQLITE_OK ) return rc;
4226
4227 sqlite3_step(p->pDelete);
4228 rc = sqlite3_reset(p->pDelete);
4229 if( rc==SQLITE_OK && sqlite3_changes(p->db)==0 ){
4230 rc = sessionConflictHandler(
4231 SQLITE_CHANGESET_DATA, p, pIter, xConflict, pCtx, pbRetry
4232 );
4233 }else if( (rc&0xff)==SQLITE_CONSTRAINT ){
4234 rc = sessionConflictHandler(
4235 SQLITE_CHANGESET_CONFLICT, p, pIter, xConflict, pCtx, 0
4236 );
4237 }
4238
4239 }else if( op==SQLITE_UPDATE ){
4240 int i;
4241 sqlite3_stmt *pUp = 0;
4242 int bPatchset = (pbRetry==0 || pIter->bPatchset);
4243
4244 rc = sessionUpdateFind(pIter, p, bPatchset, &pUp);
4245
4246 /* Bind values to the UPDATE statement. */
4247 for(i=0; rc==SQLITE_OK && i<nCol; i++){
4248 sqlite3_value *pOld = sessionChangesetOld(pIter, i);
4249 sqlite3_value *pNew = sessionChangesetNew(pIter, i);
4250 if( p->abPK[i] || (bPatchset==0 && pOld) ){
4251 rc = sessionBindValue(pUp, i*2+2, pOld);
4252 }
4253 if( rc==SQLITE_OK && pNew ){
4254 rc = sessionBindValue(pUp, i*2+1, pNew);
4255 }
4256 }
4257 if( rc!=SQLITE_OK ) return rc;
4258
4259 /* Attempt the UPDATE. In the case of a NOTFOUND or DATA conflict,
4260 ** the result will be SQLITE_OK with 0 rows modified. */
4261 sqlite3_step(pUp);
4262 rc = sqlite3_reset(pUp);
4263
4264 if( rc==SQLITE_OK && sqlite3_changes(p->db)==0 ){
4265 /* A NOTFOUND or DATA error. Search the table to see if it contains
4266 ** a row with a matching primary key. If so, this is a DATA conflict.
4267 ** Otherwise, if there is no primary key match, it is a NOTFOUND. */
4268
4269 rc = sessionConflictHandler(
4270 SQLITE_CHANGESET_DATA, p, pIter, xConflict, pCtx, pbRetry
4271 );
4272
4273 }else if( (rc&0xff)==SQLITE_CONSTRAINT ){
4274 /* This is always a CONSTRAINT conflict. */
4275 rc = sessionConflictHandler(
4276 SQLITE_CHANGESET_CONFLICT, p, pIter, xConflict, pCtx, 0
4277 );
4278 }
4279
4280 }else{
4281 assert( op==SQLITE_INSERT );
4282 if( p->bStat1 ){
4283 /* Check if there is a conflicting row. For sqlite_stat1, this needs
4284 ** to be done using a SELECT, as there is no PRIMARY KEY in the
4285 ** database schema to throw an exception if a duplicate is inserted. */
4286 rc = sessionSeekToRow(p->db, pIter, p->abPK, p->pSelect);
4287 if( rc==SQLITE_ROW ){
4288 rc = SQLITE_CONSTRAINT;
4289 sqlite3_reset(p->pSelect);
4290 }
4291 }
4292
4293 if( rc==SQLITE_OK ){
4294 rc = sessionBindRow(pIter, sqlite3changeset_new, nCol, 0, p->pInsert);
4295 if( rc!=SQLITE_OK ) return rc;
4296
4297 sqlite3_step(p->pInsert);
4298 rc = sqlite3_reset(p->pInsert);
4299 }
4300
4301 if( (rc&0xff)==SQLITE_CONSTRAINT ){
4302 rc = sessionConflictHandler(
4303 SQLITE_CHANGESET_CONFLICT, p, pIter, xConflict, pCtx, pbReplace
4304 );
4305 }
4306 }
4307
4308 return rc;
4309 }
4310
4311 /*
4312 ** Attempt to apply the change that the iterator passed as the first argument
4313 ** currently points to to the database. If a conflict is encountered, invoke
4314 ** the conflict handler callback.
4315 **
4316 ** The difference between this function and sessionApplyOne() is that this
4317 ** function handles the case where the conflict-handler is invoked and
4318 ** returns SQLITE_CHANGESET_REPLACE - indicating that the change should be
4319 ** retried in some manner.
4320 */
sessionApplyOneWithRetry(sqlite3 * db,sqlite3_changeset_iter * pIter,SessionApplyCtx * pApply,int (* xConflict)(void *,int,sqlite3_changeset_iter *),void * pCtx)4321 static int sessionApplyOneWithRetry(
4322 sqlite3 *db, /* Apply change to "main" db of this handle */
4323 sqlite3_changeset_iter *pIter, /* Changeset iterator to read change from */
4324 SessionApplyCtx *pApply, /* Apply context */
4325 int(*xConflict)(void*, int, sqlite3_changeset_iter*),
4326 void *pCtx /* First argument passed to xConflict */
4327 ){
4328 int bReplace = 0;
4329 int bRetry = 0;
4330 int rc;
4331
4332 rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, &bReplace, &bRetry);
4333 if( rc==SQLITE_OK ){
4334 /* If the bRetry flag is set, the change has not been applied due to an
4335 ** SQLITE_CHANGESET_DATA problem (i.e. this is an UPDATE or DELETE and
4336 ** a row with the correct PK is present in the db, but one or more other
4337 ** fields do not contain the expected values) and the conflict handler
4338 ** returned SQLITE_CHANGESET_REPLACE. In this case retry the operation,
4339 ** but pass NULL as the final argument so that sessionApplyOneOp() ignores
4340 ** the SQLITE_CHANGESET_DATA problem. */
4341 if( bRetry ){
4342 assert( pIter->op==SQLITE_UPDATE || pIter->op==SQLITE_DELETE );
4343 rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, 0, 0);
4344 }
4345
4346 /* If the bReplace flag is set, the change is an INSERT that has not
4347 ** been performed because the database already contains a row with the
4348 ** specified primary key and the conflict handler returned
4349 ** SQLITE_CHANGESET_REPLACE. In this case remove the conflicting row
4350 ** before reattempting the INSERT. */
4351 else if( bReplace ){
4352 assert( pIter->op==SQLITE_INSERT );
4353 rc = sqlite3_exec(db, "SAVEPOINT replace_op", 0, 0, 0);
4354 if( rc==SQLITE_OK ){
4355 rc = sessionBindRow(pIter,
4356 sqlite3changeset_new, pApply->nCol, pApply->abPK, pApply->pDelete);
4357 sqlite3_bind_int(pApply->pDelete, pApply->nCol+1, 1);
4358 }
4359 if( rc==SQLITE_OK ){
4360 sqlite3_step(pApply->pDelete);
4361 rc = sqlite3_reset(pApply->pDelete);
4362 }
4363 if( rc==SQLITE_OK ){
4364 rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, 0, 0);
4365 }
4366 if( rc==SQLITE_OK ){
4367 rc = sqlite3_exec(db, "RELEASE replace_op", 0, 0, 0);
4368 }
4369 }
4370 }
4371
4372 return rc;
4373 }
4374
4375 /*
4376 ** Retry the changes accumulated in the pApply->constraints buffer.
4377 */
sessionRetryConstraints(sqlite3 * db,int bPatchset,const char * zTab,SessionApplyCtx * pApply,int (* xConflict)(void *,int,sqlite3_changeset_iter *),void * pCtx)4378 static int sessionRetryConstraints(
4379 sqlite3 *db,
4380 int bPatchset,
4381 const char *zTab,
4382 SessionApplyCtx *pApply,
4383 int(*xConflict)(void*, int, sqlite3_changeset_iter*),
4384 void *pCtx /* First argument passed to xConflict */
4385 ){
4386 int rc = SQLITE_OK;
4387
4388 while( pApply->constraints.nBuf ){
4389 sqlite3_changeset_iter *pIter2 = 0;
4390 SessionBuffer cons = pApply->constraints;
4391 memset(&pApply->constraints, 0, sizeof(SessionBuffer));
4392
4393 rc = sessionChangesetStart(
4394 &pIter2, 0, 0, cons.nBuf, cons.aBuf, pApply->bInvertConstraints, 1
4395 );
4396 if( rc==SQLITE_OK ){
4397 size_t nByte = 2*pApply->nCol*sizeof(sqlite3_value*);
4398 int rc2;
4399 pIter2->bPatchset = bPatchset;
4400 pIter2->zTab = (char*)zTab;
4401 pIter2->nCol = pApply->nCol;
4402 pIter2->abPK = pApply->abPK;
4403 sessionBufferGrow(&pIter2->tblhdr, nByte, &rc);
4404 pIter2->apValue = (sqlite3_value**)pIter2->tblhdr.aBuf;
4405 if( rc==SQLITE_OK ) memset(pIter2->apValue, 0, nByte);
4406
4407 while( rc==SQLITE_OK && SQLITE_ROW==sqlite3changeset_next(pIter2) ){
4408 rc = sessionApplyOneWithRetry(db, pIter2, pApply, xConflict, pCtx);
4409 }
4410
4411 rc2 = sqlite3changeset_finalize(pIter2);
4412 if( rc==SQLITE_OK ) rc = rc2;
4413 }
4414 assert( pApply->bDeferConstraints || pApply->constraints.nBuf==0 );
4415
4416 sqlite3_free(cons.aBuf);
4417 if( rc!=SQLITE_OK ) break;
4418 if( pApply->constraints.nBuf>=cons.nBuf ){
4419 /* No progress was made on the last round. */
4420 pApply->bDeferConstraints = 0;
4421 }
4422 }
4423
4424 return rc;
4425 }
4426
4427 /*
4428 ** Argument pIter is a changeset iterator that has been initialized, but
4429 ** not yet passed to sqlite3changeset_next(). This function applies the
4430 ** changeset to the main database attached to handle "db". The supplied
4431 ** conflict handler callback is invoked to resolve any conflicts encountered
4432 ** while applying the change.
4433 */
sessionChangesetApply(sqlite3 * db,sqlite3_changeset_iter * pIter,int (* xFilter)(void * pCtx,const char * zTab),int (* xConflict)(void * pCtx,int eConflict,sqlite3_changeset_iter * p),void * pCtx,void ** ppRebase,int * pnRebase,int flags)4434 static int sessionChangesetApply(
4435 sqlite3 *db, /* Apply change to "main" db of this handle */
4436 sqlite3_changeset_iter *pIter, /* Changeset to apply */
4437 int(*xFilter)(
4438 void *pCtx, /* Copy of sixth arg to _apply() */
4439 const char *zTab /* Table name */
4440 ),
4441 int(*xConflict)(
4442 void *pCtx, /* Copy of fifth arg to _apply() */
4443 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4444 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4445 ),
4446 void *pCtx, /* First argument passed to xConflict */
4447 void **ppRebase, int *pnRebase, /* OUT: Rebase information */
4448 int flags /* SESSION_APPLY_XXX flags */
4449 ){
4450 int schemaMismatch = 0;
4451 int rc = SQLITE_OK; /* Return code */
4452 const char *zTab = 0; /* Name of current table */
4453 int nTab = 0; /* Result of sqlite3Strlen30(zTab) */
4454 SessionApplyCtx sApply; /* changeset_apply() context object */
4455 int bPatchset;
4456
4457 assert( xConflict!=0 );
4458
4459 pIter->in.bNoDiscard = 1;
4460 memset(&sApply, 0, sizeof(sApply));
4461 sApply.bRebase = (ppRebase && pnRebase);
4462 sApply.bInvertConstraints = !!(flags & SQLITE_CHANGESETAPPLY_INVERT);
4463 sqlite3_mutex_enter(sqlite3_db_mutex(db));
4464 if( (flags & SQLITE_CHANGESETAPPLY_NOSAVEPOINT)==0 ){
4465 rc = sqlite3_exec(db, "SAVEPOINT changeset_apply", 0, 0, 0);
4466 }
4467 if( rc==SQLITE_OK ){
4468 rc = sqlite3_exec(db, "PRAGMA defer_foreign_keys = 1", 0, 0, 0);
4469 }
4470 while( rc==SQLITE_OK && SQLITE_ROW==sqlite3changeset_next(pIter) ){
4471 int nCol;
4472 int op;
4473 const char *zNew;
4474
4475 sqlite3changeset_op(pIter, &zNew, &nCol, &op, 0);
4476
4477 if( zTab==0 || sqlite3_strnicmp(zNew, zTab, nTab+1) ){
4478 u8 *abPK;
4479
4480 rc = sessionRetryConstraints(
4481 db, pIter->bPatchset, zTab, &sApply, xConflict, pCtx
4482 );
4483 if( rc!=SQLITE_OK ) break;
4484
4485 sessionUpdateFree(&sApply);
4486 sqlite3_free((char*)sApply.azCol); /* cast works around VC++ bug */
4487 sqlite3_finalize(sApply.pDelete);
4488 sqlite3_finalize(sApply.pInsert);
4489 sqlite3_finalize(sApply.pSelect);
4490 sApply.db = db;
4491 sApply.pDelete = 0;
4492 sApply.pInsert = 0;
4493 sApply.pSelect = 0;
4494 sApply.nCol = 0;
4495 sApply.azCol = 0;
4496 sApply.abPK = 0;
4497 sApply.bStat1 = 0;
4498 sApply.bDeferConstraints = 1;
4499 sApply.bRebaseStarted = 0;
4500 memset(&sApply.constraints, 0, sizeof(SessionBuffer));
4501
4502 /* If an xFilter() callback was specified, invoke it now. If the
4503 ** xFilter callback returns zero, skip this table. If it returns
4504 ** non-zero, proceed. */
4505 schemaMismatch = (xFilter && (0==xFilter(pCtx, zNew)));
4506 if( schemaMismatch ){
4507 zTab = sqlite3_mprintf("%s", zNew);
4508 if( zTab==0 ){
4509 rc = SQLITE_NOMEM;
4510 break;
4511 }
4512 nTab = (int)strlen(zTab);
4513 sApply.azCol = (const char **)zTab;
4514 }else{
4515 int nMinCol = 0;
4516 int i;
4517
4518 sqlite3changeset_pk(pIter, &abPK, 0);
4519 rc = sessionTableInfo(0,
4520 db, "main", zNew, &sApply.nCol, &zTab, &sApply.azCol, &sApply.abPK
4521 );
4522 if( rc!=SQLITE_OK ) break;
4523 for(i=0; i<sApply.nCol; i++){
4524 if( sApply.abPK[i] ) nMinCol = i+1;
4525 }
4526
4527 if( sApply.nCol==0 ){
4528 schemaMismatch = 1;
4529 sqlite3_log(SQLITE_SCHEMA,
4530 "sqlite3changeset_apply(): no such table: %s", zTab
4531 );
4532 }
4533 else if( sApply.nCol<nCol ){
4534 schemaMismatch = 1;
4535 sqlite3_log(SQLITE_SCHEMA,
4536 "sqlite3changeset_apply(): table %s has %d columns, "
4537 "expected %d or more",
4538 zTab, sApply.nCol, nCol
4539 );
4540 }
4541 else if( nCol<nMinCol || memcmp(sApply.abPK, abPK, nCol)!=0 ){
4542 schemaMismatch = 1;
4543 sqlite3_log(SQLITE_SCHEMA, "sqlite3changeset_apply(): "
4544 "primary key mismatch for table %s", zTab
4545 );
4546 }
4547 else{
4548 sApply.nCol = nCol;
4549 if( 0==sqlite3_stricmp(zTab, "sqlite_stat1") ){
4550 if( (rc = sessionStat1Sql(db, &sApply) ) ){
4551 break;
4552 }
4553 sApply.bStat1 = 1;
4554 }else{
4555 if( (rc = sessionSelectRow(db, zTab, &sApply))
4556 || (rc = sessionDeleteRow(db, zTab, &sApply))
4557 || (rc = sessionInsertRow(db, zTab, &sApply))
4558 ){
4559 break;
4560 }
4561 sApply.bStat1 = 0;
4562 }
4563 }
4564 nTab = sqlite3Strlen30(zTab);
4565 }
4566 }
4567
4568 /* If there is a schema mismatch on the current table, proceed to the
4569 ** next change. A log message has already been issued. */
4570 if( schemaMismatch ) continue;
4571
4572 rc = sessionApplyOneWithRetry(db, pIter, &sApply, xConflict, pCtx);
4573 }
4574
4575 bPatchset = pIter->bPatchset;
4576 if( rc==SQLITE_OK ){
4577 rc = sqlite3changeset_finalize(pIter);
4578 }else{
4579 sqlite3changeset_finalize(pIter);
4580 }
4581
4582 if( rc==SQLITE_OK ){
4583 rc = sessionRetryConstraints(db, bPatchset, zTab, &sApply, xConflict, pCtx);
4584 }
4585
4586 if( rc==SQLITE_OK ){
4587 int nFk, notUsed;
4588 sqlite3_db_status(db, SQLITE_DBSTATUS_DEFERRED_FKS, &nFk, ¬Used, 0);
4589 if( nFk!=0 ){
4590 int res = SQLITE_CHANGESET_ABORT;
4591 sqlite3_changeset_iter sIter;
4592 memset(&sIter, 0, sizeof(sIter));
4593 sIter.nCol = nFk;
4594 res = xConflict(pCtx, SQLITE_CHANGESET_FOREIGN_KEY, &sIter);
4595 if( res!=SQLITE_CHANGESET_OMIT ){
4596 rc = SQLITE_CONSTRAINT;
4597 }
4598 }
4599 }
4600 sqlite3_exec(db, "PRAGMA defer_foreign_keys = 0", 0, 0, 0);
4601
4602 if( (flags & SQLITE_CHANGESETAPPLY_NOSAVEPOINT)==0 ){
4603 if( rc==SQLITE_OK ){
4604 rc = sqlite3_exec(db, "RELEASE changeset_apply", 0, 0, 0);
4605 }else{
4606 sqlite3_exec(db, "ROLLBACK TO changeset_apply", 0, 0, 0);
4607 sqlite3_exec(db, "RELEASE changeset_apply", 0, 0, 0);
4608 }
4609 }
4610
4611 assert( sApply.bRebase || sApply.rebase.nBuf==0 );
4612 if( rc==SQLITE_OK && bPatchset==0 && sApply.bRebase ){
4613 *ppRebase = (void*)sApply.rebase.aBuf;
4614 *pnRebase = sApply.rebase.nBuf;
4615 sApply.rebase.aBuf = 0;
4616 }
4617 sessionUpdateFree(&sApply);
4618 sqlite3_finalize(sApply.pInsert);
4619 sqlite3_finalize(sApply.pDelete);
4620 sqlite3_finalize(sApply.pSelect);
4621 sqlite3_free((char*)sApply.azCol); /* cast works around VC++ bug */
4622 sqlite3_free((char*)sApply.constraints.aBuf);
4623 sqlite3_free((char*)sApply.rebase.aBuf);
4624 sqlite3_mutex_leave(sqlite3_db_mutex(db));
4625 return rc;
4626 }
4627
4628 /*
4629 ** Apply the changeset passed via pChangeset/nChangeset to the main
4630 ** database attached to handle "db".
4631 */
sqlite3changeset_apply_v2(sqlite3 * db,int nChangeset,void * pChangeset,int (* xFilter)(void * pCtx,const char * zTab),int (* xConflict)(void * pCtx,int eConflict,sqlite3_changeset_iter * p),void * pCtx,void ** ppRebase,int * pnRebase,int flags)4632 int sqlite3changeset_apply_v2(
4633 sqlite3 *db, /* Apply change to "main" db of this handle */
4634 int nChangeset, /* Size of changeset in bytes */
4635 void *pChangeset, /* Changeset blob */
4636 int(*xFilter)(
4637 void *pCtx, /* Copy of sixth arg to _apply() */
4638 const char *zTab /* Table name */
4639 ),
4640 int(*xConflict)(
4641 void *pCtx, /* Copy of sixth arg to _apply() */
4642 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4643 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4644 ),
4645 void *pCtx, /* First argument passed to xConflict */
4646 void **ppRebase, int *pnRebase,
4647 int flags
4648 ){
4649 sqlite3_changeset_iter *pIter; /* Iterator to skip through changeset */
4650 int bInv = !!(flags & SQLITE_CHANGESETAPPLY_INVERT);
4651 int rc = sessionChangesetStart(&pIter, 0, 0, nChangeset, pChangeset, bInv, 1);
4652 if( rc==SQLITE_OK ){
4653 rc = sessionChangesetApply(
4654 db, pIter, xFilter, xConflict, pCtx, ppRebase, pnRebase, flags
4655 );
4656 }
4657 return rc;
4658 }
4659
4660 /*
4661 ** Apply the changeset passed via pChangeset/nChangeset to the main database
4662 ** attached to handle "db". Invoke the supplied conflict handler callback
4663 ** to resolve any conflicts encountered while applying the change.
4664 */
sqlite3changeset_apply(sqlite3 * db,int nChangeset,void * pChangeset,int (* xFilter)(void * pCtx,const char * zTab),int (* xConflict)(void * pCtx,int eConflict,sqlite3_changeset_iter * p),void * pCtx)4665 int sqlite3changeset_apply(
4666 sqlite3 *db, /* Apply change to "main" db of this handle */
4667 int nChangeset, /* Size of changeset in bytes */
4668 void *pChangeset, /* Changeset blob */
4669 int(*xFilter)(
4670 void *pCtx, /* Copy of sixth arg to _apply() */
4671 const char *zTab /* Table name */
4672 ),
4673 int(*xConflict)(
4674 void *pCtx, /* Copy of fifth arg to _apply() */
4675 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4676 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4677 ),
4678 void *pCtx /* First argument passed to xConflict */
4679 ){
4680 return sqlite3changeset_apply_v2(
4681 db, nChangeset, pChangeset, xFilter, xConflict, pCtx, 0, 0, 0
4682 );
4683 }
4684
4685 /*
4686 ** Apply the changeset passed via xInput/pIn to the main database
4687 ** attached to handle "db". Invoke the supplied conflict handler callback
4688 ** to resolve any conflicts encountered while applying the change.
4689 */
sqlite3changeset_apply_v2_strm(sqlite3 * db,int (* xInput)(void * pIn,void * pData,int * pnData),void * pIn,int (* xFilter)(void * pCtx,const char * zTab),int (* xConflict)(void * pCtx,int eConflict,sqlite3_changeset_iter * p),void * pCtx,void ** ppRebase,int * pnRebase,int flags)4690 int sqlite3changeset_apply_v2_strm(
4691 sqlite3 *db, /* Apply change to "main" db of this handle */
4692 int (*xInput)(void *pIn, void *pData, int *pnData), /* Input function */
4693 void *pIn, /* First arg for xInput */
4694 int(*xFilter)(
4695 void *pCtx, /* Copy of sixth arg to _apply() */
4696 const char *zTab /* Table name */
4697 ),
4698 int(*xConflict)(
4699 void *pCtx, /* Copy of sixth arg to _apply() */
4700 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4701 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4702 ),
4703 void *pCtx, /* First argument passed to xConflict */
4704 void **ppRebase, int *pnRebase,
4705 int flags
4706 ){
4707 sqlite3_changeset_iter *pIter; /* Iterator to skip through changeset */
4708 int bInverse = !!(flags & SQLITE_CHANGESETAPPLY_INVERT);
4709 int rc = sessionChangesetStart(&pIter, xInput, pIn, 0, 0, bInverse, 1);
4710 if( rc==SQLITE_OK ){
4711 rc = sessionChangesetApply(
4712 db, pIter, xFilter, xConflict, pCtx, ppRebase, pnRebase, flags
4713 );
4714 }
4715 return rc;
4716 }
sqlite3changeset_apply_strm(sqlite3 * db,int (* xInput)(void * pIn,void * pData,int * pnData),void * pIn,int (* xFilter)(void * pCtx,const char * zTab),int (* xConflict)(void * pCtx,int eConflict,sqlite3_changeset_iter * p),void * pCtx)4717 int sqlite3changeset_apply_strm(
4718 sqlite3 *db, /* Apply change to "main" db of this handle */
4719 int (*xInput)(void *pIn, void *pData, int *pnData), /* Input function */
4720 void *pIn, /* First arg for xInput */
4721 int(*xFilter)(
4722 void *pCtx, /* Copy of sixth arg to _apply() */
4723 const char *zTab /* Table name */
4724 ),
4725 int(*xConflict)(
4726 void *pCtx, /* Copy of sixth arg to _apply() */
4727 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4728 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4729 ),
4730 void *pCtx /* First argument passed to xConflict */
4731 ){
4732 return sqlite3changeset_apply_v2_strm(
4733 db, xInput, pIn, xFilter, xConflict, pCtx, 0, 0, 0
4734 );
4735 }
4736
4737 /*
4738 ** sqlite3_changegroup handle.
4739 */
4740 struct sqlite3_changegroup {
4741 int rc; /* Error code */
4742 int bPatch; /* True to accumulate patchsets */
4743 SessionTable *pList; /* List of tables in current patch */
4744 };
4745
4746 /*
4747 ** This function is called to merge two changes to the same row together as
4748 ** part of an sqlite3changeset_concat() operation. A new change object is
4749 ** allocated and a pointer to it stored in *ppNew.
4750 */
sessionChangeMerge(SessionTable * pTab,int bRebase,int bPatchset,SessionChange * pExist,int op2,int bIndirect,u8 * aRec,int nRec,SessionChange ** ppNew)4751 static int sessionChangeMerge(
4752 SessionTable *pTab, /* Table structure */
4753 int bRebase, /* True for a rebase hash-table */
4754 int bPatchset, /* True for patchsets */
4755 SessionChange *pExist, /* Existing change */
4756 int op2, /* Second change operation */
4757 int bIndirect, /* True if second change is indirect */
4758 u8 *aRec, /* Second change record */
4759 int nRec, /* Number of bytes in aRec */
4760 SessionChange **ppNew /* OUT: Merged change */
4761 ){
4762 SessionChange *pNew = 0;
4763 int rc = SQLITE_OK;
4764
4765 if( !pExist ){
4766 pNew = (SessionChange *)sqlite3_malloc64(sizeof(SessionChange) + nRec);
4767 if( !pNew ){
4768 return SQLITE_NOMEM;
4769 }
4770 memset(pNew, 0, sizeof(SessionChange));
4771 pNew->op = op2;
4772 pNew->bIndirect = bIndirect;
4773 pNew->aRecord = (u8*)&pNew[1];
4774 if( bIndirect==0 || bRebase==0 ){
4775 pNew->nRecord = nRec;
4776 memcpy(pNew->aRecord, aRec, nRec);
4777 }else{
4778 int i;
4779 u8 *pIn = aRec;
4780 u8 *pOut = pNew->aRecord;
4781 for(i=0; i<pTab->nCol; i++){
4782 int nIn = sessionSerialLen(pIn);
4783 if( *pIn==0 ){
4784 *pOut++ = 0;
4785 }else if( pTab->abPK[i]==0 ){
4786 *pOut++ = 0xFF;
4787 }else{
4788 memcpy(pOut, pIn, nIn);
4789 pOut += nIn;
4790 }
4791 pIn += nIn;
4792 }
4793 pNew->nRecord = pOut - pNew->aRecord;
4794 }
4795 }else if( bRebase ){
4796 if( pExist->op==SQLITE_DELETE && pExist->bIndirect ){
4797 *ppNew = pExist;
4798 }else{
4799 sqlite3_int64 nByte = nRec + pExist->nRecord + sizeof(SessionChange);
4800 pNew = (SessionChange*)sqlite3_malloc64(nByte);
4801 if( pNew==0 ){
4802 rc = SQLITE_NOMEM;
4803 }else{
4804 int i;
4805 u8 *a1 = pExist->aRecord;
4806 u8 *a2 = aRec;
4807 u8 *pOut;
4808
4809 memset(pNew, 0, nByte);
4810 pNew->bIndirect = bIndirect || pExist->bIndirect;
4811 pNew->op = op2;
4812 pOut = pNew->aRecord = (u8*)&pNew[1];
4813
4814 for(i=0; i<pTab->nCol; i++){
4815 int n1 = sessionSerialLen(a1);
4816 int n2 = sessionSerialLen(a2);
4817 if( *a1==0xFF || (pTab->abPK[i]==0 && bIndirect) ){
4818 *pOut++ = 0xFF;
4819 }else if( *a2==0 ){
4820 memcpy(pOut, a1, n1);
4821 pOut += n1;
4822 }else{
4823 memcpy(pOut, a2, n2);
4824 pOut += n2;
4825 }
4826 a1 += n1;
4827 a2 += n2;
4828 }
4829 pNew->nRecord = pOut - pNew->aRecord;
4830 }
4831 sqlite3_free(pExist);
4832 }
4833 }else{
4834 int op1 = pExist->op;
4835
4836 /*
4837 ** op1=INSERT, op2=INSERT -> Unsupported. Discard op2.
4838 ** op1=INSERT, op2=UPDATE -> INSERT.
4839 ** op1=INSERT, op2=DELETE -> (none)
4840 **
4841 ** op1=UPDATE, op2=INSERT -> Unsupported. Discard op2.
4842 ** op1=UPDATE, op2=UPDATE -> UPDATE.
4843 ** op1=UPDATE, op2=DELETE -> DELETE.
4844 **
4845 ** op1=DELETE, op2=INSERT -> UPDATE.
4846 ** op1=DELETE, op2=UPDATE -> Unsupported. Discard op2.
4847 ** op1=DELETE, op2=DELETE -> Unsupported. Discard op2.
4848 */
4849 if( (op1==SQLITE_INSERT && op2==SQLITE_INSERT)
4850 || (op1==SQLITE_UPDATE && op2==SQLITE_INSERT)
4851 || (op1==SQLITE_DELETE && op2==SQLITE_UPDATE)
4852 || (op1==SQLITE_DELETE && op2==SQLITE_DELETE)
4853 ){
4854 pNew = pExist;
4855 }else if( op1==SQLITE_INSERT && op2==SQLITE_DELETE ){
4856 sqlite3_free(pExist);
4857 assert( pNew==0 );
4858 }else{
4859 u8 *aExist = pExist->aRecord;
4860 sqlite3_int64 nByte;
4861 u8 *aCsr;
4862
4863 /* Allocate a new SessionChange object. Ensure that the aRecord[]
4864 ** buffer of the new object is large enough to hold any record that
4865 ** may be generated by combining the input records. */
4866 nByte = sizeof(SessionChange) + pExist->nRecord + nRec;
4867 pNew = (SessionChange *)sqlite3_malloc64(nByte);
4868 if( !pNew ){
4869 sqlite3_free(pExist);
4870 return SQLITE_NOMEM;
4871 }
4872 memset(pNew, 0, sizeof(SessionChange));
4873 pNew->bIndirect = (bIndirect && pExist->bIndirect);
4874 aCsr = pNew->aRecord = (u8 *)&pNew[1];
4875
4876 if( op1==SQLITE_INSERT ){ /* INSERT + UPDATE */
4877 u8 *a1 = aRec;
4878 assert( op2==SQLITE_UPDATE );
4879 pNew->op = SQLITE_INSERT;
4880 if( bPatchset==0 ) sessionSkipRecord(&a1, pTab->nCol);
4881 sessionMergeRecord(&aCsr, pTab->nCol, aExist, a1);
4882 }else if( op1==SQLITE_DELETE ){ /* DELETE + INSERT */
4883 assert( op2==SQLITE_INSERT );
4884 pNew->op = SQLITE_UPDATE;
4885 if( bPatchset ){
4886 memcpy(aCsr, aRec, nRec);
4887 aCsr += nRec;
4888 }else{
4889 if( 0==sessionMergeUpdate(&aCsr, pTab, bPatchset, aExist, 0,aRec,0) ){
4890 sqlite3_free(pNew);
4891 pNew = 0;
4892 }
4893 }
4894 }else if( op2==SQLITE_UPDATE ){ /* UPDATE + UPDATE */
4895 u8 *a1 = aExist;
4896 u8 *a2 = aRec;
4897 assert( op1==SQLITE_UPDATE );
4898 if( bPatchset==0 ){
4899 sessionSkipRecord(&a1, pTab->nCol);
4900 sessionSkipRecord(&a2, pTab->nCol);
4901 }
4902 pNew->op = SQLITE_UPDATE;
4903 if( 0==sessionMergeUpdate(&aCsr, pTab, bPatchset, aRec, aExist,a1,a2) ){
4904 sqlite3_free(pNew);
4905 pNew = 0;
4906 }
4907 }else{ /* UPDATE + DELETE */
4908 assert( op1==SQLITE_UPDATE && op2==SQLITE_DELETE );
4909 pNew->op = SQLITE_DELETE;
4910 if( bPatchset ){
4911 memcpy(aCsr, aRec, nRec);
4912 aCsr += nRec;
4913 }else{
4914 sessionMergeRecord(&aCsr, pTab->nCol, aRec, aExist);
4915 }
4916 }
4917
4918 if( pNew ){
4919 pNew->nRecord = (int)(aCsr - pNew->aRecord);
4920 }
4921 sqlite3_free(pExist);
4922 }
4923 }
4924
4925 *ppNew = pNew;
4926 return rc;
4927 }
4928
4929 /*
4930 ** Add all changes in the changeset traversed by the iterator passed as
4931 ** the first argument to the changegroup hash tables.
4932 */
sessionChangesetToHash(sqlite3_changeset_iter * pIter,sqlite3_changegroup * pGrp,int bRebase)4933 static int sessionChangesetToHash(
4934 sqlite3_changeset_iter *pIter, /* Iterator to read from */
4935 sqlite3_changegroup *pGrp, /* Changegroup object to add changeset to */
4936 int bRebase /* True if hash table is for rebasing */
4937 ){
4938 u8 *aRec;
4939 int nRec;
4940 int rc = SQLITE_OK;
4941 SessionTable *pTab = 0;
4942
4943 while( SQLITE_ROW==sessionChangesetNext(pIter, &aRec, &nRec, 0) ){
4944 const char *zNew;
4945 int nCol;
4946 int op;
4947 int iHash;
4948 int bIndirect;
4949 SessionChange *pChange;
4950 SessionChange *pExist = 0;
4951 SessionChange **pp;
4952
4953 if( pGrp->pList==0 ){
4954 pGrp->bPatch = pIter->bPatchset;
4955 }else if( pIter->bPatchset!=pGrp->bPatch ){
4956 rc = SQLITE_ERROR;
4957 break;
4958 }
4959
4960 sqlite3changeset_op(pIter, &zNew, &nCol, &op, &bIndirect);
4961 if( !pTab || sqlite3_stricmp(zNew, pTab->zName) ){
4962 /* Search the list for a matching table */
4963 int nNew = (int)strlen(zNew);
4964 u8 *abPK;
4965
4966 sqlite3changeset_pk(pIter, &abPK, 0);
4967 for(pTab = pGrp->pList; pTab; pTab=pTab->pNext){
4968 if( 0==sqlite3_strnicmp(pTab->zName, zNew, nNew+1) ) break;
4969 }
4970 if( !pTab ){
4971 SessionTable **ppTab;
4972
4973 pTab = sqlite3_malloc64(sizeof(SessionTable) + nCol + nNew+1);
4974 if( !pTab ){
4975 rc = SQLITE_NOMEM;
4976 break;
4977 }
4978 memset(pTab, 0, sizeof(SessionTable));
4979 pTab->nCol = nCol;
4980 pTab->abPK = (u8*)&pTab[1];
4981 memcpy(pTab->abPK, abPK, nCol);
4982 pTab->zName = (char*)&pTab->abPK[nCol];
4983 memcpy(pTab->zName, zNew, nNew+1);
4984
4985 /* The new object must be linked on to the end of the list, not
4986 ** simply added to the start of it. This is to ensure that the
4987 ** tables within the output of sqlite3changegroup_output() are in
4988 ** the right order. */
4989 for(ppTab=&pGrp->pList; *ppTab; ppTab=&(*ppTab)->pNext);
4990 *ppTab = pTab;
4991 }else if( pTab->nCol!=nCol || memcmp(pTab->abPK, abPK, nCol) ){
4992 rc = SQLITE_SCHEMA;
4993 break;
4994 }
4995 }
4996
4997 if( sessionGrowHash(0, pIter->bPatchset, pTab) ){
4998 rc = SQLITE_NOMEM;
4999 break;
5000 }
5001 iHash = sessionChangeHash(
5002 pTab, (pIter->bPatchset && op==SQLITE_DELETE), aRec, pTab->nChange
5003 );
5004
5005 /* Search for existing entry. If found, remove it from the hash table.
5006 ** Code below may link it back in.
5007 */
5008 for(pp=&pTab->apChange[iHash]; *pp; pp=&(*pp)->pNext){
5009 int bPkOnly1 = 0;
5010 int bPkOnly2 = 0;
5011 if( pIter->bPatchset ){
5012 bPkOnly1 = (*pp)->op==SQLITE_DELETE;
5013 bPkOnly2 = op==SQLITE_DELETE;
5014 }
5015 if( sessionChangeEqual(pTab, bPkOnly1, (*pp)->aRecord, bPkOnly2, aRec) ){
5016 pExist = *pp;
5017 *pp = (*pp)->pNext;
5018 pTab->nEntry--;
5019 break;
5020 }
5021 }
5022
5023 rc = sessionChangeMerge(pTab, bRebase,
5024 pIter->bPatchset, pExist, op, bIndirect, aRec, nRec, &pChange
5025 );
5026 if( rc ) break;
5027 if( pChange ){
5028 pChange->pNext = pTab->apChange[iHash];
5029 pTab->apChange[iHash] = pChange;
5030 pTab->nEntry++;
5031 }
5032 }
5033
5034 if( rc==SQLITE_OK ) rc = pIter->rc;
5035 return rc;
5036 }
5037
5038 /*
5039 ** Serialize a changeset (or patchset) based on all changesets (or patchsets)
5040 ** added to the changegroup object passed as the first argument.
5041 **
5042 ** If xOutput is not NULL, then the changeset/patchset is returned to the
5043 ** user via one or more calls to xOutput, as with the other streaming
5044 ** interfaces.
5045 **
5046 ** Or, if xOutput is NULL, then (*ppOut) is populated with a pointer to a
5047 ** buffer containing the output changeset before this function returns. In
5048 ** this case (*pnOut) is set to the size of the output buffer in bytes. It
5049 ** is the responsibility of the caller to free the output buffer using
5050 ** sqlite3_free() when it is no longer required.
5051 **
5052 ** If successful, SQLITE_OK is returned. Or, if an error occurs, an SQLite
5053 ** error code. If an error occurs and xOutput is NULL, (*ppOut) and (*pnOut)
5054 ** are both set to 0 before returning.
5055 */
sessionChangegroupOutput(sqlite3_changegroup * pGrp,int (* xOutput)(void * pOut,const void * pData,int nData),void * pOut,int * pnOut,void ** ppOut)5056 static int sessionChangegroupOutput(
5057 sqlite3_changegroup *pGrp,
5058 int (*xOutput)(void *pOut, const void *pData, int nData),
5059 void *pOut,
5060 int *pnOut,
5061 void **ppOut
5062 ){
5063 int rc = SQLITE_OK;
5064 SessionBuffer buf = {0, 0, 0};
5065 SessionTable *pTab;
5066 assert( xOutput==0 || (ppOut==0 && pnOut==0) );
5067
5068 /* Create the serialized output changeset based on the contents of the
5069 ** hash tables attached to the SessionTable objects in list p->pList.
5070 */
5071 for(pTab=pGrp->pList; rc==SQLITE_OK && pTab; pTab=pTab->pNext){
5072 int i;
5073 if( pTab->nEntry==0 ) continue;
5074
5075 sessionAppendTableHdr(&buf, pGrp->bPatch, pTab, &rc);
5076 for(i=0; i<pTab->nChange; i++){
5077 SessionChange *p;
5078 for(p=pTab->apChange[i]; p; p=p->pNext){
5079 sessionAppendByte(&buf, p->op, &rc);
5080 sessionAppendByte(&buf, p->bIndirect, &rc);
5081 sessionAppendBlob(&buf, p->aRecord, p->nRecord, &rc);
5082 if( rc==SQLITE_OK && xOutput && buf.nBuf>=sessions_strm_chunk_size ){
5083 rc = xOutput(pOut, buf.aBuf, buf.nBuf);
5084 buf.nBuf = 0;
5085 }
5086 }
5087 }
5088 }
5089
5090 if( rc==SQLITE_OK ){
5091 if( xOutput ){
5092 if( buf.nBuf>0 ) rc = xOutput(pOut, buf.aBuf, buf.nBuf);
5093 }else{
5094 *ppOut = buf.aBuf;
5095 *pnOut = buf.nBuf;
5096 buf.aBuf = 0;
5097 }
5098 }
5099 sqlite3_free(buf.aBuf);
5100
5101 return rc;
5102 }
5103
5104 /*
5105 ** Allocate a new, empty, sqlite3_changegroup.
5106 */
sqlite3changegroup_new(sqlite3_changegroup ** pp)5107 int sqlite3changegroup_new(sqlite3_changegroup **pp){
5108 int rc = SQLITE_OK; /* Return code */
5109 sqlite3_changegroup *p; /* New object */
5110 p = (sqlite3_changegroup*)sqlite3_malloc(sizeof(sqlite3_changegroup));
5111 if( p==0 ){
5112 rc = SQLITE_NOMEM;
5113 }else{
5114 memset(p, 0, sizeof(sqlite3_changegroup));
5115 }
5116 *pp = p;
5117 return rc;
5118 }
5119
5120 /*
5121 ** Add the changeset currently stored in buffer pData, size nData bytes,
5122 ** to changeset-group p.
5123 */
sqlite3changegroup_add(sqlite3_changegroup * pGrp,int nData,void * pData)5124 int sqlite3changegroup_add(sqlite3_changegroup *pGrp, int nData, void *pData){
5125 sqlite3_changeset_iter *pIter; /* Iterator opened on pData/nData */
5126 int rc; /* Return code */
5127
5128 rc = sqlite3changeset_start(&pIter, nData, pData);
5129 if( rc==SQLITE_OK ){
5130 rc = sessionChangesetToHash(pIter, pGrp, 0);
5131 }
5132 sqlite3changeset_finalize(pIter);
5133 return rc;
5134 }
5135
5136 /*
5137 ** Obtain a buffer containing a changeset representing the concatenation
5138 ** of all changesets added to the group so far.
5139 */
sqlite3changegroup_output(sqlite3_changegroup * pGrp,int * pnData,void ** ppData)5140 int sqlite3changegroup_output(
5141 sqlite3_changegroup *pGrp,
5142 int *pnData,
5143 void **ppData
5144 ){
5145 return sessionChangegroupOutput(pGrp, 0, 0, pnData, ppData);
5146 }
5147
5148 /*
5149 ** Streaming versions of changegroup_add().
5150 */
sqlite3changegroup_add_strm(sqlite3_changegroup * pGrp,int (* xInput)(void * pIn,void * pData,int * pnData),void * pIn)5151 int sqlite3changegroup_add_strm(
5152 sqlite3_changegroup *pGrp,
5153 int (*xInput)(void *pIn, void *pData, int *pnData),
5154 void *pIn
5155 ){
5156 sqlite3_changeset_iter *pIter; /* Iterator opened on pData/nData */
5157 int rc; /* Return code */
5158
5159 rc = sqlite3changeset_start_strm(&pIter, xInput, pIn);
5160 if( rc==SQLITE_OK ){
5161 rc = sessionChangesetToHash(pIter, pGrp, 0);
5162 }
5163 sqlite3changeset_finalize(pIter);
5164 return rc;
5165 }
5166
5167 /*
5168 ** Streaming versions of changegroup_output().
5169 */
sqlite3changegroup_output_strm(sqlite3_changegroup * pGrp,int (* xOutput)(void * pOut,const void * pData,int nData),void * pOut)5170 int sqlite3changegroup_output_strm(
5171 sqlite3_changegroup *pGrp,
5172 int (*xOutput)(void *pOut, const void *pData, int nData),
5173 void *pOut
5174 ){
5175 return sessionChangegroupOutput(pGrp, xOutput, pOut, 0, 0);
5176 }
5177
5178 /*
5179 ** Delete a changegroup object.
5180 */
sqlite3changegroup_delete(sqlite3_changegroup * pGrp)5181 void sqlite3changegroup_delete(sqlite3_changegroup *pGrp){
5182 if( pGrp ){
5183 sessionDeleteTable(0, pGrp->pList);
5184 sqlite3_free(pGrp);
5185 }
5186 }
5187
5188 /*
5189 ** Combine two changesets together.
5190 */
sqlite3changeset_concat(int nLeft,void * pLeft,int nRight,void * pRight,int * pnOut,void ** ppOut)5191 int sqlite3changeset_concat(
5192 int nLeft, /* Number of bytes in lhs input */
5193 void *pLeft, /* Lhs input changeset */
5194 int nRight /* Number of bytes in rhs input */,
5195 void *pRight, /* Rhs input changeset */
5196 int *pnOut, /* OUT: Number of bytes in output changeset */
5197 void **ppOut /* OUT: changeset (left <concat> right) */
5198 ){
5199 sqlite3_changegroup *pGrp;
5200 int rc;
5201
5202 rc = sqlite3changegroup_new(&pGrp);
5203 if( rc==SQLITE_OK ){
5204 rc = sqlite3changegroup_add(pGrp, nLeft, pLeft);
5205 }
5206 if( rc==SQLITE_OK ){
5207 rc = sqlite3changegroup_add(pGrp, nRight, pRight);
5208 }
5209 if( rc==SQLITE_OK ){
5210 rc = sqlite3changegroup_output(pGrp, pnOut, ppOut);
5211 }
5212 sqlite3changegroup_delete(pGrp);
5213
5214 return rc;
5215 }
5216
5217 /*
5218 ** Streaming version of sqlite3changeset_concat().
5219 */
sqlite3changeset_concat_strm(int (* xInputA)(void * pIn,void * pData,int * pnData),void * pInA,int (* xInputB)(void * pIn,void * pData,int * pnData),void * pInB,int (* xOutput)(void * pOut,const void * pData,int nData),void * pOut)5220 int sqlite3changeset_concat_strm(
5221 int (*xInputA)(void *pIn, void *pData, int *pnData),
5222 void *pInA,
5223 int (*xInputB)(void *pIn, void *pData, int *pnData),
5224 void *pInB,
5225 int (*xOutput)(void *pOut, const void *pData, int nData),
5226 void *pOut
5227 ){
5228 sqlite3_changegroup *pGrp;
5229 int rc;
5230
5231 rc = sqlite3changegroup_new(&pGrp);
5232 if( rc==SQLITE_OK ){
5233 rc = sqlite3changegroup_add_strm(pGrp, xInputA, pInA);
5234 }
5235 if( rc==SQLITE_OK ){
5236 rc = sqlite3changegroup_add_strm(pGrp, xInputB, pInB);
5237 }
5238 if( rc==SQLITE_OK ){
5239 rc = sqlite3changegroup_output_strm(pGrp, xOutput, pOut);
5240 }
5241 sqlite3changegroup_delete(pGrp);
5242
5243 return rc;
5244 }
5245
5246 /*
5247 ** Changeset rebaser handle.
5248 */
5249 struct sqlite3_rebaser {
5250 sqlite3_changegroup grp; /* Hash table */
5251 };
5252
5253 /*
5254 ** Buffers a1 and a2 must both contain a sessions module record nCol
5255 ** fields in size. This function appends an nCol sessions module
5256 ** record to buffer pBuf that is a copy of a1, except that for
5257 ** each field that is undefined in a1[], swap in the field from a2[].
5258 */
sessionAppendRecordMerge(SessionBuffer * pBuf,int nCol,u8 * a1,int n1,u8 * a2,int n2,int * pRc)5259 static void sessionAppendRecordMerge(
5260 SessionBuffer *pBuf, /* Buffer to append to */
5261 int nCol, /* Number of columns in each record */
5262 u8 *a1, int n1, /* Record 1 */
5263 u8 *a2, int n2, /* Record 2 */
5264 int *pRc /* IN/OUT: error code */
5265 ){
5266 sessionBufferGrow(pBuf, n1+n2, pRc);
5267 if( *pRc==SQLITE_OK ){
5268 int i;
5269 u8 *pOut = &pBuf->aBuf[pBuf->nBuf];
5270 for(i=0; i<nCol; i++){
5271 int nn1 = sessionSerialLen(a1);
5272 int nn2 = sessionSerialLen(a2);
5273 if( *a1==0 || *a1==0xFF ){
5274 memcpy(pOut, a2, nn2);
5275 pOut += nn2;
5276 }else{
5277 memcpy(pOut, a1, nn1);
5278 pOut += nn1;
5279 }
5280 a1 += nn1;
5281 a2 += nn2;
5282 }
5283
5284 pBuf->nBuf = pOut-pBuf->aBuf;
5285 assert( pBuf->nBuf<=pBuf->nAlloc );
5286 }
5287 }
5288
5289 /*
5290 ** This function is called when rebasing a local UPDATE change against one
5291 ** or more remote UPDATE changes. The aRec/nRec buffer contains the current
5292 ** old.* and new.* records for the change. The rebase buffer (a single
5293 ** record) is in aChange/nChange. The rebased change is appended to buffer
5294 ** pBuf.
5295 **
5296 ** Rebasing the UPDATE involves:
5297 **
5298 ** * Removing any changes to fields for which the corresponding field
5299 ** in the rebase buffer is set to "replaced" (type 0xFF). If this
5300 ** means the UPDATE change updates no fields, nothing is appended
5301 ** to the output buffer.
5302 **
5303 ** * For each field modified by the local change for which the
5304 ** corresponding field in the rebase buffer is not "undefined" (0x00)
5305 ** or "replaced" (0xFF), the old.* value is replaced by the value
5306 ** in the rebase buffer.
5307 */
sessionAppendPartialUpdate(SessionBuffer * pBuf,sqlite3_changeset_iter * pIter,u8 * aRec,int nRec,u8 * aChange,int nChange,int * pRc)5308 static void sessionAppendPartialUpdate(
5309 SessionBuffer *pBuf, /* Append record here */
5310 sqlite3_changeset_iter *pIter, /* Iterator pointed at local change */
5311 u8 *aRec, int nRec, /* Local change */
5312 u8 *aChange, int nChange, /* Record to rebase against */
5313 int *pRc /* IN/OUT: Return Code */
5314 ){
5315 sessionBufferGrow(pBuf, 2+nRec+nChange, pRc);
5316 if( *pRc==SQLITE_OK ){
5317 int bData = 0;
5318 u8 *pOut = &pBuf->aBuf[pBuf->nBuf];
5319 int i;
5320 u8 *a1 = aRec;
5321 u8 *a2 = aChange;
5322
5323 *pOut++ = SQLITE_UPDATE;
5324 *pOut++ = pIter->bIndirect;
5325 for(i=0; i<pIter->nCol; i++){
5326 int n1 = sessionSerialLen(a1);
5327 int n2 = sessionSerialLen(a2);
5328 if( pIter->abPK[i] || a2[0]==0 ){
5329 if( !pIter->abPK[i] && a1[0] ) bData = 1;
5330 memcpy(pOut, a1, n1);
5331 pOut += n1;
5332 }else if( a2[0]!=0xFF ){
5333 bData = 1;
5334 memcpy(pOut, a2, n2);
5335 pOut += n2;
5336 }else{
5337 *pOut++ = '\0';
5338 }
5339 a1 += n1;
5340 a2 += n2;
5341 }
5342 if( bData ){
5343 a2 = aChange;
5344 for(i=0; i<pIter->nCol; i++){
5345 int n1 = sessionSerialLen(a1);
5346 int n2 = sessionSerialLen(a2);
5347 if( pIter->abPK[i] || a2[0]!=0xFF ){
5348 memcpy(pOut, a1, n1);
5349 pOut += n1;
5350 }else{
5351 *pOut++ = '\0';
5352 }
5353 a1 += n1;
5354 a2 += n2;
5355 }
5356 pBuf->nBuf = (pOut - pBuf->aBuf);
5357 }
5358 }
5359 }
5360
5361 /*
5362 ** pIter is configured to iterate through a changeset. This function rebases
5363 ** that changeset according to the current configuration of the rebaser
5364 ** object passed as the first argument. If no error occurs and argument xOutput
5365 ** is not NULL, then the changeset is returned to the caller by invoking
5366 ** xOutput zero or more times and SQLITE_OK returned. Or, if xOutput is NULL,
5367 ** then (*ppOut) is set to point to a buffer containing the rebased changeset
5368 ** before this function returns. In this case (*pnOut) is set to the size of
5369 ** the buffer in bytes. It is the responsibility of the caller to eventually
5370 ** free the (*ppOut) buffer using sqlite3_free().
5371 **
5372 ** If an error occurs, an SQLite error code is returned. If ppOut and
5373 ** pnOut are not NULL, then the two output parameters are set to 0 before
5374 ** returning.
5375 */
sessionRebase(sqlite3_rebaser * p,sqlite3_changeset_iter * pIter,int (* xOutput)(void * pOut,const void * pData,int nData),void * pOut,int * pnOut,void ** ppOut)5376 static int sessionRebase(
5377 sqlite3_rebaser *p, /* Rebaser hash table */
5378 sqlite3_changeset_iter *pIter, /* Input data */
5379 int (*xOutput)(void *pOut, const void *pData, int nData),
5380 void *pOut, /* Context for xOutput callback */
5381 int *pnOut, /* OUT: Number of bytes in output changeset */
5382 void **ppOut /* OUT: Inverse of pChangeset */
5383 ){
5384 int rc = SQLITE_OK;
5385 u8 *aRec = 0;
5386 int nRec = 0;
5387 int bNew = 0;
5388 SessionTable *pTab = 0;
5389 SessionBuffer sOut = {0,0,0};
5390
5391 while( SQLITE_ROW==sessionChangesetNext(pIter, &aRec, &nRec, &bNew) ){
5392 SessionChange *pChange = 0;
5393 int bDone = 0;
5394
5395 if( bNew ){
5396 const char *zTab = pIter->zTab;
5397 for(pTab=p->grp.pList; pTab; pTab=pTab->pNext){
5398 if( 0==sqlite3_stricmp(pTab->zName, zTab) ) break;
5399 }
5400 bNew = 0;
5401
5402 /* A patchset may not be rebased */
5403 if( pIter->bPatchset ){
5404 rc = SQLITE_ERROR;
5405 }
5406
5407 /* Append a table header to the output for this new table */
5408 sessionAppendByte(&sOut, pIter->bPatchset ? 'P' : 'T', &rc);
5409 sessionAppendVarint(&sOut, pIter->nCol, &rc);
5410 sessionAppendBlob(&sOut, pIter->abPK, pIter->nCol, &rc);
5411 sessionAppendBlob(&sOut,(u8*)pIter->zTab,(int)strlen(pIter->zTab)+1,&rc);
5412 }
5413
5414 if( pTab && rc==SQLITE_OK ){
5415 int iHash = sessionChangeHash(pTab, 0, aRec, pTab->nChange);
5416
5417 for(pChange=pTab->apChange[iHash]; pChange; pChange=pChange->pNext){
5418 if( sessionChangeEqual(pTab, 0, aRec, 0, pChange->aRecord) ){
5419 break;
5420 }
5421 }
5422 }
5423
5424 if( pChange ){
5425 assert( pChange->op==SQLITE_DELETE || pChange->op==SQLITE_INSERT );
5426 switch( pIter->op ){
5427 case SQLITE_INSERT:
5428 if( pChange->op==SQLITE_INSERT ){
5429 bDone = 1;
5430 if( pChange->bIndirect==0 ){
5431 sessionAppendByte(&sOut, SQLITE_UPDATE, &rc);
5432 sessionAppendByte(&sOut, pIter->bIndirect, &rc);
5433 sessionAppendBlob(&sOut, pChange->aRecord, pChange->nRecord, &rc);
5434 sessionAppendBlob(&sOut, aRec, nRec, &rc);
5435 }
5436 }
5437 break;
5438
5439 case SQLITE_UPDATE:
5440 bDone = 1;
5441 if( pChange->op==SQLITE_DELETE ){
5442 if( pChange->bIndirect==0 ){
5443 u8 *pCsr = aRec;
5444 sessionSkipRecord(&pCsr, pIter->nCol);
5445 sessionAppendByte(&sOut, SQLITE_INSERT, &rc);
5446 sessionAppendByte(&sOut, pIter->bIndirect, &rc);
5447 sessionAppendRecordMerge(&sOut, pIter->nCol,
5448 pCsr, nRec-(pCsr-aRec),
5449 pChange->aRecord, pChange->nRecord, &rc
5450 );
5451 }
5452 }else{
5453 sessionAppendPartialUpdate(&sOut, pIter,
5454 aRec, nRec, pChange->aRecord, pChange->nRecord, &rc
5455 );
5456 }
5457 break;
5458
5459 default:
5460 assert( pIter->op==SQLITE_DELETE );
5461 bDone = 1;
5462 if( pChange->op==SQLITE_INSERT ){
5463 sessionAppendByte(&sOut, SQLITE_DELETE, &rc);
5464 sessionAppendByte(&sOut, pIter->bIndirect, &rc);
5465 sessionAppendRecordMerge(&sOut, pIter->nCol,
5466 pChange->aRecord, pChange->nRecord, aRec, nRec, &rc
5467 );
5468 }
5469 break;
5470 }
5471 }
5472
5473 if( bDone==0 ){
5474 sessionAppendByte(&sOut, pIter->op, &rc);
5475 sessionAppendByte(&sOut, pIter->bIndirect, &rc);
5476 sessionAppendBlob(&sOut, aRec, nRec, &rc);
5477 }
5478 if( rc==SQLITE_OK && xOutput && sOut.nBuf>sessions_strm_chunk_size ){
5479 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf);
5480 sOut.nBuf = 0;
5481 }
5482 if( rc ) break;
5483 }
5484
5485 if( rc!=SQLITE_OK ){
5486 sqlite3_free(sOut.aBuf);
5487 memset(&sOut, 0, sizeof(sOut));
5488 }
5489
5490 if( rc==SQLITE_OK ){
5491 if( xOutput ){
5492 if( sOut.nBuf>0 ){
5493 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf);
5494 }
5495 }else{
5496 *ppOut = (void*)sOut.aBuf;
5497 *pnOut = sOut.nBuf;
5498 sOut.aBuf = 0;
5499 }
5500 }
5501 sqlite3_free(sOut.aBuf);
5502 return rc;
5503 }
5504
5505 /*
5506 ** Create a new rebaser object.
5507 */
sqlite3rebaser_create(sqlite3_rebaser ** ppNew)5508 int sqlite3rebaser_create(sqlite3_rebaser **ppNew){
5509 int rc = SQLITE_OK;
5510 sqlite3_rebaser *pNew;
5511
5512 pNew = sqlite3_malloc(sizeof(sqlite3_rebaser));
5513 if( pNew==0 ){
5514 rc = SQLITE_NOMEM;
5515 }else{
5516 memset(pNew, 0, sizeof(sqlite3_rebaser));
5517 }
5518 *ppNew = pNew;
5519 return rc;
5520 }
5521
5522 /*
5523 ** Call this one or more times to configure a rebaser.
5524 */
sqlite3rebaser_configure(sqlite3_rebaser * p,int nRebase,const void * pRebase)5525 int sqlite3rebaser_configure(
5526 sqlite3_rebaser *p,
5527 int nRebase, const void *pRebase
5528 ){
5529 sqlite3_changeset_iter *pIter = 0; /* Iterator opened on pData/nData */
5530 int rc; /* Return code */
5531 rc = sqlite3changeset_start(&pIter, nRebase, (void*)pRebase);
5532 if( rc==SQLITE_OK ){
5533 rc = sessionChangesetToHash(pIter, &p->grp, 1);
5534 }
5535 sqlite3changeset_finalize(pIter);
5536 return rc;
5537 }
5538
5539 /*
5540 ** Rebase a changeset according to current rebaser configuration
5541 */
sqlite3rebaser_rebase(sqlite3_rebaser * p,int nIn,const void * pIn,int * pnOut,void ** ppOut)5542 int sqlite3rebaser_rebase(
5543 sqlite3_rebaser *p,
5544 int nIn, const void *pIn,
5545 int *pnOut, void **ppOut
5546 ){
5547 sqlite3_changeset_iter *pIter = 0; /* Iterator to skip through input */
5548 int rc = sqlite3changeset_start(&pIter, nIn, (void*)pIn);
5549
5550 if( rc==SQLITE_OK ){
5551 rc = sessionRebase(p, pIter, 0, 0, pnOut, ppOut);
5552 sqlite3changeset_finalize(pIter);
5553 }
5554
5555 return rc;
5556 }
5557
5558 /*
5559 ** Rebase a changeset according to current rebaser configuration
5560 */
sqlite3rebaser_rebase_strm(sqlite3_rebaser * p,int (* xInput)(void * pIn,void * pData,int * pnData),void * pIn,int (* xOutput)(void * pOut,const void * pData,int nData),void * pOut)5561 int sqlite3rebaser_rebase_strm(
5562 sqlite3_rebaser *p,
5563 int (*xInput)(void *pIn, void *pData, int *pnData),
5564 void *pIn,
5565 int (*xOutput)(void *pOut, const void *pData, int nData),
5566 void *pOut
5567 ){
5568 sqlite3_changeset_iter *pIter = 0; /* Iterator to skip through input */
5569 int rc = sqlite3changeset_start_strm(&pIter, xInput, pIn);
5570
5571 if( rc==SQLITE_OK ){
5572 rc = sessionRebase(p, pIter, xOutput, pOut, 0, 0);
5573 sqlite3changeset_finalize(pIter);
5574 }
5575
5576 return rc;
5577 }
5578
5579 /*
5580 ** Destroy a rebaser object
5581 */
sqlite3rebaser_delete(sqlite3_rebaser * p)5582 void sqlite3rebaser_delete(sqlite3_rebaser *p){
5583 if( p ){
5584 sessionDeleteTable(0, p->grp.pList);
5585 sqlite3_free(p);
5586 }
5587 }
5588
5589 /*
5590 ** Global configuration
5591 */
sqlite3session_config(int op,void * pArg)5592 int sqlite3session_config(int op, void *pArg){
5593 int rc = SQLITE_OK;
5594 switch( op ){
5595 case SQLITE_SESSION_CONFIG_STRMSIZE: {
5596 int *pInt = (int*)pArg;
5597 if( *pInt>0 ){
5598 sessions_strm_chunk_size = *pInt;
5599 }
5600 *pInt = sessions_strm_chunk_size;
5601 break;
5602 }
5603 default:
5604 rc = SQLITE_MISUSE;
5605 break;
5606 }
5607 return rc;
5608 }
5609
5610 #endif /* SQLITE_ENABLE_SESSION && SQLITE_ENABLE_PREUPDATE_HOOK */
5611