SQLite

  set testdir [file join [file dirname [info script]] .. .. test]
} 
source $testdir/tester.tcl
ifcapable !session {finish_test; return}

set testprefix changebatch1


proc sql_to_changeset {method sql} {

  sqlite3session S db main
  S attach *
  execsql $sql
  set ret [S $method]
  S delete
  return $ret
}

proc do_changebatch_test {tn method args} {
  set C [list]
  foreach a $args {
    lappend C [sql_to_changeset $method $a]
  }

  sqlite3changebatch cb db
  set i 1
  foreach ::cs [lrange $C 0 end-1] {
    do_test $tn.$i { cb add [set ::cs] } SQLITE_OK
    incr i
  }

  set ::cs [lindex $C end]
  do_test $tn.$i { cb add [set ::cs] } SQLITE_CONSTRAINT

  cb delete
}

proc do_changebatch_test1 {tn args} {
  uplevel do_changebatch_test $tn changeset $args
}
proc do_changebatch_test2 {tn args} {
  uplevel do_changebatch_test $tn fullchangeset $args
}

#-------------------------------------------------------------------------
# The body of the following loop contains tests for database schemas
# that do not feature multi-column UNIQUE constraints. In this case
# it doesn't matter if the changesets are generated using
# sqlite3session_changeset() or sqlite3session_fullchangeset().
#
foreach {tn testfunction} {
  1 do_changebatch_test1
  2 do_changebatch_test2
} {
  reset_db

  #-------------------------------------------------------------------------
  #
  do_execsql_test $tn.1.0 {
    CREATE TABLE t1(a PRIMARY KEY, b);
  }
  
  $testfunction $tn.1.1 {
    INSERT INTO t1 VALUES(1, 1);
  } {
    DELETE FROM t1 WHERE a=1;
  }
  
  do_execsql_test $tn.1.2.0 {
    INSERT INTO t1 VALUES(1, 1);
    INSERT INTO t1 VALUES(2, 2);
    INSERT INTO t1 VALUES(3, 3);
  }
  $testfunction $tn.1.2.1 {
    DELETE FROM t1 WHERE a=2;
  } {
    INSERT INTO t1 VALUES(2, 2);
  }
  
  #-------------------------------------------------------------------------
  #
  do_execsql_test $tn.2.0 {
    CREATE TABLE x1(a, b PRIMARY KEY, c UNIQUE);
    CREATE TABLE x2(a PRIMARY KEY, b UNIQUE, c UNIQUE);
  
    INSERT INTO x1 VALUES(1, 1, 'a');
    INSERT INTO x1 VALUES(1, 2, 'b');
    INSERT INTO x1 VALUES(1, 3, 'c');
  }
  
  $testfunction $tn.2.1 {
    DELETE FROM x1 WHERE b=2;
  } {
    UPDATE x1 SET c='b' WHERE b=3;
  }
  
  $testfunction $tn.2.2 {
    DELETE FROM x1 WHERE b=1;
  } {
    INSERT INTO x1 VALUES(1, 5, 'a');
  }
}

#-------------------------------------------------------------------------
# Test some multi-column UNIQUE constraints. First Using _changeset() to
# demonstrate the problem, then using _fullchangeset() to show that it has
# been fixed.
#
reset_db
do_execsql_test 3.0 {

  CREATE TABLE y1(a PRIMARY KEY, b, c, UNIQUE(b, c));
  INSERT INTO y1 VALUES(1, 1, 1);
  INSERT INTO y1 VALUES(2, 2, 2);
  INSERT INTO y1 VALUES(3, 3, 3);
  INSERT INTO y1 VALUES(4, 3, 4);
  BEGIN;
}

do_test 3.1.1 {
  set c1 [sql_to_changeset changeset { DELETE FROM y1 WHERE a=4    }]
  set c2 [sql_to_changeset changeset { UPDATE y1 SET c=4 WHERE a=3 }]
  sqlite3changebatch cb db
  cb add $c1
  cb add $c2
} {SQLITE_OK}
do_test 3.1.2 {
  cb delete
  execsql ROLLBACK
} {}


do_test 3.1.1 {
  set c1 [sql_to_changeset fullchangeset { DELETE FROM y1 WHERE a=4    }]
  set c2 [sql_to_changeset fullchangeset { UPDATE y1 SET c=4 WHERE a=3 }]
  sqlite3changebatch cb db

  cb add $c1
  cb add $c2
} {SQLITE_CONSTRAINT}
do_test 3.1.2 {
  cb delete
} {}



finish_test

      }
    }
  }

  *ppTab = pRet;
  return rc;
}

static int cbGetChangesetValue(
  sqlite3_changeset_iter *pIter, 
  int (*xVal)(sqlite3_changeset_iter*,int,sqlite3_value**),
  int (*xFallback)(sqlite3_changeset_iter*,int,sqlite3_value**),
  int iVal,
  sqlite3_value **ppVal
){
  int rc = xVal(pIter, iVal, ppVal);
  if( rc==SQLITE_OK && *ppVal==0 && xFallback ){
    rc = xFallback(pIter, iVal, ppVal);
  }
  return rc;
}

static int cbAddToHash(
  sqlite3_changebatch *p, 
  sqlite3_changeset_iter *pIter, 
  BatchIndex *pIdx, 
  int (*xVal)(sqlite3_changeset_iter*,int,sqlite3_value**),
  int (*xFallback)(sqlite3_changeset_iter*,int,sqlite3_value**),
  int *pbConf
){
  BatchIndexEntry *pNew;
  int sz = pIdx->nCol;
  int i;
  int iOut = 0;
  int rc = SQLITE_OK;

  for(i=0; rc==SQLITE_OK && i<pIdx->nCol; i++){
    sqlite3_value *pVal;
    rc = cbGetChangesetValue(pIter, xVal, xFallback, pIdx->aiCol[i], &pVal);
    if( rc==SQLITE_OK ){
      int eType = 0;

      if( pVal ) eType = sqlite3_value_type(pVal);

      switch( eType ){
        case 0:
        case SQLITE_NULL:
          return SQLITE_OK;

        case SQLITE_INTEGER:
          sz += 8;

  if( pNew ){
    pNew->iChangesetId = p->iChangesetId;
    pNew->iIdxId = pIdx->iId;
    pNew->szRecord = sz;

    for(i=0; rc==SQLITE_OK && i<pIdx->nCol; i++){
      sqlite3_value *pVal;
      rc = cbGetChangesetValue(pIter, xVal, xFallback, pIdx->aiCol[i], &pVal);
      if( rc==SQLITE_OK ){
        int eType = sqlite3_value_type(pVal);
        pNew->aRecord[iOut++] = eType;
        switch( eType ){
          case SQLITE_INTEGER: {
            sqlite3_int64 i64 = sqlite3_value_int64(pVal);
            memcpy(&pNew->aRecord[iOut], &i64, 8);

      sqlite3changeset_op(pIter, &zTab, &nCol, &op, 0);
      assert( op==SQLITE_INSERT || op==SQLITE_UPDATE || op==SQLITE_DELETE );

      rc = cbFindTable(p, zTab, &pTab);
      for(pIdx=pTab->pIdx; pIdx && rc==SQLITE_OK; pIdx=pIdx->pNext){
        if( op==SQLITE_UPDATE && pIdx->bPk ) continue;
        if( op==SQLITE_UPDATE || op==SQLITE_DELETE ){
          rc = cbAddToHash(p, pIter, pIdx, sqlite3changeset_old, 0, &bConf);
        }
        if( op==SQLITE_UPDATE || op==SQLITE_INSERT ){
          rc = cbAddToHash(p, pIter, pIdx, 
              sqlite3changeset_new, sqlite3changeset_old, &bConf
          );
        }
      }
      if( rc!=SQLITE_OK ) break;
    }

    rc2 = sqlite3changeset_finalize(pIter);
    if( rc==SQLITE_OK ) rc = rc2;

# ifdef SQLITE_TEST
#   define SESSIONS_STRM_CHUNK_SIZE 64
# else
#   define SESSIONS_STRM_CHUNK_SIZE 1024
# endif
#endif

/*
** The three different types of changesets generated.
*/
#define SESSIONS_PATCHSET      0
#define SESSIONS_CHANGESET     1
#define SESSIONS_FULLCHANGESET 2

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*);

**
** 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 ePatchset,                  /* 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 */

    }

    /* 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 modules is
    ** currently generating a patchset. */
    if( ePatchset!=SESSIONS_PATCHSET ){
      if( ePatchset==SESSIONS_FULLCHANGESET || 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 || (ePatchset==SESSIONS_PATCHSET && abPK[i]) ){
      sessionAppendCol(&buf2, pStmt, i, &rc);
    }else{
      sessionAppendByte(&buf2, 0, &rc);
    }

    pCsr += nAdvance;
  }

/*
** 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 eChangeset,                 /* One of SESSIONS_CHANGESET etc. */
  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( eChangeset!=SESSIONS_PATCHSET ){
    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++;

** 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 ePatchset,                  /* Use the patchset format if true */
  SessionTable *pTab,             /* Table object to append header for */
  int *pRc                        /* IN/OUT: Error code */
){
  /* Write a table header */
  sessionAppendByte(pBuf, (ePatchset==SESSIONS_PATCHSET) ? 'P' : 'T', pRc);
  sessionAppendVarint(pBuf, pTab->nCol, pRc);
  sessionAppendBlob(pBuf, pTab->abPK, pTab->nCol, pRc);
  sessionAppendBlob(pBuf, (u8 *)pTab->zName, (int)strlen(pTab->zName)+1, pRc);
}

/*
** Generate either a changeset (if argument bPatchset is zero) or a patchset
** (if it is non-zero) based on the current contents of the session object
** passed as the first argument.
**
** If no error occurs, SQLITE_OK is returned and the new changeset/patchset
** stored in output variables *pnChangeset and *ppChangeset. Or, if an error
** occurs, an SQLite error code is returned and both output variables set 
** to 0.
*/
static int sessionGenerateChangeset(
  sqlite3_session *pSession,      /* Session object */
  int ePatchset,                  /* One of SESSIONS_CHANGESET etc. */
  int (*xOutput)(void *pOut, const void *pData, int nData),
  void *pOut,                     /* First argument for xOutput */
  int *pnChangeset,               /* OUT: Size of buffer at *ppChangeset */
  void **ppChangeset              /* OUT: Buffer containing changeset */
){
  sqlite3 *db = pSession->db;     /* Source database handle */
  SessionTable *pTab;             /* Used to iterate through attached tables */

      /* Check the table schema is still Ok. */
      rc = sessionTableInfo(db, pSession->zDb, zName, &nCol, 0, &azCol, &abPK);
      if( !rc && (pTab->nCol!=nCol || memcmp(abPK, pTab->abPK, nCol)) ){
        rc = SQLITE_SCHEMA;
      }

      /* Write a table header */
      sessionAppendTableHdr(&buf, ePatchset, pTab, &rc);

      /* Build and compile a statement to execute: */
      if( rc==SQLITE_OK ){
        rc = sessionSelectStmt(
            db, pSession->zDb, zName, nCol, azCol, abPK, &pSel);
      }

              int iCol;
              sessionAppendByte(&buf, SQLITE_INSERT, &rc);
              sessionAppendByte(&buf, p->bIndirect, &rc);
              for(iCol=0; iCol<nCol; iCol++){
                sessionAppendCol(&buf, pSel, iCol, &rc);
              }
            }else{
              rc = sessionAppendUpdate(&buf, ePatchset, pSel, p, abPK);
            }
          }else if( p->op!=SQLITE_INSERT ){
            rc = sessionAppendDelete(&buf, ePatchset, p, nCol, abPK);
          }
          if( rc==SQLITE_OK ){
            rc = sqlite3_reset(pSel);
          }

          /* If the buffer is now larger than SESSIONS_STRM_CHUNK_SIZE, pass
          ** its contents to the xOutput() callback. */

** using sqlite3_free().
*/
int sqlite3session_changeset(
  sqlite3_session *pSession,      /* Session object */
  int *pnChangeset,               /* OUT: Size of buffer at *ppChangeset */
  void **ppChangeset              /* OUT: Buffer containing changeset */
){
  return sessionGenerateChangeset(
      pSession, SESSIONS_CHANGESET, 0, 0, pnChangeset, ppChangeset);
}

/*
** Streaming version of sqlite3session_changeset().
*/
int sqlite3session_changeset_strm(
  sqlite3_session *pSession,
  int (*xOutput)(void *pOut, const void *pData, int nData),
  void *pOut
){
  return sessionGenerateChangeset(
      pSession, SESSIONS_CHANGESET, xOutput, pOut, 0, 0);
}

/*
** Streaming version of sqlite3session_patchset().
*/
int sqlite3session_patchset_strm(
  sqlite3_session *pSession,
  int (*xOutput)(void *pOut, const void *pData, int nData),
  void *pOut
){
  return sessionGenerateChangeset(
      pSession, SESSIONS_PATCHSET, xOutput, pOut, 0, 0);
}

/*
** Obtain a patchset object containing all changes recorded by the 
** session object passed as the first argument.
**
** It is the responsibility of the caller to eventually free the buffer 
** using sqlite3_free().
*/
int sqlite3session_patchset(
  sqlite3_session *pSession,      /* Session object */
  int *pnPatchset,                /* OUT: Size of buffer at *ppChangeset */
  void **ppPatchset               /* OUT: Buffer containing changeset */
){
  return sessionGenerateChangeset(
      pSession, SESSIONS_PATCHSET, 0, 0, pnPatchset, ppPatchset);
}

int sqlite3session_fullchangeset(
  sqlite3_session *pSession,      /* Session object */
  int *pnChangeset,               /* OUT: Size of buffer at *ppChangeset */
  void **ppChangeset              /* OUT: Buffer containing changeset */
){
  return sessionGenerateChangeset(
      pSession, SESSIONS_FULLCHANGESET, 0, 0, pnChangeset, ppChangeset);
}


/*
** Enable or disable the session object passed as the first argument.
*/
int sqlite3session_enable(sqlite3_session *pSession, int bEnable){
  int ret;
  sqlite3_mutex_enter(sqlite3_db_mutex(pSession->db));

  SessionTable *pTab;
  assert( xOutput==0 || (ppOut==0 && pnOut==0) );

  /* Create the serialized output changeset based on the contents of the
  ** hash tables attached to the SessionTable objects in list p->pList. 
  */
  for(pTab=pGrp->pList; rc==SQLITE_OK && pTab; pTab=pTab->pNext){
    int eChangeset = pGrp->bPatch ? SESSIONS_PATCHSET : SESSIONS_CHANGESET;
    int i;
    if( pTab->nEntry==0 ) continue;

    sessionAppendTableHdr(&buf, eChangeset, pTab, &rc);
    for(i=0; i<pTab->nChange; i++){
      SessionChange *p;
      for(p=pTab->apChange[i]; p; p=p->pNext){
        sessionAppendByte(&buf, p->op, &rc);
        sessionAppendByte(&buf, p->bIndirect, &rc);
        sessionAppendBlob(&buf, p->aRecord, p->nRecord, &rc);
      }

** the same session object is disabled, no INSERT record will appear in the
** changeset, even though the delete took place while the session was disabled.
** Or, if one field of a row is updated while a session is disabled, and 
** another field of the same row is updated while the session is enabled, the
** resulting changeset will contain an UPDATE change that updates both fields.
*/
int sqlite3session_changeset(
  sqlite3_session *pSession,      /* Session object */
  int *pnChangeset,               /* OUT: Size of buffer at *ppChangeset */
  void **ppChangeset              /* OUT: Buffer containing changeset */
);

/*
** CAPI3REF: Generate A Full Changeset From A Session Object
**
** This function is similar to sqlite3session_changeset(), except that for
** each row affected by an UPDATE statement, all old.* values are recorded
** as part of the changeset, not just those modified.
*/
int sqlite3session_fullchangeset(
  sqlite3_session *pSession,      /* Session object */
  int *pnChangeset,               /* OUT: Size of buffer at *ppChangeset */
  void **ppChangeset              /* OUT: Buffer containing changeset */
);

/*
** CAPI3REF: Load The Difference Between Tables Into A Session 

** Tclcmd:  $session attach TABLE
**          $session changeset
**          $session delete
**          $session enable BOOL
**          $session indirect INTEGER
**          $session patchset
**          $session table_filter SCRIPT
**          $session fullchangeset
*/
static int SQLITE_TCLAPI test_session_cmd(
  void *clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  TestSession *p = (TestSession*)clientData;
  sqlite3_session *pSession = p->pSession;
  struct SessionSubcmd {
    const char *zSub;
    int nArg;
    const char *zMsg;

  } aSub[] = {
    { "attach",       1, "TABLE"       }, /* 0 */
    { "changeset",    0, ""            }, /* 1 */
    { "delete",       0, ""            }, /* 2 */
    { "enable",       1, "BOOL"        }, /* 3 */
    { "indirect",     1, "BOOL"        }, /* 4 */
    { "isempty",      0, ""            }, /* 5 */
    { "table_filter", 1, "SCRIPT"      }, /* 6 */
    { "patchset",     0, "",           }, /* 7 */
    { "diff",         2, "FROMDB TBL"  }, /* 8 */
    { "fullchangeset",0, ""            }, /* 9 */
    { 0 }
  };
  int iSub;
  int rc;

  if( objc<2 ){
    Tcl_WrongNumArgs(interp, 1, objv, "SUBCOMMAND ...");

      rc = sqlite3session_attach(pSession, zArg);
      if( rc!=SQLITE_OK ){
        return test_session_error(interp, rc, 0);
      }
      break;
    }

    case 9:        /* fullchangeset */
    case 7:        /* patchset */
    case 1: {      /* changeset */
      TestSessionsBlob o = {0, 0};
      if( iSub!=9 && test_tcl_integer(interp, SESSION_STREAM_TCL_VAR) ){
        void *pCtx = (void*)&o;
        if( iSub==7 ){
          rc = sqlite3session_patchset_strm(pSession, testStreamOutput, pCtx);
        }else{
          rc = sqlite3session_changeset_strm(pSession, testStreamOutput, pCtx);
        }
      }else{
        if( iSub==7 ){
          rc = sqlite3session_patchset(pSession, &o.n, &o.p);
        }else if( iSub==9 ){
          rc = sqlite3session_fullchangeset(pSession, &o.n, &o.p);
        }else{
          rc = sqlite3session_changeset(pSession, &o.n, &o.p);
        }
      }
      if( rc==SQLITE_OK ){
        Tcl_SetObjResult(interp, Tcl_NewByteArrayObj(o.p, o.n)); 
      }
      sqlite3_free(o.p);
      if( rc!=SQLITE_OK ){
        return test_session_error(interp, rc, 0);
      }
      break;
    }


    case 2:        /* delete */
      Tcl_DeleteCommand(interp, Tcl_GetString(objv[0]));
      break;

    case 3: {      /* enable */
      int val;