src/backend/access/aocs/aocs_compaction.c - cloudberry - Git at Google

 /*------------------------------------------------------------------------------
  *
  * Code dealing with the compaction of append-only tables.
  *
  * Copyright (c) 2013-Present VMware, Inc. or its affiliates.
  *
  *
  * IDENTIFICATION
  *	    src/backend/access/aocs/aocs_compaction.c
  *
  *------------------------------------------------------------------------------
  */

 #include "postgres.h"

 #include <limits.h>

 #include "access/amapi.h"
 #include "access/genam.h"
 #include "access/heapam.h"
 #include "access/transam.h"
 #include "access/aosegfiles.h"
 #include "access/aomd.h"
 #include "access/aocs_compaction.h"
 #include "access/appendonly_compaction.h"
 #include "access/appendonlywriter.h"
 #include "catalog/catalog.h"
 #include "catalog/indexing.h"
 #include "catalog/pg_appendonly.h"
 #include "catalog/pg_attribute_encoding.h"
 #include "cdb/cdbaocsam.h"
 #include "cdb/cdbvars.h"
 #include "commands/vacuum.h"
 #include "executor/executor.h"
 #include "nodes/execnodes.h"
 #include "storage/lmgr.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/relcache.h"
 #include "utils/snapmgr.h"
 #include "utils/guc.h"
 #include "miscadmin.h"
 #include "commands/progress.h"
 #include "pgstat.h"

 /*
  * Hook for plugins to get control after move or throw away tuple in
  * AOCSSegmentFileFullCompaction().
  *
  * For example, zombodb will delete the corresponding docs after the tuple
  * is moved or thrown away.
  */
 aocs_compaction_delete_hook_type aocs_compaction_delete_hook = NULL;

 /*
  * Drops a segment file.
  *
  * Actually, we just truncate the segfile to 0 bytes, to reclaim the space.
  * Before GPDB 6, we used to remove the file, but with WAL replication, we
  * no longer have a convenient function to remove a single segment of a
  * relation. An empty file is as almost as good as a non-existent file. If
  * the relation is dropped later, the code in mdunlink() will remove all
  * segments, including any empty ones we've left behind.
  */
 void
 AOCSCompaction_DropSegmentFile(Relation aorel, int segno, AOVacuumRelStats *vacrelstats)
 {
 	int			col;

 	Assert(RelationStorageIsAoCols(aorel));

 	for (col = 0; col < RelationGetNumberOfAttributes(aorel); col++)
 	{
 		char		filenamepath[MAXPGPATH];
 		int			pseudoSegNo;
 		File		fd;
 		/* Filenum for the col */
 		FileNumber  filenum = GetFilenumForAttribute(RelationGetRelid(aorel), col + 1);

 		/* Open and truncate the relation segfile */
 		MakeAOSegmentFileName(aorel, segno, filenum, &pseudoSegNo, filenamepath);

 		elogif(Debug_appendonly_print_compaction, LOG,
 			   "Drop segment file: "
 			   "segno %d",
 			   pseudoSegNo);

 		fd = OpenAOSegmentFile(aorel, filenamepath, 0);
 		if (fd >= 0)
 		{
 			TruncateAOSegmentFile(fd, aorel, pseudoSegNo, 0, vacrelstats);
 			CloseAOSegmentFile(fd, aorel);
 		}
 		else
 		{
 			/*
 			 * The file we were about to drop/truncate didn't exist. That's normal,
 			 * for example, if a column is added with ALTER TABLE ADD COLUMN.
 			 */
 			elog(DEBUG1, "could not truncate segfile %s, because it does not exist", filenamepath);
 		}
 	}
 }

 /*
  * AOCSSegmentFileTruncateToEOF()
  *
  * Truncates the files for all columns of logical segfile 'segno' to
  * the EOF values from 'vpinfo'. The caller is responsible for locking so
  * that a concurrent backend doesn't write to the segfile while we truncate
  * it.
  *
  * This is used to clean up space left behind by aborted or crashed
  * transactions.
  */
 void
 AOCSSegmentFileTruncateToEOF(Relation aorel, int segno, AOCSVPInfo *vpinfo, AOVacuumRelStats *vacrelstats)
 {
 	const char *relname = RelationGetRelationName(aorel);
 	int			j;

 	Assert(RelationStorageIsAoCols(aorel));

 	for (j = 0; j < vpinfo->nEntry; ++j)
 	{
 		int64		segeof;
 		char		filenamepath[MAXPGPATH];
 		AOCSVPInfoEntry *entry;
 		File		fd;
 		int32		fileSegNo;
 		/* Filenum for the column */
 		FileNumber  filenum = GetFilenumForAttribute(RelationGetRelid(aorel), j + 1);

 		entry = &vpinfo->entry[j];
 		segeof = entry->eof;

 		/* Open and truncate the relation segfile to its eof */
 		MakeAOSegmentFileName(aorel, segno, filenum, &fileSegNo, filenamepath);

 		elogif(Debug_appendonly_print_compaction, LOG,
 			   "Opening AO COL relation \"%s.%s\", relation id %u, relfilenode %u column #%d, logical segment #%d (physical segment file #%d, logical EOF " INT64_FORMAT ")",
 			   get_namespace_name(RelationGetNamespace(aorel)),
 			   relname,
 			   aorel->rd_id,
 			   aorel->rd_locator.relNumber,
 			   j,
 			   segno,
 			   fileSegNo,
 			   segeof);

 		fd = OpenAOSegmentFile(aorel, filenamepath, segeof);
 		if (fd >= 0)
 		{
 			TruncateAOSegmentFile(fd, aorel, fileSegNo, segeof, vacrelstats);
 			CloseAOSegmentFile(fd, aorel);

 			elogif(Debug_appendonly_print_compaction, LOG,
 				   "Successfully truncated AO COL relation \"%s.%s\", relation id %u, relfilenode %u column #%d, logical segment #%d (physical segment file #%d, logical EOF " INT64_FORMAT ")",
 				   get_namespace_name(RelationGetNamespace(aorel)),
 				   relname,
 				   aorel->rd_id,
 				   aorel->rd_locator.relNumber,
 				   j,
 				   segno,
 				   fileSegNo,
 				   segeof);
 		}
 		else
 		{
 			elogif(Debug_appendonly_print_compaction, LOG,
 				   "No gp_relation_node entry for AO COL relation \"%s.%s\", relation id %u, relfilenode %u column #%d, logical segment #%d (physical segment file #%d, logical EOF " INT64_FORMAT ")",
 				   get_namespace_name(RelationGetNamespace(aorel)),
 				   relname,
 				   aorel->rd_id,
 				   aorel->rd_locator.relNumber,
 				   j,
 				   segno,
 				   fileSegNo,
 				   segeof);
 		}
 	}
 }

 static void
 AOCSMoveTuple(TupleTableSlot *slot,
 			  AOCSInsertDesc insertDesc,
 			  ResultRelInfo *resultRelInfo,
 			  EState *estate)
 {
 	AOTupleId  *oldAoTupleId;
 	AOTupleId	newAoTupleId;

 	Assert(resultRelInfo);
 	Assert(slot);
 	Assert(estate);

 	oldAoTupleId = (AOTupleId *) &slot->tts_tid;
 	/* Extract all the values of the tuple */
 	slot_getallattrs(slot);

 	(void) aocs_insert_values(insertDesc,
 							  slot->tts_values,
 							  slot->tts_isnull,
 							  &newAoTupleId);
 	memcpy(&slot->tts_tid, &newAoTupleId, sizeof(ItemPointerData));

 	/* insert index' tuples if needed */
 	if (resultRelInfo->ri_NumIndices > 0)
 	{
 		ExecInsertIndexTuples(resultRelInfo,
 							  slot, estate, false, false,
 							  NULL, NIL, false);
 		ResetPerTupleExprContext(estate);
 	}

 	elogif(Debug_appendonly_print_compaction, DEBUG5,
 		   "Compaction: Moved tuple (%d," INT64_FORMAT ") -> (%d," INT64_FORMAT ")",
 		   AOTupleIdGet_segmentFileNum(oldAoTupleId), AOTupleIdGet_rowNum(oldAoTupleId),
 		   AOTupleIdGet_segmentFileNum(&newAoTupleId), AOTupleIdGet_rowNum(&newAoTupleId));
 }

 /*
  * Subroutine of AOCSCompact().
  */
 static bool
 AOCSSegmentFileFullCompaction(Relation aorel,
 							  AOCSInsertDesc insertDesc,
 							  AOCSFileSegInfo *fsinfo,
 							  Snapshot snapshot,
 							  AOVacuumRelStats *vacrelstats)
 {
 	const char *relname;
 	AppendOnlyVisimap visiMap;
 	AOCSScanDesc scanDesc;
 	TupleDesc	tupDesc;
 	TupleTableSlot *slot;
 	int			compact_segno;
 	ResultRelInfo *resultRelInfo;
 	EState	   *estate;
 	int64		tupleCount = 0;
 	int64		tuplePerPage = INT_MAX;
 	ItemPointerData otid;
 	AOTupleId  *aoTupleId;
 	int64		curr_num_dead_tuples = 0;
 	int64		prev_num_dead_tuples = 0;
 	int64		curr_heap_blks_scanned = 0;
 	int64		prev_heap_blks_scanned = 0;

 	Assert(Gp_role == GP_ROLE_EXECUTE || Gp_role == GP_ROLE_UTILITY);
 	Assert(RelationStorageIsAoCols(aorel));
 	Assert(insertDesc);

 	compact_segno = fsinfo->segno;
 	if (fsinfo->varblockcount > 0)
 	{
 		tuplePerPage = fsinfo->total_tupcount / fsinfo->varblockcount;
 	}
 	relname = RelationGetRelationName(aorel);

 	AppendOnlyVisimap_Init(&visiMap,
 						   insertDesc->visimaprelid,
 						   insertDesc->visimapidxid,
 						   ShareLock,
 						   snapshot);

 	elogif(Debug_appendonly_print_compaction,
 		   LOG, "Compact AO segfile %d, relation %sd",
 		   compact_segno, relname);

 	scanDesc = aocs_beginrangescan(aorel,
 								   snapshot, snapshot,
 								   &compact_segno, 1);

 	tupDesc = RelationGetDescr(aorel);
 	slot = MakeSingleTupleTableSlot(tupDesc, &TTSOpsVirtual);
 	slot->tts_tableOid = RelationGetRelid(aorel);

 	/*
 	 * We need a ResultRelInfo and an EState so we can use the regular
 	 * executor's index-entry-making machinery.
 	 */
 	estate = CreateExecutorState();
 	resultRelInfo = makeNode(ResultRelInfo);
 	resultRelInfo->ri_RangeTableIndex = 1;	/* dummy */
 	resultRelInfo->ri_RelationDesc = aorel;
 	resultRelInfo->ri_TrigDesc = NULL;	/* we don't fire triggers */
 	ExecOpenIndices(resultRelInfo, false);
 	estate->es_opened_result_relations =
 			lappend(estate->es_opened_result_relations, resultRelInfo);

 	/*
 	 * We don't want uniqueness checks to be performed while "insert"ing tuples
 	 * to a destination segfile during AOCSMoveTuple(). This is to ensure that
 	 * we can avoid spurious conflicts between the moved tuple and the original
 	 * tuple.
 	 */
 	estate->gp_bypass_unique_check = true;

 	while (aocs_getnext(scanDesc, ForwardScanDirection, slot))
 	{
 		CHECK_FOR_INTERRUPTS();

 		/*
  		 * AppendOnlyVisimap_IsVisible() has already been called in aocs_getnext().
 		 */
 		Assert(AppendOnlyVisimap_IsVisible(&scanDesc->visibilityMap,
 										   (AOTupleId *) &slot->tts_tid));

 		aoTupleId = (AOTupleId *) &slot->tts_tid;
 		otid = slot->tts_tid;

 		AOCSMoveTuple(slot,
 					  insertDesc,
 					  resultRelInfo,
 					  estate);

 		if (aocs_compaction_delete_hook)
 			(*aocs_compaction_delete_hook) (aorel, &otid);

 		/*
 		 * Check for vacuum delay point after approximatly a var block
 		 */
 		tupleCount++;
 		if (VacuumCostActive && tupleCount % tuplePerPage == 0)
 		{
 			vacuum_delay_point();
 		}

 		/*
 		 * Report that we are now scanning and compacting segment files.
 		 */
 		curr_num_dead_tuples = scanDesc->segrowsprocessed + 1 - tupleCount;
 		if (curr_num_dead_tuples > prev_num_dead_tuples)
 		{
 			pgstat_progress_update_param(PROGRESS_VACUUM_NUM_DEAD_TUPLES,
 										 vacrelstats->num_dead_tuples + curr_num_dead_tuples);
 			prev_num_dead_tuples = curr_num_dead_tuples;
 		}

 		curr_heap_blks_scanned = RelationGuessNumberOfBlocksFromSize(scanDesc->totalBytesRead);
 		if (curr_heap_blks_scanned > prev_heap_blks_scanned)
 		{
 			pgstat_progress_update_param(PROGRESS_VACUUM_HEAP_BLKS_SCANNED,
 										 curr_heap_blks_scanned);
 			prev_heap_blks_scanned = curr_heap_blks_scanned;
 		}
 	}
 	/* Accumulate total number dead tuples */
 	vacrelstats->num_dead_tuples += scanDesc->segrowsprocessed - tupleCount;

 	MarkAOCSFileSegInfoAwaitingDrop(aorel, compact_segno);

 	AppendOnlyVisimap_DeleteSegmentFile(&visiMap,
 										compact_segno);

 	/* Delete all mini pages of the segment files if block directory exists */
 	if (OidIsValid(insertDesc->blkdirrelid))
 	{
 		AppendOnlyBlockDirectory_DeleteSegmentFile(aorel,
 												   snapshot,
 												   compact_segno,
 												   0);
 	}

 	elogif(Debug_appendonly_print_compaction, LOG,
 		   "Finished compaction: "
 		   "AO segfile %d, relation %s, moved tuple count " INT64_FORMAT,
 		   compact_segno, relname, tupleCount);

 	AppendOnlyVisimap_Finish(&visiMap, NoLock);

 	ExecCloseIndices(resultRelInfo);
 	FreeExecutorState(estate);

 	ExecDropSingleTupleTableSlot(slot);

 	aocs_endscan(scanDesc);

 	return true;
 }

 /*
  * Performs a compaction of an append-only relation in column-orientation.
  *
  * The compaction segment file should be locked for this transaction in
  * the appendonlywriter.c code.
  *
  * On exit, *insert_segno will be set to the segment that was used as the
  * insertion target. The segfiles listed in 'avoid_segnos' will not be used
  * for insertion.
  *
  * The caller is required to hold either an AccessExclusiveLock (vacuum full)
  * or a ShareLock on the relation.
  */
 void
 AOCSCompact(Relation aorel,
 			int compaction_segno,
 			int *insert_segno,
 			bool isFull,
 			List *avoid_segnos,
 			AOVacuumRelStats *vacrelstats)
 {
 	const char *relname;
 	AOCSInsertDesc insertDesc = NULL;
 	AOCSFileSegInfo *fsinfo;
 	Snapshot	appendOnlyMetaDataSnapshot = RegisterSnapshot(GetCatalogSnapshot(InvalidOid));

 	Assert(RelationStorageIsAoCols(aorel));
 	Assert(Gp_role == GP_ROLE_EXECUTE || Gp_role == GP_ROLE_UTILITY);

 	relname = RelationGetRelationName(aorel);
 	elogif(Debug_appendonly_print_compaction, LOG,
 		   "Compact AO relation %s", relname);

 	/* Fetch under the write lock to get latest committed eof. */
 	fsinfo = GetAOCSFileSegInfo(aorel, appendOnlyMetaDataSnapshot, compaction_segno, true);

 	if (AppendOnlyCompaction_ShouldCompact(aorel,
 										   compaction_segno, fsinfo->total_tupcount, isFull,
 										   appendOnlyMetaDataSnapshot))
 	{
 		if (*insert_segno == -1)
 		{
 			/* get the insertion segment on first call. */
 			*insert_segno = ChooseSegnoForCompactionWrite(aorel, avoid_segnos);
 		}

 		if (*insert_segno != -1)
 		{
 			insertDesc = aocs_insert_init(aorel, *insert_segno);

 			AOCSSegmentFileFullCompaction(aorel,
 										  insertDesc,
 										  fsinfo,
 										  appendOnlyMetaDataSnapshot,
 										  vacrelstats);

 			insertDesc->skipModCountIncrement = true;
 			aocs_insert_finish(insertDesc, NULL);
 		}
 		else
 		{
 			/* FIXME: Could not find a target segment. What now? */
 		}
 	}

 	pfree(fsinfo);

 	UnregisterSnapshot(appendOnlyMetaDataSnapshot);
 }
	/*------------------------------------------------------------------------------
	*
	* Code dealing with the compaction of append-only tables.
	*
	* Copyright (c) 2013-Present VMware, Inc. or its affiliates.
	*
	*
	* IDENTIFICATION
	* src/backend/access/aocs/aocs_compaction.c
	*
	*------------------------------------------------------------------------------
	*/

	#include "postgres.h"

	#include <limits.h>

	#include "access/amapi.h"
	#include "access/genam.h"
	#include "access/heapam.h"
	#include "access/transam.h"
	#include "access/aosegfiles.h"
	#include "access/aomd.h"
	#include "access/aocs_compaction.h"
	#include "access/appendonly_compaction.h"
	#include "access/appendonlywriter.h"
	#include "catalog/catalog.h"
	#include "catalog/indexing.h"
	#include "catalog/pg_appendonly.h"
	#include "catalog/pg_attribute_encoding.h"
	#include "cdb/cdbaocsam.h"
	#include "cdb/cdbvars.h"
	#include "commands/vacuum.h"
	#include "executor/executor.h"
	#include "nodes/execnodes.h"
	#include "storage/lmgr.h"
	#include "utils/lsyscache.h"
	#include "utils/memutils.h"
	#include "utils/relcache.h"
	#include "utils/snapmgr.h"
	#include "utils/guc.h"
	#include "miscadmin.h"
	#include "commands/progress.h"
	#include "pgstat.h"

	/*
	* Hook for plugins to get control after move or throw away tuple in
	* AOCSSegmentFileFullCompaction().
	*
	* For example, zombodb will delete the corresponding docs after the tuple
	* is moved or thrown away.
	*/
	aocs_compaction_delete_hook_type aocs_compaction_delete_hook = NULL;

	/*
	* Drops a segment file.
	*
	* Actually, we just truncate the segfile to 0 bytes, to reclaim the space.
	* Before GPDB 6, we used to remove the file, but with WAL replication, we
	* no longer have a convenient function to remove a single segment of a
	* relation. An empty file is as almost as good as a non-existent file. If
	* the relation is dropped later, the code in mdunlink() will remove all
	* segments, including any empty ones we've left behind.
	*/
	void
	AOCSCompaction_DropSegmentFile(Relation aorel, int segno, AOVacuumRelStats *vacrelstats)
	{
	int col;

	Assert(RelationStorageIsAoCols(aorel));

	for (col = 0; col < RelationGetNumberOfAttributes(aorel); col++)
	{
	char filenamepath[MAXPGPATH];
	int pseudoSegNo;
	File fd;
	/* Filenum for the col */
	FileNumber filenum = GetFilenumForAttribute(RelationGetRelid(aorel), col + 1);

	/* Open and truncate the relation segfile */
	MakeAOSegmentFileName(aorel, segno, filenum, &pseudoSegNo, filenamepath);

	elogif(Debug_appendonly_print_compaction, LOG,
	"Drop segment file: "
	"segno %d",
	pseudoSegNo);

	fd = OpenAOSegmentFile(aorel, filenamepath, 0);
	if (fd >= 0)
	{
	TruncateAOSegmentFile(fd, aorel, pseudoSegNo, 0, vacrelstats);
	CloseAOSegmentFile(fd, aorel);
	}
	else
	{
	/*
	* The file we were about to drop/truncate didn't exist. That's normal,
	* for example, if a column is added with ALTER TABLE ADD COLUMN.
	*/
	elog(DEBUG1, "could not truncate segfile %s, because it does not exist", filenamepath);
	}
	}
	}

	/*
	* AOCSSegmentFileTruncateToEOF()
	*
	* Truncates the files for all columns of logical segfile 'segno' to
	* the EOF values from 'vpinfo'. The caller is responsible for locking so
	* that a concurrent backend doesn't write to the segfile while we truncate
	* it.
	*
	* This is used to clean up space left behind by aborted or crashed
	* transactions.
	*/
	void
	AOCSSegmentFileTruncateToEOF(Relation aorel, int segno, AOCSVPInfo vpinfo, AOVacuumRelStats vacrelstats)
	{
	const char *relname = RelationGetRelationName(aorel);
	int j;

	Assert(RelationStorageIsAoCols(aorel));

	for (j = 0; j < vpinfo->nEntry; ++j)
	{
	int64 segeof;
	char filenamepath[MAXPGPATH];
	AOCSVPInfoEntry *entry;
	File fd;
	int32 fileSegNo;
	/* Filenum for the column */
	FileNumber filenum = GetFilenumForAttribute(RelationGetRelid(aorel), j + 1);

	entry = &vpinfo->entry[j];
	segeof = entry->eof;

	/* Open and truncate the relation segfile to its eof */
	MakeAOSegmentFileName(aorel, segno, filenum, &fileSegNo, filenamepath);

	elogif(Debug_appendonly_print_compaction, LOG,
	"Opening AO COL relation \"%s.%s\", relation id %u, relfilenode %u column #%d, logical segment #%d (physical segment file #%d, logical EOF " INT64_FORMAT ")",
	get_namespace_name(RelationGetNamespace(aorel)),
	relname,
	aorel->rd_id,
	aorel->rd_locator.relNumber,
	j,
	segno,
	fileSegNo,
	segeof);

	fd = OpenAOSegmentFile(aorel, filenamepath, segeof);
	if (fd >= 0)
	{
	TruncateAOSegmentFile(fd, aorel, fileSegNo, segeof, vacrelstats);
	CloseAOSegmentFile(fd, aorel);

	elogif(Debug_appendonly_print_compaction, LOG,
	"Successfully truncated AO COL relation \"%s.%s\", relation id %u, relfilenode %u column #%d, logical segment #%d (physical segment file #%d, logical EOF " INT64_FORMAT ")",
	get_namespace_name(RelationGetNamespace(aorel)),
	relname,
	aorel->rd_id,
	aorel->rd_locator.relNumber,
	j,
	segno,
	fileSegNo,
	segeof);
	}
	else
	{
	elogif(Debug_appendonly_print_compaction, LOG,
	"No gp_relation_node entry for AO COL relation \"%s.%s\", relation id %u, relfilenode %u column #%d, logical segment #%d (physical segment file #%d, logical EOF " INT64_FORMAT ")",
	get_namespace_name(RelationGetNamespace(aorel)),
	relname,
	aorel->rd_id,
	aorel->rd_locator.relNumber,
	j,
	segno,
	fileSegNo,
	segeof);
	}
	}
	}

	static void
	AOCSMoveTuple(TupleTableSlot *slot,
	AOCSInsertDesc insertDesc,
	ResultRelInfo *resultRelInfo,
	EState *estate)
	{
	AOTupleId *oldAoTupleId;
	AOTupleId newAoTupleId;

	Assert(resultRelInfo);
	Assert(slot);
	Assert(estate);

	oldAoTupleId = (AOTupleId *) &slot->tts_tid;
	/* Extract all the values of the tuple */
	slot_getallattrs(slot);

	(void) aocs_insert_values(insertDesc,
	slot->tts_values,
	slot->tts_isnull,
	&newAoTupleId);
	memcpy(&slot->tts_tid, &newAoTupleId, sizeof(ItemPointerData));

	/* insert index' tuples if needed */
	if (resultRelInfo->ri_NumIndices > 0)
	{
	ExecInsertIndexTuples(resultRelInfo,
	slot, estate, false, false,
	NULL, NIL, false);
	ResetPerTupleExprContext(estate);
	}

	elogif(Debug_appendonly_print_compaction, DEBUG5,
	"Compaction: Moved tuple (%d," INT64_FORMAT ") -> (%d," INT64_FORMAT ")",
	AOTupleIdGet_segmentFileNum(oldAoTupleId), AOTupleIdGet_rowNum(oldAoTupleId),
	AOTupleIdGet_segmentFileNum(&newAoTupleId), AOTupleIdGet_rowNum(&newAoTupleId));
	}

	/*
	* Subroutine of AOCSCompact().
	*/
	static bool
	AOCSSegmentFileFullCompaction(Relation aorel,
	AOCSInsertDesc insertDesc,
	AOCSFileSegInfo *fsinfo,
	Snapshot snapshot,
	AOVacuumRelStats *vacrelstats)
	{
	const char *relname;
	AppendOnlyVisimap visiMap;
	AOCSScanDesc scanDesc;
	TupleDesc tupDesc;
	TupleTableSlot *slot;
	int compact_segno;
	ResultRelInfo *resultRelInfo;
	EState *estate;
	int64 tupleCount = 0;
	int64 tuplePerPage = INT_MAX;
	ItemPointerData otid;
	AOTupleId *aoTupleId;
	int64 curr_num_dead_tuples = 0;
	int64 prev_num_dead_tuples = 0;
	int64 curr_heap_blks_scanned = 0;
	int64 prev_heap_blks_scanned = 0;

	Assert(Gp_role == GP_ROLE_EXECUTE \|\| Gp_role == GP_ROLE_UTILITY);
	Assert(RelationStorageIsAoCols(aorel));
	Assert(insertDesc);

	compact_segno = fsinfo->segno;
	if (fsinfo->varblockcount > 0)
	{
	tuplePerPage = fsinfo->total_tupcount / fsinfo->varblockcount;
	}
	relname = RelationGetRelationName(aorel);

	AppendOnlyVisimap_Init(&visiMap,
	insertDesc->visimaprelid,
	insertDesc->visimapidxid,
	ShareLock,
	snapshot);

	elogif(Debug_appendonly_print_compaction,
	LOG, "Compact AO segfile %d, relation %sd",
	compact_segno, relname);

	scanDesc = aocs_beginrangescan(aorel,
	snapshot, snapshot,
	&compact_segno, 1);

	tupDesc = RelationGetDescr(aorel);
	slot = MakeSingleTupleTableSlot(tupDesc, &TTSOpsVirtual);
	slot->tts_tableOid = RelationGetRelid(aorel);

	/*
	* We need a ResultRelInfo and an EState so we can use the regular
	* executor's index-entry-making machinery.
	*/
	estate = CreateExecutorState();
	resultRelInfo = makeNode(ResultRelInfo);
	resultRelInfo->ri_RangeTableIndex = 1; /* dummy */
	resultRelInfo->ri_RelationDesc = aorel;
	resultRelInfo->ri_TrigDesc = NULL; /* we don't fire triggers */
	ExecOpenIndices(resultRelInfo, false);
	estate->es_opened_result_relations =
	lappend(estate->es_opened_result_relations, resultRelInfo);

	/*
	* We don't want uniqueness checks to be performed while "insert"ing tuples
	* to a destination segfile during AOCSMoveTuple(). This is to ensure that
	* we can avoid spurious conflicts between the moved tuple and the original
	* tuple.
	*/
	estate->gp_bypass_unique_check = true;

	while (aocs_getnext(scanDesc, ForwardScanDirection, slot))
	{
	CHECK_FOR_INTERRUPTS();

	/*
	* AppendOnlyVisimap_IsVisible() has already been called in aocs_getnext().
	*/
	Assert(AppendOnlyVisimap_IsVisible(&scanDesc->visibilityMap,
	(AOTupleId *) &slot->tts_tid));

	aoTupleId = (AOTupleId *) &slot->tts_tid;
	otid = slot->tts_tid;

	AOCSMoveTuple(slot,
	insertDesc,
	resultRelInfo,
	estate);

	if (aocs_compaction_delete_hook)
	(*aocs_compaction_delete_hook) (aorel, &otid);

	/*
	* Check for vacuum delay point after approximatly a var block
	*/
	tupleCount++;
	if (VacuumCostActive && tupleCount % tuplePerPage == 0)
	{
	vacuum_delay_point();
	}

	/*
	* Report that we are now scanning and compacting segment files.
	*/
	curr_num_dead_tuples = scanDesc->segrowsprocessed + 1 - tupleCount;
	if (curr_num_dead_tuples > prev_num_dead_tuples)
	{
	pgstat_progress_update_param(PROGRESS_VACUUM_NUM_DEAD_TUPLES,
	vacrelstats->num_dead_tuples + curr_num_dead_tuples);
	prev_num_dead_tuples = curr_num_dead_tuples;
	}

	curr_heap_blks_scanned = RelationGuessNumberOfBlocksFromSize(scanDesc->totalBytesRead);
	if (curr_heap_blks_scanned > prev_heap_blks_scanned)
	{
	pgstat_progress_update_param(PROGRESS_VACUUM_HEAP_BLKS_SCANNED,
	curr_heap_blks_scanned);
	prev_heap_blks_scanned = curr_heap_blks_scanned;
	}
	}
	/* Accumulate total number dead tuples */
	vacrelstats->num_dead_tuples += scanDesc->segrowsprocessed - tupleCount;

	MarkAOCSFileSegInfoAwaitingDrop(aorel, compact_segno);

	AppendOnlyVisimap_DeleteSegmentFile(&visiMap,
	compact_segno);

	/* Delete all mini pages of the segment files if block directory exists */
	if (OidIsValid(insertDesc->blkdirrelid))
	{
	AppendOnlyBlockDirectory_DeleteSegmentFile(aorel,
	snapshot,
	compact_segno,
	0);
	}

	elogif(Debug_appendonly_print_compaction, LOG,
	"Finished compaction: "
	"AO segfile %d, relation %s, moved tuple count " INT64_FORMAT,
	compact_segno, relname, tupleCount);

	AppendOnlyVisimap_Finish(&visiMap, NoLock);

	ExecCloseIndices(resultRelInfo);
	FreeExecutorState(estate);

	ExecDropSingleTupleTableSlot(slot);

	aocs_endscan(scanDesc);

	return true;
	}

	/*
	* Performs a compaction of an append-only relation in column-orientation.
	*
	* The compaction segment file should be locked for this transaction in
	* the appendonlywriter.c code.
	*
	* On exit, *insert_segno will be set to the segment that was used as the
	* insertion target. The segfiles listed in 'avoid_segnos' will not be used
	* for insertion.
	*
	* The caller is required to hold either an AccessExclusiveLock (vacuum full)
	* or a ShareLock on the relation.
	*/
	void
	AOCSCompact(Relation aorel,
	int compaction_segno,
	int *insert_segno,
	bool isFull,
	List *avoid_segnos,
	AOVacuumRelStats *vacrelstats)
	{
	const char *relname;
	AOCSInsertDesc insertDesc = NULL;
	AOCSFileSegInfo *fsinfo;
	Snapshot appendOnlyMetaDataSnapshot = RegisterSnapshot(GetCatalogSnapshot(InvalidOid));

	Assert(RelationStorageIsAoCols(aorel));
	Assert(Gp_role == GP_ROLE_EXECUTE \|\| Gp_role == GP_ROLE_UTILITY);

	relname = RelationGetRelationName(aorel);
	elogif(Debug_appendonly_print_compaction, LOG,
	"Compact AO relation %s", relname);

	/* Fetch under the write lock to get latest committed eof. */
	fsinfo = GetAOCSFileSegInfo(aorel, appendOnlyMetaDataSnapshot, compaction_segno, true);

	if (AppendOnlyCompaction_ShouldCompact(aorel,
	compaction_segno, fsinfo->total_tupcount, isFull,
	appendOnlyMetaDataSnapshot))
	{
	if (*insert_segno == -1)
	{
	/* get the insertion segment on first call. */
	*insert_segno = ChooseSegnoForCompactionWrite(aorel, avoid_segnos);
	}

	if (*insert_segno != -1)
	{
	insertDesc = aocs_insert_init(aorel, *insert_segno);

	AOCSSegmentFileFullCompaction(aorel,
	insertDesc,
	fsinfo,
	appendOnlyMetaDataSnapshot,
	vacrelstats);

	insertDesc->skipModCountIncrement = true;
	aocs_insert_finish(insertDesc, NULL);
	}
	else
	{
	/* FIXME: Could not find a target segment. What now? */
	}
	}

	pfree(fsinfo);

	UnregisterSnapshot(appendOnlyMetaDataSnapshot);
	}