ZSTD compression crashes connection

[dmkulazhenko]

Abstract

When zstd compression enabled in connection, any INSERT ... VALUES query with values provided as arguments crash connection object on next ping. In practice ClickHouse closes connection, because of invalid behavior (see artifacts).

Tested Asynch versions: 0.3.0, 0.3.1.

Artifacts

Reproduce with:

import asyncio

import asynch


async def main() -> None:
    async with asynch.Connection(
        user="default",
        password="test_password",
        host="localhost",
        database="sol_etl_data",
        port=9000,
        compression="zstd",
    ) as conn:
        async with conn.cursor() as cursor:
            await cursor.execute(
                """
                CREATE TABLE IF NOT EXISTS compression (a UInt64)
                ENGINE = MergeTree
                ORDER BY a
                """
            )

            await cursor.execute(
                "INSERT INTO compression (a) VALUES",
                [(1,)],
            )

            await cursor.execute("SELECT * FROM compression")
            print(await cursor.fetchall())


if __name__ == "__main__":
    try:
        loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    loop.run_until_complete(main())

Clickhouse logs:

2025.11.19 16:36:06.249831 [ 798 ] {} <Error> TCPHandler: Code: 101. DB::Exception: Unexpected packet Data received from client. (UNEXPECTED_PACKET_FROM_CLIENT), Stack trace (when copying this message, always include the lines below):

0. DB::Exception::Exception(DB::Exception::MessageMasked&&, int, bool) @ 0x00000000127d5c3b
1. DB::Exception::Exception(PreformattedMessage&&, int) @ 0x000000000bfe4a4c
2. DB::Exception::Exception<>(int, FormatStringHelperImpl<>) @ 0x000000000bff0e2b
3. DB::TCPHandler::receivePacketsExpectQuery(std::optional<DB::QueryState>&) @ 0x0000000018590f4c
4. DB::TCPHandler::runImpl() @ 0x0000000018582714
5. DB::TCPHandler::run() @ 0x00000000185a2d39
6. Poco::Net::TCPServerConnection::start() @ 0x000000001d918f67
7. Poco::Net::TCPServerDispatcher::run() @ 0x000000001d9193b9
8. Poco::PooledThread::run() @ 0x000000001d8e4647
9. Poco::ThreadImpl::runnableEntry(void*) @ 0x000000001d8e2be1
10. ? @ 0x0000000000094ac3
11. ? @ 0x0000000000125a74

Traceback:

Traceback (most recent call last):
  File "/home/ch_zstd_asynch.py", line 40, in <module>
    loop.run_until_complete(main())
  File "/home/.pyenv/versions/3.12.9/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/home/ch_zstd_asynch.py", line 29, in main
    await cursor.execute("SELECT * FROM compression")
  File "/home/venv/lib/python3.12/site-packages/asynch/cursors.py", line 75, in execute
    response = await execute(query, args=args, with_column_types=True, **execute_kwargs)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/venv/lib/python3.12/site-packages/asynch/proto/connection.py", line 638, in execute
    async with ExecuteContext(self, query, settings):
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/venv/lib/python3.12/site-packages/asynch/proto/context.py", line 50, in __aenter__
    await self._connection.force_connect()
  File "/home/venv/lib/python3.12/site-packages/asynch/proto/connection.py", line 761, in force_connect
    elif not await self.ping():
             ^^^^^^^^^^^^^^^^^
  File "/home/venv/lib/python3.12/site-packages/asynch/proto/connection.py", line 315, in ping
    packet_type = await self.reader.read_varint()
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/venv/lib/python3.12/site-packages/asynch/proto/streams/buffered.py", line 150, in read_varint
    raise OperationalError(msg)
asynch.errors.OperationalError: Code: None. Failed to read data from socket. Likely the connection was closed by the remote.

[nils-borrmann-tacto]

@dmkulazhenko I fixed the actual compression issue also in #153