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