Trong hệ thống lưu trữ của OpenGauss, tệp rewriteheap.cpp (nằm tại opengauss-server/src/gausskernel/storage/access/heap/rewriteheap.cpp) cung cấp các hàm hỗ trợ việc ghi lại toàn bộ nội dung bảng (heap rewrite) một cách an toàn và hiệu quả. Quá trình này đảm bảo tính nhất quán dữ liệu, duy trì chuỗi cập nhật (update chain), đồng thời xử lý các trường hợp đặc biệt như TOAST, HOT, và mã hóa trang.

Cơ chế hoạt động chính được thiết kế theo luồng sau:

begin_heap_rewrite
while (fetch next tuple) {
    if (tuple is dead)
        rewrite_heap_dead_tuple
    else {
        // optional transformation
        rewrite_heap_tuple
    }
}
end_heap_rewrite

Người gọi phải giữ khóa độc quyền trên bảng đích, vì quá trình giả định không có thao tác ghi đồng thời nào khác.

Giải thích thuật ngữ then chốt

TOAST (The Oversized-Attribute Storage Technique): Cơ chế xử lý giá trị lớn vượt quá kích thước trang 8KB. Dữ liệu được nén hoặc lưu ngoài (out-of-line) vào bảng hệ thống riêng. Các chiến lược lưu trữ bao gồm:

HOT (Heap-Only Tuple): Kỹ thuật cập nhật mà phiên bản mới của tuple không tạo mục nhập chỉ mục mới, giúp tái sử dụng không gian và giảm chi phí VACUUM.

Chuỗi cập nhật (Update chain): Chuỗi các phiên bản tuple liên kết qua trường ctid. Một chuỗi HOT bao gồm tuple gốc và các tuple con chỉ tồn tại trong heap. Cập nhật "lạnh" (cold update) sẽ tạo mục chỉ mục mới, trong khi HOT thì không.

Phân tích hàm trọng tâm

raw_heap_insert

Hàm này chèn tuple vào heap mới trong quá trình rewrite, với tối ưu cho chèn hàng loạt và xử lý đặc biệt cho TOAST:

static void raw_heap_insert(RewriteState state, HeapTuple input_tup)
{
    Page current_page = state->rs_buffer;
    Size aligned_len;
    OffsetNumber inserted_off;

    if (input_tup == NULL) return;

    HeapTuple final_tup;
    if (state->rs_new_rel->rd_rel->relkind == RELKIND_TOASTVALUE) {
        Assert(!HeapTupleHasExternal(input_tup));
        final_tup = input_tup;
    } else if (HeapTupleHasExternal(input_tup) || 
               input_tup->t_len > TOAST_TUPLE_THRESHOLD) {
        final_tup = toast_insert_or_update(
            state->rs_new_rel, input_tup, NULL,
            HEAP_INSERT_SKIP_FSM | (state->rs_use_wal ? 0 : HEAP_INSERT_SKIP_WAL),
            NULL
        );
    } else {
        final_tup = input_tup;
    }

    aligned_len = MAXALIGN(final_tup->t_len);
    if (aligned_len > MaxHeapTupleSize)
        ereport(ERROR, (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                errmsg("row too big: %lu, max %lu", 
                       (unsigned long)aligned_len, (unsigned long)MaxHeapTupleSize)));

    Size reserved_space = RelationGetTargetPageFreeSpace(
        state->rs_new_rel, HEAP_DEFAULT_FILLFACTOR);

    if (state->rs_buffer_valid) {
        if (aligned_len + reserved_space > PageGetHeapFreeSpace(current_page)) {
            rewrite_write_one_page(state, current_page);
            state->rs_blockno++;
            state->rs_buffer_valid = false;
        }
    }

    if (!state->rs_buffer_valid) {
        PageInit(current_page, BLCKSZ, 0, true);
        HeapPageHeader hdr = (HeapPageHeader)current_page;
        hdr->pd_xid_base = u_sess->utils_cxt.RecentXmin - FirstNormalTransactionId;
        hdr->pd_multi_base = 0;
        state->rs_buffer_valid = true;

        if (RelationisEncryptEnable(state->rs_new_rel) || 
            (RelationGetAlgo(state->rs_new_rel) && *RelationGetAlgo(state->rs_new_rel))) {
            hdr->pd_upper -= sizeof(TdePageInfo);
            hdr->pd_special -= sizeof(TdePageInfo);
            PageSetTDE(current_page);
        }
    }

    TransactionId xmin_val = HeapTupleGetRawXmin(final_tup);
    TransactionId xmax_val = HeapTupleGetRawXmax(final_tup);
    rewrite_page_prepare_for_xid(current_page, xmin_val, false);
    rewrite_page_prepare_for_xid(current_page, xmax_val, 
        (final_tup->t_data->t_infomask & HEAP_XMAX_IS_MULTI) != 0);

    HeapTupleCopyBaseFromPage(final_tup, current_page);
    HeapTupleSetXmin(final_tup, xmin_val);
    HeapTupleSetXmax(final_tup, xmax_val);

    inserted_off = PageAddItem(current_page, (Item)final_tup->t_data, 
                               final_tup->t_len, InvalidOffsetNumber, false, true);
    if (inserted_off == InvalidOffsetNumber)
        ereport(ERROR, (errcode(ERRCODE_DATA_CORRUPTED), errmsg("tuple insert failed")));

    ItemPointerSet(&(input_tup->t_self), state->rs_blockno, inserted_off);

    if (!ItemPointerIsValid(&input_tup->t_data->t_ctid)) {
        ItemId item_desc = PageGetItemId(current_page, inserted_off);
        HeapTupleHeader ondisk_hdr = (HeapTupleHeader)PageGetItem(current_page, item_desc);
        ondisk_hdr->t_ctid = input_tup->t_self;
    }

    if (final_tup != input_tup)
        heap_freetuple(final_tup);
}

rewrite_write_one_page

Hàm này ghi trang đã đầy ra đĩa, xử lý WAL và mã hóa nếu cần:

static void rewrite_write_one_page(RewriteState state, Page page_data)
{
    TdeInfo encryption_meta = {0};
    if (RelationisEncryptEnable(state->rs_new_rel))
        GetTdeInfoFromRel(state->rs_new_rel, &encryption_meta);

    if (IsSegmentFileNode(state->rs_new_rel->rd_node)) {
        Assert(state->rs_use_wal);
        Buffer buf = ReadBuffer(state->rs_new_rel, P_NEW);
        LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
        XLogRecPtr lsn = log_newpage(&state->rs_new_rel->rd_node, MAIN_FORKNUM,
                                     state->rs_blockno, page_data, true, &encryption_meta);
        memcpy_s(BufferGetBlock(buf), BLCKSZ, page_data, BLCKSZ);
        PageSetLSN(BufferGetPage(buf), lsn);
        MarkBufferDirty(buf);
        UnlockReleaseBuffer(buf);
    } else {
        STORAGE_SPACE_OPERATION(state->rs_new_rel, BLCKSZ);
        if (state->rs_use_wal)
            log_newpage(&state->rs_new_rel->rd_node, MAIN_FORKNUM,
                        state->rs_blockno, page_data, true, &encryption_meta);
        RelationOpenSmgr(state->rs_new_rel);
        char* write_buffer = RelationisEncryptEnable(state->rs_new_rel) ?
            PageDataEncryptIfNeed(page_data, &encryption_meta, true) : page_data;
        PageSetChecksumInplace((Page)write_buffer, state->rs_blockno);
        rewrite_flush_page(state, (Page)write_buffer);
    }
}

cmpr_heap_insert

Hàm dành riêng cho chèn tuple vào trang đã được nén:

static void cmpr_heap_insert(RewriteState state, HeapTuple tup)
{
    Page cmp_page = state->rs_cmprBuffer;
    Assert(state->rs_new_rel->rd_rel->relkind != RELKIND_TOASTVALUE);
    Assert(!HeapTupleHasExternal(tup) && !(tup->t_len > TOAST_TUPLE_THRESHOLD));

    TransactionId x_min = HeapTupleGetRawXmin(tup);
    TransactionId x_max = HeapTupleGetRawXmax(tup);
    rewrite_page_prepare_for_xid(cmp_page, x_min, false);
    rewrite_page_prepare_for_xid(cmp_page, x_max, 
        (tup->t_data->t_infomask & HEAP_XMAX_IS_MULTI) != 0);

    HeapTupleCopyBaseFromPage(tup, cmp_page);
    HeapTupleSetXmin(tup, x_min);
    HeapTupleSetXmax(tup, x_max);

    Size tuple_size = MAXALIGN(tup->t_len);
    if (tuple_size > MaxHeapTupleSize)
        ereport(ERROR, (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                errmsg("compressed row too big")));

    Assert(PageIsCompressed(cmp_page));
    OffsetNumber off_num = PageAddItem(cmp_page, (Item)tup->t_data, 
                                       tup->t_len, InvalidOffsetNumber, false, true);
    Assert(off_num != InvalidOffsetNumber);
    ItemPointerSet(&(tup->t_self), state->rs_blockno, off_num);

    if (!ItemPointerIsValid(&tup->t_data->t_ctid)) {
        ItemId item_id = PageGetItemId(cmp_page, off_num);
        HeapTupleHeader page_tuple = (HeapTupleHeader)PageGetItem(cmp_page, item_id);
        page_tuple->t_ctid = tup->t_self;
    }
}