Use interface for storage interactions in Appendonly TAM

gpcontrib/gp_replica_check/gp_replica_check.py

@@ -37,6 +37,7 @@
     import subprocess
 import threading
 import pipes  # for shell-quoting, pipes.quote()
+import time
 
 class ReplicaCheck(threading.Thread):
     def __init__(self, segrow, datname, relation_types):
@@ -58,7 +59,50 @@ def __str__(self):
 Mirror Data Directory Location: %s' % (self.getName(), self.host, self.port, self.datname,
                                           self.ploc, self.mloc)
 
+    def wait_for_wal_sync(self):
+        cmd = "PGOPTIONS='-c gp_role=utility' psql -h %s -p %s -d %s -t -A -c \"SELECT pg_current_wal_lsn() AS master_wal, replay_lsn AS standby_wal, pg_current_wal_lsn() = replay_lsn AS are_equal FROM pg_stat_replication;\"" % (self.host, self.port, pipes.quote(self.datname))
+        while True:
+            try:
+                output = subprocess.check_output(cmd, shell=True).decode().strip().split("\n")
+                print(f"Debug - Full output: {output}")  # Debug print
+
+                if not output:
+                    print("No output received from psql command.")
+                    time.sleep(5)
+                    continue
+
+                # With -t and -A options, we should get only one line of data
+                data_row = output[0].split("|")
+                if len(data_row) != 3:
+                    print(f"Unexpected data row format. Data row: {data_row}")
+                    time.sleep(5)
+                    continue
+
+                master_wal = data_row[0].strip()
+                standby_wal = data_row[1].strip()
+                are_equal = data_row[2].strip().lower() == "t"
+
+                print(f"Debug - Parsed values: master_wal={master_wal}, standby_wal={standby_wal}, are_equal={are_equal}")  # Debug print
+
+                if are_equal:
+                    print("WAL sync achieved.")
+                    break
+                else:
+                    print(f"Waiting for WAL sync. Current status: master={master_wal}, standby={standby_wal}")
+            except subprocess.CalledProcessError as e:
+                with self.lock:
+                    print(f"Error executing command: {e.cmd}")
+                    print(f"Return code: {e.returncode}")
+                    print(f"Output: {e.output}")
+            except Exception as e:
+                print(f"Unexpected error in wait_for_wal_sync: {str(e)}")
+
+            # Add a small delay before the next attempt
+            time.sleep(5)
+
+
     def run(self):
+        self.wait_for_wal_sync();
         cmd = '''PGOPTIONS='-c gp_role=utility' psql -h %s -p %s -c "select * from gp_replica_check('%s', '%s', '%s')" %s''' % (self.host, self.port,
                                                                                                                                         self.ploc, self.mloc,
                                                                                                                                         self.relation_types,

src/backend/access/aocs/aocs_compaction.c

@@ -39,6 +39,8 @@
 #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 
@@ -60,7 +62,7 @@ aocs_compaction_delete_hook_type aocs_compaction_delete_hook = NULL;
  * segments, including any empty ones we've left behind.
  */
 void
-AOCSCompaction_DropSegmentFile(Relation aorel, int segno)
+AOCSCompaction_DropSegmentFile(Relation aorel, int segno, AOVacuumRelStats *vacrelstats)
 {
 	int			col;
 
@@ -83,8 +85,8 @@ AOCSCompaction_DropSegmentFile(Relation aorel, int segno)
 		fd = OpenAOSegmentFile(aorel, filenamepath, 0);
 		if (fd >= 0)
 		{
-			TruncateAOSegmentFile(fd, aorel, pseudoSegNo, 0);
-			CloseAOSegmentFile(fd);
+			TruncateAOSegmentFile(fd, aorel, pseudoSegNo, 0, vacrelstats);
+			CloseAOSegmentFile(fd, aorel);
 		}
 		else
 		{
@@ -109,7 +111,7 @@ AOCSCompaction_DropSegmentFile(Relation aorel, int segno)
  * transactions.
  */
 void
-AOCSSegmentFileTruncateToEOF(Relation aorel, int segno, AOCSVPInfo *vpinfo)
+AOCSSegmentFileTruncateToEOF(Relation aorel, int segno, AOCSVPInfo *vpinfo, AOVacuumRelStats *vacrelstats)
 {
 	const char *relname = RelationGetRelationName(aorel);
 	int			j;
@@ -146,8 +148,8 @@ AOCSSegmentFileTruncateToEOF(Relation aorel, int segno, AOCSVPInfo *vpinfo)
 		fd = OpenAOSegmentFile(aorel, filenamepath, segeof);
 		if (fd >= 0)
 		{
-			TruncateAOSegmentFile(fd, aorel, fileSegNo, segeof);
-			CloseAOSegmentFile(fd);
+			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 %lu column #%d, logical segment #%d (physical segment file #%d, logical EOF " INT64_FORMAT ")",
@@ -221,22 +223,26 @@ static bool
 AOCSSegmentFileFullCompaction(Relation aorel,
 							  AOCSInsertDesc insertDesc,
 							  AOCSFileSegInfo *fsinfo,
-							  Snapshot snapshot)
+							  Snapshot snapshot,
+							  AOVacuumRelStats *vacrelstats)
 {
 	const char *relname;
 	AppendOnlyVisimap visiMap;
 	AOCSScanDesc scanDesc;
 	TupleDesc	tupDesc;
 	TupleTableSlot *slot;
 	int			compact_segno;
-	int64		movedTupleCount = 0;
 	ResultRelInfo *resultRelInfo;
 	MemTupleBinding *mt_bind;
 	EState	   *estate;
-	AOTupleId  *aoTupleId;
-	ItemPointerData otid;
 	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(RelationIsAoCols(aorel));
@@ -294,21 +300,19 @@ AOCSSegmentFileFullCompaction(Relation aorel,
 	{
 		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;
-		if (AppendOnlyVisimap_IsVisible(&scanDesc->visibilityMap, aoTupleId))
-		{
-			AOCSMoveTuple(slot,
-						  insertDesc,
-						  resultRelInfo,
-						  estate);
-			movedTupleCount++;
-		}
-		else
-		{
-			/* Tuple is invisible and needs to be dropped */
-			AppendOnlyThrowAwayTuple(aorel, slot, mt_bind);
-		}
+
+		AOCSMoveTuple(slot,
+					  insertDesc,
+					  resultRelInfo,
+					  estate);
 
 		if (aocs_compaction_delete_hook)
 			(*aocs_compaction_delete_hook) (aorel, &otid);
@@ -321,7 +325,28 @@ AOCSSegmentFileFullCompaction(Relation aorel,
 		{
 			vacuum_delay_point();
 		}
+
+		/*
+		 * Report that we are now scanning and compacting segment files.
+		 */
+		curr_num_dead_tuples = scanDesc->cur_seg_row + 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->cur_seg_row - tupleCount;
 
 	MarkAOCSFileSegInfoAwaitingDrop(aorel, compact_segno);
 
@@ -340,7 +365,7 @@ AOCSSegmentFileFullCompaction(Relation aorel,
 	elogif(Debug_appendonly_print_compaction, LOG,
 		   "Finished compaction: "
 		   "AO segfile %d, relation %s, moved tuple count " INT64_FORMAT,
-		   compact_segno, relname, movedTupleCount);
+		   compact_segno, relname, tupleCount);
 
 	AppendOnlyVisimap_Finish(&visiMap, NoLock);
 
@@ -361,7 +386,7 @@ AOCSSegmentFileFullCompaction(Relation aorel,
  * The compaction segment file should be locked for this transaction in
  * the appendonlywriter.c code.
  *
- * On exit, *insert_segno will be set to the the segment that was used as the
+ * 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.
  *
@@ -373,7 +398,8 @@ AOCSCompact(Relation aorel,
 			int compaction_segno,
 			int *insert_segno,
 			bool isFull,
-			List *avoid_segnos)
+			List *avoid_segnos,
+			AOVacuumRelStats *vacrelstats)
 {
 	const char *relname;
 	AOCSInsertDesc insertDesc = NULL;
@@ -407,7 +433,8 @@ AOCSCompact(Relation aorel,
 			AOCSSegmentFileFullCompaction(aorel,
 										  insertDesc,
 										  fsinfo,
-										  appendOnlyMetaDataSnapshot);
+										  appendOnlyMetaDataSnapshot,
+										  vacrelstats);
 
 			insertDesc->skipModCountIncrement = true;
 			aocs_insert_finish(insertDesc, NULL);