Skip to content

Example of migration from ClickHouseRowBinaryStream #2239

New issue
New issue
Open
Open
Example of migration from ClickHouseRowBinaryStream#2239
Labels
client-api-v2enhancementquestion

Description

@makrushin-evgenii
makrushin-evgenii
opened on Mar 21, 2025

Use case

For users, migrating from old client

Describe the solution you'd like

...

Describe the alternatives you've considered

...

Additional context

I have code, that insert (UUID, UUID, UUID, byte[]) objects with clickhouse-jdbc:0.3.1-patch:

@Override
protected ProcessorResult send(List<SinkEvent> events) throws BackendServiceFailedException {
        var sentEvents = new LinkedList<SinkEvent>();
        try (ClickHouseStatement statement = this.connection.createStatement()) {
            statement.write().send(
                    "INSERT INTO tracing_spans_local (trace_id, parent_span_id, span_id, payload)",
                    stream -> {
                        for (var event : events) {
                            write(stream, sinkEvent.getInternalEvent())
                        }
                    },
                    ClickHouseFormat.RowBinary);

@Override
protected boolean write(ClickHouseRowBinaryStream stream, Event event) throws IOException {
    // ...
    stream.writeUUID(traceId.get());
    stream.writeUUID(parentSpanId.orElse(DEFAULT_UUID));
    stream.writeUUID(spanId.get());
    stream.writeUnsignedLeb128(event.getBytes().length);
    stream.writeBytes(event.getBytes());

Now I'm trying to updage to client-v2:0.8.2 and have no idea how to rewrite this. Tried to use RowBinaryFormatWriter, but it looks very complex and have no docs. Probably, it's not for client use?

Activity

  1. makrushin-evgenii
    added
    enhancement
    on Mar 21, 2025

  2. makrushin-evgenii commented on Mar 21, 2025

    @makrushin-evgenii
    makrushin-evgenii
    on Mar 21, 2025
    Author

    Now I have this code. Works strange:

    1. With RowBinary logs " Insert succeed: 2000 rows written" when process batch of 100 events
    2. With RowBinaryWithDefaults log no errors when write no rows to table

    Main questions:

    • How to write UUID and byte[] to RowBinaryFormatWriter?
    • How to init TableSchema?

    Alternative question:

    • What prefered way to insert data, if I can't use lombok for POJOs, and column names in snake_case?

    Table

    CREATE TABLE IF NOT EXISTS default.tracing_spans_local (
    `date_hour` UInt32 DEFAULT toUInt32(toYYYYMMDDhhmmss(now()) / 10000),
    `trace_id` UUID,
    `parent_span_id` UUID,
    `span_id` UUID,
    `payload` String CODEC(LZ4)
)
ENGINE = MergeTree
PARTITION BY date_hour
ORDER BY trace_id;

    Code

    public class ClickHouseTracingSender extends NoPrepareParallelSender {
    private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseTracingSender.class);
    private static final UUID DEFAULT_UUID = new UUID(0, 0);

    private final Timer processingTimeMsTimer;
    private final Client clickhouseClient;
    private final String clickhouseTable;

    public ClickHouseTracingSender(Properties properties, MetricsCollector metricsCollector) {
        super(properties, metricsCollector);
        this.processingTimeMsTimer = metricsCollector.timer("processingTimeMs");
        this.clickhouseTable = PropertiesUtil.get(Props.CLICKHOUSE_TABLE, properties).get();

        this.clickhouseClient = new Client.Builder()
            .addEndpoint(PropertiesUtil.get(Props.CLICKHOUSE_ENDPOINT, properties).get())
            .setUsername(PropertiesUtil.get(Props.CLICKHOUSE_USERNAME, properties).get())
            .setPassword(PropertiesUtil.get(Props.CLICKHOUSE_PASSWORD, properties).get())
            .build();
    }

    @Override
    protected ProcessorResult send(List<SinkEvent> events) {
        try (var ignored = new AutoMetricStopwatch(processingTimeMsTimer, TimeUnit.MILLISECONDS);
             var outputStream = new ByteArrayOutputStream()) {

            var tableSchema = new TableSchema();
            tableSchema.addColumn("trace_id", "UUID");
            tableSchema.addColumn("span_id", "UUID");
            tableSchema.addColumn("parent_span_id", "UUID");
            tableSchema.addColumn("payload", "String");

            var writer = new RowBinaryFormatWriter(outputStream, tableSchema, ClickHouseFormat.RowBinaryWithDefaults);
            LOGGER.info("events.size()=" + events.size());
            for (var event : events) {
                writer.setValue("trace_id", ContainerUtil.extract(event.getInternalEvent().getPayload(), TraceSpanTags.TRACE_ID_TAG).orElse(DEFAULT_UUID));
                writer.setValue("span_id", ContainerUtil.extract(event.getInternalEvent().getPayload(), TraceSpanTags.SPAN_ID_TAG).orElse(DEFAULT_UUID));
                writer.setValue("parent_span_id", ContainerUtil.extract(event.getInternalEvent().getPayload(), TraceSpanTags.PARENT_SPAN_ID_TAG).orElse(DEFAULT_UUID));
                writer.setString("payload", new String(event.getInternalEvent().getBytes(), StandardCharsets.UTF_8));
                writer.commitRow();
            }

            LOGGER.info("clickhouseClient.ping()=" + clickhouseClient.ping());
            try (var inputStream = new ByteArrayInputStream(outputStream.toByteArray());
                 var response = clickhouseClient.insert(clickhouseTable, inputStream, ClickHouseFormat.RowBinary).get(60, TimeUnit.SECONDS)
            ) {
                LOGGER.info("Insert succeed: {} rows written", response.getWrittenRows());
                return ProcessorResult.ok(events, 0);
            } catch (Exception e) {
                throw new RuntimeException("Insert failure", e);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private static class Props {
        static final Parameter<String> CLICKHOUSE_ENDPOINT = Parameter.stringParameter("clickhouse.endpoint").build();
        static final Parameter<String> CLICKHOUSE_USERNAME = Parameter.stringParameter("clickhouse.username").build();
        static final Parameter<String> CLICKHOUSE_PASSWORD = Parameter.stringParameter("clickhouse.password").build();
        static final Parameter<String> CLICKHOUSE_TABLE = Parameter.stringParameter("clickhouse.table").build();
    }
}

    Log (with RowBinaryWithDefaults)

    2025-03-22T02:42:28.242+03:00 [sink-thread-1] INFO  r.k.v.h.t.s.c.ClickHouseTracingSender - events.size()=100
2025-03-22T02:42:28.245+03:00 [sink-thread-1] INFO  r.k.v.h.t.s.c.ClickHouseTracingSender - clickhouseClient.ping()=true
2025-03-22T02:42:29.327+03:00 [sink-thread-1] INFO  r.k.v.h.t.s.c.ClickHouseTracingSender - Insert succeed: 0 rows written

    Log (with RowBinary)

    2025-03-22T01:21:24.131+03:00 [sink-thread-1] INFO  r.k.v.h.t.s.c.ClickHouseTracingSender - events.size()=100
2025-03-22T01:21:24.134+03:00 [sink-thread-1] INFO  r.k.v.h.t.s.c.ClickHouseTracingSender - clickhouseClient.ping()=true
2025-03-22T01:21:25.216+03:00 [sink-thread-1] INFO  r.k.v.h.t.s.c.ClickHouseTracingSender - Insert succeed: 200 rows written
  3. chernser
    added
    question
    client-api-v2
    on Mar 22, 2025

  4. chernser commented on Mar 22, 2025

    @chernser
    chernser
    on Mar 22, 2025
    Contributor

    Good day, @makrushin-evgenii !
    Thank you for the question!

    • Yes, RowBinaryFormatWriter is still in development. But both documentation and examples are planned.
    • POJO fields are matched by names disregarding snake case (getFirstName() will work with first_name column). And there is com.clickhouse.client.api.Client.Builder#columnToMethodMatchingStrategy to set any custom matching strategy.
    • You may use TabSeparated format for the insert you have in the example. Look at recent benchmarks: https://fastformats.clickhouse.com/

  5. makrushin-evgenii commented on Mar 22, 2025

    @makrushin-evgenii
    makrushin-evgenii
    on Mar 22, 2025
    Author

    Thank you for answer

    You may use TabSeparated format for the insert you have in the example. Look at recent benchmarks:

    Looks good in case of balance of performance and ease of use. But I cant find configurations to override separators. With default \n and \t I will catch payload parse errors. This field contains something like this:

    SELECT *
FROM tracing_spans_local
LIMIT 1
FORMAT Vertical

Query id: e7a3f713-8d10-4a09-a5af-ccb8cedabfe0

Row 1:
──────
date_hour:      0
trace_id:       af209200-0000-0000-0e73-0018a6df0053
parent_span_id: 8f205a43-d77f-698a-e900-6015d9051b8b
span_id:        366f4cdb-09c9-6922-0040-69f663039169
payload:        =���i3^�t�ﾶ��W�<traceId
��S� �spanId
ZC��i�s`���� beginTimestampUtc=���de|beginTimestampUtcOffset%M8
                                                               parentSpanId

1 row in set. Elapsed: 10.116 sec.

  6. chernser commented on Mar 25, 2025

    @chernser
    chernser
    on Mar 25, 2025
    Contributor

    I see. Yes, binary format is better here.
    Is payload encoded in some well know format? For example, if Avro https://clickhouse.com/docs/interfaces/formats/Avro it can be inserted into a table. But in this case schema should be more strict.

  7. makrushin-evgenii commented on Apr 2, 2025

    @makrushin-evgenii
    makrushin-evgenii
    on Apr 2, 2025
    Author

    Is payload encoded in some well know format?

    No. It is something like protobuf, but proprietary. It's easier to think of this field as just an array of bytes or string (field payload defined as String in destination CH table)

    Yes, binary format is better here.

    Do I understand correctly? Is RowBinaryFormatWriter a right class to use in my case?

    UPD: It seems like I didn't change anything, but insert started working. Now I have this code:

    @Component
public class TracesExporter extends ExporterBase<SinkEvent<TraceEvent>> {
    private static final Logger LOGGER = LoggerFactory.getLogger(TracesExporter.class);
    private final ClickhouseClientProvider clickhouseClientProvider;

    @Autowired
    public TracesExporter(
            FinishProcessor<SinkEvent<TraceEvent>> finishProcessor,
            ClickhouseClientProvider clickhouseClientProvider) {
        super(finishProcessor);
        this.clickhouseClientProvider = clickhouseClientProvider;
    }

    @Override
    public void handle(Stream<SinkEvent<TraceEvent>> events, Request request) {
        List<SinkEvent<TraceEvent>> eventsList = events.toList();
        LOGGER.info("eventsList.size() = {}", eventsList.size());
        LOGGER.info("eventsList.get(0) = {}", eventsList.get(0));
        var clickhouseClient = clickhouseClientProvider.get();

        try (var outputStream = new ByteArrayOutputStream()) {
            write(outputStream, eventsList);
            try (var inputStream = new ByteArrayInputStream(outputStream.toByteArray())) {
                try (var response = clickhouseClient.insert("tracing_spans_local", inputStream, ClickHouseFormat.RowBinary).get()) {
                    LOGGER.info("Insert succeed: {} rows written", response.getWrittenRows());
                } catch (ExecutionException | InterruptedException e) {
                    LOGGER.warn("Insert failure", e);
                    throw new RuntimeException(e);
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        finishProcessor.handle(events, request);
    }

    private void write(OutputStream outputStream, List<SinkEvent<TraceEvent>> events) throws IOException {
        var writer = new RowBinaryFormatWriter(outputStream, getTableSchema(), ClickHouseFormat.RowBinaryWithDefaults);
//        var writer = new RowBinaryFormatWriter(outputStream, getTableSchema(), ClickHouseFormat.RowBinary); // TODO: Если меняю на RowBinary, появляется ошибка
//         com.clickhouse.client.api.ServerException: Code: 33. DB::Exception: Cannot read all data. Bytes read: 3. Bytes expected: 8.: (at row 22) : While executing BinaryRowInputFormat. (CANNOT_READ_ALL_DATA) (version 25.4.1.1604 (official build))

        for (var event : events) {
            write(writer, event.internalEvent());
        }
    }

    private TableSchema getTableSchema() {
        var tableSchema = new TableSchema();
//        tableSchema.addColumn("date_hour", "UInt32"); // TODO: Если добавляю, появляется ошибка
//        com.clickhouse.client.api.ServerException: Code: 33. DB::Exception: Cannot read all data. Bytes read: 3. Bytes expected: 8.: (at row 22) : While executing BinaryRowInputFormat. (CANNOT_READ_ALL_DATA) (version 25.4.1.1604 (official build))
        tableSchema.addColumn("trace_id", "UUID");
        tableSchema.addColumn("span_id", "UUID");
        tableSchema.addColumn("parent_span_id", "UUID");
        tableSchema.addColumn("payload", "String");
        return tableSchema;
    }

    private void write(RowBinaryFormatWriter writer, TraceEvent event) throws IOException {
//        writer.setInteger("date_hour", 2025010100); // TODO: Если добавляю, появляется ошибка
//        com.clickhouse.client.api.ServerException: Code: 33. DB::Exception: Cannot read all data. Bytes read: 3. Bytes expected: 8.: (at row 22) : While executing BinaryRowInputFormat. (CANNOT_READ_ALL_DATA) (version 25.4.1.1604 (official build))
        writer.setValue("trace_id", event.traceId());
        writer.setValue("span_id", event.spanId());
        writer.setValue("parent_span_id", emptyIfNull(event.parentSpanId()));
        writer.setString("payload", new String(event.raw()));
        writer.commitRow();
    }

    private UUID emptyIfNull(UUID value) {
        return value == null ? UUID.fromString("00000000-0000-0000-0000-000000000000") : value;
    }
}

    In this configuration (without date_hour col, with ClickHouseFormat.RowBinaryWithDefaults for RowBinaryFormatWriter while ClickHouseFormat.RowBinary for .insert() call) inserts works. But data is spoiling. It feels like the data has "shifted":

    Row 2:
──────
date_hour:      1085339904 -- 1.09 billion           < got bytes from trace_id?
trace_id:       49004d67-df7f-f76e-c861-00b6fe60552d
parent_span_id: 00006783-ad83-5c22-0000-000000000000 < last bytes lost
span_id:        00000000-0000-0000-0000-000000000000
payload:        =�_��ܿ��m01���N� traceId

    If I use ClickHouseFormat.RowBinaryWithDefaults for both RowBinaryFormatWriter and .insert() call - I got error. If I use ClickHouseFormat.RowBinary and fill the date_hour col - I got the same error:

    com.clickhouse.client.api.ServerException: Code: 33. DB::Exception: Cannot read all data. Bytes read: 2507. Bytes expected: 59727855.: (at row 3) : While executing BinaryRowInputFormat. (CANNOT_READ_ALL_DATA) (version 25.4.1.1604 (official build)) 
	at com.clickhouse.client.api.internal.HttpAPIClientHelper.readError(HttpAPIClientHelper.java:384) ~[client-v2-0.8.2.jar:client-v2 0.8.2 (revision: 1414421)]
	at com.clickhouse.client.api.internal.HttpAPIClientHelper.executeRequest(HttpAPIClientHelper.java:434) ~[client-v2-0.8.2.jar:client-v2 0.8.2 (revision: 1414421)]
	at com.clickhouse.client.api.Client.lambda$insert$8(Client.java:1497) ~[client-v2-0.8.2.jar:client-v2 0.8.2 (revision: 1414421)]
	at com.clickhouse.client.api.Client.runAsyncOperation(Client.java:2007) ~[client-v2-0.8.2.jar:client-v2 0.8.2 (revision: 1414421)]
	at com.clickhouse.client.api.Client.insert(Client.java:1540) ~[client-v2-0.8.2.jar:client-v2 0.8.2 (revision: 1414421)]
	at com.clickhouse.client.api.Client.insert(Client.java:1426) ~[client-v2-0.8.2.jar:client-v2 0.8.2 (revision: 1414421)]
	at com.clickhouse.client.api.Client.insert(Client.java:1400) ~[client-v2-0.8.2.jar:client-v2 0.8.2 (revision: 1414421)]
	at ru.kontur.vostok.hercules.traces.sink.exporter.TracesExporter.handle(TracesExporter.java:47) ~[classes/:na]
  8. chernser
    added a parent issue on Nov 4, 2025

  9. github-actions commented on Apr 3, 2026

    @github-actions
    github-actionsBot
    on Apr 3, 2026 – with GitHub Actions

    This issue has been automatically marked as stale because it has not had activity in the last year. It will be closed in 30 days if no further activity occurs. Please feel free to leave a comment if you believe the issue is still relevant. Thank you for your contributions!

  10. github-actions
    added
    stale
    on Apr 3, 2026
  11. chernser
    removed
    stale
    on Apr 3, 2026
  12. chernser
    removed a parent issue on Nov 5, 2026
  13. chernser
    added a parent issue on Nov 5, 2026
  14. chernser
    removed a parent issue on Nov 5, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    client-api-v2enhancementquestion

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions