Snapshot of upstream SQLite 3.41.0
[sqlcipher.git] / ext / session / sqlite3session.c
bloba3f28abe9ec8df184b33c0b62142a217a782070c
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 static int sessions_strm_chunk_size = SESSIONS_STRM_CHUNK_SIZE;
30 typedef struct SessionHook SessionHook;
31 struct SessionHook {
32 void *pCtx;
33 int (*xOld)(void*,int,sqlite3_value**);
34 int (*xNew)(void*,int,sqlite3_value**);
35 int (*xCount)(void*);
36 int (*xDepth)(void*);
40 ** Session handle structure.
42 struct sqlite3_session {
43 sqlite3 *db; /* Database handle session is attached to */
44 char *zDb; /* Name of database session is attached to */
45 int bEnableSize; /* True if changeset_size() enabled */
46 int bEnable; /* True if currently recording */
47 int bIndirect; /* True if all changes are indirect */
48 int bAutoAttach; /* True to auto-attach tables */
49 int rc; /* Non-zero if an error has occurred */
50 void *pFilterCtx; /* First argument to pass to xTableFilter */
51 int (*xTableFilter)(void *pCtx, const char *zTab);
52 i64 nMalloc; /* Number of bytes of data allocated */
53 i64 nMaxChangesetSize;
54 sqlite3_value *pZeroBlob; /* Value containing X'' */
55 sqlite3_session *pNext; /* Next session object on same db. */
56 SessionTable *pTable; /* List of attached tables */
57 SessionHook hook; /* APIs to grab new and old data with */
61 ** Instances of this structure are used to build strings or binary records.
63 struct SessionBuffer {
64 u8 *aBuf; /* Pointer to changeset buffer */
65 int nBuf; /* Size of buffer aBuf */
66 int nAlloc; /* Size of allocation containing aBuf */
70 ** An object of this type is used internally as an abstraction for
71 ** input data. Input data may be supplied either as a single large buffer
72 ** (e.g. sqlite3changeset_start()) or using a stream function (e.g.
73 ** sqlite3changeset_start_strm()).
75 struct SessionInput {
76 int bNoDiscard; /* If true, do not discard in InputBuffer() */
77 int iCurrent; /* Offset in aData[] of current change */
78 int iNext; /* Offset in aData[] of next change */
79 u8 *aData; /* Pointer to buffer containing changeset */
80 int nData; /* Number of bytes in aData */
82 SessionBuffer buf; /* Current read buffer */
83 int (*xInput)(void*, void*, int*); /* Input stream call (or NULL) */
84 void *pIn; /* First argument to xInput */
85 int bEof; /* Set to true after xInput finished */
89 ** Structure for changeset iterators.
91 struct sqlite3_changeset_iter {
92 SessionInput in; /* Input buffer or stream */
93 SessionBuffer tblhdr; /* Buffer to hold apValue/zTab/abPK/ */
94 int bPatchset; /* True if this is a patchset */
95 int bInvert; /* True to invert changeset */
96 int bSkipEmpty; /* Skip noop UPDATE changes */
97 int rc; /* Iterator error code */
98 sqlite3_stmt *pConflict; /* Points to conflicting row, if any */
99 char *zTab; /* Current table */
100 int nCol; /* Number of columns in zTab */
101 int op; /* Current operation */
102 int bIndirect; /* True if current change was indirect */
103 u8 *abPK; /* Primary key array */
104 sqlite3_value **apValue; /* old.* and new.* values */
108 ** Each session object maintains a set of the following structures, one
109 ** for each table the session object is monitoring. The structures are
110 ** stored in a linked list starting at sqlite3_session.pTable.
112 ** The keys of the SessionTable.aChange[] hash table are all rows that have
113 ** been modified in any way since the session object was attached to the
114 ** table.
116 ** The data associated with each hash-table entry is a structure containing
117 ** a subset of the initial values that the modified row contained at the
118 ** start of the session. Or no initial values if the row was inserted.
120 struct SessionTable {
121 SessionTable *pNext;
122 char *zName; /* Local name of table */
123 int nCol; /* Number of columns in table zName */
124 int bStat1; /* True if this is sqlite_stat1 */
125 const char **azCol; /* Column names */
126 u8 *abPK; /* Array of primary key flags */
127 int nEntry; /* Total number of entries in hash table */
128 int nChange; /* Size of apChange[] array */
129 SessionChange **apChange; /* Hash table buckets */
133 ** RECORD FORMAT:
135 ** The following record format is similar to (but not compatible with) that
136 ** used in SQLite database files. This format is used as part of the
137 ** change-set binary format, and so must be architecture independent.
139 ** Unlike the SQLite database record format, each field is self-contained -
140 ** there is no separation of header and data. Each field begins with a
141 ** single byte describing its type, as follows:
143 ** 0x00: Undefined value.
144 ** 0x01: Integer value.
145 ** 0x02: Real value.
146 ** 0x03: Text value.
147 ** 0x04: Blob value.
148 ** 0x05: SQL NULL value.
150 ** Note that the above match the definitions of SQLITE_INTEGER, SQLITE_TEXT
151 ** and so on in sqlite3.h. For undefined and NULL values, the field consists
152 ** only of the single type byte. For other types of values, the type byte
153 ** is followed by:
155 ** Text values:
156 ** A varint containing the number of bytes in the value (encoded using
157 ** UTF-8). Followed by a buffer containing the UTF-8 representation
158 ** of the text value. There is no nul terminator.
160 ** Blob values:
161 ** A varint containing the number of bytes in the value, followed by
162 ** a buffer containing the value itself.
164 ** Integer values:
165 ** An 8-byte big-endian integer value.
167 ** Real values:
168 ** An 8-byte big-endian IEEE 754-2008 real value.
170 ** Varint values are encoded in the same way as varints in the SQLite
171 ** record format.
173 ** CHANGESET FORMAT:
175 ** A changeset is a collection of DELETE, UPDATE and INSERT operations on
176 ** one or more tables. Operations on a single table are grouped together,
177 ** but may occur in any order (i.e. deletes, updates and inserts are all
178 ** mixed together).
180 ** Each group of changes begins with a table header:
182 ** 1 byte: Constant 0x54 (capital 'T')
183 ** Varint: Number of columns in the table.
184 ** nCol bytes: 0x01 for PK columns, 0x00 otherwise.
185 ** N bytes: Unqualified table name (encoded using UTF-8). Nul-terminated.
187 ** Followed by one or more changes to the table.
189 ** 1 byte: Either SQLITE_INSERT (0x12), UPDATE (0x17) or DELETE (0x09).
190 ** 1 byte: The "indirect-change" flag.
191 ** old.* record: (delete and update only)
192 ** new.* record: (insert and update only)
194 ** The "old.*" and "new.*" records, if present, are N field records in the
195 ** format described above under "RECORD FORMAT", where N is the number of
196 ** columns in the table. The i'th field of each record is associated with
197 ** the i'th column of the table, counting from left to right in the order
198 ** in which columns were declared in the CREATE TABLE statement.
200 ** The new.* record that is part of each INSERT change contains the values
201 ** that make up the new row. Similarly, the old.* record that is part of each
202 ** DELETE change contains the values that made up the row that was deleted
203 ** from the database. In the changeset format, the records that are part
204 ** of INSERT or DELETE changes never contain any undefined (type byte 0x00)
205 ** fields.
207 ** Within the old.* record associated with an UPDATE change, all fields
208 ** associated with table columns that are not PRIMARY KEY columns and are
209 ** not modified by the UPDATE change are set to "undefined". Other fields
210 ** are set to the values that made up the row before the UPDATE that the
211 ** change records took place. Within the new.* record, fields associated
212 ** with table columns modified by the UPDATE change contain the new
213 ** values. Fields associated with table columns that are not modified
214 ** are set to "undefined".
216 ** PATCHSET FORMAT:
218 ** A patchset is also a collection of changes. It is similar to a changeset,
219 ** but leaves undefined those fields that are not useful if no conflict
220 ** resolution is required when applying the changeset.
222 ** Each group of changes begins with a table header:
224 ** 1 byte: Constant 0x50 (capital 'P')
225 ** Varint: Number of columns in the table.
226 ** nCol bytes: 0x01 for PK columns, 0x00 otherwise.
227 ** N bytes: Unqualified table name (encoded using UTF-8). Nul-terminated.
229 ** Followed by one or more changes to the table.
231 ** 1 byte: Either SQLITE_INSERT (0x12), UPDATE (0x17) or DELETE (0x09).
232 ** 1 byte: The "indirect-change" flag.
233 ** single record: (PK fields for DELETE, PK and modified fields for UPDATE,
234 ** full record for INSERT).
236 ** As in the changeset format, each field of the single record that is part
237 ** of a patchset change is associated with the correspondingly positioned
238 ** table column, counting from left to right within the CREATE TABLE
239 ** statement.
241 ** For a DELETE change, all fields within the record except those associated
242 ** with PRIMARY KEY columns are omitted. The PRIMARY KEY fields contain the
243 ** values identifying the row to delete.
245 ** For an UPDATE change, all fields except those associated with PRIMARY KEY
246 ** columns and columns that are modified by the UPDATE are set to "undefined".
247 ** PRIMARY KEY fields contain the values identifying the table row to update,
248 ** and fields associated with modified columns contain the new column values.
250 ** The records associated with INSERT changes are in the same format as for
251 ** changesets. It is not possible for a record associated with an INSERT
252 ** change to contain a field set to "undefined".
254 ** REBASE BLOB FORMAT:
256 ** A rebase blob may be output by sqlite3changeset_apply_v2() and its
257 ** streaming equivalent for use with the sqlite3_rebaser APIs to rebase
258 ** existing changesets. A rebase blob contains one entry for each conflict
259 ** resolved using either the OMIT or REPLACE strategies within the apply_v2()
260 ** call.
262 ** The format used for a rebase blob is very similar to that used for
263 ** changesets. All entries related to a single table are grouped together.
265 ** Each group of entries begins with a table header in changeset format:
267 ** 1 byte: Constant 0x54 (capital 'T')
268 ** Varint: Number of columns in the table.
269 ** nCol bytes: 0x01 for PK columns, 0x00 otherwise.
270 ** N bytes: Unqualified table name (encoded using UTF-8). Nul-terminated.
272 ** Followed by one or more entries associated with the table.
274 ** 1 byte: Either SQLITE_INSERT (0x12), DELETE (0x09).
275 ** 1 byte: Flag. 0x01 for REPLACE, 0x00 for OMIT.
276 ** record: (in the record format defined above).
278 ** In a rebase blob, the first field is set to SQLITE_INSERT if the change
279 ** that caused the conflict was an INSERT or UPDATE, or to SQLITE_DELETE if
280 ** it was a DELETE. The second field is set to 0x01 if the conflict
281 ** resolution strategy was REPLACE, or 0x00 if it was OMIT.
283 ** If the change that caused the conflict was a DELETE, then the single
284 ** record is a copy of the old.* record from the original changeset. If it
285 ** was an INSERT, then the single record is a copy of the new.* record. If
286 ** the conflicting change was an UPDATE, then the single record is a copy
287 ** of the new.* record with the PK fields filled in based on the original
288 ** old.* record.
292 ** For each row modified during a session, there exists a single instance of
293 ** this structure stored in a SessionTable.aChange[] hash table.
295 struct SessionChange {
296 u8 op; /* One of UPDATE, DELETE, INSERT */
297 u8 bIndirect; /* True if this change is "indirect" */
298 int nMaxSize; /* Max size of eventual changeset record */
299 int nRecord; /* Number of bytes in buffer aRecord[] */
300 u8 *aRecord; /* Buffer containing old.* record */
301 SessionChange *pNext; /* For hash-table collisions */
305 ** Write a varint with value iVal into the buffer at aBuf. Return the
306 ** number of bytes written.
308 static int sessionVarintPut(u8 *aBuf, int iVal){
309 return putVarint32(aBuf, iVal);
313 ** Return the number of bytes required to store value iVal as a varint.
315 static int sessionVarintLen(int iVal){
316 return sqlite3VarintLen(iVal);
320 ** Read a varint value from aBuf[] into *piVal. Return the number of
321 ** bytes read.
323 static int sessionVarintGet(u8 *aBuf, int *piVal){
324 return getVarint32(aBuf, *piVal);
327 /* Load an unaligned and unsigned 32-bit integer */
328 #define SESSION_UINT32(x) (((u32)(x)[0]<<24)|((x)[1]<<16)|((x)[2]<<8)|(x)[3])
331 ** Read a 64-bit big-endian integer value from buffer aRec[]. Return
332 ** the value read.
334 static sqlite3_int64 sessionGetI64(u8 *aRec){
335 u64 x = SESSION_UINT32(aRec);
336 u32 y = SESSION_UINT32(aRec+4);
337 x = (x<<32) + y;
338 return (sqlite3_int64)x;
342 ** Write a 64-bit big-endian integer value to the buffer aBuf[].
344 static void sessionPutI64(u8 *aBuf, sqlite3_int64 i){
345 aBuf[0] = (i>>56) & 0xFF;
346 aBuf[1] = (i>>48) & 0xFF;
347 aBuf[2] = (i>>40) & 0xFF;
348 aBuf[3] = (i>>32) & 0xFF;
349 aBuf[4] = (i>>24) & 0xFF;
350 aBuf[5] = (i>>16) & 0xFF;
351 aBuf[6] = (i>> 8) & 0xFF;
352 aBuf[7] = (i>> 0) & 0xFF;
356 ** This function is used to serialize the contents of value pValue (see
357 ** comment titled "RECORD FORMAT" above).
359 ** If it is non-NULL, the serialized form of the value is written to
360 ** buffer aBuf. *pnWrite is set to the number of bytes written before
361 ** returning. Or, if aBuf is NULL, the only thing this function does is
362 ** set *pnWrite.
364 ** If no error occurs, SQLITE_OK is returned. Or, if an OOM error occurs
365 ** within a call to sqlite3_value_text() (may fail if the db is utf-16))
366 ** SQLITE_NOMEM is returned.
368 static int sessionSerializeValue(
369 u8 *aBuf, /* If non-NULL, write serialized value here */
370 sqlite3_value *pValue, /* Value to serialize */
371 sqlite3_int64 *pnWrite /* IN/OUT: Increment by bytes written */
373 int nByte; /* Size of serialized value in bytes */
375 if( pValue ){
376 int eType; /* Value type (SQLITE_NULL, TEXT etc.) */
378 eType = sqlite3_value_type(pValue);
379 if( aBuf ) aBuf[0] = eType;
381 switch( eType ){
382 case SQLITE_NULL:
383 nByte = 1;
384 break;
386 case SQLITE_INTEGER:
387 case SQLITE_FLOAT:
388 if( aBuf ){
389 /* TODO: SQLite does something special to deal with mixed-endian
390 ** floating point values (e.g. ARM7). This code probably should
391 ** too. */
392 u64 i;
393 if( eType==SQLITE_INTEGER ){
394 i = (u64)sqlite3_value_int64(pValue);
395 }else{
396 double r;
397 assert( sizeof(double)==8 && sizeof(u64)==8 );
398 r = sqlite3_value_double(pValue);
399 memcpy(&i, &r, 8);
401 sessionPutI64(&aBuf[1], i);
403 nByte = 9;
404 break;
406 default: {
407 u8 *z;
408 int n;
409 int nVarint;
411 assert( eType==SQLITE_TEXT || eType==SQLITE_BLOB );
412 if( eType==SQLITE_TEXT ){
413 z = (u8 *)sqlite3_value_text(pValue);
414 }else{
415 z = (u8 *)sqlite3_value_blob(pValue);
417 n = sqlite3_value_bytes(pValue);
418 if( z==0 && (eType!=SQLITE_BLOB || n>0) ) return SQLITE_NOMEM;
419 nVarint = sessionVarintLen(n);
421 if( aBuf ){
422 sessionVarintPut(&aBuf[1], n);
423 if( n>0 ) memcpy(&aBuf[nVarint + 1], z, n);
426 nByte = 1 + nVarint + n;
427 break;
430 }else{
431 nByte = 1;
432 if( aBuf ) aBuf[0] = '\0';
435 if( pnWrite ) *pnWrite += nByte;
436 return SQLITE_OK;
440 ** Allocate and return a pointer to a buffer nByte bytes in size. If
441 ** pSession is not NULL, increase the sqlite3_session.nMalloc variable
442 ** by the number of bytes allocated.
444 static void *sessionMalloc64(sqlite3_session *pSession, i64 nByte){
445 void *pRet = sqlite3_malloc64(nByte);
446 if( pSession ) pSession->nMalloc += sqlite3_msize(pRet);
447 return pRet;
451 ** Free buffer pFree, which must have been allocated by an earlier
452 ** call to sessionMalloc64(). If pSession is not NULL, decrease the
453 ** sqlite3_session.nMalloc counter by the number of bytes freed.
455 static void sessionFree(sqlite3_session *pSession, void *pFree){
456 if( pSession ) pSession->nMalloc -= sqlite3_msize(pFree);
457 sqlite3_free(pFree);
461 ** This macro is used to calculate hash key values for data structures. In
462 ** order to use this macro, the entire data structure must be represented
463 ** as a series of unsigned integers. In order to calculate a hash-key value
464 ** for a data structure represented as three such integers, the macro may
465 ** then be used as follows:
467 ** int hash_key_value;
468 ** hash_key_value = HASH_APPEND(0, <value 1>);
469 ** hash_key_value = HASH_APPEND(hash_key_value, <value 2>);
470 ** hash_key_value = HASH_APPEND(hash_key_value, <value 3>);
472 ** In practice, the data structures this macro is used for are the primary
473 ** key values of modified rows.
475 #define HASH_APPEND(hash, add) ((hash) << 3) ^ (hash) ^ (unsigned int)(add)
478 ** Append the hash of the 64-bit integer passed as the second argument to the
479 ** hash-key value passed as the first. Return the new hash-key value.
481 static unsigned int sessionHashAppendI64(unsigned int h, i64 i){
482 h = HASH_APPEND(h, i & 0xFFFFFFFF);
483 return HASH_APPEND(h, (i>>32)&0xFFFFFFFF);
487 ** Append the hash of the blob passed via the second and third arguments to
488 ** the hash-key value passed as the first. Return the new hash-key value.
490 static unsigned int sessionHashAppendBlob(unsigned int h, int n, const u8 *z){
491 int i;
492 for(i=0; i<n; i++) h = HASH_APPEND(h, z[i]);
493 return h;
497 ** Append the hash of the data type passed as the second argument to the
498 ** hash-key value passed as the first. Return the new hash-key value.
500 static unsigned int sessionHashAppendType(unsigned int h, int eType){
501 return HASH_APPEND(h, eType);
505 ** This function may only be called from within a pre-update callback.
506 ** It calculates a hash based on the primary key values of the old.* or
507 ** new.* row currently available and, assuming no error occurs, writes it to
508 ** *piHash before returning. If the primary key contains one or more NULL
509 ** values, *pbNullPK is set to true before returning.
511 ** If an error occurs, an SQLite error code is returned and the final values
512 ** of *piHash asn *pbNullPK are undefined. Otherwise, SQLITE_OK is returned
513 ** and the output variables are set as described above.
515 static int sessionPreupdateHash(
516 sqlite3_session *pSession, /* Session object that owns pTab */
517 SessionTable *pTab, /* Session table handle */
518 int bNew, /* True to hash the new.* PK */
519 int *piHash, /* OUT: Hash value */
520 int *pbNullPK /* OUT: True if there are NULL values in PK */
522 unsigned int h = 0; /* Hash value to return */
523 int i; /* Used to iterate through columns */
525 assert( *pbNullPK==0 );
526 assert( pTab->nCol==pSession->hook.xCount(pSession->hook.pCtx) );
527 for(i=0; i<pTab->nCol; i++){
528 if( pTab->abPK[i] ){
529 int rc;
530 int eType;
531 sqlite3_value *pVal;
533 if( bNew ){
534 rc = pSession->hook.xNew(pSession->hook.pCtx, i, &pVal);
535 }else{
536 rc = pSession->hook.xOld(pSession->hook.pCtx, i, &pVal);
538 if( rc!=SQLITE_OK ) return rc;
540 eType = sqlite3_value_type(pVal);
541 h = sessionHashAppendType(h, eType);
542 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
543 i64 iVal;
544 if( eType==SQLITE_INTEGER ){
545 iVal = sqlite3_value_int64(pVal);
546 }else{
547 double rVal = sqlite3_value_double(pVal);
548 assert( sizeof(iVal)==8 && sizeof(rVal)==8 );
549 memcpy(&iVal, &rVal, 8);
551 h = sessionHashAppendI64(h, iVal);
552 }else if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
553 const u8 *z;
554 int n;
555 if( eType==SQLITE_TEXT ){
556 z = (const u8 *)sqlite3_value_text(pVal);
557 }else{
558 z = (const u8 *)sqlite3_value_blob(pVal);
560 n = sqlite3_value_bytes(pVal);
561 if( !z && (eType!=SQLITE_BLOB || n>0) ) return SQLITE_NOMEM;
562 h = sessionHashAppendBlob(h, n, z);
563 }else{
564 assert( eType==SQLITE_NULL );
565 assert( pTab->bStat1==0 || i!=1 );
566 *pbNullPK = 1;
571 *piHash = (h % pTab->nChange);
572 return SQLITE_OK;
576 ** The buffer that the argument points to contains a serialized SQL value.
577 ** Return the number of bytes of space occupied by the value (including
578 ** the type byte).
580 static int sessionSerialLen(u8 *a){
581 int e = *a;
582 int n;
583 if( e==0 || e==0xFF ) return 1;
584 if( e==SQLITE_NULL ) return 1;
585 if( e==SQLITE_INTEGER || e==SQLITE_FLOAT ) return 9;
586 return sessionVarintGet(&a[1], &n) + 1 + n;
590 ** Based on the primary key values stored in change aRecord, calculate a
591 ** hash key. Assume the has table has nBucket buckets. The hash keys
592 ** calculated by this function are compatible with those calculated by
593 ** sessionPreupdateHash().
595 ** The bPkOnly argument is non-zero if the record at aRecord[] is from
596 ** a patchset DELETE. In this case the non-PK fields are omitted entirely.
598 static unsigned int sessionChangeHash(
599 SessionTable *pTab, /* Table handle */
600 int bPkOnly, /* Record consists of PK fields only */
601 u8 *aRecord, /* Change record */
602 int nBucket /* Assume this many buckets in hash table */
604 unsigned int h = 0; /* Value to return */
605 int i; /* Used to iterate through columns */
606 u8 *a = aRecord; /* Used to iterate through change record */
608 for(i=0; i<pTab->nCol; i++){
609 int eType = *a;
610 int isPK = pTab->abPK[i];
611 if( bPkOnly && isPK==0 ) continue;
613 /* It is not possible for eType to be SQLITE_NULL here. The session
614 ** module does not record changes for rows with NULL values stored in
615 ** primary key columns. */
616 assert( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT
617 || eType==SQLITE_TEXT || eType==SQLITE_BLOB
618 || eType==SQLITE_NULL || eType==0
620 assert( !isPK || (eType!=0 && eType!=SQLITE_NULL) );
622 if( isPK ){
623 a++;
624 h = sessionHashAppendType(h, eType);
625 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
626 h = sessionHashAppendI64(h, sessionGetI64(a));
627 a += 8;
628 }else{
629 int n;
630 a += sessionVarintGet(a, &n);
631 h = sessionHashAppendBlob(h, n, a);
632 a += n;
634 }else{
635 a += sessionSerialLen(a);
638 return (h % nBucket);
642 ** Arguments aLeft and aRight are pointers to change records for table pTab.
643 ** This function returns true if the two records apply to the same row (i.e.
644 ** have the same values stored in the primary key columns), or false
645 ** otherwise.
647 static int sessionChangeEqual(
648 SessionTable *pTab, /* Table used for PK definition */
649 int bLeftPkOnly, /* True if aLeft[] contains PK fields only */
650 u8 *aLeft, /* Change record */
651 int bRightPkOnly, /* True if aRight[] contains PK fields only */
652 u8 *aRight /* Change record */
654 u8 *a1 = aLeft; /* Cursor to iterate through aLeft */
655 u8 *a2 = aRight; /* Cursor to iterate through aRight */
656 int iCol; /* Used to iterate through table columns */
658 for(iCol=0; iCol<pTab->nCol; iCol++){
659 if( pTab->abPK[iCol] ){
660 int n1 = sessionSerialLen(a1);
661 int n2 = sessionSerialLen(a2);
663 if( n1!=n2 || memcmp(a1, a2, n1) ){
664 return 0;
666 a1 += n1;
667 a2 += n2;
668 }else{
669 if( bLeftPkOnly==0 ) a1 += sessionSerialLen(a1);
670 if( bRightPkOnly==0 ) a2 += sessionSerialLen(a2);
674 return 1;
678 ** Arguments aLeft and aRight both point to buffers containing change
679 ** records with nCol columns. This function "merges" the two records into
680 ** a single records which is written to the buffer at *paOut. *paOut is
681 ** then set to point to one byte after the last byte written before
682 ** returning.
684 ** The merging of records is done as follows: For each column, if the
685 ** aRight record contains a value for the column, copy the value from
686 ** their. Otherwise, if aLeft contains a value, copy it. If neither
687 ** record contains a value for a given column, then neither does the
688 ** output record.
690 static void sessionMergeRecord(
691 u8 **paOut,
692 int nCol,
693 u8 *aLeft,
694 u8 *aRight
696 u8 *a1 = aLeft; /* Cursor used to iterate through aLeft */
697 u8 *a2 = aRight; /* Cursor used to iterate through aRight */
698 u8 *aOut = *paOut; /* Output cursor */
699 int iCol; /* Used to iterate from 0 to nCol */
701 for(iCol=0; iCol<nCol; iCol++){
702 int n1 = sessionSerialLen(a1);
703 int n2 = sessionSerialLen(a2);
704 if( *a2 ){
705 memcpy(aOut, a2, n2);
706 aOut += n2;
707 }else{
708 memcpy(aOut, a1, n1);
709 aOut += n1;
711 a1 += n1;
712 a2 += n2;
715 *paOut = aOut;
719 ** This is a helper function used by sessionMergeUpdate().
721 ** When this function is called, both *paOne and *paTwo point to a value
722 ** within a change record. Before it returns, both have been advanced so
723 ** as to point to the next value in the record.
725 ** If, when this function is called, *paTwo points to a valid value (i.e.
726 ** *paTwo[0] is not 0x00 - the "no value" placeholder), a copy of the *paTwo
727 ** pointer is returned and *pnVal is set to the number of bytes in the
728 ** serialized value. Otherwise, a copy of *paOne is returned and *pnVal
729 ** set to the number of bytes in the value at *paOne. If *paOne points
730 ** to the "no value" placeholder, *pnVal is set to 1. In other words:
732 ** if( *paTwo is valid ) return *paTwo;
733 ** return *paOne;
736 static u8 *sessionMergeValue(
737 u8 **paOne, /* IN/OUT: Left-hand buffer pointer */
738 u8 **paTwo, /* IN/OUT: Right-hand buffer pointer */
739 int *pnVal /* OUT: Bytes in returned value */
741 u8 *a1 = *paOne;
742 u8 *a2 = *paTwo;
743 u8 *pRet = 0;
744 int n1;
746 assert( a1 );
747 if( a2 ){
748 int n2 = sessionSerialLen(a2);
749 if( *a2 ){
750 *pnVal = n2;
751 pRet = a2;
753 *paTwo = &a2[n2];
756 n1 = sessionSerialLen(a1);
757 if( pRet==0 ){
758 *pnVal = n1;
759 pRet = a1;
761 *paOne = &a1[n1];
763 return pRet;
767 ** This function is used by changeset_concat() to merge two UPDATE changes
768 ** on the same row.
770 static int sessionMergeUpdate(
771 u8 **paOut, /* IN/OUT: Pointer to output buffer */
772 SessionTable *pTab, /* Table change pertains to */
773 int bPatchset, /* True if records are patchset records */
774 u8 *aOldRecord1, /* old.* record for first change */
775 u8 *aOldRecord2, /* old.* record for second change */
776 u8 *aNewRecord1, /* new.* record for first change */
777 u8 *aNewRecord2 /* new.* record for second change */
779 u8 *aOld1 = aOldRecord1;
780 u8 *aOld2 = aOldRecord2;
781 u8 *aNew1 = aNewRecord1;
782 u8 *aNew2 = aNewRecord2;
784 u8 *aOut = *paOut;
785 int i;
787 if( bPatchset==0 ){
788 int bRequired = 0;
790 assert( aOldRecord1 && aNewRecord1 );
792 /* Write the old.* vector first. */
793 for(i=0; i<pTab->nCol; i++){
794 int nOld;
795 u8 *aOld;
796 int nNew;
797 u8 *aNew;
799 aOld = sessionMergeValue(&aOld1, &aOld2, &nOld);
800 aNew = sessionMergeValue(&aNew1, &aNew2, &nNew);
801 if( pTab->abPK[i] || nOld!=nNew || memcmp(aOld, aNew, nNew) ){
802 if( pTab->abPK[i]==0 ) bRequired = 1;
803 memcpy(aOut, aOld, nOld);
804 aOut += nOld;
805 }else{
806 *(aOut++) = '\0';
810 if( !bRequired ) return 0;
813 /* Write the new.* vector */
814 aOld1 = aOldRecord1;
815 aOld2 = aOldRecord2;
816 aNew1 = aNewRecord1;
817 aNew2 = aNewRecord2;
818 for(i=0; i<pTab->nCol; i++){
819 int nOld;
820 u8 *aOld;
821 int nNew;
822 u8 *aNew;
824 aOld = sessionMergeValue(&aOld1, &aOld2, &nOld);
825 aNew = sessionMergeValue(&aNew1, &aNew2, &nNew);
826 if( bPatchset==0
827 && (pTab->abPK[i] || (nOld==nNew && 0==memcmp(aOld, aNew, nNew)))
829 *(aOut++) = '\0';
830 }else{
831 memcpy(aOut, aNew, nNew);
832 aOut += nNew;
836 *paOut = aOut;
837 return 1;
841 ** This function is only called from within a pre-update-hook callback.
842 ** It determines if the current pre-update-hook change affects the same row
843 ** as the change stored in argument pChange. If so, it returns true. Otherwise
844 ** if the pre-update-hook does not affect the same row as pChange, it returns
845 ** false.
847 static int sessionPreupdateEqual(
848 sqlite3_session *pSession, /* Session object that owns SessionTable */
849 SessionTable *pTab, /* Table associated with change */
850 SessionChange *pChange, /* Change to compare to */
851 int op /* Current pre-update operation */
853 int iCol; /* Used to iterate through columns */
854 u8 *a = pChange->aRecord; /* Cursor used to scan change record */
856 assert( op==SQLITE_INSERT || op==SQLITE_UPDATE || op==SQLITE_DELETE );
857 for(iCol=0; iCol<pTab->nCol; iCol++){
858 if( !pTab->abPK[iCol] ){
859 a += sessionSerialLen(a);
860 }else{
861 sqlite3_value *pVal; /* Value returned by preupdate_new/old */
862 int rc; /* Error code from preupdate_new/old */
863 int eType = *a++; /* Type of value from change record */
865 /* The following calls to preupdate_new() and preupdate_old() can not
866 ** fail. This is because they cache their return values, and by the
867 ** time control flows to here they have already been called once from
868 ** within sessionPreupdateHash(). The first two asserts below verify
869 ** this (that the method has already been called). */
870 if( op==SQLITE_INSERT ){
871 /* assert( db->pPreUpdate->pNewUnpacked || db->pPreUpdate->aNew ); */
872 rc = pSession->hook.xNew(pSession->hook.pCtx, iCol, &pVal);
873 }else{
874 /* assert( db->pPreUpdate->pUnpacked ); */
875 rc = pSession->hook.xOld(pSession->hook.pCtx, iCol, &pVal);
877 assert( rc==SQLITE_OK );
878 if( sqlite3_value_type(pVal)!=eType ) return 0;
880 /* A SessionChange object never has a NULL value in a PK column */
881 assert( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT
882 || eType==SQLITE_BLOB || eType==SQLITE_TEXT
885 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
886 i64 iVal = sessionGetI64(a);
887 a += 8;
888 if( eType==SQLITE_INTEGER ){
889 if( sqlite3_value_int64(pVal)!=iVal ) return 0;
890 }else{
891 double rVal;
892 assert( sizeof(iVal)==8 && sizeof(rVal)==8 );
893 memcpy(&rVal, &iVal, 8);
894 if( sqlite3_value_double(pVal)!=rVal ) return 0;
896 }else{
897 int n;
898 const u8 *z;
899 a += sessionVarintGet(a, &n);
900 if( sqlite3_value_bytes(pVal)!=n ) return 0;
901 if( eType==SQLITE_TEXT ){
902 z = sqlite3_value_text(pVal);
903 }else{
904 z = sqlite3_value_blob(pVal);
906 if( n>0 && memcmp(a, z, n) ) return 0;
907 a += n;
912 return 1;
916 ** If required, grow the hash table used to store changes on table pTab
917 ** (part of the session pSession). If a fatal OOM error occurs, set the
918 ** session object to failed and return SQLITE_ERROR. Otherwise, return
919 ** SQLITE_OK.
921 ** It is possible that a non-fatal OOM error occurs in this function. In
922 ** that case the hash-table does not grow, but SQLITE_OK is returned anyway.
923 ** Growing the hash table in this case is a performance optimization only,
924 ** it is not required for correct operation.
926 static int sessionGrowHash(
927 sqlite3_session *pSession, /* For memory accounting. May be NULL */
928 int bPatchset,
929 SessionTable *pTab
931 if( pTab->nChange==0 || pTab->nEntry>=(pTab->nChange/2) ){
932 int i;
933 SessionChange **apNew;
934 sqlite3_int64 nNew = 2*(sqlite3_int64)(pTab->nChange ? pTab->nChange : 128);
936 apNew = (SessionChange**)sessionMalloc64(
937 pSession, sizeof(SessionChange*) * nNew
939 if( apNew==0 ){
940 if( pTab->nChange==0 ){
941 return SQLITE_ERROR;
943 return SQLITE_OK;
945 memset(apNew, 0, sizeof(SessionChange *) * nNew);
947 for(i=0; i<pTab->nChange; i++){
948 SessionChange *p;
949 SessionChange *pNext;
950 for(p=pTab->apChange[i]; p; p=pNext){
951 int bPkOnly = (p->op==SQLITE_DELETE && bPatchset);
952 int iHash = sessionChangeHash(pTab, bPkOnly, p->aRecord, nNew);
953 pNext = p->pNext;
954 p->pNext = apNew[iHash];
955 apNew[iHash] = p;
959 sessionFree(pSession, pTab->apChange);
960 pTab->nChange = nNew;
961 pTab->apChange = apNew;
964 return SQLITE_OK;
968 ** This function queries the database for the names of the columns of table
969 ** zThis, in schema zDb.
971 ** Otherwise, if they are not NULL, variable *pnCol is set to the number
972 ** of columns in the database table and variable *pzTab is set to point to a
973 ** nul-terminated copy of the table name. *pazCol (if not NULL) is set to
974 ** point to an array of pointers to column names. And *pabPK (again, if not
975 ** NULL) is set to point to an array of booleans - true if the corresponding
976 ** column is part of the primary key.
978 ** For example, if the table is declared as:
980 ** CREATE TABLE tbl1(w, x, y, z, PRIMARY KEY(w, z));
982 ** Then the four output variables are populated as follows:
984 ** *pnCol = 4
985 ** *pzTab = "tbl1"
986 ** *pazCol = {"w", "x", "y", "z"}
987 ** *pabPK = {1, 0, 0, 1}
989 ** All returned buffers are part of the same single allocation, which must
990 ** be freed using sqlite3_free() by the caller
992 static int sessionTableInfo(
993 sqlite3_session *pSession, /* For memory accounting. May be NULL */
994 sqlite3 *db, /* Database connection */
995 const char *zDb, /* Name of attached database (e.g. "main") */
996 const char *zThis, /* Table name */
997 int *pnCol, /* OUT: number of columns */
998 const char **pzTab, /* OUT: Copy of zThis */
999 const char ***pazCol, /* OUT: Array of column names for table */
1000 u8 **pabPK /* OUT: Array of booleans - true for PK col */
1002 char *zPragma;
1003 sqlite3_stmt *pStmt;
1004 int rc;
1005 sqlite3_int64 nByte;
1006 int nDbCol = 0;
1007 int nThis;
1008 int i;
1009 u8 *pAlloc = 0;
1010 char **azCol = 0;
1011 u8 *abPK = 0;
1013 assert( pazCol && pabPK );
1015 nThis = sqlite3Strlen30(zThis);
1016 if( nThis==12 && 0==sqlite3_stricmp("sqlite_stat1", zThis) ){
1017 rc = sqlite3_table_column_metadata(db, zDb, zThis, 0, 0, 0, 0, 0, 0);
1018 if( rc==SQLITE_OK ){
1019 /* For sqlite_stat1, pretend that (tbl,idx) is the PRIMARY KEY. */
1020 zPragma = sqlite3_mprintf(
1021 "SELECT 0, 'tbl', '', 0, '', 1 UNION ALL "
1022 "SELECT 1, 'idx', '', 0, '', 2 UNION ALL "
1023 "SELECT 2, 'stat', '', 0, '', 0"
1025 }else if( rc==SQLITE_ERROR ){
1026 zPragma = sqlite3_mprintf("");
1027 }else{
1028 *pazCol = 0;
1029 *pabPK = 0;
1030 *pnCol = 0;
1031 if( pzTab ) *pzTab = 0;
1032 return rc;
1034 }else{
1035 zPragma = sqlite3_mprintf("PRAGMA '%q'.table_info('%q')", zDb, zThis);
1037 if( !zPragma ){
1038 *pazCol = 0;
1039 *pabPK = 0;
1040 *pnCol = 0;
1041 if( pzTab ) *pzTab = 0;
1042 return SQLITE_NOMEM;
1045 rc = sqlite3_prepare_v2(db, zPragma, -1, &pStmt, 0);
1046 sqlite3_free(zPragma);
1047 if( rc!=SQLITE_OK ){
1048 *pazCol = 0;
1049 *pabPK = 0;
1050 *pnCol = 0;
1051 if( pzTab ) *pzTab = 0;
1052 return rc;
1055 nByte = nThis + 1;
1056 while( SQLITE_ROW==sqlite3_step(pStmt) ){
1057 nByte += sqlite3_column_bytes(pStmt, 1);
1058 nDbCol++;
1060 rc = sqlite3_reset(pStmt);
1062 if( rc==SQLITE_OK ){
1063 nByte += nDbCol * (sizeof(const char *) + sizeof(u8) + 1);
1064 pAlloc = sessionMalloc64(pSession, nByte);
1065 if( pAlloc==0 ){
1066 rc = SQLITE_NOMEM;
1069 if( rc==SQLITE_OK ){
1070 azCol = (char **)pAlloc;
1071 pAlloc = (u8 *)&azCol[nDbCol];
1072 abPK = (u8 *)pAlloc;
1073 pAlloc = &abPK[nDbCol];
1074 if( pzTab ){
1075 memcpy(pAlloc, zThis, nThis+1);
1076 *pzTab = (char *)pAlloc;
1077 pAlloc += nThis+1;
1080 i = 0;
1081 while( SQLITE_ROW==sqlite3_step(pStmt) ){
1082 int nName = sqlite3_column_bytes(pStmt, 1);
1083 const unsigned char *zName = sqlite3_column_text(pStmt, 1);
1084 if( zName==0 ) break;
1085 memcpy(pAlloc, zName, nName+1);
1086 azCol[i] = (char *)pAlloc;
1087 pAlloc += nName+1;
1088 abPK[i] = sqlite3_column_int(pStmt, 5);
1089 i++;
1091 rc = sqlite3_reset(pStmt);
1095 /* If successful, populate the output variables. Otherwise, zero them and
1096 ** free any allocation made. An error code will be returned in this case.
1098 if( rc==SQLITE_OK ){
1099 *pazCol = (const char **)azCol;
1100 *pabPK = abPK;
1101 *pnCol = nDbCol;
1102 }else{
1103 *pazCol = 0;
1104 *pabPK = 0;
1105 *pnCol = 0;
1106 if( pzTab ) *pzTab = 0;
1107 sessionFree(pSession, azCol);
1109 sqlite3_finalize(pStmt);
1110 return rc;
1114 ** This function is only called from within a pre-update handler for a
1115 ** write to table pTab, part of session pSession. If this is the first
1116 ** write to this table, initalize the SessionTable.nCol, azCol[] and
1117 ** abPK[] arrays accordingly.
1119 ** If an error occurs, an error code is stored in sqlite3_session.rc and
1120 ** non-zero returned. Or, if no error occurs but the table has no primary
1121 ** key, sqlite3_session.rc is left set to SQLITE_OK and non-zero returned to
1122 ** indicate that updates on this table should be ignored. SessionTable.abPK
1123 ** is set to NULL in this case.
1125 static int sessionInitTable(sqlite3_session *pSession, SessionTable *pTab){
1126 if( pTab->nCol==0 ){
1127 u8 *abPK;
1128 assert( pTab->azCol==0 || pTab->abPK==0 );
1129 pSession->rc = sessionTableInfo(pSession, pSession->db, pSession->zDb,
1130 pTab->zName, &pTab->nCol, 0, &pTab->azCol, &abPK
1132 if( pSession->rc==SQLITE_OK ){
1133 int i;
1134 for(i=0; i<pTab->nCol; i++){
1135 if( abPK[i] ){
1136 pTab->abPK = abPK;
1137 break;
1140 if( 0==sqlite3_stricmp("sqlite_stat1", pTab->zName) ){
1141 pTab->bStat1 = 1;
1144 if( pSession->bEnableSize ){
1145 pSession->nMaxChangesetSize += (
1146 1 + sessionVarintLen(pTab->nCol) + pTab->nCol + strlen(pTab->zName)+1
1151 return (pSession->rc || pTab->abPK==0);
1155 ** Versions of the four methods in object SessionHook for use with the
1156 ** sqlite_stat1 table. The purpose of this is to substitute a zero-length
1157 ** blob each time a NULL value is read from the "idx" column of the
1158 ** sqlite_stat1 table.
1160 typedef struct SessionStat1Ctx SessionStat1Ctx;
1161 struct SessionStat1Ctx {
1162 SessionHook hook;
1163 sqlite3_session *pSession;
1165 static int sessionStat1Old(void *pCtx, int iCol, sqlite3_value **ppVal){
1166 SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
1167 sqlite3_value *pVal = 0;
1168 int rc = p->hook.xOld(p->hook.pCtx, iCol, &pVal);
1169 if( rc==SQLITE_OK && iCol==1 && sqlite3_value_type(pVal)==SQLITE_NULL ){
1170 pVal = p->pSession->pZeroBlob;
1172 *ppVal = pVal;
1173 return rc;
1175 static int sessionStat1New(void *pCtx, int iCol, sqlite3_value **ppVal){
1176 SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
1177 sqlite3_value *pVal = 0;
1178 int rc = p->hook.xNew(p->hook.pCtx, iCol, &pVal);
1179 if( rc==SQLITE_OK && iCol==1 && sqlite3_value_type(pVal)==SQLITE_NULL ){
1180 pVal = p->pSession->pZeroBlob;
1182 *ppVal = pVal;
1183 return rc;
1185 static int sessionStat1Count(void *pCtx){
1186 SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
1187 return p->hook.xCount(p->hook.pCtx);
1189 static int sessionStat1Depth(void *pCtx){
1190 SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
1191 return p->hook.xDepth(p->hook.pCtx);
1194 static int sessionUpdateMaxSize(
1195 int op,
1196 sqlite3_session *pSession, /* Session object pTab is attached to */
1197 SessionTable *pTab, /* Table that change applies to */
1198 SessionChange *pC /* Update pC->nMaxSize */
1200 i64 nNew = 2;
1201 if( pC->op==SQLITE_INSERT ){
1202 if( op!=SQLITE_DELETE ){
1203 int ii;
1204 for(ii=0; ii<pTab->nCol; ii++){
1205 sqlite3_value *p = 0;
1206 pSession->hook.xNew(pSession->hook.pCtx, ii, &p);
1207 sessionSerializeValue(0, p, &nNew);
1210 }else if( op==SQLITE_DELETE ){
1211 nNew += pC->nRecord;
1212 if( sqlite3_preupdate_blobwrite(pSession->db)>=0 ){
1213 nNew += pC->nRecord;
1215 }else{
1216 int ii;
1217 u8 *pCsr = pC->aRecord;
1218 for(ii=0; ii<pTab->nCol; ii++){
1219 int bChanged = 1;
1220 int nOld = 0;
1221 int eType;
1222 sqlite3_value *p = 0;
1223 pSession->hook.xNew(pSession->hook.pCtx, ii, &p);
1224 if( p==0 ){
1225 return SQLITE_NOMEM;
1228 eType = *pCsr++;
1229 switch( eType ){
1230 case SQLITE_NULL:
1231 bChanged = sqlite3_value_type(p)!=SQLITE_NULL;
1232 break;
1234 case SQLITE_FLOAT:
1235 case SQLITE_INTEGER: {
1236 if( eType==sqlite3_value_type(p) ){
1237 sqlite3_int64 iVal = sessionGetI64(pCsr);
1238 if( eType==SQLITE_INTEGER ){
1239 bChanged = (iVal!=sqlite3_value_int64(p));
1240 }else{
1241 double dVal;
1242 memcpy(&dVal, &iVal, 8);
1243 bChanged = (dVal!=sqlite3_value_double(p));
1246 nOld = 8;
1247 pCsr += 8;
1248 break;
1251 default: {
1252 int nByte;
1253 nOld = sessionVarintGet(pCsr, &nByte);
1254 pCsr += nOld;
1255 nOld += nByte;
1256 assert( eType==SQLITE_TEXT || eType==SQLITE_BLOB );
1257 if( eType==sqlite3_value_type(p)
1258 && nByte==sqlite3_value_bytes(p)
1259 && (nByte==0 || 0==memcmp(pCsr, sqlite3_value_blob(p), nByte))
1261 bChanged = 0;
1263 pCsr += nByte;
1264 break;
1268 if( bChanged && pTab->abPK[ii] ){
1269 nNew = pC->nRecord + 2;
1270 break;
1273 if( bChanged ){
1274 nNew += 1 + nOld;
1275 sessionSerializeValue(0, p, &nNew);
1276 }else if( pTab->abPK[ii] ){
1277 nNew += 2 + nOld;
1278 }else{
1279 nNew += 2;
1284 if( nNew>pC->nMaxSize ){
1285 int nIncr = nNew - pC->nMaxSize;
1286 pC->nMaxSize = nNew;
1287 pSession->nMaxChangesetSize += nIncr;
1289 return SQLITE_OK;
1293 ** This function is only called from with a pre-update-hook reporting a
1294 ** change on table pTab (attached to session pSession). The type of change
1295 ** (UPDATE, INSERT, DELETE) is specified by the first argument.
1297 ** Unless one is already present or an error occurs, an entry is added
1298 ** to the changed-rows hash table associated with table pTab.
1300 static void sessionPreupdateOneChange(
1301 int op, /* One of SQLITE_UPDATE, INSERT, DELETE */
1302 sqlite3_session *pSession, /* Session object pTab is attached to */
1303 SessionTable *pTab /* Table that change applies to */
1305 int iHash;
1306 int bNull = 0;
1307 int rc = SQLITE_OK;
1308 SessionStat1Ctx stat1 = {{0,0,0,0,0},0};
1310 if( pSession->rc ) return;
1312 /* Load table details if required */
1313 if( sessionInitTable(pSession, pTab) ) return;
1315 /* Check the number of columns in this xPreUpdate call matches the
1316 ** number of columns in the table. */
1317 if( pTab->nCol!=pSession->hook.xCount(pSession->hook.pCtx) ){
1318 pSession->rc = SQLITE_SCHEMA;
1319 return;
1322 /* Grow the hash table if required */
1323 if( sessionGrowHash(pSession, 0, pTab) ){
1324 pSession->rc = SQLITE_NOMEM;
1325 return;
1328 if( pTab->bStat1 ){
1329 stat1.hook = pSession->hook;
1330 stat1.pSession = pSession;
1331 pSession->hook.pCtx = (void*)&stat1;
1332 pSession->hook.xNew = sessionStat1New;
1333 pSession->hook.xOld = sessionStat1Old;
1334 pSession->hook.xCount = sessionStat1Count;
1335 pSession->hook.xDepth = sessionStat1Depth;
1336 if( pSession->pZeroBlob==0 ){
1337 sqlite3_value *p = sqlite3ValueNew(0);
1338 if( p==0 ){
1339 rc = SQLITE_NOMEM;
1340 goto error_out;
1342 sqlite3ValueSetStr(p, 0, "", 0, SQLITE_STATIC);
1343 pSession->pZeroBlob = p;
1347 /* Calculate the hash-key for this change. If the primary key of the row
1348 ** includes a NULL value, exit early. Such changes are ignored by the
1349 ** session module. */
1350 rc = sessionPreupdateHash(pSession, pTab, op==SQLITE_INSERT, &iHash, &bNull);
1351 if( rc!=SQLITE_OK ) goto error_out;
1353 if( bNull==0 ){
1354 /* Search the hash table for an existing record for this row. */
1355 SessionChange *pC;
1356 for(pC=pTab->apChange[iHash]; pC; pC=pC->pNext){
1357 if( sessionPreupdateEqual(pSession, pTab, pC, op) ) break;
1360 if( pC==0 ){
1361 /* Create a new change object containing all the old values (if
1362 ** this is an SQLITE_UPDATE or SQLITE_DELETE), or just the PK
1363 ** values (if this is an INSERT). */
1364 sqlite3_int64 nByte; /* Number of bytes to allocate */
1365 int i; /* Used to iterate through columns */
1367 assert( rc==SQLITE_OK );
1368 pTab->nEntry++;
1370 /* Figure out how large an allocation is required */
1371 nByte = sizeof(SessionChange);
1372 for(i=0; i<pTab->nCol; i++){
1373 sqlite3_value *p = 0;
1374 if( op!=SQLITE_INSERT ){
1375 TESTONLY(int trc = ) pSession->hook.xOld(pSession->hook.pCtx, i, &p);
1376 assert( trc==SQLITE_OK );
1377 }else if( pTab->abPK[i] ){
1378 TESTONLY(int trc = ) pSession->hook.xNew(pSession->hook.pCtx, i, &p);
1379 assert( trc==SQLITE_OK );
1382 /* This may fail if SQLite value p contains a utf-16 string that must
1383 ** be converted to utf-8 and an OOM error occurs while doing so. */
1384 rc = sessionSerializeValue(0, p, &nByte);
1385 if( rc!=SQLITE_OK ) goto error_out;
1388 /* Allocate the change object */
1389 pC = (SessionChange *)sessionMalloc64(pSession, nByte);
1390 if( !pC ){
1391 rc = SQLITE_NOMEM;
1392 goto error_out;
1393 }else{
1394 memset(pC, 0, sizeof(SessionChange));
1395 pC->aRecord = (u8 *)&pC[1];
1398 /* Populate the change object. None of the preupdate_old(),
1399 ** preupdate_new() or SerializeValue() calls below may fail as all
1400 ** required values and encodings have already been cached in memory.
1401 ** It is not possible for an OOM to occur in this block. */
1402 nByte = 0;
1403 for(i=0; i<pTab->nCol; i++){
1404 sqlite3_value *p = 0;
1405 if( op!=SQLITE_INSERT ){
1406 pSession->hook.xOld(pSession->hook.pCtx, i, &p);
1407 }else if( pTab->abPK[i] ){
1408 pSession->hook.xNew(pSession->hook.pCtx, i, &p);
1410 sessionSerializeValue(&pC->aRecord[nByte], p, &nByte);
1413 /* Add the change to the hash-table */
1414 if( pSession->bIndirect || pSession->hook.xDepth(pSession->hook.pCtx) ){
1415 pC->bIndirect = 1;
1417 pC->nRecord = nByte;
1418 pC->op = op;
1419 pC->pNext = pTab->apChange[iHash];
1420 pTab->apChange[iHash] = pC;
1422 }else if( pC->bIndirect ){
1423 /* If the existing change is considered "indirect", but this current
1424 ** change is "direct", mark the change object as direct. */
1425 if( pSession->hook.xDepth(pSession->hook.pCtx)==0
1426 && pSession->bIndirect==0
1428 pC->bIndirect = 0;
1432 assert( rc==SQLITE_OK );
1433 if( pSession->bEnableSize ){
1434 rc = sessionUpdateMaxSize(op, pSession, pTab, pC);
1439 /* If an error has occurred, mark the session object as failed. */
1440 error_out:
1441 if( pTab->bStat1 ){
1442 pSession->hook = stat1.hook;
1444 if( rc!=SQLITE_OK ){
1445 pSession->rc = rc;
1449 static int sessionFindTable(
1450 sqlite3_session *pSession,
1451 const char *zName,
1452 SessionTable **ppTab
1454 int rc = SQLITE_OK;
1455 int nName = sqlite3Strlen30(zName);
1456 SessionTable *pRet;
1458 /* Search for an existing table */
1459 for(pRet=pSession->pTable; pRet; pRet=pRet->pNext){
1460 if( 0==sqlite3_strnicmp(pRet->zName, zName, nName+1) ) break;
1463 if( pRet==0 && pSession->bAutoAttach ){
1464 /* If there is a table-filter configured, invoke it. If it returns 0,
1465 ** do not automatically add the new table. */
1466 if( pSession->xTableFilter==0
1467 || pSession->xTableFilter(pSession->pFilterCtx, zName)
1469 rc = sqlite3session_attach(pSession, zName);
1470 if( rc==SQLITE_OK ){
1471 pRet = pSession->pTable;
1472 while( ALWAYS(pRet) && pRet->pNext ){
1473 pRet = pRet->pNext;
1475 assert( pRet!=0 );
1476 assert( 0==sqlite3_strnicmp(pRet->zName, zName, nName+1) );
1481 assert( rc==SQLITE_OK || pRet==0 );
1482 *ppTab = pRet;
1483 return rc;
1487 ** The 'pre-update' hook registered by this module with SQLite databases.
1489 static void xPreUpdate(
1490 void *pCtx, /* Copy of third arg to preupdate_hook() */
1491 sqlite3 *db, /* Database handle */
1492 int op, /* SQLITE_UPDATE, DELETE or INSERT */
1493 char const *zDb, /* Database name */
1494 char const *zName, /* Table name */
1495 sqlite3_int64 iKey1, /* Rowid of row about to be deleted/updated */
1496 sqlite3_int64 iKey2 /* New rowid value (for a rowid UPDATE) */
1498 sqlite3_session *pSession;
1499 int nDb = sqlite3Strlen30(zDb);
1501 assert( sqlite3_mutex_held(db->mutex) );
1502 (void)iKey1;
1503 (void)iKey2;
1505 for(pSession=(sqlite3_session *)pCtx; pSession; pSession=pSession->pNext){
1506 SessionTable *pTab;
1508 /* If this session is attached to a different database ("main", "temp"
1509 ** etc.), or if it is not currently enabled, there is nothing to do. Skip
1510 ** to the next session object attached to this database. */
1511 if( pSession->bEnable==0 ) continue;
1512 if( pSession->rc ) continue;
1513 if( sqlite3_strnicmp(zDb, pSession->zDb, nDb+1) ) continue;
1515 pSession->rc = sessionFindTable(pSession, zName, &pTab);
1516 if( pTab ){
1517 assert( pSession->rc==SQLITE_OK );
1518 sessionPreupdateOneChange(op, pSession, pTab);
1519 if( op==SQLITE_UPDATE ){
1520 sessionPreupdateOneChange(SQLITE_INSERT, pSession, pTab);
1527 ** The pre-update hook implementations.
1529 static int sessionPreupdateOld(void *pCtx, int iVal, sqlite3_value **ppVal){
1530 return sqlite3_preupdate_old((sqlite3*)pCtx, iVal, ppVal);
1532 static int sessionPreupdateNew(void *pCtx, int iVal, sqlite3_value **ppVal){
1533 return sqlite3_preupdate_new((sqlite3*)pCtx, iVal, ppVal);
1535 static int sessionPreupdateCount(void *pCtx){
1536 return sqlite3_preupdate_count((sqlite3*)pCtx);
1538 static int sessionPreupdateDepth(void *pCtx){
1539 return sqlite3_preupdate_depth((sqlite3*)pCtx);
1543 ** Install the pre-update hooks on the session object passed as the only
1544 ** argument.
1546 static void sessionPreupdateHooks(
1547 sqlite3_session *pSession
1549 pSession->hook.pCtx = (void*)pSession->db;
1550 pSession->hook.xOld = sessionPreupdateOld;
1551 pSession->hook.xNew = sessionPreupdateNew;
1552 pSession->hook.xCount = sessionPreupdateCount;
1553 pSession->hook.xDepth = sessionPreupdateDepth;
1556 typedef struct SessionDiffCtx SessionDiffCtx;
1557 struct SessionDiffCtx {
1558 sqlite3_stmt *pStmt;
1559 int nOldOff;
1563 ** The diff hook implementations.
1565 static int sessionDiffOld(void *pCtx, int iVal, sqlite3_value **ppVal){
1566 SessionDiffCtx *p = (SessionDiffCtx*)pCtx;
1567 *ppVal = sqlite3_column_value(p->pStmt, iVal+p->nOldOff);
1568 return SQLITE_OK;
1570 static int sessionDiffNew(void *pCtx, int iVal, sqlite3_value **ppVal){
1571 SessionDiffCtx *p = (SessionDiffCtx*)pCtx;
1572 *ppVal = sqlite3_column_value(p->pStmt, iVal);
1573 return SQLITE_OK;
1575 static int sessionDiffCount(void *pCtx){
1576 SessionDiffCtx *p = (SessionDiffCtx*)pCtx;
1577 return p->nOldOff ? p->nOldOff : sqlite3_column_count(p->pStmt);
1579 static int sessionDiffDepth(void *pCtx){
1580 (void)pCtx;
1581 return 0;
1585 ** Install the diff hooks on the session object passed as the only
1586 ** argument.
1588 static void sessionDiffHooks(
1589 sqlite3_session *pSession,
1590 SessionDiffCtx *pDiffCtx
1592 pSession->hook.pCtx = (void*)pDiffCtx;
1593 pSession->hook.xOld = sessionDiffOld;
1594 pSession->hook.xNew = sessionDiffNew;
1595 pSession->hook.xCount = sessionDiffCount;
1596 pSession->hook.xDepth = sessionDiffDepth;
1599 static char *sessionExprComparePK(
1600 int nCol,
1601 const char *zDb1, const char *zDb2,
1602 const char *zTab,
1603 const char **azCol, u8 *abPK
1605 int i;
1606 const char *zSep = "";
1607 char *zRet = 0;
1609 for(i=0; i<nCol; i++){
1610 if( abPK[i] ){
1611 zRet = sqlite3_mprintf("%z%s\"%w\".\"%w\".\"%w\"=\"%w\".\"%w\".\"%w\"",
1612 zRet, zSep, zDb1, zTab, azCol[i], zDb2, zTab, azCol[i]
1614 zSep = " AND ";
1615 if( zRet==0 ) break;
1619 return zRet;
1622 static char *sessionExprCompareOther(
1623 int nCol,
1624 const char *zDb1, const char *zDb2,
1625 const char *zTab,
1626 const char **azCol, u8 *abPK
1628 int i;
1629 const char *zSep = "";
1630 char *zRet = 0;
1631 int bHave = 0;
1633 for(i=0; i<nCol; i++){
1634 if( abPK[i]==0 ){
1635 bHave = 1;
1636 zRet = sqlite3_mprintf(
1637 "%z%s\"%w\".\"%w\".\"%w\" IS NOT \"%w\".\"%w\".\"%w\"",
1638 zRet, zSep, zDb1, zTab, azCol[i], zDb2, zTab, azCol[i]
1640 zSep = " OR ";
1641 if( zRet==0 ) break;
1645 if( bHave==0 ){
1646 assert( zRet==0 );
1647 zRet = sqlite3_mprintf("0");
1650 return zRet;
1653 static char *sessionSelectFindNew(
1654 const char *zDb1, /* Pick rows in this db only */
1655 const char *zDb2, /* But not in this one */
1656 const char *zTbl, /* Table name */
1657 const char *zExpr
1659 char *zRet = sqlite3_mprintf(
1660 "SELECT * FROM \"%w\".\"%w\" WHERE NOT EXISTS ("
1661 " SELECT 1 FROM \"%w\".\"%w\" WHERE %s"
1662 ")",
1663 zDb1, zTbl, zDb2, zTbl, zExpr
1665 return zRet;
1668 static int sessionDiffFindNew(
1669 int op,
1670 sqlite3_session *pSession,
1671 SessionTable *pTab,
1672 const char *zDb1,
1673 const char *zDb2,
1674 char *zExpr
1676 int rc = SQLITE_OK;
1677 char *zStmt = sessionSelectFindNew(zDb1, zDb2, pTab->zName,zExpr);
1679 if( zStmt==0 ){
1680 rc = SQLITE_NOMEM;
1681 }else{
1682 sqlite3_stmt *pStmt;
1683 rc = sqlite3_prepare(pSession->db, zStmt, -1, &pStmt, 0);
1684 if( rc==SQLITE_OK ){
1685 SessionDiffCtx *pDiffCtx = (SessionDiffCtx*)pSession->hook.pCtx;
1686 pDiffCtx->pStmt = pStmt;
1687 pDiffCtx->nOldOff = 0;
1688 while( SQLITE_ROW==sqlite3_step(pStmt) ){
1689 sessionPreupdateOneChange(op, pSession, pTab);
1691 rc = sqlite3_finalize(pStmt);
1693 sqlite3_free(zStmt);
1696 return rc;
1699 static int sessionDiffFindModified(
1700 sqlite3_session *pSession,
1701 SessionTable *pTab,
1702 const char *zFrom,
1703 const char *zExpr
1705 int rc = SQLITE_OK;
1707 char *zExpr2 = sessionExprCompareOther(pTab->nCol,
1708 pSession->zDb, zFrom, pTab->zName, pTab->azCol, pTab->abPK
1710 if( zExpr2==0 ){
1711 rc = SQLITE_NOMEM;
1712 }else{
1713 char *zStmt = sqlite3_mprintf(
1714 "SELECT * FROM \"%w\".\"%w\", \"%w\".\"%w\" WHERE %s AND (%z)",
1715 pSession->zDb, pTab->zName, zFrom, pTab->zName, zExpr, zExpr2
1717 if( zStmt==0 ){
1718 rc = SQLITE_NOMEM;
1719 }else{
1720 sqlite3_stmt *pStmt;
1721 rc = sqlite3_prepare(pSession->db, zStmt, -1, &pStmt, 0);
1723 if( rc==SQLITE_OK ){
1724 SessionDiffCtx *pDiffCtx = (SessionDiffCtx*)pSession->hook.pCtx;
1725 pDiffCtx->pStmt = pStmt;
1726 pDiffCtx->nOldOff = pTab->nCol;
1727 while( SQLITE_ROW==sqlite3_step(pStmt) ){
1728 sessionPreupdateOneChange(SQLITE_UPDATE, pSession, pTab);
1730 rc = sqlite3_finalize(pStmt);
1732 sqlite3_free(zStmt);
1736 return rc;
1739 int sqlite3session_diff(
1740 sqlite3_session *pSession,
1741 const char *zFrom,
1742 const char *zTbl,
1743 char **pzErrMsg
1745 const char *zDb = pSession->zDb;
1746 int rc = pSession->rc;
1747 SessionDiffCtx d;
1749 memset(&d, 0, sizeof(d));
1750 sessionDiffHooks(pSession, &d);
1752 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
1753 if( pzErrMsg ) *pzErrMsg = 0;
1754 if( rc==SQLITE_OK ){
1755 char *zExpr = 0;
1756 sqlite3 *db = pSession->db;
1757 SessionTable *pTo; /* Table zTbl */
1759 /* Locate and if necessary initialize the target table object */
1760 rc = sessionFindTable(pSession, zTbl, &pTo);
1761 if( pTo==0 ) goto diff_out;
1762 if( sessionInitTable(pSession, pTo) ){
1763 rc = pSession->rc;
1764 goto diff_out;
1767 /* Check the table schemas match */
1768 if( rc==SQLITE_OK ){
1769 int bHasPk = 0;
1770 int bMismatch = 0;
1771 int nCol; /* Columns in zFrom.zTbl */
1772 u8 *abPK;
1773 const char **azCol = 0;
1774 rc = sessionTableInfo(0, db, zFrom, zTbl, &nCol, 0, &azCol, &abPK);
1775 if( rc==SQLITE_OK ){
1776 if( pTo->nCol!=nCol ){
1777 bMismatch = 1;
1778 }else{
1779 int i;
1780 for(i=0; i<nCol; i++){
1781 if( pTo->abPK[i]!=abPK[i] ) bMismatch = 1;
1782 if( sqlite3_stricmp(azCol[i], pTo->azCol[i]) ) bMismatch = 1;
1783 if( abPK[i] ) bHasPk = 1;
1787 sqlite3_free((char*)azCol);
1788 if( bMismatch ){
1789 if( pzErrMsg ){
1790 *pzErrMsg = sqlite3_mprintf("table schemas do not match");
1792 rc = SQLITE_SCHEMA;
1794 if( bHasPk==0 ){
1795 /* Ignore tables with no primary keys */
1796 goto diff_out;
1800 if( rc==SQLITE_OK ){
1801 zExpr = sessionExprComparePK(pTo->nCol,
1802 zDb, zFrom, pTo->zName, pTo->azCol, pTo->abPK
1806 /* Find new rows */
1807 if( rc==SQLITE_OK ){
1808 rc = sessionDiffFindNew(SQLITE_INSERT, pSession, pTo, zDb, zFrom, zExpr);
1811 /* Find old rows */
1812 if( rc==SQLITE_OK ){
1813 rc = sessionDiffFindNew(SQLITE_DELETE, pSession, pTo, zFrom, zDb, zExpr);
1816 /* Find modified rows */
1817 if( rc==SQLITE_OK ){
1818 rc = sessionDiffFindModified(pSession, pTo, zFrom, zExpr);
1821 sqlite3_free(zExpr);
1824 diff_out:
1825 sessionPreupdateHooks(pSession);
1826 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
1827 return rc;
1831 ** Create a session object. This session object will record changes to
1832 ** database zDb attached to connection db.
1834 int sqlite3session_create(
1835 sqlite3 *db, /* Database handle */
1836 const char *zDb, /* Name of db (e.g. "main") */
1837 sqlite3_session **ppSession /* OUT: New session object */
1839 sqlite3_session *pNew; /* Newly allocated session object */
1840 sqlite3_session *pOld; /* Session object already attached to db */
1841 int nDb = sqlite3Strlen30(zDb); /* Length of zDb in bytes */
1843 /* Zero the output value in case an error occurs. */
1844 *ppSession = 0;
1846 /* Allocate and populate the new session object. */
1847 pNew = (sqlite3_session *)sqlite3_malloc64(sizeof(sqlite3_session) + nDb + 1);
1848 if( !pNew ) return SQLITE_NOMEM;
1849 memset(pNew, 0, sizeof(sqlite3_session));
1850 pNew->db = db;
1851 pNew->zDb = (char *)&pNew[1];
1852 pNew->bEnable = 1;
1853 memcpy(pNew->zDb, zDb, nDb+1);
1854 sessionPreupdateHooks(pNew);
1856 /* Add the new session object to the linked list of session objects
1857 ** attached to database handle $db. Do this under the cover of the db
1858 ** handle mutex. */
1859 sqlite3_mutex_enter(sqlite3_db_mutex(db));
1860 pOld = (sqlite3_session*)sqlite3_preupdate_hook(db, xPreUpdate, (void*)pNew);
1861 pNew->pNext = pOld;
1862 sqlite3_mutex_leave(sqlite3_db_mutex(db));
1864 *ppSession = pNew;
1865 return SQLITE_OK;
1869 ** Free the list of table objects passed as the first argument. The contents
1870 ** of the changed-rows hash tables are also deleted.
1872 static void sessionDeleteTable(sqlite3_session *pSession, SessionTable *pList){
1873 SessionTable *pNext;
1874 SessionTable *pTab;
1876 for(pTab=pList; pTab; pTab=pNext){
1877 int i;
1878 pNext = pTab->pNext;
1879 for(i=0; i<pTab->nChange; i++){
1880 SessionChange *p;
1881 SessionChange *pNextChange;
1882 for(p=pTab->apChange[i]; p; p=pNextChange){
1883 pNextChange = p->pNext;
1884 sessionFree(pSession, p);
1887 sessionFree(pSession, (char*)pTab->azCol); /* cast works around VC++ bug */
1888 sessionFree(pSession, pTab->apChange);
1889 sessionFree(pSession, pTab);
1894 ** Delete a session object previously allocated using sqlite3session_create().
1896 void sqlite3session_delete(sqlite3_session *pSession){
1897 sqlite3 *db = pSession->db;
1898 sqlite3_session *pHead;
1899 sqlite3_session **pp;
1901 /* Unlink the session from the linked list of sessions attached to the
1902 ** database handle. Hold the db mutex while doing so. */
1903 sqlite3_mutex_enter(sqlite3_db_mutex(db));
1904 pHead = (sqlite3_session*)sqlite3_preupdate_hook(db, 0, 0);
1905 for(pp=&pHead; ALWAYS((*pp)!=0); pp=&((*pp)->pNext)){
1906 if( (*pp)==pSession ){
1907 *pp = (*pp)->pNext;
1908 if( pHead ) sqlite3_preupdate_hook(db, xPreUpdate, (void*)pHead);
1909 break;
1912 sqlite3_mutex_leave(sqlite3_db_mutex(db));
1913 sqlite3ValueFree(pSession->pZeroBlob);
1915 /* Delete all attached table objects. And the contents of their
1916 ** associated hash-tables. */
1917 sessionDeleteTable(pSession, pSession->pTable);
1919 /* Assert that all allocations have been freed and then free the
1920 ** session object itself. */
1921 assert( pSession->nMalloc==0 );
1922 sqlite3_free(pSession);
1926 ** Set a table filter on a Session Object.
1928 void sqlite3session_table_filter(
1929 sqlite3_session *pSession,
1930 int(*xFilter)(void*, const char*),
1931 void *pCtx /* First argument passed to xFilter */
1933 pSession->bAutoAttach = 1;
1934 pSession->pFilterCtx = pCtx;
1935 pSession->xTableFilter = xFilter;
1939 ** Attach a table to a session. All subsequent changes made to the table
1940 ** while the session object is enabled will be recorded.
1942 ** Only tables that have a PRIMARY KEY defined may be attached. It does
1943 ** not matter if the PRIMARY KEY is an "INTEGER PRIMARY KEY" (rowid alias)
1944 ** or not.
1946 int sqlite3session_attach(
1947 sqlite3_session *pSession, /* Session object */
1948 const char *zName /* Table name */
1950 int rc = SQLITE_OK;
1951 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
1953 if( !zName ){
1954 pSession->bAutoAttach = 1;
1955 }else{
1956 SessionTable *pTab; /* New table object (if required) */
1957 int nName; /* Number of bytes in string zName */
1959 /* First search for an existing entry. If one is found, this call is
1960 ** a no-op. Return early. */
1961 nName = sqlite3Strlen30(zName);
1962 for(pTab=pSession->pTable; pTab; pTab=pTab->pNext){
1963 if( 0==sqlite3_strnicmp(pTab->zName, zName, nName+1) ) break;
1966 if( !pTab ){
1967 /* Allocate new SessionTable object. */
1968 int nByte = sizeof(SessionTable) + nName + 1;
1969 pTab = (SessionTable*)sessionMalloc64(pSession, nByte);
1970 if( !pTab ){
1971 rc = SQLITE_NOMEM;
1972 }else{
1973 /* Populate the new SessionTable object and link it into the list.
1974 ** The new object must be linked onto the end of the list, not
1975 ** simply added to the start of it in order to ensure that tables
1976 ** appear in the correct order when a changeset or patchset is
1977 ** eventually generated. */
1978 SessionTable **ppTab;
1979 memset(pTab, 0, sizeof(SessionTable));
1980 pTab->zName = (char *)&pTab[1];
1981 memcpy(pTab->zName, zName, nName+1);
1982 for(ppTab=&pSession->pTable; *ppTab; ppTab=&(*ppTab)->pNext);
1983 *ppTab = pTab;
1988 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
1989 return rc;
1993 ** Ensure that there is room in the buffer to append nByte bytes of data.
1994 ** If not, use sqlite3_realloc() to grow the buffer so that there is.
1996 ** If successful, return zero. Otherwise, if an OOM condition is encountered,
1997 ** set *pRc to SQLITE_NOMEM and return non-zero.
1999 static int sessionBufferGrow(SessionBuffer *p, i64 nByte, int *pRc){
2000 #define SESSION_MAX_BUFFER_SZ (0x7FFFFF00 - 1)
2001 i64 nReq = p->nBuf + nByte;
2002 if( *pRc==SQLITE_OK && nReq>p->nAlloc ){
2003 u8 *aNew;
2004 i64 nNew = p->nAlloc ? p->nAlloc : 128;
2006 do {
2007 nNew = nNew*2;
2008 }while( nNew<nReq );
2010 /* The value of SESSION_MAX_BUFFER_SZ is copied from the implementation
2011 ** of sqlite3_realloc64(). Allocations greater than this size in bytes
2012 ** always fail. It is used here to ensure that this routine can always
2013 ** allocate up to this limit - instead of up to the largest power of
2014 ** two smaller than the limit. */
2015 if( nNew>SESSION_MAX_BUFFER_SZ ){
2016 nNew = SESSION_MAX_BUFFER_SZ;
2017 if( nNew<nReq ){
2018 *pRc = SQLITE_NOMEM;
2019 return 1;
2023 aNew = (u8 *)sqlite3_realloc64(p->aBuf, nNew);
2024 if( 0==aNew ){
2025 *pRc = SQLITE_NOMEM;
2026 }else{
2027 p->aBuf = aNew;
2028 p->nAlloc = nNew;
2031 return (*pRc!=SQLITE_OK);
2035 ** Append the value passed as the second argument to the buffer passed
2036 ** as the first.
2038 ** This function is a no-op if *pRc is non-zero when it is called.
2039 ** Otherwise, if an error occurs, *pRc is set to an SQLite error code
2040 ** before returning.
2042 static void sessionAppendValue(SessionBuffer *p, sqlite3_value *pVal, int *pRc){
2043 int rc = *pRc;
2044 if( rc==SQLITE_OK ){
2045 sqlite3_int64 nByte = 0;
2046 rc = sessionSerializeValue(0, pVal, &nByte);
2047 sessionBufferGrow(p, nByte, &rc);
2048 if( rc==SQLITE_OK ){
2049 rc = sessionSerializeValue(&p->aBuf[p->nBuf], pVal, 0);
2050 p->nBuf += nByte;
2051 }else{
2052 *pRc = rc;
2058 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
2059 ** called. Otherwise, append a single byte to the buffer.
2061 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
2062 ** returning.
2064 static void sessionAppendByte(SessionBuffer *p, u8 v, int *pRc){
2065 if( 0==sessionBufferGrow(p, 1, pRc) ){
2066 p->aBuf[p->nBuf++] = v;
2071 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
2072 ** called. Otherwise, append a single varint to the buffer.
2074 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
2075 ** returning.
2077 static void sessionAppendVarint(SessionBuffer *p, int v, int *pRc){
2078 if( 0==sessionBufferGrow(p, 9, pRc) ){
2079 p->nBuf += sessionVarintPut(&p->aBuf[p->nBuf], v);
2084 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
2085 ** called. Otherwise, append a blob of data to the buffer.
2087 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
2088 ** returning.
2090 static void sessionAppendBlob(
2091 SessionBuffer *p,
2092 const u8 *aBlob,
2093 int nBlob,
2094 int *pRc
2096 if( nBlob>0 && 0==sessionBufferGrow(p, nBlob, pRc) ){
2097 memcpy(&p->aBuf[p->nBuf], aBlob, nBlob);
2098 p->nBuf += nBlob;
2103 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
2104 ** called. Otherwise, append a string to the buffer. All bytes in the string
2105 ** up to (but not including) the nul-terminator are written to the buffer.
2107 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
2108 ** returning.
2110 static void sessionAppendStr(
2111 SessionBuffer *p,
2112 const char *zStr,
2113 int *pRc
2115 int nStr = sqlite3Strlen30(zStr);
2116 if( 0==sessionBufferGrow(p, nStr, pRc) ){
2117 memcpy(&p->aBuf[p->nBuf], zStr, nStr);
2118 p->nBuf += nStr;
2123 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
2124 ** called. Otherwise, append the string representation of integer iVal
2125 ** to the buffer. No nul-terminator is written.
2127 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
2128 ** returning.
2130 static void sessionAppendInteger(
2131 SessionBuffer *p, /* Buffer to append to */
2132 int iVal, /* Value to write the string rep. of */
2133 int *pRc /* IN/OUT: Error code */
2135 char aBuf[24];
2136 sqlite3_snprintf(sizeof(aBuf)-1, aBuf, "%d", iVal);
2137 sessionAppendStr(p, aBuf, pRc);
2141 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
2142 ** called. Otherwise, append the string zStr enclosed in quotes (") and
2143 ** with any embedded quote characters escaped to the buffer. No
2144 ** nul-terminator byte is written.
2146 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
2147 ** returning.
2149 static void sessionAppendIdent(
2150 SessionBuffer *p, /* Buffer to a append to */
2151 const char *zStr, /* String to quote, escape and append */
2152 int *pRc /* IN/OUT: Error code */
2154 int nStr = sqlite3Strlen30(zStr)*2 + 2 + 1;
2155 if( 0==sessionBufferGrow(p, nStr, pRc) ){
2156 char *zOut = (char *)&p->aBuf[p->nBuf];
2157 const char *zIn = zStr;
2158 *zOut++ = '"';
2159 while( *zIn ){
2160 if( *zIn=='"' ) *zOut++ = '"';
2161 *zOut++ = *(zIn++);
2163 *zOut++ = '"';
2164 p->nBuf = (int)((u8 *)zOut - p->aBuf);
2169 ** This function is a no-op if *pRc is other than SQLITE_OK when it is
2170 ** called. Otherwse, it appends the serialized version of the value stored
2171 ** in column iCol of the row that SQL statement pStmt currently points
2172 ** to to the buffer.
2174 static void sessionAppendCol(
2175 SessionBuffer *p, /* Buffer to append to */
2176 sqlite3_stmt *pStmt, /* Handle pointing to row containing value */
2177 int iCol, /* Column to read value from */
2178 int *pRc /* IN/OUT: Error code */
2180 if( *pRc==SQLITE_OK ){
2181 int eType = sqlite3_column_type(pStmt, iCol);
2182 sessionAppendByte(p, (u8)eType, pRc);
2183 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
2184 sqlite3_int64 i;
2185 u8 aBuf[8];
2186 if( eType==SQLITE_INTEGER ){
2187 i = sqlite3_column_int64(pStmt, iCol);
2188 }else{
2189 double r = sqlite3_column_double(pStmt, iCol);
2190 memcpy(&i, &r, 8);
2192 sessionPutI64(aBuf, i);
2193 sessionAppendBlob(p, aBuf, 8, pRc);
2195 if( eType==SQLITE_BLOB || eType==SQLITE_TEXT ){
2196 u8 *z;
2197 int nByte;
2198 if( eType==SQLITE_BLOB ){
2199 z = (u8 *)sqlite3_column_blob(pStmt, iCol);
2200 }else{
2201 z = (u8 *)sqlite3_column_text(pStmt, iCol);
2203 nByte = sqlite3_column_bytes(pStmt, iCol);
2204 if( z || (eType==SQLITE_BLOB && nByte==0) ){
2205 sessionAppendVarint(p, nByte, pRc);
2206 sessionAppendBlob(p, z, nByte, pRc);
2207 }else{
2208 *pRc = SQLITE_NOMEM;
2216 ** This function appends an update change to the buffer (see the comments
2217 ** under "CHANGESET FORMAT" at the top of the file). An update change
2218 ** consists of:
2220 ** 1 byte: SQLITE_UPDATE (0x17)
2221 ** n bytes: old.* record (see RECORD FORMAT)
2222 ** m bytes: new.* record (see RECORD FORMAT)
2224 ** The SessionChange object passed as the third argument contains the
2225 ** values that were stored in the row when the session began (the old.*
2226 ** values). The statement handle passed as the second argument points
2227 ** at the current version of the row (the new.* values).
2229 ** If all of the old.* values are equal to their corresponding new.* value
2230 ** (i.e. nothing has changed), then no data at all is appended to the buffer.
2232 ** Otherwise, the old.* record contains all primary key values and the
2233 ** original values of any fields that have been modified. The new.* record
2234 ** contains the new values of only those fields that have been modified.
2236 static int sessionAppendUpdate(
2237 SessionBuffer *pBuf, /* Buffer to append to */
2238 int bPatchset, /* True for "patchset", 0 for "changeset" */
2239 sqlite3_stmt *pStmt, /* Statement handle pointing at new row */
2240 SessionChange *p, /* Object containing old values */
2241 u8 *abPK /* Boolean array - true for PK columns */
2243 int rc = SQLITE_OK;
2244 SessionBuffer buf2 = {0,0,0}; /* Buffer to accumulate new.* record in */
2245 int bNoop = 1; /* Set to zero if any values are modified */
2246 int nRewind = pBuf->nBuf; /* Set to zero if any values are modified */
2247 int i; /* Used to iterate through columns */
2248 u8 *pCsr = p->aRecord; /* Used to iterate through old.* values */
2250 assert( abPK!=0 );
2251 sessionAppendByte(pBuf, SQLITE_UPDATE, &rc);
2252 sessionAppendByte(pBuf, p->bIndirect, &rc);
2253 for(i=0; i<sqlite3_column_count(pStmt); i++){
2254 int bChanged = 0;
2255 int nAdvance;
2256 int eType = *pCsr;
2257 switch( eType ){
2258 case SQLITE_NULL:
2259 nAdvance = 1;
2260 if( sqlite3_column_type(pStmt, i)!=SQLITE_NULL ){
2261 bChanged = 1;
2263 break;
2265 case SQLITE_FLOAT:
2266 case SQLITE_INTEGER: {
2267 nAdvance = 9;
2268 if( eType==sqlite3_column_type(pStmt, i) ){
2269 sqlite3_int64 iVal = sessionGetI64(&pCsr[1]);
2270 if( eType==SQLITE_INTEGER ){
2271 if( iVal==sqlite3_column_int64(pStmt, i) ) break;
2272 }else{
2273 double dVal;
2274 memcpy(&dVal, &iVal, 8);
2275 if( dVal==sqlite3_column_double(pStmt, i) ) break;
2278 bChanged = 1;
2279 break;
2282 default: {
2283 int n;
2284 int nHdr = 1 + sessionVarintGet(&pCsr[1], &n);
2285 assert( eType==SQLITE_TEXT || eType==SQLITE_BLOB );
2286 nAdvance = nHdr + n;
2287 if( eType==sqlite3_column_type(pStmt, i)
2288 && n==sqlite3_column_bytes(pStmt, i)
2289 && (n==0 || 0==memcmp(&pCsr[nHdr], sqlite3_column_blob(pStmt, i), n))
2291 break;
2293 bChanged = 1;
2297 /* If at least one field has been modified, this is not a no-op. */
2298 if( bChanged ) bNoop = 0;
2300 /* Add a field to the old.* record. This is omitted if this modules is
2301 ** currently generating a patchset. */
2302 if( bPatchset==0 ){
2303 if( bChanged || abPK[i] ){
2304 sessionAppendBlob(pBuf, pCsr, nAdvance, &rc);
2305 }else{
2306 sessionAppendByte(pBuf, 0, &rc);
2310 /* Add a field to the new.* record. Or the only record if currently
2311 ** generating a patchset. */
2312 if( bChanged || (bPatchset && abPK[i]) ){
2313 sessionAppendCol(&buf2, pStmt, i, &rc);
2314 }else{
2315 sessionAppendByte(&buf2, 0, &rc);
2318 pCsr += nAdvance;
2321 if( bNoop ){
2322 pBuf->nBuf = nRewind;
2323 }else{
2324 sessionAppendBlob(pBuf, buf2.aBuf, buf2.nBuf, &rc);
2326 sqlite3_free(buf2.aBuf);
2328 return rc;
2332 ** Append a DELETE change to the buffer passed as the first argument. Use
2333 ** the changeset format if argument bPatchset is zero, or the patchset
2334 ** format otherwise.
2336 static int sessionAppendDelete(
2337 SessionBuffer *pBuf, /* Buffer to append to */
2338 int bPatchset, /* True for "patchset", 0 for "changeset" */
2339 SessionChange *p, /* Object containing old values */
2340 int nCol, /* Number of columns in table */
2341 u8 *abPK /* Boolean array - true for PK columns */
2343 int rc = SQLITE_OK;
2345 sessionAppendByte(pBuf, SQLITE_DELETE, &rc);
2346 sessionAppendByte(pBuf, p->bIndirect, &rc);
2348 if( bPatchset==0 ){
2349 sessionAppendBlob(pBuf, p->aRecord, p->nRecord, &rc);
2350 }else{
2351 int i;
2352 u8 *a = p->aRecord;
2353 for(i=0; i<nCol; i++){
2354 u8 *pStart = a;
2355 int eType = *a++;
2357 switch( eType ){
2358 case 0:
2359 case SQLITE_NULL:
2360 assert( abPK[i]==0 );
2361 break;
2363 case SQLITE_FLOAT:
2364 case SQLITE_INTEGER:
2365 a += 8;
2366 break;
2368 default: {
2369 int n;
2370 a += sessionVarintGet(a, &n);
2371 a += n;
2372 break;
2375 if( abPK[i] ){
2376 sessionAppendBlob(pBuf, pStart, (int)(a-pStart), &rc);
2379 assert( (a - p->aRecord)==p->nRecord );
2382 return rc;
2386 ** Formulate and prepare a SELECT statement to retrieve a row from table
2387 ** zTab in database zDb based on its primary key. i.e.
2389 ** SELECT * FROM zDb.zTab WHERE pk1 = ? AND pk2 = ? AND ...
2391 static int sessionSelectStmt(
2392 sqlite3 *db, /* Database handle */
2393 const char *zDb, /* Database name */
2394 const char *zTab, /* Table name */
2395 int nCol, /* Number of columns in table */
2396 const char **azCol, /* Names of table columns */
2397 u8 *abPK, /* PRIMARY KEY array */
2398 sqlite3_stmt **ppStmt /* OUT: Prepared SELECT statement */
2400 int rc = SQLITE_OK;
2401 char *zSql = 0;
2402 int nSql = -1;
2404 if( 0==sqlite3_stricmp("sqlite_stat1", zTab) ){
2405 zSql = sqlite3_mprintf(
2406 "SELECT tbl, ?2, stat FROM %Q.sqlite_stat1 WHERE tbl IS ?1 AND "
2407 "idx IS (CASE WHEN ?2=X'' THEN NULL ELSE ?2 END)", zDb
2409 if( zSql==0 ) rc = SQLITE_NOMEM;
2410 }else{
2411 int i;
2412 const char *zSep = "";
2413 SessionBuffer buf = {0, 0, 0};
2415 sessionAppendStr(&buf, "SELECT * FROM ", &rc);
2416 sessionAppendIdent(&buf, zDb, &rc);
2417 sessionAppendStr(&buf, ".", &rc);
2418 sessionAppendIdent(&buf, zTab, &rc);
2419 sessionAppendStr(&buf, " WHERE ", &rc);
2420 for(i=0; i<nCol; i++){
2421 if( abPK[i] ){
2422 sessionAppendStr(&buf, zSep, &rc);
2423 sessionAppendIdent(&buf, azCol[i], &rc);
2424 sessionAppendStr(&buf, " IS ?", &rc);
2425 sessionAppendInteger(&buf, i+1, &rc);
2426 zSep = " AND ";
2429 zSql = (char*)buf.aBuf;
2430 nSql = buf.nBuf;
2433 if( rc==SQLITE_OK ){
2434 rc = sqlite3_prepare_v2(db, zSql, nSql, ppStmt, 0);
2436 sqlite3_free(zSql);
2437 return rc;
2441 ** Bind the PRIMARY KEY values from the change passed in argument pChange
2442 ** to the SELECT statement passed as the first argument. The SELECT statement
2443 ** is as prepared by function sessionSelectStmt().
2445 ** Return SQLITE_OK if all PK values are successfully bound, or an SQLite
2446 ** error code (e.g. SQLITE_NOMEM) otherwise.
2448 static int sessionSelectBind(
2449 sqlite3_stmt *pSelect, /* SELECT from sessionSelectStmt() */
2450 int nCol, /* Number of columns in table */
2451 u8 *abPK, /* PRIMARY KEY array */
2452 SessionChange *pChange /* Change structure */
2454 int i;
2455 int rc = SQLITE_OK;
2456 u8 *a = pChange->aRecord;
2458 for(i=0; i<nCol && rc==SQLITE_OK; i++){
2459 int eType = *a++;
2461 switch( eType ){
2462 case 0:
2463 case SQLITE_NULL:
2464 assert( abPK[i]==0 );
2465 break;
2467 case SQLITE_INTEGER: {
2468 if( abPK[i] ){
2469 i64 iVal = sessionGetI64(a);
2470 rc = sqlite3_bind_int64(pSelect, i+1, iVal);
2472 a += 8;
2473 break;
2476 case SQLITE_FLOAT: {
2477 if( abPK[i] ){
2478 double rVal;
2479 i64 iVal = sessionGetI64(a);
2480 memcpy(&rVal, &iVal, 8);
2481 rc = sqlite3_bind_double(pSelect, i+1, rVal);
2483 a += 8;
2484 break;
2487 case SQLITE_TEXT: {
2488 int n;
2489 a += sessionVarintGet(a, &n);
2490 if( abPK[i] ){
2491 rc = sqlite3_bind_text(pSelect, i+1, (char *)a, n, SQLITE_TRANSIENT);
2493 a += n;
2494 break;
2497 default: {
2498 int n;
2499 assert( eType==SQLITE_BLOB );
2500 a += sessionVarintGet(a, &n);
2501 if( abPK[i] ){
2502 rc = sqlite3_bind_blob(pSelect, i+1, a, n, SQLITE_TRANSIENT);
2504 a += n;
2505 break;
2510 return rc;
2514 ** This function is a no-op if *pRc is set to other than SQLITE_OK when it
2515 ** is called. Otherwise, append a serialized table header (part of the binary
2516 ** changeset format) to buffer *pBuf. If an error occurs, set *pRc to an
2517 ** SQLite error code before returning.
2519 static void sessionAppendTableHdr(
2520 SessionBuffer *pBuf, /* Append header to this buffer */
2521 int bPatchset, /* Use the patchset format if true */
2522 SessionTable *pTab, /* Table object to append header for */
2523 int *pRc /* IN/OUT: Error code */
2525 /* Write a table header */
2526 sessionAppendByte(pBuf, (bPatchset ? 'P' : 'T'), pRc);
2527 sessionAppendVarint(pBuf, pTab->nCol, pRc);
2528 sessionAppendBlob(pBuf, pTab->abPK, pTab->nCol, pRc);
2529 sessionAppendBlob(pBuf, (u8 *)pTab->zName, (int)strlen(pTab->zName)+1, pRc);
2533 ** Generate either a changeset (if argument bPatchset is zero) or a patchset
2534 ** (if it is non-zero) based on the current contents of the session object
2535 ** passed as the first argument.
2537 ** If no error occurs, SQLITE_OK is returned and the new changeset/patchset
2538 ** stored in output variables *pnChangeset and *ppChangeset. Or, if an error
2539 ** occurs, an SQLite error code is returned and both output variables set
2540 ** to 0.
2542 static int sessionGenerateChangeset(
2543 sqlite3_session *pSession, /* Session object */
2544 int bPatchset, /* True for patchset, false for changeset */
2545 int (*xOutput)(void *pOut, const void *pData, int nData),
2546 void *pOut, /* First argument for xOutput */
2547 int *pnChangeset, /* OUT: Size of buffer at *ppChangeset */
2548 void **ppChangeset /* OUT: Buffer containing changeset */
2550 sqlite3 *db = pSession->db; /* Source database handle */
2551 SessionTable *pTab; /* Used to iterate through attached tables */
2552 SessionBuffer buf = {0,0,0}; /* Buffer in which to accumlate changeset */
2553 int rc; /* Return code */
2555 assert( xOutput==0 || (pnChangeset==0 && ppChangeset==0) );
2556 assert( xOutput!=0 || (pnChangeset!=0 && ppChangeset!=0) );
2558 /* Zero the output variables in case an error occurs. If this session
2559 ** object is already in the error state (sqlite3_session.rc != SQLITE_OK),
2560 ** this call will be a no-op. */
2561 if( xOutput==0 ){
2562 assert( pnChangeset!=0 && ppChangeset!=0 );
2563 *pnChangeset = 0;
2564 *ppChangeset = 0;
2567 if( pSession->rc ) return pSession->rc;
2568 rc = sqlite3_exec(pSession->db, "SAVEPOINT changeset", 0, 0, 0);
2569 if( rc!=SQLITE_OK ) return rc;
2571 sqlite3_mutex_enter(sqlite3_db_mutex(db));
2573 for(pTab=pSession->pTable; rc==SQLITE_OK && pTab; pTab=pTab->pNext){
2574 if( pTab->nEntry ){
2575 const char *zName = pTab->zName;
2576 int nCol = 0; /* Number of columns in table */
2577 u8 *abPK = 0; /* Primary key array */
2578 const char **azCol = 0; /* Table columns */
2579 int i; /* Used to iterate through hash buckets */
2580 sqlite3_stmt *pSel = 0; /* SELECT statement to query table pTab */
2581 int nRewind = buf.nBuf; /* Initial size of write buffer */
2582 int nNoop; /* Size of buffer after writing tbl header */
2584 /* Check the table schema is still Ok. */
2585 rc = sessionTableInfo(0, db, pSession->zDb, zName, &nCol, 0,&azCol,&abPK);
2586 if( !rc && (pTab->nCol!=nCol || memcmp(abPK, pTab->abPK, nCol)) ){
2587 rc = SQLITE_SCHEMA;
2590 /* Write a table header */
2591 sessionAppendTableHdr(&buf, bPatchset, pTab, &rc);
2593 /* Build and compile a statement to execute: */
2594 if( rc==SQLITE_OK ){
2595 rc = sessionSelectStmt(
2596 db, pSession->zDb, zName, nCol, azCol, abPK, &pSel);
2599 nNoop = buf.nBuf;
2600 for(i=0; i<pTab->nChange && rc==SQLITE_OK; i++){
2601 SessionChange *p; /* Used to iterate through changes */
2603 for(p=pTab->apChange[i]; rc==SQLITE_OK && p; p=p->pNext){
2604 rc = sessionSelectBind(pSel, nCol, abPK, p);
2605 if( rc!=SQLITE_OK ) continue;
2606 if( sqlite3_step(pSel)==SQLITE_ROW ){
2607 if( p->op==SQLITE_INSERT ){
2608 int iCol;
2609 sessionAppendByte(&buf, SQLITE_INSERT, &rc);
2610 sessionAppendByte(&buf, p->bIndirect, &rc);
2611 for(iCol=0; iCol<nCol; iCol++){
2612 sessionAppendCol(&buf, pSel, iCol, &rc);
2614 }else{
2615 assert( abPK!=0 ); /* Because sessionSelectStmt() returned ok */
2616 rc = sessionAppendUpdate(&buf, bPatchset, pSel, p, abPK);
2618 }else if( p->op!=SQLITE_INSERT ){
2619 rc = sessionAppendDelete(&buf, bPatchset, p, nCol, abPK);
2621 if( rc==SQLITE_OK ){
2622 rc = sqlite3_reset(pSel);
2625 /* If the buffer is now larger than sessions_strm_chunk_size, pass
2626 ** its contents to the xOutput() callback. */
2627 if( xOutput
2628 && rc==SQLITE_OK
2629 && buf.nBuf>nNoop
2630 && buf.nBuf>sessions_strm_chunk_size
2632 rc = xOutput(pOut, (void*)buf.aBuf, buf.nBuf);
2633 nNoop = -1;
2634 buf.nBuf = 0;
2640 sqlite3_finalize(pSel);
2641 if( buf.nBuf==nNoop ){
2642 buf.nBuf = nRewind;
2644 sqlite3_free((char*)azCol); /* cast works around VC++ bug */
2648 if( rc==SQLITE_OK ){
2649 if( xOutput==0 ){
2650 *pnChangeset = buf.nBuf;
2651 *ppChangeset = buf.aBuf;
2652 buf.aBuf = 0;
2653 }else if( buf.nBuf>0 ){
2654 rc = xOutput(pOut, (void*)buf.aBuf, buf.nBuf);
2658 sqlite3_free(buf.aBuf);
2659 sqlite3_exec(db, "RELEASE changeset", 0, 0, 0);
2660 sqlite3_mutex_leave(sqlite3_db_mutex(db));
2661 return rc;
2665 ** Obtain a changeset object containing all changes recorded by the
2666 ** session object passed as the first argument.
2668 ** It is the responsibility of the caller to eventually free the buffer
2669 ** using sqlite3_free().
2671 int sqlite3session_changeset(
2672 sqlite3_session *pSession, /* Session object */
2673 int *pnChangeset, /* OUT: Size of buffer at *ppChangeset */
2674 void **ppChangeset /* OUT: Buffer containing changeset */
2676 int rc;
2678 if( pnChangeset==0 || ppChangeset==0 ) return SQLITE_MISUSE;
2679 rc = sessionGenerateChangeset(pSession, 0, 0, 0, pnChangeset,ppChangeset);
2680 assert( rc || pnChangeset==0
2681 || pSession->bEnableSize==0 || *pnChangeset<=pSession->nMaxChangesetSize
2683 return rc;
2687 ** Streaming version of sqlite3session_changeset().
2689 int sqlite3session_changeset_strm(
2690 sqlite3_session *pSession,
2691 int (*xOutput)(void *pOut, const void *pData, int nData),
2692 void *pOut
2694 if( xOutput==0 ) return SQLITE_MISUSE;
2695 return sessionGenerateChangeset(pSession, 0, xOutput, pOut, 0, 0);
2699 ** Streaming version of sqlite3session_patchset().
2701 int sqlite3session_patchset_strm(
2702 sqlite3_session *pSession,
2703 int (*xOutput)(void *pOut, const void *pData, int nData),
2704 void *pOut
2706 if( xOutput==0 ) return SQLITE_MISUSE;
2707 return sessionGenerateChangeset(pSession, 1, xOutput, pOut, 0, 0);
2711 ** Obtain a patchset object containing all changes recorded by the
2712 ** session object passed as the first argument.
2714 ** It is the responsibility of the caller to eventually free the buffer
2715 ** using sqlite3_free().
2717 int sqlite3session_patchset(
2718 sqlite3_session *pSession, /* Session object */
2719 int *pnPatchset, /* OUT: Size of buffer at *ppChangeset */
2720 void **ppPatchset /* OUT: Buffer containing changeset */
2722 if( pnPatchset==0 || ppPatchset==0 ) return SQLITE_MISUSE;
2723 return sessionGenerateChangeset(pSession, 1, 0, 0, pnPatchset, ppPatchset);
2727 ** Enable or disable the session object passed as the first argument.
2729 int sqlite3session_enable(sqlite3_session *pSession, int bEnable){
2730 int ret;
2731 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
2732 if( bEnable>=0 ){
2733 pSession->bEnable = bEnable;
2735 ret = pSession->bEnable;
2736 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
2737 return ret;
2741 ** Enable or disable the session object passed as the first argument.
2743 int sqlite3session_indirect(sqlite3_session *pSession, int bIndirect){
2744 int ret;
2745 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
2746 if( bIndirect>=0 ){
2747 pSession->bIndirect = bIndirect;
2749 ret = pSession->bIndirect;
2750 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
2751 return ret;
2755 ** Return true if there have been no changes to monitored tables recorded
2756 ** by the session object passed as the only argument.
2758 int sqlite3session_isempty(sqlite3_session *pSession){
2759 int ret = 0;
2760 SessionTable *pTab;
2762 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
2763 for(pTab=pSession->pTable; pTab && ret==0; pTab=pTab->pNext){
2764 ret = (pTab->nEntry>0);
2766 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
2768 return (ret==0);
2772 ** Return the amount of heap memory in use.
2774 sqlite3_int64 sqlite3session_memory_used(sqlite3_session *pSession){
2775 return pSession->nMalloc;
2779 ** Configure the session object passed as the first argument.
2781 int sqlite3session_object_config(sqlite3_session *pSession, int op, void *pArg){
2782 int rc = SQLITE_OK;
2783 switch( op ){
2784 case SQLITE_SESSION_OBJCONFIG_SIZE: {
2785 int iArg = *(int*)pArg;
2786 if( iArg>=0 ){
2787 if( pSession->pTable ){
2788 rc = SQLITE_MISUSE;
2789 }else{
2790 pSession->bEnableSize = (iArg!=0);
2793 *(int*)pArg = pSession->bEnableSize;
2794 break;
2797 default:
2798 rc = SQLITE_MISUSE;
2801 return rc;
2805 ** Return the maximum size of sqlite3session_changeset() output.
2807 sqlite3_int64 sqlite3session_changeset_size(sqlite3_session *pSession){
2808 return pSession->nMaxChangesetSize;
2812 ** Do the work for either sqlite3changeset_start() or start_strm().
2814 static int sessionChangesetStart(
2815 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2816 int (*xInput)(void *pIn, void *pData, int *pnData),
2817 void *pIn,
2818 int nChangeset, /* Size of buffer pChangeset in bytes */
2819 void *pChangeset, /* Pointer to buffer containing changeset */
2820 int bInvert, /* True to invert changeset */
2821 int bSkipEmpty /* True to skip empty UPDATE changes */
2823 sqlite3_changeset_iter *pRet; /* Iterator to return */
2824 int nByte; /* Number of bytes to allocate for iterator */
2826 assert( xInput==0 || (pChangeset==0 && nChangeset==0) );
2828 /* Zero the output variable in case an error occurs. */
2829 *pp = 0;
2831 /* Allocate and initialize the iterator structure. */
2832 nByte = sizeof(sqlite3_changeset_iter);
2833 pRet = (sqlite3_changeset_iter *)sqlite3_malloc(nByte);
2834 if( !pRet ) return SQLITE_NOMEM;
2835 memset(pRet, 0, sizeof(sqlite3_changeset_iter));
2836 pRet->in.aData = (u8 *)pChangeset;
2837 pRet->in.nData = nChangeset;
2838 pRet->in.xInput = xInput;
2839 pRet->in.pIn = pIn;
2840 pRet->in.bEof = (xInput ? 0 : 1);
2841 pRet->bInvert = bInvert;
2842 pRet->bSkipEmpty = bSkipEmpty;
2844 /* Populate the output variable and return success. */
2845 *pp = pRet;
2846 return SQLITE_OK;
2850 ** Create an iterator used to iterate through the contents of a changeset.
2852 int sqlite3changeset_start(
2853 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2854 int nChangeset, /* Size of buffer pChangeset in bytes */
2855 void *pChangeset /* Pointer to buffer containing changeset */
2857 return sessionChangesetStart(pp, 0, 0, nChangeset, pChangeset, 0, 0);
2859 int sqlite3changeset_start_v2(
2860 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2861 int nChangeset, /* Size of buffer pChangeset in bytes */
2862 void *pChangeset, /* Pointer to buffer containing changeset */
2863 int flags
2865 int bInvert = !!(flags & SQLITE_CHANGESETSTART_INVERT);
2866 return sessionChangesetStart(pp, 0, 0, nChangeset, pChangeset, bInvert, 0);
2870 ** Streaming version of sqlite3changeset_start().
2872 int sqlite3changeset_start_strm(
2873 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2874 int (*xInput)(void *pIn, void *pData, int *pnData),
2875 void *pIn
2877 return sessionChangesetStart(pp, xInput, pIn, 0, 0, 0, 0);
2879 int sqlite3changeset_start_v2_strm(
2880 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */
2881 int (*xInput)(void *pIn, void *pData, int *pnData),
2882 void *pIn,
2883 int flags
2885 int bInvert = !!(flags & SQLITE_CHANGESETSTART_INVERT);
2886 return sessionChangesetStart(pp, xInput, pIn, 0, 0, bInvert, 0);
2890 ** If the SessionInput object passed as the only argument is a streaming
2891 ** object and the buffer is full, discard some data to free up space.
2893 static void sessionDiscardData(SessionInput *pIn){
2894 if( pIn->xInput && pIn->iNext>=sessions_strm_chunk_size ){
2895 int nMove = pIn->buf.nBuf - pIn->iNext;
2896 assert( nMove>=0 );
2897 if( nMove>0 ){
2898 memmove(pIn->buf.aBuf, &pIn->buf.aBuf[pIn->iNext], nMove);
2900 pIn->buf.nBuf -= pIn->iNext;
2901 pIn->iNext = 0;
2902 pIn->nData = pIn->buf.nBuf;
2907 ** Ensure that there are at least nByte bytes available in the buffer. Or,
2908 ** if there are not nByte bytes remaining in the input, that all available
2909 ** data is in the buffer.
2911 ** Return an SQLite error code if an error occurs, or SQLITE_OK otherwise.
2913 static int sessionInputBuffer(SessionInput *pIn, int nByte){
2914 int rc = SQLITE_OK;
2915 if( pIn->xInput ){
2916 while( !pIn->bEof && (pIn->iNext+nByte)>=pIn->nData && rc==SQLITE_OK ){
2917 int nNew = sessions_strm_chunk_size;
2919 if( pIn->bNoDiscard==0 ) sessionDiscardData(pIn);
2920 if( SQLITE_OK==sessionBufferGrow(&pIn->buf, nNew, &rc) ){
2921 rc = pIn->xInput(pIn->pIn, &pIn->buf.aBuf[pIn->buf.nBuf], &nNew);
2922 if( nNew==0 ){
2923 pIn->bEof = 1;
2924 }else{
2925 pIn->buf.nBuf += nNew;
2929 pIn->aData = pIn->buf.aBuf;
2930 pIn->nData = pIn->buf.nBuf;
2933 return rc;
2937 ** When this function is called, *ppRec points to the start of a record
2938 ** that contains nCol values. This function advances the pointer *ppRec
2939 ** until it points to the byte immediately following that record.
2941 static void sessionSkipRecord(
2942 u8 **ppRec, /* IN/OUT: Record pointer */
2943 int nCol /* Number of values in record */
2945 u8 *aRec = *ppRec;
2946 int i;
2947 for(i=0; i<nCol; i++){
2948 int eType = *aRec++;
2949 if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
2950 int nByte;
2951 aRec += sessionVarintGet((u8*)aRec, &nByte);
2952 aRec += nByte;
2953 }else if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
2954 aRec += 8;
2958 *ppRec = aRec;
2962 ** This function sets the value of the sqlite3_value object passed as the
2963 ** first argument to a copy of the string or blob held in the aData[]
2964 ** buffer. SQLITE_OK is returned if successful, or SQLITE_NOMEM if an OOM
2965 ** error occurs.
2967 static int sessionValueSetStr(
2968 sqlite3_value *pVal, /* Set the value of this object */
2969 u8 *aData, /* Buffer containing string or blob data */
2970 int nData, /* Size of buffer aData[] in bytes */
2971 u8 enc /* String encoding (0 for blobs) */
2973 /* In theory this code could just pass SQLITE_TRANSIENT as the final
2974 ** argument to sqlite3ValueSetStr() and have the copy created
2975 ** automatically. But doing so makes it difficult to detect any OOM
2976 ** error. Hence the code to create the copy externally. */
2977 u8 *aCopy = sqlite3_malloc64((sqlite3_int64)nData+1);
2978 if( aCopy==0 ) return SQLITE_NOMEM;
2979 memcpy(aCopy, aData, nData);
2980 sqlite3ValueSetStr(pVal, nData, (char*)aCopy, enc, sqlite3_free);
2981 return SQLITE_OK;
2985 ** Deserialize a single record from a buffer in memory. See "RECORD FORMAT"
2986 ** for details.
2988 ** When this function is called, *paChange points to the start of the record
2989 ** to deserialize. Assuming no error occurs, *paChange is set to point to
2990 ** one byte after the end of the same record before this function returns.
2991 ** If the argument abPK is NULL, then the record contains nCol values. Or,
2992 ** if abPK is other than NULL, then the record contains only the PK fields
2993 ** (in other words, it is a patchset DELETE record).
2995 ** If successful, each element of the apOut[] array (allocated by the caller)
2996 ** is set to point to an sqlite3_value object containing the value read
2997 ** from the corresponding position in the record. If that value is not
2998 ** included in the record (i.e. because the record is part of an UPDATE change
2999 ** and the field was not modified), the corresponding element of apOut[] is
3000 ** set to NULL.
3002 ** It is the responsibility of the caller to free all sqlite_value structures
3003 ** using sqlite3_free().
3005 ** If an error occurs, an SQLite error code (e.g. SQLITE_NOMEM) is returned.
3006 ** The apOut[] array may have been partially populated in this case.
3008 static int sessionReadRecord(
3009 SessionInput *pIn, /* Input data */
3010 int nCol, /* Number of values in record */
3011 u8 *abPK, /* Array of primary key flags, or NULL */
3012 sqlite3_value **apOut, /* Write values to this array */
3013 int *pbEmpty
3015 int i; /* Used to iterate through columns */
3016 int rc = SQLITE_OK;
3018 assert( pbEmpty==0 || *pbEmpty==0 );
3019 if( pbEmpty ) *pbEmpty = 1;
3020 for(i=0; i<nCol && rc==SQLITE_OK; i++){
3021 int eType = 0; /* Type of value (SQLITE_NULL, TEXT etc.) */
3022 if( abPK && abPK[i]==0 ) continue;
3023 rc = sessionInputBuffer(pIn, 9);
3024 if( rc==SQLITE_OK ){
3025 if( pIn->iNext>=pIn->nData ){
3026 rc = SQLITE_CORRUPT_BKPT;
3027 }else{
3028 eType = pIn->aData[pIn->iNext++];
3029 assert( apOut[i]==0 );
3030 if( eType ){
3031 if( pbEmpty ) *pbEmpty = 0;
3032 apOut[i] = sqlite3ValueNew(0);
3033 if( !apOut[i] ) rc = SQLITE_NOMEM;
3038 if( rc==SQLITE_OK ){
3039 u8 *aVal = &pIn->aData[pIn->iNext];
3040 if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
3041 int nByte;
3042 pIn->iNext += sessionVarintGet(aVal, &nByte);
3043 rc = sessionInputBuffer(pIn, nByte);
3044 if( rc==SQLITE_OK ){
3045 if( nByte<0 || nByte>pIn->nData-pIn->iNext ){
3046 rc = SQLITE_CORRUPT_BKPT;
3047 }else{
3048 u8 enc = (eType==SQLITE_TEXT ? SQLITE_UTF8 : 0);
3049 rc = sessionValueSetStr(apOut[i],&pIn->aData[pIn->iNext],nByte,enc);
3050 pIn->iNext += nByte;
3054 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
3055 sqlite3_int64 v = sessionGetI64(aVal);
3056 if( eType==SQLITE_INTEGER ){
3057 sqlite3VdbeMemSetInt64(apOut[i], v);
3058 }else{
3059 double d;
3060 memcpy(&d, &v, 8);
3061 sqlite3VdbeMemSetDouble(apOut[i], d);
3063 pIn->iNext += 8;
3068 return rc;
3072 ** The input pointer currently points to the second byte of a table-header.
3073 ** Specifically, to the following:
3075 ** + number of columns in table (varint)
3076 ** + array of PK flags (1 byte per column),
3077 ** + table name (nul terminated).
3079 ** This function ensures that all of the above is present in the input
3080 ** buffer (i.e. that it can be accessed without any calls to xInput()).
3081 ** If successful, SQLITE_OK is returned. Otherwise, an SQLite error code.
3082 ** The input pointer is not moved.
3084 static int sessionChangesetBufferTblhdr(SessionInput *pIn, int *pnByte){
3085 int rc = SQLITE_OK;
3086 int nCol = 0;
3087 int nRead = 0;
3089 rc = sessionInputBuffer(pIn, 9);
3090 if( rc==SQLITE_OK ){
3091 nRead += sessionVarintGet(&pIn->aData[pIn->iNext + nRead], &nCol);
3092 /* The hard upper limit for the number of columns in an SQLite
3093 ** database table is, according to sqliteLimit.h, 32676. So
3094 ** consider any table-header that purports to have more than 65536
3095 ** columns to be corrupt. This is convenient because otherwise,
3096 ** if the (nCol>65536) condition below were omitted, a sufficiently
3097 ** large value for nCol may cause nRead to wrap around and become
3098 ** negative. Leading to a crash. */
3099 if( nCol<0 || nCol>65536 ){
3100 rc = SQLITE_CORRUPT_BKPT;
3101 }else{
3102 rc = sessionInputBuffer(pIn, nRead+nCol+100);
3103 nRead += nCol;
3107 while( rc==SQLITE_OK ){
3108 while( (pIn->iNext + nRead)<pIn->nData && pIn->aData[pIn->iNext + nRead] ){
3109 nRead++;
3111 if( (pIn->iNext + nRead)<pIn->nData ) break;
3112 rc = sessionInputBuffer(pIn, nRead + 100);
3114 *pnByte = nRead+1;
3115 return rc;
3119 ** The input pointer currently points to the first byte of the first field
3120 ** of a record consisting of nCol columns. This function ensures the entire
3121 ** record is buffered. It does not move the input pointer.
3123 ** If successful, SQLITE_OK is returned and *pnByte is set to the size of
3124 ** the record in bytes. Otherwise, an SQLite error code is returned. The
3125 ** final value of *pnByte is undefined in this case.
3127 static int sessionChangesetBufferRecord(
3128 SessionInput *pIn, /* Input data */
3129 int nCol, /* Number of columns in record */
3130 int *pnByte /* OUT: Size of record in bytes */
3132 int rc = SQLITE_OK;
3133 int nByte = 0;
3134 int i;
3135 for(i=0; rc==SQLITE_OK && i<nCol; i++){
3136 int eType;
3137 rc = sessionInputBuffer(pIn, nByte + 10);
3138 if( rc==SQLITE_OK ){
3139 eType = pIn->aData[pIn->iNext + nByte++];
3140 if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
3141 int n;
3142 nByte += sessionVarintGet(&pIn->aData[pIn->iNext+nByte], &n);
3143 nByte += n;
3144 rc = sessionInputBuffer(pIn, nByte);
3145 }else if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
3146 nByte += 8;
3150 *pnByte = nByte;
3151 return rc;
3155 ** The input pointer currently points to the second byte of a table-header.
3156 ** Specifically, to the following:
3158 ** + number of columns in table (varint)
3159 ** + array of PK flags (1 byte per column),
3160 ** + table name (nul terminated).
3162 ** This function decodes the table-header and populates the p->nCol,
3163 ** p->zTab and p->abPK[] variables accordingly. The p->apValue[] array is
3164 ** also allocated or resized according to the new value of p->nCol. The
3165 ** input pointer is left pointing to the byte following the table header.
3167 ** If successful, SQLITE_OK is returned. Otherwise, an SQLite error code
3168 ** is returned and the final values of the various fields enumerated above
3169 ** are undefined.
3171 static int sessionChangesetReadTblhdr(sqlite3_changeset_iter *p){
3172 int rc;
3173 int nCopy;
3174 assert( p->rc==SQLITE_OK );
3176 rc = sessionChangesetBufferTblhdr(&p->in, &nCopy);
3177 if( rc==SQLITE_OK ){
3178 int nByte;
3179 int nVarint;
3180 nVarint = sessionVarintGet(&p->in.aData[p->in.iNext], &p->nCol);
3181 if( p->nCol>0 ){
3182 nCopy -= nVarint;
3183 p->in.iNext += nVarint;
3184 nByte = p->nCol * sizeof(sqlite3_value*) * 2 + nCopy;
3185 p->tblhdr.nBuf = 0;
3186 sessionBufferGrow(&p->tblhdr, nByte, &rc);
3187 }else{
3188 rc = SQLITE_CORRUPT_BKPT;
3192 if( rc==SQLITE_OK ){
3193 size_t iPK = sizeof(sqlite3_value*)*p->nCol*2;
3194 memset(p->tblhdr.aBuf, 0, iPK);
3195 memcpy(&p->tblhdr.aBuf[iPK], &p->in.aData[p->in.iNext], nCopy);
3196 p->in.iNext += nCopy;
3199 p->apValue = (sqlite3_value**)p->tblhdr.aBuf;
3200 if( p->apValue==0 ){
3201 p->abPK = 0;
3202 p->zTab = 0;
3203 }else{
3204 p->abPK = (u8*)&p->apValue[p->nCol*2];
3205 p->zTab = p->abPK ? (char*)&p->abPK[p->nCol] : 0;
3207 return (p->rc = rc);
3211 ** Advance the changeset iterator to the next change. The differences between
3212 ** this function and sessionChangesetNext() are that
3214 ** * If pbEmpty is not NULL and the change is a no-op UPDATE (an UPDATE
3215 ** that modifies no columns), this function sets (*pbEmpty) to 1.
3217 ** * If the iterator is configured to skip no-op UPDATEs,
3218 ** sessionChangesetNext() does that. This function does not.
3220 static int sessionChangesetNextOne(
3221 sqlite3_changeset_iter *p, /* Changeset iterator */
3222 u8 **paRec, /* If non-NULL, store record pointer here */
3223 int *pnRec, /* If non-NULL, store size of record here */
3224 int *pbNew, /* If non-NULL, true if new table */
3225 int *pbEmpty
3227 int i;
3228 u8 op;
3230 assert( (paRec==0 && pnRec==0) || (paRec && pnRec) );
3231 assert( pbEmpty==0 || *pbEmpty==0 );
3233 /* If the iterator is in the error-state, return immediately. */
3234 if( p->rc!=SQLITE_OK ) return p->rc;
3236 /* Free the current contents of p->apValue[], if any. */
3237 if( p->apValue ){
3238 for(i=0; i<p->nCol*2; i++){
3239 sqlite3ValueFree(p->apValue[i]);
3241 memset(p->apValue, 0, sizeof(sqlite3_value*)*p->nCol*2);
3244 /* Make sure the buffer contains at least 10 bytes of input data, or all
3245 ** remaining data if there are less than 10 bytes available. This is
3246 ** sufficient either for the 'T' or 'P' byte and the varint that follows
3247 ** it, or for the two single byte values otherwise. */
3248 p->rc = sessionInputBuffer(&p->in, 2);
3249 if( p->rc!=SQLITE_OK ) return p->rc;
3251 /* If the iterator is already at the end of the changeset, return DONE. */
3252 if( p->in.iNext>=p->in.nData ){
3253 return SQLITE_DONE;
3256 sessionDiscardData(&p->in);
3257 p->in.iCurrent = p->in.iNext;
3259 op = p->in.aData[p->in.iNext++];
3260 while( op=='T' || op=='P' ){
3261 if( pbNew ) *pbNew = 1;
3262 p->bPatchset = (op=='P');
3263 if( sessionChangesetReadTblhdr(p) ) return p->rc;
3264 if( (p->rc = sessionInputBuffer(&p->in, 2)) ) return p->rc;
3265 p->in.iCurrent = p->in.iNext;
3266 if( p->in.iNext>=p->in.nData ) return SQLITE_DONE;
3267 op = p->in.aData[p->in.iNext++];
3270 if( p->zTab==0 || (p->bPatchset && p->bInvert) ){
3271 /* The first record in the changeset is not a table header. Must be a
3272 ** corrupt changeset. */
3273 assert( p->in.iNext==1 || p->zTab );
3274 return (p->rc = SQLITE_CORRUPT_BKPT);
3277 p->op = op;
3278 p->bIndirect = p->in.aData[p->in.iNext++];
3279 if( p->op!=SQLITE_UPDATE && p->op!=SQLITE_DELETE && p->op!=SQLITE_INSERT ){
3280 return (p->rc = SQLITE_CORRUPT_BKPT);
3283 if( paRec ){
3284 int nVal; /* Number of values to buffer */
3285 if( p->bPatchset==0 && op==SQLITE_UPDATE ){
3286 nVal = p->nCol * 2;
3287 }else if( p->bPatchset && op==SQLITE_DELETE ){
3288 nVal = 0;
3289 for(i=0; i<p->nCol; i++) if( p->abPK[i] ) nVal++;
3290 }else{
3291 nVal = p->nCol;
3293 p->rc = sessionChangesetBufferRecord(&p->in, nVal, pnRec);
3294 if( p->rc!=SQLITE_OK ) return p->rc;
3295 *paRec = &p->in.aData[p->in.iNext];
3296 p->in.iNext += *pnRec;
3297 }else{
3298 sqlite3_value **apOld = (p->bInvert ? &p->apValue[p->nCol] : p->apValue);
3299 sqlite3_value **apNew = (p->bInvert ? p->apValue : &p->apValue[p->nCol]);
3301 /* If this is an UPDATE or DELETE, read the old.* record. */
3302 if( p->op!=SQLITE_INSERT && (p->bPatchset==0 || p->op==SQLITE_DELETE) ){
3303 u8 *abPK = p->bPatchset ? p->abPK : 0;
3304 p->rc = sessionReadRecord(&p->in, p->nCol, abPK, apOld, 0);
3305 if( p->rc!=SQLITE_OK ) return p->rc;
3308 /* If this is an INSERT or UPDATE, read the new.* record. */
3309 if( p->op!=SQLITE_DELETE ){
3310 p->rc = sessionReadRecord(&p->in, p->nCol, 0, apNew, pbEmpty);
3311 if( p->rc!=SQLITE_OK ) return p->rc;
3314 if( (p->bPatchset || p->bInvert) && p->op==SQLITE_UPDATE ){
3315 /* If this is an UPDATE that is part of a patchset, then all PK and
3316 ** modified fields are present in the new.* record. The old.* record
3317 ** is currently completely empty. This block shifts the PK fields from
3318 ** new.* to old.*, to accommodate the code that reads these arrays. */
3319 for(i=0; i<p->nCol; i++){
3320 assert( p->bPatchset==0 || p->apValue[i]==0 );
3321 if( p->abPK[i] ){
3322 assert( p->apValue[i]==0 );
3323 p->apValue[i] = p->apValue[i+p->nCol];
3324 if( p->apValue[i]==0 ) return (p->rc = SQLITE_CORRUPT_BKPT);
3325 p->apValue[i+p->nCol] = 0;
3328 }else if( p->bInvert ){
3329 if( p->op==SQLITE_INSERT ) p->op = SQLITE_DELETE;
3330 else if( p->op==SQLITE_DELETE ) p->op = SQLITE_INSERT;
3333 /* If this is an UPDATE that is part of a changeset, then check that
3334 ** there are no fields in the old.* record that are not (a) PK fields,
3335 ** or (b) also present in the new.* record.
3337 ** Such records are technically corrupt, but the rebaser was at one
3338 ** point generating them. Under most circumstances this is benign, but
3339 ** can cause spurious SQLITE_RANGE errors when applying the changeset. */
3340 if( p->bPatchset==0 && p->op==SQLITE_UPDATE){
3341 for(i=0; i<p->nCol; i++){
3342 if( p->abPK[i]==0 && p->apValue[i+p->nCol]==0 ){
3343 sqlite3ValueFree(p->apValue[i]);
3344 p->apValue[i] = 0;
3350 return SQLITE_ROW;
3354 ** Advance the changeset iterator to the next change.
3356 ** If both paRec and pnRec are NULL, then this function works like the public
3357 ** API sqlite3changeset_next(). If SQLITE_ROW is returned, then the
3358 ** sqlite3changeset_new() and old() APIs may be used to query for values.
3360 ** Otherwise, if paRec and pnRec are not NULL, then a pointer to the change
3361 ** record is written to *paRec before returning and the number of bytes in
3362 ** the record to *pnRec.
3364 ** Either way, this function returns SQLITE_ROW if the iterator is
3365 ** successfully advanced to the next change in the changeset, an SQLite
3366 ** error code if an error occurs, or SQLITE_DONE if there are no further
3367 ** changes in the changeset.
3369 static int sessionChangesetNext(
3370 sqlite3_changeset_iter *p, /* Changeset iterator */
3371 u8 **paRec, /* If non-NULL, store record pointer here */
3372 int *pnRec, /* If non-NULL, store size of record here */
3373 int *pbNew /* If non-NULL, true if new table */
3375 int bEmpty;
3376 int rc;
3377 do {
3378 bEmpty = 0;
3379 rc = sessionChangesetNextOne(p, paRec, pnRec, pbNew, &bEmpty);
3380 }while( rc==SQLITE_ROW && p->bSkipEmpty && bEmpty);
3381 return rc;
3385 ** Advance an iterator created by sqlite3changeset_start() to the next
3386 ** change in the changeset. This function may return SQLITE_ROW, SQLITE_DONE
3387 ** or SQLITE_CORRUPT.
3389 ** This function may not be called on iterators passed to a conflict handler
3390 ** callback by changeset_apply().
3392 int sqlite3changeset_next(sqlite3_changeset_iter *p){
3393 return sessionChangesetNext(p, 0, 0, 0);
3397 ** The following function extracts information on the current change
3398 ** from a changeset iterator. It may only be called after changeset_next()
3399 ** has returned SQLITE_ROW.
3401 int sqlite3changeset_op(
3402 sqlite3_changeset_iter *pIter, /* Iterator handle */
3403 const char **pzTab, /* OUT: Pointer to table name */
3404 int *pnCol, /* OUT: Number of columns in table */
3405 int *pOp, /* OUT: SQLITE_INSERT, DELETE or UPDATE */
3406 int *pbIndirect /* OUT: True if change is indirect */
3408 *pOp = pIter->op;
3409 *pnCol = pIter->nCol;
3410 *pzTab = pIter->zTab;
3411 if( pbIndirect ) *pbIndirect = pIter->bIndirect;
3412 return SQLITE_OK;
3416 ** Return information regarding the PRIMARY KEY and number of columns in
3417 ** the database table affected by the change that pIter currently points
3418 ** to. This function may only be called after changeset_next() returns
3419 ** SQLITE_ROW.
3421 int sqlite3changeset_pk(
3422 sqlite3_changeset_iter *pIter, /* Iterator object */
3423 unsigned char **pabPK, /* OUT: Array of boolean - true for PK cols */
3424 int *pnCol /* OUT: Number of entries in output array */
3426 *pabPK = pIter->abPK;
3427 if( pnCol ) *pnCol = pIter->nCol;
3428 return SQLITE_OK;
3432 ** This function may only be called while the iterator is pointing to an
3433 ** SQLITE_UPDATE or SQLITE_DELETE change (see sqlite3changeset_op()).
3434 ** Otherwise, SQLITE_MISUSE is returned.
3436 ** It sets *ppValue to point to an sqlite3_value structure containing the
3437 ** iVal'th value in the old.* record. Or, if that particular value is not
3438 ** included in the record (because the change is an UPDATE and the field
3439 ** was not modified and is not a PK column), set *ppValue to NULL.
3441 ** If value iVal is out-of-range, SQLITE_RANGE is returned and *ppValue is
3442 ** not modified. Otherwise, SQLITE_OK.
3444 int sqlite3changeset_old(
3445 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3446 int iVal, /* Index of old.* value to retrieve */
3447 sqlite3_value **ppValue /* OUT: Old value (or NULL pointer) */
3449 if( pIter->op!=SQLITE_UPDATE && pIter->op!=SQLITE_DELETE ){
3450 return SQLITE_MISUSE;
3452 if( iVal<0 || iVal>=pIter->nCol ){
3453 return SQLITE_RANGE;
3455 *ppValue = pIter->apValue[iVal];
3456 return SQLITE_OK;
3460 ** This function may only be called while the iterator is pointing to an
3461 ** SQLITE_UPDATE or SQLITE_INSERT change (see sqlite3changeset_op()).
3462 ** Otherwise, SQLITE_MISUSE is returned.
3464 ** It sets *ppValue to point to an sqlite3_value structure containing the
3465 ** iVal'th value in the new.* record. Or, if that particular value is not
3466 ** included in the record (because the change is an UPDATE and the field
3467 ** was not modified), set *ppValue to NULL.
3469 ** If value iVal is out-of-range, SQLITE_RANGE is returned and *ppValue is
3470 ** not modified. Otherwise, SQLITE_OK.
3472 int sqlite3changeset_new(
3473 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3474 int iVal, /* Index of new.* value to retrieve */
3475 sqlite3_value **ppValue /* OUT: New value (or NULL pointer) */
3477 if( pIter->op!=SQLITE_UPDATE && pIter->op!=SQLITE_INSERT ){
3478 return SQLITE_MISUSE;
3480 if( iVal<0 || iVal>=pIter->nCol ){
3481 return SQLITE_RANGE;
3483 *ppValue = pIter->apValue[pIter->nCol+iVal];
3484 return SQLITE_OK;
3488 ** The following two macros are used internally. They are similar to the
3489 ** sqlite3changeset_new() and sqlite3changeset_old() functions, except that
3490 ** they omit all error checking and return a pointer to the requested value.
3492 #define sessionChangesetNew(pIter, iVal) (pIter)->apValue[(pIter)->nCol+(iVal)]
3493 #define sessionChangesetOld(pIter, iVal) (pIter)->apValue[(iVal)]
3496 ** This function may only be called with a changeset iterator that has been
3497 ** passed to an SQLITE_CHANGESET_DATA or SQLITE_CHANGESET_CONFLICT
3498 ** conflict-handler function. Otherwise, SQLITE_MISUSE is returned.
3500 ** If successful, *ppValue is set to point to an sqlite3_value structure
3501 ** containing the iVal'th value of the conflicting record.
3503 ** If value iVal is out-of-range or some other error occurs, an SQLite error
3504 ** code is returned. Otherwise, SQLITE_OK.
3506 int sqlite3changeset_conflict(
3507 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3508 int iVal, /* Index of conflict record value to fetch */
3509 sqlite3_value **ppValue /* OUT: Value from conflicting row */
3511 if( !pIter->pConflict ){
3512 return SQLITE_MISUSE;
3514 if( iVal<0 || iVal>=pIter->nCol ){
3515 return SQLITE_RANGE;
3517 *ppValue = sqlite3_column_value(pIter->pConflict, iVal);
3518 return SQLITE_OK;
3522 ** This function may only be called with an iterator passed to an
3523 ** SQLITE_CHANGESET_FOREIGN_KEY conflict handler callback. In this case
3524 ** it sets the output variable to the total number of known foreign key
3525 ** violations in the destination database and returns SQLITE_OK.
3527 ** In all other cases this function returns SQLITE_MISUSE.
3529 int sqlite3changeset_fk_conflicts(
3530 sqlite3_changeset_iter *pIter, /* Changeset iterator */
3531 int *pnOut /* OUT: Number of FK violations */
3533 if( pIter->pConflict || pIter->apValue ){
3534 return SQLITE_MISUSE;
3536 *pnOut = pIter->nCol;
3537 return SQLITE_OK;
3542 ** Finalize an iterator allocated with sqlite3changeset_start().
3544 ** This function may not be called on iterators passed to a conflict handler
3545 ** callback by changeset_apply().
3547 int sqlite3changeset_finalize(sqlite3_changeset_iter *p){
3548 int rc = SQLITE_OK;
3549 if( p ){
3550 int i; /* Used to iterate through p->apValue[] */
3551 rc = p->rc;
3552 if( p->apValue ){
3553 for(i=0; i<p->nCol*2; i++) sqlite3ValueFree(p->apValue[i]);
3555 sqlite3_free(p->tblhdr.aBuf);
3556 sqlite3_free(p->in.buf.aBuf);
3557 sqlite3_free(p);
3559 return rc;
3562 static int sessionChangesetInvert(
3563 SessionInput *pInput, /* Input changeset */
3564 int (*xOutput)(void *pOut, const void *pData, int nData),
3565 void *pOut,
3566 int *pnInverted, /* OUT: Number of bytes in output changeset */
3567 void **ppInverted /* OUT: Inverse of pChangeset */
3569 int rc = SQLITE_OK; /* Return value */
3570 SessionBuffer sOut; /* Output buffer */
3571 int nCol = 0; /* Number of cols in current table */
3572 u8 *abPK = 0; /* PK array for current table */
3573 sqlite3_value **apVal = 0; /* Space for values for UPDATE inversion */
3574 SessionBuffer sPK = {0, 0, 0}; /* PK array for current table */
3576 /* Initialize the output buffer */
3577 memset(&sOut, 0, sizeof(SessionBuffer));
3579 /* Zero the output variables in case an error occurs. */
3580 if( ppInverted ){
3581 *ppInverted = 0;
3582 *pnInverted = 0;
3585 while( 1 ){
3586 u8 eType;
3588 /* Test for EOF. */
3589 if( (rc = sessionInputBuffer(pInput, 2)) ) goto finished_invert;
3590 if( pInput->iNext>=pInput->nData ) break;
3591 eType = pInput->aData[pInput->iNext];
3593 switch( eType ){
3594 case 'T': {
3595 /* A 'table' record consists of:
3597 ** * A constant 'T' character,
3598 ** * Number of columns in said table (a varint),
3599 ** * An array of nCol bytes (sPK),
3600 ** * A nul-terminated table name.
3602 int nByte;
3603 int nVar;
3604 pInput->iNext++;
3605 if( (rc = sessionChangesetBufferTblhdr(pInput, &nByte)) ){
3606 goto finished_invert;
3608 nVar = sessionVarintGet(&pInput->aData[pInput->iNext], &nCol);
3609 sPK.nBuf = 0;
3610 sessionAppendBlob(&sPK, &pInput->aData[pInput->iNext+nVar], nCol, &rc);
3611 sessionAppendByte(&sOut, eType, &rc);
3612 sessionAppendBlob(&sOut, &pInput->aData[pInput->iNext], nByte, &rc);
3613 if( rc ) goto finished_invert;
3615 pInput->iNext += nByte;
3616 sqlite3_free(apVal);
3617 apVal = 0;
3618 abPK = sPK.aBuf;
3619 break;
3622 case SQLITE_INSERT:
3623 case SQLITE_DELETE: {
3624 int nByte;
3625 int bIndirect = pInput->aData[pInput->iNext+1];
3626 int eType2 = (eType==SQLITE_DELETE ? SQLITE_INSERT : SQLITE_DELETE);
3627 pInput->iNext += 2;
3628 assert( rc==SQLITE_OK );
3629 rc = sessionChangesetBufferRecord(pInput, nCol, &nByte);
3630 sessionAppendByte(&sOut, eType2, &rc);
3631 sessionAppendByte(&sOut, bIndirect, &rc);
3632 sessionAppendBlob(&sOut, &pInput->aData[pInput->iNext], nByte, &rc);
3633 pInput->iNext += nByte;
3634 if( rc ) goto finished_invert;
3635 break;
3638 case SQLITE_UPDATE: {
3639 int iCol;
3641 if( 0==apVal ){
3642 apVal = (sqlite3_value **)sqlite3_malloc64(sizeof(apVal[0])*nCol*2);
3643 if( 0==apVal ){
3644 rc = SQLITE_NOMEM;
3645 goto finished_invert;
3647 memset(apVal, 0, sizeof(apVal[0])*nCol*2);
3650 /* Write the header for the new UPDATE change. Same as the original. */
3651 sessionAppendByte(&sOut, eType, &rc);
3652 sessionAppendByte(&sOut, pInput->aData[pInput->iNext+1], &rc);
3654 /* Read the old.* and new.* records for the update change. */
3655 pInput->iNext += 2;
3656 rc = sessionReadRecord(pInput, nCol, 0, &apVal[0], 0);
3657 if( rc==SQLITE_OK ){
3658 rc = sessionReadRecord(pInput, nCol, 0, &apVal[nCol], 0);
3661 /* Write the new old.* record. Consists of the PK columns from the
3662 ** original old.* record, and the other values from the original
3663 ** new.* record. */
3664 for(iCol=0; iCol<nCol; iCol++){
3665 sqlite3_value *pVal = apVal[iCol + (abPK[iCol] ? 0 : nCol)];
3666 sessionAppendValue(&sOut, pVal, &rc);
3669 /* Write the new new.* record. Consists of a copy of all values
3670 ** from the original old.* record, except for the PK columns, which
3671 ** are set to "undefined". */
3672 for(iCol=0; iCol<nCol; iCol++){
3673 sqlite3_value *pVal = (abPK[iCol] ? 0 : apVal[iCol]);
3674 sessionAppendValue(&sOut, pVal, &rc);
3677 for(iCol=0; iCol<nCol*2; iCol++){
3678 sqlite3ValueFree(apVal[iCol]);
3680 memset(apVal, 0, sizeof(apVal[0])*nCol*2);
3681 if( rc!=SQLITE_OK ){
3682 goto finished_invert;
3685 break;
3688 default:
3689 rc = SQLITE_CORRUPT_BKPT;
3690 goto finished_invert;
3693 assert( rc==SQLITE_OK );
3694 if( xOutput && sOut.nBuf>=sessions_strm_chunk_size ){
3695 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf);
3696 sOut.nBuf = 0;
3697 if( rc!=SQLITE_OK ) goto finished_invert;
3701 assert( rc==SQLITE_OK );
3702 if( pnInverted && ALWAYS(ppInverted) ){
3703 *pnInverted = sOut.nBuf;
3704 *ppInverted = sOut.aBuf;
3705 sOut.aBuf = 0;
3706 }else if( sOut.nBuf>0 && ALWAYS(xOutput!=0) ){
3707 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf);
3710 finished_invert:
3711 sqlite3_free(sOut.aBuf);
3712 sqlite3_free(apVal);
3713 sqlite3_free(sPK.aBuf);
3714 return rc;
3719 ** Invert a changeset object.
3721 int sqlite3changeset_invert(
3722 int nChangeset, /* Number of bytes in input */
3723 const void *pChangeset, /* Input changeset */
3724 int *pnInverted, /* OUT: Number of bytes in output changeset */
3725 void **ppInverted /* OUT: Inverse of pChangeset */
3727 SessionInput sInput;
3729 /* Set up the input stream */
3730 memset(&sInput, 0, sizeof(SessionInput));
3731 sInput.nData = nChangeset;
3732 sInput.aData = (u8*)pChangeset;
3734 return sessionChangesetInvert(&sInput, 0, 0, pnInverted, ppInverted);
3738 ** Streaming version of sqlite3changeset_invert().
3740 int sqlite3changeset_invert_strm(
3741 int (*xInput)(void *pIn, void *pData, int *pnData),
3742 void *pIn,
3743 int (*xOutput)(void *pOut, const void *pData, int nData),
3744 void *pOut
3746 SessionInput sInput;
3747 int rc;
3749 /* Set up the input stream */
3750 memset(&sInput, 0, sizeof(SessionInput));
3751 sInput.xInput = xInput;
3752 sInput.pIn = pIn;
3754 rc = sessionChangesetInvert(&sInput, xOutput, pOut, 0, 0);
3755 sqlite3_free(sInput.buf.aBuf);
3756 return rc;
3760 typedef struct SessionUpdate SessionUpdate;
3761 struct SessionUpdate {
3762 sqlite3_stmt *pStmt;
3763 u32 *aMask;
3764 SessionUpdate *pNext;
3767 typedef struct SessionApplyCtx SessionApplyCtx;
3768 struct SessionApplyCtx {
3769 sqlite3 *db;
3770 sqlite3_stmt *pDelete; /* DELETE statement */
3771 sqlite3_stmt *pInsert; /* INSERT statement */
3772 sqlite3_stmt *pSelect; /* SELECT statement */
3773 int nCol; /* Size of azCol[] and abPK[] arrays */
3774 const char **azCol; /* Array of column names */
3775 u8 *abPK; /* Boolean array - true if column is in PK */
3776 u32 *aUpdateMask; /* Used by sessionUpdateFind */
3777 SessionUpdate *pUp;
3778 int bStat1; /* True if table is sqlite_stat1 */
3779 int bDeferConstraints; /* True to defer constraints */
3780 int bInvertConstraints; /* Invert when iterating constraints buffer */
3781 SessionBuffer constraints; /* Deferred constraints are stored here */
3782 SessionBuffer rebase; /* Rebase information (if any) here */
3783 u8 bRebaseStarted; /* If table header is already in rebase */
3784 u8 bRebase; /* True to collect rebase information */
3787 /* Number of prepared UPDATE statements to cache. */
3788 #define SESSION_UPDATE_CACHE_SZ 12
3791 ** Find a prepared UPDATE statement suitable for the UPDATE step currently
3792 ** being visited by the iterator. The UPDATE is of the form:
3794 ** UPDATE tbl SET col = ?, col2 = ? WHERE pk1 IS ? AND pk2 IS ?
3796 static int sessionUpdateFind(
3797 sqlite3_changeset_iter *pIter,
3798 SessionApplyCtx *p,
3799 int bPatchset,
3800 sqlite3_stmt **ppStmt
3802 int rc = SQLITE_OK;
3803 SessionUpdate *pUp = 0;
3804 int nCol = pIter->nCol;
3805 int nU32 = (pIter->nCol+33)/32;
3806 int ii;
3808 if( p->aUpdateMask==0 ){
3809 p->aUpdateMask = sqlite3_malloc(nU32*sizeof(u32));
3810 if( p->aUpdateMask==0 ){
3811 rc = SQLITE_NOMEM;
3815 if( rc==SQLITE_OK ){
3816 memset(p->aUpdateMask, 0, nU32*sizeof(u32));
3817 rc = SQLITE_CORRUPT;
3818 for(ii=0; ii<pIter->nCol; ii++){
3819 if( sessionChangesetNew(pIter, ii) ){
3820 p->aUpdateMask[ii/32] |= (1<<(ii%32));
3821 rc = SQLITE_OK;
3826 if( rc==SQLITE_OK ){
3827 if( bPatchset ) p->aUpdateMask[nCol/32] |= (1<<(nCol%32));
3829 if( p->pUp ){
3830 int nUp = 0;
3831 SessionUpdate **pp = &p->pUp;
3832 while( 1 ){
3833 nUp++;
3834 if( 0==memcmp(p->aUpdateMask, (*pp)->aMask, nU32*sizeof(u32)) ){
3835 pUp = *pp;
3836 *pp = pUp->pNext;
3837 pUp->pNext = p->pUp;
3838 p->pUp = pUp;
3839 break;
3842 if( (*pp)->pNext ){
3843 pp = &(*pp)->pNext;
3844 }else{
3845 if( nUp>=SESSION_UPDATE_CACHE_SZ ){
3846 sqlite3_finalize((*pp)->pStmt);
3847 sqlite3_free(*pp);
3848 *pp = 0;
3850 break;
3855 if( pUp==0 ){
3856 int nByte = sizeof(SessionUpdate) * nU32*sizeof(u32);
3857 int bStat1 = (sqlite3_stricmp(pIter->zTab, "sqlite_stat1")==0);
3858 pUp = (SessionUpdate*)sqlite3_malloc(nByte);
3859 if( pUp==0 ){
3860 rc = SQLITE_NOMEM;
3861 }else{
3862 const char *zSep = "";
3863 SessionBuffer buf;
3865 memset(&buf, 0, sizeof(buf));
3866 pUp->aMask = (u32*)&pUp[1];
3867 memcpy(pUp->aMask, p->aUpdateMask, nU32*sizeof(u32));
3869 sessionAppendStr(&buf, "UPDATE main.", &rc);
3870 sessionAppendIdent(&buf, pIter->zTab, &rc);
3871 sessionAppendStr(&buf, " SET ", &rc);
3873 /* Create the assignments part of the UPDATE */
3874 for(ii=0; ii<pIter->nCol; ii++){
3875 if( p->abPK[ii]==0 && sessionChangesetNew(pIter, ii) ){
3876 sessionAppendStr(&buf, zSep, &rc);
3877 sessionAppendIdent(&buf, p->azCol[ii], &rc);
3878 sessionAppendStr(&buf, " = ?", &rc);
3879 sessionAppendInteger(&buf, ii*2+1, &rc);
3880 zSep = ", ";
3884 /* Create the WHERE clause part of the UPDATE */
3885 zSep = "";
3886 sessionAppendStr(&buf, " WHERE ", &rc);
3887 for(ii=0; ii<pIter->nCol; ii++){
3888 if( p->abPK[ii] || (bPatchset==0 && sessionChangesetOld(pIter, ii)) ){
3889 sessionAppendStr(&buf, zSep, &rc);
3890 if( bStat1 && ii==1 ){
3891 assert( sqlite3_stricmp(p->azCol[ii], "idx")==0 );
3892 sessionAppendStr(&buf,
3893 "idx IS CASE "
3894 "WHEN length(?4)=0 AND typeof(?4)='blob' THEN NULL "
3895 "ELSE ?4 END ", &rc
3897 }else{
3898 sessionAppendIdent(&buf, p->azCol[ii], &rc);
3899 sessionAppendStr(&buf, " IS ?", &rc);
3900 sessionAppendInteger(&buf, ii*2+2, &rc);
3902 zSep = " AND ";
3906 if( rc==SQLITE_OK ){
3907 char *zSql = (char*)buf.aBuf;
3908 rc = sqlite3_prepare_v2(p->db, zSql, buf.nBuf, &pUp->pStmt, 0);
3911 if( rc!=SQLITE_OK ){
3912 sqlite3_free(pUp);
3913 pUp = 0;
3914 }else{
3915 pUp->pNext = p->pUp;
3916 p->pUp = pUp;
3918 sqlite3_free(buf.aBuf);
3923 assert( (rc==SQLITE_OK)==(pUp!=0) );
3924 if( pUp ){
3925 *ppStmt = pUp->pStmt;
3926 }else{
3927 *ppStmt = 0;
3929 return rc;
3933 ** Free all cached UPDATE statements.
3935 static void sessionUpdateFree(SessionApplyCtx *p){
3936 SessionUpdate *pUp;
3937 SessionUpdate *pNext;
3938 for(pUp=p->pUp; pUp; pUp=pNext){
3939 pNext = pUp->pNext;
3940 sqlite3_finalize(pUp->pStmt);
3941 sqlite3_free(pUp);
3943 p->pUp = 0;
3944 sqlite3_free(p->aUpdateMask);
3945 p->aUpdateMask = 0;
3949 ** Formulate a statement to DELETE a row from database db. Assuming a table
3950 ** structure like this:
3952 ** CREATE TABLE x(a, b, c, d, PRIMARY KEY(a, c));
3954 ** The DELETE statement looks like this:
3956 ** DELETE FROM x WHERE a = :1 AND c = :3 AND (:5 OR b IS :2 AND d IS :4)
3958 ** Variable :5 (nCol+1) is a boolean. It should be set to 0 if we require
3959 ** matching b and d values, or 1 otherwise. The second case comes up if the
3960 ** conflict handler is invoked with NOTFOUND and returns CHANGESET_REPLACE.
3962 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pDelete is left
3963 ** pointing to the prepared version of the SQL statement.
3965 static int sessionDeleteRow(
3966 sqlite3 *db, /* Database handle */
3967 const char *zTab, /* Table name */
3968 SessionApplyCtx *p /* Session changeset-apply context */
3970 int i;
3971 const char *zSep = "";
3972 int rc = SQLITE_OK;
3973 SessionBuffer buf = {0, 0, 0};
3974 int nPk = 0;
3976 sessionAppendStr(&buf, "DELETE FROM main.", &rc);
3977 sessionAppendIdent(&buf, zTab, &rc);
3978 sessionAppendStr(&buf, " WHERE ", &rc);
3980 for(i=0; i<p->nCol; i++){
3981 if( p->abPK[i] ){
3982 nPk++;
3983 sessionAppendStr(&buf, zSep, &rc);
3984 sessionAppendIdent(&buf, p->azCol[i], &rc);
3985 sessionAppendStr(&buf, " = ?", &rc);
3986 sessionAppendInteger(&buf, i+1, &rc);
3987 zSep = " AND ";
3991 if( nPk<p->nCol ){
3992 sessionAppendStr(&buf, " AND (?", &rc);
3993 sessionAppendInteger(&buf, p->nCol+1, &rc);
3994 sessionAppendStr(&buf, " OR ", &rc);
3996 zSep = "";
3997 for(i=0; i<p->nCol; i++){
3998 if( !p->abPK[i] ){
3999 sessionAppendStr(&buf, zSep, &rc);
4000 sessionAppendIdent(&buf, p->azCol[i], &rc);
4001 sessionAppendStr(&buf, " IS ?", &rc);
4002 sessionAppendInteger(&buf, i+1, &rc);
4003 zSep = "AND ";
4006 sessionAppendStr(&buf, ")", &rc);
4009 if( rc==SQLITE_OK ){
4010 rc = sqlite3_prepare_v2(db, (char *)buf.aBuf, buf.nBuf, &p->pDelete, 0);
4012 sqlite3_free(buf.aBuf);
4014 return rc;
4018 ** Formulate and prepare an SQL statement to query table zTab by primary
4019 ** key. Assuming the following table structure:
4021 ** CREATE TABLE x(a, b, c, d, PRIMARY KEY(a, c));
4023 ** The SELECT statement looks like this:
4025 ** SELECT * FROM x WHERE a = ?1 AND c = ?3
4027 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pSelect is left
4028 ** pointing to the prepared version of the SQL statement.
4030 static int sessionSelectRow(
4031 sqlite3 *db, /* Database handle */
4032 const char *zTab, /* Table name */
4033 SessionApplyCtx *p /* Session changeset-apply context */
4035 return sessionSelectStmt(
4036 db, "main", zTab, p->nCol, p->azCol, p->abPK, &p->pSelect);
4040 ** Formulate and prepare an INSERT statement to add a record to table zTab.
4041 ** For example:
4043 ** INSERT INTO main."zTab" VALUES(?1, ?2, ?3 ...);
4045 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pInsert is left
4046 ** pointing to the prepared version of the SQL statement.
4048 static int sessionInsertRow(
4049 sqlite3 *db, /* Database handle */
4050 const char *zTab, /* Table name */
4051 SessionApplyCtx *p /* Session changeset-apply context */
4053 int rc = SQLITE_OK;
4054 int i;
4055 SessionBuffer buf = {0, 0, 0};
4057 sessionAppendStr(&buf, "INSERT INTO main.", &rc);
4058 sessionAppendIdent(&buf, zTab, &rc);
4059 sessionAppendStr(&buf, "(", &rc);
4060 for(i=0; i<p->nCol; i++){
4061 if( i!=0 ) sessionAppendStr(&buf, ", ", &rc);
4062 sessionAppendIdent(&buf, p->azCol[i], &rc);
4065 sessionAppendStr(&buf, ") VALUES(?", &rc);
4066 for(i=1; i<p->nCol; i++){
4067 sessionAppendStr(&buf, ", ?", &rc);
4069 sessionAppendStr(&buf, ")", &rc);
4071 if( rc==SQLITE_OK ){
4072 rc = sqlite3_prepare_v2(db, (char *)buf.aBuf, buf.nBuf, &p->pInsert, 0);
4074 sqlite3_free(buf.aBuf);
4075 return rc;
4078 static int sessionPrepare(sqlite3 *db, sqlite3_stmt **pp, const char *zSql){
4079 return sqlite3_prepare_v2(db, zSql, -1, pp, 0);
4083 ** Prepare statements for applying changes to the sqlite_stat1 table.
4084 ** These are similar to those created by sessionSelectRow(),
4085 ** sessionInsertRow(), sessionUpdateRow() and sessionDeleteRow() for
4086 ** other tables.
4088 static int sessionStat1Sql(sqlite3 *db, SessionApplyCtx *p){
4089 int rc = sessionSelectRow(db, "sqlite_stat1", p);
4090 if( rc==SQLITE_OK ){
4091 rc = sessionPrepare(db, &p->pInsert,
4092 "INSERT INTO main.sqlite_stat1 VALUES(?1, "
4093 "CASE WHEN length(?2)=0 AND typeof(?2)='blob' THEN NULL ELSE ?2 END, "
4094 "?3)"
4097 if( rc==SQLITE_OK ){
4098 rc = sessionPrepare(db, &p->pDelete,
4099 "DELETE FROM main.sqlite_stat1 WHERE tbl=?1 AND idx IS "
4100 "CASE WHEN length(?2)=0 AND typeof(?2)='blob' THEN NULL ELSE ?2 END "
4101 "AND (?4 OR stat IS ?3)"
4104 return rc;
4108 ** A wrapper around sqlite3_bind_value() that detects an extra problem.
4109 ** See comments in the body of this function for details.
4111 static int sessionBindValue(
4112 sqlite3_stmt *pStmt, /* Statement to bind value to */
4113 int i, /* Parameter number to bind to */
4114 sqlite3_value *pVal /* Value to bind */
4116 int eType = sqlite3_value_type(pVal);
4117 /* COVERAGE: The (pVal->z==0) branch is never true using current versions
4118 ** of SQLite. If a malloc fails in an sqlite3_value_xxx() function, either
4119 ** the (pVal->z) variable remains as it was or the type of the value is
4120 ** set to SQLITE_NULL. */
4121 if( (eType==SQLITE_TEXT || eType==SQLITE_BLOB) && pVal->z==0 ){
4122 /* This condition occurs when an earlier OOM in a call to
4123 ** sqlite3_value_text() or sqlite3_value_blob() (perhaps from within
4124 ** a conflict-handler) has zeroed the pVal->z pointer. Return NOMEM. */
4125 return SQLITE_NOMEM;
4127 return sqlite3_bind_value(pStmt, i, pVal);
4131 ** Iterator pIter must point to an SQLITE_INSERT entry. This function
4132 ** transfers new.* values from the current iterator entry to statement
4133 ** pStmt. The table being inserted into has nCol columns.
4135 ** New.* value $i from the iterator is bound to variable ($i+1) of
4136 ** statement pStmt. If parameter abPK is NULL, all values from 0 to (nCol-1)
4137 ** are transfered to the statement. Otherwise, if abPK is not NULL, it points
4138 ** to an array nCol elements in size. In this case only those values for
4139 ** which abPK[$i] is true are read from the iterator and bound to the
4140 ** statement.
4142 ** An SQLite error code is returned if an error occurs. Otherwise, SQLITE_OK.
4144 static int sessionBindRow(
4145 sqlite3_changeset_iter *pIter, /* Iterator to read values from */
4146 int(*xValue)(sqlite3_changeset_iter *, int, sqlite3_value **),
4147 int nCol, /* Number of columns */
4148 u8 *abPK, /* If not NULL, bind only if true */
4149 sqlite3_stmt *pStmt /* Bind values to this statement */
4151 int i;
4152 int rc = SQLITE_OK;
4154 /* Neither sqlite3changeset_old or sqlite3changeset_new can fail if the
4155 ** argument iterator points to a suitable entry. Make sure that xValue
4156 ** is one of these to guarantee that it is safe to ignore the return
4157 ** in the code below. */
4158 assert( xValue==sqlite3changeset_old || xValue==sqlite3changeset_new );
4160 for(i=0; rc==SQLITE_OK && i<nCol; i++){
4161 if( !abPK || abPK[i] ){
4162 sqlite3_value *pVal = 0;
4163 (void)xValue(pIter, i, &pVal);
4164 if( pVal==0 ){
4165 /* The value in the changeset was "undefined". This indicates a
4166 ** corrupt changeset blob. */
4167 rc = SQLITE_CORRUPT_BKPT;
4168 }else{
4169 rc = sessionBindValue(pStmt, i+1, pVal);
4173 return rc;
4177 ** SQL statement pSelect is as generated by the sessionSelectRow() function.
4178 ** This function binds the primary key values from the change that changeset
4179 ** iterator pIter points to to the SELECT and attempts to seek to the table
4180 ** entry. If a row is found, the SELECT statement left pointing at the row
4181 ** and SQLITE_ROW is returned. Otherwise, if no row is found and no error
4182 ** has occured, the statement is reset and SQLITE_OK is returned. If an
4183 ** error occurs, the statement is reset and an SQLite error code is returned.
4185 ** If this function returns SQLITE_ROW, the caller must eventually reset()
4186 ** statement pSelect. If any other value is returned, the statement does
4187 ** not require a reset().
4189 ** If the iterator currently points to an INSERT record, bind values from the
4190 ** new.* record to the SELECT statement. Or, if it points to a DELETE or
4191 ** UPDATE, bind values from the old.* record.
4193 static int sessionSeekToRow(
4194 sqlite3_changeset_iter *pIter, /* Changeset iterator */
4195 u8 *abPK, /* Primary key flags array */
4196 sqlite3_stmt *pSelect /* SELECT statement from sessionSelectRow() */
4198 int rc; /* Return code */
4199 int nCol; /* Number of columns in table */
4200 int op; /* Changset operation (SQLITE_UPDATE etc.) */
4201 const char *zDummy; /* Unused */
4203 sqlite3changeset_op(pIter, &zDummy, &nCol, &op, 0);
4204 rc = sessionBindRow(pIter,
4205 op==SQLITE_INSERT ? sqlite3changeset_new : sqlite3changeset_old,
4206 nCol, abPK, pSelect
4209 if( rc==SQLITE_OK ){
4210 rc = sqlite3_step(pSelect);
4211 if( rc!=SQLITE_ROW ) rc = sqlite3_reset(pSelect);
4214 return rc;
4218 ** This function is called from within sqlite3changeset_apply_v2() when
4219 ** a conflict is encountered and resolved using conflict resolution
4220 ** mode eType (either SQLITE_CHANGESET_OMIT or SQLITE_CHANGESET_REPLACE)..
4221 ** It adds a conflict resolution record to the buffer in
4222 ** SessionApplyCtx.rebase, which will eventually be returned to the caller
4223 ** of apply_v2() as the "rebase" buffer.
4225 ** Return SQLITE_OK if successful, or an SQLite error code otherwise.
4227 static int sessionRebaseAdd(
4228 SessionApplyCtx *p, /* Apply context */
4229 int eType, /* Conflict resolution (OMIT or REPLACE) */
4230 sqlite3_changeset_iter *pIter /* Iterator pointing at current change */
4232 int rc = SQLITE_OK;
4233 if( p->bRebase ){
4234 int i;
4235 int eOp = pIter->op;
4236 if( p->bRebaseStarted==0 ){
4237 /* Append a table-header to the rebase buffer */
4238 const char *zTab = pIter->zTab;
4239 sessionAppendByte(&p->rebase, 'T', &rc);
4240 sessionAppendVarint(&p->rebase, p->nCol, &rc);
4241 sessionAppendBlob(&p->rebase, p->abPK, p->nCol, &rc);
4242 sessionAppendBlob(&p->rebase, (u8*)zTab, (int)strlen(zTab)+1, &rc);
4243 p->bRebaseStarted = 1;
4246 assert( eType==SQLITE_CHANGESET_REPLACE||eType==SQLITE_CHANGESET_OMIT );
4247 assert( eOp==SQLITE_DELETE || eOp==SQLITE_INSERT || eOp==SQLITE_UPDATE );
4249 sessionAppendByte(&p->rebase,
4250 (eOp==SQLITE_DELETE ? SQLITE_DELETE : SQLITE_INSERT), &rc
4252 sessionAppendByte(&p->rebase, (eType==SQLITE_CHANGESET_REPLACE), &rc);
4253 for(i=0; i<p->nCol; i++){
4254 sqlite3_value *pVal = 0;
4255 if( eOp==SQLITE_DELETE || (eOp==SQLITE_UPDATE && p->abPK[i]) ){
4256 sqlite3changeset_old(pIter, i, &pVal);
4257 }else{
4258 sqlite3changeset_new(pIter, i, &pVal);
4260 sessionAppendValue(&p->rebase, pVal, &rc);
4263 return rc;
4267 ** Invoke the conflict handler for the change that the changeset iterator
4268 ** currently points to.
4270 ** Argument eType must be either CHANGESET_DATA or CHANGESET_CONFLICT.
4271 ** If argument pbReplace is NULL, then the type of conflict handler invoked
4272 ** depends solely on eType, as follows:
4274 ** eType value Value passed to xConflict
4275 ** -------------------------------------------------
4276 ** CHANGESET_DATA CHANGESET_NOTFOUND
4277 ** CHANGESET_CONFLICT CHANGESET_CONSTRAINT
4279 ** Or, if pbReplace is not NULL, then an attempt is made to find an existing
4280 ** record with the same primary key as the record about to be deleted, updated
4281 ** or inserted. If such a record can be found, it is available to the conflict
4282 ** handler as the "conflicting" record. In this case the type of conflict
4283 ** handler invoked is as follows:
4285 ** eType value PK Record found? Value passed to xConflict
4286 ** ----------------------------------------------------------------
4287 ** CHANGESET_DATA Yes CHANGESET_DATA
4288 ** CHANGESET_DATA No CHANGESET_NOTFOUND
4289 ** CHANGESET_CONFLICT Yes CHANGESET_CONFLICT
4290 ** CHANGESET_CONFLICT No CHANGESET_CONSTRAINT
4292 ** If pbReplace is not NULL, and a record with a matching PK is found, and
4293 ** the conflict handler function returns SQLITE_CHANGESET_REPLACE, *pbReplace
4294 ** is set to non-zero before returning SQLITE_OK.
4296 ** If the conflict handler returns SQLITE_CHANGESET_ABORT, SQLITE_ABORT is
4297 ** returned. Or, if the conflict handler returns an invalid value,
4298 ** SQLITE_MISUSE. If the conflict handler returns SQLITE_CHANGESET_OMIT,
4299 ** this function returns SQLITE_OK.
4301 static int sessionConflictHandler(
4302 int eType, /* Either CHANGESET_DATA or CONFLICT */
4303 SessionApplyCtx *p, /* changeset_apply() context */
4304 sqlite3_changeset_iter *pIter, /* Changeset iterator */
4305 int(*xConflict)(void *, int, sqlite3_changeset_iter*),
4306 void *pCtx, /* First argument for conflict handler */
4307 int *pbReplace /* OUT: Set to true if PK row is found */
4309 int res = 0; /* Value returned by conflict handler */
4310 int rc;
4311 int nCol;
4312 int op;
4313 const char *zDummy;
4315 sqlite3changeset_op(pIter, &zDummy, &nCol, &op, 0);
4317 assert( eType==SQLITE_CHANGESET_CONFLICT || eType==SQLITE_CHANGESET_DATA );
4318 assert( SQLITE_CHANGESET_CONFLICT+1==SQLITE_CHANGESET_CONSTRAINT );
4319 assert( SQLITE_CHANGESET_DATA+1==SQLITE_CHANGESET_NOTFOUND );
4321 /* Bind the new.* PRIMARY KEY values to the SELECT statement. */
4322 if( pbReplace ){
4323 rc = sessionSeekToRow(pIter, p->abPK, p->pSelect);
4324 }else{
4325 rc = SQLITE_OK;
4328 if( rc==SQLITE_ROW ){
4329 /* There exists another row with the new.* primary key. */
4330 pIter->pConflict = p->pSelect;
4331 res = xConflict(pCtx, eType, pIter);
4332 pIter->pConflict = 0;
4333 rc = sqlite3_reset(p->pSelect);
4334 }else if( rc==SQLITE_OK ){
4335 if( p->bDeferConstraints && eType==SQLITE_CHANGESET_CONFLICT ){
4336 /* Instead of invoking the conflict handler, append the change blob
4337 ** to the SessionApplyCtx.constraints buffer. */
4338 u8 *aBlob = &pIter->in.aData[pIter->in.iCurrent];
4339 int nBlob = pIter->in.iNext - pIter->in.iCurrent;
4340 sessionAppendBlob(&p->constraints, aBlob, nBlob, &rc);
4341 return SQLITE_OK;
4342 }else{
4343 /* No other row with the new.* primary key. */
4344 res = xConflict(pCtx, eType+1, pIter);
4345 if( res==SQLITE_CHANGESET_REPLACE ) rc = SQLITE_MISUSE;
4349 if( rc==SQLITE_OK ){
4350 switch( res ){
4351 case SQLITE_CHANGESET_REPLACE:
4352 assert( pbReplace );
4353 *pbReplace = 1;
4354 break;
4356 case SQLITE_CHANGESET_OMIT:
4357 break;
4359 case SQLITE_CHANGESET_ABORT:
4360 rc = SQLITE_ABORT;
4361 break;
4363 default:
4364 rc = SQLITE_MISUSE;
4365 break;
4367 if( rc==SQLITE_OK ){
4368 rc = sessionRebaseAdd(p, res, pIter);
4372 return rc;
4376 ** Attempt to apply the change that the iterator passed as the first argument
4377 ** currently points to to the database. If a conflict is encountered, invoke
4378 ** the conflict handler callback.
4380 ** If argument pbRetry is NULL, then ignore any CHANGESET_DATA conflict. If
4381 ** one is encountered, update or delete the row with the matching primary key
4382 ** instead. Or, if pbRetry is not NULL and a CHANGESET_DATA conflict occurs,
4383 ** invoke the conflict handler. If it returns CHANGESET_REPLACE, set *pbRetry
4384 ** to true before returning. In this case the caller will invoke this function
4385 ** again, this time with pbRetry set to NULL.
4387 ** If argument pbReplace is NULL and a CHANGESET_CONFLICT conflict is
4388 ** encountered invoke the conflict handler with CHANGESET_CONSTRAINT instead.
4389 ** Or, if pbReplace is not NULL, invoke it with CHANGESET_CONFLICT. If such
4390 ** an invocation returns SQLITE_CHANGESET_REPLACE, set *pbReplace to true
4391 ** before retrying. In this case the caller attempts to remove the conflicting
4392 ** row before invoking this function again, this time with pbReplace set
4393 ** to NULL.
4395 ** If any conflict handler returns SQLITE_CHANGESET_ABORT, this function
4396 ** returns SQLITE_ABORT. Otherwise, if no error occurs, SQLITE_OK is
4397 ** returned.
4399 static int sessionApplyOneOp(
4400 sqlite3_changeset_iter *pIter, /* Changeset iterator */
4401 SessionApplyCtx *p, /* changeset_apply() context */
4402 int(*xConflict)(void *, int, sqlite3_changeset_iter *),
4403 void *pCtx, /* First argument for the conflict handler */
4404 int *pbReplace, /* OUT: True to remove PK row and retry */
4405 int *pbRetry /* OUT: True to retry. */
4407 const char *zDummy;
4408 int op;
4409 int nCol;
4410 int rc = SQLITE_OK;
4412 assert( p->pDelete && p->pInsert && p->pSelect );
4413 assert( p->azCol && p->abPK );
4414 assert( !pbReplace || *pbReplace==0 );
4416 sqlite3changeset_op(pIter, &zDummy, &nCol, &op, 0);
4418 if( op==SQLITE_DELETE ){
4420 /* Bind values to the DELETE statement. If conflict handling is required,
4421 ** bind values for all columns and set bound variable (nCol+1) to true.
4422 ** Or, if conflict handling is not required, bind just the PK column
4423 ** values and, if it exists, set (nCol+1) to false. Conflict handling
4424 ** is not required if:
4426 ** * this is a patchset, or
4427 ** * (pbRetry==0), or
4428 ** * all columns of the table are PK columns (in this case there is
4429 ** no (nCol+1) variable to bind to).
4431 u8 *abPK = (pIter->bPatchset ? p->abPK : 0);
4432 rc = sessionBindRow(pIter, sqlite3changeset_old, nCol, abPK, p->pDelete);
4433 if( rc==SQLITE_OK && sqlite3_bind_parameter_count(p->pDelete)>nCol ){
4434 rc = sqlite3_bind_int(p->pDelete, nCol+1, (pbRetry==0 || abPK));
4436 if( rc!=SQLITE_OK ) return rc;
4438 sqlite3_step(p->pDelete);
4439 rc = sqlite3_reset(p->pDelete);
4440 if( rc==SQLITE_OK && sqlite3_changes(p->db)==0 ){
4441 rc = sessionConflictHandler(
4442 SQLITE_CHANGESET_DATA, p, pIter, xConflict, pCtx, pbRetry
4444 }else if( (rc&0xff)==SQLITE_CONSTRAINT ){
4445 rc = sessionConflictHandler(
4446 SQLITE_CHANGESET_CONFLICT, p, pIter, xConflict, pCtx, 0
4450 }else if( op==SQLITE_UPDATE ){
4451 int i;
4452 sqlite3_stmt *pUp = 0;
4453 int bPatchset = (pbRetry==0 || pIter->bPatchset);
4455 rc = sessionUpdateFind(pIter, p, bPatchset, &pUp);
4457 /* Bind values to the UPDATE statement. */
4458 for(i=0; rc==SQLITE_OK && i<nCol; i++){
4459 sqlite3_value *pOld = sessionChangesetOld(pIter, i);
4460 sqlite3_value *pNew = sessionChangesetNew(pIter, i);
4461 if( p->abPK[i] || (bPatchset==0 && pOld) ){
4462 rc = sessionBindValue(pUp, i*2+2, pOld);
4464 if( rc==SQLITE_OK && pNew ){
4465 rc = sessionBindValue(pUp, i*2+1, pNew);
4468 if( rc!=SQLITE_OK ) return rc;
4470 /* Attempt the UPDATE. In the case of a NOTFOUND or DATA conflict,
4471 ** the result will be SQLITE_OK with 0 rows modified. */
4472 sqlite3_step(pUp);
4473 rc = sqlite3_reset(pUp);
4475 if( rc==SQLITE_OK && sqlite3_changes(p->db)==0 ){
4476 /* A NOTFOUND or DATA error. Search the table to see if it contains
4477 ** a row with a matching primary key. If so, this is a DATA conflict.
4478 ** Otherwise, if there is no primary key match, it is a NOTFOUND. */
4480 rc = sessionConflictHandler(
4481 SQLITE_CHANGESET_DATA, p, pIter, xConflict, pCtx, pbRetry
4484 }else if( (rc&0xff)==SQLITE_CONSTRAINT ){
4485 /* This is always a CONSTRAINT conflict. */
4486 rc = sessionConflictHandler(
4487 SQLITE_CHANGESET_CONFLICT, p, pIter, xConflict, pCtx, 0
4491 }else{
4492 assert( op==SQLITE_INSERT );
4493 if( p->bStat1 ){
4494 /* Check if there is a conflicting row. For sqlite_stat1, this needs
4495 ** to be done using a SELECT, as there is no PRIMARY KEY in the
4496 ** database schema to throw an exception if a duplicate is inserted. */
4497 rc = sessionSeekToRow(pIter, p->abPK, p->pSelect);
4498 if( rc==SQLITE_ROW ){
4499 rc = SQLITE_CONSTRAINT;
4500 sqlite3_reset(p->pSelect);
4504 if( rc==SQLITE_OK ){
4505 rc = sessionBindRow(pIter, sqlite3changeset_new, nCol, 0, p->pInsert);
4506 if( rc!=SQLITE_OK ) return rc;
4508 sqlite3_step(p->pInsert);
4509 rc = sqlite3_reset(p->pInsert);
4512 if( (rc&0xff)==SQLITE_CONSTRAINT ){
4513 rc = sessionConflictHandler(
4514 SQLITE_CHANGESET_CONFLICT, p, pIter, xConflict, pCtx, pbReplace
4519 return rc;
4523 ** Attempt to apply the change that the iterator passed as the first argument
4524 ** currently points to to the database. If a conflict is encountered, invoke
4525 ** the conflict handler callback.
4527 ** The difference between this function and sessionApplyOne() is that this
4528 ** function handles the case where the conflict-handler is invoked and
4529 ** returns SQLITE_CHANGESET_REPLACE - indicating that the change should be
4530 ** retried in some manner.
4532 static int sessionApplyOneWithRetry(
4533 sqlite3 *db, /* Apply change to "main" db of this handle */
4534 sqlite3_changeset_iter *pIter, /* Changeset iterator to read change from */
4535 SessionApplyCtx *pApply, /* Apply context */
4536 int(*xConflict)(void*, int, sqlite3_changeset_iter*),
4537 void *pCtx /* First argument passed to xConflict */
4539 int bReplace = 0;
4540 int bRetry = 0;
4541 int rc;
4543 rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, &bReplace, &bRetry);
4544 if( rc==SQLITE_OK ){
4545 /* If the bRetry flag is set, the change has not been applied due to an
4546 ** SQLITE_CHANGESET_DATA problem (i.e. this is an UPDATE or DELETE and
4547 ** a row with the correct PK is present in the db, but one or more other
4548 ** fields do not contain the expected values) and the conflict handler
4549 ** returned SQLITE_CHANGESET_REPLACE. In this case retry the operation,
4550 ** but pass NULL as the final argument so that sessionApplyOneOp() ignores
4551 ** the SQLITE_CHANGESET_DATA problem. */
4552 if( bRetry ){
4553 assert( pIter->op==SQLITE_UPDATE || pIter->op==SQLITE_DELETE );
4554 rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, 0, 0);
4557 /* If the bReplace flag is set, the change is an INSERT that has not
4558 ** been performed because the database already contains a row with the
4559 ** specified primary key and the conflict handler returned
4560 ** SQLITE_CHANGESET_REPLACE. In this case remove the conflicting row
4561 ** before reattempting the INSERT. */
4562 else if( bReplace ){
4563 assert( pIter->op==SQLITE_INSERT );
4564 rc = sqlite3_exec(db, "SAVEPOINT replace_op", 0, 0, 0);
4565 if( rc==SQLITE_OK ){
4566 rc = sessionBindRow(pIter,
4567 sqlite3changeset_new, pApply->nCol, pApply->abPK, pApply->pDelete);
4568 sqlite3_bind_int(pApply->pDelete, pApply->nCol+1, 1);
4570 if( rc==SQLITE_OK ){
4571 sqlite3_step(pApply->pDelete);
4572 rc = sqlite3_reset(pApply->pDelete);
4574 if( rc==SQLITE_OK ){
4575 rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, 0, 0);
4577 if( rc==SQLITE_OK ){
4578 rc = sqlite3_exec(db, "RELEASE replace_op", 0, 0, 0);
4583 return rc;
4587 ** Retry the changes accumulated in the pApply->constraints buffer.
4589 static int sessionRetryConstraints(
4590 sqlite3 *db,
4591 int bPatchset,
4592 const char *zTab,
4593 SessionApplyCtx *pApply,
4594 int(*xConflict)(void*, int, sqlite3_changeset_iter*),
4595 void *pCtx /* First argument passed to xConflict */
4597 int rc = SQLITE_OK;
4599 while( pApply->constraints.nBuf ){
4600 sqlite3_changeset_iter *pIter2 = 0;
4601 SessionBuffer cons = pApply->constraints;
4602 memset(&pApply->constraints, 0, sizeof(SessionBuffer));
4604 rc = sessionChangesetStart(
4605 &pIter2, 0, 0, cons.nBuf, cons.aBuf, pApply->bInvertConstraints, 1
4607 if( rc==SQLITE_OK ){
4608 size_t nByte = 2*pApply->nCol*sizeof(sqlite3_value*);
4609 int rc2;
4610 pIter2->bPatchset = bPatchset;
4611 pIter2->zTab = (char*)zTab;
4612 pIter2->nCol = pApply->nCol;
4613 pIter2->abPK = pApply->abPK;
4614 sessionBufferGrow(&pIter2->tblhdr, nByte, &rc);
4615 pIter2->apValue = (sqlite3_value**)pIter2->tblhdr.aBuf;
4616 if( rc==SQLITE_OK ) memset(pIter2->apValue, 0, nByte);
4618 while( rc==SQLITE_OK && SQLITE_ROW==sqlite3changeset_next(pIter2) ){
4619 rc = sessionApplyOneWithRetry(db, pIter2, pApply, xConflict, pCtx);
4622 rc2 = sqlite3changeset_finalize(pIter2);
4623 if( rc==SQLITE_OK ) rc = rc2;
4625 assert( pApply->bDeferConstraints || pApply->constraints.nBuf==0 );
4627 sqlite3_free(cons.aBuf);
4628 if( rc!=SQLITE_OK ) break;
4629 if( pApply->constraints.nBuf>=cons.nBuf ){
4630 /* No progress was made on the last round. */
4631 pApply->bDeferConstraints = 0;
4635 return rc;
4639 ** Argument pIter is a changeset iterator that has been initialized, but
4640 ** not yet passed to sqlite3changeset_next(). This function applies the
4641 ** changeset to the main database attached to handle "db". The supplied
4642 ** conflict handler callback is invoked to resolve any conflicts encountered
4643 ** while applying the change.
4645 static int sessionChangesetApply(
4646 sqlite3 *db, /* Apply change to "main" db of this handle */
4647 sqlite3_changeset_iter *pIter, /* Changeset to apply */
4648 int(*xFilter)(
4649 void *pCtx, /* Copy of sixth arg to _apply() */
4650 const char *zTab /* Table name */
4652 int(*xConflict)(
4653 void *pCtx, /* Copy of fifth arg to _apply() */
4654 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4655 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4657 void *pCtx, /* First argument passed to xConflict */
4658 void **ppRebase, int *pnRebase, /* OUT: Rebase information */
4659 int flags /* SESSION_APPLY_XXX flags */
4661 int schemaMismatch = 0;
4662 int rc = SQLITE_OK; /* Return code */
4663 const char *zTab = 0; /* Name of current table */
4664 int nTab = 0; /* Result of sqlite3Strlen30(zTab) */
4665 SessionApplyCtx sApply; /* changeset_apply() context object */
4666 int bPatchset;
4668 assert( xConflict!=0 );
4670 pIter->in.bNoDiscard = 1;
4671 memset(&sApply, 0, sizeof(sApply));
4672 sApply.bRebase = (ppRebase && pnRebase);
4673 sApply.bInvertConstraints = !!(flags & SQLITE_CHANGESETAPPLY_INVERT);
4674 sqlite3_mutex_enter(sqlite3_db_mutex(db));
4675 if( (flags & SQLITE_CHANGESETAPPLY_NOSAVEPOINT)==0 ){
4676 rc = sqlite3_exec(db, "SAVEPOINT changeset_apply", 0, 0, 0);
4678 if( rc==SQLITE_OK ){
4679 rc = sqlite3_exec(db, "PRAGMA defer_foreign_keys = 1", 0, 0, 0);
4681 while( rc==SQLITE_OK && SQLITE_ROW==sqlite3changeset_next(pIter) ){
4682 int nCol;
4683 int op;
4684 const char *zNew;
4686 sqlite3changeset_op(pIter, &zNew, &nCol, &op, 0);
4688 if( zTab==0 || sqlite3_strnicmp(zNew, zTab, nTab+1) ){
4689 u8 *abPK;
4691 rc = sessionRetryConstraints(
4692 db, pIter->bPatchset, zTab, &sApply, xConflict, pCtx
4694 if( rc!=SQLITE_OK ) break;
4696 sessionUpdateFree(&sApply);
4697 sqlite3_free((char*)sApply.azCol); /* cast works around VC++ bug */
4698 sqlite3_finalize(sApply.pDelete);
4699 sqlite3_finalize(sApply.pInsert);
4700 sqlite3_finalize(sApply.pSelect);
4701 sApply.db = db;
4702 sApply.pDelete = 0;
4703 sApply.pInsert = 0;
4704 sApply.pSelect = 0;
4705 sApply.nCol = 0;
4706 sApply.azCol = 0;
4707 sApply.abPK = 0;
4708 sApply.bStat1 = 0;
4709 sApply.bDeferConstraints = 1;
4710 sApply.bRebaseStarted = 0;
4711 memset(&sApply.constraints, 0, sizeof(SessionBuffer));
4713 /* If an xFilter() callback was specified, invoke it now. If the
4714 ** xFilter callback returns zero, skip this table. If it returns
4715 ** non-zero, proceed. */
4716 schemaMismatch = (xFilter && (0==xFilter(pCtx, zNew)));
4717 if( schemaMismatch ){
4718 zTab = sqlite3_mprintf("%s", zNew);
4719 if( zTab==0 ){
4720 rc = SQLITE_NOMEM;
4721 break;
4723 nTab = (int)strlen(zTab);
4724 sApply.azCol = (const char **)zTab;
4725 }else{
4726 int nMinCol = 0;
4727 int i;
4729 sqlite3changeset_pk(pIter, &abPK, 0);
4730 rc = sessionTableInfo(0,
4731 db, "main", zNew, &sApply.nCol, &zTab, &sApply.azCol, &sApply.abPK
4733 if( rc!=SQLITE_OK ) break;
4734 for(i=0; i<sApply.nCol; i++){
4735 if( sApply.abPK[i] ) nMinCol = i+1;
4738 if( sApply.nCol==0 ){
4739 schemaMismatch = 1;
4740 sqlite3_log(SQLITE_SCHEMA,
4741 "sqlite3changeset_apply(): no such table: %s", zTab
4744 else if( sApply.nCol<nCol ){
4745 schemaMismatch = 1;
4746 sqlite3_log(SQLITE_SCHEMA,
4747 "sqlite3changeset_apply(): table %s has %d columns, "
4748 "expected %d or more",
4749 zTab, sApply.nCol, nCol
4752 else if( nCol<nMinCol || memcmp(sApply.abPK, abPK, nCol)!=0 ){
4753 schemaMismatch = 1;
4754 sqlite3_log(SQLITE_SCHEMA, "sqlite3changeset_apply(): "
4755 "primary key mismatch for table %s", zTab
4758 else{
4759 sApply.nCol = nCol;
4760 if( 0==sqlite3_stricmp(zTab, "sqlite_stat1") ){
4761 if( (rc = sessionStat1Sql(db, &sApply) ) ){
4762 break;
4764 sApply.bStat1 = 1;
4765 }else{
4766 if( (rc = sessionSelectRow(db, zTab, &sApply))
4767 || (rc = sessionDeleteRow(db, zTab, &sApply))
4768 || (rc = sessionInsertRow(db, zTab, &sApply))
4770 break;
4772 sApply.bStat1 = 0;
4775 nTab = sqlite3Strlen30(zTab);
4779 /* If there is a schema mismatch on the current table, proceed to the
4780 ** next change. A log message has already been issued. */
4781 if( schemaMismatch ) continue;
4783 rc = sessionApplyOneWithRetry(db, pIter, &sApply, xConflict, pCtx);
4786 bPatchset = pIter->bPatchset;
4787 if( rc==SQLITE_OK ){
4788 rc = sqlite3changeset_finalize(pIter);
4789 }else{
4790 sqlite3changeset_finalize(pIter);
4793 if( rc==SQLITE_OK ){
4794 rc = sessionRetryConstraints(db, bPatchset, zTab, &sApply, xConflict, pCtx);
4797 if( rc==SQLITE_OK ){
4798 int nFk, notUsed;
4799 sqlite3_db_status(db, SQLITE_DBSTATUS_DEFERRED_FKS, &nFk, &notUsed, 0);
4800 if( nFk!=0 ){
4801 int res = SQLITE_CHANGESET_ABORT;
4802 sqlite3_changeset_iter sIter;
4803 memset(&sIter, 0, sizeof(sIter));
4804 sIter.nCol = nFk;
4805 res = xConflict(pCtx, SQLITE_CHANGESET_FOREIGN_KEY, &sIter);
4806 if( res!=SQLITE_CHANGESET_OMIT ){
4807 rc = SQLITE_CONSTRAINT;
4811 sqlite3_exec(db, "PRAGMA defer_foreign_keys = 0", 0, 0, 0);
4813 if( (flags & SQLITE_CHANGESETAPPLY_NOSAVEPOINT)==0 ){
4814 if( rc==SQLITE_OK ){
4815 rc = sqlite3_exec(db, "RELEASE changeset_apply", 0, 0, 0);
4816 }else{
4817 sqlite3_exec(db, "ROLLBACK TO changeset_apply", 0, 0, 0);
4818 sqlite3_exec(db, "RELEASE changeset_apply", 0, 0, 0);
4822 assert( sApply.bRebase || sApply.rebase.nBuf==0 );
4823 if( rc==SQLITE_OK && bPatchset==0 && sApply.bRebase ){
4824 *ppRebase = (void*)sApply.rebase.aBuf;
4825 *pnRebase = sApply.rebase.nBuf;
4826 sApply.rebase.aBuf = 0;
4828 sessionUpdateFree(&sApply);
4829 sqlite3_finalize(sApply.pInsert);
4830 sqlite3_finalize(sApply.pDelete);
4831 sqlite3_finalize(sApply.pSelect);
4832 sqlite3_free((char*)sApply.azCol); /* cast works around VC++ bug */
4833 sqlite3_free((char*)sApply.constraints.aBuf);
4834 sqlite3_free((char*)sApply.rebase.aBuf);
4835 sqlite3_mutex_leave(sqlite3_db_mutex(db));
4836 return rc;
4840 ** Apply the changeset passed via pChangeset/nChangeset to the main
4841 ** database attached to handle "db".
4843 int sqlite3changeset_apply_v2(
4844 sqlite3 *db, /* Apply change to "main" db of this handle */
4845 int nChangeset, /* Size of changeset in bytes */
4846 void *pChangeset, /* Changeset blob */
4847 int(*xFilter)(
4848 void *pCtx, /* Copy of sixth arg to _apply() */
4849 const char *zTab /* Table name */
4851 int(*xConflict)(
4852 void *pCtx, /* Copy of sixth arg to _apply() */
4853 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4854 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4856 void *pCtx, /* First argument passed to xConflict */
4857 void **ppRebase, int *pnRebase,
4858 int flags
4860 sqlite3_changeset_iter *pIter; /* Iterator to skip through changeset */
4861 int bInv = !!(flags & SQLITE_CHANGESETAPPLY_INVERT);
4862 int rc = sessionChangesetStart(&pIter, 0, 0, nChangeset, pChangeset, bInv, 1);
4863 if( rc==SQLITE_OK ){
4864 rc = sessionChangesetApply(
4865 db, pIter, xFilter, xConflict, pCtx, ppRebase, pnRebase, flags
4868 return rc;
4872 ** Apply the changeset passed via pChangeset/nChangeset to the main database
4873 ** attached to handle "db". Invoke the supplied conflict handler callback
4874 ** to resolve any conflicts encountered while applying the change.
4876 int sqlite3changeset_apply(
4877 sqlite3 *db, /* Apply change to "main" db of this handle */
4878 int nChangeset, /* Size of changeset in bytes */
4879 void *pChangeset, /* Changeset blob */
4880 int(*xFilter)(
4881 void *pCtx, /* Copy of sixth arg to _apply() */
4882 const char *zTab /* Table name */
4884 int(*xConflict)(
4885 void *pCtx, /* Copy of fifth arg to _apply() */
4886 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4887 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4889 void *pCtx /* First argument passed to xConflict */
4891 return sqlite3changeset_apply_v2(
4892 db, nChangeset, pChangeset, xFilter, xConflict, pCtx, 0, 0, 0
4897 ** Apply the changeset passed via xInput/pIn to the main database
4898 ** attached to handle "db". Invoke the supplied conflict handler callback
4899 ** to resolve any conflicts encountered while applying the change.
4901 int sqlite3changeset_apply_v2_strm(
4902 sqlite3 *db, /* Apply change to "main" db of this handle */
4903 int (*xInput)(void *pIn, void *pData, int *pnData), /* Input function */
4904 void *pIn, /* First arg for xInput */
4905 int(*xFilter)(
4906 void *pCtx, /* Copy of sixth arg to _apply() */
4907 const char *zTab /* Table name */
4909 int(*xConflict)(
4910 void *pCtx, /* Copy of sixth arg to _apply() */
4911 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4912 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4914 void *pCtx, /* First argument passed to xConflict */
4915 void **ppRebase, int *pnRebase,
4916 int flags
4918 sqlite3_changeset_iter *pIter; /* Iterator to skip through changeset */
4919 int bInverse = !!(flags & SQLITE_CHANGESETAPPLY_INVERT);
4920 int rc = sessionChangesetStart(&pIter, xInput, pIn, 0, 0, bInverse, 1);
4921 if( rc==SQLITE_OK ){
4922 rc = sessionChangesetApply(
4923 db, pIter, xFilter, xConflict, pCtx, ppRebase, pnRebase, flags
4926 return rc;
4928 int sqlite3changeset_apply_strm(
4929 sqlite3 *db, /* Apply change to "main" db of this handle */
4930 int (*xInput)(void *pIn, void *pData, int *pnData), /* Input function */
4931 void *pIn, /* First arg for xInput */
4932 int(*xFilter)(
4933 void *pCtx, /* Copy of sixth arg to _apply() */
4934 const char *zTab /* Table name */
4936 int(*xConflict)(
4937 void *pCtx, /* Copy of sixth arg to _apply() */
4938 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */
4939 sqlite3_changeset_iter *p /* Handle describing change and conflict */
4941 void *pCtx /* First argument passed to xConflict */
4943 return sqlite3changeset_apply_v2_strm(
4944 db, xInput, pIn, xFilter, xConflict, pCtx, 0, 0, 0
4949 ** sqlite3_changegroup handle.
4951 struct sqlite3_changegroup {
4952 int rc; /* Error code */
4953 int bPatch; /* True to accumulate patchsets */
4954 SessionTable *pList; /* List of tables in current patch */
4958 ** This function is called to merge two changes to the same row together as
4959 ** part of an sqlite3changeset_concat() operation. A new change object is
4960 ** allocated and a pointer to it stored in *ppNew.
4962 static int sessionChangeMerge(
4963 SessionTable *pTab, /* Table structure */
4964 int bRebase, /* True for a rebase hash-table */
4965 int bPatchset, /* True for patchsets */
4966 SessionChange *pExist, /* Existing change */
4967 int op2, /* Second change operation */
4968 int bIndirect, /* True if second change is indirect */
4969 u8 *aRec, /* Second change record */
4970 int nRec, /* Number of bytes in aRec */
4971 SessionChange **ppNew /* OUT: Merged change */
4973 SessionChange *pNew = 0;
4974 int rc = SQLITE_OK;
4976 if( !pExist ){
4977 pNew = (SessionChange *)sqlite3_malloc64(sizeof(SessionChange) + nRec);
4978 if( !pNew ){
4979 return SQLITE_NOMEM;
4981 memset(pNew, 0, sizeof(SessionChange));
4982 pNew->op = op2;
4983 pNew->bIndirect = bIndirect;
4984 pNew->aRecord = (u8*)&pNew[1];
4985 if( bIndirect==0 || bRebase==0 ){
4986 pNew->nRecord = nRec;
4987 memcpy(pNew->aRecord, aRec, nRec);
4988 }else{
4989 int i;
4990 u8 *pIn = aRec;
4991 u8 *pOut = pNew->aRecord;
4992 for(i=0; i<pTab->nCol; i++){
4993 int nIn = sessionSerialLen(pIn);
4994 if( *pIn==0 ){
4995 *pOut++ = 0;
4996 }else if( pTab->abPK[i]==0 ){
4997 *pOut++ = 0xFF;
4998 }else{
4999 memcpy(pOut, pIn, nIn);
5000 pOut += nIn;
5002 pIn += nIn;
5004 pNew->nRecord = pOut - pNew->aRecord;
5006 }else if( bRebase ){
5007 if( pExist->op==SQLITE_DELETE && pExist->bIndirect ){
5008 *ppNew = pExist;
5009 }else{
5010 sqlite3_int64 nByte = nRec + pExist->nRecord + sizeof(SessionChange);
5011 pNew = (SessionChange*)sqlite3_malloc64(nByte);
5012 if( pNew==0 ){
5013 rc = SQLITE_NOMEM;
5014 }else{
5015 int i;
5016 u8 *a1 = pExist->aRecord;
5017 u8 *a2 = aRec;
5018 u8 *pOut;
5020 memset(pNew, 0, nByte);
5021 pNew->bIndirect = bIndirect || pExist->bIndirect;
5022 pNew->op = op2;
5023 pOut = pNew->aRecord = (u8*)&pNew[1];
5025 for(i=0; i<pTab->nCol; i++){
5026 int n1 = sessionSerialLen(a1);
5027 int n2 = sessionSerialLen(a2);
5028 if( *a1==0xFF || (pTab->abPK[i]==0 && bIndirect) ){
5029 *pOut++ = 0xFF;
5030 }else if( *a2==0 ){
5031 memcpy(pOut, a1, n1);
5032 pOut += n1;
5033 }else{
5034 memcpy(pOut, a2, n2);
5035 pOut += n2;
5037 a1 += n1;
5038 a2 += n2;
5040 pNew->nRecord = pOut - pNew->aRecord;
5042 sqlite3_free(pExist);
5044 }else{
5045 int op1 = pExist->op;
5048 ** op1=INSERT, op2=INSERT -> Unsupported. Discard op2.
5049 ** op1=INSERT, op2=UPDATE -> INSERT.
5050 ** op1=INSERT, op2=DELETE -> (none)
5052 ** op1=UPDATE, op2=INSERT -> Unsupported. Discard op2.
5053 ** op1=UPDATE, op2=UPDATE -> UPDATE.
5054 ** op1=UPDATE, op2=DELETE -> DELETE.
5056 ** op1=DELETE, op2=INSERT -> UPDATE.
5057 ** op1=DELETE, op2=UPDATE -> Unsupported. Discard op2.
5058 ** op1=DELETE, op2=DELETE -> Unsupported. Discard op2.
5060 if( (op1==SQLITE_INSERT && op2==SQLITE_INSERT)
5061 || (op1==SQLITE_UPDATE && op2==SQLITE_INSERT)
5062 || (op1==SQLITE_DELETE && op2==SQLITE_UPDATE)
5063 || (op1==SQLITE_DELETE && op2==SQLITE_DELETE)
5065 pNew = pExist;
5066 }else if( op1==SQLITE_INSERT && op2==SQLITE_DELETE ){
5067 sqlite3_free(pExist);
5068 assert( pNew==0 );
5069 }else{
5070 u8 *aExist = pExist->aRecord;
5071 sqlite3_int64 nByte;
5072 u8 *aCsr;
5074 /* Allocate a new SessionChange object. Ensure that the aRecord[]
5075 ** buffer of the new object is large enough to hold any record that
5076 ** may be generated by combining the input records. */
5077 nByte = sizeof(SessionChange) + pExist->nRecord + nRec;
5078 pNew = (SessionChange *)sqlite3_malloc64(nByte);
5079 if( !pNew ){
5080 sqlite3_free(pExist);
5081 return SQLITE_NOMEM;
5083 memset(pNew, 0, sizeof(SessionChange));
5084 pNew->bIndirect = (bIndirect && pExist->bIndirect);
5085 aCsr = pNew->aRecord = (u8 *)&pNew[1];
5087 if( op1==SQLITE_INSERT ){ /* INSERT + UPDATE */
5088 u8 *a1 = aRec;
5089 assert( op2==SQLITE_UPDATE );
5090 pNew->op = SQLITE_INSERT;
5091 if( bPatchset==0 ) sessionSkipRecord(&a1, pTab->nCol);
5092 sessionMergeRecord(&aCsr, pTab->nCol, aExist, a1);
5093 }else if( op1==SQLITE_DELETE ){ /* DELETE + INSERT */
5094 assert( op2==SQLITE_INSERT );
5095 pNew->op = SQLITE_UPDATE;
5096 if( bPatchset ){
5097 memcpy(aCsr, aRec, nRec);
5098 aCsr += nRec;
5099 }else{
5100 if( 0==sessionMergeUpdate(&aCsr, pTab, bPatchset, aExist, 0,aRec,0) ){
5101 sqlite3_free(pNew);
5102 pNew = 0;
5105 }else if( op2==SQLITE_UPDATE ){ /* UPDATE + UPDATE */
5106 u8 *a1 = aExist;
5107 u8 *a2 = aRec;
5108 assert( op1==SQLITE_UPDATE );
5109 if( bPatchset==0 ){
5110 sessionSkipRecord(&a1, pTab->nCol);
5111 sessionSkipRecord(&a2, pTab->nCol);
5113 pNew->op = SQLITE_UPDATE;
5114 if( 0==sessionMergeUpdate(&aCsr, pTab, bPatchset, aRec, aExist,a1,a2) ){
5115 sqlite3_free(pNew);
5116 pNew = 0;
5118 }else{ /* UPDATE + DELETE */
5119 assert( op1==SQLITE_UPDATE && op2==SQLITE_DELETE );
5120 pNew->op = SQLITE_DELETE;
5121 if( bPatchset ){
5122 memcpy(aCsr, aRec, nRec);
5123 aCsr += nRec;
5124 }else{
5125 sessionMergeRecord(&aCsr, pTab->nCol, aRec, aExist);
5129 if( pNew ){
5130 pNew->nRecord = (int)(aCsr - pNew->aRecord);
5132 sqlite3_free(pExist);
5136 *ppNew = pNew;
5137 return rc;
5141 ** Add all changes in the changeset traversed by the iterator passed as
5142 ** the first argument to the changegroup hash tables.
5144 static int sessionChangesetToHash(
5145 sqlite3_changeset_iter *pIter, /* Iterator to read from */
5146 sqlite3_changegroup *pGrp, /* Changegroup object to add changeset to */
5147 int bRebase /* True if hash table is for rebasing */
5149 u8 *aRec;
5150 int nRec;
5151 int rc = SQLITE_OK;
5152 SessionTable *pTab = 0;
5154 while( SQLITE_ROW==sessionChangesetNext(pIter, &aRec, &nRec, 0) ){
5155 const char *zNew;
5156 int nCol;
5157 int op;
5158 int iHash;
5159 int bIndirect;
5160 SessionChange *pChange;
5161 SessionChange *pExist = 0;
5162 SessionChange **pp;
5164 if( pGrp->pList==0 ){
5165 pGrp->bPatch = pIter->bPatchset;
5166 }else if( pIter->bPatchset!=pGrp->bPatch ){
5167 rc = SQLITE_ERROR;
5168 break;
5171 sqlite3changeset_op(pIter, &zNew, &nCol, &op, &bIndirect);
5172 if( !pTab || sqlite3_stricmp(zNew, pTab->zName) ){
5173 /* Search the list for a matching table */
5174 int nNew = (int)strlen(zNew);
5175 u8 *abPK;
5177 sqlite3changeset_pk(pIter, &abPK, 0);
5178 for(pTab = pGrp->pList; pTab; pTab=pTab->pNext){
5179 if( 0==sqlite3_strnicmp(pTab->zName, zNew, nNew+1) ) break;
5181 if( !pTab ){
5182 SessionTable **ppTab;
5184 pTab = sqlite3_malloc64(sizeof(SessionTable) + nCol + nNew+1);
5185 if( !pTab ){
5186 rc = SQLITE_NOMEM;
5187 break;
5189 memset(pTab, 0, sizeof(SessionTable));
5190 pTab->nCol = nCol;
5191 pTab->abPK = (u8*)&pTab[1];
5192 memcpy(pTab->abPK, abPK, nCol);
5193 pTab->zName = (char*)&pTab->abPK[nCol];
5194 memcpy(pTab->zName, zNew, nNew+1);
5196 /* The new object must be linked on to the end of the list, not
5197 ** simply added to the start of it. This is to ensure that the
5198 ** tables within the output of sqlite3changegroup_output() are in
5199 ** the right order. */
5200 for(ppTab=&pGrp->pList; *ppTab; ppTab=&(*ppTab)->pNext);
5201 *ppTab = pTab;
5202 }else if( pTab->nCol!=nCol || memcmp(pTab->abPK, abPK, nCol) ){
5203 rc = SQLITE_SCHEMA;
5204 break;
5208 if( sessionGrowHash(0, pIter->bPatchset, pTab) ){
5209 rc = SQLITE_NOMEM;
5210 break;
5212 iHash = sessionChangeHash(
5213 pTab, (pIter->bPatchset && op==SQLITE_DELETE), aRec, pTab->nChange
5216 /* Search for existing entry. If found, remove it from the hash table.
5217 ** Code below may link it back in.
5219 for(pp=&pTab->apChange[iHash]; *pp; pp=&(*pp)->pNext){
5220 int bPkOnly1 = 0;
5221 int bPkOnly2 = 0;
5222 if( pIter->bPatchset ){
5223 bPkOnly1 = (*pp)->op==SQLITE_DELETE;
5224 bPkOnly2 = op==SQLITE_DELETE;
5226 if( sessionChangeEqual(pTab, bPkOnly1, (*pp)->aRecord, bPkOnly2, aRec) ){
5227 pExist = *pp;
5228 *pp = (*pp)->pNext;
5229 pTab->nEntry--;
5230 break;
5234 rc = sessionChangeMerge(pTab, bRebase,
5235 pIter->bPatchset, pExist, op, bIndirect, aRec, nRec, &pChange
5237 if( rc ) break;
5238 if( pChange ){
5239 pChange->pNext = pTab->apChange[iHash];
5240 pTab->apChange[iHash] = pChange;
5241 pTab->nEntry++;
5245 if( rc==SQLITE_OK ) rc = pIter->rc;
5246 return rc;
5250 ** Serialize a changeset (or patchset) based on all changesets (or patchsets)
5251 ** added to the changegroup object passed as the first argument.
5253 ** If xOutput is not NULL, then the changeset/patchset is returned to the
5254 ** user via one or more calls to xOutput, as with the other streaming
5255 ** interfaces.
5257 ** Or, if xOutput is NULL, then (*ppOut) is populated with a pointer to a
5258 ** buffer containing the output changeset before this function returns. In
5259 ** this case (*pnOut) is set to the size of the output buffer in bytes. It
5260 ** is the responsibility of the caller to free the output buffer using
5261 ** sqlite3_free() when it is no longer required.
5263 ** If successful, SQLITE_OK is returned. Or, if an error occurs, an SQLite
5264 ** error code. If an error occurs and xOutput is NULL, (*ppOut) and (*pnOut)
5265 ** are both set to 0 before returning.
5267 static int sessionChangegroupOutput(
5268 sqlite3_changegroup *pGrp,
5269 int (*xOutput)(void *pOut, const void *pData, int nData),
5270 void *pOut,
5271 int *pnOut,
5272 void **ppOut
5274 int rc = SQLITE_OK;
5275 SessionBuffer buf = {0, 0, 0};
5276 SessionTable *pTab;
5277 assert( xOutput==0 || (ppOut==0 && pnOut==0) );
5279 /* Create the serialized output changeset based on the contents of the
5280 ** hash tables attached to the SessionTable objects in list p->pList.
5282 for(pTab=pGrp->pList; rc==SQLITE_OK && pTab; pTab=pTab->pNext){
5283 int i;
5284 if( pTab->nEntry==0 ) continue;
5286 sessionAppendTableHdr(&buf, pGrp->bPatch, pTab, &rc);
5287 for(i=0; i<pTab->nChange; i++){
5288 SessionChange *p;
5289 for(p=pTab->apChange[i]; p; p=p->pNext){
5290 sessionAppendByte(&buf, p->op, &rc);
5291 sessionAppendByte(&buf, p->bIndirect, &rc);
5292 sessionAppendBlob(&buf, p->aRecord, p->nRecord, &rc);
5293 if( rc==SQLITE_OK && xOutput && buf.nBuf>=sessions_strm_chunk_size ){
5294 rc = xOutput(pOut, buf.aBuf, buf.nBuf);
5295 buf.nBuf = 0;
5301 if( rc==SQLITE_OK ){
5302 if( xOutput ){
5303 if( buf.nBuf>0 ) rc = xOutput(pOut, buf.aBuf, buf.nBuf);
5304 }else if( ppOut ){
5305 *ppOut = buf.aBuf;
5306 if( pnOut ) *pnOut = buf.nBuf;
5307 buf.aBuf = 0;
5310 sqlite3_free(buf.aBuf);
5312 return rc;
5316 ** Allocate a new, empty, sqlite3_changegroup.
5318 int sqlite3changegroup_new(sqlite3_changegroup **pp){
5319 int rc = SQLITE_OK; /* Return code */
5320 sqlite3_changegroup *p; /* New object */
5321 p = (sqlite3_changegroup*)sqlite3_malloc(sizeof(sqlite3_changegroup));
5322 if( p==0 ){
5323 rc = SQLITE_NOMEM;
5324 }else{
5325 memset(p, 0, sizeof(sqlite3_changegroup));
5327 *pp = p;
5328 return rc;
5332 ** Add the changeset currently stored in buffer pData, size nData bytes,
5333 ** to changeset-group p.
5335 int sqlite3changegroup_add(sqlite3_changegroup *pGrp, int nData, void *pData){
5336 sqlite3_changeset_iter *pIter; /* Iterator opened on pData/nData */
5337 int rc; /* Return code */
5339 rc = sqlite3changeset_start(&pIter, nData, pData);
5340 if( rc==SQLITE_OK ){
5341 rc = sessionChangesetToHash(pIter, pGrp, 0);
5343 sqlite3changeset_finalize(pIter);
5344 return rc;
5348 ** Obtain a buffer containing a changeset representing the concatenation
5349 ** of all changesets added to the group so far.
5351 int sqlite3changegroup_output(
5352 sqlite3_changegroup *pGrp,
5353 int *pnData,
5354 void **ppData
5356 return sessionChangegroupOutput(pGrp, 0, 0, pnData, ppData);
5360 ** Streaming versions of changegroup_add().
5362 int sqlite3changegroup_add_strm(
5363 sqlite3_changegroup *pGrp,
5364 int (*xInput)(void *pIn, void *pData, int *pnData),
5365 void *pIn
5367 sqlite3_changeset_iter *pIter; /* Iterator opened on pData/nData */
5368 int rc; /* Return code */
5370 rc = sqlite3changeset_start_strm(&pIter, xInput, pIn);
5371 if( rc==SQLITE_OK ){
5372 rc = sessionChangesetToHash(pIter, pGrp, 0);
5374 sqlite3changeset_finalize(pIter);
5375 return rc;
5379 ** Streaming versions of changegroup_output().
5381 int sqlite3changegroup_output_strm(
5382 sqlite3_changegroup *pGrp,
5383 int (*xOutput)(void *pOut, const void *pData, int nData),
5384 void *pOut
5386 return sessionChangegroupOutput(pGrp, xOutput, pOut, 0, 0);
5390 ** Delete a changegroup object.
5392 void sqlite3changegroup_delete(sqlite3_changegroup *pGrp){
5393 if( pGrp ){
5394 sessionDeleteTable(0, pGrp->pList);
5395 sqlite3_free(pGrp);
5400 ** Combine two changesets together.
5402 int sqlite3changeset_concat(
5403 int nLeft, /* Number of bytes in lhs input */
5404 void *pLeft, /* Lhs input changeset */
5405 int nRight /* Number of bytes in rhs input */,
5406 void *pRight, /* Rhs input changeset */
5407 int *pnOut, /* OUT: Number of bytes in output changeset */
5408 void **ppOut /* OUT: changeset (left <concat> right) */
5410 sqlite3_changegroup *pGrp;
5411 int rc;
5413 rc = sqlite3changegroup_new(&pGrp);
5414 if( rc==SQLITE_OK ){
5415 rc = sqlite3changegroup_add(pGrp, nLeft, pLeft);
5417 if( rc==SQLITE_OK ){
5418 rc = sqlite3changegroup_add(pGrp, nRight, pRight);
5420 if( rc==SQLITE_OK ){
5421 rc = sqlite3changegroup_output(pGrp, pnOut, ppOut);
5423 sqlite3changegroup_delete(pGrp);
5425 return rc;
5429 ** Streaming version of sqlite3changeset_concat().
5431 int sqlite3changeset_concat_strm(
5432 int (*xInputA)(void *pIn, void *pData, int *pnData),
5433 void *pInA,
5434 int (*xInputB)(void *pIn, void *pData, int *pnData),
5435 void *pInB,
5436 int (*xOutput)(void *pOut, const void *pData, int nData),
5437 void *pOut
5439 sqlite3_changegroup *pGrp;
5440 int rc;
5442 rc = sqlite3changegroup_new(&pGrp);
5443 if( rc==SQLITE_OK ){
5444 rc = sqlite3changegroup_add_strm(pGrp, xInputA, pInA);
5446 if( rc==SQLITE_OK ){
5447 rc = sqlite3changegroup_add_strm(pGrp, xInputB, pInB);
5449 if( rc==SQLITE_OK ){
5450 rc = sqlite3changegroup_output_strm(pGrp, xOutput, pOut);
5452 sqlite3changegroup_delete(pGrp);
5454 return rc;
5458 ** Changeset rebaser handle.
5460 struct sqlite3_rebaser {
5461 sqlite3_changegroup grp; /* Hash table */
5465 ** Buffers a1 and a2 must both contain a sessions module record nCol
5466 ** fields in size. This function appends an nCol sessions module
5467 ** record to buffer pBuf that is a copy of a1, except that for
5468 ** each field that is undefined in a1[], swap in the field from a2[].
5470 static void sessionAppendRecordMerge(
5471 SessionBuffer *pBuf, /* Buffer to append to */
5472 int nCol, /* Number of columns in each record */
5473 u8 *a1, int n1, /* Record 1 */
5474 u8 *a2, int n2, /* Record 2 */
5475 int *pRc /* IN/OUT: error code */
5477 sessionBufferGrow(pBuf, n1+n2, pRc);
5478 if( *pRc==SQLITE_OK ){
5479 int i;
5480 u8 *pOut = &pBuf->aBuf[pBuf->nBuf];
5481 for(i=0; i<nCol; i++){
5482 int nn1 = sessionSerialLen(a1);
5483 int nn2 = sessionSerialLen(a2);
5484 if( *a1==0 || *a1==0xFF ){
5485 memcpy(pOut, a2, nn2);
5486 pOut += nn2;
5487 }else{
5488 memcpy(pOut, a1, nn1);
5489 pOut += nn1;
5491 a1 += nn1;
5492 a2 += nn2;
5495 pBuf->nBuf = pOut-pBuf->aBuf;
5496 assert( pBuf->nBuf<=pBuf->nAlloc );
5501 ** This function is called when rebasing a local UPDATE change against one
5502 ** or more remote UPDATE changes. The aRec/nRec buffer contains the current
5503 ** old.* and new.* records for the change. The rebase buffer (a single
5504 ** record) is in aChange/nChange. The rebased change is appended to buffer
5505 ** pBuf.
5507 ** Rebasing the UPDATE involves:
5509 ** * Removing any changes to fields for which the corresponding field
5510 ** in the rebase buffer is set to "replaced" (type 0xFF). If this
5511 ** means the UPDATE change updates no fields, nothing is appended
5512 ** to the output buffer.
5514 ** * For each field modified by the local change for which the
5515 ** corresponding field in the rebase buffer is not "undefined" (0x00)
5516 ** or "replaced" (0xFF), the old.* value is replaced by the value
5517 ** in the rebase buffer.
5519 static void sessionAppendPartialUpdate(
5520 SessionBuffer *pBuf, /* Append record here */
5521 sqlite3_changeset_iter *pIter, /* Iterator pointed at local change */
5522 u8 *aRec, int nRec, /* Local change */
5523 u8 *aChange, int nChange, /* Record to rebase against */
5524 int *pRc /* IN/OUT: Return Code */
5526 sessionBufferGrow(pBuf, 2+nRec+nChange, pRc);
5527 if( *pRc==SQLITE_OK ){
5528 int bData = 0;
5529 u8 *pOut = &pBuf->aBuf[pBuf->nBuf];
5530 int i;
5531 u8 *a1 = aRec;
5532 u8 *a2 = aChange;
5534 *pOut++ = SQLITE_UPDATE;
5535 *pOut++ = pIter->bIndirect;
5536 for(i=0; i<pIter->nCol; i++){
5537 int n1 = sessionSerialLen(a1);
5538 int n2 = sessionSerialLen(a2);
5539 if( pIter->abPK[i] || a2[0]==0 ){
5540 if( !pIter->abPK[i] && a1[0] ) bData = 1;
5541 memcpy(pOut, a1, n1);
5542 pOut += n1;
5543 }else if( a2[0]!=0xFF && a1[0] ){
5544 bData = 1;
5545 memcpy(pOut, a2, n2);
5546 pOut += n2;
5547 }else{
5548 *pOut++ = '\0';
5550 a1 += n1;
5551 a2 += n2;
5553 if( bData ){
5554 a2 = aChange;
5555 for(i=0; i<pIter->nCol; i++){
5556 int n1 = sessionSerialLen(a1);
5557 int n2 = sessionSerialLen(a2);
5558 if( pIter->abPK[i] || a2[0]!=0xFF ){
5559 memcpy(pOut, a1, n1);
5560 pOut += n1;
5561 }else{
5562 *pOut++ = '\0';
5564 a1 += n1;
5565 a2 += n2;
5567 pBuf->nBuf = (pOut - pBuf->aBuf);
5573 ** pIter is configured to iterate through a changeset. This function rebases
5574 ** that changeset according to the current configuration of the rebaser
5575 ** object passed as the first argument. If no error occurs and argument xOutput
5576 ** is not NULL, then the changeset is returned to the caller by invoking
5577 ** xOutput zero or more times and SQLITE_OK returned. Or, if xOutput is NULL,
5578 ** then (*ppOut) is set to point to a buffer containing the rebased changeset
5579 ** before this function returns. In this case (*pnOut) is set to the size of
5580 ** the buffer in bytes. It is the responsibility of the caller to eventually
5581 ** free the (*ppOut) buffer using sqlite3_free().
5583 ** If an error occurs, an SQLite error code is returned. If ppOut and
5584 ** pnOut are not NULL, then the two output parameters are set to 0 before
5585 ** returning.
5587 static int sessionRebase(
5588 sqlite3_rebaser *p, /* Rebaser hash table */
5589 sqlite3_changeset_iter *pIter, /* Input data */
5590 int (*xOutput)(void *pOut, const void *pData, int nData),
5591 void *pOut, /* Context for xOutput callback */
5592 int *pnOut, /* OUT: Number of bytes in output changeset */
5593 void **ppOut /* OUT: Inverse of pChangeset */
5595 int rc = SQLITE_OK;
5596 u8 *aRec = 0;
5597 int nRec = 0;
5598 int bNew = 0;
5599 SessionTable *pTab = 0;
5600 SessionBuffer sOut = {0,0,0};
5602 while( SQLITE_ROW==sessionChangesetNext(pIter, &aRec, &nRec, &bNew) ){
5603 SessionChange *pChange = 0;
5604 int bDone = 0;
5606 if( bNew ){
5607 const char *zTab = pIter->zTab;
5608 for(pTab=p->grp.pList; pTab; pTab=pTab->pNext){
5609 if( 0==sqlite3_stricmp(pTab->zName, zTab) ) break;
5611 bNew = 0;
5613 /* A patchset may not be rebased */
5614 if( pIter->bPatchset ){
5615 rc = SQLITE_ERROR;
5618 /* Append a table header to the output for this new table */
5619 sessionAppendByte(&sOut, pIter->bPatchset ? 'P' : 'T', &rc);
5620 sessionAppendVarint(&sOut, pIter->nCol, &rc);
5621 sessionAppendBlob(&sOut, pIter->abPK, pIter->nCol, &rc);
5622 sessionAppendBlob(&sOut,(u8*)pIter->zTab,(int)strlen(pIter->zTab)+1,&rc);
5625 if( pTab && rc==SQLITE_OK ){
5626 int iHash = sessionChangeHash(pTab, 0, aRec, pTab->nChange);
5628 for(pChange=pTab->apChange[iHash]; pChange; pChange=pChange->pNext){
5629 if( sessionChangeEqual(pTab, 0, aRec, 0, pChange->aRecord) ){
5630 break;
5635 if( pChange ){
5636 assert( pChange->op==SQLITE_DELETE || pChange->op==SQLITE_INSERT );
5637 switch( pIter->op ){
5638 case SQLITE_INSERT:
5639 if( pChange->op==SQLITE_INSERT ){
5640 bDone = 1;
5641 if( pChange->bIndirect==0 ){
5642 sessionAppendByte(&sOut, SQLITE_UPDATE, &rc);
5643 sessionAppendByte(&sOut, pIter->bIndirect, &rc);
5644 sessionAppendBlob(&sOut, pChange->aRecord, pChange->nRecord, &rc);
5645 sessionAppendBlob(&sOut, aRec, nRec, &rc);
5648 break;
5650 case SQLITE_UPDATE:
5651 bDone = 1;
5652 if( pChange->op==SQLITE_DELETE ){
5653 if( pChange->bIndirect==0 ){
5654 u8 *pCsr = aRec;
5655 sessionSkipRecord(&pCsr, pIter->nCol);
5656 sessionAppendByte(&sOut, SQLITE_INSERT, &rc);
5657 sessionAppendByte(&sOut, pIter->bIndirect, &rc);
5658 sessionAppendRecordMerge(&sOut, pIter->nCol,
5659 pCsr, nRec-(pCsr-aRec),
5660 pChange->aRecord, pChange->nRecord, &rc
5663 }else{
5664 sessionAppendPartialUpdate(&sOut, pIter,
5665 aRec, nRec, pChange->aRecord, pChange->nRecord, &rc
5668 break;
5670 default:
5671 assert( pIter->op==SQLITE_DELETE );
5672 bDone = 1;
5673 if( pChange->op==SQLITE_INSERT ){
5674 sessionAppendByte(&sOut, SQLITE_DELETE, &rc);
5675 sessionAppendByte(&sOut, pIter->bIndirect, &rc);
5676 sessionAppendRecordMerge(&sOut, pIter->nCol,
5677 pChange->aRecord, pChange->nRecord, aRec, nRec, &rc
5680 break;
5684 if( bDone==0 ){
5685 sessionAppendByte(&sOut, pIter->op, &rc);
5686 sessionAppendByte(&sOut, pIter->bIndirect, &rc);
5687 sessionAppendBlob(&sOut, aRec, nRec, &rc);
5689 if( rc==SQLITE_OK && xOutput && sOut.nBuf>sessions_strm_chunk_size ){
5690 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf);
5691 sOut.nBuf = 0;
5693 if( rc ) break;
5696 if( rc!=SQLITE_OK ){
5697 sqlite3_free(sOut.aBuf);
5698 memset(&sOut, 0, sizeof(sOut));
5701 if( rc==SQLITE_OK ){
5702 if( xOutput ){
5703 if( sOut.nBuf>0 ){
5704 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf);
5706 }else if( ppOut ){
5707 *ppOut = (void*)sOut.aBuf;
5708 *pnOut = sOut.nBuf;
5709 sOut.aBuf = 0;
5712 sqlite3_free(sOut.aBuf);
5713 return rc;
5717 ** Create a new rebaser object.
5719 int sqlite3rebaser_create(sqlite3_rebaser **ppNew){
5720 int rc = SQLITE_OK;
5721 sqlite3_rebaser *pNew;
5723 pNew = sqlite3_malloc(sizeof(sqlite3_rebaser));
5724 if( pNew==0 ){
5725 rc = SQLITE_NOMEM;
5726 }else{
5727 memset(pNew, 0, sizeof(sqlite3_rebaser));
5729 *ppNew = pNew;
5730 return rc;
5734 ** Call this one or more times to configure a rebaser.
5736 int sqlite3rebaser_configure(
5737 sqlite3_rebaser *p,
5738 int nRebase, const void *pRebase
5740 sqlite3_changeset_iter *pIter = 0; /* Iterator opened on pData/nData */
5741 int rc; /* Return code */
5742 rc = sqlite3changeset_start(&pIter, nRebase, (void*)pRebase);
5743 if( rc==SQLITE_OK ){
5744 rc = sessionChangesetToHash(pIter, &p->grp, 1);
5746 sqlite3changeset_finalize(pIter);
5747 return rc;
5751 ** Rebase a changeset according to current rebaser configuration
5753 int sqlite3rebaser_rebase(
5754 sqlite3_rebaser *p,
5755 int nIn, const void *pIn,
5756 int *pnOut, void **ppOut
5758 sqlite3_changeset_iter *pIter = 0; /* Iterator to skip through input */
5759 int rc = sqlite3changeset_start(&pIter, nIn, (void*)pIn);
5761 if( rc==SQLITE_OK ){
5762 rc = sessionRebase(p, pIter, 0, 0, pnOut, ppOut);
5763 sqlite3changeset_finalize(pIter);
5766 return rc;
5770 ** Rebase a changeset according to current rebaser configuration
5772 int sqlite3rebaser_rebase_strm(
5773 sqlite3_rebaser *p,
5774 int (*xInput)(void *pIn, void *pData, int *pnData),
5775 void *pIn,
5776 int (*xOutput)(void *pOut, const void *pData, int nData),
5777 void *pOut
5779 sqlite3_changeset_iter *pIter = 0; /* Iterator to skip through input */
5780 int rc = sqlite3changeset_start_strm(&pIter, xInput, pIn);
5782 if( rc==SQLITE_OK ){
5783 rc = sessionRebase(p, pIter, xOutput, pOut, 0, 0);
5784 sqlite3changeset_finalize(pIter);
5787 return rc;
5791 ** Destroy a rebaser object
5793 void sqlite3rebaser_delete(sqlite3_rebaser *p){
5794 if( p ){
5795 sessionDeleteTable(0, p->grp.pList);
5796 sqlite3_free(p);
5801 ** Global configuration
5803 int sqlite3session_config(int op, void *pArg){
5804 int rc = SQLITE_OK;
5805 switch( op ){
5806 case SQLITE_SESSION_CONFIG_STRMSIZE: {
5807 int *pInt = (int*)pArg;
5808 if( *pInt>0 ){
5809 sessions_strm_chunk_size = *pInt;
5811 *pInt = sessions_strm_chunk_size;
5812 break;
5814 default:
5815 rc = SQLITE_MISUSE;
5816 break;
5818 return rc;
5821 #endif /* SQLITE_ENABLE_SESSION && SQLITE_ENABLE_PREUPDATE_HOOK */