Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(361)

Side by Side Diff: third_party/sqlite/src/ext/session/sqlite3session.c

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

Powered by Google App Engine
This is Rietveld 408576698