如何解决ClickHouse中的“Unexpected content at the end of chunk”错误？

问题场景 #

我尝试同时迭代两个ClickHouse连接的ResultSet，对两张已排序的表做归并合并。两张表单独执行查询均正常，但同时进行迭代操作时，突然出现了从未见过的MalformedChunkCodingException错误。

相关代码 #

获取ClickHouse连接的工具方法 #

public static void execMany(Consumer<Connection> cons) {
    // ClickHouse server URL
    String url = "jdbc:clickhouse://myip:8123/db2";
    ClickHouseProperties properties = new ClickHouseProperties();
    properties.setSocketTimeout(6000000);
    properties.setSessionTimeout(6000000L);
    properties.setDatabase("db2");
    properties.setUser("default");
    properties.setPassword("12345");
    ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties);
    try (ClickHouseConnection connection = dataSource.getConnection()) {
        cons.accept(connection);
    } catch (SQLException e) {
        e.printStackTrace();
    }
}

归并迭代逻辑代码 #

var ref = new Object() {
    long txCnt = 0;
    long addrCnt = 0;
};
var cnt = ClickHouse.count("transactions_tmp2");
ClickHouse.execMany(conn -> ClickHouse.execMany(conn1 -> {
    try {
        var select1 = conn.createStatement().executeQuery("SELECT * FROM %s order by %s".formatted(t1,f1));
        var select2 = conn1.createStatement().executeQuery("SELECT * FROM %s order by %s".formatted(t2,f2));
        String val2 = null;
        String val1 = null;
        while (true) {
            if (ref.txCnt % 1000000 == 0)
                System.out.println((ref.txCnt) + "/" + cnt + "=" + (1. * ref.txCnt / cnt) + " addrCnt=" + ref.addrCnt + " fromAddress=" + val2 + " txAddress=" + val1);
            if (val2 == null) {
                if (!select2.next()) break;
                ref.addrCnt++;
                val2 = select2.getString(f2);
            }
            if (val1 == null) {
                if (!select1.next()) break;
                ref.txCnt++;
                val1 = select1.getString(f1);
            }
            if (val1.compareTo(val2) < 0) {
                val1 = null;
                continue;
            }
            if (val2.compareTo(val1) < 0) {
                val2 = null;
                continue;
            }
            assert val1.equals(val2);
            cons.accept(select1,select2);
        }
    } catch (SQLException e) {
        throw new RuntimeException(e);
    }
}));

错误堆栈信息 #

Exception in thread "main" java.lang.RuntimeException: java.sql.SQLException: org.apache.http.MalformedChunkCodingException: Unexpected content at the end of chunk
    at me.nanosecond.TxImporter.lambda$merge$1(TxImporter.java:123)
    at me.nanosecond.ClickHouse.execMany(ClickHouse.java:291)
    at me.nanosecond.TxImporter.lambda$merge$2(TxImporter.java:78)
    at me.nanosecond.ClickHouse.execMany(ClickHouse.java:291)
    at me.nanosecond.TxImporter.merge(TxImporter.java:78)
    at me.nanosecond.TxImporter.main(TxImporter.java:185)
Caused by: java.sql.SQLException: org.apache.http.MalformedChunkCodingException: Unexpected content at the end of chunk
    at ru.yandex.clickhouse.response.ClickHouseResultSet.hasNext(ClickHouseResultSet.java:164)
    at ru.yandex.clickhouse.response.ClickHouseResultSet.next(ClickHouseResultSet.java:200)
    at me.nanosecond.TxImporter.lambda$merge$1(TxImporter.java:90)
    ... 5 more
Caused by: org.apache.http.MalformedChunkCodingException: Unexpected content at the end of chunk
    at org.apache.http.impl.io.ChunkedInputStream.getChunkSize(ChunkedInputStream.java:254)
    at org.apache.http.impl.io.ChunkedInputStream.nextChunk(ChunkedInputStream.java:222)
    at org.apache.http.impl.io.ChunkedInputStream.read(ChunkedInputStream.java:183)
    at org.apache.http.conn.EofSensorInputStream.read(EofSensorInputStream.java:135)
    at java.base/java.io.DataInputStream.read(DataInputStream.java:158)
    at ru.yandex.clickhouse.util.Utils.readFully(Utils.java:163)
    at ru.yandex.clickhouse.util.Utils.readFully(Utils.java:147)
    at ru.yandex.clickhouse.response.ClickHouseLZ4Stream.readNextBlock(ClickHouseLZ4Stream.java:102)
    at ru.yandex.clickhouse.response.ClickHouseLZ4Stream.checkNext(ClickHouseLZ4Stream.java:75)
    at ru.yandex.clickhouse.response.ClickHouseLZ4Stream.read(ClickHouseLZ4Stream.java:61)
    at ru.yandex.clickhouse.response.StreamSplitter.readFromStream(StreamSplitter.java:92)
    at ru.yandex.clickhouse.response.StreamSplitter.next(StreamSplitter.java:63)
    at ru.yandex.clickhouse.response.ClickHouseResultSet.hasNext(ClickHouseResultSet.java:149)
    ... 7 more

采用该实现方式的原因 #

由于ClickHouse原生对两张大排序表按主键合并时性能表现不佳，因此自行实现了基于双ResultSet的归并逻辑来完成表合并操作。

内容的提问来源于stack exchange，提问作者Stepan Yakovenko