VACUUM should cleanup aosegs/aocssegs in AOSEG_STATE_AWAITING_DROP state

Due to the different phases of VACUUM, compaction phase could have
committed but the drop was cancelled. This could leave some
aoseg/aocsseg entries in AOSEG_STATE_AWAITING_DROP state which could
then be cleaned up in a subsequent VACUUM as expected... but it
currently just skips. The master keeps a hash table (AppendOnlyHash)
and uses it during VACUUM to check tuple count, state, and
aborted/committed. Our issue here is that state checking was only
being done against state AWAITING_DROP_READY. It was not including
COMPACTED_AWAITING_DROP in the state check. This commit adds the check
and updates the detailed comment block on the state transitions in
appendonlywriter.h.

src/backend/access/appendonly/appendonlywriter.c

@@ -857,8 +857,9 @@ SetSegnoForCompaction(Relation rel,
 			(errmsg("segment file %i for append-only relation \"%s\" (%d): state %d",
 					i, RelationGetRelationName(rel), RelationGetRelid(rel), segfilestat->state)));
 
-		if (segfilestat->state == AWAITING_DROP_READY && 
-				!usedByConcurrentTransaction(segfilestat, i))
+		if ((segfilestat->state == AWAITING_DROP_READY ||
+			segfilestat->state == COMPACTED_AWAITING_DROP) &&
+			!usedByConcurrentTransaction(segfilestat, i))
 		{
 			ereportif(Debug_appendonly_print_segfile_choice, LOG,
 				(errmsg("Found segment awaiting drop for append-only relation \"%s\" (%d)",

src/include/access/appendonlywriter.h

@@ -89,13 +89,14 @@ extern int	MaxAppendOnlyTables;	/* Max # of concurrently used AO rels */
  * The valid state transitions are:
  * 1 -> 3 (normal insertion or compact insertion)
  * 1 -> 4 (compaction)
- * 2 -> 7 (pseude compaction transaction started)
+ * 2 -> 7 (pseudo compaction transaction started)
  * 3 -> 1 (insertion committed or aborted)
  * 4 -> 6 (compaction transaction committed)
  * 4 -> 1 (compaction transaction aborted)
  * 5 -> 1 (drop transaction committed)
  * 5 -> 2 (drop transaction aborted)
  * 6 -> 5 (drop transaction started)
+ * 6 -> 7 (pseudo compaction transaction started)
  * 6 -> 8 (drop transaction skipped)
  * 7 -> 6 (pseudo compaction committed)
  * 7 -> 2 (pseudo compaction aborted)

src/test/isolation2/expected/vacuum_after_vacuum_skip_drop_column.out

+-- @Description Ensures next vacuum drops segfiles in AOSEG_STATE_AWAITING_DROP
+--              left over by a previous vacuum
+--
+CREATE TABLE aoco_vacuum_after_vacuum_skip_drop (a INT, b INT) WITH (appendonly=true, orientation=column);
+CREATE
+INSERT INTO aoco_vacuum_after_vacuum_skip_drop SELECT i as a, i as b FROM generate_series(1, 10) AS i;
+INSERT 10
+
+DELETE FROM aoco_vacuum_after_vacuum_skip_drop;
+DELETE 10
+
+-- We should see all aocssegs in state 1
+2U: SELECT segno, column_num, state FROM gp_toolkit.__gp_aocsseg_name('aoco_vacuum_after_vacuum_skip_drop');
+segno|column_num|state
+-----+----------+-----
+1    |0         |1    
+1    |1         |1    
+(2 rows)
+
+-- VACUUM while another session holds lock
+1: BEGIN;
+BEGIN
+1: SELECT COUNT(*) FROM aoco_vacuum_after_vacuum_skip_drop;
+count
+-----
+0    
+(1 row)
+2: VACUUM aoco_vacuum_after_vacuum_skip_drop;
+VACUUM
+1: END;
+END
+
+-- We should see an aocsseg in state 2 (AOSEG_STATE_AWAITING_DROP)
+2U: SELECT segno, column_num, state FROM gp_toolkit.__gp_aocsseg_name('aoco_vacuum_after_vacuum_skip_drop');
+segno|column_num|state
+-----+----------+-----
+1    |0         |2    
+1    |1         |2    
+2    |0         |1    
+2    |1         |1    
+(4 rows)
+
+-- The VACUUM should clean up aocssegs in state 2 (AOSEG_STATE_AWAITING_DROP)
+1: VACUUM aoco_vacuum_after_vacuum_skip_drop;
+VACUUM
+2U: SELECT segno, column_num, state FROM gp_toolkit.__gp_aocsseg_name('aoco_vacuum_after_vacuum_skip_drop');
+segno|column_num|state
+-----+----------+-----
+1    |0         |1    
+1    |1         |1    
+2    |0         |1    
+2    |1         |1    
+(4 rows)
+
+-- Check if insert goes into segno 1 instead of segno 2
+1: INSERT INTO aoco_vacuum_after_vacuum_skip_drop SELECT i as a, i as b FROM generate_series(1, 100) AS i;
+INSERT 100
+2U: SELECT segno, tupcount > 0, state FROM gp_toolkit.__gp_aocsseg_name('aoco_vacuum_after_vacuum_skip_drop');
+segno|?column?|state
+-----+--------+-----
+1    |t       |1    
+1    |t       |1    
+2    |f       |1    
+2    |f       |1    
+(4 rows)

src/test/isolation2/isolation2_schedule

@@ -96,3 +96,4 @@ test: uao/vacuum_self_serializable3_column
 test: uao/vacuum_while_insert_column
 test: uao/vacuum_while_vacuum_column
 test: add_column_after_vacuum_skip_drop_column
+test: vacuum_after_vacuum_skip_drop_column

src/test/isolation2/sql/vacuum_after_vacuum_skip_drop_column.sql

+-- @Description Ensures next vacuum drops segfiles in AOSEG_STATE_AWAITING_DROP
+--              left over by a previous vacuum
+--
+CREATE TABLE aoco_vacuum_after_vacuum_skip_drop (a INT, b INT) WITH (appendonly=true, orientation=column);
+INSERT INTO aoco_vacuum_after_vacuum_skip_drop SELECT i as a, i as b FROM generate_series(1, 10) AS i;
+
+DELETE FROM aoco_vacuum_after_vacuum_skip_drop;
+
+-- We should see all aocssegs in state 1
+2U: SELECT segno, column_num, state FROM gp_toolkit.__gp_aocsseg_name('aoco_vacuum_after_vacuum_skip_drop');
+
+-- VACUUM while another session holds lock
+1: BEGIN;
+1: SELECT COUNT(*) FROM aoco_vacuum_after_vacuum_skip_drop;
+2: VACUUM aoco_vacuum_after_vacuum_skip_drop;
+1: END;
+
+-- We should see an aocsseg in state 2 (AOSEG_STATE_AWAITING_DROP)
+2U: SELECT segno, column_num, state FROM gp_toolkit.__gp_aocsseg_name('aoco_vacuum_after_vacuum_skip_drop');
+
+-- The VACUUM should clean up aocssegs in state 2 (AOSEG_STATE_AWAITING_DROP)
+1: VACUUM aoco_vacuum_after_vacuum_skip_drop;
+2U: SELECT segno, column_num, state FROM gp_toolkit.__gp_aocsseg_name('aoco_vacuum_after_vacuum_skip_drop');
+
+-- Check if insert goes into segno 1 instead of segno 2
+1: INSERT INTO aoco_vacuum_after_vacuum_skip_drop SELECT i as a, i as b FROM generate_series(1, 100) AS i;
+2U: SELECT segno, tupcount > 0, state FROM gp_toolkit.__gp_aocsseg_name('aoco_vacuum_after_vacuum_skip_drop');