SQLite

#include <string.h>

/*
** Print a usage message and exit.
*/
void usage(const char *zArgv0){
  fprintf(stderr, 
"Usage: %s [-step NSTEP] TARGET-DB RBU-DB\n"
"Usage: %s ?OPTIONS? TARGET-DB RBU-DB\n"
"\n"
"Where options are:\n"
"\n"
"    -step NSTEP\n"
"    -vacuum\n"
"\n"
"  If the -vacuum switch is not present, argument RBU-DB must be an RBU\n"
"  Argument RBU-DB must be an RBU database containing an update suitable for\n"
"  target database TARGET-DB. If NSTEP is set to less than or equal to zero\n"
"  (the default value), this program attempts to apply the entire update to\n"
"  the target database.\n"
"  database containing an update suitable for target database TARGET-DB.\n"
"  Or, if -vacuum is specified, then TARGET-DB is a database to vacuum using\n"
"  RBU, and RBU-DB is used as the state database for the vacuum (refer to\n"
"  API documentation for details).\n"
"\n"
"  If NSTEP is set to less than or equal to zero (the default value), this \n"
"  program attempts to perform the entire update or vacuum operation before\n"
"  exiting\n"
"\n"
"  If NSTEP is greater than zero, then a maximum of NSTEP calls are made\n"
"  to sqlite3rbu_step(). If the RBU update has not been completely applied\n"
"  after the NSTEP'th call is made, the state is saved in the database RBU-DB\n"
"  and the program exits. Subsequent invocations of this (or any other RBU)\n"
"  application will use this state to resume applying the RBU update to the\n"
"  target db.\n"

  int i;
  const char *zTarget;            /* Target database to apply RBU to */
  const char *zRbu;               /* Database containing RBU */
  char zBuf[200];                 /* Buffer for printf() */
  char *zErrmsg;                  /* Error message, if any */
  sqlite3rbu *pRbu;               /* RBU handle */
  int nStep = 0;                  /* Maximum number of step() calls */
  int bVacuum = 0;
  int rc;
  sqlite3_int64 nProgress = 0;
  int nArg = argc-2;

  /* Process command line arguments. Following this block local variables 
  ** zTarget, zRbu and nStep are all set. */
  if( argc==5 ){
    size_t nArg1 = strlen(argv[1]);
    if( nArg1>5 || nArg1<2 || memcmp("-step", argv[1], nArg1) ) usage(argv[0]);
    nStep = atoi(argv[2]);
  }else if( argc!=3 ){
    usage(argv[0]);
  }
  if( argc<3 ) usage(argv[0]);
  for(i=1; i<nArg; i++){
    const char *zArg = argv[i];
    int nArg = strlen(zArg);
    if( nArg>1 && nArg<=8 && 0==memcmp(zArg, "-vacuum", nArg) ){
      bVacuum = 1;
    }else if( nArg>1 && nArg<=5 && 0==memcmp(zArg, "-step", nArg) && i<nArg-1 ){
      i++;
      nStep = atoi(argv[i]);
    }else{
      usage(argv[0]);
    }
  }

  zTarget = argv[argc-2];
  zRbu = argv[argc-1];

  report_default_vfs();

  /* Open an RBU handle. A vacuum handle if -vacuum was specified, or a
  ** regular RBU update handle otherwise.  */
  if( bVacuum ){
    pRbu = sqlite3rbu_vacuum(zTarget, zRbu);
  }else{
    pRbu = sqlite3rbu_open(zTarget, zRbu, 0);
  }
  report_rbu_vfs(pRbu);

  /* Open an RBU handle. If nStep is less than or equal to zero, call
  /* If nStep is less than or equal to zero, call
  ** sqlite3rbu_step() until either the RBU has been completely applied
  ** or an error occurs. Or, if nStep is greater than zero, call
  ** sqlite3rbu_step() a maximum of nStep times.  */
  pRbu = sqlite3rbu_open(zTarget, zRbu, 0);
  report_rbu_vfs(pRbu);
  for(i=0; (nStep<=0 || i<nStep) && sqlite3rbu_step(pRbu)==SQLITE_OK; i++);
  nProgress = sqlite3rbu_progress(pRbu);
  rc = sqlite3rbu_close(pRbu, &zErrmsg);

  /* Let the user know what happened. */
  switch( rc ){
    case SQLITE_OK:

# 2016 April 20
#
# 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.
#
#***********************************************************************
#
# This file contains fault injection tests for RBU vacuum operations.
#

source [file join [file dirname [info script]] rbu_common.tcl]
source $testdir/malloc_common.tcl
set ::testprefix rbufault3

foreach {fault errlist} {
  oom-* {
    {1 SQLITE_NOMEM}
    {1 SQLITE_IOERR_NOMEM}
    {1 {SQLITE_NOMEM - out of memory}}
  }

  ioerr-* {
    {1 {SQLITE_IOERR - disk I/O error}}
    {1 SQLITE_IOERR} 
    {1 SQLITE_IOERR_WRITE} 
    {1 SQLITE_IOERR_FSYNC} 
    {1 SQLITE_IOERR_READ} 
    {1 {SQLITE_IOERR - unable to open database: test.db2}} 
    {1 {SQLITE_ERROR - unable to open database: test.db2}} 
    {1 {SQLITE_ERROR - SQL logic error or missing database}}
  }

  cantopen* {
    {1 {SQLITE_CANTOPEN - unable to open database: test.db2}}  
    {1 {SQLITE_CANTOPEN - unable to open database: test.db2}}  
    {1 {SQLITE_CANTOPEN - unable to open database file}}  
    {1 SQLITE_CANTOPEN} 
  }

} {

  reset_db
  do_execsql_test 0 {
    CREATE TABLE target(x UNIQUE, y, z, PRIMARY KEY(y));
    INSERT INTO target VALUES(1, 2, 3);
    INSERT INTO target VALUES(4, 5, 6);
    INSERT INTO target VALUES(7, 8, 9);
    CREATE INDEX i1 ON target(z);
  }
  faultsim_save_and_close

  do_faultsim_test 1 -faults $fault -prep {
    faultsim_restore_and_reopen
    forcedelete test.db2
  } -body {
    sqlite3rbu_vacuum rbu test.db test.db2
    while {[rbu step]=="SQLITE_OK"} {}
    rbu close
  } -test {
    eval [list faultsim_test_result {0 SQLITE_DONE} {*}$::errlist]
  }

  do_faultsim_test 2 -faults $fault -prep {
    faultsim_restore_and_reopen
    forcedelete test.db2
  } -body {
    sqlite3rbu_vacuum rbu test.db test.db2
    rbu step
    rbu close
  } -test {
    eval [list faultsim_test_result {0 SQLITE_OK} {*}$::errlist]
  }

  forcedelete test.db2
  sqlite3rbu_vacuum rbu test.db test.db2
  rbu step
  rbu close
  faultsim_save_and_close

  do_faultsim_test 3 -faults $fault -prep {
    faultsim_restore_and_reopen
    forcedelete test.db2
  } -body {
    sqlite3rbu_vacuum rbu test.db test.db2
    rbu step
    rbu close
  } -test {
    eval [list faultsim_test_result {0 SQLITE_OK} {*}$::errlist]
  }

}

finish_test

# 2016 April 15
#
# 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.
#
#***********************************************************************
#
# This file contains tests for the RBU module. More specifically, it
# contains tests to ensure that the sqlite3rbu_vacuum() API works as
# expected.
#

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

proc do_rbu_vacuum_test {tn step} {
  uplevel [list do_test $tn.1 {
    if {$step==0} { sqlite3rbu_vacuum rbu test.db state.db }
    while 1 {
      if {$step==1} { sqlite3rbu_vacuum rbu test.db state.db }
      set rc [rbu step]
      if {$rc!="SQLITE_OK"} break
      if {$step==1} { rbu close }
    }
    rbu close
  } {SQLITE_DONE}]

  uplevel [list do_execsql_test $tn.2 {
    PRAGMA integrity_check
  } ok]
}

foreach step {0 1} {

  set ::testprefix rbuvacuum-step=$step
  reset_db

  # Simplest possible vacuum.
  do_execsql_test 1.0 {
    PRAGMA page_size = 1024;
    CREATE TABLE t1(a INTEGER PRIMARY KEY, b, c);
    INSERT INTO t1 VALUES(1, 2, 3);
    INSERT INTO t1 VALUES(4, 5, 6);
    INSERT INTO t1 VALUES(7, 8, 9);
    PRAGMA integrity_check;
  } {ok}
  do_rbu_vacuum_test 1.1 $step

  # A vacuum that actually reclaims space.
  do_execsql_test 1.2.1 {
    INSERT INTO t1 VALUES(8, randomblob(900), randomblob(900));
    INSERT INTO t1 VALUES(9, randomblob(900), randomblob(900));
    INSERT INTO t1 VALUES(10, randomblob(900), randomblob(900));
    INSERT INTO t1 VALUES(11, randomblob(900), randomblob(900));
    INSERT INTO t1 VALUES(12, randomblob(900), randomblob(900));
    PRAGMA page_count;
  } {12}
  do_execsql_test 1.2.2 {
    DELETE FROM t1 WHERE rowid BETWEEN 8 AND 11;
    PRAGMA page_count;
  } {12}
  do_rbu_vacuum_test 1.2.3 $step
  do_execsql_test 1.2.4 {
    PRAGMA page_count;
  } {3}
  
  # Add an index to the table.
  do_execsql_test 1.3.1 {
    CREATE INDEX t1b ON t1(b);
    INSERT INTO t1 VALUES(13, randomblob(900), randomblob(900));
    INSERT INTO t1 VALUES(14, randomblob(900), randomblob(900));
    INSERT INTO t1 VALUES(15, randomblob(900), randomblob(900));
    INSERT INTO t1 VALUES(16, randomblob(900), randomblob(900));
    PRAGMA page_count;
  } {18}
  do_execsql_test 1.3.2 {
    DELETE FROM t1 WHERE rowid BETWEEN 12 AND 15;
    PRAGMA page_count;
  } {18}
  do_rbu_vacuum_test 1.3.3 $step
  do_execsql_test 1.3.4 {
    PRAGMA page_count;
  } {5}

  # WITHOUT ROWID table.
  do_execsql_test 1.4.1 {
    CREATE TABLE t2(a, b, c, PRIMARY KEY(a, b)) WITHOUT ROWID;

    INSERT INTO t2 VALUES(randomblob(900), 1, randomblob(900));
    INSERT INTO t2 VALUES(randomblob(900), 2, randomblob(900));
    INSERT INTO t2 VALUES(randomblob(900), 3, randomblob(900));
    INSERT INTO t2 VALUES(randomblob(900), 4, randomblob(900));
    INSERT INTO t2 VALUES(randomblob(900), 6, randomblob(900));
    INSERT INTO t2 VALUES(randomblob(900), 7, randomblob(900));
    INSERT INTO t2 VALUES(randomblob(900), 8, randomblob(900));

    DELETE FROM t2 WHERE b BETWEEN 2 AND 7;
    PRAGMA page_count;
  } {20}
  do_rbu_vacuum_test 1.4.2 $step
  do_execsql_test 1.4.3 {
    PRAGMA page_count;
  } {10}
  
  # WITHOUT ROWID table with an index.
  do_execsql_test 1.4.1 {
    CREATE INDEX t2c ON t2(c);

    INSERT INTO t2 VALUES(randomblob(900), 9, randomblob(900));
    INSERT INTO t2 VALUES(randomblob(900), 10, randomblob(900));
    INSERT INTO t2 VALUES(randomblob(900), 11, randomblob(900));
    INSERT INTO t2 VALUES(randomblob(900), 12, randomblob(900));
    INSERT INTO t2 VALUES(randomblob(900), 13, randomblob(900));

    DELETE FROM t2 WHERE b BETWEEN 8 AND 12;
    PRAGMA page_count;
  } {35}
  do_rbu_vacuum_test 1.4.2 $step
  do_execsql_test 1.4.3 {
    PRAGMA page_count;
  } {15}
  do_execsql_test 1.4.4 {
    VACUUM;
    PRAGMA page_count;
  } {15}

  do_execsql_test 1.5.1 {
    CREATE TABLE t3(a, b, c);
    INSERT INTO t3 VALUES('a', 'b', 'c');
    INSERT INTO t3 VALUES('d', 'e', 'f');
    INSERT INTO t3 VALUES('g', 'h', 'i');
  }
  do_rbu_vacuum_test 1.5.2 $step
  do_execsql_test 1.5.3 {
    SELECT * FROM t3
  } {a b c d e f g h i}
  do_execsql_test 1.5.4 {
    CREATE INDEX t3a ON t3(a);
    CREATE INDEX t3b ON t3(b);
    CREATE INDEX t3c ON t3(c);
    INSERT INTO t3 VALUES('j', 'k', 'l');
    DELETE FROM t3 WHERE a = 'g';
  }
  do_rbu_vacuum_test 1.5.5 $step
  do_execsql_test 1.5.6 {
    SELECT rowid, * FROM t3 ORDER BY b
  } {1 a b c 2 d e f 4 j k l}

  do_execsql_test 1.6.1 {
    CREATE TABLE t4(a PRIMARY KEY, b, c);
    INSERT INTO t4 VALUES('a', 'b', 'c');
    INSERT INTO t4 VALUES('d', 'e', 'f');
    INSERT INTO t4 VALUES('g', 'h', 'i');
  }
  do_rbu_vacuum_test 1.6.2 $step
  do_execsql_test 1.6.3 {
    SELECT * FROM t4
  } {a b c d e f g h i}
  do_execsql_test 1.6.4 {
    CREATE INDEX t4a ON t4(a);
    CREATE INDEX t4b ON t4(b);
    CREATE INDEX t4c ON t4(c);
    
    INSERT INTO t4 VALUES('j', 'k', 'l');
    DELETE FROM t4 WHERE a='g';
  }
  do_rbu_vacuum_test 1.6.5 $step
  do_execsql_test 1.6.6 {
    SELECT * FROM t4 ORDER BY b
  } {a b c d e f j k l}

  reset_db
  do_execsql_test 1.7.0 {
    CREATE TABLE t1(a INTEGER PRIMARY KEY AUTOINCREMENT, b);
    INSERT INTO t1 VALUES(NULL, 'one');
    INSERT INTO t1 VALUES(NULL, 'two');
    DELETE FROM t1 WHERE a=2;
    INSERT INTO t1 VALUES(NULL, 'three');
    INSERT INTO t1 VALUES(NULL, 'four');
    DELETE FROM t1 WHERE a=4;
    INSERT INTO t1 VALUES(NULL, 'five');
    INSERT INTO t1 VALUES(NULL, 'six');
    DELETE FROM t1 WHERE a=6;
    SELECT * FROM t1;
  } {1 one 3 three 5 five}
  do_rbu_vacuum_test 1.7.1 $step
  do_execsql_test 1.7.2 {
    INSERT INTO t1 VALUES(NULL, 'seven');
    SELECT * FROM t1;
  } {1 one 3 three 5 five 7 seven}

  reset_db
  do_execsql_test 1.8.0 {
    CREATE TABLE t1(a INTEGER PRIMARY KEY AUTOINCREMENT, b);
    CREATE INDEX i1 ON t1(b);
    INSERT INTO t1 VALUES(NULL, 'one');
    INSERT INTO t1 VALUES(NULL, 'two');
    INSERT INTO t1 VALUES(NULL, 'three');
    INSERT INTO t1 VALUES(NULL, 'four');
    INSERT INTO t1 VALUES(NULL, 'five');
    INSERT INTO t1 VALUES(NULL, 'six');
    ANALYZE;
    SELECT * FROM sqlite_stat1;
  } {t1 i1 {6 1}}
  do_rbu_vacuum_test 1.8.1 $step
  do_execsql_test 1.7.2 {
    SELECT * FROM sqlite_stat1;
  } {t1 i1 {6 1}}

  reset_db
  do_execsql_test 1.9.0 {
    PRAGMA page_size = 8192;
    PRAGMA auto_vacuum = 2;
    PRAGMA user_version = 412;
    PRAGMA application_id = 413;

    CREATE TABLE t1(a INTEGER PRIMARY KEY AUTOINCREMENT, b);
    CREATE INDEX i1 ON t1(b);
    INSERT INTO t1 VALUES(NULL, 'one');
    INSERT INTO t1 VALUES(NULL, 'two');
    INSERT INTO t1 VALUES(NULL, 'three');
    INSERT INTO t1 VALUES(NULL, 'four');
    INSERT INTO t1 VALUES(NULL, 'five');
    INSERT INTO t1 VALUES(NULL, 'six');

    PRAGMA main.page_size;
    PRAGMA main.auto_vacuum;
    PRAGMA main.user_version;
    PRAGMA main.application_id;
  } {8192 2 412 413}

  do_rbu_vacuum_test 1.9.1 $step
  do_execsql_test 1.9.2 {
    PRAGMA main.page_size;
    PRAGMA main.auto_vacuum;
    PRAGMA main.user_version;
    PRAGMA main.application_id;
  } {8192 2 412 413}

  # Vacuum a database with a large sqlite_master table.
  #
  reset_db
  do_test 1.10.1 {
    for {set i 1} {$i < 50} {incr i} {
      execsql "PRAGMA page_size = 1024"
      execsql "CREATE TABLE t$i (a, b, c, PRIMARY KEY(a, b));"
      execsql "
        INSERT INTO t$i VALUES(1, 2, 3);
        INSERT INTO t$i VALUES(4, 5, 6);
      "
    }
  } {}
  do_rbu_vacuum_test 1.10.2 $step
}

#-------------------------------------------------------------------------
# Test some error cases:
#
#   2.1.* the db being vacuumed being in wal mode already.
#   2.2.* database modified mid vacuum.
#
reset_db
do_execsql_test 2.1.0 {
  CREATE TABLE t1(a, b);
  INSERT INTO t1 VALUES(1, 2);
  INSERT INTO t1 VALUES(3, 4);
  INSERT INTO t1 VALUES(5, 6);
  INSERT INTO t1 VALUES(7, 8);
  PRAGMA journal_mode = wal;
  INSERT INTO t1 VALUES(9, 10);
} wal
do_test 2.1.1 {
  sqlite3rbu_vacuum rbu test.db state.db
  rbu step
} {SQLITE_ERROR}
do_test 2.1.2 {
  list [catch { rbu close } msg] $msg
} {1 {SQLITE_ERROR - cannot vacuum wal mode database}}

reset_db
do_execsql_test 2.2.0 {
  CREATE TABLE tx(a PRIMARY KEY, b BLOB);
  INSERT INTO tx VALUES(1, randomblob(900));
  INSERT INTO tx SELECT a+1, randomblob(900) FROM tx;
  INSERT INTO tx SELECT a+2, randomblob(900) FROM tx;
  INSERT INTO tx SELECT a+4, randomblob(900) FROM tx;
  INSERT INTO tx SELECT a+8, randomblob(900) FROM tx;
}
db_save_and_close
for {set i 1} 1 {incr i} {
  db_restore_and_reopen

  sqlite3rbu_vacuum rbu test.db state.db
  for {set step 0} {$step<$i} {incr step} { rbu step }
  rbu close
  if {[file exists test.db-wal]} break

  execsql { INSERT INTO tx VALUES(20, 20) }

  do_test 2.2.$i.1 {
    sqlite3rbu_vacuum rbu test.db state.db 
    rbu step
  } {SQLITE_BUSY}
  do_test 2.2.$i.2 {
    list [catch { rbu close } msg] $msg
  } {1 {SQLITE_BUSY - database modified during rbu vacuum}}
}

catch { db close }
finish_test

typedef struct RbuState RbuState;
typedef struct rbu_vfs rbu_vfs;
typedef struct rbu_file rbu_file;
typedef struct RbuUpdateStmt RbuUpdateStmt;

#if !defined(SQLITE_AMALGAMATION)
typedef unsigned int u32;
typedef unsigned short u16;
typedef unsigned char u8;
typedef sqlite3_int64 i64;
#endif

/*
** These values must match the values defined in wal.c for the equivalent
** locks. These are not magic numbers as they are part of the SQLite file
** format.
*/
#define WAL_LOCK_WRITE  0
#define WAL_LOCK_CKPT   1
#define WAL_LOCK_READ0  3

#define SQLITE_FCNTL_RBUCNT    5149216

/*
** A structure to store values read from the rbu_state table in memory.
*/
struct RbuState {
  int eStage;
  char *zTbl;

  u32 mLock;
  int nFrame;                     /* Entries in aFrame[] array */
  int nFrameAlloc;                /* Allocated size of aFrame[] array */
  RbuFrame *aFrame;
  int pgsz;
  u8 *aBuf;
  i64 iWalCksum;

  /* Used in RBU vacuum mode only */
  int nRbu;                       /* Number of RBU VFS in the stack */
  rbu_file *pRbuFd;               /* Fd for main db of dbRbu */
};

/*
** An rbu VFS is implemented using an instance of this structure.
*/
struct rbu_vfs {
  sqlite3_vfs base;               /* rbu VFS shim methods */

  sqlite3_file *pReal;            /* Underlying file handle */
  rbu_vfs *pRbuVfs;               /* Pointer to the rbu_vfs object */
  sqlite3rbu *pRbu;               /* Pointer to rbu object (rbu target only) */

  int openFlags;                  /* Flags this file was opened with */
  u32 iCookie;                    /* Cookie value for main db files */
  u8 iWriteVer;                   /* "write-version" value for main db files */
  u8 bNolock;                     /* True to fail EXCLUSIVE locks */

  int nShm;                       /* Number of entries in apShm[] array */
  char **apShm;                   /* Array of mmap'd *-shm regions */
  char *zDel;                     /* Delete this when closing file */

  const char *zWal;               /* Wal filename for this main db file */
  rbu_file *pWalFd;               /* Wal file descriptor for this main db */
  rbu_file *pMainNext;            /* Next MAIN_DB file */
};

/*
** True for an RBU vacuum handle, or false otherwise.
*/
#define rbuIsVacuum(p) ((p)->zTarget==0)


/*************************************************************************
** The following three functions, found below:
**
**   rbuDeltaGetInt()
**   rbuDeltaChecksum()
**   rbuDeltaApply()

  }
  return rc;
}


/*
** The implementation of the rbu_target_name() SQL function. This function
** accepts one or two arguments. The first argument is the name of a table -
** accepts one argument - the name of a table in the RBU database. If the
** table name matches the pattern:
** the name of a table in the RBU database.  The second, if it is present, is 1
** for a view or 0 for a table. 
**
** For a non-vacuum RBU handle, if the table name matches the pattern:
**
**     data[0-9]_<name>
**
** where <name> is any sequence of 1 or more characters, <name> is returned.
** Otherwise, if the only argument does not match the above pattern, an SQL
** NULL is returned.
**
**     "data_t1"     -> "t1"
**     "data0123_t2" -> "t2"
**     "dataAB_t3"   -> NULL
**
** For an rbu vacuum handle, a copy of the first argument is returned if
** the second argument is either missing or 0 (not a view).
*/
static void rbuTargetNameFunc(
  sqlite3_context *context,
  sqlite3_context *pCtx,
  int argc,
  sqlite3_value **argv
){
  sqlite3rbu *p = sqlite3_user_data(pCtx);
  const char *zIn;
  assert( argc==1 );
  assert( argc==1 || argc==2 );

  zIn = (const char*)sqlite3_value_text(argv[0]);
  if( zIn ){
    if( rbuIsVacuum(p) ){
      if( argc==1 || 0==sqlite3_value_int(argv[1]) ){
        sqlite3_result_text(pCtx, zIn, -1, SQLITE_STATIC);
      }
    }else{
  if( zIn && strlen(zIn)>4 && memcmp("data", zIn, 4)==0 ){
    int i;
    for(i=4; zIn[i]>='0' && zIn[i]<='9'; i++);
    if( zIn[i]=='_' && zIn[i+1] ){
      sqlite3_result_text(context, &zIn[i+1], -1, SQLITE_STATIC);
      if( strlen(zIn)>4 && memcmp("data", zIn, 4)==0 ){
        int i;
        for(i=4; zIn[i]>='0' && zIn[i]<='9'; i++);
        if( zIn[i]=='_' && zIn[i+1] ){
          sqlite3_result_text(pCtx, &zIn[i+1], -1, SQLITE_STATIC);
        }
      }
    }
  }
}

/*
** Initialize the iterator structure passed as the second argument.
**
** If no error occurs, SQLITE_OK is returned and the iterator is left 
** pointing to the first entry. Otherwise, an error code and message is 
** left in the RBU handle passed as the first argument. A copy of the 
** error code is returned.
*/
static int rbuObjIterFirst(sqlite3rbu *p, RbuObjIter *pIter){
  int rc;
  memset(pIter, 0, sizeof(RbuObjIter));

  rc = prepareAndCollectError(p->dbRbu, &pIter->pTblIter, &p->zErrmsg, 
      "SELECT rbu_target_name(name, type='view') AS target, name "
      "SELECT rbu_target_name(name) AS target, name FROM sqlite_master "
      "FROM sqlite_master "
      "WHERE type IN ('table', 'view') AND target IS NOT NULL "
      "ORDER BY name"
  );

  if( rc==SQLITE_OK ){
    rc = prepareAndCollectError(p->dbMain, &pIter->pIdxIter, &p->zErrmsg,
        "SELECT name, rootpage, sql IS NULL OR substr(8, 6)=='UNIQUE' "

        bRbuRowid = 1;
      }
    }
    sqlite3_finalize(pStmt);
    pStmt = 0;

    if( p->rc==SQLITE_OK
     && rbuIsVacuum(p)==0
     && bRbuRowid!=(pIter->eType==RBU_PK_VTAB || pIter->eType==RBU_PK_NONE)
    ){
      p->rc = SQLITE_ERROR;
      p->zErrmsg = sqlite3_mprintf(
          "table %q %s rbu_rowid column", pIter->zDataTbl,
          (bRbuRowid ? "may not have" : "requires")
      );

      /* An integer primary key. If the table has an explicit IPK, use
      ** its name. Otherwise, use "rbu_rowid".  */
      if( pIter->eType==RBU_PK_IPK ){
        int i;
        for(i=0; pIter->abTblPk[i]==0; i++);
        assert( i<pIter->nTblCol );
        zCol = pIter->azTblCol[i];
      }else if( rbuIsVacuum(p) ){
        zCol = "_rowid_";
      }else{
        zCol = "rbu_rowid";
      }
      zType = "INTEGER";
    }else{
      zCol = pIter->azTblCol[iCid];
      zType = pIter->azTblType[iCid];

        p->rc = prepareFreeAndCollectError(
            p->dbMain, &pIter->pInsert, &p->zErrmsg,
          sqlite3_mprintf("INSERT INTO \"rbu_imp_%w\" VALUES(%s)", zTbl, zBind)
        );
      }

      /* And to delete index entries */
      if( p->rc==SQLITE_OK ){
      if( rbuIsVacuum(p)==0 && p->rc==SQLITE_OK ){
        p->rc = prepareFreeAndCollectError(
            p->dbMain, &pIter->pDelete, &p->zErrmsg,
          sqlite3_mprintf("DELETE FROM \"rbu_imp_%w\" WHERE %s", zTbl, zWhere)
        );
      }

      /* Create the SELECT statement to read keys in sorted order */
      if( p->rc==SQLITE_OK ){
        char *zSql;
        if( rbuIsVacuum(p) ){
          zSql = sqlite3_mprintf(
              "SELECT %s, 0 AS rbu_control FROM '%q' ORDER BY %s%s",
              zCollist, 
              pIter->zDataTbl,
              zCollist, zLimit
          );
        }else

        if( pIter->eType==RBU_PK_EXTERNAL || pIter->eType==RBU_PK_NONE ){
          zSql = sqlite3_mprintf(
              "SELECT %s, rbu_control FROM %s.'rbu_tmp_%q' ORDER BY %s%s",
              zCollist, p->zStateDb, pIter->zDataTbl,
              zCollist, zLimit
          );
        }else{

      }

      sqlite3_free(zImposterCols);
      sqlite3_free(zImposterPK);
      sqlite3_free(zWhere);
      sqlite3_free(zBind);
    }else{
      int bRbuRowid = (pIter->eType==RBU_PK_VTAB || pIter->eType==RBU_PK_NONE);
      int bRbuRowid = (pIter->eType==RBU_PK_VTAB)
                    ||(pIter->eType==RBU_PK_NONE)
                    ||(pIter->eType==RBU_PK_EXTERNAL && rbuIsVacuum(p));
      const char *zTbl = pIter->zTbl;       /* Table this step applies to */
      const char *zWrite;                   /* Imposter table name */

      char *zBindings = rbuObjIterGetBindlist(p, pIter->nTblCol + bRbuRowid);
      char *zWhere = rbuObjIterGetWhere(p, pIter);
      char *zOldlist = rbuObjIterGetOldlist(p, pIter, "old");
      char *zNewlist = rbuObjIterGetOldlist(p, pIter, "new");

            sqlite3_mprintf(
              "INSERT INTO \"%s%w\"(%s%s) VALUES(%s)", 
              zWrite, zTbl, zCollist, (bRbuRowid ? ", _rowid_" : ""), zBindings
            )
        );
      }

      /* Create the DELETE statement to write to the target PK b-tree */
      if( p->rc==SQLITE_OK ){
      /* Create the DELETE statement to write to the target PK b-tree.
      ** Because it only performs INSERT operations, this is not required for
      ** an rbu vacuum handle.  */
      if( rbuIsVacuum(p)==0 && p->rc==SQLITE_OK ){
        p->rc = prepareFreeAndCollectError(p->dbMain, &pIter->pDelete, pz,
            sqlite3_mprintf(
              "DELETE FROM \"%s%w\" WHERE %s", zWrite, zTbl, zWhere
            )
        );
      }

      if( pIter->abIndexed ){
      if( rbuIsVacuum(p)==0 && pIter->abIndexed ){
        const char *zRbuRowid = "";
        if( pIter->eType==RBU_PK_EXTERNAL || pIter->eType==RBU_PK_NONE ){
          zRbuRowid = ", rbu_rowid";
        }

        /* Create the rbu_tmp_xxx table and the triggers to populate it. */
        rbuMPrintfExec(p, p->dbRbu,

        }

        rbuObjIterPrepareTmpInsert(p, pIter, zCollist, zRbuRowid);
      }

      /* Create the SELECT statement to read keys from data_xxx */
      if( p->rc==SQLITE_OK ){
        const char *zRbuRowid = "";
        if( bRbuRowid ){
          zRbuRowid = rbuIsVacuum(p) ? ",_rowid_ " : ",rbu_rowid";
        }
        p->rc = prepareFreeAndCollectError(p->dbRbu, &pIter->pSelect, pz,
            sqlite3_mprintf(
              "SELECT %s, rbu_control%s FROM '%q'%s", 
              zCollist, (bRbuRowid ? ", rbu_rowid" : ""), 
              "SELECT %s,%s rbu_control%s FROM '%q'%s", 
              zCollist, 
              (rbuIsVacuum(p) ? "0 AS " : ""),
              zRbuRowid,
              pIter->zDataTbl, zLimit
            )
        );
      }

      sqlite3_free(zWhere);
      sqlite3_free(zOldlist);

    sqlite3_free(zWhere);
    sqlite3_free(zSet);
  }

  return p->rc;
}

static sqlite3 *rbuOpenDbhandle(sqlite3rbu *p, const char *zName){
static sqlite3 *rbuOpenDbhandle(
  sqlite3rbu *p, 
  const char *zName, 
  int bUseVfs
){
  sqlite3 *db = 0;
  if( p->rc==SQLITE_OK ){
    const int flags = SQLITE_OPEN_READWRITE|SQLITE_OPEN_CREATE|SQLITE_OPEN_URI;
    p->rc = sqlite3_open_v2(zName, &db, flags, p->zVfsName);
    p->rc = sqlite3_open_v2(zName, &db, flags, bUseVfs ? p->zVfsName : 0);
    if( p->rc ){
      p->zErrmsg = sqlite3_mprintf("%s", sqlite3_errmsg(db));
      sqlite3_close(db);
      db = 0;
    }
  }
  return db;
}

/*
** Free an RbuState object allocated by rbuLoadState().
*/
static void rbuFreeState(RbuState *p){
  if( p ){
    sqlite3_free(p->zTbl);
    sqlite3_free(p->zIdx);
    sqlite3_free(p);
  }
}

/*
** Allocate an RbuState object and load the contents of the rbu_state 
** table into it. Return a pointer to the new object. It is the 
** responsibility of the caller to eventually free the object using
** sqlite3_free().
**
** If an error occurs, leave an error code and message in the rbu handle
** and return NULL.
*/
static RbuState *rbuLoadState(sqlite3rbu *p){
  RbuState *pRet = 0;
  sqlite3_stmt *pStmt = 0;
  int rc;
  int rc2;

  pRet = (RbuState*)rbuMalloc(p, sizeof(RbuState));
  if( pRet==0 ) return 0;

  rc = prepareFreeAndCollectError(p->dbRbu, &pStmt, &p->zErrmsg, 
      sqlite3_mprintf("SELECT k, v FROM %s.rbu_state", p->zStateDb)
  );
  while( rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pStmt) ){
    switch( sqlite3_column_int(pStmt, 0) ){
      case RBU_STATE_STAGE:
        pRet->eStage = sqlite3_column_int(pStmt, 1);
        if( pRet->eStage!=RBU_STAGE_OAL
         && pRet->eStage!=RBU_STAGE_MOVE
         && pRet->eStage!=RBU_STAGE_CKPT
        ){
          p->rc = SQLITE_CORRUPT;
        }
        break;

      case RBU_STATE_TBL:
        pRet->zTbl = rbuStrndup((char*)sqlite3_column_text(pStmt, 1), &rc);
        break;

      case RBU_STATE_IDX:
        pRet->zIdx = rbuStrndup((char*)sqlite3_column_text(pStmt, 1), &rc);
        break;

      case RBU_STATE_ROW:
        pRet->nRow = sqlite3_column_int(pStmt, 1);
        break;

      case RBU_STATE_PROGRESS:
        pRet->nProgress = sqlite3_column_int64(pStmt, 1);
        break;

      case RBU_STATE_CKPT:
        pRet->iWalCksum = sqlite3_column_int64(pStmt, 1);
        break;

      case RBU_STATE_COOKIE:
        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 = sqlite3_column_int64(pStmt, 1);
        break;

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

  p->rc = rc;
  return pRet;
}


/*
** Open the database handle and attach the RBU database as "rbu". If an
** error occurs, leave an error code and message in the RBU handle.
*/
static void rbuOpenDatabase(sqlite3rbu *p){
  assert( p->rc==SQLITE_OK );
  assert( p->dbMain==0 && p->dbRbu==0 );
  assert( rbuIsVacuum(p) || p->zTarget!=0 );

  p->eStage = 0;
  p->dbMain = rbuOpenDbhandle(p, p->zTarget);
  p->dbRbu = rbuOpenDbhandle(p, p->zRbu);
  /* Open the RBU database */
  p->dbRbu = rbuOpenDbhandle(p, p->zRbu, 1);

  if( p->rc==SQLITE_OK && rbuIsVacuum(p) ){
    sqlite3_file_control(p->dbRbu, "main", SQLITE_FCNTL_RBUCNT, (void*)p);
  }

  /* If using separate RBU and state databases, attach the state database to
  ** the RBU db handle now.  */
  if( p->zState ){
    rbuMPrintfExec(p, p->dbRbu, "ATTACH %Q AS stat", p->zState);
    memcpy(p->zStateDb, "stat", 4);
  }else{
    memcpy(p->zStateDb, "main", 4);
  }

#if 0
  if( p->rc==SQLITE_OK && rbuIsVacuum(p) ){
    p->rc = sqlite3_exec(p->dbRbu, "BEGIN", 0, 0, 0);
  }
#endif

  /* If it has not already been created, create the rbu_state table */
  rbuMPrintfExec(p, p->dbRbu, RBU_CREATE_STATE, p->zStateDb);

#if 0
  if( rbuIsVacuum(p) ){
    if( p->rc==SQLITE_OK ){
      int rc2;
      int bOk = 0;
      sqlite3_stmt *pCnt = 0;
      p->rc = prepareAndCollectError(p->dbRbu, &pCnt, &p->zErrmsg,
          "SELECT count(*) FROM stat.sqlite_master"
      );
      if( p->rc==SQLITE_OK 
       && sqlite3_step(pCnt)==SQLITE_ROW
       && 1==sqlite3_column_int(pCnt, 0)
      ){
        bOk = 1;
      }
      rc2 = sqlite3_finalize(pCnt);
      if( p->rc==SQLITE_OK ) p->rc = rc2;

      if( p->rc==SQLITE_OK && bOk==0 ){
        p->rc = SQLITE_ERROR;
        p->zErrmsg = sqlite3_mprintf("invalid state database");
      }
    
      if( p->rc==SQLITE_OK ){
        p->rc = sqlite3_exec(p->dbRbu, "COMMIT", 0, 0, 0);
      }
    }
  }
#endif

  if( p->rc==SQLITE_OK && rbuIsVacuum(p) ){
    int bOpen = 0;
    int rc;
    p->nRbu = 0;
    p->pRbuFd = 0;
    rc = sqlite3_file_control(p->dbRbu, "main", SQLITE_FCNTL_RBUCNT, (void*)p);
    if( rc!=SQLITE_NOTFOUND ) p->rc = rc;
    if( p->eStage>=RBU_STAGE_MOVE ){
      bOpen = 1;
    }else{
      RbuState *pState = rbuLoadState(p);
      if( pState ){
        bOpen = (pState->eStage>RBU_STAGE_MOVE);
        rbuFreeState(pState);
      }
    }
    if( bOpen ) p->dbMain = rbuOpenDbhandle(p, p->zRbu, p->nRbu<=1);
  }

  p->eStage = 0;
  if( p->rc==SQLITE_OK && p->dbMain==0 ){
    if( !rbuIsVacuum(p) ){
      p->dbMain = rbuOpenDbhandle(p, p->zTarget, 1);
    }else if( p->pRbuFd->pWalFd ){
      p->rc = SQLITE_ERROR;
      p->zErrmsg = sqlite3_mprintf("cannot vacuum wal mode database");
    }else{
      char *zTarget;
      char *zExtra = 0;
      if( strlen(p->zRbu)>=5 && 0==memcmp("file:", p->zRbu, 5) ){
        zExtra = &p->zRbu[5];
        while( *zExtra ){
          if( *zExtra++=='?' ) break;
        }
        if( *zExtra=='\0' ) zExtra = 0;
      }

      zTarget = sqlite3_mprintf("file:%s-vacuum?rbu_memory=1%s%s", 
          sqlite3_db_filename(p->dbRbu, "main"),
          (zExtra==0 ? "" : "&"), (zExtra==0 ? "" : zExtra)
      );

      if( zTarget==0 ){
        p->rc = SQLITE_NOMEM;
        return;
      }
      p->dbMain = rbuOpenDbhandle(p, zTarget, p->nRbu<=1);
      sqlite3_free(zTarget);
    }
  }

  if( p->rc==SQLITE_OK ){
    p->rc = sqlite3_create_function(p->dbMain, 
        "rbu_tmp_insert", -1, SQLITE_UTF8, (void*)p, rbuTmpInsertFunc, 0, 0
    );
  }

  if( p->rc==SQLITE_OK ){
    p->rc = sqlite3_create_function(p->dbMain, 
        "rbu_fossil_delta", 2, SQLITE_UTF8, 0, rbuFossilDeltaFunc, 0, 0
    );
  }

  if( p->rc==SQLITE_OK ){
    p->rc = sqlite3_create_function(p->dbRbu, 
        "rbu_target_name", 1, SQLITE_UTF8, (void*)p, rbuTargetNameFunc, 0, 0
        "rbu_target_name", -1, SQLITE_UTF8, (void*)p, rbuTargetNameFunc, 0, 0
    );
  }

  if( p->rc==SQLITE_OK ){
    p->rc = sqlite3_file_control(p->dbMain, "main", SQLITE_FCNTL_RBU, (void*)p);
  }
  rbuMPrintfExec(p, p->dbMain, "SELECT * FROM sqlite_master");

** on the database file. This proc moves the *-oal file to the *-wal path,
** then reopens the database file (this time in vanilla, non-oal, WAL mode).
** If an error occurs, leave an error code and error message in the rbu 
** handle.
*/
static void rbuMoveOalFile(sqlite3rbu *p){
  const char *zBase = sqlite3_db_filename(p->dbMain, "main");
  const char *zMove = zBase;
  char *zOal;
  char *zWal;

  if( rbuIsVacuum(p) ){
    zMove = sqlite3_db_filename(p->dbRbu, "main");
  }
  char *zWal = sqlite3_mprintf("%s-wal", zBase);
  char *zOal = sqlite3_mprintf("%s-oal", zBase);
  zOal = sqlite3_mprintf("%s-oal", zMove);
  zWal = sqlite3_mprintf("%s-wal", zMove);

  assert( p->eStage==RBU_STAGE_MOVE );
  assert( p->rc==SQLITE_OK && p->zErrmsg==0 );
  if( zWal==0 || zOal==0 ){
    p->rc = SQLITE_NOMEM;
  }else{
    /* Move the *-oal file to *-wal. At this point connection p->db is
    ** holding a SHARED lock on the target database file (because it is
    ** in WAL mode). So no other connection may be writing the db. 
    **
    ** In order to ensure that there are no database readers, an EXCLUSIVE
    ** lock is obtained here before the *-oal is moved to *-wal.
    */
    rbuLockDatabase(p);
    if( p->rc==SQLITE_OK ){
      rbuFileSuffix3(zBase, zWal);
      rbuFileSuffix3(zBase, zOal);

      /* Re-open the databases. */
      rbuObjIterFinalize(&p->objiter);
      sqlite3_close(p->dbMain);
      sqlite3_close(p->dbRbu);
      sqlite3_close(p->dbRbu);
      sqlite3_close(p->dbMain);
      p->dbMain = 0;
      p->dbRbu = 0;

#if defined(_WIN32_WCE)
      {
        LPWSTR zWideOal;
        LPWSTR zWideWal;

      continue;
    }

    pVal = sqlite3_column_value(pIter->pSelect, i);
    p->rc = sqlite3_bind_value(pWriter, i+1, pVal);
    if( p->rc ) return;
  }
  if( pIter->zIdx==0
   && (pIter->eType==RBU_PK_VTAB || pIter->eType==RBU_PK_NONE) 
  ){
    /* For a virtual table, or a table with no primary key, the 
    ** SELECT statement is:
    **
    **   SELECT <cols>, rbu_control, rbu_rowid FROM ....
    **
    ** Hence column_value(pIter->nCol+1).
    */
    assertColumnName(pIter->pSelect, pIter->nCol+1, "rbu_rowid");
    pVal = sqlite3_column_value(pIter->pSelect, pIter->nCol+1);
    p->rc = sqlite3_bind_value(pWriter, pIter->nCol+1, pVal);
  if( pIter->zIdx==0 ){
    if( pIter->eType==RBU_PK_VTAB 
     || pIter->eType==RBU_PK_NONE 
     || (pIter->eType==RBU_PK_EXTERNAL && rbuIsVacuum(p)) 
    ){
      /* For a virtual table, or a table with no primary key, the 
      ** SELECT statement is:
      **
      **   SELECT <cols>, rbu_control, rbu_rowid FROM ....
      **
      ** Hence column_value(pIter->nCol+1).
      */
      assertColumnName(pIter->pSelect, pIter->nCol+1, 
          rbuIsVacuum(p) ? "rowid" : "rbu_rowid"
      );
      pVal = sqlite3_column_value(pIter->pSelect, pIter->nCol+1);
      p->rc = sqlite3_bind_value(pWriter, pIter->nCol+1, pVal);
    }
  }
  if( p->rc==SQLITE_OK ){
    sqlite3_step(pWriter);
    p->rc = resetAndCollectError(pWriter, &p->zErrmsg);
  }
}

    }
  }
  return p->rc;
}

/*
** Increment the schema cookie of the main database opened by p->dbMain.
**
** Or, if this is an RBU vacuum, set the schema cookie of the main db
** opened by p->dbMain to one more than the schema cookie of the main
** db opened by p->dbRbu.
*/
static void rbuIncrSchemaCookie(sqlite3rbu *p){
  if( p->rc==SQLITE_OK ){
    sqlite3 *dbread = (rbuIsVacuum(p) ? p->dbRbu : p->dbMain);
    int iCookie = 1000000;
    sqlite3_stmt *pStmt;

    p->rc = prepareAndCollectError(p->dbMain, &pStmt, &p->zErrmsg, 
    p->rc = prepareAndCollectError(dbread, &pStmt, &p->zErrmsg, 
        "PRAGMA schema_version"
    );
    if( p->rc==SQLITE_OK ){
      /* Coverage: it may be that this sqlite3_step() cannot fail. There
      ** is already a transaction open, so the prepared statement cannot
      ** throw an SQLITE_SCHEMA exception. The only database page the
      ** statement reads is page 1, which is guaranteed to be in the cache.

** Update the contents of the rbu_state table within the rbu database. The
** value stored in the RBU_STATE_STAGE column is eStage. All other values
** are determined by inspecting the rbu handle passed as the first argument.
*/
static void rbuSaveState(sqlite3rbu *p, int eStage){
  if( p->rc==SQLITE_OK || p->rc==SQLITE_DONE ){
    sqlite3_stmt *pInsert = 0;
    rbu_file *pFd = (rbuIsVacuum(p) ? p->pRbuFd : p->pTargetFd);
    int rc;

    assert( p->zErrmsg==0 );
    rc = prepareFreeAndCollectError(p->dbRbu, &pInsert, &p->zErrmsg, 
        sqlite3_mprintf(
          "INSERT OR REPLACE INTO %s.rbu_state(k, v) VALUES "
          "(%d, %d), "

          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_COOKIE, (i64)pFd->iCookie,
          RBU_STATE_OALSZ, p->iOalSz,
          RBU_STATE_PHASEONESTEP, p->nPhaseOneStep
      )
    );
    assert( pInsert==0 || rc==SQLITE_OK );

    if( rc==SQLITE_OK ){

        RbuObjIter *pIter = &p->objiter;
        while( p->rc==SQLITE_OK && pIter->zTbl ){

          if( pIter->bCleanup ){
            /* Clean up the rbu_tmp_xxx table for the previous table. It 
            ** cannot be dropped as there are currently active SQL statements.
            ** But the contents can be deleted.  */
            if( pIter->abIndexed ){
            if( rbuIsVacuum(p)==0 && pIter->abIndexed ){
              rbuMPrintfExec(p, p->dbRbu, 
                  "DELETE FROM %s.'rbu_tmp_%q'", p->zStateDb, pIter->zDataTbl
              );
            }
          }else{
            rbuObjIterPrepareAll(p, pIter, 0);

    }
    return p->rc;
  }else{
    return SQLITE_NOMEM;
  }
}

/*
** Free an RbuState object allocated by rbuLoadState().
*/
static void rbuFreeState(RbuState *p){
  if( p ){
    sqlite3_free(p->zTbl);
    sqlite3_free(p->zIdx);
    sqlite3_free(p);
  }
}

/*
** Allocate an RbuState object and load the contents of the rbu_state 
** table into it. Return a pointer to the new object. It is the 
** responsibility of the caller to eventually free the object using
** sqlite3_free().
**
** If an error occurs, leave an error code and message in the rbu handle
** and return NULL.
*/
static RbuState *rbuLoadState(sqlite3rbu *p){
  RbuState *pRet = 0;
  sqlite3_stmt *pStmt = 0;
  int rc;
  int rc2;

  pRet = (RbuState*)rbuMalloc(p, sizeof(RbuState));
  if( pRet==0 ) return 0;

  rc = prepareFreeAndCollectError(p->dbRbu, &pStmt, &p->zErrmsg, 
      sqlite3_mprintf("SELECT k, v FROM %s.rbu_state", p->zStateDb)
  );
  while( rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pStmt) ){
    switch( sqlite3_column_int(pStmt, 0) ){
      case RBU_STATE_STAGE:
        pRet->eStage = sqlite3_column_int(pStmt, 1);
        if( pRet->eStage!=RBU_STAGE_OAL
         && pRet->eStage!=RBU_STAGE_MOVE
         && pRet->eStage!=RBU_STAGE_CKPT
        ){
          p->rc = SQLITE_CORRUPT;
        }
        break;

      case RBU_STATE_TBL:
        pRet->zTbl = rbuStrndup((char*)sqlite3_column_text(pStmt, 1), &rc);
        break;

      case RBU_STATE_IDX:
        pRet->zIdx = rbuStrndup((char*)sqlite3_column_text(pStmt, 1), &rc);
        break;

      case RBU_STATE_ROW:
        pRet->nRow = sqlite3_column_int(pStmt, 1);
        break;

      case RBU_STATE_PROGRESS:
        pRet->nProgress = sqlite3_column_int64(pStmt, 1);
        break;

      case RBU_STATE_CKPT:
        pRet->iWalCksum = sqlite3_column_int64(pStmt, 1);
        break;

      case RBU_STATE_COOKIE:
        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 = sqlite3_column_int64(pStmt, 1);
        break;

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

  p->rc = rc;
  return pRet;
}

/*
** Compare strings z1 and z2, returning 0 if they are identical, or non-zero
** otherwise. Either or both argument may be NULL. Two NULL values are
** considered equal, and NULL is considered distinct from all other values.
*/
static int rbuStrCompare(const char *z1, const char *z2){
  if( z1==0 && z2==0 ) return 0;

        p->rc = sqlite3_finalize(pStmt);
      }
    }
  }
}

/*
** The second argument passed to this function is the name of a PRAGMA 
** setting - "page_size", "auto_vacuum", "user_version" or "application_id".
** This function executes the following on sqlite3rbu.dbRbu:
**
**   "PRAGMA main.$zPragma"
**
** where $zPragma is the string passed as the second argument, then
** on sqlite3rbu.dbMain:
**
**   "PRAGMA main.$zPragma = $val"
**
** where $val is the value returned by the first PRAGMA invocation.
**
** In short, it copies the value  of the specified PRAGMA setting from
** dbRbu to dbMain.
*/
static void rbuCopyPragma(sqlite3rbu *p, const char *zPragma){
  if( p->rc==SQLITE_OK ){
    sqlite3_stmt *pPragma = 0;
    p->rc = prepareFreeAndCollectError(p->dbRbu, &pPragma, &p->zErrmsg, 
        sqlite3_mprintf("PRAGMA main.%s", zPragma)
    );
    if( p->rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pPragma) ){
      p->rc = rbuMPrintfExec(p, p->dbMain, "PRAGMA main.%s = %d",
          zPragma, sqlite3_column_int(pPragma, 0)
      );
    }
    rbuFinalize(p, pPragma);
  }
}
** Open and return a new RBU handle. 

/*
** The RBU handle passed as the only argument has just been opened and 
** the state database is empty. If this RBU handle was opened for an
** RBU vacuum operation, create the schema in the target db.
*/
static void rbuCreateTargetSchema(sqlite3rbu *p){
  sqlite3_stmt *pSql = 0;
  sqlite3_stmt *pInsert = 0;

  assert( rbuIsVacuum(p) );
  p->rc = sqlite3_exec(p->dbMain, "PRAGMA writable_schema=1", 0,0, &p->zErrmsg);
  if( p->rc==SQLITE_OK ){
    p->rc = prepareAndCollectError(p->dbRbu, &pSql, &p->zErrmsg, 
      "SELECT sql FROM sqlite_master WHERE sql!='' AND rootpage!=0"
      " AND name!='sqlite_sequence' "
      " ORDER BY type DESC"
    );
  }

  while( p->rc==SQLITE_OK && sqlite3_step(pSql)==SQLITE_ROW ){
    const char *zSql = (const char*)sqlite3_column_text(pSql, 0);
    p->rc = sqlite3_exec(p->dbMain, zSql, 0, 0, &p->zErrmsg);
  }
  rbuFinalize(p, pSql);
  if( p->rc!=SQLITE_OK ) return;

  if( p->rc==SQLITE_OK ){
    p->rc = prepareAndCollectError(p->dbRbu, &pSql, &p->zErrmsg, 
        "SELECT * FROM sqlite_master WHERE rootpage=0 OR rootpage IS NULL" 
    );
  }

  if( p->rc==SQLITE_OK ){
    p->rc = prepareAndCollectError(p->dbMain, &pInsert, &p->zErrmsg, 
        "INSERT INTO sqlite_master VALUES(?,?,?,?,?)"
    );
  }

  while( p->rc==SQLITE_OK && sqlite3_step(pSql)==SQLITE_ROW ){
    int i;
    for(i=0; i<5; i++){
      sqlite3_bind_value(pInsert, i+1, sqlite3_column_value(pSql, i));
    }
    sqlite3_step(pInsert);
    p->rc = sqlite3_reset(pInsert);
  }
  if( p->rc==SQLITE_OK ){
    p->rc = sqlite3_exec(p->dbMain, "PRAGMA writable_schema=0",0,0,&p->zErrmsg);
  }

  rbuFinalize(p, pSql);
  rbuFinalize(p, pInsert);
}


sqlite3rbu *sqlite3rbu_open(
static sqlite3rbu *openRbuHandle(
  const char *zTarget, 
  const char *zRbu,
  const char *zState
){
  sqlite3rbu *p;
  size_t nTarget = strlen(zTarget);
  size_t nTarget = zTarget ? strlen(zTarget) : 0;
  size_t nRbu = strlen(zRbu);
  size_t nState = zState ? strlen(zState) : 0;
  size_t nByte = sizeof(sqlite3rbu) + nTarget+1 + nRbu+1+ nState+1;

  p = (sqlite3rbu*)sqlite3_malloc64(nByte);
  if( p ){
    RbuState *pState = 0;

    /* Create the custom VFS. */
    memset(p, 0, sizeof(sqlite3rbu));
    rbuCreateVfs(p);

    /* Open the target database */
    /* Open the target, RBU and state databases */
    if( p->rc==SQLITE_OK ){
      char *pCsr = (char*)&p[1];
      if( zTarget ){
      p->zTarget = (char*)&p[1];
      memcpy(p->zTarget, zTarget, nTarget+1);
      p->zRbu = &p->zTarget[nTarget+1];
        p->zTarget = pCsr;
        memcpy(p->zTarget, zTarget, nTarget+1);
        pCsr += nTarget+1;
      }
      p->zRbu = pCsr;
      memcpy(p->zRbu, zRbu, nRbu+1);
      pCsr += nRbu+1;
      if( zState ){
        p->zState = &p->zRbu[nRbu+1];
        p->zState = pCsr;
        memcpy(p->zState, zState, nState+1);
      }
      rbuOpenDatabase(p);
    }

    /* If it has not already been created, create the rbu_state table */
    rbuMPrintfExec(p, p->dbRbu, RBU_CREATE_STATE, p->zStateDb);

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

        p->zErrmsg = sqlite3_mprintf("cannot update wal mode database");
      }else if( p->eStage==RBU_STAGE_MOVE ){
        p->eStage = RBU_STAGE_CKPT;
        p->nStep = 0;
      }
    }

    if( p->rc==SQLITE_OK
    if( p->rc==SQLITE_OK 
     && (p->eStage==RBU_STAGE_OAL || p->eStage==RBU_STAGE_MOVE)
     && pState->eStage!=0 && p->pTargetFd->iCookie!=pState->iCookie
    ){   
      /* At this point (pTargetFd->iCookie) contains the value of the
      ** change-counter cookie (the thing that gets incremented when a 
      ** transaction is committed in rollback mode) currently stored on 
      ** page 1 of the database file. */
      p->rc = SQLITE_BUSY;
      p->zErrmsg = sqlite3_mprintf("database modified during rbu update");
     && pState->eStage!=0
    ){
      rbu_file *pFd = (rbuIsVacuum(p) ? p->pRbuFd : p->pTargetFd);
      if( pFd->iCookie!=pState->iCookie ){   
        /* At this point (pTargetFd->iCookie) contains the value of the
        ** change-counter cookie (the thing that gets incremented when a 
        ** transaction is committed in rollback mode) currently stored on 
        ** page 1 of the database file. */
        p->rc = SQLITE_BUSY;
        p->zErrmsg = sqlite3_mprintf("database modified during rbu %s",
            (rbuIsVacuum(p) ? "vacuum" : "update")
        );
      }
    }

    if( p->rc==SQLITE_OK ){
      if( p->eStage==RBU_STAGE_OAL ){
        sqlite3 *db = p->dbMain;

        if( pState->eStage==0 && rbuIsVacuum(p) ){
          rbuCopyPragma(p, "page_size");
          rbuCopyPragma(p, "auto_vacuum");
        }

        /* Open transactions both databases. The *-oal file is opened or
        ** created at this point. */
        if( p->rc==SQLITE_OK ){
        p->rc = sqlite3_exec(db, "BEGIN IMMEDIATE", 0, 0, &p->zErrmsg);
          p->rc = sqlite3_exec(db, "BEGIN IMMEDIATE", 0, 0, &p->zErrmsg);
        }
        if( p->rc==SQLITE_OK ){
          p->rc = sqlite3_exec(p->dbRbu, "BEGIN IMMEDIATE", 0, 0, &p->zErrmsg);
          p->rc = sqlite3_exec(p->dbRbu, "BEGIN", 0, 0, &p->zErrmsg);
        }

        /* Check if the main database is a zipvfs db. If it is, set the upper
        ** level pager to use "journal_mode=off". This prevents it from 
        ** generating a large journal using a temp file.  */
        if( p->rc==SQLITE_OK ){
          int frc = sqlite3_file_control(db, "main", SQLITE_FCNTL_ZIPVFS, 0);
          if( frc==SQLITE_OK ){
            p->rc = sqlite3_exec(db, "PRAGMA journal_mode=off",0,0,&p->zErrmsg);
          }
        }

        /* If this is an RBU vacuum operation and the state table was empty
        ** when this handle was opened, create the target database schema. */
        if( p->rc==SQLITE_OK && pState->eStage==0 && rbuIsVacuum(p) ){
          rbuCreateTargetSchema(p);
          rbuCopyPragma(p, "user_version");
          rbuCopyPragma(p, "application_id");
        }

        /* Point the object iterator at the first object */
        if( p->rc==SQLITE_OK ){
          p->rc = rbuObjIterFirst(p, &p->objiter);
        }

        /* If the RBU database contains no data_xxx tables, declare the RBU

    rbuFreeState(pState);
  }

  return p;
}

/*
** Open and return a new RBU handle. 
*/
sqlite3rbu *sqlite3rbu_open(
  const char *zTarget, 
  const char *zRbu,
  const char *zState
){
  /* TODO: Check that zTarget and zRbu are non-NULL */
  return openRbuHandle(zTarget, zRbu, zState);
}

/*
** Open a handle to begin or resume an RBU VACUUM operation.
*/
sqlite3rbu *sqlite3rbu_vacuum(
  const char *zTarget, 
  const char *zState
){
  /* TODO: Check that both arguments are non-NULL */
  return openRbuHandle(0, zTarget, zState);
}

/*
** Return the database handle used by pRbu.
*/
sqlite3 *sqlite3rbu_db(sqlite3rbu *pRbu, int bRbu){
  sqlite3 *db = 0;
  if( pRbu ){

    if( p->rc==SQLITE_OK && p->eStage==RBU_STAGE_OAL ){
      p->rc = sqlite3_exec(p->dbRbu, "COMMIT", 0, 0, &p->zErrmsg);
    }

    /* Close any open statement handles. */
    rbuObjIterFinalize(&p->objiter);

    /* If this is an RBU vacuum handle and the vacuum has either finished
    ** successfully or encountered an error, delete the contents of the 
    ** state table. This causes the next call to sqlite3rbu_vacuum() 
    ** specifying the current target and state databases to start a new
    ** vacuum from scratch.  */
    if( rbuIsVacuum(p) && p->rc!=SQLITE_OK && p->dbRbu ){
      int rc2 = sqlite3_exec(p->dbRbu, "DELETE FROM stat.rbu_state", 0, 0, 0);
      if( p->rc==SQLITE_DONE && rc2!=SQLITE_OK ) p->rc = rc2;
    }

    /* Close the open database handle and VFS object. */
    sqlite3_close(p->dbMain);
    sqlite3_close(p->dbRbu);
    sqlite3_close(p->dbRbu);
    sqlite3_close(p->dbMain);
    rbuDeleteVfs(p);
    sqlite3_free(p->aBuf);
    sqlite3_free(p->aFrame);

    rbuEditErrmsg(p);
    rc = p->rc;
    *pzErrmsg = p->zErrmsg;

*/
static u32 rbuGetU32(u8 *aBuf){
  return ((u32)aBuf[0] << 24)
       + ((u32)aBuf[1] << 16)
       + ((u32)aBuf[2] <<  8)
       + ((u32)aBuf[3]);
}

/*
** Write an unsigned 32-bit value in big-endian format to the supplied
** buffer.
*/
static void rbuPutU32(u8 *aBuf, u32 iVal){
  aBuf[0] = (iVal >> 24) & 0xFF;
  aBuf[1] = (iVal >> 16) & 0xFF;
  aBuf[2] = (iVal >>  8) & 0xFF;
  aBuf[3] = (iVal >>  0) & 0xFF;
}

static void rbuPutU16(u8 *aBuf, u16 iVal){
  aBuf[0] = (iVal >>  8) & 0xFF;
  aBuf[1] = (iVal >>  0) & 0xFF;
}

/*
** Read data from an rbuVfs-file.
*/
static int rbuVfsRead(
  sqlite3_file *pFile, 
  void *zBuf, 

     && (p->openFlags & SQLITE_OPEN_WAL) 
     && iOfst>=pRbu->iOalSz 
    ){
      rc = SQLITE_OK;
      memset(zBuf, 0, iAmt);
    }else{
      rc = p->pReal->pMethods->xRead(p->pReal, zBuf, iAmt, iOfst);
#if 1
      /* If this is being called to read the first page of the target 
      ** database as part of an rbu vacuum operation, synthesize the 
      ** contents of the first page if it does not yet exist. Otherwise,
      ** SQLite will not check for a *-wal file.  */
      if( pRbu && rbuIsVacuum(pRbu) 
          && rc==SQLITE_IOERR_SHORT_READ && iOfst==0
          && (p->openFlags & SQLITE_OPEN_MAIN_DB)
          && pRbu->rc==SQLITE_OK
      ){
        sqlite3_file *pFd = (sqlite3_file*)pRbu->pRbuFd;
        rc = pFd->pMethods->xRead(pFd, zBuf, iAmt, iOfst);
        if( rc==SQLITE_OK ){
          u8 *aBuf = (u8*)zBuf;
          u32 iRoot = rbuGetU32(&aBuf[52]) ? 1 : 0;
          rbuPutU32(&aBuf[52], iRoot);      /* largest root page number */
          rbuPutU32(&aBuf[36], 0);          /* number of free pages */
          rbuPutU32(&aBuf[32], 0);          /* first page on free list trunk */
          rbuPutU32(&aBuf[28], 1);          /* size of db file in pages */
          rbuPutU32(&aBuf[24], pRbu->pRbuFd->iCookie+1);  /* Change counter */

          if( iAmt>100 ){
            memset(&aBuf[100], 0, iAmt-100);
            rbuPutU16(&aBuf[105], iAmt & 0xFFFF);
            aBuf[100] = 0x0D;
          }
        }
      }
#endif
    }
    if( rc==SQLITE_OK && iOfst==0 && (p->openFlags & SQLITE_OPEN_MAIN_DB) ){
      /* These look like magic numbers. But they are stable, as they are part
       ** of the definition of the SQLite file format, which may not change. */
      u8 *pBuf = (u8*)zBuf;
      p->iCookie = rbuGetU32(&pBuf[24]);
      p->iWriteVer = pBuf[19];

}

/*
** Return the current file-size of an rbuVfs-file.
*/
static int rbuVfsFileSize(sqlite3_file *pFile, sqlite_int64 *pSize){
  rbu_file *p = (rbu_file *)pFile;
  int rc;
  return p->pReal->pMethods->xFileSize(p->pReal, pSize);
  rc = p->pReal->pMethods->xFileSize(p->pReal, pSize);

  /* If this is an RBU vacuum operation and this is the target database,
  ** pretend that it has at least one page. Otherwise, SQLite will not
  ** check for the existance of a *-wal file. rbuVfsRead() contains 
  ** similar logic.  */
  if( rc==SQLITE_OK && *pSize==0 
   && p->pRbu && rbuIsVacuum(p->pRbu) 
   && (p->openFlags & SQLITE_OPEN_MAIN_DB)
  ){
    *pSize = 1024;
  }
  return rc;
}

/*
** Lock an rbuVfs-file.
*/
static int rbuVfsLock(sqlite3_file *pFile, int eLock){
  rbu_file *p = (rbu_file*)pFile;
  sqlite3rbu *pRbu = p->pRbu;
  int rc = SQLITE_OK;

  assert( p->openFlags & (SQLITE_OPEN_MAIN_DB|SQLITE_OPEN_TEMP_DB) );
  if( pRbu && eLock==SQLITE_LOCK_EXCLUSIVE && pRbu->eStage!=RBU_STAGE_DONE ){
  if( eLock==SQLITE_LOCK_EXCLUSIVE 
   && (p->bNolock || (pRbu && pRbu->eStage!=RBU_STAGE_DONE))
  ){
    /* Do not allow EXCLUSIVE locks. Preventing SQLite from taking this 
    ** prevents it from checkpointing the database from sqlite3_close(). */
    rc = SQLITE_BUSY;
  }else{
    rc = p->pReal->pMethods->xLock(p->pReal, eLock);
  }

        pRbu->pTargetFd = p;
        p->pRbu = pRbu;
        if( p->pWalFd ) p->pWalFd->pRbu = pRbu;
        rc = SQLITE_OK;
      }
    }
    return rc;
  }
  else if( op==SQLITE_FCNTL_RBUCNT ){
    sqlite3rbu *pRbu = (sqlite3rbu*)pArg;
    pRbu->nRbu++;
    pRbu->pRbuFd = p;
    p->bNolock = 1;
  }

  rc = xControl(p->pReal, op, pArg);
  if( rc==SQLITE_OK && op==SQLITE_FCNTL_VFSNAME ){
    rbu_vfs *pRbuVfs = p->pRbuVfs;
    char *zIn = *(char**)pArg;
    char *zOut = sqlite3_mprintf("rbu(%s)/%z", pRbuVfs->base.zName, zIn);

static rbu_file *rbuFindMaindb(rbu_vfs *pRbuVfs, const char *zWal){
  rbu_file *pDb;
  sqlite3_mutex_enter(pRbuVfs->mutex);
  for(pDb=pRbuVfs->pMain; pDb && pDb->zWal!=zWal; pDb=pDb->pMainNext){}
  sqlite3_mutex_leave(pRbuVfs->mutex);
  return pDb;
}

/* 
** A main database named zName has just been opened. The following 
** function returns a pointer to a buffer owned by SQLite that contains
** the name of the *-wal file this db connection will use. SQLite
** happens to pass a pointer to this buffer when using xAccess()
** or xOpen() to operate on the *-wal file.  
*/
static const char *rbuMainToWal(const char *zName, int flags){
  int n = (int)strlen(zName);
  const char *z = &zName[n];
  if( flags & SQLITE_OPEN_URI ){
    int odd = 0;
    while( 1 ){
      if( z[0]==0 ){
        odd = 1 - odd;
        if( odd && z[1]==0 ) break;
      }
      z++;
    }
    z += 2;
  }else{
    while( *z==0 ) z++;
  }
  z += (n + 8 + 1);
  return z;
}

/*
** Open an rbu file handle.
*/
static int rbuVfsOpen(
  sqlite3_vfs *pVfs,
  const char *zName,

    0, 0                          /* xFetch, xUnfetch */
  };
  rbu_vfs *pRbuVfs = (rbu_vfs*)pVfs;
  sqlite3_vfs *pRealVfs = pRbuVfs->pRealVfs;
  rbu_file *pFd = (rbu_file *)pFile;
  int rc = SQLITE_OK;
  const char *zOpen = zName;
  int oflags = flags;

  memset(pFd, 0, sizeof(rbu_file));
  pFd->pReal = (sqlite3_file*)&pFd[1];
  pFd->pRbuVfs = pRbuVfs;
  pFd->openFlags = flags;
  if( zName ){
    if( flags & SQLITE_OPEN_MAIN_DB ){
      /* A main database has just been opened. The following block sets
      ** (pFd->zWal) to point to a buffer owned by SQLite that contains
      ** the name of the *-wal file this db connection will use. SQLite
      ** happens to pass a pointer to this buffer when using xAccess()
      ** or xOpen() to operate on the *-wal file.  */
      int n = (int)strlen(zName);
      const char *z = &zName[n];
      if( flags & SQLITE_OPEN_URI ){
        int odd = 0;
        while( 1 ){
          if( z[0]==0 ){
            odd = 1 - odd;
            if( odd && z[1]==0 ) break;
          }
          z++;
        }
        z += 2;
      }else{
        while( *z==0 ) z++;
      }
      z += (n + 8 + 1);
      pFd->zWal = z;
      pFd->zWal = rbuMainToWal(zName, flags);
    }
    else if( flags & SQLITE_OPEN_WAL ){
      rbu_file *pDb = rbuFindMaindb(pRbuVfs, zName);
      if( pDb ){
        if( pDb->pRbu && pDb->pRbu->eStage==RBU_STAGE_OAL ){
          /* This call is to open a *-wal file. Intead, open the *-oal. This
          ** code ensures that the string passed to xOpen() is terminated by a
          ** pair of '\0' bytes in case the VFS attempts to extract a URI 
          ** parameter from it.  */
          const char *zBase = zName;
          size_t nCopy = strlen(zName);
          char *zCopy = sqlite3_malloc64(nCopy+2);
          size_t nCopy;
          char *zCopy;
          if( rbuIsVacuum(pDb->pRbu) ){
            zBase = sqlite3_db_filename(pDb->pRbu->dbRbu, "main");
            zBase = rbuMainToWal(zBase, SQLITE_OPEN_URI);
          }
          nCopy = strlen(zBase);
          zCopy = sqlite3_malloc64(nCopy+2);
          if( zCopy ){
            memcpy(zCopy, zName, nCopy);
            memcpy(zCopy, zBase, nCopy);
            zCopy[nCopy-3] = 'o';
            zCopy[nCopy] = '\0';
            zCopy[nCopy+1] = '\0';
            zOpen = (const char*)(pFd->zDel = zCopy);
          }else{
            rc = SQLITE_NOMEM;
          }
          pFd->pRbu = pDb->pRbu;
        }
        pDb->pWalFd = pFd;
      }
    }
  }

  if( oflags & SQLITE_OPEN_MAIN_DB 
   && sqlite3_uri_boolean(zName, "rbu_memory", 0) 
  ){
    assert( oflags & SQLITE_OPEN_MAIN_DB );
    oflags =  SQLITE_OPEN_TEMP_DB | SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE |
              SQLITE_OPEN_EXCLUSIVE | SQLITE_OPEN_DELETEONCLOSE;
    zOpen = 0;
  }

  if( rc==SQLITE_OK ){
    rc = pRealVfs->xOpen(pRealVfs, zOpen, pFd->pReal, flags, pOutFlags);
    rc = pRealVfs->xOpen(pRealVfs, zOpen, pFd->pReal, oflags, pOutFlags);
  }
  if( pFd->pReal->pMethods ){
    /* The xOpen() operation has succeeded. Set the sqlite3_file.pMethods
    ** pointer and, if the file is a main database file, link it into the
    ** mutex protected linked list of all such files.  */
    pFile->pMethods = &rbuvfs_io_methods;
    if( flags & SQLITE_OPEN_MAIN_DB ){

** the zipvfs_create_vfs() API below for details on using RBU with zipvfs.
*/
sqlite3rbu *sqlite3rbu_open(
  const char *zTarget, 
  const char *zRbu,
  const char *zState
);

/*
** Open an RBU handle to perform an RBU vacuum on database file zTarget.
** An RBU vacuum is similar to SQLite's built-in VACUUM command, except
** that it can be suspended and resumed like an RBU update.
**
** The second argument to this function, which may not be NULL, identifies 
** a database in which to store the state of the RBU vacuum operation if
** it is suspended. The first time sqlite3rbu_vacuum() is called, to start
** an RBU vacuum operation, the state database should either not exist or
** be empty (contain no tables). If an RBU vacuum is suspended by calling
** sqlite3rbu_close() on the RBU handle before sqlite3rbu_step() has
** returned SQLITE_DONE, the vacuum state is stored in the state database. 
** The vacuum can be resumed by calling this function to open a new RBU
** handle specifying the same target and state databases.
**
** This function does not delete the state database after an RBU vacuum
** is completed, even if it created it. However, if the call to
** sqlite3rbu_close() returns any value other than SQLITE_OK, the contents
** of the state tables within the state database are zeroed. This way,
** the next call to sqlite3rbu_vacuum() opens a handle that starts a 
** new RBU vacuum operation.
**
** As with sqlite3rbu_open(), Zipvfs users should rever to the comment
** describing the sqlite3rbu_create_vfs() API function below for 
** a description of the complications associated with using RBU with 
** zipvfs databases.
*/
sqlite3rbu *sqlite3rbu_vacuum(
  const char *zTarget, 
  const char *zState
);

/*
** Internally, each RBU connection uses a separate SQLite database 
** connection to access the target and rbu update databases. This
** API allows the application direct access to these database handles.
**
** The first argument passed to this function must be a valid, open, RBU

  if( objc==5 ) zStateDb = Tcl_GetString(objv[4]);

  pRbu = sqlite3rbu_open(zTarget, zRbu, zStateDb);
  Tcl_CreateObjCommand(interp, zCmd, test_sqlite3rbu_cmd, (ClientData)pRbu, 0);
  Tcl_SetObjResult(interp, objv[1]);
  return TCL_OK;
}

/*
** Tclcmd: sqlite3rbu_vacuum CMD <target-db> <state-db>
*/
static int test_sqlite3rbu_vacuum(
  ClientData clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  sqlite3rbu *pRbu = 0;
  const char *zCmd;
  const char *zTarget;
  const char *zStateDb = 0;

  if( objc!=4 ){
    Tcl_WrongNumArgs(interp, 1, objv, "NAME TARGET-DB STATE-DB");
    return TCL_ERROR;
  }
  zCmd = Tcl_GetString(objv[1]);
  zTarget = Tcl_GetString(objv[2]);
  zStateDb = Tcl_GetString(objv[3]);

  pRbu = sqlite3rbu_vacuum(zTarget, zStateDb);
  Tcl_CreateObjCommand(interp, zCmd, test_sqlite3rbu_cmd, (ClientData)pRbu, 0);
  Tcl_SetObjResult(interp, objv[1]);
  return TCL_OK;
}

/*
** Tclcmd: sqlite3rbu_create_vfs ?-default? NAME PARENT
*/
static int test_sqlite3rbu_create_vfs(
  ClientData clientData,
  Tcl_Interp *interp,

int SqliteRbu_Init(Tcl_Interp *interp){ 
  static struct {
     char *zName;
     Tcl_ObjCmdProc *xProc;
  } aObjCmd[] = {
    { "sqlite3rbu", test_sqlite3rbu },
    { "sqlite3rbu_vacuum", test_sqlite3rbu_vacuum },
    { "sqlite3rbu_create_vfs", test_sqlite3rbu_create_vfs },
    { "sqlite3rbu_destroy_vfs", test_sqlite3rbu_destroy_vfs },
    { "sqlite3rbu_internal_test", test_sqlite3rbu_internal_test },
  };
  int i;
  for(i=0; i<sizeof(aObjCmd)/sizeof(aObjCmd[0]); i++){
    Tcl_CreateObjCommand(interp, aObjCmd[i].zName, aObjCmd[i].xProc, 0, 0);

    if( pParse->nTempReg<ArraySize(pParse->aTempReg) ){
      pParse->aTempReg[pParse->nTempReg++] = p->iReg;
    }
    p->tempReg = 0;
  }
  p->iReg = 0;
  pParse->nColCache--;
  assert( cacheIsValid(pParse) );
  assert( pParse->db->mallocFailed || cacheIsValid(pParse) );
}


/*
** Record in the column cache that a particular column from a
** particular table is stored in a particular register.
*/

      p->iLevel = pParse->iCacheLevel;
      p->iTable = iTab;
      p->iColumn = iCol;
      p->iReg = iReg;
      p->tempReg = 0;
      p->lru = pParse->iCacheCnt++;
      pParse->nColCache++;
      assert( cacheIsValid(pParse) );
      assert( pParse->db->mallocFailed || cacheIsValid(pParse) );
      return;
    }
  }

  /* Replace the last recently used */
  minLru = 0x7fffffff;
  idxLru = -1;