Fix client-v2: consume Nullable null marker when reading into a primitive POJO field

CHANGELOG.md

@@ -44,6 +44,13 @@
 
 ### Bug Fixes 
 
+- **[client-v2]** Fixed a `Nullable(T)` column bound to a **primitive** POJO field silently corrupting a row on the
+  POJO read path. The compiled setter went straight to a primitive read method without consuming the `Nullable`
+  null-marker byte, which is on the wire for every value of a nullable column regardless of the value, so the stream
+  stayed shifted by one byte per row and the nullable column and every column after it decoded from the wrong offset
+  without any error being raised. The generated setter now consumes the marker; a value that is actually `NULL` cannot
+  be held by a primitive field and is reported with a `NullValueException`. Boxed POJO fields are unaffected.
+  (https://github.com/ClickHouse/clickhouse-java/issues/2993)
 - **[client-v2]** Fixed LZ4 input streams not closing their underlying HTTP response stream. Closing an LZ4 stream
   returned by `QueryResponse.getInputStream()` now releases the wrapped transport stream, including after a partial
   read. (https://github.com/ClickHouse/clickhouse-java/issues/2985)

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

@@ -2,6 +2,7 @@
 
 import com.clickhouse.client.api.ClientException;
 import com.clickhouse.client.api.DataTypeUtils;
+import com.clickhouse.client.api.query.NullValueException;
 import com.clickhouse.data.ClickHouseColumn;
 import com.clickhouse.data.ClickHouseDataType;
 import com.clickhouse.data.ClickHouseEnum;
@@ -1357,6 +1358,26 @@ public byte[] allocate(int size) {
         }
     }
 
+    /**
+     * Consumes the NULL marker that precedes every value of a nullable column when that value is read
+     * into a primitive variable. Does nothing for a column that is not nullable. Primitives cannot hold
+     * NULL, so a NULL value is reported instead of being silently replaced with a default value.
+     *
+     * @param column - column information
+     * @param targetType - name of the primitive type the value is read into
+     * @throws IOException when IO error occurs
+     */
+    public void readNullMarkerForPrimitive(ClickHouseColumn column, String targetType) throws IOException {
+        if (!column.isNullable()) {
+            return;
+        }
+
+        if (readByteOrEOF(input) == 1) {
+            throw new NullValueException("Column " + column.getColumnName()
+                    + " has null value and it cannot be cast to " + targetType);
+        }
+    }
+
     public static boolean isReadToPrimitive(ClickHouseDataType dataType) {
         switch (dataType) {
             case Int8:

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

@@ -53,6 +53,7 @@
 import static org.objectweb.asm.Opcodes.ACC_PUBLIC;
 import static org.objectweb.asm.Opcodes.ALOAD;
 import static org.objectweb.asm.Opcodes.CHECKCAST;
+import static org.objectweb.asm.Opcodes.DUP;
 import static org.objectweb.asm.Opcodes.INVOKESPECIAL;
 import static org.objectweb.asm.Opcodes.INVOKESTATIC;
 import static org.objectweb.asm.Opcodes.INVOKEVIRTUAL;
@@ -1176,6 +1177,7 @@ public static List<?> convertArrayValueToList(Object value) {
      * @see SerializerUtils#longToOpcode(Class)
      * @see SerializerUtils#floatToOpcode(Class)
      * @see SerializerUtils#doubleToOpcode(Class)
+     * @see BinaryStreamReader#readNullMarkerForPrimitive(ClickHouseColumn, String)
      * @see BinaryStreamReader#readValue(ClickHouseColumn, Class)
      * @see BinaryStreamReader#readByte()
      * @see BinaryStreamReader#readUnsignedByte()
@@ -1234,9 +1236,11 @@ public static POJOFieldDeserializer compilePOJOSetter(Method setterMethod, Click
             mv.visitVarInsn(ALOAD, 2); // load reader
 
             if (targetType.isPrimitive() && BinaryStreamReader.isReadToPrimitive(column.getDataType())) {
+                nullMarkerReaderForPrimitive(mv, targetType);
                 binaryReaderMethodForType(mv,
                         targetPrimitiveType, column.getDataType());
             } else if (targetType.isPrimitive() && column.getDataType() == ClickHouseDataType.UInt64) {
+                nullMarkerReaderForPrimitive(mv, targetType);
                 mv.visitTypeInsn(CHECKCAST, Type.getInternalName(BigInteger.class));
                 mv.visitMethodInsn(INVOKEVIRTUAL,
                         Type.getInternalName(BigInteger.class),
@@ -1305,6 +1309,28 @@ public static POJOFieldDeserializer compilePOJOSetter(Method setterMethod, Click
         }
     }
 
+    /**
+     * Emits a call to {@link BinaryStreamReader#readNullMarkerForPrimitive(ClickHouseColumn, String)}. Values of a
+     * nullable column are prefixed with a NULL marker on the wire regardless of the value itself, so a reader that goes
+     * directly to a primitive read method has to consume that marker - otherwise the stream stays shifted by one byte
+     * for the rest of the row. Nullability is decided by the callee from the column being read, not at compile time,
+     * because the column of the result being read may differ from the one the setter was compiled for.
+     *
+     * @param mv - visitor of the method being generated
+     * @param targetType - primitive type the value is read into
+     */
+    private static void nullMarkerReaderForPrimitive(MethodVisitor mv, Class<?> targetType) {
+        mv.visitInsn(DUP); // reader
+        mv.visitVarInsn(ALOAD, 3); // column
+        mv.visitLdcInsn(targetType.getName());
+        mv.visitMethodInsn(INVOKEVIRTUAL,
+                Type.getInternalName(BinaryStreamReader.class),
+                "readNullMarkerForPrimitive",
+                Type.getMethodDescriptor(Type.VOID_TYPE,
+                        Type.getType(ClickHouseColumn.class), Type.getType(String.class)),
+                false);
+    }
+
     private static void binaryReaderMethodForType(MethodVisitor mv, Class<?> targetType, ClickHouseDataType dataType) {
         String readerMethod = null;
         String readerMethodReturnType = null;

client-v2/src/test/java/com/clickhouse/client/datatypes/DataTypeTests.java

@@ -13,6 +13,7 @@
 import com.clickhouse.client.api.insert.InsertSettings;
 import com.clickhouse.client.api.metadata.TableSchema;
 import com.clickhouse.client.api.query.GenericRecord;
+import com.clickhouse.client.api.query.NullValueException;
 import com.clickhouse.client.api.query.QueryResponse;
 import com.clickhouse.client.api.query.QuerySettings;
 import com.clickhouse.client.api.sql.SQLUtils;
@@ -2444,6 +2445,144 @@ public Object[][] testJSONSubPathAccess_dp() {
         };
     }
 
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    public static class DTOForNullablePrimitivesTests {
+        private int rowId;
+        private byte int8;
+        private short uint8;
+        private short int16;
+        private int uint16;
+        private int int32;
+        private long uint32;
+        private long int64;
+        private float float32;
+        private double float64;
+        private boolean bool;
+        private byte enum8;
+        private short enum16;
+        private long trailing;
+    }
+
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    public static class DTOForNullablePrimitiveBFloat16Tests {
+        private int rowId;
+        private float bFloat16;
+        private long trailing;
+    }
+
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    public static class DTOForNullPrimitiveTests {
+        private int rowId;
+        private long int64;
+        private long trailing;
+    }
+
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    public static class DTOForNullBoxedTests {
+        private int rowId;
+        private Long int64;
+        private long trailing;
+    }
+
+    private static final String NULLABLE_PRIMITIVES_SQL =
+            "SELECT toInt32(number) AS rowId" +
+                    ", toNullable(toInt8(-128)) AS int8" +
+                    ", toNullable(toUInt8(255)) AS uint8" +
+                    ", toNullable(toInt16(-32768)) AS int16" +
+                    ", toNullable(toUInt16(65535)) AS uint16" +
+                    ", toNullable(toInt32(-2147483648)) AS int32" +
+                    ", toNullable(toUInt32(4294967295)) AS uint32" +
+                    ", toNullable(toInt64(-9223372036854775808)) AS int64" +
+                    ", toNullable(toFloat32(1.5)) AS float32" +
+                    ", toNullable(toFloat64(2.25)) AS float64" +
+                    ", toNullable(true) AS bool" +
+                    ", toNullable(CAST('b', 'Enum8(''a'' = 1, ''b'' = 2)')) AS enum8" +
+                    ", toNullable(CAST('y', 'Enum16(''x'' = 1000, ''y'' = 2000)')) AS enum16" +
+                    ", toInt64(777) AS trailing FROM numbers(3)";
+
+    private static final String NULL_VALUE_SQL =
+            "SELECT toInt32(1) AS rowId, CAST(NULL, 'Nullable(Int64)') AS int64, toInt64(777) AS trailing";
+
+    @Test(groups = {"integration"})
+    public void testReadNullableColumnsIntoPrimitivePojoFields() {
+        TableSchema schema = client.getTableSchemaFromQuery(NULLABLE_PRIMITIVES_SQL);
+        client.register(DTOForNullablePrimitivesTests.class, schema);
+
+        List<DTOForNullablePrimitivesTests> rows =
+                client.queryAll(NULLABLE_PRIMITIVES_SQL, DTOForNullablePrimitivesTests.class, schema);
+
+        List<DTOForNullablePrimitivesTests> expected = new ArrayList<>();
+        for (int rowId = 0; rowId < 3; rowId++) {
+            expected.add(new DTOForNullablePrimitivesTests(rowId, Byte.MIN_VALUE, (short) 255, Short.MIN_VALUE, 65535,
+                    Integer.MIN_VALUE, 4294967295L, Long.MIN_VALUE, 1.5f, 2.25d, true, (byte) 2, (short) 2000, 777L));
+        }
+        Assert.assertEquals(rows, expected);
+    }
+
+    @Test(groups = {"integration"})
+    public void testReadNullableBFloat16IntoPrimitivePojoField() {
+        if (isVersionMatch(BFLOAT16_UNSUPPORTED_VERSIONS)) {
+            throw new SkipException("BFloat16 requires ClickHouse 24.11+");
+        }
+
+        final String sql = "SELECT toInt32(1) AS rowId, CAST(1.5, 'Nullable(BFloat16)') AS bFloat16"
+                + ", toInt64(777) AS trailing";
+        TableSchema schema = client.getTableSchemaFromQuery(sql);
+        client.register(DTOForNullablePrimitiveBFloat16Tests.class, schema);
+
+        Assert.assertEquals(client.queryAll(sql, DTOForNullablePrimitiveBFloat16Tests.class, schema),
+                Collections.singletonList(new DTOForNullablePrimitiveBFloat16Tests(1, 1.5f, 777L)));
+    }
+
+    @Test(groups = {"integration"})
+    public void testReadNonNullableColumnIntoPrimitivePojoFieldRegisteredAsNullable() throws Exception {
+        final String table = "test_nullable_primitive_pojo_field";
+        client.execute("DROP TABLE IF EXISTS " + table).get();
+        client.execute(tableDefinition(table, "rowId Int32", "int64 Nullable(Int64)", "trailing Int64")).get();
+        client.execute("INSERT INTO " + table + " VALUES (1, -9223372036854775808, 777)").get();
+
+        TableSchema schema = client.getTableSchema(table);
+        client.register(DTOForNullPrimitiveTests.class, schema);
+
+        List<DTOForNullPrimitiveTests> rows = client.queryAll(
+                "SELECT rowId, assumeNotNull(int64) AS int64, trailing FROM " + table,
+                DTOForNullPrimitiveTests.class, schema);
+
+        Assert.assertEquals(rows, Collections.singletonList(new DTOForNullPrimitiveTests(1, Long.MIN_VALUE, 777L)));
+    }
+
+    @Test(groups = {"integration"})
+    public void testReadNullValueIntoPrimitivePojoField() {
+        TableSchema schema = client.getTableSchemaFromQuery(NULL_VALUE_SQL);
+        client.register(DTOForNullPrimitiveTests.class, schema);
+
+        ClientException exception = Assert.expectThrows(ClientException.class,
+                () -> client.queryAll(NULL_VALUE_SQL, DTOForNullPrimitiveTests.class, schema));
+        Throwable cause = exception.getCause();
+        while (cause != null && !(cause instanceof NullValueException)) {
+            cause = cause.getCause();
+        }
+        Assert.assertNotNull(cause, "NullValueException is not in the cause chain of " + exception);
+        Assert.assertEquals(cause.getMessage(), "Column int64 has null value and it cannot be cast to long");
+    }
+
+    @Test(groups = {"integration"})
+    public void testReadNullValueIntoBoxedPojoField() {
+        TableSchema schema = client.getTableSchemaFromQuery(NULL_VALUE_SQL);
+        client.register(DTOForNullBoxedTests.class, schema);
+
+        Assert.assertEquals(client.queryAll(NULL_VALUE_SQL, DTOForNullBoxedTests.class, schema),
+                Collections.singletonList(new DTOForNullBoxedTests(1, null, 777L)));
+    }
+
     public static String tableDefinition(String table, String... columns) {
         StringBuilder sb = new StringBuilder();
         sb.append("CREATE TABLE " + table + " ( ");

docs/features.md

@@ -16,7 +16,7 @@ This document lists stable, user-visible behavior in `client-v2` and `jdbc-v2` t
 - Query execution: Executes SQL asynchronously and returns streaming query responses with response metadata and metrics.
 - Query settings: Supports per-query database selection, output format, execution limits, roles, log comments, headers, reusable `Session` objects, session settings, server settings, and network timeout overrides. Settings explicitly set to `null` will not be sent to the server.
 - Parameterized SQL: Accepts named query parameters and can send them through supported HTTP request encodings.
-- Result materialization helpers: Provides streaming `Records`, generic row access, and convenience APIs that materialize all rows into generic records or typed POJOs.
+- Result materialization helpers: Provides streaming `Records`, generic row access, and convenience APIs that materialize all rows into generic records or typed POJOs. A `Nullable(T)` column may be bound to a POJO field of a primitive type (e.g. `long`), which reads any non-`NULL` value without boxing; a value that is actually `NULL` cannot be held by such a field and is reported with a `NullValueException` (use the boxed type to accept `NULL`).
 - Binary format readers: Reads ClickHouse binary result formats including `Native`, `RowBinary`, `RowBinaryWithNames`, and `RowBinaryWithNamesAndTypes`.
 - Binary string support: Opt-in through the `binary_string_support` property (or `Client.Builder#binaryStringSupport(boolean)`), disabled by default. The setting is resolved per operation from the merged client and query settings, so it can be overridden for a single request by setting the `binary_string_support` option on the operation's settings (e.g. `QuerySettings#setOption(ClientConfigProperties.BINARY_STRING_SUPPORT.getKey(), true)`) regardless of the client-level default. When enabled, untyped reads (e.g. `GenericRecord.getObject(...)`/`BinaryStreamReader.readValue(...)`) of top-level `String` and `FixedString` columns return a `StringValue`, which preserves the raw bytes (`toByteArray()`/`asByteBuffer()`) and lazily decodes a `String` (`asString()`), instead of eagerly decoding to a `String`. `StringValue` is a read-time holder, not a supported POJO field type: typed `queryAll(...)`/`readToPOJO` binding still maps these columns to `String` (decoded) or `byte[]` (raw bytes) according to the POJO field type. Strings nested inside containers (`Array`, `Map`, `Tuple`, `Nested`, `Variant`, `JSON`) are still read as `String`.
 - JSONEachRow text reader: Can stream `JSONEachRow` responses through a caller-supplied `JsonParser`, with Jackson and Gson parser factory implementations available as optional classpath dependencies, and infers a best-effort schema from the first row.