【 OpenGauss源码学习 —— 列存储（CStore）（三）】

源码学习 —— 列存储（CStore）（三）】"/>

【 OpenGauss源码学习 —— 列存储（CStore）（三）】

列存储（CStore）（三）

• 概述
• CStore::GetCUDesc函数
• • systable_beginscan_ordered 函数
  • heap_deform_tuple 函数
• CStore::GetCUDeleteMaskIfNeed 函数
• CStore::GetCURowCount 函数
• CStore::GetLivedRowNumbers 函数
• CStore::GetCUData 函数

声明：本文的部分内容参考了他人的文章。在编写过程中，我们尊重他人的知识产权和学术成果，力求遵循合理使用原则，并在适用的情况下注明引用来源。
本文主要参考了 OpenGauss1.1.0 的开源代码和《OpenGauss数据库源码解析》一书以及OpenGauss社区学习文档和一些学习资料

概述

  本章我们继续在【 OpenGauss源码学习 —— 列存储（CStore）（二）】基础上进行进一步学习，我们将继续介绍 CStore 类中的部分公有成员函数。

CStore::GetCUDesc函数

  CStore::GetCUDesc 函数用于从数据库中获取与给定列（col）和 CUID（cuid）相关的 CUDesc（Column Update Description）信息。以下是代码的一些关键部分和功能：

1. 打开关系和索引：代码首先打开了 CUDesc 关系和相应的索引，以便后续的数据检索。
2. 设置扫描键：代码为扫描操作设置了两个扫描键，一个是CUDescColIDAttr（列标识）的等于条件，另一个是 CUDescCUIDAttr（CUID标识）的等于条件。这些条件用于从索引中检索符合条件的数据。
3. 执行系统扫描：代码使用 systable_beginscan_ordered 函数执行系统扫描操作，以获取符合指定条件的 CUDesc 数据。
4. 处理检索到的数据：代码在循环中处理检索到的 CUDesc 数据，包括获取 CUID、xmin、最小值、最大值、行数、CUMode、CU大小、CU指针、magic 等信息，并将这些信息存储在 cuDescPtr 结构中。
5. 关闭关系和索引：最后，代码在完成数据检索后关闭了关系和索引，释放相应的资源。

  需要注意的是，这段代码是在数据库管理系统的上下文中编写的，特定于数据库的数据结构和函数被使用，如 Relation、HeapTuple、systable_beginscan_ordered 等。这段代码的目的是根据给定的列和 CUID 来获取与之相关的 CUDesc 信息，以便在数据库操作中使用。函数源码如下所示：（路径：src/gausskernel/storage/cstore/cstore_am.cpp）

/** Get CUDesc of column according to cuid.* 根据cuid获取列的CUDesc（Column Update Description）。*/
bool CStore::GetCUDesc(_in_ int col, _in_ uint32 cuid, _out_ CUDesc* cuDescPtr, _in_ Snapshot snapShot)
{ScanKeyData key[2];HeapTuple tup;bool found = false;errno_t rc = EOK;Assert(col >= 0);// 当切换到下一批CUDesc数据时，我们将重置m_perScanMemCnxt。// 因此，只用于此批次的空间应由m_perScanMemCnxt管理。AutoContextSwitch newMemCnxt(m_perScanMemCnxt);/** 打开CUDesc关系和其索引*/Relation cudesc_rel = heap_open(m_relation->rd_rel->relcudescrelid, AccessShareLock);TupleDesc cudesc_tupdesc = cudesc_rel->rd_att;Relation idx_rel = index_open(cudesc_rel->rd_rel->relcudescidx, AccessShareLock);bool isFixedLen = m_relation->rd_att->attrs[col]->attlen > 0 ? true : false;// 将逻辑id转换为属性的物理idint attid = m_relation->rd_att->attrs[col]->attnum;/** 设置扫描键以从索引中按attid检索。*/ScanKeyInit(&key[0], (AttrNumber)CUDescColIDAttr, BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(attid));ScanKeyInit(&key[1], (AttrNumber)CUDescCUIDAttr, BTEqualStrategyNumber, F_OIDEQ, UInt32GetDatum(cuid));snapShot = (snapShot == NULL) ? GetActiveSnapshot() : snapShot;Assert(snapShot != NULL);// 执行系统扫描操作，以获取符合指定条件的 CUDesc 数据SysScanDesc cudesc_scan = systable_beginscan_ordered(cudesc_rel, idx_rel, snapShot, 2, key);// 只循环一次while ((tup = systable_getnext_ordered(cudesc_scan, ForwardScanDirection)) != NULL) {Datum values[CUDescCUExtraAttr] = {0};bool isnull[CUDescCUExtraAttr] = {0};char* valPtr = NULL;heap_deform_tuple(tup, cudesc_tupdesc, values, isnull);uint32 cu_id = DatumGetUInt32(values[CUDescCUIDAttr - 1]);Assert(!isnull[CUDescCUIDAttr - 1] && cu_id == cuid && found == false);cuDescPtr->xmin = HeapTupleGetRawXmin(tup);cuDescPtr->cu_id = cu_id;// 将最小值放入cudesc->cu_minif (!isnull[CUDescMinAttr - 1]) {char* minPtr = cuDescPtr->cu_min;char len_1 = MIN_MAX_LEN;valPtr = DatumGetPointer(values[CUDescMinAttr - 1]);if (!isFixedLen) {*minPtr = (char)VARSIZE_ANY_EXHDR(valPtr);minPtr = minPtr + 1;len_1 -= 1;}rc = memcpy_s(minPtr, len_1, VARDATA_ANY(valPtr), VARSIZE_ANY_EXHDR(valPtr));securec_check(rc, "", "");}// 将最大值放入cudesc->maxif (!isnull[CUDescMaxAttr - 1]) {char* maxPtr = cuDescPtr->cu_max;char len_2 = MIN_MAX_LEN;valPtr = DatumGetPointer(values[CUDescMaxAttr - 1]);if (!isFixedLen) {*maxPtr = VARSIZE_ANY_EXHDR(valPtr);maxPtr = maxPtr + 1;len_2 -= 1;}rc = memcpy_s(maxPtr, len_2, VARDATA_ANY(valPtr), VARSIZE_ANY_EXHDR(valPtr));securec_check(rc, "", "");}cuDescPtr->row_count = DatumGetInt32(values[CUDescRowCountAttr - 1]);Assert(!isnull[CUDescRowCountAttr - 1]);// 将CUMode放入cudesc->cumodecuDescPtr->cu_mode = DatumGetInt32(values[CUDescCUModeAttr - 1]);Assert(!isnull[CUDescCUModeAttr - 1]);// 将cusize放入cudesc->cu_sizecuDescPtr->cu_size = DatumGetInt32(values[CUDescSizeAttr - 1]);Assert(!isnull[CUDescSizeAttr - 1]);// 将CUPointer放入cudesc->cuPointerchar* cu_ptr = DatumGetPointer(values[CUDescCUPointerAttr - 1]);Assert(!isnull[CUDescCUPointerAttr - 1] && cu_ptr);rc = memcpy_s(&cuDescPtr->cu_pointer, sizeof(CUPointer), VARDATA_ANY(cu_ptr), sizeof(CUPointer));securec_check(rc, "", "");Assert(VARSIZE_ANY_EXHDR(cu_ptr) == sizeof(CUPointer));cuDescPtr->magic = DatumGetUInt32(values[CUDescCUMagicAttr - 1]);Assert(!isnull[CUDescCUMagicAttr - 1]);found = true;}systable_endscan_ordered(cudesc_scan);index_close(idx_rel, AccessShareLock);heap_close(cudesc_rel, AccessShareLock);return found;
}

systable_beginscan_ordered 函数

  该函数的作用是用于在数据库中设置有序的系统目录扫描，以确保按照索引的顺序返回匹配的元组。该函数会对输入参数进行一些检查，然后设置扫描所需的数据结构，并调用适当的函数来初始化和执行有序的索引扫描。函数源码如下所示：（路径：src\gausskernel\storage\access\index\genam.cpp）

函数入参解释：

1. Relation heap_relation：这是一个指向要进行扫描的表的关系对象的指针。这是扫描的目标表。
2. Relation index_relation：这是一个指向要进行扫描的索引的关系对象的指针。这是扫描目标索引。
3. Snapshot snapshot：这是一个数据库快照对象，用于确定扫描的数据版本。这是扫描的数据一致性的一部分。
4. int nkeys：这是整数，表示搜索键的数量。搜索键是用于限定扫描结果的条件。
5. ScanKey key：这是一个指向扫描键的数组的指针。扫描键是用于确定匹配的条件，每个键包括列号、操作符和要匹配的值。

// 函数目的：设置有序系统目录扫描，确保按照索引顺序返回匹配的元组
SysScanDesc systable_beginscan_ordered(Relation heap_relation, Relation index_relation, Snapshot snapshot, int nkeys,ScanKey key)
{SysScanDesc sysscan;int i;// 检查索引是否正在重新构建，如果是，抛出错误if (ReindexIsProcessingIndex(RelationGetRelid(index_relation)))ereport(ERROR, (errcode(ERRCODE_INVALID_OPERATION),errmsg("无法对索引 \"%s\" 执行有序扫描，因为正在重新构建",RelationGetRelationName(index_relation))));// 检查是否启用了 IgnoreSystemIndexes 配置，如果是，发出警告if (u_sess->attr.attr_common.IgnoreSystemIndexes) {elog(WARNING, "尽管 IgnoreSystemIndexes 设置为真，仍在使用索引 \"%s\"", RelationGetRelationName(index_relation));}// 分配内存以存储 SysScanDesc 结构sysscan = (SysScanDesc)palloc(sizeof(SysScanDescData));// 设置 SysScanDesc 结构的 heap_rel 字段为表引用sysscan->heap_rel = heap_relation;// 设置 SysScanDesc 结构的 irel 字段为索引引用sysscan->irel = index_relation;// 调整搜索键中的属性号以匹配索引列号for (i = 0; i < nkeys; i++) {int j;for (j = 0; j < IndexRelationGetNumberOfAttributes(index_relation); j++) {if (key[i].sk_attno == index_relation->rd_index->indkey.values[j]) {key[i].sk_attno = j + 1;break;}}if (j == IndexRelationGetNumberOfAttributes(index_relation))ereport(ERROR, (errcode(ERRCODE_INDEX_CORRUPTED), errmsg("列不在索引中")));}// 调用 index_beginscan 设置索引扫描sysscan->iscan = (IndexScanDesc)index_beginscan(heap_relation, index_relation, snapshot, nkeys, 0);index_rescan(sysscan->iscan, key, nkeys, NULL, 0);sysscan->scan = NULL;return sysscan;
}

heap_deform_tuple 函数

  heap_deform_tuple 函数用于从堆元组（HeapTuple）中提取数据，并将数据存储到值（values）和空值标志（isnull）数组中。函数的主要功能是根据元组描述（tupleDesc）和堆元组（tuple）中的数据提取相应的值和空值标志。它会遍历元组的各个字段，并根据字段的数据类型和长度，将数据复制到 values 数组中，并相应地设置 isnull 数组的标志。函数源码路径如下：（路径：src\gausskernel\storage\access\common\heaptuple.cpp）

/** heap_deform_tuple* 从元组中提取数据，并将数据放入值数组和空值标志数组中；这是 heap_form_tuple 的逆过程。** 值数组和空值标志数组的存储由调用者提供；其大小应根据 tupleDesc->natts 和* HeapTupleHeaderGetNatts(tuple->t_data, tupleDesc) 来确定。** 需要注意，对于传引用数据类型（pass-by-reference datatypes），放入 Datum 中的指针将指向给定元组中的数据。** 当需要提取元组的所有或大多数字段时，这个函数将比循环使用 heap_getattr 快得多；* 一旦涉及到任何不可缓存属性偏移时，循环将变成 O(N^2)。*/
void heap_deform_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *values, bool *isnull)
{// 获取堆元组的头部信息HeapTupleHeader tup = tuple->t_data;// 检查堆元组是否包含空值标志bool hasnulls = HeapTupleHasNulls(tuple);// 获取元组描述中的属性数组Form_pg_attribute *att = tupleDesc->attrs;// 元组描述中的属性数量uint32 tdesc_natts = tupleDesc->natts;// 要提取的属性数量uint32 natts;// 属性编号uint32 attnum;// 指向元组数据的指针char *tp = NULL;// 数据偏移long off;// 指向元组中的空值位图的指针bits8 *bp = tup->t_bits;// 是否需要慢速路径（无法使用 attcacheoff）bool slow = false;// 确保堆元组没有被压缩Assert(!HEAP_TUPLE_IS_COMPRESSED(tup));// 获取元组的属性数量natts = HeapTupleHeaderGetNatts(tup, tupleDesc);/** 在继承情况下，给定的元组可能实际上具有比调用者期望的更多字段。* 不要超出调用者的数组边界。*/natts = Min(natts, tdesc_natts);// 检查属性数量是否超过限制if (natts > MaxTupleAttributeNumber) {ereport(ERROR, (errcode(ERRCODE_TOO_MANY_COLUMNS),errmsg("列数 (%u) 超过限制 (%d)", natts, MaxTupleAttributeNumber)));}// 初始化元组数据指针tp = (char *)tup + tup->t_hoff;// 初始化数据偏移off = 0;// 遍历属性for (attnum = 0; attnum < natts; attnum++) {// 获取当前属性Form_pg_attribute thisatt = att[attnum];// 如果包含空值，检查并设置相应的标志if (hasnulls && att_isnull(attnum, bp)) {values[attnum] = (Datum)0;isnull[attnum] = true;slow = true; /* 无法再使用 attcacheoff */continue;}// 标记属性不为空isnull[attnum] = false;// 如果不需要慢速路径且属性具有缓存偏移（attcacheoff），使用缓存偏移if (!slow && thisatt->attcacheoff >= 0) {off = thisatt->attcacheoff;} else if (thisatt->attlen == -1) {/** 只有在偏移已经适当对齐的情况下，我们才能缓存 varlena 属性的偏移，* 这样无论偏移适合还是不适合，偏移都对齐。然后，偏移将适用于已对齐或未对齐的值。*/if (!slow && (uintptr_t)(off) == att_align_nominal(off, thisatt->attalign)) {thisatt->attcacheoff = off;} else {off = att_align_pointer(off, thisatt->attalign, -1, tp + off);slow = true;}} else {/* 非 varlena 类型，可以安全使用 att_align_nominal */off = att_align_nominal(off, thisatt->attalign);if (!slow)thisatt->attcacheoff = off;}// 提取属性的值values[attnum] = fetchatt(thisatt, tp + off);// 更新偏移以跳过当前属性的数据off = att_addlength_pointer(off, thisatt->attlen, tp + off);// 如果属性长度小于等于0，不能再使用 attcacheoffif (thisatt->attlen <= 0) {slow = true;}}// 如果元组不包含元组描述中的所有属性，将其余属性读取为空值for (; attnum < tdesc_natts; attnum++) {// 从元组描述中获取初始默认值values[attnum] = heapGetInitDefVal(attnum + 1, tupleDesc, &isnull[attnum]);}
}

CStore::GetCUDeleteMaskIfNeed 函数

  CStore::GetCUDeleteMaskIfNeed 函数这段代码用于从数据库中提取删除掩码信息，以便后续的操作可以正确地处理已删除的行。它还包括了一些错误处理逻辑，以确保数据的一致性和完整性。这段代码执行以下操作：

1. 首先，它检查是否已经加载了特定 cuid 的删除掩码。如果已经加载，它直接返回。
2. 然后，它创建一个新的内存上下文，用于管理批次数据的内存空间。
3. 打开了与 CUDesc 表和相关索引相关的数据库关系。
4. 配置扫描键，以从索引中提取数据。
5. 开始有序扫描 CUDesc 表，尝试查找匹配给定 cuid 的记录。
6. 如果找到匹配记录，它会提取 CUPointer 并存储在 CStore 对象中。
7. 如果没有找到匹配记录，它会根据快照的时间戳和其他条件，决定如何处理。

  函数源码如下所示：（路径：src/gausskernel/storage/cstore/cstore_am.cpp）

void CStore::GetCUDeleteMaskIfNeed(_in_ uint32 cuid, _in_ Snapshot snapShot)
{// 定义扫描键数组ScanKeyData key[2];// 堆元组和相关变量HeapTuple tup;bool isnull = false;errno_t rc = EOK;bool found = false;// 如果删除掩码已加载，则直接返回if (m_delMaskCUId == cuid)return;// 切换到新的内存上下文，用于管理批次数据的内存空间AutoContextSwitch newMemCnxt(m_perScanMemCnxt);// 打开 CUDesc 表及其索引Relation cudesc_rel = heap_open(m_relation->rd_rel->relcudescrelid, AccessShareLock);TupleDesc cudesc_tupdesc = cudesc_rel->rd_att;Relation idx_rel = index_open(cudesc_rel->rd_rel->relcudescidx, AccessShareLock);// 设置用于从索引中提取数据的扫描键ScanKeyInit(&key[0], (AttrNumber)CUDescColIDAttr, BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(VitrualDelColID));ScanKeyInit(&key[1], (AttrNumber)CUDescCUIDAttr, BTEqualStrategyNumber, F_OIDEQ, UInt32GetDatum(cuid));// 如果快照为空，获取活动快照snapShot = (snapShot == NULL) ? GetActiveSnapshot() : snapShot;Assert(snapShot != NULL);// 开始有序扫描 CUDesc 表SysScanDesc cudesc_scan = systable_beginscan_ordered(cudesc_rel, idx_rel, snapShot, 2, key);// 从扫描中获取下一个元组if ((tup = systable_getnext_ordered(cudesc_scan, ForwardScanDirection)) != NULL) {// 获取 CUPointer 并存储在 CStore 对象中Datum v = fastgetattr(tup, CUDescCUPointerAttr, cudesc_tupdesc, &isnull);if (isnull)m_hasDeadRow = false;else {m_hasDeadRow = true;int8* bitmap = (int8*)PG_DETOAST_DATUM(DatumGetPointer(v));rc = memcpy_s(m_cuDelMask, MaxDelBitmapSize, VARDATA_ANY(bitmap), VARSIZE_ANY_EXHDR(bitmap));securec_check(rc, "", "");// 由于可能创建了新内存，因此需要检查并及时释放if ((Pointer)bitmap != DatumGetPointer(v)) {pfree_ext(bitmap);}}found = true;}// 结束扫描systable_endscan_ordered(cudesc_scan);// 关闭索引index_close(idx_rel, AccessShareLock);// 关闭 CUDesc 表heap_close(cudesc_rel, AccessShareLock);// 如果没有找到匹配的记录if (!found) {TransactionId currGlobalXmin = pg_atomic_read_u64(&t_thrd.xact_cxt.ShmemVariableCache->recentGlobalXmin);Assert(snapShot->xmin > 0);// 如果快照太旧，抛出错误if (TransactionIdPrecedes(snapShot->xmin, currGlobalXmin))ereport(ERROR,(errcode(ERRCODE_SNAPSHOT_INVALID),(errmsg("快照过旧。"),errdetail("无法获取旧版本的 CUDeleteBitmap，RecentGlobalXmin: %lu，snapShot->xmin: %lu，snapShot->xmax: %lu",currGlobalXmin,snapShot->xmin,snapShot->xmax),errhint("这是一个安全的错误报告，不会影响数据一致性，如果需要，请重试您的查询。"))));else {if (m_useBtreeIndex)m_delMaskCUId = InValidCUID;else {ereport(PANIC,(errmsg("CU 删除位图丢失。"),errdetail("可能存在有关 cu %u 删除位图的问题，请联系 HW 工程师获取支持。",cuid)));}}} else {m_delMaskCUId = cuid;}return;
}

CStore::GetCURowCount 函数

  CStore::GetCURowCount 函数用于扫描虚拟的 CUDesc（Column Unit Description） 表，以计算特定列的行数。它遵循以下步骤：

1. 首先，它检查输入参数的有效性，并创建新的内存上下文以管理内存空间。
2. 然后，它初始化加载信息并获取与特定列相关的属性标识和关系信息。
3. 之后，它配置用于索引扫描的扫描键，以查找列单元的描述信息。
4. 使用有序扫描的方式遍历 CUDesc 表，获取匹配的行数据，并将它们存储在加载信息中。
5. 最后，它返回 true 表示需要重新加载更多数据，或者返回 false 表示加载完成。

  总的来说，这段代码用于在列存储数据库中获取特定列的行数信息，以支持后续查询和分析操作。函数源码如下所示：（路径：src/gausskernel/storage/cstore/cstore_am.cpp）

/** @Description: 扫描虚拟的 CUDesc 表以计算行数* @Param[IN] col: 列标识* @Param[IN/OUT] loadCUDescInfoPtr: CUDesc 加载信息指针* @Param[IN] snapShot: 扫描快照* @Return: true -- 需要重新加载; false -- 加载完成* @See also: 仅由 GetLivedRowNumbers 调用*/
bool CStore::GetCURowCount(_in_ int col, __inout LoadCUDescCtl* loadCUDescInfoPtr, _in_ Snapshot snapShot)
{// 定义扫描键数组ScanKeyData key[2];// 堆元组和相关变量HeapTuple tup;bool isnull = false;bool found = false;// 断言列编号大于等于0Assert(col >= 0);// 断言加载信息指针有效Assert(loadCUDescInfoPtr);// 创建新的内存上下文，用于管理批次数据的内存空间AutoContextSwitch newMemCnxt(m_perScanMemCnxt);// 重置加载信息中的计数值loadCUDescInfoPtr->lastLoadNum = 0;loadCUDescInfoPtr->curLoadNum = 0;// 获取 CUDesc 数组CUDesc* cuDescArray = loadCUDescInfoPtr->cuDescArray;// 获取列属性标识int attid = m_relation->rd_att->attrs[col]->attnum;// 打开 CUDesc 表及其索引Relation cudesc_rel = heap_open(m_relation->rd_rel->relcudescrelid, AccessShareLock);TupleDesc cudesc_tupdesc = cudesc_rel->rd_att;Relation idx_rel = index_open(cudesc_rel->rd_rel->relcudescidx, AccessShareLock);// 初始化用于索引扫描的扫描键ScanKeyInit(&key[0], (AttrNumber)CUDescColIDAttr, BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(attid));ScanKeyInit(&key[1],(AttrNumber)CUDescCUIDAttr,BTGreaterEqualStrategyNumber,F_OIDGE,UInt32GetDatum(loadCUDescInfoPtr->nextCUID));// 如果快照为空，获取活动快照snapShot = (snapShot == NULL) ? GetActiveSnapshot() : snapShot;// 开始有序扫描 CUDesc 表SysScanDesc cudesc_scan = systable_beginscan_ordered(cudesc_rel, idx_rel, snapShot, 2, key);// 遍历扫描结果while ((tup = systable_getnext_ordered(cudesc_scan, ForwardScanDirection)) != NULL) {uint32 cu_id = DatumGetUInt32(fastgetattr(tup, CUDescCUIDAttr, cudesc_tupdesc, &isnull));Assert(!isnull);// 如果是 Dictionary-based VCU（值编码单元），则跳过if (IsDicVCU(cu_id))continue;// 如果加载信息中没有空闲槽位，退出循环if (!loadCUDescInfoPtr->HasFreeSlot())break;// 存储 CU ID 到加载信息中cuDescArray[loadCUDescInfoPtr->curLoadNum].cu_id = cu_id;loadCUDescInfoPtr->nextCUID = cu_id;// 获取行数并存储到加载信息中cuDescArray[loadCUDescInfoPtr->curLoadNum].row_count =DatumGetInt32(fastgetattr(tup, CUDescRowCountAttr, cudesc_tupdesc, &isnull));Assert(!isnull);// 增加当前加载数量loadCUDescInfoPtr->curLoadNum++;found = true;}// 结束扫描systable_endscan_ordered(cudesc_scan);// 关闭索引index_close(idx_rel, AccessShareLock);// 关闭 CUDesc 表heap_close(cudesc_rel, AccessShareLock);// 如果找到匹配记录if (found) {// 下一个 CUID 必须大于已加载的 CUIDloadCUDescInfoPtr->nextCUID++;return true;}// 加载完成，返回 falsereturn false;
}

CStore::GetLivedRowNumbers 函数

  CStore::GetLivedRowNumbers 函数的主要功能是遍历列存储关系的 CUDesc 表，获取存活行数和死亡行数的统计信息。它通过循环加载 CUDesc 数据，检查每个 CU 的存活和死亡行数，最终计算出总的存活行数和死亡行数。这对于数据库的存储管理和查询优化非常重要，因为它提供了有关表中数据的重要统计信息。函数源码如下所示：（路径：src/gausskernel/storage/cstore/cstore_am.cpp）

/** 获取关系的存活行数。*/
int64 CStore::GetLivedRowNumbers(int64* totaldeadrows)
{int64 rowNumbers = 0; // 存储存活行数LoadCUDescCtl loadInfo(m_startCUID); // 创建 CUDesc 加载信息对象，从指定 CUID 开始加载*totaldeadrows = 0; // 初始化总死亡行数为0// 循环获取列的行数信息while (GetCURowCount(m_firstColIdx, &loadInfo, m_snapshot)) {// 获取加载信息中的 CUDesc 数组CUDesc* cuDescArray = loadInfo.cuDescArray;// 遍历加载信息中的当前加载数量for (uint32 i = 0; i < loadInfo.curLoadNum; ++i) {// 获取 CUDeleteBitmap（如果需要）以检查死亡行GetCUDeleteMaskIfNeed(cuDescArray[i].cu_id, m_snapshot);// 增加存活行数rowNumbers += cuDescArray[i].row_count;// 如果存在死亡行if (m_hasDeadRow) {int nBytes = (cuDescArray[i].row_count + 7) / 8;// 遍历 CUDeleteBitmap 的字节，计算死亡行数for (int j = 0; j < nBytes; ++j) {*totaldeadrows += NumberOfBit1Set[m_cuDelMask[j]];rowNumbers -= NumberOfBit1Set[m_cuDelMask[j]];}}}}loadInfo.Destroy(); // 销毁加载信息对象return rowNumbers; // 返回存活行数
}

注：(cuDescArray[i].row_count + 7) / 8 是一种常见的计算方式，用于确定位图所需的字节数，以便有效地表示一组布尔值。

CStore::GetCUData 函数

  CStore::GetCUData 函数用于从 CU 缓存中获取列存储数据，以提高查询性能。它通过缓存 CU 数据来避免多次从磁盘读取相同的数据，并在需要时进行解压缩。如果数据未在缓存中找到，它将尝试从磁盘加载并解压缩数据，并将数据存储在缓存中以供以后使用。这可以减少 I/O 开销并提高查询性能。以下是代码的主要步骤和作用：

1. 代码首先检查列是否已经被删除，如果已删除则会抛出错误。
2. 切换到专门用于扫描内存的内存上下文（m_perScanMemCnxt）。
3. 初始化一些变量，包括用于记录是否找到 CU 的标志（hasFound）以及一个用于标识 CU 的数据槽标签（dataSlotTag）。
4. 记录一个 CU 的获取操作。
5. 尝试在 CU 缓存中查找 CU（使用 CUCache->FindDataBlock 方法），如果能够找到，直接返回。
6. 如果在 CU 缓存中找不到 CU，尝试在缓存中为其保留一个槽位（使用 CUCache->ReserveDataBlock 方法）。
7. 获取 CU 的缓冲区，将其标记为在 CU 缓存中，并设置其属性信息。
8. 如果 CU 已经在缓存中，返回 CU 数据。否则，继续下面的步骤。
9. 如果 CU 不在缓存中，需要从磁盘上加载 CU 数据。加载前会等待 I/O 操作完成。
10. 如果 CU 在缓存中，返回已加载的 CU 数据。如果没有在缓存中，继续下面的步骤。
11. 如果 CU 需要解压缩，使用 CUCache->StartUncompressCU 方法进行解压缩。如果在解压缩过程中发生错误，会进行错误处理。
12. 校验加载的 CU 数据的 CRC 校验值。
13. 如果 CU 没有在缓存中，并且加载成功，将 CU 数据返回。
14. 校验 CU 数据的一致性。

  函数源码如下所示：（路径：src/gausskernel/storage/cstore/cstore_am.cpp）

// 将CU放入缓存并返回CU数据的指针。
// 返回的CU被固定，调用者在使用完后必须取消固定。
// 1. 记录一个获取（读取）操作。
// 2. 首先通过FindDataBlock()在缓存中查找CU。
//    这通常会成功，而且速度很快。
// 3. 如果FindDataBlock()无法获取CU，则使用InsertCU()。
// 4. 如果FindDataBlock()或InsertCU()发现CU已经在缓存中，
//    则记录缓存命中，返回CU缓冲区和缓存条目。
// 5. 如果InsertCU()没有找到条目，它会保留内存，
//    一个CU描述符槽和一个CU数据槽。
// 6. 从磁盘加载CU并设置CU数据槽，然后检查CRC。
// 7. 解压缩CU数据缓冲区（如果需要）。
// 8. 释放压缩的缓冲区。
// 9. 更新内存保留情况。
// 10. 恢复繁忙的CU缓冲区，唤醒等待缓存条目的线程。
CU* CStore::GetCUData(CUDesc* cuDescPtr, int colIdx, int valSize, int& slotId)
{/** 当切换到下一批cudesc数据时，我们将重置m_PerScanMemCnxt。* 因此，仅应由m_PerScanMemCnxt管理为该批次使用的内存空间，* 包括解压缩中使用的内存空间片段。*/if (m_relation->rd_att->attrs[colIdx]->attisdropped) {ereport(ERROR,(errcode(ERRCODE_INVALID_OPERATION),(errmsg("Cannot get CUData for a dropped column \"%s\" of table \"%s\"",NameStr(m_relation->rd_att->attrs[colIdx]->attname),RelationGetRelationName(m_relation)))));}AutoContextSwitch newMemCnxt(this->m_perScanMemCnxt);CU* cuPtr = NULL;Form_pg_attribute* attrs = m_relation->rd_att->attrs;CUUncompressedRetCode retCode = CU_OK;bool hasFound = false;DataSlotTag dataSlotTag =CUCache->InitCUSlotTag((RelFileNodeOld *)&m_relation->rd_node, colIdx, cuDescPtr->cu_id, cuDescPtr->cu_pointer);// 记录一个获取（读取）操作。// 获取计数是命中和读取次数的总和。if (m_rowCursorInCU == 0) {pgstat_count_buffer_read(m_relation);}RETRY_LOAD_CU:// 首先查找缓存中的CU，这是快速且通常会成功的。slotId = CUCache->FindDataBlock(&dataSlotTag, (m_rowCursorInCU == 0));// 如果CU不在缓存中，则进行预留。// 获取一个缓存槽，预留内存，并将其放入哈希表中。// ReserveDataBlock()可能会因等待空间或CU缓存槽而阻塞。if (IsValidCacheSlotID(slotId)) {hasFound = true;} else {hasFound = false;slotId = CUCache->ReserveDataBlock(&dataSlotTag, cuDescPtr->cu_size, hasFound);}// 使用缓存中的CUcuPtr = CUCache->GetCUBuf(slotId);cuPtr->m_inCUCache = true;cuPtr->SetAttInfo(valSize, attrs[colIdx]->atttypmod, attrs[colIdx]->atttypid);// 如果CU已经在缓存中，直接返回它。if (hasFound) {// 如果仍在进行中，则等待读取完成。if (CUCache->DataBlockWaitIO(slotId)) {CUCache->UnPinDataBlock(slotId);ereport(LOG,(errmodule(MOD_CACHE),errmsg("CU wait IO find an error, need to reload! table(%s), column(%s), relfilenode(%u/%u/%u), ""cuid(%u)",RelationGetRelationName(m_relation),NameStr(m_relation->rd_att->attrs[colIdx]->attname),m_relation->rd_node.spcNode,m_relation->rd_node.dbNode,m_relation->rd_node.relNode,cuDescPtr->cu_id)));goto RETRY_LOAD_CU;}// 当CStore扫描首次访问CU时，计算内存命中。if (m_rowCursorInCU == 0) {// 记录缓存命中。pgstat_count_buffer_hit(m_relation);// 统计CU SSD命中。pgstatCountCUMemHit4SessionLevel();pgstat_count_cu_mem_hit(m_relation);}if (!cuPtr->m_cache_compressed) {CheckConsistenceOfCUData(cuDescPtr, cuPtr, (AttrNumber)(colIdx + 1));return cuPtr;}if (cuPtr->m_cache_compressed) {retCode = CUCache->StartUncompressCU(cuDescPtr, slotId, this->m_plan_node_id, this->m_timing_on, ALIGNOF_CUSIZE);if (retCode == CU_RELOADING) {CUCache->UnPinDataBlock(slotId);ereport(LOG, (errmodule(MOD_CACHE),errmsg("The CU is being reloaded by remote read thread. Retry to load CU! table(%s), ""column(%s), relfilenode(%u/%u/%u), cuid(%u)",RelationGetRelationName(m_relation), NameStr(m_relation->rd_att->attrs[colIdx]->attname),m_relation->rd_node.spcNode, m_relation->rd_node.dbNode, m_relation->rd_node.relNode,cuDescPtr->cu_id)));goto RETRY_LOAD_CU;} else if (retCode == CU_ERR_ADIO) {ereport(ERROR,(errcode(ERRCODE_IO_ERROR),errmodule(MOD_ADIO),errmsg("Load CU failed in adio! table(%s), column(%s), relfilenode(%u/%u/%u), cuid(%u)",RelationGetRelationName(m_relation),NameStr(m_relation->rd_att->attrs[colIdx]->attname),m_relation->rd_node.spcNode,m_relation->rd_node.dbNode,m_relation->rd_node.relNode,cuDescPtr->cu_id)));} else if (retCode == CU_ERR_CRC || retCode == CU_ERR_MAGIC) {/* 预提取的CU包含不正确的校验和 */addBadBlockStat(&m_cuStorage[colIdx]->m_cnode.m_rnode, ColumnId2ColForkNum(m_cuStorage[colIdx]->m_cnode.m_attid));if (RelationNeedsWAL(m_relation) && CanRemoteRead()) {/* 清除CacheBlockInProgressIO和CacheBlockInProgressUncompress，但不释放CU缓冲区 */CUCache->TerminateCU(false);ereport(WARNING,(errcode(ERRCODE_DATA_CORRUPTED),(errmsg("invalid CU in cu_id %u of relation %s file %s offset %lu, prefetch %s, try to ""remote read",cuDescPtr->cu_id,RelationGetRelationName(m_relation),relcolpath(m_cuStorage[colIdx]),cuDescPtr->cu_pointer,GetUncompressErrMsg(retCode))),handle_in_client(true)));/* 远程加载CU */retCode = GetCUDataFromRemote(cuDescPtr, cuPtr, colIdx, valSize, slotId);if (retCode == CU_RELOADING) {/* 其他线程在远程读取 */CUCache->UnPinDataBlock(slotId);ereport(LOG, (errmodule(MOD_CACHE),errmsg("The CU is being reloaded by remote read thread. Retry to load CU! table(%s), ""column(%s), relfilenode(%u/%u/%u), cuid(%u)",RelationGetRelationName(m_relation),NameStr(m_relation->rd_att->attrs[colIdx]->attname), m_relation->rd_node.spcNode,m_relation->rd_node.dbNode, m_relation->rd_node.relNode, cuDescPtr->cu_id)));goto RETRY_LOAD_CU;}} else {// 无记录表不能进行远程读取CUCache->TerminateCU(true);ereport(ERROR,(errcode(ERRCODE_DATA_CORRUPTED),(errmsg("invalid CU in cu_id %u of relation %s file %s offset %lu, prefetch %s",cuDescPtr->cu_id,RelationGetRelationName(m_relation),relcolpath(m_cuStorage[colIdx]),cuDescPtr->cu_pointer,GetUncompressErrMsg(retCode)),errdetail("Can not remote read for unlogged/temp table. Should truncate table and ""re-import data."),handle_in_client(true))));}} else {Assert(retCode == CU_OK);}}CheckConsistenceOfCUData(cuDescPtr, cuPtr, (AttrNumber)(colIdx + 1));return cuPtr;}// stat CU hdd sync readpgstatCountCUHDDSyncRead4SessionLevel();pgstat_count_cu_hdd_sync(m_relation);m_cuStorage[colIdx]->LoadCU(cuPtr, cuDescPtr->cu_pointer, cuDescPtr->cu_size, g_instance.attr.attr_storage.enable_adio_function, true);ADIO_RUN(){ereport(DEBUG1,(errmodule(MOD_ADIO),errmsg("GetCUData:relation(%s), colIdx(%d), load cuid(%u), slotId(%d)",RelationGetRelationName(m_relation),colIdx,cuDescPtr->cu_id,slotId)));}ADIO_END();// Mark the CU as no longer io busy, and wake any waitersCUCache->DataBlockCompleteIO(slotId);retCode = CUCache->StartUncompressCU(cuDescPtr, slotId, this->m_plan_node_id, this->m_timing_on, ALIGNOF_CUSIZE);if (retCode == CU_RELOADING) {CUCache->UnPinDataBlock(slotId);ereport(LOG,(errmodule(MOD_CACHE),errmsg("The CU is being reloaded by remote read thread. Retry to load CU! table(%s), column(%s), ""relfilenode(%u/%u/%u), cuid(%u)",RelationGetRelationName(m_relation),NameStr(m_relation->rd_att->attrs[colIdx]->attname),m_relation->rd_node.spcNode,m_relation->rd_node.dbNode,m_relation->rd_node.relNode,cuDescPtr->cu_id)));goto RETRY_LOAD_CU;} else if (retCode == CU_ERR_CRC || retCode == CU_ERR_MAGIC) {/* Sync load CU contains incorrect checksum */addBadBlockStat(&m_cuStorage[colIdx]->m_cnode.m_rnode, ColumnId2ColForkNum(m_cuStorage[colIdx]->m_cnode.m_attid));if (RelationNeedsWAL(m_relation) && CanRemoteRead()) {/* clear CacheBlockInProgressIO and CacheBlockInProgressUncompress but not free cu buffer */CUCache->TerminateCU(false);ereport(WARNING,(errcode(ERRCODE_DATA_CORRUPTED),(errmsg("invalid CU in cu_id %u of relation %s file %s offset %lu, sync load %s, try to remote read",cuDescPtr->cu_id,RelationGetRelationName(m_relation),relcolpath(m_cuStorage[colIdx]),cuDescPtr->cu_pointer,GetUncompressErrMsg(retCode)),handle_in_client(true))));/* remote load cu */retCode = GetCUDataFromRemote(cuDescPtr, cuPtr, colIdx, valSize, slotId);if (retCode == CU_RELOADING) {/* other thread in remote read */CUCache->UnPinDataBlock(slotId);ereport(LOG,(errmodule(MOD_CACHE),errmsg("The CU is being reloaded by remote read thread. Retry to load CU! table(%s), ""column(%s), relfilenode(%u/%u/%u), cuid(%u)",RelationGetRelationName(m_relation), NameStr(m_relation->rd_att->attrs[colIdx]->attname),m_relation->rd_node.spcNode, m_relation->rd_node.dbNode, m_relation->rd_node.relNode,cuDescPtr->cu_id)));goto RETRY_LOAD_CU;}} else {// unlogged table can not remote readCUCache->TerminateCU(true);ereport(ERROR, (errcode(ERRCODE_DATA_CORRUPTED),(errmsg("invalid CU in cu_id %u of relation %s file %s offset %lu, sync load %s", cuDescPtr->cu_id,RelationGetRelationName(m_relation), relcolpath(m_cuStorage[colIdx]), cuDescPtr->cu_pointer,GetUncompressErrMsg(retCode)),errdetail("Can not remote read for unlogged/temp table. Should truncate table and re-import ""data."))));}}Assert(retCode == CU_OK);if (t_thrd.vacuum_cxt.VacuumCostActive) {// cu cache misses, so we update vacuum statst_thrd.vacuum_cxt.VacuumCostBalance += u_sess->attr.attr_storage.VacuumCostPageMiss;}CheckConsistenceOfCUData(cuDescPtr, cuPtr, (AttrNumber)(colIdx + 1));return cuPtr;
}