Fix LOGICAL_ERROR on race between DROP and INSERT with materialized views

In case of parallel INSERT (max_insert_threads > 1) it is possible for
VIEW to be DROP/DETACH'ed while building pipeline for various paralell
streams, and in this case the header will not match since when you have
VIEW you will got empty header and non-empty header otherwise.

And this leads to LOGICAL_ERROR later, while checking that output
headers are the same (in QueryPipelineBuilder::addChains() -> Pipe::addChains()).

However it also makes the pipeline different for various parallel
streams, and it looks like it is better to fail in this case, so instead
of always returning empty header from buildChainImpl() explicit check
had been added.

Note, that I wasn't able to reproduce the issue with the added test,
but CI may have more "luck" (although I've verified it manually).

Fixes: #35902
Cc: @KochetovNicolai
Signed-off-by: Azat Khuzhin <a.khuzhin@semrush.com>

src/Interpreters/InterpreterInsertQuery.cpp

@@ -44,6 +44,7 @@ namespace ErrorCodes
     extern const int NO_SUCH_COLUMN_IN_TABLE;
     extern const int ILLEGAL_COLUMN;
     extern const int DUPLICATE_COLUMN;
+    extern const int TABLE_IS_DROPPED;
 }
 
 InterpreterInsertQuery::InterpreterInsertQuery(
@@ -424,6 +425,15 @@ BlockIO InterpreterInsertQuery::execute()
         for (size_t i = 0; i < out_streams_size; ++i)
         {
             auto out = buildChainImpl(table, metadata_snapshot, query_sample_block, nullptr, nullptr);
+            if (!out_chains.empty())
+            {
+                if (out.getProcessors().size() != out_chains.back().getProcessors().size())
+                {
+                    throw Exception(ErrorCodes::TABLE_IS_DROPPED,
+                        "Some VIEW is gone in between ({} vs {} processors, on {} parallel stream)",
+                        out.getProcessors().size(), out_chains.back().getProcessors().size(), i);
+                }
+            }
             out_chains.emplace_back(std::move(out));
         }
     }

tests/queries/0_stateless/02380_insert_mv_race.sh

@@ -0,0 +1,66 @@
+#!/usr/bin/env bash
+# Tags: long, race
+
+# Regression test for INSERT into table with MV attached,
+# to avoid possible errors if some table will disappears,
+# in case of multiple streams was used (i.e. max_insert_threads>1)
+
+CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CUR_DIR"/../shell_config.sh
+
+function bootstrap()
+{
+    $CLICKHOUSE_CLIENT -nm -q "
+        DROP TABLE IF EXISTS null;
+        CREATE TABLE null (key Int) ENGINE = Null;
+
+        DROP TABLE IF EXISTS mv;
+        CREATE MATERIALIZED VIEW mv ENGINE = Null() AS SELECT * FROM null;
+    "
+}
+
+function insert_thread()
+{
+    local opts=(
+        --max_insert_threads 100
+        --max_threads 100
+    )
+    local patterns=(
+        -e UNKNOWN_TABLE
+        -e TABLE_IS_DROPPED
+    )
+
+    while :; do
+        $CLICKHOUSE_CLIENT "${opts[@]}" -q "INSERT INTO null SELECT * FROM numbers_mt(1e6)" |& {
+            grep -F "DB::Exception: " | grep -v -F "${patterns[@]}"
+        }
+    done
+}
+export -f insert_thread
+
+function drop_thread()
+{
+    local opts=(
+        --database_atomic_wait_for_drop_and_detach_synchronously 1
+    )
+
+    while :; do
+        $CLICKHOUSE_CLIENT -nm "${opts[@]}" -q "DETACH TABLE mv"
+        sleep 0.01
+        $CLICKHOUSE_CLIENT -nm "${opts[@]}" -q "ATTACH TABLE mv"
+    done
+}
+export -f drop_thread
+
+function main()
+{
+    local test_timeout=1m
+
+    bootstrap
+    timeout "$test_timeout" bash -c insert_thread &
+    timeout "$test_timeout" bash -c drop_thread &
+
+    wait
+}
+main "$@"