From ed0e317e29a9ab2a5e1eec07cfb5a22acf31c96e Mon Sep 17 00:00:00 2001 From: piyush15102003 Date: Thu, 5 Nov 2026 00:43:20 +0530 Subject: [PATCH] [client-v2] Keep row buffers across Native blocks so rows at a block boundary are not lost Closes #3151 --- .../internal/AbstractBinaryFormatReader.java | 17 +++- .../NativeFormatReaderBlockBoundaryTest.java | 93 +++++++++++++++++++ history/latest/3151.md | 8 ++ 3 files changed, 114 insertions(+), 4 deletions(-) create mode 100644 client-v2/src/test/java/com/clickhouse/client/api/data_formats/NativeFormatReaderBlockBoundaryTest.java create mode 100644 history/latest/3151.md diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/AbstractBinaryFormatReader.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/AbstractBinaryFormatReader.java index 6c175e2f4e..5bb924e016 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/AbstractBinaryFormatReader.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/AbstractBinaryFormatReader.java @@ -274,13 +274,16 @@ public Map next() { Object[] tmp = currentRecord; currentRecord = nextRecord; nextRecord = tmp; + // The read-ahead may start a new block and re-publish the schema, so take the record first. + Object[] record = currentRecord; readNextRecord(); - return new RecordWrapper(currentRecord, schema); + return new RecordWrapper(record, schema); } else { try { if (readRecord(currentRecord)) { + Object[] record = currentRecord; readNextRecord(); - return new RecordWrapper(currentRecord, schema); + return new RecordWrapper(record, schema); } else { currentRecord = null; return null; @@ -305,8 +308,14 @@ protected void setSchema(TableSchema schema) { this.columns = schema.getColumns().toArray(ClickHouseColumn.EMPTY_ARRAY); this.convertions = new Map[columns.length]; - this.currentRecord = new Object[columns.length]; - this.nextRecord = new Object[columns.length]; + // Native re-publishes the schema for every block while a row may already be read into these + // buffers, so they are kept when the shape is unchanged. + if (currentRecord == null || currentRecord.length != columns.length) { + this.currentRecord = new Object[columns.length]; + } + if (nextRecord == null || nextRecord.length != columns.length) { + this.nextRecord = new Object[columns.length]; + } for (int i = 0; i < columns.length; i++) { ClickHouseColumn column = columns[i]; diff --git a/client-v2/src/test/java/com/clickhouse/client/api/data_formats/NativeFormatReaderBlockBoundaryTest.java b/client-v2/src/test/java/com/clickhouse/client/api/data_formats/NativeFormatReaderBlockBoundaryTest.java new file mode 100644 index 0000000000..6d95a07698 --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/data_formats/NativeFormatReaderBlockBoundaryTest.java @@ -0,0 +1,93 @@ +package com.clickhouse.client.api.data_formats; + +import com.clickhouse.client.api.data_formats.internal.BinaryStreamReader; +import com.clickhouse.client.api.query.QuerySettings; +import com.clickhouse.data.format.BinaryStreamUtils; + +import org.testng.Assert; +import org.testng.annotations.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** + * Rows around a Native block boundary. The reader reads one row ahead and the next block re-publishes the + * schema, so these check every value, not only the row count (which was right even when rows were lost). + */ +public class NativeFormatReaderBlockBoundaryTest { + + @Test + public void testNextReturnsEveryRowAcrossBlocks() throws IOException { + // 25 rows split like SELECT toInt32(number) FROM numbers(25) SETTINGS max_block_size = 10 + NativeFormatReader reader = nativeReader(blocks(25, 10)); + + List values = new ArrayList<>(); + Map row; + while ((row = reader.next()) != null) { + values.add(row.get("n")); + } + + Assert.assertEquals(values, expected(25)); + } + + @Test + public void testGettersReturnEveryRowAcrossBlocks() throws IOException { + NativeFormatReader reader = nativeReader(blocks(25, 10)); + + List values = new ArrayList<>(); + while (reader.hasNext()) { + reader.next(); + values.add(reader.getInteger("n")); + } + + Assert.assertEquals(values, expected(25)); + } + + @Test + public void testSingleRowBlocks() throws IOException { + // Every row is both the last row of one block and followed by a new block. + NativeFormatReader reader = nativeReader(blocks(5, 1)); + + List values = new ArrayList<>(); + Map row; + while ((row = reader.next()) != null) { + values.add(row.get("n")); + } + + Assert.assertEquals(values, expected(5)); + } + + private static List expected(int rows) { + List values = new ArrayList<>(); + for (int i = 0; i < rows; i++) { + values.add(i); + } + return values; + } + + // Native blocks with a single Int32 column "n" holding 0..rows-1, at most blockSize rows per block. + private static byte[] blocks(int rows, int blockSize) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + for (int start = 0; start < rows; start += blockSize) { + int n = Math.min(blockSize, rows - start); + BinaryStreamUtils.writeVarInt(out, 1); // one column + BinaryStreamUtils.writeVarInt(out, n); + BinaryStreamUtils.writeString(out, "n"); + BinaryStreamUtils.writeString(out, "Int32"); + for (int i = start; i < start + n; i++) { + BinaryStreamUtils.writeInt32(out, i); + } + } + return out.toByteArray(); + } + + private static NativeFormatReader nativeReader(byte[] data) { + return new NativeFormatReader(new ByteArrayInputStream(data), + new QuerySettings().setUseTimeZone("UTC"), + new BinaryStreamReader.CachingByteBufferAllocator()); + } +} diff --git a/history/latest/3151.md b/history/latest/3151.md new file mode 100644 index 0000000000..bd60477d7e --- /dev/null +++ b/history/latest/3151.md @@ -0,0 +1,8 @@ +## Bug Fixes + +- **[client-v2]** Reading a `Native` result that spans more than one block no longer loses rows at + each block boundary. The last row of a block and the first row of the next were dropped and + returned as two all-`null` rows, with the row count unchanged and no error. The reader reads one + row ahead, and each new block re-published the schema, which replaced the row buffers while a + row was still held in them. The buffers are now kept when the column count is unchanged. + (https://github.com/ClickHouse/clickhouse-java/issues/3151)