memset(pNew, 0, sizeof(KVLsm));
    pNew->base.pStoreVfunc = &kvlsmMethods;
    pNew->base.pEnv = pEnv;
    rc = lsm_new(0, &pNew->pDb);
    if( rc==SQLITE4_OK ){
      int i;


      for(i=0; i<ArraySize(aConfig); i++){
        const char *zVal = sqlite4_uri_parameter(zName, aConfig[i].zParam);
        if( zVal ){
          int nVal = sqlite4Atoi(zVal);
          lsm_config(pNew->pDb, aConfig[i].eParam, &nVal);
        }
      }

    memset(pNew, 0, sizeof(KVLsm));
    pNew->base.pStoreVfunc = &kvlsmMethods;
    pNew->base.pEnv = pEnv;
    rc = lsm_new(0, &pNew->pDb);
    if( rc==SQLITE4_OK ){
      int i;
      int bMmap = 0;
      lsm_config(pNew->pDb, LSM_CONFIG_MMAP, &bMmap);
      for(i=0; i<ArraySize(aConfig); i++){
        const char *zVal = sqlite4_uri_parameter(zName, aConfig[i].zParam);
        if( zVal ){
          int nVal = sqlite4Atoi(zVal);
          lsm_config(pNew->pDb, aConfig[i].eParam, &nVal);
        }
      }

    multiCursorVisitFreelist(pCsr, pnOvfl);
    rc = multiCursorAddTree(pCsr, pDb->pWorker, eTree);
    if( rc==LSM_OK && pNext && pNext->pMerge==0 && pNext->lhs.iRoot ){
      pDel = &pNext->lhs;
      rc = btreeCursorNew(pDb, pDel, &pCsr->pBtCsr);
      iLeftPtr = pNext->lhs.iFirst;
    }




  }

  if( rc!=LSM_OK ){
    lsmMCursorClose(pCsr);
  }else{
    Merge merge;                  /* Merge object used to create new level */
    MergeWorker mergeworker;      /* MergeWorker object for the same purpose */
................................................................................
#endif
  return rc;
}

/*
** The database connection passed as the first argument must be a worker
** connection. This function checks if there exists an "old" in-memory tree
** ready to be flushed to disk. If so, *pbOut is set to true before 
** returning. Otherwise false.
**

** Normally, LSM_OK is returned. Or, if an error occurs, an LSM error code.
*/
static int sortedTreeHasOld(lsm_db *pDb, int *pbOut){
  int rc = LSM_OK;


  assert( pDb->pWorker );

  if( pDb->nTransOpen==0 ){
    rc = lsmTreeLoadHeader(pDb, 0);
  }

  if( rc==LSM_OK 
   && pDb->treehdr.iOldShmid
   && pDb->treehdr.iOldLog!=pDb->pWorker->iLogOff 
  ){
    *pbOut = 1;
  }else{
    *pbOut = 0;
  }



  return rc;
}

static int doLsmSingleWork(
  lsm_db *pDb, 
  int bShutdown,
  int flags, 
  int nPage,                      /* Number of pages to write to disk */
................................................................................
    nMax = (pDb->nAutockpt/nPgsz) - (nUnsync-nSync);
    if( nMax<nRem ){
      bCkpt = 1;
      nRem = LSM_MAX(nMax, 0);
    }
  }

  /* If the FLUSH flag is set, there exists in-memory ready to be flushed
  ** to disk and there are lsm_db.nMerge or fewer age=0 levels, flush the 
  ** data to disk now.  */
  if( (flags & LSM_WORK_FLUSH) ){
    int bOld;
    rc = sortedTreeHasOld(pDb, &bOld);
    if( bOld ){
      if( sortedDbIsFull(pDb) ){
        int nPg = 0;
        rc = sortedWork(pDb, nRem, 0, 1, &nPg);
        nRem -= nPg;
        assert( rc!=LSM_OK || nRem<=0 || !sortedDbIsFull(pDb) );
        bToplevel = 1;
      }

      if( rc==LSM_OK && nRem>0 ){
        int nPg = 0;
        rc = sortedNewToplevel(pDb, TREE_OLD, &nOvfl, &nPg);
        nRem -= nPg;
        if( rc==LSM_OK && pDb->nTransOpen>0 ){
          lsmTreeDiscardOld(pDb);
        }
        bFlush = 1;
        bToplevel = 0;
      }
    }
  }

  /* If nPage is still greater than zero, do some merging. */
  if( rc==LSM_OK && nRem>0 && bShutdown==0 ){
    int nPg = 0;
    int bOptimize = ((flags & LSM_WORK_OPTIMIZE) ? 1 : 0);

    multiCursorVisitFreelist(pCsr, pnOvfl);
    rc = multiCursorAddTree(pCsr, pDb->pWorker, eTree);
    if( rc==LSM_OK && pNext && pNext->pMerge==0 && pNext->lhs.iRoot ){
      pDel = &pNext->lhs;
      rc = btreeCursorNew(pDb, pDel, &pCsr->pBtCsr);
      iLeftPtr = pNext->lhs.iFirst;
    }

    if( pNext==0 ){
      multiCursorIgnoreDelete(pCsr);
    }
  }

  if( rc!=LSM_OK ){
    lsmMCursorClose(pCsr);
  }else{
    Merge merge;                  /* Merge object used to create new level */
    MergeWorker mergeworker;      /* MergeWorker object for the same purpose */
................................................................................
#endif
  return rc;
}

/*
** The database connection passed as the first argument must be a worker
** connection. This function checks if there exists an "old" in-memory tree
** ready to be flushed to disk. If so, true is returned. Otherwise false.

**
** If an error occurs, *pRc is set to an LSM error code before returning.
** It is assumed that *pRc is set to LSM_OK when this function is called.
*/
static int sortedTreeHasOld(lsm_db *pDb, int *pRc){
  int rc = LSM_OK;
  int bRet = 0;

  assert( pDb->pWorker );
  if( *pRc==LSM_OK ){
    if( pDb->nTransOpen==0 ){
      rc = lsmTreeLoadHeader(pDb, 0);
    }

    if( rc==LSM_OK 
        && pDb->treehdr.iOldShmid
        && pDb->treehdr.iOldLog!=pDb->pWorker->iLogOff 
      ){
      bRet = 1;
    }else{
      bRet = 0;
    }
    *pRc = rc;
  }
  assert( *pRc==LSM_OK || bRet==0 );
  return bRet;
}

static int doLsmSingleWork(
  lsm_db *pDb, 
  int bShutdown,
  int flags, 
  int nPage,                      /* Number of pages to write to disk */
................................................................................
    nMax = (pDb->nAutockpt/nPgsz) - (nUnsync-nSync);
    if( nMax<nRem ){
      bCkpt = 1;
      nRem = LSM_MAX(nMax, 0);
    }
  }

  /* If there exists in-memory data ready to be flushed to disk, attempt
  ** to flush it now.  */



  if( sortedTreeHasOld(pDb, &rc) ){

    if( sortedDbIsFull(pDb) ){
      int nPg = 0;
      rc = sortedWork(pDb, nRem, 0, 1, &nPg);
      nRem -= nPg;
      assert( rc!=LSM_OK || nRem<=0 || !sortedDbIsFull(pDb) );
      bToplevel = 1;
    }

    if( rc==LSM_OK && nRem>0 ){
      int nPg = 0;
      rc = sortedNewToplevel(pDb, TREE_OLD, &nOvfl, &nPg);
      nRem -= nPg;
      if( rc==LSM_OK && pDb->nTransOpen>0 ){
        lsmTreeDiscardOld(pDb);
      }
      bFlush = 1;
      bToplevel = 0;

    }
  }

  /* If nPage is still greater than zero, do some merging. */
  if( rc==LSM_OK && nRem>0 && bShutdown==0 ){
    int nPg = 0;
    int bOptimize = ((flags & LSM_WORK_OPTIMIZE) ? 1 : 0);

populate_db
do_execsql_test 2.1 { 
  BEGIN;
    INSERT INTO t1 VALUES(10, 100);
}
do_test 2.2 { 
  sqlite4 db2 ./test.db
  list [catch { sqlite4_lsm_work db2 main -flush 0 } msg] $msg
} {1 SQLITE4_BUSY}

do_execsql_test 2.3 { COMMIT }
do_test 2.4 { sqlite4_lsm_work db2 main -flush 0 } {0}
db2 close


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

populate_db
do_execsql_test 2.1 { 
  BEGIN;
    INSERT INTO t1 VALUES(10, 100);
}
do_test 2.2 { 
  sqlite4 db2 ./test.db
  list [catch { db2 eval { BEGIN ; INSERT INTO t1 VALUES(1, 2) } } msg] $msg
} {1 {database is locked}}

do_execsql_test 2.3 { COMMIT }
do_test 2.4 { sqlite4_lsm_work db2 main -flush 0 } {0}
db2 close


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

  execsql { SELECT count(*) FROM t1 } db2
} {256}
db2 close

reset_db
do_execsql_test 3.5 { CREATE TABLE t1(a, b) }
do_test 3.6 {
  sqlite4_lsm_work db main -flush -checkpoint 0
  for {set i 0} {$i < 203} {incr i} {
    execsql { INSERT INTO t1 VALUES(randstr(100,100), randstr(100,100)) }
  }
  execsql { SELECT count(*) FROM t1 }
} {203}

do_test 3.7 {
................................................................................
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
}
do_filesize_test 8.2   0 776

do_test          8.3 { sqlite4_lsm_work db main -flush } 0
do_execsql_test  8.4 { INSERT INTO x VALUES(randstr(10,10), randstr(100,100)) }
do_filesize_test 8.5   12288 915
do_test          8.6 { sqlite4_lsm_work db main -checkpoint } 0

do_test 8.7 {
  copy_db_files test.db test.db2
  sqlite4 db2 test.db2
  execsql { SELECT count(*) FROM x ; PRAGMA integrity_check } db2
} {6 ok}

................................................................................
}
do_test 10.7 { sqlite4_lsm_info db main log-structure } {0 0 0 0 0 556}
do_test 10.8 { sqlite4_lsm_work db main -flush } 0
do_execsql_test 10.9 {
  INSERT INTO t1 VALUES(randstr(10,10), randstr(100,100));
}
do_test 10.9  { sqlite4_lsm_info db main log-structure } {0 0 0 0 0 695}
do_test 10.10 { sqlite4_lsm_work db main -checkpoint } 0
do_test 10.11 { sqlite4_lsm_info db main log-structure } {0 0 0 0 556 695}

#-------------------------------------------------------------------------
#
reset_db
do_test         11.1 { sqlite4_lsm_config db main log-size 800 } 800
do_test         11.2 { sqlite4_lsm_config db main log-size     } 800
................................................................................

do_test 11.4 { sqlite4_lsm_info db main log-structure } {0 0 0 0 0 1335}
do_test 11.5 { sqlite4_lsm_work db main -flush } 0
do_execsql_test 11.6 {
  INSERT INTO t1 VALUES(randstr(10,10), randstr(100,100));
}
do_test 11.7 { sqlite4_lsm_info db main log-structure } {0 0 0 0 0 1474}
do_test 11.8 { sqlite4_lsm_work db main -checkpoint } 0
do_test 11.9 { sqlite4_lsm_info db main log-structure } {0 0 0 0 1335 1474}
do_execsql_test 11.10 {
  INSERT INTO t1 VALUES(randstr(10,10), randstr(100,100));
}
do_test 11.11 { sqlite4_lsm_info db main log-structure } {1335 1482 0 0 0 139}
do_test 11.12 {
  execsql { SELECT count(*) FROM t1 ; PRAGMA integrity_check } 
................................................................................
db eval {SELECT randstr(5,5)}
do_execsql_test 11.22 {
  INSERT INTO t1 VALUES(randstr(10,10), randstr(100,100));
}
do_test 11.23 { 
  sqlite4_lsm_info db main log-structure 
} {1335 1482 0 1259 1483 1908}
do_test 11.24 { sqlite4_lsm_work db main -checkpoint } {0}
do_test 11.25 { 
  sqlite4_lsm_info db main log-structure 
} {0 0 0 0 1769 1908}

#-------------------------------------------------------------------------
#
reset_db
................................................................................
}
for {set iTest 1} {$iTest<=150} {incr iTest} {
  expr srand(0)
  do_test 12.3.$iTest {
    for {set i 0} {$i < 10} {incr i} {
      execsql { INSERT INTO t1 VALUES(randstr(20,20), randstr(100,100)) }
      if { int(rand()*10.0)==0 } { sqlite4_lsm_work db main -flush }
      if { int(rand()*10.0)==0 } { sqlite4_lsm_work db main -checkpoint }
    }
    copy_db_files test.db test.db2
    sqlite4 db2 test.db2
    set sql "SELECT count(*) FROM t1 ; "
    if {0==($iTest % 25)} {
      append sql "PRAGMA integrity_check"
    } else {

  execsql { SELECT count(*) FROM t1 } db2
} {256}
db2 close

reset_db
do_execsql_test 3.5 { CREATE TABLE t1(a, b) }
do_test 3.6 {
  sqlite4_lsm_checkpoint db main
  for {set i 0} {$i < 203} {incr i} {
    execsql { INSERT INTO t1 VALUES(randstr(100,100), randstr(100,100)) }
  }
  execsql { SELECT count(*) FROM t1 }
} {203}

do_test 3.7 {
................................................................................
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
}
do_filesize_test 8.2   0 776
do_test          8.3.1 { sqlite4_lsm_flush db main } {}
do_test          8.3.2 { sqlite4_lsm_work db main } 0
do_execsql_test  8.4 { INSERT INTO x VALUES(randstr(10,10), randstr(100,100)) }
do_filesize_test 8.5   12288 915
do_test          8.6 { sqlite4_lsm_checkpoint db main } {}

do_test 8.7 {
  copy_db_files test.db test.db2
  sqlite4 db2 test.db2
  execsql { SELECT count(*) FROM x ; PRAGMA integrity_check } db2
} {6 ok}

................................................................................
}
do_test 10.7 { sqlite4_lsm_info db main log-structure } {0 0 0 0 0 556}
do_test 10.8 { sqlite4_lsm_work db main -flush } 0
do_execsql_test 10.9 {
  INSERT INTO t1 VALUES(randstr(10,10), randstr(100,100));
}
do_test 10.9  { sqlite4_lsm_info db main log-structure } {0 0 0 0 0 695}
do_test 10.10 { sqlite4_lsm_checkpoint db main } {}
do_test 10.11 { sqlite4_lsm_info db main log-structure } {0 0 0 0 556 695}

#-------------------------------------------------------------------------
#
reset_db
do_test         11.1 { sqlite4_lsm_config db main log-size 800 } 800
do_test         11.2 { sqlite4_lsm_config db main log-size     } 800
................................................................................

do_test 11.4 { sqlite4_lsm_info db main log-structure } {0 0 0 0 0 1335}
do_test 11.5 { sqlite4_lsm_work db main -flush } 0
do_execsql_test 11.6 {
  INSERT INTO t1 VALUES(randstr(10,10), randstr(100,100));
}
do_test 11.7 { sqlite4_lsm_info db main log-structure } {0 0 0 0 0 1474}
do_test 11.8 { sqlite4_lsm_checkpoint db main } {}
do_test 11.9 { sqlite4_lsm_info db main log-structure } {0 0 0 0 1335 1474}
do_execsql_test 11.10 {
  INSERT INTO t1 VALUES(randstr(10,10), randstr(100,100));
}
do_test 11.11 { sqlite4_lsm_info db main log-structure } {1335 1482 0 0 0 139}
do_test 11.12 {
  execsql { SELECT count(*) FROM t1 ; PRAGMA integrity_check } 
................................................................................
db eval {SELECT randstr(5,5)}
do_execsql_test 11.22 {
  INSERT INTO t1 VALUES(randstr(10,10), randstr(100,100));
}
do_test 11.23 { 
  sqlite4_lsm_info db main log-structure 
} {1335 1482 0 1259 1483 1908}
do_test 11.24 { sqlite4_lsm_checkpoint db main } {}
do_test 11.25 { 
  sqlite4_lsm_info db main log-structure 
} {0 0 0 0 1769 1908}

#-------------------------------------------------------------------------
#
reset_db
................................................................................
}
for {set iTest 1} {$iTest<=150} {incr iTest} {
  expr srand(0)
  do_test 12.3.$iTest {
    for {set i 0} {$i < 10} {incr i} {
      execsql { INSERT INTO t1 VALUES(randstr(20,20), randstr(100,100)) }
      if { int(rand()*10.0)==0 } { sqlite4_lsm_work db main -flush }
      if { int(rand()*10.0)==0 } { sqlite4_lsm_checkpoint db main }
    }
    copy_db_files test.db test.db2
    sqlite4 db2 test.db2
    set sql "SELECT count(*) FROM t1 ; "
    if {0==($iTest % 25)} {
      append sql "PRAGMA integrity_check"
    } else {

    INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50));
    INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50));
    INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50));
  COMMIT;
} {}
do_filesize_test 2.5   0 2048

do_test         2.6 { sqlite4_lsm_work db main -flush 0 } {0}
do_execsql_test 2.7 { INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50)) }
do_test         2.8 { sqlite4_lsm_work db main -check 0 } {0}
do_test 2.9 { sqlite4_lsm_info db main log-structure } {0 0 0 0 2048 2560}

for {set i 1} {$i <= 6} {incr i} {
  do_execsql_test 2.10.$i.1 {
    INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50));
  }
  do_execsql_test 2.10.$i.2 { SELECT count(*) FROM t1 } [expr 8 + $i]
  do_recover_test 2.10.$i.3 { SELECT count(*) FROM t1 } [expr 8 + $i]
}

do_test 2.11 { 
  sqlite4_lsm_info db main log-structure 
} {2048 2568 0 1704 3072 4608}


finish_test

    INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50));
    INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50));
    INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50));
  COMMIT;
} {}
do_filesize_test 2.5   0 2048

do_test         2.6 { sqlite4_lsm_flush db main } {}
do_execsql_test 2.7 { INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50)) }
do_test         2.8 { sqlite4_lsm_checkpoint db main } {}
do_test 2.9 { sqlite4_lsm_info db main log-structure } {0 0 0 0 2560 3072}

for {set i 1} {$i <= 6} {incr i} {
  do_execsql_test 2.10.$i.1 {
    INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50));
  }
  do_execsql_test 2.10.$i.2 { SELECT count(*) FROM t1 } [expr 8 + $i]
  do_recover_test 2.10.$i.3 { SELECT count(*) FROM t1 } [expr 8 + $i]
}

do_test 2.11 { 
  sqlite4_lsm_info db main log-structure 
} {2560 3080 0 2216 3584 4608}


finish_test

#   full
#
lappend ::testsuitelist xxx

test_suite "src4" -prefix "" -description {
} -files {
  simple.test simple2.test
  log1.test log2.test log3.test 
  csr1.test
  ckpt1.test
  mc1.test

  aggerror.test
  attach.test
  autoindex1.test

#   full
#
lappend ::testsuitelist xxx

test_suite "src4" -prefix "" -description {
} -files {
  simple.test simple2.test
  log3.test 
  csr1.test
  ckpt1.test
  mc1.test

  aggerror.test
  attach.test
  autoindex1.test

do_catchsql_test 70.3 {
  select * from maintable, joinme INDEXED by joinme_id_text_idx
} {1 {cannot use index: joinme_id_text_idx}}

#-------------------------------------------------------------------------
# This is testing that the "phantom" runs feature works.



reset_db
do_execsql_test 71.1 {
  CREATE TABLE t1(x);
  INSERT INTO t1 VALUES(randomblob(1024));           --   1
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   2
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   4
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   8
................................................................................
do_execsql_test 71.3 { SELECT count(*) FROM t1 } 64
do_test 71.4 { 
  expr {[file size test.db] < 256*1024}
} {1}

#-------------------------------------------------------------------------
# This is testing that the "phantom" runs feature works with mmap.



reset_db

do_test 72.0.1 { sqlite4_lsm_config db main mmap   } 0
do_test 72.0.2 { sqlite4_lsm_config db main mmap 1 } 1
do_test 72.0.3 { sqlite4_lsm_config db main mmap   } 1

do_execsql_test 72.1 {
  CREATE TABLE t1(x);
  INSERT INTO t1 VALUES(randomblob(1024));           --   1
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   2
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   4
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   8

do_catchsql_test 70.3 {
  select * from maintable, joinme INDEXED by joinme_id_text_idx
} {1 {cannot use index: joinme_id_text_idx}}

#-------------------------------------------------------------------------
# This is testing that the "phantom" runs feature works.
#
# UPDATE: Said feature was dropped early in development. But the test 
# remains valid.
reset_db
do_execsql_test 71.1 {
  CREATE TABLE t1(x);
  INSERT INTO t1 VALUES(randomblob(1024));           --   1
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   2
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   4
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   8
................................................................................
do_execsql_test 71.3 { SELECT count(*) FROM t1 } 64
do_test 71.4 { 
  expr {[file size test.db] < 256*1024}
} {1}

#-------------------------------------------------------------------------
# This is testing that the "phantom" runs feature works with mmap.
#
# UPDATE: Said feature was dropped early in development. But the test 
# remains valid.
reset_db

#do_test 72.0.1 { sqlite4_lsm_config db main mmap   } 0
#do_test 72.0.2 { sqlite4_lsm_config db main mmap 1 } 1
#do_test 72.0.3 { sqlite4_lsm_config db main mmap   } 1

do_execsql_test 72.1 {
  CREATE TABLE t1(x);
  INSERT INTO t1 VALUES(randomblob(1024));           --   1
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   2
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   4
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   8

  Tcl_Obj *CONST objv[]
){
  struct Switch {
    const char *zSwitch;
    int flags;
  } aSwitch[] = {
    { "-flush",      LSM_WORK_FLUSH }, 
    { "-checkpoint", LSM_WORK_CHECKPOINT }, 
    { "-optimize",   LSM_WORK_OPTIMIZE }, 
    { 0, 0 }
  };

  int flags = 0;
  int nPage = 0;
  const char *zDb;
................................................................................
    Tcl_SetResult(interp, (char *)sqlite4TestErrorName(rc), TCL_STATIC);
    return TCL_ERROR;
  }

  Tcl_SetObjResult(interp, Tcl_NewIntObj(nWork));
  return TCL_OK;
}



















































































int SqlitetestLsm_Init(Tcl_Interp *interp){
  struct SyscallCmd {
    const char *zName;
    Tcl_ObjCmdProc *xCmd;
  } aCmd[] = {
    { "sqlite4_lsm_work",     test_lsm_work                },


    { "sqlite4_lsm_info",     test_lsm_info                },
    { "sqlite4_lsm_config",   test_lsm_config              },
  };
  int i;

  for(i=0; i<sizeof(aCmd)/sizeof(aCmd[0]); i++){
    Tcl_CreateObjCommand(interp, aCmd[i].zName, aCmd[i].xCmd, 0, 0);
  }
  return TCL_OK;
}

  Tcl_Obj *CONST objv[]
){
  struct Switch {
    const char *zSwitch;
    int flags;
  } aSwitch[] = {
    { "-flush",      LSM_WORK_FLUSH }, 

    { "-optimize",   LSM_WORK_OPTIMIZE }, 
    { 0, 0 }
  };

  int flags = 0;
  int nPage = 0;
  const char *zDb;
................................................................................
    Tcl_SetResult(interp, (char *)sqlite4TestErrorName(rc), TCL_STATIC);
    return TCL_ERROR;
  }

  Tcl_SetObjResult(interp, Tcl_NewIntObj(nWork));
  return TCL_OK;
}

/*
** TCLCMD:    sqlite4_lsm_checkpoint DB DBNAME 
*/
static int test_lsm_checkpoint(
  void * clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  const char *zDb;
  const char *zName;
  int rc;
  sqlite4 *db;
  lsm_db *pLsm;

  if( objc!=3 ){
    Tcl_WrongNumArgs(interp, 1, objv, "DB DBNAME");
    return TCL_ERROR;
  }
  zDb = Tcl_GetString(objv[1]);
  zName = Tcl_GetString(objv[2]);

  rc = getDbPointer(interp, zDb, &db);
  if( rc!=TCL_OK ) return rc;

  rc = sqlite4_kvstore_control(db, zName, SQLITE4_KVCTRL_LSM_HANDLE, &pLsm);
  if( rc==SQLITE4_OK ){
    rc = lsm_checkpoint(pLsm, 0);
  }
  if( rc!=SQLITE4_OK ){
    Tcl_SetResult(interp, (char *)sqlite4TestErrorName(rc), TCL_STATIC);
    return TCL_ERROR;
  }

  Tcl_ResetResult(interp);
  return TCL_OK;
}

/*
** TCLCMD:    sqlite4_lsm_flush DB DBNAME 
*/
static int test_lsm_flush(
  void * clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  const char *zDb;
  const char *zName;
  int rc;
  sqlite4 *db;
  lsm_db *pLsm;

  if( objc!=3 ){
    Tcl_WrongNumArgs(interp, 1, objv, "DB DBNAME");
    return TCL_ERROR;
  }
  zDb = Tcl_GetString(objv[1]);
  zName = Tcl_GetString(objv[2]);

  rc = getDbPointer(interp, zDb, &db);
  if( rc!=TCL_OK ) return rc;

  rc = sqlite4_kvstore_control(db, zName, SQLITE4_KVCTRL_LSM_HANDLE, &pLsm);
  if( rc==SQLITE4_OK ){
    int nZero = 0;
    int nOrig = -1;
    lsm_config(pLsm, LSM_CONFIG_WRITE_BUFFER, &nOrig);
    lsm_config(pLsm, LSM_CONFIG_WRITE_BUFFER, &nZero);
    rc = lsm_begin(pLsm, 1);
    if( rc==LSM_OK ) rc = lsm_commit(pLsm, 0);
    lsm_config(pLsm, LSM_CONFIG_WRITE_BUFFER, &nOrig);
  }
  if( rc!=SQLITE4_OK ){
    Tcl_SetResult(interp, (char *)sqlite4TestErrorName(rc), TCL_STATIC);
    return TCL_ERROR;
  }

  Tcl_ResetResult(interp);
  return TCL_OK;
}

int SqlitetestLsm_Init(Tcl_Interp *interp){
  struct SyscallCmd {
    const char *zName;
    Tcl_ObjCmdProc *xCmd;
  } aCmd[] = {
    { "sqlite4_lsm_work",       test_lsm_work                },
    { "sqlite4_lsm_checkpoint", test_lsm_checkpoint          },
    { "sqlite4_lsm_flush",      test_lsm_flush               },
    { "sqlite4_lsm_info",       test_lsm_info                },
    { "sqlite4_lsm_config",     test_lsm_config              },
  };
  int i;

  for(i=0; i<sizeof(aCmd)/sizeof(aCmd[0]); i++){
    Tcl_CreateObjCommand(interp, aCmd[i].zName, aCmd[i].xCmd, 0, 0);
  }
  return TCL_OK;
}

# Flush the in-memory tree to disk and merge all runs together into
# a single b-tree structure. Because this annihilates all delete keys,
# the next rowid allocated for each table with an IPK will be as expected
# by SQLite 3 tests.
#
proc optimize_db {} { 


  catch { sqlite4_lsm_work db main -checkpoint -opt -flush 100000 }


  return ""
}


# If the library is compiled with the SQLITE4_DEFAULT_AUTOVACUUM macro set
# to non-zero, then set the global variable $AUTOVACUUM to 1.
set AUTOVACUUM $sqlite_options(default_autovacuum)

source $testdir/malloc_common.tcl

# Flush the in-memory tree to disk and merge all runs together into
# a single b-tree structure. Because this annihilates all delete keys,
# the next rowid allocated for each table with an IPK will be as expected
# by SQLite 3 tests.
#
proc optimize_db {} { 
  #catch { 
    sqlite4_lsm_flush db main 
    sqlite4_lsm_work db main -opt -flush 100000 
    sqlite4_lsm_checkpoint db main
  #}
  return ""
}


# If the library is compiled with the SQLITE4_DEFAULT_AUTOVACUUM macro set
# to non-zero, then set the global variable $AUTOVACUUM to 1.
set AUTOVACUUM $sqlite_options(default_autovacuum)

source $testdir/malloc_common.tcl