MongoDB write写(增、删、改)模块源码实现
source link: https://mongoing.com/archives/78795
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
前面的《transport_layer网络传输层模块源码实现》和《command命令处理模块源码实现》详细的分析了MongoDB内核网络数据收发过程以及命令解析处理的整个过程，本文将继续分析该系列的第三个子模块-《write写(增、删、改)模块源码实现》。
1. write写模块与command命令处理模块衔接回顾
上面两图是command命令处理模块的大体流程，最终经过command模块处理后，会执行对应的命令run接口，本文要分析的write模块也将从本入口入手。增、删、改三个最基本的写操作对应的命令入口如下表：
操作类型 命令Run()入口 增 CmdInsert::runImpl() 删 CmdDelete::runImpl() 改 CmdUpdate::runImpl()MongoDB内核write模块主要由如下目录代码实现：
下面章节将分析增删改操作的详细内核实现流程，注意包括请求序列化解析存储、insert写入流程、update更新计划执行器、delete删除计划执行器等。
2. 增、删、改序列化解析及结构化统一存储
本章节详细分析增、删、改三个操作的序列化解析及结构化统一存储核心实现过程。
2.1 增删改写入操作语法及其主要含义说明
insert插入语法及说明
insert主要完成数据的写入操作，其命令语法如下：
{
insert: <collection>,
documents: [ <document>, <document>, <document>, ... ],
ordered: <boolean>,
writeConcern: { <write concern> },
bypassDocumentValidation: <boolean>
}
insert操作主要由五个字段类型组成，具体字段功能说明如下：
字段名 功能说明 insert 集合名 documents 具体的文档内容 ordered 一次性插入多条文档数据，前面的数据写入失败，是否继续后面的数据写入 writeConcern writeConcern写策略配置，写多少个节点才算成功 bypassDocumentValidation 是否进行validator相关schema文档验证update更新语法及说明
update操作实现数据更新操作，其命令语法如下：
{
update: <collection>,
updates: [
{ q: <query>, u: <update>, upsert: <boolean>, multi: <boolean>,
collation: <document>, arrayFilters: <array> },
{ q: <query>, u: <update>, upsert: <boolean>, multi: <boolean>,
collation: <document>, arrayFilters: <array> },
{ q: <query>, u: <update>, upsert: <boolean>, multi: <boolean>,
collation: <document>, arrayFilters: <array> },
...
],
ordered: <boolean>,
writeConcern: { <write concern> },
bypassDocumentValidation: <boolean>
}
上述语法各字段功能说明如表：
字段名 功能说明 update 对那个表做update操作 updates.q 查询条件 updates.u 更新操作方法 updates.upsert 如果需要更新的数据不存在，是否直接插入新数据 updates.multi query满足条件数据有多条，是只更新一条还是多条一起更新 updates.collation 根据不同语言定义不同排序规则 updates.arrayFilters 数组中成员内容跟新 ordered 一次更新多条文档数据，前面的数据更新失败，是否继续后面的数据更新操作 writeConcern 更新多少个节点成功才返回OK bypassDocumentValidation 是否进行validator相关schema文档验证update更新语法及说明
delete删除操作对应语法如下：
{
delete: <collection>,
deletes: [
{ q : <query>, limit : <integer>, collation: <document> },
{ q : <query>, limit : <integer>, collation: <document> },
{ q : <query>, limit : <integer>, collation: <document> },
...
],
ordered: <boolean>,
writeConcern: { <write concern> }
}
如上，delete语法各个字段功能说明如下：
字段名 功能说明 delete 对那个表做delete操作 deletes.q 需要删除那一部分数据，也就是删除数据的条件 deletes.limit 删除所有满足条件的数据还是只删除一条，取值0或1 deletes.collation 根据不同语言定义不同排序规则 ordered 删除一批数据，如果前面某数据删除失败，是否还需要删除后面满足条件的数据 writeConcern 删除多少个节点成功才返回OK2.2 增、删、改序列化解析
2.2.1 增、删、改核心数据结构
从上面的insert、delete、update语法可以看出，这三个操作有一部分字段名是一样的，内核在代码实现的时候也重复利用了这一特定，把这部分成员抽象为公共类，不同的字段则在各自操作类中封装。
最终，三个操作的字段信息通过公用类WriteCommandBase和各自私有类Insert、Update、Delete保持及解析封装。如下图所示：
公共基类由WriteCommandBase类实现，如下：
class WriteCommandBase {
public:
//基类接口
......
//MongoDB字段验证规则（schema validation）
bool _bypassDocumentValidation{false};
//一次对多条数据进行插入或者删除或者更新的时候，前面的数据操作失败，是否继续后面的操作
bool _ordered{true};
//事务相关，等4.2版本回头分析
boost::optional<std::vector<std::int32_t>> _stmtIds;
}
Insert类包含WriteCommandBase类成员，同时包括Insert操作对应的私有成员信息，如下：
class Insert {
public:
......
//也就是db.collection
NamespaceString _nss;
//公共结构信息
WriteCommandBase _writeCommandBase;
//真正的文档在这里documents
std::vector<mongo::BSONObj> _documents;
//库信息
std::string _dbName;
//是否有documents
}
delete删除操作对应Delete类核心成员信息如下：
class Delete {
public:
......
//DB.COLLECTION信息
NamespaceString _nss;
WriteCommandBase _writeCommandBase;
//具体的delete内容在这里
std::vector<DeleteOpEntry> _deletes;
}
update更新操作对应的Update类核心成员信息如下:
class Update {
public:
......
//db.collection信息，也就是库.表信息
NamespaceString _nss;
WriteCommandBase _writeCommandBase;
//需要更新的具体内容在该成员中
std::vector<UpdateOpEntry> _updates;
}
上面的类结构中，documents、deletes、_updates三个成员分别对应增、删、改操作的集体操作信息，都是数组类型，可以一次进行多条数据操作。
2.2.2 增、删、改解析过程
增删改三个操作对应三个不同的类，由这三个类来完成各自操作的协议解析及封装，整体代码实现大同小异，本文只分析insert解析及封装过程，主要代码实现如下:
Insert Insert::parse(const IDLParserErrorContext& ctxt, const BSONObj& bsonObject) {
......
//调用Insert::parseProtected
object.parseProtected(ctxt, bsonObject);
return object;
}
void Insert::parseProtected(...)
{
//解析出insert类的对应成员信息
for (const auto& element :request.body) {
const auto fieldName = element.fieldNameStringData();
//解析bypassDocumentValidation信息
if (fieldName == kBypassDocumentValidationFieldName) {
......
}
//解析ordered信息
else if (fieldName == kOrderedFieldName) {
......
}
//解析stmtIds信息
else if (fieldName == kStmtIdsFieldName) {
......
}
//解析需要插入的文档信息
else if (fieldName == kDocumentsFieldName) {
//解析的文档保持到_documents数组
_documents = std::move(values);
}
//解析db名
else if (fieldName == kDbNameFieldName) {
......
}
......
}
//从request中解析出_writeCommandBase基础成员内容
_writeCommandBase = WriteCommandBase::parse(ctxt, request.body);
......
//根据db+collection构造出db.collection字符串
_nss = ctxt.parseNSCollectionRequired(_dbName, commandElement);
}
和insert操作类似，update和delete操作的解析过程与insert流程一样比较简单，因此不在分析。
最终，所有解析出的数据保存到各自类中，总结如下图所示：
此外，增删改操作的序列化封装由write_ops_gen.cpp中的Insert::serialize()、Update::serialize()、Delete::serialize()完成，主要根据各自类完成Bson统一封装，整个实现过程比较简单，这里不在详细分析。
增删改接口解析及序列化相关几个核心接口功能说明如下：
类 函数接口 功能说明 write_ops::Insert InsertOp::parse(…) insert操作解析 Insert::toBSON(…) insert Bson序列化
write_ops::Update UpdateOp::parse(…) update操作解析 Update::toBSON(…) update Bson序列化
write_ops::Delete DeleteOp::parse(…) delete操作解析 DeleteOp::toBSON(…) delete Bson序列化
注意：在insert、update、delete中还有如下一个细节，为何不见writeConcern相关成员存储？原因是writeConcern解析放到了外层runCommandImpl中通过setWriteConcern()保持到该请求对应得opCtx操作上下文中。
3. Insert数据写操作核心实现
insert处理和command命令处理模块通过CmdInsert::runImpl()衔接，该接口代码实现如下：
//插入文档会走这里面 CmdInsert::runImpl
void runImpl(...) final {
//从request中解析出write_ops::Insert类成员信息
const auto batch = InsertOp::parse(request);
const auto reply = performInserts(opCtx, batch);
......
}
InsertOp::parse()在前面章节已经分析，主要完成数据的统一解析存储。insert请求解析存储到write_ops::Insert类后，开始调用performInserts(…)处理。在该接口中完成如下流程：分批数据组装、批量数据写入、事务封装、写入存储引擎等。
3.1 数据分批组装
由于inset一次可以插入多条数据，为了最大化满足性能要求，当写入数据很多的时候，MongoDB内核通过把这些数据按照指定规则拆分到多个batch中，这样每个batch代表一批数据，然后进行统一处理。分批数据组装拆分过程核心代码实现如下:
//数据分批写入核心代码实现
WriteResult performInserts(OperationContext* opCtx, const write_ops::Insert& wholeOp) {
.......
//写入数据成功后的会掉处理
//主要完成表级tps及时延统计
ON_BLOCK_EXIT([&] {
//performInserts执行完成后调用，记录执行结束时间
curOp.done();
//表级tps及时延统计
Top::get(opCtx->getServiceContext())
.record(...);
});
......
size_t bytesInBatch = 0;
//batch数组
std::vector<InsertStatement> batch;
//默认64,可以通过db.adminCommand( { setParameter: 1, internalInsertMaxBatchSize:xx } )配置
const size_t maxBatchSize = internalInsertMaxBatchSize.load();
//当写入的数据小于64时，也就是一个batch即可一起处理
//batch最大限制为写入数据大于64或者batch中总字节数超过256K
batch.reserve(std::min(wholeOp.getDocuments().size(), maxBatchSize));
for (auto&& doc : wholeOp.getDocuments()) {
......
//doc检查，例如是否嵌套过多，是否一个doc带有多个_id等
auto fixedDoc = fixDocumentForInsert(opCtx->getServiceContext(), doc);
//如果这个文档检测有异常，则跳过这个文档，进行下一个文档操作
if (!fixedDoc.isOK()) {
//啥也不做，直接忽略该doc
} else {
//事务相关，先忽略，以后会回头专门分析事务
const auto stmtId = getStmtIdForWriteOp(opCtx, wholeOp, stmtIdIndex++);
......
//把文档插入到batch数组
BSONObj toInsert = fixedDoc.getValue().isEmpty() ? doc : std::move(fixedDoc.getValue());
batch.emplace_back(stmtId, toInsert);
bytesInBatch += batch.back().doc.objsize();
//这里continue，就是为了把批量插入的文档组成到一个batch数组中，到达一定量一次性插入
//batch里面一次最多插入64个文档或者总字节数256K，则后续的数据拆分到下一个batch
if (!isLastDoc && batch.size() < maxBatchSize && bytesInBatch < insertVectorMaxBytes)
continue; // Add more to batch before inserting.
}
//把本batch中的数据交由该接口统一处理
bool canContinue = insertBatchAndHandleErrors(opCtx, wholeOp, batch, &lastOpFixer, &out);
//清空batch，开始下一轮处理
batch.clear();
bytesInBatch = 0;
......
}
上面的代码可以总结为以下图形：
说明，上面假设64条数据总大小不超过256KB的batch图，如果64条doc文档数据总大小超过256kb，这时候阀值则以总数据256K为限制。单个batch最大上限限制条件如下：
-
最多64个doc文档数据。
-
单个batch总数据长度不超过256Kb。
3.2 batch数据事务写入流程及其异常补偿机制
一批数据通过分批拆分存入多个batch后，调用insertBatchAndHandleErrors()接口来完成单个batch的数据写入。整个batch数据写入可以在一个transaction事务完成，也可以一条数据一个事务来完成写入，具体核心代码实现如下:
bool insertBatchAndHandleErrors(...) {
......
try {
//如果对应collection不存在则创建
acquireCollection(); //执行上面定义的函数
//如果collection不是固定capped集合，并且batch中数据大于一条
//则试着在一个事务中一次性写入所有的数据
if (!collection->getCollection()->isCapped() && batch.size() > 1) {
......
//为什么这里没有检查返回值？默认全部成功？ 实际上通过try catch获取到异常后，再后续改为一条一条插入
insertDocuments(opCtx, collection->getCollection(), batch.begin(), batch.end());
//insert统计计数及返回值赋值
globalOpCounters.gotInserts(batch.size());
......
std::fill_n(std::back_inserter(out->results), batch.size(), std::move(result));
curOp.debug().ninserted += batch.size();
//一个事务写入多个doc成功，直接返回
return true;
}
} catch (const DBException&) { //批量写入失败，则后面一条一条的写
collection.reset();
//注意这里没有return，在后续一条一个事务写入
}
//这里循环解析batch，实现一条数据一个在一个事务中处理
for (auto it = batch.begin(); it != batch.end(); ++it) {
globalOpCounters.gotInsert(); //insert操作计数
try {
//log() << "yang test ............getNamespace().ns():" << wholeOp.getNamespace().ns();
//writeConflictRetry里面会执行{}中的函数体
writeConflictRetry(opCtx, "insert", wholeOp.getNamespace().ns(), [&] {
try {
......
//把该条文档插入
insertDocuments(opCtx, collection->getCollection(), it, it + 1);
//统计计数处理
SingleWriteResult result;
result.setN(1);
out->results.emplace_back(std::move(result));
curOp.debug().ninserted++;
} catch (...) {
......
}
});
} catch (const DBException& ex) {//写入异常
//注意这里，如果失败是否还可以继续后续数据的写入
bool canContinue =
handleError(opCtx, ex, wholeOp.getNamespace(), wholeOp.getWriteCommandBase(), out);
if (!canContinue)
return false; //注意这里直接退出循环，也就是本批次数据后续数据没有写入了
}
}
return true;
}
一批batch数据(假设64条)写入过程，如果不是capped固定集合，则这64条数据首先放入一个transaction事务中完成写入。如果写入异常，则继续一个事务一条数据写入。数据放入事务执行流程如下:
void insertDocuments(OperationContext* opCtx, Collection* collection, std::vector<InsertStatement>::iterator begin, std::vector<InsertStatement>::iterator end) //事务开始 WriteUnitOfWork wuow(opCtx); ...... //把数组begin到end之间的所有doc文档数据放入该事务中 uassertStatusOK(collection->insertDocuments( opCtx, begin, end, &CurOp::get(opCtx)->debug(), /*enforceQuota*/ true)); //事务结束 wuow.commit(); //WriteUnitOfWork::commit }
到这里后，insert操作在write模块中的流程就结束了，后续的doc写入流程存储引擎将交由storage模块实现。
上面的核心代码分析可以总结为如下总结：
当这个batch中的数据放入同一个事务执行失败后，则改为一条一个事务循环处理，如下图所示：
3.3 中间数据写入异常如何处理
假设一个batch数据64条数据，如果第23条数据写入失败了，后续的第24-64条数据是否需要继续写入，这就是本章节需要分析的问题。MongoDB内核实现的时候通过handleError()接口判断是否需要继续写入，该接口代码如下：
//前面数据写入失败，是否可以继续后续数据写入
bool handleError(...) {
......
//判断是什么原因引起的异常，从而返回不同的值
//如果是isInterruption错误，直接返回true,意思是不需要后续数据写入
if (ErrorCodes::isInterruption(ex.code())) {
//如果是interrupt异常，则整批数据写失败，也就是不进行后续数据写入
throw; // These have always failed the whole batch.
}
......
//如果ordered为false则忽略这条写入失败的数据，继续后续数据写入
return !wholeOp.getOrdered();
}
从上面的代码可以看出，只要出现以下异常情况，就不可继续后续数据insert写入操作了，如下：
Interruption错误：包括Interrupted、InterruptedAtShutdown、ExceededTimeLimit、InterruptedDueToReplStateChange四种异常，其他异常情况可以继续写入。
ordered参数配置为false: 如果该配置为false则遇到异常不继续处理后续doc写入。
写入异常后是否继续写总结如下图所示：
3.4 后续
通过前面的分析可以得出，MongoDB内核把多条doc文档按照指定限制把文档封装到不同batch中，然后一个batch一个batch分批处理。最终，这些batch对应数据将会通过MongoDB内核的storage存储模块来完成insert事务处理，最终在CollectionImpl::insertDocuments()实现。
Insert写入流程核心接口调用关系图如下：
说明：数据如何组装存入wiredtiger存储引擎将在后续《storage存储模块源码实现》中详细分析。
4. delete删除操作核心实现
delete数据删除通过命令处理模块中的CmdDelete::runImpl(…) ->performDeletes接口完成和write写模块delete操作对接，下面我们分析该接口核心代码实现，如下：
WriteResult performDeletes(...)
{
......
//singleOp类型为DeleteOpEntry write_ops::Delete::getDeletes
for (auto&& singleOp : wholeOp.getDeletes()) {
//事务相关，先跳过，以后相关章节专门分析
const auto stmtId = getStmtIdForWriteOp(opCtx, wholeOp, stmtIdIndex++);
......
//该函数接口执行完后执行该finishCurOp
//finishCurOp实现表级QPS及时延统计 本op操作的慢日志记录等
ON_BLOCK_EXIT([&] { finishCurOp(opCtx, &curOp); });
try {
lastOpFixer.startingOp();
out.results.emplace_back(
//该delete op操作真正执行在这里，singleOp类型为DeleteOpEntry
performSingleDeleteOp(opCtx, wholeOp.getNamespace(), stmtId, singleOp));
lastOpFixer.finishedOpSuccessfully();
} catch (const DBException& ex) {
......
}
return out;
}
从上面代码分析可以看出，如果wholeOp携带有多个DeleteOpEntry(也就是singleOp )操作，则循环对singleOp 进行处理，这个处理过程由performSingleDeleteOp(…)接口实现，具体如下:
performSingleDeleteOp(…)接口核心代码实现如下：
static SingleWriteResult performSingleDeleteOp(...) {
......
//根据ns构造DeleteReques
//根据请求相关信息初始化赋值DeleteRequest
DeleteRequest request(ns);
request.setQuery(op.getQ());
request.setCollation(write_ops::collationOf(op));
request.setMulti(op.getMulti());
request.setYieldPolicy(PlanExecutor::YIELD_AUTO); // ParsedDelete overrides this for $isolated.
request.setStmtId(stmtId);
//根据DeleteRequest构造ParsedDelete
ParsedDelete parsedDelete(opCtx, &request);
//从request解析出对应成员存入parsedDelete
uassertStatusOK(parsedDelete.parseRequest());
//检查该请求是否已经被kill掉了
opCtx->checkForInterrupt();
......
//写必须走主节点判断及版本判断
assertCanWrite_inlock(opCtx, ns);
//从查询引擎中获取delete执行器
auto exec = uassertStatusOK(
getExecutorDelete(opCtx, &curOp.debug(), collection.getCollection(), &parsedDelete));
{
stdx::lock_guard<Client> lk(*opCtx->getClient());
CurOp::get(opCtx)->setPlanSummary_inlock(Explain::getPlanSummary(exec.get()));
}
//运行该执行器
uassertStatusOK(exec->executePlan());
//下面流程是记录各种统计信息
long long n = DeleteStage::getNumDeleted(*exec);
curOp.debug().ndeleted = n;
PlanSummaryStats summary;
//获取执行器运行过程中的各种统计信息
Explain::getSummaryStats(*exec, &summary);
if (collection.getCollection()) {
collection.getCollection()->infoCache()->notifyOfQuery(opCtx, summary.indexesUsed);
}
curOp.debug().setPlanSummaryMetrics(summary);
//统计信息序列化
if (curOp.shouldDBProfile()) {
BSONObjBuilder execStatsBob;
Explain::getWinningPlanStats(exec.get(), &execStatsBob);
curOp.debug().execStats = execStatsBob.obj();
}
......
return result;
}
该接口最核心的部分为获取delete执行器并运行，执行器由query查询引擎模块实现，因此getExecutorDelete(…)获取delete执行器及其运行过程具体实现流程将在后续《query查询引擎模块实现原理》章节详细分析，这里暂时跳过这一逻辑。write模块中delete操作主要接口调用流程如下：
5. update更新操作核心实现
update数据更新操作过程和delete操作过程类似，这里不在累述，其核心接口调用流程如下图所示：
6. 下期预告
下期将分析《storage存储模块源码实现》，storage模块分析完成后将分析MongoDB最复杂的《query查询引擎源码实现》，敬请关注。
更多文章：
MongoDB网络传输处理源码实现及性能调优-体验内核性能极致设计
OPPO百万级高并发MongoDB集群性能数十倍提升优化实践
盘点 2020 | 我要为分布式数据库 MongoDB 在国内影响力提升及推广做点事
作者：杨亚洲
前滴滴出行技术专家，现任OPPO文档数据库MongoDB负责人，负责oppo千万级峰值TPS/十万亿级数据量文档数据库MongoDB内核研发及运维工作，一直专注于分布式缓存、高性能服务端、数据库、中间件等相关研发。Github账号地址:
5月29日来长沙参加MongoDB中文社区2021年第一场线下大会吧！互联网证券及金融系统实践案例、物联网实时数据融合平台案例分享，存储引擎、数据迁移同步等诸多MongoDB一手实践干货在现场等你来！
长沙大会速递Conference
时间：2021年5月29日（周六） 13:30 – 17:30
地点：长沙市岳麓区中南大学科技园（研发）总部1号栋3楼309多功能厅
名额： 150人
报名链接： https://sourl.cn/EJNURJ
* 9.9元票请添加小芒果（微信ID：mongoingcom）按照说明获取。
* 分享活动还有机会领取MongoDB周边礼品！
您也可以直接阅读相关活动介绍：2021年MongoDB中文社区长沙大会 | 相聚岳麓，共话mongo！
大会奖品Prizes
Recommend
About Joyk
Aggregate valuable and interesting links.
Joyk means Joy of geeK