<commit_before>#include "StorageFile.h"
#include "StorageFileHeader.h"
#include "StorageDataCache.h"
#include "StorageCursor.h"
#include "StorageRecoveryLog.h"
#include "StorageDefaults.h"
#include "System/FileSystem.h"
#include "stdio.h"

#define INDEXPAGE_OFFSET    (STORAGEFILE_HEADER_LENGTH+INDEXPAGE_HEADER_SIZE)

#define DATAPAGE_OFFSET(idx) (INDEXPAGE_OFFSET+indexPageSize+(idx)*dataPageSize)
#define DATAPAGE_INDEX(offs) (((offs)-INDEXPAGE_OFFSET-indexPageSize)/dataPageSize)
#define FILE_TYPE           "ScalienDB data file"
#define FILE_VERSION_MAJOR  0
#define FILE_VERSION_MINOR  1

#define ST_FIRSTKEY_ASSERT(expr) \
{ \
    if (!(expr)) \
    { \
        indexPage.DumpKeys(); \
        DumpKeys(dataPages); \
        ST_ASSERT(expr); \
    } \
}

#ifdef DEBUG
#define ASSERT_INDEX_CONSISTENCY AssertIndexConsistency
//#define ASSERT_INDEX_CONSISTENCY()
#define ASSERT_FILEINDEX_CONSISTENCY AssertFileIndexConsistency
#else
#define ASSERT_INDEX_CONSISTENCY()
#define ASSERT_FILEINDEX_CONSISTENCY()
#endif

//static void DumpKeys(StorageDataPage** dataPages)
//{
//    ReadBuffer  firstKey;
//    
//    for (int i = 0; i < 256; i++)
//    {
//        if (dataPages[i] == NULL)
//            printf("StorageData: %i: NULL\n", i);
//        else
//        {
//            firstKey = dataPages[i]->FirstKey();
//            printf("StorageData: %i: %.*s\n", i, P(&firstKey));
//        }
//    }
//}

static int KeyCmp(const uint32_t a, const uint32_t b)
{
    if (a < b)
        return -1;
    else if (a > b)
        return 1;
    else
        return 0;       
}

static uint32_t Key(StoragePage* page)
{
    return page->GetOffset();
}

StorageFile::StorageFile()
{
    uint32_t i;
        
    indexPageSize = DEFAULT_INDEXPAGE_SIZE;
    dataPageSize = DEFAULT_DATAPAGE_SIZE;
    numDataPageSlots = DEFAULT_NUM_DATAPAGES;

    isOverflowing = false;
    isDeleted = false;
    newFile = true;
    
    indexPage.SetOffset(INDEXPAGE_OFFSET);
    indexPage.SetPageSize(indexPageSize);
    indexPage.SetNumDataPageSlots(numDataPageSlots);

    dataPages = (StorageDataPage**) malloc(sizeof(StorageDataPage*) * numDataPageSlots);
    for (i = 0; i < numDataPageSlots; i++)
        dataPages[i] = NULL;
    numDataPages = 0;
    
    fd = INVALID_FD;
}

StorageFile::~StorageFile()
{
    // dirtyPages' destructor automatically calls this
    // but it is only running after all pages are free'd.
    dirtyPages.Clear();
    
    for (uint32_t u = 0; u < numDataPages; u++)
    {
        if (dataPages[u] != NULL)
        {
            if (!dataPages[u]->IsDirty())
                DCACHE->Checkout(dataPages[u]);

            DCACHE->FreePage(dataPages[u]);
        }
    }
    free(dataPages);
}

void StorageFile::Open(const char* filepath_)
{
    filepath.Write(filepath_);
    filepath.NullTerminate();

    fd = FS_Open(filepath.GetBuffer(), FS_READWRITE | FS_CREATE);

    if (fd == INVALID_FD)
        STOP_FAIL(1, "Cannot open file (%s)", filepath.GetBuffer());

    if (FS_FileSize(fd) > 0)
        Read();
}

void StorageFile::Close()
{
    FS_FileClose(fd);
    
    fd = INVALID_FD;
}

void StorageFile::SetStorageFileIndex(uint32_t fileIndex_)
{
    StoragePage*    it;
    
    fileIndex = fileIndex_;
    
    indexPage.SetStorageFileIndex(fileIndex);
    for (it = dirtyPages.First(); it != NULL; it = dirtyPages.Next(it))
        it->SetStorageFileIndex(fileIndex);
    
    ASSERT_FILEINDEX_CONSISTENCY();
}

bool StorageFile::Get(ReadBuffer& key, ReadBuffer& value)
{
    int32_t index;
    bool    ret;
    
    index = Locate(key);
    
    if (index < 0)
        return false;
    
    ret = dataPages[index]->Get(key, value);
    if (ret && !dataPages[index]->IsDirty())
        DCACHE->RegisterHit(dataPages[index]);

    return ret;
}

bool StorageFile::Set(ReadBuffer& key, ReadBuffer& value, bool copy)
{
    int32_t             index;
    ReadBuffer          rb;
    StorageDataPage*    dataPage;

    ASSERT_INDEX_CONSISTENCY();
    
    if (key.GetLength() + value.GetLength() > DATAPAGE_MAX_KV_SIZE(dataPageSize))
        return false;
    
    index = Locate(key);
    
    if (index < 0)
    {
        // file is empty, allocate first data page
        index = 0;
        ST_ASSERT(numDataPages == 0);
        ST_ASSERT(dataPages[index] == NULL);
        dataPage = dataPages[index] = DCACHE->GetPage();
        DCACHE->Checkin(dataPage);
        dataPage->SetStorageFileIndex(fileIndex);
        dataPage->SetOffset(DATAPAGE_OFFSET(index));
        dataPage->SetPageSize(dataPageSize);
        dataPage->SetFile(this);
        ST_ASSERT(numDataPages == numDataPageSlots - indexPage.freeDataPages.GetLength());
        numDataPages++;
        indexPage.Add(key, index, true); // TODO
        MarkPageDirty(&indexPage);
    }
    else
    {
        dataPage = dataPages[index];
        rb = dataPage->FirstKey();
        if (ReadBuffer::LessThan(key, rb))
        {
            STORAGE_TRACE("Changing index entry for %u to %R", index, &key);
            indexPage.Update(key, index, true);
            MarkPageDirty(&indexPage);
        }
    }
    
    if (dataPage->HasCursors())
    {
        // no cache because cursor has the detached copy
        dataPage->Detach();
        ST_ASSERT(dataPage->HasCursors() == false);
    }
    
    if (!dataPage->Set(key, value, copy))
    {
        if (!dataPage->IsDirty())
            DCACHE->RegisterHit(dataPage);

        ASSERT_INDEX_CONSISTENCY();
        return true; // nothing changed
    }
    
    MarkPageDirty(dataPage);
    
    if (dataPage->IsOverflowing())
        SplitDataPage(index);
    
    ASSERT_INDEX_CONSISTENCY();
    return true;
}

void StorageFile::Delete(ReadBuffer& key)
{
    bool        updateIndex;
    int32_t     index;
    ReadBuffer  firstKey;
    
    index = Locate(key);
    
    if (index < 0)
        return;

    ASSERT_INDEX_CONSISTENCY();

    updateIndex = false;
    firstKey = dataPages[index]->FirstKey();
    if (BUFCMP(&key, &firstKey))
        updateIndex = true;

    if (dataPages[index]->HasCursors())
    {
        dataPages[index]->Detach();
        ST_ASSERT(dataPages[index]->HasCursors() == false);
    }

    dataPages[index]->Delete(key);
    MarkPageDirty(dataPages[index]);

    if (dataPages[index]->IsEmpty())
    {
        indexPage.Remove(key);
        MarkPageDirty(&indexPage);
        numDataPages--;
        dataPages[index]->SetDeleted(true);
        dataPages[index] = NULL;
        STORAGE_TRACE("Delete empty");
    }
    else if (updateIndex)
    {
        firstKey = dataPages[index]->FirstKey();
        indexPage.Update(firstKey, index, true);
        MarkPageDirty(&indexPage);
        STORAGE_TRACE("Delete updateIndex");
    }
    else
        STORAGE_TRACE("Delete else");

    ASSERT_INDEX_CONSISTENCY();
}

ReadBuffer StorageFile::FirstKey()
{
    return indexPage.FirstKey();
}

bool StorageFile::IsEmpty()
{
    return indexPage.IsEmpty();
}

bool StorageFile::IsNew()
{
    return newFile;
}

bool StorageFile::IsOverflowing()
{
    return isOverflowing || indexPage.IsOverflowing();
}

StorageFile* StorageFile::SplitFile()
{
    StorageFile*    newFile;
    uint32_t        index, newIndex, num;
    
    ST_ASSERT(numDataPageSlots == numDataPages);
    ASSERT_INDEX_CONSISTENCY();
    
    // TODO: do thing for cursors
    
    newFile = new StorageFile;
    newFile->indexPageSize = indexPageSize;
    newFile->dataPageSize = dataPageSize;
    newFile->numDataPageSlots = numDataPageSlots;
    newFile->indexPage.SetPageSize(indexPageSize);
    newFile->indexPage.SetNumDataPageSlots(numDataPageSlots);
    
    ReorderPages();
    
    num = numDataPages;
    for (index = numDataPageSlots / 2, newIndex = 0; index < num; index++, newIndex++)
    {
        ST_ASSERT(dataPages[index]->IsEmpty() != true);
        newFile->dataPages[newIndex] = dataPages[index];
        dataPages[index] = NULL;
        newFile->dataPages[newIndex]->SetOffset(DATAPAGE_OFFSET(newIndex));
        newFile->dataPages[newIndex]->SetFile(newFile);

        ST_ASSERT(numDataPages == numDataPageSlots - indexPage.freeDataPages.GetLength());
        ST_ASSERT(newFile->numDataPages == newFile->numDataPageSlots - newFile->indexPage.freeDataPages.GetLength());
        numDataPages--;
        newFile->numDataPages++;
        ST_ASSERT(newFile->numDataPages < newFile->numDataPageSlots);

        indexPage.Remove(newFile->dataPages[newIndex]->FirstKey());
        newFile->indexPage.Add(newFile->dataPages[newIndex]->FirstKey(), newIndex, true);
    }
    
    isOverflowing = false;
    for (index = 0; index < numDataPages; index++)
    {
        if (dataPages[index]->IsOverflowing())
            SplitDataPage(index);
    }
    ST_ASSERT(isOverflowing == false);
    
    newFile->isOverflowing = false;
    for (index = 0; index < newFile->numDataPages; index++)
    {
        if (newFile->dataPages[index]->IsOverflowing())
            newFile->SplitDataPage(index);
    }
    ST_ASSERT(newFile->isOverflowing == false);
    
    ReorderFile();
    newFile->ReorderFile();

    ASSERT_INDEX_CONSISTENCY();
    newFile->AssertIndexConsistency();

    return newFile;
}

StorageFile* StorageFile::SplitFileByKey(ReadBuffer& startKey)
{
    StorageDataPage*    page;
    StorageDataPage*    newPage;
    StorageFile*        newFile;
    uint32_t            index;
    uint32_t            splitIndex;
    int32_t             ret;
    int                 cmp;
    
    newFile = new StorageFile;
    newFile->indexPageSize = indexPageSize;
    newFile->dataPageSize = dataPageSize;
    newFile->numDataPageSlots = numDataPageSlots;
    newFile->indexPage.SetPageSize(indexPageSize);
    newFile->indexPage.SetNumDataPageSlots(numDataPageSlots);

    ReadRest();
    ReorderPages();
    
    ret = Locate(startKey);
    ST_ASSERT(ret >= 0);
    splitIndex = (uint32_t) ret;
    
    for (index = 0; index < numDataPageSlots; index++)
    {
        page = dataPages[index];
        if (page == NULL)
            break;          // the pages are already ordered
    
        cmp = ReadBuffer::Cmp(startKey, page->FirstKey());
        if (cmp <= 0)
        {
            if (cmp < 0 && index == splitIndex)
            {
                // startKey is not the first key of the page so we have to split it
                newPage = page->SplitDataPageByKey(startKey);
                newFile->dataPages[index] = newPage;
            }
            else
            {
                ST_ASSERT(numDataPages == numDataPageSlots - indexPage.freeDataPages.GetLength());
                newFile->dataPages[index] = dataPages[index];
                indexPage.Remove(dataPages[index]->FirstKey()); 
                dataPages[index]->SetDeleted(true);
                dataPages[index] = NULL;
                numDataPages--;
            }

            ST_ASSERT(newFile->numDataPages == newFile->numDataPageSlots - newFile->indexPage.freeDataPages.GetLength());
            newFile->dataPages[index]->SetOffset(DATAPAGE_OFFSET(index));
            newFile->dataPages[index]->SetFile(newFile);
            newFile->dataPages[index]->Invalidate();
            MarkPageDirty(newFile->dataPages[index]);
            newFile->indexPage.Add(newFile->dataPages[index]->FirstKey(), index, true);     
            newFile->numDataPages++;
        }
    }
    
    ReorderFile();
    newFile->ReorderFile();
    
    return newFile;
}

void StorageFile::Read()
{
    unsigned            i;
    int                 length;
    Buffer              buffer;
    ReadBuffer          readBuffer;
    StorageFileHeader   header;
    bool                ret;

    ASSERT_INDEX_CONSISTENCY();
    
    newFile = false;

    // read file header
    buffer.Allocate(STORAGEFILE_HEADER_LENGTH);
    if (FS_FileRead(fd, (void*) buffer.GetBuffer(), STORAGEFILE_HEADER_LENGTH) != STORAGEFILE_HEADER_LENGTH)
        STOP_FAIL(1, "Invalid file header (%s)", filepath.GetBuffer());
    buffer.SetLength(STORAGEFILE_HEADER_LENGTH);
    if (!header.Read(buffer))
        STOP_FAIL(1, "Invalid file header (%s)", filepath.GetBuffer());
    
    // read index header
    buffer.Allocate(INDEXPAGE_HEADER_SIZE);
    if (FS_FileRead(fd, (void*) buffer.GetBuffer(), INDEXPAGE_HEADER_SIZE) != INDEXPAGE_HEADER_SIZE)
        STOP_FAIL(1, "Invalid file header (%s)", filepath.GetBuffer());
    readBuffer.Wrap(buffer);
    ret = readBuffer.ReadLittle32(indexPageSize);
    ST_ASSERT(ret == true);
    ST_ASSERT(indexPageSize != 0);
    
    readBuffer.Advance(sizeof(uint32_t));
    ret = readBuffer.ReadLittle32(dataPageSize);
    ST_ASSERT(ret == true);
    ST_ASSERT(dataPageSize != 0);

    readBuffer.Advance(sizeof(uint32_t));
    ret = readBuffer.ReadLittle32(numDataPageSlots);
    ST_ASSERT(ret == true);
    ST_ASSERT(numDataPageSlots != 0);

    indexPage.SetOffset(INDEXPAGE_OFFSET);
    indexPage.SetPageSize(indexPageSize);
    indexPage.SetNew(false);
    indexPage.SetNumDataPageSlots(numDataPageSlots);
    
    // read index page
    buffer.Allocate(indexPageSize);
    length = FS_FileRead(fd, (void*) buffer.GetBuffer(), indexPageSize);
    ST_ASSERT(length == (int) indexPageSize);
    buffer.SetLength(length);
    readBuffer.Wrap(buffer);
    indexPage.Read(readBuffer);

    ST_DEBUG_ASSERT(GetSize() <= (uint64_t) FS_FileSize(fd));
    
    // allocate memory for data page slots
    if (dataPages != NULL)
        free(dataPages);
    dataPages = (StorageDataPage**) malloc(sizeof(StorageDataPage*) * numDataPageSlots);
    for (i = 0; i < numDataPageSlots; i++)
        dataPages[i] = NULL;
    numDataPages = indexPage.NumEntries();  
    
    ST_ASSERT(numDataPages == numDataPageSlots - indexPage.freeDataPages.GetLength());
    ASSERT_INDEX_CONSISTENCY();
}

void StorageFile::ReadRest()
{
    StorageKeyIndex*        it;
    uint32_t*               uit;
    SortedList<uint32_t>    indexes;

    ASSERT_INDEX_CONSISTENCY();

    // As IO occurs in order, sort by index rather than by name
    FOREACH (it, indexPage.keys)
    {
        if (dataPages[it->index] == NULL)
            indexes.Add(it->index);
    }

    FOREACH (uit, indexes)
        LoadDataPage(*uit);

    ASSERT_INDEX_CONSISTENCY();
}

void StorageFile::WriteRecovery(StorageRecoveryLog& recoveryLog)
{
    Buffer          buffer;
    StoragePage*    it;

    ASSERT_FILEINDEX_CONSISTENCY();

    if (IsNew())
    {
        buffer.AppendLittle32(fileIndex);
        if (!recoveryLog.WriteOp(RECOVERY_OP_FILE, sizeof(fileIndex), buffer))
            STOP_FAIL(1, "Recovery failed when writing file (%s)", recoveryLog.GetFilename());
        return;
    }

    FOREACH (it, dirtyPages)
    {
        if (it->IsNew())
            continue;
        ST_ASSERT(it->buffer.GetLength() <= it->GetPageSize());
        ST_ASSERT(it->buffer.GetLength() >= DATAPAGE_FIX_OVERHEAD);
        // it->buffer contains the old page
        
//        buffer.Allocate(it->GetPageSize());
//        if (!it->CheckWrite(buffer))
//            continue;

        if (it->NeedRewrite())
        {
            if (!recoveryLog.WriteOp(RECOVERY_OP_PAGE, it->GetPageSize(), it->buffer))
                STOP_FAIL(1, "Recovery failed when writing file (%s)", recoveryLog.GetFilename());
        }
        else
        {
            it->WriteHeader(buffer);
            if (!recoveryLog.WriteOp(RECOVERY_OP_PAGE, buffer.GetLength(), buffer))
                STOP_FAIL(1, "Recovery failed when writing file (%s)", recoveryLog.GetFilename());
        }
    }<commit_msg>Under regular (rewrite) conditions, only write actual page into the recovery page, not always an entire pageSize block.<commit_after>#include "StorageFile.h"
#include "StorageFileHeader.h"
#include "StorageDataCache.h"
#include "StorageCursor.h"
#include "StorageRecoveryLog.h"
#include "StorageDefaults.h"
#include "System/FileSystem.h"
#include "stdio.h"

#define INDEXPAGE_OFFSET    (STORAGEFILE_HEADER_LENGTH+INDEXPAGE_HEADER_SIZE)

#define DATAPAGE_OFFSET(idx) (INDEXPAGE_OFFSET+indexPageSize+(idx)*dataPageSize)
#define DATAPAGE_INDEX(offs) (((offs)-INDEXPAGE_OFFSET-indexPageSize)/dataPageSize)
#define FILE_TYPE           "ScalienDB data file"
#define FILE_VERSION_MAJOR  0
#define FILE_VERSION_MINOR  1

#define ST_FIRSTKEY_ASSERT(expr) \
{ \
    if (!(expr)) \
    { \
        indexPage.DumpKeys(); \
        DumpKeys(dataPages); \
        ST_ASSERT(expr); \
    } \
}

#ifdef DEBUG
#define ASSERT_INDEX_CONSISTENCY AssertIndexConsistency
//#define ASSERT_INDEX_CONSISTENCY()
#define ASSERT_FILEINDEX_CONSISTENCY AssertFileIndexConsistency
#else
#define ASSERT_INDEX_CONSISTENCY()
#define ASSERT_FILEINDEX_CONSISTENCY()
#endif

//static void DumpKeys(StorageDataPage** dataPages)
//{
//    ReadBuffer  firstKey;
//    
//    for (int i = 0; i < 256; i++)
//    {
//        if (dataPages[i] == NULL)
//            printf("StorageData: %i: NULL\n", i);
//        else
//        {
//            firstKey = dataPages[i]->FirstKey();
//            printf("StorageData: %i: %.*s\n", i, P(&firstKey));
//        }
//    }
//}

static int KeyCmp(const uint32_t a, const uint32_t b)
{
    if (a < b)
        return -1;
    else if (a > b)
        return 1;
    else
        return 0;       
}

static uint32_t Key(StoragePage* page)
{
    return page->GetOffset();
}

StorageFile::StorageFile()
{
    uint32_t i;
        
    indexPageSize = DEFAULT_INDEXPAGE_SIZE;
    dataPageSize = DEFAULT_DATAPAGE_SIZE;
    numDataPageSlots = DEFAULT_NUM_DATAPAGES;

    isOverflowing = false;
    isDeleted = false;
    newFile = true;
    
    indexPage.SetOffset(INDEXPAGE_OFFSET);
    indexPage.SetPageSize(indexPageSize);
    indexPage.SetNumDataPageSlots(numDataPageSlots);

    dataPages = (StorageDataPage**) malloc(sizeof(StorageDataPage*) * numDataPageSlots);
    for (i = 0; i < numDataPageSlots; i++)
        dataPages[i] = NULL;
    numDataPages = 0;
    
    fd = INVALID_FD;
}

StorageFile::~StorageFile()
{
    // dirtyPages' destructor automatically calls this
    // but it is only running after all pages are free'd.
    dirtyPages.Clear();
    
    for (uint32_t u = 0; u < numDataPages; u++)
    {
        if (dataPages[u] != NULL)
        {
            if (!dataPages[u]->IsDirty())
                DCACHE->Checkout(dataPages[u]);

            DCACHE->FreePage(dataPages[u]);
        }
    }
    free(dataPages);
}

void StorageFile::Open(const char* filepath_)
{
    filepath.Write(filepath_);
    filepath.NullTerminate();

    fd = FS_Open(filepath.GetBuffer(), FS_READWRITE | FS_CREATE);

    if (fd == INVALID_FD)
        STOP_FAIL(1, "Cannot open file (%s)", filepath.GetBuffer());

    if (FS_FileSize(fd) > 0)
        Read();
}

void StorageFile::Close()
{
    FS_FileClose(fd);
    
    fd = INVALID_FD;
}

void StorageFile::SetStorageFileIndex(uint32_t fileIndex_)
{
    StoragePage*    it;
    
    fileIndex = fileIndex_;
    
    indexPage.SetStorageFileIndex(fileIndex);
    for (it = dirtyPages.First(); it != NULL; it = dirtyPages.Next(it))
        it->SetStorageFileIndex(fileIndex);
    
    ASSERT_FILEINDEX_CONSISTENCY();
}

bool StorageFile::Get(ReadBuffer& key, ReadBuffer& value)
{
    int32_t index;
    bool    ret;
    
    index = Locate(key);
    
    if (index < 0)
        return false;
    
    ret = dataPages[index]->Get(key, value);
    if (ret && !dataPages[index]->IsDirty())
        DCACHE->RegisterHit(dataPages[index]);

    return ret;
}

bool StorageFile::Set(ReadBuffer& key, ReadBuffer& value, bool copy)
{
    int32_t             index;
    ReadBuffer          rb;
    StorageDataPage*    dataPage;

    ASSERT_INDEX_CONSISTENCY();
    
    if (key.GetLength() + value.GetLength() > DATAPAGE_MAX_KV_SIZE(dataPageSize))
        return false;
    
    index = Locate(key);
    
    if (index < 0)
    {
        // file is empty, allocate first data page
        index = 0;
        ST_ASSERT(numDataPages == 0);
        ST_ASSERT(dataPages[index] == NULL);
        dataPage = dataPages[index] = DCACHE->GetPage();
        DCACHE->Checkin(dataPage);
        dataPage->SetStorageFileIndex(fileIndex);
        dataPage->SetOffset(DATAPAGE_OFFSET(index));
        dataPage->SetPageSize(dataPageSize);
        dataPage->SetFile(this);
        ST_ASSERT(numDataPages == numDataPageSlots - indexPage.freeDataPages.GetLength());
        numDataPages++;
        indexPage.Add(key, index, true); // TODO
        MarkPageDirty(&indexPage);
    }
    else
    {
        dataPage = dataPages[index];
        rb = dataPage->FirstKey();
        if (ReadBuffer::LessThan(key, rb))
        {
            STORAGE_TRACE("Changing index entry for %u to %R", index, &key);
            indexPage.Update(key, index, true);
            MarkPageDirty(&indexPage);
        }
    }
    
    if (dataPage->HasCursors())
    {
        // no cache because cursor has the detached copy
        dataPage->Detach();
        ST_ASSERT(dataPage->HasCursors() == false);
    }
    
    if (!dataPage->Set(key, value, copy))
    {
        if (!dataPage->IsDirty())
            DCACHE->RegisterHit(dataPage);

        ASSERT_INDEX_CONSISTENCY();
        return true; // nothing changed
    }
    
    MarkPageDirty(dataPage);
    
    if (dataPage->IsOverflowing())
        SplitDataPage(index);
    
    ASSERT_INDEX_CONSISTENCY();
    return true;
}

void StorageFile::Delete(ReadBuffer& key)
{
    bool        updateIndex;
    int32_t     index;
    ReadBuffer  firstKey;
    
    index = Locate(key);
    
    if (index < 0)
        return;

    ASSERT_INDEX_CONSISTENCY();

    updateIndex = false;
    firstKey = dataPages[index]->FirstKey();
    if (BUFCMP(&key, &firstKey))
        updateIndex = true;

    if (dataPages[index]->HasCursors())
    {
        dataPages[index]->Detach();
        ST_ASSERT(dataPages[index]->HasCursors() == false);
    }

    dataPages[index]->Delete(key);
    MarkPageDirty(dataPages[index]);

    if (dataPages[index]->IsEmpty())
    {
        indexPage.Remove(key);
        MarkPageDirty(&indexPage);
        numDataPages--;
        dataPages[index]->SetDeleted(true);
        dataPages[index] = NULL;
        STORAGE_TRACE("Delete empty");
    }
    else if (updateIndex)
    {
        firstKey = dataPages[index]->FirstKey();
        indexPage.Update(firstKey, index, true);
        MarkPageDirty(&indexPage);
        STORAGE_TRACE("Delete updateIndex");
    }
    else
        STORAGE_TRACE("Delete else");

    ASSERT_INDEX_CONSISTENCY();
}

ReadBuffer StorageFile::FirstKey()
{
    return indexPage.FirstKey();
}

bool StorageFile::IsEmpty()
{
    return indexPage.IsEmpty();
}

bool StorageFile::IsNew()
{
    return newFile;
}

bool StorageFile::IsOverflowing()
{
    return isOverflowing || indexPage.IsOverflowing();
}

StorageFile* StorageFile::SplitFile()
{
    StorageFile*    newFile;
    uint32_t        index, newIndex, num;
    
    ST_ASSERT(numDataPageSlots == numDataPages);
    ASSERT_INDEX_CONSISTENCY();
    
    // TODO: do thing for cursors
    
    newFile = new StorageFile;
    newFile->indexPageSize = indexPageSize;
    newFile->dataPageSize = dataPageSize;
    newFile->numDataPageSlots = numDataPageSlots;
    newFile->indexPage.SetPageSize(indexPageSize);
    newFile->indexPage.SetNumDataPageSlots(numDataPageSlots);
    
    ReorderPages();
    
    num = numDataPages;
    for (index = numDataPageSlots / 2, newIndex = 0; index < num; index++, newIndex++)
    {
        ST_ASSERT(dataPages[index]->IsEmpty() != true);
        newFile->dataPages[newIndex] = dataPages[index];
        dataPages[index] = NULL;
        newFile->dataPages[newIndex]->SetOffset(DATAPAGE_OFFSET(newIndex));
        newFile->dataPages[newIndex]->SetFile(newFile);

        ST_ASSERT(numDataPages == numDataPageSlots - indexPage.freeDataPages.GetLength());
        ST_ASSERT(newFile->numDataPages == newFile->numDataPageSlots - newFile->indexPage.freeDataPages.GetLength());
        numDataPages--;
        newFile->numDataPages++;
        ST_ASSERT(newFile->numDataPages < newFile->numDataPageSlots);

        indexPage.Remove(newFile->dataPages[newIndex]->FirstKey());
        newFile->indexPage.Add(newFile->dataPages[newIndex]->FirstKey(), newIndex, true);
    }
    
    isOverflowing = false;
    for (index = 0; index < numDataPages; index++)
    {
        if (dataPages[index]->IsOverflowing())
            SplitDataPage(index);
    }
    ST_ASSERT(isOverflowing == false);
    
    newFile->isOverflowing = false;
    for (index = 0; index < newFile->numDataPages; index++)
    {
        if (newFile->dataPages[index]->IsOverflowing())
            newFile->SplitDataPage(index);
    }
    ST_ASSERT(newFile->isOverflowing == false);
    
    ReorderFile();
    newFile->ReorderFile();

    ASSERT_INDEX_CONSISTENCY();
    newFile->AssertIndexConsistency();

    return newFile;
}

StorageFile* StorageFile::SplitFileByKey(ReadBuffer& startKey)
{
    StorageDataPage*    page;
    StorageDataPage*    newPage;
    StorageFile*        newFile;
    uint32_t            index;
    uint32_t            splitIndex;
    int32_t             ret;
    int                 cmp;
    
    newFile = new StorageFile;
    newFile->indexPageSize = indexPageSize;
    newFile->dataPageSize = dataPageSize;
    newFile->numDataPageSlots = numDataPageSlots;
    newFile->indexPage.SetPageSize(indexPageSize);
    newFile->indexPage.SetNumDataPageSlots(numDataPageSlots);

    ReadRest();
    ReorderPages();
    
    ret = Locate(startKey);
    ST_ASSERT(ret >= 0);
    splitIndex = (uint32_t) ret;
    
    for (index = 0; index < numDataPageSlots; index++)
    {
        page = dataPages[index];
        if (page == NULL)
            break;          // the pages are already ordered
    
        cmp = ReadBuffer::Cmp(startKey, page->FirstKey());
        if (cmp <= 0)
        {
            if (cmp < 0 && index == splitIndex)
            {
                // startKey is not the first key of the page so we have to split it
                newPage = page->SplitDataPageByKey(startKey);
                newFile->dataPages[index] = newPage;
            }
            else
            {
                ST_ASSERT(numDataPages == numDataPageSlots - indexPage.freeDataPages.GetLength());
                newFile->dataPages[index] = dataPages[index];
                indexPage.Remove(dataPages[index]->FirstKey()); 
                dataPages[index]->SetDeleted(true);
                dataPages[index] = NULL;
                numDataPages--;
            }

            ST_ASSERT(newFile->numDataPages == newFile->numDataPageSlots - newFile->indexPage.freeDataPages.GetLength());
            newFile->dataPages[index]->SetOffset(DATAPAGE_OFFSET(index));
            newFile->dataPages[index]->SetFile(newFile);
            newFile->dataPages[index]->Invalidate();
            MarkPageDirty(newFile->dataPages[index]);
            newFile->indexPage.Add(newFile->dataPages[index]->FirstKey(), index, true);     
            newFile->numDataPages++;
        }
    }
    
    ReorderFile();
    newFile->ReorderFile();
    
    return newFile;
}

void StorageFile::Read()
{
    unsigned            i;
    int                 length;
    Buffer              buffer;
    ReadBuffer          readBuffer;
    StorageFileHeader   header;
    bool                ret;

    ASSERT_INDEX_CONSISTENCY();
    
    newFile = false;

    // read file header
    buffer.Allocate(STORAGEFILE_HEADER_LENGTH);
    if (FS_FileRead(fd, (void*) buffer.GetBuffer(), STORAGEFILE_HEADER_LENGTH) != STORAGEFILE_HEADER_LENGTH)
        STOP_FAIL(1, "Invalid file header (%s)", filepath.GetBuffer());
    buffer.SetLength(STORAGEFILE_HEADER_LENGTH);
    if (!header.Read(buffer))
        STOP_FAIL(1, "Invalid file header (%s)", filepath.GetBuffer());
    
    // read index header
    buffer.Allocate(INDEXPAGE_HEADER_SIZE);
    if (FS_FileRead(fd, (void*) buffer.GetBuffer(), INDEXPAGE_HEADER_SIZE) != INDEXPAGE_HEADER_SIZE)
        STOP_FAIL(1, "Invalid file header (%s)", filepath.GetBuffer());
    readBuffer.Wrap(buffer);
    ret = readBuffer.ReadLittle32(indexPageSize);
    ST_ASSERT(ret == true);
    ST_ASSERT(indexPageSize != 0);
    
    readBuffer.Advance(sizeof(uint32_t));
    ret = readBuffer.ReadLittle32(dataPageSize);
    ST_ASSERT(ret == true);
    ST_ASSERT(dataPageSize != 0);

    readBuffer.Advance(sizeof(uint32_t));
    ret = readBuffer.ReadLittle32(numDataPageSlots);
    ST_ASSERT(ret == true);
    ST_ASSERT(numDataPageSlots != 0);

    indexPage.SetOffset(INDEXPAGE_OFFSET);
    indexPage.SetPageSize(indexPageSize);
    indexPage.SetNew(false);
    indexPage.SetNumDataPageSlots(numDataPageSlots);
    
    // read index page
    buffer.Allocate(indexPageSize);
    length = FS_FileRead(fd, (void*) buffer.GetBuffer(), indexPageSize);
    ST_ASSERT(length == (int) indexPageSize);
    buffer.SetLength(length);
    readBuffer.Wrap(buffer);
    indexPage.Read(readBuffer);

    ST_DEBUG_ASSERT(GetSize() <= (uint64_t) FS_FileSize(fd));
    
    // allocate memory for data page slots
    if (dataPages != NULL)
        free(dataPages);
    dataPages = (StorageDataPage**) malloc(sizeof(StorageDataPage*) * numDataPageSlots);
    for (i = 0; i < numDataPageSlots; i++)
        dataPages[i] = NULL;
    numDataPages = indexPage.NumEntries();  
    
    ST_ASSERT(numDataPages == numDataPageSlots - indexPage.freeDataPages.GetLength());
    ASSERT_INDEX_CONSISTENCY();
}

void StorageFile::ReadRest()
{
    StorageKeyIndex*        it;
    uint32_t*               uit;
    SortedList<uint32_t>    indexes;

    ASSERT_INDEX_CONSISTENCY();

    // As IO occurs in order, sort by index rather than by name
    FOREACH (it, indexPage.keys)
    {
        if (dataPages[it->index] == NULL)
            indexes.Add(it->index);
    }

    FOREACH (uit, indexes)
        LoadDataPage(*uit);

    ASSERT_INDEX_CONSISTENCY();
}

void StorageFile::WriteRecovery(StorageRecoveryLog& recoveryLog)
{
    Buffer          buffer;
    StoragePage*    it;

    ASSERT_FILEINDEX_CONSISTENCY();

    if (IsNew())
    {
        buffer.AppendLittle32(fileIndex);
        if (!recoveryLog.WriteOp(RECOVERY_OP_FILE, sizeof(fileIndex), buffer))
            STOP_FAIL(1, "Recovery failed when writing file (%s)", recoveryLog.GetFilename());
        return;
    }

    FOREACH (it, dirtyPages)
    {
        if (it->IsNew())
            continue;
        ST_ASSERT(it->buffer.GetLength() <= it->GetPageSize());
        ST_ASSERT(it->buffer.GetLength() >= DATAPAGE_FIX_OVERHEAD);
        // it->buffer contains the old page
        
//        buffer.Allocate(it->GetPageSize());
//        if (!it->CheckWrite(buffer))
//            continue;

        if (it->NeedRewrite())
        {
            if (!recoveryLog.WriteOp(RECOVERY_OP_PAGE, it->buffer.GetLength(), it->buffer))
                STOP_FAIL(1, "Recovery failed when writing file (%s)", recoveryLog.GetFilename());
        }
        else
        {
            it->WriteHeader(buffer);
            if (!recoveryLog.WriteOp(RECOVERY_OP_PAGE, buffer.GetLength(), buffer))
                STOP_FAIL(1, "Recovery failed when writing file (%s)", recoveryLog.GetFilename());
        }
    }
{'fraction_non_alphanumeric': 0.12472189081997871, 'fraction_numerical': 0.005352594073453068, 'mean_word_length': 2.6642249527410207, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 217, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6944392', 'n_tokens_mistral': 9983, 'n_tokens_neox': 9721, 'n_words': 1759}