[client-v2] Keep row buffers across Native blocks so rows at a block boundary are not lost

client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/AbstractBinaryFormatReader.java

@@ -274,13 +274,16 @@ public Map<String, Object> 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];

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<Object> values = new ArrayList<>();
+        Map<String, Object> 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<Object> 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<Object> values = new ArrayList<>();
+        Map<String, Object> row;
+        while ((row = reader.next()) != null) {
+            values.add(row.get("n"));
+        }
+
+        Assert.assertEquals(values, expected(5));
+    }
+
+    private static List<Object> expected(int rows) {
+        List<Object> 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());
+    }
+}

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)