Remove unnecessary calls to WAL::flushAllPages and clear the dirty flag when flushing pages

src/include/processor/operator/persistent/batch_insert.h

@@ -44,10 +44,7 @@ struct BatchInsertSharedState {
     }
     inline common::row_idx_t getNumRows() { return numRows.load(); }
     // NOLINTNEXTLINE(readability-make-member-function-const): Semantically non-const.
-    inline void logBatchInsertWALRecord() {
-        wal->logCopyTableRecord(table->getTableID());
-        wal->flushAllPages();
-    }
+    inline void logBatchInsertWALRecord() { wal->logCopyTableRecord(table->getTableID()); }
     inline void updateNumTuplesForTable() { table->updateNumTuplesByValue(getNumRows()); }
 };
 

src/include/storage/buffer_manager/bm_file_handle.h

@@ -80,6 +80,9 @@ class PageState {
         KU_ASSERT(getState(stateAndVersion.load()) == LOCKED);
         stateAndVersion &= ~DIRTY_MASK;
     }
+    // Meant to be used when flushing in a single thread.
+    // Should not be used if other threads are modifying the page state
+    inline void clearDirtyWithoutLock() { stateAndVersion &= ~DIRTY_MASK; }
     inline bool isDirty() const { return stateAndVersion & DIRTY_MASK; }
     uint64_t getStateAndVersion() const { return stateAndVersion.load(); }
 

src/main/database.cpp

@@ -217,8 +217,8 @@ void Database::commit(Transaction* transaction, bool skipCheckpointForTestingRec
     // Note: committing and stopping new transactions can be done in any order. This
     // order allows us to throw exceptions if we have to wait a lot to stop.
     transactionManager->commitButKeepActiveWriteTransaction(transaction);
-    wal->flushAllPages();
     if (skipCheckpointForTestingRecovery) {
+        wal->flushAllPages();
         transactionManager->allowReceivingNewTransactions();
         return;
     }

src/processor/operator/persistent/node_batch_insert.cpp

@@ -237,8 +237,6 @@ void NodeBatchInsert::finalize(ExecutionContext* context) {
     if (nodeSharedState->globalIndexBuilder) {
         nodeSharedState->globalIndexBuilder->finalize(context);
     }
-    // Batch Insert wal record needs to be logged after PrimaryKeyIndex::prepareCommit
-    // so that the wal pages get flushed before the index is re-initialized.
     sharedState->logBatchInsertWALRecord();
     auto outputMsg = stringFormat("{} tuples have been copied to the {} table.",
         sharedState->getNumRows(), info->tableEntry->getName());

src/storage/buffer_manager/buffer_manager.cpp

@@ -322,6 +322,7 @@ void BufferManager::flushIfDirtyWithoutLock(BMFileHandle& fileHandle, page_idx_t
     if (pageState->isDirty()) {
         fileHandle.getFileInfo()->writeFile(getFrame(fileHandle, pageIdx), fileHandle.getPageSize(),
             pageIdx * fileHandle.getPageSize());
+        pageState->clearDirtyWithoutLock();
     }
 }