Merge free pages after rollback

Previously only happened during a checkpoint
But if a rollback occurs after data has been written, there would be a bunch of small page ranges in the free page cache from data appended to the end of the file, which can be merged and truncated

Author: benjaminwinger

src/include/storage/free_space_manager.h

@@ -48,6 +48,8 @@ class FreeSpaceManager {
     // each checkpoint we remove any already-evicted pages.
     void clearEvictedBufferManagerEntriesIfNeeded(BufferManager* bufferManager);
 
+    void mergeFreePages(FileHandle* fileHandle);
+
 private:
     PageRange splitPageRange(PageRange chunk, common::page_idx_t numRequiredPages);
     void mergePageRanges(free_list_t newInitialEntries, FileHandle* fileHandle);

src/include/storage/page_manager.h

@@ -40,6 +40,8 @@ class PageManager : public PageAllocator {
     void finalizeCheckpoint();
     void rollbackCheckpoint() { freeSpaceManager->rollbackCheckpoint(); }
 
+    void mergeFreePages(FileHandle* fileHandle);
+
     common::row_idx_t getNumFreeEntries() const { return freeSpaceManager->getNumEntries(); }
     std::vector<PageRange> getFreeEntries(common::row_idx_t startOffset,
         common::row_idx_t endOffset) const {

src/storage/free_space_manager.cpp

@@ -18,7 +18,8 @@ static FreeSpaceManager::sorted_free_list_t& getFreeList(
     return freeLists[level];
 }
 
-FreeSpaceManager::FreeSpaceManager() : freeLists{}, numEntries(0), needClearEvictedEntries(false){};
+FreeSpaceManager::FreeSpaceManager()
+    : freeLists{}, numEntries(0), needClearEvictedEntries(false) {};
 
 common::idx_t FreeSpaceManager::getLevel(common::page_idx_t numPages) {
     // level is exponent of largest power of 2 that is <= numPages
@@ -201,6 +202,10 @@ void FreeSpaceManager::finalizeCheckpoint(FileHandle* fileHandle) {
     uncheckpointedFreePageRanges.clear();
 }
 
+void FreeSpaceManager::mergeFreePages(FileHandle* fileHandle) {
+    mergePageRanges(std::move(uncheckpointedFreePageRanges), fileHandle);
+}
+
 void FreeSpaceManager::resetFreeLists() {
     freeLists.clear();
     numEntries = 0;

src/storage/local_storage/local_storage.cpp

@@ -88,8 +88,9 @@ void LocalStorage::rollback() {
     for (auto& optimisticAllocator : optimisticAllocators) {
         optimisticAllocator->rollback();
     }
-    auto* bufferManager = mm->getBufferManager();
-    PageManager::Get(clientContext)->clearEvictedBMEntriesIfNeeded(bufferManager);
+    auto& pageManager = *PageManager::Get(clientContext);
+    pageManager.mergeFreePages(pageManager.getDataFH());
+    pageManager.clearEvictedBMEntriesIfNeeded(mm->getBufferManager());
 }
 
 } // namespace storage

src/storage/page_manager.cpp

@@ -43,6 +43,14 @@ void PageManager::freeImmediatelyRewritablePageRange(FileHandle* fileHandle, Pag
     }
 }
 
+void PageManager::mergeFreePages(FileHandle* fileHandle) {
+    if constexpr (ENABLE_FSM) {
+        common::UniqLock lck{mtx};
+        freeSpaceManager->mergeFreePages(fileHandle);
+        ++version;
+    }
+}
+
 void PageManager::serialize(common::Serializer& serializer) {
     freeSpaceManager->serialize(serializer);
 }