Refactor code

Signed-off-by: Jin Hai <[email protected]>

Author: JinHai-CN

src/storage/catalog/new_catalog_static.cpp

@@ -114,11 +114,7 @@ Status NewCatalog::InitCatalog(KVInstance *kv_instance, TxnTimeStamp checkpoint_
     }
 
     auto InitBlockColumn = [&](ColumnMeta &column_meta) {
-        Status status = column_meta.LoadSet();
-        if (!status.ok()) {
-            return status;
-        }
-        return Status::OK();
+        return column_meta.LoadSet();
     };
     auto InitBlock = [&](BlockMeta &block_meta) {
         SharedPtr<Vector<SharedPtr<ColumnDef>>> column_defs_ptr;
@@ -133,12 +129,7 @@ Status NewCatalog::InitCatalog(KVInstance *kv_instance, TxnTimeStamp checkpoint_
                 return status;
             }
         }
-        Status status = block_meta.LoadSet(checkpoint_ts);
-        if (!status.ok()) {
-            return status;
-        }
-
-        return Status::OK();
+        return block_meta.LoadSet(checkpoint_ts);
     };
     auto InitSegment = [&](SegmentMeta &segment_meta) {
         auto [block_ids, status] = segment_meta.GetBlockIDs1();
@@ -376,9 +367,9 @@ Status NewCatalog::GetAllMemIndexes(KVInstance *kv_instance,
             SegmentIndexMeta segment_index_meta(segment_id, table_index_meta);
 
             SharedPtr<MemIndex> mem_index;
-            Status status = segment_index_meta.GetMemIndex(mem_index);
-            if (!status.ok()) {
-                return status;
+            Status status_mem_index = segment_index_meta.GetMemIndex(mem_index);
+            if (!status_mem_index.ok()) {
+                return status_mem_index;
             }
 
             mem_indexes.push_back(mem_index);
@@ -433,7 +424,7 @@ Status NewCatalog::GetAllMemIndexes(KVInstance *kv_instance,
         const String &db_id_str = (*db_id_strs_ptr)[i];
         const String &db_name = (*db_names_ptr)[i];
         DBMeeta db_meta(db_id_str, *kv_instance);
-        Status status = TraverseDB(db_meta, db_name);
+        status = TraverseDB(db_meta, db_name);
         if (!status.ok()) {
             return status;
         }
@@ -509,7 +500,7 @@ Status NewCatalog::AddNewTable(DBMeeta &db_meta,
                                TxnTimeStamp commit_ts,
                                const SharedPtr<TableDef> &table_def,
                                Optional<TableMeeta> &table_meta) {
-    // Create table key value pair
+    // Create table a key value pair
     KVInstance &kv_instance = db_meta.kv_instance();
     String table_key = KeyEncode::CatalogTableKey(db_meta.db_id_str(), *table_def->table_name(), commit_ts);
     Status status = kv_instance.Put(table_key, table_id_str);
@@ -584,7 +575,7 @@ Status NewCatalog::AddNewTableIndex(TableMeeta &table_meta,
                                     TxnTimeStamp commit_ts,
                                     const SharedPtr<IndexBase> &index_base,
                                     Optional<TableIndexMeeta> &table_index_meta) {
-    // Create index key value pair
+    // Create index a key value pair
     KVInstance &kv_instance = table_meta.kv_instance();
     const String &index_name = *index_base->index_name_;
     String index_key = KeyEncode::CatalogIndexKey(table_meta.db_id_str(), table_meta.table_id_str(), index_name, commit_ts);
@@ -664,12 +655,7 @@ Status NewCatalog::CleanTableIndex(TableIndexMeeta &table_index_meta,
         }
     }
 
-    Status status = table_index_meta.UninitSet1(usage_flag);
-    if (!status.ok()) {
-        return status;
-    }
-
-    return Status::OK();
+    return table_index_meta.UninitSet1(usage_flag);
 }
 
 // Status NewCatalog::AddNewSegment(TableMeeta &table_meta, SegmentID segment_id, Optional<SegmentMeta> &segment_meta) {
@@ -697,11 +683,7 @@ Status NewCatalog::AddNewSegment1(TableMeeta &table_meta, TxnTimeStamp commit_ts
         return status;
     }
     segment_meta.emplace(segment_id, table_meta);
-    status = segment_meta->InitSet();
-    if (!status.ok()) {
-        return status;
-    }
-    return Status::OK();
+    return segment_meta->InitSet();
 }
 
 Status NewCatalog::AddNewSegmentWithID(TableMeeta &table_meta, TxnTimeStamp commit_ts, Optional<SegmentMeta> &segment_meta, SegmentID segment_id) {
@@ -710,11 +692,7 @@ Status NewCatalog::AddNewSegmentWithID(TableMeeta &table_meta, TxnTimeStamp comm
         return status;
     }
     segment_meta.emplace(segment_id, table_meta);
-    status = segment_meta->InitSet();
-    if (!status.ok()) {
-        return status;
-    }
-    return Status::OK();
+    return segment_meta->InitSet();
 }
 
 Status NewCatalog::LoadFlushedSegment1(TableMeeta &table_meta, const WalSegmentInfo &segment_info, TxnTimeStamp checkpoint_ts) {
@@ -859,7 +837,7 @@ Status NewCatalog::AddNewBlockForTransform(SegmentMeta &segment_meta, TxnTimeSta
     return Status::OK();
 }
 //
-// Status NewCatalog::AddNewTableIndexForTransform(TableMeeta &table_meta,TxnTimeStamp commit_ts,Optional<TableIndexMeta> &table_index_meta) {
+// Status NewCatalog::AddNewTableIndexForTransform(TableMeeta &table_meta, TxnTimeStamp commit_ts, Optional<TableIndexMeta> &table_index_meta) {
 //     Status status;
 //     BlockID block_id;
 //     std::tie(block_id, status) = table_meta.AddBlockID1(commit_ts);
@@ -1246,7 +1224,7 @@ Status NewCatalog::GetBlockVisibleRange(BlockMeta &block_meta, TxnTimeStamp begi
     BufferHandle buffer_handle = version_buffer->Load();
     SharedPtr<BlockLock> block_lock;
     {
-        Status status = block_meta.GetBlockLock(block_lock);
+        status = block_meta.GetBlockLock(block_lock);
         if (!status.ok()) {
             return status;
         }
@@ -1371,9 +1349,9 @@ Status NewCatalog::GetBlockFilePaths(BlockMeta &block_meta, Vector<String> &file
     if (!status.ok()) {
         return status;
     }
-    if (!column_def) {
-        for (const auto &column_def : *column_defs_ptr) {
-            ColumnMeta column_meta(column_def->id(), block_meta);
+    if (column_def == nullptr) {
+        for (const auto &column_def_ptr : *column_defs_ptr) {
+            ColumnMeta column_meta(column_def_ptr->id(), block_meta);
             status = GetBlockColumnFilePaths(column_meta, file_paths);
             if (!status.ok()) {
                 return status;
@@ -1539,7 +1517,7 @@ Status NewCatalog::CheckSegmentRowsVisible(SegmentMeta &segment_meta, TxnTimeSta
     }
     for (BlockID block_id : *block_ids_ptr) {
         BlockMeta block_meta(block_id, segment_meta);
-        Status status = NewCatalog::SetBlockDeleteBitmask(block_meta, begin_ts, bitmask);
+        status = NewCatalog::SetBlockDeleteBitmask(block_meta, begin_ts, bitmask);
         if (!status.ok()) {
             return status;
         }

src/storage/new_txn/new_txn_manager.cpp

@@ -192,7 +192,7 @@ UniquePtr<NewTxn> NewTxnManager::BeginRecoveryTxn() {
 
     //    current_ts_ += 2;
     prepare_commit_ts_ = current_ts_ + 2;
-    TxnTimeStamp commit_ts = current_ts_ + 2; // Will not be used, actually.
+    TxnTimeStamp commit_ts = current_ts_ + 2; // Will not be used
     TxnTimeStamp begin_ts = current_ts_ + 1;  // current_ts_ > 0
 
     // Create txn instance
@@ -239,14 +239,6 @@ TxnTimeStamp NewTxnManager::GetWriteCommitTS(SharedPtr<NewTxn> txn) {
     return commit_ts;
 }
 
-TxnTimeStamp NewTxnManager::GetReplayWriteCommitTS(NewTxn *txn) {
-    std::lock_guard guard(locker_);
-    prepare_commit_ts_ += 2;
-    TxnTimeStamp commit_ts = prepare_commit_ts_;
-    txn->SetTxnWrite();
-    return commit_ts;
-}
-
 bool NewTxnManager::CheckConflict1(NewTxn *txn, String &conflict_reason, bool &retry_query) {
     TxnTimeStamp begin_ts = txn->BeginTS();
     TxnTimeStamp commit_ts = txn->CommitTS();
@@ -423,7 +415,7 @@ void NewTxnManager::CommitBottom(NewTxn *txn) {
 
     // ensure notify top half orderly per commit_ts
     while (!bottom_txns_.empty()) {
-        auto iter = bottom_txns_.begin();
+        iter = bottom_txns_.begin();
         TxnTimeStamp it_ts = iter->first;
         SharedPtr<NewTxn> it_txn = iter->second;
         if (current_ts_ > it_ts || it_ts > prepare_commit_ts_) {
@@ -509,7 +501,7 @@ void NewTxnManager::CleanupTxn(NewTxn *txn) {
     TransactionID txn_id = txn->TxnID();
     LOG_DEBUG(fmt::format("Cleanup txn, id: {}, begin_ts: {}", txn_id, begin_ts));
     if (is_write_transaction) {
-        // For write txn, we need to update the state: committing->committed, rollbacking->rollbacked
+        // For writing txn, we need to update the state: committing->committed, rollbacking->rollbacked
         TxnState txn_state = txn->GetTxnState();
         switch (txn_state) {
             case TxnState::kCommitting: {
@@ -580,16 +572,6 @@ void NewTxnManager::CleanupTxnBottomNolock(TransactionID txn_id, TxnTimeStamp be
     }
 }
 
-bool NewTxnManager::InCheckpointProcess(TxnTimeStamp commit_ts) {
-    std::lock_guard guard(locker_);
-    if (commit_ts > ckp_begin_ts_) {
-        LOG_TRACE(fmt::format("Full/Delta checkpoint begin at {}, cur txn commit_ts: {}, swap to new wal file", ckp_begin_ts_, commit_ts));
-        ckp_begin_ts_ = UNCOMMIT_TS;
-        return true;
-    }
-    return false;
-}
-
 void NewTxnManager::PrintAllKeyValue() const {
     std::cout << String("All store key and value: ") << std::endl;
     std::cout << kv_store_->ToString() << std::endl;

src/storage/new_txn/new_txn_manager.cppm

@@ -64,8 +64,6 @@ public:
 
     TxnTimeStamp GetWriteCommitTS(SharedPtr<NewTxn> txn);
 
-    TxnTimeStamp GetReplayWriteCommitTS(NewTxn *txn);
-
     // Optional<String> CheckTxnConflict(NewTxn *txn);
 
     bool CheckConflict1(NewTxn *txn, String &conflict_reason, bool &retry_query);
@@ -118,8 +116,6 @@ private:
     void CleanupTxnBottomNolock(TransactionID txn_id, TxnTimeStamp begin_ts);
 
 public:
-    bool InCheckpointProcess(TxnTimeStamp commit_ts);
-
     // Only used by follower and learner when received the replicated log from leader
     void SetStartTS(TxnTimeStamp new_start_ts) { current_ts_ = new_start_ts; }
 

src/storage/storage.cppm

@@ -15,15 +15,8 @@
 module;
 
 import stl;
-import config;
-import catalog;
-import txn_manager;
-import buffer_manager;
 import wal_manager;
-import object_storage_process;
 import log_file;
-import persistence_manager;
-import virtual_store;
 import status;
 
 export module storage;
@@ -33,13 +26,19 @@ namespace infinity {
 class CleanupInfoTracer;
 class ResultCacheManager;
 class NewCatalog;
+class Catalog;
 class NewTxnManager;
+class TxnManager;
 class KVStore;
 class KVInstance;
 class PeriodicTriggerThread;
 class CompactionProcessor;
 class BGTaskProcessor;
 class BGMemIndexTracer;
+class ObjectStorageProcess;
+class Config;
+class BufferManager;
+class PersistenceManager;
 
 export enum class ReaderInitPhase {
     kInvalid,