SQLite

# 2016 March 18
#
# The author disclaims copyright to this source code.  In place of
# a legal notice, here is a blessing:
#
#    May you do good and not evil.
#    May you find forgiveness for yourself and forgive others.
#    May you share freely, never taking more than you give.
#
#***********************************************************************
#

source [file join [file dirname [info script]] rbu_common.tcl]
set ::testprefix rbuprogress


# Create a simple RBU database. That expects to write to a table:
#
#   CREATE TABLE t1(a INTEGER PRIMARY KEY, b, c);
#
proc create_rbu1 {filename} {
  forcedelete $filename
  sqlite3 rbu1 $filename  
  rbu1 eval {
    CREATE TABLE data_t1(a, b, c, rbu_control);
    INSERT INTO data_t1 VALUES(1, 2, 3, 0);
    INSERT INTO data_t1 VALUES(2, 'two', 'three', 0);
    INSERT INTO data_t1 VALUES(3, NULL, 8.2, 0);

    CREATE TABLE rbu_count(tbl, cnt);
    INSERT INTO rbu_count VALUES('data_t1', 3);
  }
  rbu1 close
  return $filename
}


do_execsql_test 1.0 {
  CREATE TABLE t1(a INTEGER PRIMARY KEY, b, c);
}

do_test 1.1 {
  create_rbu1 rbu.db
  sqlite3rbu rbu test.db rbu.db
  rbu stage_progress
} {0 0}
do_test 1.2 { rbu step ; rbu stage_progress } {3333 0}
do_test 1.3 { rbu step ; rbu stage_progress } {6666 0}
do_test 1.4 { rbu step ; rbu stage_progress } {10000 0}
do_test 1.5 { rbu step ; rbu stage_progress } {10000 0}
do_test 1.6 { rbu step ; rbu stage_progress } {10000 0}
do_test 1.7 { rbu step ; rbu stage_progress } {10000 5000}
do_test 1.8 { rbu step ; rbu stage_progress } {10000 10000}
do_test 1.9 { rbu step ; rbu stage_progress } {10000 10000}

do_test 1.10 {
  rbu close
} {SQLITE_DONE}

#-------------------------------------------------------------------------

proc do_sp_test {tn target rbu reslist} {
  uplevel [list do_test $tn [subst -nocommands {
    sqlite3rbu rbu $target $rbu
    set res [list]
    while 1 {
      set rc [rbu step]
      if {[set rc] != "SQLITE_OK"} { error "error 1" }
      lappend res [lindex [rbu stage_progress] 0]
      if {[lindex [set res] end]==10000} break
    }
    if {[set res] != [list $reslist]} {
      error "reslist is incorrect (expect=$reslist got=[set res])"
    }

    # One step to clean up the temporary tables used to update the only
    # target table in the rbu database. And one more to move the *-oal 
    # file to *-wal.
    rbu step
    rbu step

    # Do the checkpoint.
    while {[rbu step]=="SQLITE_OK"} { }

    rbu close
  }] {SQLITE_DONE}]
}

proc create_db_file {filename sql} {
  forcedelete $filename
  sqlite3 tmpdb $filename  
  tmpdb eval $sql
  tmpdb close
}

reset_db
do_test 2.1.0 {
  execsql {
    CREATE TABLE t1(a INTEGER PRIMARY KEY, b, c);
  }
  create_db_file rbu.db {
    CREATE TABLE data_t1(a, b, c, rbu_control);
    INSERT INTO data_t1 VALUES(4, 4, 4, 0);
    INSERT INTO data_t1 VALUES(5, 5, 5, 0);

    CREATE TABLE rbu_count(tbl, cnt);
    INSERT INTO rbu_count VALUES('data_t1', 2);
  }
} {}
do_sp_test 2.1.1 test.db rbu.db {5000 10000}

reset_db
do_test 2.2.0 {
  execsql { CREATE TABLE t1(a INTEGER PRIMARY KEY, b, c) }
  create_rbu1 rbu.db
} {rbu.db}
do_sp_test 2.2.1 test.db rbu.db {3333 6666 10000}

reset_db
do_test 2.3.0 {
  execsql { 
    CREATE TABLE t1(a INTEGER PRIMARY KEY, b, c);
    CREATE INDEX i1 ON t1(b);
    INSERT INTO t1 VALUES(1, 1, 1);
    INSERT INTO t1 VALUES(2, 2, 2);
    INSERT INTO t1 VALUES(3, 3, 3);
  }
  create_db_file rbu.db {
    CREATE TABLE data_t1(a, b, c, rbu_control);
    INSERT INTO data_t1 VALUES(4, 4, 4, 0);
    INSERT INTO data_t1 VALUES(2, NULL, NULL, 1);
    INSERT INTO data_t1 VALUES(5, NULL, NULL, 1);

    CREATE TABLE rbu_count(tbl, cnt);
    INSERT INTO rbu_count VALUES('data_t1', 3);
  }
} {}
do_sp_test 2.3.1 test.db rbu.db {1666 3333 6000 8000 10000}

reset_db
do_test 2.4.0 {
  execsql { 
    CREATE TABLE t1(a INTEGER PRIMARY KEY, b, c);
    CREATE INDEX i1 ON t1(b);
    INSERT INTO t1 VALUES(1, 1, 1);
    INSERT INTO t1 VALUES(2, 2, 2);
    INSERT INTO t1 VALUES(3, 3, 3);
  }
  create_db_file rbu.db {
    CREATE TABLE data_t1(a, b, c, rbu_control);
    INSERT INTO data_t1 VALUES(2, 4, 4, '.xx');

    CREATE TABLE rbu_count(tbl, cnt);
    INSERT INTO rbu_count VALUES('data_t1', 1);
  }
} {}
do_sp_test 2.4.1 test.db rbu.db {3333 6666 10000}

reset_db
do_test 2.5.0 {
  execsql { 
    CREATE TABLE t1(a INTEGER PRIMARY KEY, b, c);
    CREATE INDEX i1 ON t1(b);
    INSERT INTO t1 VALUES(1, 1, 1);
    INSERT INTO t1 VALUES(2, 2, 2);
    INSERT INTO t1 VALUES(3, 3, 3);
  }
  create_db_file rbu.db {
    CREATE TABLE data_t1(a, b, c, rbu_control);
    INSERT INTO data_t1 VALUES(4, NULL, 4, '.xx');

    CREATE TABLE rbu_count(tbl, cnt);
    INSERT INTO rbu_count VALUES('data_t1', 1);
  }
} {}
do_sp_test 2.5.1 test.db rbu.db {10000}

finish_test

** RBU_STATE_COOKIE:
**   Valid if STAGE==1. The current change-counter cookie value in the 
**   target db file.
**
** RBU_STATE_OALSZ:
**   Valid if STAGE==1. The size in bytes of the *-oal file.
*/
#define RBU_STATE_STAGE        1
#define RBU_STATE_TBL          2
#define RBU_STATE_IDX          3
#define RBU_STATE_ROW          4
#define RBU_STATE_PROGRESS     5
#define RBU_STATE_CKPT         6
#define RBU_STATE_COOKIE       7
#define RBU_STATE_OALSZ        8
#define RBU_STATE_PHASEONESTEP 9

#define RBU_STAGE_OAL         1
#define RBU_STAGE_MOVE        2
#define RBU_STAGE_CAPTURE     3
#define RBU_STAGE_CKPT        4
#define RBU_STAGE_DONE        5

  char *zTbl;
  char *zIdx;
  i64 iWalCksum;
  int nRow;
  i64 nProgress;
  u32 iCookie;
  i64 iOalSz;
  i64 nPhaseOneStep;
};

struct RbuUpdateStmt {
  char *zMask;                    /* Copy of update mask used with pUpdate */
  sqlite3_stmt *pUpdate;          /* Last update statement (or NULL) */
  RbuUpdateStmt *pNext;
};

  int bCleanup;                   /* True in "cleanup" state */
  const char *zTbl;               /* Name of target db table */
  const char *zDataTbl;           /* Name of rbu db table (or null) */
  const char *zIdx;               /* Name of target db index (or null) */
  int iTnum;                      /* Root page of current object */
  int iPkTnum;                    /* If eType==EXTERNAL, root of PK index */
  int bUnique;                    /* Current index is unique */
  int nIndex;                     /* Number of aux. indexes on table zTbl */

  /* Statements created by rbuObjIterPrepareAll() */
  int nCol;                       /* Number of columns in current object */
  sqlite3_stmt *pSelect;          /* Source data */
  sqlite3_stmt *pInsert;          /* Statement for INSERT operations */
  sqlite3_stmt *pDelete;          /* Statement for DELETE ops */
  sqlite3_stmt *pTmpInsert;       /* Insert into rbu_tmp_$zDataTbl */

  char *zErrmsg;                  /* Error message if rc!=SQLITE_OK */
  int nStep;                      /* Rows processed for current object */
  int nProgress;                  /* Rows processed for all objects */
  RbuObjIter objiter;             /* Iterator for skipping through tbl/idx */
  const char *zVfsName;           /* Name of automatically created rbu vfs */
  rbu_file *pTargetFd;            /* File handle open on target db */
  i64 iOalSz;
  i64 nPhaseOneStep;

  /* The following state variables are used as part of the incremental
  ** checkpoint stage (eStage==RBU_STAGE_CKPT). See comments surrounding
  ** function rbuSetupCheckpoint() for details.  */
  u32 iMaxFrame;                  /* Largest iWalFrame value in aFrame[] */
  u32 mLock;
  int nFrame;                     /* Entries in aFrame[] array */

  if( p->rc==SQLITE_OK ){
    memcpy(pIter->abIndexed, pIter->abTblPk, sizeof(u8)*pIter->nTblCol);
    p->rc = prepareFreeAndCollectError(p->dbMain, &pList, &p->zErrmsg,
        sqlite3_mprintf("PRAGMA main.index_list = %Q", pIter->zTbl)
    );
  }

  pIter->nIndex = 0;
  while( p->rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pList) ){
    const char *zIdx = (const char*)sqlite3_column_text(pList, 1);
    sqlite3_stmt *pXInfo = 0;
    if( zIdx==0 ) break;
    p->rc = prepareFreeAndCollectError(p->dbMain, &pXInfo, &p->zErrmsg,
        sqlite3_mprintf("PRAGMA main.index_xinfo = %Q", zIdx)
    );
    while( p->rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pXInfo) ){
      int iCid = sqlite3_column_int(pXInfo, 1);
      if( iCid>=0 ) pIter->abIndexed[iCid] = 1;
    }
    rbuFinalize(p, pXInfo);
    bIndex = 1;
    pIter->nIndex++;
  }

  rbuFinalize(p, pList);
  if( bIndex==0 ) pIter->abIndexed = 0;
}

  sqlite3_context *pCtx, 
  int nVal,
  sqlite3_value **apVal
){
  sqlite3rbu *p = sqlite3_user_data(pCtx);
  int rc = SQLITE_OK;
  int i;

  assert( sqlite3_value_int(apVal[0])!=0
      || p->objiter.eType==RBU_PK_EXTERNAL 
      || p->objiter.eType==RBU_PK_NONE 
  );
  if( sqlite3_value_int(apVal[0])!=0 ){
    p->nPhaseOneStep += p->objiter.nIndex;
  }

  for(i=0; rc==SQLITE_OK && i<nVal; i++){
    rc = sqlite3_bind_value(p->objiter.pTmpInsert, i+1, apVal[i]);
  }
  if( rc==SQLITE_OK ){
    sqlite3_step(p->objiter.pTmpInsert);
    rc = sqlite3_reset(p->objiter.pTmpInsert);

  RbuObjIter *pIter = &p->objiter;
  sqlite3_value *pVal;
  sqlite3_stmt *pWriter;
  int i;

  assert( p->rc==SQLITE_OK );
  assert( eType!=RBU_DELETE || pIter->zIdx==0 );
  assert( eType==RBU_DELETE || eType==RBU_IDX_DELETE
       || eType==RBU_INSERT || eType==RBU_IDX_INSERT
  );

  /* If this is a delete, decrement nPhaseOneStep by nIndex. If the DELETE
  ** statement below does actually delete a row, nPhaseOneStep will be
  ** incremented by the same amount when SQL function rbu_tmp_insert()
  ** is invoked by the trigger.  */
  if( eType==RBU_DELETE ){
    p->nPhaseOneStep -= p->objiter.nIndex;
  }

  if( eType==RBU_IDX_DELETE || eType==RBU_DELETE ){
    pWriter = pIter->pDelete;
  }else{
    pWriter = pIter->pInsert;
  }

    );
    assert( eType!=RBU_UPDATE || pIter->zIdx==0 );

    if( pIter->zIdx==0 && eType==RBU_IDX_DELETE ){
      rbuBadControlError(p);
    }
    else if( eType==RBU_REPLACE ){
      if( pIter->zIdx==0 ){
        p->nPhaseOneStep += p->objiter.nIndex;
        rbuStepOneOp(p, RBU_DELETE);
      }
      if( p->rc==SQLITE_OK ) rbuStepOneOp(p, RBU_INSERT);
    }
    else if( eType!=RBU_UPDATE ){
      rbuStepOneOp(p, eType);
    }
    else{
      sqlite3_value *pVal;
      sqlite3_stmt *pUpdate = 0;
      assert( eType==RBU_UPDATE );
      p->nPhaseOneStep -= p->objiter.nIndex;
      rbuGetUpdateStmt(p, pIter, zMask, &pUpdate);
      if( pUpdate ){
        int i;
        for(i=0; p->rc==SQLITE_OK && i<pIter->nCol; i++){
          char c = zMask[pIter->aiSrcOrder[i]];
          pVal = sqlite3_column_value(pIter->pSelect, i);
          if( pIter->abTblPk[i] || c!='.' ){

          "(%d, %d), "
          "(%d, %Q), "
          "(%d, %Q), "
          "(%d, %d), "
          "(%d, %d), "
          "(%d, %lld), "
          "(%d, %lld), "
          "(%d, %lld), "
          "(%d, %lld) ",
          p->zStateDb,
          RBU_STATE_STAGE, eStage,
          RBU_STATE_TBL, p->objiter.zTbl, 
          RBU_STATE_IDX, p->objiter.zIdx, 
          RBU_STATE_ROW, p->nStep, 
          RBU_STATE_PROGRESS, p->nProgress,
          RBU_STATE_CKPT, p->iWalCksum,
          RBU_STATE_COOKIE, (i64)p->pTargetFd->iCookie,
          RBU_STATE_OALSZ, p->iOalSz,
          RBU_STATE_PHASEONESTEP, p->nPhaseOneStep
      )
    );
    assert( pInsert==0 || rc==SQLITE_OK );

    if( rc==SQLITE_OK ){
      sqlite3_step(pInsert);
      rc = sqlite3_finalize(pInsert);

        pRet->iCookie = (u32)sqlite3_column_int64(pStmt, 1);
        break;

      case RBU_STATE_OALSZ:
        pRet->iOalSz = (u32)sqlite3_column_int64(pStmt, 1);
        break;

      case RBU_STATE_PHASEONESTEP:
        pRet->nPhaseOneStep = (u32)sqlite3_column_int64(pStmt, 1);
        break;

      default:
        rc = SQLITE_CORRUPT;
        break;
    }
  }
  rc2 = sqlite3_finalize(pStmt);
  if( rc==SQLITE_OK ) rc = rc2;

*/
static void rbuDeleteVfs(sqlite3rbu *p){
  if( p->zVfsName ){
    sqlite3rbu_destroy_vfs(p->zVfsName);
    p->zVfsName = 0;
  }
}

/*
**
*/
static void rbuIndexCntFunc(
  sqlite3_context *pCtx, 
  int nVal,
  sqlite3_value **apVal
){
  sqlite3rbu *p = (sqlite3rbu*)sqlite3_user_data(pCtx);
  sqlite3_stmt *pStmt = 0;
  char *zErrmsg = 0;
  int rc;

  assert( nVal==1 );
  
  rc = prepareFreeAndCollectError(p->dbMain, &pStmt, &zErrmsg, 
      sqlite3_mprintf("PRAGMA index_list = %Q", sqlite3_value_text(apVal[0]))
  );
  if( rc!=SQLITE_OK ){
    sqlite3_result_error(pCtx, zErrmsg, -1);
  }else{
    int nIndex = 0;
    while( SQLITE_ROW==sqlite3_step(pStmt) ){
      nIndex++;
    }
    rc = sqlite3_finalize(pStmt);
    if( rc==SQLITE_OK ){
      sqlite3_result_int(pCtx, nIndex);
    }else{
      sqlite3_result_error(pCtx, sqlite3_errmsg(p->dbMain), -1);
    }
  }

  sqlite3_free(zErrmsg);
}

/*
** If the RBU database contains the rbu_count table, use it to initialize
** the sqlite3rbu.nPhaseOneStep variable. The schema of the rbu_count table
** is assumed to contain the same columns as:
**
**   CREATE TABLE rbu_count(tbl TEXT PRIMARY KEY, cnt INTEGER) WITHOUT ROWID;
**
** There should be one row in the table for each data_xxx table in the
** database. The 'tbl' column should contain the name of a data_xxx table,
** and the cnt column the number of rows it contains.
**
** sqlite3rbu.nPhaseOneStep is initialized to the sum of (1 + nIndex) * cnt
** for all rows in the rbu_count table, where nIndex is the number of 
** indexes on the corresponding target database table.
*/
static void rbuInitPhaseOneSteps(sqlite3rbu *p){
  if( p->rc==SQLITE_OK ){
    sqlite3_stmt *pStmt = 0;
    int bExists = 0;                /* True if rbu_count exists */

    p->nPhaseOneStep = -1;

    p->rc = sqlite3_create_function(p->dbRbu, 
        "rbu_index_cnt", 1, SQLITE_UTF8, (void*)p, rbuIndexCntFunc, 0, 0
    );
  
    /* Check for the rbu_count table. If it does not exist, or if an error
    ** occurs, nPhaseOneStep will be left set to -1. */
    if( p->rc==SQLITE_OK ){
      p->rc = prepareAndCollectError(p->dbRbu, &pStmt, &p->zErrmsg,
          "SELECT 1 FROM sqlite_master WHERE tbl_name = 'rbu_count'"
      );
    }
    if( p->rc==SQLITE_OK ){
      if( SQLITE_ROW==sqlite3_step(pStmt) ){
        bExists = 1;
      }
      p->rc = sqlite3_finalize(pStmt);
    }
  
    if( p->rc==SQLITE_OK && bExists ){
      p->rc = prepareAndCollectError(p->dbRbu, &pStmt, &p->zErrmsg,
          "SELECT sum(cnt * (1 + rbu_index_cnt(rbu_target_name(tbl))))"
          "FROM rbu_count"
      );
      if( p->rc==SQLITE_OK ){
        if( SQLITE_ROW==sqlite3_step(pStmt) ){
          p->nPhaseOneStep = sqlite3_column_int64(pStmt, 0);
        }
        p->rc = sqlite3_finalize(pStmt);
      }
    }
  }
}

/*
** Open and return a new RBU handle. 
*/
sqlite3rbu *sqlite3rbu_open(
  const char *zTarget, 
  const char *zRbu,

    if( p->rc==SQLITE_OK ){
      pState = rbuLoadState(p);
      assert( pState || p->rc!=SQLITE_OK );
      if( p->rc==SQLITE_OK ){

        if( pState->eStage==0 ){ 
          rbuDeleteOalFile(p);
          rbuInitPhaseOneSteps(p);
          p->eStage = RBU_STAGE_OAL;
        }else{
          p->eStage = pState->eStage;
        }
        p->nProgress = pState->nProgress;
        p->iOalSz = pState->iOalSz;
      }

** Return the total number of key-value operations (inserts, deletes or 
** updates) that have been performed on the target database since the
** current RBU update was started.
*/
sqlite3_int64 sqlite3rbu_progress(sqlite3rbu *pRbu){
  return pRbu->nProgress;
}

void sqlite3rbu_stage_progress(sqlite3rbu *p, int *pnOne, int *pnTwo){
  const int MAX_PROGRESS = 10000;
  switch( p->eStage ){
    case RBU_STAGE_OAL:
      if( p->nPhaseOneStep>0 ){
        *pnOne = (int)(MAX_PROGRESS * (i64)p->nProgress/(i64)p->nPhaseOneStep);
      }else{
        *pnOne = -1;
      }
      *pnTwo = 0;
      break;

    case RBU_STAGE_MOVE:
      *pnOne = MAX_PROGRESS;
      *pnTwo = 0;
      break;

    case RBU_STAGE_CKPT:
      *pnOne = MAX_PROGRESS;
      *pnTwo = (int)(MAX_PROGRESS * (i64)p->nStep / (i64)p->nFrame);
      break;

    case RBU_STAGE_DONE:
      *pnOne = MAX_PROGRESS;
      *pnTwo = MAX_PROGRESS;
      break;

    default:
      assert( 0 );
  }
}

int sqlite3rbu_savestate(sqlite3rbu *p){
  int rc = p->rc;
  
  if( rc==SQLITE_DONE ) return SQLITE_OK;

  assert( p->eStage>=RBU_STAGE_OAL && p->eStage<=RBU_STAGE_DONE );

/*
** Return the total number of key-value operations (inserts, deletes or 
** updates) that have been performed on the target database since the
** current RBU update was started.
*/
sqlite3_int64 sqlite3rbu_progress(sqlite3rbu *pRbu);

void sqlite3rbu_stage_progress(sqlite3rbu *pRbu, int *pnOne, int *pnTwo);

/*
** Create an RBU VFS named zName that accesses the underlying file-system
** via existing VFS zParent. Or, if the zParent parameter is passed NULL, 
** then the new RBU VFS uses the default system VFS to access the file-system.
** The new object is registered as a non-default VFS with SQLite before 
** returning.

    const char *zUsage;
  } aCmd[] = {
    {"step", 2, ""},              /* 0 */
    {"close", 2, ""},             /* 1 */
    {"create_rbu_delta", 2, ""},  /* 2 */
    {"savestate", 2, ""},         /* 3 */
    {"dbMain_eval", 3, "SQL"},    /* 4 */
    {"stage_progress", 2, ""},    /* 5 */
    {0,0,0}
  };
  int iCmd;

  if( objc<2 ){
    Tcl_WrongNumArgs(interp, 1, objv, "METHOD");
    return TCL_ERROR;

      int rc = sqlite3_exec(db, Tcl_GetString(objv[2]), 0, 0, 0);
      if( rc!=SQLITE_OK ){
        Tcl_SetObjResult(interp, Tcl_NewStringObj(sqlite3_errmsg(db), -1));
        ret = TCL_ERROR;
      }
      break;
    }

    case 5: /* stage_progress */ {
      int one, two;
      Tcl_Obj *pObj;
      sqlite3rbu_stage_progress(pRbu, &one, &two);

      pObj = Tcl_NewObj();
      Tcl_ListObjAppendElement(interp, pObj, Tcl_NewIntObj(one));
      Tcl_ListObjAppendElement(interp, pObj, Tcl_NewIntObj(two));
      Tcl_SetObjResult(interp, pObj);
      break;
    }

    default: /* seems unlikely */
      assert( !"cannot happen" );
      break;
  }

  return ret;