Add a "flags" parameter to experimental API sqlite3changeset_apply_v2(). Also
[sqlite.git] / ext / session / sqlite3session.c
blobe432b894ca5cf7ae8bc2306809203337065f8929
2 #if defined(SQLITE_ENABLE_SESSION) && defined(SQLITE_ENABLE_PREUPDATE_HOOK)
3 #include "sqlite3session.h"
4 #include <assert.h>
5 #include <string.h>
7 #ifndef SQLITE_AMALGAMATION
8 # include "sqliteInt.h"
9 # include "vdbeInt.h"
10 #endif
12 typedef struct SessionTable SessionTable;
13 typedef struct SessionChange SessionChange;
14 typedef struct SessionBuffer SessionBuffer;
15 typedef struct SessionInput SessionInput;
18 ** Minimum chunk size used by streaming versions of functions.
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
28 typedef struct SessionHook SessionHook;
29 struct SessionHook {
30 void *pCtx;
31 int (*xOld)(void*,int,sqlite3_value**);
32 int (*xNew)(void*,int,sqlite3_value**);
33 int (*xCount)(void*);
34 int (*xDepth)(void*);
38 ** Session handle structure.
40 struct sqlite3_session {
41 sqlite3 *db; /* Database handle session is attached to */
42 char *zDb; /* Name of database session is attached to */
43 int bEnable; /* True if currently recording */
44 int bIndirect; /* True if all changes are indirect */
45 int bAutoAttach; /* True to auto-attach tables */
46 int rc; /* Non-zero if an error has occurred */
47 void *pFilterCtx; /* First argument to pass to xTableFilter */
48 int (*xTableFilter)(void *pCtx, const char *zTab);
49 sqlite3_value *pZeroBlob; /* Value containing X'' */
50 sqlite3_session *pNext; /* Next session object on same db. */
51 SessionTable *pTable; /* List of attached tables */
52 SessionHook hook; /* APIs to grab new and old data with */
56 ** Instances of this structure are used to build strings or binary records.
58 struct SessionBuffer {
59 u8 *aBuf; /* Pointer to changeset buffer */
60 int nBuf; /* Size of buffer aBuf */
61 int nAlloc; /* Size of allocation containing aBuf */
65 ** An object of this type is used internally as an abstraction for
66 ** input data. Input data may be supplied either as a single large buffer
67 ** (e.g. sqlite3changeset_start()) or using a stream function (e.g.
68 ** sqlite3changeset_start_strm()).
70 struct SessionInput {
71 int bNoDiscard; /* If true, do not discard in InputBuffer() */
72 int iCurrent; /* Offset in aData[] of current change */
73 int iNext; /* Offset in aData[] of next change */
74 u8 *aData; /* Pointer to buffer containing changeset */
75 int nData; /* Number of bytes in aData */
77 SessionBuffer buf; /* Current read buffer */
78 int (*xInput)(void*, void*, int*); /* Input stream call (or NULL) */
79 void *pIn; /* First argument to xInput */
80 int bEof; /* Set to true after xInput finished */
84 ** Structure for changeset iterators.
86 struct sqlite3_changeset_iter {
87 SessionInput in; /* Input buffer or stream */
88 SessionBuffer tblhdr; /* Buffer to hold apValue/zTab/abPK/ */
89 int bPatchset; /* True if this is a patchset */
90 int rc; /* Iterator error code */
91 sqlite3_stmt *pConflict; /* Points to conflicting row, if any */
92 char *zTab; /* Current table */
93 int nCol; /* Number of columns in zTab */
94 int op; /* Current operation */
95 int bIndirect; /* True if current change was indirect */
96 u8 *abPK; /* Primary key array */
97 sqlite3_value **apValue; /* old.* and new.* values */
101 ** Each session object maintains a set of the following structures, one
102 ** for each table the session object is monitoring. The structures are
103 ** stored in a linked list starting at sqlite3_session.pTable.
105 ** The keys of the SessionTable.aChange[] hash table are all rows that have
106 ** been modified in any way since the session object was attached to the
107 ** table.
109 ** The data associated with each hash-table entry is a structure containing
110 ** a subset of the initial values that the modified row contained at the
111 ** start of the session. Or no initial values if the row was inserted.
113 struct SessionTable {
114 SessionTable *pNext;
115 char *zName; /* Local name of table */
116 int nCol; /* Number of columns in table zName */
117 int bStat1; /* True if this is sqlite_stat1 */
118 const char **azCol; /* Column names */
119 u8 *abPK; /* Array of primary key flags */
120 int nEntry; /* Total number of entries in hash table */
121 int nChange; /* Size of apChange[] array */
122 SessionChange **apChange; /* Hash table buckets */
126 ** RECORD FORMAT:
128 ** The following record format is similar to (but not compatible with) that
129 ** used in SQLite database files. This format is used as part of the
130 ** change-set binary format, and so must be architecture independent.
132 ** Unlike the SQLite database record format, each field is self-contained -
133 ** there is no separation of header and data. Each field begins with a
134 ** single byte describing its type, as follows:
136 ** 0x00: Undefined value.
137 ** 0x01: Integer value.
138 ** 0x02: Real value.
139 ** 0x03: Text value.
140 ** 0x04: Blob value.
141 ** 0x05: SQL NULL value.
143 ** Note that the above match the definitions of SQLITE_INTEGER, SQLITE_TEXT
144 ** and so on in sqlite3.h. For undefined and NULL values, the field consists
145 ** only of the single type byte. For other types of values, the type byte
146 ** is followed by:
148 ** Text values:
149 ** A varint containing the number of bytes in the value (encoded using
150 ** UTF-8). Followed by a buffer containing the UTF-8 representation
151 ** of the text value. There is no nul terminator.
153 ** Blob values:
154 ** A varint containing the number of bytes in the value, followed by
155 ** a buffer containing the value itself.
157 ** Integer values:
158 ** An 8-byte big-endian integer value.
160 ** Real values:
161 ** An 8-byte big-endian IEEE 754-2008 real value.
163 ** Varint values are encoded in the same way as varints in the SQLite
164 ** record format.
166 ** CHANGESET FORMAT:
168 ** A changeset is a collection of DELETE, UPDATE and INSERT operations on
169 ** one or more tables. Operations on a single table are grouped together,
170 ** but may occur in any order (i.e. deletes, updates and inserts are all
171 ** mixed together).
173 ** Each group of changes begins with a table header:
175 ** 1 byte: Constant 0x54 (capital 'T')
176 ** Varint: Number of columns in the table.
177 ** nCol bytes: 0x01 for PK columns, 0x00 otherwise.
178 ** N bytes: Unqualified table name (encoded using UTF-8). Nul-terminated.
180 ** Followed by one or more changes to the table.
182 ** 1 byte: Either SQLITE_INSERT (0x12), UPDATE (0x17) or DELETE (0x09).
183 ** 1 byte: The "indirect-change" flag.
184 ** old.* record: (delete and update only)
185 ** new.* record: (insert and update only)
187 ** The "old.*" and "new.*" records, if present, are N field records in the
188 ** format described above under "RECORD FORMAT", where N is the number of
189 ** columns in the table. The i'th field of each record is associated with
190 ** the i'th column of the table, counting from left to right in the order
191 ** in which columns were declared in the CREATE TABLE statement.
193 ** The new.* record that is part of each INSERT change contains the values
194 ** that make up the new row. Similarly, the old.* record that is part of each
195 ** DELETE change contains the values that made up the row that was deleted
196 ** from the database. In the changeset format, the records that are part
197 ** of INSERT or DELETE changes never contain any undefined (type byte 0x00)
198 ** fields.
200 ** Within the old.* record associated with an UPDATE change, all fields
201 ** associated with table columns that are not PRIMARY KEY columns and are
202 ** not modified by the UPDATE change are set to "undefined". Other fields
203 ** are set to the values that made up the row before the UPDATE that the
204 ** change records took place. Within the new.* record, fields associated
205 ** with table columns modified by the UPDATE change contain the new
206 ** values. Fields associated with table columns that are not modified
207 ** are set to "undefined".
209 ** PATCHSET FORMAT:
211 ** A patchset is also a collection of changes. It is similar to a changeset,
212 ** but leaves undefined those fields that are not useful if no conflict
213 ** resolution is required when applying the changeset.
215 ** Each group of changes begins with a table header:
217 ** 1 byte: Constant 0x50 (capital 'P')
218 ** Varint: Number of columns in the table.
219 ** nCol bytes: 0x01 for PK columns, 0x00 otherwise.
220 ** N bytes: Unqualified table name (encoded using UTF-8). Nul-terminated.
222 ** Followed by one or more changes to the table.
224 ** 1 byte: Either SQLITE_INSERT (0x12), UPDATE (0x17) or DELETE (0x09).
225 ** 1 byte: The "indirect-change" flag.
226 ** single record: (PK fields for DELETE, PK and modified fields for UPDATE,
227 ** full record for INSERT).
229 ** As in the changeset format, each field of the single record that is part
230 ** of a patchset change is associated with the correspondingly positioned
231 ** table column, counting from left to right within the CREATE TABLE
232 ** statement.
234 ** For a DELETE change, all fields within the record except those associated
235 ** with PRIMARY KEY columns are omitted. The PRIMARY KEY fields contain the
236 ** values identifying the row to delete.
238 ** For an UPDATE change, all fields except those associated with PRIMARY KEY
239 ** columns and columns that are modified by the UPDATE are set to "undefined".
240 ** PRIMARY KEY fields contain the values identifying the table row to update,
241 ** and fields associated with modified columns contain the new column values.
243 ** The records associated with INSERT changes are in the same format as for
244 ** changesets. It is not possible for a record associated with an INSERT
245 ** change to contain a field set to "undefined".
249 ** For each row modified during a session, there exists a single instance of
250 ** this structure stored in a SessionTable.aChange[] hash table.
252 struct SessionChange {
253 int op; /* One of UPDATE, DELETE, INSERT */
254 int bIndirect; /* True if this change is "indirect" */
255 int nRecord; /* Number of bytes in buffer aRecord[] */
256 u8 *aRecord; /* Buffer containing old.* record */
257 SessionChange *pNext; /* For hash-table collisions */
261 ** Write a varint with value iVal into the buffer at aBuf. Return the
262 ** number of bytes written.
264 static int sessionVarintPut(u8 *aBuf, int iVal){
265 return putVarint32(aBuf, iVal);
269 ** Return the number of bytes required to store value iVal as a varint.
271 static int sessionVarintLen(int iVal){
272 return sqlite3VarintLen(iVal);
276 ** Read a varint value from aBuf[] into *piVal. Return the number of
277 ** bytes read.
279 static int sessionVarintGet(u8 *aBuf, int *piVal){
280 return getVarint32(aBuf, *piVal);
283 /* Load an unaligned and unsigned 32-bit integer */
284 #define SESSION_UINT32(x) (((u32)(x)[0]<<24)|((x)[1]<<16)|((x)[2]<<8)|(x)[3])
287 ** Read a 64-bit big-endian integer value from buffer aRec[]. Return
288 ** the value read.
290 static sqlite3_int64 sessionGetI64(u8 *aRec){
291 u64 x = SESSION_UINT32(aRec);
292 u32 y = SESSION_UINT32(aRec+4);
293 x = (x<<32) + y;
294 return (sqlite3_int64)x;
298 ** Write a 64-bit big-endian integer value to the buffer aBuf[].
300 static void sessionPutI64(u8 *aBuf, sqlite3_int64 i){
301 aBuf[0] = (i>>56) & 0xFF;
302 aBuf[1] = (i>>48) & 0xFF;
303 aBuf[2] = (i>>40) & 0xFF;
304 aBuf[3] = (i>>32) & 0xFF;
305 aBuf[4] = (i>>24) & 0xFF;
306 aBuf[5] = (i>>16) & 0xFF;
307 aBuf[6] = (i>> 8) & 0xFF;
308 aBuf[7] = (i>> 0) & 0xFF;
312 ** This function is used to serialize the contents of value pValue (see
313 ** comment titled "RECORD FORMAT" above).
315 ** If it is non-NULL, the serialized form of the value is written to
316 ** buffer aBuf. *pnWrite is set to the number of bytes written before
317 ** returning. Or, if aBuf is NULL, the only thing this function does is
318 ** set *pnWrite.
320 ** If no error occurs, SQLITE_OK is returned. Or, if an OOM error occurs
321 ** within a call to sqlite3_value_text() (may fail if the db is utf-16))
322 ** SQLITE_NOMEM is returned.
324 static int sessionSerializeValue(
325 u8 *aBuf, /* If non-NULL, write serialized value here */
326 sqlite3_value *pValue, /* Value to serialize */
327 int *pnWrite /* IN/OUT: Increment by bytes written */
329 int nByte; /* Size of serialized value in bytes */
331 if( pValue ){
332 int eType; /* Value type (SQLITE_NULL, TEXT etc.) */
334 eType = sqlite3_value_type(pValue);
335 if( aBuf ) aBuf[0] = eType;
337 switch( eType ){
338 case SQLITE_NULL:
339 nByte = 1;
340 break;
342 case SQLITE_INTEGER:
343 case SQLITE_FLOAT:
344 if( aBuf ){
345 /* TODO: SQLite does something special to deal with mixed-endian
346 ** floating point values (e.g. ARM7). This code probably should
347 ** too. */
348 u64 i;
349 if( eType==SQLITE_INTEGER ){
350 i = (u64)sqlite3_value_int64(pValue);
351 }else{
352 double r;
353 assert( sizeof(double)==8 && sizeof(u64)==8 );
354 r = sqlite3_value_double(pValue);
355 memcpy(&i, &r, 8);
357 sessionPutI64(&aBuf[1], i);
359 nByte = 9;
360 break;
362 default: {
363 u8 *z;
364 int n;
365 int nVarint;
367 assert( eType==SQLITE_TEXT || eType==SQLITE_BLOB );
368 if( eType==SQLITE_TEXT ){
369 z = (u8 *)sqlite3_value_text(pValue);
370 }else{
371 z = (u8 *)sqlite3_value_blob(pValue);
373 n = sqlite3_value_bytes(pValue);
374 if( z==0 && (eType!=SQLITE_BLOB || n>0) ) return SQLITE_NOMEM;
375 nVarint = sessionVarintLen(n);
377 if( aBuf ){
378 sessionVarintPut(&aBuf[1], n);
379 if( n ) memcpy(&aBuf[nVarint + 1], z, n);
382 nByte = 1 + nVarint + n;
383 break;
386 }else{
387 nByte = 1;
388 if( aBuf ) aBuf[0] = '\0';
391 if( pnWrite ) *pnWrite += nByte;
392 return SQLITE_OK;
397 ** This macro is used to calculate hash key values for data structures. In
398 ** order to use this macro, the entire data structure must be represented
399 ** as a series of unsigned integers. In order to calculate a hash-key value
400 ** for a data structure represented as three such integers, the macro may
401 ** then be used as follows:
403 ** int hash_key_value;
404 ** hash_key_value = HASH_APPEND(0, <value 1>);
405 ** hash_key_value = HASH_APPEND(hash_key_value, <value 2>);
406 ** hash_key_value = HASH_APPEND(hash_key_value, <value 3>);
408 ** In practice, the data structures this macro is used for are the primary
409 ** key values of modified rows.
411 #define HASH_APPEND(hash, add) ((hash) << 3) ^ (hash) ^ (unsigned int)(add)
414 ** Append the hash of the 64-bit integer passed as the second argument to the
415 ** hash-key value passed as the first. Return the new hash-key value.
417 static unsigned int sessionHashAppendI64(unsigned int h, i64 i){
418 h = HASH_APPEND(h, i & 0xFFFFFFFF);
419 return HASH_APPEND(h, (i>>32)&0xFFFFFFFF);
423 ** Append the hash of the blob passed via the second and third arguments to
424 ** the hash-key value passed as the first. Return the new hash-key value.
426 static unsigned int sessionHashAppendBlob(unsigned int h, int n, const u8 *z){
427 int i;
428 for(i=0; i<n; i++) h = HASH_APPEND(h, z[i]);
429 return h;
433 ** Append the hash of the data type passed as the second argument to the
434 ** hash-key value passed as the first. Return the new hash-key value.
436 static unsigned int sessionHashAppendType(unsigned int h, int eType){
437 return HASH_APPEND(h, eType);
441 ** This function may only be called from within a pre-update callback.
442 ** It calculates a hash based on the primary key values of the old.* or
443 ** new.* row currently available and, assuming no error occurs, writes it to
444 ** *piHash before returning. If the primary key contains one or more NULL
445 ** values, *pbNullPK is set to true before returning.
447 ** If an error occurs, an SQLite error code is returned and the final values
448 ** of *piHash asn *pbNullPK are undefined. Otherwise, SQLITE_OK is returned
449 ** and the output variables are set as described above.
451 static int sessionPreupdateHash(
452 sqlite3_session *pSession, /* Session object that owns pTab */
453 SessionTable *pTab, /* Session table handle */
454 int bNew, /* True to hash the new.* PK */
455 int *piHash, /* OUT: Hash value */
456 int *pbNullPK /* OUT: True if there are NULL values in PK */
458 unsigned int h = 0; /* Hash value to return */
459 int i; /* Used to iterate through columns */
461 assert( *pbNullPK==0 );
462 assert( pTab->nCol==pSession->hook.xCount(pSession->hook.pCtx) );
463 for(i=0; i<pTab->nCol; i++){
464 if( pTab->abPK[i] ){
465 int rc;
466 int eType;
467 sqlite3_value *pVal;
469 if( bNew ){
470 rc = pSession->hook.xNew(pSession->hook.pCtx, i, &pVal);
471 }else{
472 rc = pSession->hook.xOld(pSession->hook.pCtx, i, &pVal);
474 if( rc!=SQLITE_OK ) return rc;
476 eType = sqlite3_value_type(pVal);
477 h = sessionHashAppendType(h, eType);
478 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
479 i64 iVal;
480 if( eType==SQLITE_INTEGER ){
481 iVal = sqlite3_value_int64(pVal);
482 }else{
483 double rVal = sqlite3_value_double(pVal);
484 assert( sizeof(iVal)==8 && sizeof(rVal)==8 );
485 memcpy(&iVal, &rVal, 8);
487 h = sessionHashAppendI64(h, iVal);
488 }else if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
489 const u8 *z;
490 int n;
491 if( eType==SQLITE_TEXT ){
492 z = (const u8 *)sqlite3_value_text(pVal);
493 }else{
494 z = (const u8 *)sqlite3_value_blob(pVal);
496 n = sqlite3_value_bytes(pVal);
497 if( !z && (eType!=SQLITE_BLOB || n>0) ) return SQLITE_NOMEM;
498 h = sessionHashAppendBlob(h, n, z);
499 }else{
500 assert( eType==SQLITE_NULL );
501 assert( pTab->bStat1==0 || i!=1 );
502 *pbNullPK = 1;
507 *piHash = (h % pTab->nChange);
508 return SQLITE_OK;
512 ** The buffer that the argument points to contains a serialized SQL value.
513 ** Return the number of bytes of space occupied by the value (including
514 ** the type byte).
516 static int sessionSerialLen(u8 *a){
517 int e = *a;
518 int n;
519 if( e==0 || e==0xFF ) return 1;
520 if( e==SQLITE_NULL ) return 1;
521 if( e==SQLITE_INTEGER || e==SQLITE_FLOAT ) return 9;
522 return sessionVarintGet(&a[1], &n) + 1 + n;
526 ** Based on the primary key values stored in change aRecord, calculate a
527 ** hash key. Assume the has table has nBucket buckets. The hash keys
528 ** calculated by this function are compatible with those calculated by
529 ** sessionPreupdateHash().
531 ** The bPkOnly argument is non-zero if the record at aRecord[] is from
532 ** a patchset DELETE. In this case the non-PK fields are omitted entirely.
534 static unsigned int sessionChangeHash(
535 SessionTable *pTab, /* Table handle */
536 int bPkOnly, /* Record consists of PK fields only */
537 u8 *aRecord, /* Change record */
538 int nBucket /* Assume this many buckets in hash table */
540 unsigned int h = 0; /* Value to return */
541 int i; /* Used to iterate through columns */
542 u8 *a = aRecord; /* Used to iterate through change record */
544 for(i=0; i<pTab->nCol; i++){
545 int eType = *a;
546 int isPK = pTab->abPK[i];
547 if( bPkOnly && isPK==0 ) continue;
549 /* It is not possible for eType to be SQLITE_NULL here. The session
550 ** module does not record changes for rows with NULL values stored in
551 ** primary key columns. */
552 assert( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT
553 || eType==SQLITE_TEXT || eType==SQLITE_BLOB
554 || eType==SQLITE_NULL || eType==0
556 assert( !isPK || (eType!=0 && eType!=SQLITE_NULL) );
558 if( isPK ){
559 a++;
560 h = sessionHashAppendType(h, eType);
561 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
562 h = sessionHashAppendI64(h, sessionGetI64(a));
563 a += 8;
564 }else{
565 int n;
566 a += sessionVarintGet(a, &n);
567 h = sessionHashAppendBlob(h, n, a);
568 a += n;
570 }else{
571 a += sessionSerialLen(a);
574 return (h % nBucket);
578 ** Arguments aLeft and aRight are pointers to change records for table pTab.
579 ** This function returns true if the two records apply to the same row (i.e.
580 ** have the same values stored in the primary key columns), or false
581 ** otherwise.
583 static int sessionChangeEqual(
584 SessionTable *pTab, /* Table used for PK definition */
585 int bLeftPkOnly, /* True if aLeft[] contains PK fields only */
586 u8 *aLeft, /* Change record */
587 int bRightPkOnly, /* True if aRight[] contains PK fields only */
588 u8 *aRight /* Change record */
590 u8 *a1 = aLeft; /* Cursor to iterate through aLeft */
591 u8 *a2 = aRight; /* Cursor to iterate through aRight */
592 int iCol; /* Used to iterate through table columns */
594 for(iCol=0; iCol<pTab->nCol; iCol++){
595 if( pTab->abPK[iCol] ){
596 int n1 = sessionSerialLen(a1);
597 int n2 = sessionSerialLen(a2);
599 if( n1!=n2 || memcmp(a1, a2, n1) ){
600 return 0;
602 a1 += n1;
603 a2 += n2;
604 }else{
605 if( bLeftPkOnly==0 ) a1 += sessionSerialLen(a1);
606 if( bRightPkOnly==0 ) a2 += sessionSerialLen(a2);
610 return 1;
614 ** Arguments aLeft and aRight both point to buffers containing change
615 ** records with nCol columns. This function "merges" the two records into
616 ** a single records which is written to the buffer at *paOut. *paOut is
617 ** then set to point to one byte after the last byte written before
618 ** returning.
620 ** The merging of records is done as follows: For each column, if the
621 ** aRight record contains a value for the column, copy the value from
622 ** their. Otherwise, if aLeft contains a value, copy it. If neither
623 ** record contains a value for a given column, then neither does the
624 ** output record.
626 static void sessionMergeRecord(
627 u8 **paOut,
628 int nCol,
629 u8 *aLeft,
630 u8 *aRight
632 u8 *a1 = aLeft; /* Cursor used to iterate through aLeft */
633 u8 *a2 = aRight; /* Cursor used to iterate through aRight */
634 u8 *aOut = *paOut; /* Output cursor */
635 int iCol; /* Used to iterate from 0 to nCol */
637 for(iCol=0; iCol<nCol; iCol++){
638 int n1 = sessionSerialLen(a1);
639 int n2 = sessionSerialLen(a2);
640 if( *a2 ){
641 memcpy(aOut, a2, n2);
642 aOut += n2;
643 }else{
644 memcpy(aOut, a1, n1);
645 aOut += n1;
647 a1 += n1;
648 a2 += n2;
651 *paOut = aOut;
655 ** This is a helper function used by sessionMergeUpdate().
657 ** When this function is called, both *paOne and *paTwo point to a value
658 ** within a change record. Before it returns, both have been advanced so
659 ** as to point to the next value in the record.
661 ** If, when this function is called, *paTwo points to a valid value (i.e.
662 ** *paTwo[0] is not 0x00 - the "no value" placeholder), a copy of the *paTwo
663 ** pointer is returned and *pnVal is set to the number of bytes in the
664 ** serialized value. Otherwise, a copy of *paOne is returned and *pnVal
665 ** set to the number of bytes in the value at *paOne. If *paOne points
666 ** to the "no value" placeholder, *pnVal is set to 1. In other words:
668 ** if( *paTwo is valid ) return *paTwo;
669 ** return *paOne;
672 static u8 *sessionMergeValue(
673 u8 **paOne, /* IN/OUT: Left-hand buffer pointer */
674 u8 **paTwo, /* IN/OUT: Right-hand buffer pointer */
675 int *pnVal /* OUT: Bytes in returned value */
677 u8 *a1 = *paOne;
678 u8 *a2 = *paTwo;
679 u8 *pRet = 0;
680 int n1;
682 assert( a1 );
683 if( a2 ){
684 int n2 = sessionSerialLen(a2);
685 if( *a2 ){
686 *pnVal = n2;
687 pRet = a2;
689 *paTwo = &a2[n2];
692 n1 = sessionSerialLen(a1);
693 if( pRet==0 ){
694 *pnVal = n1;
695 pRet = a1;
697 *paOne = &a1[n1];
699 return pRet;
703 ** This function is used by changeset_concat() to merge two UPDATE changes
704 ** on the same row.
706 static int sessionMergeUpdate(
707 u8 **paOut, /* IN/OUT: Pointer to output buffer */
708 SessionTable *pTab, /* Table change pertains to */
709 int bPatchset, /* True if records are patchset records */
710 u8 *aOldRecord1, /* old.* record for first change */
711 u8 *aOldRecord2, /* old.* record for second change */
712 u8 *aNewRecord1, /* new.* record for first change */
713 u8 *aNewRecord2 /* new.* record for second change */
715 u8 *aOld1 = aOldRecord1;
716 u8 *aOld2 = aOldRecord2;
717 u8 *aNew1 = aNewRecord1;
718 u8 *aNew2 = aNewRecord2;
720 u8 *aOut = *paOut;
721 int i;
723 if( bPatchset==0 ){
724 int bRequired = 0;
726 assert( aOldRecord1 && aNewRecord1 );
728 /* Write the old.* vector first. */
729 for(i=0; i<pTab->nCol; i++){
730 int nOld;
731 u8 *aOld;
732 int nNew;
733 u8 *aNew;
735 aOld = sessionMergeValue(&aOld1, &aOld2, &nOld);
736 aNew = sessionMergeValue(&aNew1, &aNew2, &nNew);
737 if( pTab->abPK[i] || nOld!=nNew || memcmp(aOld, aNew, nNew) ){
738 if( pTab->abPK[i]==0 ) bRequired = 1;
739 memcpy(aOut, aOld, nOld);
740 aOut += nOld;
741 }else{
742 *(aOut++) = '\0';
746 if( !bRequired ) return 0;
749 /* Write the new.* vector */
750 aOld1 = aOldRecord1;
751 aOld2 = aOldRecord2;
752 aNew1 = aNewRecord1;
753 aNew2 = aNewRecord2;
754 for(i=0; i<pTab->nCol; i++){
755 int nOld;
756 u8 *aOld;
757 int nNew;
758 u8 *aNew;
760 aOld = sessionMergeValue(&aOld1, &aOld2, &nOld);
761 aNew = sessionMergeValue(&aNew1, &aNew2, &nNew);
762 if( bPatchset==0
763 && (pTab->abPK[i] || (nOld==nNew && 0==memcmp(aOld, aNew, nNew)))
765 *(aOut++) = '\0';
766 }else{
767 memcpy(aOut, aNew, nNew);
768 aOut += nNew;
772 *paOut = aOut;
773 return 1;
777 ** This function is only called from within a pre-update-hook callback.
778 ** It determines if the current pre-update-hook change affects the same row
779 ** as the change stored in argument pChange. If so, it returns true. Otherwise
780 ** if the pre-update-hook does not affect the same row as pChange, it returns
781 ** false.
783 static int sessionPreupdateEqual(
784 sqlite3_session *pSession, /* Session object that owns SessionTable */
785 SessionTable *pTab, /* Table associated with change */
786 SessionChange *pChange, /* Change to compare to */
787 int op /* Current pre-update operation */
789 int iCol; /* Used to iterate through columns */
790 u8 *a = pChange->aRecord; /* Cursor used to scan change record */
792 assert( op==SQLITE_INSERT || op==SQLITE_UPDATE || op==SQLITE_DELETE );
793 for(iCol=0; iCol<pTab->nCol; iCol++){
794 if( !pTab->abPK[iCol] ){
795 a += sessionSerialLen(a);
796 }else{
797 sqlite3_value *pVal; /* Value returned by preupdate_new/old */
798 int rc; /* Error code from preupdate_new/old */
799 int eType = *a++; /* Type of value from change record */
801 /* The following calls to preupdate_new() and preupdate_old() can not
802 ** fail. This is because they cache their return values, and by the
803 ** time control flows to here they have already been called once from
804 ** within sessionPreupdateHash(). The first two asserts below verify
805 ** this (that the method has already been called). */
806 if( op==SQLITE_INSERT ){
807 /* assert( db->pPreUpdate->pNewUnpacked || db->pPreUpdate->aNew ); */
808 rc = pSession->hook.xNew(pSession->hook.pCtx, iCol, &pVal);
809 }else{
810 /* assert( db->pPreUpdate->pUnpacked ); */
811 rc = pSession->hook.xOld(pSession->hook.pCtx, iCol, &pVal);
813 assert( rc==SQLITE_OK );
814 if( sqlite3_value_type(pVal)!=eType ) return 0;
816 /* A SessionChange object never has a NULL value in a PK column */
817 assert( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT
818 || eType==SQLITE_BLOB || eType==SQLITE_TEXT
821 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
822 i64 iVal = sessionGetI64(a);
823 a += 8;
824 if( eType==SQLITE_INTEGER ){
825 if( sqlite3_value_int64(pVal)!=iVal ) return 0;
826 }else{
827 double rVal;
828 assert( sizeof(iVal)==8 && sizeof(rVal)==8 );
829 memcpy(&rVal, &iVal, 8);
830 if( sqlite3_value_double(pVal)!=rVal ) return 0;
832 }else{
833 int n;
834 const u8 *z;
835 a += sessionVarintGet(a, &n);
836 if( sqlite3_value_bytes(pVal)!=n ) return 0;
837 if( eType==SQLITE_TEXT ){
838 z = sqlite3_value_text(pVal);
839 }else{
840 z = sqlite3_value_blob(pVal);
842 if( n>0 && memcmp(a, z, n) ) return 0;
843 a += n;
848 return 1;
852 ** If required, grow the hash table used to store changes on table pTab
853 ** (part of the session pSession). If a fatal OOM error occurs, set the
854 ** session object to failed and return SQLITE_ERROR. Otherwise, return
855 ** SQLITE_OK.
857 ** It is possible that a non-fatal OOM error occurs in this function. In
858 ** that case the hash-table does not grow, but SQLITE_OK is returned anyway.
859 ** Growing the hash table in this case is a performance optimization only,
860 ** it is not required for correct operation.
862 static int sessionGrowHash(int bPatchset, SessionTable *pTab){
863 if( pTab->nChange==0 || pTab->nEntry>=(pTab->nChange/2) ){
864 int i;
865 SessionChange **apNew;
866 int nNew = (pTab->nChange ? pTab->nChange : 128) * 2;
868 apNew = (SessionChange **)sqlite3_malloc(sizeof(SessionChange *) * nNew);
869 if( apNew==0 ){
870 if( pTab->nChange==0 ){
871 return SQLITE_ERROR;
873 return SQLITE_OK;
875 memset(apNew, 0, sizeof(SessionChange *) * nNew);
877 for(i=0; i<pTab->nChange; i++){
878 SessionChange *p;
879 SessionChange *pNext;
880 for(p=pTab->apChange[i]; p; p=pNext){
881 int bPkOnly = (p->op==SQLITE_DELETE && bPatchset);
882 int iHash = sessionChangeHash(pTab, bPkOnly, p->aRecord, nNew);
883 pNext = p->pNext;
884 p->pNext = apNew[iHash];
885 apNew[iHash] = p;
889 sqlite3_free(pTab->apChange);
890 pTab->nChange = nNew;
891 pTab->apChange = apNew;
894 return SQLITE_OK;
898 ** This function queries the database for the names of the columns of table
899 ** zThis, in schema zDb.
901 ** Otherwise, if they are not NULL, variable *pnCol is set to the number
902 ** of columns in the database table and variable *pzTab is set to point to a
903 ** nul-terminated copy of the table name. *pazCol (if not NULL) is set to
904 ** point to an array of pointers to column names. And *pabPK (again, if not
905 ** NULL) is set to point to an array of booleans - true if the corresponding
906 ** column is part of the primary key.
908 ** For example, if the table is declared as:
910 ** CREATE TABLE tbl1(w, x, y, z, PRIMARY KEY(w, z));
912 ** Then the four output variables are populated as follows:
914 ** *pnCol = 4
915 ** *pzTab = "tbl1"
916 ** *pazCol = {"w", "x", "y", "z"}
917 ** *pabPK = {1, 0, 0, 1}
919 ** All returned buffers are part of the same single allocation, which must
920 ** be freed using sqlite3_free() by the caller
922 static int sessionTableInfo(
923 sqlite3 *db, /* Database connection */
924 const char *zDb, /* Name of attached database (e.g. "main") */
925 const char *zThis, /* Table name */
926 int *pnCol, /* OUT: number of columns */
927 const char **pzTab, /* OUT: Copy of zThis */
928 const char ***pazCol, /* OUT: Array of column names for table */
929 u8 **pabPK /* OUT: Array of booleans - true for PK col */
931 char *zPragma;
932 sqlite3_stmt *pStmt;
933 int rc;
934 int nByte;
935 int nDbCol = 0;
936 int nThis;
937 int i;
938 u8 *pAlloc = 0;
939 char **azCol = 0;
940 u8 *abPK = 0;
942 assert( pazCol && pabPK );
944 nThis = sqlite3Strlen30(zThis);
945 if( nThis==12 && 0==sqlite3_stricmp("sqlite_stat1", zThis) ){
946 rc = sqlite3_table_column_metadata(db, zDb, zThis, 0, 0, 0, 0, 0, 0);
947 if( rc==SQLITE_OK ){
948 /* For sqlite_stat1, pretend that (tbl,idx) is the PRIMARY KEY. */
949 zPragma = sqlite3_mprintf(
950 "SELECT 0, 'tbl', '', 0, '', 1 UNION ALL "
951 "SELECT 1, 'idx', '', 0, '', 2 UNION ALL "
952 "SELECT 2, 'stat', '', 0, '', 0"
954 }else if( rc==SQLITE_ERROR ){
955 zPragma = sqlite3_mprintf("");
956 }else{
957 return rc;
959 }else{
960 zPragma = sqlite3_mprintf("PRAGMA '%q'.table_info('%q')", zDb, zThis);
962 if( !zPragma ) return SQLITE_NOMEM;
964 rc = sqlite3_prepare_v2(db, zPragma, -1, &pStmt, 0);
965 sqlite3_free(zPragma);
966 if( rc!=SQLITE_OK ) return rc;
968 nByte = nThis + 1;
969 while( SQLITE_ROW==sqlite3_step(pStmt) ){
970 nByte += sqlite3_column_bytes(pStmt, 1);
971 nDbCol++;
973 rc = sqlite3_reset(pStmt);
975 if( rc==SQLITE_OK ){
976 nByte += nDbCol * (sizeof(const char *) + sizeof(u8) + 1);
977 pAlloc = sqlite3_malloc(nByte);
978 if( pAlloc==0 ){
979 rc = SQLITE_NOMEM;
982 if( rc==SQLITE_OK ){
983 azCol = (char **)pAlloc;
984 pAlloc = (u8 *)&azCol[nDbCol];
985 abPK = (u8 *)pAlloc;
986 pAlloc = &abPK[nDbCol];
987 if( pzTab ){
988 memcpy(pAlloc, zThis, nThis+1);
989 *pzTab = (char *)pAlloc;
990 pAlloc += nThis+1;
993 i = 0;
994 while( SQLITE_ROW==sqlite3_step(pStmt) ){
995 int nName = sqlite3_column_bytes(pStmt, 1);
996 const unsigned char *zName = sqlite3_column_text(pStmt, 1);
997 if( zName==0 ) break;
998 memcpy(pAlloc, zName, nName+1);
999 azCol[i] = (char *)pAlloc;
1000 pAlloc += nName+1;
1001 abPK[i] = sqlite3_column_int(pStmt, 5);
1002 i++;
1004 rc = sqlite3_reset(pStmt);
1008 /* If successful, populate the output variables. Otherwise, zero them and
1009 ** free any allocation made. An error code will be returned in this case.
1011 if( rc==SQLITE_OK ){
1012 *pazCol = (const char **)azCol;
1013 *pabPK = abPK;
1014 *pnCol = nDbCol;
1015 }else{
1016 *pazCol = 0;
1017 *pabPK = 0;
1018 *pnCol = 0;
1019 if( pzTab ) *pzTab = 0;
1020 sqlite3_free(azCol);
1022 sqlite3_finalize(pStmt);
1023 return rc;
1027 ** This function is only called from within a pre-update handler for a
1028 ** write to table pTab, part of session pSession. If this is the first
1029 ** write to this table, initalize the SessionTable.nCol, azCol[] and
1030 ** abPK[] arrays accordingly.
1032 ** If an error occurs, an error code is stored in sqlite3_session.rc and
1033 ** non-zero returned. Or, if no error occurs but the table has no primary
1034 ** key, sqlite3_session.rc is left set to SQLITE_OK and non-zero returned to
1035 ** indicate that updates on this table should be ignored. SessionTable.abPK
1036 ** is set to NULL in this case.
1038 static int sessionInitTable(sqlite3_session *pSession, SessionTable *pTab){
1039 if( pTab->nCol==0 ){
1040 u8 *abPK;
1041 assert( pTab->azCol==0 || pTab->abPK==0 );
1042 pSession->rc = sessionTableInfo(pSession->db, pSession->zDb,
1043 pTab->zName, &pTab->nCol, 0, &pTab->azCol, &abPK
1045 if( pSession->rc==SQLITE_OK ){
1046 int i;
1047 for(i=0; i<pTab->nCol; i++){
1048 if( abPK[i] ){
1049 pTab->abPK = abPK;
1050 break;
1053 if( 0==sqlite3_stricmp("sqlite_stat1", pTab->zName) ){
1054 pTab->bStat1 = 1;
1058 return (pSession->rc || pTab->abPK==0);
1062 ** Versions of the four methods in object SessionHook for use with the
1063 ** sqlite_stat1 table. The purpose of this is to substitute a zero-length
1064 ** blob each time a NULL value is read from the "idx" column of the
1065 ** sqlite_stat1 table.
1067 typedef struct SessionStat1Ctx SessionStat1Ctx;
1068 struct SessionStat1Ctx {
1069 SessionHook hook;
1070 sqlite3_session *pSession;
1072 static int sessionStat1Old(void *pCtx, int iCol, sqlite3_value **ppVal){
1073 SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
1074 sqlite3_value *pVal = 0;
1075 int rc = p->hook.xOld(p->hook.pCtx, iCol, &pVal);
1076 if( rc==SQLITE_OK && iCol==1 && sqlite3_value_type(pVal)==SQLITE_NULL ){
1077 pVal = p->pSession->pZeroBlob;
1079 *ppVal = pVal;
1080 return rc;
1082 static int sessionStat1New(void *pCtx, int iCol, sqlite3_value **ppVal){
1083 SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
1084 sqlite3_value *pVal = 0;
1085 int rc = p->hook.xNew(p->hook.pCtx, iCol, &pVal);
1086 if( rc==SQLITE_OK && iCol==1 && sqlite3_value_type(pVal)==SQLITE_NULL ){
1087 pVal = p->pSession->pZeroBlob;
1089 *ppVal = pVal;
1090 return rc;
1092 static int sessionStat1Count(void *pCtx){
1093 SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
1094 return p->hook.xCount(p->hook.pCtx);
1096 static int sessionStat1Depth(void *pCtx){
1097 SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
1098 return p->hook.xDepth(p->hook.pCtx);
1103 ** This function is only called from with a pre-update-hook reporting a
1104 ** change on table pTab (attached to session pSession). The type of change
1105 ** (UPDATE, INSERT, DELETE) is specified by the first argument.
1107 ** Unless one is already present or an error occurs, an entry is added
1108 ** to the changed-rows hash table associated with table pTab.
1110 static void sessionPreupdateOneChange(
1111 int op, /* One of SQLITE_UPDATE, INSERT, DELETE */
1112 sqlite3_session *pSession, /* Session object pTab is attached to */
1113 SessionTable *pTab /* Table that change applies to */
1115 int iHash;
1116 int bNull = 0;
1117 int rc = SQLITE_OK;
1118 SessionStat1Ctx stat1 = {0};
1120 if( pSession->rc ) return;
1122 /* Load table details if required */
1123 if( sessionInitTable(pSession, pTab) ) return;
1125 /* Check the number of columns in this xPreUpdate call matches the
1126 ** number of columns in the table. */
1127 if( pTab->nCol!=pSession->hook.xCount(pSession->hook.pCtx) ){
1128 pSession->rc = SQLITE_SCHEMA;
1129 return;
1132 /* Grow the hash table if required */
1133 if( sessionGrowHash(0, pTab) ){
1134 pSession->rc = SQLITE_NOMEM;
1135 return;
1138 if( pTab->bStat1 ){
1139 stat1.hook = pSession->hook;
1140 stat1.pSession = pSession;
1141 pSession->hook.pCtx = (void*)&stat1;
1142 pSession->hook.xNew = sessionStat1New;
1143 pSession->hook.xOld = sessionStat1Old;
1144 pSession->hook.xCount = sessionStat1Count;
1145 pSession->hook.xDepth = sessionStat1Depth;
1146 if( pSession->pZeroBlob==0 ){
1147 sqlite3_value *p = sqlite3ValueNew(0);
1148 if( p==0 ){
1149 rc = SQLITE_NOMEM;
1150 goto error_out;
1152 sqlite3ValueSetStr(p, 0, "", 0, SQLITE_STATIC);
1153 pSession->pZeroBlob = p;
1157 /* Calculate the hash-key for this change. If the primary key of the row
1158 ** includes a NULL value, exit early. Such changes are ignored by the
1159 ** session module. */
1160 rc = sessionPreupdateHash(pSession, pTab, op==SQLITE_INSERT, &iHash, &bNull);
1161 if( rc!=SQLITE_OK ) goto error_out;
1163 if( bNull==0 ){
1164 /* Search the hash table for an existing record for this row. */
1165 SessionChange *pC;
1166 for(pC=pTab->apChange[iHash]; pC; pC=pC->pNext){
1167 if( sessionPreupdateEqual(pSession, pTab, pC, op) ) break;
1170 if( pC==0 ){
1171 /* Create a new change object containing all the old values (if
1172 ** this is an SQLITE_UPDATE or SQLITE_DELETE), or just the PK
1173 ** values (if this is an INSERT). */
1174 SessionChange *pChange; /* New change object */
1175 int nByte; /* Number of bytes to allocate */
1176 int i; /* Used to iterate through columns */
1178 assert( rc==SQLITE_OK );
1179 pTab->nEntry++;
1181 /* Figure out how large an allocation is required */
1182 nByte = sizeof(SessionChange);
1183 for(i=0; i<pTab->nCol; i++){
1184 sqlite3_value *p = 0;
1185 if( op!=SQLITE_INSERT ){
1186 TESTONLY(int trc = ) pSession->hook.xOld(pSession->hook.pCtx, i, &p);
1187 assert( trc==SQLITE_OK );
1188 }else if( pTab->abPK[i] ){
1189 TESTONLY(int trc = ) pSession->hook.xNew(pSession->hook.pCtx, i, &p);
1190 assert( trc==SQLITE_OK );
1193 /* This may fail if SQLite value p contains a utf-16 string that must
1194 ** be converted to utf-8 and an OOM error occurs while doing so. */
1195 rc = sessionSerializeValue(0, p, &nByte);
1196 if( rc!=SQLITE_OK ) goto error_out;
1199 /* Allocate the change object */
1200 pChange = (SessionChange *)sqlite3_malloc(nByte);
1201 if( !pChange ){
1202 rc = SQLITE_NOMEM;
1203 goto error_out;
1204 }else{
1205 memset(pChange, 0, sizeof(SessionChange));
1206 pChange->aRecord = (u8 *)&pChange[1];
1209 /* Populate the change object. None of the preupdate_old(),
1210 ** preupdate_new() or SerializeValue() calls below may fail as all
1211 ** required values and encodings have already been cached in memory.
1212 ** It is not possible for an OOM to occur in this block. */
1213 nByte = 0;
1214 for(i=0; i<pTab->nCol; i++){
1215 sqlite3_value *p = 0;
1216 if( op!=SQLITE_INSERT ){
1217 pSession->hook.xOld(pSession->hook.pCtx, i, &p);
1218 }else if( pTab->abPK[i] ){
1219 pSession->hook.xNew(pSession->hook.pCtx, i, &p);
1221 sessionSerializeValue(&pChange->aRecord[nByte], p, &nByte);
1224 /* Add the change to the hash-table */
1225 if( pSession->bIndirect || pSession->hook.xDepth(pSession->hook.pCtx) ){
1226 pChange->bIndirect = 1;
1228 pChange->nRecord = nByte;
1229 pChange->op = op;
1230 pChange->pNext = pTab->apChange[iHash];
1231 pTab->apChange[iHash] = pChange;
1233 }else if( pC->bIndirect ){
1234 /* If the existing change is considered "indirect", but this current
1235 ** change is "direct", mark the change object as direct. */
1236 if( pSession->hook.xDepth(pSession->hook.pCtx)==0
1237 && pSession->bIndirect==0
1239 pC->bIndirect = 0;
1244 /* If an error has occurred, mark the session object as failed. */
1245 error_out:
1246 if( pTab->bStat1 ){
1247 pSession->hook = stat1.hook;
1249 if( rc!=SQLITE_OK ){
1250 pSession->rc = rc;
1254 static int sessionFindTable(
1255 sqlite3_session *pSession,
1256 const char *zName,
1257 SessionTable **ppTab
1259 int rc = SQLITE_OK;
1260 int nName = sqlite3Strlen30(zName);
1261 SessionTable *pRet;
1263 /* Search for an existing table */
1264 for(pRet=pSession->pTable; pRet; pRet=pRet->pNext){
1265 if( 0==sqlite3_strnicmp(pRet->zName, zName, nName+1) ) break;
1268 if( pRet==0 && pSession->bAutoAttach ){
1269 /* If there is a table-filter configured, invoke it. If it returns 0,
1270 ** do not automatically add the new table. */
1271 if( pSession->xTableFilter==0
1272 || pSession->xTableFilter(pSession->pFilterCtx, zName)
1274 rc = sqlite3session_attach(pSession, zName);
1275 if( rc==SQLITE_OK ){
1276 for(pRet=pSession->pTable; pRet->pNext; pRet=pRet->pNext);
1277 assert( 0==sqlite3_strnicmp(pRet->zName, zName, nName+1) );
1282 assert( rc==SQLITE_OK || pRet==0 );
1283 *ppTab = pRet;
1284 return rc;
1288 ** The 'pre-update' hook registered by this module with SQLite databases.
1290 static void xPreUpdate(
1291 void *pCtx, /* Copy of third arg to preupdate_hook() */
1292 sqlite3 *db, /* Database handle */
1293 int op, /* SQLITE_UPDATE, DELETE or INSERT */
1294 char const *zDb, /* Database name */
1295 char const *zName, /* Table name */
1296 sqlite3_int64 iKey1, /* Rowid of row about to be deleted/updated */
1297 sqlite3_int64 iKey2 /* New rowid value (for a rowid UPDATE) */
1299 sqlite3_session *pSession;
1300 int nDb = sqlite3Strlen30(zDb);
1302 assert( sqlite3_mutex_held(db->mutex) );
1304 for(pSession=(sqlite3_session *)pCtx; pSession; pSession=pSession->pNext){
1305 SessionTable *pTab;
1307 /* If this session is attached to a different database ("main", "temp"
1308 ** etc.), or if it is not currently enabled, there is nothing to do. Skip
1309 ** to the next session object attached to this database. */
1310 if( pSession->bEnable==0 ) continue;
1311 if( pSession->rc ) continue;
1312 if( sqlite3_strnicmp(zDb, pSession->zDb, nDb+1) ) continue;
1314 pSession->rc = sessionFindTable(pSession, zName, &pTab);
1315 if( pTab ){
1316 assert( pSession->rc==SQLITE_OK );
1317 sessionPreupdateOneChange(op, pSession, pTab);
1318 if( op==SQLITE_UPDATE ){
1319 sessionPreupdateOneChange(SQLITE_INSERT, pSession, pTab);
1326 ** The pre-update hook implementations.
1328 static int sessionPreupdateOld(void *pCtx, int iVal, sqlite3_value **ppVal){
1329 return sqlite3_preupdate_old((sqlite3*)pCtx, iVal, ppVal);
1331 static int sessionPreupdateNew(void *pCtx, int iVal, sqlite3_value **ppVal){
1332 return sqlite3_preupdate_new((sqlite3*)pCtx, iVal, ppVal);
1334 static int sessionPreupdateCount(void *pCtx){
1335 return sqlite3_preupdate_count((sqlite3*)pCtx);
1337 static int sessionPreupdateDepth(void *pCtx){
1338 return sqlite3_preupdate_depth((sqlite3*)pCtx);
1342 ** Install the pre-update hooks on the session object passed as the only
1343 ** argument.
1345 static void sessionPreupdateHooks(
1346 sqlite3_session *pSession
1348 pSession->hook.pCtx = (void*)pSession->db;
1349 pSession->hook.xOld = sessionPreupdateOld;
1350 pSession->hook.xNew = sessionPreupdateNew;
1351 pSession->hook.xCount = sessionPreupdateCount;
1352 pSession->hook.xDepth = sessionPreupdateDepth;
1355 typedef struct SessionDiffCtx SessionDiffCtx;
1356 struct SessionDiffCtx {
1357 sqlite3_stmt *pStmt;
1358 int nOldOff;
1362 ** The diff hook implementations.
1364 static int sessionDiffOld(void *pCtx, int iVal, sqlite3_value **ppVal){
1365 SessionDiffCtx *p = (SessionDiffCtx*)pCtx;
1366 *ppVal = sqlite3_column_value(p->pStmt, iVal+p->nOldOff);
1367 return SQLITE_OK;
1369 static int sessionDiffNew(void *pCtx, int iVal, sqlite3_value **ppVal){
1370 SessionDiffCtx *p = (SessionDiffCtx*)pCtx;
1371 *ppVal = sqlite3_column_value(p->pStmt, iVal);
1372 return SQLITE_OK;
1374 static int sessionDiffCount(void *pCtx){
1375 SessionDiffCtx *p = (SessionDiffCtx*)pCtx;
1376 return p->nOldOff ? p->nOldOff : sqlite3_column_count(p->pStmt);
1378 static int sessionDiffDepth(void *pCtx){
1379 return 0;
1383 ** Install the diff hooks on the session object passed as the only
1384 ** argument.
1386 static void sessionDiffHooks(
1387 sqlite3_session *pSession,
1388 SessionDiffCtx *pDiffCtx
1390 pSession->hook.pCtx = (void*)pDiffCtx;
1391 pSession->hook.xOld = sessionDiffOld;
1392 pSession->hook.xNew = sessionDiffNew;
1393 pSession->hook.xCount = sessionDiffCount;
1394 pSession->hook.xDepth = sessionDiffDepth;
1397 static char *sessionExprComparePK(
1398 int nCol,
1399 const char *zDb1, const char *zDb2,
1400 const char *zTab,
1401 const char **azCol, u8 *abPK
1403 int i;
1404 const char *zSep = "";
1405 char *zRet = 0;
1407 for(i=0; i<nCol; i++){
1408 if( abPK[i] ){
1409 zRet = sqlite3_mprintf("%z%s\"%w\".\"%w\".\"%w\"=\"%w\".\"%w\".\"%w\"",
1410 zRet, zSep, zDb1, zTab, azCol[i], zDb2, zTab, azCol[i]
1412 zSep = " AND ";
1413 if( zRet==0 ) break;
1417 return zRet;
1420 static char *sessionExprCompareOther(
1421 int nCol,
1422 const char *zDb1, const char *zDb2,
1423 const char *zTab,
1424 const char **azCol, u8 *abPK
1426 int i;
1427 const char *zSep = "";
1428 char *zRet = 0;
1429 int bHave = 0;
1431 for(i=0; i<nCol; i++){
1432 if( abPK[i]==0 ){
1433 bHave = 1;
1434 zRet = sqlite3_mprintf(
1435 "%z%s\"%w\".\"%w\".\"%w\" IS NOT \"%w\".\"%w\".\"%w\"",
1436 zRet, zSep, zDb1, zTab, azCol[i], zDb2, zTab, azCol[i]
1438 zSep = " OR ";
1439 if( zRet==0 ) break;
1443 if( bHave==0 ){
1444 assert( zRet==0 );
1445 zRet = sqlite3_mprintf("0");
1448 return zRet;
1451 static char *sessionSelectFindNew(
1452 int nCol,
1453 const char *zDb1, /* Pick rows in this db only */
1454 const char *zDb2, /* But not in this one */
1455 const char *zTbl, /* Table name */
1456 const char *zExpr
1458 char *zRet = sqlite3_mprintf(
1459 "SELECT * FROM \"%w\".\"%w\" WHERE NOT EXISTS ("
1460 " SELECT 1 FROM \"%w\".\"%w\" WHERE %s"
1461 ")",
1462 zDb1, zTbl, zDb2, zTbl, zExpr
1464 return zRet;
1467 static int sessionDiffFindNew(
1468 int op,
1469 sqlite3_session *pSession,
1470 SessionTable *pTab,
1471 const char *zDb1,
1472 const char *zDb2,
1473 char *zExpr
1475 int rc = SQLITE_OK;
1476 char *zStmt = sessionSelectFindNew(pTab->nCol, zDb1, zDb2, pTab->zName,zExpr);
1478 if( zStmt==0 ){
1479 rc = SQLITE_NOMEM;
1480 }else{
1481 sqlite3_stmt *pStmt;
1482 rc = sqlite3_prepare(pSession->db, zStmt, -1, &pStmt, 0);
1483 if( rc==SQLITE_OK ){
1484 SessionDiffCtx *pDiffCtx = (SessionDiffCtx*)pSession->hook.pCtx;
1485 pDiffCtx->pStmt = pStmt;
1486 pDiffCtx->nOldOff = 0;
1487 while( SQLITE_ROW==sqlite3_step(pStmt) ){
1488 sessionPreupdateOneChange(op, pSession, pTab);
1490 rc = sqlite3_finalize(pStmt);
1492 sqlite3_free(zStmt);
1495 return rc;
1498 static int sessionDiffFindModified(
1499 sqlite3_session *pSession,
1500 SessionTable *pTab,
1501 const char *zFrom,
1502 const char *zExpr
1504 int rc = SQLITE_OK;
1506 char *zExpr2 = sessionExprCompareOther(pTab->nCol,
1507 pSession->zDb, zFrom, pTab->zName, pTab->azCol, pTab->abPK
1509 if( zExpr2==0 ){
1510 rc = SQLITE_NOMEM;
1511 }else{
1512 char *zStmt = sqlite3_mprintf(
1513 "SELECT * FROM \"%w\".\"%w\", \"%w\".\"%w\" WHERE %s AND (%z)",
1514 pSession->zDb, pTab->zName, zFrom, pTab->zName, zExpr, zExpr2
1516 if( zStmt==0 ){
1517 rc = SQLITE_NOMEM;
1518 }else{
1519 sqlite3_stmt *pStmt;
1520 rc = sqlite3_prepare(pSession->db, zStmt, -1, &pStmt, 0);
1522 if( rc==SQLITE_OK ){
1523 SessionDiffCtx *pDiffCtx = (SessionDiffCtx*)pSession->hook.pCtx;
1524 pDiffCtx->pStmt = pStmt;
1525 pDiffCtx->nOldOff = pTab->nCol;
1526 while( SQLITE_ROW==sqlite3_step(pStmt) ){
1527 sessionPreupdateOneChange(SQLITE_UPDATE, pSession, pTab);
1529 rc = sqlite3_finalize(pStmt);
1531 sqlite3_free(zStmt);
1535 return rc;
1538 int sqlite3session_diff(
1539 sqlite3_session *pSession,
1540 const char *zFrom,
1541 const char *zTbl,
1542 char **pzErrMsg
1544 const char *zDb = pSession->zDb;
1545 int rc = pSession->rc;
1546 SessionDiffCtx d;
1548 memset(&d, 0, sizeof(d));
1549 sessionDiffHooks(pSession, &d);
1551 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
1552 if( pzErrMsg ) *pzErrMsg = 0;
1553 if( rc==SQLITE_OK ){
1554 char *zExpr = 0;
1555 sqlite3 *db = pSession->db;
1556 SessionTable *pTo; /* Table zTbl */
1558 /* Locate and if necessary initialize the target table object */
1559 rc = sessionFindTable(pSession, zTbl, &pTo);
1560 if( pTo==0 ) goto diff_out;
1561 if( sessionInitTable(pSession, pTo) ){
1562 rc = pSession->rc;
1563 goto diff_out;
1566 /* Check the table schemas match */
1567 if( rc==SQLITE_OK ){
1568 int bHasPk = 0;
1569 int bMismatch = 0;
1570 int nCol; /* Columns in zFrom.zTbl */
1571 u8 *abPK;
1572 const char **azCol = 0;
1573 rc = sessionTableInfo(db, zFrom, zTbl, &nCol, 0, &azCol, &abPK);
1574 if( rc==SQLITE_OK ){
1575 if( pTo->nCol!=nCol ){
1576 bMismatch = 1;
1577 }else{
1578 int i;
1579 for(i=0; i<nCol; i++){
1580 if( pTo->abPK[i]!=abPK[i] ) bMismatch = 1;
1581 if( sqlite3_stricmp(azCol[i], pTo->azCol[i]) ) bMismatch = 1;
1582 if( abPK[i] ) bHasPk = 1;
1586 sqlite3_free((char*)azCol);
1587 if( bMismatch ){
1588 *pzErrMsg = sqlite3_mprintf("table schemas do not match");
1589 rc = SQLITE_SCHEMA;
1591 if( bHasPk==0 ){
1592 /* Ignore tables with no primary keys */
1593 goto diff_out;
1597 if( rc==SQLITE_OK ){
1598 zExpr = sessionExprComparePK(pTo->nCol,
1599 zDb, zFrom, pTo->zName, pTo->azCol, pTo->abPK
1603 /* Find new rows */
1604 if( rc==SQLITE_OK ){
1605 rc = sessionDiffFindNew(SQLITE_INSERT, pSession, pTo, zDb, zFrom, zExpr);
1608 /* Find old rows */
1609 if( rc==SQLITE_OK ){
1610 rc = sessionDiffFindNew(SQLITE_DELETE, pSession, pTo, zFrom, zDb, zExpr);
1613 /* Find modified rows */
1614 if( rc==SQLITE_OK ){
1615 rc = sessionDiffFindModified(pSession, pTo, zFrom, zExpr);
1618 sqlite3_free(zExpr);
1621 diff_out:
1622 sessionPreupdateHooks(pSession);
1623 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
1624 return rc;
1628 ** Create a session object. This session object will record changes to
1629 ** database zDb attached to connection db.
1631 int sqlite3session_create(
1632 sqlite3 *db, /* Database handle */
1633 const char *zDb, /* Name of db (e.g. "main") */
1634 sqlite3_session **ppSession /* OUT: New session object */
1636 sqlite3_session *pNew; /* Newly allocated session object */
1637 sqlite3_session *pOld; /* Session object already attached to db */
1638 int nDb = sqlite3Strlen30(zDb); /* Length of zDb in bytes */
1640 /* Zero the output value in case an error occurs. */
1641 *ppSession = 0;
1643 /* Allocate and populate the new session object. */
1644 pNew = (sqlite3_session *)sqlite3_malloc(sizeof(sqlite3_session) + nDb + 1);
1645 if( !pNew ) return SQLITE_NOMEM;
1646 memset(pNew, 0, sizeof(sqlite3_session));
1647 pNew->db = db;
1648 pNew->zDb = (char *)&pNew[1];
1649 pNew->bEnable = 1;
1650 memcpy(pNew->zDb, zDb, nDb+1);
1651 sessionPreupdateHooks(pNew);
1653 /* Add the new session object to the linked list of session objects
1654 ** attached to database handle $db. Do this under the cover of the db
1655 ** handle mutex. */
1656 sqlite3_mutex_enter(sqlite3_db_mutex(db));
1657 pOld = (sqlite3_session*)sqlite3_preupdate_hook(db, xPreUpdate, (void*)pNew);
1658 pNew->pNext = pOld;
1659 sqlite3_mutex_leave(sqlite3_db_mutex(db));
1661 *ppSession = pNew;
1662 return SQLITE_OK;
1666 ** Free the list of table objects passed as the first argument. The contents
1667 ** of the changed-rows hash tables are also deleted.
1669 static void sessionDeleteTable(SessionTable *pList){
1670 SessionTable *pNext;
1671 SessionTable *pTab;
1673 for(pTab=pList; pTab; pTab=pNext){
1674 int i;
1675 pNext = pTab->pNext;
1676 for(i=0; i<pTab->nChange; i++){
1677 SessionChange *p;
1678 SessionChange *pNextChange;
1679 for(p=pTab->apChange[i]; p; p=pNextChange){
1680 pNextChange = p->pNext;
1681 sqlite3_free(p);
1684 sqlite3_free((char*)pTab->azCol); /* cast works around VC++ bug */
1685 sqlite3_free(pTab->apChange);
1686 sqlite3_free(pTab);
1691 ** Delete a session object previously allocated using sqlite3session_create().
1693 void sqlite3session_delete(sqlite3_session *pSession){
1694 sqlite3 *db = pSession->db;
1695 sqlite3_session *pHead;
1696 sqlite3_session **pp;
1698 /* Unlink the session from the linked list of sessions attached to the
1699 ** database handle. Hold the db mutex while doing so. */
1700 sqlite3_mutex_enter(sqlite3_db_mutex(db));
1701 pHead = (sqlite3_session*)sqlite3_preupdate_hook(db, 0, 0);
1702 for(pp=&pHead; ALWAYS((*pp)!=0); pp=&((*pp)->pNext)){
1703 if( (*pp)==pSession ){
1704 *pp = (*pp)->pNext;
1705 if( pHead ) sqlite3_preupdate_hook(db, xPreUpdate, (void*)pHead);
1706 break;
1709 sqlite3_mutex_leave(sqlite3_db_mutex(db));
1710 sqlite3ValueFree(pSession->pZeroBlob);
1712 /* Delete all attached table objects. And the contents of their
1713 ** associated hash-tables. */
1714 sessionDeleteTable(pSession->pTable);
1716 /* Free the session object itself. */
1717 sqlite3_free(pSession);
1721 ** Set a table filter on a Session Object.
1723 void sqlite3session_table_filter(
1724 sqlite3_session *pSession,
1725 int(*xFilter)(void*, const char*),
1726 void *pCtx /* First argument passed to xFilter */
1728 pSession->bAutoAttach = 1;
1729 pSession->pFilterCtx = pCtx;
1730 pSession->xTableFilter = xFilter;
1734 ** Attach a table to a session. All subsequent changes made to the table
1735 ** while the session object is enabled will be recorded.
1737 ** Only tables that have a PRIMARY KEY defined may be attached. It does
1738 ** not matter if the PRIMARY KEY is an "INTEGER PRIMARY KEY" (rowid alias)
1739 ** or not.
1741 int sqlite3session_attach(
1742 sqlite3_session *pSession, /* Session object */
1743 const char *zName /* Table name */
1745 int rc = SQLITE_OK;
1746 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
1748 if( !zName ){
1749 pSession->bAutoAttach = 1;
1750 }else{
1751 SessionTable *pTab; /* New table object (if required) */
1752 int nName; /* Number of bytes in string zName */
1754 /* First search for an existing entry. If one is found, this call is
1755 ** a no-op. Return early. */
1756 nName = sqlite3Strlen30(zName);
1757 for(pTab=pSession->pTable; pTab; pTab=pTab->pNext){
1758 if( 0==sqlite3_strnicmp(pTab->zName, zName, nName+1) ) break;
1761 if( !pTab ){
1762 /* Allocate new SessionTable object. */
1763 pTab = (SessionTable *)sqlite3_malloc(sizeof(SessionTable) + nName + 1);
1764 if( !pTab ){
1765 rc = SQLITE_NOMEM;
1766 }else{
1767 /* Populate the new SessionTable object and link it into the list.
1768 ** The new object must be linked onto the end of the list, not
1769 ** simply added to the start of it in order to ensure that tables
1770 ** appear in the correct order when a changeset or patchset is
1771 ** eventually generated. */
1772 SessionTable **ppTab;
1773 memset(pTab, 0, sizeof(SessionTable));
1774 pTab->zName = (char *)&pTab[1];
1775 memcpy(pTab->zName, zName, nName+1);
1776 for(ppTab=&pSession->pTable; *ppTab; ppTab=&(*ppTab)->pNext);
1777 *ppTab = pTab;
1782 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
1783 return rc;
1787 ** Ensure that there is room in the buffer to append nByte bytes of data.
1788 ** If not, use sqlite3_realloc() to grow the buffer so that there is.
1790 ** If successful, return zero. Otherwise, if an OOM condition is encountered,
1791 ** set *pRc to SQLITE_NOMEM and return non-zero.
1793 static int sessionBufferGrow(SessionBuffer *p, int nByte, int *pRc){
1794 if( *pRc==SQLITE_OK && p->nAlloc-p->nBuf<nByte ){
1795 u8 *aNew;
1796 int nNew = p->nAlloc ? p->nAlloc : 128;
1797 do {
1798 nNew = nNew*2;
1799 }while( nNew<(p->nBuf+nByte) );
1801 aNew = (u8 *)sqlite3_realloc(p->aBuf, nNew);
1802 if( 0==aNew ){
1803 *pRc = SQLITE_NOMEM;
1804 }else{
1805 p->aBuf = aNew;
1806 p->nAlloc = nNew;
1809 return (*pRc!=SQLITE_OK);
1813 ** Append the value passed as the second argument to the buffer passed
1814 ** as the first.
1816 ** This function is a no-op if *pRc is non-zero when it is called.
1817 ** Otherwise, if an error occurs, *pRc is set to an SQLite error code
1818 ** before returning.
1820 static void sessionAppendValue(SessionBuffer *p, sqlite3_value *pVal, int *pRc){
1821 int rc = *pRc;
1822 if( rc==SQLITE_OK ){
1823 int nByte = 0;
1824 rc = sessionSerializeValue(0, pVal, &nByte);
1825 sessionBufferGrow(p, nByte, &rc);
1826 if( rc==SQLITE_OK ){
1827 rc = sessionSerializeValue(&p->aBuf[p->nBuf], pVal, 0);
1828 p->nBuf += nByte;
1829 }else{
1830 *pRc = rc;
1836 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1837 ** called. Otherwise, append a single byte to the buffer.
1839 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
1840 ** returning.
1842 static void sessionAppendByte(SessionBuffer *p, u8 v, int *pRc){
1843 if( 0==sessionBufferGrow(p, 1, pRc) ){
1844 p->aBuf[p->nBuf++] = v;
1849 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1850 ** called. Otherwise, append a single varint to the buffer.
1852 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
1853 ** returning.
1855 static void sessionAppendVarint(SessionBuffer *p, int v, int *pRc){
1856 if( 0==sessionBufferGrow(p, 9, pRc) ){
1857 p->nBuf += sessionVarintPut(&p->aBuf[p->nBuf], v);
1862 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1863 ** called. Otherwise, append a blob of data to the buffer.
1865 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
1866 ** returning.
1868 static void sessionAppendBlob(
1869 SessionBuffer *p,
1870 const u8 *aBlob,
1871 int nBlob,
1872 int *pRc
1874 if( nBlob>0 && 0==sessionBufferGrow(p, nBlob, pRc) ){
1875 memcpy(&p->aBuf[p->nBuf], aBlob, nBlob);
1876 p->nBuf += nBlob;
1881 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1882 ** called. Otherwise, append a string to the buffer. All bytes in the string
1883 ** up to (but not including) the nul-terminator are written to the buffer.
1885 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
1886 ** returning.
1888 static void sessionAppendStr(
1889 SessionBuffer *p,
1890 const char *zStr,
1891 int *pRc
1893 int nStr = sqlite3Strlen30(zStr);
1894 if( 0==sessionBufferGrow(p, nStr, pRc) ){
1895 memcpy(&p->aBuf[p->nBuf], zStr, nStr);
1896 p->nBuf += nStr;
1901 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1902 ** called. Otherwise, append the string representation of integer iVal
1903 ** to the buffer. No nul-terminator is written.
1905 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
1906 ** returning.
1908 static void sessionAppendInteger(
1909 SessionBuffer *p, /* Buffer to append to */
1910 int iVal, /* Value to write the string rep. of */
1911 int *pRc /* IN/OUT: Error code */
1913 char aBuf[24];
1914 sqlite3_snprintf(sizeof(aBuf)-1, aBuf, "%d", iVal);
1915 sessionAppendStr(p, aBuf, pRc);
1919 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1920 ** called. Otherwise, append the string zStr enclosed in quotes (") and
1921 ** with any embedded quote characters escaped to the buffer. No
1922 ** nul-terminator byte is written.
1924 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
1925 ** returning.
1927 static void sessionAppendIdent(
1928 SessionBuffer *p, /* Buffer to a append to */
1929 const char *zStr, /* String to quote, escape and append */
1930 int *pRc /* IN/OUT: Error code */
1932 int nStr = sqlite3Strlen30(zStr)*2 + 2 + 1;
1933 if( 0==sessionBufferGrow(p, nStr, pRc) ){
1934 char *zOut = (char *)&p->aBuf[p->nBuf];
1935 const char *zIn = zStr;
1936 *zOut++ = '"';
1937 while( *zIn ){
1938 if( *zIn=='"' ) *zOut++ = '"';
1939 *zOut++ = *(zIn++);
1941 *zOut++ = '"';
1942 p->nBuf = (int)((u8 *)zOut - p->aBuf);
1947 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
1948 ** called. Otherwse, it appends the serialized version of the value stored
1949 ** in column iCol of the row that SQL statement pStmt currently points
1950 ** to to the buffer.
1952 static void sessionAppendCol(
1953 SessionBuffer *p, /* Buffer to append to */
1954 sqlite3_stmt *pStmt, /* Handle pointing to row containing value */
1955 int iCol, /* Column to read value from */
1956 int *pRc /* IN/OUT: Error code */
1958 if( *pRc==SQLITE_OK ){
1959 int eType = sqlite3_column_type(pStmt, iCol);
1960 sessionAppendByte(p, (u8)eType, pRc);
1961 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
1962 sqlite3_int64 i;
1963 u8 aBuf[8];
1964 if( eType==SQLITE_INTEGER ){
1965 i = sqlite3_column_int64(pStmt, iCol);
1966 }else{
1967 double r = sqlite3_column_double(pStmt, iCol);
1968 memcpy(&i, &r, 8);
1970 sessionPutI64(aBuf, i);
1971 sessionAppendBlob(p, aBuf, 8, pRc);
1973 if( eType==SQLITE_BLOB || eType==SQLITE_TEXT ){
1974 u8 *z;
1975 int nByte;
1976 if( eType==SQLITE_BLOB ){
1977 z = (u8 *)sqlite3_column_blob(pStmt, iCol);
1978 }else{
1979 z = (u8 *)sqlite3_column_text(pStmt, iCol);
1981 nByte = sqlite3_column_bytes(pStmt, iCol);
1982 if( z || (eType==SQLITE_BLOB && nByte==0) ){
1983 sessionAppendVarint(p, nByte, pRc);
1984 sessionAppendBlob(p, z, nByte, pRc);
1985 }else{
1986 *pRc = SQLITE_NOMEM;
1994 ** This function appends an update change to the buffer (see the comments
1995 ** under "CHANGESET FORMAT" at the top of the file). An update change
1996 ** consists of:
1998 ** 1 byte: SQLITE_UPDATE (0x17)
1999 ** n bytes: old.* record (see RECORD FORMAT)
2000 ** m bytes: new.* record (see RECORD FORMAT)
2002 ** The SessionChange object passed as the third argument contains the
2003 ** values that were stored in the row when the session began (the old.*
2004 ** values). The statement handle passed as the second argument points
2005 ** at the current version of the row (the new.* values).
2007 ** If all of the old.* values are equal to their corresponding new.* value
2008 ** (i.e. nothing has changed), then no data at all is appended to the buffer.
2010 ** Otherwise, the old.* record contains all primary key values and the
2011 ** original values of any fields that have been modified. The new.* record
2012 ** contains the new values of only those fields that have been modified.
2014 static int sessionAppendUpdate(
2015 SessionBuffer *pBuf, /* Buffer to append to */
2016 int bPatchset, /* True for "patchset", 0 for "changeset" */
2017 sqlite3_stmt *pStmt, /* Statement handle pointing at new row */
2018 SessionChange *p, /* Object containing old values */
2019 u8 *abPK /* Boolean array - true for PK columns */
2021 int rc = SQLITE_OK;
2022 SessionBuffer buf2 = {0,0,0}; /* Buffer to accumulate new.* record in */
2023 int bNoop = 1; /* Set to zero if any values are modified */
2024 int nRewind = pBuf->nBuf; /* Set to zero if any values are modified */
2025 int i; /* Used to iterate through columns */
2026 u8 *pCsr = p->aRecord; /* Used to iterate through old.* values */
2028 sessionAppendByte(pBuf, SQLITE_UPDATE, &rc);
2029 sessionAppendByte(pBuf, p->bIndirect, &rc);
2030 for(i=0; i<sqlite3_column_count(pStmt); i++){
2031 int bChanged = 0;
2032 int nAdvance;
2033 int eType = *pCsr;
2034 switch( eType ){
2035 case SQLITE_NULL:
2036 nAdvance = 1;
2037 if( sqlite3_column_type(pStmt, i)!=SQLITE_NULL ){
2038 bChanged = 1;
2040 break;
2042 case SQLITE_FLOAT:
2043 case SQLITE_INTEGER: {
2044 nAdvance = 9;
2045 if( eType==sqlite3_column_type(pStmt, i) ){
2046 sqlite3_int64 iVal = sessionGetI64(&pCsr[1]);
2047 if( eType==SQLITE_INTEGER ){
2048 if( iVal==sqlite3_column_int64(pStmt, i) ) break;
2049 }else{
2050 double dVal;
2051 memcpy(&dVal, &iVal, 8);
2052 if( dVal==sqlite3_column_double(pStmt, i) ) break;
2055 bChanged = 1;
2056 break;
2059 default: {
2060 int n;
2061 int nHdr = 1 + sessionVarintGet(&pCsr[1], &n);
2062 assert( eType==SQLITE_TEXT || eType==SQLITE_BLOB );
2063 nAdvance = nHdr + n;
2064 if( eType==sqlite3_column_type(pStmt, i)
2065 && n==sqlite3_column_bytes(pStmt, i)
2066 && (n==0 || 0==memcmp(&pCsr[nHdr], sqlite3_column_blob(pStmt, i), n))
2068 break;
2070 bChanged = 1;
2074 /* If at least one field has been modified, this is not a no-op. */
2075 if( bChanged ) bNoop = 0;
2077 /* Add a field to the old.* record. This is omitted if this modules is
2078 ** currently generating a patchset. */
2079 if( bPatchset==0 ){
2080 if( bChanged || abPK[i] ){
2081 sessionAppendBlob(pBuf, pCsr, nAdvance, &rc);
2082 }else{
2083 sessionAppendByte(pBuf, 0, &rc);
2087 /* Add a field to the new.* record. Or the only record if currently
2088 ** generating a patchset. */
2089 if( bChanged || (bPatchset && abPK[i]) ){
2090 sessionAppendCol(&buf2, pStmt, i, &rc);
2091 }else{
2092 sessionAppendByte(&buf2, 0, &rc);
2095 pCsr += nAdvance;
2098 if( bNoop ){
2099 pBuf->nBuf = nRewind;
2100 }else{
2101 sessionAppendBlob(pBuf, buf2.aBuf, buf2.nBuf, &rc);
2103 sqlite3_free(buf2.aBuf);
2105 return rc;
2109 ** Append a DELETE change to the buffer passed as the first argument. Use
2110 ** the changeset format if argument bPatchset is zero, or the patchset
2111 ** format otherwise.
2113 static int sessionAppendDelete(
2114 SessionBuffer *pBuf, /* Buffer to append to */
2115 int bPatchset, /* True for "patchset", 0 for "changeset" */
2116 SessionChange *p, /* Object containing old values */
2117 int nCol, /* Number of columns in table */
2118 u8 *abPK /* Boolean array - true for PK columns */
2120 int rc = SQLITE_OK;
2122 sessionAppendByte(pBuf, SQLITE_DELETE, &rc);
2123 sessionAppendByte(pBuf, p->bIndirect, &rc);
2125 if( bPatchset==0 ){
2126 sessionAppendBlob(pBuf, p->aRecord, p->nRecord, &rc);
2127 }else{
2128 int i;
2129 u8 *a = p->aRecord;
2130 for(i=0; i<nCol; i++){
2131 u8 *pStart = a;
2132 int eType = *a++;
2134 switch( eType ){
2135 case 0:
2136 case SQLITE_NULL:
2137 assert( abPK[i]==0 );
2138 break;
2140 case SQLITE_FLOAT:
2141 case SQLITE_INTEGER:
2142 a += 8;
2143 break;
2145 default: {
2146 int n;
2147 a += sessionVarintGet(a, &n);
2148 a += n;
2149 break;
2152 if( abPK[i] ){
2153 sessionAppendBlob(pBuf, pStart, (int)(a-pStart), &rc);
2156 assert( (a - p->aRecord)==p->nRecord );
2159 return rc;
2163 ** Formulate and prepare a SELECT statement to retrieve a row from table
2164 ** zTab in database zDb based on its primary key. i.e.
2166 ** SELECT * FROM zDb.zTab WHERE pk1 = ? AND pk2 = ? AND ...
2168 static int sessionSelectStmt(
2169 sqlite3 *db, /* Database handle */
2170 const char *zDb, /* Database name */
2171 const char *zTab, /* Table name */
2172 int nCol, /* Number of columns in table */
2173 const char **azCol, /* Names of table columns */
2174 u8 *abPK, /* PRIMARY KEY array */
2175 sqlite3_stmt **ppStmt /* OUT: Prepared SELECT statement */
2177 int rc = SQLITE_OK;
2178 char *zSql = 0;
2179 int nSql = -1;
2181 if( 0==sqlite3_stricmp("sqlite_stat1", zTab) ){
2182 zSql = sqlite3_mprintf(
2183 "SELECT tbl, ?2, stat FROM %Q.sqlite_stat1 WHERE tbl IS ?1 AND "
2184 "idx IS (CASE WHEN ?2=X'' THEN NULL ELSE ?2 END)", zDb
2186 if( zSql==0 ) rc = SQLITE_NOMEM;
2187 }else{
2188 int i;
2189 const char *zSep = "";
2190 SessionBuffer buf = {0, 0, 0};
2192 sessionAppendStr(&buf, "SELECT * FROM ", &rc);
2193 sessionAppendIdent(&buf, zDb, &rc);
2194 sessionAppendStr(&buf, ".", &rc);
2195 sessionAppendIdent(&buf, zTab, &rc);
2196 sessionAppendStr(&buf, " WHERE ", &rc);
2197 for(i=0; i<nCol; i++){
2198 if( abPK[i] ){
2199 sessionAppendStr(&buf, zSep, &rc);
2200 sessionAppendIdent(&buf, azCol[i], &rc);
2201 sessionAppendStr(&buf, " IS ?", &rc);
2202 sessionAppendInteger(&buf, i+1, &rc);
2203 zSep = " AND ";
2206 zSql = (char*)buf.aBuf;
2207 nSql = buf.nBuf;
2210 if( rc==SQLITE_OK ){
2211 rc = sqlite3_prepare_v2(db, zSql, nSql, ppStmt, 0);
2213 sqlite3_free(zSql);
2214 return rc;
2218 ** Bind the PRIMARY KEY values from the change passed in argument pChange
2219 ** to the SELECT statement passed as the first argument. The SELECT statement
2220 ** is as prepared by function sessionSelectStmt().
2222 ** Return SQLITE_OK if all PK values are successfully bound, or an SQLite
2223 ** error code (e.g. SQLITE_NOMEM) otherwise.
2225 static int sessionSelectBind(
2226 sqlite3_stmt *pSelect, /* SELECT from sessionSelectStmt() */
2227 int nCol, /* Number of columns in table */
2228 u8 *abPK, /* PRIMARY KEY array */
2229 SessionChange *pChange /* Change structure */
2231 int i;
2232 int rc = SQLITE_OK;
2233 u8 *a = pChange->aRecord;
2235 for(i=0; i<nCol && rc==SQLITE_OK; i++){
2236 int eType = *a++;
2238 switch( eType ){
2239 case 0:
2240 case SQLITE_NULL:
2241 assert( abPK[i]==0 );
2242 break;
2244 case SQLITE_INTEGER: {
2245 if( abPK[i] ){
2246 i64 iVal = sessionGetI64(a);
2247 rc = sqlite3_bind_int64(pSelect, i+1, iVal);
2249 a += 8;
2250 break;
2253 case SQLITE_FLOAT: {
2254 if( abPK[i] ){
2255 double rVal;
2256 i64 iVal = sessionGetI64(a);
2257 memcpy(&rVal, &iVal, 8);
2258 rc = sqlite3_bind_double(pSelect, i+1, rVal);
2260 a += 8;
2261 break;
2264 case SQLITE_TEXT: {
2265 int n;
2266 a += sessionVarintGet(a, &n);
2267 if( abPK[i] ){
2268 rc = sqlite3_bind_text(pSelect, i+1, (char *)a, n, SQLITE_TRANSIENT);
2270 a += n;
2271 break;
2274 default: {
2275 int n;
2276 assert( eType==SQLITE_BLOB );
2277 a += sessionVarintGet(a, &n);
2278 if( abPK[i] ){
2279 rc = sqlite3_bind_blob(pSelect, i+1, a, n, SQLITE_TRANSIENT);
2281 a += n;
2282 break;
2287 return rc;
2291 ** This function is a no-op if *pRc is set to other than SQLITE_OK when it
2292 ** is called. Otherwise, append a serialized table header (part of the binary
2293 ** changeset format) to buffer *pBuf. If an error occurs, set *pRc to an
2294 ** SQLite error code before returning.
2296 static void sessionAppendTableHdr(
2297 SessionBuffer *pBuf, /* Append header to this buffer */
2298 int bPatchset, /* Use the patchset format if true */
2299 SessionTable *pTab, /* Table object to append header for */
2300 int *pRc /* IN/OUT: Error code */
2302 /* Write a table header */
2303 sessionAppendByte(pBuf, (bPatchset ? 'P' : 'T'), pRc);
2304 sessionAppendVarint(pBuf, pTab->nCol, pRc);
2305 sessionAppendBlob(pBuf, pTab->abPK, pTab->nCol, pRc);
2306 sessionAppendBlob(pBuf, (u8 *)pTab->zName, (int)strlen(pTab->zName)+1, pRc);
2310 ** Generate either a changeset (if argument bPatchset is zero) or a patchset
2311 ** (if it is non-zero) based on the current contents of the session object
2312 ** passed as the first argument.
2314 ** If no error occurs, SQLITE_OK is returned and the new changeset/patchset
2315 ** stored in output variables *pnChangeset and *ppChangeset. Or, if an error
2316 ** occurs, an SQLite error code is returned and both output variables set
2317 ** to 0.
2319 static int sessionGenerateChangeset(
2320 sqlite3_session *pSession, /* Session object */
2321 int bPatchset, /* True for patchset, false for changeset */
2322 int (*xOutput)(void *pOut, const void *pData, int nData),
2323 void *pOut, /* First argument for xOutput */
2324 int *pnChangeset, /* OUT: Size of buffer at *ppChangeset */
2325 void **ppChangeset /* OUT: Buffer containing changeset */
2327 sqlite3 *db = pSession->db; /* Source database handle */
2328 SessionTable *pTab; /* Used to iterate through attached tables */
2329 SessionBuffer buf = {0,0,0}; /* Buffer in which to accumlate changeset */
2330 int rc; /* Return code */
2332 assert( xOutput==0 || (pnChangeset==0 && ppChangeset==0 ) );
2334 /* Zero the output variables in case an error occurs. If this session
2335 ** object is already in the error state (sqlite3_session.rc != SQLITE_OK),
2336 ** this call will be a no-op. */
2337 if( xOutput==0 ){
2338 *pnChangeset = 0;
2339 *ppChangeset = 0;
2342 if( pSession->rc ) return pSession->rc;
2343 rc = sqlite3_exec(pSession->db, "SAVEPOINT changeset", 0, 0, 0);
2344 if( rc!=SQLITE_OK ) return rc;
2346 sqlite3_mutex_enter(sqlite3_db_mutex(db));
2348 for(pTab=pSession->pTable; rc==SQLITE_OK && pTab; pTab=pTab->pNext){
2349 if( pTab->nEntry ){
2350 const char *zName = pTab->zName;
2351 int nCol; /* Number of columns in table */
2352 u8 *abPK; /* Primary key array */
2353 const char **azCol = 0; /* Table columns */
2354 int i; /* Used to iterate through hash buckets */
2355 sqlite3_stmt *pSel = 0; /* SELECT statement to query table pTab */
2356 int nRewind = buf.nBuf; /* Initial size of write buffer */
2357 int nNoop; /* Size of buffer after writing tbl header */
2359 /* Check the table schema is still Ok. */
2360 rc = sessionTableInfo(db, pSession->zDb, zName, &nCol, 0, &azCol, &abPK);
2361 if( !rc && (pTab->nCol!=nCol || memcmp(abPK, pTab->abPK, nCol)) ){
2362 rc = SQLITE_SCHEMA;
2365 /* Write a table header */
2366 sessionAppendTableHdr(&buf, bPatchset, pTab, &rc);
2368 /* Build and compile a statement to execute: */
2369 if( rc==SQLITE_OK ){
2370 rc = sessionSelectStmt(
2371 db, pSession->zDb, zName, nCol, azCol, abPK, &pSel);
2374 nNoop = buf.nBuf;
2375 for(i=0; i<pTab->nChange && rc==SQLITE_OK; i++){
2376 SessionChange *p; /* Used to iterate through changes */
2378 for(p=pTab->apChange[i]; rc==SQLITE_OK && p; p=p->pNext){
2379 rc = sessionSelectBind(pSel, nCol, abPK, p);
2380 if( rc!=SQLITE_OK ) continue;
2381 if( sqlite3_step(pSel)==SQLITE_ROW ){
2382 if( p->op==SQLITE_INSERT ){
2383 int iCol;
2384 sessionAppendByte(&buf, SQLITE_INSERT, &rc);
2385 sessionAppendByte(&buf, p->bIndirect, &rc);
2386 for(iCol=0; iCol<nCol; iCol++){
2387 sessionAppendCol(&buf, pSel, iCol, &rc);
2389 }else{
2390 rc = sessionAppendUpdate(&buf, bPatchset, pSel, p, abPK);
2392 }else if( p->op!=SQLITE_INSERT ){
2393 rc = sessionAppendDelete(&buf, bPatchset, p, nCol, abPK);
2395 if( rc==SQLITE_OK ){
2396 rc = sqlite3_reset(pSel);
2399 /* If the buffer is now larger than SESSIONS_STRM_CHUNK_SIZE, pass
2400 ** its contents to the xOutput() callback. */
2401 if( xOutput
2402 && rc==SQLITE_OK
2403 && buf.nBuf>nNoop
2404 && buf.nBuf>SESSIONS_STRM_CHUNK_SIZE
2406 rc = xOutput(pOut, (void*)buf.aBuf, buf.nBuf);
2407 nNoop = -1;
2408 buf.nBuf = 0;
2414 sqlite3_finalize(pSel);
2415 if( buf.nBuf==nNoop ){
2416 buf.nBuf = nRewind;
2418 sqlite3_free((char*)azCol); /* cast works around VC++ bug */
2422 if( rc==SQLITE_OK ){
2423 if( xOutput==0 ){
2424 *pnChangeset = buf.nBuf;
2425 *ppChangeset = buf.aBuf;
2426 buf.aBuf = 0;
2427 }else if( buf.nBuf>0 ){
2428 rc = xOutput(pOut, (void*)buf.aBuf, buf.nBuf);
2432 sqlite3_free(buf.aBuf);
2433 sqlite3_exec(db, "RELEASE changeset", 0, 0, 0);
2434 sqlite3_mutex_leave(sqlite3_db_mutex(db));
2435 return rc;
2439 ** Obtain a changeset object containing all changes recorded by the
2440 ** session object passed as the first argument.
2442 ** It is the responsibility of the caller to eventually free the buffer
2443 ** using sqlite3_free().
2445 int sqlite3session_changeset(
2446 sqlite3_session *pSession, /* Session object */
2447 int *pnChangeset, /* OUT: Size of buffer at *ppChangeset */
2448 void **ppChangeset /* OUT: Buffer containing changeset */
2450 return sessionGenerateChangeset(pSession, 0, 0, 0, pnChangeset, ppChangeset);
2454 ** Streaming version of sqlite3session_changeset().
2456 int sqlite3session_changeset_strm(
2457 sqlite3_session *pSession,
2458 int (*xOutput)(void *pOut, const void *pData, int nData),
2459 void *pOut
2461 return sessionGenerateChangeset(pSession, 0, xOutput, pOut, 0, 0);
2465 ** Streaming version of sqlite3session_patchset().
2467 int sqlite3session_patchset_strm(
2468 sqlite3_session *pSession,
2469 int (*xOutput)(void *pOut, const void *pData, int nData),
2470 void *pOut
2472 return sessionGenerateChangeset(pSession, 1, xOutput, pOut, 0, 0);
2476 ** Obtain a patchset object containing all changes recorded by the
2477 ** session object passed as the first argument.
2479 ** It is the responsibility of the caller to eventually free the buffer
2480 ** using sqlite3_free().
2482 int sqlite3session_patchset(
2483 sqlite3_session *pSession, /* Session object */
2484 int *pnPatchset, /* OUT: Size of buffer at *ppChangeset */
2485 void **ppPatchset /* OUT: Buffer containing changeset */
2487 return sessionGenerateChangeset(pSession, 1, 0, 0, pnPatchset, ppPatchset);
2491 ** Enable or disable the session object passed as the first argument.
2493 int sqlite3session_enable(sqlite3_session *pSession, int bEnable){
2494 int ret;
2495 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
2496 if( bEnable>=0 ){
2497 pSession->bEnable = bEnable;
2499 ret = pSession->bEnable;
2500 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
2501 return ret;
2505 ** Enable or disable the session object passed as the first argument.
2507 int sqlite3session_indirect(sqlite3_session *pSession, int bIndirect){
2508 int ret;
2509 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
2510 if( bIndirect>=0 ){
2511 pSession->bIndirect = bIndirect;
2513 ret = pSession->bIndirect;
2514 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
2515 return ret;
2519 ** Return true if there have been no changes to monitored tables recorded
2520 ** by the session object passed as the only argument.
2522 int sqlite3session_isempty(sqlite3_session *pSession){
2523 int ret = 0;
2524 SessionTable *pTab;
2526 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
2527 for(pTab=pSession->pTable; pTab && ret==0; pTab=pTab->pNext){
2528 ret = (pTab->nEntry>0);
2530 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
2532 return (ret==0);
2536 ** Do the work for either sqlite3changeset_start() or start_strm().
2538 static int sessionChangesetStart(
2539 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2540 int (*xInput)(void *pIn, void *pData, int *pnData),
2541 void *pIn,
2542 int nChangeset, /* Size of buffer pChangeset in bytes */
2543 void *pChangeset /* Pointer to buffer containing changeset */
2545 sqlite3_changeset_iter *pRet; /* Iterator to return */
2546 int nByte; /* Number of bytes to allocate for iterator */
2548 assert( xInput==0 || (pChangeset==0 && nChangeset==0) );
2550 /* Zero the output variable in case an error occurs. */
2551 *pp = 0;
2553 /* Allocate and initialize the iterator structure. */
2554 nByte = sizeof(sqlite3_changeset_iter);
2555 pRet = (sqlite3_changeset_iter *)sqlite3_malloc(nByte);
2556 if( !pRet ) return SQLITE_NOMEM;
2557 memset(pRet, 0, sizeof(sqlite3_changeset_iter));
2558 pRet->in.aData = (u8 *)pChangeset;
2559 pRet->in.nData = nChangeset;
2560 pRet->in.xInput = xInput;
2561 pRet->in.pIn = pIn;
2562 pRet->in.bEof = (xInput ? 0 : 1);
2564 /* Populate the output variable and return success. */
2565 *pp = pRet;
2566 return SQLITE_OK;
2570 ** Create an iterator used to iterate through the contents of a changeset.
2572 int sqlite3changeset_start(
2573 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2574 int nChangeset, /* Size of buffer pChangeset in bytes */
2575 void *pChangeset /* Pointer to buffer containing changeset */
2577 return sessionChangesetStart(pp, 0, 0, nChangeset, pChangeset);
2581 ** Streaming version of sqlite3changeset_start().
2583 int sqlite3changeset_start_strm(
2584 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2585 int (*xInput)(void *pIn, void *pData, int *pnData),
2586 void *pIn
2588 return sessionChangesetStart(pp, xInput, pIn, 0, 0);
2592 ** If the SessionInput object passed as the only argument is a streaming
2593 ** object and the buffer is full, discard some data to free up space.
2595 static void sessionDiscardData(SessionInput *pIn){
2596 if( pIn->xInput && pIn->iNext>=SESSIONS_STRM_CHUNK_SIZE ){
2597 int nMove = pIn->buf.nBuf - pIn->iNext;
2598 assert( nMove>=0 );
2599 if( nMove>0 ){
2600 memmove(pIn->buf.aBuf, &pIn->buf.aBuf[pIn->iNext], nMove);
2602 pIn->buf.nBuf -= pIn->iNext;
2603 pIn->iNext = 0;
2604 pIn->nData = pIn->buf.nBuf;
2609 ** Ensure that there are at least nByte bytes available in the buffer. Or,
2610 ** if there are not nByte bytes remaining in the input, that all available
2611 ** data is in the buffer.
2613 ** Return an SQLite error code if an error occurs, or SQLITE_OK otherwise.
2615 static int sessionInputBuffer(SessionInput *pIn, int nByte){
2616 int rc = SQLITE_OK;
2617 if( pIn->xInput ){
2618 while( !pIn->bEof && (pIn->iNext+nByte)>=pIn->nData && rc==SQLITE_OK ){
2619 int nNew = SESSIONS_STRM_CHUNK_SIZE;
2621 if( pIn->bNoDiscard==0 ) sessionDiscardData(pIn);
2622 if( SQLITE_OK==sessionBufferGrow(&pIn->buf, nNew, &rc) ){
2623 rc = pIn->xInput(pIn->pIn, &pIn->buf.aBuf[pIn->buf.nBuf], &nNew);
2624 if( nNew==0 ){
2625 pIn->bEof = 1;
2626 }else{
2627 pIn->buf.nBuf += nNew;
2631 pIn->aData = pIn->buf.aBuf;
2632 pIn->nData = pIn->buf.nBuf;
2635 return rc;
2639 ** When this function is called, *ppRec points to the start of a record
2640 ** that contains nCol values. This function advances the pointer *ppRec
2641 ** until it points to the byte immediately following that record.
2643 static void sessionSkipRecord(
2644 u8 **ppRec, /* IN/OUT: Record pointer */
2645 int nCol /* Number of values in record */
2647 u8 *aRec = *ppRec;
2648 int i;
2649 for(i=0; i<nCol; i++){
2650 int eType = *aRec++;
2651 if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
2652 int nByte;
2653 aRec += sessionVarintGet((u8*)aRec, &nByte);
2654 aRec += nByte;
2655 }else if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
2656 aRec += 8;
2660 *ppRec = aRec;
2664 ** This function sets the value of the sqlite3_value object passed as the
2665 ** first argument to a copy of the string or blob held in the aData[]
2666 ** buffer. SQLITE_OK is returned if successful, or SQLITE_NOMEM if an OOM
2667 ** error occurs.
2669 static int sessionValueSetStr(
2670 sqlite3_value *pVal, /* Set the value of this object */
2671 u8 *aData, /* Buffer containing string or blob data */
2672 int nData, /* Size of buffer aData[] in bytes */
2673 u8 enc /* String encoding (0 for blobs) */
2675 /* In theory this code could just pass SQLITE_TRANSIENT as the final
2676 ** argument to sqlite3ValueSetStr() and have the copy created
2677 ** automatically. But doing so makes it difficult to detect any OOM
2678 ** error. Hence the code to create the copy externally. */
2679 u8 *aCopy = sqlite3_malloc(nData+1);
2680 if( aCopy==0 ) return SQLITE_NOMEM;
2681 memcpy(aCopy, aData, nData);
2682 sqlite3ValueSetStr(pVal, nData, (char*)aCopy, enc, sqlite3_free);
2683 return SQLITE_OK;
2687 ** Deserialize a single record from a buffer in memory. See "RECORD FORMAT"
2688 ** for details.
2690 ** When this function is called, *paChange points to the start of the record
2691 ** to deserialize. Assuming no error occurs, *paChange is set to point to
2692 ** one byte after the end of the same record before this function returns.
2693 ** If the argument abPK is NULL, then the record contains nCol values. Or,
2694 ** if abPK is other than NULL, then the record contains only the PK fields
2695 ** (in other words, it is a patchset DELETE record).
2697 ** If successful, each element of the apOut[] array (allocated by the caller)
2698 ** is set to point to an sqlite3_value object containing the value read
2699 ** from the corresponding position in the record. If that value is not
2700 ** included in the record (i.e. because the record is part of an UPDATE change
2701 ** and the field was not modified), the corresponding element of apOut[] is
2702 ** set to NULL.
2704 ** It is the responsibility of the caller to free all sqlite_value structures
2705 ** using sqlite3_free().
2707 ** If an error occurs, an SQLite error code (e.g. SQLITE_NOMEM) is returned.
2708 ** The apOut[] array may have been partially populated in this case.
2710 static int sessionReadRecord(
2711 SessionInput *pIn, /* Input data */
2712 int nCol, /* Number of values in record */
2713 u8 *abPK, /* Array of primary key flags, or NULL */
2714 sqlite3_value **apOut /* Write values to this array */
2716 int i; /* Used to iterate through columns */
2717 int rc = SQLITE_OK;
2719 for(i=0; i<nCol && rc==SQLITE_OK; i++){
2720 int eType = 0; /* Type of value (SQLITE_NULL, TEXT etc.) */
2721 if( abPK && abPK[i]==0 ) continue;
2722 rc = sessionInputBuffer(pIn, 9);
2723 if( rc==SQLITE_OK ){
2724 if( pIn->iNext>=pIn->nData ){
2725 rc = SQLITE_CORRUPT_BKPT;
2726 }else{
2727 eType = pIn->aData[pIn->iNext++];
2728 assert( apOut[i]==0 );
2729 if( eType ){
2730 apOut[i] = sqlite3ValueNew(0);
2731 if( !apOut[i] ) rc = SQLITE_NOMEM;
2736 if( rc==SQLITE_OK ){
2737 u8 *aVal = &pIn->aData[pIn->iNext];
2738 if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
2739 int nByte;
2740 pIn->iNext += sessionVarintGet(aVal, &nByte);
2741 rc = sessionInputBuffer(pIn, nByte);
2742 if( rc==SQLITE_OK ){
2743 if( nByte<0 || nByte>pIn->nData-pIn->iNext ){
2744 rc = SQLITE_CORRUPT_BKPT;
2745 }else{
2746 u8 enc = (eType==SQLITE_TEXT ? SQLITE_UTF8 : 0);
2747 rc = sessionValueSetStr(apOut[i],&pIn->aData[pIn->iNext],nByte,enc);
2748 pIn->iNext += nByte;
2752 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
2753 sqlite3_int64 v = sessionGetI64(aVal);
2754 if( eType==SQLITE_INTEGER ){
2755 sqlite3VdbeMemSetInt64(apOut[i], v);
2756 }else{
2757 double d;
2758 memcpy(&d, &v, 8);
2759 sqlite3VdbeMemSetDouble(apOut[i], d);
2761 pIn->iNext += 8;
2766 return rc;
2770 ** The input pointer currently points to the second byte of a table-header.
2771 ** Specifically, to the following:
2773 ** + number of columns in table (varint)
2774 ** + array of PK flags (1 byte per column),
2775 ** + table name (nul terminated).
2777 ** This function ensures that all of the above is present in the input
2778 ** buffer (i.e. that it can be accessed without any calls to xInput()).
2779 ** If successful, SQLITE_OK is returned. Otherwise, an SQLite error code.
2780 ** The input pointer is not moved.
2782 static int sessionChangesetBufferTblhdr(SessionInput *pIn, int *pnByte){
2783 int rc = SQLITE_OK;
2784 int nCol = 0;
2785 int nRead = 0;
2787 rc = sessionInputBuffer(pIn, 9);
2788 if( rc==SQLITE_OK ){
2789 nRead += sessionVarintGet(&pIn->aData[pIn->iNext + nRead], &nCol);
2790 /* The hard upper limit for the number of columns in an SQLite
2791 ** database table is, according to sqliteLimit.h, 32676. So
2792 ** consider any table-header that purports to have more than 65536
2793 ** columns to be corrupt. This is convenient because otherwise,
2794 ** if the (nCol>65536) condition below were omitted, a sufficiently
2795 ** large value for nCol may cause nRead to wrap around and become
2796 ** negative. Leading to a crash. */
2797 if( nCol<0 || nCol>65536 ){
2798 rc = SQLITE_CORRUPT_BKPT;
2799 }else{
2800 rc = sessionInputBuffer(pIn, nRead+nCol+100);
2801 nRead += nCol;
2805 while( rc==SQLITE_OK ){
2806 while( (pIn->iNext + nRead)<pIn->nData && pIn->aData[pIn->iNext + nRead] ){
2807 nRead++;
2809 if( (pIn->iNext + nRead)<pIn->nData ) break;
2810 rc = sessionInputBuffer(pIn, nRead + 100);
2812 *pnByte = nRead+1;
2813 return rc;
2817 ** The input pointer currently points to the first byte of the first field
2818 ** of a record consisting of nCol columns. This function ensures the entire
2819 ** record is buffered. It does not move the input pointer.
2821 ** If successful, SQLITE_OK is returned and *pnByte is set to the size of
2822 ** the record in bytes. Otherwise, an SQLite error code is returned. The
2823 ** final value of *pnByte is undefined in this case.
2825 static int sessionChangesetBufferRecord(
2826 SessionInput *pIn, /* Input data */
2827 int nCol, /* Number of columns in record */
2828 int *pnByte /* OUT: Size of record in bytes */
2830 int rc = SQLITE_OK;
2831 int nByte = 0;
2832 int i;
2833 for(i=0; rc==SQLITE_OK && i<nCol; i++){
2834 int eType;
2835 rc = sessionInputBuffer(pIn, nByte + 10);
2836 if( rc==SQLITE_OK ){
2837 eType = pIn->aData[pIn->iNext + nByte++];
2838 if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
2839 int n;
2840 nByte += sessionVarintGet(&pIn->aData[pIn->iNext+nByte], &n);
2841 nByte += n;
2842 rc = sessionInputBuffer(pIn, nByte);
2843 }else if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
2844 nByte += 8;
2848 *pnByte = nByte;
2849 return rc;
2853 ** The input pointer currently points to the second byte of a table-header.
2854 ** Specifically, to the following:
2856 ** + number of columns in table (varint)
2857 ** + array of PK flags (1 byte per column),
2858 ** + table name (nul terminated).
2860 ** This function decodes the table-header and populates the p->nCol,
2861 ** p->zTab and p->abPK[] variables accordingly. The p->apValue[] array is
2862 ** also allocated or resized according to the new value of p->nCol. The
2863 ** input pointer is left pointing to the byte following the table header.
2865 ** If successful, SQLITE_OK is returned. Otherwise, an SQLite error code
2866 ** is returned and the final values of the various fields enumerated above
2867 ** are undefined.
2869 static int sessionChangesetReadTblhdr(sqlite3_changeset_iter *p){
2870 int rc;
2871 int nCopy;
2872 assert( p->rc==SQLITE_OK );
2874 rc = sessionChangesetBufferTblhdr(&p->in, &nCopy);
2875 if( rc==SQLITE_OK ){
2876 int nByte;
2877 int nVarint;
2878 nVarint = sessionVarintGet(&p->in.aData[p->in.iNext], &p->nCol);
2879 if( p->nCol>0 ){
2880 nCopy -= nVarint;
2881 p->in.iNext += nVarint;
2882 nByte = p->nCol * sizeof(sqlite3_value*) * 2 + nCopy;
2883 p->tblhdr.nBuf = 0;
2884 sessionBufferGrow(&p->tblhdr, nByte, &rc);
2885 }else{
2886 rc = SQLITE_CORRUPT_BKPT;
2890 if( rc==SQLITE_OK ){
2891 int iPK = sizeof(sqlite3_value*)*p->nCol*2;
2892 memset(p->tblhdr.aBuf, 0, iPK);
2893 memcpy(&p->tblhdr.aBuf[iPK], &p->in.aData[p->in.iNext], nCopy);
2894 p->in.iNext += nCopy;
2897 p->apValue = (sqlite3_value**)p->tblhdr.aBuf;
2898 p->abPK = (u8*)&p->apValue[p->nCol*2];
2899 p->zTab = (char*)&p->abPK[p->nCol];
2900 return (p->rc = rc);
2904 ** Advance the changeset iterator to the next change.
2906 ** If both paRec and pnRec are NULL, then this function works like the public
2907 ** API sqlite3changeset_next(). If SQLITE_ROW is returned, then the
2908 ** sqlite3changeset_new() and old() APIs may be used to query for values.
2910 ** Otherwise, if paRec and pnRec are not NULL, then a pointer to the change
2911 ** record is written to *paRec before returning and the number of bytes in
2912 ** the record to *pnRec.
2914 ** Either way, this function returns SQLITE_ROW if the iterator is
2915 ** successfully advanced to the next change in the changeset, an SQLite
2916 ** error code if an error occurs, or SQLITE_DONE if there are no further
2917 ** changes in the changeset.
2919 static int sessionChangesetNext(
2920 sqlite3_changeset_iter *p, /* Changeset iterator */
2921 u8 **paRec, /* If non-NULL, store record pointer here */
2922 int *pnRec, /* If non-NULL, store size of record here */
2923 int *pbNew /* If non-NULL, true if new table */
2925 int i;
2926 u8 op;
2928 assert( (paRec==0 && pnRec==0) || (paRec && pnRec) );
2930 /* If the iterator is in the error-state, return immediately. */
2931 if( p->rc!=SQLITE_OK ) return p->rc;
2933 /* Free the current contents of p->apValue[], if any. */
2934 if( p->apValue ){
2935 for(i=0; i<p->nCol*2; i++){
2936 sqlite3ValueFree(p->apValue[i]);
2938 memset(p->apValue, 0, sizeof(sqlite3_value*)*p->nCol*2);
2941 /* Make sure the buffer contains at least 10 bytes of input data, or all
2942 ** remaining data if there are less than 10 bytes available. This is
2943 ** sufficient either for the 'T' or 'P' byte and the varint that follows
2944 ** it, or for the two single byte values otherwise. */
2945 p->rc = sessionInputBuffer(&p->in, 2);
2946 if( p->rc!=SQLITE_OK ) return p->rc;
2948 /* If the iterator is already at the end of the changeset, return DONE. */
2949 if( p->in.iNext>=p->in.nData ){
2950 return SQLITE_DONE;
2953 sessionDiscardData(&p->in);
2954 p->in.iCurrent = p->in.iNext;
2956 op = p->in.aData[p->in.iNext++];
2957 while( op=='T' || op=='P' ){
2958 if( pbNew ) *pbNew = 1;
2959 p->bPatchset = (op=='P');
2960 if( sessionChangesetReadTblhdr(p) ) return p->rc;
2961 if( (p->rc = sessionInputBuffer(&p->in, 2)) ) return p->rc;
2962 p->in.iCurrent = p->in.iNext;
2963 if( p->in.iNext>=p->in.nData ) return SQLITE_DONE;
2964 op = p->in.aData[p->in.iNext++];
2967 if( p->zTab==0 ){
2968 /* The first record in the changeset is not a table header. Must be a
2969 ** corrupt changeset. */
2970 assert( p->in.iNext==1 );
2971 return (p->rc = SQLITE_CORRUPT_BKPT);
2974 p->op = op;
2975 p->bIndirect = p->in.aData[p->in.iNext++];
2976 if( p->op!=SQLITE_UPDATE && p->op!=SQLITE_DELETE && p->op!=SQLITE_INSERT ){
2977 return (p->rc = SQLITE_CORRUPT_BKPT);
2980 if( paRec ){
2981 int nVal; /* Number of values to buffer */
2982 if( p->bPatchset==0 && op==SQLITE_UPDATE ){
2983 nVal = p->nCol * 2;
2984 }else if( p->bPatchset && op==SQLITE_DELETE ){
2985 nVal = 0;
2986 for(i=0; i<p->nCol; i++) if( p->abPK[i] ) nVal++;
2987 }else{
2988 nVal = p->nCol;
2990 p->rc = sessionChangesetBufferRecord(&p->in, nVal, pnRec);
2991 if( p->rc!=SQLITE_OK ) return p->rc;
2992 *paRec = &p->in.aData[p->in.iNext];
2993 p->in.iNext += *pnRec;
2994 }else{
2996 /* If this is an UPDATE or DELETE, read the old.* record. */
2997 if( p->op!=SQLITE_INSERT && (p->bPatchset==0 || p->op==SQLITE_DELETE) ){
2998 u8 *abPK = p->bPatchset ? p->abPK : 0;
2999 p->rc = sessionReadRecord(&p->in, p->nCol, abPK, p->apValue);
3000 if( p->rc!=SQLITE_OK ) return p->rc;
3003 /* If this is an INSERT or UPDATE, read the new.* record. */
3004 if( p->op!=SQLITE_DELETE ){
3005 p->rc = sessionReadRecord(&p->in, p->nCol, 0, &p->apValue[p->nCol]);
3006 if( p->rc!=SQLITE_OK ) return p->rc;
3009 if( p->bPatchset && p->op==SQLITE_UPDATE ){
3010 /* If this is an UPDATE that is part of a patchset, then all PK and
3011 ** modified fields are present in the new.* record. The old.* record
3012 ** is currently completely empty. This block shifts the PK fields from
3013 ** new.* to old.*, to accommodate the code that reads these arrays. */
3014 for(i=0; i<p->nCol; i++){
3015 assert( p->apValue[i]==0 );
3016 if( p->abPK[i] ){
3017 p->apValue[i] = p->apValue[i+p->nCol];
3018 if( p->apValue[i]==0 ) return (p->rc = SQLITE_CORRUPT_BKPT);
3019 p->apValue[i+p->nCol] = 0;
3025 return SQLITE_ROW;
3029 ** Advance an iterator created by sqlite3changeset_start() to the next
3030 ** change in the changeset. This function may return SQLITE_ROW, SQLITE_DONE
3031 ** or SQLITE_CORRUPT.
3033 ** This function may not be called on iterators passed to a conflict handler
3034 ** callback by changeset_apply().
3036 int sqlite3changeset_next(sqlite3_changeset_iter *p){
3037 return sessionChangesetNext(p, 0, 0, 0);
3041 ** The following function extracts information on the current change
3042 ** from a changeset iterator. It may only be called after changeset_next()
3043 ** has returned SQLITE_ROW.
3045 int sqlite3changeset_op(
3046 sqlite3_changeset_iter *pIter, /* Iterator handle */
3047 const char **pzTab, /* OUT: Pointer to table name */
3048 int *pnCol, /* OUT: Number of columns in table */
3049 int *pOp, /* OUT: SQLITE_INSERT, DELETE or UPDATE */
3050 int *pbIndirect /* OUT: True if change is indirect */
3052 *pOp = pIter->op;
3053 *pnCol = pIter->nCol;
3054 *pzTab = pIter->zTab;
3055 if( pbIndirect ) *pbIndirect = pIter->bIndirect;
3056 return SQLITE_OK;
3060 ** Return information regarding the PRIMARY KEY and number of columns in
3061 ** the database table affected by the change that pIter currently points
3062 ** to. This function may only be called after changeset_next() returns
3063 ** SQLITE_ROW.
3065 int sqlite3changeset_pk(
3066 sqlite3_changeset_iter *pIter, /* Iterator object */
3067 unsigned char **pabPK, /* OUT: Array of boolean - true for PK cols */
3068 int *pnCol /* OUT: Number of entries in output array */
3070 *pabPK = pIter->abPK;
3071 if( pnCol ) *pnCol = pIter->nCol;
3072 return SQLITE_OK;
3076 ** This function may only be called while the iterator is pointing to an
3077 ** SQLITE_UPDATE or SQLITE_DELETE change (see sqlite3changeset_op()).
3078 ** Otherwise, SQLITE_MISUSE is returned.
3080 ** It sets *ppValue to point to an sqlite3_value structure containing the
3081 ** iVal'th value in the old.* record. Or, if that particular value is not
3082 ** included in the record (because the change is an UPDATE and the field
3083 ** was not modified and is not a PK column), set *ppValue to NULL.
3085 ** If value iVal is out-of-range, SQLITE_RANGE is returned and *ppValue is
3086 ** not modified. Otherwise, SQLITE_OK.
3088 int sqlite3changeset_old(
3089 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3090 int iVal, /* Index of old.* value to retrieve */
3091 sqlite3_value **ppValue /* OUT: Old value (or NULL pointer) */
3093 if( pIter->op!=SQLITE_UPDATE && pIter->op!=SQLITE_DELETE ){
3094 return SQLITE_MISUSE;
3096 if( iVal<0 || iVal>=pIter->nCol ){
3097 return SQLITE_RANGE;
3099 *ppValue = pIter->apValue[iVal];
3100 return SQLITE_OK;
3104 ** This function may only be called while the iterator is pointing to an
3105 ** SQLITE_UPDATE or SQLITE_INSERT change (see sqlite3changeset_op()).
3106 ** Otherwise, SQLITE_MISUSE is returned.
3108 ** It sets *ppValue to point to an sqlite3_value structure containing the
3109 ** iVal'th value in the new.* record. Or, if that particular value is not
3110 ** included in the record (because the change is an UPDATE and the field
3111 ** was not modified), set *ppValue to NULL.
3113 ** If value iVal is out-of-range, SQLITE_RANGE is returned and *ppValue is
3114 ** not modified. Otherwise, SQLITE_OK.
3116 int sqlite3changeset_new(
3117 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3118 int iVal, /* Index of new.* value to retrieve */
3119 sqlite3_value **ppValue /* OUT: New value (or NULL pointer) */
3121 if( pIter->op!=SQLITE_UPDATE && pIter->op!=SQLITE_INSERT ){
3122 return SQLITE_MISUSE;
3124 if( iVal<0 || iVal>=pIter->nCol ){
3125 return SQLITE_RANGE;
3127 *ppValue = pIter->apValue[pIter->nCol+iVal];
3128 return SQLITE_OK;
3132 ** The following two macros are used internally. They are similar to the
3133 ** sqlite3changeset_new() and sqlite3changeset_old() functions, except that
3134 ** they omit all error checking and return a pointer to the requested value.
3136 #define sessionChangesetNew(pIter, iVal) (pIter)->apValue[(pIter)->nCol+(iVal)]
3137 #define sessionChangesetOld(pIter, iVal) (pIter)->apValue[(iVal)]
3140 ** This function may only be called with a changeset iterator that has been
3141 ** passed to an SQLITE_CHANGESET_DATA or SQLITE_CHANGESET_CONFLICT
3142 ** conflict-handler function. Otherwise, SQLITE_MISUSE is returned.
3144 ** If successful, *ppValue is set to point to an sqlite3_value structure
3145 ** containing the iVal'th value of the conflicting record.
3147 ** If value iVal is out-of-range or some other error occurs, an SQLite error
3148 ** code is returned. Otherwise, SQLITE_OK.
3150 int sqlite3changeset_conflict(
3151 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3152 int iVal, /* Index of conflict record value to fetch */
3153 sqlite3_value **ppValue /* OUT: Value from conflicting row */
3155 if( !pIter->pConflict ){
3156 return SQLITE_MISUSE;
3158 if( iVal<0 || iVal>=pIter->nCol ){
3159 return SQLITE_RANGE;
3161 *ppValue = sqlite3_column_value(pIter->pConflict, iVal);
3162 return SQLITE_OK;
3166 ** This function may only be called with an iterator passed to an
3167 ** SQLITE_CHANGESET_FOREIGN_KEY conflict handler callback. In this case
3168 ** it sets the output variable to the total number of known foreign key
3169 ** violations in the destination database and returns SQLITE_OK.
3171 ** In all other cases this function returns SQLITE_MISUSE.
3173 int sqlite3changeset_fk_conflicts(
3174 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3175 int *pnOut /* OUT: Number of FK violations */
3177 if( pIter->pConflict || pIter->apValue ){
3178 return SQLITE_MISUSE;
3180 *pnOut = pIter->nCol;
3181 return SQLITE_OK;
3186 ** Finalize an iterator allocated with sqlite3changeset_start().
3188 ** This function may not be called on iterators passed to a conflict handler
3189 ** callback by changeset_apply().
3191 int sqlite3changeset_finalize(sqlite3_changeset_iter *p){
3192 int rc = SQLITE_OK;
3193 if( p ){
3194 int i; /* Used to iterate through p->apValue[] */
3195 rc = p->rc;
3196 if( p->apValue ){
3197 for(i=0; i<p->nCol*2; i++) sqlite3ValueFree(p->apValue[i]);
3199 sqlite3_free(p->tblhdr.aBuf);
3200 sqlite3_free(p->in.buf.aBuf);
3201 sqlite3_free(p);
3203 return rc;
3206 static int sessionChangesetInvert(
3207 SessionInput *pInput, /* Input changeset */
3208 int (*xOutput)(void *pOut, const void *pData, int nData),
3209 void *pOut,
3210 int *pnInverted, /* OUT: Number of bytes in output changeset */
3211 void **ppInverted /* OUT: Inverse of pChangeset */
3213 int rc = SQLITE_OK; /* Return value */
3214 SessionBuffer sOut; /* Output buffer */
3215 int nCol = 0; /* Number of cols in current table */
3216 u8 *abPK = 0; /* PK array for current table */
3217 sqlite3_value **apVal = 0; /* Space for values for UPDATE inversion */
3218 SessionBuffer sPK = {0, 0, 0}; /* PK array for current table */
3220 /* Initialize the output buffer */
3221 memset(&sOut, 0, sizeof(SessionBuffer));
3223 /* Zero the output variables in case an error occurs. */
3224 if( ppInverted ){
3225 *ppInverted = 0;
3226 *pnInverted = 0;
3229 while( 1 ){
3230 u8 eType;
3232 /* Test for EOF. */
3233 if( (rc = sessionInputBuffer(pInput, 2)) ) goto finished_invert;
3234 if( pInput->iNext>=pInput->nData ) break;
3235 eType = pInput->aData[pInput->iNext];
3237 switch( eType ){
3238 case 'T': {
3239 /* A 'table' record consists of:
3241 ** * A constant 'T' character,
3242 ** * Number of columns in said table (a varint),
3243 ** * An array of nCol bytes (sPK),
3244 ** * A nul-terminated table name.
3246 int nByte;
3247 int nVar;
3248 pInput->iNext++;
3249 if( (rc = sessionChangesetBufferTblhdr(pInput, &nByte)) ){
3250 goto finished_invert;
3252 nVar = sessionVarintGet(&pInput->aData[pInput->iNext], &nCol);
3253 sPK.nBuf = 0;
3254 sessionAppendBlob(&sPK, &pInput->aData[pInput->iNext+nVar], nCol, &rc);
3255 sessionAppendByte(&sOut, eType, &rc);
3256 sessionAppendBlob(&sOut, &pInput->aData[pInput->iNext], nByte, &rc);
3257 if( rc ) goto finished_invert;
3259 pInput->iNext += nByte;
3260 sqlite3_free(apVal);
3261 apVal = 0;
3262 abPK = sPK.aBuf;
3263 break;
3266 case SQLITE_INSERT:
3267 case SQLITE_DELETE: {
3268 int nByte;
3269 int bIndirect = pInput->aData[pInput->iNext+1];
3270 int eType2 = (eType==SQLITE_DELETE ? SQLITE_INSERT : SQLITE_DELETE);
3271 pInput->iNext += 2;
3272 assert( rc==SQLITE_OK );
3273 rc = sessionChangesetBufferRecord(pInput, nCol, &nByte);
3274 sessionAppendByte(&sOut, eType2, &rc);
3275 sessionAppendByte(&sOut, bIndirect, &rc);
3276 sessionAppendBlob(&sOut, &pInput->aData[pInput->iNext], nByte, &rc);
3277 pInput->iNext += nByte;
3278 if( rc ) goto finished_invert;
3279 break;
3282 case SQLITE_UPDATE: {
3283 int iCol;
3285 if( 0==apVal ){
3286 apVal = (sqlite3_value **)sqlite3_malloc(sizeof(apVal[0])*nCol*2);
3287 if( 0==apVal ){
3288 rc = SQLITE_NOMEM;
3289 goto finished_invert;
3291 memset(apVal, 0, sizeof(apVal[0])*nCol*2);
3294 /* Write the header for the new UPDATE change. Same as the original. */
3295 sessionAppendByte(&sOut, eType, &rc);
3296 sessionAppendByte(&sOut, pInput->aData[pInput->iNext+1], &rc);
3298 /* Read the old.* and new.* records for the update change. */
3299 pInput->iNext += 2;
3300 rc = sessionReadRecord(pInput, nCol, 0, &apVal[0]);
3301 if( rc==SQLITE_OK ){
3302 rc = sessionReadRecord(pInput, nCol, 0, &apVal[nCol]);
3305 /* Write the new old.* record. Consists of the PK columns from the
3306 ** original old.* record, and the other values from the original
3307 ** new.* record. */
3308 for(iCol=0; iCol<nCol; iCol++){
3309 sqlite3_value *pVal = apVal[iCol + (abPK[iCol] ? 0 : nCol)];
3310 sessionAppendValue(&sOut, pVal, &rc);
3313 /* Write the new new.* record. Consists of a copy of all values
3314 ** from the original old.* record, except for the PK columns, which
3315 ** are set to "undefined". */
3316 for(iCol=0; iCol<nCol; iCol++){
3317 sqlite3_value *pVal = (abPK[iCol] ? 0 : apVal[iCol]);
3318 sessionAppendValue(&sOut, pVal, &rc);
3321 for(iCol=0; iCol<nCol*2; iCol++){
3322 sqlite3ValueFree(apVal[iCol]);
3324 memset(apVal, 0, sizeof(apVal[0])*nCol*2);
3325 if( rc!=SQLITE_OK ){
3326 goto finished_invert;
3329 break;
3332 default:
3333 rc = SQLITE_CORRUPT_BKPT;
3334 goto finished_invert;
3337 assert( rc==SQLITE_OK );
3338 if( xOutput && sOut.nBuf>=SESSIONS_STRM_CHUNK_SIZE ){
3339 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf);
3340 sOut.nBuf = 0;
3341 if( rc!=SQLITE_OK ) goto finished_invert;
3345 assert( rc==SQLITE_OK );
3346 if( pnInverted ){
3347 *pnInverted = sOut.nBuf;
3348 *ppInverted = sOut.aBuf;
3349 sOut.aBuf = 0;
3350 }else if( sOut.nBuf>0 ){
3351 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf);
3354 finished_invert:
3355 sqlite3_free(sOut.aBuf);
3356 sqlite3_free(apVal);
3357 sqlite3_free(sPK.aBuf);
3358 return rc;
3363 ** Invert a changeset object.
3365 int sqlite3changeset_invert(
3366 int nChangeset, /* Number of bytes in input */
3367 const void *pChangeset, /* Input changeset */
3368 int *pnInverted, /* OUT: Number of bytes in output changeset */
3369 void **ppInverted /* OUT: Inverse of pChangeset */
3371 SessionInput sInput;
3373 /* Set up the input stream */
3374 memset(&sInput, 0, sizeof(SessionInput));
3375 sInput.nData = nChangeset;
3376 sInput.aData = (u8*)pChangeset;
3378 return sessionChangesetInvert(&sInput, 0, 0, pnInverted, ppInverted);
3382 ** Streaming version of sqlite3changeset_invert().
3384 int sqlite3changeset_invert_strm(
3385 int (*xInput)(void *pIn, void *pData, int *pnData),
3386 void *pIn,
3387 int (*xOutput)(void *pOut, const void *pData, int nData),
3388 void *pOut
3390 SessionInput sInput;
3391 int rc;
3393 /* Set up the input stream */
3394 memset(&sInput, 0, sizeof(SessionInput));
3395 sInput.xInput = xInput;
3396 sInput.pIn = pIn;
3398 rc = sessionChangesetInvert(&sInput, xOutput, pOut, 0, 0);
3399 sqlite3_free(sInput.buf.aBuf);
3400 return rc;
3403 typedef struct SessionApplyCtx SessionApplyCtx;
3404 struct SessionApplyCtx {
3405 sqlite3 *db;
3406 sqlite3_stmt *pDelete; /* DELETE statement */
3407 sqlite3_stmt *pUpdate; /* UPDATE statement */
3408 sqlite3_stmt *pInsert; /* INSERT statement */
3409 sqlite3_stmt *pSelect; /* SELECT statement */
3410 int nCol; /* Size of azCol[] and abPK[] arrays */
3411 const char **azCol; /* Array of column names */
3412 u8 *abPK; /* Boolean array - true if column is in PK */
3413 int bStat1; /* True if table is sqlite_stat1 */
3414 int bDeferConstraints; /* True to defer constraints */
3415 SessionBuffer constraints; /* Deferred constraints are stored here */
3416 SessionBuffer rebase; /* Rebase information (if any) here */
3417 int bRebaseStarted; /* If table header is already in rebase */
3421 ** Formulate a statement to DELETE a row from database db. Assuming a table
3422 ** structure like this:
3424 ** CREATE TABLE x(a, b, c, d, PRIMARY KEY(a, c));
3426 ** The DELETE statement looks like this:
3428 ** DELETE FROM x WHERE a = :1 AND c = :3 AND (:5 OR b IS :2 AND d IS :4)
3430 ** Variable :5 (nCol+1) is a boolean. It should be set to 0 if we require
3431 ** matching b and d values, or 1 otherwise. The second case comes up if the
3432 ** conflict handler is invoked with NOTFOUND and returns CHANGESET_REPLACE.
3434 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pDelete is left
3435 ** pointing to the prepared version of the SQL statement.
3437 static int sessionDeleteRow(
3438 sqlite3 *db, /* Database handle */
3439 const char *zTab, /* Table name */
3440 SessionApplyCtx *p /* Session changeset-apply context */
3442 int i;
3443 const char *zSep = "";
3444 int rc = SQLITE_OK;
3445 SessionBuffer buf = {0, 0, 0};
3446 int nPk = 0;
3448 sessionAppendStr(&buf, "DELETE FROM ", &rc);
3449 sessionAppendIdent(&buf, zTab, &rc);
3450 sessionAppendStr(&buf, " WHERE ", &rc);
3452 for(i=0; i<p->nCol; i++){
3453 if( p->abPK[i] ){
3454 nPk++;
3455 sessionAppendStr(&buf, zSep, &rc);
3456 sessionAppendIdent(&buf, p->azCol[i], &rc);
3457 sessionAppendStr(&buf, " = ?", &rc);
3458 sessionAppendInteger(&buf, i+1, &rc);
3459 zSep = " AND ";
3463 if( nPk<p->nCol ){
3464 sessionAppendStr(&buf, " AND (?", &rc);
3465 sessionAppendInteger(&buf, p->nCol+1, &rc);
3466 sessionAppendStr(&buf, " OR ", &rc);
3468 zSep = "";
3469 for(i=0; i<p->nCol; i++){
3470 if( !p->abPK[i] ){
3471 sessionAppendStr(&buf, zSep, &rc);
3472 sessionAppendIdent(&buf, p->azCol[i], &rc);
3473 sessionAppendStr(&buf, " IS ?", &rc);
3474 sessionAppendInteger(&buf, i+1, &rc);
3475 zSep = "AND ";
3478 sessionAppendStr(&buf, ")", &rc);
3481 if( rc==SQLITE_OK ){
3482 rc = sqlite3_prepare_v2(db, (char *)buf.aBuf, buf.nBuf, &p->pDelete, 0);
3484 sqlite3_free(buf.aBuf);
3486 return rc;
3490 ** Formulate and prepare a statement to UPDATE a row from database db.
3491 ** Assuming a table structure like this:
3493 ** CREATE TABLE x(a, b, c, d, PRIMARY KEY(a, c));
3495 ** The UPDATE statement looks like this:
3497 ** UPDATE x SET
3498 ** a = CASE WHEN ?2 THEN ?3 ELSE a END,
3499 ** b = CASE WHEN ?5 THEN ?6 ELSE b END,
3500 ** c = CASE WHEN ?8 THEN ?9 ELSE c END,
3501 ** d = CASE WHEN ?11 THEN ?12 ELSE d END
3502 ** WHERE a = ?1 AND c = ?7 AND (?13 OR
3503 ** (?5==0 OR b IS ?4) AND (?11==0 OR d IS ?10) AND
3504 ** )
3506 ** For each column in the table, there are three variables to bind:
3508 ** ?(i*3+1) The old.* value of the column, if any.
3509 ** ?(i*3+2) A boolean flag indicating that the value is being modified.
3510 ** ?(i*3+3) The new.* value of the column, if any.
3512 ** Also, a boolean flag that, if set to true, causes the statement to update
3513 ** a row even if the non-PK values do not match. This is required if the
3514 ** conflict-handler is invoked with CHANGESET_DATA and returns
3515 ** CHANGESET_REPLACE. This is variable "?(nCol*3+1)".
3517 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pUpdate is left
3518 ** pointing to the prepared version of the SQL statement.
3520 static int sessionUpdateRow(
3521 sqlite3 *db, /* Database handle */
3522 const char *zTab, /* Table name */
3523 SessionApplyCtx *p /* Session changeset-apply context */
3525 int rc = SQLITE_OK;
3526 int i;
3527 const char *zSep = "";
3528 SessionBuffer buf = {0, 0, 0};
3530 /* Append "UPDATE tbl SET " */
3531 sessionAppendStr(&buf, "UPDATE ", &rc);
3532 sessionAppendIdent(&buf, zTab, &rc);
3533 sessionAppendStr(&buf, " SET ", &rc);
3535 /* Append the assignments */
3536 for(i=0; i<p->nCol; i++){
3537 sessionAppendStr(&buf, zSep, &rc);
3538 sessionAppendIdent(&buf, p->azCol[i], &rc);
3539 sessionAppendStr(&buf, " = CASE WHEN ?", &rc);
3540 sessionAppendInteger(&buf, i*3+2, &rc);
3541 sessionAppendStr(&buf, " THEN ?", &rc);
3542 sessionAppendInteger(&buf, i*3+3, &rc);
3543 sessionAppendStr(&buf, " ELSE ", &rc);
3544 sessionAppendIdent(&buf, p->azCol[i], &rc);
3545 sessionAppendStr(&buf, " END", &rc);
3546 zSep = ", ";
3549 /* Append the PK part of the WHERE clause */
3550 sessionAppendStr(&buf, " WHERE ", &rc);
3551 for(i=0; i<p->nCol; i++){
3552 if( p->abPK[i] ){
3553 sessionAppendIdent(&buf, p->azCol[i], &rc);
3554 sessionAppendStr(&buf, " = ?", &rc);
3555 sessionAppendInteger(&buf, i*3+1, &rc);
3556 sessionAppendStr(&buf, " AND ", &rc);
3560 /* Append the non-PK part of the WHERE clause */
3561 sessionAppendStr(&buf, " (?", &rc);
3562 sessionAppendInteger(&buf, p->nCol*3+1, &rc);
3563 sessionAppendStr(&buf, " OR 1", &rc);
3564 for(i=0; i<p->nCol; i++){
3565 if( !p->abPK[i] ){
3566 sessionAppendStr(&buf, " AND (?", &rc);
3567 sessionAppendInteger(&buf, i*3+2, &rc);
3568 sessionAppendStr(&buf, "=0 OR ", &rc);
3569 sessionAppendIdent(&buf, p->azCol[i], &rc);
3570 sessionAppendStr(&buf, " IS ?", &rc);
3571 sessionAppendInteger(&buf, i*3+1, &rc);
3572 sessionAppendStr(&buf, ")", &rc);
3575 sessionAppendStr(&buf, ")", &rc);
3577 if( rc==SQLITE_OK ){
3578 rc = sqlite3_prepare_v2(db, (char *)buf.aBuf, buf.nBuf, &p->pUpdate, 0);
3580 sqlite3_free(buf.aBuf);
3582 return rc;
3587 ** Formulate and prepare an SQL statement to query table zTab by primary
3588 ** key. Assuming the following table structure:
3590 ** CREATE TABLE x(a, b, c, d, PRIMARY KEY(a, c));
3592 ** The SELECT statement looks like this:
3594 ** SELECT * FROM x WHERE a = ?1 AND c = ?3
3596 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pSelect is left
3597 ** pointing to the prepared version of the SQL statement.
3599 static int sessionSelectRow(
3600 sqlite3 *db, /* Database handle */
3601 const char *zTab, /* Table name */
3602 SessionApplyCtx *p /* Session changeset-apply context */
3604 return sessionSelectStmt(
3605 db, "main", zTab, p->nCol, p->azCol, p->abPK, &p->pSelect);
3609 ** Formulate and prepare an INSERT statement to add a record to table zTab.
3610 ** For example:
3612 ** INSERT INTO main."zTab" VALUES(?1, ?2, ?3 ...);
3614 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pInsert is left
3615 ** pointing to the prepared version of the SQL statement.
3617 static int sessionInsertRow(
3618 sqlite3 *db, /* Database handle */
3619 const char *zTab, /* Table name */
3620 SessionApplyCtx *p /* Session changeset-apply context */
3622 int rc = SQLITE_OK;
3623 int i;
3624 SessionBuffer buf = {0, 0, 0};
3626 sessionAppendStr(&buf, "INSERT INTO main.", &rc);
3627 sessionAppendIdent(&buf, zTab, &rc);
3628 sessionAppendStr(&buf, "(", &rc);
3629 for(i=0; i<p->nCol; i++){
3630 if( i!=0 ) sessionAppendStr(&buf, ", ", &rc);
3631 sessionAppendIdent(&buf, p->azCol[i], &rc);
3634 sessionAppendStr(&buf, ") VALUES(?", &rc);
3635 for(i=1; i<p->nCol; i++){
3636 sessionAppendStr(&buf, ", ?", &rc);
3638 sessionAppendStr(&buf, ")", &rc);
3640 if( rc==SQLITE_OK ){
3641 rc = sqlite3_prepare_v2(db, (char *)buf.aBuf, buf.nBuf, &p->pInsert, 0);
3643 sqlite3_free(buf.aBuf);
3644 return rc;
3647 static int sessionPrepare(sqlite3 *db, sqlite3_stmt **pp, const char *zSql){
3648 return sqlite3_prepare_v2(db, zSql, -1, pp, 0);
3652 ** Prepare statements for applying changes to the sqlite_stat1 table.
3653 ** These are similar to those created by sessionSelectRow(),
3654 ** sessionInsertRow(), sessionUpdateRow() and sessionDeleteRow() for
3655 ** other tables.
3657 static int sessionStat1Sql(sqlite3 *db, SessionApplyCtx *p){
3658 int rc = sessionSelectRow(db, "sqlite_stat1", p);
3659 if( rc==SQLITE_OK ){
3660 rc = sessionPrepare(db, &p->pInsert,
3661 "INSERT INTO main.sqlite_stat1 VALUES(?1, "
3662 "CASE WHEN length(?2)=0 AND typeof(?2)='blob' THEN NULL ELSE ?2 END, "
3663 "?3)"
3666 if( rc==SQLITE_OK ){
3667 rc = sessionPrepare(db, &p->pUpdate,
3668 "UPDATE main.sqlite_stat1 SET "
3669 "tbl = CASE WHEN ?2 THEN ?3 ELSE tbl END, "
3670 "idx = CASE WHEN ?5 THEN ?6 ELSE idx END, "
3671 "stat = CASE WHEN ?8 THEN ?9 ELSE stat END "
3672 "WHERE tbl=?1 AND idx IS "
3673 "CASE WHEN length(?4)=0 AND typeof(?4)='blob' THEN NULL ELSE ?4 END "
3674 "AND (?10 OR ?8=0 OR stat IS ?7)"
3677 if( rc==SQLITE_OK ){
3678 rc = sessionPrepare(db, &p->pDelete,
3679 "DELETE FROM main.sqlite_stat1 WHERE tbl=?1 AND idx IS "
3680 "CASE WHEN length(?2)=0 AND typeof(?2)='blob' THEN NULL ELSE ?2 END "
3681 "AND (?4 OR stat IS ?3)"
3684 return rc;
3688 ** A wrapper around sqlite3_bind_value() that detects an extra problem.
3689 ** See comments in the body of this function for details.
3691 static int sessionBindValue(
3692 sqlite3_stmt *pStmt, /* Statement to bind value to */
3693 int i, /* Parameter number to bind to */
3694 sqlite3_value *pVal /* Value to bind */
3696 int eType = sqlite3_value_type(pVal);
3697 /* COVERAGE: The (pVal->z==0) branch is never true using current versions
3698 ** of SQLite. If a malloc fails in an sqlite3_value_xxx() function, either
3699 ** the (pVal->z) variable remains as it was or the type of the value is
3700 ** set to SQLITE_NULL. */
3701 if( (eType==SQLITE_TEXT || eType==SQLITE_BLOB) && pVal->z==0 ){
3702 /* This condition occurs when an earlier OOM in a call to
3703 ** sqlite3_value_text() or sqlite3_value_blob() (perhaps from within
3704 ** a conflict-handler) has zeroed the pVal->z pointer. Return NOMEM. */
3705 return SQLITE_NOMEM;
3707 return sqlite3_bind_value(pStmt, i, pVal);
3711 ** Iterator pIter must point to an SQLITE_INSERT entry. This function
3712 ** transfers new.* values from the current iterator entry to statement
3713 ** pStmt. The table being inserted into has nCol columns.
3715 ** New.* value $i from the iterator is bound to variable ($i+1) of
3716 ** statement pStmt. If parameter abPK is NULL, all values from 0 to (nCol-1)
3717 ** are transfered to the statement. Otherwise, if abPK is not NULL, it points
3718 ** to an array nCol elements in size. In this case only those values for
3719 ** which abPK[$i] is true are read from the iterator and bound to the
3720 ** statement.
3722 ** An SQLite error code is returned if an error occurs. Otherwise, SQLITE_OK.
3724 static int sessionBindRow(
3725 sqlite3_changeset_iter *pIter, /* Iterator to read values from */
3726 int(*xValue)(sqlite3_changeset_iter *, int, sqlite3_value **),
3727 int nCol, /* Number of columns */
3728 u8 *abPK, /* If not NULL, bind only if true */
3729 sqlite3_stmt *pStmt /* Bind values to this statement */
3731 int i;
3732 int rc = SQLITE_OK;
3734 /* Neither sqlite3changeset_old or sqlite3changeset_new can fail if the
3735 ** argument iterator points to a suitable entry. Make sure that xValue
3736 ** is one of these to guarantee that it is safe to ignore the return
3737 ** in the code below. */
3738 assert( xValue==sqlite3changeset_old || xValue==sqlite3changeset_new );
3740 for(i=0; rc==SQLITE_OK && i<nCol; i++){
3741 if( !abPK || abPK[i] ){
3742 sqlite3_value *pVal;
3743 (void)xValue(pIter, i, &pVal);
3744 if( pVal==0 ){
3745 /* The value in the changeset was "undefined". This indicates a
3746 ** corrupt changeset blob. */
3747 rc = SQLITE_CORRUPT_BKPT;
3748 }else{
3749 rc = sessionBindValue(pStmt, i+1, pVal);
3753 return rc;
3757 ** SQL statement pSelect is as generated by the sessionSelectRow() function.
3758 ** This function binds the primary key values from the change that changeset
3759 ** iterator pIter points to to the SELECT and attempts to seek to the table
3760 ** entry. If a row is found, the SELECT statement left pointing at the row
3761 ** and SQLITE_ROW is returned. Otherwise, if no row is found and no error
3762 ** has occured, the statement is reset and SQLITE_OK is returned. If an
3763 ** error occurs, the statement is reset and an SQLite error code is returned.
3765 ** If this function returns SQLITE_ROW, the caller must eventually reset()
3766 ** statement pSelect. If any other value is returned, the statement does
3767 ** not require a reset().
3769 ** If the iterator currently points to an INSERT record, bind values from the
3770 ** new.* record to the SELECT statement. Or, if it points to a DELETE or
3771 ** UPDATE, bind values from the old.* record.
3773 static int sessionSeekToRow(
3774 sqlite3 *db, /* Database handle */
3775 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3776 u8 *abPK, /* Primary key flags array */
3777 sqlite3_stmt *pSelect /* SELECT statement from sessionSelectRow() */
3779 int rc; /* Return code */
3780 int nCol; /* Number of columns in table */
3781 int op; /* Changset operation (SQLITE_UPDATE etc.) */
3782 const char *zDummy; /* Unused */
3784 sqlite3changeset_op(pIter, &zDummy, &nCol, &op, 0);
3785 rc = sessionBindRow(pIter,
3786 op==SQLITE_INSERT ? sqlite3changeset_new : sqlite3changeset_old,
3787 nCol, abPK, pSelect
3790 if( rc==SQLITE_OK ){
3791 rc = sqlite3_step(pSelect);
3792 if( rc!=SQLITE_ROW ) rc = sqlite3_reset(pSelect);
3795 return rc;
3799 ** This function is called from within sqlite3changset_apply_v2() when
3800 ** a conflict is encountered and resolved using conflict resolution
3801 ** mode eType (either SQLITE_CHANGESET_OMIT or SQLITE_CHANGESET_REPLACE)..
3802 ** It adds a conflict resolution record to the buffer in
3803 ** SessionApplyCtx.rebase, which will eventually be returned to the caller
3804 ** of apply_v2() as the "rebase" buffer.
3806 ** Return SQLITE_OK if successful, or an SQLite error code otherwise.
3808 static int sessionRebaseAdd(
3809 SessionApplyCtx *p, /* Apply context */
3810 int eType, /* Conflict resolution (OMIT or REPLACE) */
3811 sqlite3_changeset_iter *pIter /* Iterator pointing at current change */
3813 int rc = SQLITE_OK;
3814 int i;
3815 int eOp = pIter->op;
3816 if( p->bRebaseStarted==0 ){
3817 /* Append a table-header to the rebase buffer */
3818 const char *zTab = pIter->zTab;
3819 sessionAppendByte(&p->rebase, 'T', &rc);
3820 sessionAppendVarint(&p->rebase, p->nCol, &rc);
3821 sessionAppendBlob(&p->rebase, p->abPK, p->nCol, &rc);
3822 sessionAppendBlob(&p->rebase, (u8*)zTab, (int)strlen(zTab)+1, &rc);
3823 p->bRebaseStarted = 1;
3826 assert( eType==SQLITE_CHANGESET_REPLACE||eType==SQLITE_CHANGESET_OMIT );
3827 assert( eOp==SQLITE_DELETE || eOp==SQLITE_INSERT || eOp==SQLITE_UPDATE );
3829 sessionAppendByte(&p->rebase,
3830 (eOp==SQLITE_DELETE ? SQLITE_DELETE : SQLITE_INSERT), &rc
3832 sessionAppendByte(&p->rebase, (eType==SQLITE_CHANGESET_REPLACE), &rc);
3833 for(i=0; i<p->nCol; i++){
3834 sqlite3_value *pVal = 0;
3835 if( eOp==SQLITE_DELETE || (eOp==SQLITE_UPDATE && p->abPK[i]) ){
3836 sqlite3changeset_old(pIter, i, &pVal);
3837 }else{
3838 sqlite3changeset_new(pIter, i, &pVal);
3840 sessionAppendValue(&p->rebase, pVal, &rc);
3843 return rc;
3847 ** Invoke the conflict handler for the change that the changeset iterator
3848 ** currently points to.
3850 ** Argument eType must be either CHANGESET_DATA or CHANGESET_CONFLICT.
3851 ** If argument pbReplace is NULL, then the type of conflict handler invoked
3852 ** depends solely on eType, as follows:
3854 ** eType value Value passed to xConflict
3855 ** -------------------------------------------------
3856 ** CHANGESET_DATA CHANGESET_NOTFOUND
3857 ** CHANGESET_CONFLICT CHANGESET_CONSTRAINT
3859 ** Or, if pbReplace is not NULL, then an attempt is made to find an existing
3860 ** record with the same primary key as the record about to be deleted, updated
3861 ** or inserted. If such a record can be found, it is available to the conflict
3862 ** handler as the "conflicting" record. In this case the type of conflict
3863 ** handler invoked is as follows:
3865 ** eType value PK Record found? Value passed to xConflict
3866 ** ----------------------------------------------------------------
3867 ** CHANGESET_DATA Yes CHANGESET_DATA
3868 ** CHANGESET_DATA No CHANGESET_NOTFOUND
3869 ** CHANGESET_CONFLICT Yes CHANGESET_CONFLICT
3870 ** CHANGESET_CONFLICT No CHANGESET_CONSTRAINT
3872 ** If pbReplace is not NULL, and a record with a matching PK is found, and
3873 ** the conflict handler function returns SQLITE_CHANGESET_REPLACE, *pbReplace
3874 ** is set to non-zero before returning SQLITE_OK.
3876 ** If the conflict handler returns SQLITE_CHANGESET_ABORT, SQLITE_ABORT is
3877 ** returned. Or, if the conflict handler returns an invalid value,
3878 ** SQLITE_MISUSE. If the conflict handler returns SQLITE_CHANGESET_OMIT,
3879 ** this function returns SQLITE_OK.
3881 static int sessionConflictHandler(
3882 int eType, /* Either CHANGESET_DATA or CONFLICT */
3883 SessionApplyCtx *p, /* changeset_apply() context */
3884 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3885 int(*xConflict)(void *, int, sqlite3_changeset_iter*),
3886 void *pCtx, /* First argument for conflict handler */
3887 int *pbReplace /* OUT: Set to true if PK row is found */
3889 int res = 0; /* Value returned by conflict handler */
3890 int rc;
3891 int nCol;
3892 int op;
3893 const char *zDummy;
3895 sqlite3changeset_op(pIter, &zDummy, &nCol, &op, 0);
3897 assert( eType==SQLITE_CHANGESET_CONFLICT || eType==SQLITE_CHANGESET_DATA );
3898 assert( SQLITE_CHANGESET_CONFLICT+1==SQLITE_CHANGESET_CONSTRAINT );
3899 assert( SQLITE_CHANGESET_DATA+1==SQLITE_CHANGESET_NOTFOUND );
3901 /* Bind the new.* PRIMARY KEY values to the SELECT statement. */
3902 if( pbReplace ){
3903 rc = sessionSeekToRow(p->db, pIter, p->abPK, p->pSelect);
3904 }else{
3905 rc = SQLITE_OK;
3908 if( rc==SQLITE_ROW ){
3909 /* There exists another row with the new.* primary key. */
3910 pIter->pConflict = p->pSelect;
3911 res = xConflict(pCtx, eType, pIter);
3912 pIter->pConflict = 0;
3913 rc = sqlite3_reset(p->pSelect);
3914 }else if( rc==SQLITE_OK ){
3915 if( p->bDeferConstraints && eType==SQLITE_CHANGESET_CONFLICT ){
3916 /* Instead of invoking the conflict handler, append the change blob
3917 ** to the SessionApplyCtx.constraints buffer. */
3918 u8 *aBlob = &pIter->in.aData[pIter->in.iCurrent];
3919 int nBlob = pIter->in.iNext - pIter->in.iCurrent;
3920 sessionAppendBlob(&p->constraints, aBlob, nBlob, &rc);
3921 return SQLITE_OK;
3922 }else{
3923 /* No other row with the new.* primary key. */
3924 res = xConflict(pCtx, eType+1, pIter);
3925 if( res==SQLITE_CHANGESET_REPLACE ) rc = SQLITE_MISUSE;
3929 if( rc==SQLITE_OK ){
3930 switch( res ){
3931 case SQLITE_CHANGESET_REPLACE:
3932 assert( pbReplace );
3933 *pbReplace = 1;
3934 break;
3936 case SQLITE_CHANGESET_OMIT:
3937 break;
3939 case SQLITE_CHANGESET_ABORT:
3940 rc = SQLITE_ABORT;
3941 break;
3943 default:
3944 rc = SQLITE_MISUSE;
3945 break;
3947 if( rc==SQLITE_OK ){
3948 rc = sessionRebaseAdd(p, res, pIter);
3952 return rc;
3956 ** Attempt to apply the change that the iterator passed as the first argument
3957 ** currently points to to the database. If a conflict is encountered, invoke
3958 ** the conflict handler callback.
3960 ** If argument pbRetry is NULL, then ignore any CHANGESET_DATA conflict. If
3961 ** one is encountered, update or delete the row with the matching primary key
3962 ** instead. Or, if pbRetry is not NULL and a CHANGESET_DATA conflict occurs,
3963 ** invoke the conflict handler. If it returns CHANGESET_REPLACE, set *pbRetry
3964 ** to true before returning. In this case the caller will invoke this function
3965 ** again, this time with pbRetry set to NULL.
3967 ** If argument pbReplace is NULL and a CHANGESET_CONFLICT conflict is
3968 ** encountered invoke the conflict handler with CHANGESET_CONSTRAINT instead.
3969 ** Or, if pbReplace is not NULL, invoke it with CHANGESET_CONFLICT. If such
3970 ** an invocation returns SQLITE_CHANGESET_REPLACE, set *pbReplace to true
3971 ** before retrying. In this case the caller attempts to remove the conflicting
3972 ** row before invoking this function again, this time with pbReplace set
3973 ** to NULL.
3975 ** If any conflict handler returns SQLITE_CHANGESET_ABORT, this function
3976 ** returns SQLITE_ABORT. Otherwise, if no error occurs, SQLITE_OK is
3977 ** returned.
3979 static int sessionApplyOneOp(
3980 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3981 SessionApplyCtx *p, /* changeset_apply() context */
3982 int(*xConflict)(void *, int, sqlite3_changeset_iter *),
3983 void *pCtx, /* First argument for the conflict handler */
3984 int *pbReplace, /* OUT: True to remove PK row and retry */
3985 int *pbRetry /* OUT: True to retry. */
3987 const char *zDummy;
3988 int op;
3989 int nCol;
3990 int rc = SQLITE_OK;
3992 assert( p->pDelete && p->pUpdate && p->pInsert && p->pSelect );
3993 assert( p->azCol && p->abPK );
3994 assert( !pbReplace || *pbReplace==0 );
3996 sqlite3changeset_op(pIter, &zDummy, &nCol, &op, 0);
3998 if( op==SQLITE_DELETE ){
4000 /* Bind values to the DELETE statement. If conflict handling is required,
4001 ** bind values for all columns and set bound variable (nCol+1) to true.
4002 ** Or, if conflict handling is not required, bind just the PK column
4003 ** values and, if it exists, set (nCol+1) to false. Conflict handling
4004 ** is not required if:
4006 ** * this is a patchset, or
4007 ** * (pbRetry==0), or
4008 ** * all columns of the table are PK columns (in this case there is
4009 ** no (nCol+1) variable to bind to).
4011 u8 *abPK = (pIter->bPatchset ? p->abPK : 0);
4012 rc = sessionBindRow(pIter, sqlite3changeset_old, nCol, abPK, p->pDelete);
4013 if( rc==SQLITE_OK && sqlite3_bind_parameter_count(p->pDelete)>nCol ){
4014 rc = sqlite3_bind_int(p->pDelete, nCol+1, (pbRetry==0 || abPK));
4016 if( rc!=SQLITE_OK ) return rc;
4018 sqlite3_step(p->pDelete);
4019 rc = sqlite3_reset(p->pDelete);
4020 if( rc==SQLITE_OK && sqlite3_changes(p->db)==0 ){
4021 rc = sessionConflictHandler(
4022 SQLITE_CHANGESET_DATA, p, pIter, xConflict, pCtx, pbRetry
4024 }else if( (rc&0xff)==SQLITE_CONSTRAINT ){
4025 rc = sessionConflictHandler(
4026 SQLITE_CHANGESET_CONFLICT, p, pIter, xConflict, pCtx, 0
4030 }else if( op==SQLITE_UPDATE ){
4031 int i;
4033 /* Bind values to the UPDATE statement. */
4034 for(i=0; rc==SQLITE_OK && i<nCol; i++){
4035 sqlite3_value *pOld = sessionChangesetOld(pIter, i);
4036 sqlite3_value *pNew = sessionChangesetNew(pIter, i);
4038 sqlite3_bind_int(p->pUpdate, i*3+2, !!pNew);
4039 if( pOld ){
4040 rc = sessionBindValue(p->pUpdate, i*3+1, pOld);
4042 if( rc==SQLITE_OK && pNew ){
4043 rc = sessionBindValue(p->pUpdate, i*3+3, pNew);
4046 if( rc==SQLITE_OK ){
4047 sqlite3_bind_int(p->pUpdate, nCol*3+1, pbRetry==0 || pIter->bPatchset);
4049 if( rc!=SQLITE_OK ) return rc;
4051 /* Attempt the UPDATE. In the case of a NOTFOUND or DATA conflict,
4052 ** the result will be SQLITE_OK with 0 rows modified. */
4053 sqlite3_step(p->pUpdate);
4054 rc = sqlite3_reset(p->pUpdate);
4056 if( rc==SQLITE_OK && sqlite3_changes(p->db)==0 ){
4057 /* A NOTFOUND or DATA error. Search the table to see if it contains
4058 ** a row with a matching primary key. If so, this is a DATA conflict.
4059 ** Otherwise, if there is no primary key match, it is a NOTFOUND. */
4061 rc = sessionConflictHandler(
4062 SQLITE_CHANGESET_DATA, p, pIter, xConflict, pCtx, pbRetry
4065 }else if( (rc&0xff)==SQLITE_CONSTRAINT ){
4066 /* This is always a CONSTRAINT conflict. */
4067 rc = sessionConflictHandler(
4068 SQLITE_CHANGESET_CONFLICT, p, pIter, xConflict, pCtx, 0
4072 }else{
4073 assert( op==SQLITE_INSERT );
4074 if( p->bStat1 ){
4075 /* Check if there is a conflicting row. For sqlite_stat1, this needs
4076 ** to be done using a SELECT, as there is no PRIMARY KEY in the
4077 ** database schema to throw an exception if a duplicate is inserted. */
4078 rc = sessionSeekToRow(p->db, pIter, p->abPK, p->pSelect);
4079 if( rc==SQLITE_ROW ){
4080 rc = SQLITE_CONSTRAINT;
4081 sqlite3_reset(p->pSelect);
4085 if( rc==SQLITE_OK ){
4086 rc = sessionBindRow(pIter, sqlite3changeset_new, nCol, 0, p->pInsert);
4087 if( rc!=SQLITE_OK ) return rc;
4089 sqlite3_step(p->pInsert);
4090 rc = sqlite3_reset(p->pInsert);
4093 if( (rc&0xff)==SQLITE_CONSTRAINT ){
4094 rc = sessionConflictHandler(
4095 SQLITE_CHANGESET_CONFLICT, p, pIter, xConflict, pCtx, pbReplace
4100 return rc;
4104 ** Attempt to apply the change that the iterator passed as the first argument
4105 ** currently points to to the database. If a conflict is encountered, invoke
4106 ** the conflict handler callback.
4108 ** The difference between this function and sessionApplyOne() is that this
4109 ** function handles the case where the conflict-handler is invoked and
4110 ** returns SQLITE_CHANGESET_REPLACE - indicating that the change should be
4111 ** retried in some manner.
4113 static int sessionApplyOneWithRetry(
4114 sqlite3 *db, /* Apply change to "main" db of this handle */
4115 sqlite3_changeset_iter *pIter, /* Changeset iterator to read change from */
4116 SessionApplyCtx *pApply, /* Apply context */
4117 int(*xConflict)(void*, int, sqlite3_changeset_iter*),
4118 void *pCtx /* First argument passed to xConflict */
4120 int bReplace = 0;
4121 int bRetry = 0;
4122 int rc;
4124 rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, &bReplace, &bRetry);
4125 if( rc==SQLITE_OK ){
4126 /* If the bRetry flag is set, the change has not been applied due to an
4127 ** SQLITE_CHANGESET_DATA problem (i.e. this is an UPDATE or DELETE and
4128 ** a row with the correct PK is present in the db, but one or more other
4129 ** fields do not contain the expected values) and the conflict handler
4130 ** returned SQLITE_CHANGESET_REPLACE. In this case retry the operation,
4131 ** but pass NULL as the final argument so that sessionApplyOneOp() ignores
4132 ** the SQLITE_CHANGESET_DATA problem. */
4133 if( bRetry ){
4134 assert( pIter->op==SQLITE_UPDATE || pIter->op==SQLITE_DELETE );
4135 rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, 0, 0);
4138 /* If the bReplace flag is set, the change is an INSERT that has not
4139 ** been performed because the database already contains a row with the
4140 ** specified primary key and the conflict handler returned
4141 ** SQLITE_CHANGESET_REPLACE. In this case remove the conflicting row
4142 ** before reattempting the INSERT. */
4143 else if( bReplace ){
4144 assert( pIter->op==SQLITE_INSERT );
4145 rc = sqlite3_exec(db, "SAVEPOINT replace_op", 0, 0, 0);
4146 if( rc==SQLITE_OK ){
4147 rc = sessionBindRow(pIter,
4148 sqlite3changeset_new, pApply->nCol, pApply->abPK, pApply->pDelete);
4149 sqlite3_bind_int(pApply->pDelete, pApply->nCol+1, 1);
4151 if( rc==SQLITE_OK ){
4152 sqlite3_step(pApply->pDelete);
4153 rc = sqlite3_reset(pApply->pDelete);
4155 if( rc==SQLITE_OK ){
4156 rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, 0, 0);
4158 if( rc==SQLITE_OK ){
4159 rc = sqlite3_exec(db, "RELEASE replace_op", 0, 0, 0);
4164 return rc;
4168 ** Retry the changes accumulated in the pApply->constraints buffer.
4170 static int sessionRetryConstraints(
4171 sqlite3 *db,
4172 int bPatchset,
4173 const char *zTab,
4174 SessionApplyCtx *pApply,
4175 int(*xConflict)(void*, int, sqlite3_changeset_iter*),
4176 void *pCtx /* First argument passed to xConflict */
4178 int rc = SQLITE_OK;
4180 while( pApply->constraints.nBuf ){
4181 sqlite3_changeset_iter *pIter2 = 0;
4182 SessionBuffer cons = pApply->constraints;
4183 memset(&pApply->constraints, 0, sizeof(SessionBuffer));
4185 rc = sessionChangesetStart(&pIter2, 0, 0, cons.nBuf, cons.aBuf);
4186 if( rc==SQLITE_OK ){
4187 int nByte = 2*pApply->nCol*sizeof(sqlite3_value*);
4188 int rc2;
4189 pIter2->bPatchset = bPatchset;
4190 pIter2->zTab = (char*)zTab;
4191 pIter2->nCol = pApply->nCol;
4192 pIter2->abPK = pApply->abPK;
4193 sessionBufferGrow(&pIter2->tblhdr, nByte, &rc);
4194 pIter2->apValue = (sqlite3_value**)pIter2->tblhdr.aBuf;
4195 if( rc==SQLITE_OK ) memset(pIter2->apValue, 0, nByte);
4197 while( rc==SQLITE_OK && SQLITE_ROW==sqlite3changeset_next(pIter2) ){
4198 rc = sessionApplyOneWithRetry(db, pIter2, pApply, xConflict, pCtx);
4201 rc2 = sqlite3changeset_finalize(pIter2);
4202 if( rc==SQLITE_OK ) rc = rc2;
4204 assert( pApply->bDeferConstraints || pApply->constraints.nBuf==0 );
4206 sqlite3_free(cons.aBuf);
4207 if( rc!=SQLITE_OK ) break;
4208 if( pApply->constraints.nBuf>=cons.nBuf ){
4209 /* No progress was made on the last round. */
4210 pApply->bDeferConstraints = 0;
4214 return rc;
4218 ** Argument pIter is a changeset iterator that has been initialized, but
4219 ** not yet passed to sqlite3changeset_next(). This function applies the
4220 ** changeset to the main database attached to handle "db". The supplied
4221 ** conflict handler callback is invoked to resolve any conflicts encountered
4222 ** while applying the change.
4224 static int sessionChangesetApply(
4225 sqlite3 *db, /* Apply change to "main" db of this handle */
4226 sqlite3_changeset_iter *pIter, /* Changeset to apply */
4227 int(*xFilter)(
4228 void *pCtx, /* Copy of sixth arg to _apply() */
4229 const char *zTab /* Table name */
4231 int(*xConflict)(
4232 void *pCtx, /* Copy of fifth arg to _apply() */
4233 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4234 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4236 void *pCtx, /* First argument passed to xConflict */
4237 void **ppRebase, int *pnRebase, /* OUT: Rebase information */
4238 int flags /* SESSION_APPLY_XXX flags */
4240 int schemaMismatch = 0;
4241 int rc = SQLITE_OK; /* Return code */
4242 const char *zTab = 0; /* Name of current table */
4243 int nTab = 0; /* Result of sqlite3Strlen30(zTab) */
4244 SessionApplyCtx sApply; /* changeset_apply() context object */
4245 int bPatchset;
4247 assert( xConflict!=0 );
4249 pIter->in.bNoDiscard = 1;
4250 memset(&sApply, 0, sizeof(sApply));
4251 sqlite3_mutex_enter(sqlite3_db_mutex(db));
4252 if( (flags & SQLITE_CHANGESETAPPLY_NOSAVEPOINT)==0 ){
4253 rc = sqlite3_exec(db, "SAVEPOINT changeset_apply", 0, 0, 0);
4255 if( rc==SQLITE_OK ){
4256 rc = sqlite3_exec(db, "PRAGMA defer_foreign_keys = 1", 0, 0, 0);
4258 while( rc==SQLITE_OK && SQLITE_ROW==sqlite3changeset_next(pIter) ){
4259 int nCol;
4260 int op;
4261 const char *zNew;
4263 sqlite3changeset_op(pIter, &zNew, &nCol, &op, 0);
4265 if( zTab==0 || sqlite3_strnicmp(zNew, zTab, nTab+1) ){
4266 u8 *abPK;
4268 rc = sessionRetryConstraints(
4269 db, pIter->bPatchset, zTab, &sApply, xConflict, pCtx
4271 if( rc!=SQLITE_OK ) break;
4273 sqlite3_free((char*)sApply.azCol); /* cast works around VC++ bug */
4274 sqlite3_finalize(sApply.pDelete);
4275 sqlite3_finalize(sApply.pUpdate);
4276 sqlite3_finalize(sApply.pInsert);
4277 sqlite3_finalize(sApply.pSelect);
4278 sApply.db = db;
4279 sApply.pDelete = 0;
4280 sApply.pUpdate = 0;
4281 sApply.pInsert = 0;
4282 sApply.pSelect = 0;
4283 sApply.nCol = 0;
4284 sApply.azCol = 0;
4285 sApply.abPK = 0;
4286 sApply.bStat1 = 0;
4287 sApply.bDeferConstraints = 1;
4288 sApply.bRebaseStarted = 0;
4289 memset(&sApply.constraints, 0, sizeof(SessionBuffer));
4291 /* If an xFilter() callback was specified, invoke it now. If the
4292 ** xFilter callback returns zero, skip this table. If it returns
4293 ** non-zero, proceed. */
4294 schemaMismatch = (xFilter && (0==xFilter(pCtx, zNew)));
4295 if( schemaMismatch ){
4296 zTab = sqlite3_mprintf("%s", zNew);
4297 if( zTab==0 ){
4298 rc = SQLITE_NOMEM;
4299 break;
4301 nTab = (int)strlen(zTab);
4302 sApply.azCol = (const char **)zTab;
4303 }else{
4304 int nMinCol = 0;
4305 int i;
4307 sqlite3changeset_pk(pIter, &abPK, 0);
4308 rc = sessionTableInfo(
4309 db, "main", zNew, &sApply.nCol, &zTab, &sApply.azCol, &sApply.abPK
4311 if( rc!=SQLITE_OK ) break;
4312 for(i=0; i<sApply.nCol; i++){
4313 if( sApply.abPK[i] ) nMinCol = i+1;
4316 if( sApply.nCol==0 ){
4317 schemaMismatch = 1;
4318 sqlite3_log(SQLITE_SCHEMA,
4319 "sqlite3changeset_apply(): no such table: %s", zTab
4322 else if( sApply.nCol<nCol ){
4323 schemaMismatch = 1;
4324 sqlite3_log(SQLITE_SCHEMA,
4325 "sqlite3changeset_apply(): table %s has %d columns, "
4326 "expected %d or more",
4327 zTab, sApply.nCol, nCol
4330 else if( nCol<nMinCol || memcmp(sApply.abPK, abPK, nCol)!=0 ){
4331 schemaMismatch = 1;
4332 sqlite3_log(SQLITE_SCHEMA, "sqlite3changeset_apply(): "
4333 "primary key mismatch for table %s", zTab
4336 else{
4337 sApply.nCol = nCol;
4338 if( 0==sqlite3_stricmp(zTab, "sqlite_stat1") ){
4339 if( (rc = sessionStat1Sql(db, &sApply) ) ){
4340 break;
4342 sApply.bStat1 = 1;
4343 }else{
4344 if((rc = sessionSelectRow(db, zTab, &sApply))
4345 || (rc = sessionUpdateRow(db, zTab, &sApply))
4346 || (rc = sessionDeleteRow(db, zTab, &sApply))
4347 || (rc = sessionInsertRow(db, zTab, &sApply))
4349 break;
4351 sApply.bStat1 = 0;
4354 nTab = sqlite3Strlen30(zTab);
4358 /* If there is a schema mismatch on the current table, proceed to the
4359 ** next change. A log message has already been issued. */
4360 if( schemaMismatch ) continue;
4362 rc = sessionApplyOneWithRetry(db, pIter, &sApply, xConflict, pCtx);
4365 bPatchset = pIter->bPatchset;
4366 if( rc==SQLITE_OK ){
4367 rc = sqlite3changeset_finalize(pIter);
4368 }else{
4369 sqlite3changeset_finalize(pIter);
4372 if( rc==SQLITE_OK ){
4373 rc = sessionRetryConstraints(db, bPatchset, zTab, &sApply, xConflict, pCtx);
4376 if( rc==SQLITE_OK ){
4377 int nFk, notUsed;
4378 sqlite3_db_status(db, SQLITE_DBSTATUS_DEFERRED_FKS, &nFk, &notUsed, 0);
4379 if( nFk!=0 ){
4380 int res = SQLITE_CHANGESET_ABORT;
4381 sqlite3_changeset_iter sIter;
4382 memset(&sIter, 0, sizeof(sIter));
4383 sIter.nCol = nFk;
4384 res = xConflict(pCtx, SQLITE_CHANGESET_FOREIGN_KEY, &sIter);
4385 if( res!=SQLITE_CHANGESET_OMIT ){
4386 rc = SQLITE_CONSTRAINT;
4390 sqlite3_exec(db, "PRAGMA defer_foreign_keys = 0", 0, 0, 0);
4392 if( (flags & SQLITE_CHANGESETAPPLY_NOSAVEPOINT)==0 ){
4393 if( rc==SQLITE_OK ){
4394 rc = sqlite3_exec(db, "RELEASE changeset_apply", 0, 0, 0);
4395 }else{
4396 sqlite3_exec(db, "ROLLBACK TO changeset_apply", 0, 0, 0);
4397 sqlite3_exec(db, "RELEASE changeset_apply", 0, 0, 0);
4401 if( rc==SQLITE_OK && bPatchset==0 && ppRebase && pnRebase ){
4402 *ppRebase = (void*)sApply.rebase.aBuf;
4403 *pnRebase = sApply.rebase.nBuf;
4404 sApply.rebase.aBuf = 0;
4406 sqlite3_finalize(sApply.pInsert);
4407 sqlite3_finalize(sApply.pDelete);
4408 sqlite3_finalize(sApply.pUpdate);
4409 sqlite3_finalize(sApply.pSelect);
4410 sqlite3_free((char*)sApply.azCol); /* cast works around VC++ bug */
4411 sqlite3_free((char*)sApply.constraints.aBuf);
4412 sqlite3_free((char*)sApply.rebase.aBuf);
4413 sqlite3_mutex_leave(sqlite3_db_mutex(db));
4414 return rc;
4418 ** Apply the changeset passed via pChangeset/nChangeset to the main
4419 ** database attached to handle "db".
4421 int sqlite3changeset_apply_v2(
4422 sqlite3 *db, /* Apply change to "main" db of this handle */
4423 int nChangeset, /* Size of changeset in bytes */
4424 void *pChangeset, /* Changeset blob */
4425 int(*xFilter)(
4426 void *pCtx, /* Copy of sixth arg to _apply() */
4427 const char *zTab /* Table name */
4429 int(*xConflict)(
4430 void *pCtx, /* Copy of sixth arg to _apply() */
4431 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4432 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4434 void *pCtx, /* First argument passed to xConflict */
4435 void **ppRebase, int *pnRebase,
4436 int flags
4438 sqlite3_changeset_iter *pIter; /* Iterator to skip through changeset */
4439 int rc = sqlite3changeset_start(&pIter, nChangeset, pChangeset);
4440 if( rc==SQLITE_OK ){
4441 rc = sessionChangesetApply(
4442 db, pIter, xFilter, xConflict, pCtx, ppRebase, pnRebase, flags
4445 return rc;
4449 ** Apply the changeset passed via pChangeset/nChangeset to the main database
4450 ** attached to handle "db". Invoke the supplied conflict handler callback
4451 ** to resolve any conflicts encountered while applying the change.
4453 int sqlite3changeset_apply(
4454 sqlite3 *db, /* Apply change to "main" db of this handle */
4455 int nChangeset, /* Size of changeset in bytes */
4456 void *pChangeset, /* Changeset blob */
4457 int(*xFilter)(
4458 void *pCtx, /* Copy of sixth arg to _apply() */
4459 const char *zTab /* Table name */
4461 int(*xConflict)(
4462 void *pCtx, /* Copy of fifth arg to _apply() */
4463 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4464 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4466 void *pCtx /* First argument passed to xConflict */
4468 return sqlite3changeset_apply_v2(
4469 db, nChangeset, pChangeset, xFilter, xConflict, pCtx, 0, 0, 0
4474 ** Apply the changeset passed via xInput/pIn to the main database
4475 ** attached to handle "db". Invoke the supplied conflict handler callback
4476 ** to resolve any conflicts encountered while applying the change.
4478 int sqlite3changeset_apply_v2_strm(
4479 sqlite3 *db, /* Apply change to "main" db of this handle */
4480 int (*xInput)(void *pIn, void *pData, int *pnData), /* Input function */
4481 void *pIn, /* First arg for xInput */
4482 int(*xFilter)(
4483 void *pCtx, /* Copy of sixth arg to _apply() */
4484 const char *zTab /* Table name */
4486 int(*xConflict)(
4487 void *pCtx, /* Copy of sixth arg to _apply() */
4488 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4489 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4491 void *pCtx, /* First argument passed to xConflict */
4492 void **ppRebase, int *pnRebase,
4493 int flags
4495 sqlite3_changeset_iter *pIter; /* Iterator to skip through changeset */
4496 int rc = sqlite3changeset_start_strm(&pIter, xInput, pIn);
4497 if( rc==SQLITE_OK ){
4498 rc = sessionChangesetApply(
4499 db, pIter, xFilter, xConflict, pCtx, ppRebase, pnRebase, flags
4502 return rc;
4504 int sqlite3changeset_apply_strm(
4505 sqlite3 *db, /* Apply change to "main" db of this handle */
4506 int (*xInput)(void *pIn, void *pData, int *pnData), /* Input function */
4507 void *pIn, /* First arg for xInput */
4508 int(*xFilter)(
4509 void *pCtx, /* Copy of sixth arg to _apply() */
4510 const char *zTab /* Table name */
4512 int(*xConflict)(
4513 void *pCtx, /* Copy of sixth arg to _apply() */
4514 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4515 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4517 void *pCtx /* First argument passed to xConflict */
4519 return sqlite3changeset_apply_v2_strm(
4520 db, xInput, pIn, xFilter, xConflict, pCtx, 0, 0, 0
4525 ** sqlite3_changegroup handle.
4527 struct sqlite3_changegroup {
4528 int rc; /* Error code */
4529 int bPatch; /* True to accumulate patchsets */
4530 SessionTable *pList; /* List of tables in current patch */
4534 ** This function is called to merge two changes to the same row together as
4535 ** part of an sqlite3changeset_concat() operation. A new change object is
4536 ** allocated and a pointer to it stored in *ppNew.
4538 static int sessionChangeMerge(
4539 SessionTable *pTab, /* Table structure */
4540 int bRebase, /* True for a rebase hash-table */
4541 int bPatchset, /* True for patchsets */
4542 SessionChange *pExist, /* Existing change */
4543 int op2, /* Second change operation */
4544 int bIndirect, /* True if second change is indirect */
4545 u8 *aRec, /* Second change record */
4546 int nRec, /* Number of bytes in aRec */
4547 SessionChange **ppNew /* OUT: Merged change */
4549 SessionChange *pNew = 0;
4550 int rc = SQLITE_OK;
4552 if( !pExist ){
4553 pNew = (SessionChange *)sqlite3_malloc(sizeof(SessionChange) + nRec);
4554 if( !pNew ){
4555 return SQLITE_NOMEM;
4557 memset(pNew, 0, sizeof(SessionChange));
4558 pNew->op = op2;
4559 pNew->bIndirect = bIndirect;
4560 pNew->aRecord = (u8*)&pNew[1];
4561 if( bIndirect==0 || bRebase==0 ){
4562 pNew->nRecord = nRec;
4563 memcpy(pNew->aRecord, aRec, nRec);
4564 }else{
4565 int i;
4566 u8 *pIn = aRec;
4567 u8 *pOut = pNew->aRecord;
4568 for(i=0; i<pTab->nCol; i++){
4569 int nIn = sessionSerialLen(pIn);
4570 if( *pIn==0 ){
4571 *pOut++ = 0;
4572 }else if( pTab->abPK[i]==0 ){
4573 *pOut++ = 0xFF;
4574 }else{
4575 memcpy(pOut, pIn, nIn);
4576 pOut += nIn;
4578 pIn += nIn;
4580 pNew->nRecord = pOut - pNew->aRecord;
4582 }else if( bRebase ){
4583 if( pExist->op==SQLITE_DELETE && pExist->bIndirect ){
4584 *ppNew = pExist;
4585 }else{
4586 int nByte = nRec + pExist->nRecord + sizeof(SessionChange);
4587 pNew = (SessionChange*)sqlite3_malloc(nByte);
4588 if( pNew==0 ){
4589 rc = SQLITE_NOMEM;
4590 }else{
4591 int i;
4592 u8 *a1 = pExist->aRecord;
4593 u8 *a2 = aRec;
4594 u8 *pOut;
4596 memset(pNew, 0, nByte);
4597 pNew->bIndirect = bIndirect || pExist->bIndirect;
4598 pNew->op = op2;
4599 pOut = pNew->aRecord = (u8*)&pNew[1];
4601 for(i=0; i<pTab->nCol; i++){
4602 int n1 = sessionSerialLen(a1);
4603 int n2 = sessionSerialLen(a2);
4604 if( *a1==0xFF || (pTab->abPK[i]==0 && bIndirect) ){
4605 *pOut++ = 0xFF;
4606 }else if( *a2==0 ){
4607 memcpy(pOut, a1, n1);
4608 pOut += n1;
4609 }else{
4610 memcpy(pOut, a2, n2);
4611 pOut += n2;
4613 a1 += n1;
4614 a2 += n2;
4616 pNew->nRecord = pOut - pNew->aRecord;
4618 sqlite3_free(pExist);
4620 }else{
4621 int op1 = pExist->op;
4624 ** op1=INSERT, op2=INSERT -> Unsupported. Discard op2.
4625 ** op1=INSERT, op2=UPDATE -> INSERT.
4626 ** op1=INSERT, op2=DELETE -> (none)
4628 ** op1=UPDATE, op2=INSERT -> Unsupported. Discard op2.
4629 ** op1=UPDATE, op2=UPDATE -> UPDATE.
4630 ** op1=UPDATE, op2=DELETE -> DELETE.
4632 ** op1=DELETE, op2=INSERT -> UPDATE.
4633 ** op1=DELETE, op2=UPDATE -> Unsupported. Discard op2.
4634 ** op1=DELETE, op2=DELETE -> Unsupported. Discard op2.
4636 if( (op1==SQLITE_INSERT && op2==SQLITE_INSERT)
4637 || (op1==SQLITE_UPDATE && op2==SQLITE_INSERT)
4638 || (op1==SQLITE_DELETE && op2==SQLITE_UPDATE)
4639 || (op1==SQLITE_DELETE && op2==SQLITE_DELETE)
4641 pNew = pExist;
4642 }else if( op1==SQLITE_INSERT && op2==SQLITE_DELETE ){
4643 sqlite3_free(pExist);
4644 assert( pNew==0 );
4645 }else{
4646 u8 *aExist = pExist->aRecord;
4647 int nByte;
4648 u8 *aCsr;
4650 /* Allocate a new SessionChange object. Ensure that the aRecord[]
4651 ** buffer of the new object is large enough to hold any record that
4652 ** may be generated by combining the input records. */
4653 nByte = sizeof(SessionChange) + pExist->nRecord + nRec;
4654 pNew = (SessionChange *)sqlite3_malloc(nByte);
4655 if( !pNew ){
4656 sqlite3_free(pExist);
4657 return SQLITE_NOMEM;
4659 memset(pNew, 0, sizeof(SessionChange));
4660 pNew->bIndirect = (bIndirect && pExist->bIndirect);
4661 aCsr = pNew->aRecord = (u8 *)&pNew[1];
4663 if( op1==SQLITE_INSERT ){ /* INSERT + UPDATE */
4664 u8 *a1 = aRec;
4665 assert( op2==SQLITE_UPDATE );
4666 pNew->op = SQLITE_INSERT;
4667 if( bPatchset==0 ) sessionSkipRecord(&a1, pTab->nCol);
4668 sessionMergeRecord(&aCsr, pTab->nCol, aExist, a1);
4669 }else if( op1==SQLITE_DELETE ){ /* DELETE + INSERT */
4670 assert( op2==SQLITE_INSERT );
4671 pNew->op = SQLITE_UPDATE;
4672 if( bPatchset ){
4673 memcpy(aCsr, aRec, nRec);
4674 aCsr += nRec;
4675 }else{
4676 if( 0==sessionMergeUpdate(&aCsr, pTab, bPatchset, aExist, 0,aRec,0) ){
4677 sqlite3_free(pNew);
4678 pNew = 0;
4681 }else if( op2==SQLITE_UPDATE ){ /* UPDATE + UPDATE */
4682 u8 *a1 = aExist;
4683 u8 *a2 = aRec;
4684 assert( op1==SQLITE_UPDATE );
4685 if( bPatchset==0 ){
4686 sessionSkipRecord(&a1, pTab->nCol);
4687 sessionSkipRecord(&a2, pTab->nCol);
4689 pNew->op = SQLITE_UPDATE;
4690 if( 0==sessionMergeUpdate(&aCsr, pTab, bPatchset, aRec, aExist,a1,a2) ){
4691 sqlite3_free(pNew);
4692 pNew = 0;
4694 }else{ /* UPDATE + DELETE */
4695 assert( op1==SQLITE_UPDATE && op2==SQLITE_DELETE );
4696 pNew->op = SQLITE_DELETE;
4697 if( bPatchset ){
4698 memcpy(aCsr, aRec, nRec);
4699 aCsr += nRec;
4700 }else{
4701 sessionMergeRecord(&aCsr, pTab->nCol, aRec, aExist);
4705 if( pNew ){
4706 pNew->nRecord = (int)(aCsr - pNew->aRecord);
4708 sqlite3_free(pExist);
4712 *ppNew = pNew;
4713 return rc;
4717 ** Add all changes in the changeset traversed by the iterator passed as
4718 ** the first argument to the changegroup hash tables.
4720 static int sessionChangesetToHash(
4721 sqlite3_changeset_iter *pIter, /* Iterator to read from */
4722 sqlite3_changegroup *pGrp, /* Changegroup object to add changeset to */
4723 int bRebase /* True if hash table is for rebasing */
4725 u8 *aRec;
4726 int nRec;
4727 int rc = SQLITE_OK;
4728 SessionTable *pTab = 0;
4730 while( SQLITE_ROW==sessionChangesetNext(pIter, &aRec, &nRec, 0) ){
4731 const char *zNew;
4732 int nCol;
4733 int op;
4734 int iHash;
4735 int bIndirect;
4736 SessionChange *pChange;
4737 SessionChange *pExist = 0;
4738 SessionChange **pp;
4740 if( pGrp->pList==0 ){
4741 pGrp->bPatch = pIter->bPatchset;
4742 }else if( pIter->bPatchset!=pGrp->bPatch ){
4743 rc = SQLITE_ERROR;
4744 break;
4747 sqlite3changeset_op(pIter, &zNew, &nCol, &op, &bIndirect);
4748 if( !pTab || sqlite3_stricmp(zNew, pTab->zName) ){
4749 /* Search the list for a matching table */
4750 int nNew = (int)strlen(zNew);
4751 u8 *abPK;
4753 sqlite3changeset_pk(pIter, &abPK, 0);
4754 for(pTab = pGrp->pList; pTab; pTab=pTab->pNext){
4755 if( 0==sqlite3_strnicmp(pTab->zName, zNew, nNew+1) ) break;
4757 if( !pTab ){
4758 SessionTable **ppTab;
4760 pTab = sqlite3_malloc(sizeof(SessionTable) + nCol + nNew+1);
4761 if( !pTab ){
4762 rc = SQLITE_NOMEM;
4763 break;
4765 memset(pTab, 0, sizeof(SessionTable));
4766 pTab->nCol = nCol;
4767 pTab->abPK = (u8*)&pTab[1];
4768 memcpy(pTab->abPK, abPK, nCol);
4769 pTab->zName = (char*)&pTab->abPK[nCol];
4770 memcpy(pTab->zName, zNew, nNew+1);
4772 /* The new object must be linked on to the end of the list, not
4773 ** simply added to the start of it. This is to ensure that the
4774 ** tables within the output of sqlite3changegroup_output() are in
4775 ** the right order. */
4776 for(ppTab=&pGrp->pList; *ppTab; ppTab=&(*ppTab)->pNext);
4777 *ppTab = pTab;
4778 }else if( pTab->nCol!=nCol || memcmp(pTab->abPK, abPK, nCol) ){
4779 rc = SQLITE_SCHEMA;
4780 break;
4784 if( sessionGrowHash(pIter->bPatchset, pTab) ){
4785 rc = SQLITE_NOMEM;
4786 break;
4788 iHash = sessionChangeHash(
4789 pTab, (pIter->bPatchset && op==SQLITE_DELETE), aRec, pTab->nChange
4792 /* Search for existing entry. If found, remove it from the hash table.
4793 ** Code below may link it back in.
4795 for(pp=&pTab->apChange[iHash]; *pp; pp=&(*pp)->pNext){
4796 int bPkOnly1 = 0;
4797 int bPkOnly2 = 0;
4798 if( pIter->bPatchset ){
4799 bPkOnly1 = (*pp)->op==SQLITE_DELETE;
4800 bPkOnly2 = op==SQLITE_DELETE;
4802 if( sessionChangeEqual(pTab, bPkOnly1, (*pp)->aRecord, bPkOnly2, aRec) ){
4803 pExist = *pp;
4804 *pp = (*pp)->pNext;
4805 pTab->nEntry--;
4806 break;
4810 rc = sessionChangeMerge(pTab, bRebase,
4811 pIter->bPatchset, pExist, op, bIndirect, aRec, nRec, &pChange
4813 if( rc ) break;
4814 if( pChange ){
4815 pChange->pNext = pTab->apChange[iHash];
4816 pTab->apChange[iHash] = pChange;
4817 pTab->nEntry++;
4821 if( rc==SQLITE_OK ) rc = pIter->rc;
4822 return rc;
4826 ** Serialize a changeset (or patchset) based on all changesets (or patchsets)
4827 ** added to the changegroup object passed as the first argument.
4829 ** If xOutput is not NULL, then the changeset/patchset is returned to the
4830 ** user via one or more calls to xOutput, as with the other streaming
4831 ** interfaces.
4833 ** Or, if xOutput is NULL, then (*ppOut) is populated with a pointer to a
4834 ** buffer containing the output changeset before this function returns. In
4835 ** this case (*pnOut) is set to the size of the output buffer in bytes. It
4836 ** is the responsibility of the caller to free the output buffer using
4837 ** sqlite3_free() when it is no longer required.
4839 ** If successful, SQLITE_OK is returned. Or, if an error occurs, an SQLite
4840 ** error code. If an error occurs and xOutput is NULL, (*ppOut) and (*pnOut)
4841 ** are both set to 0 before returning.
4843 static int sessionChangegroupOutput(
4844 sqlite3_changegroup *pGrp,
4845 int (*xOutput)(void *pOut, const void *pData, int nData),
4846 void *pOut,
4847 int *pnOut,
4848 void **ppOut
4850 int rc = SQLITE_OK;
4851 SessionBuffer buf = {0, 0, 0};
4852 SessionTable *pTab;
4853 assert( xOutput==0 || (ppOut==0 && pnOut==0) );
4855 /* Create the serialized output changeset based on the contents of the
4856 ** hash tables attached to the SessionTable objects in list p->pList.
4858 for(pTab=pGrp->pList; rc==SQLITE_OK && pTab; pTab=pTab->pNext){
4859 int i;
4860 if( pTab->nEntry==0 ) continue;
4862 sessionAppendTableHdr(&buf, pGrp->bPatch, pTab, &rc);
4863 for(i=0; i<pTab->nChange; i++){
4864 SessionChange *p;
4865 for(p=pTab->apChange[i]; p; p=p->pNext){
4866 sessionAppendByte(&buf, p->op, &rc);
4867 sessionAppendByte(&buf, p->bIndirect, &rc);
4868 sessionAppendBlob(&buf, p->aRecord, p->nRecord, &rc);
4872 if( rc==SQLITE_OK && xOutput && buf.nBuf>=SESSIONS_STRM_CHUNK_SIZE ){
4873 rc = xOutput(pOut, buf.aBuf, buf.nBuf);
4874 buf.nBuf = 0;
4878 if( rc==SQLITE_OK ){
4879 if( xOutput ){
4880 if( buf.nBuf>0 ) rc = xOutput(pOut, buf.aBuf, buf.nBuf);
4881 }else{
4882 *ppOut = buf.aBuf;
4883 *pnOut = buf.nBuf;
4884 buf.aBuf = 0;
4887 sqlite3_free(buf.aBuf);
4889 return rc;
4893 ** Allocate a new, empty, sqlite3_changegroup.
4895 int sqlite3changegroup_new(sqlite3_changegroup **pp){
4896 int rc = SQLITE_OK; /* Return code */
4897 sqlite3_changegroup *p; /* New object */
4898 p = (sqlite3_changegroup*)sqlite3_malloc(sizeof(sqlite3_changegroup));
4899 if( p==0 ){
4900 rc = SQLITE_NOMEM;
4901 }else{
4902 memset(p, 0, sizeof(sqlite3_changegroup));
4904 *pp = p;
4905 return rc;
4909 ** Add the changeset currently stored in buffer pData, size nData bytes,
4910 ** to changeset-group p.
4912 int sqlite3changegroup_add(sqlite3_changegroup *pGrp, int nData, void *pData){
4913 sqlite3_changeset_iter *pIter; /* Iterator opened on pData/nData */
4914 int rc; /* Return code */
4916 rc = sqlite3changeset_start(&pIter, nData, pData);
4917 if( rc==SQLITE_OK ){
4918 rc = sessionChangesetToHash(pIter, pGrp, 0);
4920 sqlite3changeset_finalize(pIter);
4921 return rc;
4925 ** Obtain a buffer containing a changeset representing the concatenation
4926 ** of all changesets added to the group so far.
4928 int sqlite3changegroup_output(
4929 sqlite3_changegroup *pGrp,
4930 int *pnData,
4931 void **ppData
4933 return sessionChangegroupOutput(pGrp, 0, 0, pnData, ppData);
4937 ** Streaming versions of changegroup_add().
4939 int sqlite3changegroup_add_strm(
4940 sqlite3_changegroup *pGrp,
4941 int (*xInput)(void *pIn, void *pData, int *pnData),
4942 void *pIn
4944 sqlite3_changeset_iter *pIter; /* Iterator opened on pData/nData */
4945 int rc; /* Return code */
4947 rc = sqlite3changeset_start_strm(&pIter, xInput, pIn);
4948 if( rc==SQLITE_OK ){
4949 rc = sessionChangesetToHash(pIter, pGrp, 0);
4951 sqlite3changeset_finalize(pIter);
4952 return rc;
4956 ** Streaming versions of changegroup_output().
4958 int sqlite3changegroup_output_strm(
4959 sqlite3_changegroup *pGrp,
4960 int (*xOutput)(void *pOut, const void *pData, int nData),
4961 void *pOut
4963 return sessionChangegroupOutput(pGrp, xOutput, pOut, 0, 0);
4967 ** Delete a changegroup object.
4969 void sqlite3changegroup_delete(sqlite3_changegroup *pGrp){
4970 if( pGrp ){
4971 sessionDeleteTable(pGrp->pList);
4972 sqlite3_free(pGrp);
4977 ** Combine two changesets together.
4979 int sqlite3changeset_concat(
4980 int nLeft, /* Number of bytes in lhs input */
4981 void *pLeft, /* Lhs input changeset */
4982 int nRight /* Number of bytes in rhs input */,
4983 void *pRight, /* Rhs input changeset */
4984 int *pnOut, /* OUT: Number of bytes in output changeset */
4985 void **ppOut /* OUT: changeset (left <concat> right) */
4987 sqlite3_changegroup *pGrp;
4988 int rc;
4990 rc = sqlite3changegroup_new(&pGrp);
4991 if( rc==SQLITE_OK ){
4992 rc = sqlite3changegroup_add(pGrp, nLeft, pLeft);
4994 if( rc==SQLITE_OK ){
4995 rc = sqlite3changegroup_add(pGrp, nRight, pRight);
4997 if( rc==SQLITE_OK ){
4998 rc = sqlite3changegroup_output(pGrp, pnOut, ppOut);
5000 sqlite3changegroup_delete(pGrp);
5002 return rc;
5006 ** Streaming version of sqlite3changeset_concat().
5008 int sqlite3changeset_concat_strm(
5009 int (*xInputA)(void *pIn, void *pData, int *pnData),
5010 void *pInA,
5011 int (*xInputB)(void *pIn, void *pData, int *pnData),
5012 void *pInB,
5013 int (*xOutput)(void *pOut, const void *pData, int nData),
5014 void *pOut
5016 sqlite3_changegroup *pGrp;
5017 int rc;
5019 rc = sqlite3changegroup_new(&pGrp);
5020 if( rc==SQLITE_OK ){
5021 rc = sqlite3changegroup_add_strm(pGrp, xInputA, pInA);
5023 if( rc==SQLITE_OK ){
5024 rc = sqlite3changegroup_add_strm(pGrp, xInputB, pInB);
5026 if( rc==SQLITE_OK ){
5027 rc = sqlite3changegroup_output_strm(pGrp, xOutput, pOut);
5029 sqlite3changegroup_delete(pGrp);
5031 return rc;
5035 ** Changeset rebaser handle.
5037 struct sqlite3_rebaser {
5038 sqlite3_changegroup grp; /* Hash table */
5042 ** Buffers a1 and a2 must both contain a sessions module record nCol
5043 ** fields in size. This function appends an nCol sessions module
5044 ** record to buffer pBuf that is a copy of a1, except that for
5045 ** each field that is undefined in a1[], swap in the field from a2[].
5047 static void sessionAppendRecordMerge(
5048 SessionBuffer *pBuf, /* Buffer to append to */
5049 int nCol, /* Number of columns in each record */
5050 u8 *a1, int n1, /* Record 1 */
5051 u8 *a2, int n2, /* Record 2 */
5052 int *pRc /* IN/OUT: error code */
5054 sessionBufferGrow(pBuf, n1+n2, pRc);
5055 if( *pRc==SQLITE_OK ){
5056 int i;
5057 u8 *pOut = &pBuf->aBuf[pBuf->nBuf];
5058 for(i=0; i<nCol; i++){
5059 int nn1 = sessionSerialLen(a1);
5060 int nn2 = sessionSerialLen(a2);
5061 if( *a1==0 || *a1==0xFF ){
5062 memcpy(pOut, a2, nn2);
5063 pOut += nn2;
5064 }else{
5065 memcpy(pOut, a1, nn1);
5066 pOut += nn1;
5068 a1 += nn1;
5069 a2 += nn2;
5072 pBuf->nBuf = pOut-pBuf->aBuf;
5073 assert( pBuf->nBuf<=pBuf->nAlloc );
5078 ** This function is called when rebasing a local UPDATE change against one
5079 ** or more remote UPDATE changes. The aRec/nRec buffer contains the current
5080 ** old.* and new.* records for the change. The rebase buffer (a single
5081 ** record) is in aChange/nChange. The rebased change is appended to buffer
5082 ** pBuf.
5084 ** Rebasing the UPDATE involves:
5086 ** * Removing any changes to fields for which the corresponding field
5087 ** in the rebase buffer is set to "replaced" (type 0xFF). If this
5088 ** means the UPDATE change updates no fields, nothing is appended
5089 ** to the output buffer.
5091 ** * For each field modified by the local change for which the
5092 ** corresponding field in the rebase buffer is not "undefined" (0x00)
5093 ** or "replaced" (0xFF), the old.* value is replaced by the value
5094 ** in the rebase buffer.
5096 static void sessionAppendPartialUpdate(
5097 SessionBuffer *pBuf, /* Append record here */
5098 sqlite3_changeset_iter *pIter, /* Iterator pointed at local change */
5099 u8 *aRec, int nRec, /* Local change */
5100 u8 *aChange, int nChange, /* Record to rebase against */
5101 int *pRc /* IN/OUT: Return Code */
5103 sessionBufferGrow(pBuf, 2+nRec+nChange, pRc);
5104 if( *pRc==SQLITE_OK ){
5105 int bData = 0;
5106 u8 *pOut = &pBuf->aBuf[pBuf->nBuf];
5107 int i;
5108 u8 *a1 = aRec;
5109 u8 *a2 = aChange;
5111 *pOut++ = SQLITE_UPDATE;
5112 *pOut++ = pIter->bIndirect;
5113 for(i=0; i<pIter->nCol; i++){
5114 int n1 = sessionSerialLen(a1);
5115 int n2 = sessionSerialLen(a2);
5116 if( pIter->abPK[i] || a2[0]==0 ){
5117 if( !pIter->abPK[i] ) bData = 1;
5118 memcpy(pOut, a1, n1);
5119 pOut += n1;
5120 }else if( a2[0]!=0xFF ){
5121 bData = 1;
5122 memcpy(pOut, a2, n2);
5123 pOut += n2;
5124 }else{
5125 *pOut++ = '\0';
5127 a1 += n1;
5128 a2 += n2;
5130 if( bData ){
5131 a2 = aChange;
5132 for(i=0; i<pIter->nCol; i++){
5133 int n1 = sessionSerialLen(a1);
5134 int n2 = sessionSerialLen(a2);
5135 if( pIter->abPK[i] || a2[0]!=0xFF ){
5136 memcpy(pOut, a1, n1);
5137 pOut += n1;
5138 }else{
5139 *pOut++ = '\0';
5141 a1 += n1;
5142 a2 += n2;
5144 pBuf->nBuf = (pOut - pBuf->aBuf);
5150 ** pIter is configured to iterate through a changeset. This function rebases
5151 ** that changeset according to the current configuration of the rebaser
5152 ** object passed as the first argument. If no error occurs and argument xOutput
5153 ** is not NULL, then the changeset is returned to the caller by invoking
5154 ** xOutput zero or more times and SQLITE_OK returned. Or, if xOutput is NULL,
5155 ** then (*ppOut) is set to point to a buffer containing the rebased changeset
5156 ** before this function returns. In this case (*pnOut) is set to the size of
5157 ** the buffer in bytes. It is the responsibility of the caller to eventually
5158 ** free the (*ppOut) buffer using sqlite3_free().
5160 ** If an error occurs, an SQLite error code is returned. If ppOut and
5161 ** pnOut are not NULL, then the two output parameters are set to 0 before
5162 ** returning.
5164 static int sessionRebase(
5165 sqlite3_rebaser *p, /* Rebaser hash table */
5166 sqlite3_changeset_iter *pIter, /* Input data */
5167 int (*xOutput)(void *pOut, const void *pData, int nData),
5168 void *pOut, /* Context for xOutput callback */
5169 int *pnOut, /* OUT: Number of bytes in output changeset */
5170 void **ppOut /* OUT: Inverse of pChangeset */
5172 int rc = SQLITE_OK;
5173 u8 *aRec = 0;
5174 int nRec = 0;
5175 int bNew = 0;
5176 SessionTable *pTab = 0;
5177 SessionBuffer sOut = {0,0,0};
5179 while( SQLITE_ROW==sessionChangesetNext(pIter, &aRec, &nRec, &bNew) ){
5180 SessionChange *pChange = 0;
5181 int bDone = 0;
5183 if( bNew ){
5184 const char *zTab = pIter->zTab;
5185 for(pTab=p->grp.pList; pTab; pTab=pTab->pNext){
5186 if( 0==sqlite3_stricmp(pTab->zName, zTab) ) break;
5188 bNew = 0;
5190 /* A patchset may not be rebased */
5191 if( pIter->bPatchset ){
5192 rc = SQLITE_ERROR;
5195 /* Append a table header to the output for this new table */
5196 sessionAppendByte(&sOut, pIter->bPatchset ? 'P' : 'T', &rc);
5197 sessionAppendVarint(&sOut, pIter->nCol, &rc);
5198 sessionAppendBlob(&sOut, pIter->abPK, pIter->nCol, &rc);
5199 sessionAppendBlob(&sOut,(u8*)pIter->zTab,(int)strlen(pIter->zTab)+1,&rc);
5202 if( pTab && rc==SQLITE_OK ){
5203 int iHash = sessionChangeHash(pTab, 0, aRec, pTab->nChange);
5205 for(pChange=pTab->apChange[iHash]; pChange; pChange=pChange->pNext){
5206 if( sessionChangeEqual(pTab, 0, aRec, 0, pChange->aRecord) ){
5207 break;
5212 if( pChange ){
5213 assert( pChange->op==SQLITE_DELETE || pChange->op==SQLITE_INSERT );
5214 switch( pIter->op ){
5215 case SQLITE_INSERT:
5216 if( pChange->op==SQLITE_INSERT ){
5217 bDone = 1;
5218 if( pChange->bIndirect==0 ){
5219 sessionAppendByte(&sOut, SQLITE_UPDATE, &rc);
5220 sessionAppendByte(&sOut, pIter->bIndirect, &rc);
5221 sessionAppendBlob(&sOut, pChange->aRecord, pChange->nRecord, &rc);
5222 sessionAppendBlob(&sOut, aRec, nRec, &rc);
5225 break;
5227 case SQLITE_UPDATE:
5228 bDone = 1;
5229 if( pChange->op==SQLITE_DELETE ){
5230 if( pChange->bIndirect==0 ){
5231 u8 *pCsr = aRec;
5232 sessionSkipRecord(&pCsr, pIter->nCol);
5233 sessionAppendByte(&sOut, SQLITE_INSERT, &rc);
5234 sessionAppendByte(&sOut, pIter->bIndirect, &rc);
5235 sessionAppendRecordMerge(&sOut, pIter->nCol,
5236 pCsr, nRec-(pCsr-aRec),
5237 pChange->aRecord, pChange->nRecord, &rc
5240 }else{
5241 sessionAppendPartialUpdate(&sOut, pIter,
5242 aRec, nRec, pChange->aRecord, pChange->nRecord, &rc
5245 break;
5247 default:
5248 assert( pIter->op==SQLITE_DELETE );
5249 bDone = 1;
5250 if( pChange->op==SQLITE_INSERT ){
5251 sessionAppendByte(&sOut, SQLITE_DELETE, &rc);
5252 sessionAppendByte(&sOut, pIter->bIndirect, &rc);
5253 sessionAppendRecordMerge(&sOut, pIter->nCol,
5254 pChange->aRecord, pChange->nRecord, aRec, nRec, &rc
5257 break;
5261 if( bDone==0 ){
5262 sessionAppendByte(&sOut, pIter->op, &rc);
5263 sessionAppendByte(&sOut, pIter->bIndirect, &rc);
5264 sessionAppendBlob(&sOut, aRec, nRec, &rc);
5266 if( rc==SQLITE_OK && xOutput && sOut.nBuf>SESSIONS_STRM_CHUNK_SIZE ){
5267 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf);
5268 sOut.nBuf = 0;
5270 if( rc ) break;
5273 if( rc!=SQLITE_OK ){
5274 sqlite3_free(sOut.aBuf);
5275 memset(&sOut, 0, sizeof(sOut));
5278 if( rc==SQLITE_OK ){
5279 if( xOutput ){
5280 if( sOut.nBuf>0 ){
5281 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf);
5283 }else{
5284 *ppOut = (void*)sOut.aBuf;
5285 *pnOut = sOut.nBuf;
5286 sOut.aBuf = 0;
5289 sqlite3_free(sOut.aBuf);
5290 return rc;
5294 ** Create a new rebaser object.
5296 int sqlite3rebaser_create(sqlite3_rebaser **ppNew){
5297 int rc = SQLITE_OK;
5298 sqlite3_rebaser *pNew;
5300 pNew = sqlite3_malloc(sizeof(sqlite3_rebaser));
5301 if( pNew==0 ){
5302 rc = SQLITE_NOMEM;
5303 }else{
5304 memset(pNew, 0, sizeof(sqlite3_rebaser));
5306 *ppNew = pNew;
5307 return rc;
5311 ** Call this one or more times to configure a rebaser.
5313 int sqlite3rebaser_configure(
5314 sqlite3_rebaser *p,
5315 int nRebase, const void *pRebase
5317 sqlite3_changeset_iter *pIter = 0; /* Iterator opened on pData/nData */
5318 int rc; /* Return code */
5319 rc = sqlite3changeset_start(&pIter, nRebase, (void*)pRebase);
5320 if( rc==SQLITE_OK ){
5321 rc = sessionChangesetToHash(pIter, &p->grp, 1);
5323 sqlite3changeset_finalize(pIter);
5324 return rc;
5328 ** Rebase a changeset according to current rebaser configuration
5330 int sqlite3rebaser_rebase(
5331 sqlite3_rebaser *p,
5332 int nIn, const void *pIn,
5333 int *pnOut, void **ppOut
5335 sqlite3_changeset_iter *pIter = 0; /* Iterator to skip through input */
5336 int rc = sqlite3changeset_start(&pIter, nIn, (void*)pIn);
5338 if( rc==SQLITE_OK ){
5339 rc = sessionRebase(p, pIter, 0, 0, pnOut, ppOut);
5340 sqlite3changeset_finalize(pIter);
5343 return rc;
5347 ** Rebase a changeset according to current rebaser configuration
5349 int sqlite3rebaser_rebase_strm(
5350 sqlite3_rebaser *p,
5351 int (*xInput)(void *pIn, void *pData, int *pnData),
5352 void *pIn,
5353 int (*xOutput)(void *pOut, const void *pData, int nData),
5354 void *pOut
5356 sqlite3_changeset_iter *pIter = 0; /* Iterator to skip through input */
5357 int rc = sqlite3changeset_start_strm(&pIter, xInput, pIn);
5359 if( rc==SQLITE_OK ){
5360 rc = sessionRebase(p, pIter, xOutput, pOut, 0, 0);
5361 sqlite3changeset_finalize(pIter);
5364 return rc;
5368 ** Destroy a rebaser object
5370 void sqlite3rebaser_delete(sqlite3_rebaser *p){
5371 if( p ){
5372 sessionDeleteTable(p->grp.pList);
5373 sqlite3_free(p);
5377 #endif /* SQLITE_ENABLE_SESSION && SQLITE_ENABLE_PREUPDATE_HOOK */