Many hyperlinks are disabled.
Use anonymous login
to enable hyperlinks.
Overview
| Comment: | Write a delete-key into the top level of the fast-insert tree when an item is deleted. Change the seek code so that if a delete-key is encountered SQLITE4_NOTFOUND is returned to the caller. |
|---|---|
| Downloads: | Tarball | ZIP archive | SQL archive |
| Timelines: | family | ancestors | descendants | both | trunk |
| Files: | files | file ages | folders |
| SHA1: |
d9fa045dd73419a940de66ec6cae9a55 |
| User & Date: | dan 2013-11-28 18:24:18 |
Context
|
2013-12-05
| ||
| 20:08 | Support scan queries on fast-insert data. Still some problems. check-in: 0cd2ab7e9e user: dan tags: trunk | |
|
2013-11-28
| ||
| 18:24 | Write a delete-key into the top level of the fast-insert tree when an item is deleted. Change the seek code so that if a delete-key is encountered SQLITE4_NOTFOUND is returned to the caller. check-in: d9fa045dd7 user: dan tags: trunk | |
| 15:23 | Make a small change to the bt cell formats to accommodate delete keys. check-in: a5186d0b0a user: dan tags: trunk | |
Changes
Changes to src/bt_main.c.
492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 ... 870 871 872 873 874 875 876 877 878 879 880 881 882 883 ... 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 .... 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 .... 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 .... 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 .... 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 .... 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 .... 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910 1911 1912 1913 1914 .... 2767 2768 2769 2770 2771 2772 2773 2774 2775 2776 2777 2778 2779 2780 2781 2782 2783 2784 2785 2786 2787 2788 2789 2790 2791 2792 2793 2794 2795 2796 2797 2798 2799 2800 2801 2802 2803 2804 2805 2806 2807 2808 2809 2810 2811 2812 2813 2814 2815 2816 2817 2818 2819 2820 |
if( btFlags(aData) & BT_PGFLAGS_INTERNAL ){
sqlite4BtBufAppendf(pBuf, " child=%d ", (int)btGetU32(&pCell[j]));
}else{
int nVal;
pCell += nKey;
sqlite4BtBufAppendf(pBuf, " ");
pCell += sqlite4BtVarintGet32(pCell, &nVal);
for(j=0; j<(nVal-1); j++){
sqlite4BtBufAppendf(pBuf, "%02X", (int)pCell[j]);
}
}
sqlite4BtBufAppendf(pBuf, "\n");
}
}
int sqlite4BtDebugPage(sqlite4_buffer *pBuf, u32 pgno, char *aData, int nData){
................................................................................
return rc;
}
static u32 btBlockToRoot(BtDbHdr *pHdr, u32 iBlk){
assert( iBlk>0 );
return (iBlk - 1) * (pHdr->blksz / pHdr->pgsz) + 1;
}
/*
** Seek a fast-insert cursor.
*/
static int fiCsrSeek(FiCursor *pCsr, const void *pK, int nK, int eSeek){
int rc = SQLITE4_OK; /* Return code */
bt_db *db = pCsr->base.pDb; /* Database handle */
................................................................................
btCsrSetup(db, btBlockToRoot(pHdr, iBlk), pSub);
rc = btCsrSeek(pSub, pK, nK, BT_SEEK_EQ, BT_CSRSEEK_SEEK);
if( rc!=SQLITE4_NOTFOUND && rc!=SQLITE4_INEXACT ){
/* A hit on the requested key or an error has occurred. Either
** way, break out of the loop. If this is a hit, set iBt to
** zero so that the BtCsrKey() and BtCsrData() routines know
** to return data from the first (only) sub-cursor. */
if( rc==SQLITE4_OK ) pCsr->iBt = 0;
break;
}
}
btCsrReset(&mcsr, 1);
}
}else{
/* Deal with this later... */
................................................................................
p += sqlite4BtVarintGet32(p, &nKey);
if( nKey==0 ){
/* Type (b) cell */
p += sqlite4BtVarintGet32(p, &nKey);
p += nKey;
p += sqlite4BtVarintGet32(p, &nKey);
p += btOverflowArrayLen(p);
}else{
/* Type (a) cell */
p += (nKey-2);
}
}
return (p-pCell);
}
................................................................................
assert( pKV->eType==KV_CELL || pKV->eType==KV_VALUE );
if( pKV->eType==KV_CELL ){
nByte = pKV->nV;
}else{
if( pKV->pgno ){
nByte = sqlite4BtVarintLen32(pKV->nK) + pKV->nK + 4;
}else{
nByte =
sqlite4BtVarintLen32(pKV->nK)
+ sqlite4BtVarintLen32(pKV->nV+2)
+ pKV->nV + pKV->nK;
}
}
return nByte;
}
/*
** Write a cell based on *pKV to buffer aBuffer. Return the number
................................................................................
memcpy(aBuf, pKV->pV, i);
}else{
i += sqlite4BtVarintPut32(&aBuf[i], pKV->nK);
memcpy(&aBuf[i], pKV->pK, pKV->nK); i += pKV->nK;
if( pKV->pgno==0 ){
i += sqlite4BtVarintPut32(&aBuf[i], pKV->nV+2);
memcpy(&aBuf[i], pKV->pV, pKV->nV);
i += pKV->nV;
}else{
btPutU32(&aBuf[i], pKV->pgno);
i += 4;
}
}
assert( i==btKVCellSize(pKV) );
................................................................................
rc = sqlite4BtPageAllocate(db->pPager, ppPg);
}
return rc;
}
/*
** Trim a non-overflow page.
**
** This function is a simple wrapper around sqlite4BtPageAllocate(),
** except that if the database is currenly in fast-insert mode the
** BtDbHdr.nSubPg counter is incremented.
*/
static int btTrimNonOverflow(bt_db *db, BtPage *pPg){
int rc = sqlite4BtPageTrim(pPg);
if( rc==SQLITE4_OK && db->bFastInsertOp ){
BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);
pHdr->nSubPg--;
}
return rc;
}
/*
** Allocate and zero an overflow page.
*/
................................................................................
assert( pKV->pgno==0 && pKV->eType==KV_VALUE );
/* Check if this is a type (a) cell - one that can fit entirely on a
** leaf page. If so, do nothing. */
nReq = btKVCellSize(pKV);
if( nReq > nMaxSize ){
int nArraySz = btOverflowArraySz(pgsz, pKV->nK + pKV->nV);
u8 *pBuf = 0; /* Buffer containing formatted cell */
int nKeyOvfl; /* Bytes of key that overflow */
int nValOvfl; /* Bytes of value that overflow */
/* Check if the entire key can fit on a leaf page. If so, this is a
** type (b) page - entire key and partial value on the leaf page,
** overflow pages contain the rest of the value.
**
** This expression uses sqlite4BtVarintLen32() to calculate an upper
** bound for the size of the varint that indicates the number of bytes
** of the value stored locally. */
nReq = 1 + sqlite4BtVarintLen32(pKV->nK) + pKV->nK
+ 1 + sqlite4BtVarintLen32(pKV->nV) + nArraySz;
if( nReq<nMaxSize ){
/* nSpc is initialized to the amount of space available to store:
**
** * varint containing number of bytes stored locally (nLVal).
** * nLVal bytes of content.
** * varint containing number of bytes in overflow pages.
*/
int nLVal; /* Bytes of value data on main page */
................................................................................
*pOut++ = 0x00;
}
pOut += sqlite4BtVarintPut32(pOut, nLKey);
memcpy(pOut, pKV->pK, nLKey);
pOut += nLKey;
if( nKeyOvfl==0 ){
/* Type (b) cell */
*pOut++ = 0x00;
pOut += sqlite4BtVarintPut32(pOut, nLVal);
memcpy(pOut, pKV->pV, nLVal);
pOut += nLVal;
}else{
/* Type (c) cell */
pOut += sqlite4BtVarintPut32(pOut, nKeyOvfl);
}
pOut += sqlite4BtVarintPut32(pOut, nValOvfl + (nKeyOvfl>0));
rc = btOverflowArrayPopulate(db, &pOut,
(u8*)(pKV->pK) + nLKey, nKeyOvfl,
(u8*)(pKV->pV) + nLVal, nValOvfl
);
if( rc==SQLITE4_OK ){
memset(pKV, 0, sizeof(*pKV));
pKV->pV = pBuf;
pKV->nV = pOut - pBuf;
pKV->eType = KV_CELL;
pBuf = 0;
................................................................................
}
if( rc==SQLITE4_OK ){
/* The cursor currently points to an entry with key pK/nK. This call
** should therefore replace that entry. So delete it and then re-seek
** the cursor. */
rc = sqlite4BtDelete(&csr.base);
if( rc==SQLITE4_OK && nV>=0 ){
rc = btCsrSeek(&csr, pK, nK, BT_SEEK_GE, BT_CSRSEEK_UPDATE);
if( rc==SQLITE4_OK ) rc = btErrorBkpt(SQLITE4_CORRUPT);
}
}
if( nV>=0 && (rc==SQLITE4_NOTFOUND || rc==SQLITE4_INEXACT ) ){
KeyValue kv;
kv.pgno = 0;
kv.eType = KV_VALUE;
kv.pK = pK; kv.nK = nK;
kv.pV = pV; kv.nV = nV;
rc = btOverflowAssign(db, &kv);
if( rc==SQLITE4_OK ){
do{
/* Insert the new KV pair into the current leaf. */
rc = btInsertAndBalance(&csr, 1, &kv);
/* Unless this is a block-full error, break out of the loop */
if( rc!=BT_BLOCKFULL ) break;
assert( iRoot==0 );
rc = btFastInsertRoot(db, pHdr, &iRootPg);
if( rc==SQLITE4_OK ){
btCsrReset(&csr, 1);
btCsrSetup(db, iRootPg, &csr);
rc = btCsrSeek(&csr, pK, nK, BT_SEEK_GE, BT_CSRSEEK_UPDATE);
}
}while( rc==SQLITE4_NOTFOUND || rc==SQLITE4_INEXACT );
}
if( kv.eType==KV_CELL ){
sqlite4_free(db->pEnv, (void*)kv.pV);
}
}
btCsrReset(&csr, 1);
return rc;
}
static int btAllocateNewRoot(bt_db *db, u32 *piNew){
u32 iNew = 0;
BtPage *pPg;
int rc;
|
> | | > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > | > > > > > > | < > | > | | > | | | | | | | < < < | | > | < | > | > > > > > | | | | | | | | | | | | | | | | | | | | | | | | | | < > |
492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 ... 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 ... 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 .... 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 .... 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 .... 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 .... 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 .... 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 .... 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 .... 2809 2810 2811 2812 2813 2814 2815 2816 2817 2818 2819 2820 2821 2822 2823 2824 2825 2826 2827 2828 2829 2830 2831 2832 2833 2834 2835 2836 2837 2838 2839 2840 2841 2842 2843 2844 2845 2846 2847 2848 2849 2850 2851 2852 2853 2854 2855 2856 2857 2858 2859 2860 2861 2862 2863 2864 2865 2866 2867 |
if( btFlags(aData) & BT_PGFLAGS_INTERNAL ){
sqlite4BtBufAppendf(pBuf, " child=%d ", (int)btGetU32(&pCell[j]));
}else{
int nVal;
pCell += nKey;
sqlite4BtBufAppendf(pBuf, " ");
pCell += sqlite4BtVarintGet32(pCell, &nVal);
if( nVal>=2 ){
for(j=0; j<(nVal-2); j++){
sqlite4BtBufAppendf(pBuf, "%02X", (int)pCell[j]);
}
}else{
sqlite4BtBufAppendf(pBuf, "delete-key");
}
}
sqlite4BtBufAppendf(pBuf, "\n");
}
}
int sqlite4BtDebugPage(sqlite4_buffer *pBuf, u32 pgno, char *aData, int nData){
................................................................................
return rc;
}
static u32 btBlockToRoot(BtDbHdr *pHdr, u32 iBlk){
assert( iBlk>0 );
return (iBlk - 1) * (pHdr->blksz / pHdr->pgsz) + 1;
}
/*
** Return true if the cell that the argument cursor currently points to
** is a delete marker.
*/
static int fiCsrIsDelete(BtCursor *pCsr){
const int pgsz = sqlite4BtPagerPagesize(pCsr->base.pDb->pPager);
int bRet; /* Return value */
u8 *aData;
u8 *pCell;
int n;
aData = sqlite4BtPageData(pCsr->apPage[pCsr->nPg-1]);
pCell = btCellFind(aData, pgsz, pCsr->aiCell[pCsr->nPg-1]);
pCell += sqlite4BtVarintGet32(pCell, &n);
if( n==0 ){
/* Type (c) cell */
pCell += sqlite4BtVarintGet32(pCell, &n);
pCell += n;
pCell += sqlite4BtVarintGet32(pCell, &n);
pCell += sqlite4BtVarintGet32(pCell, &n);
bRet = (n==0);
}else{
pCell += n;
pCell += sqlite4BtVarintGet32(pCell, &n);
bRet = (n==1);
}
return bRet;
}
/*
** Seek a fast-insert cursor.
*/
static int fiCsrSeek(FiCursor *pCsr, const void *pK, int nK, int eSeek){
int rc = SQLITE4_OK; /* Return code */
bt_db *db = pCsr->base.pDb; /* Database handle */
................................................................................
btCsrSetup(db, btBlockToRoot(pHdr, iBlk), pSub);
rc = btCsrSeek(pSub, pK, nK, BT_SEEK_EQ, BT_CSRSEEK_SEEK);
if( rc!=SQLITE4_NOTFOUND && rc!=SQLITE4_INEXACT ){
/* A hit on the requested key or an error has occurred. Either
** way, break out of the loop. If this is a hit, set iBt to
** zero so that the BtCsrKey() and BtCsrData() routines know
** to return data from the first (only) sub-cursor. */
assert( pCsr->iBt<0 );
if( rc==SQLITE4_OK ){
if( 0==fiCsrIsDelete(pSub) ){
pCsr->iBt = 0;
}else{
rc = SQLITE4_NOTFOUND;
}
}
break;
}
}
btCsrReset(&mcsr, 1);
}
}else{
/* Deal with this later... */
................................................................................
p += sqlite4BtVarintGet32(p, &nKey);
if( nKey==0 ){
/* Type (b) cell */
p += sqlite4BtVarintGet32(p, &nKey);
p += nKey;
p += sqlite4BtVarintGet32(p, &nKey);
p += btOverflowArrayLen(p);
}else if( nKey>=2 ){
p += (nKey-2);
}
}
return (p-pCell);
}
................................................................................
assert( pKV->eType==KV_CELL || pKV->eType==KV_VALUE );
if( pKV->eType==KV_CELL ){
nByte = pKV->nV;
}else{
if( pKV->pgno ){
nByte = sqlite4BtVarintLen32(pKV->nK) + pKV->nK + 4;
}else{
assert( pKV->nV>=0 || pKV->pV==0 );
nByte =
sqlite4BtVarintLen32(pKV->nK)
+ sqlite4BtVarintLen32(pKV->nV+2)
+ MAX(pKV->nV, 0) + pKV->nK;
}
}
return nByte;
}
/*
** Write a cell based on *pKV to buffer aBuffer. Return the number
................................................................................
memcpy(aBuf, pKV->pV, i);
}else{
i += sqlite4BtVarintPut32(&aBuf[i], pKV->nK);
memcpy(&aBuf[i], pKV->pK, pKV->nK); i += pKV->nK;
if( pKV->pgno==0 ){
i += sqlite4BtVarintPut32(&aBuf[i], pKV->nV+2);
if( pKV->nV>0 ){
memcpy(&aBuf[i], pKV->pV, pKV->nV);
i += pKV->nV;
}
}else{
btPutU32(&aBuf[i], pKV->pgno);
i += 4;
}
}
assert( i==btKVCellSize(pKV) );
................................................................................
rc = sqlite4BtPageAllocate(db->pPager, ppPg);
}
return rc;
}
/*
** Trim a non-overflow page.
*/
static int btTrimNonOverflow(bt_db *db, BtPage *pPg){
int rc; /* Return code */
if( db->bFastInsertOp==0 ){
rc = sqlite4BtPageTrim(pPg);
}else{
rc = sqlite4BtPageRelease(pPg);
}
return rc;
}
/*
** Allocate and zero an overflow page.
*/
................................................................................
assert( pKV->pgno==0 && pKV->eType==KV_VALUE );
/* Check if this is a type (a) cell - one that can fit entirely on a
** leaf page. If so, do nothing. */
nReq = btKVCellSize(pKV);
if( nReq > nMaxSize ){
int nArraySz = btOverflowArraySz(pgsz, pKV->nK + MAX(0, pKV->nV));
u8 *pBuf = 0; /* Buffer containing formatted cell */
int nKeyOvfl; /* Bytes of key that overflow */
int nValOvfl; /* Bytes of value that overflow */
/* Check if the entire key can fit on a leaf page. If so, this is a
** type (b) page - entire key and partial value on the leaf page,
** overflow pages contain the rest of the value.
**
** This expression uses sqlite4BtVarintLen32() to calculate an upper
** bound for the size of the varint that indicates the number of bytes
** of the value stored locally. */
nReq = 1 + sqlite4BtVarintLen32(pKV->nK) + pKV->nK
+ 1 + sqlite4BtVarintLen32(pKV->nV) + nArraySz;
if( nReq<nMaxSize && pKV->nV>=0 ){
/* nSpc is initialized to the amount of space available to store:
**
** * varint containing number of bytes stored locally (nLVal).
** * nLVal bytes of content.
** * varint containing number of bytes in overflow pages.
*/
int nLVal; /* Bytes of value data on main page */
................................................................................
*pOut++ = 0x00;
}
pOut += sqlite4BtVarintPut32(pOut, nLKey);
memcpy(pOut, pKV->pK, nLKey);
pOut += nLKey;
if( nKeyOvfl==0 ){
/* Type (b) cell */
assert( pKV->nV>=0 );
*pOut++ = 0x00;
pOut += sqlite4BtVarintPut32(pOut, nLVal);
memcpy(pOut, pKV->pV, nLVal);
pOut += nLVal;
}else{
/* Type (c) cell */
pOut += sqlite4BtVarintPut32(pOut, nKeyOvfl);
}
pOut += sqlite4BtVarintPut32(pOut, nValOvfl + (nKeyOvfl>0));
rc = btOverflowArrayPopulate(db, &pOut,
(u8*)(pKV->pK) + nLKey, nKeyOvfl,
(u8*)(pKV->pV) + nLVal, MAX(0, nValOvfl)
);
if( rc==SQLITE4_OK ){
memset(pKV, 0, sizeof(*pKV));
pKV->pV = pBuf;
pKV->nV = pOut - pBuf;
pKV->eType = KV_CELL;
pBuf = 0;
................................................................................
}
if( rc==SQLITE4_OK ){
/* The cursor currently points to an entry with key pK/nK. This call
** should therefore replace that entry. So delete it and then re-seek
** the cursor. */
rc = sqlite4BtDelete(&csr.base);
if( rc==SQLITE4_OK && (nV>=0 || iRoot==0) ){
rc = btCsrSeek(&csr, pK, nK, BT_SEEK_GE, BT_CSRSEEK_UPDATE);
if( rc==SQLITE4_OK ) rc = btErrorBkpt(SQLITE4_CORRUPT);
}
}
if( rc==SQLITE4_NOTFOUND || rc==SQLITE4_INEXACT ){
if( nV<0 && iRoot!=0 ){
/* This is a delete on the regular b-tree (not the fast-insert tree).
** Nothing more to do. */
rc = SQLITE4_OK;
}else{
KeyValue kv;
kv.pgno = 0;
kv.eType = KV_VALUE;
kv.pK = pK; kv.nK = nK;
kv.pV = pV; kv.nV = nV;
rc = btOverflowAssign(db, &kv);
if( rc==SQLITE4_OK ){
do{
/* Insert the new KV pair into the current leaf. */
rc = btInsertAndBalance(&csr, 1, &kv);
/* Unless this is a block-full error, break out of the loop */
if( rc!=BT_BLOCKFULL ) break;
assert( iRoot==0 );
rc = btFastInsertRoot(db, pHdr, &iRootPg);
if( rc==SQLITE4_OK ){
btCsrReset(&csr, 1);
btCsrSetup(db, iRootPg, &csr);
rc = btCsrSeek(&csr, pK, nK, BT_SEEK_GE, BT_CSRSEEK_UPDATE);
}
}while( rc==SQLITE4_NOTFOUND || rc==SQLITE4_INEXACT );
}
if( kv.eType==KV_CELL ){
sqlite4_free(db->pEnv, (void*)kv.pV);
}
}
}
btCsrReset(&csr, 1);
return rc;
}
static int btAllocateNewRoot(bt_db *db, u32 *piNew){
u32 iNew = 0;
BtPage *pPg;
int rc;
|
Changes to www/bt.wiki.
538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 |
that use overflow pages for the value only, and (c) cells that use overflow
pages for the key and value.
<p>Cell type (a):
<ul>
<li> Number of bytes of the entries key (nKey), as a varint.
<li> nKey bytes of key data.
<li> Number of bytes in entries value plus one (nValue+2), as a varint. Or,
for a delete key, a single 0x01 byte.
<li> Unless this is a delete key, nValue bytes of value data.
</ul>
<p>Cell type (b):
<ul>
<li> Number of bytes in entries key (nKey), as a varint.
|
| |
538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 |
that use overflow pages for the value only, and (c) cells that use overflow
pages for the key and value.
<p>Cell type (a):
<ul>
<li> Number of bytes of the entries key (nKey), as a varint.
<li> nKey bytes of key data.
<li> Number of bytes in entries value plus two (nValue+2), as a varint. Or,
for a delete key, a single 0x01 byte.
<li> Unless this is a delete key, nValue bytes of value data.
</ul>
<p>Cell type (b):
<ul>
<li> Number of bytes in entries key (nKey), as a varint.
|