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