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