I am trying to move to `async_insert` to avoid having a batching component merge many small inserts (the exact use case of `async_insert`). It works as intended over the HTTP protocol using both row-based input formats (CSV, TSV) and the column-based Parquet format (for testing, I generate Parquet data by piping CSV through clickhouse-local). However, if I try the more efficient and easier to generate Native input format, the HTTP query succeeds and returns a query id via the `X-ClickHouse-Query-Id` header, a corresponding entry in the `system.asynchronous_inserts` table is created and then deleted, and CH logs appear to indicate that the insert succeeded (see below), but no data is inserted into the table. If `async_insert` is set to 0, everything works correctly.

**How to reproduce**
CH version 21.12.3.32 (official build).

```
clickhouse-client -q "CREATE TABLE async_inserts (id UInt32, s String) ENGINE = Memory"
function insert() {
  curl -i --data-binary @- "http://localhost:8123?query=INSERT+INTO+async_inserts+FORMAT+$1&async_insert=$2"
}
function convert() {
  clickhouse-local --input-format=CSV -N t -S 'id UInt32,s String' -q "SELECT * FROM t FORMAT $1"
}

# with async_insert=1
echo '1, "a"' | insert CSV 1
echo '2, "b"' | convert Parquet | insert Parquet 1
echo '3, "c"' | convert Native | insert Native 1

# with async_insert=0
echo '4, "a"' | insert CSV 0
echo '5, "b"' | convert Parquet | insert Parquet 0
echo '6, "c"' | convert Native | insert Native 0

clickhouse-client -q "SELECT * FROM async_inserts FORMAT TSV"
# output:
# 1   a
# 2   b
# 4   a
# 5   b
# 6   c
```

**Example records from /var/log/clickhouse-server/clickhouse-server.log**
2022.01.11 21:35:54.615405 [ 9514 ] {d8f833c7-1b68-49ad-a8b4-abbb4e79c3e8} <Debug> executeQuery: (from [::ffff:127.0.0.1]:53242) INSERT INTO async_inserts FORMAT Native
2022.01.11 21:35:54.615449 [ 9514 ] {d8f833c7-1b68-49ad-a8b4-abbb4e79c3e8} <Trace> ContextAccess (default): Access granted: INSERT(id, s) ON default.async_inserts
2022.01.11 21:35:54.615500 [ 9514 ] {d8f833c7-1b68-49ad-a8b4-abbb4e79c3e8} <Trace> AsynchronousInsertQueue: Have 1 pending inserts with total 27 bytes of data for query 'INSERT INTO default.async_inserts FORMAT Native'
2022.01.11 21:35:54.817042 [ 9514 ] {d8f833c7-1b68-49ad-a8b4-abbb4e79c3e8} <Debug> DynamicQueryHandler: Done processing query
2022.01.11 21:35:54.817072 [ 9514 ] {d8f833c7-1b68-49ad-a8b4-abbb4e79c3e8} <Debug> MemoryTracker: Peak memory usage (for query): 3.00 MiB.
#include <Formats/NativeReader.h>
#include <Formats/NativeWriter.h>
#include <Formats/FormatFactory.h>
#include <Processors/Formats/IInputFormat.h>
#include <Processors/Formats/IOutputFormat.h>
#include <Processors/Formats/ISchemaReader.h>
#include <Processors/Transforms/AggregatingTransform.h>
namespace DB
{
class NativeInputFormat final : public IInputFormat
{
public:
    NativeInputFormat(ReadBuffer & buf, const Block & header)
        : IInputFormat(header, buf)
        , reader(buf, header, 0) {}
    String getName() const override { return "Native"; }
    void resetParser() override
    {
        IInputFormat::resetParser();
        reader.resetParser();
    }
    Chunk generate() override