Merge branch 'main' into citus_pause_node

src/backend/distributed/deparser/deparse_table_stmts.c

@@ -627,6 +627,10 @@ AppendAlterTableCmdAddColumn(StringInfo buf, AlterTableCmd *alterTableCmd,
 		{
 			appendStringInfoString(buf, " NOT NULL");
 		}
+		else if (constraint->contype == CONSTR_NULL)
+		{
+			appendStringInfoString(buf, " NULL");
+		}
 		else if (constraint->contype == CONSTR_DEFAULT)
 		{
 			char attgenerated = '\0';

src/backend/distributed/executor/adaptive_executor.c

@@ -519,7 +519,9 @@ typedef enum TaskExecutionState
 /*
  * PlacementExecutionOrder indicates whether a command should be executed
  * on any replica, on all replicas sequentially (in order), or on all
- * replicas in parallel.
+ * replicas in parallel. In other words, EXECUTION_ORDER_ANY is used for
+ * SELECTs, EXECUTION_ORDER_SEQUENTIAL/EXECUTION_ORDER_PARALLEL is used for
+ * DML/DDL.
  */
 typedef enum PlacementExecutionOrder
 {
@@ -4687,6 +4689,10 @@ TaskExecutionStateMachine(ShardCommandExecution *shardCommandExecution)
 	{
 		currentTaskExecutionState = TASK_EXECUTION_FAILED;
 	}
+	else if (executionOrder != EXECUTION_ORDER_ANY && failedPlacementCount > 0)
+	{
+		currentTaskExecutionState = TASK_EXECUTION_FAILED;
+	}
 	else if (executionOrder == EXECUTION_ORDER_ANY && donePlacementCount > 0)
 	{
 		currentTaskExecutionState = TASK_EXECUTION_FINISHED;

src/test/regress/expected/alter_table_add_column.out

@@ -35,7 +35,7 @@ ALTER TABLE referencing ADD COLUMN test_1 integer DEFAULT (alter_table_add_colum
 ALTER TABLE referencing ADD COLUMN test_2 integer UNIQUE REFERENCES referenced(int_col) ON UPDATE CASCADE ON DELETE SET DEFAULT NOT DEFERRABLE INITIALLY IMMEDIATE;
 ALTER TABLE referencing ADD COLUMN test_3 integer GENERATED ALWAYS AS (test_1 * alter_table_add_column_other_schema.my_random(1)) STORED UNIQUE REFERENCES referenced(int_col) MATCH FULL;
 ALTER TABLE referencing ADD COLUMN test_4 integer PRIMARY KEY WITH (fillfactor=70) NOT NULL REFERENCES referenced(int_col) MATCH SIMPLE ON UPDATE CASCADE ON DELETE SET DEFAULT;
-ALTER TABLE referencing ADD COLUMN test_5 integer CONSTRAINT unique_c UNIQUE WITH (fillfactor=50);
+ALTER TABLE referencing ADD COLUMN test_5 integer CONSTRAINT unique_c UNIQUE WITH (fillfactor=50) NULL;
 ALTER TABLE referencing ADD COLUMN test_6 text COMPRESSION pglz COLLATE caseinsensitive NOT NULL;
 ALTER TABLE referencing ADD COLUMN "test_\'!7" "simple_!\'custom_type";
 -- we give up deparsing ALTER TABLE command if it needs to create a check constraint, and we fallback to legacy behavior

src/test/regress/expected/executor_local_failure.out

@@ -0,0 +1,47 @@
+CREATE SCHEMA failure_local_modification;
+SET search_path TO failure_local_modification;
+SET citus.next_shard_id TO 1989000;
+SET citus.shard_replication_factor TO 1;
+CREATE TABLE failover_to_local (key int PRIMARY KEY, value varchar(10));
+SELECT create_reference_table('failover_to_local');
+ create_reference_table
+---------------------------------------------------------------------
+
+(1 row)
+
+\c - - - :worker_2_port
+SET search_path TO failure_local_modification;
+-- prevent local connection establishment, imitate
+-- a failure
+ALTER SYSTEM SET citus.local_shared_pool_size TO -1;
+SELECT pg_reload_conf();
+ pg_reload_conf
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT pg_sleep(0.2);
+ pg_sleep
+---------------------------------------------------------------------
+
+(1 row)
+
+BEGIN;
+	-- we force the execution to use connections (e.g., remote execution)
+	-- however, we do not allow connections as local_shared_pool_size=-1
+	-- so, properly error out
+	SET LOCAL citus.enable_local_execution TO false;
+	INSERT INTO failover_to_local VALUES (1,'1'), (2,'2'),(3,'3'),(4,'4');
+ERROR:  the total number of connections on the server is more than max_connections(100)
+HINT:  This command supports local execution. Consider enabling local execution using SET citus.enable_local_execution TO true;
+ROLLBACK;
+ALTER SYSTEM RESET citus.local_shared_pool_size;
+SELECT pg_reload_conf();
+ pg_reload_conf
+---------------------------------------------------------------------
+ t
+(1 row)
+
+\c - - - :master_port
+SET client_min_messages TO ERROR;
+DROP SCHEMA failure_local_modification cascade;

src/test/regress/multi_mx_schedule

@@ -69,6 +69,7 @@ test: local_shard_execution_dropped_column
 test: metadata_sync_helpers
 
 test: issue_6592
+test: executor_local_failure
 
 # test that no tests leaked intermediate results. This should always be last
 test: ensure_no_intermediate_data_leak

src/test/regress/sql/alter_table_add_column.sql

@@ -33,7 +33,7 @@ ALTER TABLE referencing ADD COLUMN test_1 integer DEFAULT (alter_table_add_colum
 ALTER TABLE referencing ADD COLUMN test_2 integer UNIQUE REFERENCES referenced(int_col) ON UPDATE CASCADE ON DELETE SET DEFAULT NOT DEFERRABLE INITIALLY IMMEDIATE;
 ALTER TABLE referencing ADD COLUMN test_3 integer GENERATED ALWAYS AS (test_1 * alter_table_add_column_other_schema.my_random(1)) STORED UNIQUE REFERENCES referenced(int_col) MATCH FULL;
 ALTER TABLE referencing ADD COLUMN test_4 integer PRIMARY KEY WITH (fillfactor=70) NOT NULL REFERENCES referenced(int_col) MATCH SIMPLE ON UPDATE CASCADE ON DELETE SET DEFAULT;
-ALTER TABLE referencing ADD COLUMN test_5 integer CONSTRAINT unique_c UNIQUE WITH (fillfactor=50);
+ALTER TABLE referencing ADD COLUMN test_5 integer CONSTRAINT unique_c UNIQUE WITH (fillfactor=50) NULL;
 ALTER TABLE referencing ADD COLUMN test_6 text COMPRESSION pglz COLLATE caseinsensitive NOT NULL;
 ALTER TABLE referencing ADD COLUMN "test_\'!7" "simple_!\'custom_type";
 

src/test/regress/sql/executor_local_failure.sql

@@ -0,0 +1,31 @@
+CREATE SCHEMA failure_local_modification;
+SET search_path TO failure_local_modification;
+SET citus.next_shard_id TO 1989000;
+
+SET citus.shard_replication_factor TO 1;
+CREATE TABLE failover_to_local (key int PRIMARY KEY, value varchar(10));
+SELECT create_reference_table('failover_to_local');
+
+\c - - - :worker_2_port
+
+SET search_path TO failure_local_modification;
+
+-- prevent local connection establishment, imitate
+-- a failure
+ALTER SYSTEM SET citus.local_shared_pool_size TO -1;
+SELECT pg_reload_conf();
+SELECT pg_sleep(0.2);
+BEGIN;
+	-- we force the execution to use connections (e.g., remote execution)
+	-- however, we do not allow connections as local_shared_pool_size=-1
+	-- so, properly error out
+	SET LOCAL citus.enable_local_execution TO false;
+	INSERT INTO failover_to_local VALUES (1,'1'), (2,'2'),(3,'3'),(4,'4');
+ROLLBACK;
+
+ALTER SYSTEM RESET citus.local_shared_pool_size;
+SELECT pg_reload_conf();
+
+\c - - - :master_port
+SET client_min_messages TO ERROR;
+DROP SCHEMA failure_local_modification cascade;