LOGICAL_ERROR Pipeline for PushingPipelineExecutor was finished before all data was inserted

[nickitat]

I'm able to reproduce smth similar with the following:

diff --git a/src/Server/TCPHandler.cpp b/src/Server/TCPHandler.cpp
index ae2f150c4a1..d0911804063 100644
--- a/src/Server/TCPHandler.cpp
+++ b/src/Server/TCPHandler.cpp
@@ -910,6 +910,8 @@ void TCPHandler::processInsertQuery()

     auto run_executor = [&](auto & executor, Block processed_data)
     {
+        std::this_thread::sleep_for(std::chrono::seconds(1));
+
         /// Made above the rest of the lines,
         /// so that in case of `start` function throws an exception,
         /// client receive exception before sending data.

❯ cat repro.sql
drop table if exists mv_table_1;
drop table if exists mv_table;
drop table if exists null_table;

set max_execution_time = 0.1, timeout_overflow_mode = 'break';

CREATE TABLE null_table (str String) ENGINE = Null;
CREATE MATERIALIZED VIEW mv_table (str String) ENGINE = MergeTree ORDER BY str AS SELECT str AS str FROM null_table;
CREATE MATERIALIZED VIEW mv_table_1 (str Decimal(38, 7)) ENGINE = MergeTree ORDER BY str AS SELECT str AS str FROM null_table;

INSERT INTO null_table VALUES ('test');

Code: 49. DB::Exception: Received from localhost:9000. DB::Exception: Pipeline for PushingPipelineExecutor was finished before all data was inserted. (LOGICAL_ERROR)

---

Stepped on this while trying to reproduce https://s3.amazonaws.com/clickhouse-test-reports/63861/af84e3e1a9c8c3b62f738f231e457efc8ea301ea/ast_fuzzer__tsan_/fatal.log

Not sure if it is connected anyhow with #52234 or not. Probably yes, since it requires max_execution_time = 0.1, timeout_overflow_mode = 'break'.

[nickitat]

another interesting behaviour is this:

❯ gd | cat
diff --git a/src/Server/TCPHandler.cpp b/src/Server/TCPHandler.cpp
index ae2f150c4a1..dfc18bd90bb 100644
--- a/src/Server/TCPHandler.cpp
+++ b/src/Server/TCPHandler.cpp
@@ -923,6 +923,8 @@ void TCPHandler::processInsertQuery()
         while (readDataNext())
             executor.push(std::move(state.block_for_insert));

+        std::this_thread::sleep_for(std::chrono::seconds(1));
+
         if (state.cancellation_status == CancellationStatus::FULLY_CANCELLED)
             executor.cancel();
         else

produces no error, while without any sleep we will obviously get an error:

❯ clickhouse client --queries-file repro.sql
Received exception from server (version 24.5.1):
Code: 6. DB::Exception: Received from localhost:9000. DB::Exception: Cannot parse string 'test' as Decimal(38, 7): syntax error at begin of string: while converting source column str to destination column str: while executing 'FUNCTION _CAST(str :: 0, Decimal(38, 7) :: 1) -> _CAST(str, Decimal(38, 7)) Decimal(38, 7) : 2': while pushing to view default.mv_table_1 (318bed11-79cd-4168-b3ac-b8341f8fd678). (CANNOT_PARSE_TEXT)
(query: INSERT INTO null_table VALUES ('test');)

so, we effectively loosing data: https://pastila.nl/?008ee639/0bb65a2827875c58b98c62a4cf731a59#C/nUnY22Jtcrb4U9dUv/tA==