diff --git a/src/gausskernel/cbb/communication/libcomm_utils/libcomm_memory.cpp b/src/gausskernel/cbb/communication/libcomm_utils/libcomm_memory.cpp
index 388c88e83..e436a4052 100644
--- a/src/gausskernel/cbb/communication/libcomm_utils/libcomm_memory.cpp
+++ b/src/gausskernel/cbb/communication/libcomm_utils/libcomm_memory.cpp
@@ -65,7 +65,6 @@
 #define static
 #endif
 
-
 #define STREAM_SCAN_FINISH 'F'
 #define STREAM_SCAN_WAIT 'W'
 #define STREAM_SCAN_DATA 'D'
@@ -73,9 +72,12 @@
 extern bool executorEarlyStop();
 
 /* release memory of communication layer, just for LLT */
+// 此函数的功能是释放通信层的内存
 int gs_release_comm_memory()
 {
+    // 构造一个通信上下文的对象，切换到全局通信内存上下文，用于管理通信内存。
     AutoContextSwitch commContext(g_instance.comm_cxt.comm_global_mem_cxt);
+    // 调用函数gs_r_release_comm_memory来释放通信内存。
     gs_r_release_comm_memory();
     return 0;
 }
@@ -85,27 +87,41 @@ int gs_release_comm_memory()
  *
  * @param[IN] key_s: stream key
  */
-void gs_memory_init_entry(StreamSharedContext* sharedContext, int consumerNum, int producerNum)
+// 此函数用于初始化一个关于流和内存使用情况的哈希表条目
+void gs_memory_init_entry(StreamSharedContext *sharedContext, int consumerNum, int producerNum)
 {
-    struct hash_entry* entry = NULL;
-    struct hash_entry** poll_entrys = NULL;
-    struct hash_entry*** quota_entrys = NULL;
+    // 定义一个指向哈希表条目的指针
+    struct hash_entry *entry = NULL;
+    // 定义一个指向哈希表条目的指针的指针，用于存储与消费者相关的条目
+    struct hash_entry **poll_entrys = NULL;
+    // 定义一个指向哈希表条目的指针的指针的指针，用于存储与生产者相关的条目
+    struct hash_entry ***quota_entrys = NULL;
 
-    poll_entrys = (struct hash_entry**)palloc(sizeof(struct hash_entry*) * consumerNum);
-    quota_entrys = (struct hash_entry***)palloc(sizeof(struct hash_entry**) * consumerNum);
-
-    for (int i = 0; i < consumerNum; i++) {
-        entry = (struct hash_entry*)palloc(sizeof(struct hash_entry));
+    // 为每个消费者分配内存以存储哈希表条目指针
+    poll_entrys = (struct hash_entry **)palloc(sizeof(struct hash_entry *) * consumerNum);
+    // 为每个消费者分配内存以存储生产者相关的哈希表条目指针
+    quota_entrys = (struct hash_entry ***)palloc(sizeof(struct hash_entry **) * consumerNum);
+    // 遍历所有消费者
+    for (int i = 0; i < consumerNum; i++)
+    {
+        // 为每个消费者分配一个哈希表条目，并初始化它
+        entry = (struct hash_entry *)palloc(sizeof(struct hash_entry));
         (void)entry->_init();
+        // 将当前消费者的哈希表条目存储在poll_entrys中
         poll_entrys[i] = entry;
-        quota_entrys[i] = (struct hash_entry**)palloc(sizeof(struct hash_entry*) * producerNum);
-        for (int j = 0; j < producerNum; j++) {
-            entry = (struct hash_entry*)palloc(sizeof(struct hash_entry));
+        // 为每个生产者分配内存以存储哈希表条目指针
+        quota_entrys[i] = (struct hash_entry **)palloc(sizeof(struct hash_entry *) * producerNum);
+        // 遍历所有的生产者
+        for (int j = 0; j < producerNum; j++)
+        {
+            // 为每个生产者分配一个哈希表条目，并初始化它
+            entry = (struct hash_entry *)palloc(sizeof(struct hash_entry));
             (void)entry->_init();
+            // 将当前生产者的哈希表条目存储在quota_entrys中
             quota_entrys[i][j] = entry;
         }
     }
-
+    // 将poll_entrys和quota_entrys存储在共享上下文中，以便后续使用
     sharedContext->poll_entrys = poll_entrys;
     sharedContext->quota_entrys = quota_entrys;
 }
@@ -117,38 +133,51 @@ void gs_memory_init_entry(StreamSharedContext* sharedContext, int consumerNum, i
  * @param[IN] sharedContext: context for shared memory stream
  * @param[IN] nthChannel: destination consumer
  */
-void gs_message_by_memory(StringInfo buf, StreamSharedContext* sharedContext, int nthChannel)
+// 此函数的作用是通过内存发送错误/通知消息，其中参数buf为错误或者通知的字符串，sharedContext为共享内存流上下文，nthChannel为目标消费者
+void gs_message_by_memory(StringInfo buf, StreamSharedContext *sharedContext, int nthChannel)
 {
+    // 目标缓冲区，用于存储要发送的消息
     StringInfo buf_dst = NULL;
-    struct hash_entry* entry = NULL;
+    // 哈希表条目，用于管理共享内存流
+    struct hash_entry *entry = NULL;
 
-    /* Copy Error/Notice messages to shared context. */
+    // 将错误/通知消息复制到共享上下文中
     buf_dst = sharedContext->messages[nthChannel][u_sess->stream_cxt.smp_id];
 
-    /*
-     * If producer is waked up and shared buffer has been consumed while waiting,
-     * it can continue to append data to its messages of sharedContext.
-     */
+    // 如果生产者在等待期间被唤醒，并且共享缓冲区已被消耗，它可以继续将其数据追加到sharedContext的消息中
     entry = sharedContext->quota_entrys[nthChannel][u_sess->stream_cxt.smp_id];
-    while (buf_dst->len > 0) {
+    // 当目标缓冲区中还有未处理的数据时
+    while (buf_dst->len > 0)
+    {
+        // 等待一段时间，直到可以继续处理数据
         (void)entry->_timewait(SINGLE_WAITQUOTA);
     }
+    // 将源缓冲区的数据追加到目标缓冲区中
     appendBinaryStringInfo(buf_dst, buf->data, buf->len);
+    // 更新目标缓冲区的游标位置
     buf_dst->cursor = buf->cursor;
 
-    /* Send signal to dest consumer. */
+    // 向目标消费者发送信号。
     entry = sharedContext->poll_entrys[nthChannel];
+    // 发送信号通知目标消费者有新消息到达
     entry->_signal();
 
+    // 释放源缓冲区的数据内存
     pfree(buf->data);
+    // 将源缓冲区的数据指针置为NULL，避免悬挂指针
     buf->data = NULL;
 }
 
-void gs_memory_disconnect(StreamSharedContext* sharedContext, int nthChannel)
+// 此函数的作用是断开内存连接
+void gs_memory_disconnect(StreamSharedContext *sharedContext, int nthChannel)
 {
-    struct hash_entry* entry = NULL;
+    // 定义一个指向哈希表条目的指针
+    struct hash_entry *entry = NULL;
+    // 将指定通道的数据状态设置为连接错误
     sharedContext->dataStatus[nthChannel][u_sess->stream_cxt.smp_id] = CONN_ERR;
+    // 获取指定通道的轮询条目
     entry = sharedContext->poll_entrys[nthChannel];
+    // 向轮询条目发送信号，通常用于通知其他进程或线程发生了某种事件或状态变化
     entry->_signal();
 }
 
@@ -159,16 +188,27 @@ void gs_memory_disconnect(StreamSharedContext* sharedContext, int nthChannel)
  * @param[IN] sharedContext: context for shared memory stream
  * @param[IN] nthChannel: destination consumer
  */
-bool gs_is_databuff_empty(StreamSharedContext* sharedContext, int nthChannel)
+// 此函数的作用是判断数据缓冲区是否为空，其中参数sharedContext为共享内存流上下文，nthChannel为目标消费者
+bool gs_is_databuff_empty(StreamSharedContext *sharedContext, int nthChannel)
 {
-    if (sharedContext->vectorized) {
-        VectorBatch* batch = sharedContext->sharedBatches[nthChannel][u_sess->stream_cxt.smp_id];
-        if (batch->m_rows == 0) {
+    // 判断是否启用了向量化处理
+    if (sharedContext->vectorized)
+    {
+        // 获取指定通道和会话的共享批处理对象
+        VectorBatch *batch = sharedContext->sharedBatches[nthChannel][u_sess->stream_cxt.smp_id];
+        // 如果批处理的行数为0，则缓冲区为空
+        if (batch->m_rows == 0)
+        {
             return true;
         }
-    } else {
-        TupleVector* tupleVec = sharedContext->sharedTuples[nthChannel][u_sess->stream_cxt.smp_id];
-        if (tupleVec->tuplePointer == 0) {
+    }
+    else
+    {
+        // 获取指定通道和会话的共享元组向量对象
+        TupleVector *tupleVec = sharedContext->sharedTuples[nthChannel][u_sess->stream_cxt.smp_id];
+        // 如果元组指针为0，则缓冲区为空
+        if (tupleVec->tuplePointer == 0)
+        {
             return true;
         }
     }
@@ -185,94 +225,134 @@ bool gs_is_databuff_empty(StreamSharedContext* sharedContext, int nthChannel)
  * @param[IN] nthChannel: destination consumer
  * @param[IN] nthRow: the Nth row to be sent in batch
  */
+// 此函数的作用是通过共享内存向本地消费者发送消息，参数tuple为要发送的元组，batchsrc为要发送的批次，sharedContext为共享内存流上下文，nthChannel为目标消费者，nthRow为批次中要发送的第n行
 void gs_memory_send(
-    TupleTableSlot* tuple, VectorBatch* batchsrc, StreamSharedContext* sharedContext, int nthChannel, int nthRow)
+    TupleTableSlot *tuple, VectorBatch *batchsrc, StreamSharedContext *sharedContext, int nthChannel, int nthRow)
 {
-    VectorBatch* batch = NULL;
-    TupleVector* tupleVec = NULL;
+    // 定义一个批处理指针，用于存储批处理对象
+    VectorBatch *batch = NULL;
+    // 定义一个元组向量指针，用于存储元组向量对象
+    TupleVector *tupleVec = NULL;
+    // 定义一个布尔型变量，用于存储是否可以发送的状态
     bool ready_to_send = false;
+    // 定义一个数据状态变量，用于存储数据的状态
     DataStatus dataStatus;
-    struct hash_entry* entry = NULL;
-
+    // 定义一个哈希表条目指针，用于存储共享内存流上下文的哈希表条目
+    struct hash_entry *entry = NULL;
+    // 报告等待状态，将当前状态设置为等待刷新数据状态
     WaitState oldStatus = pgstat_report_waitstatus_comm(STATE_WAIT_FLUSH_DATA,
-        u_sess->pgxc_cxt.PGXCNodeId,
-        -1,
-        u_sess->stream_cxt.producer_obj->getParentPlanNodeId(),
-        global_node_definition ? global_node_definition->num_nodes : -1);
-
+                                                        u_sess->pgxc_cxt.PGXCNodeId,
+                                                        -1,
+                                                        u_sess->stream_cxt.producer_obj->getParentPlanNodeId(),
+                                                        global_node_definition ? global_node_definition->num_nodes : -1);
+    // 记录时间，开始发送数据
     StreamTimeSendStart(t_thrd.pgxc_cxt.GlobalNetInstr);
+    // 获取指定通道和会话的共享内存流上下文的哈希表条目
     entry = sharedContext->quota_entrys[nthChannel][u_sess->stream_cxt.smp_id];
-    for (;;) {
-        /* Check for interrupt at the beginning of the loop. */
+    // 进入无限循环，直到发送完成或发生中断等条件退出循环
+    for (;;)
+    {
+        // 在循环开始处检查中断。如果发生中断，则立即退出循环
         CHECK_FOR_INTERRUPTS();
 
-        /* Check if we should early stop. */
-        /* Quit if the connection close, especially in a early close case. */
-        if (executorEarlyStop() || sharedContext->is_connect_end[nthChannel][u_sess->stream_cxt.smp_id]) {
+        // 检查是否需要提前停止。如果连接关闭，特别是在提前关闭的情况下，则退出循环
+        if (executorEarlyStop() || sharedContext->is_connect_end[nthChannel][u_sess->stream_cxt.smp_id])
+        {
+            // 恢复等待状态为原始状态
             (void)pgstat_report_waitstatus(oldStatus);
             return;
         }
-
+        // 获取指定通道和会话的数据状态
         dataStatus = sharedContext->dataStatus[nthChannel][u_sess->stream_cxt.smp_id];
-        /* Break the loop if we find quota. */
+        // 如果数据状态为DATA_EMPTY且（在__aarch64__架构下，数据缓冲区为空），或者数据状态为DATA_PREPARE，则跳出循环
         if ((dataStatus == DATA_EMPTY
 #ifdef __aarch64__
              && gs_is_databuff_empty(sharedContext, nthChannel)
 #endif
-             ) ||
-            dataStatus == DATA_PREPARE) {
+                 ) ||
+            dataStatus == DATA_PREPARE)
+        {
             break;
         }
-
+        // 记录时间，开始等待配额
         StreamTimeWaitQuotaStart(t_thrd.pgxc_cxt.GlobalNetInstr);
+        // 调用entry的_timewait方法，传入SINGLE_WAITQUOTA作为参数，等待配额
         (void)entry->_timewait(SINGLE_WAITQUOTA);
+        // 记录时间，结束等待配额
         StreamTimeWaitQuotaEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
     }
-
+    // 记录时间，开始复制数据
     StreamTimeCopyStart(t_thrd.pgxc_cxt.GlobalNetInstr);
-    /* Copy data to shared context. */
-    if (sharedContext->vectorized) {
+    // 将数据复制到共享上下文
+    if (sharedContext->vectorized)
+    {
+        // 如果启用了向量化处理，则断言共享批处理对象不为空
         Assert(sharedContext->sharedBatches != NULL);
+        // 获取指定通道和会话的共享批处理对象
         batch = sharedContext->sharedBatches[nthChannel][u_sess->stream_cxt.smp_id];
-        /* data copy */
-        if (-1 == nthRow) {
-            /* Do deep copy of all rows, for local roundrobin & local broadcast. */
+        // 如果nthRow为-1，则对所有行进行深度复制，用于本地循环和本地广播
+        if (-1 == nthRow)
+        {
+            // Assert批处理的行数为0，因为要进行所有行的深度复制
             Assert(batch->m_rows == 0);
+            // 进行深度复制
             batch->Copy<true, false>(batchsrc);
+            // 设置可以发送的状态为true
             ready_to_send = true;
-        } else {
+        }
+        else
+        {
+            // 复制指定行的数据
             batch->CopyNth(batchsrc, nthRow);
-            if (BatchMaxSize == batch->m_rows) {
+            // 如果批处理的行数等于BatchMaxSize，则设置可以发送的状态为true
+            if (BatchMaxSize == batch->m_rows)
+            {
                 ready_to_send = true;
             }
         }
-    } else {
+    }
+    else
+    {
+        // 如果未启用向量化处理，则断言共享元组向量对象不为空
         Assert(sharedContext->sharedTuples != NULL);
+        // 获取指定通道和会话的共享元组向量对象
         tupleVec = sharedContext->sharedTuples[nthChannel][u_sess->stream_cxt.smp_id];
+        // 获取元组指针
         int n = tupleVec->tuplePointer;
+        // 复制元组到元组向量
         ExecCopySlot(tupleVec->tupleVector[n], tuple);
+        // 元组指针加1
         tupleVec->tuplePointer++;
-        if (TupleVectorMaxSize == tupleVec->tuplePointer) {
+        // 如果元组指针等于TupleVectorMaxSize，则设置可以发送的状态为true
+        if (TupleVectorMaxSize == tupleVec->tuplePointer)
+        {
             ready_to_send = true;
         }
     }
+    // 记录时间，结束复制数据
     StreamTimeCopyEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
 
-    /* send the signal if copy finished */
-    if (ready_to_send) {
+    // 如果数据已经准备好发送，则执行以下代码块
+    if (ready_to_send)
+    {
 #ifdef __aarch64__
+        // 在__aarch64__架构下，执行内存屏障操作，确保内存操作的正确顺序
         pg_memory_barrier();
 #endif
-        /* set flag */
+        // 设置数据状态为DATA_READY，表示数据已经准备好
         sharedContext->dataStatus[nthChannel][u_sess->stream_cxt.smp_id] = DATA_READY;
         /* send signal */
         entry = sharedContext->poll_entrys[nthChannel];
         entry->_signal();
-    } else {
+    }
+    else
+    {
+        // 如果数据还没有准备好，则将数据状态设置为DATA_PREPARE，表示数据正在准备中
         sharedContext->dataStatus[nthChannel][u_sess->stream_cxt.smp_id] = DATA_PREPARE;
     }
+    // 记录时间，结束发送数据
     StreamTimeSendEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
-
+    // 恢复等待状态为原始状态
     (void)pgstat_report_waitstatus(oldStatus);
 }
 
@@ -282,17 +362,22 @@ void gs_memory_send(
  * @param[IN] node: stream state
  * @return bool: true -- found data
  */
+// 此函数用于从流状态的缓冲区中获取一个元组，参数node为流状态
 FORCE_INLINE
-bool gs_return_tuple(StreamState* node)
+bool gs_return_tuple(StreamState *node)
 {
-    TupleVector* tupleVec = node->tempTupleVec;
-
-    if (tupleVec->tuplePointer == 0) {
+    // 获取流状态的临时元组向量
+    TupleVector *tupleVec = node->tempTupleVec;
+    // 如果元组指针为0，表示没有数据可返回
+    if (tupleVec->tuplePointer == 0)
+    {
         return false;
     }
-
+    // 元组指针减1，因为我们要返回的是当前指针指向的元组
     tupleVec->tuplePointer--;
+    // 获取当前元组指针的索引
     int n = tupleVec->tuplePointer;
+    // 将结果元组槽设置为当前元组指针指向的元组
     node->ss.ps.ps_ResultTupleSlot = tupleVec->tupleVector[n];
 
     return true;
@@ -305,53 +390,67 @@ bool gs_return_tuple(StreamState* node)
  * @param[IN] loc: data location
  * @return bool: true -- found data
  */
-bool gs_consume_memory_data(StreamState* node, int loc)
+// 此函数的作用是从共享内存中消费本地生产者的数据，参数node为流状态，loc为数据位置
+bool gs_consume_memory_data(StreamState *node, int loc)
 {
-    StreamSharedContext* sharedContext = node->sharedContext;
-
+    // 获取流状态的共享上下文
+    StreamSharedContext *sharedContext = node->sharedContext;
+    // 记录时间，开始网络工作时间拷贝
     NetWorkTimeCopyStart(t_thrd.pgxc_cxt.GlobalNetInstr);
-    /* Take data from the shared context. */
-    if (sharedContext->vectorized) {
-        VectorBatch* batchsrc = sharedContext->sharedBatches[u_sess->stream_cxt.smp_id][loc];
-        VectorBatch* batchdst = ((VecStreamState*)node)->m_CurrentBatch;
-
-        if (batchsrc->m_rows == 0) {
+    // 如果共享上下文已经向量化,从共享上下文中获取数据
+    if (sharedContext->vectorized)
+    {
+        // 获取源批处理对象和目标批处理对象
+        VectorBatch *batchsrc = sharedContext->sharedBatches[u_sess->stream_cxt.smp_id][loc];
+        VectorBatch *batchdst = ((VecStreamState *)node)->m_CurrentBatch;
+        // 如果源批处理的行数为0，表示没有数据可消费，返回false
+        if (batchsrc->m_rows == 0)
+        {
             return false;
         }
-
+        // 将源批处理的数据复制到目标批处理，进行深复制，不重置源批处理
         batchdst->Copy<true, false>(batchsrc);
-
+        // 重置源批处理
         batchsrc->Reset();
-    } else {
-        TupleVector* tuplesrc = sharedContext->sharedTuples[u_sess->stream_cxt.smp_id][loc];
-        TupleVector* tupledst = node->tempTupleVec;
+    }
+    else
+    {
+        // 如果共享上下文未向量化,获取源元组向量对象和目标元组向量对象
+        TupleVector *tuplesrc = sharedContext->sharedTuples[u_sess->stream_cxt.smp_id][loc];
+        TupleVector *tupledst = node->tempTupleVec;
 
-        if (tuplesrc->tuplePointer == 0) {
+        // 如果源元组指针为0，表示没有数据可消费，返回false
+        if (tuplesrc->tuplePointer == 0)
+        {
             return false;
         }
-
-        for (int i = 0; i < tuplesrc->tuplePointer; i++) {
+        // 将源元组向量的数据复制到目标元组向量
+        for (int i = 0; i < tuplesrc->tuplePointer; i++)
+        {
             (void)ExecCopySlot(tupledst->tupleVector[i], tuplesrc->tupleVector[i]);
         }
-
+        // 设置目标元组指针为源元组指针，重置源元组指针
         tupledst->tuplePointer = tuplesrc->tuplePointer;
         tuplesrc->tuplePointer = 0;
+        // 返回元组
         (void)gs_return_tuple(node);
     }
+    // 记录时间，结束网络工作时间拷贝
     NetWorkTimeCopyEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
 
-    struct hash_entry* entry = NULL;
+    struct hash_entry *entry = NULL;
+    // 获取配额条目
     entry = sharedContext->quota_entrys[u_sess->stream_cxt.smp_id][loc];
-
+    // 如果编译在aarch64架构下，执行内存屏障操作，确保内存操作的正确顺序
 #ifdef __aarch64__
     pg_memory_barrier();
 #endif
-    /* Reset flag */
+    // 重置标志位
     sharedContext->dataStatus[u_sess->stream_cxt.smp_id][loc] = DATA_EMPTY;
 
-    /* send signal */
+    // 发送信号
     entry->_signal();
-
+    // 更新扫描位置
     node->sharedContext->scanLoc[u_sess->stream_cxt.smp_id] = loc;
     return true;
 }
@@ -364,97 +463,139 @@ bool gs_consume_memory_data(StreamState* node, int loc)
  *                  STREAM_SCAN_WAIT -- still need to poll to wait for data.
  *                 STREAM_SCAN_FINISH -- stream scan finished.
  */
-char gs_find_memory_data(StreamState* node, int* waitnode_count)
+// 此函数的作用是从生产者状态扫描数据，参数node为流状态
+char gs_find_memory_data(StreamState *node, int *waitnode_count)
 {
+    // 定义一个数据状态变量
     DataStatus dataStatus;
+    // 定义一个字符串信息变量，初始值为NULL
     StringInfo buf = NULL;
+    // 获取上次扫描的位置
     int scanLoc = node->sharedContext->scanLoc[u_sess->stream_cxt.smp_id];
+    // 定义一个计数器变量，初始值为上次扫描的位置
     int i = scanLoc;
+    // 定义一个标志位，表示扫描是否完成，初始值为true
     bool finished = true;
+    // 定义一个标志位，表示连接是否结束，初始值为false
     bool is_conn_end = false;
+    // 定义一个计数器，用于统计需要等待的节点数量，初始值为0
     int waitnodeCount = 0;
-    struct hash_entry* entry = NULL;
+    // 定义一个哈希表条目指针，初始值为NULL
+    struct hash_entry *entry = NULL;
 
-    /* Check if there is available data, and scan from last time location. */
-    do {
+    // 检查是否存在目标数据，并且从最后位置开始寻找
+    do
+    {
         i++;
-        if (i == node->conn_count) {
+        // 如果计数器等于连接数，则重置为0
+        if (i == node->conn_count)
+        {
             i = 0;
         }
 
-        /* Update scan location. */
+        // 更新扫描位置
         node->sharedContext->scanLoc[u_sess->stream_cxt.smp_id] = i;
+        // 获取当前位置的数据状态
         dataStatus = node->sharedContext->dataStatus[u_sess->stream_cxt.smp_id][i];
+        // 获取当前位置的连接是否结束状态
         is_conn_end = node->sharedContext->is_connect_end[u_sess->stream_cxt.smp_id][i];
 
-        if (!is_conn_end) {
+        if (!is_conn_end)
+        {
+            // 设置扫描完成标志位为false
             finished = false;
+            // 需要等待的节点数量加1
             waitnodeCount++;
         }
 
-        /*
-         * Firstly, we handle error or notice messages.
-         * If an error occured, we should stop scan now.
-         * If an notice occured, we can still receive data.
-         */
+        // 首先，我们处理错误或通知消息。如果错误，我们应该立即停止。如果发生了通知，我们仍然可以接收数据
+        // 从共享上下文中获取消息，这是一个字符串信息（StringInfo）结构，其中包含了消息的数据和长度等信息
         buf = node->sharedContext->messages[u_sess->stream_cxt.smp_id][i];
-        if (buf->len > 0) {
-            if (buf->cursor == 'E') {
+        // 如果消息的长度大于0，即存在消息
+        if (buf->len > 0)
+        {
+            // 如果消息的游标为'E'，表示这是一个错误消息
+            if (buf->cursor == 'E')
+            {
+                // 调用函数处理流错误，参数为节点，错误消息的数据和长度
                 HandleStreamError(node, buf->data, buf->len);
+                // 返回一个标识，表示流扫描结束
                 return STREAM_SCAN_FINISH;
-            } else if (buf->cursor == 'N') {
+            }
+            // 如果消息的游标为'N'，表示这是一个通知消息
+            else if (buf->cursor == 'N')
+            {
+                // 调用函数处理流通知，参数为节点，通知消息的数据和长度
                 HandleStreamNotice(node, buf->data, buf->len);
+                // 重置字符串信息，清空游标和数据
                 resetStringInfo(buf);
 
-                /* After one notice message has handled, send signal and wake up the dest producer. */
+                // 在处理完一个通知消息后，发送信号并唤醒目标生产者
                 entry = node->sharedContext->quota_entrys[u_sess->stream_cxt.smp_id][i];
+                // 获取配额条目，可能是为了记录或控制生产者的行为
                 entry->_signal();
-
+                // 返回一个标识，表示流扫描需要等待
                 return STREAM_SCAN_WAIT;
             }
         }
-
-        switch (dataStatus) {
-            case DATA_EMPTY:
-                break;
-
-            case DATA_PREPARE:
-                /* Take the rest data away when the connection is end. */
-                if (is_conn_end) {
-                    /* Return data if any. */
-                    if (gs_consume_memory_data(node, i)) {
-                        return STREAM_SCAN_DATA;
-                    }
-                }
-                break;
-
-            case DATA_READY:
-                if (gs_consume_memory_data(node, i)) {
+        // 根据dataStatus的值选择执行的代码块
+        switch (dataStatus)
+        {
+        // 如果dataStatus的值为DATA_EMPTY,不执行任何操作
+        case DATA_EMPTY:
+            break;
+        // 如果dataStatus的值为DATA_PREPARE
+        case DATA_PREPARE:
+            // 当连接结束的时候带走其余的数据
+            if (is_conn_end)
+            {
+                /* Return data if any. */
+                // 如果有数据，通过调用gs_consume_memory_data函数来消耗数据
+                if (gs_consume_memory_data(node, i))
+                {
+                    // 返回STREAM_SCAN_DATA，表示成功从生产者找到数据
                     return STREAM_SCAN_DATA;
-                } else {
-                    break;
                 }
+            }
+            break;
 
-            case CONN_ERR:
-                ereport(ERROR,
+        case DATA_READY:
+            // 通过调用gs_consume_memory_data函数来消耗数据
+            if (gs_consume_memory_data(node, i))
+            {
+                // 返回STREAM_SCAN_DATA，表示成功从生产者找到数据
+                return STREAM_SCAN_DATA;
+            }
+            else
+            {
+                break;
+            }
+        // 如果dataStatus的值为CONN_ERR，生成一个错误报告
+        case CONN_ERR:
+            ereport(ERROR,
                     (errcode(ERRCODE_STREAM_REMOTE_CLOSE_SOCKET),
-                        errmsg("Failed to read response from Local Stream Node,"
-                               " Detail: Node %s, Plan Node ID %u, SMP ID %d",
+                     errmsg("Failed to read response from Local Stream Node,"
+                            " Detail: Node %s, Plan Node ID %u, SMP ID %d",
                             g_instance.attr.attr_common.PGXCNodeName,
                             node->sharedContext->key_s.planNodeId,
                             i)));
-                break;
-            // dataStatus is enum,
-            default:
-                break;
+            break;
+        // dataStatus is enum,
+        default:
+            break;
         }
     } while (i != scanLoc);
-
+    // 将waitnodeCount的值赋给指针waitnode_count所指向的变量，为了返回等待节点的数量
     *waitnode_count = waitnodeCount;
 
-    if (finished) {
+    if (finished)
+    {
+        // 返回STREAM_SCAN_FINISH，表示流扫描完成
         return STREAM_SCAN_FINISH;
-    } else {
+    }
+    else
+    {
+        // 返回STREAM_SCAN_WAIT，表示仍然需要轮询等待数据
         return STREAM_SCAN_WAIT;
     }
 }
@@ -466,51 +607,68 @@ char gs_find_memory_data(StreamState* node, int* waitnode_count)
  * @return bool: true -- successed to find data and need more data.
  *                  false -- all connection finished or recerive error.
  */
-bool gs_memory_recv(StreamState* node)
+// 此函数的作用是从共享内存中接收本地流的数据，返回是否成功找到数据并需要更多数据，或者所有连接已完成或接收错误
+bool gs_memory_recv(StreamState *node)
 {
+    // 存储操作结果的字符变量
     char result;
-    struct hash_entry* entry = NULL;
+    // 哈希表条目指针，初始化为NULL
+    struct hash_entry *entry = NULL;
+    // 获取流状态对应的哈希表条目
     entry = node->sharedContext->poll_entrys[u_sess->stream_cxt.smp_id];
     bool re = true;
+    // 初始化等待节点数为0
     int waitnode_count = 0;
 
-    /* If there is already tuple in buffer, return the data at once. */
-    if (!node->sharedContext->vectorized && gs_return_tuple(node)) {
+    // 如果缓冲区中已有元组，则立即返回数据
+    if (!node->sharedContext->vectorized && gs_return_tuple(node))
+    {
         return true;
     }
 
-    for (;;) {
-        /* Check for interrupt at the beginning of the loop. */
+    for (;;)
+    {
+        // 检查是否有中断请求
         CHECK_FOR_INTERRUPTS();
 
-        /* Check if we can early stop now. */
-        if (executorEarlyStop()) {
+        // 检查是否可以提前结束循环
+        if (executorEarlyStop())
+        {
             re = false;
             break;
         }
 
-        /* Search all producers to find data. */
+        // 搜索所有生产者以查找数据，同时更新等待节点数
         result = gs_find_memory_data(node, &waitnode_count);
-        if (result == STREAM_SCAN_DATA) {
+        if (result == STREAM_SCAN_DATA)
+        {
             re = true;
             break;
-        } else if (result == STREAM_SCAN_FINISH) {
+        }
+        // 如果所有连接已完成或接收错误
+        else if (result == STREAM_SCAN_FINISH)
+        {
             re = false;
             break;
         }
-
+        // 定义一个旧的等待状态阶段变量，初始化为PHASE_NONE，表示当前没有等待状态
         WaitStatePhase oldPhase = pgstat_report_waitstatus_phase(PHASE_NONE, true);
+        // 定义一个旧的等待状态变量，通过调用pgstat_report_waitstatus_comm函数来初始化。
+        // 该函数将等待状态设置为STATE_WAIT_NODE，表示当前正在等待节点响应。
+        // 还将当前节点的ID、等待节点的数量、计划节点的ID以及全局节点定义的数量作为参数传递给该函数
         WaitState oldStatus = pgstat_report_waitstatus_comm(STATE_WAIT_NODE,
-            u_sess->pgxc_cxt.PGXCNodeId,
-            waitnode_count,
-            node->sharedContext->key_s.planNodeId,
-            global_node_definition ? global_node_definition->num_nodes : -1);
+                                                            u_sess->pgxc_cxt.PGXCNodeId,
+                                                            waitnode_count,
+                                                            node->sharedContext->key_s.planNodeId,
+                                                            global_node_definition ? global_node_definition->num_nodes : -1);
 
-        /* Poll to wait data from producers. */
+        // 开始网络时间轮询，用于度量网络操作的耗时
         NetWorkTimePollStart(t_thrd.pgxc_cxt.GlobalNetInstr);
+        // 调用entry的_timewait方法，传入SINGLE_WAITQUOTA作为参数，用于等待生产者提供数据
         (void)entry->_timewait(SINGLE_WAITQUOTA);
+        // 结束网络时间轮询
         NetWorkTimePollEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
-
+        // 重置等待状态阶段和等待状态为旧的状态
         pgstat_reset_waitStatePhase(oldStatus, oldPhase);
     }
 
@@ -523,16 +681,20 @@ bool gs_memory_recv(StreamState* node)
  * @param[IN] sharedContext: context for shared memory stream
  * @param[IN] connNum: producer connection number
  */
-void gs_memory_send_finish(StreamSharedContext* sharedContext, int connNum)
+// 此函数用于通知所有相关的消费者没有更多数据可发送
+void gs_memory_send_finish(StreamSharedContext *sharedContext, int connNum)
 {
-    struct hash_entry* entry = NULL;
+    // 定义一个哈希表条目指针，初始化为NULL
+    struct hash_entry *entry = NULL;
 
-    for (int i = 0; i < connNum; i++) {
-        /* Set flags. */
+    for (int i = 0; i < connNum; i++)
+    {
+        // 设置标志位，表示连接已经结束
         sharedContext->is_connect_end[i][u_sess->stream_cxt.smp_id] = true;
 
-        /* send signal */
+        // 获取当前连接对应的哈希表条目
         entry = sharedContext->poll_entrys[i];
+        // 调用哈希表条目的_signal方法，发送信号且通知等操作
         entry->_signal();
     }
 }
@@ -544,12 +706,15 @@ void gs_memory_send_finish(StreamSharedContext* sharedContext, int connNum)
  * @param[IN] connNum: producer connection number
  * @param[IN] smpId: producer smp id
  */
-void gs_memory_close_conn(StreamSharedContext* sharedContext, int connNum, int consumerId)
+// 此函数用于设置与特定生产者的所有连接关闭
+void gs_memory_close_conn(StreamSharedContext *sharedContext, int connNum, int consumerId)
 {
-    struct hash_entry* entry = NULL;
+    // 定义一个哈希表条目指针，初始化为NULL
+    struct hash_entry *entry = NULL;
 
-    for (int i = 0; i < connNum; i++) {
-        /* Set flags. */
+    for (int i = 0; i < connNum; i++)
+    {
+        // 设置标志位，表示与特定生产者的连接已结束
         sharedContext->is_connect_end[consumerId][i] = true;
 
         /*
@@ -557,8 +722,7 @@ void gs_memory_close_conn(StreamSharedContext* sharedContext, int connNum, int c
          * in a query like "limit XXX", when consumer don't need data anymore,
          * but the producers haven't send all data yet.
          */
-        entry = sharedContext->quota_entrys[consumerId][i];
-        entry->_signal();
+        entry = sharedContext->quota_entrys[consumerId][i]; // 获取当前连接对应的哈希表条目
+        entry->_signal();                                   // 调用哈希表条目的_signal方法，发送信号且通知等操作
     }
-}
-
+}
\ No newline at end of file