blob: 1cf1bf4669bb916f5b2bc1c17c2dd2f0de21f15c [file] [log] [blame]
#if defined(SQLITE_ENABLE_SESSION) && defined(SQLITE_ENABLE_PREUPDATE_HOOK)
#include "sqlite3session.h"
#include <assert.h>
#include <string.h>
#ifndef SQLITE_AMALGAMATION
# include "sqliteInt.h"
# include "vdbeInt.h"
#endif
typedef struct SessionTable SessionTable;
typedef struct SessionChange SessionChange;
typedef struct SessionBuffer SessionBuffer;
typedef struct SessionInput SessionInput;
/*
** Minimum chunk size used by streaming versions of functions.
*/
#ifndef SESSIONS_STRM_CHUNK_SIZE
# ifdef SQLITE_TEST
# define SESSIONS_STRM_CHUNK_SIZE 64
# else
# define SESSIONS_STRM_CHUNK_SIZE 1024
# endif
#endif
#define SESSIONS_ROWID "_rowid_"
static int sessions_strm_chunk_size = SESSIONS_STRM_CHUNK_SIZE;
typedef struct SessionHook SessionHook;
struct SessionHook {
void *pCtx;
int (*xOld)(void*,int,sqlite3_value**);
int (*xNew)(void*,int,sqlite3_value**);
int (*xCount)(void*);
int (*xDepth)(void*);
};
/*
** Session handle structure.
*/
struct sqlite3_session {
sqlite3 *db; /* Database handle session is attached to */
char *zDb; /* Name of database session is attached to */
int bEnableSize; /* True if changeset_size() enabled */
int bEnable; /* True if currently recording */
int bIndirect; /* True if all changes are indirect */
int bAutoAttach; /* True to auto-attach tables */
int bImplicitPK; /* True to handle tables with implicit PK */
int rc; /* Non-zero if an error has occurred */
void *pFilterCtx; /* First argument to pass to xTableFilter */
int (*xTableFilter)(void *pCtx, const char *zTab);
i64 nMalloc; /* Number of bytes of data allocated */
i64 nMaxChangesetSize;
sqlite3_value *pZeroBlob; /* Value containing X'' */
sqlite3_session *pNext; /* Next session object on same db. */
SessionTable *pTable; /* List of attached tables */
SessionHook hook; /* APIs to grab new and old data with */
};
/*
** Instances of this structure are used to build strings or binary records.
*/
struct SessionBuffer {
u8 *aBuf; /* Pointer to changeset buffer */
int nBuf; /* Size of buffer aBuf */
int nAlloc; /* Size of allocation containing aBuf */
};
/*
** An object of this type is used internally as an abstraction for
** input data. Input data may be supplied either as a single large buffer
** (e.g. sqlite3changeset_start()) or using a stream function (e.g.
** sqlite3changeset_start_strm()).
**
** bNoDiscard:
** If true, then the only time data is discarded is as a result of explicit
** sessionDiscardData() calls. Not within every sessionInputBuffer() call.
*/
struct SessionInput {
int bNoDiscard; /* If true, do not discard in InputBuffer() */
int iCurrent; /* Offset in aData[] of current change */
int iNext; /* Offset in aData[] of next change */
u8 *aData; /* Pointer to buffer containing changeset */
int nData; /* Number of bytes in aData */
SessionBuffer buf; /* Current read buffer */
int (*xInput)(void*, void*, int*); /* Input stream call (or NULL) */
void *pIn; /* First argument to xInput */
int bEof; /* Set to true after xInput finished */
};
/*
** Structure for changeset iterators.
*/
struct sqlite3_changeset_iter {
SessionInput in; /* Input buffer or stream */
SessionBuffer tblhdr; /* Buffer to hold apValue/zTab/abPK/ */
int bPatchset; /* True if this is a patchset */
int bInvert; /* True to invert changeset */
int bSkipEmpty; /* Skip noop UPDATE changes */
int rc; /* Iterator error code */
sqlite3_stmt *pConflict; /* Points to conflicting row, if any */
char *zTab; /* Current table */
int nCol; /* Number of columns in zTab */
int op; /* Current operation */
int bIndirect; /* True if current change was indirect */
u8 *abPK; /* Primary key array */
sqlite3_value **apValue; /* old.* and new.* values */
};
/*
** Each session object maintains a set of the following structures, one
** for each table the session object is monitoring. The structures are
** stored in a linked list starting at sqlite3_session.pTable.
**
** The keys of the SessionTable.aChange[] hash table are all rows that have
** been modified in any way since the session object was attached to the
** table.
**
** The data associated with each hash-table entry is a structure containing
** a subset of the initial values that the modified row contained at the
** start of the session. Or no initial values if the row was inserted.
**
** pDfltStmt:
** This is only used by the sqlite3changegroup_xxx() APIs, not by
** regular sqlite3_session objects. It is a SELECT statement that
** selects the default value for each table column. For example,
** if the table is
**
** CREATE TABLE xx(a DEFAULT 1, b, c DEFAULT 'abc')
**
** then this variable is the compiled version of:
**
** SELECT 1, NULL, 'abc'
*/
struct SessionTable {
SessionTable *pNext;
char *zName; /* Local name of table */
int nCol; /* Number of non-hidden columns */
int nTotalCol; /* Number of columns including hidden */
int bStat1; /* True if this is sqlite_stat1 */
int bRowid; /* True if this table uses rowid for PK */
const char **azCol; /* Column names */
const char **azDflt; /* Default value expressions */
int *aiIdx; /* Index to pass to xNew/xOld */
u8 *abPK; /* Array of primary key flags */
int nEntry; /* Total number of entries in hash table */
int nChange; /* Size of apChange[] array */
SessionChange **apChange; /* Hash table buckets */
sqlite3_stmt *pDfltStmt;
};
/*
** RECORD FORMAT:
**
** The following record format is similar to (but not compatible with) that
** used in SQLite database files. This format is used as part of the
** change-set binary format, and so must be architecture independent.
**
** Unlike the SQLite database record format, each field is self-contained -
** there is no separation of header and data. Each field begins with a
** single byte describing its type, as follows:
**
** 0x00: Undefined value.
** 0x01: Integer value.
** 0x02: Real value.
** 0x03: Text value.
** 0x04: Blob value.
** 0x05: SQL NULL value.
**
** Note that the above match the definitions of SQLITE_INTEGER, SQLITE_TEXT
** and so on in sqlite3.h. For undefined and NULL values, the field consists
** only of the single type byte. For other types of values, the type byte
** is followed by:
**
** Text values:
** A varint containing the number of bytes in the value (encoded using
** UTF-8). Followed by a buffer containing the UTF-8 representation
** of the text value. There is no nul terminator.
**
** Blob values:
** A varint containing the number of bytes in the value, followed by
** a buffer containing the value itself.
**
** Integer values:
** An 8-byte big-endian integer value.
**
** Real values:
** An 8-byte big-endian IEEE 754-2008 real value.
**
** Varint values are encoded in the same way as varints in the SQLite
** record format.
**
** CHANGESET FORMAT:
**
** A changeset is a collection of DELETE, UPDATE and INSERT operations on
** one or more tables. Operations on a single table are grouped together,
** but may occur in any order (i.e. deletes, updates and inserts are all
** mixed together).
**
** Each group of changes begins with a table header:
**
** 1 byte: Constant 0x54 (capital 'T')
** Varint: Number of columns in the table.
** nCol bytes: 0x01 for PK columns, 0x00 otherwise.
** N bytes: Unqualified table name (encoded using UTF-8). Nul-terminated.
**
** Followed by one or more changes to the table.
**
** 1 byte: Either SQLITE_INSERT (0x12), UPDATE (0x17) or DELETE (0x09).
** 1 byte: The "indirect-change" flag.
** old.* record: (delete and update only)
** new.* record: (insert and update only)
**
** The "old.*" and "new.*" records, if present, are N field records in the
** format described above under "RECORD FORMAT", where N is the number of
** columns in the table. The i'th field of each record is associated with
** the i'th column of the table, counting from left to right in the order
** in which columns were declared in the CREATE TABLE statement.
**
** The new.* record that is part of each INSERT change contains the values
** that make up the new row. Similarly, the old.* record that is part of each
** DELETE change contains the values that made up the row that was deleted
** from the database. In the changeset format, the records that are part
** of INSERT or DELETE changes never contain any undefined (type byte 0x00)
** fields.
**
** Within the old.* record associated with an UPDATE change, all fields
** associated with table columns that are not PRIMARY KEY columns and are
** not modified by the UPDATE change are set to "undefined". Other fields
** are set to the values that made up the row before the UPDATE that the
** change records took place. Within the new.* record, fields associated
** with table columns modified by the UPDATE change contain the new
** values. Fields associated with table columns that are not modified
** are set to "undefined".
**
** PATCHSET FORMAT:
**
** A patchset is also a collection of changes. It is similar to a changeset,
** but leaves undefined those fields that are not useful if no conflict
** resolution is required when applying the changeset.
**
** Each group of changes begins with a table header:
**
** 1 byte: Constant 0x50 (capital 'P')
** Varint: Number of columns in the table.
** nCol bytes: 0x01 for PK columns, 0x00 otherwise.
** N bytes: Unqualified table name (encoded using UTF-8). Nul-terminated.
**
** Followed by one or more changes to the table.
**
** 1 byte: Either SQLITE_INSERT (0x12), UPDATE (0x17) or DELETE (0x09).
** 1 byte: The "indirect-change" flag.
** single record: (PK fields for DELETE, PK and modified fields for UPDATE,
** full record for INSERT).
**
** As in the changeset format, each field of the single record that is part
** of a patchset change is associated with the correspondingly positioned
** table column, counting from left to right within the CREATE TABLE
** statement.
**
** For a DELETE change, all fields within the record except those associated
** with PRIMARY KEY columns are omitted. The PRIMARY KEY fields contain the
** values identifying the row to delete.
**
** For an UPDATE change, all fields except those associated with PRIMARY KEY
** columns and columns that are modified by the UPDATE are set to "undefined".
** PRIMARY KEY fields contain the values identifying the table row to update,
** and fields associated with modified columns contain the new column values.
**
** The records associated with INSERT changes are in the same format as for
** changesets. It is not possible for a record associated with an INSERT
** change to contain a field set to "undefined".
**
** REBASE BLOB FORMAT:
**
** A rebase blob may be output by sqlite3changeset_apply_v2() and its
** streaming equivalent for use with the sqlite3_rebaser APIs to rebase
** existing changesets. A rebase blob contains one entry for each conflict
** resolved using either the OMIT or REPLACE strategies within the apply_v2()
** call.
**
** The format used for a rebase blob is very similar to that used for
** changesets. All entries related to a single table are grouped together.
**
** Each group of entries begins with a table header in changeset format:
**
** 1 byte: Constant 0x54 (capital 'T')
** Varint: Number of columns in the table.
** nCol bytes: 0x01 for PK columns, 0x00 otherwise.
** N bytes: Unqualified table name (encoded using UTF-8). Nul-terminated.
**
** Followed by one or more entries associated with the table.
**
** 1 byte: Either SQLITE_INSERT (0x12), DELETE (0x09).
** 1 byte: Flag. 0x01 for REPLACE, 0x00 for OMIT.
** record: (in the record format defined above).
**
** In a rebase blob, the first field is set to SQLITE_INSERT if the change
** that caused the conflict was an INSERT or UPDATE, or to SQLITE_DELETE if
** it was a DELETE. The second field is set to 0x01 if the conflict
** resolution strategy was REPLACE, or 0x00 if it was OMIT.
**
** If the change that caused the conflict was a DELETE, then the single
** record is a copy of the old.* record from the original changeset. If it
** was an INSERT, then the single record is a copy of the new.* record. If
** the conflicting change was an UPDATE, then the single record is a copy
** of the new.* record with the PK fields filled in based on the original
** old.* record.
*/
/*
** For each row modified during a session, there exists a single instance of
** this structure stored in a SessionTable.aChange[] hash table.
*/
struct SessionChange {
u8 op; /* One of UPDATE, DELETE, INSERT */
u8 bIndirect; /* True if this change is "indirect" */
u16 nRecordField; /* Number of fields in aRecord[] */
int nMaxSize; /* Max size of eventual changeset record */
int nRecord; /* Number of bytes in buffer aRecord[] */
u8 *aRecord; /* Buffer containing old.* record */
SessionChange *pNext; /* For hash-table collisions */
};
/*
** Write a varint with value iVal into the buffer at aBuf. Return the
** number of bytes written.
*/
static int sessionVarintPut(u8 *aBuf, int iVal){
return putVarint32(aBuf, iVal);
}
/*
** Return the number of bytes required to store value iVal as a varint.
*/
static int sessionVarintLen(int iVal){
return sqlite3VarintLen(iVal);
}
/*
** Read a varint value from aBuf[] into *piVal. Return the number of
** bytes read.
*/
static int sessionVarintGet(const u8 *aBuf, int *piVal){
return getVarint32(aBuf, *piVal);
}
/* Load an unaligned and unsigned 32-bit integer */
#define SESSION_UINT32(x) (((u32)(x)[0]<<24)|((x)[1]<<16)|((x)[2]<<8)|(x)[3])
/*
** Read a 64-bit big-endian integer value from buffer aRec[]. Return
** the value read.
*/
static sqlite3_int64 sessionGetI64(u8 *aRec){
u64 x = SESSION_UINT32(aRec);
u32 y = SESSION_UINT32(aRec+4);
x = (x<<32) + y;
return (sqlite3_int64)x;
}
/*
** Write a 64-bit big-endian integer value to the buffer aBuf[].
*/
static void sessionPutI64(u8 *aBuf, sqlite3_int64 i){
aBuf[0] = (i>>56) & 0xFF;
aBuf[1] = (i>>48) & 0xFF;
aBuf[2] = (i>>40) & 0xFF;
aBuf[3] = (i>>32) & 0xFF;
aBuf[4] = (i>>24) & 0xFF;
aBuf[5] = (i>>16) & 0xFF;
aBuf[6] = (i>> 8) & 0xFF;
aBuf[7] = (i>> 0) & 0xFF;
}
/*
** This function is used to serialize the contents of value pValue (see
** comment titled "RECORD FORMAT" above).
**
** If it is non-NULL, the serialized form of the value is written to
** buffer aBuf. *pnWrite is set to the number of bytes written before
** returning. Or, if aBuf is NULL, the only thing this function does is
** set *pnWrite.
**
** If no error occurs, SQLITE_OK is returned. Or, if an OOM error occurs
** within a call to sqlite3_value_text() (may fail if the db is utf-16))
** SQLITE_NOMEM is returned.
*/
static int sessionSerializeValue(
u8 *aBuf, /* If non-NULL, write serialized value here */
sqlite3_value *pValue, /* Value to serialize */
sqlite3_int64 *pnWrite /* IN/OUT: Increment by bytes written */
){
int nByte; /* Size of serialized value in bytes */
if( pValue ){
int eType; /* Value type (SQLITE_NULL, TEXT etc.) */
eType = sqlite3_value_type(pValue);
if( aBuf ) aBuf[0] = eType;
switch( eType ){
case SQLITE_NULL:
nByte = 1;
break;
case SQLITE_INTEGER:
case SQLITE_FLOAT:
if( aBuf ){
/* TODO: SQLite does something special to deal with mixed-endian
** floating point values (e.g. ARM7). This code probably should
** too. */
u64 i;
if( eType==SQLITE_INTEGER ){
i = (u64)sqlite3_value_int64(pValue);
}else{
double r;
assert( sizeof(double)==8 && sizeof(u64)==8 );
r = sqlite3_value_double(pValue);
memcpy(&i, &r, 8);
}
sessionPutI64(&aBuf[1], i);
}
nByte = 9;
break;
default: {
u8 *z;
int n;
int nVarint;
assert( eType==SQLITE_TEXT || eType==SQLITE_BLOB );
if( eType==SQLITE_TEXT ){
z = (u8 *)sqlite3_value_text(pValue);
}else{
z = (u8 *)sqlite3_value_blob(pValue);
}
n = sqlite3_value_bytes(pValue);
if( z==0 && (eType!=SQLITE_BLOB || n>0) ) return SQLITE_NOMEM;
nVarint = sessionVarintLen(n);
if( aBuf ){
sessionVarintPut(&aBuf[1], n);
if( n>0 ) memcpy(&aBuf[nVarint + 1], z, n);
}
nByte = 1 + nVarint + n;
break;
}
}
}else{
nByte = 1;
if( aBuf ) aBuf[0] = '\0';
}
if( pnWrite ) *pnWrite += nByte;
return SQLITE_OK;
}
/*
** Allocate and return a pointer to a buffer nByte bytes in size. If
** pSession is not NULL, increase the sqlite3_session.nMalloc variable
** by the number of bytes allocated.
*/
static void *sessionMalloc64(sqlite3_session *pSession, i64 nByte){
void *pRet = sqlite3_malloc64(nByte);
if( pSession ) pSession->nMalloc += sqlite3_msize(pRet);
return pRet;
}
/*
** Free buffer pFree, which must have been allocated by an earlier
** call to sessionMalloc64(). If pSession is not NULL, decrease the
** sqlite3_session.nMalloc counter by the number of bytes freed.
*/
static void sessionFree(sqlite3_session *pSession, void *pFree){
if( pSession ) pSession->nMalloc -= sqlite3_msize(pFree);
sqlite3_free(pFree);
}
/*
** This macro is used to calculate hash key values for data structures. In
** order to use this macro, the entire data structure must be represented
** as a series of unsigned integers. In order to calculate a hash-key value
** for a data structure represented as three such integers, the macro may
** then be used as follows:
**
** int hash_key_value;
** hash_key_value = HASH_APPEND(0, <value 1>);
** hash_key_value = HASH_APPEND(hash_key_value, <value 2>);
** hash_key_value = HASH_APPEND(hash_key_value, <value 3>);
**
** In practice, the data structures this macro is used for are the primary
** key values of modified rows.
*/
#define HASH_APPEND(hash, add) ((hash) << 3) ^ (hash) ^ (unsigned int)(add)
/*
** Append the hash of the 64-bit integer passed as the second argument to the
** hash-key value passed as the first. Return the new hash-key value.
*/
static unsigned int sessionHashAppendI64(unsigned int h, i64 i){
h = HASH_APPEND(h, i & 0xFFFFFFFF);
return HASH_APPEND(h, (i>>32)&0xFFFFFFFF);
}
/*
** Append the hash of the blob passed via the second and third arguments to
** the hash-key value passed as the first. Return the new hash-key value.
*/
static unsigned int sessionHashAppendBlob(unsigned int h, int n, const u8 *z){
int i;
for(i=0; i<n; i++) h = HASH_APPEND(h, z[i]);
return h;
}
/*
** Append the hash of the data type passed as the second argument to the
** hash-key value passed as the first. Return the new hash-key value.
*/
static unsigned int sessionHashAppendType(unsigned int h, int eType){
return HASH_APPEND(h, eType);
}
/*
** This function may only be called from within a pre-update callback.
** It calculates a hash based on the primary key values of the old.* or
** new.* row currently available and, assuming no error occurs, writes it to
** *piHash before returning. If the primary key contains one or more NULL
** values, *pbNullPK is set to true before returning.
**
** If an error occurs, an SQLite error code is returned and the final values
** of *piHash asn *pbNullPK are undefined. Otherwise, SQLITE_OK is returned
** and the output variables are set as described above.
*/
static int sessionPreupdateHash(
sqlite3_session *pSession, /* Session object that owns pTab */
i64 iRowid,
SessionTable *pTab, /* Session table handle */
int bNew, /* True to hash the new.* PK */
int *piHash, /* OUT: Hash value */
int *pbNullPK /* OUT: True if there are NULL values in PK */
){
unsigned int h = 0; /* Hash value to return */
int i; /* Used to iterate through columns */
assert( pTab->nTotalCol==pSession->hook.xCount(pSession->hook.pCtx) );
if( pTab->bRowid ){
h = sessionHashAppendI64(h, iRowid);
}else{
assert( *pbNullPK==0 );
for(i=0; i<pTab->nCol; i++){
if( pTab->abPK[i] ){
int rc;
int eType;
sqlite3_value *pVal;
int iIdx = pTab->aiIdx[i];
if( bNew ){
rc = pSession->hook.xNew(pSession->hook.pCtx, iIdx, &pVal);
}else{
rc = pSession->hook.xOld(pSession->hook.pCtx, iIdx, &pVal);
}
if( rc!=SQLITE_OK ) return rc;
eType = sqlite3_value_type(pVal);
h = sessionHashAppendType(h, eType);
if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
i64 iVal;
if( eType==SQLITE_INTEGER ){
iVal = sqlite3_value_int64(pVal);
}else{
double rVal = sqlite3_value_double(pVal);
assert( sizeof(iVal)==8 && sizeof(rVal)==8 );
memcpy(&iVal, &rVal, 8);
}
h = sessionHashAppendI64(h, iVal);
}else if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){
const u8 *z;
int n;
if( eType==SQLITE_TEXT ){
z = (const u8 *)sqlite3_value_text(pVal);
}else{
z = (const u8 *)sqlite3_value_blob(pVal);
}
n = sqlite3_value_bytes(pVal);
if( !z && (eType!=SQLITE_BLOB || n>0) ) return SQLITE_NOMEM;
h = sessionHashAppendBlob(h, n, z);
}else{
assert( eType==SQLITE_NULL );
assert( pTab->bStat1==0 || i!=1 );
*pbNullPK = 1;
}
}
}
}
*piHash = (h % pTab->nChange);
return SQLITE_OK;
}
/*
** The buffer that the argument points to contains a serialized SQL value.
** Return the number of bytes of space occupied by the value (including
** the type byte).
*/
static int sessionSerialLen(const u8 *a){
int e;
int n;
assert( a!=0 );
e = *a;
if( e==0 || e==0xFF ) return 1;
if( e==SQLITE_NULL ) return 1;
if( e==SQLITE_INTEGER || e==SQLITE_FLOAT ) return 9;
return sessionVarintGet(&a[1], &n) + 1 + n;
}
/*
** Based on the primary key values stored in change aRecord, calculate a
** hash key. Assume the has table has nBucket buckets. The hash keys
** calculated by this function are compatible with those calculated by
** sessionPreupdateHash().
**
** The bPkOnly argument is non-zero if the record at aRecord[] is from
** a patchset DELETE. In this case the non-PK fields are omitted entirely.
*/
static unsigned int sessionChangeHash(
SessionTable *pTab, /* Table handle */
int bPkOnly, /* Record consists of PK fields only */
u8 *aRecord, /* Change record */
int nBucket /* Assume this many buckets in hash table */
){
unsigned int h = 0; /* Value to return */
int i; /* Used to iterate through columns */
u8 *a = aRecord; /* Used to iterate through change record */
for(i=0; i<pTab->nCol; i++){
int eType = *a;
int isPK = pTab->abPK[i];
if( bPkOnly && isPK==0 ) continue;
/* It is not possible for eType to be SQLITE_NULL here. The session
** module does not record changes for rows with NULL values stored in
** primary key columns. */
assert( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT
|| eType==SQLITE_TEXT || eType==SQLITE_BLOB
|| eType==SQLITE_NULL || eType==0
);
assert( !isPK || (eType!=0 && eType!=SQLITE_NULL) );
if( isPK ){
a++;
h = sessionHashAppendType(h, eType);
if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
h = sessionHashAppendI64(h, sessionGetI64(a));
a += 8;
}else{
int n;
a += sessionVarintGet(a, &n);
h = sessionHashAppendBlob(h, n, a);
a += n;
}
}else{
a += sessionSerialLen(a);
}
}
return (h % nBucket);
}
/*
** Arguments aLeft and aRight are pointers to change records for table pTab.
** This function returns true if the two records apply to the same row (i.e.
** have the same values stored in the primary key columns), or false
** otherwise.
*/
static int sessionChangeEqual(
SessionTable *pTab, /* Table used for PK definition */
int bLeftPkOnly, /* True if aLeft[] contains PK fields only */
u8 *aLeft, /* Change record */
int bRightPkOnly, /* True if aRight[] contains PK fields only */
u8 *aRight /* Change record */
){
u8 *a1 = aLeft; /* Cursor to iterate through aLeft */
u8 *a2 = aRight; /* Cursor to iterate through aRight */
int iCol; /* Used to iterate through table columns */
for(iCol=0; iCol<pTab->nCol; iCol++){
if( pTab->abPK[iCol] ){
int n1 = sessionSerialLen(a1);
int n2 = sessionSerialLen(a2);
if( n1!=n2 || memcmp(a1, a2, n1) ){
return 0;
}
a1 += n1;
a2 += n2;
}else{
if( bLeftPkOnly==0 ) a1 += sessionSerialLen(a1);
if( bRightPkOnly==0 ) a2 += sessionSerialLen(a2);
}
}
return 1;
}
/*
** Arguments aLeft and aRight both point to buffers containing change
** records with nCol columns. This function "merges" the two records into
** a single records which is written to the buffer at *paOut. *paOut is
** then set to point to one byte after the last byte written before
** returning.
**
** The merging of records is done as follows: For each column, if the
** aRight record contains a value for the column, copy the value from
** their. Otherwise, if aLeft contains a value, copy it. If neither
** record contains a value for a given column, then neither does the
** output record.
*/
static void sessionMergeRecord(
u8 **paOut,
int nCol,
u8 *aLeft,
u8 *aRight
){
u8 *a1 = aLeft; /* Cursor used to iterate through aLeft */
u8 *a2 = aRight; /* Cursor used to iterate through aRight */
u8 *aOut = *paOut; /* Output cursor */
int iCol; /* Used to iterate from 0 to nCol */
for(iCol=0; iCol<nCol; iCol++){
int n1 = sessionSerialLen(a1);
int n2 = sessionSerialLen(a2);
if( *a2 ){
memcpy(aOut, a2, n2);
aOut += n2;
}else{
memcpy(aOut, a1, n1);
aOut += n1;
}
a1 += n1;
a2 += n2;
}
*paOut = aOut;
}
/*
** This is a helper function used by sessionMergeUpdate().
**
** When this function is called, both *paOne and *paTwo point to a value
** within a change record. Before it returns, both have been advanced so
** as to point to the next value in the record.
**
** If, when this function is called, *paTwo points to a valid value (i.e.
** *paTwo[0] is not 0x00 - the "no value" placeholder), a copy of the *paTwo
** pointer is returned and *pnVal is set to the number of bytes in the
** serialized value. Otherwise, a copy of *paOne is returned and *pnVal
** set to the number of bytes in the value at *paOne. If *paOne points
** to the "no value" placeholder, *pnVal is set to 1. In other words:
**
** if( *paTwo is valid ) return *paTwo;
** return *paOne;
**
*/
static u8 *sessionMergeValue(
u8 **paOne, /* IN/OUT: Left-hand buffer pointer */
u8 **paTwo, /* IN/OUT: Right-hand buffer pointer */
int *pnVal /* OUT: Bytes in returned value */
){
u8 *a1 = *paOne;
u8 *a2 = *paTwo;
u8 *pRet = 0;
int n1;
assert( a1 );
if( a2 ){
int n2 = sessionSerialLen(a2);
if( *a2 ){
*pnVal = n2;
pRet = a2;
}
*paTwo = &a2[n2];
}
n1 = sessionSerialLen(a1);
if( pRet==0 ){
*pnVal = n1;
pRet = a1;
}
*paOne = &a1[n1];
return pRet;
}
/*
** This function is used by changeset_concat() to merge two UPDATE changes
** on the same row.
*/
static int sessionMergeUpdate(
u8 **paOut, /* IN/OUT: Pointer to output buffer */
SessionTable *pTab, /* Table change pertains to */
int bPatchset, /* True if records are patchset records */
u8 *aOldRecord1, /* old.* record for first change */
u8 *aOldRecord2, /* old.* record for second change */
u8 *aNewRecord1, /* new.* record for first change */
u8 *aNewRecord2 /* new.* record for second change */
){
u8 *aOld1 = aOldRecord1;
u8 *aOld2 = aOldRecord2;
u8 *aNew1 = aNewRecord1;
u8 *aNew2 = aNewRecord2;
u8 *aOut = *paOut;
int i;
if( bPatchset==0 ){
int bRequired = 0;
assert( aOldRecord1 && aNewRecord1 );
/* Write the old.* vector first. */
for(i=0; i<pTab->nCol; i++){
int nOld;
u8 *aOld;
int nNew;
u8 *aNew;
aOld = sessionMergeValue(&aOld1, &aOld2, &nOld);
aNew = sessionMergeValue(&aNew1, &aNew2, &nNew);
if( pTab->abPK[i] || nOld!=nNew || memcmp(aOld, aNew, nNew) ){
if( pTab->abPK[i]==0 ) bRequired = 1;
memcpy(aOut, aOld, nOld);
aOut += nOld;
}else{
*(aOut++) = '\0';
}
}
if( !bRequired ) return 0;
}
/* Write the new.* vector */
aOld1 = aOldRecord1;
aOld2 = aOldRecord2;
aNew1 = aNewRecord1;
aNew2 = aNewRecord2;
for(i=0; i<pTab->nCol; i++){
int nOld;
u8 *aOld;
int nNew;
u8 *aNew;
aOld = sessionMergeValue(&aOld1, &aOld2, &nOld);
aNew = sessionMergeValue(&aNew1, &aNew2, &nNew);
if( bPatchset==0
&& (pTab->abPK[i] || (nOld==nNew && 0==memcmp(aOld, aNew, nNew)))
){
*(aOut++) = '\0';
}else{
memcpy(aOut, aNew, nNew);
aOut += nNew;
}
}
*paOut = aOut;
return 1;
}
/*
** This function is only called from within a pre-update-hook callback.
** It determines if the current pre-update-hook change affects the same row
** as the change stored in argument pChange. If so, it returns true. Otherwise
** if the pre-update-hook does not affect the same row as pChange, it returns
** false.
*/
static int sessionPreupdateEqual(
sqlite3_session *pSession, /* Session object that owns SessionTable */
i64 iRowid, /* Rowid value if pTab->bRowid */
SessionTable *pTab, /* Table associated with change */
SessionChange *pChange, /* Change to compare to */
int op /* Current pre-update operation */
){
int iCol; /* Used to iterate through columns */
u8 *a = pChange->aRecord; /* Cursor used to scan change record */
if( pTab->bRowid ){
if( a[0]!=SQLITE_INTEGER ) return 0;
return sessionGetI64(&a[1])==iRowid;
}
assert( op==SQLITE_INSERT || op==SQLITE_UPDATE || op==SQLITE_DELETE );
for(iCol=0; iCol<pTab->nCol; iCol++){
if( !pTab->abPK[iCol] ){
a += sessionSerialLen(a);
}else{
sqlite3_value *pVal; /* Value returned by preupdate_new/old */
int rc; /* Error code from preupdate_new/old */
int eType = *a++; /* Type of value from change record */
int iIdx = pTab->aiIdx[iCol];
/* The following calls to preupdate_new() and preupdate_old() can not
** fail. This is because they cache their return values, and by the
** time control flows to here they have already been called once from
** within sessionPreupdateHash(). The first two asserts below verify
** this (that the method has already been called). */
if( op==SQLITE_INSERT ){
/* assert( db->pPreUpdate->pNewUnpacked || db->pPreUpdate->aNew ); */
rc = pSession->hook.xNew(pSession->hook.pCtx, iIdx, &pVal);
}else{
/* assert( db->pPreUpdate->pUnpacked ); */
rc = pSession->hook.xOld(pSession->hook.pCtx, iIdx, &pVal);
}
assert( rc==SQLITE_OK );
(void)rc; /* Suppress warning about unused variable */
if( sqlite3_value_type(pVal)!=eType ) return 0;
/* A SessionChange object never has a NULL value in a PK column */
assert( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT
|| eType==SQLITE_BLOB || eType==SQLITE_TEXT
);
if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
i64 iVal = sessionGetI64(a);
a += 8;
if( eType==SQLITE_INTEGER ){
if( sqlite3_value_int64(pVal)!=iVal ) return 0;
}else{
double rVal;
assert( sizeof(iVal)==8 && sizeof(rVal)==8 );
memcpy(&rVal, &iVal, 8);
if( sqlite3_value_double(pVal)!=rVal ) return 0;
}
}else{
int n;
const u8 *z;
a += sessionVarintGet(a, &n);
if( sqlite3_value_bytes(pVal)!=n ) return 0;
if( eType==SQLITE_TEXT ){
z = sqlite3_value_text(pVal);
}else{
z = sqlite3_value_blob(pVal);
}
if( n>0 && memcmp(a, z, n) ) return 0;
a += n;
}
}
}
return 1;
}
/*
** If required, grow the hash table used to store changes on table pTab
** (part of the session pSession). If a fatal OOM error occurs, set the
** session object to failed and return SQLITE_ERROR. Otherwise, return
** SQLITE_OK.
**
** It is possible that a non-fatal OOM error occurs in this function. In
** that case the hash-table does not grow, but SQLITE_OK is returned anyway.
** Growing the hash table in this case is a performance optimization only,
** it is not required for correct operation.
*/
static int sessionGrowHash(
sqlite3_session *pSession, /* For memory accounting. May be NULL */
int bPatchset,
SessionTable *pTab
){
if( pTab->nChange==0 || pTab->nEntry>=(pTab->nChange/2) ){
int i;
SessionChange **apNew;
sqlite3_int64 nNew = 2*(sqlite3_int64)(pTab->nChange ? pTab->nChange : 128);
apNew = (SessionChange**)sessionMalloc64(
pSession, sizeof(SessionChange*) * nNew
);
if( apNew==0 ){
if( pTab->nChange==0 ){
return SQLITE_ERROR;
}
return SQLITE_OK;
}
memset(apNew, 0, sizeof(SessionChange *) * nNew);
for(i=0; i<pTab->nChange; i++){
SessionChange *p;
SessionChange *pNext;
for(p=pTab->apChange[i]; p; p=pNext){
int bPkOnly = (p->op==SQLITE_DELETE && bPatchset);
int iHash = sessionChangeHash(pTab, bPkOnly, p->aRecord, nNew);
pNext = p->pNext;
p->pNext = apNew[iHash];
apNew[iHash] = p;
}
}
sessionFree(pSession, pTab->apChange);
pTab->nChange = nNew;
pTab->apChange = apNew;
}
return SQLITE_OK;
}
/*
** This function queries the database for the names of the columns of table
** zThis, in schema zDb.
**
** Otherwise, if they are not NULL, variable *pnCol is set to the number
** of columns in the database table and variable *pzTab is set to point to a
** nul-terminated copy of the table name. *pazCol (if not NULL) is set to
** point to an array of pointers to column names. And *pabPK (again, if not
** NULL) is set to point to an array of booleans - true if the corresponding
** column is part of the primary key.
**
** For example, if the table is declared as:
**
** CREATE TABLE tbl1(w, x DEFAULT 'abc', y, z, PRIMARY KEY(w, z));
**
** Then the five output variables are populated as follows:
**
** *pnCol = 4
** *pzTab = "tbl1"
** *pazCol = {"w", "x", "y", "z"}
** *pazDflt = {NULL, 'abc', NULL, NULL}
** *pabPK = {1, 0, 0, 1}
**
** All returned buffers are part of the same single allocation, which must
** be freed using sqlite3_free() by the caller
*/
static int sessionTableInfo(
sqlite3_session *pSession, /* For memory accounting. May be NULL */
sqlite3 *db, /* Database connection */
const char *zDb, /* Name of attached database (e.g. "main") */
const char *zThis, /* Table name */
int *pnCol, /* OUT: number of columns */
int *pnTotalCol, /* OUT: number of hidden columns */
const char **pzTab, /* OUT: Copy of zThis */
const char ***pazCol, /* OUT: Array of column names for table */
const char ***pazDflt, /* OUT: Array of default value expressions */
int **paiIdx, /* OUT: Array of xNew/xOld indexes */
u8 **pabPK, /* OUT: Array of booleans - true for PK col */
int *pbRowid /* OUT: True if only PK is a rowid */
){
char *zPragma;
sqlite3_stmt *pStmt;
int rc;
sqlite3_int64 nByte;
int nDbCol = 0;
int nThis;
int i;
u8 *pAlloc = 0;
char **azCol = 0;
char **azDflt = 0;
u8 *abPK = 0;
int *aiIdx = 0;
int bRowid = 0; /* Set to true to use rowid as PK */
assert( pazCol && pabPK );
*pazCol = 0;
*pabPK = 0;
*pnCol = 0;
if( pnTotalCol ) *pnTotalCol = 0;
if( paiIdx ) *paiIdx = 0;
if( pzTab ) *pzTab = 0;
if( pazDflt ) *pazDflt = 0;
nThis = sqlite3Strlen30(zThis);
if( nThis==12 && 0==sqlite3_stricmp("sqlite_stat1", zThis) ){
rc = sqlite3_table_column_metadata(db, zDb, zThis, 0, 0, 0, 0, 0, 0);
if( rc==SQLITE_OK ){
/* For sqlite_stat1, pretend that (tbl,idx) is the PRIMARY KEY. */
zPragma = sqlite3_mprintf(
"SELECT 0, 'tbl', '', 0, '', 1, 0 UNION ALL "
"SELECT 1, 'idx', '', 0, '', 2, 0 UNION ALL "
"SELECT 2, 'stat', '', 0, '', 0, 0"
);
}else if( rc==SQLITE_ERROR ){
zPragma = sqlite3_mprintf("");
}else{
return rc;
}
}else{
zPragma = sqlite3_mprintf("PRAGMA '%q'.table_xinfo('%q')", zDb, zThis);
}
if( !zPragma ){
return SQLITE_NOMEM;
}
rc = sqlite3_prepare_v2(db, zPragma, -1, &pStmt, 0);
sqlite3_free(zPragma);
if( rc!=SQLITE_OK ){
return rc;
}
nByte = nThis + 1;
bRowid = (pbRowid!=0);
while( SQLITE_ROW==sqlite3_step(pStmt) ){
nByte += sqlite3_column_bytes(pStmt, 1); /* name */
nByte += sqlite3_column_bytes(pStmt, 4); /* dflt_value */
if( sqlite3_column_int(pStmt, 6)==0 ){ /* !hidden */
nDbCol++;
}
if( sqlite3_column_int(pStmt, 5) ) bRowid = 0; /* pk */
}
if( nDbCol==0 ) bRowid = 0;
nDbCol += bRowid;
nByte += strlen(SESSIONS_ROWID);
rc = sqlite3_reset(pStmt);
if( rc==SQLITE_OK ){
nByte += nDbCol * (sizeof(const char *)*2 +sizeof(int)+sizeof(u8) + 1 + 1);
pAlloc = sessionMalloc64(pSession, nByte);
if( pAlloc==0 ){
rc = SQLITE_NOMEM;
}else{
memset(pAlloc, 0, nByte);
}
}
if( rc==SQLITE_OK ){
azCol = (char **)pAlloc;
azDflt = (char**)&azCol[nDbCol];
aiIdx = (int*)&azDflt[nDbCol];
abPK = (u8 *)&aiIdx[nDbCol];
pAlloc = &abPK[nDbCol];
if( pzTab ){
memcpy(pAlloc, zThis, nThis+1);
*pzTab = (char *)pAlloc;
pAlloc += nThis+1;
}
i = 0;
if( bRowid ){
size_t nName = strlen(SESSIONS_ROWID);
memcpy(pAlloc, SESSIONS_ROWID, nName+1);
azCol[i] = (char*)pAlloc;
pAlloc += nName+1;
abPK[i] = 1;
aiIdx[i] = -1;
i++;
}
while( SQLITE_ROW==sqlite3_step(pStmt) ){
if( sqlite3_column_int(pStmt, 6)==0 ){ /* !hidden */
int nName = sqlite3_column_bytes(pStmt, 1);
int nDflt = sqlite3_column_bytes(pStmt, 4);
const unsigned char *zName = sqlite3_column_text(pStmt, 1);
const unsigned char *zDflt = sqlite3_column_text(pStmt, 4);
if( zName==0 ) break;
memcpy(pAlloc, zName, nName+1);
azCol[i] = (char *)pAlloc;
pAlloc += nName+1;
if( zDflt ){
memcpy(pAlloc, zDflt, nDflt+1);
azDflt[i] = (char *)pAlloc;
pAlloc += nDflt+1;
}else{
azDflt[i] = 0;
}
abPK[i] = sqlite3_column_int(pStmt, 5);
aiIdx[i] = sqlite3_column_int(pStmt, 0);
i++;
}
if( pnTotalCol ) (*pnTotalCol)++;
}
rc = sqlite3_reset(pStmt);
}
/* If successful, populate the output variables. Otherwise, zero them and
** free any allocation made. An error code will be returned in this case.
*/
if( rc==SQLITE_OK ){
*pazCol = (const char**)azCol;
if( pazDflt ) *pazDflt = (const char**)azDflt;
*pabPK = abPK;
*pnCol = nDbCol;
if( paiIdx ) *paiIdx = aiIdx;
}else{
sessionFree(pSession, azCol);
}
if( pbRowid ) *pbRowid = bRowid;
sqlite3_finalize(pStmt);
return rc;
}
/*
** This function is called to initialize the SessionTable.nCol, azCol[]
** abPK[] and azDflt[] members of SessionTable object pTab. If these
** fields are already initilialized, this function is a no-op.
**
** If an error occurs, an error code is stored in sqlite3_session.rc and
** non-zero returned. Or, if no error occurs but the table has no primary
** key, sqlite3_session.rc is left set to SQLITE_OK and non-zero returned to
** indicate that updates on this table should be ignored. SessionTable.abPK
** is set to NULL in this case.
*/
static int sessionInitTable(
sqlite3_session *pSession, /* Optional session handle */
SessionTable *pTab, /* Table object to initialize */
sqlite3 *db, /* Database handle to read schema from */
const char *zDb /* Name of db - "main", "temp" etc. */
){
int rc = SQLITE_OK;
if( pTab->nCol==0 ){
u8 *abPK;
assert( pTab->azCol==0 || pTab->abPK==0 );
rc = sessionTableInfo(pSession, db, zDb,
pTab->zName, &pTab->nCol, &pTab->nTotalCol, 0, &pTab->azCol,
&pTab->azDflt, &pTab->aiIdx, &abPK,
((pSession==0 || pSession->bImplicitPK) ? &pTab->bRowid : 0)
);
if( rc==SQLITE_OK ){
int i;
for(i=0; i<pTab->nCol; i++){
if( abPK[i] ){
pTab->abPK = abPK;
break;
}
}
if( 0==sqlite3_stricmp("sqlite_stat1", pTab->zName) ){
pTab->bStat1 = 1;
}
if( pSession && pSession->bEnableSize ){
pSession->nMaxChangesetSize += (
1 + sessionVarintLen(pTab->nCol) + pTab->nCol + strlen(pTab->zName)+1
);
}
}
}
if( pSession ){
pSession->rc = rc;
return (rc || pTab->abPK==0);
}
return rc;
}
/*
** Re-initialize table object pTab.
*/
static int sessionReinitTable(sqlite3_session *pSession, SessionTable *pTab){
int nCol = 0;
int nTotalCol = 0;
const char **azCol = 0;
const char **azDflt = 0;
int *aiIdx = 0;
u8 *abPK = 0;
int bRowid = 0;
assert( pSession->rc==SQLITE_OK );
pSession->rc = sessionTableInfo(pSession, pSession->db, pSession->zDb,
pTab->zName, &nCol, &nTotalCol, 0, &azCol, &azDflt, &aiIdx, &abPK,
(pSession->bImplicitPK ? &bRowid : 0)
);
if( pSession->rc==SQLITE_OK ){
if( pTab->nCol>nCol || pTab->bRowid!=bRowid ){
pSession->rc = SQLITE_SCHEMA;
}else{
int ii;
int nOldCol = pTab->nCol;
for(ii=0; ii<nCol; ii++){
if( ii<pTab->nCol ){
if( pTab->abPK[ii]!=abPK[ii] ){
pSession->rc = SQLITE_SCHEMA;
}
}else if( abPK[ii] ){
pSession->rc = SQLITE_SCHEMA;
}
}
if( pSession->rc==SQLITE_OK ){
const char **a = pTab->azCol;
pTab->azCol = azCol;
pTab->nCol = nCol;
pTab->nTotalCol = nTotalCol;
pTab->azDflt = azDflt;
pTab->abPK = abPK;
pTab->aiIdx = aiIdx;
azCol = a;
}
if( pSession->bEnableSize ){
pSession->nMaxChangesetSize += (nCol - nOldCol);
pSession->nMaxChangesetSize += sessionVarintLen(nCol);
pSession->nMaxChangesetSize -= sessionVarintLen(nOldCol);
}
}
}
sqlite3_free((char*)azCol);
return pSession->rc;
}
/*
** Session-change object (*pp) contains an old.* record with fewer than
** nCol fields. This function updates it with the default values for
** the missing fields.
*/
static void sessionUpdateOneChange(
sqlite3_session *pSession, /* For memory accounting */
int *pRc, /* IN/OUT: Error code */
SessionChange **pp, /* IN/OUT: Change object to update */
int nCol, /* Number of columns now in table */
sqlite3_stmt *pDflt /* SELECT <default-values...> */
){
SessionChange *pOld = *pp;
while( pOld->nRecordField<nCol ){
SessionChange *pNew = 0;
int nByte = 0;
int nIncr = 0;
int iField = pOld->nRecordField;
int eType = sqlite3_column_type(pDflt, iField);
switch( eType ){
case SQLITE_NULL:
nIncr = 1;
break;
case SQLITE_INTEGER:
case SQLITE_FLOAT:
nIncr = 9;
break;
default: {
int n = sqlite3_column_bytes(pDflt, iField);
nIncr = 1 + sessionVarintLen(n) + n;
assert( eType==SQLITE_TEXT || eType==SQLITE_BLOB );
break;
}
}
nByte = nIncr + (sizeof(SessionChange) + pOld->nRecord);
pNew = sessionMalloc64(pSession, nByte);
if( pNew==0 ){
*pRc = SQLITE_NOMEM;
return;
}else{
memcpy(pNew, pOld, sizeof(SessionChange));
pNew->aRecord = (u8*)&pNew[1];
memcpy(pNew->aRecord, pOld->aRecord, pOld->nRecord);
pNew->aRecord[pNew->nRecord++] = (u8)eType;
switch( eType ){
case SQLITE_INTEGER: {
i64 iVal = sqlite3_column_int64(pDflt, iField);
sessionPutI64(&pNew->aRecord[pNew->nRecord], iVal);
pNew->nRecord += 8;
break;
}
case SQLITE_FLOAT: {
double rVal = sqlite3_column_double(pDflt, iField);
i64 iVal = 0;
memcpy(&iVal, &rVal, sizeof(rVal));
sessionPutI64(&pNew->aRecord[pNew->nRecord], iVal);
pNew->nRecord += 8;
break;
}
case SQLITE_TEXT: {
int n = sqlite3_column_bytes(pDflt, iField);
const char *z = (const char*)sqlite3_column_text(pDflt, iField);
pNew->nRecord += sessionVarintPut(&pNew->aRecord[pNew->nRecord], n);
memcpy(&pNew->aRecord[pNew->nRecord], z, n);
pNew->nRecord += n;
break;
}
case SQLITE_BLOB: {
int n = sqlite3_column_bytes(pDflt, iField);
const u8 *z = (const u8*)sqlite3_column_blob(pDflt, iField);
pNew->nRecord += sessionVarintPut(&pNew->aRecord[pNew->nRecord], n);
memcpy(&pNew->aRecord[pNew->nRecord], z, n);
pNew->nRecord += n;
break;
}
default:
assert( eType==SQLITE_NULL );
break;
}
sessionFree(pSession, pOld);
*pp = pOld = pNew;
pNew->nRecordField++;
pNew->nMaxSize += nIncr;
if( pSession ){
pSession->nMaxChangesetSize += nIncr;
}
}
}
}
/*
** Ensure that there is room in the buffer to append nByte bytes of data.
** If not, use sqlite3_realloc() to grow the buffer so that there is.
**
** If successful, return zero. Otherwise, if an OOM condition is encountered,
** set *pRc to SQLITE_NOMEM and return non-zero.
*/
static int sessionBufferGrow(SessionBuffer *p, i64 nByte, int *pRc){
#define SESSION_MAX_BUFFER_SZ (0x7FFFFF00 - 1)
i64 nReq = p->nBuf + nByte;
if( *pRc==SQLITE_OK && nReq>p->nAlloc ){
u8 *aNew;
i64 nNew = p->nAlloc ? p->nAlloc : 128;
do {
nNew = nNew*2;
}while( nNew<nReq );
/* The value of SESSION_MAX_BUFFER_SZ is copied from the implementation
** of sqlite3_realloc64(). Allocations greater than this size in bytes
** always fail. It is used here to ensure that this routine can always
** allocate up to this limit - instead of up to the largest power of
** two smaller than the limit. */
if( nNew>SESSION_MAX_BUFFER_SZ ){
nNew = SESSION_MAX_BUFFER_SZ;
if( nNew<nReq ){
*pRc = SQLITE_NOMEM;
return 1;
}
}
aNew = (u8 *)sqlite3_realloc64(p->aBuf, nNew);
if( 0==aNew ){
*pRc = SQLITE_NOMEM;
}else{
p->aBuf = aNew;
p->nAlloc = nNew;
}
}
return (*pRc!=SQLITE_OK);
}
/*
** This function is a no-op if *pRc is other than SQLITE_OK when it is
** called. Otherwise, append a string to the buffer. All bytes in the string
** up to (but not including) the nul-terminator are written to the buffer.
**
** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
** returning.
*/
static void sessionAppendStr(
SessionBuffer *p,
const char *zStr,
int *pRc
){
int nStr = sqlite3Strlen30(zStr);
if( 0==sessionBufferGrow(p, nStr+1, pRc) ){
memcpy(&p->aBuf[p->nBuf], zStr, nStr);
p->nBuf += nStr;
p->aBuf[p->nBuf] = 0x00;
}
}
/*
** Format a string using printf() style formatting and then append it to the
** buffer using sessionAppendString().
*/
static void sessionAppendPrintf(
SessionBuffer *p, /* Buffer to append to */
int *pRc,
const char *zFmt,
...
){
if( *pRc==SQLITE_OK ){
char *zApp = 0;
va_list ap;
va_start(ap, zFmt);
zApp = sqlite3_vmprintf(zFmt, ap);
if( zApp==0 ){
*pRc = SQLITE_NOMEM;
}else{
sessionAppendStr(p, zApp, pRc);
}
va_end(ap);
sqlite3_free(zApp);
}
}
/*
** Prepare a statement against database handle db that SELECTs a single
** row containing the default values for each column in table pTab. For
** example, if pTab is declared as:
**
** CREATE TABLE pTab(a PRIMARY KEY, b DEFAULT 123, c DEFAULT 'abcd');
**
** Then this function prepares and returns the SQL statement:
**
** SELECT NULL, 123, 'abcd';
*/
static int sessionPrepareDfltStmt(
sqlite3 *db, /* Database handle */
SessionTable *pTab, /* Table to prepare statement for */
sqlite3_stmt **ppStmt /* OUT: Statement handle */
){
SessionBuffer sql = {0,0,0};
int rc = SQLITE_OK;
const char *zSep = " ";
int ii = 0;
*ppStmt = 0;
sessionAppendPrintf(&sql, &rc, "SELECT");
for(ii=0; ii<pTab->nCol; ii++){
const char *zDflt = pTab->azDflt[ii] ? pTab->azDflt[ii] : "NULL";
sessionAppendPrintf(&sql, &rc, "%s%s", zSep, zDflt);
zSep = ", ";
}
if( rc==SQLITE_OK ){
rc = sqlite3_prepare_v2(db, (const char*)sql.aBuf, -1, ppStmt, 0);
}
sqlite3_free(sql.aBuf);
return rc;
}
/*
** Table pTab has one or more existing change-records with old.* records
** with fewer than pTab->nCol columns. This function updates all such
** change-records with the default values for the missing columns.
*/
static int sessionUpdateChanges(sqlite3_session *pSession, SessionTable *pTab){
sqlite3_stmt *pStmt = 0;
int rc = pSession->rc;
rc = sessionPrepareDfltStmt(pSession->db, pTab, &pStmt);
if( rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pStmt) ){
int ii = 0;
SessionChange **pp = 0;
for(ii=0; ii<pTab->nChange; ii++){
for(pp=&pTab->apChange[ii]; *pp; pp=&((*pp)->pNext)){
if( (*pp)->nRecordField!=pTab->nCol ){
sessionUpdateOneChange(pSession, &rc, pp, pTab->nCol, pStmt);
}
}
}
}
pSession->rc = rc;
rc = sqlite3_finalize(pStmt);
if( pSession->rc==SQLITE_OK ) pSession->rc = rc;
return pSession->rc;
}
/*
** Versions of the four methods in object SessionHook for use with the
** sqlite_stat1 table. The purpose of this is to substitute a zero-length
** blob each time a NULL value is read from the "idx" column of the
** sqlite_stat1 table.
*/
typedef struct SessionStat1Ctx SessionStat1Ctx;
struct SessionStat1Ctx {
SessionHook hook;
sqlite3_session *pSession;
};
static int sessionStat1Old(void *pCtx, int iCol, sqlite3_value **ppVal){
SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
sqlite3_value *pVal = 0;
int rc = p->hook.xOld(p->hook.pCtx, iCol, &pVal);
if( rc==SQLITE_OK && iCol==1 && sqlite3_value_type(pVal)==SQLITE_NULL ){
pVal = p->pSession->pZeroBlob;
}
*ppVal = pVal;
return rc;
}
static int sessionStat1New(void *pCtx, int iCol, sqlite3_value **ppVal){
SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
sqlite3_value *pVal = 0;
int rc = p->hook.xNew(p->hook.pCtx, iCol, &pVal);
if( rc==SQLITE_OK && iCol==1 && sqlite3_value_type(pVal)==SQLITE_NULL ){
pVal = p->pSession->pZeroBlob;
}
*ppVal = pVal;
return rc;
}
static int sessionStat1Count(void *pCtx){
SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
return p->hook.xCount(p->hook.pCtx);
}
static int sessionStat1Depth(void *pCtx){
SessionStat1Ctx *p = (SessionStat1Ctx*)pCtx;
return p->hook.xDepth(p->hook.pCtx);
}
static int sessionUpdateMaxSize(
int op,
sqlite3_session *pSession, /* Session object pTab is attached to */
SessionTable *pTab, /* Table that change applies to */
SessionChange *pC /* Update pC->nMaxSize */
){
i64 nNew = 2;
if( pC->op==SQLITE_INSERT ){
if( pTab->bRowid ) nNew += 9;
if( op!=SQLITE_DELETE ){
int ii;
for(ii=0; ii<pTab->nCol; ii++){
sqlite3_value *p = 0;
pSession->hook.xNew(pSession->hook.pCtx, pTab->aiIdx[ii], &p);
sessionSerializeValue(0, p, &nNew);
}
}
}else if( op==SQLITE_DELETE ){
nNew += pC->nRecord;
if( sqlite3_preupdate_blobwrite(pSession->db)>=0 ){
nNew += pC->nRecord;
}
}else{
int ii;
u8 *pCsr = pC->aRecord;
if( pTab->bRowid ){
nNew += 9 + 1;
pCsr += 9;
}
for(ii=pTab->bRowid; ii<pTab->nCol; ii++){
int bChanged = 1;
int nOld = 0;
int eType;
int iIdx = pTab->aiIdx[ii];
sqlite3_value *p = 0;
pSession->hook.xNew(pSession->hook.pCtx, iIdx, &p);
if( p==0 ){
return SQLITE_NOMEM;
}
eType = *pCsr++;
switch( eType ){
case SQLITE_NULL:
bChanged = sqlite3_value_type(p)!=SQLITE_NULL;
break;
case SQLITE_FLOAT:
case SQLITE_INTEGER: {
if( eType==sqlite3_value_type(p) ){
sqlite3_int64 iVal = sessionGetI64(pCsr);
if( eType==SQLITE_INTEGER ){
bChanged = (iVal!=sqlite3_value_int64(p));
}else{
double dVal;
memcpy(&dVal, &iVal, 8);
bChanged = (dVal!=sqlite3_value_double(p));
}
}
nOld = 8;
pCsr += 8;
break;
}
default: {
int nByte;
nOld = sessionVarintGet(pCsr, &nByte);
pCsr += nOld;
nOld += nByte;
assert( eType==SQLITE_TEXT || eType==SQLITE_BLOB );
if( eType==sqlite3_value_type(p)
&& nByte==sqlite3_value_bytes(p)
&& (nByte==0 || 0==memcmp(pCsr, sqlite3_value_blob(p), nByte))
){
bChanged = 0;
}
pCsr += nByte;
break;
}
}
if( bChanged && pTab->abPK[ii] ){
nNew = pC->nRecord + 2;
break;
}
if( bChanged ){
nNew += 1 + nOld;
sessionSerializeValue(0, p, &nNew);
}else if( pTab->abPK[ii] ){
nNew += 2 + nOld;
}else{
nNew += 2;
}
}
}
if( nNew>pC->nMaxSize ){
int nIncr = nNew - pC->nMaxSize;
pC->nMaxSize = nNew;
pSession->nMaxChangesetSize += nIncr;
}
return SQLITE_OK;
}
/*
** This function is only called from with a pre-update-hook reporting a
** change on table pTab (attached to session pSession). The type of change
** (UPDATE, INSERT, DELETE) is specified by the first argument.
**
** Unless one is already present or an error occurs, an entry is added
** to the changed-rows hash table associated with table pTab.
*/
static void sessionPreupdateOneChange(
int op, /* One of SQLITE_UPDATE, INSERT, DELETE */
i64 iRowid,
sqlite3_session *pSession, /* Session object pTab is attached to */
SessionTable *pTab /* Table that change applies to */
){
int iHash;
int bNull = 0;
int rc = SQLITE_OK;
int nExpect = 0;
SessionStat1Ctx stat1 = {{0,0,0,0,0},0};
if( pSession->rc ) return;
/* Load table details if required */
if( sessionInitTable(pSession, pTab, pSession->db, pSession->zDb) ) return;
/* Check the number of columns in this xPreUpdate call matches the
** number of columns in the table. */
nExpect = pSession->hook.xCount(pSession->hook.pCtx);
if( pTab->nTotalCol<nExpect ){
if( sessionReinitTable(pSession, pTab) ) return;
if( sessionUpdateChanges(pSession, pTab) ) return;
}
if( pTab->nTotalCol!=nExpect ){
pSession->rc = SQLITE_SCHEMA;
return;
}
/* Grow the hash table if required */
if( sessionGrowHash(pSession, 0, pTab) ){
pSession->rc = SQLITE_NOMEM;
return;
}
if( pTab->bStat1 ){
stat1.hook = pSession->hook;
stat1.pSession = pSession;
pSession->hook.pCtx = (void*)&stat1;
pSession->hook.xNew = sessionStat1New;
pSession->hook.xOld = sessionStat1Old;
pSession->hook.xCount = sessionStat1Count;
pSession->hook.xDepth = sessionStat1Depth;
if( pSession->pZeroBlob==0 ){
sqlite3_value *p = sqlite3ValueNew(0);
if( p==0 ){
rc = SQLITE_NOMEM;
goto error_out;
}
sqlite3ValueSetStr(p, 0, "", 0, SQLITE_STATIC);
pSession->pZeroBlob = p;
}
}
/* Calculate the hash-key for this change. If the primary key of the row
** includes a NULL value, exit early. Such changes are ignored by the
** session module. */
rc = sessionPreupdateHash(
pSession, iRowid, pTab, op==SQLITE_INSERT, &iHash, &bNull
);
if( rc!=SQLITE_OK ) goto error_out;
if( bNull==0 ){
/* Search the hash table for an existing record for this row. */
SessionChange *pC;
for(pC=pTab->apChange[iHash]; pC; pC=pC->pNext){
if( sessionPreupdateEqual(pSession, iRowid, pTab, pC, op) ) break;
}
if( pC==0 ){
/* Create a new change object containing all the old values (if
** this is an SQLITE_UPDATE or SQLITE_DELETE), or just the PK
** values (if this is an INSERT). */
sqlite3_int64 nByte; /* Number of bytes to allocate */
int i; /* Used to iterate through columns */
assert( rc==SQLITE_OK );
pTab->nEntry++;
/* Figure out how large an allocation is required */
nByte = sizeof(SessionChange);
for(i=pTab->bRowid; i<pTab->nCol; i++){
int iIdx = pTab->aiIdx[i];
sqlite3_value *p = 0;
if( op!=SQLITE_INSERT ){
/* This may fail if the column has a non-NULL default and was added
** using ALTER TABLE ADD COLUMN after this record was created. */
rc = pSession->hook.xOld(pSession->hook.pCtx, iIdx, &p);
}else if( pTab->abPK[i] ){
TESTONLY(int trc = ) pSession->hook.xNew(pSession->hook.pCtx,iIdx,&p);
assert( trc==SQLITE_OK );
}
if( rc==SQLITE_OK ){
/* This may fail if SQLite value p contains a utf-16 string that must
** be converted to utf-8 and an OOM error occurs while doing so. */
rc = sessionSerializeValue(0, p, &nByte);
}
if( rc!=SQLITE_OK ) goto error_out;
}
if( pTab->bRowid ){
nByte += 9; /* Size of rowid field - an integer */
}
/* Allocate the change object */
pC = (SessionChange*)sessionMalloc64(pSession, nByte);
if( !pC ){
rc = SQLITE_NOMEM;
goto error_out;
}else{
memset(pC, 0, sizeof(SessionChange));
pC->aRecord = (u8 *)&pC[1];
}
/* Populate the change object. None of the preupdate_old(),
** preupdate_new() or SerializeValue() calls below may fail as all
** required values and encodings have already been cached in memory.
** It is not possible for an OOM to occur in this block. */
nByte = 0;
if( pTab->bRowid ){
pC->aRecord[0] = SQLITE_INTEGER;
sessionPutI64(&pC->aRecord[1], iRowid);
nByte = 9;
}
for(i=pTab->bRowid; i<pTab->nCol; i++){
sqlite3_value *p = 0;
int iIdx = pTab->aiIdx[i];
if( op!=SQLITE_INSERT ){
pSession->hook.xOld(pSession->hook.pCtx, iIdx, &p);
}else if( pTab->abPK[i] ){
pSession->hook.xNew(pSession->hook.pCtx, iIdx, &p);
}
sessionSerializeValue(&pC->aRecord[nByte], p, &nByte);
}
/* Add the change to the hash-table */
if( pSession->bIndirect || pSession->hook.xDepth(pSession->hook.pCtx) ){
pC->bIndirect = 1;
}
pC->nRecordField = pTab->nCol;
pC->nRecord = nByte;
pC->op = op;
pC->pNext = pTab->apChange[iHash];
pTab->apChange[iHash] = pC;
}else if( pC->bIndirect ){
/* If the existing change is considered "indirect", but this current
** change is "direct", mark the change object as direct. */
if( pSession->hook.xDepth(pSession->hook.pCtx)==0
&& pSession->bIndirect==0
){
pC->bIndirect = 0;
}
}
assert( rc==SQLITE_OK );
if( pSession->bEnableSize ){
rc = sessionUpdateMaxSize(op, pSession, pTab, pC);
}
}
/* If an error has occurred, mark the session object as failed. */
error_out:
if( pTab->bStat1 ){
pSession->hook = stat1.hook;
}
if( rc!=SQLITE_OK ){
pSession->rc = rc;
}
}
static int sessionFindTable(
sqlite3_session *pSession,
const char *zName,
SessionTable **ppTab
){
int rc = SQLITE_OK;
int nName = sqlite3Strlen30(zName);
SessionTable *pRet;
/* Search for an existing table */
for(pRet=pSession->pTable; pRet; pRet=pRet->pNext){
if( 0==sqlite3_strnicmp(pRet->zName, zName, nName+1) ) break;
}
if( pRet==0 && pSession->bAutoAttach ){
/* If there is a table-filter configured, invoke it. If it returns 0,
** do not automatically add the new table. */
if( pSession->xTableFilter==0
|| pSession->xTableFilter(pSession->pFilterCtx, zName)
){
rc = sqlite3session_attach(pSession, zName);
if( rc==SQLITE_OK ){
pRet = pSession->pTable;
while( ALWAYS(pRet) && pRet->pNext ){
pRet = pRet->pNext;
}
assert( pRet!=0 );
assert( 0==sqlite3_strnicmp(pRet->zName, zName, nName+1) );
}
}
}
assert( rc==SQLITE_OK || pRet==0 );
*ppTab = pRet;
return rc;
}
/*
** The 'pre-update' hook registered by this module with SQLite databases.
*/
static void xPreUpdate(
void *pCtx, /* Copy of third arg to preupdate_hook() */
sqlite3 *db, /* Database handle */
int op, /* SQLITE_UPDATE, DELETE or INSERT */
char const *zDb, /* Database name */
char const *zName, /* Table name */
sqlite3_int64 iKey1, /* Rowid of row about to be deleted/updated */
sqlite3_int64 iKey2 /* New rowid value (for a rowid UPDATE) */
){
sqlite3_session *pSession;
int nDb = sqlite3Strlen30(zDb);
assert( sqlite3_mutex_held(db->mutex) );
(void)iKey1;
(void)iKey2;
for(pSession=(sqlite3_session *)pCtx; pSession; pSession=pSession->pNext){
SessionTable *pTab;
/* If this session is attached to a different database ("main", "temp"
** etc.), or if it is not currently enabled, there is nothing to do. Skip
** to the next session object attached to this database. */
if( pSession->bEnable==0 ) continue;
if( pSession->rc ) continue;
if( sqlite3_strnicmp(zDb, pSession->zDb, nDb+1) ) continue;
pSession->rc = sessionFindTable(pSession, zName, &pTab);
if( pTab ){
assert( pSession->rc==SQLITE_OK );
assert( op==SQLITE_UPDATE || iKey1==iKey2 );
sessionPreupdateOneChange(op, iKey1, pSession, pTab);
if( op==SQLITE_UPDATE ){
sessionPreupdateOneChange(SQLITE_INSERT, iKey2, pSession, pTab);
}
}
}
}
/*
** The pre-update hook implementations.
*/
static int sessionPreupdateOld(void *pCtx, int iVal, sqlite3_value **ppVal){
return sqlite3_preupdate_old((sqlite3*)pCtx, iVal, ppVal);
}
static int sessionPreupdateNew(void *pCtx, int iVal, sqlite3_value **ppVal){
return sqlite3_preupdate_new((sqlite3*)pCtx, iVal, ppVal);
}
static int sessionPreupdateCount(void *pCtx){
return sqlite3_preupdate_count((sqlite3*)pCtx);
}
static int sessionPreupdateDepth(void *pCtx){
return sqlite3_preupdate_depth((sqlite3*)pCtx);
}
/*
** Install the pre-update hooks on the session object passed as the only
** argument.
*/
static void sessionPreupdateHooks(
sqlite3_session *pSession
){
pSession->hook.pCtx = (void*)pSession->db;
pSession->hook.xOld = sessionPreupdateOld;
pSession->hook.xNew = sessionPreupdateNew;
pSession->hook.xCount = sessionPreupdateCount;
pSession->hook.xDepth = sessionPreupdateDepth;
}
typedef struct SessionDiffCtx SessionDiffCtx;
struct SessionDiffCtx {
sqlite3_stmt *pStmt;
int bRowid;
int nOldOff;
};
/*
** The diff hook implementations.
*/
static int sessionDiffOld(void *pCtx, int iVal, sqlite3_value **ppVal){
SessionDiffCtx *p = (SessionDiffCtx*)pCtx;
*ppVal = sqlite3_column_value(p->pStmt, iVal+p->nOldOff+p->bRowid);
return SQLITE_OK;
}
static int sessionDiffNew(void *pCtx, int iVal, sqlite3_value **ppVal){
SessionDiffCtx *p = (SessionDiffCtx*)pCtx;
*ppVal = sqlite3_column_value(p->pStmt, iVal+p->bRowid);
return SQLITE_OK;
}
static int sessionDiffCount(void *pCtx){
SessionDiffCtx *p = (SessionDiffCtx*)pCtx;
return (p->nOldOff ? p->nOldOff : sqlite3_column_count(p->pStmt)) - p->bRowid;
}
static int sessionDiffDepth(void *pCtx){
(void)pCtx;
return 0;
}
/*
** Install the diff hooks on the session object passed as the only
** argument.
*/
static void sessionDiffHooks(
sqlite3_session *pSession,
SessionDiffCtx *pDiffCtx
){
pSession->hook.pCtx = (void*)pDiffCtx;
pSession->hook.xOld = sessionDiffOld;
pSession->hook.xNew = sessionDiffNew;
pSession->hook.xCount = sessionDiffCount;
pSession->hook.xDepth = sessionDiffDepth;
}
static char *sessionExprComparePK(
int nCol,
const char *zDb1, const char *zDb2,
const char *zTab,
const char **azCol, u8 *abPK
){
int i;
const char *zSep = "";
char *zRet = 0;
for(i=0; i<nCol; i++){
if( abPK[i] ){
zRet = sqlite3_mprintf("%z%s\"%w\".\"%w\".\"%w\"=\"%w\".\"%w\".\"%w\"",
zRet, zSep, zDb1, zTab, azCol[i], zDb2, zTab, azCol[i]
);
zSep = " AND ";
if( zRet==0 ) break;
}
}
return zRet;
}
static char *sessionExprCompareOther(
int nCol,
const char *zDb1, const char *zDb2,
const char *zTab,
const char **azCol, u8 *abPK
){
int i;
const char *zSep = "";
char *zRet = 0;
int bHave = 0;
for(i=0; i<nCol; i++){
if( abPK[i]==0 ){
bHave = 1;
zRet = sqlite3_mprintf(
"%z%s\"%w\".\"%w\".\"%w\" IS NOT \"%w\".\"%w\".\"%w\"",
zRet, zSep, zDb1, zTab, azCol[i], zDb2, zTab, azCol[i]
);
zSep = " OR ";
if( zRet==0 ) break;
}
}
if( bHave==0 ){
assert( zRet==0 );
zRet = sqlite3_mprintf("0");
}
return zRet;
}
static char *sessionSelectFindNew(
const char *zDb1, /* Pick rows in this db only */
const char *zDb2, /* But not in this one */
int bRowid,
const char *zTbl, /* Table name */
const char *zExpr
){
const char *zSel = (bRowid ? SESSIONS_ROWID ", *" : "*");
char *zRet = sqlite3_mprintf(
"SELECT %s FROM \"%w\".\"%w\" WHERE NOT EXISTS ("
" SELECT 1 FROM \"%w\".\"%w\" WHERE %s"
")",
zSel, zDb1, zTbl, zDb2, zTbl, zExpr
);
return zRet;
}
static int sessionDiffFindNew(
int op,
sqlite3_session *pSession,
SessionTable *pTab,
const char *zDb1,
const char *zDb2,
char *zExpr
){
int rc = SQLITE_OK;
char *zStmt = sessionSelectFindNew(
zDb1, zDb2, pTab->bRowid, pTab->zName, zExpr
);
if( zStmt==0 ){
rc = SQLITE_NOMEM;
}else{
sqlite3_stmt *pStmt;
rc = sqlite3_prepare(pSession->db, zStmt, -1, &pStmt, 0);
if( rc==SQLITE_OK ){
SessionDiffCtx *pDiffCtx = (SessionDiffCtx*)pSession->hook.pCtx;
pDiffCtx->pStmt = pStmt;
pDiffCtx->nOldOff = 0;
pDiffCtx->bRowid = pTab->bRowid;
while( SQLITE_ROW==sqlite3_step(pStmt) ){
i64 iRowid = (pTab->bRowid ? sqlite3_column_int64(pStmt, 0) : 0);
sessionPreupdateOneChange(op, iRowid, pSession, pTab);
}
rc = sqlite3_finalize(pStmt);
}
sqlite3_free(zStmt);
}
return rc;
}
/*
** Return a comma-separated list of the fully-qualified (with both database
** and table name) column names from table pTab. e.g.
**
** "main"."t1"."a", "main"."t1"."b", "main"."t1"."c"
*/
static char *sessionAllCols(
const char *zDb,
SessionTable *pTab
){
int ii;
char *zRet = 0;
for(ii=0; ii<pTab->nCol; ii++){
zRet = sqlite3_mprintf("%z%s\"%w\".\"%w\".\"%w\"",
zRet, (zRet ? ", " : ""), zDb, pTab->zName, pTab->azCol[ii]
);
if( !zRet ) break;
}
return zRet;
}
static int sessionDiffFindModified(
sqlite3_session *pSession,
SessionTable *pTab,
const char *zFrom,
const char *zExpr
){
int rc = SQLITE_OK;
char *zExpr2 = sessionExprCompareOther(pTab->nCol,
pSession->zDb, zFrom, pTab->zName, pTab->azCol, pTab->abPK
);
if( zExpr2==0 ){
rc = SQLITE_NOMEM;
}else{
char *z1 = sessionAllCols(pSession->zDb, pTab);
char *z2 = sessionAllCols(zFrom, pTab);
char *zStmt = sqlite3_mprintf(
"SELECT %s,%s FROM \"%w\".\"%w\", \"%w\".\"%w\" WHERE %s AND (%z)",
z1, z2, pSession->zDb, pTab->zName, zFrom, pTab->zName, zExpr, zExpr2
);
if( zStmt==0 || z1==0 || z2==0 ){
rc = SQLITE_NOMEM;
}else{
sqlite3_stmt *pStmt;
rc = sqlite3_prepare(pSession->db, zStmt, -1, &pStmt, 0);
if( rc==SQLITE_OK ){
SessionDiffCtx *pDiffCtx = (SessionDiffCtx*)pSession->hook.pCtx;
pDiffCtx->pStmt = pStmt;
pDiffCtx->nOldOff = pTab->nCol;
while( SQLITE_ROW==sqlite3_step(pStmt) ){
i64 iRowid = (pTab->bRowid ? sqlite3_column_int64(pStmt, 0) : 0);
sessionPreupdateOneChange(SQLITE_UPDATE, iRowid, pSession, pTab);
}
rc = sqlite3_finalize(pStmt);
}
}
sqlite3_free(zStmt);
sqlite3_free(z1);
sqlite3_free(z2);
}
return rc;
}
int sqlite3session_diff(
sqlite3_session *pSession,
const char *zFrom,
const char *zTbl,
char **pzErrMsg
){
const char *zDb = pSession->zDb;
int rc = pSession->rc;
SessionDiffCtx d;
memset(&d, 0, sizeof(d));
sessionDiffHooks(pSession, &d);
sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
if( pzErrMsg ) *pzErrMsg = 0;
if( rc==SQLITE_OK ){
char *zExpr = 0;
sqlite3 *db = pSession->db;
SessionTable *pTo; /* Table zTbl */
/* Locate and if necessary initialize the target table object */
rc = sessionFindTable(pSession, zTbl, &pTo);
if( pTo==0 ) goto diff_out;
if( sessionInitTable(pSession, pTo, pSession->db, pSession->zDb) ){
rc = pSession->rc;
goto diff_out;
}
/* Check the table schemas match */
if( rc==SQLITE_OK ){
int bHasPk = 0;
int bMismatch = 0;
int nCol; /* Columns in zFrom.zTbl */
int bRowid = 0;
u8 *abPK;
const char **azCol = 0;
rc = sessionTableInfo(0, db, zFrom, zTbl,
&nCol, 0, 0, &azCol, 0, 0, &abPK,
pSession->bImplicitPK ? &bRowid : 0
);
if( rc==SQLITE_OK ){
if( pTo->nCol!=nCol ){
bMismatch = 1;
}else{
int i;
for(i=0; i<nCol; i++){
if( pTo->abPK[i]!=abPK[i] ) bMismatch = 1;
if( sqlite3_stricmp(azCol[i], pTo->azCol[i]) ) bMismatch = 1;
if( abPK[i] ) bHasPk = 1;
}
}
}
sqlite3_free((char*)azCol);
if( bMismatch ){
if( pzErrMsg ){
*pzErrMsg = sqlite3_mprintf("table schemas do not match");
}
rc = SQLITE_SCHEMA;
}
if( bHasPk==0 ){
/* Ignore tables with no primary keys */
goto diff_out;
}
}
if( rc==SQLITE_OK ){
zExpr = sessionExprComparePK(pTo->nCol,
zDb, zFrom, pTo->zName, pTo->azCol, pTo->abPK
);
}
/* Find new rows */
if( rc==SQLITE_OK ){
rc = sessionDiffFindNew(SQLITE_INSERT, pSession, pTo, zDb, zFrom, zExpr);
}
/* Find old rows */
if( rc==SQLITE_OK ){
rc = sessionDiffFindNew(SQLITE_DELETE, pSession, pTo, zFrom, zDb, zExpr);
}
/* Find modified rows */
if( rc==SQLITE_OK ){
rc = sessionDiffFindModified(pSession, pTo, zFrom, zExpr);
}
sqlite3_free(zExpr);
}
diff_out:
sessionPreupdateHooks(pSession);
sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
return rc;
}
/*
** Create a session object. This session object will record changes to
** database zDb attached to connection db.
*/
int sqlite3session_create(
sqlite3 *db, /* Database handle */
const char *zDb, /* Name of db (e.g. "main") */
sqlite3_session **ppSession /* OUT: New session object */
){
sqlite3_session *pNew; /* Newly allocated session object */
sqlite3_session *pOld; /* Session object already attached to db */
int nDb = sqlite3Strlen30(zDb); /* Length of zDb in bytes */
/* Zero the output value in case an error occurs. */
*ppSession = 0;
/* Allocate and populate the new session object. */
pNew = (sqlite3_session *)sqlite3_malloc64(sizeof(sqlite3_session) + nDb + 1);
if( !pNew ) return SQLITE_NOMEM;
memset(pNew, 0, sizeof(sqlite3_session));
pNew->db = db;
pNew->zDb = (char *)&pNew[1];
pNew->bEnable = 1;
memcpy(pNew->zDb, zDb, nDb+1);
sessionPreupdateHooks(pNew);
/* Add the new session object to the linked list of session objects
** attached to database handle $db. Do this under the cover of the db
** handle mutex. */
sqlite3_mutex_enter(sqlite3_db_mutex(db));
pOld = (sqlite3_session*)sqlite3_preupdate_hook(db, xPreUpdate, (void*)pNew);
pNew->pNext = pOld;
sqlite3_mutex_leave(sqlite3_db_mutex(db));
*ppSession = pNew;
return SQLITE_OK;
}
/*
** Free the list of table objects passed as the first argument. The contents
** of the changed-rows hash tables are also deleted.
*/
static void sessionDeleteTable(sqlite3_session *pSession, SessionTable *pList){
SessionTable *pNext;
SessionTable *pTab;
for(pTab=pList; pTab; pTab=pNext){
int i;
pNext = pTab->pNext;
for(i=0; i<pTab->nChange; i++){
SessionChange *p;
SessionChange *pNextChange;
for(p=pTab->apChange[i]; p; p=pNextChange){
pNextChange = p->pNext;
sessionFree(pSession, p);
}
}
sqlite3_finalize(pTab->pDfltStmt);
sessionFree(pSession, (char*)pTab->azCol); /* cast works around VC++ bug */
sessionFree(pSession, pTab->apChange);
sessionFree(pSession, pTab);
}
}
/*
** Delete a session object previously allocated using sqlite3session_create().
*/
void sqlite3session_delete(sqlite3_session *pSession){
sqlite3 *db = pSession->db;
sqlite3_session *pHead;
sqlite3_session **pp;
/* Unlink the session from the linked list of sessions attached to the
** database handle. Hold the db mutex while doing so. */
sqlite3_mutex_enter(sqlite3_db_mutex(db));
pHead = (sqlite3_session*)sqlite3_preupdate_hook(db, 0, 0);
for(pp=&pHead; ALWAYS((*pp)!=0); pp=&((*pp)->pNext)){
if( (*pp)==pSession ){
*pp = (*pp)->pNext;
if( pHead ) sqlite3_preupdate_hook(db, xPreUpdate, (void*)pHead);
break;
}
}
sqlite3_mutex_leave(sqlite3_db_mutex(db));
sqlite3ValueFree(pSession->pZeroBlob);
/* Delete all attached table objects. And the contents of their
** associated hash-tables. */
sessionDeleteTable(pSession, pSession->pTable);
/* Free the session object. */
sqlite3_free(pSession);
}
/*
** Set a table filter on a Session Object.
*/
void sqlite3session_table_filter(
sqlite3_session *pSession,
int(*xFilter)(void*, const char*),
void *pCtx /* First argument passed to xFilter */
){
pSession->bAutoAttach = 1;
pSession->pFilterCtx = pCtx;
pSession->xTableFilter = xFilter;
}
/*
** Attach a table to a session. All subsequent changes made to the table
** while the session object is enabled will be recorded.
**
** Only tables that have a PRIMARY KEY defined may be attached. It does
** not matter if the PRIMARY KEY is an "INTEGER PRIMARY KEY" (rowid alias)
** or not.
*/
int sqlite3session_attach(
sqlite3_session *pSession, /* Session object */
const char *zName /* Table name */
){
int rc = SQLITE_OK;
sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));
if( !zName ){
pSession->bAutoAttach = 1;
}else{
SessionTable *pTab; /* New table object (if required) */
int nName; /* Number of bytes in string zName */
/* First search for an existing entry. If one is found, this call is
** a no-op. Return early. */
nName = sqlite3Strlen30(zName);
for(pTab=pSession->pTable; pTab; pTab=pTab->pNext){
if( 0==sqlite3_strnicmp(pTab->zName, zName, nName+1) ) break;
}
if( !pTab ){
/* Allocate new SessionTable object. */
int nByte = sizeof(SessionTable) + nName + 1;
pTab = (SessionTable*)sessionMalloc64(pSession, nByte);
if( !pTab ){
rc = SQLITE_NOMEM;
}else{
/* Populate the new SessionTable object and link it into the list.
** The new object must be linked onto the end of the list, not
** simply added to the start of it in order to ensure that tables
** appear in the correct order when a changeset or patchset is
** eventually generated. */
SessionTable **ppTab;
memset(pTab, 0, sizeof(SessionTable));
pTab->zName = (char *)&pTab[1];
memcpy(pTab->zName, zName, nName+1);
for(ppTab=&pSession->pTable; *ppTab; ppTab=&(*ppTab)->pNext);
*ppTab = pTab;
}
}
}
sqlite3_mutex_leave(sqlite3_db_mutex(pSession->db));
return rc;
}
/*
** Append the value passed as the second argument to the buffer passed
** as the first.
**
** This function is a no-op if *pRc is non-zero when it is called.
** Otherwise, if an error occurs, *pRc is set to an SQLite error code
** before returning.
*/
static void sessionAppendValue(SessionBuffer *p, sqlite3_value *pVal, int *pRc){
int rc = *pRc;
if( rc==SQLITE_OK ){
sqlite3_int64 nByte = 0;
rc = sessionSerializeValue(0, pVal, &nByte);
sessionBufferGrow(p, nByte, &rc);
if( rc==SQLITE_OK ){
rc = sessionSerializeValue(&p->aBuf[p->nBuf], pVal, 0);
p->nBuf += nByte;
}else{
*pRc = rc;
}
}
}
/*
** This function is a no-op if *pRc is other than SQLITE_OK when it is
** called. Otherwise, append a single byte to the buffer.
**
** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
** returning.
*/
static void sessionAppendByte(SessionBuffer *p, u8 v, int *pRc){
if( 0==sessionBufferGrow(p, 1, pRc) ){
p->aBuf[p->nBuf++] = v;
}
}
/*
** This function is a no-op if *pRc is other than SQLITE_OK when it is
** called. Otherwise, append a single varint to the buffer.
**
** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
** returning.
*/
static void sessionAppendVarint(SessionBuffer *p, int v, int *pRc){
if( 0==sessionBufferGrow(p, 9, pRc) ){
p->nBuf += sessionVarintPut(&p->aBuf[p->nBuf], v);
}
}
/*
** This function is a no-op if *pRc is other than SQLITE_OK when it is
** called. Otherwise, append a blob of data to the buffer.
**
** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
** returning.
*/
static void sessionAppendBlob(
SessionBuffer *p,
const u8 *aBlob,
int nBlob,
int *pRc
){
if( nBlob>0 && 0==sessionBufferGrow(p, nBlob, pRc) ){
memcpy(&p->aBuf[p->nBuf], aBlob, nBlob);
p->nBuf += nBlob;
}
}
/*
** This function is a no-op if *pRc is other than SQLITE_OK when it is
** called. Otherwise, append the string representation of integer iVal
** to the buffer. No nul-terminator is written.
**
** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
** returning.
*/
static void sessionAppendInteger(
SessionBuffer *p, /* Buffer to append to */
int iVal, /* Value to write the string rep. of */
int *pRc /* IN/OUT: Error code */
){
char aBuf[24];
sqlite3_snprintf(sizeof(aBuf)-1, aBuf, "%d", iVal);
sessionAppendStr(p, aBuf, pRc);
}
/*
** This function is a no-op if *pRc is other than SQLITE_OK when it is
** called. Otherwise, append the string zStr enclosed in quotes (") and
** with any embedded quote characters escaped to the buffer. No
** nul-terminator byte is written.
**
** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before
** returning.
*/
static void sessionAppendIdent(
SessionBuffer *p, /* Buffer to a append to */
const char *zStr, /* String to quote, escape and append */
int *pRc /* IN/OUT: Error code */
){
int nStr = sqlite3Strlen30(zStr)*2 + 2 + 2;
if( 0==sessionBufferGrow(p, nStr, pRc) ){
char *zOut = (char *)&p->aBuf[p->nBuf];
const char *zIn = zStr;
*zOut++ = '"';
if( zIn!=0 ){
while( *zIn ){
if( *zIn=='"' ) *zOut++ = '"';
*zOut++ = *(zIn++);
}
}
*zOut++ = '"';
p->nBuf = (int)((u8 *)zOut - p->aBuf);
p->aBuf[p->nBuf] = 0x00;
}
}
/*
** This function is a no-op if *pRc is other than SQLITE_OK when it is
** called. Otherwse, it appends the serialized version of the value stored
** in column iCol of the row that SQL statement pStmt currently points
** to to the buffer.
*/
static void sessionAppendCol(
SessionBuffer *p, /* Buffer to append to */
sqlite3_stmt *pStmt, /* Handle pointing to row containing value */
int iCol, /* Column to read value from */
int *pRc /* IN/OUT: Error code */
){
if( *pRc==SQLITE_OK ){
int eType = sqlite3_column_type(pStmt, iCol);
sessionAppendByte(p, (u8)eType, pRc);
if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){
sqlite3_int64 i;
u8 aBuf[8];
if( eType==SQLITE_INTEGER ){
i = sqlite3_column_int64(pStmt, iCol);
}else{
double r = sqlite3_column_double(pStmt, iCol);
memcpy(&i, &r, 8);
}
sessionPutI64(aBuf, i);
sessionAppendBlob(p, aBuf, 8, pRc);
}
if( eType==SQLITE_BLOB || eType==SQLITE_TEXT ){
u8 *z;
int nByte;
if( eType==SQLITE_BLOB ){
z = (u8 *)sqlite3_column_blob(pStmt, iCol);
}else{
z = (u8 *)sqlite3_column_text(pStmt, iCol);
}
nByte = sqlite3_column_bytes(pStmt, iCol);
if( z || (eType==SQLITE_BLOB && nByte==0) ){
sessionAppendVarint(p, nByte, pRc);
sessionAppendBlob(p, z, nByte, pRc);
}else{
*pRc = SQLITE_NOMEM;
}
}
}
}
/*
**
** This function appends an update change to the buffer (see the comments
** under "CHANGESET FORMAT" at the top of the file). An update change
** consists of:
**
** 1 byte: SQLITE_UPDATE (0x17)
** n bytes: old.* record (see RECORD FORMAT)
** m bytes: new.* record (see RECORD FORMAT)
**
** The SessionChange object passed as the third argument contains the
** values that were stored in the row when the session began (the old.*
** values). The statement handle passed as the second argument points
** at the current version of the row (the new.* values).
**
** If all of the old.* values are equal to their corresponding new.* value
** (i.e. nothing has changed), then no data at all is appended to the buffer.
**
** Otherwise, the old.* record contains all primary key values and the
** original values of any fields that have been modified. The new.* record
** contains the new values of only those fields that have been modified.
*/
static int sessionAppendUpdate(
SessionBuffer *pBuf, /* Buffer to append to */
int bPatchset, /* True for "patchset", 0 for "changeset" */
sqlite3_stmt *pStmt, /* Statement handle pointing at new row */
SessionChange *p, /* Object containing old values */
u8 *abPK /* Boolean array - true for PK columns */
){
int rc = SQLITE_OK;
SessionBuffer buf2 = {0,0,0}; /* Buffer to accumulate new.* record in */
int bNoop = 1; /* Set to zero if any values are modified */
int nRewind = pBuf->nBuf; /* Set to zero if any values are modified */
int i; /* Used to iterate through columns */
u8 *pCsr = p->aRecord; /* Used to iterate through old.* values */
assert( abPK!=0 );
sessionAppendByte(pBuf, SQLITE_UPDATE, &rc);
sessionAppendByte(pBuf, p->bIndirect, &rc);
for(i=0; i<sqlite3_column_count(pStmt); i++){
int bChanged = 0;
int nAdvance;
int eType = *pCsr;
switch( eType ){
case SQLITE_NULL:
nAdvance = 1;
if( sqlite3_column_type(pStmt, i)!=SQLITE_NULL ){
bChanged = 1;
}
break;
case SQLITE_FLOAT:
case SQLITE_INTEGER: {
nAdvance = 9;
if( eType==sqlite3_column_type(pStmt, i) ){
sqlite3_int64 iVal = sessionGetI64(&pCsr[1]);
if( eType==SQLITE_INTEGER ){
if( iVal==sqlite3_column_int64(pStmt, i) ) break;
}else{
double dVal;
memcpy(&dVal, &iVal, 8);
if( dVal==sqlite3_column_double(pStmt, i) ) break;
}
}
bChanged = 1;
break;
}
default: {
int n;
int nHdr = 1 + sessionVarintGet(&pCsr[1], &n);
assert( eType==SQLITE_TEXT || eType==SQLITE_BLOB );
nAdvance = nHdr + n;
if( eType==sqlite3_column_type(pStmt, i)
&& n==sqlite3_column_bytes(pStmt, i)
&& (n==0 || 0==memcmp(&pCsr[nHdr], sqlite3_column_blob(pStmt, i), n))
){
break;
}
bChanged = 1;
}
}
/* If at least one field has been modified, this is not a no-op. */
if( bChanged ) bNoop = 0;
/* Add a field to the old.* record. This is omitted if this module is
** currently generating a patchset. */
if( bPatchset==0 ){
if( bChanged || abPK[i] ){
sessionAppendBlob(pBuf, pCsr, nAdvance, &rc);
}else{
sessionAppendByte(pBuf, 0, &rc);
}
}
/* Add a field to the new.* record. Or the only record if currently
** generating a patchset. */
if( bChanged || (bPatchset && abPK[i]) ){
sessionAppendCol(&buf2, pStmt, i, &rc);
}else{
sessionAppendByte(&buf2, 0, &rc);
}
pCsr += nAdvance;
}
if( bNoop ){
pBuf->nBuf = nRewind;
}else{
sessionAppendBlob(pBuf, buf2.aBuf, buf2.nBuf, &rc);
}
sqlite3_free(buf2.aBuf);
return rc;
}
/*
** Append a DELETE change to the buffer passed as the first argument. Use
** the changeset format if argument bPatchset is zero, or the patchset
** format otherwise.
*/
static int sessionAppendDelete(
SessionBuffer *pBuf, /* Buffer to append to */
int bPatchset, /* True for "patchset", 0 for "changeset" */
SessionChange *p, /* Object containing old values */
int nCol, /* Number of columns in table */
u8 *abPK /* Boolean array - true for PK columns */
){
int rc = SQLITE_OK;
sessionAppendByte(pBuf, SQLITE_DELETE, &rc);
sessionAppendByte(pBuf, p->bIndirect, &rc);
if( bPatchset==0 ){
sessionAppendBlob(pBuf, p->aRecord, p->nRecord, &rc);
}else{
int i;
u8 *a = p->aRecord;
for(i=0; i<nCol; i++){
u8 *pStart = a;
int eType = *a++;
switch( eType ){
case 0:
case SQLITE_NULL:
assert( abPK[i]==0 );
break;
case SQLITE_FLOAT:
case SQLITE_INTEGER:
a += 8;
break;
default: {
int n;
a += sessionVarintGet(a, &n);
a += n;
break;
}
}
if( abPK[i] ){
sessionAppendBlob(pBuf, pStart, (int)(a-pStart), &rc);
}
}
assert( (a - p->aRecord)==p->nRecord );
}
return rc;
}
/*
** Formulate and prepare a SELECT statement to retrieve a row from table
** zTab in database zDb based on its primary key. i.e.
**
** SELECT *, <noop-test> FROM zDb.zTab WHERE (pk1, pk2,...) IS (?1, ?2,...)
**
** where <noop-test> is:
**
** 1 AND (?A OR ?1 IS <column>) AND ...
**
** for each non-pk <column>.
*/
static int sessionSelectStmt(
sqlite3 *db, /* Database handle */
int bIgnoreNoop,
const char *zDb, /* Database name */
const char *zTab, /* Table name */
int bRowid,
int nCol, /* Number of columns in table */
const char **azCol, /* Names of table columns */
u8 *abPK, /* PRIMARY KEY array */
sqlite3_stmt **ppStmt /* OUT: Prepared SELECT statement */
){
int rc = SQLITE_OK;
char *zSql = 0;
const char *zSep = "";
int nSql = -1;
int i;
SessionBuffer cols = {0, 0, 0};
SessionBuffer nooptest = {0, 0, 0};
SessionBuffer pkfield = {0, 0, 0};
SessionBuffer pkvar = {0, 0, 0};
sessionAppendStr(&nooptest, ", 1", &rc);
if( 0==sqlite3_stricmp("sqlite_stat1", zTab) ){
sessionAppendStr(&nooptest, " AND (?6 OR ?3 IS stat)", &rc);
sessionAppendStr(&pkfield, "tbl, idx", &rc);
sessionAppendStr(&pkvar,
"?1, (CASE WHEN ?2=X'' THEN NULL ELSE ?2 END)", &rc
);
sessionAppendStr(&cols, "tbl, ?2, stat", &rc);
}else{
#if 0
if( bRowid ){
sessionAppendStr(&cols, SESSIONS_ROWID, &rc);
}
#endif
for(i=0; i<nCol; i++){
if( cols.nBuf ) sessionAppendStr(&cols, ", ", &rc);
sessionAppendIdent(&cols, azCol[i], &rc);
if( abPK[i] ){
sessionAppendStr(&pkfield, zSep, &rc);
sessionAppendStr(&pkvar, zSep, &rc);
zSep = ", ";
sessionAppendIdent(&pkfield, azCol[i], &rc);
sessionAppendPrintf(&pkvar, &rc, "?%d", i+1);
}else{
sessionAppendPrintf(&nooptest, &rc,
" AND (?%d OR ?%d IS %w.%w)", i+1+nCol, i+1, zTab, azCol[i]
);
}
}
}
if( rc==SQLITE_OK ){
zSql = sqlite3_mprintf(
"SELECT %s%s FROM %Q.%Q WHERE (%s) IS (%s)",
(char*)cols.aBuf, (bIgnoreNoop ? (char*)nooptest.aBuf : ""),
zDb, zTab, (char*)pkfield.aBuf, (char*)pkvar.aBuf
);
if( zSql==0 ) rc = SQLITE_NOMEM;
}
#if 0
if( 0==sqlite3_stricmp("sqlite_stat1", zTab) ){
zSql = sqlite3_mprintf(
"SELECT tbl, ?2, stat FROM %Q.sqlite_stat1 WHERE tbl IS ?1 AND "
"idx IS (CASE WHEN ?2=X'' THEN NULL ELSE ?2 END)", zDb
);
if( zSql==0 ) rc = SQLITE_NOMEM;
}else{
const char *zSep = "";
SessionBuffer buf = {0, 0, 0};
sessionAppendStr(&buf, "SELECT * FROM ", &rc);
sessionAppendIdent(&buf, zDb, &rc);
sessionAppendStr(&buf, ".", &rc);
sessionAppendIdent(&buf, zTab, &rc);
sessionAppendStr(&buf, " WHERE ", &rc);
for(i=0; i<nCol; i++){
if( abPK[i] ){
sessionAppendStr(&buf, zSep, &rc);
sessionAppendIdent(&buf, azCol[i], &rc);
sessionAppendStr(&buf, " IS ?", &rc);
sessionAppendInteger(&buf, i+1, &rc);
zSep = " AND ";
}
}
zSql = (char*)buf.aBuf;
nSql = buf.nBuf;
}
#endif
if( rc==SQLITE_OK ){
rc = sqlite3_prepare_v2(db, zSql, nSql, ppStmt, 0);
}
sqlite3_free(zSql);
sqlite3_free(nooptest.aBuf);
sqlite3_free(pkfield.aBuf);
sqlite3_free(pkvar.aBuf);
sqlite3_free(cols.aBuf);
return rc;
}
/*
** Bind the PRIMARY KEY values from the change passed in argument pChange
** to the SELECT statement passed as the first argument. The SELECT statement
** is as prepared by function sessionSelectStmt().
**
** Return SQLITE_OK if all PK values are successfully bound, or an SQLite
** error code (e.g. SQLITE_NOMEM) otherwise.
*/
static int sessionSelectBind(
sqlite3_stmt *pSelect, /* SELECT from sessionSelectStmt() */
int nCol, /* Number of columns in table */
u8 *abPK, /* PRIMARY KEY array */
SessionChange *pChange /* Change structure */
){
int i;
int rc = SQLITE_OK;
u8 *a = pChange->aRecord;
for(i=0; i<nCol && rc==SQLITE_OK; i++){
int eType = *a++;
switch( eType ){
case 0:
case SQLITE_NULL:
assert( abPK[i]==0 );
break;
case SQLITE_INTEGER: {
if( abPK[i] ){
i64 iVal = sessionGetI64(a);
rc = sqlite3_bind_int64(pSelect, i+1, iVal);
}
a += 8;
break;
}
case SQLITE_FLOAT: {
if( abPK[i] ){
double rVal;
i64 iVal = sessionGetI64(a);
memcpy(&rVal, &iVal, 8);
rc = sqlite3_bind_double(pSelect, i+1, rVal);
}
a += 8;
break;
}
case SQLITE_TEXT: {
int n;
a += sessionVarintGet(a, &n);
if( abPK[i] ){
rc = sqlite3_bind_text(pSelect, i+1, (char *)a, n, SQLITE_TRANSIENT);
}
a += n;
break;
}
default: {
int n;
assert( eType==SQLITE_BLOB );
a += sessionVarintGet(a, &n);
if( abPK[i] ){
rc = sqlite3_bind_blob(pSelect, i+1, a, n, SQLITE_TRANSIENT);
}
a += n;
break;
}
}
}
return rc;
}
/*
** This function is a no-op if *pRc is set to other than SQLITE_OK when it
** is called. Otherwise, append a serialized table header (part of the binary
** changeset format) to buffer *pBuf. If an error occurs, set *pRc to an
** SQLite error code before returning.
*/
static void sessionAppendTableHdr(
SessionBuffer *pBuf, /* Append header to this buffer */
int bPatchset, /* Use the patchset format if true */
SessionTable *pTab, /* Table object to append header for */