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