Changelog: handle VACUUM PARALLEL option

Postgres 13 added a new VACUUM option, PARALLEL. It is now supported
in our code as well.

Relevant changelog message on postgres:
Allow VACUUM to process indexes in parallel (Masahiko Sawada, Amit Kapila)

src/backend/distributed/commands/vacuum.c

@@ -29,7 +29,10 @@
 #include "storage/lmgr.h"
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
+#include "postmaster/bgworker_internals.h"
 
+
+#define VACUUM_PARALLEL_NOTSET -2
 /*
  * Subset of VacuumParams we care about
  */
@@ -40,8 +43,11 @@ typedef struct CitusVacuumParams
 	VacOptTernaryValue truncate;
 	VacOptTernaryValue index_cleanup;
 #endif
-} CitusVacuumParams;
 
+#if PG_VERSION_NUM >= PG_VERSION_13
+	int nworkers;
+#endif
+} CitusVacuumParams;
 
 /* Local functions forward declarations for processing distributed table commands */
 static bool IsDistributedVacuumStmt(int vacuumOptions, List *vacuumRelationIdList);
@@ -285,6 +291,9 @@ DeparseVacuumStmtPrefix(CitusVacuumParams vacuumParams)
 		&& vacuumParams.truncate == VACOPT_TERNARY_DEFAULT &&
 		vacuumParams.index_cleanup == VACOPT_TERNARY_DEFAULT
 #endif
+#if PG_VERSION_NUM >= PG_VERSION_13
+		&& vacuumParams.nworkers == VACUUM_PARALLEL_NOTSET
+#endif		
 		)
 	{
 		return vacuumPrefix->data;
@@ -341,6 +350,12 @@ DeparseVacuumStmtPrefix(CitusVacuumParams vacuumParams)
 	}
 #endif
 
+#if PG_VERSION_NUM >= PG_VERSION_13
+	if (vacuumParams.nworkers != VACUUM_PARALLEL_NOTSET) {
+		appendStringInfo(vacuumPrefix, "PARALLEL %d,", vacuumParams.nworkers);
+	}
+#endif
+
 	vacuumPrefix->data[vacuumPrefix->len - 1] = ')';
 
 	appendStringInfoChar(vacuumPrefix, ' ');
@@ -421,6 +436,8 @@ ExtractVacuumTargetRels(VacuumStmt *vacuumStmt)
 
 /*
  * This is mostly ExecVacuum from Postgres's commands/vacuum.c
+ * Note that ExecVacuum does an actual vacuum as well and we don't want
+ * that to happen in the coordinator hence we copied the rest here.
  */
 static CitusVacuumParams
 VacuumStmtParams(VacuumStmt *vacstmt)
@@ -436,6 +453,9 @@ VacuumStmtParams(VacuumStmt *vacstmt)
 	/* Set default value */
 	params.index_cleanup = VACOPT_TERNARY_DEFAULT;
 	params.truncate = VACOPT_TERNARY_DEFAULT;
+	#if PG_VERSION_NUM >= PG_VERSION_13
+		params.nworkers = VACUUM_PARALLEL_NOTSET;
+	#endif
 
 	/* Parse options list */
 	DefElem *opt = NULL;
@@ -484,6 +504,30 @@ VacuumStmtParams(VacuumStmt *vacstmt)
 			params.truncate = defGetBoolean(opt) ? VACOPT_TERNARY_ENABLED :
 							  VACOPT_TERNARY_DISABLED;
 		}
+		#if PG_VERSION_NUM >= PG_VERSION_13
+		else if (strcmp(opt->defname, "parallel") == 0) {
+			
+			if (opt->arg == NULL)
+			{
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("parallel option requires a value between 0 and %d",
+								MAX_PARALLEL_WORKER_LIMIT)));
+			}
+			else
+			{
+				int			nworkers;
+				nworkers = defGetInt32(opt);
+				if (nworkers < 0 || nworkers > MAX_PARALLEL_WORKER_LIMIT)
+					ereport(ERROR,
+							(errcode(ERRCODE_SYNTAX_ERROR),
+							 errmsg("parallel vacuum degree must be between 0 and %d",
+									MAX_PARALLEL_WORKER_LIMIT)));
+
+				params.nworkers = nworkers;
+			}
+		}
+		#endif
 		else
 		{
 			ereport(ERROR,

src/test/regress/expected/pg13.out

@@ -0,0 +1,36 @@
+create schema test_pg13;
+set search_path to test_pg13;
+SET citus.shard_replication_factor to 1;
+SET citus.shard_count to 2;
+SET citus.next_shard_id TO 65000;
+CREATE TABLE dist_table (name char, age int);
+CREATE INDEX name_index on dist_table(name);
+SELECT create_distributed_table('dist_table', 'name');
+ create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
+SET client_min_messages to DEBUG1;
+SET citus.log_remote_commands to ON;
+-- make sure vacuum parallel doesn't error out
+VACUUM (PARALLEL 2) dist_table;
+NOTICE:  issuing VACUUM (PARALLEL 2) test_pg13.dist_table_65000
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing VACUUM (PARALLEL 2) test_pg13.dist_table_65001
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+VACUUM (PARALLEL 0) dist_table;
+NOTICE:  issuing VACUUM (PARALLEL 0) test_pg13.dist_table_65000
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing VACUUM (PARALLEL 0) test_pg13.dist_table_65001
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+-- This should error out since -5 is not valid.
+VACUUM (PARALLEL -5) dist_table;
+ERROR:  parallel vacuum degree must be between 0 and 1024
+-- This should error out since no number is given
+VACUUM (PARALLEL) dist_table;
+ERROR:  parallel option requires a value between 0 and 1024
+RESET client_min_messages;
+RESET citus.log_remote_commands;
+drop schema test_pg13 cascade;
+NOTICE:  drop cascades to table dist_table

src/test/regress/multi_schedule

@@ -80,7 +80,7 @@ test: set_operation_and_local_tables
 
 test: subqueries_deep subquery_view subquery_partitioning subquery_complex_target_list subqueries_not_supported subquery_in_where
 test: non_colocated_leaf_subquery_joins non_colocated_subquery_joins non_colocated_join_order
-test: subquery_prepared_statements pg12 cte_inline
+test: subquery_prepared_statements pg12 cte_inline pg13
 
 # ----------
 # Miscellaneous tests to check our query planning behavior

src/test/regress/sql/pg13.sql

@@ -0,0 +1,27 @@
+
+create schema test_pg13;
+set search_path to test_pg13;
+
+SET citus.shard_replication_factor to 1;
+SET citus.shard_count to 2;
+SET citus.next_shard_id TO 65000;
+
+CREATE TABLE dist_table (name char, age int);
+CREATE INDEX name_index on dist_table(name);
+
+SELECT create_distributed_table('dist_table', 'name');
+
+SET client_min_messages to DEBUG1;
+SET citus.log_remote_commands to ON;
+-- make sure vacuum parallel doesn't error out
+VACUUM (PARALLEL 2) dist_table;
+VACUUM (PARALLEL 0) dist_table;
+-- This should error out since -5 is not valid.
+VACUUM (PARALLEL -5) dist_table;
+-- This should error out since no number is given
+VACUUM (PARALLEL) dist_table;
+
+RESET client_min_messages;
+RESET citus.log_remote_commands;
+
+drop schema test_pg13 cascade;