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, size_t nByte, int *pRc){ 1978 if( *pRc==SQLITE_OK && (size_t)(p->nAlloc-p->nBuf)<nByte ){ 1979 u8 *aNew; 1980 i64 nNew = p->nAlloc ? p->nAlloc : 128; 1981 do { 1982 nNew = nNew*2; 1983 }while( (size_t)(nNew-p->nBuf)<nByte ); 1984 1985 aNew = (u8 *)sqlite3_realloc64(p->aBuf, nNew); 1986 if( 0==aNew ){ 1987 *pRc = SQLITE_NOMEM; 1988 }else{ 1989 p->aBuf = aNew; 1990 p->nAlloc = nNew; 1991 } 1992 } 1993 return (*pRc!=SQLITE_OK); 1994 } 1995 1996 /* 1997 ** Append the value passed as the second argument to the buffer passed 1998 ** as the first. 1999 ** 2000 ** This function is a no-op if *pRc is non-zero when it is called. 2001 ** Otherwise, if an error occurs, *pRc is set to an SQLite error code 2002 ** before returning. 2003 */ 2004 static void sessionAppendValue(SessionBuffer *p, sqlite3_value *pVal, int *pRc){ 2005 int rc = *pRc; 2006 if( rc==SQLITE_OK ){ 2007 sqlite3_int64 nByte = 0; 2008 rc = sessionSerializeValue(0, pVal, &nByte); 2009 sessionBufferGrow(p, nByte, &rc); 2010 if( rc==SQLITE_OK ){ 2011 rc = sessionSerializeValue(&p->aBuf[p->nBuf], pVal, 0); 2012 p->nBuf += nByte; 2013 }else{ 2014 *pRc = rc; 2015 } 2016 } 2017 } 2018 2019 /* 2020 ** This function is a no-op if *pRc is other than SQLITE_OK when it is 2021 ** called. Otherwise, append a single byte to the buffer. 2022 ** 2023 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before 2024 ** returning. 2025 */ 2026 static void sessionAppendByte(SessionBuffer *p, u8 v, int *pRc){ 2027 if( 0==sessionBufferGrow(p, 1, pRc) ){ 2028 p->aBuf[p->nBuf++] = v; 2029 } 2030 } 2031 2032 /* 2033 ** This function is a no-op if *pRc is other than SQLITE_OK when it is 2034 ** called. Otherwise, append a single varint to the buffer. 2035 ** 2036 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before 2037 ** returning. 2038 */ 2039 static void sessionAppendVarint(SessionBuffer *p, int v, int *pRc){ 2040 if( 0==sessionBufferGrow(p, 9, pRc) ){ 2041 p->nBuf += sessionVarintPut(&p->aBuf[p->nBuf], v); 2042 } 2043 } 2044 2045 /* 2046 ** This function is a no-op if *pRc is other than SQLITE_OK when it is 2047 ** called. Otherwise, append a blob of data to the buffer. 2048 ** 2049 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before 2050 ** returning. 2051 */ 2052 static void sessionAppendBlob( 2053 SessionBuffer *p, 2054 const u8 *aBlob, 2055 int nBlob, 2056 int *pRc 2057 ){ 2058 if( nBlob>0 && 0==sessionBufferGrow(p, nBlob, pRc) ){ 2059 memcpy(&p->aBuf[p->nBuf], aBlob, nBlob); 2060 p->nBuf += nBlob; 2061 } 2062 } 2063 2064 /* 2065 ** This function is a no-op if *pRc is other than SQLITE_OK when it is 2066 ** called. Otherwise, append a string to the buffer. All bytes in the string 2067 ** up to (but not including) the nul-terminator are written to the buffer. 2068 ** 2069 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before 2070 ** returning. 2071 */ 2072 static void sessionAppendStr( 2073 SessionBuffer *p, 2074 const char *zStr, 2075 int *pRc 2076 ){ 2077 int nStr = sqlite3Strlen30(zStr); 2078 if( 0==sessionBufferGrow(p, nStr, pRc) ){ 2079 memcpy(&p->aBuf[p->nBuf], zStr, nStr); 2080 p->nBuf += nStr; 2081 } 2082 } 2083 2084 /* 2085 ** This function is a no-op if *pRc is other than SQLITE_OK when it is 2086 ** called. Otherwise, append the string representation of integer iVal 2087 ** to the buffer. No nul-terminator is written. 2088 ** 2089 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before 2090 ** returning. 2091 */ 2092 static void sessionAppendInteger( 2093 SessionBuffer *p, /* Buffer to append to */ 2094 int iVal, /* Value to write the string rep. of */ 2095 int *pRc /* IN/OUT: Error code */ 2096 ){ 2097 char aBuf[24]; 2098 sqlite3_snprintf(sizeof(aBuf)-1, aBuf, "%d", iVal); 2099 sessionAppendStr(p, aBuf, pRc); 2100 } 2101 2102 /* 2103 ** This function is a no-op if *pRc is other than SQLITE_OK when it is 2104 ** called. Otherwise, append the string zStr enclosed in quotes (") and 2105 ** with any embedded quote characters escaped to the buffer. No 2106 ** nul-terminator byte is written. 2107 ** 2108 ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before 2109 ** returning. 2110 */ 2111 static void sessionAppendIdent( 2112 SessionBuffer *p, /* Buffer to a append to */ 2113 const char *zStr, /* String to quote, escape and append */ 2114 int *pRc /* IN/OUT: Error code */ 2115 ){ 2116 int nStr = sqlite3Strlen30(zStr)*2 + 2 + 1; 2117 if( 0==sessionBufferGrow(p, nStr, pRc) ){ 2118 char *zOut = (char *)&p->aBuf[p->nBuf]; 2119 const char *zIn = zStr; 2120 *zOut++ = '"'; 2121 while( *zIn ){ 2122 if( *zIn=='"' ) *zOut++ = '"'; 2123 *zOut++ = *(zIn++); 2124 } 2125 *zOut++ = '"'; 2126 p->nBuf = (int)((u8 *)zOut - p->aBuf); 2127 } 2128 } 2129 2130 /* 2131 ** This function is a no-op if *pRc is other than SQLITE_OK when it is 2132 ** called. Otherwse, it appends the serialized version of the value stored 2133 ** in column iCol of the row that SQL statement pStmt currently points 2134 ** to to the buffer. 2135 */ 2136 static void sessionAppendCol( 2137 SessionBuffer *p, /* Buffer to append to */ 2138 sqlite3_stmt *pStmt, /* Handle pointing to row containing value */ 2139 int iCol, /* Column to read value from */ 2140 int *pRc /* IN/OUT: Error code */ 2141 ){ 2142 if( *pRc==SQLITE_OK ){ 2143 int eType = sqlite3_column_type(pStmt, iCol); 2144 sessionAppendByte(p, (u8)eType, pRc); 2145 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){ 2146 sqlite3_int64 i; 2147 u8 aBuf[8]; 2148 if( eType==SQLITE_INTEGER ){ 2149 i = sqlite3_column_int64(pStmt, iCol); 2150 }else{ 2151 double r = sqlite3_column_double(pStmt, iCol); 2152 memcpy(&i, &r, 8); 2153 } 2154 sessionPutI64(aBuf, i); 2155 sessionAppendBlob(p, aBuf, 8, pRc); 2156 } 2157 if( eType==SQLITE_BLOB || eType==SQLITE_TEXT ){ 2158 u8 *z; 2159 int nByte; 2160 if( eType==SQLITE_BLOB ){ 2161 z = (u8 *)sqlite3_column_blob(pStmt, iCol); 2162 }else{ 2163 z = (u8 *)sqlite3_column_text(pStmt, iCol); 2164 } 2165 nByte = sqlite3_column_bytes(pStmt, iCol); 2166 if( z || (eType==SQLITE_BLOB && nByte==0) ){ 2167 sessionAppendVarint(p, nByte, pRc); 2168 sessionAppendBlob(p, z, nByte, pRc); 2169 }else{ 2170 *pRc = SQLITE_NOMEM; 2171 } 2172 } 2173 } 2174 } 2175 2176 /* 2177 ** 2178 ** This function appends an update change to the buffer (see the comments 2179 ** under "CHANGESET FORMAT" at the top of the file). An update change 2180 ** consists of: 2181 ** 2182 ** 1 byte: SQLITE_UPDATE (0x17) 2183 ** n bytes: old.* record (see RECORD FORMAT) 2184 ** m bytes: new.* record (see RECORD FORMAT) 2185 ** 2186 ** The SessionChange object passed as the third argument contains the 2187 ** values that were stored in the row when the session began (the old.* 2188 ** values). The statement handle passed as the second argument points 2189 ** at the current version of the row (the new.* values). 2190 ** 2191 ** If all of the old.* values are equal to their corresponding new.* value 2192 ** (i.e. nothing has changed), then no data at all is appended to the buffer. 2193 ** 2194 ** Otherwise, the old.* record contains all primary key values and the 2195 ** original values of any fields that have been modified. The new.* record 2196 ** contains the new values of only those fields that have been modified. 2197 */ 2198 static int sessionAppendUpdate( 2199 SessionBuffer *pBuf, /* Buffer to append to */ 2200 int bPatchset, /* True for "patchset", 0 for "changeset" */ 2201 sqlite3_stmt *pStmt, /* Statement handle pointing at new row */ 2202 SessionChange *p, /* Object containing old values */ 2203 u8 *abPK /* Boolean array - true for PK columns */ 2204 ){ 2205 int rc = SQLITE_OK; 2206 SessionBuffer buf2 = {0,0,0}; /* Buffer to accumulate new.* record in */ 2207 int bNoop = 1; /* Set to zero if any values are modified */ 2208 int nRewind = pBuf->nBuf; /* Set to zero if any values are modified */ 2209 int i; /* Used to iterate through columns */ 2210 u8 *pCsr = p->aRecord; /* Used to iterate through old.* values */ 2211 2212 sessionAppendByte(pBuf, SQLITE_UPDATE, &rc); 2213 sessionAppendByte(pBuf, p->bIndirect, &rc); 2214 for(i=0; i<sqlite3_column_count(pStmt); i++){ 2215 int bChanged = 0; 2216 int nAdvance; 2217 int eType = *pCsr; 2218 switch( eType ){ 2219 case SQLITE_NULL: 2220 nAdvance = 1; 2221 if( sqlite3_column_type(pStmt, i)!=SQLITE_NULL ){ 2222 bChanged = 1; 2223 } 2224 break; 2225 2226 case SQLITE_FLOAT: 2227 case SQLITE_INTEGER: { 2228 nAdvance = 9; 2229 if( eType==sqlite3_column_type(pStmt, i) ){ 2230 sqlite3_int64 iVal = sessionGetI64(&pCsr[1]); 2231 if( eType==SQLITE_INTEGER ){ 2232 if( iVal==sqlite3_column_int64(pStmt, i) ) break; 2233 }else{ 2234 double dVal; 2235 memcpy(&dVal, &iVal, 8); 2236 if( dVal==sqlite3_column_double(pStmt, i) ) break; 2237 } 2238 } 2239 bChanged = 1; 2240 break; 2241 } 2242 2243 default: { 2244 int n; 2245 int nHdr = 1 + sessionVarintGet(&pCsr[1], &n); 2246 assert( eType==SQLITE_TEXT || eType==SQLITE_BLOB ); 2247 nAdvance = nHdr + n; 2248 if( eType==sqlite3_column_type(pStmt, i) 2249 && n==sqlite3_column_bytes(pStmt, i) 2250 && (n==0 || 0==memcmp(&pCsr[nHdr], sqlite3_column_blob(pStmt, i), n)) 2251 ){ 2252 break; 2253 } 2254 bChanged = 1; 2255 } 2256 } 2257 2258 /* If at least one field has been modified, this is not a no-op. */ 2259 if( bChanged ) bNoop = 0; 2260 2261 /* Add a field to the old.* record. This is omitted if this modules is 2262 ** currently generating a patchset. */ 2263 if( bPatchset==0 ){ 2264 if( bChanged || abPK[i] ){ 2265 sessionAppendBlob(pBuf, pCsr, nAdvance, &rc); 2266 }else{ 2267 sessionAppendByte(pBuf, 0, &rc); 2268 } 2269 } 2270 2271 /* Add a field to the new.* record. Or the only record if currently 2272 ** generating a patchset. */ 2273 if( bChanged || (bPatchset && abPK[i]) ){ 2274 sessionAppendCol(&buf2, pStmt, i, &rc); 2275 }else{ 2276 sessionAppendByte(&buf2, 0, &rc); 2277 } 2278 2279 pCsr += nAdvance; 2280 } 2281 2282 if( bNoop ){ 2283 pBuf->nBuf = nRewind; 2284 }else{ 2285 sessionAppendBlob(pBuf, buf2.aBuf, buf2.nBuf, &rc); 2286 } 2287 sqlite3_free(buf2.aBuf); 2288 2289 return rc; 2290 } 2291 2292 /* 2293 ** Append a DELETE change to the buffer passed as the first argument. Use 2294 ** the changeset format if argument bPatchset is zero, or the patchset 2295 ** format otherwise. 2296 */ 2297 static int sessionAppendDelete( 2298 SessionBuffer *pBuf, /* Buffer to append to */ 2299 int bPatchset, /* True for "patchset", 0 for "changeset" */ 2300 SessionChange *p, /* Object containing old values */ 2301 int nCol, /* Number of columns in table */ 2302 u8 *abPK /* Boolean array - true for PK columns */ 2303 ){ 2304 int rc = SQLITE_OK; 2305 2306 sessionAppendByte(pBuf, SQLITE_DELETE, &rc); 2307 sessionAppendByte(pBuf, p->bIndirect, &rc); 2308 2309 if( bPatchset==0 ){ 2310 sessionAppendBlob(pBuf, p->aRecord, p->nRecord, &rc); 2311 }else{ 2312 int i; 2313 u8 *a = p->aRecord; 2314 for(i=0; i<nCol; i++){ 2315 u8 *pStart = a; 2316 int eType = *a++; 2317 2318 switch( eType ){ 2319 case 0: 2320 case SQLITE_NULL: 2321 assert( abPK[i]==0 ); 2322 break; 2323 2324 case SQLITE_FLOAT: 2325 case SQLITE_INTEGER: 2326 a += 8; 2327 break; 2328 2329 default: { 2330 int n; 2331 a += sessionVarintGet(a, &n); 2332 a += n; 2333 break; 2334 } 2335 } 2336 if( abPK[i] ){ 2337 sessionAppendBlob(pBuf, pStart, (int)(a-pStart), &rc); 2338 } 2339 } 2340 assert( (a - p->aRecord)==p->nRecord ); 2341 } 2342 2343 return rc; 2344 } 2345 2346 /* 2347 ** Formulate and prepare a SELECT statement to retrieve a row from table 2348 ** zTab in database zDb based on its primary key. i.e. 2349 ** 2350 ** SELECT * FROM zDb.zTab WHERE pk1 = ? AND pk2 = ? AND ... 2351 */ 2352 static int sessionSelectStmt( 2353 sqlite3 *db, /* Database handle */ 2354 const char *zDb, /* Database name */ 2355 const char *zTab, /* Table name */ 2356 int nCol, /* Number of columns in table */ 2357 const char **azCol, /* Names of table columns */ 2358 u8 *abPK, /* PRIMARY KEY array */ 2359 sqlite3_stmt **ppStmt /* OUT: Prepared SELECT statement */ 2360 ){ 2361 int rc = SQLITE_OK; 2362 char *zSql = 0; 2363 int nSql = -1; 2364 2365 if( 0==sqlite3_stricmp("sqlite_stat1", zTab) ){ 2366 zSql = sqlite3_mprintf( 2367 "SELECT tbl, ?2, stat FROM %Q.sqlite_stat1 WHERE tbl IS ?1 AND " 2368 "idx IS (CASE WHEN ?2=X'' THEN NULL ELSE ?2 END)", zDb 2369 ); 2370 if( zSql==0 ) rc = SQLITE_NOMEM; 2371 }else{ 2372 int i; 2373 const char *zSep = ""; 2374 SessionBuffer buf = {0, 0, 0}; 2375 2376 sessionAppendStr(&buf, "SELECT * FROM ", &rc); 2377 sessionAppendIdent(&buf, zDb, &rc); 2378 sessionAppendStr(&buf, ".", &rc); 2379 sessionAppendIdent(&buf, zTab, &rc); 2380 sessionAppendStr(&buf, " WHERE ", &rc); 2381 for(i=0; i<nCol; i++){ 2382 if( abPK[i] ){ 2383 sessionAppendStr(&buf, zSep, &rc); 2384 sessionAppendIdent(&buf, azCol[i], &rc); 2385 sessionAppendStr(&buf, " IS ?", &rc); 2386 sessionAppendInteger(&buf, i+1, &rc); 2387 zSep = " AND "; 2388 } 2389 } 2390 zSql = (char*)buf.aBuf; 2391 nSql = buf.nBuf; 2392 } 2393 2394 if( rc==SQLITE_OK ){ 2395 rc = sqlite3_prepare_v2(db, zSql, nSql, ppStmt, 0); 2396 } 2397 sqlite3_free(zSql); 2398 return rc; 2399 } 2400 2401 /* 2402 ** Bind the PRIMARY KEY values from the change passed in argument pChange 2403 ** to the SELECT statement passed as the first argument. The SELECT statement 2404 ** is as prepared by function sessionSelectStmt(). 2405 ** 2406 ** Return SQLITE_OK if all PK values are successfully bound, or an SQLite 2407 ** error code (e.g. SQLITE_NOMEM) otherwise. 2408 */ 2409 static int sessionSelectBind( 2410 sqlite3_stmt *pSelect, /* SELECT from sessionSelectStmt() */ 2411 int nCol, /* Number of columns in table */ 2412 u8 *abPK, /* PRIMARY KEY array */ 2413 SessionChange *pChange /* Change structure */ 2414 ){ 2415 int i; 2416 int rc = SQLITE_OK; 2417 u8 *a = pChange->aRecord; 2418 2419 for(i=0; i<nCol && rc==SQLITE_OK; i++){ 2420 int eType = *a++; 2421 2422 switch( eType ){ 2423 case 0: 2424 case SQLITE_NULL: 2425 assert( abPK[i]==0 ); 2426 break; 2427 2428 case SQLITE_INTEGER: { 2429 if( abPK[i] ){ 2430 i64 iVal = sessionGetI64(a); 2431 rc = sqlite3_bind_int64(pSelect, i+1, iVal); 2432 } 2433 a += 8; 2434 break; 2435 } 2436 2437 case SQLITE_FLOAT: { 2438 if( abPK[i] ){ 2439 double rVal; 2440 i64 iVal = sessionGetI64(a); 2441 memcpy(&rVal, &iVal, 8); 2442 rc = sqlite3_bind_double(pSelect, i+1, rVal); 2443 } 2444 a += 8; 2445 break; 2446 } 2447 2448 case SQLITE_TEXT: { 2449 int n; 2450 a += sessionVarintGet(a, &n); 2451 if( abPK[i] ){ 2452 rc = sqlite3_bind_text(pSelect, i+1, (char *)a, n, SQLITE_TRANSIENT); 2453 } 2454 a += n; 2455 break; 2456 } 2457 2458 default: { 2459 int n; 2460 assert( eType==SQLITE_BLOB ); 2461 a += sessionVarintGet(a, &n); 2462 if( abPK[i] ){ 2463 rc = sqlite3_bind_blob(pSelect, i+1, a, n, SQLITE_TRANSIENT); 2464 } 2465 a += n; 2466 break; 2467 } 2468 } 2469 } 2470 2471 return rc; 2472 } 2473 2474 /* 2475 ** This function is a no-op if *pRc is set to other than SQLITE_OK when it 2476 ** is called. Otherwise, append a serialized table header (part of the binary 2477 ** changeset format) to buffer *pBuf. If an error occurs, set *pRc to an 2478 ** SQLite error code before returning. 2479 */ 2480 static void sessionAppendTableHdr( 2481 SessionBuffer *pBuf, /* Append header to this buffer */ 2482 int bPatchset, /* Use the patchset format if true */ 2483 SessionTable *pTab, /* Table object to append header for */ 2484 int *pRc /* IN/OUT: Error code */ 2485 ){ 2486 /* Write a table header */ 2487 sessionAppendByte(pBuf, (bPatchset ? 'P' : 'T'), pRc); 2488 sessionAppendVarint(pBuf, pTab->nCol, pRc); 2489 sessionAppendBlob(pBuf, pTab->abPK, pTab->nCol, pRc); 2490 sessionAppendBlob(pBuf, (u8 *)pTab->zName, (int)strlen(pTab->zName)+1, pRc); 2491 } 2492 2493 /* 2494 ** Generate either a changeset (if argument bPatchset is zero) or a patchset 2495 ** (if it is non-zero) based on the current contents of the session object 2496 ** passed as the first argument. 2497 ** 2498 ** If no error occurs, SQLITE_OK is returned and the new changeset/patchset 2499 ** stored in output variables *pnChangeset and *ppChangeset. Or, if an error 2500 ** occurs, an SQLite error code is returned and both output variables set 2501 ** to 0. 2502 */ 2503 static int sessionGenerateChangeset( 2504 sqlite3_session *pSession, /* Session object */ 2505 int bPatchset, /* True for patchset, false for changeset */ 2506 int (*xOutput)(void *pOut, const void *pData, int nData), 2507 void *pOut, /* First argument for xOutput */ 2508 int *pnChangeset, /* OUT: Size of buffer at *ppChangeset */ 2509 void **ppChangeset /* OUT: Buffer containing changeset */ 2510 ){ 2511 sqlite3 *db = pSession->db; /* Source database handle */ 2512 SessionTable *pTab; /* Used to iterate through attached tables */ 2513 SessionBuffer buf = {0,0,0}; /* Buffer in which to accumlate changeset */ 2514 int rc; /* Return code */ 2515 2516 assert( xOutput==0 || (pnChangeset==0 && ppChangeset==0 ) ); 2517 2518 /* Zero the output variables in case an error occurs. If this session 2519 ** object is already in the error state (sqlite3_session.rc != SQLITE_OK), 2520 ** this call will be a no-op. */ 2521 if( xOutput==0 ){ 2522 *pnChangeset = 0; 2523 *ppChangeset = 0; 2524 } 2525 2526 if( pSession->rc ) return pSession->rc; 2527 rc = sqlite3_exec(pSession->db, "SAVEPOINT changeset", 0, 0, 0); 2528 if( rc!=SQLITE_OK ) return rc; 2529 2530 sqlite3_mutex_enter(sqlite3_db_mutex(db)); 2531 2532 for(pTab=pSession->pTable; rc==SQLITE_OK && pTab; pTab=pTab->pNext){ 2533 if( pTab->nEntry ){ 2534 const char *zName = pTab->zName; 2535 int nCol; /* Number of columns in table */ 2536 u8 *abPK; /* Primary key array */ 2537 const char **azCol = 0; /* Table columns */ 2538 int i; /* Used to iterate through hash buckets */ 2539 sqlite3_stmt *pSel = 0; /* SELECT statement to query table pTab */ 2540 int nRewind = buf.nBuf; /* Initial size of write buffer */ 2541 int nNoop; /* Size of buffer after writing tbl header */ 2542 2543 /* Check the table schema is still Ok. */ 2544 rc = sessionTableInfo(0, db, pSession->zDb, zName, &nCol, 0,&azCol,&abPK); 2545 if( !rc && (pTab->nCol!=nCol || memcmp(abPK, pTab->abPK, nCol)) ){ 2546 rc = SQLITE_SCHEMA; 2547 } 2548 2549 /* Write a table header */ 2550 sessionAppendTableHdr(&buf, bPatchset, pTab, &rc); 2551 2552 /* Build and compile a statement to execute: */ 2553 if( rc==SQLITE_OK ){ 2554 rc = sessionSelectStmt( 2555 db, pSession->zDb, zName, nCol, azCol, abPK, &pSel); 2556 } 2557 2558 nNoop = buf.nBuf; 2559 for(i=0; i<pTab->nChange && rc==SQLITE_OK; i++){ 2560 SessionChange *p; /* Used to iterate through changes */ 2561 2562 for(p=pTab->apChange[i]; rc==SQLITE_OK && p; p=p->pNext){ 2563 rc = sessionSelectBind(pSel, nCol, abPK, p); 2564 if( rc!=SQLITE_OK ) continue; 2565 if( sqlite3_step(pSel)==SQLITE_ROW ){ 2566 if( p->op==SQLITE_INSERT ){ 2567 int iCol; 2568 sessionAppendByte(&buf, SQLITE_INSERT, &rc); 2569 sessionAppendByte(&buf, p->bIndirect, &rc); 2570 for(iCol=0; iCol<nCol; iCol++){ 2571 sessionAppendCol(&buf, pSel, iCol, &rc); 2572 } 2573 }else{ 2574 rc = sessionAppendUpdate(&buf, bPatchset, pSel, p, abPK); 2575 } 2576 }else if( p->op!=SQLITE_INSERT ){ 2577 rc = sessionAppendDelete(&buf, bPatchset, p, nCol, abPK); 2578 } 2579 if( rc==SQLITE_OK ){ 2580 rc = sqlite3_reset(pSel); 2581 } 2582 2583 /* If the buffer is now larger than sessions_strm_chunk_size, pass 2584 ** its contents to the xOutput() callback. */ 2585 if( xOutput 2586 && rc==SQLITE_OK 2587 && buf.nBuf>nNoop 2588 && buf.nBuf>sessions_strm_chunk_size 2589 ){ 2590 rc = xOutput(pOut, (void*)buf.aBuf, buf.nBuf); 2591 nNoop = -1; 2592 buf.nBuf = 0; 2593 } 2594 2595 } 2596 } 2597 2598 sqlite3_finalize(pSel); 2599 if( buf.nBuf==nNoop ){ 2600 buf.nBuf = nRewind; 2601 } 2602 sqlite3_free((char*)azCol); /* cast works around VC++ bug */ 2603 } 2604 } 2605 2606 if( rc==SQLITE_OK ){ 2607 if( xOutput==0 ){ 2608 *pnChangeset = buf.nBuf; 2609 *ppChangeset = buf.aBuf; 2610 buf.aBuf = 0; 2611 }else if( buf.nBuf>0 ){ 2612 rc = xOutput(pOut, (void*)buf.aBuf, buf.nBuf); 2613 } 2614 } 2615 2616 sqlite3_free(buf.aBuf); 2617 sqlite3_exec(db, "RELEASE changeset", 0, 0, 0); 2618 sqlite3_mutex_leave(sqlite3_db_mutex(db)); 2619 return rc; 2620 } 2621 2622 /* 2623 ** Obtain a changeset object containing all changes recorded by the 2624 ** session object passed as the first argument. 2625 ** 2626 ** It is the responsibility of the caller to eventually free the buffer 2627 ** using sqlite3_free(). 2628 */ 2629 int sqlite3session_changeset( 2630 sqlite3_session *pSession, /* Session object */ 2631 int *pnChangeset, /* OUT: Size of buffer at *ppChangeset */ 2632 void **ppChangeset /* OUT: Buffer containing changeset */ 2633 ){ 2634 int rc = sessionGenerateChangeset(pSession, 0, 0, 0, pnChangeset,ppChangeset); 2635 assert( rc || pnChangeset==0 2636 || pSession->bEnableSize==0 || *pnChangeset<=pSession->nMaxChangesetSize 2637 ); 2638 return rc; 2639 } 2640 2641 /* 2642 ** Streaming version of sqlite3session_changeset(). 2643 */ 2644 int sqlite3session_changeset_strm( 2645 sqlite3_session *pSession, 2646 int (*xOutput)(void *pOut, const void *pData, int nData), 2647 void *pOut 2648 ){ 2649 return sessionGenerateChangeset(pSession, 0, xOutput, pOut, 0, 0); 2650 } 2651 2652 /* 2653 ** Streaming version of sqlite3session_patchset(). 2654 */ 2655 int sqlite3session_patchset_strm( 2656 sqlite3_session *pSession, 2657 int (*xOutput)(void *pOut, const void *pData, int nData), 2658 void *pOut 2659 ){ 2660 return sessionGenerateChangeset(pSession, 1, xOutput, pOut, 0, 0); 2661 } 2662 2663 /* 2664 ** Obtain a patchset object containing all changes recorded by the 2665 ** session object passed as the first argument. 2666 ** 2667 ** It is the responsibility of the caller to eventually free the buffer 2668 ** using sqlite3_free(). 2669 */ 2670 int sqlite3session_patchset( 2671 sqlite3_session *pSession, /* Session object */ 2672 int *pnPatchset, /* OUT: Size of buffer at *ppChangeset */ 2673 void **ppPatchset /* OUT: Buffer containing changeset */ 2674 ){ 2675 return sessionGenerateChangeset(pSession, 1, 0, 0, pnPatchset, ppPatchset); 2676 } 2677 2678 /* 2679 ** Enable or disable the session object passed as the first argument. 2680 */ 2681 int sqlite3session_enable(sqlite3_session *pSession, int bEnable){ 2682 int ret; 2683 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db)); 2684 if( bEnable>=0 ){ 2685 pSession->bEnable = bEnable; 2686 } 2687 ret = pSession->bEnable; 2688 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db)); 2689 return ret; 2690 } 2691 2692 /* 2693 ** Enable or disable the session object passed as the first argument. 2694 */ 2695 int sqlite3session_indirect(sqlite3_session *pSession, int bIndirect){ 2696 int ret; 2697 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db)); 2698 if( bIndirect>=0 ){ 2699 pSession->bIndirect = bIndirect; 2700 } 2701 ret = pSession->bIndirect; 2702 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db)); 2703 return ret; 2704 } 2705 2706 /* 2707 ** Return true if there have been no changes to monitored tables recorded 2708 ** by the session object passed as the only argument. 2709 */ 2710 int sqlite3session_isempty(sqlite3_session *pSession){ 2711 int ret = 0; 2712 SessionTable *pTab; 2713 2714 sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db)); 2715 for(pTab=pSession->pTable; pTab && ret==0; pTab=pTab->pNext){ 2716 ret = (pTab->nEntry>0); 2717 } 2718 sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db)); 2719 2720 return (ret==0); 2721 } 2722 2723 /* 2724 ** Return the amount of heap memory in use. 2725 */ 2726 sqlite3_int64 sqlite3session_memory_used(sqlite3_session *pSession){ 2727 return pSession->nMalloc; 2728 } 2729 2730 /* 2731 ** Configure the session object passed as the first argument. 2732 */ 2733 int sqlite3session_object_config(sqlite3_session *pSession, int op, void *pArg){ 2734 int rc = SQLITE_OK; 2735 switch( op ){ 2736 case SQLITE_SESSION_OBJCONFIG_SIZE: { 2737 int iArg = *(int*)pArg; 2738 if( iArg>=0 ){ 2739 if( pSession->pTable ){ 2740 rc = SQLITE_MISUSE; 2741 }else{ 2742 pSession->bEnableSize = (iArg!=0); 2743 } 2744 } 2745 *(int*)pArg = pSession->bEnableSize; 2746 break; 2747 } 2748 2749 default: 2750 rc = SQLITE_MISUSE; 2751 } 2752 2753 return rc; 2754 } 2755 2756 /* 2757 ** Return the maximum size of sqlite3session_changeset() output. 2758 */ 2759 sqlite3_int64 sqlite3session_changeset_size(sqlite3_session *pSession){ 2760 return pSession->nMaxChangesetSize; 2761 } 2762 2763 /* 2764 ** Do the work for either sqlite3changeset_start() or start_strm(). 2765 */ 2766 static int sessionChangesetStart( 2767 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */ 2768 int (*xInput)(void *pIn, void *pData, int *pnData), 2769 void *pIn, 2770 int nChangeset, /* Size of buffer pChangeset in bytes */ 2771 void *pChangeset, /* Pointer to buffer containing changeset */ 2772 int bInvert, /* True to invert changeset */ 2773 int bSkipEmpty /* True to skip empty UPDATE changes */ 2774 ){ 2775 sqlite3_changeset_iter *pRet; /* Iterator to return */ 2776 int nByte; /* Number of bytes to allocate for iterator */ 2777 2778 assert( xInput==0 || (pChangeset==0 && nChangeset==0) ); 2779 2780 /* Zero the output variable in case an error occurs. */ 2781 *pp = 0; 2782 2783 /* Allocate and initialize the iterator structure. */ 2784 nByte = sizeof(sqlite3_changeset_iter); 2785 pRet = (sqlite3_changeset_iter *)sqlite3_malloc(nByte); 2786 if( !pRet ) return SQLITE_NOMEM; 2787 memset(pRet, 0, sizeof(sqlite3_changeset_iter)); 2788 pRet->in.aData = (u8 *)pChangeset; 2789 pRet->in.nData = nChangeset; 2790 pRet->in.xInput = xInput; 2791 pRet->in.pIn = pIn; 2792 pRet->in.bEof = (xInput ? 0 : 1); 2793 pRet->bInvert = bInvert; 2794 pRet->bSkipEmpty = bSkipEmpty; 2795 2796 /* Populate the output variable and return success. */ 2797 *pp = pRet; 2798 return SQLITE_OK; 2799 } 2800 2801 /* 2802 ** Create an iterator used to iterate through the contents of a changeset. 2803 */ 2804 int sqlite3changeset_start( 2805 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */ 2806 int nChangeset, /* Size of buffer pChangeset in bytes */ 2807 void *pChangeset /* Pointer to buffer containing changeset */ 2808 ){ 2809 return sessionChangesetStart(pp, 0, 0, nChangeset, pChangeset, 0, 0); 2810 } 2811 int sqlite3changeset_start_v2( 2812 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */ 2813 int nChangeset, /* Size of buffer pChangeset in bytes */ 2814 void *pChangeset, /* Pointer to buffer containing changeset */ 2815 int flags 2816 ){ 2817 int bInvert = !!(flags & SQLITE_CHANGESETSTART_INVERT); 2818 return sessionChangesetStart(pp, 0, 0, nChangeset, pChangeset, bInvert, 0); 2819 } 2820 2821 /* 2822 ** Streaming version of sqlite3changeset_start(). 2823 */ 2824 int sqlite3changeset_start_strm( 2825 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */ 2826 int (*xInput)(void *pIn, void *pData, int *pnData), 2827 void *pIn 2828 ){ 2829 return sessionChangesetStart(pp, xInput, pIn, 0, 0, 0, 0); 2830 } 2831 int sqlite3changeset_start_v2_strm( 2832 sqlite3_changeset_iter **pp, /* OUT: Changeset iterator handle */ 2833 int (*xInput)(void *pIn, void *pData, int *pnData), 2834 void *pIn, 2835 int flags 2836 ){ 2837 int bInvert = !!(flags & SQLITE_CHANGESETSTART_INVERT); 2838 return sessionChangesetStart(pp, xInput, pIn, 0, 0, bInvert, 0); 2839 } 2840 2841 /* 2842 ** If the SessionInput object passed as the only argument is a streaming 2843 ** object and the buffer is full, discard some data to free up space. 2844 */ 2845 static void sessionDiscardData(SessionInput *pIn){ 2846 if( pIn->xInput && pIn->iNext>=sessions_strm_chunk_size ){ 2847 int nMove = pIn->buf.nBuf - pIn->iNext; 2848 assert( nMove>=0 ); 2849 if( nMove>0 ){ 2850 memmove(pIn->buf.aBuf, &pIn->buf.aBuf[pIn->iNext], nMove); 2851 } 2852 pIn->buf.nBuf -= pIn->iNext; 2853 pIn->iNext = 0; 2854 pIn->nData = pIn->buf.nBuf; 2855 } 2856 } 2857 2858 /* 2859 ** Ensure that there are at least nByte bytes available in the buffer. Or, 2860 ** if there are not nByte bytes remaining in the input, that all available 2861 ** data is in the buffer. 2862 ** 2863 ** Return an SQLite error code if an error occurs, or SQLITE_OK otherwise. 2864 */ 2865 static int sessionInputBuffer(SessionInput *pIn, int nByte){ 2866 int rc = SQLITE_OK; 2867 if( pIn->xInput ){ 2868 while( !pIn->bEof && (pIn->iNext+nByte)>=pIn->nData && rc==SQLITE_OK ){ 2869 int nNew = sessions_strm_chunk_size; 2870 2871 if( pIn->bNoDiscard==0 ) sessionDiscardData(pIn); 2872 if( SQLITE_OK==sessionBufferGrow(&pIn->buf, nNew, &rc) ){ 2873 rc = pIn->xInput(pIn->pIn, &pIn->buf.aBuf[pIn->buf.nBuf], &nNew); 2874 if( nNew==0 ){ 2875 pIn->bEof = 1; 2876 }else{ 2877 pIn->buf.nBuf += nNew; 2878 } 2879 } 2880 2881 pIn->aData = pIn->buf.aBuf; 2882 pIn->nData = pIn->buf.nBuf; 2883 } 2884 } 2885 return rc; 2886 } 2887 2888 /* 2889 ** When this function is called, *ppRec points to the start of a record 2890 ** that contains nCol values. This function advances the pointer *ppRec 2891 ** until it points to the byte immediately following that record. 2892 */ 2893 static void sessionSkipRecord( 2894 u8 **ppRec, /* IN/OUT: Record pointer */ 2895 int nCol /* Number of values in record */ 2896 ){ 2897 u8 *aRec = *ppRec; 2898 int i; 2899 for(i=0; i<nCol; i++){ 2900 int eType = *aRec++; 2901 if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){ 2902 int nByte; 2903 aRec += sessionVarintGet((u8*)aRec, &nByte); 2904 aRec += nByte; 2905 }else if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){ 2906 aRec += 8; 2907 } 2908 } 2909 2910 *ppRec = aRec; 2911 } 2912 2913 /* 2914 ** This function sets the value of the sqlite3_value object passed as the 2915 ** first argument to a copy of the string or blob held in the aData[] 2916 ** buffer. SQLITE_OK is returned if successful, or SQLITE_NOMEM if an OOM 2917 ** error occurs. 2918 */ 2919 static int sessionValueSetStr( 2920 sqlite3_value *pVal, /* Set the value of this object */ 2921 u8 *aData, /* Buffer containing string or blob data */ 2922 int nData, /* Size of buffer aData[] in bytes */ 2923 u8 enc /* String encoding (0 for blobs) */ 2924 ){ 2925 /* In theory this code could just pass SQLITE_TRANSIENT as the final 2926 ** argument to sqlite3ValueSetStr() and have the copy created 2927 ** automatically. But doing so makes it difficult to detect any OOM 2928 ** error. Hence the code to create the copy externally. */ 2929 u8 *aCopy = sqlite3_malloc64((sqlite3_int64)nData+1); 2930 if( aCopy==0 ) return SQLITE_NOMEM; 2931 memcpy(aCopy, aData, nData); 2932 sqlite3ValueSetStr(pVal, nData, (char*)aCopy, enc, sqlite3_free); 2933 return SQLITE_OK; 2934 } 2935 2936 /* 2937 ** Deserialize a single record from a buffer in memory. See "RECORD FORMAT" 2938 ** for details. 2939 ** 2940 ** When this function is called, *paChange points to the start of the record 2941 ** to deserialize. Assuming no error occurs, *paChange is set to point to 2942 ** one byte after the end of the same record before this function returns. 2943 ** If the argument abPK is NULL, then the record contains nCol values. Or, 2944 ** if abPK is other than NULL, then the record contains only the PK fields 2945 ** (in other words, it is a patchset DELETE record). 2946 ** 2947 ** If successful, each element of the apOut[] array (allocated by the caller) 2948 ** is set to point to an sqlite3_value object containing the value read 2949 ** from the corresponding position in the record. If that value is not 2950 ** included in the record (i.e. because the record is part of an UPDATE change 2951 ** and the field was not modified), the corresponding element of apOut[] is 2952 ** set to NULL. 2953 ** 2954 ** It is the responsibility of the caller to free all sqlite_value structures 2955 ** using sqlite3_free(). 2956 ** 2957 ** If an error occurs, an SQLite error code (e.g. SQLITE_NOMEM) is returned. 2958 ** The apOut[] array may have been partially populated in this case. 2959 */ 2960 static int sessionReadRecord( 2961 SessionInput *pIn, /* Input data */ 2962 int nCol, /* Number of values in record */ 2963 u8 *abPK, /* Array of primary key flags, or NULL */ 2964 sqlite3_value **apOut, /* Write values to this array */ 2965 int *pbEmpty 2966 ){ 2967 int i; /* Used to iterate through columns */ 2968 int rc = SQLITE_OK; 2969 2970 assert( pbEmpty==0 || *pbEmpty==0 ); 2971 if( pbEmpty ) *pbEmpty = 1; 2972 for(i=0; i<nCol && rc==SQLITE_OK; i++){ 2973 int eType = 0; /* Type of value (SQLITE_NULL, TEXT etc.) */ 2974 if( abPK && abPK[i]==0 ) continue; 2975 rc = sessionInputBuffer(pIn, 9); 2976 if( rc==SQLITE_OK ){ 2977 if( pIn->iNext>=pIn->nData ){ 2978 rc = SQLITE_CORRUPT_BKPT; 2979 }else{ 2980 eType = pIn->aData[pIn->iNext++]; 2981 assert( apOut[i]==0 ); 2982 if( eType ){ 2983 if( pbEmpty ) *pbEmpty = 0; 2984 apOut[i] = sqlite3ValueNew(0); 2985 if( !apOut[i] ) rc = SQLITE_NOMEM; 2986 } 2987 } 2988 } 2989 2990 if( rc==SQLITE_OK ){ 2991 u8 *aVal = &pIn->aData[pIn->iNext]; 2992 if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){ 2993 int nByte; 2994 pIn->iNext += sessionVarintGet(aVal, &nByte); 2995 rc = sessionInputBuffer(pIn, nByte); 2996 if( rc==SQLITE_OK ){ 2997 if( nByte<0 || nByte>pIn->nData-pIn->iNext ){ 2998 rc = SQLITE_CORRUPT_BKPT; 2999 }else{ 3000 u8 enc = (eType==SQLITE_TEXT ? SQLITE_UTF8 : 0); 3001 rc = sessionValueSetStr(apOut[i],&pIn->aData[pIn->iNext],nByte,enc); 3002 pIn->iNext += nByte; 3003 } 3004 } 3005 } 3006 if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){ 3007 sqlite3_int64 v = sessionGetI64(aVal); 3008 if( eType==SQLITE_INTEGER ){ 3009 sqlite3VdbeMemSetInt64(apOut[i], v); 3010 }else{ 3011 double d; 3012 memcpy(&d, &v, 8); 3013 sqlite3VdbeMemSetDouble(apOut[i], d); 3014 } 3015 pIn->iNext += 8; 3016 } 3017 } 3018 } 3019 3020 return rc; 3021 } 3022 3023 /* 3024 ** The input pointer currently points to the second byte of a table-header. 3025 ** Specifically, to the following: 3026 ** 3027 ** + number of columns in table (varint) 3028 ** + array of PK flags (1 byte per column), 3029 ** + table name (nul terminated). 3030 ** 3031 ** This function ensures that all of the above is present in the input 3032 ** buffer (i.e. that it can be accessed without any calls to xInput()). 3033 ** If successful, SQLITE_OK is returned. Otherwise, an SQLite error code. 3034 ** The input pointer is not moved. 3035 */ 3036 static int sessionChangesetBufferTblhdr(SessionInput *pIn, int *pnByte){ 3037 int rc = SQLITE_OK; 3038 int nCol = 0; 3039 int nRead = 0; 3040 3041 rc = sessionInputBuffer(pIn, 9); 3042 if( rc==SQLITE_OK ){ 3043 nRead += sessionVarintGet(&pIn->aData[pIn->iNext + nRead], &nCol); 3044 /* The hard upper limit for the number of columns in an SQLite 3045 ** database table is, according to sqliteLimit.h, 32676. So 3046 ** consider any table-header that purports to have more than 65536 3047 ** columns to be corrupt. This is convenient because otherwise, 3048 ** if the (nCol>65536) condition below were omitted, a sufficiently 3049 ** large value for nCol may cause nRead to wrap around and become 3050 ** negative. Leading to a crash. */ 3051 if( nCol<0 || nCol>65536 ){ 3052 rc = SQLITE_CORRUPT_BKPT; 3053 }else{ 3054 rc = sessionInputBuffer(pIn, nRead+nCol+100); 3055 nRead += nCol; 3056 } 3057 } 3058 3059 while( rc==SQLITE_OK ){ 3060 while( (pIn->iNext + nRead)<pIn->nData && pIn->aData[pIn->iNext + nRead] ){ 3061 nRead++; 3062 } 3063 if( (pIn->iNext + nRead)<pIn->nData ) break; 3064 rc = sessionInputBuffer(pIn, nRead + 100); 3065 } 3066 *pnByte = nRead+1; 3067 return rc; 3068 } 3069 3070 /* 3071 ** The input pointer currently points to the first byte of the first field 3072 ** of a record consisting of nCol columns. This function ensures the entire 3073 ** record is buffered. It does not move the input pointer. 3074 ** 3075 ** If successful, SQLITE_OK is returned and *pnByte is set to the size of 3076 ** the record in bytes. Otherwise, an SQLite error code is returned. The 3077 ** final value of *pnByte is undefined in this case. 3078 */ 3079 static int sessionChangesetBufferRecord( 3080 SessionInput *pIn, /* Input data */ 3081 int nCol, /* Number of columns in record */ 3082 int *pnByte /* OUT: Size of record in bytes */ 3083 ){ 3084 int rc = SQLITE_OK; 3085 int nByte = 0; 3086 int i; 3087 for(i=0; rc==SQLITE_OK && i<nCol; i++){ 3088 int eType; 3089 rc = sessionInputBuffer(pIn, nByte + 10); 3090 if( rc==SQLITE_OK ){ 3091 eType = pIn->aData[pIn->iNext + nByte++]; 3092 if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){ 3093 int n; 3094 nByte += sessionVarintGet(&pIn->aData[pIn->iNext+nByte], &n); 3095 nByte += n; 3096 rc = sessionInputBuffer(pIn, nByte); 3097 }else if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){ 3098 nByte += 8; 3099 } 3100 } 3101 } 3102 *pnByte = nByte; 3103 return rc; 3104 } 3105 3106 /* 3107 ** The input pointer currently points to the second byte of a table-header. 3108 ** Specifically, to the following: 3109 ** 3110 ** + number of columns in table (varint) 3111 ** + array of PK flags (1 byte per column), 3112 ** + table name (nul terminated). 3113 ** 3114 ** This function decodes the table-header and populates the p->nCol, 3115 ** p->zTab and p->abPK[] variables accordingly. The p->apValue[] array is 3116 ** also allocated or resized according to the new value of p->nCol. The 3117 ** input pointer is left pointing to the byte following the table header. 3118 ** 3119 ** If successful, SQLITE_OK is returned. Otherwise, an SQLite error code 3120 ** is returned and the final values of the various fields enumerated above 3121 ** are undefined. 3122 */ 3123 static int sessionChangesetReadTblhdr(sqlite3_changeset_iter *p){ 3124 int rc; 3125 int nCopy; 3126 assert( p->rc==SQLITE_OK ); 3127 3128 rc = sessionChangesetBufferTblhdr(&p->in, &nCopy); 3129 if( rc==SQLITE_OK ){ 3130 int nByte; 3131 int nVarint; 3132 nVarint = sessionVarintGet(&p->in.aData[p->in.iNext], &p->nCol); 3133 if( p->nCol>0 ){ 3134 nCopy -= nVarint; 3135 p->in.iNext += nVarint; 3136 nByte = p->nCol * sizeof(sqlite3_value*) * 2 + nCopy; 3137 p->tblhdr.nBuf = 0; 3138 sessionBufferGrow(&p->tblhdr, nByte, &rc); 3139 }else{ 3140 rc = SQLITE_CORRUPT_BKPT; 3141 } 3142 } 3143 3144 if( rc==SQLITE_OK ){ 3145 size_t iPK = sizeof(sqlite3_value*)*p->nCol*2; 3146 memset(p->tblhdr.aBuf, 0, iPK); 3147 memcpy(&p->tblhdr.aBuf[iPK], &p->in.aData[p->in.iNext], nCopy); 3148 p->in.iNext += nCopy; 3149 } 3150 3151 p->apValue = (sqlite3_value**)p->tblhdr.aBuf; 3152 if( p->apValue==0 ){ 3153 p->abPK = 0; 3154 p->zTab = 0; 3155 }else{ 3156 p->abPK = (u8*)&p->apValue[p->nCol*2]; 3157 p->zTab = p->abPK ? (char*)&p->abPK[p->nCol] : 0; 3158 } 3159 return (p->rc = rc); 3160 } 3161 3162 /* 3163 ** Advance the changeset iterator to the next change. The differences between 3164 ** this function and sessionChangesetNext() are that 3165 ** 3166 ** * If pbEmpty is not NULL and the change is a no-op UPDATE (an UPDATE 3167 ** that modifies no columns), this function sets (*pbEmpty) to 1. 3168 ** 3169 ** * If the iterator is configured to skip no-op UPDATEs, 3170 ** sessionChangesetNext() does that. This function does not. 3171 */ 3172 static int sessionChangesetNextOne( 3173 sqlite3_changeset_iter *p, /* Changeset iterator */ 3174 u8 **paRec, /* If non-NULL, store record pointer here */ 3175 int *pnRec, /* If non-NULL, store size of record here */ 3176 int *pbNew, /* If non-NULL, true if new table */ 3177 int *pbEmpty 3178 ){ 3179 int i; 3180 u8 op; 3181 3182 assert( (paRec==0 && pnRec==0) || (paRec && pnRec) ); 3183 assert( pbEmpty==0 || *pbEmpty==0 ); 3184 3185 /* If the iterator is in the error-state, return immediately. */ 3186 if( p->rc!=SQLITE_OK ) return p->rc; 3187 3188 /* Free the current contents of p->apValue[], if any. */ 3189 if( p->apValue ){ 3190 for(i=0; i<p->nCol*2; i++){ 3191 sqlite3ValueFree(p->apValue[i]); 3192 } 3193 memset(p->apValue, 0, sizeof(sqlite3_value*)*p->nCol*2); 3194 } 3195 3196 /* Make sure the buffer contains at least 10 bytes of input data, or all 3197 ** remaining data if there are less than 10 bytes available. This is 3198 ** sufficient either for the 'T' or 'P' byte and the varint that follows 3199 ** it, or for the two single byte values otherwise. */ 3200 p->rc = sessionInputBuffer(&p->in, 2); 3201 if( p->rc!=SQLITE_OK ) return p->rc; 3202 3203 /* If the iterator is already at the end of the changeset, return DONE. */ 3204 if( p->in.iNext>=p->in.nData ){ 3205 return SQLITE_DONE; 3206 } 3207 3208 sessionDiscardData(&p->in); 3209 p->in.iCurrent = p->in.iNext; 3210 3211 op = p->in.aData[p->in.iNext++]; 3212 while( op=='T' || op=='P' ){ 3213 if( pbNew ) *pbNew = 1; 3214 p->bPatchset = (op=='P'); 3215 if( sessionChangesetReadTblhdr(p) ) return p->rc; 3216 if( (p->rc = sessionInputBuffer(&p->in, 2)) ) return p->rc; 3217 p->in.iCurrent = p->in.iNext; 3218 if( p->in.iNext>=p->in.nData ) return SQLITE_DONE; 3219 op = p->in.aData[p->in.iNext++]; 3220 } 3221 3222 if( p->zTab==0 || (p->bPatchset && p->bInvert) ){ 3223 /* The first record in the changeset is not a table header. Must be a 3224 ** corrupt changeset. */ 3225 assert( p->in.iNext==1 || p->zTab ); 3226 return (p->rc = SQLITE_CORRUPT_BKPT); 3227 } 3228 3229 p->op = op; 3230 p->bIndirect = p->in.aData[p->in.iNext++]; 3231 if( p->op!=SQLITE_UPDATE && p->op!=SQLITE_DELETE && p->op!=SQLITE_INSERT ){ 3232 return (p->rc = SQLITE_CORRUPT_BKPT); 3233 } 3234 3235 if( paRec ){ 3236 int nVal; /* Number of values to buffer */ 3237 if( p->bPatchset==0 && op==SQLITE_UPDATE ){ 3238 nVal = p->nCol * 2; 3239 }else if( p->bPatchset && op==SQLITE_DELETE ){ 3240 nVal = 0; 3241 for(i=0; i<p->nCol; i++) if( p->abPK[i] ) nVal++; 3242 }else{ 3243 nVal = p->nCol; 3244 } 3245 p->rc = sessionChangesetBufferRecord(&p->in, nVal, pnRec); 3246 if( p->rc!=SQLITE_OK ) return p->rc; 3247 *paRec = &p->in.aData[p->in.iNext]; 3248 p->in.iNext += *pnRec; 3249 }else{ 3250 sqlite3_value **apOld = (p->bInvert ? &p->apValue[p->nCol] : p->apValue); 3251 sqlite3_value **apNew = (p->bInvert ? p->apValue : &p->apValue[p->nCol]); 3252 3253 /* If this is an UPDATE or DELETE, read the old.* record. */ 3254 if( p->op!=SQLITE_INSERT && (p->bPatchset==0 || p->op==SQLITE_DELETE) ){ 3255 u8 *abPK = p->bPatchset ? p->abPK : 0; 3256 p->rc = sessionReadRecord(&p->in, p->nCol, abPK, apOld, 0); 3257 if( p->rc!=SQLITE_OK ) return p->rc; 3258 } 3259 3260 /* If this is an INSERT or UPDATE, read the new.* record. */ 3261 if( p->op!=SQLITE_DELETE ){ 3262 p->rc = sessionReadRecord(&p->in, p->nCol, 0, apNew, pbEmpty); 3263 if( p->rc!=SQLITE_OK ) return p->rc; 3264 } 3265 3266 if( (p->bPatchset || p->bInvert) && p->op==SQLITE_UPDATE ){ 3267 /* If this is an UPDATE that is part of a patchset, then all PK and 3268 ** modified fields are present in the new.* record. The old.* record 3269 ** is currently completely empty. This block shifts the PK fields from 3270 ** new.* to old.*, to accommodate the code that reads these arrays. */ 3271 for(i=0; i<p->nCol; i++){ 3272 assert( p->bPatchset==0 || p->apValue[i]==0 ); 3273 if( p->abPK[i] ){ 3274 assert( p->apValue[i]==0 ); 3275 p->apValue[i] = p->apValue[i+p->nCol]; 3276 if( p->apValue[i]==0 ) return (p->rc = SQLITE_CORRUPT_BKPT); 3277 p->apValue[i+p->nCol] = 0; 3278 } 3279 } 3280 }else if( p->bInvert ){ 3281 if( p->op==SQLITE_INSERT ) p->op = SQLITE_DELETE; 3282 else if( p->op==SQLITE_DELETE ) p->op = SQLITE_INSERT; 3283 } 3284 } 3285 3286 return SQLITE_ROW; 3287 } 3288 3289 /* 3290 ** Advance the changeset iterator to the next change. 3291 ** 3292 ** If both paRec and pnRec are NULL, then this function works like the public 3293 ** API sqlite3changeset_next(). If SQLITE_ROW is returned, then the 3294 ** sqlite3changeset_new() and old() APIs may be used to query for values. 3295 ** 3296 ** Otherwise, if paRec and pnRec are not NULL, then a pointer to the change 3297 ** record is written to *paRec before returning and the number of bytes in 3298 ** the record to *pnRec. 3299 ** 3300 ** Either way, this function returns SQLITE_ROW if the iterator is 3301 ** successfully advanced to the next change in the changeset, an SQLite 3302 ** error code if an error occurs, or SQLITE_DONE if there are no further 3303 ** changes in the changeset. 3304 */ 3305 static int sessionChangesetNext( 3306 sqlite3_changeset_iter *p, /* Changeset iterator */ 3307 u8 **paRec, /* If non-NULL, store record pointer here */ 3308 int *pnRec, /* If non-NULL, store size of record here */ 3309 int *pbNew /* If non-NULL, true if new table */ 3310 ){ 3311 int bEmpty; 3312 int rc; 3313 do { 3314 bEmpty = 0; 3315 rc = sessionChangesetNextOne(p, paRec, pnRec, pbNew, &bEmpty); 3316 }while( rc==SQLITE_ROW && p->bSkipEmpty && bEmpty); 3317 return rc; 3318 } 3319 3320 /* 3321 ** Advance an iterator created by sqlite3changeset_start() to the next 3322 ** change in the changeset. This function may return SQLITE_ROW, SQLITE_DONE 3323 ** or SQLITE_CORRUPT. 3324 ** 3325 ** This function may not be called on iterators passed to a conflict handler 3326 ** callback by changeset_apply(). 3327 */ 3328 int sqlite3changeset_next(sqlite3_changeset_iter *p){ 3329 return sessionChangesetNext(p, 0, 0, 0); 3330 } 3331 3332 /* 3333 ** The following function extracts information on the current change 3334 ** from a changeset iterator. It may only be called after changeset_next() 3335 ** has returned SQLITE_ROW. 3336 */ 3337 int sqlite3changeset_op( 3338 sqlite3_changeset_iter *pIter, /* Iterator handle */ 3339 const char **pzTab, /* OUT: Pointer to table name */ 3340 int *pnCol, /* OUT: Number of columns in table */ 3341 int *pOp, /* OUT: SQLITE_INSERT, DELETE or UPDATE */ 3342 int *pbIndirect /* OUT: True if change is indirect */ 3343 ){ 3344 *pOp = pIter->op; 3345 *pnCol = pIter->nCol; 3346 *pzTab = pIter->zTab; 3347 if( pbIndirect ) *pbIndirect = pIter->bIndirect; 3348 return SQLITE_OK; 3349 } 3350 3351 /* 3352 ** Return information regarding the PRIMARY KEY and number of columns in 3353 ** the database table affected by the change that pIter currently points 3354 ** to. This function may only be called after changeset_next() returns 3355 ** SQLITE_ROW. 3356 */ 3357 int sqlite3changeset_pk( 3358 sqlite3_changeset_iter *pIter, /* Iterator object */ 3359 unsigned char **pabPK, /* OUT: Array of boolean - true for PK cols */ 3360 int *pnCol /* OUT: Number of entries in output array */ 3361 ){ 3362 *pabPK = pIter->abPK; 3363 if( pnCol ) *pnCol = pIter->nCol; 3364 return SQLITE_OK; 3365 } 3366 3367 /* 3368 ** This function may only be called while the iterator is pointing to an 3369 ** SQLITE_UPDATE or SQLITE_DELETE change (see sqlite3changeset_op()). 3370 ** Otherwise, SQLITE_MISUSE is returned. 3371 ** 3372 ** It sets *ppValue to point to an sqlite3_value structure containing the 3373 ** iVal'th value in the old.* record. Or, if that particular value is not 3374 ** included in the record (because the change is an UPDATE and the field 3375 ** was not modified and is not a PK column), set *ppValue to NULL. 3376 ** 3377 ** If value iVal is out-of-range, SQLITE_RANGE is returned and *ppValue is 3378 ** not modified. Otherwise, SQLITE_OK. 3379 */ 3380 int sqlite3changeset_old( 3381 sqlite3_changeset_iter *pIter, /* Changeset iterator */ 3382 int iVal, /* Index of old.* value to retrieve */ 3383 sqlite3_value **ppValue /* OUT: Old value (or NULL pointer) */ 3384 ){ 3385 if( pIter->op!=SQLITE_UPDATE && pIter->op!=SQLITE_DELETE ){ 3386 return SQLITE_MISUSE; 3387 } 3388 if( iVal<0 || iVal>=pIter->nCol ){ 3389 return SQLITE_RANGE; 3390 } 3391 *ppValue = pIter->apValue[iVal]; 3392 return SQLITE_OK; 3393 } 3394 3395 /* 3396 ** This function may only be called while the iterator is pointing to an 3397 ** SQLITE_UPDATE or SQLITE_INSERT change (see sqlite3changeset_op()). 3398 ** Otherwise, SQLITE_MISUSE is returned. 3399 ** 3400 ** It sets *ppValue to point to an sqlite3_value structure containing the 3401 ** iVal'th value in the new.* record. Or, if that particular value is not 3402 ** included in the record (because the change is an UPDATE and the field 3403 ** was not modified), set *ppValue to NULL. 3404 ** 3405 ** If value iVal is out-of-range, SQLITE_RANGE is returned and *ppValue is 3406 ** not modified. Otherwise, SQLITE_OK. 3407 */ 3408 int sqlite3changeset_new( 3409 sqlite3_changeset_iter *pIter, /* Changeset iterator */ 3410 int iVal, /* Index of new.* value to retrieve */ 3411 sqlite3_value **ppValue /* OUT: New value (or NULL pointer) */ 3412 ){ 3413 if( pIter->op!=SQLITE_UPDATE && pIter->op!=SQLITE_INSERT ){ 3414 return SQLITE_MISUSE; 3415 } 3416 if( iVal<0 || iVal>=pIter->nCol ){ 3417 return SQLITE_RANGE; 3418 } 3419 *ppValue = pIter->apValue[pIter->nCol+iVal]; 3420 return SQLITE_OK; 3421 } 3422 3423 /* 3424 ** The following two macros are used internally. They are similar to the 3425 ** sqlite3changeset_new() and sqlite3changeset_old() functions, except that 3426 ** they omit all error checking and return a pointer to the requested value. 3427 */ 3428 #define sessionChangesetNew(pIter, iVal) (pIter)->apValue[(pIter)->nCol+(iVal)] 3429 #define sessionChangesetOld(pIter, iVal) (pIter)->apValue[(iVal)] 3430 3431 /* 3432 ** This function may only be called with a changeset iterator that has been 3433 ** passed to an SQLITE_CHANGESET_DATA or SQLITE_CHANGESET_CONFLICT 3434 ** conflict-handler function. Otherwise, SQLITE_MISUSE is returned. 3435 ** 3436 ** If successful, *ppValue is set to point to an sqlite3_value structure 3437 ** containing the iVal'th value of the conflicting record. 3438 ** 3439 ** If value iVal is out-of-range or some other error occurs, an SQLite error 3440 ** code is returned. Otherwise, SQLITE_OK. 3441 */ 3442 int sqlite3changeset_conflict( 3443 sqlite3_changeset_iter *pIter, /* Changeset iterator */ 3444 int iVal, /* Index of conflict record value to fetch */ 3445 sqlite3_value **ppValue /* OUT: Value from conflicting row */ 3446 ){ 3447 if( !pIter->pConflict ){ 3448 return SQLITE_MISUSE; 3449 } 3450 if( iVal<0 || iVal>=pIter->nCol ){ 3451 return SQLITE_RANGE; 3452 } 3453 *ppValue = sqlite3_column_value(pIter->pConflict, iVal); 3454 return SQLITE_OK; 3455 } 3456 3457 /* 3458 ** This function may only be called with an iterator passed to an 3459 ** SQLITE_CHANGESET_FOREIGN_KEY conflict handler callback. In this case 3460 ** it sets the output variable to the total number of known foreign key 3461 ** violations in the destination database and returns SQLITE_OK. 3462 ** 3463 ** In all other cases this function returns SQLITE_MISUSE. 3464 */ 3465 int sqlite3changeset_fk_conflicts( 3466 sqlite3_changeset_iter *pIter, /* Changeset iterator */ 3467 int *pnOut /* OUT: Number of FK violations */ 3468 ){ 3469 if( pIter->pConflict || pIter->apValue ){ 3470 return SQLITE_MISUSE; 3471 } 3472 *pnOut = pIter->nCol; 3473 return SQLITE_OK; 3474 } 3475 3476 3477 /* 3478 ** Finalize an iterator allocated with sqlite3changeset_start(). 3479 ** 3480 ** This function may not be called on iterators passed to a conflict handler 3481 ** callback by changeset_apply(). 3482 */ 3483 int sqlite3changeset_finalize(sqlite3_changeset_iter *p){ 3484 int rc = SQLITE_OK; 3485 if( p ){ 3486 int i; /* Used to iterate through p->apValue[] */ 3487 rc = p->rc; 3488 if( p->apValue ){ 3489 for(i=0; i<p->nCol*2; i++) sqlite3ValueFree(p->apValue[i]); 3490 } 3491 sqlite3_free(p->tblhdr.aBuf); 3492 sqlite3_free(p->in.buf.aBuf); 3493 sqlite3_free(p); 3494 } 3495 return rc; 3496 } 3497 3498 static int sessionChangesetInvert( 3499 SessionInput *pInput, /* Input changeset */ 3500 int (*xOutput)(void *pOut, const void *pData, int nData), 3501 void *pOut, 3502 int *pnInverted, /* OUT: Number of bytes in output changeset */ 3503 void **ppInverted /* OUT: Inverse of pChangeset */ 3504 ){ 3505 int rc = SQLITE_OK; /* Return value */ 3506 SessionBuffer sOut; /* Output buffer */ 3507 int nCol = 0; /* Number of cols in current table */ 3508 u8 *abPK = 0; /* PK array for current table */ 3509 sqlite3_value **apVal = 0; /* Space for values for UPDATE inversion */ 3510 SessionBuffer sPK = {0, 0, 0}; /* PK array for current table */ 3511 3512 /* Initialize the output buffer */ 3513 memset(&sOut, 0, sizeof(SessionBuffer)); 3514 3515 /* Zero the output variables in case an error occurs. */ 3516 if( ppInverted ){ 3517 *ppInverted = 0; 3518 *pnInverted = 0; 3519 } 3520 3521 while( 1 ){ 3522 u8 eType; 3523 3524 /* Test for EOF. */ 3525 if( (rc = sessionInputBuffer(pInput, 2)) ) goto finished_invert; 3526 if( pInput->iNext>=pInput->nData ) break; 3527 eType = pInput->aData[pInput->iNext]; 3528 3529 switch( eType ){ 3530 case 'T': { 3531 /* A 'table' record consists of: 3532 ** 3533 ** * A constant 'T' character, 3534 ** * Number of columns in said table (a varint), 3535 ** * An array of nCol bytes (sPK), 3536 ** * A nul-terminated table name. 3537 */ 3538 int nByte; 3539 int nVar; 3540 pInput->iNext++; 3541 if( (rc = sessionChangesetBufferTblhdr(pInput, &nByte)) ){ 3542 goto finished_invert; 3543 } 3544 nVar = sessionVarintGet(&pInput->aData[pInput->iNext], &nCol); 3545 sPK.nBuf = 0; 3546 sessionAppendBlob(&sPK, &pInput->aData[pInput->iNext+nVar], nCol, &rc); 3547 sessionAppendByte(&sOut, eType, &rc); 3548 sessionAppendBlob(&sOut, &pInput->aData[pInput->iNext], nByte, &rc); 3549 if( rc ) goto finished_invert; 3550 3551 pInput->iNext += nByte; 3552 sqlite3_free(apVal); 3553 apVal = 0; 3554 abPK = sPK.aBuf; 3555 break; 3556 } 3557 3558 case SQLITE_INSERT: 3559 case SQLITE_DELETE: { 3560 int nByte; 3561 int bIndirect = pInput->aData[pInput->iNext+1]; 3562 int eType2 = (eType==SQLITE_DELETE ? SQLITE_INSERT : SQLITE_DELETE); 3563 pInput->iNext += 2; 3564 assert( rc==SQLITE_OK ); 3565 rc = sessionChangesetBufferRecord(pInput, nCol, &nByte); 3566 sessionAppendByte(&sOut, eType2, &rc); 3567 sessionAppendByte(&sOut, bIndirect, &rc); 3568 sessionAppendBlob(&sOut, &pInput->aData[pInput->iNext], nByte, &rc); 3569 pInput->iNext += nByte; 3570 if( rc ) goto finished_invert; 3571 break; 3572 } 3573 3574 case SQLITE_UPDATE: { 3575 int iCol; 3576 3577 if( 0==apVal ){ 3578 apVal = (sqlite3_value **)sqlite3_malloc64(sizeof(apVal[0])*nCol*2); 3579 if( 0==apVal ){ 3580 rc = SQLITE_NOMEM; 3581 goto finished_invert; 3582 } 3583 memset(apVal, 0, sizeof(apVal[0])*nCol*2); 3584 } 3585 3586 /* Write the header for the new UPDATE change. Same as the original. */ 3587 sessionAppendByte(&sOut, eType, &rc); 3588 sessionAppendByte(&sOut, pInput->aData[pInput->iNext+1], &rc); 3589 3590 /* Read the old.* and new.* records for the update change. */ 3591 pInput->iNext += 2; 3592 rc = sessionReadRecord(pInput, nCol, 0, &apVal[0], 0); 3593 if( rc==SQLITE_OK ){ 3594 rc = sessionReadRecord(pInput, nCol, 0, &apVal[nCol], 0); 3595 } 3596 3597 /* Write the new old.* record. Consists of the PK columns from the 3598 ** original old.* record, and the other values from the original 3599 ** new.* record. */ 3600 for(iCol=0; iCol<nCol; iCol++){ 3601 sqlite3_value *pVal = apVal[iCol + (abPK[iCol] ? 0 : nCol)]; 3602 sessionAppendValue(&sOut, pVal, &rc); 3603 } 3604 3605 /* Write the new new.* record. Consists of a copy of all values 3606 ** from the original old.* record, except for the PK columns, which 3607 ** are set to "undefined". */ 3608 for(iCol=0; iCol<nCol; iCol++){ 3609 sqlite3_value *pVal = (abPK[iCol] ? 0 : apVal[iCol]); 3610 sessionAppendValue(&sOut, pVal, &rc); 3611 } 3612 3613 for(iCol=0; iCol<nCol*2; iCol++){ 3614 sqlite3ValueFree(apVal[iCol]); 3615 } 3616 memset(apVal, 0, sizeof(apVal[0])*nCol*2); 3617 if( rc!=SQLITE_OK ){ 3618 goto finished_invert; 3619 } 3620 3621 break; 3622 } 3623 3624 default: 3625 rc = SQLITE_CORRUPT_BKPT; 3626 goto finished_invert; 3627 } 3628 3629 assert( rc==SQLITE_OK ); 3630 if( xOutput && sOut.nBuf>=sessions_strm_chunk_size ){ 3631 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf); 3632 sOut.nBuf = 0; 3633 if( rc!=SQLITE_OK ) goto finished_invert; 3634 } 3635 } 3636 3637 assert( rc==SQLITE_OK ); 3638 if( pnInverted ){ 3639 *pnInverted = sOut.nBuf; 3640 *ppInverted = sOut.aBuf; 3641 sOut.aBuf = 0; 3642 }else if( sOut.nBuf>0 ){ 3643 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf); 3644 } 3645 3646 finished_invert: 3647 sqlite3_free(sOut.aBuf); 3648 sqlite3_free(apVal); 3649 sqlite3_free(sPK.aBuf); 3650 return rc; 3651 } 3652 3653 3654 /* 3655 ** Invert a changeset object. 3656 */ 3657 int sqlite3changeset_invert( 3658 int nChangeset, /* Number of bytes in input */ 3659 const void *pChangeset, /* Input changeset */ 3660 int *pnInverted, /* OUT: Number of bytes in output changeset */ 3661 void **ppInverted /* OUT: Inverse of pChangeset */ 3662 ){ 3663 SessionInput sInput; 3664 3665 /* Set up the input stream */ 3666 memset(&sInput, 0, sizeof(SessionInput)); 3667 sInput.nData = nChangeset; 3668 sInput.aData = (u8*)pChangeset; 3669 3670 return sessionChangesetInvert(&sInput, 0, 0, pnInverted, ppInverted); 3671 } 3672 3673 /* 3674 ** Streaming version of sqlite3changeset_invert(). 3675 */ 3676 int sqlite3changeset_invert_strm( 3677 int (*xInput)(void *pIn, void *pData, int *pnData), 3678 void *pIn, 3679 int (*xOutput)(void *pOut, const void *pData, int nData), 3680 void *pOut 3681 ){ 3682 SessionInput sInput; 3683 int rc; 3684 3685 /* Set up the input stream */ 3686 memset(&sInput, 0, sizeof(SessionInput)); 3687 sInput.xInput = xInput; 3688 sInput.pIn = pIn; 3689 3690 rc = sessionChangesetInvert(&sInput, xOutput, pOut, 0, 0); 3691 sqlite3_free(sInput.buf.aBuf); 3692 return rc; 3693 } 3694 3695 3696 typedef struct SessionUpdate SessionUpdate; 3697 struct SessionUpdate { 3698 sqlite3_stmt *pStmt; 3699 u32 *aMask; 3700 SessionUpdate *pNext; 3701 }; 3702 3703 typedef struct SessionApplyCtx SessionApplyCtx; 3704 struct SessionApplyCtx { 3705 sqlite3 *db; 3706 sqlite3_stmt *pDelete; /* DELETE statement */ 3707 sqlite3_stmt *pInsert; /* INSERT statement */ 3708 sqlite3_stmt *pSelect; /* SELECT statement */ 3709 int nCol; /* Size of azCol[] and abPK[] arrays */ 3710 const char **azCol; /* Array of column names */ 3711 u8 *abPK; /* Boolean array - true if column is in PK */ 3712 u32 *aUpdateMask; /* Used by sessionUpdateFind */ 3713 SessionUpdate *pUp; 3714 int bStat1; /* True if table is sqlite_stat1 */ 3715 int bDeferConstraints; /* True to defer constraints */ 3716 int bInvertConstraints; /* Invert when iterating constraints buffer */ 3717 SessionBuffer constraints; /* Deferred constraints are stored here */ 3718 SessionBuffer rebase; /* Rebase information (if any) here */ 3719 u8 bRebaseStarted; /* If table header is already in rebase */ 3720 u8 bRebase; /* True to collect rebase information */ 3721 }; 3722 3723 /* Number of prepared UPDATE statements to cache. */ 3724 #define SESSION_UPDATE_CACHE_SZ 12 3725 3726 /* 3727 ** Find a prepared UPDATE statement suitable for the UPDATE step currently 3728 ** being visited by the iterator. The UPDATE is of the form: 3729 ** 3730 ** UPDATE tbl SET col = ?, col2 = ? WHERE pk1 IS ? AND pk2 IS ? 3731 */ 3732 static int sessionUpdateFind( 3733 sqlite3_changeset_iter *pIter, 3734 SessionApplyCtx *p, 3735 int bPatchset, 3736 sqlite3_stmt **ppStmt 3737 ){ 3738 int rc = SQLITE_OK; 3739 SessionUpdate *pUp = 0; 3740 int nCol = pIter->nCol; 3741 int nU32 = (pIter->nCol+33)/32; 3742 int ii; 3743 3744 if( p->aUpdateMask==0 ){ 3745 p->aUpdateMask = sqlite3_malloc(nU32*sizeof(u32)); 3746 if( p->aUpdateMask==0 ){ 3747 rc = SQLITE_NOMEM; 3748 } 3749 } 3750 3751 if( rc==SQLITE_OK ){ 3752 memset(p->aUpdateMask, 0, nU32*sizeof(u32)); 3753 rc = SQLITE_CORRUPT; 3754 for(ii=0; ii<pIter->nCol; ii++){ 3755 if( sessionChangesetNew(pIter, ii) ){ 3756 p->aUpdateMask[ii/32] |= (1<<(ii%32)); 3757 rc = SQLITE_OK; 3758 } 3759 } 3760 } 3761 3762 if( rc==SQLITE_OK ){ 3763 if( bPatchset ) p->aUpdateMask[nCol/32] |= (1<<(nCol%32)); 3764 3765 if( p->pUp ){ 3766 int nUp = 0; 3767 SessionUpdate **pp = &p->pUp; 3768 while( 1 ){ 3769 nUp++; 3770 if( 0==memcmp(p->aUpdateMask, (*pp)->aMask, nU32*sizeof(u32)) ){ 3771 pUp = *pp; 3772 *pp = pUp->pNext; 3773 pUp->pNext = p->pUp; 3774 p->pUp = pUp; 3775 break; 3776 } 3777 3778 if( (*pp)->pNext ){ 3779 pp = &(*pp)->pNext; 3780 }else{ 3781 if( nUp>=SESSION_UPDATE_CACHE_SZ ){ 3782 sqlite3_finalize((*pp)->pStmt); 3783 sqlite3_free(*pp); 3784 *pp = 0; 3785 } 3786 break; 3787 } 3788 } 3789 } 3790 3791 if( pUp==0 ){ 3792 int nByte = sizeof(SessionUpdate) * nU32*sizeof(u32); 3793 int bStat1 = (sqlite3_stricmp(pIter->zTab, "sqlite_stat1")==0); 3794 pUp = (SessionUpdate*)sqlite3_malloc(nByte); 3795 if( pUp==0 ){ 3796 rc = SQLITE_NOMEM; 3797 }else{ 3798 const char *zSep = ""; 3799 SessionBuffer buf; 3800 3801 memset(&buf, 0, sizeof(buf)); 3802 pUp->aMask = (u32*)&pUp[1]; 3803 memcpy(pUp->aMask, p->aUpdateMask, nU32*sizeof(u32)); 3804 3805 sessionAppendStr(&buf, "UPDATE main.", &rc); 3806 sessionAppendIdent(&buf, pIter->zTab, &rc); 3807 sessionAppendStr(&buf, " SET ", &rc); 3808 3809 /* Create the assignments part of the UPDATE */ 3810 for(ii=0; ii<pIter->nCol; ii++){ 3811 if( p->abPK[ii]==0 && sessionChangesetNew(pIter, ii) ){ 3812 sessionAppendStr(&buf, zSep, &rc); 3813 sessionAppendIdent(&buf, p->azCol[ii], &rc); 3814 sessionAppendStr(&buf, " = ?", &rc); 3815 sessionAppendInteger(&buf, ii*2+1, &rc); 3816 zSep = ", "; 3817 } 3818 } 3819 3820 /* Create the WHERE clause part of the UPDATE */ 3821 zSep = ""; 3822 sessionAppendStr(&buf, " WHERE ", &rc); 3823 for(ii=0; ii<pIter->nCol; ii++){ 3824 if( p->abPK[ii] || (bPatchset==0 && sessionChangesetOld(pIter, ii)) ){ 3825 sessionAppendStr(&buf, zSep, &rc); 3826 if( bStat1 && ii==1 ){ 3827 assert( sqlite3_stricmp(p->azCol[ii], "idx")==0 ); 3828 sessionAppendStr(&buf, 3829 "idx IS CASE " 3830 "WHEN length(?4)=0 AND typeof(?4)='blob' THEN NULL " 3831 "ELSE ?4 END ", &rc 3832 ); 3833 }else{ 3834 sessionAppendIdent(&buf, p->azCol[ii], &rc); 3835 sessionAppendStr(&buf, " IS ?", &rc); 3836 sessionAppendInteger(&buf, ii*2+2, &rc); 3837 } 3838 zSep = " AND "; 3839 } 3840 } 3841 3842 if( rc==SQLITE_OK ){ 3843 char *zSql = (char*)buf.aBuf; 3844 rc = sqlite3_prepare_v2(p->db, zSql, buf.nBuf, &pUp->pStmt, 0); 3845 } 3846 3847 if( rc!=SQLITE_OK ){ 3848 sqlite3_free(pUp); 3849 pUp = 0; 3850 }else{ 3851 pUp->pNext = p->pUp; 3852 p->pUp = pUp; 3853 } 3854 sqlite3_free(buf.aBuf); 3855 } 3856 } 3857 } 3858 3859 assert( (rc==SQLITE_OK)==(pUp!=0) ); 3860 if( pUp ){ 3861 *ppStmt = pUp->pStmt; 3862 }else{ 3863 *ppStmt = 0; 3864 } 3865 return rc; 3866 } 3867 3868 /* 3869 ** Free all cached UPDATE statements. 3870 */ 3871 static void sessionUpdateFree(SessionApplyCtx *p){ 3872 SessionUpdate *pUp; 3873 SessionUpdate *pNext; 3874 for(pUp=p->pUp; pUp; pUp=pNext){ 3875 pNext = pUp->pNext; 3876 sqlite3_finalize(pUp->pStmt); 3877 sqlite3_free(pUp); 3878 } 3879 p->pUp = 0; 3880 sqlite3_free(p->aUpdateMask); 3881 p->aUpdateMask = 0; 3882 } 3883 3884 /* 3885 ** Formulate a statement to DELETE a row from database db. Assuming a table 3886 ** structure like this: 3887 ** 3888 ** CREATE TABLE x(a, b, c, d, PRIMARY KEY(a, c)); 3889 ** 3890 ** The DELETE statement looks like this: 3891 ** 3892 ** DELETE FROM x WHERE a = :1 AND c = :3 AND (:5 OR b IS :2 AND d IS :4) 3893 ** 3894 ** Variable :5 (nCol+1) is a boolean. It should be set to 0 if we require 3895 ** matching b and d values, or 1 otherwise. The second case comes up if the 3896 ** conflict handler is invoked with NOTFOUND and returns CHANGESET_REPLACE. 3897 ** 3898 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pDelete is left 3899 ** pointing to the prepared version of the SQL statement. 3900 */ 3901 static int sessionDeleteRow( 3902 sqlite3 *db, /* Database handle */ 3903 const char *zTab, /* Table name */ 3904 SessionApplyCtx *p /* Session changeset-apply context */ 3905 ){ 3906 int i; 3907 const char *zSep = ""; 3908 int rc = SQLITE_OK; 3909 SessionBuffer buf = {0, 0, 0}; 3910 int nPk = 0; 3911 3912 sessionAppendStr(&buf, "DELETE FROM main.", &rc); 3913 sessionAppendIdent(&buf, zTab, &rc); 3914 sessionAppendStr(&buf, " WHERE ", &rc); 3915 3916 for(i=0; i<p->nCol; i++){ 3917 if( p->abPK[i] ){ 3918 nPk++; 3919 sessionAppendStr(&buf, zSep, &rc); 3920 sessionAppendIdent(&buf, p->azCol[i], &rc); 3921 sessionAppendStr(&buf, " = ?", &rc); 3922 sessionAppendInteger(&buf, i+1, &rc); 3923 zSep = " AND "; 3924 } 3925 } 3926 3927 if( nPk<p->nCol ){ 3928 sessionAppendStr(&buf, " AND (?", &rc); 3929 sessionAppendInteger(&buf, p->nCol+1, &rc); 3930 sessionAppendStr(&buf, " OR ", &rc); 3931 3932 zSep = ""; 3933 for(i=0; i<p->nCol; i++){ 3934 if( !p->abPK[i] ){ 3935 sessionAppendStr(&buf, zSep, &rc); 3936 sessionAppendIdent(&buf, p->azCol[i], &rc); 3937 sessionAppendStr(&buf, " IS ?", &rc); 3938 sessionAppendInteger(&buf, i+1, &rc); 3939 zSep = "AND "; 3940 } 3941 } 3942 sessionAppendStr(&buf, ")", &rc); 3943 } 3944 3945 if( rc==SQLITE_OK ){ 3946 rc = sqlite3_prepare_v2(db, (char *)buf.aBuf, buf.nBuf, &p->pDelete, 0); 3947 } 3948 sqlite3_free(buf.aBuf); 3949 3950 return rc; 3951 } 3952 3953 /* 3954 ** Formulate and prepare an SQL statement to query table zTab by primary 3955 ** key. Assuming the following table structure: 3956 ** 3957 ** CREATE TABLE x(a, b, c, d, PRIMARY KEY(a, c)); 3958 ** 3959 ** The SELECT statement looks like this: 3960 ** 3961 ** SELECT * FROM x WHERE a = ?1 AND c = ?3 3962 ** 3963 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pSelect is left 3964 ** pointing to the prepared version of the SQL statement. 3965 */ 3966 static int sessionSelectRow( 3967 sqlite3 *db, /* Database handle */ 3968 const char *zTab, /* Table name */ 3969 SessionApplyCtx *p /* Session changeset-apply context */ 3970 ){ 3971 return sessionSelectStmt( 3972 db, "main", zTab, p->nCol, p->azCol, p->abPK, &p->pSelect); 3973 } 3974 3975 /* 3976 ** Formulate and prepare an INSERT statement to add a record to table zTab. 3977 ** For example: 3978 ** 3979 ** INSERT INTO main."zTab" VALUES(?1, ?2, ?3 ...); 3980 ** 3981 ** If successful, SQLITE_OK is returned and SessionApplyCtx.pInsert is left 3982 ** pointing to the prepared version of the SQL statement. 3983 */ 3984 static int sessionInsertRow( 3985 sqlite3 *db, /* Database handle */ 3986 const char *zTab, /* Table name */ 3987 SessionApplyCtx *p /* Session changeset-apply context */ 3988 ){ 3989 int rc = SQLITE_OK; 3990 int i; 3991 SessionBuffer buf = {0, 0, 0}; 3992 3993 sessionAppendStr(&buf, "INSERT INTO main.", &rc); 3994 sessionAppendIdent(&buf, zTab, &rc); 3995 sessionAppendStr(&buf, "(", &rc); 3996 for(i=0; i<p->nCol; i++){ 3997 if( i!=0 ) sessionAppendStr(&buf, ", ", &rc); 3998 sessionAppendIdent(&buf, p->azCol[i], &rc); 3999 } 4000 4001 sessionAppendStr(&buf, ") VALUES(?", &rc); 4002 for(i=1; i<p->nCol; i++){ 4003 sessionAppendStr(&buf, ", ?", &rc); 4004 } 4005 sessionAppendStr(&buf, ")", &rc); 4006 4007 if( rc==SQLITE_OK ){ 4008 rc = sqlite3_prepare_v2(db, (char *)buf.aBuf, buf.nBuf, &p->pInsert, 0); 4009 } 4010 sqlite3_free(buf.aBuf); 4011 return rc; 4012 } 4013 4014 static int sessionPrepare(sqlite3 *db, sqlite3_stmt **pp, const char *zSql){ 4015 return sqlite3_prepare_v2(db, zSql, -1, pp, 0); 4016 } 4017 4018 /* 4019 ** Prepare statements for applying changes to the sqlite_stat1 table. 4020 ** These are similar to those created by sessionSelectRow(), 4021 ** sessionInsertRow(), sessionUpdateRow() and sessionDeleteRow() for 4022 ** other tables. 4023 */ 4024 static int sessionStat1Sql(sqlite3 *db, SessionApplyCtx *p){ 4025 int rc = sessionSelectRow(db, "sqlite_stat1", p); 4026 if( rc==SQLITE_OK ){ 4027 rc = sessionPrepare(db, &p->pInsert, 4028 "INSERT INTO main.sqlite_stat1 VALUES(?1, " 4029 "CASE WHEN length(?2)=0 AND typeof(?2)='blob' THEN NULL ELSE ?2 END, " 4030 "?3)" 4031 ); 4032 } 4033 if( rc==SQLITE_OK ){ 4034 rc = sessionPrepare(db, &p->pDelete, 4035 "DELETE FROM main.sqlite_stat1 WHERE tbl=?1 AND idx IS " 4036 "CASE WHEN length(?2)=0 AND typeof(?2)='blob' THEN NULL ELSE ?2 END " 4037 "AND (?4 OR stat IS ?3)" 4038 ); 4039 } 4040 return rc; 4041 } 4042 4043 /* 4044 ** A wrapper around sqlite3_bind_value() that detects an extra problem. 4045 ** See comments in the body of this function for details. 4046 */ 4047 static int sessionBindValue( 4048 sqlite3_stmt *pStmt, /* Statement to bind value to */ 4049 int i, /* Parameter number to bind to */ 4050 sqlite3_value *pVal /* Value to bind */ 4051 ){ 4052 int eType = sqlite3_value_type(pVal); 4053 /* COVERAGE: The (pVal->z==0) branch is never true using current versions 4054 ** of SQLite. If a malloc fails in an sqlite3_value_xxx() function, either 4055 ** the (pVal->z) variable remains as it was or the type of the value is 4056 ** set to SQLITE_NULL. */ 4057 if( (eType==SQLITE_TEXT || eType==SQLITE_BLOB) && pVal->z==0 ){ 4058 /* This condition occurs when an earlier OOM in a call to 4059 ** sqlite3_value_text() or sqlite3_value_blob() (perhaps from within 4060 ** a conflict-handler) has zeroed the pVal->z pointer. Return NOMEM. */ 4061 return SQLITE_NOMEM; 4062 } 4063 return sqlite3_bind_value(pStmt, i, pVal); 4064 } 4065 4066 /* 4067 ** Iterator pIter must point to an SQLITE_INSERT entry. This function 4068 ** transfers new.* values from the current iterator entry to statement 4069 ** pStmt. The table being inserted into has nCol columns. 4070 ** 4071 ** New.* value $i from the iterator is bound to variable ($i+1) of 4072 ** statement pStmt. If parameter abPK is NULL, all values from 0 to (nCol-1) 4073 ** are transfered to the statement. Otherwise, if abPK is not NULL, it points 4074 ** to an array nCol elements in size. In this case only those values for 4075 ** which abPK[$i] is true are read from the iterator and bound to the 4076 ** statement. 4077 ** 4078 ** An SQLite error code is returned if an error occurs. Otherwise, SQLITE_OK. 4079 */ 4080 static int sessionBindRow( 4081 sqlite3_changeset_iter *pIter, /* Iterator to read values from */ 4082 int(*xValue)(sqlite3_changeset_iter *, int, sqlite3_value **), 4083 int nCol, /* Number of columns */ 4084 u8 *abPK, /* If not NULL, bind only if true */ 4085 sqlite3_stmt *pStmt /* Bind values to this statement */ 4086 ){ 4087 int i; 4088 int rc = SQLITE_OK; 4089 4090 /* Neither sqlite3changeset_old or sqlite3changeset_new can fail if the 4091 ** argument iterator points to a suitable entry. Make sure that xValue 4092 ** is one of these to guarantee that it is safe to ignore the return 4093 ** in the code below. */ 4094 assert( xValue==sqlite3changeset_old || xValue==sqlite3changeset_new ); 4095 4096 for(i=0; rc==SQLITE_OK && i<nCol; i++){ 4097 if( !abPK || abPK[i] ){ 4098 sqlite3_value *pVal; 4099 (void)xValue(pIter, i, &pVal); 4100 if( pVal==0 ){ 4101 /* The value in the changeset was "undefined". This indicates a 4102 ** corrupt changeset blob. */ 4103 rc = SQLITE_CORRUPT_BKPT; 4104 }else{ 4105 rc = sessionBindValue(pStmt, i+1, pVal); 4106 } 4107 } 4108 } 4109 return rc; 4110 } 4111 4112 /* 4113 ** SQL statement pSelect is as generated by the sessionSelectRow() function. 4114 ** This function binds the primary key values from the change that changeset 4115 ** iterator pIter points to to the SELECT and attempts to seek to the table 4116 ** entry. If a row is found, the SELECT statement left pointing at the row 4117 ** and SQLITE_ROW is returned. Otherwise, if no row is found and no error 4118 ** has occured, the statement is reset and SQLITE_OK is returned. If an 4119 ** error occurs, the statement is reset and an SQLite error code is returned. 4120 ** 4121 ** If this function returns SQLITE_ROW, the caller must eventually reset() 4122 ** statement pSelect. If any other value is returned, the statement does 4123 ** not require a reset(). 4124 ** 4125 ** If the iterator currently points to an INSERT record, bind values from the 4126 ** new.* record to the SELECT statement. Or, if it points to a DELETE or 4127 ** UPDATE, bind values from the old.* record. 4128 */ 4129 static int sessionSeekToRow( 4130 sqlite3 *db, /* Database handle */ 4131 sqlite3_changeset_iter *pIter, /* Changeset iterator */ 4132 u8 *abPK, /* Primary key flags array */ 4133 sqlite3_stmt *pSelect /* SELECT statement from sessionSelectRow() */ 4134 ){ 4135 int rc; /* Return code */ 4136 int nCol; /* Number of columns in table */ 4137 int op; /* Changset operation (SQLITE_UPDATE etc.) */ 4138 const char *zDummy; /* Unused */ 4139 4140 sqlite3changeset_op(pIter, &zDummy, &nCol, &op, 0); 4141 rc = sessionBindRow(pIter, 4142 op==SQLITE_INSERT ? sqlite3changeset_new : sqlite3changeset_old, 4143 nCol, abPK, pSelect 4144 ); 4145 4146 if( rc==SQLITE_OK ){ 4147 rc = sqlite3_step(pSelect); 4148 if( rc!=SQLITE_ROW ) rc = sqlite3_reset(pSelect); 4149 } 4150 4151 return rc; 4152 } 4153 4154 /* 4155 ** This function is called from within sqlite3changeset_apply_v2() when 4156 ** a conflict is encountered and resolved using conflict resolution 4157 ** mode eType (either SQLITE_CHANGESET_OMIT or SQLITE_CHANGESET_REPLACE).. 4158 ** It adds a conflict resolution record to the buffer in 4159 ** SessionApplyCtx.rebase, which will eventually be returned to the caller 4160 ** of apply_v2() as the "rebase" buffer. 4161 ** 4162 ** Return SQLITE_OK if successful, or an SQLite error code otherwise. 4163 */ 4164 static int sessionRebaseAdd( 4165 SessionApplyCtx *p, /* Apply context */ 4166 int eType, /* Conflict resolution (OMIT or REPLACE) */ 4167 sqlite3_changeset_iter *pIter /* Iterator pointing at current change */ 4168 ){ 4169 int rc = SQLITE_OK; 4170 if( p->bRebase ){ 4171 int i; 4172 int eOp = pIter->op; 4173 if( p->bRebaseStarted==0 ){ 4174 /* Append a table-header to the rebase buffer */ 4175 const char *zTab = pIter->zTab; 4176 sessionAppendByte(&p->rebase, 'T', &rc); 4177 sessionAppendVarint(&p->rebase, p->nCol, &rc); 4178 sessionAppendBlob(&p->rebase, p->abPK, p->nCol, &rc); 4179 sessionAppendBlob(&p->rebase, (u8*)zTab, (int)strlen(zTab)+1, &rc); 4180 p->bRebaseStarted = 1; 4181 } 4182 4183 assert( eType==SQLITE_CHANGESET_REPLACE||eType==SQLITE_CHANGESET_OMIT ); 4184 assert( eOp==SQLITE_DELETE || eOp==SQLITE_INSERT || eOp==SQLITE_UPDATE ); 4185 4186 sessionAppendByte(&p->rebase, 4187 (eOp==SQLITE_DELETE ? SQLITE_DELETE : SQLITE_INSERT), &rc 4188 ); 4189 sessionAppendByte(&p->rebase, (eType==SQLITE_CHANGESET_REPLACE), &rc); 4190 for(i=0; i<p->nCol; i++){ 4191 sqlite3_value *pVal = 0; 4192 if( eOp==SQLITE_DELETE || (eOp==SQLITE_UPDATE && p->abPK[i]) ){ 4193 sqlite3changeset_old(pIter, i, &pVal); 4194 }else{ 4195 sqlite3changeset_new(pIter, i, &pVal); 4196 } 4197 sessionAppendValue(&p->rebase, pVal, &rc); 4198 } 4199 } 4200 return rc; 4201 } 4202 4203 /* 4204 ** Invoke the conflict handler for the change that the changeset iterator 4205 ** currently points to. 4206 ** 4207 ** Argument eType must be either CHANGESET_DATA or CHANGESET_CONFLICT. 4208 ** If argument pbReplace is NULL, then the type of conflict handler invoked 4209 ** depends solely on eType, as follows: 4210 ** 4211 ** eType value Value passed to xConflict 4212 ** ------------------------------------------------- 4213 ** CHANGESET_DATA CHANGESET_NOTFOUND 4214 ** CHANGESET_CONFLICT CHANGESET_CONSTRAINT 4215 ** 4216 ** Or, if pbReplace is not NULL, then an attempt is made to find an existing 4217 ** record with the same primary key as the record about to be deleted, updated 4218 ** or inserted. If such a record can be found, it is available to the conflict 4219 ** handler as the "conflicting" record. In this case the type of conflict 4220 ** handler invoked is as follows: 4221 ** 4222 ** eType value PK Record found? Value passed to xConflict 4223 ** ---------------------------------------------------------------- 4224 ** CHANGESET_DATA Yes CHANGESET_DATA 4225 ** CHANGESET_DATA No CHANGESET_NOTFOUND 4226 ** CHANGESET_CONFLICT Yes CHANGESET_CONFLICT 4227 ** CHANGESET_CONFLICT No CHANGESET_CONSTRAINT 4228 ** 4229 ** If pbReplace is not NULL, and a record with a matching PK is found, and 4230 ** the conflict handler function returns SQLITE_CHANGESET_REPLACE, *pbReplace 4231 ** is set to non-zero before returning SQLITE_OK. 4232 ** 4233 ** If the conflict handler returns SQLITE_CHANGESET_ABORT, SQLITE_ABORT is 4234 ** returned. Or, if the conflict handler returns an invalid value, 4235 ** SQLITE_MISUSE. If the conflict handler returns SQLITE_CHANGESET_OMIT, 4236 ** this function returns SQLITE_OK. 4237 */ 4238 static int sessionConflictHandler( 4239 int eType, /* Either CHANGESET_DATA or CONFLICT */ 4240 SessionApplyCtx *p, /* changeset_apply() context */ 4241 sqlite3_changeset_iter *pIter, /* Changeset iterator */ 4242 int(*xConflict)(void *, int, sqlite3_changeset_iter*), 4243 void *pCtx, /* First argument for conflict handler */ 4244 int *pbReplace /* OUT: Set to true if PK row is found */ 4245 ){ 4246 int res = 0; /* Value returned by conflict handler */ 4247 int rc; 4248 int nCol; 4249 int op; 4250 const char *zDummy; 4251 4252 sqlite3changeset_op(pIter, &zDummy, &nCol, &op, 0); 4253 4254 assert( eType==SQLITE_CHANGESET_CONFLICT || eType==SQLITE_CHANGESET_DATA ); 4255 assert( SQLITE_CHANGESET_CONFLICT+1==SQLITE_CHANGESET_CONSTRAINT ); 4256 assert( SQLITE_CHANGESET_DATA+1==SQLITE_CHANGESET_NOTFOUND ); 4257 4258 /* Bind the new.* PRIMARY KEY values to the SELECT statement. */ 4259 if( pbReplace ){ 4260 rc = sessionSeekToRow(p->db, pIter, p->abPK, p->pSelect); 4261 }else{ 4262 rc = SQLITE_OK; 4263 } 4264 4265 if( rc==SQLITE_ROW ){ 4266 /* There exists another row with the new.* primary key. */ 4267 pIter->pConflict = p->pSelect; 4268 res = xConflict(pCtx, eType, pIter); 4269 pIter->pConflict = 0; 4270 rc = sqlite3_reset(p->pSelect); 4271 }else if( rc==SQLITE_OK ){ 4272 if( p->bDeferConstraints && eType==SQLITE_CHANGESET_CONFLICT ){ 4273 /* Instead of invoking the conflict handler, append the change blob 4274 ** to the SessionApplyCtx.constraints buffer. */ 4275 u8 *aBlob = &pIter->in.aData[pIter->in.iCurrent]; 4276 int nBlob = pIter->in.iNext - pIter->in.iCurrent; 4277 sessionAppendBlob(&p->constraints, aBlob, nBlob, &rc); 4278 return SQLITE_OK; 4279 }else{ 4280 /* No other row with the new.* primary key. */ 4281 res = xConflict(pCtx, eType+1, pIter); 4282 if( res==SQLITE_CHANGESET_REPLACE ) rc = SQLITE_MISUSE; 4283 } 4284 } 4285 4286 if( rc==SQLITE_OK ){ 4287 switch( res ){ 4288 case SQLITE_CHANGESET_REPLACE: 4289 assert( pbReplace ); 4290 *pbReplace = 1; 4291 break; 4292 4293 case SQLITE_CHANGESET_OMIT: 4294 break; 4295 4296 case SQLITE_CHANGESET_ABORT: 4297 rc = SQLITE_ABORT; 4298 break; 4299 4300 default: 4301 rc = SQLITE_MISUSE; 4302 break; 4303 } 4304 if( rc==SQLITE_OK ){ 4305 rc = sessionRebaseAdd(p, res, pIter); 4306 } 4307 } 4308 4309 return rc; 4310 } 4311 4312 /* 4313 ** Attempt to apply the change that the iterator passed as the first argument 4314 ** currently points to to the database. If a conflict is encountered, invoke 4315 ** the conflict handler callback. 4316 ** 4317 ** If argument pbRetry is NULL, then ignore any CHANGESET_DATA conflict. If 4318 ** one is encountered, update or delete the row with the matching primary key 4319 ** instead. Or, if pbRetry is not NULL and a CHANGESET_DATA conflict occurs, 4320 ** invoke the conflict handler. If it returns CHANGESET_REPLACE, set *pbRetry 4321 ** to true before returning. In this case the caller will invoke this function 4322 ** again, this time with pbRetry set to NULL. 4323 ** 4324 ** If argument pbReplace is NULL and a CHANGESET_CONFLICT conflict is 4325 ** encountered invoke the conflict handler with CHANGESET_CONSTRAINT instead. 4326 ** Or, if pbReplace is not NULL, invoke it with CHANGESET_CONFLICT. If such 4327 ** an invocation returns SQLITE_CHANGESET_REPLACE, set *pbReplace to true 4328 ** before retrying. In this case the caller attempts to remove the conflicting 4329 ** row before invoking this function again, this time with pbReplace set 4330 ** to NULL. 4331 ** 4332 ** If any conflict handler returns SQLITE_CHANGESET_ABORT, this function 4333 ** returns SQLITE_ABORT. Otherwise, if no error occurs, SQLITE_OK is 4334 ** returned. 4335 */ 4336 static int sessionApplyOneOp( 4337 sqlite3_changeset_iter *pIter, /* Changeset iterator */ 4338 SessionApplyCtx *p, /* changeset_apply() context */ 4339 int(*xConflict)(void *, int, sqlite3_changeset_iter *), 4340 void *pCtx, /* First argument for the conflict handler */ 4341 int *pbReplace, /* OUT: True to remove PK row and retry */ 4342 int *pbRetry /* OUT: True to retry. */ 4343 ){ 4344 const char *zDummy; 4345 int op; 4346 int nCol; 4347 int rc = SQLITE_OK; 4348 4349 assert( p->pDelete && p->pInsert && p->pSelect ); 4350 assert( p->azCol && p->abPK ); 4351 assert( !pbReplace || *pbReplace==0 ); 4352 4353 sqlite3changeset_op(pIter, &zDummy, &nCol, &op, 0); 4354 4355 if( op==SQLITE_DELETE ){ 4356 4357 /* Bind values to the DELETE statement. If conflict handling is required, 4358 ** bind values for all columns and set bound variable (nCol+1) to true. 4359 ** Or, if conflict handling is not required, bind just the PK column 4360 ** values and, if it exists, set (nCol+1) to false. Conflict handling 4361 ** is not required if: 4362 ** 4363 ** * this is a patchset, or 4364 ** * (pbRetry==0), or 4365 ** * all columns of the table are PK columns (in this case there is 4366 ** no (nCol+1) variable to bind to). 4367 */ 4368 u8 *abPK = (pIter->bPatchset ? p->abPK : 0); 4369 rc = sessionBindRow(pIter, sqlite3changeset_old, nCol, abPK, p->pDelete); 4370 if( rc==SQLITE_OK && sqlite3_bind_parameter_count(p->pDelete)>nCol ){ 4371 rc = sqlite3_bind_int(p->pDelete, nCol+1, (pbRetry==0 || abPK)); 4372 } 4373 if( rc!=SQLITE_OK ) return rc; 4374 4375 sqlite3_step(p->pDelete); 4376 rc = sqlite3_reset(p->pDelete); 4377 if( rc==SQLITE_OK && sqlite3_changes(p->db)==0 ){ 4378 rc = sessionConflictHandler( 4379 SQLITE_CHANGESET_DATA, p, pIter, xConflict, pCtx, pbRetry 4380 ); 4381 }else if( (rc&0xff)==SQLITE_CONSTRAINT ){ 4382 rc = sessionConflictHandler( 4383 SQLITE_CHANGESET_CONFLICT, p, pIter, xConflict, pCtx, 0 4384 ); 4385 } 4386 4387 }else if( op==SQLITE_UPDATE ){ 4388 int i; 4389 sqlite3_stmt *pUp = 0; 4390 int bPatchset = (pbRetry==0 || pIter->bPatchset); 4391 4392 rc = sessionUpdateFind(pIter, p, bPatchset, &pUp); 4393 4394 /* Bind values to the UPDATE statement. */ 4395 for(i=0; rc==SQLITE_OK && i<nCol; i++){ 4396 sqlite3_value *pOld = sessionChangesetOld(pIter, i); 4397 sqlite3_value *pNew = sessionChangesetNew(pIter, i); 4398 if( p->abPK[i] || (bPatchset==0 && pOld) ){ 4399 rc = sessionBindValue(pUp, i*2+2, pOld); 4400 } 4401 if( rc==SQLITE_OK && pNew ){ 4402 rc = sessionBindValue(pUp, i*2+1, pNew); 4403 } 4404 } 4405 if( rc!=SQLITE_OK ) return rc; 4406 4407 /* Attempt the UPDATE. In the case of a NOTFOUND or DATA conflict, 4408 ** the result will be SQLITE_OK with 0 rows modified. */ 4409 sqlite3_step(pUp); 4410 rc = sqlite3_reset(pUp); 4411 4412 if( rc==SQLITE_OK && sqlite3_changes(p->db)==0 ){ 4413 /* A NOTFOUND or DATA error. Search the table to see if it contains 4414 ** a row with a matching primary key. If so, this is a DATA conflict. 4415 ** Otherwise, if there is no primary key match, it is a NOTFOUND. */ 4416 4417 rc = sessionConflictHandler( 4418 SQLITE_CHANGESET_DATA, p, pIter, xConflict, pCtx, pbRetry 4419 ); 4420 4421 }else if( (rc&0xff)==SQLITE_CONSTRAINT ){ 4422 /* This is always a CONSTRAINT conflict. */ 4423 rc = sessionConflictHandler( 4424 SQLITE_CHANGESET_CONFLICT, p, pIter, xConflict, pCtx, 0 4425 ); 4426 } 4427 4428 }else{ 4429 assert( op==SQLITE_INSERT ); 4430 if( p->bStat1 ){ 4431 /* Check if there is a conflicting row. For sqlite_stat1, this needs 4432 ** to be done using a SELECT, as there is no PRIMARY KEY in the 4433 ** database schema to throw an exception if a duplicate is inserted. */ 4434 rc = sessionSeekToRow(p->db, pIter, p->abPK, p->pSelect); 4435 if( rc==SQLITE_ROW ){ 4436 rc = SQLITE_CONSTRAINT; 4437 sqlite3_reset(p->pSelect); 4438 } 4439 } 4440 4441 if( rc==SQLITE_OK ){ 4442 rc = sessionBindRow(pIter, sqlite3changeset_new, nCol, 0, p->pInsert); 4443 if( rc!=SQLITE_OK ) return rc; 4444 4445 sqlite3_step(p->pInsert); 4446 rc = sqlite3_reset(p->pInsert); 4447 } 4448 4449 if( (rc&0xff)==SQLITE_CONSTRAINT ){ 4450 rc = sessionConflictHandler( 4451 SQLITE_CHANGESET_CONFLICT, p, pIter, xConflict, pCtx, pbReplace 4452 ); 4453 } 4454 } 4455 4456 return rc; 4457 } 4458 4459 /* 4460 ** Attempt to apply the change that the iterator passed as the first argument 4461 ** currently points to to the database. If a conflict is encountered, invoke 4462 ** the conflict handler callback. 4463 ** 4464 ** The difference between this function and sessionApplyOne() is that this 4465 ** function handles the case where the conflict-handler is invoked and 4466 ** returns SQLITE_CHANGESET_REPLACE - indicating that the change should be 4467 ** retried in some manner. 4468 */ 4469 static int sessionApplyOneWithRetry( 4470 sqlite3 *db, /* Apply change to "main" db of this handle */ 4471 sqlite3_changeset_iter *pIter, /* Changeset iterator to read change from */ 4472 SessionApplyCtx *pApply, /* Apply context */ 4473 int(*xConflict)(void*, int, sqlite3_changeset_iter*), 4474 void *pCtx /* First argument passed to xConflict */ 4475 ){ 4476 int bReplace = 0; 4477 int bRetry = 0; 4478 int rc; 4479 4480 rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, &bReplace, &bRetry); 4481 if( rc==SQLITE_OK ){ 4482 /* If the bRetry flag is set, the change has not been applied due to an 4483 ** SQLITE_CHANGESET_DATA problem (i.e. this is an UPDATE or DELETE and 4484 ** a row with the correct PK is present in the db, but one or more other 4485 ** fields do not contain the expected values) and the conflict handler 4486 ** returned SQLITE_CHANGESET_REPLACE. In this case retry the operation, 4487 ** but pass NULL as the final argument so that sessionApplyOneOp() ignores 4488 ** the SQLITE_CHANGESET_DATA problem. */ 4489 if( bRetry ){ 4490 assert( pIter->op==SQLITE_UPDATE || pIter->op==SQLITE_DELETE ); 4491 rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, 0, 0); 4492 } 4493 4494 /* If the bReplace flag is set, the change is an INSERT that has not 4495 ** been performed because the database already contains a row with the 4496 ** specified primary key and the conflict handler returned 4497 ** SQLITE_CHANGESET_REPLACE. In this case remove the conflicting row 4498 ** before reattempting the INSERT. */ 4499 else if( bReplace ){ 4500 assert( pIter->op==SQLITE_INSERT ); 4501 rc = sqlite3_exec(db, "SAVEPOINT replace_op", 0, 0, 0); 4502 if( rc==SQLITE_OK ){ 4503 rc = sessionBindRow(pIter, 4504 sqlite3changeset_new, pApply->nCol, pApply->abPK, pApply->pDelete); 4505 sqlite3_bind_int(pApply->pDelete, pApply->nCol+1, 1); 4506 } 4507 if( rc==SQLITE_OK ){ 4508 sqlite3_step(pApply->pDelete); 4509 rc = sqlite3_reset(pApply->pDelete); 4510 } 4511 if( rc==SQLITE_OK ){ 4512 rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, 0, 0); 4513 } 4514 if( rc==SQLITE_OK ){ 4515 rc = sqlite3_exec(db, "RELEASE replace_op", 0, 0, 0); 4516 } 4517 } 4518 } 4519 4520 return rc; 4521 } 4522 4523 /* 4524 ** Retry the changes accumulated in the pApply->constraints buffer. 4525 */ 4526 static int sessionRetryConstraints( 4527 sqlite3 *db, 4528 int bPatchset, 4529 const char *zTab, 4530 SessionApplyCtx *pApply, 4531 int(*xConflict)(void*, int, sqlite3_changeset_iter*), 4532 void *pCtx /* First argument passed to xConflict */ 4533 ){ 4534 int rc = SQLITE_OK; 4535 4536 while( pApply->constraints.nBuf ){ 4537 sqlite3_changeset_iter *pIter2 = 0; 4538 SessionBuffer cons = pApply->constraints; 4539 memset(&pApply->constraints, 0, sizeof(SessionBuffer)); 4540 4541 rc = sessionChangesetStart( 4542 &pIter2, 0, 0, cons.nBuf, cons.aBuf, pApply->bInvertConstraints, 1 4543 ); 4544 if( rc==SQLITE_OK ){ 4545 size_t nByte = 2*pApply->nCol*sizeof(sqlite3_value*); 4546 int rc2; 4547 pIter2->bPatchset = bPatchset; 4548 pIter2->zTab = (char*)zTab; 4549 pIter2->nCol = pApply->nCol; 4550 pIter2->abPK = pApply->abPK; 4551 sessionBufferGrow(&pIter2->tblhdr, nByte, &rc); 4552 pIter2->apValue = (sqlite3_value**)pIter2->tblhdr.aBuf; 4553 if( rc==SQLITE_OK ) memset(pIter2->apValue, 0, nByte); 4554 4555 while( rc==SQLITE_OK && SQLITE_ROW==sqlite3changeset_next(pIter2) ){ 4556 rc = sessionApplyOneWithRetry(db, pIter2, pApply, xConflict, pCtx); 4557 } 4558 4559 rc2 = sqlite3changeset_finalize(pIter2); 4560 if( rc==SQLITE_OK ) rc = rc2; 4561 } 4562 assert( pApply->bDeferConstraints || pApply->constraints.nBuf==0 ); 4563 4564 sqlite3_free(cons.aBuf); 4565 if( rc!=SQLITE_OK ) break; 4566 if( pApply->constraints.nBuf>=cons.nBuf ){ 4567 /* No progress was made on the last round. */ 4568 pApply->bDeferConstraints = 0; 4569 } 4570 } 4571 4572 return rc; 4573 } 4574 4575 /* 4576 ** Argument pIter is a changeset iterator that has been initialized, but 4577 ** not yet passed to sqlite3changeset_next(). This function applies the 4578 ** changeset to the main database attached to handle "db". The supplied 4579 ** conflict handler callback is invoked to resolve any conflicts encountered 4580 ** while applying the change. 4581 */ 4582 static int sessionChangesetApply( 4583 sqlite3 *db, /* Apply change to "main" db of this handle */ 4584 sqlite3_changeset_iter *pIter, /* Changeset to apply */ 4585 int(*xFilter)( 4586 void *pCtx, /* Copy of sixth arg to _apply() */ 4587 const char *zTab /* Table name */ 4588 ), 4589 int(*xConflict)( 4590 void *pCtx, /* Copy of fifth arg to _apply() */ 4591 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ 4592 sqlite3_changeset_iter *p /* Handle describing change and conflict */ 4593 ), 4594 void *pCtx, /* First argument passed to xConflict */ 4595 void **ppRebase, int *pnRebase, /* OUT: Rebase information */ 4596 int flags /* SESSION_APPLY_XXX flags */ 4597 ){ 4598 int schemaMismatch = 0; 4599 int rc = SQLITE_OK; /* Return code */ 4600 const char *zTab = 0; /* Name of current table */ 4601 int nTab = 0; /* Result of sqlite3Strlen30(zTab) */ 4602 SessionApplyCtx sApply; /* changeset_apply() context object */ 4603 int bPatchset; 4604 4605 assert( xConflict!=0 ); 4606 4607 pIter->in.bNoDiscard = 1; 4608 memset(&sApply, 0, sizeof(sApply)); 4609 sApply.bRebase = (ppRebase && pnRebase); 4610 sApply.bInvertConstraints = !!(flags & SQLITE_CHANGESETAPPLY_INVERT); 4611 sqlite3_mutex_enter(sqlite3_db_mutex(db)); 4612 if( (flags & SQLITE_CHANGESETAPPLY_NOSAVEPOINT)==0 ){ 4613 rc = sqlite3_exec(db, "SAVEPOINT changeset_apply", 0, 0, 0); 4614 } 4615 if( rc==SQLITE_OK ){ 4616 rc = sqlite3_exec(db, "PRAGMA defer_foreign_keys = 1", 0, 0, 0); 4617 } 4618 while( rc==SQLITE_OK && SQLITE_ROW==sqlite3changeset_next(pIter) ){ 4619 int nCol; 4620 int op; 4621 const char *zNew; 4622 4623 sqlite3changeset_op(pIter, &zNew, &nCol, &op, 0); 4624 4625 if( zTab==0 || sqlite3_strnicmp(zNew, zTab, nTab+1) ){ 4626 u8 *abPK; 4627 4628 rc = sessionRetryConstraints( 4629 db, pIter->bPatchset, zTab, &sApply, xConflict, pCtx 4630 ); 4631 if( rc!=SQLITE_OK ) break; 4632 4633 sessionUpdateFree(&sApply); 4634 sqlite3_free((char*)sApply.azCol); /* cast works around VC++ bug */ 4635 sqlite3_finalize(sApply.pDelete); 4636 sqlite3_finalize(sApply.pInsert); 4637 sqlite3_finalize(sApply.pSelect); 4638 sApply.db = db; 4639 sApply.pDelete = 0; 4640 sApply.pInsert = 0; 4641 sApply.pSelect = 0; 4642 sApply.nCol = 0; 4643 sApply.azCol = 0; 4644 sApply.abPK = 0; 4645 sApply.bStat1 = 0; 4646 sApply.bDeferConstraints = 1; 4647 sApply.bRebaseStarted = 0; 4648 memset(&sApply.constraints, 0, sizeof(SessionBuffer)); 4649 4650 /* If an xFilter() callback was specified, invoke it now. If the 4651 ** xFilter callback returns zero, skip this table. If it returns 4652 ** non-zero, proceed. */ 4653 schemaMismatch = (xFilter && (0==xFilter(pCtx, zNew))); 4654 if( schemaMismatch ){ 4655 zTab = sqlite3_mprintf("%s", zNew); 4656 if( zTab==0 ){ 4657 rc = SQLITE_NOMEM; 4658 break; 4659 } 4660 nTab = (int)strlen(zTab); 4661 sApply.azCol = (const char **)zTab; 4662 }else{ 4663 int nMinCol = 0; 4664 int i; 4665 4666 sqlite3changeset_pk(pIter, &abPK, 0); 4667 rc = sessionTableInfo(0, 4668 db, "main", zNew, &sApply.nCol, &zTab, &sApply.azCol, &sApply.abPK 4669 ); 4670 if( rc!=SQLITE_OK ) break; 4671 for(i=0; i<sApply.nCol; i++){ 4672 if( sApply.abPK[i] ) nMinCol = i+1; 4673 } 4674 4675 if( sApply.nCol==0 ){ 4676 schemaMismatch = 1; 4677 sqlite3_log(SQLITE_SCHEMA, 4678 "sqlite3changeset_apply(): no such table: %s", zTab 4679 ); 4680 } 4681 else if( sApply.nCol<nCol ){ 4682 schemaMismatch = 1; 4683 sqlite3_log(SQLITE_SCHEMA, 4684 "sqlite3changeset_apply(): table %s has %d columns, " 4685 "expected %d or more", 4686 zTab, sApply.nCol, nCol 4687 ); 4688 } 4689 else if( nCol<nMinCol || memcmp(sApply.abPK, abPK, nCol)!=0 ){ 4690 schemaMismatch = 1; 4691 sqlite3_log(SQLITE_SCHEMA, "sqlite3changeset_apply(): " 4692 "primary key mismatch for table %s", zTab 4693 ); 4694 } 4695 else{ 4696 sApply.nCol = nCol; 4697 if( 0==sqlite3_stricmp(zTab, "sqlite_stat1") ){ 4698 if( (rc = sessionStat1Sql(db, &sApply) ) ){ 4699 break; 4700 } 4701 sApply.bStat1 = 1; 4702 }else{ 4703 if( (rc = sessionSelectRow(db, zTab, &sApply)) 4704 || (rc = sessionDeleteRow(db, zTab, &sApply)) 4705 || (rc = sessionInsertRow(db, zTab, &sApply)) 4706 ){ 4707 break; 4708 } 4709 sApply.bStat1 = 0; 4710 } 4711 } 4712 nTab = sqlite3Strlen30(zTab); 4713 } 4714 } 4715 4716 /* If there is a schema mismatch on the current table, proceed to the 4717 ** next change. A log message has already been issued. */ 4718 if( schemaMismatch ) continue; 4719 4720 rc = sessionApplyOneWithRetry(db, pIter, &sApply, xConflict, pCtx); 4721 } 4722 4723 bPatchset = pIter->bPatchset; 4724 if( rc==SQLITE_OK ){ 4725 rc = sqlite3changeset_finalize(pIter); 4726 }else{ 4727 sqlite3changeset_finalize(pIter); 4728 } 4729 4730 if( rc==SQLITE_OK ){ 4731 rc = sessionRetryConstraints(db, bPatchset, zTab, &sApply, xConflict, pCtx); 4732 } 4733 4734 if( rc==SQLITE_OK ){ 4735 int nFk, notUsed; 4736 sqlite3_db_status(db, SQLITE_DBSTATUS_DEFERRED_FKS, &nFk, ¬Used, 0); 4737 if( nFk!=0 ){ 4738 int res = SQLITE_CHANGESET_ABORT; 4739 sqlite3_changeset_iter sIter; 4740 memset(&sIter, 0, sizeof(sIter)); 4741 sIter.nCol = nFk; 4742 res = xConflict(pCtx, SQLITE_CHANGESET_FOREIGN_KEY, &sIter); 4743 if( res!=SQLITE_CHANGESET_OMIT ){ 4744 rc = SQLITE_CONSTRAINT; 4745 } 4746 } 4747 } 4748 sqlite3_exec(db, "PRAGMA defer_foreign_keys = 0", 0, 0, 0); 4749 4750 if( (flags & SQLITE_CHANGESETAPPLY_NOSAVEPOINT)==0 ){ 4751 if( rc==SQLITE_OK ){ 4752 rc = sqlite3_exec(db, "RELEASE changeset_apply", 0, 0, 0); 4753 }else{ 4754 sqlite3_exec(db, "ROLLBACK TO changeset_apply", 0, 0, 0); 4755 sqlite3_exec(db, "RELEASE changeset_apply", 0, 0, 0); 4756 } 4757 } 4758 4759 assert( sApply.bRebase || sApply.rebase.nBuf==0 ); 4760 if( rc==SQLITE_OK && bPatchset==0 && sApply.bRebase ){ 4761 *ppRebase = (void*)sApply.rebase.aBuf; 4762 *pnRebase = sApply.rebase.nBuf; 4763 sApply.rebase.aBuf = 0; 4764 } 4765 sessionUpdateFree(&sApply); 4766 sqlite3_finalize(sApply.pInsert); 4767 sqlite3_finalize(sApply.pDelete); 4768 sqlite3_finalize(sApply.pSelect); 4769 sqlite3_free((char*)sApply.azCol); /* cast works around VC++ bug */ 4770 sqlite3_free((char*)sApply.constraints.aBuf); 4771 sqlite3_free((char*)sApply.rebase.aBuf); 4772 sqlite3_mutex_leave(sqlite3_db_mutex(db)); 4773 return rc; 4774 } 4775 4776 /* 4777 ** Apply the changeset passed via pChangeset/nChangeset to the main 4778 ** database attached to handle "db". 4779 */ 4780 int sqlite3changeset_apply_v2( 4781 sqlite3 *db, /* Apply change to "main" db of this handle */ 4782 int nChangeset, /* Size of changeset in bytes */ 4783 void *pChangeset, /* Changeset blob */ 4784 int(*xFilter)( 4785 void *pCtx, /* Copy of sixth arg to _apply() */ 4786 const char *zTab /* Table name */ 4787 ), 4788 int(*xConflict)( 4789 void *pCtx, /* Copy of sixth arg to _apply() */ 4790 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ 4791 sqlite3_changeset_iter *p /* Handle describing change and conflict */ 4792 ), 4793 void *pCtx, /* First argument passed to xConflict */ 4794 void **ppRebase, int *pnRebase, 4795 int flags 4796 ){ 4797 sqlite3_changeset_iter *pIter; /* Iterator to skip through changeset */ 4798 int bInv = !!(flags & SQLITE_CHANGESETAPPLY_INVERT); 4799 int rc = sessionChangesetStart(&pIter, 0, 0, nChangeset, pChangeset, bInv, 1); 4800 if( rc==SQLITE_OK ){ 4801 rc = sessionChangesetApply( 4802 db, pIter, xFilter, xConflict, pCtx, ppRebase, pnRebase, flags 4803 ); 4804 } 4805 return rc; 4806 } 4807 4808 /* 4809 ** Apply the changeset passed via pChangeset/nChangeset to the main database 4810 ** attached to handle "db". Invoke the supplied conflict handler callback 4811 ** to resolve any conflicts encountered while applying the change. 4812 */ 4813 int sqlite3changeset_apply( 4814 sqlite3 *db, /* Apply change to "main" db of this handle */ 4815 int nChangeset, /* Size of changeset in bytes */ 4816 void *pChangeset, /* Changeset blob */ 4817 int(*xFilter)( 4818 void *pCtx, /* Copy of sixth arg to _apply() */ 4819 const char *zTab /* Table name */ 4820 ), 4821 int(*xConflict)( 4822 void *pCtx, /* Copy of fifth arg to _apply() */ 4823 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ 4824 sqlite3_changeset_iter *p /* Handle describing change and conflict */ 4825 ), 4826 void *pCtx /* First argument passed to xConflict */ 4827 ){ 4828 return sqlite3changeset_apply_v2( 4829 db, nChangeset, pChangeset, xFilter, xConflict, pCtx, 0, 0, 0 4830 ); 4831 } 4832 4833 /* 4834 ** Apply the changeset passed via xInput/pIn to the main database 4835 ** attached to handle "db". Invoke the supplied conflict handler callback 4836 ** to resolve any conflicts encountered while applying the change. 4837 */ 4838 int sqlite3changeset_apply_v2_strm( 4839 sqlite3 *db, /* Apply change to "main" db of this handle */ 4840 int (*xInput)(void *pIn, void *pData, int *pnData), /* Input function */ 4841 void *pIn, /* First arg for xInput */ 4842 int(*xFilter)( 4843 void *pCtx, /* Copy of sixth arg to _apply() */ 4844 const char *zTab /* Table name */ 4845 ), 4846 int(*xConflict)( 4847 void *pCtx, /* Copy of sixth arg to _apply() */ 4848 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ 4849 sqlite3_changeset_iter *p /* Handle describing change and conflict */ 4850 ), 4851 void *pCtx, /* First argument passed to xConflict */ 4852 void **ppRebase, int *pnRebase, 4853 int flags 4854 ){ 4855 sqlite3_changeset_iter *pIter; /* Iterator to skip through changeset */ 4856 int bInverse = !!(flags & SQLITE_CHANGESETAPPLY_INVERT); 4857 int rc = sessionChangesetStart(&pIter, xInput, pIn, 0, 0, bInverse, 1); 4858 if( rc==SQLITE_OK ){ 4859 rc = sessionChangesetApply( 4860 db, pIter, xFilter, xConflict, pCtx, ppRebase, pnRebase, flags 4861 ); 4862 } 4863 return rc; 4864 } 4865 int sqlite3changeset_apply_strm( 4866 sqlite3 *db, /* Apply change to "main" db of this handle */ 4867 int (*xInput)(void *pIn, void *pData, int *pnData), /* Input function */ 4868 void *pIn, /* First arg for xInput */ 4869 int(*xFilter)( 4870 void *pCtx, /* Copy of sixth arg to _apply() */ 4871 const char *zTab /* Table name */ 4872 ), 4873 int(*xConflict)( 4874 void *pCtx, /* Copy of sixth arg to _apply() */ 4875 int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ 4876 sqlite3_changeset_iter *p /* Handle describing change and conflict */ 4877 ), 4878 void *pCtx /* First argument passed to xConflict */ 4879 ){ 4880 return sqlite3changeset_apply_v2_strm( 4881 db, xInput, pIn, xFilter, xConflict, pCtx, 0, 0, 0 4882 ); 4883 } 4884 4885 /* 4886 ** sqlite3_changegroup handle. 4887 */ 4888 struct sqlite3_changegroup { 4889 int rc; /* Error code */ 4890 int bPatch; /* True to accumulate patchsets */ 4891 SessionTable *pList; /* List of tables in current patch */ 4892 }; 4893 4894 /* 4895 ** This function is called to merge two changes to the same row together as 4896 ** part of an sqlite3changeset_concat() operation. A new change object is 4897 ** allocated and a pointer to it stored in *ppNew. 4898 */ 4899 static int sessionChangeMerge( 4900 SessionTable *pTab, /* Table structure */ 4901 int bRebase, /* True for a rebase hash-table */ 4902 int bPatchset, /* True for patchsets */ 4903 SessionChange *pExist, /* Existing change */ 4904 int op2, /* Second change operation */ 4905 int bIndirect, /* True if second change is indirect */ 4906 u8 *aRec, /* Second change record */ 4907 int nRec, /* Number of bytes in aRec */ 4908 SessionChange **ppNew /* OUT: Merged change */ 4909 ){ 4910 SessionChange *pNew = 0; 4911 int rc = SQLITE_OK; 4912 4913 if( !pExist ){ 4914 pNew = (SessionChange *)sqlite3_malloc64(sizeof(SessionChange) + nRec); 4915 if( !pNew ){ 4916 return SQLITE_NOMEM; 4917 } 4918 memset(pNew, 0, sizeof(SessionChange)); 4919 pNew->op = op2; 4920 pNew->bIndirect = bIndirect; 4921 pNew->aRecord = (u8*)&pNew[1]; 4922 if( bIndirect==0 || bRebase==0 ){ 4923 pNew->nRecord = nRec; 4924 memcpy(pNew->aRecord, aRec, nRec); 4925 }else{ 4926 int i; 4927 u8 *pIn = aRec; 4928 u8 *pOut = pNew->aRecord; 4929 for(i=0; i<pTab->nCol; i++){ 4930 int nIn = sessionSerialLen(pIn); 4931 if( *pIn==0 ){ 4932 *pOut++ = 0; 4933 }else if( pTab->abPK[i]==0 ){ 4934 *pOut++ = 0xFF; 4935 }else{ 4936 memcpy(pOut, pIn, nIn); 4937 pOut += nIn; 4938 } 4939 pIn += nIn; 4940 } 4941 pNew->nRecord = pOut - pNew->aRecord; 4942 } 4943 }else if( bRebase ){ 4944 if( pExist->op==SQLITE_DELETE && pExist->bIndirect ){ 4945 *ppNew = pExist; 4946 }else{ 4947 sqlite3_int64 nByte = nRec + pExist->nRecord + sizeof(SessionChange); 4948 pNew = (SessionChange*)sqlite3_malloc64(nByte); 4949 if( pNew==0 ){ 4950 rc = SQLITE_NOMEM; 4951 }else{ 4952 int i; 4953 u8 *a1 = pExist->aRecord; 4954 u8 *a2 = aRec; 4955 u8 *pOut; 4956 4957 memset(pNew, 0, nByte); 4958 pNew->bIndirect = bIndirect || pExist->bIndirect; 4959 pNew->op = op2; 4960 pOut = pNew->aRecord = (u8*)&pNew[1]; 4961 4962 for(i=0; i<pTab->nCol; i++){ 4963 int n1 = sessionSerialLen(a1); 4964 int n2 = sessionSerialLen(a2); 4965 if( *a1==0xFF || (pTab->abPK[i]==0 && bIndirect) ){ 4966 *pOut++ = 0xFF; 4967 }else if( *a2==0 ){ 4968 memcpy(pOut, a1, n1); 4969 pOut += n1; 4970 }else{ 4971 memcpy(pOut, a2, n2); 4972 pOut += n2; 4973 } 4974 a1 += n1; 4975 a2 += n2; 4976 } 4977 pNew->nRecord = pOut - pNew->aRecord; 4978 } 4979 sqlite3_free(pExist); 4980 } 4981 }else{ 4982 int op1 = pExist->op; 4983 4984 /* 4985 ** op1=INSERT, op2=INSERT -> Unsupported. Discard op2. 4986 ** op1=INSERT, op2=UPDATE -> INSERT. 4987 ** op1=INSERT, op2=DELETE -> (none) 4988 ** 4989 ** op1=UPDATE, op2=INSERT -> Unsupported. Discard op2. 4990 ** op1=UPDATE, op2=UPDATE -> UPDATE. 4991 ** op1=UPDATE, op2=DELETE -> DELETE. 4992 ** 4993 ** op1=DELETE, op2=INSERT -> UPDATE. 4994 ** op1=DELETE, op2=UPDATE -> Unsupported. Discard op2. 4995 ** op1=DELETE, op2=DELETE -> Unsupported. Discard op2. 4996 */ 4997 if( (op1==SQLITE_INSERT && op2==SQLITE_INSERT) 4998 || (op1==SQLITE_UPDATE && op2==SQLITE_INSERT) 4999 || (op1==SQLITE_DELETE && op2==SQLITE_UPDATE) 5000 || (op1==SQLITE_DELETE && op2==SQLITE_DELETE) 5001 ){ 5002 pNew = pExist; 5003 }else if( op1==SQLITE_INSERT && op2==SQLITE_DELETE ){ 5004 sqlite3_free(pExist); 5005 assert( pNew==0 ); 5006 }else{ 5007 u8 *aExist = pExist->aRecord; 5008 sqlite3_int64 nByte; 5009 u8 *aCsr; 5010 5011 /* Allocate a new SessionChange object. Ensure that the aRecord[] 5012 ** buffer of the new object is large enough to hold any record that 5013 ** may be generated by combining the input records. */ 5014 nByte = sizeof(SessionChange) + pExist->nRecord + nRec; 5015 pNew = (SessionChange *)sqlite3_malloc64(nByte); 5016 if( !pNew ){ 5017 sqlite3_free(pExist); 5018 return SQLITE_NOMEM; 5019 } 5020 memset(pNew, 0, sizeof(SessionChange)); 5021 pNew->bIndirect = (bIndirect && pExist->bIndirect); 5022 aCsr = pNew->aRecord = (u8 *)&pNew[1]; 5023 5024 if( op1==SQLITE_INSERT ){ /* INSERT + UPDATE */ 5025 u8 *a1 = aRec; 5026 assert( op2==SQLITE_UPDATE ); 5027 pNew->op = SQLITE_INSERT; 5028 if( bPatchset==0 ) sessionSkipRecord(&a1, pTab->nCol); 5029 sessionMergeRecord(&aCsr, pTab->nCol, aExist, a1); 5030 }else if( op1==SQLITE_DELETE ){ /* DELETE + INSERT */ 5031 assert( op2==SQLITE_INSERT ); 5032 pNew->op = SQLITE_UPDATE; 5033 if( bPatchset ){ 5034 memcpy(aCsr, aRec, nRec); 5035 aCsr += nRec; 5036 }else{ 5037 if( 0==sessionMergeUpdate(&aCsr, pTab, bPatchset, aExist, 0,aRec,0) ){ 5038 sqlite3_free(pNew); 5039 pNew = 0; 5040 } 5041 } 5042 }else if( op2==SQLITE_UPDATE ){ /* UPDATE + UPDATE */ 5043 u8 *a1 = aExist; 5044 u8 *a2 = aRec; 5045 assert( op1==SQLITE_UPDATE ); 5046 if( bPatchset==0 ){ 5047 sessionSkipRecord(&a1, pTab->nCol); 5048 sessionSkipRecord(&a2, pTab->nCol); 5049 } 5050 pNew->op = SQLITE_UPDATE; 5051 if( 0==sessionMergeUpdate(&aCsr, pTab, bPatchset, aRec, aExist,a1,a2) ){ 5052 sqlite3_free(pNew); 5053 pNew = 0; 5054 } 5055 }else{ /* UPDATE + DELETE */ 5056 assert( op1==SQLITE_UPDATE && op2==SQLITE_DELETE ); 5057 pNew->op = SQLITE_DELETE; 5058 if( bPatchset ){ 5059 memcpy(aCsr, aRec, nRec); 5060 aCsr += nRec; 5061 }else{ 5062 sessionMergeRecord(&aCsr, pTab->nCol, aRec, aExist); 5063 } 5064 } 5065 5066 if( pNew ){ 5067 pNew->nRecord = (int)(aCsr - pNew->aRecord); 5068 } 5069 sqlite3_free(pExist); 5070 } 5071 } 5072 5073 *ppNew = pNew; 5074 return rc; 5075 } 5076 5077 /* 5078 ** Add all changes in the changeset traversed by the iterator passed as 5079 ** the first argument to the changegroup hash tables. 5080 */ 5081 static int sessionChangesetToHash( 5082 sqlite3_changeset_iter *pIter, /* Iterator to read from */ 5083 sqlite3_changegroup *pGrp, /* Changegroup object to add changeset to */ 5084 int bRebase /* True if hash table is for rebasing */ 5085 ){ 5086 u8 *aRec; 5087 int nRec; 5088 int rc = SQLITE_OK; 5089 SessionTable *pTab = 0; 5090 5091 while( SQLITE_ROW==sessionChangesetNext(pIter, &aRec, &nRec, 0) ){ 5092 const char *zNew; 5093 int nCol; 5094 int op; 5095 int iHash; 5096 int bIndirect; 5097 SessionChange *pChange; 5098 SessionChange *pExist = 0; 5099 SessionChange **pp; 5100 5101 if( pGrp->pList==0 ){ 5102 pGrp->bPatch = pIter->bPatchset; 5103 }else if( pIter->bPatchset!=pGrp->bPatch ){ 5104 rc = SQLITE_ERROR; 5105 break; 5106 } 5107 5108 sqlite3changeset_op(pIter, &zNew, &nCol, &op, &bIndirect); 5109 if( !pTab || sqlite3_stricmp(zNew, pTab->zName) ){ 5110 /* Search the list for a matching table */ 5111 int nNew = (int)strlen(zNew); 5112 u8 *abPK; 5113 5114 sqlite3changeset_pk(pIter, &abPK, 0); 5115 for(pTab = pGrp->pList; pTab; pTab=pTab->pNext){ 5116 if( 0==sqlite3_strnicmp(pTab->zName, zNew, nNew+1) ) break; 5117 } 5118 if( !pTab ){ 5119 SessionTable **ppTab; 5120 5121 pTab = sqlite3_malloc64(sizeof(SessionTable) + nCol + nNew+1); 5122 if( !pTab ){ 5123 rc = SQLITE_NOMEM; 5124 break; 5125 } 5126 memset(pTab, 0, sizeof(SessionTable)); 5127 pTab->nCol = nCol; 5128 pTab->abPK = (u8*)&pTab[1]; 5129 memcpy(pTab->abPK, abPK, nCol); 5130 pTab->zName = (char*)&pTab->abPK[nCol]; 5131 memcpy(pTab->zName, zNew, nNew+1); 5132 5133 /* The new object must be linked on to the end of the list, not 5134 ** simply added to the start of it. This is to ensure that the 5135 ** tables within the output of sqlite3changegroup_output() are in 5136 ** the right order. */ 5137 for(ppTab=&pGrp->pList; *ppTab; ppTab=&(*ppTab)->pNext); 5138 *ppTab = pTab; 5139 }else if( pTab->nCol!=nCol || memcmp(pTab->abPK, abPK, nCol) ){ 5140 rc = SQLITE_SCHEMA; 5141 break; 5142 } 5143 } 5144 5145 if( sessionGrowHash(0, pIter->bPatchset, pTab) ){ 5146 rc = SQLITE_NOMEM; 5147 break; 5148 } 5149 iHash = sessionChangeHash( 5150 pTab, (pIter->bPatchset && op==SQLITE_DELETE), aRec, pTab->nChange 5151 ); 5152 5153 /* Search for existing entry. If found, remove it from the hash table. 5154 ** Code below may link it back in. 5155 */ 5156 for(pp=&pTab->apChange[iHash]; *pp; pp=&(*pp)->pNext){ 5157 int bPkOnly1 = 0; 5158 int bPkOnly2 = 0; 5159 if( pIter->bPatchset ){ 5160 bPkOnly1 = (*pp)->op==SQLITE_DELETE; 5161 bPkOnly2 = op==SQLITE_DELETE; 5162 } 5163 if( sessionChangeEqual(pTab, bPkOnly1, (*pp)->aRecord, bPkOnly2, aRec) ){ 5164 pExist = *pp; 5165 *pp = (*pp)->pNext; 5166 pTab->nEntry--; 5167 break; 5168 } 5169 } 5170 5171 rc = sessionChangeMerge(pTab, bRebase, 5172 pIter->bPatchset, pExist, op, bIndirect, aRec, nRec, &pChange 5173 ); 5174 if( rc ) break; 5175 if( pChange ){ 5176 pChange->pNext = pTab->apChange[iHash]; 5177 pTab->apChange[iHash] = pChange; 5178 pTab->nEntry++; 5179 } 5180 } 5181 5182 if( rc==SQLITE_OK ) rc = pIter->rc; 5183 return rc; 5184 } 5185 5186 /* 5187 ** Serialize a changeset (or patchset) based on all changesets (or patchsets) 5188 ** added to the changegroup object passed as the first argument. 5189 ** 5190 ** If xOutput is not NULL, then the changeset/patchset is returned to the 5191 ** user via one or more calls to xOutput, as with the other streaming 5192 ** interfaces. 5193 ** 5194 ** Or, if xOutput is NULL, then (*ppOut) is populated with a pointer to a 5195 ** buffer containing the output changeset before this function returns. In 5196 ** this case (*pnOut) is set to the size of the output buffer in bytes. It 5197 ** is the responsibility of the caller to free the output buffer using 5198 ** sqlite3_free() when it is no longer required. 5199 ** 5200 ** If successful, SQLITE_OK is returned. Or, if an error occurs, an SQLite 5201 ** error code. If an error occurs and xOutput is NULL, (*ppOut) and (*pnOut) 5202 ** are both set to 0 before returning. 5203 */ 5204 static int sessionChangegroupOutput( 5205 sqlite3_changegroup *pGrp, 5206 int (*xOutput)(void *pOut, const void *pData, int nData), 5207 void *pOut, 5208 int *pnOut, 5209 void **ppOut 5210 ){ 5211 int rc = SQLITE_OK; 5212 SessionBuffer buf = {0, 0, 0}; 5213 SessionTable *pTab; 5214 assert( xOutput==0 || (ppOut==0 && pnOut==0) ); 5215 5216 /* Create the serialized output changeset based on the contents of the 5217 ** hash tables attached to the SessionTable objects in list p->pList. 5218 */ 5219 for(pTab=pGrp->pList; rc==SQLITE_OK && pTab; pTab=pTab->pNext){ 5220 int i; 5221 if( pTab->nEntry==0 ) continue; 5222 5223 sessionAppendTableHdr(&buf, pGrp->bPatch, pTab, &rc); 5224 for(i=0; i<pTab->nChange; i++){ 5225 SessionChange *p; 5226 for(p=pTab->apChange[i]; p; p=p->pNext){ 5227 sessionAppendByte(&buf, p->op, &rc); 5228 sessionAppendByte(&buf, p->bIndirect, &rc); 5229 sessionAppendBlob(&buf, p->aRecord, p->nRecord, &rc); 5230 if( rc==SQLITE_OK && xOutput && buf.nBuf>=sessions_strm_chunk_size ){ 5231 rc = xOutput(pOut, buf.aBuf, buf.nBuf); 5232 buf.nBuf = 0; 5233 } 5234 } 5235 } 5236 } 5237 5238 if( rc==SQLITE_OK ){ 5239 if( xOutput ){ 5240 if( buf.nBuf>0 ) rc = xOutput(pOut, buf.aBuf, buf.nBuf); 5241 }else{ 5242 *ppOut = buf.aBuf; 5243 *pnOut = buf.nBuf; 5244 buf.aBuf = 0; 5245 } 5246 } 5247 sqlite3_free(buf.aBuf); 5248 5249 return rc; 5250 } 5251 5252 /* 5253 ** Allocate a new, empty, sqlite3_changegroup. 5254 */ 5255 int sqlite3changegroup_new(sqlite3_changegroup **pp){ 5256 int rc = SQLITE_OK; /* Return code */ 5257 sqlite3_changegroup *p; /* New object */ 5258 p = (sqlite3_changegroup*)sqlite3_malloc(sizeof(sqlite3_changegroup)); 5259 if( p==0 ){ 5260 rc = SQLITE_NOMEM; 5261 }else{ 5262 memset(p, 0, sizeof(sqlite3_changegroup)); 5263 } 5264 *pp = p; 5265 return rc; 5266 } 5267 5268 /* 5269 ** Add the changeset currently stored in buffer pData, size nData bytes, 5270 ** to changeset-group p. 5271 */ 5272 int sqlite3changegroup_add(sqlite3_changegroup *pGrp, int nData, void *pData){ 5273 sqlite3_changeset_iter *pIter; /* Iterator opened on pData/nData */ 5274 int rc; /* Return code */ 5275 5276 rc = sqlite3changeset_start(&pIter, nData, pData); 5277 if( rc==SQLITE_OK ){ 5278 rc = sessionChangesetToHash(pIter, pGrp, 0); 5279 } 5280 sqlite3changeset_finalize(pIter); 5281 return rc; 5282 } 5283 5284 /* 5285 ** Obtain a buffer containing a changeset representing the concatenation 5286 ** of all changesets added to the group so far. 5287 */ 5288 int sqlite3changegroup_output( 5289 sqlite3_changegroup *pGrp, 5290 int *pnData, 5291 void **ppData 5292 ){ 5293 return sessionChangegroupOutput(pGrp, 0, 0, pnData, ppData); 5294 } 5295 5296 /* 5297 ** Streaming versions of changegroup_add(). 5298 */ 5299 int sqlite3changegroup_add_strm( 5300 sqlite3_changegroup *pGrp, 5301 int (*xInput)(void *pIn, void *pData, int *pnData), 5302 void *pIn 5303 ){ 5304 sqlite3_changeset_iter *pIter; /* Iterator opened on pData/nData */ 5305 int rc; /* Return code */ 5306 5307 rc = sqlite3changeset_start_strm(&pIter, xInput, pIn); 5308 if( rc==SQLITE_OK ){ 5309 rc = sessionChangesetToHash(pIter, pGrp, 0); 5310 } 5311 sqlite3changeset_finalize(pIter); 5312 return rc; 5313 } 5314 5315 /* 5316 ** Streaming versions of changegroup_output(). 5317 */ 5318 int sqlite3changegroup_output_strm( 5319 sqlite3_changegroup *pGrp, 5320 int (*xOutput)(void *pOut, const void *pData, int nData), 5321 void *pOut 5322 ){ 5323 return sessionChangegroupOutput(pGrp, xOutput, pOut, 0, 0); 5324 } 5325 5326 /* 5327 ** Delete a changegroup object. 5328 */ 5329 void sqlite3changegroup_delete(sqlite3_changegroup *pGrp){ 5330 if( pGrp ){ 5331 sessionDeleteTable(0, pGrp->pList); 5332 sqlite3_free(pGrp); 5333 } 5334 } 5335 5336 /* 5337 ** Combine two changesets together. 5338 */ 5339 int sqlite3changeset_concat( 5340 int nLeft, /* Number of bytes in lhs input */ 5341 void *pLeft, /* Lhs input changeset */ 5342 int nRight /* Number of bytes in rhs input */, 5343 void *pRight, /* Rhs input changeset */ 5344 int *pnOut, /* OUT: Number of bytes in output changeset */ 5345 void **ppOut /* OUT: changeset (left <concat> right) */ 5346 ){ 5347 sqlite3_changegroup *pGrp; 5348 int rc; 5349 5350 rc = sqlite3changegroup_new(&pGrp); 5351 if( rc==SQLITE_OK ){ 5352 rc = sqlite3changegroup_add(pGrp, nLeft, pLeft); 5353 } 5354 if( rc==SQLITE_OK ){ 5355 rc = sqlite3changegroup_add(pGrp, nRight, pRight); 5356 } 5357 if( rc==SQLITE_OK ){ 5358 rc = sqlite3changegroup_output(pGrp, pnOut, ppOut); 5359 } 5360 sqlite3changegroup_delete(pGrp); 5361 5362 return rc; 5363 } 5364 5365 /* 5366 ** Streaming version of sqlite3changeset_concat(). 5367 */ 5368 int sqlite3changeset_concat_strm( 5369 int (*xInputA)(void *pIn, void *pData, int *pnData), 5370 void *pInA, 5371 int (*xInputB)(void *pIn, void *pData, int *pnData), 5372 void *pInB, 5373 int (*xOutput)(void *pOut, const void *pData, int nData), 5374 void *pOut 5375 ){ 5376 sqlite3_changegroup *pGrp; 5377 int rc; 5378 5379 rc = sqlite3changegroup_new(&pGrp); 5380 if( rc==SQLITE_OK ){ 5381 rc = sqlite3changegroup_add_strm(pGrp, xInputA, pInA); 5382 } 5383 if( rc==SQLITE_OK ){ 5384 rc = sqlite3changegroup_add_strm(pGrp, xInputB, pInB); 5385 } 5386 if( rc==SQLITE_OK ){ 5387 rc = sqlite3changegroup_output_strm(pGrp, xOutput, pOut); 5388 } 5389 sqlite3changegroup_delete(pGrp); 5390 5391 return rc; 5392 } 5393 5394 /* 5395 ** Changeset rebaser handle. 5396 */ 5397 struct sqlite3_rebaser { 5398 sqlite3_changegroup grp; /* Hash table */ 5399 }; 5400 5401 /* 5402 ** Buffers a1 and a2 must both contain a sessions module record nCol 5403 ** fields in size. This function appends an nCol sessions module 5404 ** record to buffer pBuf that is a copy of a1, except that for 5405 ** each field that is undefined in a1[], swap in the field from a2[]. 5406 */ 5407 static void sessionAppendRecordMerge( 5408 SessionBuffer *pBuf, /* Buffer to append to */ 5409 int nCol, /* Number of columns in each record */ 5410 u8 *a1, int n1, /* Record 1 */ 5411 u8 *a2, int n2, /* Record 2 */ 5412 int *pRc /* IN/OUT: error code */ 5413 ){ 5414 sessionBufferGrow(pBuf, n1+n2, pRc); 5415 if( *pRc==SQLITE_OK ){ 5416 int i; 5417 u8 *pOut = &pBuf->aBuf[pBuf->nBuf]; 5418 for(i=0; i<nCol; i++){ 5419 int nn1 = sessionSerialLen(a1); 5420 int nn2 = sessionSerialLen(a2); 5421 if( *a1==0 || *a1==0xFF ){ 5422 memcpy(pOut, a2, nn2); 5423 pOut += nn2; 5424 }else{ 5425 memcpy(pOut, a1, nn1); 5426 pOut += nn1; 5427 } 5428 a1 += nn1; 5429 a2 += nn2; 5430 } 5431 5432 pBuf->nBuf = pOut-pBuf->aBuf; 5433 assert( pBuf->nBuf<=pBuf->nAlloc ); 5434 } 5435 } 5436 5437 /* 5438 ** This function is called when rebasing a local UPDATE change against one 5439 ** or more remote UPDATE changes. The aRec/nRec buffer contains the current 5440 ** old.* and new.* records for the change. The rebase buffer (a single 5441 ** record) is in aChange/nChange. The rebased change is appended to buffer 5442 ** pBuf. 5443 ** 5444 ** Rebasing the UPDATE involves: 5445 ** 5446 ** * Removing any changes to fields for which the corresponding field 5447 ** in the rebase buffer is set to "replaced" (type 0xFF). If this 5448 ** means the UPDATE change updates no fields, nothing is appended 5449 ** to the output buffer. 5450 ** 5451 ** * For each field modified by the local change for which the 5452 ** corresponding field in the rebase buffer is not "undefined" (0x00) 5453 ** or "replaced" (0xFF), the old.* value is replaced by the value 5454 ** in the rebase buffer. 5455 */ 5456 static void sessionAppendPartialUpdate( 5457 SessionBuffer *pBuf, /* Append record here */ 5458 sqlite3_changeset_iter *pIter, /* Iterator pointed at local change */ 5459 u8 *aRec, int nRec, /* Local change */ 5460 u8 *aChange, int nChange, /* Record to rebase against */ 5461 int *pRc /* IN/OUT: Return Code */ 5462 ){ 5463 sessionBufferGrow(pBuf, 2+nRec+nChange, pRc); 5464 if( *pRc==SQLITE_OK ){ 5465 int bData = 0; 5466 u8 *pOut = &pBuf->aBuf[pBuf->nBuf]; 5467 int i; 5468 u8 *a1 = aRec; 5469 u8 *a2 = aChange; 5470 5471 *pOut++ = SQLITE_UPDATE; 5472 *pOut++ = pIter->bIndirect; 5473 for(i=0; i<pIter->nCol; i++){ 5474 int n1 = sessionSerialLen(a1); 5475 int n2 = sessionSerialLen(a2); 5476 if( pIter->abPK[i] || a2[0]==0 ){ 5477 if( !pIter->abPK[i] && a1[0] ) bData = 1; 5478 memcpy(pOut, a1, n1); 5479 pOut += n1; 5480 }else if( a2[0]!=0xFF ){ 5481 bData = 1; 5482 memcpy(pOut, a2, n2); 5483 pOut += n2; 5484 }else{ 5485 *pOut++ = '\0'; 5486 } 5487 a1 += n1; 5488 a2 += n2; 5489 } 5490 if( bData ){ 5491 a2 = aChange; 5492 for(i=0; i<pIter->nCol; i++){ 5493 int n1 = sessionSerialLen(a1); 5494 int n2 = sessionSerialLen(a2); 5495 if( pIter->abPK[i] || a2[0]!=0xFF ){ 5496 memcpy(pOut, a1, n1); 5497 pOut += n1; 5498 }else{ 5499 *pOut++ = '\0'; 5500 } 5501 a1 += n1; 5502 a2 += n2; 5503 } 5504 pBuf->nBuf = (pOut - pBuf->aBuf); 5505 } 5506 } 5507 } 5508 5509 /* 5510 ** pIter is configured to iterate through a changeset. This function rebases 5511 ** that changeset according to the current configuration of the rebaser 5512 ** object passed as the first argument. If no error occurs and argument xOutput 5513 ** is not NULL, then the changeset is returned to the caller by invoking 5514 ** xOutput zero or more times and SQLITE_OK returned. Or, if xOutput is NULL, 5515 ** then (*ppOut) is set to point to a buffer containing the rebased changeset 5516 ** before this function returns. In this case (*pnOut) is set to the size of 5517 ** the buffer in bytes. It is the responsibility of the caller to eventually 5518 ** free the (*ppOut) buffer using sqlite3_free(). 5519 ** 5520 ** If an error occurs, an SQLite error code is returned. If ppOut and 5521 ** pnOut are not NULL, then the two output parameters are set to 0 before 5522 ** returning. 5523 */ 5524 static int sessionRebase( 5525 sqlite3_rebaser *p, /* Rebaser hash table */ 5526 sqlite3_changeset_iter *pIter, /* Input data */ 5527 int (*xOutput)(void *pOut, const void *pData, int nData), 5528 void *pOut, /* Context for xOutput callback */ 5529 int *pnOut, /* OUT: Number of bytes in output changeset */ 5530 void **ppOut /* OUT: Inverse of pChangeset */ 5531 ){ 5532 int rc = SQLITE_OK; 5533 u8 *aRec = 0; 5534 int nRec = 0; 5535 int bNew = 0; 5536 SessionTable *pTab = 0; 5537 SessionBuffer sOut = {0,0,0}; 5538 5539 while( SQLITE_ROW==sessionChangesetNext(pIter, &aRec, &nRec, &bNew) ){ 5540 SessionChange *pChange = 0; 5541 int bDone = 0; 5542 5543 if( bNew ){ 5544 const char *zTab = pIter->zTab; 5545 for(pTab=p->grp.pList; pTab; pTab=pTab->pNext){ 5546 if( 0==sqlite3_stricmp(pTab->zName, zTab) ) break; 5547 } 5548 bNew = 0; 5549 5550 /* A patchset may not be rebased */ 5551 if( pIter->bPatchset ){ 5552 rc = SQLITE_ERROR; 5553 } 5554 5555 /* Append a table header to the output for this new table */ 5556 sessionAppendByte(&sOut, pIter->bPatchset ? 'P' : 'T', &rc); 5557 sessionAppendVarint(&sOut, pIter->nCol, &rc); 5558 sessionAppendBlob(&sOut, pIter->abPK, pIter->nCol, &rc); 5559 sessionAppendBlob(&sOut,(u8*)pIter->zTab,(int)strlen(pIter->zTab)+1,&rc); 5560 } 5561 5562 if( pTab && rc==SQLITE_OK ){ 5563 int iHash = sessionChangeHash(pTab, 0, aRec, pTab->nChange); 5564 5565 for(pChange=pTab->apChange[iHash]; pChange; pChange=pChange->pNext){ 5566 if( sessionChangeEqual(pTab, 0, aRec, 0, pChange->aRecord) ){ 5567 break; 5568 } 5569 } 5570 } 5571 5572 if( pChange ){ 5573 assert( pChange->op==SQLITE_DELETE || pChange->op==SQLITE_INSERT ); 5574 switch( pIter->op ){ 5575 case SQLITE_INSERT: 5576 if( pChange->op==SQLITE_INSERT ){ 5577 bDone = 1; 5578 if( pChange->bIndirect==0 ){ 5579 sessionAppendByte(&sOut, SQLITE_UPDATE, &rc); 5580 sessionAppendByte(&sOut, pIter->bIndirect, &rc); 5581 sessionAppendBlob(&sOut, pChange->aRecord, pChange->nRecord, &rc); 5582 sessionAppendBlob(&sOut, aRec, nRec, &rc); 5583 } 5584 } 5585 break; 5586 5587 case SQLITE_UPDATE: 5588 bDone = 1; 5589 if( pChange->op==SQLITE_DELETE ){ 5590 if( pChange->bIndirect==0 ){ 5591 u8 *pCsr = aRec; 5592 sessionSkipRecord(&pCsr, pIter->nCol); 5593 sessionAppendByte(&sOut, SQLITE_INSERT, &rc); 5594 sessionAppendByte(&sOut, pIter->bIndirect, &rc); 5595 sessionAppendRecordMerge(&sOut, pIter->nCol, 5596 pCsr, nRec-(pCsr-aRec), 5597 pChange->aRecord, pChange->nRecord, &rc 5598 ); 5599 } 5600 }else{ 5601 sessionAppendPartialUpdate(&sOut, pIter, 5602 aRec, nRec, pChange->aRecord, pChange->nRecord, &rc 5603 ); 5604 } 5605 break; 5606 5607 default: 5608 assert( pIter->op==SQLITE_DELETE ); 5609 bDone = 1; 5610 if( pChange->op==SQLITE_INSERT ){ 5611 sessionAppendByte(&sOut, SQLITE_DELETE, &rc); 5612 sessionAppendByte(&sOut, pIter->bIndirect, &rc); 5613 sessionAppendRecordMerge(&sOut, pIter->nCol, 5614 pChange->aRecord, pChange->nRecord, aRec, nRec, &rc 5615 ); 5616 } 5617 break; 5618 } 5619 } 5620 5621 if( bDone==0 ){ 5622 sessionAppendByte(&sOut, pIter->op, &rc); 5623 sessionAppendByte(&sOut, pIter->bIndirect, &rc); 5624 sessionAppendBlob(&sOut, aRec, nRec, &rc); 5625 } 5626 if( rc==SQLITE_OK && xOutput && sOut.nBuf>sessions_strm_chunk_size ){ 5627 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf); 5628 sOut.nBuf = 0; 5629 } 5630 if( rc ) break; 5631 } 5632 5633 if( rc!=SQLITE_OK ){ 5634 sqlite3_free(sOut.aBuf); 5635 memset(&sOut, 0, sizeof(sOut)); 5636 } 5637 5638 if( rc==SQLITE_OK ){ 5639 if( xOutput ){ 5640 if( sOut.nBuf>0 ){ 5641 rc = xOutput(pOut, sOut.aBuf, sOut.nBuf); 5642 } 5643 }else{ 5644 *ppOut = (void*)sOut.aBuf; 5645 *pnOut = sOut.nBuf; 5646 sOut.aBuf = 0; 5647 } 5648 } 5649 sqlite3_free(sOut.aBuf); 5650 return rc; 5651 } 5652 5653 /* 5654 ** Create a new rebaser object. 5655 */ 5656 int sqlite3rebaser_create(sqlite3_rebaser **ppNew){ 5657 int rc = SQLITE_OK; 5658 sqlite3_rebaser *pNew; 5659 5660 pNew = sqlite3_malloc(sizeof(sqlite3_rebaser)); 5661 if( pNew==0 ){ 5662 rc = SQLITE_NOMEM; 5663 }else{ 5664 memset(pNew, 0, sizeof(sqlite3_rebaser)); 5665 } 5666 *ppNew = pNew; 5667 return rc; 5668 } 5669 5670 /* 5671 ** Call this one or more times to configure a rebaser. 5672 */ 5673 int sqlite3rebaser_configure( 5674 sqlite3_rebaser *p, 5675 int nRebase, const void *pRebase 5676 ){ 5677 sqlite3_changeset_iter *pIter = 0; /* Iterator opened on pData/nData */ 5678 int rc; /* Return code */ 5679 rc = sqlite3changeset_start(&pIter, nRebase, (void*)pRebase); 5680 if( rc==SQLITE_OK ){ 5681 rc = sessionChangesetToHash(pIter, &p->grp, 1); 5682 } 5683 sqlite3changeset_finalize(pIter); 5684 return rc; 5685 } 5686 5687 /* 5688 ** Rebase a changeset according to current rebaser configuration 5689 */ 5690 int sqlite3rebaser_rebase( 5691 sqlite3_rebaser *p, 5692 int nIn, const void *pIn, 5693 int *pnOut, void **ppOut 5694 ){ 5695 sqlite3_changeset_iter *pIter = 0; /* Iterator to skip through input */ 5696 int rc = sqlite3changeset_start(&pIter, nIn, (void*)pIn); 5697 5698 if( rc==SQLITE_OK ){ 5699 rc = sessionRebase(p, pIter, 0, 0, pnOut, ppOut); 5700 sqlite3changeset_finalize(pIter); 5701 } 5702 5703 return rc; 5704 } 5705 5706 /* 5707 ** Rebase a changeset according to current rebaser configuration 5708 */ 5709 int sqlite3rebaser_rebase_strm( 5710 sqlite3_rebaser *p, 5711 int (*xInput)(void *pIn, void *pData, int *pnData), 5712 void *pIn, 5713 int (*xOutput)(void *pOut, const void *pData, int nData), 5714 void *pOut 5715 ){ 5716 sqlite3_changeset_iter *pIter = 0; /* Iterator to skip through input */ 5717 int rc = sqlite3changeset_start_strm(&pIter, xInput, pIn); 5718 5719 if( rc==SQLITE_OK ){ 5720 rc = sessionRebase(p, pIter, xOutput, pOut, 0, 0); 5721 sqlite3changeset_finalize(pIter); 5722 } 5723 5724 return rc; 5725 } 5726 5727 /* 5728 ** Destroy a rebaser object 5729 */ 5730 void sqlite3rebaser_delete(sqlite3_rebaser *p){ 5731 if( p ){ 5732 sessionDeleteTable(0, p->grp.pList); 5733 sqlite3_free(p); 5734 } 5735 } 5736 5737 /* 5738 ** Global configuration 5739 */ 5740 int sqlite3session_config(int op, void *pArg){ 5741 int rc = SQLITE_OK; 5742 switch( op ){ 5743 case SQLITE_SESSION_CONFIG_STRMSIZE: { 5744 int *pInt = (int*)pArg; 5745 if( *pInt>0 ){ 5746 sessions_strm_chunk_size = *pInt; 5747 } 5748 *pInt = sessions_strm_chunk_size; 5749 break; 5750 } 5751 default: 5752 rc = SQLITE_MISUSE; 5753 break; 5754 } 5755 return rc; 5756 } 5757 5758 #endif /* SQLITE_ENABLE_SESSION && SQLITE_ENABLE_PREUPDATE_HOOK */ 5759