// 创建storageServer并启动
StorageDaemon::main(){
    auto storageServer = std::make_unique<nebula::storage::StorageServer>(
      localhost, metaAddrsRet.value(), paths, FLAGS_wal_path, FLAGS_listener_path);
    storageServer->start()  
}

// 初始化kv存储引擎
StorageServer::start(){
  LOG(INFO) << "Init kvstore";
  kvstore_ = getStoreInstance();
}

// 实例化nebula存储引擎并初始化
StorageServer::getStoreInstance(){
    auto nbStore = std::make_unique<kvstore::NebulaStore>(
        std::move(options), ioThreadPool_, localHost_, workers_);
    nbStore->init()        
}

// 从data_path中载入part分区数据
NebulaStore::init(){
    loadPartFromDataPath()
}

//遍历data_path目录为每个space创建一个rocksdb引擎
NebulaStore::loadPartFromDataPath(){
    std::vector<folly::Future<std::pair<GraphSpaceID, std::unique_ptr<KVEngine>>>> futures;
  std::vector<std::string> enginesPath;
  for (auto& path : options_.dataPaths_) {
    //这里其实就是data_path，比如:/usr/local/nebula/data
    auto rootPath = folly::stringPrintf("%s/nebula", path.c_str());
    //获取每个data_pah子目录，其实就是space
    // 比如 /usr/local/nebula/data/1、/usr/local/nebula/data/2
    auto dirs = fs::FileUtils::listAllDirsInDir(rootPath.c_str());
    //遍历space目录，为每个space目录创建一个rocksdb
    for (auto& dir : dirs) {
      //目录名称就是spaceId，比如/usr/local/nebula/data/1中的1就是spaceId
      GraphSpaceID spaceId = folly::to<GraphSpaceID>(dir);
      // 目录为0的跳过
      if (spaceId == 0) {
        // skip the system space, only handle data space here.
        continue;
      }
      enginesPath.emplace_back(rootPath + "/" + dir);
      //异步创建rocksdb存储引擎
      futures.emplace_back(newEngineAsync(spaceId, path, options_.walPath_));
    }
    // space和kvEngine列表的映射关系
    std::unordered_map<GraphSpaceID, std::vector<std::unique_ptr<KVEngine>>> spaceEngines;
    // 等待并获取前面的异步创建引擎列表
    auto tries = folly::collectAll(futures).get();
    for (auto& t : tries) {
        //p就是个二元组，第一个元素为space，第二个元素为kvEngine
        auto&& p = t.value();
        auto spaceIt = spaceEngines.find(p.first);
        if (spaceIt == spaceEngines.end()) {
        spaceIt = spaceEngines.emplace(p.first, std::vector<std::unique_ptr<KVEngine>>()).first;
      }
        spaceIt->second.emplace_back(std::move(p.second));
    }
    // 遍历所有space存储引擎
    for (auto& spaceEngine : spaceEngines) {
        GraphSpaceID spaceId = spaceEngine.first;
        for (auto& engine : spaceEngine.second) {
          // part分片与part副本节点集合，后面raft需要用到
          std::map<PartitionID, Peers> partRaftPeers;
          partRaftPeers.emplace(partId, raftPeers);
          //建立起partId和part映射关系，后面操作某partId的数据时就是通过part进行的，part中其实保留了kvEngine的引用
          for (auto& it : partRaftPeers) {
                auto part = newPart(spaceId, partId, enginePtr, isLearner, addrs);
                auto ret = iter->second->parts_.emplace(partId, part);
          }
        }
    }
  }
}
// 创建rocksdb引擎
NebulaStore::newEngineAsync() {
      engine = std::make_unique<RocksEngine>(
          spaceId, vIdLen, dataPath, walPath, options_.mergeOp_, cfFactory);
}

// 通过上下文获取到storageClient然后调用获取属性的方法
GetVerticesExecutor::getVertices() {
    StorageClient *storageClient = qctx()->getStorageClient();
    storageClient->getProps(param,
                 std::move(vertices),
                 gv->props(),
                 nullptr,
                 gv->exprs(),
                 gv->dedup(),
                 gv->orderBy(),
                 gv->getValidLimit(),
                 gv->filter())
}

// 通过vertexId获取到part所在节点信息
// 请求对应节点的vertexId信息
StorageRpcRespFuture<cpp2::GetPropResponse> StorageClient::getProps(){
    // 建立vertexId和part节点映射关系
    auto status = clusterIdsToHosts(param.space, input.rows, std::move(cbStatus).value());
  // 为不同的节点构建对应的request
  cpp2::GetPropRequest> requests;
  for (auto& c : clusters) {
    auto& host = c.first;
    auto& req = requests[host];
    req.space_id_ref() = param.space;
    req.parts_ref() = std::move(c.second);
  }
  //调用rpc请求storaged获取点的属性信息
  return collectResponse(
      param.evb, std::move(requests), [](ThriftClientType* client, const cpp2::GetPropRequest& r) {
        return client->future_getProps(r);
      });
}

//通过（vid % numParts + 1）计算出对应的part
// 这种方式也预示着space设定的分区数量一旦创建不可更改，否则数据就会映射不上
StorageClientBase::clusterIdsToHosts(){
    //获取space空间有多少个part
  auto status = metaClient_->partsNum(spaceId);
  std::unordered_map<PartitionID, HostAddr> leaders;
    //建立起partId和leader节点的映射关系
  for (int32_t partId = 1; partId <= numParts; ++partId) {
    auto leader = getLeader(spaceId, partId);
    leaders[partId] = std::move(leader).value();
  }
  //
   for (auto& id : ids) {
        // 计算vid对应的partId
        PartitionID part = vid % numParts + 1;
        const auto& leader = leaders[part];
        clusters[leader][part].emplace_back(std::move(id));
  }
  return clusters;
}

folly::Future<cpp2::GetPropResponse> GraphStorageServiceHandler::future_getProps(
    const cpp2::GetPropRequest& req) {
  auto* processor = GetPropProcessor::instance(env_, &kGetPropCounters, readerPool_.get());
  RETURN_FUTURE(processor);
}

GetPropProcessor::doProcess(){
    if (!FLAGS_query_concurrently) {
     //为了方便理解，这里直接选择单线程
      runInSingleThread(req);
    } else {
      runInMultipleThread(req);
    }
}

GetPropProcessor::runInSingleThread(){
    // 构建tag查询计划,这个计划相当于构建一个DAG
    // 里面有完整的依赖信息
    auto plan = buildTagPlan(&contexts_.front(), &resultDataSet_);
    // 遍历待查询的part分区
    for (const auto& partEntry : req.get_parts()) {
      auto partId = partEntry.first;
      for (const auto& row : partEntry.second) {
        auto vId = row.values[0].getStr();
        // 执行查询计划
        plan.go(partId, vId);
      }
    }    
}

// 构建执行计划DAG
GetPropProcessor::buildTagPlan(RuntimeContext* context,
                                                     nebula::DataSet* result) {
  StoragePlan<VertexID> plan;
  std::vector<TagNode*> tags;
  for (const auto& tc : tagContext_.propContexts_) {
    // TagNode获取vid指定tag的属性列表，TagNode will return a DataSet of specified props of tagId
    // 这里面就有从rocksdb中读取tag的value属性
    // 查询语句指定多少个tag，这里就会有多少个TagNode
    // 比如：fetch prop on tag_1,tag_2,tag_3 idxxx,那这里就有3个TagNode
    auto tag = std::make_unique<TagNode>(context, &tagContext_, tc.first, &tc.second);
    tags.emplace_back(tag.get());
    plan.addNode(std::move(tag));
  }
  // GetTagPropNode用于收集vid对应多个tag的属性
  auto output = std::make_unique<GetTagPropNode>(
      context, tags, result, filter_ == nullptr ? nullptr : filter_->clone(), limit_, &tagContext_);
  for (auto* tag : tags) {
    output->addDependency(tag);
  }
  plan.addNode(std::move(output));
  return plan;
}

TagNode::doExecute(){
    // 根据vid和tagId构建rocksdb中key
    key_ = NebulaKeyUtils::tagKey(context_->vIdLen(), partId, vId, tagId_);
    // 从响应分片的rocksdb中获取value，此value为经过RowWriter编码的value
    ret = context_->env()->kvstore_->get(context_->spaceId(), partId, key_, &value_);
    // RowReaderWrapper设置schema和对应的value，后面用于GetTagPropNode收集prop做准备
    reader_.reset(*schemas_, value_);
}


// 收集vid对应所有tag的属性值列表
GetTagPropNode::doExecute(){
    // 这个集合用于保存查询到的vid对应不同tag的属性集合
    List row;
    // vertexId is the first column
    if (context_->isIntId()) {
      row.emplace_back(*reinterpret_cast<const int64_t*>(vId.data()));
    } else {
      row.emplace_back(vId);
    }
    // 遍历本次要查询的所有tag
    for (auto* tagNode : tagNodes_) {
      ret = tagNode->collectTagPropsIfValid(
          // 收集tagNode中属性到row中
          [&row, vIdLen, isIntId, tagNode, this](){
            auto status = QueryUtils::collectVertexProps(
                key, vIdLen, isIntId, reader, props, row, expCtx_.get(), tagNode->getTagName());
          });
    }
}

QueryUtils::collectVertexProps(){
  // 遍历tag中的属性列表，挨个的取出来放入到list中，这个list就是GetTagPropNode::doExecute中的row
  for (const auto& prop : *props) {
      auto value = QueryUtils::readVertexProp(key, vIdLen, isIntId, reader, prop);
      list.emplace_back(value.value());
    }
}

// 读取属性值
QueryUtils::readVertexProp(){
  switch (prop.propInKeyType_) {
      // prop in value
      case PropContext::PropInKeyType::NONE: {
        // ddl中的属性，这里的readValue就是RowReader根据schame获取字段的offset，然后获取对应的field值
        return readValue(reader, prop.name_, prop.field_);
      }
      case PropContext::PropInKeyType::VID: {
        // 属性类型为vid
        auto vId = NebulaKeyUtils::getVertexId(vIdLen, key);
        if (isIntId) {
          return *reinterpret_cast<const int64_t*>(vId.data());
        } else {
          return vId.subpiece(0, vId.find_first_of('\0')).toString();
        }
      }
      case PropContext::PropInKeyType::TAG: {
        // 属性类型为tagId
        auto tag = NebulaKeyUtils::getTagId(vIdLen, key);
        return tag;
      }
    }
}

// KVFilter在启动的rocksdb的时候就需要设置进去，所以这里直接看rocksdb启动的代码即可
StorageServer::getStoreInstance(){
	kvstore::KVOptions options;
	// 这个StorageCompactionFilter就是nebula给予VKFilter实现的
	options.compaction_filter_factory = std::make_unique<StorageCompactionFilterFactoryBuilder>(schemaMan_.get(), indexMan_.get());
	status = rocksdb::DB::Open(options, path, &db);
}

class StorageCompactionFilter final : public kvstore::KVFilter {

  //这里是filter的入口
  bool filter(int level,
              GraphSpaceID spaceId,
              const folly::StringPiece& key,
              const folly::StringPiece& val) const override {
    if (level < FLAGS_min_level_for_custom_filter) {
      // for upper level such as L0/L1, we don't go through the custom
      // validation to achieve better performance
      return false;
    }
    if (NebulaKeyUtils::isTag(vIdLen_, key)) {
      return !tagValid(spaceId, key, val);
    } else if (NebulaKeyUtils::isEdge(vIdLen_, key)) {
      return !edgeValid(spaceId, key, val);
    } else if (IndexKeyUtils::isIndexKey(key)) {
      return !indexValid(spaceId, key, val);
    } else if (!FLAGS_use_vertex_key && NebulaKeyUtils::isVertex(key)) {
      return true;
    } else if (NebulaKeyUtils::isLock(vIdLen_, key)) {
      return !lockValid(spaceId, key);
    } else {
      // skip uuid/system/operation
      VLOG(3) << "Skip the system key inside, key " << key;
    }
    return false;
  }
 
 //判断tag是否有效,这里其实就是比对ttl字段的值对比当前时间是否超过ttl
 private:
  bool tagValid(GraphSpaceID spaceId,
                const folly::StringPiece& key,
                const folly::StringPiece& val) const {
    ......
    if (ttlExpired(schema.get(), reader.get())) {
      VLOG(3) << "Ttl expired";
      return false;
    }
    return true;
  }
 
  //判断边是否过期
  bool edgeValid(GraphSpaceID spaceId,
                 const folly::StringPiece& key,
                 const folly::StringPiece& val) const {
    ....
    if (ttlExpired(schema.get(), reader.get())) {
      VLOG(3) << "Ttl expired";
      return false;
    }
    return true;
  }

 
  // TODO(panda) Optimize the method in the future
  bool ttlExpired(const meta::NebulaSchemaProvider* schema,
                  nebula::RowReaderWrapper* reader) const {
    if (schema == nullptr) {
      return true;
    }
    auto ttl = CommonUtils::ttlProps(schema);
    // Only support the specified ttl_col mode
    // Not specifying or non-positive ttl_duration behaves like ttl_duration =
    // infinity
    if (!ttl.first) {
      return false;
    }
    //这里比对数据的ttl字段的值和当前时间，如果过期就返回true
    return CommonUtils::checkDataExpiredForTTL(schema, reader, ttl.second.second, ttl.second.first);
  }

  bool ttlExpired(const meta::NebulaSchemaProvider* schema, const Value& v) const {
    if (schema == nullptr) {
      return true;
    }
    auto ttl = CommonUtils::ttlProps(schema);
    if (!ttl.first) {
      return false;
    }
    return CommonUtils::checkDataExpiredForTTL(schema, v, ttl.second.second, ttl.second.first);
  }
};

//判断ttl是否过期
bool CommonUtils::checkDataExpiredForTTL(const meta::NebulaSchemaProvider* schema,
                                         const Value& v,
                                         const std::string& ttlCol,
                                         int64_t ttlDuration) {
  const auto& ftype = schema->getFieldType(ttlCol);
  if (ftype != nebula::cpp2::PropertyType::TIMESTAMP &&
      ftype != nebula::cpp2::PropertyType::INT64) {
    return false;
  }

  int64_t now;
  // The unit of ttl expiration unit is controlled by user, we just use a gflag here.
  if (!FLAGS_ttl_use_ms) {
    now = std::time(nullptr);
  } else {
    auto t = std::chrono::system_clock::now();
    now = std::chrono::duration_cast<std::chrono::milliseconds>(t.time_since_epoch()).count();
  }

  // if the value is not INT type (sush as NULL), it will never expire.
  // TODO (sky) : DateTime
  if (v.isInt() && (now > (v.getInt() + ttlDuration))) {
    VLOG(2) << "ttl expired";
    return true;
  }
  return false;
}

//从ddl的schema里面读取ttl_duration和ttl_col字段
std::pair<bool, std::pair<int64_t, std::string>> CommonUtils::ttlProps(
    const meta::NebulaSchemaProvider* schema) {
  DCHECK(schema != nullptr);
  const auto* ns = dynamic_cast<const meta::NebulaSchemaProvider*>(schema);
  const auto sp = ns->getProp();
  int64_t duration = 0;
  if (sp.get_ttl_duration()) {
    duration = *sp.get_ttl_duration();
  }
  std::string col;
  if (sp.get_ttl_col()) {
    col = *sp.get_ttl_col();
  }
  return std::make_pair(!(duration <= 0 || col.empty()), std::make_pair(duration, col));
}

//获取ttl_col字段的值
StatusOr<Value> CommonUtils::ttlValue(const meta::NebulaSchemaProvider* schema,
                                      RowReaderWrapper* reader) {
  DCHECK(schema != nullptr);
  const auto* ns = dynamic_cast<const meta::NebulaSchemaProvider*>(schema);
  auto ttlProp = ttlProps(ns);
  if (!ttlProp.first) {
    return Status::Error();
  }
  return reader->getValueByName(std::move(ttlProp).second.second);
}

GraphService::future_authenticate(const std::string& username,const std::string& password){
  // 请求nebula-metad校验用户名密码
  auto authResult = auth(username, password);
  if (!authResult.ok()) {
    //用户密码错误
    ctx->resp().errorCode = ErrorCode::E_BAD_USERNAME_PASSWORD;
    return future;
  }
  if (sessionManager_->isOutOfConnections()) {
    //session创建过多
    ctx->resp().errorCode = ErrorCode::E_TOO_MANY_CONNECTIONS;
    return future;
  }
  //请求nebula-metad创建会话
  sessionManager_->createSession(username, clientIp, getThreadManager())
}

GraphSessionManager::createSession(){
  std::string key = userName + clientIp;
  //nebula-graphd.conf中配置的max_sessions_per_ip_per_user参数值
  auto maxSessions = FLAGS_max_sessions_per_ip_per_user;
  //获取相同用户IP创建session数量
  auto uiscFindPtr = sessionCnt(key);
  if (uiscFindPtr->get() > maxSessions - 1) {
    return Status::Error("Create Session failed: Too many sessions created");
  }
  //通过metaClient创建会话
  return metaClient_->createSession(userName, myAddr_, clientIp)
}

MetaClient::createSession(){
  //像nebula-metad发起RPC请求创建session
  client->future_createSession(request)
}

//接收到session创建请求并进行处理
MetaServiceHandler::future_createSession(
    const cpp2::CreateSessionReq& req) {
  auto* processor = CreateSessionProcessor::instance(kvstore_);
  RETURN_FUTURE(processor);
}

CreateSessionProcessor::process(){
  cpp2::Session session;
  // 创建sessionId，以当前时间微妙作为值
  session.session_id_ref() = time::WallClock::fastNowInMicroSec();
  session.create_time_ref() = session.get_session_id();
  session.update_time_ref() = session.get_create_time();
  session.user_name_ref() = user;
  session.graph_addr_ref() = req.get_graph_addr();
  session.client_ip_ref() = req.get_client_ip();
  std::vector<kvstore::KV> data;
  //构建kv引擎键值对，key为sessionId，value为会话信息
  data.emplace_back(MetaKeyUtils::sessionKey(session.get_session_id()),
                    MetaKeyUtils::sessionVal(session));
  resp_.session_ref() = session;
  //调用kv引擎的put方法
  ret = doSyncPut(std::move(data));
}

static const PartitionID kDefaultPartId = 0;
static const GraphSpaceID kDefaultSpaceId = 0;

//调用NebulaStore的put方法
BaseProcessor<RESP>::doSyncPut(){
  kvstore_->asyncMultiPut(kDefaultSpaceId,kDefaultPartId)
}

NebulaStore::asyncMultiPut(){
  //获取session会话分片信息，spaceId和 partId都为0
  auto ret = part(spaceId, partId);
  auto part = nebula::value(ret);
  //分片数据写入
  part->asyncMultiPut(std::move(keyValues), std::move(cb));
}

Part::asyncMultiPut(){
  //简直编码
  std::string log = encodeMultiValues(OP_MULTI_PUT, keyValues);
  //通过raft协议进行数据写入
  appendLogAsync(source, LogType::NORMAL, std::move(log))
}

GraphSessionManager::threadFunc() {
  //回收过期会话
  reclaimExpiredSessions();
  //启动下次回收任务
  scavenger_->addDelayTask(
      FLAGS_session_reclaim_interval_secs * 1000, &GraphSessionManager::threadFunc, this);
}

GraphSessionManager::reclaimExpiredSessions() {
  //过期session列表
  std::vector<SessionID> expiredSessions;
  // activeSessions_为当前有效session列表，在创建session的时候进行填充
  for (const auto& iter : activeSessions_) {
    int32_t idleSecs = iter.second->idleSeconds();
    if (idleSecs < FLAGS_session_idle_timeout_secs) {
      continue;
    }
    //空闲时间大于session_idle_timeout_secs就会加入到过期列表中
    expiredSessions.emplace_back(iter.first);
  }
}
//请求nebula-metad进行会话清除，根据之前创建session可知其实就是通过raft协议从rocksdb中删除sessionId为key的值
metaClient_->removeSessions(std::move(expiredSessions))