huawei
/
openGauss-server
0
0
Fork 77
Code Issues Pull Requests 24 Projects Releases Wiki Activity

stOOrz部分代码注释 #30

Open
p5kenirwc wants to merge 16 commits from p5kenirwc/openGauss-server:master into master
pull from: p5kenirwc/openGauss-server:master
merge into: huawei:master
Conversation 0 Commits 16 Files Changed 15 +522 -92
15 changed files with 522 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
src/common/backend/utils/time/snapmgr.cpp
View File

@ -178,10 +178,11 @@ static void RecheckXidFinish(TransactionId xid, CommitSeqNo csn)
/*
* XidVisibleInSnapshot
* Is the given XID visible according to the snapshot?
* This code is a function named XidVisibleInSnapshot, which accepts five parameters: TransactionId xid, Snapshot snapshot,
* TransactionIdStatus* hintstatus, Buffer buffer, and bool* sync
* The main purpose of this function is to check whether a transaction ID (xid) is visible in a given snapshot (snapshot).
* It does this by obtaining the commit sequence number (csn) of the transaction ID and comparing it with the commit sequence number of the snapshot.
*
* On return, *hintstatus is set to indicate if the transaction had committed,
* or aborted, whether or not it's not visible to us.
*/
bool XidVisibleInSnapshot(TransactionId xid, Snapshot snapshot, TransactionIdStatus* hintstatus, Buffer buffer, bool* sync)
{
@ -202,6 +203,7 @@ bool XidVisibleInSnapshot(TransactionId xid, Snapshot snapshot, TransactionIdSta
loop:
csn = TransactionIdGetCommitSeqNo(xid, false, true, false, snapshot);
//fetch CSN of specified transaction id
#ifdef XIDVIS_DEBUG
ereport(DEBUG1,
@ -220,10 +222,23 @@ loop:
return true;
else
return false;
//if the commit sequence number has been committed,
//and is less than the commit sequence number of the sanpshot,
//then the function return true ,else return false
} else if (COMMITSEQNO_IS_COMMITTING(csn)) {
//if the commit sequece number is committing,
//the function performs some additonal checks and operations
//including :synchronously waiting for the transaction to end
if (looped) {
ereport(DEBUG1, (errmsg("transaction id %lu's csn %ld is changed to ABORT after lockwait.", xid, csn)));
/* recheck if transaction id is finished */
/*
* If a loop has already been performed (looped is true),
* then the function reports that the csn of the transaction ID has been changed to ABORT
* after waiting for a lock. Then, it rechecks whether the transaction ID has finished,
* sets the csn of the transaction ID to ABORTED, updates the latest fetch state of the transaction ID to ABORTED,
* sets hintstatus to XID_ABORTED, and returns false (false).
*/
RecheckXidFinish(xid, csn);
CSNLogSetCommitSeqNo(xid, 0, NULL, COMMITSEQNO_ABORTED);
SetLatestFetchState(xid, COMMITSEQNO_ABORTED);
@ -249,7 +264,7 @@ loop:
if (u_sess->attr.attr_common.xc_maintenance_mode || t_thrd.xact_cxt.bInAbortTransaction) {
return false;
}
/* Then, if it is currently in maintenance mode or in the process of aborting a transaction, then the function returns false */
/* Wait for txn end and check again. */
if (sync != NULL) {
*sync = true;
@ -896,14 +911,22 @@ Snapshot GetActiveSnapshot(void)
if (!u_sess->utils_cxt.ActiveSnapshot && IS_PGXC_COORDINATOR && !IsConnFromCoord())
return NULL;
#endif
//This code is checking if the current snapshot (ActiveSnapshot) is null.
if (u_sess->utils_cxt.ActiveSnapshot == NULL) {
ereport(ERROR,
(errmodule(MOD_TRANS_SNAPSHOT),
errcode(ERRCODE_INVALID_STATUS),
errmsg("snapshot is not active")));
// Specifically, if the current snapshot is null (i.e., there is no active snapshot)
// then the code will report an error, indicating "snapshot not active".
// This might mean that an attempt was made to read data without starting a transaction,
// which is not allowed.
}
return u_sess->utils_cxt.ActiveSnapshot->as_snap;
// If the current snapshot is not null, then the code will
// return the snapshot value of the current snapshot.
}
/*

12
src/gausskernel/storage/access/archive/archive_am.cpp
View File

@ -38,7 +38,8 @@ size_t ArchiveRead(const char* fileName, const int offset, char *buffer, const i
return 0;
}
// 从指定的文件名和偏移量开始，读取指定长度的数据到缓冲区。
//如果媒体类型是OBS，则调用obsRead，如果是NAS，则调用NasRead。
int ArchiveWrite(const char* fileName, const char *buffer, const int bufferLength, ArchiveConfig *archive_config)
{
int ret = -1;
@ -54,7 +55,8 @@ int ArchiveWrite(const char* fileName, const char *buffer, const int bufferLengt
return ret;
}
//将缓冲区中的数据写入指定的文件。
//如果媒体类型是OBS，则调用obsWrite，如果是NAS，则调用NasWrite
int ArchiveDelete(const char* fileName, ArchiveConfig *archive_config)
{
int ret = -1;
@ -70,7 +72,7 @@ int ArchiveDelete(const char* fileName, ArchiveConfig *archive_config)
return ret;
}
//删除指定的文件。如果媒体类型是OBS，则调用obsDelete，如果是NAS，则调用NasDelete。
List* ArchiveList(const char* prefix, ArchiveConfig *archive_config, bool reportError, bool shortenConnTime)
{
List* fileNameList = NIL;
@ -86,7 +88,7 @@ List* ArchiveList(const char* prefix, ArchiveConfig *archive_config, bool report
return fileNameList;
}
//列出与指定前缀匹配的所有文件。如果媒体类型是OBS，则调用obsList，如果是NAS，则调用NasList。
bool ArchiveFileExist(const char* file_path, ArchiveConfig *archive_config)
{
bool ret = false;
@ -102,4 +104,4 @@ bool ArchiveFileExist(const char* file_path, ArchiveConfig *archive_config)
}
return ret;
}
}//检查指定的文件是否存在。如果媒体类型是OBS，则调用checkOBSFileExist，如果是NAS，则调用checkNASFileExist

118
src/gausskernel/storage/access/archive/nas_am.cpp
View File

@ -48,7 +48,26 @@
#define MAX_PATH_LEN 1024
static int headerLen = 22;
/*
function name: NasRead
description: This code defines a function named NasRead that reads data from a specified file
arguments:
fileName: The name of the file to read.
offset: The position in the file to start reading from.
buffer: The buffer to store the read data.
length: The number of bytes to read.
nas_config: The configuration for the NAS return
value: Returns readLength if the file is read successfully, otherwise returns zero.
Note: The main process is as follows:
It checks if fileName and buffer are not NULL, and if nas_config is NULL, it retrieves the archive configuration.
It constructs the full file path based on the file name and archive prefix.
It checks if the file exists and if it can be accessed. If not, it returns an error.
It opens the file in binary read mode and reads data from it into the buffer.
If there’s an error during reading or if the file size is larger than the buffer length, it closes the file and returns
an error.
Finally, it closes the file and returns the number of bytes read.
date: 2023/9/17
*/
size_t NasRead(const char* fileName, const int offset, char *buffer, const int length, ArchiveConfig *nas_config)
{
size_t readLength = 0;
@ -62,7 +81,6 @@ size_t NasRead(const char* fileName, const int offset, char *buffer, const int l
ereport(ERROR, (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
errmsg("The parameter cannot be NULL")));
}
if (nas_config != NULL) {
archive_nas = nas_config;
} else {
@ -73,10 +91,10 @@ size_t NasRead(const char* fileName, const int offset, char *buffer, const int l
ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("Cannot get archive config from replication slots")));
}
if (strncmp(fileName, "global_barrier_records", headerLen) != 0) {
ret = snprintf_s(file_path, MAXPGPATH, MAXPGPATH - 1, "%s/%s", archive_nas->archive_prefix, fileName);
securec_check_ss(ret, "\0", "\0");
//"\0" is the messgae to be printed when an error occurs.In this example ,no message will be printed when an error occurs;
} else {
char pathPrefix[MAXPGPATH] = {0};
ret = strcpy_s(pathPrefix, MAXPGPATH, archive_nas->archive_prefix);
@ -117,7 +135,31 @@ size_t NasRead(const char* fileName, const int offset, char *buffer, const int l
fclose(fp);
return readLength;
}
/*
function name: NasWrite
description: This function first checks the validity of the input parameters.
Then, it constructs the file path based on the archive configuration and file name
fileName: The name of the file to read.
bufferLength: The number of bytes to write.
buffer: The buffer containing the data to write.
nas_config: The configuration for the NAS return
value: Returns readLength if the file is read successfully, otherwise returns zero.
Note: The main process is as follows:
It first checks if fileName and buffer are not NULL, and if nas_config is NULL, and retrieves the NAS configuration.
It constructs the full file path based on the file name and archive prefix.If the file nme is "global_barrier_records",it hs a special handing. It checks if the obtained configuration is NULL. If it is, the
function errors out
It creates a file path based on the filename and the prefix in the configuration.
It checks if the file path exists. If it doesn’t, it creates it.
It opens the file for writing. If it can’t open the file, the function frees memory, logs an error, and returns -1.
It writes the data from the buffer to the file. If it fails to write, the function logs an error, frees memory, closes
the file, and returns -1.
It flushes the file to ensure all data has been written. If it fails to flush, the function logs an error, closes the
file, frees memory, and returns -1.
It renames the backup file to the final filename. If it fails to rename, the function logs an error, closes the file,
frees memory, and returns -1.
Finally, it closes the file and returns the number of bytes read.
date: 2023/9/17
*/
int NasWrite(const char* fileName, const char *buffer, const int bufferLength, ArchiveConfig *nas_config)
{
int ret = 0;
@ -206,7 +248,18 @@ int NasWrite(const char* fileName, const char *buffer, const int bufferLength, A
fclose(fp);
return 0;
}
/*
function name: NasDelete
description:This code takes two parameters: a filename (fileName) and a pointer
to an ArchiveConfig structure (nas_config). The main purpose of this function is to delete a file or directory. Here are the
main steps of this function:
If nas_config is not NULL, it uses it. Otherwise, it calls getArchiveConfig() to get the configuration. It checks if the obtained configuration is NULL. If it is, the function errors out.
It creates a file path based on the filename and the prefix in the configuration.
It checks if the file or directory at the file path exists using lstat(). If it doesn’t exist, it logs an error and
returns -1.
If the file path points to a directory (S_ISDIR(statbuf.st_mode)), it tries to remove the directory using rmdir().
Otherwise, it tries to delete the file using unlink().
*/
int NasDelete(const char* fileName, ArchiveConfig *nas_config)
{
int ret = 0;
@ -259,6 +312,22 @@ int NasDelete(const char* fileName, ArchiveConfig *nas_config)
* 2. filename
* 3. the prefix of filename
*/
/*
function name: GetNasFileList.
prefix:It is a pointer to a character, indicating the prefix to be searched.
This prefix can be a path, a filename, or the prefix of a filename.
nax_config:This is a pointer to the ArchiveConfig structure, which contains NAS configuration information.
description:It takes a prefix and a nas_config as parameters and returns a list of files from a NAS (Network Attached Storage) that match the given prefix.
The prefix can be a path, a filename, or the prefix of a filename.
Note: The main process is as follows:
STEP1: checks if the parameters are not NULL.
STEP2: it constructs the file path using the archive prefix from the NAS
configuration and the provided prefix.
STEP3:If the file path corresponds to a directory, it lists all files in that
directory. If the file path is not a directory, it treats the file path as a filename or the prefix of a filename and
lists all matching files in the base directory.
TIPS:this function does not sort the returned list of files
*/
static List* GetNasFileList(const char* prefix, ArchiveConfig *nas_config)
{
int ret = 0;
@ -334,8 +403,19 @@ static int CompareFileNames(const void* a, const void* b)
return strcmp(fna, fnb);
}
/*
function: SortFileList
file_list:a pointer to a List structure,which contains a list of lies.Each item in this list is a filename.
description:this function returns stored list of files.
main steps of this function:
First,it check the length of the file list,and return null if the list is null.
Second,it allocates memory for an array of character pointes with the same length as the file list.It fills the
array with the filename from the list.
Third, it use qsort to sort the array.
After sorting ,if creates a new list and fills it.
After all, it frees the memory and returns the sorted list
Tips:This function uses some PostgreSQL-specific functions, such as palloc0, pfree_ext, lappend, and pstrdup. PostgreSQL is a feature-rich free software object-relational database management system.
*/
static List* SortFileList(List* file_list)
{
int file_num;
@ -361,7 +441,20 @@ static List* SortFileList(List* file_list)
pfree_ext(files);
return result;
}
/*
function: NasList
prefix:It is a pointer to a character, indicating the prefix to be searched.
This prefix can be a path, a filename, or the prefix of a filename.
nas_config:This is a pointer to the ArchiveConfig structure, which contains NAS configuration information.
Description:returns a sorted list of files from a NAS that match the given prefix.
main steps of this function:
Firstly it checks if the nas_config is not NULL. If it is NULL, it retrieves the NAS configuration using the
getArchiveConfig function.
Secondly,it checks if the NAS configuration was successfully retrieved. If not, it reports an error.
Next, it retrieves a list of files that match the given prefix using the GetNasFileList function.
Then it sorts this list using the SortFileList function.u
Uimately, it frees the memory allocated for the unsorted list and returns the sorted list.
*/
List* NasList(const char* prefix, ArchiveConfig *nas_config)
{
List* fileNameList = NIL;
@ -387,7 +480,12 @@ List* NasList(const char* prefix, ArchiveConfig *nas_config)
fileNameListTmp = NIL;
return fileNameList;
}
/*
function: checkNASFileExist
file_path:the path of the file.
nas_config:This is a pointer to the ArchiveConfig structure, which contains NAS configuration information.
Description: checks whether a file with the given path exists on the NAS (Network Attached Storage).
*/
bool checkNASFileExist(const char* file_path, ArchiveConfig *nas_config)
{
struct stat buf;
@ -420,4 +518,4 @@ bool checkNASFileExist(const char* file_path, ArchiveConfig *nas_config)
}
return true;
}
}

32
src/gausskernel/storage/access/cbtree/cbtree.cpp
View File

@ -145,19 +145,34 @@ Datum cbtreecanreturn(PG_FUNCTION_ARGS)
{
PG_RETURN_BOOL(true);
}
/*
function name:cbtreeoptions
description:This function is part of handling the CBTree index.
Note: The main process is as follows:
STEP1:To get the default relationship options, if the `validate` parameter is true, then the `default_reloptions` function will validate the validity of the options.
STEP2:If filledOption is null,return Datum(0),else return Datum(filledOption)
date: 2023/9/17
*/
Datum cbtreeoptions(PG_FUNCTION_ARGS)
{
Datum indexRelOptions = PG_GETARG_DATUM(0);
bool validate = PG_GETARG_BOOL(1);
bytea *filledOption = default_reloptions(indexRelOptions, validate, RELOPT_KIND_CBTREE);
bytea *filledOption = default_reloptions(indexRelOptions, validate, RELOPT_KIND_CBTREE);//To get the default relationship options, if the `validate` parameter is true, then the `default_reloptions` function will validate the validity of the options.
if (filledOption != NULL)
PG_RETURN_BYTEA_P(filledOption);
PG_RETURN_NULL();
}
/*
function name:cbtreegettuple
description:This function is part of handling the CBTree index.
Note: The main process is as follows:
STEP1:The function first checks whether the scan parameter is NULL. If it is, it will report an error and exit.
STEP2:The function calls the _bt_gettuple_internal function to get the tuple.And returns a boolean value indicating whether the tuple was successfully obtained.
STEP3:The cbtreegettuple function returns the result of the _bt_gettuple_internal function
date: 2023/9/17
*/
Datum cbtreegettuple(PG_FUNCTION_ARGS)
{
IndexScanDesc scan = (IndexScanDesc)PG_GETARG_POINTER(0);
@ -182,6 +197,17 @@ Datum cbtreegettuple(PG_FUNCTION_ARGS)
* @IN param isnulls: the container to use temprarily
* @IN param transferFuncs: the transfer functions array
*/
/*
function name:InsertToBtree
description:This function is part of handling the CBTree index.
Note: The main process is as follows:
STEP1:It gets the number of rows in the batch and adds this to reltuples.
STEP2:For each row in the batch, it does the following:checking if the value is null. If it is, it sets the corresponding entry in isnulls to true. Otherwise, it sets it to false and converts the scalar value to a datum using the corresponding transfer function.
And it gets the item pointer (tid) for the row.
STEP3:It calls _bt_spool to add the tid and values to the B-tree spool.
STEP4:add buildstate.indtuples.
date: 2023/9/17
*/
static void InsertToBtree(VectorBatch *vecScanBatch, BTBuildState &buildstate, IndexInfo *indexInfo, double &reltuples,
Datum *values, bool *isnulls, ScalarToDatum *transferFuncs)
{

35
src/gausskernel/storage/access/common/heaptuple.cpp
View File

@ -111,7 +111,7 @@ Size heap_compute_data_size(TupleDesc tupleDesc, Datum *values, const bool *isnu
if (isnull[i]) {
continue;
}
//if the current value is null,we skip getting its imformation.
val = values[i];
if (ATT_IS_PACKABLE(att[i]) && VARATT_CAN_MAKE_SHORT(DatumGetPointer(val))) {
@ -138,6 +138,16 @@ Size heap_compute_data_size(TupleDesc tupleDesc, Datum *values, const bool *isnu
*
* NOTE: it is now REQUIRED that the caller have pre-zeroed the data area.
*/
/*
*
TupleDesc tupleDesc: A tuple descriptor that contains information about the tuple’s attributes.
Datum *values: An array containing the data values to be stored in the tuple.
const bool *isnull: A boolean array indicating whether the corresponding data values are null.
char *data: A pointer to the memory area where the tuple data is to be stored.
Size data_size: The size of the memory area available for storing the tuple data.
uint16 *infomask: A pointer to an information mask used to store some state information about the tuple.
bits8 *bit: A bitmap used to indicate which data values are null
*/
void heap_fill_tuple(TupleDesc tupleDesc, Datum *values, const bool *isnull, char *data, Size data_size,
uint16 *infomask, bits8 *bit)
{
@ -161,7 +171,11 @@ void heap_fill_tuple(TupleDesc tupleDesc, Datum *values, const bool *isnull, cha
bitP = NULL;
bitmask = 0;
}
/*
The key here is understanding the use of pointers and bit masks.
Pointers are typically used for accessing and manipulating memory,
while bit masks are typically used for manipulating specific bits of binary data.
*/
*infomask &= ~(HEAP_HASNULL | HEAP_HASVARWIDTH | HEAP_HASEXTERNAL);
for (i = 0; i < numberOfAttributes; i++) {
@ -184,7 +198,6 @@ void heap_fill_tuple(TupleDesc tupleDesc, Datum *values, const bool *isnull, cha
*bitP |= bitmask;
}
/*
* XXX we use the att_align macros on the pointer value itself, not on
* an offset. This is a bit of a hack.
@ -225,6 +238,12 @@ void heap_fill_tuple(TupleDesc tupleDesc, Datum *values, const bool *isnull, cha
rc = memcpy_s(data, remain_length, val, data_length);
securec_check(rc, "\0", "\0");
}
/*
* When the length of the attribute (att[i]->attlen) is -1, it indicates that the attribute is a variable-length type.
* For such types of attributes, the function checks whether it is externally stored (VARATT_IS_EXTERNAL),
* whether it is a short varlena (VARATT_IS_SHORT), or whether it can be converted to a short varlena (VARLENA_ATT_IS_PACKABLE and VARATT_CAN_MAKE_SHORT).
* Then, depending on the different situations, it adopts different ways to copy the data.
*/
} else if (att[i]->attlen == -2) {
/* cstring ... never needs alignment */
*infomask |= HEAP_HASVARWIDTH;
@ -623,24 +642,28 @@ HeapTuple heap_copytuple(HeapTuple tuple)
if (!HeapTupleIsValid(tuple) || tuple->t_data == NULL) {
return NULL;
}
}//If the input HeapTuple is invalid or its data part is empty, it returns NULL.
Assert(!HEAP_TUPLE_IS_COMPRESSED(tuple->t_data));
// Assert that the data portion of the input HeapTuple is not compressed.
newTuple = (HeapTuple)palloc(HEAPTUPLESIZE + tuple->t_len);
// Allocate a new memory chunk to store a new HeapTuple.
newTuple->tupTableType = HEAP_TUPLE;
newTuple->t_len = tuple->t_len;
newTuple->t_self = tuple->t_self;
newTuple->t_tableOid = tuple->t_tableOid;
newTuple->t_bucketId = tuple->t_bucketId;
HeapTupleCopyBase(newTuple, tuple);
//The size of the new memory should be equal to the sum of the fixed part and variable-length data part of the HeapTuple.
#ifdef PGXC
newTuple->t_xc_node_id = tuple->t_xc_node_id;
#endif
// Set the pointer to the data portion of the new HeapTuple.
newTuple->t_data = (HeapTupleHeader)((char *)newTuple + HEAPTUPLESIZE);
rc = memcpy_s((char *)newTuple->t_data, tuple->t_len, (char *)tuple->t_data, tuple->t_len);
securec_check(rc, "\0", "\0");
//check the function memcpy_s is executed correctly.if it runs error,the function report the error.
return newTuple;
}

67
src/gausskernel/storage/access/heap/heapam.cpp
View File

@ -154,6 +154,14 @@ static bool DoesMultiXactIdConflict(MultiXactId multi, LockTupleMode lockmode);
* initscan - scan code common to heap_beginscan and heap_rescan
* ----------------
*/
/*
*Function name:InitScanBlocks
*scan:This is a descriptor for the heap table scan. It contains information about the relation being scanned, the snapshot to use, and other scan parameters.
* rangeScanInRedis:This is a structure that indicates whether the scan is a range scan in Redis and contains information about the total number of slices (sliceTotal) and the index of the current slice (sliceIndex).
*Description:This code primarily initializes the block range of a HeapScanDesc (heap scan descriptor).
*First, it determines the number of blocks (nblocks) that need to be scanned.
*This value can be determined once at the start of the scan, as any tuples added during the scan process would be invisible to the current snapshot anyway.
*/
static inline void InitScanBlocks(HeapScanDesc scan, RangeScanInRedis rangeScanInRedis)
{
BlockNumber nblocks;
@ -174,6 +182,8 @@ static inline void InitScanBlocks(HeapScanDesc scan, RangeScanInRedis rangeScanI
nblocks = InvalidBlockNumber;
} else if (scan->rs_parallel != NULL && scan->rs_parallel->isplain) {
nblocks = scan->rs_parallel->phs_nblocks;
//If it's not a partitioned table, but the scan is parallel and is a plain scan,
//then it will use the number of blocks from the parallel scan structure.
}
else {
nblocks = RelationGetNumberOfBlocks(scan->rs_base.rs_rd);
@ -181,12 +191,13 @@ static inline void InitScanBlocks(HeapScanDesc scan, RangeScanInRedis rangeScanI
if (nblocks > 0 && rangeScanInRedis.isRangeScanInRedis) {
ItemPointerData start_ctid;
ItemPointerData end_ctid;
RelationGetCtids(scan->rs_base.rs_rd, &start_ctid, &end_ctid);
//get the start_ctid and the end_ctid
if (rangeScanInRedis.sliceTotal <= 1) {
Assert(rangeScanInRedis.sliceIndex == 0);
scan->rs_base.rs_nblocks = RedisCtidGetBlockNumber(&end_ctid) - RedisCtidGetBlockNumber(&start_ctid) + 1;
scan->rs_base.rs_startblock = RedisCtidGetBlockNumber(&start_ctid);
//calculate the start block and the number of blocks for the scan if the rangeScanInRedis.sliceTotal<=1
} else {
ItemPointer sctid = eval_redis_func_direct_slice(&start_ctid, &end_ctid, true,
rangeScanInRedis.sliceTotal,
@ -197,6 +208,7 @@ static inline void InitScanBlocks(HeapScanDesc scan, RangeScanInRedis rangeScanI
scan->rs_base.rs_startblock = RedisCtidGetBlockNumber(sctid);
scan->rs_base.rs_nblocks = RedisCtidGetBlockNumber(ectid) - scan->rs_base.rs_startblock + 1;
}
//calculate the start and end CTID for each slice
ereport(LOG, (errmsg("start block is %d, nblock is %d, start_ctid is %d, end_ctid is %d, sliceTotal is %d, "
"sliceIndex is %d", scan->rs_base.rs_startblock, scan->rs_base.rs_nblocks, RedisCtidGetBlockNumber(&start_ctid),
RedisCtidGetBlockNumber(&end_ctid), rangeScanInRedis.sliceTotal, rangeScanInRedis.sliceIndex)));
@ -7425,6 +7437,10 @@ static TM_Result heap_lock_updated_tuple(Relation rel, HeapTuple tuple, ItemPoin
* tuple is an in-memory tuple structure containing the data to be written
* over the target tuple. Also, tuple->t_self identifies the target tuple.
*/
/*
* This function, heap_inplace_update, is used to update a tuple in-place within
a OpenGauss database.
*/
void heap_inplace_update(Relation relation, HeapTuple tuple, bool waitFlush)
{
Buffer buffer;
@ -7448,9 +7464,11 @@ void heap_inplace_update(Relation relation, HeapTuple tuple, bool waitFlush)
}
buffer = ReadBuffer(relation, ItemPointerGetBlockNumber(&(tuple->t_self)));
//Reads the buffer: It reads the buffer that contains the block where the tuple to be updated is located
LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
//Locks the buffer: It locks the buffer exclusively to prevent other processes from accessing it during the update
page = (Page)BufferGetPage(buffer);
// Get the pages from the Buffer
offnum = ItemPointerGetOffsetNumber(&(tuple->t_self));
maxoff = PageGetMaxOffsetNumber(page);
if (maxoff >= offnum) {
@ -7460,7 +7478,7 @@ void heap_inplace_update(Relation relation, HeapTuple tuple, bool waitFlush)
if (maxoff < offnum || !ItemIdIsNormal(lp)) {
ereport(ERROR, (errcode(ERRCODE_DATA_CORRUPTED), errmsg("heap_inplace_update: invalid lp")));
}
//Retrieve the offset of the tuple on the page and check its validity.
htup = (HeapTupleHeader)PageGetItem(page, lp);
oldlen = ItemIdGetLength(lp) - htup->t_hoff;
@ -7471,14 +7489,15 @@ void heap_inplace_update(Relation relation, HeapTuple tuple, bool waitFlush)
/* NO EREPORT(ERROR) from here till changes are logged */
START_CRIT_SECTION();
//This is a critical section that includes operations for updating tuple data and marking the buffer as dirty.
rc = memcpy_s((char*)htup + htup->t_hoff, newlen, (char*)tuple->t_data + tuple->t_data->t_hoff, newlen);
securec_check(rc, "\0", "\0");
MarkBufferDirty(buffer);
XLogRecPtr recptr = InvalidXLogRecPtr;
//This line of code defines a log record pointer and initializes it with an invalid value.
//This pointer will be used later in the code to record the log position.
/* XLOG stuff */
if (RelationNeedsWAL(relation)) {
xl_heap_inplace xlrec;
@ -7495,7 +7514,7 @@ void heap_inplace_update(Relation relation, HeapTuple tuple, bool waitFlush)
PageSetLSN(page, recptr);
}
//If Write-Ahead Logging (WAL) is enabled for the relation, it constructs a WAL record for the update and inserts it into the WAL buffer
END_CRIT_SECTION();
UnlockReleaseBuffer(buffer);
@ -8861,14 +8880,24 @@ static void heap_xlog_visible(XLogReaderState* record)
UnlockReleaseBuffer(vmbuffer.buf);
}
}
/*
* Function name:heap_bcm_redo
* xlrec:This is a pointer to the xl_heap_bcm structure, which contains information about the Block Change Map (BCM),
such as block numbers, statuses, and counts.
* node:This is a relation file node, which identifies a relation (i.e., a table) within the database. In PostgreSQL and openGauss, each table is composed of one or more files,
each with an associated relation file node
* lsn:This is a Log Sequence Number (LSN), which identifies a position within the log. During database recovery, LSN is used to determine which operations need to be redone.
* Description:This function is used to redo modifications to the Block Change Map (BCM) during the database recovery process.
* The BCM is a data structure used to track which blocks in the database have been modified, allowing for efficient checking of these blocks during database crash recovery.
*/
void heap_bcm_redo(xl_heap_bcm* xlrec, RelFileNode node, XLogRecPtr lsn)
{
int col = xlrec->col;
Relation reln = CreateFakeRelcacheEntry(node);
Buffer bcmbuffer = InvalidBuffer;
//If the number of columns is greater than 0, it indicates a column-stored table.
//In this code block, the function processes each BCM for every column.
if (col > 0) { /* cloumn store */
BlockNumber curBcmBlock = 0;
BlockNumber nextBcmBlock = 0;
@ -8910,13 +8939,15 @@ void heap_bcm_redo(xl_heap_bcm* xlrec, RelFileNode node, XLogRecPtr lsn)
nextBcmBlock = HEAPBLK_TO_BCMBLOCK(xlrec->block + i);
} while (i < xlrec->count);
UnlockReleaseBuffer(bcmbuffer);
UnlockReleaseBuffer(bcmbuffer);//Unlock and release the buffer.
} else { /* row store */
BCM_pin(reln, xlrec->block, &bcmbuffer);
//Get the particular BCM buffer.
LockBuffer(bcmbuffer, BUFFER_LOCK_EXCLUSIVE);
if (!XLByteLE(lsn, PageGetLSN(BufferGetPage(bcmbuffer)))) {
BCMSetStatusBit(reln, xlrec->block, bcmbuffer, xlrec->status, col);
}
}//Judge the lsn to ensure that only after the modifications to BCM have been recorded in the log
//will they be redone during the recovery process.
UnlockReleaseBuffer(bcmbuffer);
}
@ -9771,20 +9802,32 @@ Partition partitionOpen(Relation relation, Oid partition_id, LOCKMODE lockmode,
}
return p;
}
/*
* Functiion name: TrySubPartitionOidGetPartition
* rel：a relation that represents the partitioned table
* subPartOid：the OID of the sub-partition to be retrieved
* lockmode：the lock mode to be applied when opening the partition
* Description:This function is used to get a sub-partition from a partitioned table in a PostgreSQL database
*/
static Partition TrySubPartitionOidGetPartition(Relation rel, Oid subPartOid, LOCKMODE lockmode)
{
Oid parentOid = partid_get_parentid(subPartOid);
Assert(rel->rd_id == partid_get_parentid(parentOid));
//It retrieves the parent partition OID of the given sub-partition OID using the partid_get_parentid function.
//And it asserts that the relation’s OID is equal to the parent partition’s OID.
Partition part = tryPartitionOpen(rel, parentOid, lockmode);
//It try to open the parent partition with the given lock mode using the tryPartitionOpen function.
if (part == NULL) {
return NULL;
}
Relation partRel = partitionGetRelation(rel, part);
//If the parent partition is successfully opened, it gets the relation of the parent partition using the partitionGetRelation function.
Partition subPart = tryPartitionOpen(partRel, subPartOid, lockmode);
//The function then tries to open the sub-partition with the given lock mode.
releaseDummyRelation(&partRel);
//It free the dummy relation of the parent partition using releaseDummyRelation.
partitionClose(rel, part, NoLock);
//It closes the parent partition using partitionClose and returns the sub-partition.
return subPart;
}

14
src/gausskernel/storage/access/heap/heapam_visibility.cpp
View File

@ -1018,7 +1018,7 @@ static bool HeapTupleSatisfiesMVCC(HeapTuple htup, Snapshot snapshot, Buffer buf
bool visible = false;
TransactionIdStatus hintstatus;
Page page = BufferGetPage(buffer);
//check the basic information of heaptuple
ereport(DEBUG1,
(errmsg("HeapTupleSatisfiesMVCC self(%d,%d) ctid(%d,%d) cur_xid %ld xmin %ld"
" xmax %ld csn %lu",
@ -1093,17 +1093,25 @@ static bool HeapTupleSatisfiesMVCC(HeapTuple htup, Snapshot snapshot, Buffer buf
LatestTransactionStatusError(HeapTupleHeaderGetXmin(page, tuple),
snapshot,
"HeapTupleSatisfiesMVCC set HEAP_XMIN_INVALID xid don't abort");
//If LatestFetchCSNDidAbort() does not return true, then it calls the function LatestTransactionStatusError().
//This function reports an error with the message “HeapTupleSatisfiesMVCC set HEAP_XMIN_INVALID xid don’t abort”.
//The parameters for this function include: Xmin obtained from the page and tuple
//(HeapTupleHeaderGetXmin(page, tuple)), and the current snapshot .
SetHintBits(tuple, buffer, HEAP_XMIN_INVALID, InvalidTransactionId);
}
if (!visible) {
//This code first checks if the variable visible is false.
//If visible is false, it performs the following operations
if (!GTM_LITE_MODE || u_sess->attr.attr_common.xc_maintenance_mode ||
snapshot->gtm_snapshot_type != GTM_SNAPSHOT_TYPE_LOCAL ||
!IsXidVisibleInGtmLiteLocalSnapshot(HeapTupleHeaderGetXmin(page, tuple), snapshot, hintstatus,
HeapTupleHeaderGetXmin(page, tuple) == HeapTupleHeaderGetXmax(page, tuple), buffer, NULL)) {
return false;
}
//It checks if it is in GTM_LITE_MODE, or if the session’s attribute u_sess->attr.attr_common.xc_maintenance_mode is true,
//or if the snapshot type snapshot->gtm_snapshot_type is not equal to GTM_SNAPSHOT_TYPE_LOCAL.
//If any of these conditions are true, then the code returns false
}
}
} else {
@ -1286,7 +1294,7 @@ HTSV_Result HeapTupleSatisfiesVacuum(HeapTuple htup, TransactionId OldestXmin, B
*/
if (!HeapTupleHeaderXminCommitted(tuple)) {
if (HeapTupleHeaderXminInvalid(tuple))
return HEAPTUPLE_DEAD;
return HEAPTUPLE_DEAD;
xidstatus = TransactionIdGetStatus(HeapTupleGetRawXmin(htup));
if (TransactionIdIsCurrentTransactionId(HeapTupleGetRawXmin(htup))) {
if (tuple->t_infomask & HEAP_XMAX_INVALID) /* xid invalid */

56
src/gausskernel/storage/access/heap/rewriteheap.cpp
View File

@ -330,25 +330,41 @@ RewriteState begin_heap_rewrite(Relation old_heap, Relation new_heap, Transactio
return state;
}
/*
* Function name:rewrite_write_one_page
* Description:This function is mainly used in the data page rewriting process
* of a database system. After a data page is modified, it needs to write back
* the modified data page to disk. In this process, it may involve some special
* operations such as encryption, logging, etc.
*/
static void rewrite_write_one_page(RewriteState state, Page page)
{
TdeInfo tde_info = {0};
if (RelationisEncryptEnable(state->rs_new_rel)) {
GetTdeInfoFromRel(state->rs_new_rel, &tde_info);
}
/*
* If the new relation (`state->rs_new_rel`) has encryption enabled,
* it gets the TDE information from this relation.
*/
if (IsSegmentFileNode(state->rs_new_rel->rd_node)) {
//checks if the new relation is a segment file node
Assert(state->rs_use_wal);
Buffer buf = ReadBuffer(state->rs_new_rel, P_NEW);
//reads a buffer of the new relation
#ifdef USE_ASSERT_CHECKING
BufferDesc *buf_desc = GetBufferDescriptor(buf - 1);
Assert(buf_desc->tag.blockNum == state->rs_blockno);
#endif
#endif
//check the rs_blockno if the USE_ASSERT_CHEKING is defined.
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
XLogRecPtr xlog_ptr = log_newpage(&state->rs_new_rel->rd_node, MAIN_FORKNUM, state->rs_blockno, page, true,
&tde_info);
//locks this buffer
errno_t rc = memcpy_s(BufferGetBlock(buf), BLCKSZ, page, BLCKSZ);
//copier the page data to the buffer block
securec_check(rc, "\0", "\0");
PageSetLSN(BufferGetPage(buf), xlog_ptr);
PageSetLSN(BufferGetPage(buf), xlog_ptr);//sets the LSN of the buffer block
MarkBufferDirty(buf);
UnlockReleaseBuffer(buf);
} else {
@ -358,19 +374,20 @@ static void rewrite_write_one_page(RewriteState state, Page page)
if (state->rs_use_wal) {
log_newpage(&state->rs_new_rel->rd_node, MAIN_FORKNUM, state->rs_blockno, page, true, &tde_info);
}
// if the WAL is used,it logs a new page.
RelationOpenSmgr(state->rs_new_rel);
//open the storage manager of the new relation.
char *bufToWrite = NULL;
if (RelationisEncryptEnable(state->rs_new_rel)) {
bufToWrite = PageDataEncryptIfNeed(page, &tde_info, true);
} else {
bufToWrite = page;
}
//it needs to encrypt the page data when the new relation has encryption enabled.
PageSetChecksumInplace((Page)bufToWrite, state->rs_blockno);
//set page checksum in the correct location .
rewrite_flush_page(state, (Page)bufToWrite);
//flushes the page state
}
}
@ -746,7 +763,12 @@ static void prepare_cmpr_buffer(RewriteState state, Size meta_size, const char *
typedef void (*insert_tuple_func)(RewriteState state, HeapTuple tuple);
/* Insert a tuple to the new relation after compression. */
/*
* Function name:cmpr_heap_insert
* Description:This function is mainly used in the data page rewriting process of a database system.
* After a data page is modified, it needs to write back the modified data page to disk.
* In this process, it may involve some special operations such as compression, transaction handling, etc.
*/
static void cmpr_heap_insert(RewriteState state, HeapTuple tup)
{
Page page = state->rs_cmprBuffer;
@ -761,7 +783,7 @@ static void cmpr_heap_insert(RewriteState state, HeapTuple tup)
xmax = HeapTupleGetRawXmax(tup);
rewrite_page_prepare_for_xid(page, xmin, false);
(void)rewrite_page_prepare_for_xid(page, xmax, (tup->t_data->t_infomask & HEAP_XMAX_IS_MULTI) ? true : false);
//It gets the xmin and xmax transaction IDs of the tuple and prepares the page to handle these transaction IDs
HeapTupleCopyBaseFromPage(tup, page);
HeapTupleSetXmin(tup, xmin);
HeapTupleSetXmax(tup, xmax);
@ -771,9 +793,9 @@ static void cmpr_heap_insert(RewriteState state, HeapTuple tup)
ereport(ERROR,
(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), errmsg("row is too big: size %lu, maximum size %lu",
(unsigned long)len, (unsigned long)MaxHeapTupleSize)));
//It checks if the length of the tuple exceeds the maximum heap tuple size. If it does, it reports an error.
Assert(PageIsCompressed(page));
//ensure page has been compressed
/* And now we can insert the tuple into the page */
newoff = PageAddItem(page, (Item)tup->t_data, tup->t_len, InvalidOffsetNumber, false, true);
Assert(newoff != InvalidOffsetNumber);
@ -781,7 +803,7 @@ static void cmpr_heap_insert(RewriteState state, HeapTuple tup)
/* Update caller's t_self to the actual position where it was stored */
ItemPointerSet(&(tup->t_self), state->rs_blockno, newoff);
/* Now Insert the correct position into CTID of the stored tuple too. */
// If the CTID of the stored tuple is invalid, it updates the CTID of the stored tuple on the page to its actual location.
if (!ItemPointerIsValid(&tup->t_data->t_ctid)) {
ItemId newitemid = PageGetItemId(page, newoff);
HeapTupleHeader onpage_tup = (HeapTupleHeader)PageGetItem(page, newitemid);
@ -1167,7 +1189,11 @@ static void rewrite_end_flush_page(RewriteState state)
}
#endif
}
/*
* This function is mainly used to insert tuples during heap rewrite and handles some special cases,
* such as TOAST values and tuples greater than the threshold.
* At the same time, it also handles some error situations and ensures data integrity
*/
/*
* Insert a tuple to the new relation. This has to track heap_insert
* and its subsidiary functions!
@ -1191,6 +1217,8 @@ static void raw_heap_insert(RewriteState state, HeapTuple tup)
ereport(DEBUG5, (errmodule(MOD_TBLSPC), errmsg("tuple is null")));
return;
}
//If tup is not empty, it checks if tup is a heap tuple. If tup is empty,
// it reports an error at DEBUG5 level
/*
* If the new tuple is too big for storage or contains already toasted
@ -1230,7 +1258,7 @@ static void raw_heap_insert(RewriteState state, HeapTuple tup)
state->rs_buffer_valid = false;
}
}
//If the buffer is invalid, it initializes the page and sets some basic attributes
if (!state->rs_buffer_valid) {
HeapPageHeader phdr = (HeapPageHeader)page;
/* Initialize a new empty page */
@ -1249,7 +1277,7 @@ static void raw_heap_insert(RewriteState state, HeapTuple tup)
PageSetTDE(page);
}
}
// prepares the page to get xmin and xmax
xmin = HeapTupleGetRawXmin(heaptup);
xmax = HeapTupleGetRawXmax(heaptup);
rewrite_page_prepare_for_xid(page, xmin, false);

23
src/gausskernel/storage/access/transam/csnlog.cpp
View File

@ -122,13 +122,18 @@ void CSNLogSetCommitSeqNo(TransactionId xid, int nsubxids, TransactionId *subxid
csn <= COMMITSEQNO_ABORTED) {
return;
}
//if csn is invlid or xid is eqaul to BootstrapTransactionId which is typically used to identify transactions
//during the system startup or initialization process
if (csn == InvalidCommitSeqNo || xid == BootstrapTransactionId) {
if (IsBootstrapProcessingMode())
csn = COMMITSEQNO_FROZEN;
//if the system is currently in bootstrao processing mode
//then the CSN will be ste to a frozen state.
else
ereport(ERROR, (errcode(ERRCODE_IO_ERROR),
errmsg("cannot mark transaction %lu committed without CSN %lu", xid, csn)));
//else the code will report a error which means that the XID can not be set as committed
//without CSN.
}
/*
@ -138,14 +143,20 @@ void CSNLogSetCommitSeqNo(TransactionId xid, int nsubxids, TransactionId *subxid
*/
pageno = TransactionIdToCSNPage(xid);
for (;;) {
for (;;) {//it deal with the subxids in the unlimited loop.and log the CSN of each
//sub-transaction to the corresponding log page.
int num_on_page = 0;
//it iterates over the array of sub-transaction ID,incrementing the counter for each
// whose CSN page number matches the current page number (pageno), and continues to
// process the next sub-transaction ID
while (i < nsubxids && (int64)TransactionIdToCSNPage(subxids[i]) == pageno) {
num_on_page++;
i++;
}
//When it encounters a new CSN page number or has iterated over all sub-transaction IDs,
//the code calls the CSNLogSetPageStatus function to log the CSN of all transactions on
//the current page. Then, if there are still unprocessed sub-transaction IDs, the code updates the page number and transaction ID and continues processing in the next loop iteration
CSNLogSetPageStatus(xid, num_on_page, subxids + offset, csn, pageno, topxid);
if (i >= nsubxids) {
break;
@ -155,9 +166,13 @@ void CSNLogSetCommitSeqNo(TransactionId xid, int nsubxids, TransactionId *subxid
pageno = TransactionIdToCSNPage(subxids[offset]);
xid = InvalidTransactionId;
}
//Generally ,this code seeks to ensure all sub-transaction id have been logged in the CSN log
if (IS_DISASTER_RECOVER_MODE && COMMITSEQNO_IS_COMMITTED(csn)) {
UpdateXLogMaxCSN(csn);
}
//ultimately,this function check the system is on the disaster_recover condition
//or other serious condition.SO the system will attempt to recover to a consistent
//state to operate normally.
}
/**

7
src/gausskernel/storage/access/transam/xact.cpp
View File

@ -8094,8 +8094,13 @@ CommitSeqNo SetXact2CommitInProgress(TransactionId xid, CommitSeqNo csn)
return InvalidCommitSeqNo;
nchildren = xactGetCommittedChildren(&children);
// get the number of subTransaction.
CSNLogSetCommitSeqNo(xid, nchildren, children, COMMITSEQNO_COMMIT_INPROGRESS | latestCSN);
/*
* Record the status and CSN of transaction entries in the transaction commit log of the
* transaction and its child transaction tree. Be careful to ensure that this operation * is as efficient and atomic
* as possible..
*/
ereport(
DEBUG1, (errmsg("Set %lu to commit in progress, latest csn is %lu", xid, latestCSN)));
return latestCSN;

98
src/gausskernel/storage/mot/core/concurrency_control/occ_transaction_manager.cpp
View File

@ -59,14 +59,24 @@ bool OccTransactionManager::Init()
bool result = true;
return result;
}
/*
* CheckVersion() ---
* check transaction versions to maintain the consistency and concurrency control of the database
* Param [IN] access：contains information about the transaction, including its version number or timestamp
* Returns [OUT] : bool.
*/
bool OccTransactionManager::CheckVersion(const Access* access)
{
// We always validate on committed rows!
const Row* row = access->GetRowFromHeader();
return (row->m_rowHeader.GetCSN() == access->m_tid);
}
/*
* QuickHeaderValidation() ---
* perform a quick header validation. It takes a pointer to an Access object as a parameter
* Param [IN] access：contains information about the transaction, including its version number or timestamp
* Returns [OUT] : bool,returns a boolean value indicating whether the quick header validation has passed.
*/
bool OccTransactionManager::QuickHeaderValidation(const Access* access)
{
if (access->m_type != INS) {
@ -103,7 +113,13 @@ bool OccTransactionManager::QuickHeaderValidation(const Access* access)
return true;
}
/*
* ValidateReadSet() ---
* validate the read set of a transaction to ensure its validity and thereby maintain data consistency
* Param [IN] txMan： a pointer to a transaction manager (TxnManager*). This pointer is used to access
* transaction-related information for the purpose of validating the data items in the read set.
* Returns [OUT] : bool,returns a boolean value indicating whether the quick header validation has passed.
*/
bool OccTransactionManager::ValidateReadSet(TxnManager* txMan)
{
TxnOrderedSet_t& orderedSet = txMan->m_accessMgr->GetOrderedRowSet();
@ -119,7 +135,14 @@ bool OccTransactionManager::ValidateReadSet(TxnManager* txMan)
return true;
}
/*
* ValidateWriteSet() ---
* validate the write set.
* Param [IN] txMan： representing the transaction manager that needs to validate the write set
* through this parameter, the function gains access to information and access objects within the transaction manager,
* allowing it to validate the validity of data items in the write set
* Returns [OUT] : bool,returns a boolean value indicating whether the write set valid
*/
bool OccTransactionManager::ValidateWriteSet(TxnManager* txMan)
{
TxnOrderedSet_t& orderedSet = txMan->m_accessMgr->GetOrderedRowSet();
@ -135,7 +158,15 @@ bool OccTransactionManager::ValidateWriteSet(TxnManager* txMan)
}
return true;
}
/*
* LockRows() ---
* lock rows in the database
* Param [IN] txMan： representing the transaction manager that needs to validate the write set
* through this parameter, the function gains access to information and access objects within the transaction manager,
* allowing it to validate the validity of data items in the write set。
* Param [IN] numRowsLock： record the number of locked rows.
* Returns [OUT] : indicate the success of the operation.
*/
RC OccTransactionManager::LockRows(TxnManager* txMan, uint32_t& numRowsLock)
{
RC rc = RC_OK;
@ -156,7 +187,15 @@ RC OccTransactionManager::LockRows(TxnManager* txMan, uint32_t& numRowsLock)
return rc;
}
/*
* LockHeadersNoWait() ---
* try to lock the information of header
* Param [IN] txMan： representing the transaction manager that needs to validate the write set
* through this parameter, the function gains access to information and access objects within the transaction manager,
* allowing it to validate the validity of data items in the write set。
* Param [IN] numRowsLock： record the number of locked rows.
* Returns [OUT] : indicate the success of the operation.
*/
bool OccTransactionManager::LockHeadersNoWait(TxnManager* txMan, uint32_t& numSentinelsLock)
{
uint64_t sleepTime = 1;
@ -210,7 +249,15 @@ bool OccTransactionManager::LockHeadersNoWait(TxnManager* txMan, uint32_t& numSe
return true;
}
/*
* LockHeaders() ---
* try to lock the information of header
* Param [IN] txMan： representing the transaction manager that needs to validate the write set
* through this parameter, the function gains access to information and access objects within the transaction manager,
* allowing it to validate the validity of data items in the write set。
* Param [IN] numRowsLock： record the number of locked rows.
* Returns [OUT] : indicate the success of the operation.
*/
RC OccTransactionManager::LockHeaders(TxnManager* txMan, uint32_t& numSentinelsLock)
{
RC rc = RC_OK;
@ -245,7 +292,14 @@ RC OccTransactionManager::LockHeaders(TxnManager* txMan, uint32_t& numSentinelsL
final:
return rc;
}
/*
* PreAllocStableRow() ---
* The typical purpose of the PreAllocStableRow function is to preallocate memory for stable (persistent) row data structures within a database transaction.
* Param [IN] txMan： representing the transaction manager that needs to validate the write set
* through this parameter, the function gains access to information and access objects within the transaction manager,
* allowing it to validate the validity of data items in the write set.
* Returns [OUT] ：indicate the success of the operation.
*/
bool OccTransactionManager::PreAllocStableRow(TxnManager* txMan)
{
if (GetGlobalConfiguration().m_enableCheckpoint) {
@ -268,7 +322,14 @@ bool OccTransactionManager::PreAllocStableRow(TxnManager* txMan)
}
return true;
}
/*
* QuickVersionCheck() ---
* perform a quick version check to determine if a transaction can proceed
* Param [IN] txMan： representing the transaction manager that needs to validate the write set
* through this parameter, the function gains access to information and access objects within the transaction manager,
* allowing it to validate the validity of data items in the write set.
* Returns [OUT] ：indicate the success of the operation.
*/
bool OccTransactionManager::QuickVersionCheck(TxnManager* txMan, uint32_t& readSetSize)
{
int isolationLevel = txMan->GetTxnIsoLevel();
@ -311,7 +372,14 @@ bool OccTransactionManager::QuickVersionCheck(TxnManager* txMan, uint32_t& readS
}
return true;
}
/*
* ValidateOcc() ---
* validate whether a transaction in the OCC (Optimistic Concurrency Control) concurrency control protocol can proceed
* Param [IN] txMan： representing the transaction manager that needs to validate the write set
* through this parameter, the function gains access to information and access objects within the transaction manager,
* allowing it to validate the validity of data items in the write set.
* Returns [OUT] RC：indicate the success of the operation.
*/
RC OccTransactionManager::ValidateOcc(TxnManager* txMan)
{
uint32_t numSentinelLock = 0;
@ -379,8 +447,14 @@ final:
return rc;
}
void OccTransactionManager::RollbackInserts(TxnManager* txMan)
/*
* RollbackInsertsc() ---
* rollback insert operations that have been executed but not yet committed
* Param [IN] txMan： representing the transaction manager that needs to validate the write set through this
* parameter, the function gains access to information and access objects within the transaction manager, allowing it to
* validate the validity of data items in the write set. Returns [OUT] RC：indicate the success of the operation.
*/
void OccTransactionManager::RollbackInserts(TxnManager *txMan)
{
return txMan->UndoInserts();
}

52
src/gausskernel/storage/mot/core/storage/index/masstree/mot_masstree_kvthread.cpp
View File

@ -39,9 +39,17 @@
// This is the thread info which serves the current masstree operation. It is set before the operation starts.
__thread threadinfo* mtSessionThreadInfo = nullptr;
/**
* @class threadinfo
* @details Maintains key-value thread information
*/
volatile mrcu_epoch_type globalepoch;
/**
* @brief create the threadinfo subject
* @param purpose:express the purpose of the thread
* @param index
* @param rcu_max_free_count:the max number allowed to free for the RCU(read-copy-update)
*/
inline threadinfo::threadinfo(int purpose, int index, int rcu_max_free_count)
{
errno_t erc = memset_s(this, sizeof(*this), 0, sizeof(*this));
@ -53,34 +61,50 @@ inline threadinfo::threadinfo(int purpose, int index, int rcu_max_free_count)
ts_ = 2;
}
/**
* @brief create the threadinfo subject
* @param purpose
* @param index
* @param rcu_max_free_count the max number allowed to free for the RCU(read-copy-update)
* @return the pointer which point to a threadinfo subject
*/
threadinfo* threadinfo::make(void* obj_mem, int purpose, int index, int rcu_max_free_count)
{
threadinfo* ti = new (obj_mem) threadinfo(purpose, index, rcu_max_free_count);
//cerate the new threadinfo object at the given memory location
//then ,it checks whether to use the memory pool
if (use_pool()) {
void* limbo_space = ti->allocate(MAX_MEMTAG_MASSTREE_LIMBO_GROUP_ALLOCATION_SIZE, memtag_limbo);
if (!limbo_space) {
return nullptr;
}
//If memory allocation is successful, it calls the member function mark() to mark some status,
//and creates a new mt_limbo_group object on the just allocated memory space
ti->mark(tc_limbo_slots, mt_limbo_group::capacity);
ti->limbo_head_ = ti->limbo_tail_ = new (limbo_space) mt_limbo_group;
}
//regardless of whether a memory pool is used, the function returns the pointer to the newly created object
return ti;
}
/**
* @brief create the threadinfo subject
* @param sz:it represents the requested memory size
* @param tag:it is used to mark the allocated memory
* @param actual_size:return the actual allocated memory size
*/
void* threadinfo::allocate(size_t sz, memtag tag, size_t* actual_size)
{
int size = sz;
void* p = nullptr;
//first it initializes the val
if (likely(!use_pool())) {
p = cur_working_index->AllocateMem(size, tag);
} else {
p = malloc(sz + memdebug_size);
//allocate the memory which can hold the size and the memdebug_size used to carry debug information
}
//check whether to use the memory pool,if not ,call the AllocateMem
p = memdebug::make(p, sz, tag);
if (p) {
if (actual_size) {
@ -89,11 +113,18 @@ void* threadinfo::allocate(size_t sz, memtag tag, size_t* actual_size)
mark(threadcounter(tc_alloc + (tag > memtag_value)), sz);
}
return p;
//Finally, it returns the pointer to the allocated memory
}
/**
* @brief create the threadinfo subject
* @param p: This is a pointer to the memory block to be freed
* @param size_t sz:representing the size of the memory block to be freed
* @param tag: used to tag or categorize the memory to be freed
*/
void threadinfo::deallocate(void* p, size_t sz, memtag tag)
{
MOT_ASSERT(p);
// it asserts that the pointer p is not null.
p = memdebug::check_free(p, sz, tag);
if (likely(!use_pool())) {
cur_working_index->DeallocateMem(p, sz, tag);
@ -101,6 +132,7 @@ void threadinfo::deallocate(void* p, size_t sz, memtag tag)
free(p);
}
mark(threadcounter(tc_alloc + (tag > memtag_value)), -sz);
//mark some states,like threadcounter
}
void threadinfo::ng_record_rcu(void* p, int sz, memtag tag)
@ -109,6 +141,10 @@ void threadinfo::ng_record_rcu(void* p, int sz, memtag tag)
memdebug::check_rcu(p, sz, tag);
cur_working_index->RecordMemRcu(p, sz, tag);
mark(threadcounter(tc_alloc + (tag > memtag_value)), -sz);
//first use assert to check the p is not null
//second,use check_rcu to ensure the memory can be free
//third,recode the MEMRCU
//finally,mark the stats
}
void threadinfo::set_gc_session(MOT::GcManager* gc_session)

12
src/gausskernel/storage/mot/core/storage/index/masstree/mot_masstree_struct.hpp
View File

@ -48,8 +48,10 @@ public:
{
value->print(f, prefix, indent, key, initial_timestamp, suffix);
}
//the function called print will be rewrite in the sub-class.
};
// Provide a specialized value output function for the parameter of type `uint64_t`,
// to distinguish it from the previous generic template `value_print`.
template <>
class value_print<unsigned char*> {
public:
@ -59,7 +61,8 @@ public:
fprintf(f, "%s%*s%.*s = %p%s\n", prefix, indent, "", key.len, key.s, value, suffix);
}
};
//Provide a specialized value output function for the parameter of type `uint64_t`,
//to distinguish it from the previous generic template `value_print`.
template <>
class value_print<uint64_t> {
public:
@ -77,7 +80,8 @@ public:
securec_check_ss(erc, "\0", "\0");
return erc;
}
};
//The function converts the integer key value of the key into a string, and writes it into the provided buffer `buf`.
//Then, it returns the number of characters written.
typedef key_unparse_unsigned key_unparse_type;
@ -86,7 +90,7 @@ void leaf<P>::print(FILE* f, const char* prefix, int depth, int kdepth) const
{
return;
}
//it maybe rewrite in the sub-class,sounds like handling with leaf node.
template <typename P>
void internode<P>::print(FILE* f, const char* prefix, int depth, int kdepth) const
{

42
src/gausskernel/storage/mot/core/storage/index/masstree_index.h
View File

@ -209,6 +209,7 @@ public:
/**
* @brief Default constructor.
*/
MasstreePrimaryIndex()
: Index(MOT::IndexOrder::INDEX_ORDER_PRIMARY, IndexingMethod::INDEXING_METHOD_TREE),
m_leafsPool(nullptr),
@ -226,12 +227,14 @@ public:
m_initialized = false;
DestroyPools();
}
//check the pools is not initialized,or DestroyPools and set the m_minitialized=false
}
/**
* @brief Calculate the Index memory consumption.
* @return The amount of memory the Index consumes.
*/
virtual uint64_t GetIndexSize() override;
/**
@ -309,7 +312,20 @@ public:
m_internodesPool->Print("Internode pool", level);
m_ksuffixSlab->Print("Ksuffix slab", level);
}
/*
* GetLeafsPoolStats() ---
* return statistics about the leaf node pool
* The implementation of this virtual function should be provided in derived classes
* to compute statistics about the leaf node pool and populate the results into the provided reference parameters.
* Different derived classes can provide different statistics based on their specific implementations.
* This mechanism allows for polymorphism, where the correct implementation in a derived class is chosen at runtime
* to obtain the appropriate leaf node pool statistics.
* Param [IN] objSize：the basic structure of object description information.
* Param [IN] numUsedObj: address information for the object to be renamed.
* Param [IN] totalSize: name after renamed.
* Param [IN] netto: name after renamed.
* Returns [OUT] : void.
*/
virtual void GetLeafsPoolStats(uint64_t& objSize, uint64_t& numUsedObj, uint64_t& totalSize, uint64_t& netto)
{
PoolStatsSt stats = {};
@ -320,7 +336,19 @@ public:
totalSize = stats.m_poolCount * stats.m_poolGrossSize;
netto = numUsedObj * objSize;
}
/*
* GetInternodesPoolStats() ---
* retrieve statistics about the "Internodes Pool."
* The purpose of this function is to obtain statistics information from the internal nodes pool
* and return this information through the reference parameters, including object size, the
* count of objects in use, total size, and net size. These statistics are valuable for purposes
* such as performance analysis and memory management.
* Param [IN] objSize：the basic structure of object description information.
* Param [IN] numUsedObj: address information for the object to be renamed.
* Param [IN] totalSize: name after renamed.
* Param [IN] netto: name after renamed.
* Returns [OUT] : void.
*/
virtual void GetInternodesPoolStats(uint64_t& objSize, uint64_t& numUsedObj, uint64_t& totalSize, uint64_t& netto)
{
PoolStatsSt stats = {};
@ -331,7 +359,15 @@ public:
totalSize = stats.m_poolCount * stats.m_poolGrossSize;
netto = numUsedObj * objSize;
}
/*
* GetKsuffixSlabStats() ---
* retrieve statistics about the "Ksuffix Slab" and returns a pointer to a PoolStatsSt structure.
* In summary, the function's purpose is to obtain statistics about the "Ksuffix Slab" by calling
* the GetStats method of the m_ksuffixSlab object and return this information as a
* pointer to a PoolStatsSt structure. This allows external code to access and use these
* statistics for further processing or display.
* Returns [OUT] : This means that the function returns a pointer to a PoolStatsSt structure object in memory,
*/
virtual PoolStatsSt* GetKsuffixSlabStats()
{
return m_ksuffixSlab->GetStats();

11
src/gausskernel/storage/mot/core/storage/key.h
View File

@ -110,7 +110,11 @@ public:
MOT_ASSERT(newLen <= MAX_KEY_SIZE);
m_keyLen = newLen;
}
/**
* @brief copy the key
* @param buf the opinter of the data which is waiting for copy
* @param len Used to ensure the length waiting for copying
*/
inline void CpKey(const uint8_t* buf, uint16_t len)
{
MOT_ASSERT(len <= m_keyLen);
@ -162,6 +166,10 @@ public:
return true;
}
/**
* @brief copy the key
* @param key pass the information to use the other CpKey function
*/
inline void CpKey(const Key& key)
{
CpKey(key.GetKeyBuf(), key.GetKeyLength());
@ -177,6 +185,7 @@ public:
return HexStr(m_keyBuf, m_keyLen).c_str();
}
// give the way to judge the size between of the two Key
bool operator==(const Key& key) const
{
MOT_ASSERT(m_keyLen == key.GetKeyLength());