Introduces IsReindexWithParam_compat macro

In ReindexStmt concurrent field is moved to options and then options are converted to params list.
This macro uses previous fields for previous versions and the new params list with a new function named IsReindexWithParam for PG14

Relevant PG commits:
844c05abc3f1c1703bf17cf44ab66351ed9711d2
b5913f6120792465f4394b93c15c2e2ac0c08376

src/backend/distributed/commands/index.c

@@ -529,8 +529,8 @@ PreprocessReindexStmt(Node *node, const char *reindexCommand,
 	{
 		Relation relation = NULL;
 		Oid relationId = InvalidOid;
-		LOCKMODE lockmode = reindexStatement->concurrent ? ShareUpdateExclusiveLock :
-							AccessExclusiveLock;
+		LOCKMODE lockmode = IsReindexWithParam_compat(reindexStatement, "concurrently") ?
+							ShareUpdateExclusiveLock : AccessExclusiveLock;
 		MemoryContext relationContext = NULL;
 
 		Assert(reindexStatement->kind == REINDEX_OBJECT_INDEX ||
@@ -539,7 +539,8 @@ PreprocessReindexStmt(Node *node, const char *reindexCommand,
 		if (reindexStatement->kind == REINDEX_OBJECT_INDEX)
 		{
 			struct ReindexIndexCallbackState state;
-			state.concurrent = reindexStatement->concurrent;
+			state.concurrent = IsReindexWithParam_compat(reindexStatement,
+														 "concurrently");
 			state.locked_table_oid = InvalidOid;
 
 			Oid indOid = RangeVarGetRelidExtended(reindexStatement->relation,
@@ -590,8 +591,10 @@ PreprocessReindexStmt(Node *node, const char *reindexCommand,
 		{
 			DDLJob *ddlJob = palloc0(sizeof(DDLJob));
 			ddlJob->targetRelationId = relationId;
-			ddlJob->concurrentIndexCmd = reindexStatement->concurrent;
-			ddlJob->startNewTransaction = reindexStatement->concurrent;
+			ddlJob->concurrentIndexCmd = IsReindexWithParam_compat(reindexStatement,
+																   "concurrently");
+			ddlJob->startNewTransaction = IsReindexWithParam_compat(reindexStatement,
+																	"concurrently");
 			ddlJob->commandString = reindexCommand;
 			ddlJob->taskList = CreateReindexTaskList(relationId, reindexStatement);
 

src/backend/distributed/deparser/citus_ruleutils.c

@@ -39,6 +39,7 @@
 #include "commands/defrem.h"
 #include "commands/extension.h"
 #include "distributed/citus_ruleutils.h"
+#include "distributed/commands.h"
 #include "distributed/listutils.h"
 #include "distributed/multi_partitioning_utils.h"
 #include "distributed/metadata_cache.h"
@@ -740,7 +741,8 @@ deparse_shard_reindex_statement(ReindexStmt *origStmt, Oid distrelid, int64 shar
 {
 	ReindexStmt *reindexStmt = copyObject(origStmt); /* copy to avoid modifications */
 	char *relationName = NULL;
-	const char *concurrentlyString = reindexStmt->concurrent ? "CONCURRENTLY " : "";
+	const char *concurrentlyString =
+		IsReindexWithParam_compat(reindexStmt, "concurrently") ? "CONCURRENTLY " : "";
 
 
 	if (reindexStmt->kind == REINDEX_OBJECT_INDEX ||
@@ -754,7 +756,7 @@ deparse_shard_reindex_statement(ReindexStmt *origStmt, Oid distrelid, int64 shar
 
 	appendStringInfoString(buffer, "REINDEX ");
 
-	if (reindexStmt->options == REINDEXOPT_VERBOSE)
+	if (IsReindexWithParam_compat(reindexStmt, "verbose"))
 	{
 		appendStringInfoString(buffer, "(VERBOSE) ");
 	}
@@ -800,6 +802,32 @@ deparse_shard_reindex_statement(ReindexStmt *origStmt, Oid distrelid, int64 shar
 	}
 }
 
+/*
+ * IsReindexWithParam_compat returns true if the given parameter
+ * exists for the given reindexStmt.
+ */
+bool IsReindexWithParam_compat(ReindexStmt* reindexStmt, char* param) {
+#if PG_VERSION_NUM < PG_VERSION_14
+	if (strcmp(param, "concurrently") == 0) {
+		return reindexStmt->concurrent;
+	}else if (strcmp(param, "verbose") == 0) {
+		return reindexStmt->options & REINDEXOPT_VERBOSE;
+	}
+	return false;
+#else
+	DefElem *opt = NULL;
+	foreach_ptr(opt, reindexStmt->params)
+	{
+		if (strcmp(opt->defname, param) == 0)
+		{
+			return defGetBoolean(opt);
+		}
+	}
+	return false;	
+#endif
+
+} 
+
 
 /* deparse_index_columns appends index or include parameters to the provided buffer */
 static void

src/include/distributed/commands.h

@@ -288,6 +288,7 @@ extern void ErrorIfUnsupportedAlterIndexStmt(AlterTableStmt *alterTableStatement
 extern void MarkIndexValid(IndexStmt *indexStmt);
 extern List * ExecuteFunctionOnEachTableIndex(Oid relationId, PGIndexProcessor
 											  pgIndexProcessor, int flags);
+extern bool IsReindexWithParam_compat(ReindexStmt *stmt, char *paramName);
 
 /* objectaddress.c - forward declarations */
 extern ObjectAddress CreateExtensionStmtObjectAddress(Node *stmt, bool missing_ok);

src/include/distributed/version_compat.h

@@ -47,6 +47,7 @@
 #define VACOPTVALUE_UNSPECIFIED_COMPAT VACOPTVALUE_UNSPECIFIED
 #define VACOPTVALUE_DISABLED_COMPAT VACOPTVALUE_DISABLED
 #define VACOPTVALUE_ENABLED_COMPAT VACOPTVALUE_ENABLED
+#define IsReindexWithParam_compat(reindex, param) IsReindexWithParam(reindex, param)
 #else
 #define AlterTableStmtObjType(a) ((a)->relkind)
 #define F_NEXTVAL_COMPAT F_NEXTVAL_OID
@@ -64,6 +65,10 @@
 #define VACOPTVALUE_UNSPECIFIED_COMPAT VACOPT_TERNARY_DEFAULT
 #define VACOPTVALUE_DISABLED_COMPAT VACOPT_TERNARY_DISABLED
 #define VACOPTVALUE_ENABLED_COMPAT VACOPT_TERNARY_ENABLED
+#define IsReindexWithParam_compat(reindex, param) \
+	((strcmp(param, "concurrently") == 0) ? ((reindex)->concurrent) : \
+	 ((strcmp(param, "verbose") == 0) ? ((reindex)->options == REINDEXOPT_VERBOSE) : \
+	  false))
 #endif
 
 #if PG_VERSION_NUM >= PG_VERSION_13