Fix client-v2 Native reader misreading multi-row Array columns

CHANGELOG.md

@@ -44,6 +44,12 @@
 
 ### Bug Fixes 
 
+- **[client-v2]** Fixed the `Native` format reader (`NativeFormatReader`) misreading `Array` columns in multi-row
+  results whose rows have different lengths. Native encodes an array column as cumulative row offsets followed by the
+  flattened elements, but the reader used the first row's offset as the element count for every row — truncating later
+  rows and desyncing the columns that follow the array in the same block. Each row's length is now derived from the
+  difference between consecutive offsets, and empty array rows (`len == 0`) no longer read a phantom element. Results
+  with uniform array lengths were unaffected. (https://github.com/ClickHouse/clickhouse-java/issues/2955)
 - **[client-v2, jdbc-v2]** Reduced noisy and potentially sensitive logging; SQL that fails to parse is no
   longer logged at `WARN` (it could contain credentials/PII). (https://github.com/ClickHouse/clickhouse-java/issues/2970)
 - **[client-v2]** Fixed `BigDecimal` values written into a `Dynamic` column being silently truncated when the

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

@@ -108,12 +108,18 @@ private boolean readBlock() throws IOException {
 
             List<Object> values = new ArrayList<>(nRows);
             if (column.isArray()) {
-                int[] sizes = new int[nRows];
+                // Native encodes an Array column as nRows cumulative offsets followed by the
+                // flattened elements; each row's element count is the delta between consecutive
+                // offsets, not the first offset.
+                long[] offsets = new long[nRows];
                 for (int j = 0; j < nRows; j++) {
-                    sizes[j] = Math.toIntExact(binaryStreamReader.readLongLE());
+                    offsets[j] = binaryStreamReader.readLongLE();
                 }
+                long prevOffset = 0;
                 for (int j = 0; j < nRows; j++) {
-                    values.add(binaryStreamReader.readArrayItem(column.getNestedColumns().get(0), sizes[0]));
+                    int len = Math.toIntExact(offsets[j] - prevOffset);
+                    values.add(binaryStreamReader.readArrayItem(column.getNestedColumns().get(0), len));
+                    prevOffset = offsets[j];
                 }
             } else {
                 for (int j = 0; j < nRows; j++) {

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

@@ -689,6 +689,11 @@ public ArrayValue readArray(ClickHouseColumn column) throws IOException {
     }
 
     public ArrayValue readArrayItem(ClickHouseColumn itemTypeColumn, int len) throws IOException {
+        if (len == 0) {
+            // Nothing to read for an empty array; typing it via resolveArrayItemClass avoids the
+            // primitive branch below reading a phantom element and indexing a zero-length array.
+            return new ArrayValue(resolveArrayItemClass(itemTypeColumn), 0);
+        }
         ArrayValue array;
         if (itemTypeColumn.isNullable()) {
             Class<?> itemClass = resolveArrayItemClass(itemTypeColumn);

client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java

@@ -482,6 +482,52 @@
         }
     }
 
+    @DataProvider(name = "multiRowArrayCases")
+    Object[][] getMultiRowArrayCases() {
+        String nonUniform = "SELECT id, arr, tag FROM values("
+                + "'id UInt32, arr Array(Int32), tag Int32', "
+                + "(1, [10], 100), (2, [20, 21], 200), (3, [], 300), (4, [30, 31, 32], 400), (5, [40], 500)"
+                + ") ORDER BY id";
+        List<Object[]> nonUniformRows = Arrays.asList(
+                new Object[]{1L, Arrays.asList(10), 100},
+                new Object[]{2L, Arrays.asList(20, 21), 200},
+                new Object[]{3L, Collections.emptyList(), 300},
+                new Object[]{4L, Arrays.asList(30, 31, 32), 400},
+                new Object[]{5L, Arrays.asList(40), 500});
+
+        String uniform = "SELECT id, arr, tag FROM values("
+                + "'id UInt32, arr Array(Int32), tag Int32', "
+                + "(1, [10, 11], 100), (2, [20, 21], 200), (3, [30, 31], 300)"
+                + ") ORDER BY id";
+        List<Object[]> uniformRows = Arrays.asList(
+                new Object[]{1L, Arrays.asList(10, 11), 100},
+                new Object[]{2L, Arrays.asList(20, 21), 200},
+                new Object[]{3L, Arrays.asList(30, 31), 300});
+
+        return new Object[][]{
+                {ClickHouseFormat.Native, nonUniform, nonUniformRows},
+                {ClickHouseFormat.Native, uniform, uniformRows},
+                {ClickHouseFormat.RowBinaryWithNamesAndTypes, nonUniform, nonUniformRows},
+        };
+    }
+
+    @Test(groups = {"integration"}, dataProvider = "multiRowArrayCases")
+    public void testReadingMultiRowArrays(ClickHouseFormat format, String sql, List<Object[]> expectedRows)
+            throws Exception {
+        QuerySettings settings = new QuerySettings().setFormat(format);
+        try (QueryResponse response = client.query(sql, settings).get()) {
+            ClickHouseBinaryFormatReader reader = client.newBinaryFormatReader(response);
+            for (Object[] expected : expectedRows) {
+                Map<String, Object> record = reader.next();
+                Assert.assertNotNull(record, "Expected a row for id " + expected[0]);
+                Assert.assertEquals(record.get("id"), expected[0]);
+                Assert.assertEquals(((BinaryStreamReader.ArrayValue) record.get("arr")).asList(), expected[1]);
+                Assert.assertEquals(record.get("tag"), expected[2]);
+            }
+            Assert.assertNull(reader.next());
+        }
+    }
+
     @Test(groups = {"integration"})
     public void testBinaryStreamReader() throws Exception {
         final String table = "dynamic_schema_test_table";