首页 > 数据库 > 国内数据库 > OceanBase 存储层代码解读(二)微块存储格式
作者:
公祺,一个专注于 OBKV 的程序员
海芊,一个致力于当网红的 OceanBase 文档工程师。个人频道: Amber loves OB


// src/storage/blocksstable/ob_macro_block_writer.cpp
// 该函数主要逻辑为:
// 1. 将行数据写入到微块中,并更新 bloomfilter 和 checksum
// 2. 在当前微块写满的情况下:构建当前的微块,并切换微块写入器
int ObMacroBlockWriter::append_row(const ObStoreRow& row, const int64_t split_size, const bool ignore_lob)
{
int ret = OB_SUCCESS;
const ObStoreRow* row_to_append = &row;
// ... 省略参数检查的相关代码if (OB_SUCC(ret)) {// 向当前的微块中写入一行数据if (OB_FAIL(micro_writer_->append_row(*row_to_append))) { if (OB_BUF_NOT_ENOUGH == ret) {
if (0 == micro_writer_->get_row_count()) {
// 不支持超过 16KB 的行
ret = OB_NOT_SUPPORTED;
STORAGE_LOG(ERROR, "The single row is too large, ", K(ret), K(row));
} else if (OB_FAIL(build_micro_block(false))) { // 当前微块写满后,需要构建该微块:编码、压缩等// for compatibilitySTORAGE_LOG(WARN, "Fail to build micro block, ", K(ret));
} else if (OB_FAIL(micro_writer_->append_row(*row_to_append))) { // 继续写当前行数据STORAGE_LOG(ERROR, "Fail to append row to micro block, ", K(ret), K(row));
} else if (data_store_desc_->need_calc_column_checksum_
&& OB_FAIL(add_row_checksum(row_to_append->row_val_))) { // 计算 checksumSTORAGE_LOG(WARN, "fail to add column checksum", K(ret));
}
if (OB_SUCC(ret) && data_store_desc_->need_prebuild_bloomfilter_) {
// 根据 rowkey 构建 bloomfilter
const ObStoreRowkey rowkey(row_to_append->row_val_.cells_, data_store_desc_->bloomfilter_rowkey_prefix_);
if (OB_FAIL(micro_rowkey_hashs_.push_back(static_cast<uint32_t>(rowkey.murmurhash(0))))) {
STORAGE_LOG(WARN, "Fail to put rowkey hash to array ", K(ret), K(rowkey));
micro_rowkey_hashs_.reuse();
ret = OB_SUCCESS;
}
}
} else {
STORAGE_LOG(WARN, "Fail to append row to micro block, ", K(ret), K(row));
}
} else {
if (data_store_desc_->need_prebuild_bloomfilter_) {
// 根据 rowkey 构建 bloomfilter
const ObStoreRowkey rowkey(row_to_append->row_val_.cells_, data_store_desc_->bloomfilter_rowkey_prefix_);
if (OB_FAIL(micro_rowkey_hashs_.push_back(static_cast<uint32_t>(rowkey.murmurhash(0))))) {
STORAGE_LOG(WARN, "Fail to put rowkey hash to array ", K(ret), K(rowkey));
micro_rowkey_hashs_.reuse();
ret = OB_SUCCESS;
}
}
// 计算 checksumif (data_store_desc_->need_calc_column_checksum_ && OB_FAIL(add_row_checksum(row_to_append->row_val_))) {STORAGE_LOG(WARN, "fail to add column checksum", K(ret));
} else if (micro_writer_->get_block_size() >= split_size) {
// 如果当前微块已经满 16KB,则构建微块if (OB_FAIL(build_micro_block())) {STORAGE_LOG(WARN, "Fail to build micro block, ", K(ret));
}
}
}
}
return ret;
}
// src/storage/blocksstable/ob_row_writer.cpp
int ObRowWriter::write(const int64_t rowkey_column_count, const ObStoreRow& row, char* buf,
const int64_t buf_size, int64_t& pos, int64_t& rowkey_start_pos,
int64_t& rowkey_length, const bool only_row_key)
{
int ret = OB_SUCCESS;
if (row.is_sparse_row_) { // 使用稀疏行存储格式写入行数据if (OB_FAIL(write_sparse_row(
rowkey_column_count, row, buf, buf_size, pos, rowkey_start_pos, rowkey_length, only_row_key))) {
STORAGE_LOG(WARN, "write sparse row failed", K(ret), K(row), K(OB_P(buf)), K(buf_size));
}
} else if (OB_FAIL(write_flat_row(rowkey_column_count,
row,
buf,
buf_size,
pos,
rowkey_start_pos,
rowkey_length,
only_row_key))) { // 使用稠密行存储格式写入行数据
STORAGE_LOG(WARN, "write flat row failed", K(ret), K(row), K(OB_P(buf)), K(buf_size));
}
return ret;
}
// src/storage/blocksstable/ob_row_writer.cpp
// 使用稠密行存储格式写入行数据
int ObRowWriter::write_flat_row(const int64_t rowkey_column_count, const ObStoreRow& row, char* buf,
const int64_t buf_size, int64_t& pos, int64_t& rowkey_start_pos, int64_t& rowkey_length, const bool only_row_key)
{
int ret = OB_SUCCESS;
int64_t tmp_rowkey_start_pos = 0;
int64_t tmp_rowkey_length = 0;
if (!row.is_valid() || rowkey_column_count <= 0 || rowkey_column_count > row.row_val_.count_) {
ret = OB_INVALID_ARGUMENT;
STORAGE_LOG(ERROR, "invalid input argument.", K(buf), K(buf_size), K(row), K(rowkey_column_count), K(ret));
} else if (OB_FAIL(init_common(buf, buf_size, pos))) { // buf 参数初始化
STORAGE_LOG(WARN, "row writer fail to init common.", K(ret), K(row), K(OB_P(buf)), K(buf_size), K(pos));
} else if (OB_FAIL(init_store_row(row, rowkey_column_count))) { // 做一些参数检查
STORAGE_LOG(WARN, "row writer fail to init store row.", K(ret), K(rowkey_column_count));
} else if (OB_FAIL(append_row_header(row))) { // 填充 row header,不包括:列索引占用的字节数if (OB_BUF_NOT_ENOUGH != ret) {
STORAGE_LOG(WARN, "row writer fail to append row header.", K(ret));
}
} else if (OB_FAIL(
append_store_row(rowkey_column_count, row, only_row_key, tmp_rowkey_start_pos, tmp_rowkey_length))) { // 填充行数据,具体代码见下面的代码:ObRowWriter::append_store_rowif (OB_BUF_NOT_ENOUGH != ret) {
STORAGE_LOG(WARN, "row writer fail to append store row.", K(ret), K(rowkey_column_count));
}
} else if (OB_FAIL(append_column_index())) { // 填充列索引数据以及 row_header_.column_index_bytes_if (OB_BUF_NOT_ENOUGH != ret) {
STORAGE_LOG(WARN, "row writer fail to append column index.", K(ret));
}
} else {
pos = pos_;
rowkey_start_pos = tmp_rowkey_start_pos;
rowkey_length = tmp_rowkey_length;
}
return ret;
}
// 填充行数据,包括每一列的值
int ObRowWriter::append_store_row(const int64_t rowkey_column_count, const ObStoreRow& row, const bool only_row_key,
int64_t& rowkey_start_pos, int64_t& rowkey_length)
{
int ret = OB_SUCCESS;
const int64_t end_index = only_row_key ? rowkey_column_count : row.row_val_.count_;
rowkey_start_pos = pos_;
if (!row.is_valid() || rowkey_column_count <= 0 || rowkey_column_count > row.row_val_.count_) {
ret = OB_INVALID_ARGUMENT;
STORAGE_LOG(WARN, "invalid row input argument.", K(row), K(rowkey_column_count), K(ret));
} else {
for (int64_t i = 0; OB_SUCC(ret) && i < end_index; ++i) {
// 记录列索引,当前还没有确定列索引占用的字节数// 所以记录 3 种类型的列索引数组,在 append_column_index(...) 时,// 会最终确认列索引占用的字节数,以及使用哪个数组const int64_t column_index = pos_ - start_pos_;
column_indexs_8_[i] = static_cast<int8_t>(column_index);
column_indexs_16_[i] = static_cast<int16_t>(column_index);
column_indexs_32_[i] = static_cast<int32_t>(column_index);
// 填充列数据,包括:列类型、列值长度、列值if (OB_FAIL(append_column(row.row_val_.cells_[i]))) {if (OB_BUF_NOT_ENOUGH != ret) {
// 行数据太长了
STORAGE_LOG(WARN, "row writer fail to append column.", K(ret), K(i), K(row.row_val_.cells_[i]));
}
break;
}
}
}
if (OB_SUCC(ret)) {
column_index_count_ += end_index;
if (rowkey_column_count < end_index) {
rowkey_length = column_indexs_32_[rowkey_column_count] + start_pos_ - rowkey_start_pos;
} else {
rowkey_length = pos_ - rowkey_start_pos;
}
}
return ret;
}
// src/storage/blocksstable/ob_macro_block_writer.cpp
int ObMacroBlockWriter::build_micro_block(const bool force_split)
{
int ret = OB_SUCCESS;
char* block_buffer = NULL;
int64_t block_size = 0;
bool mark_deletion = false;
ObMicroBlockDesc micro_block_desc;
if (micro_writer_->get_row_count() <= 0) {
ret = OB_INNER_STAT_ERROR;
STORAGE_LOG(WARN, "micro_block_writer is empty", K(ret));
} else if (OB_FAIL(micro_writer_->build_block(block_buffer, block_size))) { // 序列化
STORAGE_LOG(WARN, "Fail to build block, ", K(ret));
} else if (OB_FAIL(
compressor_.compress(block_buffer, block_size, micro_block_desc.buf_, micro_block_desc.buf_size_))) { // 使用压缩算法,对微块序列化好的 buffer 进行压缩
STORAGE_LOG(WARN, "macro block writer fail to compress.", K(ret), K(OB_P(block_buffer)), K(block_size));
} else if (MICRO_BLOCK_MERGE_VERIFY_LEVEL::NONE != micro_writer_->get_micro_block_merge_verify_level() &&
OB_FAIL(check_micro_block( // 检查压缩是否正确
micro_block_desc.buf_, micro_block_desc.buf_size_, block_buffer, block_size, micro_writer_))) {
STORAGE_LOG(WARN, "failed to check micro block", K(ret));
} else if (OB_FAIL(can_mark_deletion(pre_micro_last_key_,
last_key_, mark_deletion))) { // 设置宏块元数据:微块 delete mark
STORAGE_LOG(WARN, "fail to run can mark deletion", K(ret));
} else if (OB_FAIL(save_pre_micro_last_key(last_key_))) { // 设置宏块的元数据:微块的 endkey
STORAGE_LOG(WARN, "Fail to save pre micro last key, ", K(ret), K_(last_key));
} else {
micro_block_desc.last_rowkey_ = micro_writer_->get_last_rowkey();
micro_block_desc.data_size_ = micro_writer_->get_data_size();
micro_block_desc.row_count_ = micro_writer_->get_row_count();
micro_block_desc.column_count_ = micro_writer_->get_column_count();
micro_block_desc.row_count_delta_ = micro_writer_->get_row_count_delta();
micro_block_desc.can_mark_deletion_ = mark_deletion;
micro_block_desc.column_checksums_ =
data_store_desc_->need_calc_column_checksum_ ? curr_micro_column_checksum_ : NULL;
if (data_store_desc_->is_multi_version_minor_sstable()) {
micro_block_desc.max_merged_trans_version_ = micro_writer_->get_max_merged_trans_version();
micro_block_desc.contain_uncommitted_row_ = micro_writer_->is_contain_uncommitted_row();
}
// 将该微块写入宏块,包括:// 1. 将该微块放入到宏块的微块数组中// 2. 更新宏块的 bloomfilterif (OB_FAIL(write_micro_block(micro_block_desc, force_split))) {
STORAGE_LOG(WARN, "build_micro_block failed", K(micro_block_desc), K(force_split), K(ret));
} else {
// 重置微块的写入器,后续可以复用micro_writer_->reuse();if (data_store_desc_->need_prebuild_bloomfilter_ && micro_rowkey_hashs_.count() > 0) {
micro_rowkey_hashs_.reuse();
}
}
}
return ret;
}
// deps/oblib/src/lib/compress/ob_compressor.h
class ObCompressor {
public:
static const char* none_compressor_name;
public:
ObCompressor()
{}
virtual ~ObCompressor()
{}
// 压缩接口virtual int compress(const char* src_buffer, const int64_t src_data_size, char* dst_buffer,
const int64_t dst_buffer_size, int64_t& dst_data_size) = 0;
// 解压接口virtual int decompress(const char* src_buffer, const int64_t src_data_size, char* dst_buffer,
const int64_t dst_buffer_size, int64_t& dst_data_size) = 0;
// 获取 max(压缩后的数据大小 - 原始数据大小)// 主要用来确定 dst_buffer_size: src_data_size + max_overflow_size// 用来预分配压缩后的内存 buffervirtual int get_max_overflow_size(const int64_t src_data_size, int64_t& max_overflow_size) const = 0;
// 获取压缩算法名称接口virtual const char* get_compressor_name() const = 0;
};
// src/storage/blocksstable/ob_macro_block.h
class ObMicroBlockCompressor {
public:
ObMicroBlockCompressor();
virtual ~ObMicroBlockCompressor();
void reset();
// 根据微块大小、压缩算法,初始化压缩器int init(const int64_t micro_block_size, const char* comp_name);
// 压缩接口,通过调用 ObCompressor 的实现类的 compress 接口int compress(const char* in, const int64_t in_size, const char*& out, int64_t& out_size);
// 解压接口,通过调用 ObCompressor 的实现类的 decompress 接口int decompress(const char* in, const int64_t in_size, const int64_t uncomp_size, const char*& out, int64_t& out_size);
private:
bool is_none_;
int64_t micro_block_size_; // 微块大小
common::ObCompressor* compressor_; // 指向压缩算法的实现类
ObSelfBufferWriter comp_buf_; // 压缩 buffer
ObSelfBufferWriter decomp_buf_; // 解压 buffer// ObSelfBufferWriter 是可以自动扩展的内存 buffer 类
};
CREATE TABLE `t` ( `id` int(11) NOT NULL, `k` int(11) DEFAULT NULL, PRIMARY KEY (`id`) ) DEFAULT CHARSET = utf8mb4 ROW_FORMAT = DYNAMIC COMPRESSION = 'zstd_1.0' REPLICA_NUM = 1 BLOCK_SIZE = 16384 USE_BLOOM_FILTER = FALSE TABLET_SIZE = 134217728 PCTFREE = 0


来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70005215/viewspace-2795678/ ,如需转载,请注明出处,否则将追究法律责任。