# mongo-rocks data format

## mongo-rocks 存储数据的底层格式

### 1. mongo-rocks

[mongo-rocks](https://github.com/mongodb-partners/mongo-rocks)是基于mongo对存储引擎的规定的接口，用rocksdb实现的一个存储插件的；类似于mysql本身底层的存储引擎可以是插件化的，mongo本身也支持的；mongo-rocks就是基于rocksdb为mongodb提供存储的功能的；当然目前版本的mongo的默认的存储引擎是wt引擎，官方支持的；mongo-rocks这个库目前也在不断的更新用来支持新版本的mongo版本的；当然由于官方版本的版本迭代速度实在太快了，目前最新的mongo-rocks的能支持的最新的稳定版本是`4.2.5`，如果是第一次用mongo的话，推荐还是使用官方的wt引擎吧，别折腾自己；而自己分析这个东西的原因主要是因为我司的mongo版本算是走了弯路：1. 版本是3.4（mongo默认是放弃维护的） 2. 底层存储引擎是rocksdb；所以还是需要了解这个东西是如何组织在底层rocksdb；

### 2. Mongodb存储接口

在具体了解底层rocksdb的组织结构之前，先对一个存储引擎需要提供哪些接口进行分析，这部分的代码在mongodb的工程的`percona-server-mongodb/src/mongo/db/storage/kv/` 目录下面的，这部分定义了如果是用kv存储来支持mongo需要实现的接口有哪些；

1. `KVEngine (kv_engine.h)`: kv存储引擎定义接口的地方

```C++
// 1. 指定ns和ident来获得对应表的元信息;    
virtual std::unique_ptr<RecordStore> getRecordStore(OperationContext* opCtx,
                                                        StringData ns,
                                                        StringData ident,
                                                        const CollectionOptions& options) = 0;
// 2. 指定ns和索引信息返回一个可以进行索引的数据访问接口
virtual SortedDataInterface* getSortedDataInterface(OperationContext* opCtx,
                                                        StringData ident,
                                                        const IndexDescriptor* desc) = 0;
// 3. 创建一张表
virtual Status createRecordStore(OperationContext* opCtx,
                                     StringData ns,
                                     StringData ident,
                                     const CollectionOptions& options) = 0;
// 4. 为ns创建索引
virtual Status createSortedDataInterface(OperationContext* opCtx,
                                             StringData ident,
                                             const IndexDescriptor* desc) = 0;
// 5. 获得ns的大小
virtual int64_t getIdentSize(OperationContext* opCtx, StringData ident) = 0;

// 6. 删除一张表
virtual Status dropIdent(OperationContext* opCtx, StringData ident) = 0;

// 其他还有，但是并不是很重要
```

2. `RecordStore`: 对应的是某一个表的读写操作；

    上面`KVEngine`没有定义如何写入或者查询数据，这些定义的接口都在这个类中`RecordStore`; 主要关心的就是所谓的`CRUD` + 一些表的元信息的获得;

```C++
// 获得collection的数据大小,
virtual long long dataSize(OperationContext* txn) const = 0;
// 获得collection有多少条记录
virtual long long numRecords(OperationContext* txn) const = 0;
// 是否是限定容量的表，类似于oplog
virtual bool isCapped() const = 0;
// 设置如果容量达到上限之后如何处理的回调函数
virtual void setCappedCallback(CappedCallback*) {
        invariant(false);
}
// 数据在存储引擎中的真实大小，大部分时候都不大准;
virtual int64_t storageSize(OperationContext* txn,
                                BSONObjBuilder* extraInfo = NULL,
                                int infoLevel = 0) const = 0;

//Query
virtual bool findRecord(OperationContext* txn, const RecordId& loc, RecordData* out) const {}
//Delete
virtual void deleteRecord(OperationContext* txn, const RecordId& dl) = 0;
// Insert
virtual StatusWith<RecordId> insertRecord(OperationContext* txn,
                                              const char* data,
                                              int len,
                                              bool enforceQuota) = 0;
// Update
virtual Status updateRecord(OperationContext* txn,
                                const RecordId& oldLocation,
                                const char* data,
                                int len,
                                bool enforceQuota,
                                UpdateNotifier* notifier) = 0;


// 当然还是其他的接口，比如批量insert或者mongo的一些东西，但是这次重点还是以上的接口
```

3. `mongodb`是如何初始化整个存储引擎并且如何调用上面的这些接口

```C++
// 1. 核心类,这些成员并非所有，但是是最重要的
class KVStorageEngine final : public StorageEngine {
    // This must be the first member so it is destroyed last.
  // 真实的存储引擎
    std::unique_ptr<KVEngine> _engine;
  // 存储元信息的表；上面介绍过RecordStore是表的read/write操作
    std::unique_ptr<RecordStore> _catalogRecordStore;
  // mongo的db/coll的元信息的操作，比如创建表、删除、修改、list等等
    std::unique_ptr<KVCatalog> _catalog;
  // database的管理;
    typedef std::map<std::string, KVDatabaseCatalogEntry*> DBMap;
    DBMap _dbs;
}
//用来存放元信息的表名
const std::string catalogInfo = "_mdb_catalog";
//下面分析KVStorageEngine初始化过程
KVStorageEngine::KVStorageEngine(KVEngine* engine, const KVStorageEngineOptions& options)
    : _options(options), _engine(engine), _supportsDocLocking(_engine->supportsDocLocking()) {

    //底层存储中是否包含了存储mongodb表信息的表
    bool catalogExists = engine->hasIdent(&opCtx, catalogInfo);
    if (options.forRepair && catalogExists) {
        log() << "Repairing catalog metadata";
        // TODO should also validate all BSON in the catalog.
        engine->repairIdent(&opCtx, catalogInfo);
    }

    if (!catalogExists) {
        WriteUnitOfWork uow(&opCtx);

        Status status =
            _engine->createRecordStore(&opCtx, catalogInfo, catalogInfo, CollectionOptions());
        // BadValue is usually caused by invalid configuration string.
        // We still fassert() but without a stack trace.
        if (status.code() == ErrorCodes::BadValue) {
            fassertFailedNoTrace(28562);
        }
        fassert(28520, status);
        uow.commit();
    }
    // 2. 从底层引擎中获得对应的元信息表的对象
    _catalogRecordStore =
        _engine->getRecordStore(&opCtx, catalogInfo, catalogInfo, CollectionOptions());
    // _catalog本身是基于_catalogRecordStore来提供对表的一些访问、list、删除等等的操作
    _catalog.reset(new KVCatalog(
        _catalogRecordStore.get(), _options.directoryPerDB, _options.directoryForIndexes));
    _catalog->init(&opCtx);

    std::vector<std::string> collections;
    _catalog->getAllCollections(&collections);

    for (size_t i = 0; i < collections.size(); i++) {
        std::string coll = collections[i];
        NamespaceString nss(coll);
        string dbName = nss.db().toString();

        // No rollback since this is only for committed dbs.
        KVDatabaseCatalogEntry*& db = _dbs[dbName];
        if (!db) {
            db = new KVDatabaseCatalogEntry(dbName, this);
        }

        db->initCollection(&opCtx, coll, options.forRepair);
    }
	
     //清理无用的表
    {
        // get all idents
        std::set<std::string> allIdents;
        {
            std::vector<std::string> v = _engine->getAllIdents(&opCtx);
            allIdents.insert(v.begin(), v.end());
            allIdents.erase(catalogInfo);
        }

        // remove ones still in use
        {
            vector<string> idents = _catalog->getAllIdents(&opCtx);
            for (size_t i = 0; i < idents.size(); i++) {
                allIdents.erase(idents[i]);
            }
        }

        for (std::set<std::string>::const_iterator it = allIdents.begin(); it != allIdents.end();
             ++it) {
            const std::string& toRemove = *it;
            if (!_catalog->isUserDataIdent(toRemove))
                continue;
            log() << "dropping unused ident: " << toRemove;
            WriteUnitOfWork wuow(&opCtx);
            _engine->dropIdent(&opCtx, toRemove);
            wuow.commit();
        }
    }
}
```

4. `mongodb`创建表的一个具体的过程,中间会涉及到一些参数

```C++
// 用来生成collection的ident，最终的样子应该是: collection-2-一个随机数
std::string KVCatalog::_newUniqueIdent(StringData ns, const char* kind) {
    // If this changes to not put _rand at the end, _hasEntryCollidingWithRand will need fixing.
    StringBuilder buf;
    if (_directoryPerDb) {
        buf << NamespaceString::escapeDbName(nsToDatabaseSubstring(ns)) << '/';
    }
    buf << kind;
    buf << (_directoryForIndexes ? '/' : '-');
    buf << _next.fetchAndAdd(1) << '-' << _rand;
    return buf.str();
}
// 创建一张表
Status KVCatalog::newCollection(OperationContext* opCtx,
                                StringData ns,
                                const CollectionOptions& options) {
    invariant(opCtx->lockState() == NULL ||
              opCtx->lockState()->isDbLockedForMode(nsToDatabaseSubstring(ns), MODE_X));
   //1. 获得这个ns的一个唯一的标识符
    const string ident = _newUniqueIdent(ns, "collection");

    stdx::lock_guard<stdx::mutex> lk(_identsLock);
    Entry& old = _idents[ns.toString()];
    if (!old.ident.empty()) {
        return Status(ErrorCodes::NamespaceExists, "collection already exists");
    }
	  // 一个表信息存储到底层的信息;大概是这个样子，option中会包含更多的一些关于collection的元信息;
    /**
    {
    "ns":"xx",
    "ident":"uniq",
    "md":{
        "ns":"xxx",
        "options":{

        }
    	}
		}
    */
    BSONObj obj;
    {
        BSONObjBuilder b;
        b.append("ns", ns);
        b.append("ident", ident);
        BSONCollectionCatalogEntry::MetaData md;
        md.ns = ns.toString();
        md.options = options;
        b.append("md", md.toBSON());
        obj = b.obj();
    }

    // 直接把表的信息存储到底层的元信息的表中;
    StatusWith<RecordId> res = _rs->insertRecord(opCtx, obj.objdata(), obj.objsize(), false);
    if (!res.isOK())
        return res.getStatus();

    old = Entry(ident, res.getValue());
    LOG(1) << "stored meta data for " << ns << " @ " << res.getValue();
    return Status::OK();
}
```

上面几个类的分析，基本上能知道mongo是如何使用这些接口，主要是也解释了一些参数的生成，等到分析底层的rocksdb的数据组织形式的时候就会比较好的理解；

### 3. Mongo-Rocks是如何组织数据

因为我们使用的版本实在是很老，对应到的mongo版本是3.4的，所以存储的格式和目前的版本差别很大的；尤其是新版本都是用的`Column family`的方式来存储不同的表, 等下次我也总结直接的区别；

具体的目录如下: `percona-server-mongodb/src/mongo/db/storage/kv/rocks/src`下面，当然也可以直接clone mongo-rocks的代码; 下面集中几个点来分析，分别为：

1. collection存储
2. data的存储
3. index的存储
4. capped的表是如何控制size

#### 1. Collection在rocksdb中如何存储

```C++
const std::string RocksEngine::kMetadataPrefix("\0\0\0\0metadata-", 12);    
//  创建一个表
    Status RocksEngine::createRecordStore(OperationContext* opCtx, StringData ns, StringData ident,
                                          const CollectionOptions& options) {
        BSONObjBuilder configBuilder;
        //构建一个rocksdb的key，结构为:KmetadataPrefix + ident,value就是{"prefix":<uint32>}, prefix应该是rockdb自己统计的目前最大的prefix; 并且会把这个prefix编码也存储到rocksdb中，可能是为了加速找到最大的prefix, 这里我不管oplog的特殊性，所以下面的逻辑可以忽略
        auto s = _createIdent(ident, &configBuilder);
        if (s.isOK() && NamespaceString::oplog(ns)) {
            _oplogIdent = ident.toString();
            // oplog needs two prefixes, so we also reserve the next one
            uint64_t oplogTrackerPrefix = 0;
            {
                stdx::lock_guard<stdx::mutex> lk(_identMapMutex);
                oplogTrackerPrefix = ++_maxPrefix;
            }
            // we also need to write out the new prefix to the database. this is just an
            // optimization, 
            std::string encodedPrefix(encodePrefix(oplogTrackerPrefix));
            s = rocksToMongoStatus(
                _db->Put(rocksdb::WriteOptions(), encodedPrefix, rocksdb::Slice()));
        }
        return s;
    }
//读取一个表的信息
std::unique_ptr<RecordStore> RocksEngine::getRecordStore(OperationContext* opCtx, StringData ns,
                                             StringData ident, const CollectionOptions& options) {
        if (NamespaceString::oplog(ns)) {
            _oplogIdent = ident.toString();
        }

        auto config = _getIdentConfig(ident);
      	// 从config配置里面获得对应的prefix，上面有介绍过，每一个collection在创建的时候会对应一个prefix，是一个递增的uint32
        std::string prefix = _extractPrefix(config);
				
  		// 构建RocksRecordStore，这个类是RecordStore的一个实现类，用rocksdb来存储，本质也是用来操作crud的
        std::unique_ptr<RocksRecordStore> recordStore =
            options.capped
                ? stdx::make_unique<RocksRecordStore>(
                      ns, ident, _db.get(), _counterManager.get(), _durabilityManager.get(),
                      _compactionScheduler.get(), prefix,
                      true, options.cappedSize ? options.cappedSize : 4096,  // default size
                      options.cappedMaxDocs ? options.cappedMaxDocs : -1)
                : stdx::make_unique<RocksRecordStore>(ns, ident, _db.get(), _counterManager.get(),
                                                      _durabilityManager.get(), _compactionScheduler.get(),
                                                      prefix);

        {
            stdx::lock_guard<stdx::mutex> lk(_identObjectMapMutex);
            _identCollectionMap[ident] = recordStore.get();
        }
        return std::move(recordStore);
    }
```

存储Collection的元信息的总结：

1. 每一个`collection`对应会存在一个prefix,后面会解释有什么用
2. 元信息的key的格式为:`kMetadataPrefix` + `ident`, value：存储上面的`prefix`
3. 将`prefix` encode之后也存储在rocksdb，目前还不是很清楚是做什么

#### 2. collection中的数据如何存储

主要是看`RocksRecordStore`，因为它操作了collection的crud的操作；这里面我们将看到真实mongodb的存储结构，尤其是recordId的这个生成，为什么比较重要，原因在于不管是mongo的_id还是其他的索引，理论上来说最终都会转化成recordId，逻辑上接近在物理上可能相差很远，这是之前我遇到的一个问题;

```C++
        rocksdb::DB* _db;                      // rocksdb
        std::string _prefix; // 每一个collectuin都存在一个prefix,uniq
        const bool _isCapped; //是否是容量限制的collection
        const int64_t _cappedMaxSize; //max size
        const int64_t _cappedMaxSizeSlack;  // when to start applying backpressure
        const int64_t _cappedMaxDocs; // max doc num
        CappedCallback* _cappedCallback; // callback
        int _cappedDeleteCheckCount;      // 有多少的数据要被删
        const bool _isOplog; //是否是oplog

        // invariant: there is no live records earlier than _cappedOldestKeyHint. There might be
        // some records that are dead after _cappedOldestKeyHint.
        // SeekToFirst() on an capped collection is an expensive operation because bunch of keys at
        // the start are deleted. To reduce the overhead, we remember the next key to delete and
        // seek directly to it. This will not work correctly if somebody inserted a key before this
        // _cappedOldestKeyHint. However, we prevent this from happening by using
        // _cappedVisibilityManager and checking isCappedHidden() during deletions
        RecordId _cappedOldestKeyHint; //capped中最早的一条数据的recordId
        std::string _ident; //collection's ident
        AtomicUInt64 _nextIdNum; //recordId的生成依赖
        std::atomic<long long> _dataSize; //data size
        std::atomic<long long> _numRecords; //num record

        const std::string _dataSizeKey; // 存储datasize的rocksdb的key
        const std::string _numRecordsKey; // 存储record num的record的key

// 如何生成recordId,是一个原子变量的递增
    RecordId RocksRecordStore::_nextId() {
        invariant(!_isOplog);
        return RecordId(_nextIdNum.fetchAndAdd(1));
    }
// 查看如何初始化的_nextIdNum,主要思考如何持久化的; 下面这段代码来源于RocksRecordStore在初始化的时候使用，本质上是通过查找表中最后一条记录获得对应的Id，然后+1； TODO：是否会存在已经被使用过的recordId再次被重用呢？理论上应该是会的，但是recordId本质上不对外，所以即使重用了，也是说明之前的数据是被删除的，理论上问题不大的；
        bool emptyCollection = !iter->Valid();
        if (!emptyCollection) {
            // if it's not empty, find next RecordId
            iter->SeekToLast();
            dassert(iter->Valid());
            rocksdb::Slice lastSlice = iter->key();
            RecordId lastId = _makeRecordId(lastSlice);
            if (_isOplog || _isCapped) {
                _cappedVisibilityManager->updateHighestSeen(lastId);
            }
            _nextIdNum.store(lastId.repr() + 1);
        } else {
            // Need to start at 1 so we are always higher than RecordId::min()
            _nextIdNum.store(1);
        }

// 如何写入一条数据,这边主要关心key的构建, 这是在insertRecord中最终往rocksdb中使用的;
ru->writeBatch()->Put(_makePrefixedKey(_prefix, loc), rocksdb::Slice(data, len));
//一个key的构成: encord(prefix) + encord(recordId), 其实理论上在rocksdb中，所有表的数据是聚集在一起的;
std::string RocksRecordStore::_makePrefixedKey(const std::string& prefix, const RecordId& loc) {
        int64_t storage;
        auto encodedLoc = _makeKey(loc, &storage);
        std::string key(prefix);
        key.append(encodedLoc.data(), encodedLoc.size());
        return key;
}
//recordId统一编码
rocksdb::Slice RocksRecordStore::_makeKey(const RecordId& loc, int64_t* storage) {
        *storage = endian::nativeToBig(loc.repr());
        return rocksdb::Slice(reinterpret_cast<const char*>(storage), sizeof(*storage));
}

//查找一个记录,比较简单的，就是构建key，直接在rocksdb中去找即可
RecordData RocksRecordStore::_getDataFor(rocksdb::DB* db, const std::string& prefix,
                                             OperationContext* txn, const RecordId& loc) {
        RocksRecoveryUnit* ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);

        std::string valueStorage;
        auto status = ru->Get(_makePrefixedKey(prefix, loc), &valueStorage);
        if (status.IsNotFound()) {
            return RecordData(nullptr, 0);
        }
        invariantRocksOK(status);

        SharedBuffer data = SharedBuffer::allocate(valueStorage.size());
        memcpy(data.get(), valueStorage.data(), valueStorage.size());
        return RecordData(data, valueStorage.size());
}

//update + delete和正常的rocksdb操作都类似；
```

总结存储数据的:

1. `data key`的组成: `encode(prefix) + encode(recordId)`, prefix是每一个collection的uniq id, Len:32byte
2. `prefix`: 每次mongo重启的时候会去当前数据库找到最大的前缀，后面的新生成的prefix就递增;
3. `data size`的前缀: `std::string("\0\0\0\0", 4) + "datasize-" + id.toString()`,id = collection的ident
4. `record num`的前缀: `std::string("\0\0\0\0", 4) + "numrecords-" + id.toString()`, id = collection的ident
5. `prefix`: 每一个mongodb的meta信息的一个前缀，为了区别各个元数据
6. `ident`:每一个mongodb的meta信息的名字，比如collection、索引之类的，名字上还是有点区别