diff --git a/src/Storages/Distributed/DistributedSettings.h b/src/Storages/Distributed/DistributedSettings.h
index 86b77707dee..85e64cabfd7 100644
--- a/src/Storages/Distributed/DistributedSettings.h
+++ b/src/Storages/Distributed/DistributedSettings.h
@@ -26,6 +26,7 @@ class ASTStorage;
     M(UInt64, monitor_split_batch_on_failure, 0, "Default - distributed_directory_monitor_split_batch_on_failure", 0) \
     M(Milliseconds, monitor_sleep_time_ms, 0, "Default - distributed_directory_monitor_sleep_time_ms", 0) \
     M(Milliseconds, monitor_max_sleep_time_ms, 0, "Default - distributed_directory_monitor_max_sleep_time_ms", 0) \
+    M(Bool, flush_on_detach, true, "Flush data to remote nodes on DETACH/DROP/server shutdown", 0) \
 
 DECLARE_SETTINGS_TRAITS(DistributedSettingsTraits, LIST_OF_DISTRIBUTED_SETTINGS)
 
diff --git a/src/Storages/StorageDistributed.cpp b/src/Storages/StorageDistributed.cpp
index f80e498efa8..6f0072c4560 100644
--- a/src/Storages/StorageDistributed.cpp
+++ b/src/Storages/StorageDistributed.cpp
@@ -1438,6 +1438,12 @@ ActionLock StorageDistributed::getActionLock(StorageActionBlockType type)
 
 void StorageDistributed::flushAndPrepareForShutdown()
 {
+    if (!getDistributedSettingsRef().flush_on_detach)
+    {
+        LOG_INFO(log, "Skip flushing data (due to flush_on_detach=0)");
+        return;
+    }
+
     try
     {
         flushClusterNodesAllData(getContext());
diff --git a/tests/queries/0_stateless/02860_distributed_flush_on_detach.reference b/tests/queries/0_stateless/02860_distributed_flush_on_detach.reference
new file mode 100644
index 00000000000..8e5815188e7
--- /dev/null
+++ b/tests/queries/0_stateless/02860_distributed_flush_on_detach.reference
@@ -0,0 +1,27 @@
+-- { echoOn }
+
+create table data (key Int) engine=Memory();
+create table dist (key Int) engine=Distributed(default, currentDatabase(), data);
+system stop distributed sends dist;
+-- check that FLUSH DISTRIBUTED does flushing anyway
+insert into dist values (1);
+select * from data;
+system flush distributed dist;
+select * from data;
+1
+truncate table data;
+-- check that flush_on_detach=1 by default
+insert into dist values (1);
+detach table dist;
+select * from data;
+1
+attach table dist;
+truncate table data;
+-- check flush_on_detach=0
+drop table dist;
+create table dist (key Int) engine=Distributed(default, currentDatabase(), data) settings flush_on_detach=0;
+system stop distributed sends dist;
+insert into dist values (1);
+detach table dist;
+select * from data;
+attach table dist;
diff --git a/tests/queries/0_stateless/02860_distributed_flush_on_detach.sql b/tests/queries/0_stateless/02860_distributed_flush_on_detach.sql
new file mode 100644
index 00000000000..5ba45d72c08
--- /dev/null
+++ b/tests/queries/0_stateless/02860_distributed_flush_on_detach.sql
@@ -0,0 +1,33 @@
+set prefer_localhost_replica=0;
+
+drop table if exists data;
+drop table if exists dist;
+
+-- { echoOn }
+
+create table data (key Int) engine=Memory();
+create table dist (key Int) engine=Distributed(default, currentDatabase(), data);
+system stop distributed sends dist;
+
+-- check that FLUSH DISTRIBUTED does flushing anyway
+insert into dist values (1);
+select * from data;
+system flush distributed dist;
+select * from data;
+truncate table data;
+
+-- check that flush_on_detach=1 by default
+insert into dist values (1);
+detach table dist;
+select * from data;
+attach table dist;
+truncate table data;
+
+-- check flush_on_detach=0
+drop table dist;
+create table dist (key Int) engine=Distributed(default, currentDatabase(), data) settings flush_on_detach=0;
+system stop distributed sends dist;
+insert into dist values (1);
+detach table dist;
+select * from data;
+attach table dist;