SQLite4  Check-in [9f83998312]

/*
** TODO: Rearrange things so these are not required!
*/
static int fiLoadSummary(bt_db *, BtCursor *, const u8 **, int *);
static void btReadSummary(const u8 *, int, u16 *, u16 *, u16 *);
static int btCsrData(BtCursor *, int, int, const void **, int *);
static int btReadSchedule(bt_db *, u8 *, BtSchedule *);
static int btCsrEnd(BtCursor *pCsr, int bLast);
static int btCsrStep(BtCursor *pCsr, int bNext);
static int btCsrKey(BtCursor *pCsr, const void **ppK, int *pnK);

/* 
** Meta-table summary key.
*/
static const u8 aSummaryKey[] = {0xFF, 0xFF, 0xFF, 0xFF};

#ifndef btErrorBkpt

  btPageToAscii(pHdr->iMRoot, bAscii, pPager, aData, nData, &buf);
  sqlite4_buffer_append(&buf, "", 1);

  fprintf(stderr, "%s", (char*)buf.p);
  sqlite4_buffer_clear(&buf);
  sqlite4BtPageRelease(pPg);
}


static void btDumpCsr(sqlite4_buffer *pBuf, BtCursor *pCsr){
  assert( pCsr->nPg>=0 );
  if( pCsr->nPg==0 ){
    sqlite4BtBufAppendf(pBuf, "EOF");
  }else{
    int rc;
    const void *pKey = 0;
    int nKey = 0;

    rc = btCsrKey(pCsr, &pKey, &nKey);
    assert( rc==SQLITE4_OK );
    btBufferAppendBlob(pBuf, 0, (u8*)pKey, nKey);
  }
}

static void fiDumpCsr(FiCursor *pCsr){
  int iBt;

  sqlite4_buffer buf;
  sqlite4_buffer_init(&buf, 0);
  for(iBt=0; iBt<pCsr->nBt; iBt++){
    FiSubCursor *pSub = &pCsr->aSub[iBt];

    sqlite4BtBufAppendf(&buf, "%d prefix: ", iBt);
    btBufferAppendBlob(&buf, 0, pSub->aPrefix, sizeof(pSub->aPrefix));

    sqlite4BtBufAppendf(&buf, "\n%d mcsr  : ", iBt);
    btDumpCsr(&buf, &pSub->mcsr);

    sqlite4BtBufAppendf(&buf, "\n%d csr   : ", iBt);
    btDumpCsr(&buf, &pSub->csr);
    sqlite4BtBufAppendf(&buf, "\n");
  }

  fprintf(stderr, "%s", (char*)buf.p);
  sqlite4_buffer_clear(&buf);
}
#endif

#ifndef NDEBUG
/*
** This function is really just a big assert() statement. It contributes 
** nothing to the operation of the library.
**
** The assert() fails if the summary-record is not consistent with the
** actual contents of the meta-tree.

      assert( iMin==0 && nLevel==0 );
    }
  }

  btCsrReset(&csr, 1);
  btCsrReset(&mcsr, 1);
}

static void assert_ficursor_ok(FiCursor *p, int crc){
  int iBt;

  if( crc!=SQLITE4_OK && crc!=SQLITE4_NOTFOUND && crc!=SQLITE4_INEXACT ) return;
  if( (p->base.flags & (CSR_NEXT_OK|CSR_PREV_OK))==0 ) return;

  for(iBt=0; iBt<p->nBt; iBt++){
    FiSubCursor *pSub = &p->aSub[iBt];

    if( pSub->mcsr.nPg>0 ){
      const void *pKey = 0; 
      int nKey = 0;
      assert( pSub->csr.nPg>0 );

      btCsrKey(&pSub->mcsr, &pKey, &nKey);
      assert( nKey>=8 && 0==memcmp(pSub->aPrefix, pKey, 8) );
    }else{
      assert( pSub->csr.nPg==0 );
    }
  }
}

#else
# define assert_summary_ok(x, rc) 
# define assert_ficursor_ok(p, rc)
#endif


/*
** This function compares the key passed via parameters pK and nK to the
** key that cursor pCsr currently points to.
**

    }
  }

  if( iMin<0 && rc==SQLITE4_OK ) rc = SQLITE4_NOTFOUND;
  pCsr->iBt = iMin;
  return rc;
}

static int fiSubCsrCheckPrefix(FiSubCursor *pSub){
  const int nPrefix = sizeof(pSub->aPrefix);
  const void *pK = 0;
  int nK;
  int rc;

  rc = btCsrKey(&pSub->mcsr, &pK, &nK);
  if( rc==SQLITE4_OK && (nK<nPrefix || memcmp(pK, pSub->aPrefix, nPrefix)) ){
    rc = SQLITE4_NOTFOUND;
    btCsrReset(&pSub->mcsr, 0);
  }
  return rc;
}

/*
** Return SQLITE4_OK if the cursor is successfully stepped, or 
** SQLITE4_NOTFOUND if an EOF is encountered.
**
** If an error occurs (e.g. an IO error or OOM condition), return the
** relevant error code.

#endif

    rc = btCsrStep(&pSub->mcsr, bNext);
    if( rc==SQLITE4_OK ){
      rc = btCsrKey(&pSub->mcsr, &pV, &nV);
    }
    if( rc==SQLITE4_OK ){
      rc = fiSubCsrCheckPrefix(pSub);


    }
    if( rc==SQLITE4_OK ){

      rc = btCsrData(&pSub->mcsr, 0, 4, &pV, &nV);

    }
    if( rc==SQLITE4_OK ){
      pSub->csr.iRoot = sqlite4BtGetU32((const u8*)pV);
      rc = btCsrEnd(&pSub->csr, !bNext);
    }
  }

#ifndef NDEBUG
  sqlite4_buffer buf;
  sqlite4_buffer_init(&buf, 0);
  sqlite4BtCsrKey(&pCsr->base, &pKey, &nKey);
  sqlite4_buffer_set(&buf, pKey, nKey);
#endif

  assert_ficursor_ok(pCsr, rc);
  assert( pCsr->base.flags & (CSR_NEXT_OK | CSR_PREV_OK) );
  assert( pCsr->iBt>=0 );

  do{
    /* Load the current key in to pKey/nKey. Then advance all sub-cursors 
    ** that share a key with the current sub-cursor. */
    rc = sqlite4BtCsrKey(&pCsr->base, &pKey, &nKey);

#ifndef NDEBUG
  if( rc==SQLITE4_OK ){
    sqlite4BtCsrKey(&pCsr->base, &pKey, &nKey);
    assert( btKeyCompare(buf.p, buf.n, pKey, nKey) * (bNext?1:-1) < 0 );
  }
  sqlite4_buffer_clear(&buf);
#endif
  assert_ficursor_ok(pCsr, rc);

  return rc;
}

typedef struct FiLevelIter FiLevelIter;
struct FiLevelIter {
  /* Used internally */

  btPutU32(&aPrefix[4], ~(u32)iLvl);
}

/*
** Seek a fast-insert cursor.
*/
static int fiCsrSeek(FiCursor *pCsr, const void *pK, int nK, int eSeek){

  int rc = SQLITE4_NOTFOUND;      /* Return code */
  bt_db *db = pCsr->base.pDb;     /* Database handle */
  BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);

  assert( eSeek==BT_SEEK_LE || eSeek==BT_SEEK_EQ || eSeek==BT_SEEK_GE );
  assert( (pCsr->base.flags & CSR_VISIT_DEL)==0 || eSeek==BT_SEEK_GE );
  fiCsrReset(pCsr);

  if( pHdr->iMRoot ){
    FiLevelIter iter;

    /* Initialize the iterator used to skip through database levels */
    rc = fiLevelIterInit(db, &iter);
    if( rc!=SQLITE4_OK ) return rc;

    if( eSeek==BT_SEEK_EQ ){
      FiSubCursor *pSub;
      BtCursor *pM;

      pCsr->base.flags &= ~(CSR_NEXT_OK | CSR_PREV_OK);

      /* A BT_SEEK_EQ is a special case. There is no need to set up a cursor
      ** that can be advanced (in either direction) in this case. All that
      ** is required is to search each level in order for the requested key 
      ** (or a corresponding delete marker). Once a match is found, there
      ** is no need to search any further. As a result, only a single
      ** sub-cursor is required.  */
      rc = fiCsrAllocateSubs(db, pCsr, 1);
      pSub = pCsr->aSub;
      pM = &pSub->mcsr;

      btCsrSetup(db, pHdr->iMRoot, pM);
      while( 0==fiLevelIterNext(&iter) ){

        fiFormatPrefix(pSub->aPrefix, iter.iAge, iter.iLvl);
        rc = btCsrSeek(pM, pSub->aPrefix, pK, nK, BT_SEEK_LE, BT_CSRSEEK_SEEK);

        if( rc==SQLITE4_INEXACT ){
          rc = fiSubCsrCheckPrefix(pSub);
        }

        if( rc==SQLITE4_NOTFOUND ){
          /* All keys in this level are greater than pK/nK. */
          /* no-op */
        }else if( rc==SQLITE4_OK || rc==SQLITE4_INEXACT ){
          const void *pV;
          int nV;

          }
        }
      }
    }else{
      int bMatch = 0;           /* Found an exact match */
      int bHit = 0;             /* Found at least one entry */

      pCsr->base.flags |= (eSeek==BT_SEEK_GE ? CSR_NEXT_OK : CSR_PREV_OK);

      /* Allocate required sub-cursors. */
      if( rc==SQLITE4_OK ){
        rc = fiCsrAllocateSubs(db, pCsr, iter.nSub);
      }

      /* This loop runs once for each sub-cursor */
      while( rc==SQLITE4_OK && 0==fiLevelIterNext(&iter) ){
        FiSubCursor *pSub = &pCsr->aSub[iter.iSub];
        BtCursor *pM = &pSub->mcsr;
        btCsrSetup(db, pHdr->iMRoot, pM);

        btPutU32(&pSub->aPrefix[0], (u32)iter.iAge);
        btPutU32(&pSub->aPrefix[4], ~(u32)iter.iLvl);

        rc = btCsrSeek(pM, pSub->aPrefix, pK, nK, BT_SEEK_LE, BT_CSRSEEK_SEEK);
        if( rc==SQLITE4_NOTFOUND && eSeek==BT_SEEK_GE ){
          rc = btCsrSeek(pM, pSub->aPrefix, 0, 0, BT_SEEK_GE, BT_CSRSEEK_SEEK);
        }

        if( rc==SQLITE4_INEXACT ){
          rc = fiSubCsrCheckPrefix(pSub);
        }

        if( rc==SQLITE4_NOTFOUND ){
          /* No keys to visit in this level */
          assert( pSub->mcsr.nPg==0 );
          assert( pSub->csr.nPg==0 );
          rc = SQLITE4_OK;
        }else if( rc==SQLITE4_OK || rc==SQLITE4_INEXACT ){
          const void *pV;
          int nV;
          u32 iRoot;
          sqlite4BtCsrData(&pM->base, 0, 4, &pV, &nV);
          iRoot = sqlite4BtGetU32((const u8*)pV);
          btCsrReset(&pSub->csr, 1);
          btCsrSetup(db, iRoot, &pSub->csr);

          rc = btCsrSeek(&pSub->csr, 0, pK, nK, eSeek, BT_CSRSEEK_SEEK);
          if( rc==SQLITE4_NOTFOUND ){
            if( eSeek==BT_SEEK_LE ){
              rc = fiSubCsrStep(pCsr, pSub, 0);
            }else{
              btCsrReset(pM, 0);
            }
          }else{
            if( rc==SQLITE4_OK ) bMatch = 1;
            if( rc==SQLITE4_INEXACT ) bHit = 1;

          }

          if( rc==SQLITE4_INEXACT || rc==SQLITE4_NOTFOUND ) rc = SQLITE4_OK;
        }else{
          /* An error */
        }
      }
      assert( rc!=SQLITE4_OK || iter.iSub==iter.nSub );

      if( rc==SQLITE4_OK ){

        rc = fiCsrSetCurrent(pCsr);
        if( rc==SQLITE4_OK ){
          if( fiCsrIsDelete(pCsr) ){
            rc = fiCsrStep(pCsr);
            if( rc==SQLITE4_OK ) rc = SQLITE4_INEXACT;
          }else if( bMatch==0 ){
            rc = (bHit ? SQLITE4_INEXACT : SQLITE4_NOTFOUND);

){
  int rc;
  btCheckPageRefs(pBase->pDb);
  if( IsBtCsr(pBase) ){
    BtCursor *pCsr = (BtCursor*)pBase;
    rc = btCsrSeek(pCsr, 0, pK, nK, eSeek, BT_CSRSEEK_SEEK);
  }else{
    FiCursor *pCsr = (FiCursor*)pBase;
    rc = fiCsrSeek(pCsr, pK, nK, eSeek);
    assert_ficursor_ok(pCsr, rc);
  }
  btCheckPageRefs(pBase->pDb);
  return rc;
}

/*
** Position cursor pCsr to point to the smallest key in the database.

  return rc;
}

typedef struct BalanceCtx BalanceCtx;
struct BalanceCtx {
  int pgsz;                       /* Database page size */
  int bLeaf;                      /* True if we are rebalancing leaf data */
  u8 flags;                       /* Flags byte for new sibling pages */
  BtCursor *pCsr;                 /* Cursor identifying where to insert pKV */
  int nKV;                        /* Number of KV pairs */
  KeyValue *apKV;                 /* New KV pairs being inserted */

  /* Populated by btGatherSiblings */
  int nIn;                        /* Number of sibling pages */
  BtPage *apPg[5];                /* Array of sibling pages */

    btPutU16(&aOut[p->pgsz-6], iOff);

    if( (iCell+1)==p->anOut[p->iOut] ){
      /* That was the last cell for this page. Fill in the rest of the 
      ** output page footer and the flags byte at the start of the page.  */
      int nFree;                    /* Free space remaining on output page */
      nFree = p->pgsz - iOff - (6 + 2*(nCell+1));
      aOut[0] = p->flags;
      btPutU16(&aOut[p->pgsz-4], nFree);

      /* If the siblings are leaf pages, increment BalanceCtx.iOut here.
      ** for internal nodes, it will be incremented by the next call to
      ** this function, after a divider cell is pushed into the parent 
      ** node.  */
      p->iOut += p->bLeaf;

  BalanceCtx ctx;
  memset(&ctx, 0, sizeof(ctx));
  ctx.pCsr = pCsr;
  ctx.nKV = nKV;
  ctx.apKV = apKV;
  ctx.pgsz = pgsz;
  ctx.bLeaf = bLeaf;
  ctx.flags = *(u8*)sqlite4BtPageData(pCsr->apPage[pCsr->nPg-1]);

  memset(anByteOut, 0, sizeof(anByteOut));

  /* Gather the sibling pages from which cells will be redistributed into
  ** the ctx.apPg[] array.  */
  assert( bLeaf==0 || bLeaf==1 );
  assert( pCsr->nPg>1 );

  btFreeBuffer(pDb, ctx.pTmp);
  sqlite4_free(pDb->pEnv, ctx.anCellSz);
  return rc;
}

static int btExtendTree(BtCursor *pCsr){
  bt_db * const pDb = pCsr->base.pDb;
  BtDbHdr *pHdr = sqlite4BtPagerDbhdr(pDb->pPager);
  const int pgsz = pHdr->pgsz;
  int rc;                         /* Return code */
  BtPage *pNew;                   /* New (and only) child of root page */
  BtPage *pRoot = pCsr->apPage[0];

  assert( pCsr->nPg==1 );

  rc = sqlite4BtPageWrite(pRoot);
  if( rc==SQLITE4_OK ){
    rc = btAllocateNonOverflow(pDb, &pNew);
  }
  if( rc==SQLITE4_OK ){
    u8 *aRoot = sqlite4BtPageData(pRoot);
    u8 *aData = sqlite4BtPageData(pNew);

    memcpy(aData, aRoot, pgsz);
    aRoot[0] = BT_PGFLAGS_INTERNAL;
    if( pHdr->iMRoot==pCsr->iRoot ) aRoot[0] |= BT_PGFLAGS_METATREE;
    btPutU32(&aRoot[1], sqlite4BtPagePgno(pNew));
    btPutU16(&aRoot[pgsz-2], 0);
    btPutU16(&aRoot[pgsz-4], 5);
    btPutU16(&aRoot[pgsz-6], pgsz - 5 - 6);

    pCsr->nPg = 2;
    pCsr->aiCell[1] = pCsr->aiCell[0];

      u8 *aNew;
      int nByte = nSum;
      if( iBestAge+1>=(nSum/6) ) nByte += 6;

      rc = SQLITE4_OK;
      aNew = (u8*)btMalloc(db, nByte, &rc);
      if( rc==SQLITE4_OK ){


        /* Create a copy of the summary record */
        memcpy(aNew, aSum, nSum);
        if( nByte>nSum ) btWriteSummary(aNew, iBestAge+1, 0, 0, 0);

        /* Find the input age and maximum level */
        btReadSummary(aSum, iBestAge, &iMin, &nLevel, &iMerge);
        *piMinLevel = (u32)iMin;
        *piMaxLevel = (u32)(iMin + nLevel - 1);
        *piAge = iBestAge;

        /* Find the output level */
        btReadSummary(aNew, iBestAge+1, &iMin, &nLevel, &iMerge);
        *piOutLevel = iMin + nLevel;










      }
    }
  }

  btCsrReset(&csr, 1);
  return rc;
}

  BtCursor mcsr;                  /* Cursor for reading the meta-tree */
  const void *pKey = 0;
  const u8 *aSum; int nSum;       /* Summary value */
  int nKey = 0;
  sqlite4_buffer buf;
  u32 iLvl;
  int iBlk;

  /* Save the value of the fast-insert flag. It will be restored before
  ** this function returns. Leaving it set here interferes with page 
  ** allocation if the meta-tree needs to be extended.  */
  const int bFastInsertOp = db->bFastInsertOp;
  db->bFastInsertOp = 0;
  
#if 0
  fprintf(stderr, "BEFORE!\n");
  btPrintMetaTree(db->pPager, 1, pHdr);
#endif
  assert_summary_ok(db, SQLITE4_OK);

#if 0
  if( rc==SQLITE4_OK ){
    btPrintMetaTree(db->pPager, 1, pHdr);
  }
#endif
  assert_summary_ok(db, SQLITE4_OK);
  db->bFastInsertOp = bFastInsertOp;
  return rc;
}

/*
** If possible, schedule a merge operation. 
**
** The merge operation is selected based on the following criteria:

    if( rc==SQLITE4_INEXACT ){
      const int nPrefix = sizeof(pSub->aPrefix);
      rc = btCsrKey(pM, &pKey, &nKey);
      if( rc==SQLITE4_OK ){
        if( nKey<nPrefix || memcmp(pKey, pSub->aPrefix, nPrefix) ){
          /* Level is completely empty. Nothing to do for this level. */
          btCsrReset(pM, 0);
          rc = SQLITE4_NOTFOUND;
        }else{
          nKey -= nPrefix;
          pKey = (const void*)(((const u8*)pKey) + nPrefix);
        }
      }
    }

        btCsrSetup(db, iRoot, &pSub->csr);
        rc = btCsrSeek(&pSub->csr, 0, pKey, nKey, BT_SEEK_GE, 0);
        if( rc==SQLITE4_INEXACT ) rc = SQLITE4_OK;
        if( rc==SQLITE4_NOTFOUND ) rc = btErrorBkpt(SQLITE4_CORRUPT);
      }
    }else if( rc==SQLITE4_NOTFOUND ){
      assert( pSub->csr.nPg==0 );
      assert( pSub->mcsr.nPg==0 );
      rc = SQLITE4_OK;
    }
  }

  if( rc==SQLITE4_OK ){
    rc = fiCsrSetCurrent(pCsr);
  }

  btReadSchedule(db, aSched, &s);
  if( s.eBusy==BT_SCHEDULE_BUSY ){
    FiCursor fcsr;                /* FiCursor used to read input */
    FiWriter writer;              /* FiWriter used to write output */

    rc = fiSetupMergeCsr(db, pHdr, &s, &fcsr);
    assert( rc!=SQLITE4_NOTFOUND );
    assert_ficursor_ok(&fcsr, rc);
    fiWriterInit(db, &s, &writer, &rc);

    /* The following loop runs once for each key copied from the input to
    ** the output segments. It terminates either when the input is exhausted
    ** or when all available output blocks are full.  */
    while( rc==SQLITE4_OK ){
      const void *pCell;          /* Cell to copy to output */

      db->bFastInsertOp = 1;
      break;
    }
  }

  return rc;
}