openGauss源码学习（四）SeqScan扫描算子 执行器部分

上一篇：openGauss源码学习（三）SeqScan扫描算子 优化器部分

---

---

前言

    上篇文章介绍了一个简单查询里如何为基表生成SeqScan计划，这篇文章从执行器和存储引擎的角度介绍一下SeqScan算子是如何执行的。
    执行器框架有很多博主都做了全面并且详细的分析，可以参考博文PostgreSQL内核学习（十五）—— （ExecutorRun）

---

一、执行器部分

    执行器关键函数是ExecScan，OpenGauss源码学习 —— 执行算子（SeqScan算子）博文中对于整个SeqScan算子都有一个全面的介绍，本文就不再赘述了。
    SeqScan算子顺序扫描表时，是通过seq_scan_getnext函数从存储引擎获取可见的记录，关键函数分别是heap_getnext和heap_slot_store_heap_tuple。其中heap_getnext函数负责扫描页面并检查可见性并返回可见的HeapTuple，而heap_slot_store_heap_tuple则是负责将物理存储格式的数据转换为执行器格式的TupleTableSlot，屏蔽了底层存储引擎的差异。

void seq_scan_getnext(TableScanDesc scan,  TupleTableSlot* slot, ScanDirection direction,
        bool* has_cur_xact_write)
{
    Tuple tuple;
    tuple =  (Tuple)heap_getnext(scan, direction, has_cur_xact_write);
    if (tuple != NULL) {
         Assert(slot != NULL);
         Assert(slot->tts_tupleDescriptor != NULL);
         slot->tts_tam_ops = GetTableAmRoutine(TAM_HEAP);
         heap_slot_store_heap_tuple((HeapTuple)tuple, slot, scan->rs_cbuf, false, false);
    } else {
         ExecClearTuple(slot);
    }
}

二、存储引擎

    介绍heap_getnext之前先结合图片（来源于5.3. Inserting, Deleting and Updating Tuples）简单说明一下page的结构，有助于理解后面的代码。SeqScan就是对表的页面进行顺序扫描，每读完一个完整页面再读取下一个页面。page主要包含两部分：

1. page header，记录了页面的概要信息，比如xlog的lsn，pd_flags记录的页面属性等等。
2. HeapTuple，物理数据结构，用于记录事务信息和实际用户数据。扫描页面主要是结合tuple的事务信息检查可见性并返回本页面上所有可见的HeapTuple。

那么下面我们对代码进行走读，结合源码分析下存储引擎是如何从一个物理页面读取数据的。

2.1 heapgettup_pagemode

heapgettup_pagemode是主要入口函数，主要的入参分别是：

• HeapScanDesc scan 执行器传入，说明表的扫描信息，包括当前页面可见的总元组数等。
• ScanDirection dir：扫描的方向，是正向还是反向。
• nkeys&keys：检查key，一般用于索引扫描，SeqScan不需要检查。

static void heapgettup_pagemode(HeapScanDesc scan, ScanDirection dir, int nkeys, ScanKey key,
                                bool* has_cur_xact_write = NULL)

首先，需要获取表的page。

		/* 
		 * 首先检查一下是否已经初始化，未初始化则需要去加载页面，
		 * 否则直接从已加载的页面顺序扫描tuple即可。 
		 */
        if (!scan->rs_base.rs_inited) {
            /* return null immediately if relation is empty */
            if (scan->rs_base.rs_nblocks == 0) {
                Assert(!BufferIsValid(scan->rs_base.rs_cbuf));
                tuple->t_data = NULL;
                return;
            }
            if (scan->rs_parallel != NULL) {
            	/*
            	 * 并行SeqScan部分逻辑，如果开启query_dop并生成了并行计划，
            	 * 那么SeqScan的时候对pagenum取模，每个线程扫描对应的page。
            	 */
                HeapParallelscanStartblockInit(scan);
                page = HeapParallelscanNextpage(scan);

                /* Other processes might have already finished the scan. */
                if (page == InvalidBlockNumber) {
                    Assert(!BufferIsValid(scan->rs_base.rs_cbuf));
                    tuple->t_data = NULL;
                    return;
                }
            } else {
                page = scan->rs_base.rs_startblock; /* first page */
            }
            /* heapgetpage中负责将表对应的page读取到缓存中 */
            heapgetpage((TableScanDesc)scan, page, has_cur_xact_write);
            line_index = 0;
            scan->rs_base.rs_inited = true;
        } else {
        	/* page已加载，直接读取加载后的page */
            /* continue from previously returned page/tuple */
            page = scan->rs_base.rs_cblock; /* current page */
            line_index = scan->rs_base.rs_cindex + 1;
        }

        dp = (Page)BufferGetPage(scan->rs_base.rs_cbuf);
        /* page中所有记录的总数 */
        lines = scan->rs_base.rs_ntuples;
        /* page and line_index now reference the next visible tid */
        /* page中剩余的记录数 */
        lines_left = lines - line_index;

接下来是顺序扫描page中所有的页面，如上面图片所示，page中tuple的存储类似于数组，扫描的时候也是按照下标顺序去取每一个tuple。

    for (;;) {
        /* 首先检查当前页面剩余的tuple是否还存在可见的 */
        while (lines_left > 0) {
            line_off = scan->rs_base.rs_vistuples[line_index];
            lpp = HeapPageGetItemId(dp, line_off);
            Assert(ItemIdIsNormal(lpp));

			/* 保存当前tuple，如果符合要求则返回给执行器 */
            tuple->t_data = (HeapTupleHeader)PageGetItem((Page)dp, lpp);
            tuple->t_len = ItemIdGetLength(lpp);
            ItemPointerSet(&(tuple->t_self), page, line_off);
            HeapTupleCopyBaseFromPage(tuple, dp);

            /*
             * if the tuple is compressed, uncompress it first, because
             * 1. reduce the UNCOMPRESS number within HeapKeyTest();
             * 2. maybe reduce the number of palloc() within HeapKeyTest();
             */
            if (scan->rs_base.rs_rd->is_compressed && HEAP_TUPLE_IS_COMPRESSED(tuple->t_data)) {
                DECOMPRESS_HEAP_TUPLE(
                    true, tuple, &(scan->rs_ctbuf_hdr), (scan->rs_tupdesc), dp);
            }
            /* if current tuple qualifies, return it. */
            if (key != NULL) {
                bool valid = false;

                HeapKeyTest(tuple, (scan->rs_tupdesc), nkeys, key, valid);

                if (valid) {
                    scan->rs_base.rs_cindex = line_index;
                    return;
                }
            } else {
                /* 找到符合要求的tuple，记录下当前tuple下标，下次从该下标继续扫描。 */
                scan->rs_base.rs_cindex = line_index;
                return;
            }

            /* otherwise move to the next item on the page */
            --lines_left;
            if (backward) {
                --line_index;
            } else {
                ++line_index;
            }
        }
        /*
         * if we get here, it means we've exhausted the items on this page and
         * it's time to move to the next.
         */
         /* 获取下一个page，如果没有则说明表已经扫描完。 */
        finished = next_page(scan, dir, page);
        /* return NULL if we've exhausted all the pages */
        if (finished) {
            if (BufferIsValid(scan->rs_base.rs_cbuf)) {
                ReleaseBuffer(scan->rs_base.rs_cbuf);
            }
            scan->rs_base.rs_cbuf = InvalidBuffer;
            scan->rs_base.rs_cblock = InvalidBlockNumber;
            tuple->t_data = NULL;
            scan->rs_base.rs_inited = false;
            return;
        }

        heapgetpage((TableScanDesc)scan, page, has_cur_xact_write);

        dp = (Page)BufferGetPage(scan->rs_base.rs_cbuf);
        lines = scan->rs_base.rs_ntuples;
        lines_left = lines;
        if (backward) {
            line_index = lines - 1;
        } else {
            line_index = 0;
        }
    }

2.2 heapgetpage

函数主要逻辑是将page从物理文件加载到buffer中，其中涉及到BufferPool的管理和可见性检查，本博客暂不对此部分展开说明。

    /* read page using selected strategy */
    scan->rs_base.rs_cbuf = ReadBufferExtended(scan->rs_base.rs_rd, MAIN_FORKNUM, page, RBM_NORMAL, scan->rs_base.rs_strategy);
    scan->rs_base.rs_cblock = page;
    
    ...
    
    /*
     * 如果页面是全部可见的，那么对于这个页面的所有tuple就不需要检查可见性。
     * 通过page的header标记位判断，vacuum时检查并赋值。
     * 所以对于大量导入数据后的场景，可以尝试vacuum来提升扫描效率。
     */
    all_visible = PageIsAllVisible(dp) && !snapshot->takenDuringRecovery;

    for (line_off = FirstOffsetNumber, lpp = HeapPageGetItemId(dp, line_off); line_off <= lines; line_off++, lpp++) {
        if (ItemIdIsNormal(lpp)) {
        	/* 遍历page中的tuple，并记录可见的tuple到rs_vistuples数组中 */
            HeapTupleData loctup;
            bool valid = false;

            if (likely(all_visible && (!IsSerializableXact()))) {
                scan->rs_base.rs_vistuples[ntup++] = line_off;
                continue;
            }

            loctup.t_tableOid = RelationGetRelid(scan->rs_base.rs_rd);
            loctup.t_bucketId = RelationGetBktid(scan->rs_base.rs_rd);
            loctup.t_data = (HeapTupleHeader)PageGetItem((Page)dp, lpp);
            loctup.t_len = ItemIdGetLength(lpp);
            HeapTupleCopyBaseFromPage(&loctup, dp);
            ItemPointerSet(&(loctup.t_self), page, line_off);

            if (all_visible)
                valid = true;
            else
                /* MVCC可见性检查 */
                valid = HeapTupleSatisfiesVisibility(&loctup, snapshot, buffer, has_cur_xact_write);

            CheckForSerializableConflictOut(valid, scan->rs_base.rs_rd, (void*)&loctup, buffer, snapshot);

            if (valid) {
                scan->rs_base.rs_vistuples[ntup++] = line_off;
            }

            ereport(DEBUG1, (errmsg("heapgetpage xid %lu ctid(%u,%d) valid %d", GetCurrentTransactionIdIfAny(), page,
                line_off, valid)));
        }
    }

---

总结

    本篇博客简单介绍了OpenGauss执行器和astore存储引擎的SeqScan逻辑，可以清晰看到火山模型运行过程中是如何处理一行记录的。总的来说SeqScan是以页面为单位的，每次存储引擎读取一整个页面后加载到BufferPool中并记录所有可见的记录，然后将每一条记录返回给执行器的SeqScan算子。每次读取并缓存一个页面有很多好处，比如小表可以减少IO，提升扫描效率；并发时可以共享page，减少内存的消耗等等。

参考文档

OpenGauss源码学习 —— 执行算子（SeqScan算子）