diff --git a/src/backend/distributed/executor/adaptive_executor.c b/src/backend/distributed/executor/adaptive_executor.c
index 039475735..18dbed49d 100644
--- a/src/backend/distributed/executor/adaptive_executor.c
+++ b/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;
diff --git a/src/test/regress/expected/executor_local_failure.out b/src/test/regress/expected/executor_local_failure.out
new file mode 100644
index 000000000..1e8d577ba
--- /dev/null
+++ b/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;
diff --git a/src/test/regress/multi_mx_schedule b/src/test/regress/multi_mx_schedule
index 682379b78..6654b4ab0 100644
--- a/src/test/regress/multi_mx_schedule
+++ b/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
diff --git a/src/test/regress/sql/executor_local_failure.sql b/src/test/regress/sql/executor_local_failure.sql
new file mode 100644
index 000000000..0d1c56ca1
--- /dev/null
+++ b/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;