石头队小组代码评注2 #45

Open
Cachuela wants to merge 19 commits from yangke1125/openGauss-server:master into master
10 changed files with 2500 additions and 1636 deletions

File diff suppressed because it is too large Load Diff

View File

@ -30,47 +30,56 @@
* index_ tuple interface routines
* ----------------------------------------------------------------
*/
// 函数用于在索引元组中查找指定属性号attrno的属性值
static inline bool index_findattr(Relation irel, IndexTuple itup, AttrNumber attrno, Datum *value)
{
bool isnull = false;
bool isnull = false; // 初始化是否为NULL标志
TupleDesc tupdesc = RelationGetDescr(irel);
int nattrs = IndexRelationGetNumberOfAttributes(irel);
TupleDesc tupdesc = RelationGetDescr(irel); // 获取索引关系的元组描述符
int nattrs = IndexRelationGetNumberOfAttributes(irel); // 获取索引关系的属性数目
for (int i = 0; i < nattrs; i++) {
// 检查索引属性的键indkey中是否包含要查找的属性号
if (irel->rd_index->indkey.values[i] == attrno) {
// 如果匹配使用index_getattr函数获取属性值
*value = index_getattr(itup, i + 1, tupdesc, &isnull);
break;
}
}
return !isnull;
return !isnull; // 返回是否找到非NULL的属性值
}
// 函数用于获取索引元组的表OID
Oid index_getattr_tableoid(Relation irel, IndexTuple itup)
{
Datum val = 0;
Oid tableoid = InvalidOid;
Datum val = 0; // 初始化属性值
Oid tableoid = InvalidOid; // 初始化表OID为无效OID
Assert(RelationIsIndex(irel));
Assert(RelationIsIndex(irel)); // 断言,确保传入的关系是索引关系
// 调用index_findattr函数查找表OID属性值
if (index_findattr(irel, itup, TableOidAttributeNumber, &val)) {
tableoid = DatumGetUInt32(val);
tableoid = DatumGetUInt32(val); // 将属性值转换为Oid类型
}
return tableoid;
return tableoid; // 返回表OID
}
// 函数用于获取索引元组的桶ID
int2 index_getattr_bucketid(Relation irel, IndexTuple itup)
{
Datum val = 0;
int2 bucketid = InvalidBktId;
Datum val = 0; // 初始化属性值
int2 bucketid = InvalidBktId; // 初始化桶ID为无效桶ID
Assert(RelationIsIndex(irel));
Assert(RelationIsIndex(irel)); // 断言,确保传入的关系是索引关系
// 调用index_findattr函数查找桶ID属性值
if (index_findattr(irel, itup, BucketIdAttributeNumber, &val)) {
bucketid = DatumGetInt16(val);
bucketid = DatumGetInt16(val); // 将属性值转换为int2类型
}
return bucketid;
return bucketid; // 返回桶ID
}
/* ----------------
@ -84,39 +93,43 @@ IndexTuple index_form_tuple(TupleDesc tuple_descriptor, Datum* values, const boo
{
char *tp = NULL; /* tuple pointer */
IndexTuple tuple = NULL; /* return tuple */
Size size, data_size, hoff;
int i;
unsigned short infomask = 0;
bool hasnull = false;
uint16 tupmask = 0;
int attributeNum = tuple_descriptor->natts;
Size size, data_size, hoff; // 声明一些用于计算大小和偏移量的变量
int i;
unsigned short infomask = 0; // 初始化元组信息掩码
bool hasnull = false; // 初始化是否包含NULL值的标志
uint16 tupmask = 0; // 初始化元组掩码
int attributeNum = tuple_descriptor->natts; // 获取属性数目
Size (*computedatasize_tuple)(TupleDesc tuple_desc, Datum* values, const bool* isnull);
void (*filltuple)(TupleDesc tuple_desc, Datum* values, const bool* isnull, char* data, Size data_size, uint16* infomask, bits8* bit);
// 声明函数指针,用于计算数据大小和填充元组数据
Size (*computedatasize_tuple)(TupleDesc tuple_desc, Datum* values, const bool* isnull);
void (*filltuple)(TupleDesc tuple_desc, Datum* values, const bool* isnull, char* data, Size data_size, uint16* infomask, bits8* bit);
computedatasize_tuple = &heap_compute_data_size;
filltuple = &heap_fill_tuple;
// 初始化函数指针
computedatasize_tuple = &heap_compute_data_size;
filltuple = &heap_fill_tuple;
#ifdef TOAST_INDEX_HACK
Datum untoasted_values[INDEX_MAX_KEYS];
bool untoasted_free[INDEX_MAX_KEYS];
Datum untoasted_values[INDEX_MAX_KEYS]; // 声明未压缩的属性值数组
bool untoasted_free[INDEX_MAX_KEYS]; // 声明未压缩的属性值释放标志数组
#endif
if (attributeNum > INDEX_MAX_KEYS)
ereport(ERROR,
(errcode(ERRCODE_TOO_MANY_COLUMNS),
errmsg("number of index columns (%d) exceeds limit (%d)", attributeNum, INDEX_MAX_KEYS)));
// 检查属性数目是否超过索引键的最大限制
if (attributeNum > INDEX_MAX_KEYS)
ereport(ERROR,
(errcode(ERRCODE_TOO_MANY_COLUMNS),
errmsg("number of index columns (%d) exceeds limit (%d)", attributeNum, INDEX_MAX_KEYS)));
#ifdef TOAST_INDEX_HACK
uint32 toastTarget = TOAST_INDEX_TARGET;
if (tuple_descriptor->tdTableAmType == TAM_USTORE) {
toastTarget = UTOAST_INDEX_TARGET;
}
for (i = 0; i < attributeNum; i++) {
Form_pg_attribute att = tuple_descriptor->attrs[i];
uint32 toastTarget = TOAST_INDEX_TARGET; // 初始化TOAST目标大小
if (tuple_descriptor->tdTableAmType == TAM_USTORE) {
toastTarget = UTOAST_INDEX_TARGET; // 如果表AM类型是UTable则使用UTable的TOAST目标大小
}
for (i = 0; i < attributeNum; i++) {
Form_pg_attribute att = tuple_descriptor->attrs[i];
untoasted_values[i] = values[i];
untoasted_free[i] = false;
untoasted_values[i] = values[i]; // 复制未压缩的属性值
untoasted_free[i] = false; // 初始化未压缩的属性值释放标志
}
/* Do nothing if value is NULL or not of varlena type */
if (isnull[i] || att->attlen != -1)
@ -137,60 +150,71 @@ IndexTuple index_form_tuple(TupleDesc tuple_descriptor, Datum* values, const boo
* If value is above size target, and is of a compressible datatype,
* try to compress it in-line.
*/
if (!VARATT_IS_EXTENDED(DatumGetPointer(untoasted_values[i])) &&
VARSIZE(DatumGetPointer(untoasted_values[i])) > toastTarget &&
(att->attstorage == 'x' || att->attstorage == 'm')) {
Datum cvalue = toast_compress_datum(untoasted_values[i]);
if (DatumGetPointer(cvalue) != NULL) {
// 遍历所有属性值检查是否需要进行TOAST压缩
for (i = 0; i < attributeNum; i++) {
// 如果属性值不是扩展形式并且大于TOAST目标大小并且属性存储类型是扩展形式或主存储形式
if (!VARATT_IS_EXTENDED(DatumGetPointer(untoasted_values[i])) &&
VARSIZE(DatumGetPointer(untoasted_values[i])) > toastTarget &&
(att->attstorage == 'x' || att->attstorage == 'm')) {
// 对属性值进行TOAST压缩
Datum cvalue = toast_compress_datum(untoasted_values[i]);
if (DatumGetPointer(cvalue) != NULL) {
/* successful compression */
if (untoasted_free[i])
pfree(DatumGetPointer(untoasted_values[i]));
untoasted_values[i] = cvalue;
untoasted_free[i] = true;
}
if (untoasted_free[i])
pfree(DatumGetPointer(untoasted_values[i]));
untoasted_values[i] = cvalue;
untoasted_free[i] = true;
}
}
#endif
}
for (i = 0; i < attributeNum; i++) {
if (isnull[i]) {
hasnull = true;
break;
}
// 检查是否有NULL属性值
for (i = 0; i < attributeNum; i++) {
if (isnull[i]) {
hasnull = true;
break;
}
}
if (hasnull)
infomask |= INDEX_NULL_MASK;
if (hasnull)
infomask |= INDEX_NULL_MASK;
hoff = IndexInfoFindDataOffset(infomask);
// 计算数据在元组中的偏移量
hoff = IndexInfoFindDataOffset(infomask);
// 计算数据大小
#ifdef TOAST_INDEX_HACK
data_size = computedatasize_tuple(tuple_descriptor, untoasted_values, isnull);
data_size = computedatasize_tuple(tuple_descriptor, untoasted_values, isnull);
#else
data_size = computedatasize_tuple(tuple_descriptor, values, isnull);
data_size = computedatasize_tuple(tuple_descriptor, values, isnull);
#endif
size = hoff + data_size;
size = MAXALIGN(size); /* be conservative */
tp = (char*)palloc0(size);
tuple = (IndexTuple)tp;
size = hoff + data_size; // 计算总大小
size = MAXALIGN(size); // 对齐到最大值
filltuple(tuple_descriptor,
// 分配元组数据的内存
tp = (char*)palloc0(size);
tuple = (IndexTuple)tp;
// 填充元组数据
filltuple(tuple_descriptor,
#ifdef TOAST_INDEX_HACK
untoasted_values,
untoasted_values,
#else
values,
values,
#endif
isnull,
(char*)tp + hoff,
data_size,
&tupmask,
(hasnull ? (bits8*)tp + sizeof(IndexTupleData) : NULL));
isnull,
(char*)tp + hoff, // 数据的偏移量
data_size,
&tupmask,
(hasnull ? (bits8*)tp + sizeof(IndexTupleData) : NULL)); // 如果包含NULL值则设置位图
// 如果使用了TOAST压缩释放未压缩的属性值
#ifdef TOAST_INDEX_HACK
for (i = 0; i < attributeNum; i++) {
if (untoasted_free[i])
pfree(DatumGetPointer(untoasted_values[i]));
}
for (i = 0; i < attributeNum; i++) {
if (untoasted_free[i])
pfree(DatumGetPointer(untoasted_values[i]));
}
#endif
/*
@ -206,14 +230,17 @@ IndexTuple index_form_tuple(TupleDesc tuple_descriptor, Datum* values, const boo
* Here we make sure that the size will fit in the field reserved for it
* in t_info.
*/
if ((size & INDEX_SIZE_MASK) != size)
ereport(ERROR,
(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
errmsg("index row requires %lu bytes, maximum size is %lu",
(unsigned long)size,
(unsigned long)INDEX_SIZE_MASK)));
// 检查索引行大小是否超过了索引行大小限制INDEX_SIZE_MASK
if ((size & INDEX_SIZE_MASK) != size) {
ereport(ERROR,
(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
errmsg("index row requires %lu bytes, maximum size is %lu",
(unsigned long)size,
(unsigned long)INDEX_SIZE_MASK)));
}
infomask |= size;
// 将大小信息添加到infomask中
infomask |= size;
/*
* initialize metadata
@ -337,25 +364,34 @@ Datum nocache_index_getattr(IndexTuple tup, uint32 attnum, TupleDesc tuple_desc)
att[0]->attcacheoff = 0;
/* we might have set some offsets in the slow path previously */
while (j < natts && att[j]->attcacheoff > 0)
j++;
// 循环直到找到第一个具有正值attcacheoff的属性或达到属性数上限
while (j < natts && att[j]->attcacheoff > 0)
j++;
off = att[j - 1]->attcacheoff + att[j - 1]->attlen;
// 计算偏移量,从上一步找到的第一个属性之后开始
off = att[j - 1]->attcacheoff + att[j - 1]->attlen;
for (; j < natts; j++) {
if (att[j]->attlen <= 0)
break;
// 遍历属性,为每个属性计算偏移量
for (; j < natts; j++) {
// 如果属性的长度小于等于0退出循环
if (att[j]->attlen <= 0)
break;
off = att_align_nominal((uint32)off, att[j]->attalign);
// 根据属性的对齐方式对偏移量进行对齐
off = att_align_nominal((uint32)off, att[j]->attalign);
att[j]->attcacheoff = off;
// 将计算后的偏移量存储在属性描述符的attcacheoff字段中
att[j]->attcacheoff = off;
off += att[j]->attlen;
}
// 增加偏移量,以包括当前属性
off += att[j]->attlen;
}
Assert(j > attnum);
// 断言确保找到的属性数量j大于原始属性编号attnum
Assert(j > attnum);
off = att[attnum]->attcacheoff;
// 将指定属性attnum的偏移量存储在off中
off = att[attnum]->attcacheoff;
} else {
bool usecache = true;
uint32 i;
@ -387,27 +423,28 @@ Datum nocache_index_getattr(IndexTuple tup, uint32 attnum, TupleDesc tuple_desc)
* no pad bytes in any case: then the offset will be valid for
* either an aligned or unaligned value.
*/
if (usecache && (uintptr_t)(off) == att_align_nominal((uint32)off, att[i]->attalign))
att[i]->attcacheoff = off;
else {
off = att_align_pointer((uint32)off, att[i]->attalign, -1, tp + off);
usecache = false;
}
if (usecache && (uintptr_t)(off) == att_align_nominal((uint32)off, att[i]->attalign))
att[i]->attcacheoff = off; // 如果使用缓存且与属性对齐,将缓存偏移设置为 off
else {
off = att_align_pointer((uint32)off, att[i]->attalign, -1, tp + off); // 否则,重新计算 off 以满足属性对齐,并禁用缓存
usecache = false;
}
} else {
/* not varlena, so safe to use att_align_nominal */
off = att_align_nominal((uint32)off, att[i]->attalign);
if (usecache)
att[i]->attcacheoff = off;
}
}// 如果允许缓存,将缓存偏移设置为 off
if (i == attnum)
break;
if (i == attnum)
break; // 如果 i 等于目标属性编号,跳出循环
off = att_addlength_pointer(off, att[i]->attlen, tp + off);
off = att_addlength_pointer(off, att[i]->attlen, tp + off); // 根据属性长度调整 off
if (usecache && att[i]->attlen <= 0)
usecache = false; // 如果允许缓存且属性长度小于等于0禁用缓存
if (usecache && att[i]->attlen <= 0)
usecache = false;
}
}
@ -441,13 +478,14 @@ IndexTuple CopyIndexTuple(IndexTuple source)
Size size;
errno_t rc = EOK;
size = IndexTupleSize(source);
result = (IndexTuple)palloc(size);
rc = memcpy_s(result, size, source, size);
securec_check(rc, "\0", "\0");
return result;
size = IndexTupleSize(source); // 计算源索引元组的大小
result = (IndexTuple)palloc(size); // 为结果分配内存空间
rc = memcpy_s(result, size, source, size); // 复制源索引元组到结果
securec_check(rc, "\0", "\0"); // 检查复制是否成功
return result; // 返回复制的索引元组
}
/*
* Create a palloc'd copy of an index tuple with a reserved space.
*/
@ -457,12 +495,12 @@ IndexTuple CopyIndexTupleAndReserveSpace(IndexTuple source, Size reserved_size)
Size size;
errno_t rc = EOK;
size = IndexTupleSize(source);
result = (IndexTuple)palloc0(size + reserved_size);
rc = memcpy_s(result, size, source, size);
securec_check(rc, "\0", "\0");
IndexTupleSetSize(result, size + reserved_size);
return result;
size = IndexTupleSize(source); // 计算源索引元组的大小
result = (IndexTuple)palloc0(size + reserved_size); // 为结果分配内存空间,并清零
rc = memcpy_s(result, size, source, size); // 复制源索引元组到结果
securec_check(rc, "\0", "\0"); // 检查复制是否成功
IndexTupleSetSize(result, size + reserved_size); // 设置结果的大小
return result; // 返回带有保留空间的复制索引元组
}
/*
@ -471,25 +509,25 @@ IndexTuple CopyIndexTupleAndReserveSpace(IndexTuple source, Size reserved_size)
*/
IndexTuple index_truncate_tuple(TupleDesc tupleDescriptor, IndexTuple olditup, int new_indnatts)
{
TupleDesc itupdesc = CreateTupleDescCopyConstr(tupleDescriptor);
TupleDesc itupdesc = CreateTupleDescCopyConstr(tupleDescriptor); // 创建源元组描述符的副本
Datum values[INDEX_MAX_KEYS];
bool isnull[INDEX_MAX_KEYS];
IndexTuple newitup;
Assert(tupleDescriptor->natts <= INDEX_MAX_KEYS);
Assert(new_indnatts > 0);
Assert(new_indnatts < tupleDescriptor->natts);
Assert(tupleDescriptor->natts <= INDEX_MAX_KEYS); // 断言检查索引列数不超过最大限制
Assert(new_indnatts > 0); // 断言检查新索引列数大于0
Assert(new_indnatts < tupleDescriptor->natts); // 断言检查新索引列数小于原索引列数
index_deform_tuple(olditup, tupleDescriptor, values, isnull);
index_deform_tuple(olditup, tupleDescriptor, values, isnull); // 解析原索引元组
/* form new tuple that will contain only key attributes */
itupdesc->natts = new_indnatts;
newitup = index_form_tuple(itupdesc, values, isnull);
newitup->t_tid = olditup->t_tid;
/* form new tuple that will contain only key attributes */
itupdesc->natts = new_indnatts; // 设置新描述符的属性数
newitup = index_form_tuple(itupdesc, values, isnull); // 创建新索引元组
newitup->t_tid = olditup->t_tid; // 设置新索引元组的 TID 与原索引元组相同
FreeTupleDesc(itupdesc);
Assert(IndexTupleSize(newitup) <= IndexTupleSize(olditup));
return newitup;
FreeTupleDesc(itupdesc); // 释放新元组描述符
Assert(IndexTupleSize(newitup) <= IndexTupleSize(olditup)); // 断言检查新索引元组的大小不超过原索引元组
return newitup; // 返回新的索引元组
}
/*
@ -502,16 +540,16 @@ IndexTuple UBTreeIndexTruncateTuple(TupleDesc tupleDescriptor, IndexTuple olditu
Datum values[INDEX_MAX_KEYS];
bool isnull[INDEX_MAX_KEYS];
IndexTuple newitup;
int indnatts = tupleDescriptor->natts;
int indnatts = tupleDescriptor->natts;// 创建源元组描述符的副本
if (indnatts > INDEX_MAX_KEYS) {
if (indnatts > INDEX_MAX_KEYS) {
ereport(ERROR,
(errcode(ERRCODE_TOO_MANY_COLUMNS),
errmsg("number of index columns (%d) exceeds limit (%d)", indnatts, INDEX_MAX_KEYS)));
}
} // 如果索引列数超过了最大限制,报错
Assert(leavenatts > 0);
Assert(leavenatts <= indnatts);
Assert(leavenatts > 0); // 断言检查要保留的新索引列数大于0
Assert(leavenatts <= indnatts); // 断言检查要保留的新索引列数不超过原索引列数
/* Easy case: no truncation actually required */
if (leavenatts == indnatts) {
@ -535,13 +573,13 @@ IndexTuple UBTreeIndexTruncateTuple(TupleDesc tupleDescriptor, IndexTuple olditu
index_deform_tuple(olditup, tupleDescriptor, values, isnull);
/* form new tuple that will contain only key attributes */
itupdesc->natts = leavenatts;
newitup = index_form_tuple(itupdesc, values, isnull);
newitup->t_tid = olditup->t_tid;
Assert(IndexTupleSize(newitup) <= IndexTupleSize(olditup));
itupdesc->natts = leavenatts; // 设置新描述符的属性数
newitup = index_form_tuple(itupdesc, values, isnull); // 创建新索引元组
newitup->t_tid = olditup->t_tid; // 设置新索引元组的 TID 与原索引元组相同
Assert(IndexTupleSize(newitup) <= IndexTupleSize(olditup)); // 断言检查新索引元组的大小不超过原索引元组
FreeTupleDesc(itupdesc);
FreeTupleDesc(itupdesc); // 释放新元组描述符
return newitup;
return newitup; // 返回新的索引元组
}

View File

@ -89,70 +89,69 @@ DestReceiver *createStreamDestReceiver(CommandDest dest)
streamReceiver *self = (streamReceiver *)palloc0(sizeof(streamReceiver));
/* Assign data send function based on the stream type. */
switch (dest) {
case DestTupleBroadCast:
self->pub.receiveSlot = printBroadCastTuple;
break;
switch (dest) {
case DestTupleBroadCast:
self->pub.receiveSlot = printBroadCastTuple; // 如果 dest 为 DestTupleBroadCast设置 receiveSlot 为 printBroadCastTuple 函数
break;
case DestTupleLocalBroadCast:
self->pub.receiveSlot = printLocalBroadCastTuple;
break;
case DestTupleLocalBroadCast:
self->pub.receiveSlot = printLocalBroadCastTuple; // 如果 dest 为 DestTupleLocalBroadCast设置 receiveSlot 为 printLocalBroadCastTuple 函数
break;
case DestTupleRedistribute:
self->pub.receiveSlot = printRedistributeTuple;
break;
case DestTupleRedistribute:
self->pub.receiveSlot = printRedistributeTuple; // 如果 dest 为 DestTupleRedistribute设置 receiveSlot 为 printRedistributeTuple 函数
break;
case DestTupleLocalRedistribute:
self->pub.receiveSlot = printLocalRedistributeTuple;
break;
case DestTupleLocalRedistribute:
self->pub.receiveSlot = printLocalRedistributeTuple; // 如果 dest 为 DestTupleLocalRedistribute设置 receiveSlot 为 printLocalRedistributeTuple 函数
break;
case DestTupleLocalRoundRobin:
self->pub.receiveSlot = printLocalRoundRobinTuple;
break;
case DestTupleLocalRoundRobin:
self->pub.receiveSlot = printLocalRoundRobinTuple; // 如果 dest 为 DestTupleLocalRoundRobin设置 receiveSlot 为 printLocalRoundRobinTuple 函数
break;
case DestTupleHybrid:
self->pub.receiveSlot = printHybridTuple;
break;
case DestTupleHybrid:
self->pub.receiveSlot = printHybridTuple; // 如果 dest 为 DestTupleHybrid设置 receiveSlot 为 printHybridTuple 函数
break;
case DestBatchBroadCast:
self->pub.sendBatch = printBroadCastBatchCompress;
break;
case DestBatchBroadCast:
self->pub.sendBatch = printBroadCastBatchCompress; // 如果 dest 为 DestBatchBroadCast设置 sendBatch 为 printBroadCastBatchCompress 函数
break;
case DestBatchLocalBroadCast:
self->pub.sendBatch = printLocalBroadCastBatch;
break;
case DestBatchLocalBroadCast:
self->pub.sendBatch = printLocalBroadCastBatch; // 如果 dest 为 DestBatchLocalBroadCast设置 sendBatch 为 printLocalBroadCastBatch 函数
break;
case DestBatchRedistribute:
self->pub.sendBatch = printRedistributeBatch;
break;
case DestBatchRedistribute:
self->pub.sendBatch = printRedistributeBatch; // 如果 dest 为 DestBatchRedistribute设置 sendBatch 为 printRedistributeBatch 函数
break;
case DestBatchLocalRedistribute:
self->pub.sendBatch = printLocalRedistributeBatch;
break;
case DestBatchLocalRedistribute:
self->pub.sendBatch = printLocalRedistributeBatch; // 如果 dest 为 DestBatchLocalRedistribute设置 sendBatch 为 printLocalRedistributeBatch 函数
break;
case DestBatchLocalRoundRobin:
self->pub.sendBatch = printLocalRoundRobinBatch;
break;
case DestBatchLocalRoundRobin:
self->pub.sendBatch = printLocalRoundRobinBatch; // 如果 dest 为 DestBatchLocalRoundRobin设置 sendBatch 为 printLocalRoundRobinBatch 函数
break;
case DestBatchHybrid:
self->pub.sendBatch = printHybridBatch;
break;
case DestBatchHybrid:
self->pub.sendBatch = printHybridBatch; // 如果 dest 为 DestBatchHybrid设置 sendBatch 为 printHybridBatch 函数
break;
default:
Assert(false);
break;
}
self->pub.rStartup = printStreamStartup;
self->pub.rShutdown = printStreamShutdown;
self->pub.rDestroy = printtup_destroy;
self->pub.finalizeLocalStream = NULL;
self->pub.mydest = dest;
self->pub.tmpContext = NULL;
return (DestReceiver *)self;
default:
Assert(false); // 如果 dest 未匹配到任何已知的值,断言报错
break;
}
self->pub.rStartup = printStreamStartup; // 设置 rStartup 为 printStreamStartup 函数
self->pub.rShutdown = printStreamShutdown; // 设置 rShutdown 为 printStreamShutdown 函数
self->pub.rDestroy = printtup_destroy; // 设置 rDestroy 为 printtup_destroy 函数
self->pub.finalizeLocalStream = NULL; // 设置 finalizeLocalStream 为 NULL
self->pub.mydest = dest; // 设置 mydest 为 dest
self->pub.tmpContext = NULL; // 设置 tmpContext 为 NULL
return (DestReceiver *)self; // 返回 DestReceiver 类型的指针,指向 self
}
/*
* @Description: Flush data in the buffer
*
@ -568,16 +567,36 @@ static void SendRowDescriptionCols_3(StringInfo buf, TupleDesc typeinfo, List *t
/* Do we have a non-resjunk tlist item? */
while (tlist_item &&
#ifdef STREAMPLAN
StreamTopConsumerAmI() == false && StreamThreadAmI() == false &&
// 如果定义了 STREAMPLAN 宏,执行以下条件判断
StreamTopConsumerAmI() == false && StreamThreadAmI() == false &&
#endif
((TargetEntry *)lfirst(tlist_item))->resjunk)
tlist_item = lnext(tlist_item);
if (tlist_item != NULL) {
TargetEntry *tle = (TargetEntry *)lfirst(tlist_item);
// 判断当前是否为流式计划中的最终消费者,如果是则跳过下面的逻辑,否则继续
// 判断当前是否为流式计划中的线程,如果是则跳过下面的逻辑,否则继续
// 上述两个条件都满足,才会继续执行下面的逻辑
// 如果它们不满足,则不会执行下面的逻辑
// 注意:这是一段条件编译,具体情况依赖于是否定义了 STREAMPLAN 宏
// 检查当前目标项是否为 "resjunk"(通常表示结果列是否为垃圾列)
((TargetEntry *)lfirst(tlist_item))->resjunk)
// 如果目标项为垃圾列,则跳过当前项
tlist_item = lnext(tlist_item);
// 继续执行下面的逻辑,不再受到上面条件的限制
if (tlist_item != NULL) {
// 如果 tlist_item 不为空
TargetEntry *tle = (TargetEntry *)lfirst(tlist_item);
// 使用缓冲区 buf 写入 tle->resorigtbl 的值,通常是一个表的原始 ID
pq_writeint32(buf, tle->resorigtbl);
// 使用缓冲区 buf 写入 tle->resorigcol 的值,通常是一个列的原始 ID
pq_writeint16(buf, tle->resorigcol);
// 移动 tlist_item 到下一个目标项
tlist_item = lnext(tlist_item);
}
pq_writeint32(buf, tle->resorigtbl);
pq_writeint16(buf, tle->resorigcol);
tlist_item = lnext(tlist_item);
} else {
/* No info available, so send zeroes */
pq_writeint32(buf, 0);
@ -615,16 +634,17 @@ static void SendRowDescriptionCols_3(StringInfo buf, TupleDesc typeinfo, List *t
*/
static void SendRowDescriptionCols_2(StringInfo buf, TupleDesc typeinfo, List *targetlist, int16 *formats)
{
Form_pg_attribute *attrs = typeinfo->attrs;
int natts = typeinfo->natts;
Form_pg_attribute *attrs = typeinfo->attrs; // 从 TupleDesc 中获取属性数组
int natts = typeinfo->natts; // 获取属性数量
int i;
// 遍历属性列表
for (i = 0; i < natts; ++i) {
Oid atttypid = attrs[i]->atttypid;
int32 atttypmod = attrs[i]->atttypmod;
Oid atttypid = attrs[i]->atttypid; // 获取属性的数据类型 ID
int32 atttypmod = attrs[i]->atttypmod; // 获取属性的类型修饰符
writeString(buf, NameStr(attrs[i]->attname), false);
// 使用 writeString 函数将属性的名称写入缓冲区 buf第三个参数表示不添加引号
#ifdef PGXC
/*
* for analyze global stats, because DN will send sample rows to CN,
@ -722,22 +742,23 @@ static void printtup_prepare_info(DR_printtup *myState, TupleDesc typeinfo, int
int16 *formats = myState->portal != NULL ? myState->portal->formats : myState->formats;
int i;
/* get rid of any old data */
// 清除旧数据
if (myState->myinfo != NULL) {
pfree(myState->myinfo);
pfree(myState->myinfo); // 释放之前分配的内存
}
myState->myinfo = NULL;
myState->myinfo = NULL; // 将 myinfo 设置为 NULL
myState->attrinfo = typeinfo; // 设置 attrinfo 为给定的 TupleDesc
myState->nattrs = numAttrs; // 设置属性数量
myState->attrinfo = typeinfo;
myState->nattrs = numAttrs;
if (numAttrs <= 0) {
return;
return; // 如果属性数量小于等于0直接返回
}
/* get rid of any old data */
if (myState->portal != NULL && myState->portal->tupDesc != NULL) {
#ifdef USE_ASSERT_CHECKING
Assert(numAttrs <= myState->portal->tupDesc->natts);
Assert(numAttrs <= myState->portal->tupDesc->natts); // 使用断言检查属性数量是否小于等于 tupDesc 中的属性数量
#else
if (numAttrs > myState->portal->tupDesc->natts) {
ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
@ -746,12 +767,14 @@ static void printtup_prepare_info(DR_printtup *myState, TupleDesc typeinfo, int
#endif
}
// 分配并初始化 myinfo 数组
myState->myinfo = (PrinttupAttrInfo *)palloc0(numAttrs * sizeof(PrinttupAttrInfo));
for (i = 0; i < numAttrs; i++) {
PrinttupAttrInfo *thisState = myState->myinfo + i;
int16 format = (formats ? formats[i] : 0);
PrinttupAttrInfo *thisState = myState->myinfo + i; // 获取当前属性信息的指针
int16 format = (formats ? formats[i] : 0); // 获取格式信息,如果 formats 为空则默认为0
}
}
/*
* for analyze global stats, because DN will send sample rows to CN,
* if we encounter droped columns, we should send it to CN. but atttypid of dropped column
@ -763,13 +786,20 @@ static void printtup_prepare_info(DR_printtup *myState, TupleDesc typeinfo, int
thisState->format = format;
if (format == 0) {
getTypeOutputInfo(typeinfo->attrs[i]->atttypid, &thisState->typoutput, &thisState->typisvarlena);
fmgr_info(thisState->typoutput, &thisState->finfo);
} else if (format == 1) {
getTypeBinaryOutputInfo(typeinfo->attrs[i]->atttypid, &thisState->typsend, &thisState->typisvarlena);
fmgr_info(thisState->typsend, &thisState->finfo);
} else {
ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("unsupported format code: %d", format)));
// 如果格式为0表示文本格式
getTypeOutputInfo(typeinfo->attrs[i]->atttypid, &thisState->typoutput, &thisState->typisvarlena);
// 获取当前属性的输出函数及是否为可变长度数据类型
fmgr_info(thisState->typoutput, &thisState->finfo);
// 获取输出函数的信息
} else if (format == 1) {
// 如果格式为1表示二进制格式
getTypeBinaryOutputInfo(typeinfo->attrs[i]->atttypid, &thisState->typsend, &thisState->typisvarlena);
// 获取当前属性的二进制输出函数及是否为可变长度数据类型
fmgr_info(thisState->typsend, &thisState->finfo);
// 获取二进制输出函数的信息
} else {
// 如果格式不是0或1报错
ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("unsupported format code: %d", format)));
}
}
}
@ -782,48 +812,56 @@ static void printtup_prepare_info_for_stream(DR_printtup *myState, TupleDesc typ
{
int i;
/* get rid of any old data */
/* get rid of any old data */
if (myState->myinfo != NULL) {
pfree(myState->myinfo);
pfree(myState->myinfo); // 释放之前分配的内存
}
myState->myinfo = NULL;
myState->myinfo = NULL; // 将 myinfo 设置为 NULL
myState->attrinfo = typeinfo; // 设置 attrinfo 为给定的 TupleDesc
myState->nattrs = numAttrs; // 设置属性数量
myState->attrinfo = typeinfo;
myState->nattrs = numAttrs;
if (numAttrs <= 0) {
return;
return; // 如果属性数量小于等于0直接返回
}
// 分配并初始化 myinfo 数组
myState->myinfo = (PrinttupAttrInfo *)palloc0(numAttrs * sizeof(PrinttupAttrInfo));
/* let's say for stream thread only support format = 0; */
for (i = 0; i < numAttrs; i++) {
PrinttupAttrInfo *thisState = myState->myinfo + i;
thisState->format = 0;
PrinttupAttrInfo *thisState = myState->myinfo + i; // 获取当前属性信息的指针
thisState->format = 0; // 设置输出格式为0文本格式
getTypeOutputInfo(typeinfo->attrs[i]->atttypid, &thisState->typoutput, &thisState->typisvarlena);
// 获取当前属性的输出函数及是否为可变长度数据类型
fmgr_info(thisState->typoutput, &thisState->finfo);
// 获取输出函数的信息
}
}
inline MemoryContext changeToTmpContext(DestReceiver *self)
{
MemoryContext old_context = CurrentMemoryContext;
if (self->tmpContext != NULL) {
old_context = MemoryContextSwitchTo(self->tmpContext);
MemoryContext old_context = CurrentMemoryContext; // 保存当前内存上下文
if (self->tmpContext != NULL) { // 如果存在临时内存上下文
old_context = MemoryContextSwitchTo(self->tmpContext); // 切换到临时内存上下文
}
return old_context;
return old_context; // 返回切换前的内存上下文
}
void assembleStreamMessage(TupleTableSlot *slot, DestReceiver *self, StringInfo buf)
{
TupleDesc typeinfo = slot->tts_tupleDescriptor;
DR_printtup *myState = (DR_printtup *)self;
int natts = typeinfo->natts;
TupleDesc typeinfo = slot->tts_tupleDescriptor; // 获取槽中的元组描述符
DR_printtup *myState = (DR_printtup *)self; // 将 DestReceiver 转换为 DR_printtup 类型
int natts = typeinfo->natts; // 获取属性数量
int i;
StreamTimeSerilizeStart(t_thrd.pgxc_cxt.GlobalNetInstr);
if (slot->tts_dataRow) {
Assert(buf->len == 0);
StreamTimeSerilizeStart(t_thrd.pgxc_cxt.GlobalNetInstr); // 开始流式时间序列化
if (slot->tts_dataRow) { // 如果槽中包含数据行
Assert(buf->len == 0); // 使用断言确保缓冲区长度为0
// 通常,这个条件检查用于确保开始新的数据行时缓冲区为空
/*
* Prepare a DataRow message
@ -901,13 +939,13 @@ void assembleStreamMessage(TupleTableSlot *slot, DestReceiver *self, StringInfo
*/
void printtupStream(TupleTableSlot *slot, DestReceiver *self)
{
TupleDesc typeinfo = slot->tts_tupleDescriptor;
DR_printtup *myState = (DR_printtup *)self;
StringInfo buf = &myState->buf;
int natts = typeinfo->natts;
TupleDesc typeinfo = slot->tts_tupleDescriptor; // 获取槽中的元组描述符
DR_printtup *myState = (DR_printtup *)self; // 将 DestReceiver 转换为 DR_printtup 类型
StringInfo buf = &myState->buf; // 获取 DR_printtup 结构中的缓冲区
int natts = typeinfo->natts; // 获取属性数量
int i;
StreamTimeSerilizeStart(t_thrd.pgxc_cxt.GlobalNetInstr);
StreamTimeSerilizeStart(t_thrd.pgxc_cxt.GlobalNetInstr);// 开始流式时间序列化
#ifdef PGXC
/*
@ -915,13 +953,14 @@ void printtupStream(TupleTableSlot *slot, DestReceiver *self)
* values, just send over the DataRow message as we received it from the
* Datanode
*/
if (slot->tts_dataRow) {
pq_beginmessage_reuse(buf, 'D');
appendBinaryStringInfo(buf, slot->tts_dataRow, slot->tts_dataLen);
AddCheckInfo(buf);
pq_endmessage_reuse(buf);
StreamTimeSerilizeEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
return;
if (slot->tts_dataRow) { // 如果槽中包含数据行
pq_beginmessage_reuse(buf, 'D'); // 开始一个数据行消息
appendBinaryStringInfo(buf, slot->tts_dataRow, slot->tts_dataLen); // 将数据行追加到消息中
AddCheckInfo(buf); // 添加校验信息(这部分功能可能需要上下文)
pq_endmessage_reuse(buf); // 结束消息
StreamTimeSerilizeEnd(t_thrd.pgxc_cxt.GlobalNetInstr); // 结束流式时间序列化
return; // 返回
}
#endif
/* Set or update my derived attribute info, if needed */
@ -996,26 +1035,32 @@ void printtupStream(TupleTableSlot *slot, DestReceiver *self)
*/
void printBatch(VectorBatch *batch, DestReceiver *self)
{
DR_printtup *myState = (DR_printtup *)self;
StringInfo buf = &myState->buf;
pq_beginmessage_reuse(buf, 'B');
batch->SerializeWithLZ4Compress(buf);
AddCheckInfo(buf);
pq_endmessage_reuse(buf);
DR_printtup *myState = (DR_printtup *)self; // 将 DestReceiver 转换为 DR_printtup 类型
StringInfo buf = &myState->buf; // 获取 DR_printtup 结构中的缓冲区
pq_beginmessage_reuse(buf, 'B'); // 开始一个批处理消息,使用字符 'B'
batch->SerializeWithLZ4Compress(buf); // 将批处理数据进行 LZ4 压缩后追加到消息中
AddCheckInfo(buf); // 添加校验信息(这部分功能可能需要上下文)
pq_endmessage_reuse(buf); // 结束消息
}
/* ----------------
* printtup --- print a tuple in protocol 3.0
* ----------------
*/
void printtup(TupleTableSlot *slot, DestReceiver *self)
{
TupleDesc typeinfo = slot->tts_tupleDescriptor;
DR_printtup *myState = (DR_printtup *)self;
StringInfo buf = &myState->buf;
int natts = typeinfo->natts;
TupleDesc typeinfo = slot->tts_tupleDescriptor; // 获取槽中的元组描述符
DR_printtup *myState = (DR_printtup *)self; // 将 DestReceiver 转换为 DR_printtup 类型
StringInfo buf = &myState->buf; // 获取 DR_printtup 结构中的缓冲区
int natts = typeinfo->natts; // 获取属性数量
int i;
bool binary = false;
bool binary = false; // 初始化 binary 标志为 false表示输出为文本格式
/* just as we define in backend/commands/analyze.cpp */
#define WIDTH_THRESHOLD 1024
@ -1045,13 +1090,16 @@ void printtup(TupleTableSlot *slot, DestReceiver *self)
* Datanode
*/
if (slot->tts_dataRow != NULL && (pg_get_client_encoding() == GetDatabaseEncoding()) && !binary) {
pq_beginmessage_reuse(buf, 'D');
appendBinaryStringInfo(buf, slot->tts_dataRow, slot->tts_dataLen);
AddCheckInfo(buf);
pq_endmessage_reuse(buf);
StreamTimeSerilizeEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
return;
}
// 如果槽中包含数据行,客户端编码与数据库编码匹配,且输出格式为文本(非二进制)
pq_beginmessage_reuse(buf, 'D'); // 开始一个数据行消息,使用字符 'D'
appendBinaryStringInfo(buf, slot->tts_dataRow, slot->tts_dataLen); // 将数据行追加到消息中
AddCheckInfo(buf); // 添加校验信息(这部分功能可能需要上下文)
pq_endmessage_reuse(buf); // 结束消息
StreamTimeSerilizeEnd(t_thrd.pgxc_cxt.GlobalNetInstr); // 结束流式时间序列化
return; // 返回
}
#endif
/* Make sure the tuple is fully deconstructed */
@ -1104,59 +1152,63 @@ void printtup(TupleTableSlot *slot, DestReceiver *self)
char *outputstr = NULL;
outputstr = OutputFunctionCall(&thisState->finfo, attr);
if (thisState->typisvarlena && self->forAnalyzeSampleTuple &&
(typeinfo->attrs[i]->atttypid == BYTEAOID || typeinfo->attrs[i]->atttypid == CHAROID ||
typeinfo->attrs[i]->atttypid == TEXTOID || typeinfo->attrs[i]->atttypid == BLOBOID ||
typeinfo->attrs[i]->atttypid == CLOBOID || typeinfo->attrs[i]->atttypid == RAWOID ||
typeinfo->attrs[i]->atttypid == BPCHAROID || typeinfo->attrs[i]->atttypid == VARCHAROID ||
typeinfo->attrs[i]->atttypid == NVARCHAR2OID) &&
strlen(outputstr) > WIDTH_THRESHOLD * 2) {
if (thisState->typisvarlena && self->forAnalyzeSampleTuple &&
(typeinfo->attrs[i]->atttypid == BYTEAOID || typeinfo->attrs[i]->atttypid == CHAROID ||
typeinfo->attrs[i]->atttypid == TEXTOID || typeinfo->attrs[i]->atttypid == BLOBOID ||
typeinfo->attrs[i]->atttypid == CLOBOID || typeinfo->attrs[i]->atttypid == RAWOID ||
typeinfo->attrs[i]->atttypid == BPCHAROID || typeinfo->attrs[i]->atttypid == VARCHAROID ||
typeinfo->attrs[i]->atttypid == NVARCHAR2OID) &&
strlen(outputstr) > WIDTH_THRESHOLD * 2)
{
// 检查当前属性是否为可变长度数据类型,并且处于分析示例元组的模式中
// 同时检查当前属性的数据类型是否为指定的文本类型之一BYTEA、CHAR、TEXT、BLOB、CLOB、RAW、BPCHAR、VARCHAR、NVARCHAR2
// 最后,检查输出字符串的长度是否超过某个阈值的两倍
}
{
/*
* in compute_scalar_stats, we just skip detoast value if value size is
* bigger than WIDTH_THRESHOLD to avoid consuming too much memory
* during analysis, so we just send as WIDTH_THRESHOLD + 4 to cn so that
* it can use as little memory as we can to satisfy the threshold
*/
const int length = WIDTH_THRESHOLD + 4;
text *txt = NULL;
Datum str;
text *result = NULL;
const int length = WIDTH_THRESHOLD + 4; // 设置一个常量长度
text *txt = NULL; // 声明一个 text 类型指针,并初始化为 NULL
Datum str; // 声明一个 Datum 类型变量
text *result = NULL; // 声明一个 text 类型指针,并初始化为 NULL
txt = cstring_to_text(outputstr);
pfree(outputstr);
txt = cstring_to_text(outputstr); // 将字符串转换为 text 类型
pfree(outputstr); // 释放原始字符串的内存
str = DirectFunctionCall3(substrb_with_lenth, PointerGetDatum(txt), Int32GetDatum(0),
Int32GetDatum(length));
result = DatumGetTextP(str);
if (result != txt)
pfree(txt);
str = DirectFunctionCall3(substrb_with_lenth, PointerGetDatum(txt), Int32GetDatum(0), Int32GetDatum(length)); // 调用 substrb_with_lenth 函数
result = DatumGetTextP(str); // 将返回的 Datum 转换为 text
outputstr = TextDatumGetCString(str);
pfree(result);
if (result != txt)
pfree(txt); // 如果结果不等于原始的 text则释放原始 text 的内存
outputstr = TextDatumGetCString(str); // 将 text 转换为 C 字符串
pfree(result); // 释放结果 text 的内存
}
pq_sendcountedtext(buf, outputstr, strlen(outputstr), false);
pfree(outputstr);
} else {
/* Binary output */
bytea *outputbytes = NULL;
bytea *outputbytes = NULL; // 声明一个 bytea 类型指针,并初始化为 NULL
outputbytes = SendFunctionCall(&thisState->finfo, attr);
pq_sendint32(buf, VARSIZE(outputbytes) - VARHDRSZ);
pq_sendbytes(buf, VARDATA(outputbytes), VARSIZE(outputbytes) - VARHDRSZ);
pfree(outputbytes);
outputbytes = SendFunctionCall(&thisState->finfo, attr); // 调用 SendFunctionCall 函数
pq_sendint32(buf, VARSIZE(outputbytes) - VARHDRSZ); // 发送 bytea 数据的长度
pq_sendbytes(buf, VARDATA(outputbytes), VARSIZE(outputbytes) - VARHDRSZ); // 发送 bytea 数据内容
pfree(outputbytes); // 释放 bytea 数据的内存
}
/* Clean up detoasted copy, if any */
if (DatumGetPointer(attr) != DatumGetPointer(origattr))
pfree(DatumGetPointer(attr));
}
}
pfree(DatumGetPointer(attr)); // 释放去掉 TOAST 的拷贝的内存
(void)MemoryContextSwitchTo(old_context);
StreamTimeSerilizeEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
(void)MemoryContextSwitchTo(old_context); // 切换回之前的内存上下文
StreamTimeSerilizeEnd(t_thrd.pgxc_cxt.GlobalNetInstr); // 结束流式时间序列化
AddCheckInfo(buf);
pq_endmessage_reuse(buf);
AddCheckInfo(buf); // 添加校验信息(这部分功能可能需要上下文)
pq_endmessage_reuse(buf); // 结束消息
}
/* ----------------
@ -1246,15 +1298,16 @@ static void printtup_20(TupleTableSlot *slot, DestReceiver *self)
*/
static void printtup_shutdown(DestReceiver *self)
{
DR_printtup *myState = (DR_printtup *)self;
DR_printtup *myState = (DR_printtup *)self; // 将 DestReceiver 转换为 DR_printtup 类型
if (myState->myinfo != NULL)
pfree(myState->myinfo);
myState->myinfo = NULL;
if (myState->myinfo != NULL) // 如果 myinfo 不为 NULL
pfree(myState->myinfo); // 释放 myinfo 的内存
myState->myinfo = NULL; // 将 myinfo 设置为 NULL
myState->attrinfo = NULL;
myState->attrinfo = NULL; // 将 attrinfo 设置为 NULL
}
/* ----------------
* printtup_destroy
* ----------------
@ -1300,37 +1353,41 @@ void debugStartup(DestReceiver *self, int operation, TupleDesc typeinfo)
*/
void debugtup(TupleTableSlot *slot, DestReceiver *self)
{
TupleDesc typeinfo = slot->tts_tupleDescriptor;
int natts = typeinfo->natts;
TupleDesc typeinfo = slot->tts_tupleDescriptor; // 获取槽中的元组描述符
int natts = typeinfo->natts; // 获取属性数量
int i;
Datum origattr, attr;
char *value = NULL;
bool isnull = false;
char *value = NULL; // 初始化属性值字符串为 NULL
bool isnull = false; // 初始化是否为 NULL 值的标志为 false
Oid typoutput;
bool typisvarlena = false;
for (i = 0; i < natts; ++i) {
origattr = tableam_tslot_getattr(slot, i + 1, &isnull);
origattr = tableam_tslot_getattr(slot, i + 1, &isnull); // 获取属性的原始值和是否为 NULL
if (isnull) {
continue;
continue; // 如果属性值为 NULL跳过此次循环
}
getTypeOutputInfo(typeinfo->attrs[i]->atttypid, &typoutput, &typisvarlena);
getTypeOutputInfo(typeinfo->attrs[i]->atttypid, &typoutput, &typisvarlena); // 获取属性的输出信息
/*
* If we have a toasted datum, forcibly detoast it here to avoid
* memory leakage inside the type's output routine.
*/
if (typisvarlena) {
attr = PointerGetDatum(PG_DETOAST_DATUM(origattr));
} else {
attr = origattr;
}
if (typisvarlena) {
// 如果属性类型为可变长度VARLENA解压缩属性值
attr = PointerGetDatum(PG_DETOAST_DATUM(origattr));
} else {
// 否则,使用原始属性值
attr = origattr;
}
value = OidOutputFunctionCall(typoutput, attr);
value = OidOutputFunctionCall(typoutput, attr); // 将属性值转换为字符串形式
printatt((unsigned)i + 1, typeinfo->attrs[i], value);
printatt((unsigned)i + 1, typeinfo->attrs[i], value); // 打印属性信息
pfree(value); // 释放属性值字符串的内存
pfree(value);
/* Clean up detoasted copy, if any */
if (DatumGetPointer(attr) != DatumGetPointer(origattr)) {
@ -1436,43 +1493,46 @@ static void printtup_internal_20(TupleTableSlot *slot, DestReceiver *self)
*/
void assembleStreamBatchMessage(BatchCompressType ctype, VectorBatch *batch, StringInfo buf)
{
buf->cursor = 'B';
buf->cursor = 'B'; // 设置消息缓冲区的标记为 'B'
switch (ctype) {
case BCT_NOCOMP:
// 如果压缩类型为 BCT_NOCOMP使用无压缩方式序列化向量批次数据
batch->SerializeWithoutCompress(buf);
break;
case BCT_LZ4:
// 如果压缩类型为 BCT_LZ4使用 LZ4 压缩方式序列化向量批次数据
batch->SerializeWithLZ4Compress(buf);
break;
default:
// 如果压缩类型不被识别,报告错误
ereport(ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("unrecognized batch compress type")));
}
}
inline void AddCheckInfo(StringInfo buf)
{
StringInfoData buf_check;
bool is_check_added = false;
StringInfoData buf_check; // 声明一个 StringInfoData 结构体
bool is_check_added = false; // 初始化一个标志变量为 false
/* add check info for datanode and coordinator */
if (IsConnFromCoord()) {
if (IsConnFromCoord()) { // 检查是否来自协调节点的连接
#ifdef USE_ASSERT_CHECKING
initStringInfo(&buf_check);
AddCheckMessage(&buf_check, buf, false);
is_check_added = true;
initStringInfo(&buf_check); // 初始化 buf_check 字符串缓冲区
AddCheckMessage(&buf_check, buf, false); // 添加校验信息到 buf_check
is_check_added = true; // 设置校验信息已添加的标志为 true
#else
if (anls_opt_is_on(ANLS_STREAM_DATA_CHECK)) {
initStringInfo(&buf_check);
AddCheckMessage(&buf_check, buf, false);
is_check_added = true;
initStringInfo(&buf_check); // 初始化 buf_check 字符串缓冲区
AddCheckMessage(&buf_check, buf, false); // 添加校验信息到 buf_check
is_check_added = true; // 设置校验信息已添加的标志为 true
}
#endif
if (unlikely(is_check_added)) {
pfree(buf->data);
buf->len = buf_check.len;
buf->maxlen = buf_check.maxlen;
buf->data = buf_check.data;
pfree(buf->data); // 释放原始缓冲区的数据内存
buf->len = buf_check.len; // 更新缓冲区的长度
buf->maxlen = buf_check.maxlen; // 更新缓冲区的最大长度
buf->data = buf_check.data; // 更新缓冲区的数据指针
}
}
}

File diff suppressed because it is too large Load Diff

View File

@ -73,40 +73,38 @@ void InitBufferPool(void)
bool found_buf_ckpt = false;
uint64 buffer_size;
// 初始化一个存储buffer描述符的共享内存区域
t_thrd.storage_cxt.BufferDescriptors = (BufferDescPadded *)CACHELINEALIGN(
ShmemInitStruct("Buffer Descriptors",
TOTAL_BUFFER_NUM * sizeof(BufferDescPadded) + PG_CACHE_LINE_SIZE,
&found_descs));
/* Init candidate buffer list and candidate buffer free map */
// 初始化候选buffer列表和候选buffer的空闲映射
candidate_buf_init();
#ifdef __aarch64__
// 针对 ARM 架构计算buffer大小并初始化buffer区块的共享内存区域
buffer_size = TOTAL_BUFFER_NUM * (Size)BLCKSZ + PG_CACHE_LINE_SIZE;
t_thrd.storage_cxt.BufferBlocks =
(char *)CACHELINEALIGN(ShmemInitStruct("Buffer Blocks", buffer_size, &found_bufs));
#else
// 针对其他架构计算buffer大小并初始化buffer区块的共享内存区域
buffer_size = TOTAL_BUFFER_NUM * (Size)BLCKSZ;
t_thrd.storage_cxt.BufferBlocks = (char *)ShmemInitStruct("Buffer Blocks", buffer_size, &found_bufs);
#endif
if (BBOX_BLACKLIST_SHARE_BUFFER) {
/* Segment Buffer is exclued from the black list, as it contains many critical information for debug */
// 如果设置了将共享buffer段添加到黑名单中执行该操作
bbox_blacklist_add(SHARED_BUFFER, t_thrd.storage_cxt.BufferBlocks, NORMAL_SHARED_BUFFER_NUM * (Size)BLCKSZ);
}
/*
* The array used to sort to-be-checkpointed buffer ids is located in
* shared memory, to avoid having to allocate significant amounts of
* memory at runtime. As that'd be in the middle of a checkpoint, or when
* the checkpointer is restarted, memory allocation failures would be
* painful.
*/
// 初始化一个用于排序即将被检查点checkpoint操作处理的buffer ID的数组
g_instance.ckpt_cxt_ctl->CkptBufferIds =
(CkptSortItem *)ShmemInitStruct("Checkpoint BufferIds",
TOTAL_BUFFER_NUM * sizeof(CkptSortItem), &found_buf_ckpt);
if (ENABLE_INCRE_CKPT && g_instance.ckpt_cxt_ctl->dirty_page_queue == NULL) {
// 如果启用了增量检查点,并且脏页队列为空,则初始化该队列
g_instance.ckpt_cxt_ctl->dirty_page_queue_size = TOTAL_BUFFER_NUM *
PAGE_QUEUE_SLOT_MULTI_NBUFFERS;
MemoryContext oldcontext = MemoryContextSwitchTo(g_instance.increCheckPoint_context);
@ -115,30 +113,29 @@ void InitBufferPool(void)
g_instance.ckpt_cxt_ctl->dirty_page_queue =
(DirtyPageQueueSlot *)palloc_huge(CurrentMemoryContext, queue_mem_size);
/* The memory of the memset sometimes exceeds 2 GB. so, memset_s cannot be used. */
MemSet((char*)g_instance.ckpt_cxt_ctl->dirty_page_queue, 0, queue_mem_size);
(void)MemoryContextSwitchTo(oldcontext);
}
if (g_instance.bgwriter_cxt.unlink_rel_hashtbl == NULL) {
// 如果未链接关系哈希表为空,则创建它
g_instance.bgwriter_cxt.unlink_rel_hashtbl = relfilenode_hashtbl_create("unlink_rel_hashtbl", true);
}
if (g_instance.bgwriter_cxt.unlink_rel_fork_hashtbl == NULL) {
// 如果未链接关系分支哈希表为空,则创建它
g_instance.bgwriter_cxt.unlink_rel_fork_hashtbl =
relfilenode_fork_hashtbl_create("unlink_rel_one_fork_hashtbl", true);
}
if (found_descs || found_bufs || found_buf_ckpt) {
/* both should be present or neither */
// 确保三者都存在或都不存在
Assert(found_descs && found_bufs && found_buf_ckpt);
/* note: this path is only taken in EXEC_BACKEND case */
// 注意:只有在 EXEC_BACKEND 情况下才走这条路径
} else {
int i;
/*
* Initialize all the buffer headers.
*/
// 初始化所有buffer头部信息
for (i = 0; i < TOTAL_BUFFER_NUM; i++) {
BufferDesc *buf = GetBufferDescriptor(i);
CLEAR_BUFFERTAG(buf->tag);
@ -153,23 +150,25 @@ void InitBufferPool(void)
buf->dirty_queue_loc = PG_UINT64_MAX;
buf->encrypt = false;
}
// 分配解锁关系表的锁
g_instance.bgwriter_cxt.rel_hashtbl_lock = LWLockAssign(LWTRANCHE_UNLINK_REL_TBL);
g_instance.bgwriter_cxt.rel_one_fork_hashtbl_lock = LWLockAssign(LWTRANCHE_UNLINK_REL_FORK_TBL);
}
/* Init other shared buffer-management stuff */
// 初始化其他的共享buffer管理相关的内容
StrategyInitialize(!found_descs);
/* Init Vector Buffer management stuff */
// 初始化向量Buffer管理相关的内容
DataCacheMgr::NewSingletonInstance();
/* Init Meta data cache management stuff */
// 初始化元数据缓存管理相关的内容
MetaCacheMgr::NewSingletonInstance();
/* Initialize per-backend file flush context */
// 初始化每个后端文件刷新上下文
WritebackContextInit(t_thrd.storage_cxt.BackendWritebackContext, &u_sess->attr.attr_common.backend_flush_after);
}
/*
* BufferShmemSize
*
@ -178,29 +177,38 @@ void InitBufferPool(void)
*/
Size BufferShmemSize(void)
{
Size size = 0;
Size size = 0; // 初始化一个大小为0的变量用于存储总共需要的共享内存大小
/* size of buffer descriptors */
size = add_size(size, mul_size(TOTAL_BUFFER_NUM, sizeof(BufferDescPadded)));
size = add_size(size, PG_CACHE_LINE_SIZE);
size = add_size(size, mul_size(TOTAL_BUFFER_NUM, sizeof(BufferDescPadded)));
// 计算缓冲区描述符的大小TOTAL_BUFFER_NUM是缓冲区的总数sizeof(BufferDescPadded)是每个描述符的大小
size = add_size(size, PG_CACHE_LINE_SIZE);
// 添加一个缓存行的大小(通常用于内存对齐)
/* size of data pages */
size = add_size(size, mul_size(TOTAL_BUFFER_NUM, BLCKSZ));
size = add_size(size, mul_size(TOTAL_BUFFER_NUM, BLCKSZ));
// 计算数据页的大小TOTAL_BUFFER_NUM是数据页的总数BLCKSZ是每个数据页的大小
#ifdef __aarch64__
size = add_size(size, PG_CACHE_LINE_SIZE);
size = add_size(size, PG_CACHE_LINE_SIZE);
// 如果是在aarch64架构下再添加一个缓存行的大小
#endif
/* size of stuff controlled by freelist.c */
size = add_size(size, StrategyShmemSize());
size = add_size(size, StrategyShmemSize());
// 计算由freelist.c控制的内容的大小
/* size of checkpoint sort array in bufmgr.c */
size = add_size(size, mul_size(TOTAL_BUFFER_NUM, sizeof(CkptSortItem)));
size = add_size(size, mul_size(TOTAL_BUFFER_NUM, sizeof(CkptSortItem)));
// 计算bufmgr.c中的检查点排序数组的大小
/* size of candidate buffers */
size = add_size(size, mul_size(TOTAL_BUFFER_NUM, sizeof(Buffer)));
size = add_size(size, mul_size(TOTAL_BUFFER_NUM, sizeof(Buffer)));
// 计算候选缓冲区的大小
/* size of candidate free map */
size = add_size(size, mul_size(TOTAL_BUFFER_NUM, sizeof(bool)));
size = add_size(size, mul_size(TOTAL_BUFFER_NUM, sizeof(bool)));
// 计算候选空闲映射的大小
return size;
return size; // 返回计算得到的总共需要的共享内存大小
}

View File

@ -88,17 +88,20 @@ uint32 BufTableHashCode(BufferTag *tagPtr)
*/
int BufTableLookup(BufferTag *tag, uint32 hashcode)
{
BufferLookupEnt *result = NULL;
BufferLookupEnt *result = NULL; // 初始化一个指向BufferLookupEnt结构的指针用于存储查找结果
result = (BufferLookupEnt *)buf_hash_operate<HASH_FIND>(t_thrd.storage_cxt.SharedBufHash, tag, hashcode, NULL);
// 调用buf_hash_operate函数进行查找操作t_thrd.storage_cxt.SharedBufHash是缓冲区的哈希表tag是要查找的缓冲区标签hashcode是哈希码
if (SECUREC_UNLIKELY(result == NULL)) {
// 如果查找结果为空(未找到对应的缓冲区),则返回-1表示未找到
return -1;
}
return result->id;
return result->id; // 如果找到了对应的缓冲区返回该缓冲区的id
}
/*
* BufTableInsert
* Insert a hashtable entry for given tag and buffer ID,
@ -111,23 +114,26 @@ int BufTableLookup(BufferTag *tag, uint32 hashcode)
*/
int BufTableInsert(BufferTag *tag, uint32 hashcode, int buf_id)
{
BufferLookupEnt *result = NULL;
bool found = false;
BufferLookupEnt *result = NULL; // 初始化一个指向BufferLookupEnt结构的指针用于存储查找结果
bool found = false; // 初始化一个布尔变量,用于标识是否找到了相应的缓冲区
Assert(buf_id >= 0); /* -1 is reserved for not-in-table */
Assert(tag->blockNum != P_NEW); /* invalid tag */
Assert(buf_id >= 0); // 确保buf_id不小于0因为-1通常用于表示不在表中的情况
Assert(tag->blockNum != P_NEW); // 确保标签中的块号不是P_NEW因为这是无效的标签
result = (BufferLookupEnt *)buf_hash_operate<HASH_ENTER>(t_thrd.storage_cxt.SharedBufHash, tag, hashcode, &found);
// 调用buf_hash_operate函数进行插入操作t_thrd.storage_cxt.SharedBufHash是缓冲区的哈希表tag是要插入的缓冲区标签hashcode是哈希码&found是一个标志位用于表示是否找到相应的缓冲区
if (found) { /* found something already in the table */
// 如果在表中找到了相应的缓冲区则返回该缓冲区的id
return result->id;
}
result->id = buf_id;
result->id = buf_id; // 如果没有找到相应的缓冲区则将缓冲区的id设置为传入的buf_id
return -1;
return -1; // 返回-1表示插入成功
}
/*
* BufTableDelete
* Delete the hashtable entry for given tag (which must exist)
@ -136,11 +142,13 @@ int BufTableInsert(BufferTag *tag, uint32 hashcode, int buf_id)
*/
void BufTableDelete(BufferTag *tag, uint32 hashcode)
{
BufferLookupEnt *result = NULL;
BufferLookupEnt *result = NULL; // 初始化一个指向BufferLookupEnt结构的指针用于存储查找结果
result = (BufferLookupEnt *)buf_hash_operate<HASH_REMOVE>(t_thrd.storage_cxt.SharedBufHash, tag, hashcode, NULL);
// 调用buf_hash_operate函数进行删除操作t_thrd.storage_cxt.SharedBufHash是缓冲区的哈希表tag是要删除的缓冲区标签hashcode是哈希码NULL表示不需要返回结果
if (result == NULL) { /* shouldn't happen */
// 如果删除操作未找到相应的缓冲区,通常表示哈希表已经损坏或发生了异常情况,应该抛出错误
ereport(ERROR, (errcode(ERRCODE_DATA_CORRUPTED), (errmsg("shared buffer hash table corrupted."))));
}
}

View File

@ -137,33 +137,40 @@ static Buffer ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumb
*
* Only works for shared buffers.
*/
//GetPrivateRefCountEntryFast 函数用于快速获取给定缓冲区buffer的私有引用计数条目Private Refcount Entry的指针。
//私有引用计数主要用于跟踪某个后端(即数据库的一个客户端连接进程)对某个缓冲区的引用计数。
static PrivateRefCountEntry* GetPrivateRefCountEntryFast(Buffer buffer, PrivateRefCountEntry* &free_entry)
{
PrivateRefCountEntry* res = NULL;
PrivateRefCountEntry* res = NULL; // 用于存储找到的引用计数条目的指针
int i;
// 确保输入的buffer是有效的并且不是本地的
Assert(BufferIsValid(buffer));
Assert(!BufferIsLocal(buffer));
/*
* First search for references in the array, that'll be sufficient in the
* majority of cases.
*
*/
for (i = 0; i < REFCOUNT_ARRAY_ENTRIES; i++) {
// 获取当前遍历到的引用计数条目的指针
res = &t_thrd.storage_cxt.PrivateRefCountArray[i];
// 如果找到了与输入buffer匹配的条目则返回它的指针
if (res->buffer == buffer) {
return res;
}
/* Remember where to put a new refcount, should it become necessary. */
/* 如果尚未找到空闲的引用计数条目且当前条目是无效的,则记下它的位置,可能稍后需要在这里放置一个新的引用计数 */
if (free_entry == NULL && res->buffer == InvalidBuffer) {
free_entry = res;
}
}
// 如果在数组中没有找到引用计数条目则返回NULL
return NULL;
}
/*
* Return the PrivateRefCount entry for the passed buffer.
*
@ -185,20 +192,18 @@ static PrivateRefCountEntry* GetPrivateRefCountEntryFast(Buffer buffer, PrivateR
static PrivateRefCountEntry* GetPrivateRefCountEntrySlow(Buffer buffer,
bool create, bool do_move, PrivateRefCountEntry* free_entry)
{
Assert(!create || do_move);
Assert(BufferIsValid(buffer));
Assert(!BufferIsLocal(buffer));
Assert(!create || do_move); // 确保如果指定创建新条目,则必须进行移动
Assert(BufferIsValid(buffer)); // 确保缓冲区是有效的
Assert(!BufferIsLocal(buffer)); // 确保缓冲区不是本地的
/*
* By here we know that the buffer, if already pinned, isn't residing in
* the array.
* pinned
*/
PrivateRefCountEntry* res = NULL;
bool found = false;
/*
* Look up the buffer in the hashtable if we've previously overflowed into
* it.
*
*/
if (t_thrd.storage_cxt.PrivateRefCountOverflowed > 0) {
res = (PrivateRefCountEntry *)hash_search(t_thrd.storage_cxt.PrivateRefCountHash, (void *)&buffer, HASH_FIND,
@ -207,34 +212,33 @@ static PrivateRefCountEntry* GetPrivateRefCountEntrySlow(Buffer buffer,
if (!found) {
if (!create) {
/* Neither array nor hash have an entry and no new entry is needed */
// 数组和哈希表都没有条目,并且不需要新条目
return NULL;
} else if (free_entry != NULL) {
/* add entry into the free array slot */
// 将条目添加到空闲的数组插槽中
free_entry->buffer = buffer;
free_entry->refcount = 0;
return free_entry;
} else {
/*
* Move entry from the current clock position in the array into the
* hashtable. Use that slot.
* 使
*/
PrivateRefCountEntry *array_ent = NULL;
PrivateRefCountEntry *hash_ent = NULL;
/* select victim slot */
// 选择受害者插槽
array_ent = &t_thrd.storage_cxt
.PrivateRefCountArray[t_thrd.storage_cxt.PrivateRefCountClock++ % REFCOUNT_ARRAY_ENTRIES];
Assert(array_ent->buffer != InvalidBuffer);
/* enter victim array entry into hashtable */
// 将受害者数组条目输入到哈希表中
hash_ent = (PrivateRefCountEntry *)hash_search(t_thrd.storage_cxt.PrivateRefCountHash,
(void *)&array_ent->buffer, HASH_ENTER, &found);
Assert(!found);
hash_ent->refcount = array_ent->refcount;
/* fill the now free array slot */
// 填充现在空闲的数组插槽
array_ent->buffer = buffer;
array_ent->refcount = 0;
@ -246,14 +250,14 @@ static PrivateRefCountEntry* GetPrivateRefCountEntrySlow(Buffer buffer,
if (!do_move) {
return res;
} else if (found && free_entry != NULL) {
/* move buffer from hashtable into the free array slot
/* 从哈希表中移动缓冲区到空闲的数组插槽
*
* fill array slot
*
*/
free_entry->buffer = buffer;
free_entry->refcount = res->refcount;
/* delete from hashtable */
// 从哈希表中删除
(void)hash_search(t_thrd.storage_cxt.PrivateRefCountHash, (void *)&buffer, HASH_REMOVE, &found);
Assert(found);
Assert(t_thrd.storage_cxt.PrivateRefCountOverflowed > 0);
@ -262,32 +266,31 @@ static PrivateRefCountEntry* GetPrivateRefCountEntrySlow(Buffer buffer,
return free_entry;
} else {
/*
* Swap the entry in the hash table with the one in the array at the
* current clock position.
*
*/
PrivateRefCountEntry *array_ent = NULL;
PrivateRefCountEntry *hash_ent = NULL;
/* select victim slot */
// 选择受害者插槽
array_ent = &t_thrd.storage_cxt
.PrivateRefCountArray[t_thrd.storage_cxt.PrivateRefCountClock++ % REFCOUNT_ARRAY_ENTRIES];
Assert(array_ent->buffer != InvalidBuffer);
/* enter victim entry into the hashtable */
// 将受害者条目输入到哈希表中
hash_ent = (PrivateRefCountEntry *)hash_search(t_thrd.storage_cxt.PrivateRefCountHash,
(void *)&array_ent->buffer, HASH_ENTER, &found);
Assert(!found);
hash_ent->refcount = array_ent->refcount;
/* fill now free array entry with previously searched entry */
// 用之前搜索到的条目填充现在空闲的数组条目
array_ent->buffer = res->buffer;
array_ent->refcount = res->refcount;
/* and remove the old entry */
// 并删除旧条目
(void)hash_search(t_thrd.storage_cxt.PrivateRefCountHash, (void *)&array_ent->buffer, HASH_REMOVE, &found);
Assert(found);
/* PrivateRefCountOverflowed stays the same -1 + +1 = 0 */
// PrivateRefCountOverflowed保持不变 -1 + +1 = 0
return array_ent;
}
}
@ -295,64 +298,96 @@ static PrivateRefCountEntry* GetPrivateRefCountEntrySlow(Buffer buffer,
return NULL;
}
/* A combination of GetPrivateRefCountEntryFast & GetPrivateRefCountEntrySlow. */
//这个函数的主要作用是获取缓冲区的私有引用计数项。它首先尝试从快速路径GetPrivateRefCountEntryFast获取引用计数项
//如果失败则进入慢速路径GetPrivateRefCountEntrySlow来获取。
//这通常用于管理缓冲区的引用计数,以确保正确地释放和管理缓冲区的使用。
PrivateRefCountEntry *GetPrivateRefCountEntry(Buffer buffer, bool create, bool do_move)
{
PrivateRefCountEntry *free_entry = NULL;
PrivateRefCountEntry *ref = NULL;
PrivateRefCountEntry *free_entry = NULL; // 初始化一个指向PrivateRefCountEntry的指针用于存储可用的空闲项
PrivateRefCountEntry *ref = NULL; // 初始化一个指向PrivateRefCountEntry的指针用于存储最终的引用计数项
// 尝试从快速路径获取引用计数项
ref = GetPrivateRefCountEntryFast(buffer, free_entry);
// 如果快速路径获取失败,进入慢速路径
if (ref == NULL) {
ref = GetPrivateRefCountEntrySlow(buffer, create, do_move, free_entry);
}
return ref;
return ref; // 返回获取到的引用计数项
}
/*
* Returns how many times the passed buffer is pinned by this backend.
*
* Only works for shared memory buffers!
*/
//这个函数的主要作用是获取缓冲区的私有引用计数。
//首先,它会验证缓冲区的有效性和是否为本地缓冲区。
//然后它尝试从快速路径或慢速路径获取引用计数项如果无法获取则返回0表示引用计数为0。
//如果成功获取引用计数项,它将返回引用计数项中的引用计数值。
//私有引用计数通常用于跟踪缓冲区的引用情况,以确保正确释放和管理缓冲区的使用。
static int32 GetPrivateRefCount(Buffer buffer)
{
PrivateRefCountEntry *ref = NULL;
PrivateRefCountEntry *ref = NULL; // 初始化一个指向PrivateRefCountEntry的指针用于存储引用计数项
Assert(BufferIsValid(buffer));
Assert(!BufferIsLocal(buffer));
Assert(BufferIsValid(buffer)); // 断言缓冲区有效性
Assert(!BufferIsLocal(buffer)); // 断言缓冲区不是本地缓冲区
PrivateRefCountEntry *free_entry = NULL;
ref = GetPrivateRefCountEntryFast(buffer, free_entry);
PrivateRefCountEntry *free_entry = NULL; // 初始化一个指向PrivateRefCountEntry的指针用于存储可用的空闲项
ref = GetPrivateRefCountEntryFast(buffer, free_entry); // 尝试从快速路径获取引用计数项
// 如果快速路径获取失败,进入慢速路径
if (ref == NULL) {
ref = GetPrivateRefCountEntrySlow(buffer, false, false, free_entry);
// 第一个false表示不要创建新的引用计数项第二个false表示不要移动引用计数项
}
if (ref == NULL) {
return 0;
return 0; // 如果无法获取引用计数项返回0表示引用计数为0
}
return ref->refcount;
return ref->refcount; // 返回引用计数项中的引用计数值
}
/*
* Release resources used to track the reference count of a buffer which we no
* longer have pinned and don't want to pin again immediately.
*/
//这个函数的主要作用是忘记清除私有引用计数项通常是在引用计数为0时进行。根据引用计数项的存储位置它有两种不同的处理方式
//如果引用计数项在PrivateRefCountArray数组内部它将引用计数项的缓冲区字段设置为InvalidBuffer表示忘记了该引用计数项。
//如果引用计数项不在PrivateRefCountArray数组内部它将从PrivateRefCountHash哈希表中删除引用计数项的记录并更新溢出计数器以表示删除了一个溢出的引用计数项。
//这个函数通常用于管理私有引用计数项的生命周期,以确保正确地释放和管理内存。
void ForgetPrivateRefCountEntry(PrivateRefCountEntry *ref)
{
Assert(ref->refcount == 0);
Assert(ref->refcount == 0); // 断言引用计数项的引用计数为0
if (ref >= &t_thrd.storage_cxt.PrivateRefCountArray[0] &&
ref < &t_thrd.storage_cxt.PrivateRefCountArray[REFCOUNT_ARRAY_ENTRIES]) {
ref->buffer = InvalidBuffer;
// 如果引用计数项在PrivateRefCountArray数组内部
ref->buffer = InvalidBuffer; // 将引用计数项的缓冲区字段设置为InvalidBuffer表示忘记引用计数项
} else {
// 如果引用计数项不在PrivateRefCountArray数组内部
bool found = false;
Buffer buffer = ref->buffer;
(void)hash_search(t_thrd.storage_cxt.PrivateRefCountHash, (void *)&buffer, HASH_REMOVE, &found);
Assert(found);
// 从PrivateRefCountHash哈希表中删除引用计数项的记录
Assert(found); // 确保找到了要删除的引用计数项
Assert(t_thrd.storage_cxt.PrivateRefCountOverflowed > 0);
t_thrd.storage_cxt.PrivateRefCountOverflowed--;
// 更新溢出计数器,表示删除了一个溢出的引用计数项
}
}
static void BufferSync(int flags);
static uint32 WaitBufHdrUnlocked(BufferDesc* buf);
static void WaitIO(BufferDesc* buf);
@ -383,59 +418,60 @@ static bool ConditionalStartBufferIO(BufferDesc* buf, bool forInput);
* block will not be delayed by the I/O. Prefetching is optional.
* No-op if prefetching isn't compiled in.
*/
//这段代码首先检查是否已定义了USE_PREFETCH和USE_POSIX_FADVISE
//如果定义了这两个宏,才会执行预取操作。然后,它会验证关系的有效性和块号的有效性,
//打开关系的存储管理器smgr并在需要的情况下执行预取操作。
void PrefetchBuffer(Relation reln, ForkNumber forkNum, BlockNumber blockNum)
{
#if defined(USE_PREFETCH) && defined(USE_POSIX_FADVISE)
Assert(RelationIsValid(reln));
Assert(BlockNumberIsValid(blockNum));
Assert(RelationIsValid(reln)); // 断言关系有效
Assert(BlockNumberIsValid(blockNum)); // 断言块号有效
/* Open it at the smgr level if not already done */
/* 在smgr级别打开关系如果尚未打开 */
RelationOpenSmgr(reln);
if (RelationUsesLocalBuffers(reln)) {
/* see comments in ReadBufferExtended */
/* 请参阅ReadBufferExtended中的注释 */
if (RELATION_IS_OTHER_TEMP(reln)) {
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot access temporary tables of other sessions")));
}
}
BufferTag new_tag; /* identity of requested block */
uint32 new_hash; /* hash value for newTag */
LWLock *new_partition_lock; /* buffer partition lock for it */
BufferTag new_tag; /* 请求块的标识 */
uint32 new_hash; /* newTag的哈希值 */
LWLock *new_partition_lock; /* 用于新标签的缓冲区分区锁 */
int buf_id;
/* create a tag so we can lookup the buffer */
/* 创建一个标签以便我们可以查找缓冲区 */
INIT_BUFFERTAG(new_tag, reln->rd_smgr->smgr_rnode.node, forkNum, blockNum);
/* determine its hash code and partition lock ID */
/* 确定其哈希代码和分区锁ID */
new_hash = BufTableHashCode(&new_tag);
new_partition_lock = BufMappingPartitionLock(new_hash);
/* see if the block is in the buffer pool already */
/* 查看块是否已经在缓冲池中 */
(void)LWLockAcquire(new_partition_lock, LW_SHARED);
buf_id = BufTableLookup(&new_tag, new_hash);
LWLockRelease(new_partition_lock);
/* If not in buffers, initiate prefetch */
/* 如果不在缓冲区中,则启动预取 */
if (buf_id < 0) {
smgrprefetch(reln->rd_smgr, forkNum, blockNum);
}
/*
* If the block *is* in buffers, we do nothing. This is not really
* ideal: the block might be just about to be evicted, which would be
* stupid since we know we are going to need it soon. But the only
* easy answer is to bump the usage_count, which does not seem like a
* great solution: when the caller does ultimately touch the block,
* usage_count would get bumped again, resulting in too much
* favoritism for blocks that are involved in a prefetch sequence. A
* real fix would involve some additional per-buffer state, and it's
* not clear that there's enough of a problem to justify that.
*
*
* usage_count
* usage_count会再次增加
*
*
*/
#endif /* USE_PREFETCH && USE_POSIX_FADVISE */
}
/*
* @Description: ConditionalStartBufferIO: conditionally begin and Asynchronous Prefetch or
* WriteBack I/O on this buffer.
@ -459,22 +495,21 @@ void PrefetchBuffer(Relation reln, ForkNumber forkNum, BlockNumber blockNum)
* @Return: true -- lock sucess; false-- lock failed
* @See also:
*/
static bool ConditionalStartBufferIO(BufferDesc *buf, bool for_input)
static bool ConditionalStartBufferIO(BufferDesc *buf, bool for_input)//这个函数的目的是协调并控制缓冲区的I/O操作以避免多个线程同时对同一个缓冲区进行读取或写入操作。
{
uint32 buf_state;
/*
* Grab the io_in_progress lock so that other processes can wait for
* me to finish the I/O. If we cannot acquire the lock it means
* I/O we want to do is in progress on this buffer.
* io_in_progress_lock锁便I/O操作
* I/O操作已经在该缓冲区上进行中
*/
if (LWLockConditionalAcquire(buf->io_in_progress_lock, LW_EXCLUSIVE) == true) {
/* Got the lock */
/* 获取了锁 */
buf_state = LockBufHdr(buf);
/*
* If BM_IO_IN_PROGRESS and the lock isn't held,
* it means the i/o was in progress and an error occured,
* and some other thread should take care of this.
* BM_IO_IN_PROGRESS标志被设置
* I/O操作正在进行中线
*
*/
if (buf_state & BM_IO_IN_PROGRESS) {
UnlockBufHdr(buf, buf_state);
@ -483,20 +518,20 @@ static bool ConditionalStartBufferIO(BufferDesc *buf, bool for_input)
}
} else {
/*
* Could not get the lock...
* Another thread is currently attempting to read
* or write this buffer. We don't need to do our I/O.
* ...
* 线
* I/O操作
*/
return false;
}
/*
* At this point, there is no I/O active on this buffer
* We are holding the BufHdr lock and the io_in_progress_lock.
* I/O操作
* BufHdr锁和io_in_progress_lock锁
*/
buf_state = pg_atomic_read_u32(&buf->state);
if (for_input ? (buf_state & BM_VALID) : !(buf_state & BM_DIRTY)) {
/* Another thread already did the I/O */
/* 另一个线程已经完成了I/O操作 */
UnlockBufHdr(buf, buf_state);
LWLockRelease(buf->io_in_progress_lock);
return false;
@ -507,12 +542,13 @@ static bool ConditionalStartBufferIO(BufferDesc *buf, bool for_input)
UnlockBufHdr(buf, buf_state);
/*
* Return holding the io_in_progress_lock,
* The caller is expected to perform the actual I/O.
* io_in_progress_lock锁
* I/O操作
*/
return true;
}
/*
* @Description: PageListBufferAlloc--
*
@ -788,36 +824,46 @@ static volatile BufferDesc *PageListBufferAlloc(SMgrRelation smgr, char relpersi
* @Param[IN] reln: relation
* @See also:
*/
void PageRangePrefetch(Relation reln, ForkNumber fork_num, BlockNumber block_num, int32 n, uint32 flags = 0,
uint32 col = 0)
//这个函数的目的是预取一定范围内的多个块,它采取以下步骤:
//分配一个块号列表block_list用于存储要预取的块号。这个列表只在函数的上下文中需要因此使用palloc动态分配内存来存储块号列表。
//使用循环填充块号列表从block_num开始预取n个连续的块。这些块号被添加到block_list中。
//调用PageListPrefetch函数来处理块号列表该函数将根据列表中的块号执行实际的预取操作。
//最后,释放分配的块号列表内存,以避免内存泄漏。
//这个函数通常用于优化数据访问性能,以减少等待时间,并提高从磁盘读取数据的效率。
void PageRangePrefetch(Relation reln, ForkNumber fork_num, BlockNumber block_num, int32 n, uint32 flags = 0, uint32 col = 0)
{
BlockNumber *block_list = NULL;
BlockNumber *block_ptr = NULL;
BlockNumber *block_list = NULL; // 用于存储块号列表的指针
BlockNumber *block_ptr = NULL; // 指向块号列表的当前位置
BlockNumber block, end;
/*
* Allocate the block list. It is only required
* within the context of this function.
*
*/
block_list = (BlockNumber *)palloc(sizeof(BlockNumber) * n);
/*
* Fill the blockList and call PageListPrefetch to process it.
* PageListPrefetch来处理它
*/
for (block = block_num, end = block_num + n, block_ptr = block_list; block < end; block++) {
*(block_ptr++) = block;
*(block_ptr++) = block; // 将块号添加到块列表中
}
/*
* Call PageListPrefetch to process the list
* PageListPrefetch来处理列表
*/
PageListPrefetch(reln, fork_num, block_list, n, flags, col);
pfree(block_list);
pfree(block_list); // 释放分配的块列表内存
return;
}
/*
* @Description: PageListPrefetch
* The dispatch list of AioDispatchDesc_t structures is released

View File

@ -81,17 +81,24 @@ void PageListBackWrite(uint32* bufList, int32 n,
int32* bufs_reusable = NULL); /* opt reusable count returned */
static BufferDesc* get_buf_from_candidate_list(BufferAccessStrategy strategy, uint32* buf_state);
static void perform_delay(StrategyDelayStatus *status)
static void perform_delay(StrategyDelayStatus *status)//这个函数的目的是在满足一定条件时执行延迟操作。条件包括已经重试的次数超过最大次数MAX_RETRY_TIMES并且脏页的数量超过了一个阈值g_instance.attr.attr_storage.NBuffers * NEED_DELAY_RETRY_GET_BUF
{
if (++(status->retry_times) > MAX_RETRY_TIMES &&
get_dirty_page_num() > g_instance.attr.attr_storage.NBuffers * NEED_DELAY_RETRY_GET_BUF) {
// 如果已经重试了最大<E69C80><E5A4A7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>并且脏页数量超过了阈值
if (status->cur_delay_time == 0) {
// 如果当前延迟时间为0则初始化为最小延迟时间
status->cur_delay_time = MIN_DELAY_RETRY;
}
// 通过睡眠来延迟操作
pg_usleep(status->cur_delay_time);
/* increase delay by a random fraction between 1X and 2X */
/* 增加延迟时间增加的时间在1X和2X之间的随机分数 */
status->cur_delay_time += (int)(status->cur_delay_time * ((double)random() / (double)MAX_RANDOM_VALUE) + 0.5);
// 如果当前延迟时间超过最大延迟时间,则重新初始化为最小延迟时间
if (status->cur_delay_time > MAX_DELAY_RETRY) {
status->cur_delay_time = MIN_DELAY_RETRY;
}
@ -100,33 +107,38 @@ static void perform_delay(StrategyDelayStatus *status)
}
/*
* ClockSweepTick - Helper routine for StrategyGetBuffer()
*
* Move the clock hand one buffer ahead of its current position and return the
* id of the buffer now under the hand.
*/
//这个函数的目的是根据时钟扫描算法选择要回收的缓冲区。
//函数首先通过原子操作增加一个称为"victim"的计数器,
//该计数器用于标识下一个要回收的缓冲区。
//如果"victim"的值超过了可用的缓冲区数量,就需要对其进行包装,
//以确保在BufferDescriptors中查找时不会越界。
static inline uint32 ClockSweepTick(int max_nbuffer_can_use)
{
uint32 victim;
/*
* Atomically move hand ahead one buffer - if there's several processes
* doing this, this can lead to buffers being returned slightly out of
* apparent order.
* -
*
*/
victim = pg_atomic_fetch_add_u32(&t_thrd.storage_cxt.StrategyControl->nextVictimBuffer, 1);
if (victim >= (uint32)max_nbuffer_can_use) {
uint32 original_victim = victim;
/* always wrap what we look up in BufferDescriptors */
/* 始终对BufferDescriptors中查找的内容进行包装 */
victim = victim % max_nbuffer_can_use;
/*
* If we're the one that just caused a wraparound, force
* completePasses to be incremented while holding the spinlock. We
* need the spinlock so StrategySyncStart() can return a consistent
* value consisting of nextVictimBuffer and completePasses.
*
* completePasses便StrategySyncStart()
* nextVictimBuffer和completePasses组成的一致值
*/
if (victim == 0) {
uint32 expected;
@ -137,12 +149,10 @@ static inline uint32 ClockSweepTick(int max_nbuffer_can_use)
while (!success) {
/*
* Acquire the spinlock while increasing completePasses. That
* allows other readers to read nextVictimBuffer and
* completePasses in a consistent manner which is required for
* StrategySyncStart(). In theory delaying the increment
* could lead to a overflow of nextVictimBuffers, but that's
* highly unlikely and wouldn't be particularly harmful.
* completePasses时获取自旋锁
* nextVictimBuffer和completePassesStrategySyncStart()
* nextVictimBuffers溢出
*
*/
SpinLockAcquire(&t_thrd.storage_cxt.StrategyControl->buffer_strategy_lock);
@ -159,6 +169,7 @@ static inline uint32 ClockSweepTick(int max_nbuffer_can_use)
return victim;
}
/*
* StrategyGetBuffer
*
@ -177,20 +188,30 @@ static inline uint32 ClockSweepTick(int max_nbuffer_can_use)
* If the fraction is too small, we will increase dynamiclly to avoid elog(ERROR)
* in `Startup' process because of ERROR will promote to FATAL.
*/
/*
使使
使
*/
BufferDesc* StrategyGetBuffer(BufferAccessStrategy strategy, uint32* buf_state)
{
BufferDesc *buf = NULL;
int bgwproc_no;
int try_counter;
uint32 local_buf_state = 0; /* to avoid repeated (de-)referencing */
uint32 local_buf_state = 0; /* 用于避免重复引用 */
int max_buffer_can_use;
bool am_standby = RecoveryInProgress();
StrategyDelayStatus retry_lock_status = { 0, 0 };
StrategyDelayStatus retry_buf_status = { 0, 0 };
/*
* If given a strategy object, see whether it can select a buffer. We
* assume strategy objects don't need buffer_strategy_lock.
* 使
* buffer_strategy_lock
*/
if (strategy != NULL) {
buf = GetBufferFromRing(strategy, buf_state);
@ -200,38 +221,28 @@ BufferDesc* StrategyGetBuffer(BufferAccessStrategy strategy, uint32* buf_state)
}
/*
* If asked, we need to waken the bgwriter. Since we don't want to rely on
* a spinlock for this we force a read from shared memory once, and then
* set the latch based on that value. We need to go through that length
* because otherwise bgprocno might be reset while/after we check because
* the compiler might just reread from memory.
*
* This can possibly set the latch of the wrong process if the bgwriter
* dies in the wrong moment. But since PGPROC->procLatch is never
* deallocated the worst consequence of that is that we set the latch of
* some arbitrary process.
* bgwriter
*/
bgwproc_no = INT_ACCESS_ONCE(t_thrd.storage_cxt.StrategyControl->bgwprocno);
if (bgwproc_no != -1) {
/* reset bgwprocno first, before setting the latch */
/* 先重置bgwprocno然后设置latch */
t_thrd.storage_cxt.StrategyControl->bgwprocno = -1;
/*
* Not acquiring ProcArrayLock here which is slightly icky. It's
* actually fine because procLatch isn't ever freed, so we just can
* potentially set the wrong process' (or no process') latch.
* ProcArrayLock
* procLatch永远不会被释放latch设置为错误的进程
* latch
*/
SetLatch(&g_instance.proc_base_all_procs[bgwproc_no]->procLatch);
}
/*
* We count buffer allocation requests so that the bgwriter can estimate
* the rate of buffer consumption. Note that buffers recycled by a
* strategy object are intentionally not counted here.
* 便bgwriter使
*
*/
(void)pg_atomic_fetch_add_u32(&t_thrd.storage_cxt.StrategyControl->numBufferAllocs, 1);
/* Check the Candidate list */
/* 检查候选列表 */
if (ENABLE_INCRE_CKPT && pg_atomic_read_u32(&g_instance.ckpt_cxt_ctl->current_page_writer_count) > 1) {
if (NEED_CONSIDER_USECOUNT) {
const uint32 MAX_RETRY_SCAN_CANDIDATE_LISTS = 5;
@ -259,7 +270,7 @@ BufferDesc* StrategyGetBuffer(BufferAccessStrategy strategy, uint32* buf_state)
}
retry:
/* Nothing on the freelist, so run the "clock sweep" algorithm */
/* 在自由列表上没有可用的缓冲区,因此运行“时钟扫描”算法 */
if (am_standby)
max_buffer_can_use = int(NORMAL_SHARED_BUFFER_NUM * u_sess->attr.attr_storage.shared_buffers_fraction);
else
@ -269,7 +280,7 @@ retry:
for (;;) {
buf = GetBufferDescriptor(ClockSweepTick(max_buffer_can_use));
/*
* If the buffer is pinned, we cannot use it.
* pinned使
*/
if (!retryLockBufHdr(buf, &local_buf_state)) {
if (--try_get_loc_times == 0) {
@ -284,7 +295,7 @@ retry:
retry_lock_status.retry_times = 0;
if (BUF_STATE_GET_REFCOUNT(local_buf_state) == 0 && !(local_buf_state & BM_IS_META) &&
(backend_can_flush_dirty_page() || !(local_buf_state & BM_DIRTY))) {
/* Found a usable buffer */
/* 找到可用的缓冲区 */
if (strategy != NULL)
AddBufferToRing(strategy, buf);
*buf_state = local_buf_state;
@ -292,11 +303,9 @@ retry:
return buf;
} else if (--try_counter == 0) {
/*
* We've scanned all the buffers without making any state changes,
* so all the buffers are pinned (or were when we looked at them).
* We could hope that someone will free one eventually, but it's
* probably better to fail than to risk getting stuck in an
* infinite loop.
*
*
*
*/
UnlockBufHdr(buf, local_buf_state);
@ -322,10 +331,11 @@ retry:
perform_delay(&retry_buf_status);
}
/* not reached */
/* 不会执行到这里 */
return NULL;
}
/*
* StrategySyncStart -- tell BufferSync where to start syncing
*
@ -337,31 +347,51 @@ retry:
* allocs if non-NULL pointers are passed. The alloc count is reset after
* being read.
*/
/*
*/
int StrategySyncStart(uint32 *complete_passes, uint32 *num_buf_alloc)
{
uint32 next_victim_buffer;
int result;
/* 获取缓冲区策略控制锁 */
SpinLockAcquire(&t_thrd.storage_cxt.StrategyControl->buffer_strategy_lock);
/* 获取下一个待回收的缓冲区的编号 */
next_victim_buffer = pg_atomic_read_u32(&t_thrd.storage_cxt.StrategyControl->nextVictimBuffer);
/* 计算结果,用于返回 */
result = next_victim_buffer % TOTAL_BUFFER_NUM;
/* 获取完成的扫描轮数如果complete_passes不为NULL */
if (complete_passes != NULL) {
*complete_passes = t_thrd.storage_cxt.StrategyControl->completePasses;
/*
* Additionally add the number of wraparounds that happened before
* completePasses could be incremented. C.f. ClockSweepTick().
* completePasses被增加之前发生的循环次数
* ClockSweepTick()
*/
*complete_passes += next_victim_buffer / (unsigned int) NORMAL_SHARED_BUFFER_NUM;
*complete_passes += next_victim_buffer / (unsigned int)NORMAL_SHARED_BUFFER_NUM;
}
/* 获取缓冲区分配请求的计数如果num_buf_alloc不为NULL */
if (num_buf_alloc != NULL) {
*num_buf_alloc = pg_atomic_exchange_u32(&t_thrd.storage_cxt.StrategyControl->numBufferAllocs, 0);
}
/* 释放缓冲区策略控制锁 */
SpinLockRelease(&t_thrd.storage_cxt.StrategyControl->buffer_strategy_lock);
/* 返回结果 */
return result;
}
/*
* StrategyNotifyBgWriter -- set or clear allocation notification latch
*
@ -370,15 +400,25 @@ int StrategySyncStart(uint32 *complete_passes, uint32 *num_buf_alloc)
* happens. This feature is used by the bgwriter process to wake itself up
* from hibernation, and is not meant for anybody else to use.
*/
/*
bgwriter进程的编号
便
*/
void StrategyNotifyBgWriter(int bgwproc_no)
{
/*
* We acquire the BufFreelistLock just to ensure that the store appears
* atomic to StrategyGetBuffer. The bgwriter should call this rather
* infrequently, so there's no performance penalty from being safe.
* BufFreelistLock仅仅是为了确保存储看起来是原子的
* StrategyGetBuffer来说bgwriter应该相对不频繁地调用这个函数
*
*/
SpinLockAcquire(&t_thrd.storage_cxt.StrategyControl->buffer_strategy_lock);
/* 设置bgwriter进程的编号通知它有关缓冲区管理策略的信息 */
t_thrd.storage_cxt.StrategyControl->bgwprocno = bgwproc_no;
/* 释放BufFreelistLock */
SpinLockRelease(&t_thrd.storage_cxt.StrategyControl->buffer_strategy_lock);
}
@ -390,19 +430,30 @@ void StrategyNotifyBgWriter(int bgwproc_no)
* Note: for somewhat historical reasons, the buffer lookup hashtable size
* is also determined here.
*/
Size StrategyShmemSize(void)
/*
BufTableShmemSize
TOTAL_BUFFER_NUM
NUM_BUFFER_PARTITIONS
使 MAXALIGN
便
*/
Size StrategyShmemSize(void)
{
Size size = 0;
/* size of lookup hash table ... see comment in StrategyInitialize */
/* 计算查找哈希表的共享内存大小,详情见 StrategyInitialize 函数的注释 */
size = add_size(size, BufTableShmemSize(TOTAL_BUFFER_NUM + NUM_BUFFER_PARTITIONS));
/* size of the shared replacement strategy control block */
/* 计算共享替换策略控制块的大小,需要对齐到 MAXALIGN 大小 */
size = add_size(size, MAXALIGN(sizeof(BufferStrategyControl)));
return size;
}
}
/*
* StrategyInitialize -- initialize the buffer cache replacement
* strategy.
@ -410,49 +461,56 @@ Size StrategyShmemSize(void)
* Assumes: All of the buffers are already built into a linked list.
* Only called by postmaster and only during initialization.
*/
/*
InitBufTable
TOTAL_BUFFER_NUM + NUM_BUFFER_PARTITIONS
ShmemInitStruct t_thrd.storage_cxt.StrategyControl
PostgreSQL
*/
void StrategyInitialize(bool init)
{
bool found = false;
/*
* Initialize the shared buffer lookup hashtable.
*
*
* Since we can't tolerate running out of lookup table entries, we must be
* sure to specify an adequate table size here. The maximum steady-state
* usage is of course NBuffers entries, but BufferAlloc() tries to insert
* a new entry before deleting the old. In principle this could be
* happening in each partition concurrently, so we could need as many as
* NBuffers + NUM_BUFFER_PARTITIONS entries.
* 使
* NBuffers BufferAlloc()
* NBuffers + NUM_BUFFER_PARTITIONS
*/
InitBufTable(TOTAL_BUFFER_NUM + NUM_BUFFER_PARTITIONS);
/*
* Get or create the shared strategy control block
*
*/
t_thrd.storage_cxt.StrategyControl =
(BufferStrategyControl *)ShmemInitStruct("Buffer Strategy Status", sizeof(BufferStrategyControl), &found);
if (!found) {
/*
* Only done once, usually in postmaster
* postmaster
*/
Assert(init);
SpinLockInit(&t_thrd.storage_cxt.StrategyControl->buffer_strategy_lock);
/* Initialize the clock sweep pointer */
/* 初始化时钟扫描指针 */
pg_atomic_init_u32(&t_thrd.storage_cxt.StrategyControl->nextVictimBuffer, 0);
/* Clear statistics */
/* 清空统计信息 */
t_thrd.storage_cxt.StrategyControl->completePasses = 0;
pg_atomic_init_u32(&t_thrd.storage_cxt.StrategyControl->numBufferAllocs, 0);
/* No pending notification */
/* 没有挂起的通知 */
t_thrd.storage_cxt.StrategyControl->bgwprocno = -1;
} else {
Assert(!init);
}
}
const int MIN_REPAIR_FILE_SLOT_NUM = 32;
/* ----------------------------------------------------------------
* Backend-private buffer ring management
@ -463,20 +521,26 @@ const int MIN_REPAIR_FILE_SLOT_NUM = 32;
*
* The object is allocated in the current memory context.
*/
/*
访 btype ring_size
BufferAccessStrategy
访
访访
*/
BufferAccessStrategy GetAccessStrategy(BufferAccessStrategyType btype)
{
BufferAccessStrategy strategy;
int ring_size;
/*
* Select ring size to use. See buffer/README for rationales.
* 使buffer/README中的原理说明
*
* Note: if you change the ring size for BAS_BULKREAD, see also
* SYNC_SCAN_REPORT_INTERVAL in access/heap/syncscan.c.
* BAS_BULKREAD的环大小access/heap/syncscan.c中的SYNC_SCAN_REPORT_INTERVAL
*/
switch (btype) {
case BAS_NORMAL:
/* if someone asks for NORMAL, just give 'em a "default" object */
/* 如果有人要求NORMAL只需给他们一个“默认”对象 */
return NULL;
case BAS_BULKREAD:
@ -495,22 +559,22 @@ BufferAccessStrategy GetAccessStrategy(BufferAccessStrategyType btype)
default:
ereport(ERROR, (errcode(ERRCODE_INVALID_OPERATION),
(errmsg("unrecognized buffer access strategy: %d", (int)btype))));
return NULL; /* keep compiler quiet */
return NULL; /* 保持编译器安静 */
}
/* If the shared buffers is too small, make sure ring size not equal zero. */
/* 如果共享缓冲区太小,请确保环大小不等于零。 */
ring_size = Max(ring_size, 4);
/* Make sure ring isn't an undue fraction of shared buffers */
/* 确保环不是共享缓冲区的过大比例 */
if (btype != BAS_BULKWRITE && btype != BAS_BULKREAD)
ring_size = Min(g_instance.attr.attr_storage.NBuffers / 8, ring_size);
else
ring_size = Min(g_instance.attr.attr_storage.NBuffers / 4, ring_size);
/* Allocate the object and initialize all elements to zeroes */
/* 分配对象并将所有元素初始化为零 */
strategy = (BufferAccessStrategy)palloc0(offsetof(BufferAccessStrategyData, buffers) + ring_size * sizeof(Buffer));
/* Set fields that don't start out zero */
/* 设置初始不为零的字段 */
strategy->btype = btype;
strategy->ring_size = ring_size;
strategy->flush_rate = Min(u_sess->attr.attr_storage.backwrite_quantity, ring_size);
@ -518,6 +582,7 @@ BufferAccessStrategy GetAccessStrategy(BufferAccessStrategyType btype)
return strategy;
}
/*
* FreeAccessStrategy -- release a BufferAccessStrategy object
*
@ -526,13 +591,14 @@ BufferAccessStrategy GetAccessStrategy(BufferAccessStrategyType btype)
*/
void FreeAccessStrategy(BufferAccessStrategy strategy)
{
/* don't crash if called on a "default" strategy */
/* 不要在“默认”策略上调用时崩溃 */
if (strategy != NULL) {
pfree(strategy);
strategy = NULL;
pfree(strategy); // 释放策略对象占用的内存
strategy = NULL; // 将策略对象指针设置为 NULL以避免引用已释放的内存
}
}
const int MAX_RETRY_RING_TIMES = 100;
const float MAX_RETRY_RING_PCT = 0.1;
/*
@ -541,15 +607,21 @@ const float MAX_RETRY_RING_PCT = 0.1;
*
* The bufhdr spin lock is held on the returned buffer.
*/
/*
使
*/
static BufferDesc *GetBufferFromRing(BufferAccessStrategy strategy, uint32 *buf_state)
{
BufferDesc *buf = NULL;
Buffer buf_num;
uint32 local_buf_state; /* to avoid repeated (de-)referencing */
uint16 retry_times = 0;
BufferDesc *buf = NULL; // 用于存储缓冲区描述符的指针
Buffer buf_num; // 用于存储缓冲区编号的变量
uint32 local_buf_state; // 用于存储缓冲区状态的变量,以避免重复引用
uint16 retry_times = 0; // 用于记录重试次数的变量
RETRY:
/* Advance to next ring slot */
/* 移动到下一个环形槽位 */
if (++strategy->current >= strategy->ring_size)
strategy->current = 0;
retry_times++;
@ -557,9 +629,9 @@ RETRY:
ADIO_RUN()
{
/*
* Flush out buffers asynchronously from behind the current slot.
* This is a kludge because the PageListBackWrite() is not strictly
* asynchronous and this function really shouldn't be doing the actual I/O.
*
* PageListBackWrite()
* I/O
*/
if (AioCompltrIsReady() &&
((strategy->btype == BAS_BULKWRITE) && (strategy->current % strategy->flush_rate == 0))) {
@ -585,9 +657,8 @@ RETRY:
ADIO_END();
/*
* If the slot hasn't been filled yet, tell the caller to allocate a new
* buffer with the normal allocation strategy. He will then fill this
* slot by calling AddBufferToRing with the new buffer.
* 使
* AddBufferToRing
*/
buf_num = strategy->buffers[strategy->current];
if (buf_num == InvalidBuffer) {
@ -596,13 +667,10 @@ RETRY:
}
/*
* If the buffer is pinned we cannot use it under any circumstances.
* 使
*
* If usage_count is 0 or 1 then the buffer is fair game (we expect 1,
* since our own previous usage of the ring element would have left it
* there, but it might've been decremented by clock sweep since then). A
* higher usage_count indicates someone else has touched the buffer, so we
* shouldn't re-use it.
* usage_count 0 1使 1使
* usage_count 访
*/
buf = GetBufferDescriptor(buf_num - 1);
if (pg_atomic_read_u32(&buf->state) & (BM_DIRTY | BM_IS_META)) {
@ -626,13 +694,14 @@ RETRY:
UnlockBufHdr(buf, local_buf_state);
/*
* Tell caller to allocate a new buffer with the normal allocation
* strategy. He'll then replace this ring element via AddBufferToRing.
* 使
* AddBufferToRing
*/
strategy->current_was_in_ring = false;
return NULL;
}
/*
* AddBufferToRing -- add a buffer to the buffer ring
*
@ -655,105 +724,156 @@ static void AddBufferToRing(BufferAccessStrategy strategy, volatile BufferDesc *
* Returns true if buffer manager should ask for a new victim, and false
* if this buffer should be written and re-used.
*/
/*
true
false
*/
bool StrategyRejectBuffer(BufferAccessStrategy strategy, BufferDesc *buf)
{
/* We only do this in bulkread mode */
/* 只在批量读取模式下执行此操作 */
if (strategy->btype != BAS_BULKREAD)
return false;
return false; // 如果不是批量读取模式,则不进行拒绝操作
/* Don't muck with behavior of normal buffer-replacement strategy */
/* 不要改变正常缓冲区替换策略的行为 */
if (!strategy->current_was_in_ring || strategy->buffers[strategy->current] != BufferDescriptorGetBuffer(buf))
return false;
return false; // 如果当前槽位不在环中,或者环中的缓冲区与给定的缓冲区描述符不匹配,则不进行拒绝操作
/*
* Remove the dirty buffer from the ring; necessary to prevent infinite
* loop if all ring members are dirty.
*
*/
strategy->buffers[strategy->current] = InvalidBuffer;
return true;
return true; // 返回true表示已经拒绝了缓冲区
}
/*
访访
quantity trigger
*/
void StrategyGetRingPrefetchQuantityAndTrigger(BufferAccessStrategy strategy, int *quantity, int *trigger)
{
int threshold;
int prefetch_trigger = u_sess->attr.attr_storage.prefetch_quantity;
int threshold; // 阈值,用于确定预取数量和触发值
int prefetch_trigger = u_sess->attr.attr_storage.prefetch_quantity; // 预取触发值,从全局配置获取
// 如果策略为空或者不是批量读取模式,直接返回
if (strategy == NULL || strategy->btype != BAS_BULKREAD) {
return;
}
// 计算阈值为环形缓冲区大小的四分之一
threshold = strategy->ring_size / 4;
// 如果 quantity 不为空,则将其设置为预取数量或阈值中的较小值
if (quantity != NULL) {
*quantity = (threshold > u_sess->attr.attr_storage.prefetch_quantity)
? u_sess->attr.attr_storage.prefetch_quantity
: threshold;
}
// 如果 trigger 不为空,则将其设置为预取触发值或阈值中的较小值
if (trigger != NULL) {
*trigger = (threshold > prefetch_trigger) ? prefetch_trigger : threshold;
}
}
}
/*
线便
线线
线线
线
线
*/
void wakeup_pagewriter_thread()
{
PageWriterProc *pgwr = &g_instance.ckpt_cxt_ctl->pgwr_procs.writer_proc[0];
/* The current candidate list is empty, wake up the buffer writer. */
/* 如果当前的候选列表为空,唤醒页写入线程。 */
if (pgwr->proc != NULL) {
SetLatch(&pgwr->proc->procLatch);
SetLatch(&pgwr->proc->procLatch); // 设置页写入线程的进程标志位,以唤醒线程
}
return;
}
const int CANDIDATE_DIRTY_LIST_LEN = 100;
const float HIGH_WATER = 0.75;
/*
线 NULL
使
*/
static BufferDesc* get_buf_from_candidate_list(BufferAccessStrategy strategy, uint32* buf_state)
{
BufferDesc* buf = NULL;
uint32 local_buf_state;
int buf_id = 0;
int list_num = g_instance.ckpt_cxt_ctl->pgwr_procs.sub_num;
int list_id = 0;
volatile PgBackendStatus* beentry = t_thrd.shemem_ptr_cxt.MyBEEntry;
Buffer *candidate_dirty_list = NULL;
int dirty_list_num = 0;
bool enable_available = false;
bool need_push_dirst_list = false;
BufferDesc* buf = NULL; // 用于存储缓冲区描述符的指针
uint32 local_buf_state; // 用于存储缓冲区状态的变量
int buf_id = 0; // 缓冲区的编号
int list_num = g_instance.ckpt_cxt_ctl->pgwr_procs.sub_num; // 获取子线程数量
int list_id = 0; // 子线程的ID
volatile PgBackendStatus* beentry = t_thrd.shemem_ptr_cxt.MyBEEntry; // 获取当前后台进程的状态
Buffer *candidate_dirty_list = NULL; // 用于存储候选的脏缓冲区列表
int dirty_list_num = 0; // 脏缓冲区列表的数量
bool enable_available = false; // 缓冲区是否可用的标志
bool need_push_dirty_list = false; // 是否需要将缓冲区添加到脏缓冲区列表的标志
bool need_scan_dirty =
(g_instance.ckpt_cxt_ctl->actual_dirty_page_num / (float)(g_instance.attr.attr_storage.NBuffers) > HIGH_WATER)
&& backend_can_flush_dirty_page();
&& backend_can_flush_dirty_page(); // 是否需要扫描脏缓冲区的标志
// 如果需要扫描脏缓冲区,分配一个候选脏缓冲区列表
if (need_scan_dirty) {
/*Not return the dirty page when there are few dirty pages */
candidate_dirty_list = (Buffer*)palloc0(sizeof(Buffer) * CANDIDATE_DIRTY_LIST_LEN);
}
// 根据当前进程的线程号或会话号计算子线程的ID
list_id = beentry->st_tid > 0 ? (beentry->st_tid % list_num) : (beentry->st_sessionid % list_num);
// 遍历所有子线程的候选缓冲区列表
for (int i = 0; i < list_num; i++) {
/* the pagewriter sub thread store normal buffer pool, sub thread starts from 1 */
// 计算子线程的实际ID从1开始
int thread_id = (list_id + i) % list_num + 1;
Assert(thread_id > 0 && thread_id <= list_num);
// 循环弹出候选缓冲区,直到列表为空
while (candidate_buf_pop(&buf_id, thread_id)) {
Assert(buf_id < SegmentBufferStartID);
buf = GetBufferDescriptor(buf_id);
local_buf_state = LockBufHdr(buf);
// 检查缓冲区是否可用
if (g_instance.ckpt_cxt_ctl->candidate_free_map[buf_id]) {
g_instance.ckpt_cxt_ctl->candidate_free_map[buf_id] = false;
enable_available = BUF_STATE_GET_REFCOUNT(local_buf_state) == 0 && !(local_buf_state & BM_IS_META);
need_push_dirst_list = need_scan_dirty && dirty_list_num < CANDIDATE_DIRTY_LIST_LEN &&
need_push_dirty_list = need_scan_dirty && dirty_list_num < CANDIDATE_DIRTY_LIST_LEN &&
free_space_enough(buf_id);
if (enable_available) {
// 如果需要考虑 usage count减少 usage count
if (NEED_CONSIDER_USECOUNT && BUF_STATE_GET_USAGECOUNT(local_buf_state) != 0) {
local_buf_state -= BUF_USAGECOUNT_ONE;
} else if (!(local_buf_state & BM_DIRTY)) {
}
// 如果缓冲区不是脏的,将其添加到环形缓冲区
else if (!(local_buf_state & BM_DIRTY)) {
if (strategy != NULL) {
AddBufferToRing(strategy, buf);
}
*buf_state = local_buf_state;
// 释放候选脏缓冲区列表的内存并返回找到的缓冲区描述符
if (candidate_dirty_list != NULL) {
pfree(candidate_dirty_list);
}
return buf;
} else if (need_push_dirst_list) {
}
// 如果需要将缓冲区添加到脏缓冲区列表
else if (need_push_dirty_list) {
candidate_dirty_list[dirty_list_num++] = buf_id;
}
}
@ -762,8 +882,10 @@ static BufferDesc* get_buf_from_candidate_list(BufferAccessStrategy strategy, ui
}
}
// 唤醒页写入线程以处理脏缓冲区
wakeup_pagewriter_thread();
// 遍历候选脏缓冲区列表
if (need_scan_dirty) {
for (int i = 0; i < dirty_list_num; i++) {
buf_id = candidate_dirty_list[i];
@ -771,11 +893,14 @@ static BufferDesc* get_buf_from_candidate_list(BufferAccessStrategy strategy, ui
local_buf_state = LockBufHdr(buf);
enable_available = (BUF_STATE_GET_REFCOUNT(local_buf_state) == 0) && !(local_buf_state & BM_IS_META)
&& free_space_enough(buf_id);
if (enable_available) {
// 如果需要将缓冲区添加到环形缓冲区
if (strategy != NULL) {
AddBufferToRing(strategy, buf);
}
*buf_state = local_buf_state;
// 释放候选脏缓冲区列表的内存并返回找到的缓冲区描述符
pfree(candidate_dirty_list);
return buf;
}
@ -783,9 +908,11 @@ static BufferDesc* get_buf_from_candidate_list(BufferAccessStrategy strategy, ui
}
}
// 释放候选脏缓冲区列表的内存
if (candidate_dirty_list != NULL) {
pfree(candidate_dirty_list);
}
// 如果未找到可用缓冲区则返回NULL
return NULL;
}

View File

@ -47,77 +47,115 @@ static Block GetLocalBufferStorage(void);
* Do PrefetchBuffer's work for temporary relations.
* No-op if prefetching isn't compiled in.
*/
/*
访
访
*/
void LocalPrefetchBuffer(SMgrRelation smgr, ForkNumber forkNum, BlockNumber blockNum)
{
#ifdef USE_PREFETCH
BufferTag new_tag; /* identity of requested block */
BufferTag new_tag; /* 请求块的标识 */
LocalBufferLookupEnt *hresult = NULL;
// 初始化请求块的标识
INIT_BUFFERTAG(new_tag, smgr->smgr_rnode.node, forkNum, blockNum);
/* Initialize local buffers if first request in this session */
// 如果在当前会话中首次请求本地缓冲区,初始化本地缓冲区
if (u_sess->storage_cxt.LocalBufHash == NULL)
InitLocalBuffers();
/* See if the desired buffer already exists */
// 查看所需的缓冲区是否已经存在
hresult = (LocalBufferLookupEnt*)hash_search(u_sess->storage_cxt.LocalBufHash, (void*)&new_tag, HASH_FIND, NULL);
if (hresult != NULL) {
/* Yes, so nothing to do */
/* 是的,所以不需要进行预取操作 */
return;
}
/* Not in buffers, so initiate prefetch */
// 如果缓冲区中不存在所需块,启动预取操作
smgrprefetch(smgr, forkNum, blockNum);
#endif /* USE_PREFETCH */
}
/*
SMgrRelation
smgrwrite fork
*/
void LocalBufferWrite(BufferDesc *bufHdr)
{
SMgrRelation oreln;
Page localpage = (char *)LocalBufHdrGetBlock(bufHdr);
char *bufToWrite = NULL;
/* Find smgr relation for buffer */
/* 查找缓冲区对应的SMgrRelation */
oreln = smgropen(bufHdr->tag.rnode, BackendIdForTempRelations);
/* data encrypt */
/* 对缓冲区中的数据进行加密data encrypt */
bufToWrite = PageDataEncryptForBuffer(localpage, bufHdr);
/* 计算并设置数据页的校验和 */
PageSetChecksumInplace((Page)bufToWrite, bufHdr->tag.blockNum);
/* And write... */
/* 将数据写入磁盘 */
smgrwrite(oreln, bufHdr->tag.forkNum, bufHdr->tag.blockNum, bufToWrite, false);
}
void LocalBufferFlushForExtremRTO(BufferDesc *bufHdr)
{
if (dw_enabled()) {
/* double write */
/* 如果启用了双写double write机制执行相应的操作 */
/* 双写是一种数据持久性保护机制,确保数据写入磁盘的安全性 */
/* 在此处可能包括双写操作的代码 */
}
/* 调用 FlushBuffer 函数将本地缓冲区刷新到磁盘 */
/* WITH_LOCAL_CACHE 参数表示使用本地缓存 */
FlushBuffer(bufHdr, NULL, WITH_LOCAL_CACHE);
}
/*
LocalBufferFlushForExtremRTO
*/
void LocalBufferFlushAllBuffer()
{
int i;
// 遍历所有本地缓冲区
for (i = 0; i < u_sess->storage_cxt.NLocBuffer; i++) {
BufferDesc *bufHdr = &u_sess->storage_cxt.LocalBufferDescriptors[i];
uint32 buf_state;
// 读取缓冲区状态
buf_state = pg_atomic_read_u32(&bufHdr->state);
// 确保本地引用计数为0表示没有任何进程正在使用该缓冲区
Assert(u_sess->storage_cxt.LocalRefCount[i] == 0);
// 如果缓冲区是有效的且脏的
if ((buf_state & BM_VALID) && (buf_state & BM_DIRTY)) {
// 调用 LocalBufferFlushForExtremRTO 函数刷新缓冲区到磁盘
LocalBufferFlushForExtremRTO(bufHdr);
// 清除缓冲区的脏标志
buf_state &= ~BM_DIRTY;
pg_atomic_write_u32(&bufHdr->state, buf_state);
// 更新本地缓冲区写入计数
u_sess->instr_cxt.pg_buffer_usage->local_blks_written++;
}
}
}
static void LocalBufferSanityCheck(BufferTag tag1, BufferTag tag2)
{
if (!BUFFERTAGS_EQUAL(tag1, tag2)) {
@ -135,9 +173,17 @@ static void LocalBufferSanityCheck(BufferTag tag1, BufferTag tag2)
* does not get set. Lastly, we support only default access strategy
* (hence, usage_count is always advanced).
*/
/*
使
*/
BufferDesc *LocalBufferAlloc(SMgrRelation smgr, ForkNumber forkNum, BlockNumber blockNum, bool *foundPtr)
{
BufferTag new_tag; /* identity of requested block */
BufferTag new_tag; /* 请求块的标识 */
LocalBufferLookupEnt *hresult = NULL;
BufferDesc *buf_desc = NULL;
int b;
@ -145,24 +191,22 @@ BufferDesc *LocalBufferAlloc(SMgrRelation smgr, ForkNumber forkNum, BlockNumber
bool found = false;
uint32 buf_state;
// 初始化请求块的标识
INIT_BUFFERTAG(new_tag, smgr->smgr_rnode.node, forkNum, blockNum);
/* Initialize local buffers if first request in this session */
// 如果在当前会话中首次请求本地缓冲区,初始化本地缓冲区
if (u_sess->storage_cxt.LocalBufHash == NULL)
InitLocalBuffers();
/* See if the desired buffer already exists */
// 查看所需的缓冲区是否已经存在
hresult = (LocalBufferLookupEnt*)hash_search(u_sess->storage_cxt.LocalBufHash, (void*)&new_tag, HASH_FIND, NULL);
if (hresult != NULL) {
b = hresult->id;
buf_desc = &u_sess->storage_cxt.LocalBufferDescriptors[b];
LocalBufferSanityCheck(buf_desc->tag, new_tag);
#ifdef LBDEBUG
fprintf(stderr, "LB ALLOC (%u,%d,%d) %d\n", smgr->smgr_rnode.node.relNode, forkNum, blockNum, -b - 1);
#endif
buf_state = pg_atomic_read_u32(&buf_desc->state);
/* this part is equivalent to PinBuffer for a shared buffer */
// 如果缓冲区已经存在,增加本地引用计数,标记缓冲区为使用中
buf_state = pg_atomic_read_u32(&buf_desc->state);
if (u_sess->storage_cxt.LocalRefCount[b] == 0) {
if (BUF_STATE_GET_USAGECOUNT(buf_state) < BM_MAX_USAGE_COUNT) {
buf_state += BUF_USAGECOUNT_ONE;
@ -171,26 +215,12 @@ BufferDesc *LocalBufferAlloc(SMgrRelation smgr, ForkNumber forkNum, BlockNumber
}
u_sess->storage_cxt.LocalRefCount[b]++;
ResourceOwnerRememberBuffer(t_thrd.utils_cxt.CurrentResourceOwner, BufferDescriptorGetBuffer(buf_desc));
*foundPtr = (buf_state & BM_VALID) ? TRUE : FALSE; /* If previous read attempt have failed; try again */
#ifdef EXTREME_RTO_DEBUG
ereport(LOG, (errmsg("LocalBufferAlloc %u/%u/%u %u %u find in local buf %u/%u/%u %u %u id %d state %X, lsn %lu",
smgr->smgr_rnode.node.spcNode, smgr->smgr_rnode.node.dbNode, smgr->smgr_rnode.node.relNode,
forkNum, blockNum, hresult->key.rnode.spcNode, hresult->key.rnode.dbNode,
hresult->key.rnode.relNode, hresult->key.forkNum, hresult->key.blockNum, hresult->id,
buf_state, LocalBufGetLSN(buf_desc))));
#endif
*foundPtr = (buf_state & BM_VALID) ? TRUE : FALSE; /* 如果以前的读取尝试失败,则再次尝试 */
return buf_desc;
}
#ifdef LBDEBUG
fprintf(stderr, "LB ALLOC (%u,%d,%d) %d\n", smgr->smgr_rnode.node.relNode, forkNum, blockNum,
-t_thrd.storage_cxt.nextFreeLocalBuf - 1);
#endif
/*
* Need to get a new buffer. We use a clock sweep algorithm (essentially
* the same as what freelist.c does now...)
*/
// 如果缓冲区不存在,需要分配新的缓冲区
try_counter = u_sess->storage_cxt.NLocBuffer;
for (;;) {
b = u_sess->storage_cxt.nextFreeLocalBuf;
@ -208,19 +238,18 @@ BufferDesc *LocalBufferAlloc(SMgrRelation smgr, ForkNumber forkNum, BlockNumber
pg_atomic_write_u32(&buf_desc->state, buf_state);
try_counter = u_sess->storage_cxt.NLocBuffer;
} else {
/* Found a usable buffer */
/* 找到可用的缓冲区,增加引用计数,标记为使用中 */
u_sess->storage_cxt.LocalRefCount[b]++;
ResourceOwnerRememberBuffer(t_thrd.utils_cxt.CurrentResourceOwner, BufferDescriptorGetBuffer(buf_desc));
break;
}
} else if (--try_counter == 0)
} else if (--try_counter == 0) {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_RESOURCES), errmsg("no empty local buffer available")));
}
}
/*
* this buffer is not referenced but it might still be dirty. if that's
* the case, write it out before reusing it!
*/
// 如果缓冲区存在脏数据,需要在重用之前写回磁盘
buf_state = pg_atomic_read_u32(&buf_desc->state);
if (buf_state & BM_DIRTY) {
if (AmPageRedoProcess()) {
LocalBufferFlushForExtremRTO(buf_desc);
@ -228,30 +257,26 @@ BufferDesc *LocalBufferAlloc(SMgrRelation smgr, ForkNumber forkNum, BlockNumber
LocalBufferWrite(buf_desc);
}
/* Mark not-dirty now in case we error out below */
// 清除脏标志
buf_state &= ~BM_DIRTY;
pg_atomic_write_u32(&buf_desc->state, buf_state);
u_sess->instr_cxt.pg_buffer_usage->local_blks_written++;
}
/*
* lazy memory allocation: allocate space on first use of a buffer.
*/
// 懒惰内存分配:在第一次使用缓冲区时分配空间
if (LocalBufHdrGetBlock(buf_desc) == NULL) {
/* Set pointer for use by BufferGetBlock() macro */
/* 为 BufferGetBlock 宏设置指针 */
LocalBufHdrGetBlock(buf_desc) = GetLocalBufferStorage();
}
/*
* Update the hash table: remove old entry, if any, and make new one.
*/
// 更新哈希表:删除旧条目(如果存在)并创建新条目
if (buf_state & BM_TAG_VALID) {
hresult = (LocalBufferLookupEnt *)hash_search(u_sess->storage_cxt.LocalBufHash, (void *)&buf_desc->tag,
HASH_REMOVE, NULL);
if (hresult == NULL) /* shouldn't happen */
if (hresult == NULL) /* 不应该发生 */
ereport(ERROR, (errcode(ERRCODE_DATA_CORRUPTED), (errmsg("local buffer hash table corrupted."))));
/* mark buffer invalid just in case hash insert fails */
/* 清除缓冲区标签以防哈希插入失败 */
CLEAR_BUFFERTAG(buf_desc->tag);
buf_state &= ~(BM_VALID | BM_TAG_VALID);
pg_atomic_write_u32(&buf_desc->state, buf_state);
@ -259,15 +284,12 @@ BufferDesc *LocalBufferAlloc(SMgrRelation smgr, ForkNumber forkNum, BlockNumber
hresult = (LocalBufferLookupEnt *)hash_search(u_sess->storage_cxt.LocalBufHash, (void *)&new_tag, HASH_ENTER,
&found);
if (found) /* shouldn't happen */
if (found) /* 不应该发生 */
ereport(ERROR, (errcode(ERRCODE_DATA_CORRUPTED), (errmsg("local buffer hash table corrupted."))));
hresult->id = b;
/*
* it's all ours now.
*/
buf_desc->tag = new_tag;
buf_desc->encrypt = smgr->encrypt ? true : false; /* set tde flag */
buf_desc->encrypt = smgr->encrypt ? true : false; /* 设置 TDE 标志 */
buf_state &= ~(BM_VALID | BM_DIRTY | BM_JUST_DIRTIED | BM_IO_ERROR);
buf_state |= BM_TAG_VALID;
buf_state &= ~BUF_USAGECOUNT_MASK;
@ -280,35 +302,52 @@ BufferDesc *LocalBufferAlloc(SMgrRelation smgr, ForkNumber forkNum, BlockNumber
return buf_desc;
}
/*
* MarkLocalBufferDirty -
* mark a local buffer dirty
*/
/*
使 BM_DIRTY
*/
void MarkLocalBufferDirty(Buffer buffer)
{
int buf_id;
BufferDesc *buf_desc = NULL;
uint32 buf_state;
// 确保传入的缓冲区是本地缓冲区
Assert(BufferIsLocal(buffer));
#ifdef LBDEBUG
fprintf(stderr, "LB DIRTY %d\n", buffer);
#endif
// 通过缓冲区号计算缓冲区描述符的索引
buf_id = -(buffer + 1);
// 确保本地引用计数大于0表示缓冲区正在使用中
Assert(u_sess->storage_cxt.LocalRefCount[buf_id] > 0);
// 获取缓冲区描述符
buf_desc = &u_sess->storage_cxt.LocalBufferDescriptors[buf_id];
// 使用原子操作将缓冲区的状态标记为脏数据
buf_state = pg_atomic_fetch_or_u32(&buf_desc->state, BM_DIRTY);
// 如果之前未标记为脏数据,增加本地缓冲区脏数据统计
if (!(buf_state & BM_DIRTY)) {
u_sess->instr_cxt.pg_buffer_usage->local_blks_dirtied++;
pgstatCountLocalBlocksDirtied4SessionLevel();
}
}
/*
* DropRelFileNodeLocalBuffers
* This function removes from the buffer pool all the pages of the
@ -320,6 +359,13 @@ void MarkLocalBufferDirty(Buffer buffer)
*
* See DropRelFileNodeBuffers in bufmgr.c for more notes.
*/
/*
便
*/
void DropRelFileNodeLocalBuffers(const RelFileNode &rnode, ForkNumber forkNum, BlockNumber firstDelBlock)
{
int i;
@ -329,10 +375,14 @@ void DropRelFileNodeLocalBuffers(const RelFileNode &rnode, ForkNumber forkNum, B
LocalBufferLookupEnt* hresult = NULL;
uint32 buf_state;
// 读取缓冲区的状态信息
buf_state = pg_atomic_read_u32(&buf_desc->state);
// 检查缓冲区的标签是否有效,且与传入的关系文件节点、分支号、以及块号条件匹配
if ((buf_state & BM_TAG_VALID) && RelFileNodeEquals(rnode, buf_desc->tag.rnode) &&
buf_desc->tag.forkNum == forkNum && buf_desc->tag.blockNum >= firstDelBlock) {
// 如果缓冲区仍然被引用,报错
if (u_sess->storage_cxt.LocalRefCount[i] != 0) {
ereport(ERROR,
(errcode(ERRCODE_INVALID_BUFFER_REFERENCE),
@ -341,13 +391,17 @@ void DropRelFileNodeLocalBuffers(const RelFileNode &rnode, ForkNumber forkNum, B
relpathbackend(buf_desc->tag.rnode, BackendIdForTempRelations, buf_desc->tag.forkNum),
u_sess->storage_cxt.LocalRefCount[i]))));
}
/* Remove entry from hashtable */
// 从哈希表中移除缓冲区的条目
hresult = (LocalBufferLookupEnt*)hash_search(
u_sess->storage_cxt.LocalBufHash, (void*)&buf_desc->tag, HASH_REMOVE, NULL);
if (hresult == NULL) /* shouldn't happen */
if (hresult == NULL) /* 不应该发生 */
ereport(ERROR, (errcode(ERRCODE_DATA_CORRUPTED), (errmsg("local buffer hash table corrupted."))));
/* Mark buffer invalid */
// 清除缓冲区的标签
CLEAR_BUFFERTAG(buf_desc->tag);
// 标记缓冲区为无效状态
buf_state &= ~BUF_FLAG_MASK;
buf_state &= ~BUF_USAGECOUNT_MASK;
pg_atomic_write_u32(&buf_desc->state, buf_state);
@ -355,6 +409,7 @@ void DropRelFileNodeLocalBuffers(const RelFileNode &rnode, ForkNumber forkNum, B
}
}
/*
* DropRelFileNodeAllLocalBuffers
* This function removes from the buffer pool all pages of all forks
@ -362,6 +417,14 @@ void DropRelFileNodeLocalBuffers(const RelFileNode &rnode, ForkNumber forkNum, B
*
* See DropRelFileNodeAllBuffers in bufmgr.c for more notes.
*/
/*
便
*/
void DropRelFileNodeAllLocalBuffers(const RelFileNode &rnode)
{
int i;
@ -371,9 +434,13 @@ void DropRelFileNodeAllLocalBuffers(const RelFileNode &rnode)
LocalBufferLookupEnt* hresult = NULL;
uint32 buf_state;
// 读取缓冲区的状态信息
buf_state = pg_atomic_read_u32(&buf_desc->state);
// 检查缓冲区的标签是否有效,且与传入的关系文件节点条件匹配
if ((buf_state & BM_TAG_VALID) && RelFileNodeEquals(rnode, buf_desc->tag.rnode)) {
// 如果缓冲区仍然被引用,报错
if (u_sess->storage_cxt.LocalRefCount[i] != 0) {
if (buf_desc->tag.forkNum < 0) {
ereport(ERROR, (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
@ -386,13 +453,17 @@ void DropRelFileNodeAllLocalBuffers(const RelFileNode &rnode)
relpathbackend(buf_desc->tag.rnode, BackendIdForTempRelations, buf_desc->tag.forkNum),
u_sess->storage_cxt.LocalRefCount[i]))));
}
/* Remove entry from hashtable */
// 从哈希表中移除缓冲区的条目
hresult = (LocalBufferLookupEnt*)hash_search(
u_sess->storage_cxt.LocalBufHash, (void*)&buf_desc->tag, HASH_REMOVE, NULL);
if (hresult == NULL) /* shouldn't happen */
if (hresult == NULL) /* 不应该发生 */
ereport(ERROR, (errcode(ERRCODE_DATA_CORRUPTED), (errmsg("local buffer hash table corrupted."))));
/* Mark buffer invalid */
// 清除缓冲区的标签
CLEAR_BUFFERTAG(buf_desc->tag);
// 标记缓冲区为无效状态
buf_state &= ~BUF_FLAG_MASK;
buf_state &= ~BUF_USAGECOUNT_MASK;
pg_atomic_write_u32(&buf_desc->state, buf_state);
@ -400,19 +471,27 @@ void DropRelFileNodeAllLocalBuffers(const RelFileNode &rnode)
}
}
/*
* InitLocalBuffers -
* init the local buffer cache. Since most queries (esp. multi-user ones)
* don't involve local buffers, we delay allocating actual memory for the
* buffers until we need them; just make the buffer headers here.
*/
/*
ID
使
*/
static void InitLocalBuffers(void)
{
int nbufs = u_sess->attr.attr_storage.num_temp_buffers;
int nbufs = u_sess->attr.attr_storage.num_temp_buffers; // 获取本地缓冲区的数量
HASHCTL info;
int i;
/* Allocate and zero buffer headers and auxiliary arrays */
/* 分配并清零缓冲区头和辅助数组的内存空间 */
u_sess->storage_cxt.LocalBufferDescriptors = (BufferDesc*)MemoryContextAllocZero(
SESS_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_STORAGE), (unsigned int)nbufs * sizeof(BufferDesc));
u_sess->storage_cxt.LocalBufferBlockPointers = (Block*)MemoryContextAllocZero(
@ -425,20 +504,18 @@ static void InitLocalBuffers(void)
u_sess->storage_cxt.nextFreeLocalBuf = 0;
/* initialize fields that need to start off nonzero */
/* 初始化需要初始值非零的字段 */
for (i = 0; i < nbufs; i++) {
BufferDesc* buf = &u_sess->storage_cxt.LocalBufferDescriptors[i];
/*
* negative to indicate local buffer. This is tricky: shared buffers
* start with 0. We have to start with -2. (Note that the routine
* BufferDescriptorGetBuffer adds 1 to buf_id so our first buffer id
* is -1.)
* 使0-2
* BufferDescriptorGetBuffer函数会将buf_id加1ID是-1
*/
buf->buf_id = -i - 2;
}
/* Create the lookup hash table */
/* 创建查找哈希表 */
errno_t ret = memset_s(&info, sizeof(info), 0, sizeof(info));
securec_check(ret, "\0", "\0");
info.keysize = sizeof(BufferTag);
@ -454,10 +531,11 @@ static void InitLocalBuffers(void)
(errmsg("could not initialize local buffer hash table."))));
}
/* Initialization done, mark buffers allocated */
/* 初始化完成,标记分配了缓冲区 */
u_sess->storage_cxt.NLocBuffer = nbufs;
}
/*
* GetLocalBufferStorage - allocate memory for a local buffer
*
@ -467,6 +545,14 @@ static void InitLocalBuffers(void)
* within a particular process, no point in burdening memmgr with separately
* managed chunks.
*/
/*
*/
static Block GetLocalBufferStorage(void)
{
char *this_buf = NULL;
@ -474,13 +560,13 @@ static Block GetLocalBufferStorage(void)
Assert(u_sess->storage_cxt.total_bufs_allocated < u_sess->storage_cxt.NLocBuffer);
if (u_sess->storage_cxt.next_buf_in_block >= u_sess->storage_cxt.num_bufs_in_block) {
/* Need to make a new request to memmgr */
/* 需要向内存管理器请求新的内存块 */
int num_bufs;
/*
* We allocate local buffers in a context of their own, so that the
* space eaten for them is easily recognizable in MemoryContextStats
* output. Create the context on first use.
* 便 MemoryContextStats
* 使
*/
if (u_sess->storage_cxt.LocalBufferContext == NULL)
u_sess->storage_cxt.LocalBufferContext = AllocSetContextCreate(u_sess->top_mem_cxt,
@ -489,11 +575,11 @@ static Block GetLocalBufferStorage(void)
ALLOCSET_DEFAULT_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
/* Start with a 16-buffer request; subsequent ones double each time */
/* 从一个16个缓冲区的请求开始后续请求每次翻倍 */
num_bufs = Max(u_sess->storage_cxt.num_bufs_in_block * 2, 16);
/* But not more than what we need for all remaining local bufs */
/* 但不能超过剩余本地缓冲区所需的数量 */
num_bufs = Min(num_bufs, u_sess->storage_cxt.NLocBuffer - u_sess->storage_cxt.total_bufs_allocated);
/* And don't overflow MaxAllocSize, either */
/* 也不能超过 MaxAllocSize */
num_bufs = Min((unsigned int)(num_bufs), MaxAllocSize / BLCKSZ);
u_sess->storage_cxt.cur_block =
@ -502,7 +588,7 @@ static Block GetLocalBufferStorage(void)
u_sess->storage_cxt.num_bufs_in_block = num_bufs;
}
/* Allocate next buffer in current memory block */
/* 在当前内存块中分配下一个缓冲区 */
this_buf = u_sess->storage_cxt.cur_block + u_sess->storage_cxt.next_buf_in_block * BLCKSZ;
u_sess->storage_cxt.next_buf_in_block++;
u_sess->storage_cxt.total_bufs_allocated++;
@ -510,6 +596,7 @@ static Block GetLocalBufferStorage(void)
return (Block)this_buf;
}
/*
* AtEOXact_LocalBuffers - clean up at end of transaction.
*
@ -560,32 +647,32 @@ void AtProcExit_LocalBuffers(void)
void ForgetLocalBuffer(RelFileNode rnode, ForkNumber forkNum, BlockNumber blockNum)
{
SMgrRelation smgr = smgropen(rnode, t_thrd.proc_cxt.MyBackendId);
BufferTag tag; /* identity of target block */
BufferTag tag; /* 目标块的标识 */
LocalBufferLookupEnt *hresult;
BufferDesc *bufHdr;
uint32 bufState;
/*
* If somehow this is the first request in the session, there's nothing to
* do. (This probably shouldn't happen, though.)
*
*
*/
if (t_thrd.storage_cxt.LocalBufHash == NULL) {
return;
}
/* create a tag so we can lookup the buffer */
/* 创建一个标签以便查找缓冲块 */
INIT_BUFFERTAG(tag, smgr->smgr_rnode.node, forkNum, blockNum);
/* see if the block is in the local buffer pool */
/* 查看块是否在本地缓冲池中 */
hresult = (LocalBufferLookupEnt *)
hash_search(t_thrd.storage_cxt.LocalBufHash, (void *) &tag, HASH_REMOVE, NULL);
/* didn't find it, so nothing to do */
/* 没找到,所以无需处理 */
if (!hresult) {
return;
}
/* mark buffer invalid */
/* 将缓冲块标记为无效 */
bufHdr = GetLocalBufferDescriptor(hresult->id);
CLEAR_BUFFERTAG(bufHdr->tag);
bufState = pg_atomic_read_u32(&bufHdr->state);

View File

@ -31,24 +31,31 @@ namespace dfs {
* @Return: dfs connecort
* @See also:
*/
// 创建一个 DFSConnector 连接器对象
DFSConnector *createConnector(MemoryContext ctx, Oid foreignTableId)
{
// 获取外部表的服务器类型
ServerTypeOption srvType = getServerType(foreignTableId);
switch (srvType) {
case T_OBS_SERVER:
case T_TXT_CSV_OBS_SERVER: {
#ifndef ENABLE_LITE_MODE
// 如果服务器类型是 OBS 或 TXT_CSV_OBS则创建一个 OBSConnector 对象
// 在 LITE_MODE 未启用时创建连接器对象
return New(ctx) OBSConnector(ctx, foreignTableId);
#else
// 如果启用了 LITE_MODE则报告不支持的特性并返回 NULL
FEATURE_ON_LITE_MODE_NOT_SUPPORTED();
return NULL;
#endif
}
case T_HDFS_SERVER: {
// 如果服务器类型是 HDFS则报告不支持的错误并返回 NULL
FEATURE_NOT_PUBLIC_ERROR("HDFS is not yet supported.");
return NULL;
}
default: {
// 如果服务器类型未知或不支持,则报告错误,并返回 NULL
ereport(ERROR, (errcode(ERRCODE_INVALID_OBJECT_DEFINITION), errmodule(MOD_DFS),
errmsg("dfs foreign server type error. type = %d", srvType)));
return NULL;
@ -56,6 +63,7 @@ DFSConnector *createConnector(MemoryContext ctx, Oid foreignTableId)
}
}
/*
* @Description: create connector for foreign table
* @IN ctx: memory context
@ -63,31 +71,39 @@ DFSConnector *createConnector(MemoryContext ctx, Oid foreignTableId)
* @Return: dfs connector
* @See also:
*/
// 创建 DFSConnector 连接器对象
DFSConnector *createConnector(MemoryContext ctx, ServerTypeOption srvType, void *options)
{
// 根据服务器类型进行分支处理
switch (srvType) {
case T_OBS_SERVER: {
#ifndef ENABLE_LITE_MODE
// 如果服务器类型为 OBS
// 在非 LITE_MODE 模式下,创建一个 OBSConnector 连接器对象
return New(ctx) OBSConnector(ctx, (ObsOptions *)options);
#else
// 如果启用了 LITE_MODE报告不支持的特性并返回 NULL
FEATURE_ON_LITE_MODE_NOT_SUPPORTED();
return NULL;
#endif
break;
break; // 可选的,不过由于函数会在 return 之后退出,所以不会执行到这里
}
case T_HDFS_SERVER: {
// 如果服务器类型为 HDFS报告不支持的错误并返回 NULL
FEATURE_NOT_PUBLIC_ERROR("HDFS is not yet supported.");
return NULL;
break;
break; // 可选的,不过由于函数会在 return 之后退出,所以不会执行到这里
}
default: {
// 如果服务器类型未知或不支持,报告错误,并返回 NULL
ereport(ERROR, (errcode(ERRCODE_INVALID_OBJECT_DEFINITION), errmodule(MOD_DFS),
errmsg("dfs forign server type error. type = %d", srvType)));
errmsg("dfs foreign server type error. type = %d", srvType)));
return NULL;
}
}
}
/*
* @Description: create connctor for table space, only support hdfs
* @IN ctx: memory context
@ -110,40 +126,51 @@ DFSConnector *createConnector(MemoryContext ctx, DfsSrvOptions *srvOptions, Oid
* @Return: true for skip, false for not skip
* @See also:
*/
// 检查文件是否应该被跳过的函数
bool checkFileShouldSkip(char *fileName, int end_pos)
{
// 如果文件名为空或者为空字符串,则不跳过文件,返回 false
if (fileName == NULL || *fileName == '\0') {
return false;
}
// 使用断言确保 end_pos 不超过文件名的长度
Assert(end_pos <= (int)strlen(fileName));
/* start with . _ or # will be skip */
// 如果文件名以 '.'、'_' 或 '#' 开头,则跳过文件,返回 true
if (*fileName == '.' || *fileName == '_' || *fileName == '#') {
return true;
}
/* end with ~ will be skip */
// 获取文件名的最后一个字符
char *endChar = fileName + end_pos - 1;
// 如果文件名以 '~' 结尾,则跳过文件,返回 true
if (*endChar == '~') {
return true;
}
// 否则,不跳过文件,返回 false
return false;
}
/*
* @Description: check file path should skip
* @IN fileName: file name
* @Return: true for skip, false for not skip
* @See also:
*/
// 封装函数,检查文件是否应该被跳过
bool checkFileShouldSkip(char *fileName)
{
// 如果文件名为空或者为空字符串,则不跳过文件,返回 false
if (fileName == NULL || *fileName == '\0') {
return false;
}
// 调用之前的 checkFileShouldSkip 函数,传递文件名和文件名的长度作为参数
// 获取返回值并返回
bool isSkip = checkFileShouldSkip(fileName, strlen(fileName));
return isSkip;
@ -155,45 +182,55 @@ bool checkFileShouldSkip(char *fileName)
* @Return: true for skip, false for not skip
* @See also:
*/
// 检查路径中的文件是否应该被跳过的函数
bool checkPathShouldSkip(char *pathName)
{
// 如果路径名为空或为空字符串,则不跳过文件,返回 false
if (pathName == NULL || *pathName == '\0') {
return false;
}
int end_pos = 0;
bool isSkip = false;
char *pos = pathName;
int end_pos = 0; // 用于存储文件名的结束位置
bool isSkip = false; // 用于存储是否应该跳过文件的标志
char *pos = pathName; // 用于迭代遍历路径名中的各个文件名
/* first char is slash */
// 如果路径名的第一个字符是斜杠 '/',则跳过它
if (*pos == '/') {
pos += 1;
}
/* find next slash */
// 查找下一个斜杠 '/'
char *next_slash = strchr(pos, '/');
/* get next slash position or the last char position */
// 获取下一个斜杠的位置或路径名的最后一个字符位置
end_pos = (int)((next_slash != NULL) ? (next_slash - pos) : strlen(pos));
/* check path name */
// 检查当前文件名是否应该跳过
isSkip = checkFileShouldSkip(pos, end_pos);
// 遍历路径中的各个文件名,直到找到一个应该跳过的文件名或遍历完整个路径
while (!isSkip) {
pos = next_slash;
// 如果 pos 为 NULL 或为空字符,则退出循环
if (pos == NULL || *pos == '\0') {
break;
}
pos += 1;
/* find next slash */
// 查找下一个斜杠 '/'
next_slash = strchr(pos, '/');
/* get next slash position or the last char position */
// 获取下一个斜杠的位置或路径名的最后一个字符位置
end_pos = (int)((next_slash != NULL) ? (next_slash - pos) : strlen(pos));
/* check path name */
// 检查当前文件名是否应该跳过
isSkip = checkFileShouldSkip(pos, end_pos);
}
// 返回是否应该跳过路径中的文件
return isSkip;
}