Backport #46350 to 23.1: Fix LOGICAL_ERROR in async inserts with invalid data in format VALUES

src/Processors/Formats/Impl/ValuesBlockInputFormat.cpp

@@ -602,6 +602,17 @@ void ValuesBlockInputFormat::readSuffix()
 
 void ValuesBlockInputFormat::resetParser()
 {
+    if (got_exception)
+    {
+        /// In case of exception always reset the templates and parser type,
+        /// because they may be in the invalid state.
+        for (size_t i = 0; i < num_columns; ++i)
+        {
+            templates[i].reset();
+            parser_type_for_column[i] = ParserType::Streaming;
+        }
+    }
+
     IInputFormat::resetParser();
     // I'm not resetting parser modes here.
     // There is a good chance that all messages have the same format.

src/Processors/ISource.cpp

@@ -110,6 +110,7 @@ void ISource::work()
     catch (...)
     {
         finished = true;
+        got_exception = true;
         throw;
     }
 }

tests/queries/0_stateless/02563_async_insert_bad_data.reference

@@ -0,0 +1,2 @@
+SYNTAX_ERROR
+good string

tests/queries/0_stateless/02563_async_insert_bad_data.sh

@@ -0,0 +1,23 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CURDIR"/../shell_config.sh
+
+${CLICKHOUSE_CLIENT} -q "DROP TABLE IF EXISTS t_async_insert_bad_data"
+${CLICKHOUSE_CLIENT} -q "CREATE TABLE t_async_insert_bad_data (d Date, s String) ENGINE = Memory"
+
+${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}&async_insert=1" -X POST \
+    --data-binary "INSERT INTO t_async_insert_bad_data VALUES (now(), 'bad'string')" 2>&1 | grep -o SYNTAX_ERROR &
+
+# Sleep to avoid reordering of inserts. It doesn't affect the correctness
+# of test, but allows to reproduce a bug more often in old versions.
+sleep 0.02
+
+${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}&async_insert=1" -X POST  \
+    --data-binary "INSERT INTO t_async_insert_bad_data VALUES (now(), 'good string')" &
+
+wait
+
+${CLICKHOUSE_CLIENT} -q "SELECT s FROM t_async_insert_bad_data ORDER BY s"
+${CLICKHOUSE_CLIENT} -q "DROP TABLE IF EXISTS t_async_insert_bad_data"