SQLite4  Check-in [e5edba9caa]

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

    if( db ){
      int nDone;
      fprintf(stderr, "lsm_work() start...\n");
      fflush(stderr);
      do {
        nDone = 0;
        rc = lsm_work(db, 1, (1<<30), &nDone);
      }while( rc==0 && nDone>0 );
      fprintf(stderr, "lsm_work() done...\n");
      fflush(stderr);
    }

if( i+1<p->nRepeat ){
    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);

int test_lsm_lomem_open(
  const char *zFilename, 
  int bClear, 
  TestDb **ppDb
){
  const char *zCfg = 
    "page_size=256 block_size=65536 autoflush=16384 "
    "max_freelist=4 autocheckpoint=32768 "
    "mmap=0 "
  ;
  return testLsmOpen(zCfg, zFilename, bClear, ppDb);
}

int test_lsm_zip_open(
  const char *zFilename, 

      iRet = iPg;
    }
  }
  return iRet;
}

#ifndef NDEBUG
/*
** Return true if page iPg, which is a part of segment p, lies on
** a redirected block. 
*/
static int fsPageRedirects(FileSystem *pFS, Segment *p, Pgno iPg){
  return (iPg!=0 && iPg!=lsmFsRedirectPage(pFS, p->pRedirect, iPg));
}

/*
** Return true if the second argument is not NULL and any of the first
** last or root pages lie on a redirected block. 
*/
static int fsSegmentRedirects(FileSystem *pFS, Segment *p){
  return (p && (
      fsPageRedirects(pFS, p, p->iFirst)
   || fsPageRedirects(pFS, p, p->iRoot)
   || fsPageRedirects(pFS, p, p->iLastPg)
  ));
}
#endif

/*
** Argument aPgno is an array of nPgno page numbers. All pages belong to
** the segment pRun. This function gobbles from the start of the run to the
** first page that appears in aPgno[] (i.e. so that the aPgno[] entry is

  u8 *aUsed
){
  if( pSeg ){
    if( pSeg && pSeg->nSize>0 ){
      Redirect *pRedir = pSeg->pRedirect;
      int rc;
      int iBlk;                   /* Current block (during iteration) */
      int iLastBlk;               /* Last block of segment */
      int iFirstBlk;              /* First block of segment */
      int bLastIsLastOnBlock;     /* True iLast is the last on its block */

      assert( 0==fsSegmentRedirects(pFS, pSeg) );
      iBlk = iFirstBlk = fsPageToBlock(pFS, pSeg->iFirst);
      iLastBlk = fsPageToBlock(pFS, pSeg->iLastPg);

      bLastIsLastOnBlock = (fsLastPageOnBlock(pFS, iLastBlk)==pSeg->iLastPg);
      assert( iBlk>0 );









      do {
        /* iBlk is a part of this sorted run. */
        aUsed[iBlk-1] |= INTEGRITY_CHECK_USED;

        /* If the first page of this block is also part of the segment,
        ** set the flag to indicate that the first page of iBlk is in use.  
        */
        if( fsFirstPageOnBlock(pFS, iBlk)==pSeg->iFirst || iBlk!=iFirstBlk ){
          assert( (aUsed[iBlk-1] & INTEGRITY_CHECK_FIRST_PG)==0 );
          aUsed[iBlk-1] |= INTEGRITY_CHECK_FIRST_PG;
        }

        /* Unless the sorted run finishes before the last page on this block, 
        ** the last page of this block is also in use.  */
        if( iBlk!=iLastBlk || bLastIsLastOnBlock ){
          assert( (aUsed[iBlk-1] & INTEGRITY_CHECK_LAST_PG)==0 );
          aUsed[iBlk-1] |= INTEGRITY_CHECK_LAST_PG;
        }

  DbTruncateCtx *p = (DbTruncateCtx *)pCtx;
  if( iBlk!=p->nBlock || (p->iInUse>=0 && iSnapshot>=p->iInUse) ) return 1;
  p->nBlock--;
  return 0;
}

static int dbTruncate(lsm_db *pDb, i64 iInUse){
  int rc = LSM_OK;
#if 0
  int i;
  DbTruncateCtx ctx;

  assert( pDb->pWorker );
  ctx.nBlock = pDb->pWorker->nBlock;
  ctx.iInUse = iInUse;

      lsmLogMessage(pDb, 0, 
          "dbTruncate(): truncated db to %d blocks",ctx.nBlock
      );
    }
#endif
    pDb->pWorker->nBlock = ctx.nBlock;
  }
#endif
  return rc;
}


/*
** This function is called during database shutdown (when the number of
** connections drops from one to zero). It truncates the database file

  return rc;
}


static int multiCursorSetupTree(MultiCursor *pCsr, int bRev){
  int rc;



  rc = multiCursorAllocTree(pCsr);
  if( rc==LSM_OK ){
    int i;
    for(i=pCsr->nTree-1; i>0; i--){
      multiCursorDoCompare(pCsr, i, bRev);
    }
  }

  iVal = pCsr->aTree[1];
  mergeRangeDeletes(pCsr, &iVal, &eType);

  if( eType!=0 ){
    if( pMW->aGobble ){
      int iGobble = pCsr->aTree[1] - CURSOR_DATA_SEGMENT;
      if( iGobble<pCsr->nPtr && iGobble>=0 ){
        SegmentPtr *pGobble = &pCsr->aPtr[iGobble];
        if( (pGobble->flags & PGFTR_SKIP_THIS_FLAG)==0 ){
          pMW->aGobble[iGobble] = lsmFsPageNumber(pGobble->pPg);
        }
      }
    }

    }
  }

  /* Advance the cursor to the next input record (assuming one exists). */
  assert( lsmMCursorValid(pMW->pCsr) );
  if( rc==LSM_OK ) rc = lsmMCursorNext(pMW->pCsr);

























  return rc;
}

static int mergeWorkerDone(MergeWorker *pMW){
  return pMW->pCsr==0 || !lsmMCursorValid(pMW->pCsr);
}

    /* Do the work to create the new merged segment on disk */
    if( rc==LSM_OK ) rc = lsmMCursorFirst(pCsr);
    while( rc==LSM_OK && mergeWorkerDone(&mergeworker)==0 ){
      rc = mergeWorkerStep(&mergeworker);
    }
    assert( rc!=LSM_OK || mergeworker.nWork==0 || pNew->lhs.iFirst );
    mergeWorkerShutdown(&mergeworker, &rc);
    if( rc==LSM_OK && pNew->lhs.iFirst ){
      rc = lsmFsSortedFinish(pDb->pFS, &pNew->lhs);
    }
    nWrite = mergeworker.nWork;

    pNew->flags &= ~LEVEL_INCOMPLETE;
    if( eTree==TREE_NONE ){
      pNew->flags |= LEVEL_FREELIST_ONLY;
    }
    pNew->pMerge = 0;
  }

    }else if( pDel ){
      assert( pNew->pNext==pDel );
      pNew->pNext = pDel->pNext;
      lsmFsSortedDelete(pDb->pFS, pDb->pWorker, 1, &pDel->lhs);
      sortedFreeLevel(pDb->pEnv, pDel);
    }

#if 1
    lsmSortedDumpStructure(pDb, pDb->pWorker, 0, 0, "new-toplevel");
#endif

    if( freelist.nEntry ){
      Freelist *p = &pDb->pWorker->freelist;
      lsmFree(pDb->pEnv, p->aEntry);
      memcpy(p, &freelist, sizeof(freelist));

      rc = lsmBlockFree(pDb, iFrom);

      *pnWrite = lsmFsBlockSize(pDb->pFS) / lsmFsPageSize(pDb->pFS);
      pLvl->lhs.pRedirect = &p->redirect;
    }
  }

#if 1
  if( rc==LSM_OK ){
    lsmSortedDumpStructure(pDb, pDb->pWorker, 0, 0, "move-block");
  }
#endif
  return rc;
}

        rc = sortedMoveBlock(pDb, &nDone);
      }
      nRemaining -= nDone;

      /* Could not find any work to do. Finished. */
      if( nDone==0 ) break;
    }else{
      int bSave = 0;
      Freelist freelist = {0, 0, 0};
      MergeWorker mergeworker;    /* State used to work on the level merge */

      assert( pDb->bIncrMerge==0 );
      assert( pDb->pFreelist==0 && pDb->bUseFreelist==0 );

      pDb->bIncrMerge = 1;
      rc = mergeWorkerInit(pDb, pLevel, &mergeworker);
      assert( mergeworker.nWork==0 );
      
      while( rc==LSM_OK 
          && 0==mergeWorkerDone(&mergeworker) 
          && (mergeworker.nWork<nRemaining || pDb->bUseFreelist)
      ){
        int eType = rtTopic(mergeworker.pCsr->eType);
        rc = mergeWorkerStep(&mergeworker);

        if( rc==LSM_OK 
         && nMerge==1 
         && mergeworker.pLevel==pWorker->pLevel
         && eType==0
         && (rtTopic(mergeworker.pCsr->eType) || mergeWorkerDone(&mergeworker))
        ){
          assert( pDb->pFreelist==0 && pDb->bUseFreelist==0 );
          rc = multiCursorVisitFreelist(mergeworker.pCsr);
          if( rc==LSM_OK ){
            rc = multiCursorSetupTree(mergeworker.pCsr, 0);
            pDb->pFreelist = &freelist;
            pDb->bUseFreelist = 1;
          }
        }
      }
      nRemaining -= LSM_MAX(mergeworker.nWork, 1);


      if( rc==LSM_OK ){
        /* Check if the merge operation is completely finished. If not,


        ** gobble up (declare eligible for recycling) any pages from rhs
        ** segments for which the content has been completely merged into 
        ** the lhs of the level.  */


        if( mergeWorkerDone(&mergeworker)==0 ){
          int i;
          for(i=0; i<pLevel->nRight; i++){
            SegmentPtr *pGobble = &mergeworker.pCsr->aPtr[i];
            if( pGobble->pSeg->iRoot ){
              rc = sortedBtreeGobble(pDb, mergeworker.pCsr, i);
            }else if( mergeworker.aGobble[i] ){
              lsmFsGobble(pDb, pGobble->pSeg, &mergeworker.aGobble[i], 1);
            }
          }
        }else{
          int i;
          int bEmpty;
          mergeWorkerShutdown(&mergeworker, &rc);
          bEmpty = (pLevel->lhs.iFirst==0);

          if( bEmpty==0 && rc==LSM_OK ){
            rc = lsmFsSortedFinish(pDb->pFS, &pLevel->lhs);
          }

          if( pDb->bUseFreelist ){
            Freelist *p = &pDb->pWorker->freelist;
            lsmFree(pDb->pEnv, p->aEntry);
            memcpy(p, &freelist, sizeof(freelist));
            pDb->bUseFreelist = 0;
            pDb->pFreelist = 0;
            bSave = 1;
          }

          for(i=0; i<pLevel->nRight; i++){
            lsmFsSortedDelete(pDb->pFS, pWorker, 1, &pLevel->aRhs[i]);
          }

          if( bEmpty ){
            /* If the new level is completely empty, remove it from the 
            ** database snapshot. This can only happen if all input keys were
            ** annihilated. Since keys are only annihilated if the new level
            ** is the last in the linked list (contains the most ancient of
            ** database content), this guarantees that pLevel->pNext==0.  */ 

            Level *pTop;          /* Top level of worker snapshot */
            Level **pp;           /* Read/write iterator for Level.pNext list */

            assert( pLevel->pNext==0 );

            /* Remove the level from the worker snapshot. */
            pTop = lsmDbSnapshotLevel(pWorker);
            for(pp=&pTop; *pp!=pLevel; pp=&((*pp)->pNext));
            *pp = pLevel->pNext;
            lsmDbSnapshotSetLevel(pWorker, pTop);

            /* Free the Level structure. */




            sortedFreeLevel(pDb->pEnv, pLevel);
          }else{


            /* Free the separators of the next level, if required. */
            if( pLevel->pMerge->nInput > pLevel->nRight ){
              assert( pLevel->pNext->lhs.iRoot );
              pLevel->pNext->lhs.iRoot = 0;
            }

            /* Zero the right-hand-side of pLevel */



            lsmFree(pDb->pEnv, pLevel->aRhs);
            pLevel->nRight = 0;
            pLevel->aRhs = 0;

            /* Free the Merge object */
            lsmFree(pDb->pEnv, pLevel->pMerge);
            pLevel->pMerge = 0;
          }

          if( bSave && rc==LSM_OK ){
            pDb->bIncrMerge = 0;
            rc = lsmSaveWorker(pDb, 0);
          }
        }
      }

      /* Clean up the MergeWorker object initialized above. If no error
      ** has occurred, invoke the work-hook to inform the application that
      ** the database structure has changed. */
      mergeWorkerShutdown(&mergeworker, &rc);
      pDb->bIncrMerge = 0;
      if( rc==LSM_OK ) sortedInvokeWorkHook(pDb);

#if 1
      lsmSortedDumpStructure(pDb, pDb->pWorker, 0, 0, "work");
#endif
      assertBtreeOk(pDb, &pLevel->lhs);
      assertRunInOrder(pDb, &pLevel->lhs);

      /* If bFlush is true and the database is no longer considered "full",
      ** break out of the loop even if nRemaining is still greater than

    static int nCall = 0;
    Level *pLevel;
    int iLevel = 0;

    nCall++;
    lsmLogMessage(pDb, LSM_OK, "Database structure %d (%s)", nCall, zWhy);

    if( nCall==275 || nCall==276 ) bKeys = 1;

    for(pLevel=pTopLevel; pLevel; pLevel=pLevel->pNext){
      char zLeft[1024];
      char zRight[1024];
      int i = 0;

      Segment *aLeft[24];  

        sortedDumpSegment(pDb, &pLevel->lhs, bVals);
        for(i=0; i<pLevel->nRight; i++){
          sortedDumpSegment(pDb, &pLevel->aRhs[i], bVals);
        }
      }
    }
  }

  assert( lsmFsIntegrityCheck(pDb) );
}

void lsmSortedFreeLevel(lsm_env *pEnv, Level *pLevel){
  Level *pNext;
  Level *p;

  for(p=pLevel; p; p=pNext){