Support some of the DDLs for schema-based-sharding from any node

configure

@@ -1,6 +1,6 @@
 #! /bin/sh
 # Guess values for system-dependent variables and create Makefiles.
-# Generated by GNU Autoconf 2.69 for Citus 14.0.0.
+# Generated by GNU Autoconf 2.69 for Citus 14.1.0.
 #
 #
 # Copyright (C) 1992-1996, 1998-2012 Free Software Foundation, Inc.
@@ -579,8 +579,8 @@ MAKEFLAGS=
 # Identity of this package.
 PACKAGE_NAME='Citus'
 PACKAGE_TARNAME='citus'
-PACKAGE_VERSION='14.0.0'
-PACKAGE_STRING='Citus 14.0.0'
+PACKAGE_VERSION='14.1.0'
+PACKAGE_STRING='Citus 14.1.0'
 PACKAGE_BUGREPORT=''
 PACKAGE_URL=''
 
@@ -1262,7 +1262,7 @@ if test "$ac_init_help" = "long"; then
   # Omit some internal or obsolete options to make the list less imposing.
   # This message is too long to be a string in the A/UX 3.1 sh.
   cat <<_ACEOF
-\`configure' configures Citus 14.0.0 to adapt to many kinds of systems.
+\`configure' configures Citus 14.1.0 to adapt to many kinds of systems.
 
 Usage: $0 [OPTION]... [VAR=VALUE]...
 
@@ -1324,7 +1324,7 @@ fi
 
 if test -n "$ac_init_help"; then
   case $ac_init_help in
-     short | recursive ) echo "Configuration of Citus 14.0.0:";;
+     short | recursive ) echo "Configuration of Citus 14.1.0:";;
    esac
   cat <<\_ACEOF
 
@@ -1429,7 +1429,7 @@ fi
 test -n "$ac_init_help" && exit $ac_status
 if $ac_init_version; then
   cat <<\_ACEOF
-Citus configure 14.0.0
+Citus configure 14.1.0
 generated by GNU Autoconf 2.69
 
 Copyright (C) 2012 Free Software Foundation, Inc.
@@ -1912,7 +1912,7 @@ cat >config.log <<_ACEOF
 This file contains any messages produced by compilers while
 running configure, to aid debugging if configure makes a mistake.
 
-It was created by Citus $as_me 14.0.0, which was
+It was created by Citus $as_me 14.1.0, which was
 generated by GNU Autoconf 2.69.  Invocation command line was
 
   $ $0 $@
@@ -5395,7 +5395,7 @@ cat >>$CONFIG_STATUS <<\_ACEOF || ac_write_fail=1
 # report actual input values of CONFIG_FILES etc. instead of their
 # values after options handling.
 ac_log="
-This file was extended by Citus $as_me 14.0.0, which was
+This file was extended by Citus $as_me 14.1.0, which was
 generated by GNU Autoconf 2.69.  Invocation command line was
 
   CONFIG_FILES    = $CONFIG_FILES
@@ -5457,7 +5457,7 @@ _ACEOF
 cat >>$CONFIG_STATUS <<_ACEOF || ac_write_fail=1
 ac_cs_config="`$as_echo "$ac_configure_args" | sed 's/^ //; s/[\\""\`\$]/\\\\&/g'`"
 ac_cs_version="\\
-Citus config.status 14.0.0
+Citus config.status 14.1.0
 configured by $0, generated by GNU Autoconf 2.69,
   with options \\"\$ac_cs_config\\"
 
@@ -6162,4 +6162,3 @@ if test -n "$ac_unrecognized_opts" && test "$enable_option_checking" != no; then
   { $as_echo "$as_me:${as_lineno-$LINENO}: WARNING: unrecognized options: $ac_unrecognized_opts" >&5
 $as_echo "$as_me: WARNING: unrecognized options: $ac_unrecognized_opts" >&2;}
 fi
-

configure.ac

@@ -5,7 +5,7 @@
 # everyone needing autoconf installed, the resulting files are checked
 # into the SCM.
 
-AC_INIT([Citus], [14.0.0])
+AC_INIT([Citus], [14.1.0])
 AC_COPYRIGHT([Copyright (c) Citus Data, Inc.])
 
 # we'll need sed and awk for some of the version commands

src/backend/columnar/citus_columnar.control

@@ -1,6 +1,6 @@
 # Columnar extension
 comment = 'Citus Columnar extension'
-default_version = '14.0-1'
+default_version = '14.1-1'
 module_pathname = '$libdir/citus_columnar'
 relocatable = false
 schema = pg_catalog

src/backend/columnar/mod.c

@@ -22,7 +22,7 @@
 
 
 #if PG_VERSION_NUM >= PG_VERSION_18
-PG_MODULE_MAGIC_EXT(.name = "citus_columnar", .version = "14.0.0");
+PG_MODULE_MAGIC_EXT(.name = "citus_columnar", .version = "14.1.0");
 #else
 PG_MODULE_MAGIC;
 #endif

src/backend/columnar/sql/citus_columnar--14.0-1--14.1-1.sql

@@ -0,0 +1,2 @@
+-- citus_columnar--14.0-1--14.1-1
+-- bump version to 14.1-1

src/backend/columnar/sql/downgrades/citus_columnar--14.1-1--14.0-1.sql

@@ -0,0 +1,2 @@
+-- citus_columnar--14.1-1--14.0-1
+-- downgrade version to 14.0-1

src/backend/distributed/README.md

@@ -1819,13 +1819,13 @@ Each field in the struct is documented in the comments within the `DistributeObj
 
 - **Returning tasks for `preprocess` and `postprocess`**: Ensure that either `preprocess` or `postprocess` returns a list of "DDLJob"s. If both functions return non-empty lists, then you would get an assertion failure.
 
-- **Generic `preprocess` and `postprocess` methods**: The generic methods, `PreprocessAlterDistributedObjectStmt` and `PostprocessAlterDistributedObjectStmt`, serve as generic pre and post methods utilized for various statements. Both of these methods find application in distributed object operations.
+- **Generic `preprocess` and `postprocess` methods**: The generic methods, `PreprocessAlterDistributedObjectStmtFromCoordinator` and `PostprocessAlterDistributedObjectStmtFromCoordinator`, serve as generic pre and post methods utilized for various statements. Both of these methods find application in distributed object operations.
 
-  - The `PreprocessAlterDistributedObjectStmt` method carries out the following operations:
+  - The `PreprocessAlterDistributedObjectStmtFromCoordinator` method carries out the following operations:
     - Performs a qualification operation.
     - Deparses the statement and generates a task list.
 
-  - As for the `PostprocessAlterDistributedObjectStmt` method, it:
+  - As for the `PostprocessAlterDistributedObjectStmtFromCoordinator` method, it:
     - Invokes the `EnsureAllObjectDependenciesExistOnAllNodes` function to propagate missing dependencies, both on the coordinator and the worker.
 
   - Before defining new `preprocess` or `postprocess` methods, it is advisable to assess whether the generic methods can be employed in your specific case.

src/backend/distributed/citus.control

@@ -1,6 +1,6 @@
 # Citus extension
 comment = 'Citus distributed database'
-default_version = '14.0-1'
+default_version = '14.1-1'
 module_pathname = '$libdir/citus'
 relocatable = false
 schema = pg_catalog

src/backend/distributed/commands/alter_table.c

@@ -1525,7 +1525,9 @@ CreateCitusTableLike(TableConversionState *con)
 				.colocateWithTableName = quote_qualified_identifier(con->schemaName,
 																	con->relationName)
 			};
-			CreateSingleShardTable(con->newRelationId, colocationParam);
+			bool allowFromWorkers = false;
+			CreateSingleShardTable(con->newRelationId, colocationParam,
+								   allowFromWorkers);
 		}
 		else
 		{

src/backend/distributed/commands/common.c

@@ -31,6 +31,13 @@
 #include "distributed/worker_transaction.h"
 
 
+static List * PreprocessAlterDistributedObjectStmtInternal(Node *stmt,
+														   const char *queryString,
+														   ProcessUtilityContext
+														   processUtilityContext,
+														   bool allowFromWorkers);
+
+
 /*
  * PostprocessCreateDistributedObjectFromCatalogStmt is a common function that can be used
  * for most objects during their creation phase. After the creation has happened locally
@@ -104,27 +111,59 @@ PostprocessCreateDistributedObjectFromCatalogStmt(Node *stmt, const char *queryS
 
 
 /*
- * PreprocessAlterDistributedObjectStmt handles any updates to distributed objects by
- * creating the fully qualified sql to apply to all workers after checking all
- * predconditions that apply to propagating changes.
+ * PreprocessAlterDistributedObjectStmtFromCoordinator is a wrapper around
+ * PreprocessAlterDistributedObjectStmtInternal to be used when altering distributed
+ * objects that we allow altering only from the coordinator.
+ */
+List *
+PreprocessAlterDistributedObjectStmtFromCoordinator(Node *stmt, const char *queryString,
+													ProcessUtilityContext
+													processUtilityContext)
+{
+	return PreprocessAlterDistributedObjectStmtInternal(stmt, queryString,
+														processUtilityContext, false);
+}
+
+
+/*
+ * PreprocessAlterDistributedObjectStmtFromAnyNode is a wrapper around
+ * PreprocessAlterDistributedObjectStmtInternal to be used when altering distributed
+ * objects that we allow altering from any node.
+ */
+List *
+PreprocessAlterDistributedObjectStmtFromAnyNode(Node *stmt,
+												const char *queryString,
+												ProcessUtilityContext
+												processUtilityContext)
+{
+	return PreprocessAlterDistributedObjectStmtInternal(stmt, queryString,
+														processUtilityContext, true);
+}
+
+
+/*
+ * PreprocessAlterDistributedObjectStmtInternal handles any updates to distributed
+ * objects by creating the fully qualified sql to apply to all other nodes after checking
+ * all predconditions that apply to propagating changes.
  *
  * Preconditions are (in order):
  *  - not in a CREATE/ALTER EXTENSION code block
  *  - citus.enable_metadata_sync is turned on
  *  - object being altered is distributed
  *  - any object specific feature flag is turned on when a feature flag is available
  *
- * Once we conclude to propagate the changes to the workers we make sure that the command
- * has been executed on the coordinator and force any ongoing transaction to run in
+ * Once we conclude to propagate the changes to other nodes we make sure that the command
+ * has been executed on the local node and force any ongoing transaction to run in
  * sequential mode. If any of these steps fail we raise an error to inform the user.
  *
  * Lastly we recreate a fully qualified version of the original sql and prepare the tasks
- * to send these sql commands to the workers. These tasks include instructions to prevent
+ * to send these sql commands to other nodes. These tasks include instructions to prevent
  * recursion of propagation with Citus' MX functionality.
  */
-List *
-PreprocessAlterDistributedObjectStmt(Node *stmt, const char *queryString,
-									 ProcessUtilityContext processUtilityContext)
+static List *
+PreprocessAlterDistributedObjectStmtInternal(Node *stmt, const char *queryString,
+											 ProcessUtilityContext processUtilityContext,
+											 bool allowFromWorkers)
 {
 	const DistributeObjectOps *ops = GetDistributeObjectOps(stmt);
 	Assert(ops != NULL);
@@ -145,7 +184,15 @@ PreprocessAlterDistributedObjectStmt(Node *stmt, const char *queryString,
 		return NIL;
 	}
 
-	EnsureCoordinator();
+	if (allowFromWorkers)
+	{
+		EnsurePropagationToCoordinator();
+	}
+	else
+	{
+		EnsureCoordinator();
+	}
+
 	EnsureSequentialMode(ops->objectType);
 
 	QualifyTreeNode(stmt);
@@ -155,14 +202,14 @@ PreprocessAlterDistributedObjectStmt(Node *stmt, const char *queryString,
 								(void *) sql,
 								ENABLE_DDL_PROPAGATION);
 
-	return NodeDDLTaskList(NON_COORDINATOR_NODES, commands);
+	return NodeDDLTaskList(REMOTE_NODES, commands);
 }
 
 
 /*
- * PostprocessAlterDistributedObjectStmt is the counter part of
- * PreprocessAlterDistributedObjectStmt that should be executed after the object has been
- * changed locally.
+ * PostprocessAlterDistributedObjectStmtFromCoordinator is the counter part of
+ * PreprocessAlterDistributedObjectStmtFromCoordinator that should be executed after the
+ * object has been changed locally.
  *
  * We perform the same precondition checks as before to skip this operation if any of the
  * failed during preprocessing. Since we already raised an error on other checks we don't
@@ -173,7 +220,7 @@ PreprocessAlterDistributedObjectStmt(Node *stmt, const char *queryString,
  * they get created on the workers before we send the command list to the workers.
  */
 List *
-PostprocessAlterDistributedObjectStmt(Node *stmt, const char *queryString)
+PostprocessAlterDistributedObjectStmtFromCoordinator(Node *stmt, const char *queryString)
 {
 	const DistributeObjectOps *ops = GetDistributeObjectOps(stmt);
 	Assert(ops != NULL);