SQLite4  Check-in [eec16b0f2f]

              the system recovers and other clients proceed unaffected if
              a process fails in the middle of a write transaction.

              The difference from lsmtest2.c is that this file tests
              live-recovery (recovery from a failure that occurs while other
              clients are still running) whereas lsmtest2.c tests recovery
              from a system or power failure.

  lsmtest9.c: More data tests. These focus on testing that calling
              lsm_work(nMerge=1) to compact the database does not corrupt it.
              In other words, that databases containing block-redirects
              can be read and written.

void test_data_1(const char *, const char *, int *pRc);
void test_data_2(const char *, const char *, int *pRc);
void test_data_3(const char *, const char *, int *pRc);
void testDbContents(TestDb *, Datasource *, int, int, int, int, int, int *);
void testCaseProgress(int, int, int, int *);
int testCaseNDot(void);

void testCompareDb(Datasource *, int, int, TestDb *, TestDb *, int *);
int testControlDb(TestDb **ppDb);

typedef struct CksumDb CksumDb;
CksumDb *testCksumArrayNew(Datasource *, int, int, int);
char *testCksumArrayGet(CksumDb *, int);
void testCksumArrayFree(CksumDb *);
void testCaseStart(int *pRc, char *zFmt, ...);
void testCaseFinish(int rc);
void testCaseSkip(void);
int testCaseBegin(int *, const char *, const char *, ...);

#define TEST_CKSUM_BYTES 29
int testCksumDatabase(TestDb *pDb, char *zOut);
int testCountDatabase(TestDb *pDb);
void testCompareInt(int, int, int *);
void testCompareStr(const char *z1, const char *z2, int *pRc);

/* lsmtest9.c */
void test_data_4(const char *, const char *, int *pRc);


/*
** Similar to the Tcl_GetIndexFromObjStruct() Tcl library function.
*/
#define testArgSelect(w,x,y,z) testArgSelectX(w,x,sizeof(w[0]),y,z)
int testArgSelectX(void *, const char *, int, const char *, int *);

#ifdef __cplusplus
}  /* End of the 'extern "C"' block */
#endif

#endif

  zRet = testMallocPrintf("data.%s.%s.%d.%d", 
      zSystem, zData, pTest->nRow, pTest->nVerify
  );
  testFree(zData);
  return zRet;
}

int testControlDb(TestDb **ppDb){
#ifdef HAVE_KYOTOCABINET
  return tdb_open("kyotocabinet", "tmp.db", 1, ppDb);
#else
  return tdb_open("sqlite3", ":memory:", 1, ppDb);
#endif
}

    if( testCaseBegin(pRc, zPattern, "%s", zName) ){
      doDataTest1(zSystem, &aTest[i], pRc);
    }
    testFree(zName);
  }
}

void testCompareDb(
  Datasource *pData,
  int nData,
  int iSeed,
  TestDb *pControl,
  TestDb *pDb,
  int *pRc
){

#include "lsmtest.h"

#define DATA_SEQUENTIAL TEST_DATASOURCE_SEQUENCE
#define DATA_RANDOM     TEST_DATASOURCE_RANDOM

typedef struct Datatest4 Datatest4;

/*
** Test overview:
**
**   1. Insert (Datatest4.nRec) records into a database.
**
**   2. Repeat (Datatest4.nRepeat) times:
**
**      2a. Delete 2/3 of the records in the database.
**
**      2b. Run lsm_work(nMerge=1).
**
**      2c. Insert as many records as were deleted in 2a.
**
**      2d. Check database content is as expected.
**
**      2e. If (Datatest4.bReopen) is true, close and reopen the database.
*/
struct Datatest4 {
  /* Datasource definition */
  DatasourceDefn defn;

  int nRec;
  int nRepeat;
  int bReopen;
};

static void doDataTest4(
  const char *zSystem,            /* Database system to test */
  Datatest4 *p,                   /* Structure containing test parameters */
  int *pRc                        /* OUT: Error code */
){
  lsm_db *db = 0;
  TestDb *pDb;
  TestDb *pControl;
  Datasource *pData;
  int i;
  int rc = 0;
  int iDot = 0;

  int nRecOn3 = (p->nRec / 3);
  int iData = 0;

  /* Start the test case, open a database and allocate the datasource. */
  rc = testControlDb(&pControl);
  pDb = testOpen(zSystem, 1, &rc);
  pData = testDatasourceNew(&p->defn);
  if( rc==0 ) db = tdb_lsm(pDb);

  testWriteDatasourceRange(pControl, pData, iData, nRecOn3*3, &rc);
  testWriteDatasourceRange(pDb,      pData, iData, nRecOn3*3, &rc);

  for(i=0; rc==0 && i<p->nRepeat; i++){

    testDeleteDatasourceRange(pControl, pData, iData, nRecOn3*2, &rc);
    testDeleteDatasourceRange(pDb,      pData, iData, nRecOn3*2, &rc);

    if( db ){
      int nDone;
      do {
        nDone = 0;
        rc = lsm_work(db, 1, 100000, &nDone);
      }while( rc==0 && nDone>0 );
    }

    iData += (nRecOn3*2);
    testWriteDatasourceRange(pControl, pData, iData+nRecOn3, nRecOn3*2, &rc);
    testWriteDatasourceRange(pDb,      pData, iData+nRecOn3, nRecOn3*2, &rc);

    testCompareDb(pData, nRecOn3*3, iData, pControl, pDb, &rc);

    /* If Datatest4.bReopen is true, close and reopen the database */
    if( p->bReopen ){
      testReopen(&pDb, &rc);
      if( rc==0 ) db = tdb_lsm(pDb);
    }

    /* Update the progress dots... */
    testCaseProgress(i, p->nRepeat, testCaseNDot(), &iDot);
  }

  testClose(&pDb);
  testClose(&pControl);
  testDatasourceFree(pData);
  testCaseFinish(rc);
  *pRc = rc;
}

static char *getName4(const char *zSystem, Datatest4 *pTest){
  char *zRet;
  char *zData;
  zData = testDatasourceName(&pTest->defn);
  zRet = testMallocPrintf("data4.%s.%s.%d.%d.%d", 
      zSystem, zData, pTest->nRec, pTest->nRepeat, pTest->bReopen
  );
  testFree(zData);
  return zRet;
}

void test_data_4(
  const char *zSystem,            /* Database system name */
  const char *zPattern,           /* Run test cases that match this pattern */
  int *pRc                        /* IN/OUT: Error code */
){
  Datatest4 aTest[] = {
      /* defn,                                 nRec, nRepeat, bReopen */
    { {DATA_RANDOM,     20,25,     100,200}, 10000,      10,       0   },
    { {DATA_RANDOM,     20,25,     100,200}, 10000,      10,       1   },
  };

  int i;

  for(i=0; *pRc==LSM_OK && i<ArraySize(aTest); i++){
    char *zName = getName4(zSystem, &aTest[i]);
    if( testCaseBegin(pRc, zPattern, "%s", zName) ){
      doDataTest4(zSystem, &aTest[i], pRc);
    }
    testFree(zName);
  }
}

  for(j=0; tdb_system_name(j); j++){
    rc = 0;

    test_data_1(tdb_system_name(j), zPattern, &rc);
    test_data_2(tdb_system_name(j), zPattern, &rc);
    test_data_3(tdb_system_name(j), zPattern, &rc);
    test_data_4(tdb_system_name(j), zPattern, &rc);
    test_rollback(tdb_system_name(j), zPattern, &rc);
    test_mc(tdb_system_name(j), zPattern, &rc);
    test_mt(tdb_system_name(j), zPattern, &rc);

    if( rc ) nFail++;
  }

EXTHDR += \
  $(TOP)/ext/icu/sqliteicu.h

LSMTESTSRC = $(TOP)/lsm-test/lsmtest1.c $(TOP)/lsm-test/lsmtest2.c           \
             $(TOP)/lsm-test/lsmtest3.c $(TOP)/lsm-test/lsmtest4.c           \
             $(TOP)/lsm-test/lsmtest5.c $(TOP)/lsm-test/lsmtest6.c           \
             $(TOP)/lsm-test/lsmtest7.c $(TOP)/lsm-test/lsmtest8.c           \
             $(TOP)/lsm-test/lsmtest9.c                                      \
             $(TOP)/lsm-test/lsmtest_datasource.c \
             $(TOP)/lsm-test/lsmtest_func.c $(TOP)/lsm-test/lsmtest_io.c     \
             $(TOP)/lsm-test/lsmtest_main.c $(TOP)/lsm-test/lsmtest_mem.c    \
             $(TOP)/lsm-test/lsmtest_tdb.c $(TOP)/lsm-test/lsmtest_tdb3.c    \
             $(TOP)/lsm-test/lsmtest_util.c 

LSMTESTHDR = $(TOP)/lsm-test/lsmtest.h $(TOP)/lsm-test/lsmtest_tdb.h

void lsmEnvShmUnmap(lsm_env *, lsm_file *, int);

void lsmEnvSleep(lsm_env *, int);

int lsmFsReadSyncedId(lsm_db *db, int, i64 *piVal);

int lsmFsSegmentContainsPg(FileSystem *pFS, Segment *, Pgno, int *);
Pgno lsmFsRedirectPage(FileSystem *, Redirect *, Pgno);

/*
** End of functions from "lsm_file.c".
**************************************************************************/

/* 
** Functions from file "lsm_sorted.c".

    for(i=0; i<p->n; i++){
      if( iBlk==p->a[i].iFrom ) return p->a[i].iTo;
    }
  }
  return iBlk;
}

Pgno lsmFsRedirectPage(FileSystem *pFS, Redirect *pRedir, Pgno iPg){
  Pgno iReal = iPg;

  if( pRedir ){
    const int nPagePerBlock = (pFS->nBlocksize / pFS->nPagesize);
    int iBlk = fsPageToBlock(pFS, iPg);
    int i;
    for(i=0; i<pRedir->n; i++){

        rc = fsBlockNext(pFS, pDel, iBlk, &iNext);
      }else if( bZero==0 && pDel->iLastPg!=fsLastPageOnBlock(pFS, iLastBlk) ){
        break;
      }
      rc = fsFreeBlock(pFS, pSnapshot, pDel, iBlk);
      iBlk = iNext;
    }

    if( pDel->pRedirect ){
      assert( pDel->pRedirect==&pSnapshot->redirect );
      pSnapshot->redirect.n = 0;
    }

    if( bZero ) memset(pDel, 0, sizeof(Segment));
  }
  return LSM_OK;
}

static Pgno firstOnBlock(FileSystem *pFS, int iBlk, Pgno *aPgno, int nPgno){

        assert( pPg->flags & PAGE_HASPREV );
        iPg = fsLastPageOnBlock(pFS, lsmGetU32(&pPg->aData[-4]));
      }else{
        iPg--;
      }
    }else{
      if( pRun ){
        Pgno iLast = lsmFsRedirectPage(pFS, pRedir, pRun->iLastPg);
        if( iPg==iLast ){
          *ppNext = 0;
          return LSM_OK;
        }
      }

      if( fsIsLast(pFS, iPg) ){

      Pgno iFirst;                /* First page of segment (post-redirection) */
      int iBlk;                   /* Current block (during iteration) */
      int iLastBlk;               /* Last real block of segment */
      int bLastIsLastOnBlock;     /* True iLast is the last on its block */

      iBlk = fsRedirectBlock(pRedir, fsPageToBlock(pFS, pSeg->iFirst));
      iLastBlk = fsRedirectBlock(pRedir, fsPageToBlock(pFS, pSeg->iLastPg));
      iLast = lsmFsRedirectPage(pFS, pRedir, pSeg->iLastPg);
      iFirst = lsmFsRedirectPage(pFS, pRedir, pSeg->iFirst);

      bLastIsLastOnBlock = (fsLastPageOnBlock(pFS, iLastBlk)==iLast);
      assert( iBlk>0 );

      /* If the first page of this run is also the first page of its first
      ** block, set the flag to indicate that the first page of iBlk is 
      ** in use.  */

  int rc = LSM_OK;

  if( p->iPg ){
    lsm_env *pEnv = lsmFsEnv(pCsr->pFS);
    int iCell;                    /* Current cell number on leaf page */
    Pgno iLeaf;                   /* Page number of current leaf page */
    int nDepth;                   /* Depth of b-tree structure */
    Segment *pSeg = pCsr->pSeg;

    /* Decode the MergeInput structure */
    iLeaf = p->iPg;
    nDepth = (p->iCell & 0x00FF);
    iCell = (p->iCell >> 8) - 1;

    /* Allocate the BtreeCursor.aPg[] array */
    assert( pCsr->aPg==0 );
    pCsr->aPg = (BtreePg *)lsmMallocZeroRc(pEnv, sizeof(BtreePg) * nDepth, &rc);

    /* Populate the last entry of the aPg[] array */
    if( rc==LSM_OK ){
      Page **pp = &pCsr->aPg[nDepth-1].pPage;
      pCsr->iPg = nDepth-1;
      pCsr->nDepth = nDepth;
      pCsr->aPg[pCsr->iPg].iCell = iCell;
      rc = lsmFsDbPageGet(pCsr->pFS, pSeg, iLeaf, pp);
    }

    /* Populate any other aPg[] array entries */
    if( rc==LSM_OK && nDepth>1 ){
      Blob blob = {0,0,0};
      void *pSeek;
      int nSeek;
      int iTopicSeek;
      int iPg = 0;
      int iLoad = pSeg->iRoot;
      Page *pPg = pCsr->aPg[nDepth-1].pPage;
 
      if( pageObjGetNRec(pPg)==0 ){
        /* This can happen when pPg is the right-most leaf in the b-tree.
        ** In this case, set the iTopicSeek/pSeek/nSeek key to a value
        ** greater than any real key.  */
        assert( iCell==-1 );
        iTopicSeek = 1000;
        pSeek = 0;
        nSeek = 0;
      }else{
        Pgno dummy;
        rc = pageGetBtreeKey(pSeg, pPg,
            0, &dummy, &iTopicSeek, &pSeek, &nSeek, &pCsr->blob
        );
      }

      do {
        Page *pPg;
        rc = lsmFsDbPageGet(pCsr->pFS, pSeg, iLoad, &pPg);
        assert( rc==LSM_OK || pPg==0 );
        if( rc==LSM_OK ){
          u8 *aData;                  /* Buffer containing page data */
          int nData;                  /* Size of aData[] in bytes */
          int iMin;
          int iMax;
          int iCell;

            int iTry = (iMin+iMax)/2;
            void *pKey; int nKey;         /* Key for cell iTry */
            int iTopic;                   /* Topic for key pKeyT/nKeyT */
            Pgno iPtr;                    /* Pointer for cell iTry */
            int res;                      /* (pSeek - pKeyT) */

            rc = pageGetBtreeKey(
                pSeg, pPg, iTry, &iPtr, &iTopic, &pKey, &nKey, &blob
            );
            if( rc!=LSM_OK ) break;

            res = sortedKeyCompare(
                xCmp, iTopicSeek, pSeek, nSeek, iTopic, pKey, nKey
            );
            assert( res!=0 );

            if( res<0 ){
              iLoad = iPtr;
              iCell = iTry;
              iMax = iTry-1;
            }else{
              iMin = iTry+1;
            }
          }

          pCsr->aPg[iPg].pPage = pPg;
          pCsr->aPg[iPg].iCell = iCell;
          iPg++;
          assert( iPg!=nDepth-1 
               || lsmFsRedirectPage(pCsr->pFS, pSeg->pRedirect, iLoad)==iLeaf
          );
        }
      }while( rc==LSM_OK && iPg<(nDepth-1) );
      sortedBlobFree(&blob);
    }

    /* Load the current key and pointer */
    if( rc==LSM_OK ){

      if( pBtreePg->iCell<0 ){
        Pgno dummy;
        int i;
        for(i=pCsr->iPg-1; i>=0; i--){
          if( pCsr->aPg[i].iCell>0 ) break;
        }
        assert( i>=0 );
        rc = pageGetBtreeKey(pSeg,
            pCsr->aPg[i].pPage, pCsr->aPg[i].iCell-1,
            &dummy, &pCsr->eType, &pCsr->pKey, &pCsr->nKey, &pCsr->blob
        );
        pCsr->eType |= LSM_SEPARATOR;

      }else{
        rc = btreeCursorLoadKey(pCsr);

    if( rc==LSM_OK ){
      rc = sortedNewFreelistOnly(pDb);
    }
    nRem -= nPg;
    if( nPg ) bDirty = 1;
  }









  if( rc==LSM_OK ){
    *pnWrite = (nMax - nRem);
    *pbCkpt = (bCkpt && nRem<=0);
    if( nMerge==1 && pDb->nAutockpt>0 && *pnWrite>0
     && pWorker->pLevel 
     && pWorker->pLevel->nRight==0 
     && pWorker->pLevel->pNext==0 
    ){
      *pbCkpt = 1;
    }
  }

  if( rc==LSM_OK && bDirty ){
    lsmFinishWork(pDb, 0, &rc);
  }else{
    int rcdummy = LSM_BUSY;
    lsmFinishWork(pDb, 0, &rcdummy);
    *pnWrite = 0;
    *pbCkpt = 0;
  }
  assert( pDb->pWorker==0 );
  return rc;
}

static int doLsmWork(lsm_db *pDb, int nMerge, int nPage, int *pnWrite){
  int rc;
  int nWrite = 0;
  int bCkpt = 0;