openGauss行存储引擎

openGauss的行存储引擎设计上支持MVCC（multi-version concurrency control，多版本并发控制），采用集中式垃圾版本回收机制，可以提供OLTP业务系统的高并发读写要求。架构如图1-6所示。

图1-6 行存储架构
行存储引擎的关键技术有：
（1） 基于CSN（commit sequence number，待提交事务的序列号，它是一个64位递增无符号数）的MVCC并发控制机制，进行集中式垃圾数据清理。
（2） 并行刷新日志，并行恢复。传统数据库一般都采用串行刷日志的设计，因为日志有顺序依赖关系，例如一个由事务产生的redo/undo log是有前后依赖关系的。openGauss的日志系统采用多个logwriter线程并行写的机制，充分发挥SSD的多通道I/O能力。
（3） 基于大内存设计的缓冲管理器。
行存储缓冲区主流程代码如下：
/* buffer/bufmgr.cpp, 基础行存储管理 */
...
/* 查找或创建一个缓冲区 */
Buffer ReadBufferExtended(
Relation reln, ForkNumber fork_num, BlockNumber block_num, ReadBufferMode mode, BufferAccessStrategy strategy)
{
bool hit = false;
Buffer buf;
if (block_num == P_NEW) {
STORAGE_SPACE_OPERATION(reln, BLCKSZ);
}
/* 以smgr(存储管理器)级别打开一个缓冲区 */
RelationOpenSmgr(reln);
/* 拒绝读取非局部临时关系的请求，因为可能会获得监控不到的错误数据 */
if (RELATION_IS_OTHER_TEMP(reln) && fork_num <= INIT_FORKNUM)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("cannot access temporary tables of other sessions")));
/* 读取缓冲区，更新统计信息数量反馈缓存命中与否情况 */
pgstat_count_buffer_read(reln);
pgstatCountBlocksFetched4SessionLevel();
buf = ReadBuffer_common(reln->rd_smgr, reln->rd_rel->relpersistence, fork_num, block_num, mode, strategy, &hit);
if (hit) {
pgstat_count_buffer_hit(reln);
}
return buf;
}
/* 释放一个缓冲区 */
void ReleaseBuffer(Buffer buffer)
{
BufferDesc* buf_desc = NULL;
PrivateRefCountEntry* ref = NULL;
/* 错误释放处理 */
if (!BufferIsValid(buffer)) {
ereport(ERROR, (errcode(ERRCODE_INVALID_BUFFER), (errmsg("bad buffer ID: %d", buffer))));
}
ResourceOwnerForgetBuffer(t_thrd.utils_cxt.CurrentResourceOwner, buffer);
if (BufferIsLocal(buffer)) {
Assert(u_sess->storage_cxt.LocalRefCount[-buffer - 1] > 0);
u_sess->storage_cxt.LocalRefCount[-buffer - 1]--;
return;
}
/* 释放当前缓冲区 */
buf_desc = GetBufferDescriptor(buffer - 1);
PrivateRefCountEntry *free_entry = NULL;
ref = GetPrivateRefCountEntryFast(buffer, free_entry);
if (ref == NULL) {
ref = GetPrivateRefCountEntrySlow(buffer, false, false, free_entry);}
Assert(ref != NULL);
Assert(ref->refcount > 0);
if (ref->refcount > 1) {
ref->refcount--;
} else {
UnpinBuffer(buf_desc, false);
}
}
/* 标记写脏缓冲区 */
void MarkBufferDirty(Buffer buffer)
{
BufferDesc* buf_desc = NULL;
uint32 buf_state;
uint32 old_buf_state;
if (!BufferIsValid(buffer)) {
ereport(ERROR, (errcode(ERRCODE_INVALID_BUFFER), (errmsg("bad buffer ID: %d", buffer))));}
if (BufferIsLocal(buffer)) {
MarkLocalBufferDirty(buffer);
return;
}
buf_desc = GetBufferDescriptor(buffer - 1);
Assert(BufferIsPinned(buffer));
Assert(LWLockHeldByMe(buf_desc->content_lock));
old_buf_state = LockBufHdr(buf_desc);
buf_state = old_buf_state | (BM_DIRTY | BM_JUST_DIRTIED);
/* 将未入队的脏页入队 */
if (g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
for (;;) {
buf_state = old_buf_state | (BM_DIRTY | BM_JUST_DIRTIED);
if (!XLogRecPtrIsInvalid(pg_atomic_read_u64(&buf_desc->rec_lsn))) {
break;
}
if (!is_dirty_page_queue_full(buf_desc) && push_pending_flush_queue(buffer)) {
break;
}
UnlockBufHdr(buf_desc, old_buf_state);
pg_usleep(TEN_MICROSECOND);
old_buf_state = LockBufHdr(buf_desc);
}
}
UnlockBufHdr(buf_desc, buf_state);
/* 如果缓冲区不是“脏”状态，则更新相关计数 */
if (!(old_buf_state & BM_DIRTY)) {
t_thrd.vacuum_cxt.VacuumPageDirty++;
u_sess->instr_cxt.pg_buffer_usage->shared_blks_dirtied++;
pgstatCountSharedBlocksDirtied4SessionLevel();
if (t_thrd.vacuum_cxt.VacuumCostActive) {
t_thrd.vacuum_cxt.VacuumCostBalance += u_sess->attr.attr_storage.VacuumCostPageDirty;
}
}
}