[DNM] Raft integration and TMT storage review

dbms/src/Raft/RaftService.cpp

@@ -25,6 +25,7 @@ RaftService::RaftService(const std::string & address_, DB::Context & db_context_
 
 RaftService::~RaftService()
 {
+    // REVIEW: can we remove this mutex?
     std::lock_guard<std::mutex> lock{mutex};
     grpc_server->Shutdown();
     grpc_server->Wait();
@@ -42,10 +43,12 @@ grpc::Status RaftService::ApplyCommandBatch(grpc::ServerContext * grpc_context,
     {
         kvstore->report(rctx);
 
+        // REVIEW: persisting interval is?
         persist_handle = background_pool.addTask([&, this] { return kvstore->tryPersistAndReport(rctx); });
         flush_handle = background_pool.addTask([&] { return region_table.tryFlushRegions(); });
 
         enginepb::CommandRequestBatch request;
+        // REVIEW: should we use EOS flag?
         while (stream->Read(&request))
         {
             applyCommand(rctx, request);
@@ -56,11 +59,13 @@ grpc::Status RaftService::ApplyCommandBatch(grpc::ServerContext * grpc_context,
         tryLogCurrentException(log, "gRPC ApplyCommandBatch on " + address + " error");
     }
 
+    // REVIEW: is this removing will cause persisting missing? For example, call write some data, and then this conn broke before flushing
     if (persist_handle)
         background_pool.removeTask(persist_handle);
     if (flush_handle)
         background_pool.removeTask(flush_handle);
 
+    // REVIEW: should we use OK?
     return grpc::Status::CANCELLED;
 }
 

dbms/src/Storages/StorageMergeTree.cpp

@@ -134,6 +134,7 @@ BlockInputStreams StorageMergeTree::read(
     auto res = reader.read(column_names, query_info, context, processed_stage, max_block_size, num_streams, 0);
     const ASTSelectQuery * select_query = typeid_cast<const ASTSelectQuery *>(query_info.query.get());
 
+    // REVIEW: move engine specified processing to SelectExecutor may be better
     if (data.merging_params.mode == MergeTreeData::MergingParams::Mutable ||
         data.merging_params.mode == MergeTreeData::MergingParams::Txn)
     {
@@ -180,6 +181,10 @@ BlockOutputStreamPtr StorageMergeTree::write(const ASTPtr & query, const Setting
     const ASTInsertQuery * insert_query = typeid_cast<const ASTInsertQuery *>(&*query);
     const ASTDeleteQuery * delete_query = typeid_cast<const ASTDeleteQuery *>(&*query);
 
+    // REVIEW: move engine specified processing to lower layer may be better:
+    //   switch (engine type) -> MergeTreeBlockOutputStream
+    //                        -> TxnMergeTreeBlockOutputStream
+    //                        -> MutableMergeTreeBlockOutputStream
     if (data.merging_params.mode == MergeTreeData::MergingParams::Txn)
     {
         res = std::make_shared<TxnMergeTreeBlockOutputStream>(*this);

dbms/src/Storages/StorageMergeTree.h

@@ -88,6 +88,7 @@ friend class TxnMergeTreeBlockOutputStream;
 
     bool checkTableCanBeDropped() const override;
 
+    // REVIEW: get/set TableInfo => data
     const TableInfo & getTableInfo() const { return data.table_info; }
     void setTableInfo(const TableInfo & table_info_) { data.table_info = table_info_; }
 

dbms/src/Storages/Transaction/KVStore.cpp

@@ -66,6 +66,7 @@ void KVStore::onSnapshot(RegionPtr new_region, Context * context)
         {
             LOG_DEBUG(log, "KVStore::onSnapshot: previous " << old_region->toString(true) << " ; new " << new_region->toString(true));
 
+            // REVIEW: in what case this can happen? rngine crushed?
             if (old_region->getProbableIndex() >= new_region->getProbableIndex())
             {
                 LOG_DEBUG(log, "KVStore::onSnapshot: discard new region because of index is outdated");
@@ -82,11 +83,13 @@ void KVStore::onSnapshot(RegionPtr new_region, Context * context)
 
         if (new_region->isPendingRemove())
         {
+            // REVIEW: here we remove the region in persister, then below the region is added to persister again
             removeRegion(region_id, context);
             return;
         }
     }
 
+    // REVIEW: we don't need to persist region on any change.
     region_persister.persist(new_region);
 
     if (tmt_ctx)
@@ -160,6 +163,7 @@ void KVStore::onServiceCommand(const enginepb::CommandRequestBatch & cmds, RaftC
                     auto [it, ok] = regions.emplace(new_region->id(), new_region);
                     if (!ok)
                     {
+                        // REVIEW: do we need to compare the old one and the new one?
                         // definitely, any region's index is greater or equal than the initial one, discard it.
                         continue;
                     }
@@ -177,11 +181,17 @@ void KVStore::onServiceCommand(const enginepb::CommandRequestBatch & cmds, RaftC
                     region_persister.persist(region);
             }
 
+            // REVIEW: if process crushed here ...
+
             if (tmt_ctx)
                 tmt_ctx->region_table.splitRegion(curr_region, split_regions);
+
+            // REVIEW: do region_table need to updateRegion of the splitted regions?
         }
         else
         {
+            // REVIEW: is the persisting order OK?
+
             if (tmt_ctx)
                 tmt_ctx->region_table.updateRegion(curr_region, table_ids);
 
@@ -272,6 +282,9 @@ void KVStore::removeRegion(RegionID region_id, Context * context)
     }
 
     region_persister.drop(region_id);
+
+    // REVIEW: if process crushed here, then when the process start again, the region_table will not find this region, is it OK?
+
     if (context)
         context->getTMTContext().region_table.removeRegion(region);
 }

dbms/src/Storages/Transaction/KVStore.h

@@ -58,6 +58,7 @@ class KVStore final : private boost::noncopyable
 
     std::atomic<Timepoint> last_try_persist_time = Clock::now();
 
+    // REVIEW: the task_mutex looks should be one mutex per region than one big lock in kvstore
     // onServiceCommand and onSnapshot should not be called concurrently
     mutable std::mutex task_mutex;
 

dbms/src/Storages/Transaction/Region.cpp

@@ -541,6 +541,7 @@ void Region::setPendingRemove()
 
 size_t Region::dataSize() const { return cf_data_size; }
 
+// REVIEW: reset persist_parm here?
 void Region::markPersisted() { last_persist_time = Clock::now(); }
 
 Timepoint Region::lastPersistTime() const { return last_persist_time; }

dbms/src/Storages/Transaction/Region.h

@@ -139,6 +139,7 @@ class Region : public std::enable_shared_from_this<Region>
     TableID insert(const std::string & cf, const TiKVKey & key, const TiKVValue & value);
     TableID remove(const std::string & cf, const TiKVKey & key);
 
+    // REVIEW: 'node' seems not a good name here
     using BatchInsertNode = std::tuple<const TiKVKey *, const TiKVValue *, const String *>;
     void batchInsert(std::function<bool(BatchInsertNode &)> && f);
 
@@ -164,6 +165,8 @@ class Region : public std::enable_shared_from_this<Region>
 
     void markPersisted();
     Timepoint lastPersistTime() const;
+
+    // REVIEW: use a dirty flag instead?
     size_t persistParm() const;
     void decPersistParm(size_t x);
     void incPersistParm();

dbms/src/Storages/Transaction/RegionMeta.cpp

@@ -42,6 +42,8 @@ RegionMeta RegionMeta::deserialize(ReadBuffer & buf)
     return RegionMeta(peer, region, apply_state, applied_term, pending_remove);
 }
 
+// REVIEW: lock? be carefull, can be easily deadlock.
+//  or use member `const RegionID region_id`
 RegionID RegionMeta::regionId() const { return region.id(); }
 
 UInt64 RegionMeta::peerId() const
@@ -101,6 +103,7 @@ void RegionMeta::doSetApplied(UInt64 index, UInt64 term)
     applied_term = term;
 }
 
+// REVIEW: should move this notification to doSetApplied?
 void RegionMeta::notifyAll() { cv.notify_all(); }
 
 UInt64 RegionMeta::appliedIndex() const
@@ -127,9 +130,11 @@ enginepb::CommandResponse RegionMeta::toCommandResponse() const
 
 RegionMeta::RegionMeta(RegionMeta && rhs) : region_id(rhs.regionId())
 {
+    // REVIEW: lock rhs
     peer = std::move(rhs.peer);
     region = std::move(rhs.region);
     apply_state = std::move(rhs.apply_state);
+    // REVIEW: set rhs.* to init state
     applied_term = rhs.applied_term;
     pending_remove = rhs.pending_remove;
 }
@@ -176,6 +181,7 @@ void RegionMeta::doSetPendingRemove() { pending_remove = true; }
 void RegionMeta::waitIndex(UInt64 index)
 {
     std::unique_lock<std::mutex> lock(mutex);
+    // REVIEW: should we lock inside the closure function?
     cv.wait(lock, [this, index] { return pending_remove || apply_state.applied_index() >= index; });
 }
 
@@ -193,6 +199,7 @@ UInt64 RegionMeta::confVer() const
 
 void RegionMeta::reset(RegionMeta && rhs)
 {
+    // REVIEW: lock rhs
     std::lock_guard<std::mutex> lock(mutex);
 
     peer = std::move(rhs.peer);
@@ -225,6 +232,7 @@ void RegionMeta::execChangePeer(
 
     switch (change_peer_request.change_type())
     {
+        // REVIEW: throws when meet `AddNode`?
         case eraftpb::ConfChangeType::AddNode:
         case eraftpb::ConfChangeType::AddLearnerNode:
         {

dbms/src/Storages/Transaction/RegionMeta.h

@@ -70,6 +70,7 @@ class RegionMeta
     bool isPendingRemove() const;
     void setPendingRemove();
 
+    // REVIEW: assign is a better name
     void reset(RegionMeta && other);
 
     friend bool operator==(const RegionMeta & meta1, const RegionMeta & meta2)
@@ -94,6 +95,7 @@ class RegionMeta
     void doSetApplied(UInt64 index, UInt64 term);
 
 private:
+    // REVIEW: we should make sure all these member are deepcopy, (eg: region.peers())
     metapb::Peer peer;
     metapb::Region region;
     raft_serverpb::RaftApplyState apply_state;
@@ -110,7 +112,9 @@ class RegionMeta
 
 // When we create a region peer, we should initialize its log term/index > 0,
 // so that we can force the follower peer to sync the snapshot first.
+// REVIEW: why initialize term = 5?
 static constexpr UInt64 RAFT_INIT_LOG_TERM = 5;
+// REVIEW: raft log initialize position should be 0, after received snapshot, the position should move to 5
 static constexpr UInt64 RAFT_INIT_LOG_INDEX = 5;
 
 inline raft_serverpb::RaftApplyState initialApplyState()

dbms/src/Storages/Transaction/RegionPersister.cpp

@@ -11,6 +11,8 @@ extern const int LOGICAL_ERROR;
 
 void RegionPersister::drop(RegionID region_id)
 {
+    // REVIEW: need mutex?
+
     WriteBatch wb;
     wb.delPage(region_id);
     page_storage.write(wb);
@@ -21,6 +23,7 @@ void RegionPersister::persist(const RegionPtr & region, enginepb::CommandRespons
     // Support only on thread persist.
     std::lock_guard<std::mutex> lock(mutex);
 
+    // REVIEW: can we just region->resetPersistParm(void)?
     size_t persist_parm = region->persistParm();
     doPersist(region, response);
     region->markPersisted();
@@ -40,6 +43,7 @@ void RegionPersister::doPersist(const RegionPtr & region, enginepb::CommandRespo
     }
 
     MemoryWriteBuffer buffer;
+    // REVIEW: add some warning log here if region is too large
     size_t region_size = region->serialize(buffer, response);
     if (unlikely(region_size > std::numeric_limits<UInt32>::max()))
         throw Exception("Region is too big to persist", ErrorCodes::LOGICAL_ERROR);

dbms/src/Storages/Transaction/applySnapshot.cpp

@@ -15,12 +15,16 @@ void applySnapshot(KVStorePtr kvstore, RequestReader read, Context * context)
 
     enginepb::SnapshotRequest request;
     auto ok = read(&request);
+    // REVIEW: use two 'throw' here, 'not ok' and 'no state'
     if (!ok || !request.has_state())
         throw Exception("Failed to read snapshot state", ErrorCodes::LOGICAL_ERROR);
+
     const auto & state = request.state();
     pingcap::kv::RegionClientPtr region_client = nullptr;
     auto meta = RegionMeta(state.peer(), state.region(), state.apply_state());
+
     Region::RegionClientCreateFunc region_client_create = [&](pingcap::kv::RegionVerID id) -> pingcap::kv::RegionClientPtr {
+        // context may be null in test cases.
         if (context)
         {
             auto & tmt_ctx = context->getTMTContext();
@@ -44,6 +48,8 @@ void applySnapshot(KVStorePtr kvstore, RequestReader read, Context * context)
             auto cf_name = data.cf();
             auto key = TiKVKey();
             auto value = TiKVValue();
+            // REVIEW: the batch inserting logic is OK, but the calling stack is weird.
+            //   May be we should just do lock action on each node-inserting, it's also fast when lock contention is low.
             region->batchInsert([&](Region::BatchInsertNode & node) -> bool {
                 if (it == cf_data.end())
                     return false;