From 1602daba5937311a5d70a45dbae78435da9a8200 Mon Sep 17 00:00:00 2001
From: alexey-milovidov <milovidov@yandex-team.ru>
Date: Thu, 12 Sep 2019 00:36:59 +0300
Subject: [PATCH] Merge pull request #6911 from
 yandex/fix-insert-select-data-loss

Fix insert select data loss

(cherry picked from commit 6fbf9ca7ab88f02b39328f0c5cc83ee8c5cfdfd6)
---
 dbms/src/DataStreams/copyData.cpp                 |  2 --
 dbms/src/Interpreters/executeQuery.cpp            | 15 ++++++++++++++-
 .../01009_insert_select_data_loss.reference       |  2 ++
 .../0_stateless/01009_insert_select_data_loss.sql |  5 +++++
 4 files changed, 21 insertions(+), 3 deletions(-)
 create mode 100644 dbms/tests/queries/0_stateless/01009_insert_select_data_loss.reference
 create mode 100644 dbms/tests/queries/0_stateless/01009_insert_select_data_loss.sql

diff --git a/dbms/src/DataStreams/copyData.cpp b/dbms/src/DataStreams/copyData.cpp
index 5000c87be7c..9d17596fc8d 100644
--- a/dbms/src/DataStreams/copyData.cpp
+++ b/dbms/src/DataStreams/copyData.cpp
@@ -28,8 +28,6 @@ void copyDataImpl(IBlockInputStream & from, IBlockOutputStream & to, TCancelCall
             break;
 
         to.write(block);
-        if (!block.rows())
-            to.flush();
         progress(block);
     }
 
diff --git a/dbms/src/Interpreters/executeQuery.cpp b/dbms/src/Interpreters/executeQuery.cpp
index 3793f2f79c9..85130437155 100644
--- a/dbms/src/Interpreters/executeQuery.cpp
+++ b/dbms/src/Interpreters/executeQuery.cpp
@@ -35,6 +35,7 @@
 #include <Processors/Transforms/LimitsCheckingTransform.h>
 #include <Processors/Transforms/MaterializingTransform.h>
 #include <Processors/Formats/IOutputFormat.h>
+#include <Parsers/ASTWatchQuery.h>
 
 namespace ProfileEvents
 {
@@ -631,7 +632,19 @@ void executeQuery(
             if (set_query_id)
                 set_query_id(context.getClientInfo().current_query_id);
 
-            copyData(*streams.in, *out);
+            if (ast->as<ASTWatchQuery>())
+            {
+                /// For Watch query, flush data if block is empty (to send data to client).
+                auto flush_callback = [&out](const Block & block)
+                {
+                    if (block.rows() == 0)
+                        out->flush();
+                };
+
+                copyData(*streams.in, *out, [](){ return false; }, std::move(flush_callback));
+            }
+            else
+                copyData(*streams.in, *out);
         }
 
         if (pipeline.initialized())
diff --git a/dbms/tests/queries/0_stateless/01009_insert_select_data_loss.reference b/dbms/tests/queries/0_stateless/01009_insert_select_data_loss.reference
new file mode 100644
index 00000000000..25e7f55667e
--- /dev/null
+++ b/dbms/tests/queries/0_stateless/01009_insert_select_data_loss.reference
@@ -0,0 +1,2 @@
+0
+10
diff --git a/dbms/tests/queries/0_stateless/01009_insert_select_data_loss.sql b/dbms/tests/queries/0_stateless/01009_insert_select_data_loss.sql
new file mode 100644
index 00000000000..9a754d94323
--- /dev/null
+++ b/dbms/tests/queries/0_stateless/01009_insert_select_data_loss.sql
@@ -0,0 +1,5 @@
+drop table if exists tab;
+create table tab (x UInt64) engine = MergeTree order by tuple();
+
+insert into tab select number as n from numbers(20) any inner join (select number * 10 as n from numbers(2)) using(n) settings any_join_distinct_right_table_keys = 1, max_block_size = 5;
+select * from tab order by x;