Rocksdb iterator和snapshot 接口

Rocksdb提供疊代器來來通路整個db中的資料，就像STL中的疊代器功能一樣，用來通路容器中的具體的資料。

通路形式以及通路接口有如下幾種：

周遊所有的key-value

//打開db，并初始化一個疊代器指針
rocksdb::Iterator* it = db->NewIterator(rocksdb::ReadOptions());
for (it->SeekToFirst(); it->Valid(); it->Next()) {
    cout << it->key().ToString() << ": " << it->value().ToString() << endl;
}
assert(it->status().ok()); // Check for any errors found during the scan
delete it;

輸出一個範圍内的key-value，[small, big)

for (it->Seek(small);
   it->Valid() && it->key().ToString() < big;
   it->Next()) {
...
}
assert(it->status().ok()); // Check for any errors found during the scan

反向周遊db中的元素

for (it->SeekToLast(); it->Valid(); it->Prev()) {
...
}
assert(it->status().ok()); // Check for any errors found during the scan

反向周遊一個指定範圍的key,如(small, big]

for (it->SeekForPrev(start);
   it->Valid() && it->key().ToString() > limit;
   it->Prev()) {
...
}
assert(it->status().ok()); // Check for any errors found during the scan

疊代器的接口可以算是 rocksdb針對用戶端的核心接口，主要是提供排序以及高效查找的功能。

測試代碼如下:

#include <iostream>
#include <string>
#include <rocksdb/db.h>
#include <rocksdb/iterator.h>
#include <rocksdb/table.h>
#include <rocksdb/options.h>
#include <rocksdb/env.h>

using namespace std;


static string rand_key(unsigned long long key_range) {
    char buff[30];
    unsigned long long n = 1;

    for (int i =1; i <= 4; ++i) {
        n *= (unsigned long long ) rand();
    }

    sprintf(buff, "%llu", n % key_range);

    string k(buff);
    return k;
}

int main() {
    rocksdb::DB *db;
    rocksdb::Options option;

    option.create_if_missing = true;
    option.compression = rocksdb::CompressionType::kNoCompression;

    rocksdb::Status s = rocksdb::DB::Open(option, "./iterator_db", &db);
    if (!s.ok()) {
        cout << "Open failed with " << s.ToString() << endl;
        exit(1);
    }

    rocksdb::DestroyDB("./iterator_db", option);

	cout << "seek all keys : " << endl;
    for(int i = 0; i < 5; i ++) {
        rocksdb::Status s = db->Put(rocksdb::WriteOptions(), 
                                rand_key(9), string(10, 'a' + (i % 26)) );

        if (!s.ok()) {
            cout << "Put failed with " << s.ToString() << endl;
            exit(1);
        }
    }   
    
    /* traverse rocksdb key-value */
    rocksdb::Iterator *it = db->NewIterator(rocksdb::ReadOptions());
    for (it->SeekToFirst(); it->Valid(); it->Next()) {
        cout << it->key().ToString() << ": " << it->value().ToString() << endl;
    }

    string limit="4";
    string start="2";
    cout << "seek from '2' to '4' : " << endl;
    for(it->Seek(start); it->Valid()&&it->key().ToString() < limit;
        it->Next()) {
            cout << it->key().ToString() << ": " << it->value().ToString() << endl;
        } 
    assert(it->status().ok());

    cout << "seek from last to start :" << endl;
    for (it->SeekToLast(); it->Valid(); it->Prev()) {
        cout << it->key().ToString() << ": " << it->value().ToString() << endl;
    }
    assert(it->status().ok());

    cout << "seek from '4' to '2' :" << endl;
    for(it->SeekForPrev(limit); it->Valid()&&it->key().ToString() > start;
        it->Prev()) {
            cout << it->key().ToString() << ": " << it->value().ToString() << endl;
        } 
    assert(it->status().ok());
        delete it;

    db->Close();
    delete db;

    return 0;
}

輸出如下:

seek all keys : 
3: cccccccccc
4: dddddddddd
7: bbbbbbbbbb
8: eeeeeeeeee
seek from '2' to '4' : 
3: cccccccccc
seek from last to start :
8: eeeeeeeeee
7: bbbbbbbbbb
4: dddddddddd
3: cccccccccc
seek from '4' to '2' :
4: dddddddddd
3: cccccccccc

且上層使用rocksdb疊代器接口時一般會和

snapshot

接口一同使用，用來實作MVCC的版本控制功能。

關于snapshot的實作，我們在Rocksdb事務：隔離性的實作中有提到，感興趣的可以看看。

關于snapshot的用戶端接口主要有:

sp1 = db->GetSnapshot(); 在目前db狀态下建立一個snapshot，添加到内部維護的一個全局的snapshotImpl的雙向連結清單中，并傳回該snapshot的對象
read_option.snapshot = sp1; 将擷取到的snapshot 傳給read_option,進行Get操作
db->ReleaseSnapshot(sp1); 釋放snapshot相關的資源（從雙向連結清單中删除該節點）

隔離性的測試代碼如下：

#include <iostream>
#include <string>
#include <rocksdb/db.h>
#include <rocksdb/iterator.h>
#include <rocksdb/table.h>
#include <rocksdb/options.h>
#include <rocksdb/env.h>

using namespace std;

int main() {
    rocksdb::DB *db;
    rocksdb::Options option;

    option.create_if_missing = true;
    option.compression = rocksdb::CompressionType::kNoCompression;

    rocksdb::Status s = rocksdb::DB::Open(option, "./iterator_db", &db);
    if (!s.ok()) {
        cout << "Open failed with " << s.ToString() << endl;
        exit(1);
    }
    // set a snapshot before put
    const rocksdb::Snapshot *sp1 = db->GetSnapshot(); 

    s = db->Put(rocksdb::WriteOptions(), "sp2", "value_sp2");
    assert(s.ok());

	// set a snapshot after put
    const rocksdb::Snapshot *sp2 = db->GetSnapshot();

    rocksdb::ReadOptions read_option;
    read_option.snapshot = sp1;
    string value = "";
    //預期擷取不到sp2的value，因為這裡用的是sp1的快照
    s = db->Get(read_option, "sp2", &value); 
    if(value == "") {
        cout << "Can't get sp2 at sp1!" << endl;
    }

    read_option.snapshot = sp2;
    // 能夠擷取到，使用的是sp2的快照，其是在put之後設定的
    s = db->Get(read_option, "sp2", &value); 
    assert(s.ok());
    if(value != "") {
        cout << "Got sp2's value: " << value << endl;
    }

    db->ReleaseSnapshot(sp1);
    db->ReleaseSnapshot(sp2);

輸出如下：

Can't get sp2 at sp1!
Got sp2's value: value_sp2

當然rocksdb也提供了更為複雜的mvcc特性，來以事務的方式支援不同的隔離級别。

//打開db，并初始化一個疊代器指針
rocksdb::Iterator* it = db->NewIterator(rocksdb::ReadOptions());
for (it->SeekToFirst(); it->Valid(); it->Next()) {
    cout << it->key().ToString() << ": " << it->value().ToString() << endl;
}
assert(it->status().ok()); // Check for any errors found during the scan
delete it;

for (it->Seek(small);
   it->Valid() && it->key().ToString() < big;
   it->Next()) {
...
}
assert(it->status().ok()); // Check for any errors found during the scan

for (it->SeekToLast(); it->Valid(); it->Prev()) {
...
}
assert(it->status().ok()); // Check for any errors found during the scan

for (it->SeekForPrev(start);
   it->Valid() && it->key().ToString() > limit;
   it->Prev()) {
...
}
assert(it->status().ok()); // Check for any errors found during the scan

#include <iostream>
#include <string>
#include <rocksdb/db.h>
#include <rocksdb/iterator.h>
#include <rocksdb/table.h>
#include <rocksdb/options.h>
#include <rocksdb/env.h>

using namespace std;


static string rand_key(unsigned long long key_range) {
    char buff[30];
    unsigned long long n = 1;

    for (int i =1; i <= 4; ++i) {
        n *= (unsigned long long ) rand();
    }

    sprintf(buff, "%llu", n % key_range);

    string k(buff);
    return k;
}

int main() {
    rocksdb::DB *db;
    rocksdb::Options option;

    option.create_if_missing = true;
    option.compression = rocksdb::CompressionType::kNoCompression;

    rocksdb::Status s = rocksdb::DB::Open(option, "./iterator_db", &db);
    if (!s.ok()) {
        cout << "Open failed with " << s.ToString() << endl;
        exit(1);
    }

    rocksdb::DestroyDB("./iterator_db", option);

	cout << "seek all keys : " << endl;
    for(int i = 0; i < 5; i ++) {
        rocksdb::Status s = db->Put(rocksdb::WriteOptions(), 
                                rand_key(9), string(10, 'a' + (i % 26)) );

        if (!s.ok()) {
            cout << "Put failed with " << s.ToString() << endl;
            exit(1);
        }
    }   
    
    /* traverse rocksdb key-value */
    rocksdb::Iterator *it = db->NewIterator(rocksdb::ReadOptions());
    for (it->SeekToFirst(); it->Valid(); it->Next()) {
        cout << it->key().ToString() << ": " << it->value().ToString() << endl;
    }

    string limit="4";
    string start="2";
    cout << "seek from '2' to '4' : " << endl;
    for(it->Seek(start); it->Valid()&&it->key().ToString() < limit;
        it->Next()) {
            cout << it->key().ToString() << ": " << it->value().ToString() << endl;
        } 
    assert(it->status().ok());

    cout << "seek from last to start :" << endl;
    for (it->SeekToLast(); it->Valid(); it->Prev()) {
        cout << it->key().ToString() << ": " << it->value().ToString() << endl;
    }
    assert(it->status().ok());

    cout << "seek from '4' to '2' :" << endl;
    for(it->SeekForPrev(limit); it->Valid()&&it->key().ToString() > start;
        it->Prev()) {
            cout << it->key().ToString() << ": " << it->value().ToString() << endl;
        } 
    assert(it->status().ok());
        delete it;

    db->Close();
    delete db;

    return 0;
}

seek all keys : 
3: cccccccccc
4: dddddddddd
7: bbbbbbbbbb
8: eeeeeeeeee
seek from '2' to '4' : 
3: cccccccccc
seek from last to start :
8: eeeeeeeeee
7: bbbbbbbbbb
4: dddddddddd
3: cccccccccc
seek from '4' to '2' :
4: dddddddddd
3: cccccccccc

snapshot

sp1 = db->GetSnapshot();
read_option.snapshot = sp1;
db->ReleaseSnapshot(sp1);

#include <iostream>
#include <string>
#include <rocksdb/db.h>
#include <rocksdb/iterator.h>
#include <rocksdb/table.h>
#include <rocksdb/options.h>
#include <rocksdb/env.h>

using namespace std;

int main() {
    rocksdb::DB *db;
    rocksdb::Options option;

    option.create_if_missing = true;
    option.compression = rocksdb::CompressionType::kNoCompression;

    rocksdb::Status s = rocksdb::DB::Open(option, "./iterator_db", &db);
    if (!s.ok()) {
        cout << "Open failed with " << s.ToString() << endl;
        exit(1);
    }
    // set a snapshot before put
    const rocksdb::Snapshot *sp1 = db->GetSnapshot(); 

    s = db->Put(rocksdb::WriteOptions(), "sp2", "value_sp2");
    assert(s.ok());

	// set a snapshot after put
    const rocksdb::Snapshot *sp2 = db->GetSnapshot();

    rocksdb::ReadOptions read_option;
    read_option.snapshot = sp1;
    string value = "";
    //預期擷取不到sp2的value，因為這裡用的是sp1的快照
    s = db->Get(read_option, "sp2", &value); 
    if(value == "") {
        cout << "Can't get sp2 at sp1!" << endl;
    }

    read_option.snapshot = sp2;
    // 能夠擷取到，使用的是sp2的快照，其是在put之後設定的
    s = db->Get(read_option, "sp2", &value); 
    assert(s.ok());
    if(value != "") {
        cout << "Got sp2's value: " << value << endl;
    }

    db->ReleaseSnapshot(sp1);
    db->ReleaseSnapshot(sp2);

Can't get sp2 at sp1!
Got sp2's value: value_sp2

Rocksdb iterator和snapshot 接口

繼續閱讀

LSM HOOK 學習及踩坑（recipe for target ‘__modpost‘ failed）LSM HOOK 學習及踩坑

「從零單排HBase 09」HBase的那些資料結構和算法1.LSM樹2.用跳表實作LSM的記憶體部分3.用多路歸并實作LSM樹的檔案合并4.布隆過濾器

RocksDB報錯：Compression type Snappy is not linked with the binary.

Flink的容錯機制第三篇狀态管理及容錯

levelDB的LSM檔案樹

讀寫鎖 share_mutex

RocksDB原理介紹RocksDB原理

[HBase基礎]--詳解SSTable結構和LSMTree索引

ZNS : 解決傳統SSD問題的高性能存儲棧設計（fs--＞io--＞device）

LSM核心源代碼分析與測試（一）

LSM樹原理及應用到HBase的索引

殘缺的LSM架構啟動分析前言LSM 簡介LSM 啟動分析附錄

資料結構（四）、LSM樹（日志結構合并樹）背景存儲引擎原理尾巴

為什麼大部分NOSQL資料庫選擇使用LSM樹而非B+樹？LSM Tree架構介紹核心流程LSM-Tree VS B+ Teee

RocksDB的Compaction : Leveled Compaction 和 Universal CompactionCompaction算法RocksDB的Compaction