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