Fix client-v2 RowBinary: write Nullable marker for nested Tuple/Map elements

A Nullable element nested inside a Tuple or Map was serialized without its
leading null-marker byte for present (non-null) values, corrupting the
RowBinary stream so the server (and the client's own reader) mis-parsed the
following bytes. Top-level columns get the marker from writeValuePreamble and
serializeArrayData already wrote per-element markers, but serializeTupleData
and serializeMapData called serializeData directly and skipped it.

Add a serializeNestedData helper that writes the 0x00 (present) / 0x01 (null)
marker for nested Nullable elements, and route Tuple elements and Map values
through it. The array and top-level paths are unchanged.

Fixes: #2721

Author: polyglotAI-bot

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

@@ -111,6 +111,25 @@ public static void serializeData(OutputStream stream, Object value, ClickHouseCo
         }
     }
 
+    /**
+     * Serializes a value that is nested inside a container ({@code Tuple} or {@code Map}). A nested
+     * {@code Nullable} element is prefixed with its null-marker byte ({@code 0x00} when present,
+     * {@code 0x01} when null), as the server expects for {@code Nullable} sub-columns in
+     * {@code RowBinary}. For a top-level column this marker is instead written by
+     * {@link com.clickhouse.client.api.data_formats.RowBinaryFormatSerializer#writeValuePreamble},
+     * so this helper must only be used for nested elements.
+     */
+    private static void serializeNestedData(OutputStream stream, Object value, ClickHouseColumn column) throws IOException {
+        if (column.isNullable()) {
+            if (value == null) {
+                writeNull(stream);
+                return;
+            }
+            writeNonNull(stream);
+        }
+        serializeData(stream, value, column);
+    }
+
     private static final Map<Class<?>, ClickHouseColumn> PREDEFINED_TYPE_COLUMNS = getPredefinedTypeColumnsMap();
 
     private static Map<Class<?>, ClickHouseColumn> getPredefinedTypeColumnsMap() {
@@ -460,12 +479,12 @@ public static void serializeTupleData(OutputStream stream, Object value, ClickHo
         if (value instanceof List) {
             List<?> values = (List<?>) value;
             for (int i = 0; i < values.size(); i++) {
-                serializeData(stream, values.get(i), column.getNestedColumns().get(i));
+                serializeNestedData(stream, values.get(i), column.getNestedColumns().get(i));
             }
         } else if (value.getClass().isArray()) {
             // TODO: this code uses reflection - we might need to measure it and find faster solution.
             for (int i = 0; i < Array.getLength(value); i++) {
-                serializeData(stream, Array.get(value, i), column.getNestedColumns().get(i));
+                serializeNestedData(stream, Array.get(value, i), column.getNestedColumns().get(i));
             }
         } else {
             throw new IllegalArgumentException("Cannot serialize " + value + " as a tuple");
@@ -483,7 +502,7 @@ public static void serializeMapData(OutputStream stream, Object value, ClickHous
         map.forEach((key, val) -> {
             try {
                 serializePrimitiveData(stream, key, Objects.requireNonNull(column.getKeyInfo()));
-                serializeData(stream, val, Objects.requireNonNull(column.getValueInfo()));
+                serializeNestedData(stream, val, Objects.requireNonNull(column.getValueInfo()));
             } catch (IOException e) {
                 throw new RuntimeException(e);
             }

client-v2/src/test/java/com/clickhouse/client/api/data_formats/internal/SerializerUtilsTest.java

@@ -9,7 +9,9 @@
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.util.Arrays;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.TimeZone;
 
 @Test(groups = {"unit"})
@@ -134,6 +136,74 @@ public void testGeometrySerializationRejectsMalformedList() {
                         ClickHouseColumn.of("v", "Geometry")));
     }
 
+    @Test
+    public void testTupleWithNullableElementsRoundTrip() throws Exception {
+        // Regression for #2721: a Nullable element nested inside a Tuple must be prefixed with its
+        // null-marker byte (0x00 present, 0x01 null). Before the fix the present-value marker was
+        // missing, so the reader mis-parsed the bytes that followed. The non-nullable sibling element
+        // must keep being written without a marker.
+        ClickHouseColumn column = ClickHouseColumn.of("v", "Tuple(Nullable(String), String)");
+
+        ByteArrayOutputStream present = new ByteArrayOutputStream();
+        SerializerUtils.serializeData(present, Arrays.asList("optional-value", "value-2"), column);
+        Assert.assertEquals((Object[]) newReader(present.toByteArray()).readValue(column),
+                new Object[] {"optional-value", "value-2"});
+
+        ByteArrayOutputStream absent = new ByteArrayOutputStream();
+        SerializerUtils.serializeData(absent, Arrays.asList(null, "value-2"), column);
+        Assert.assertEquals((Object[]) newReader(absent.toByteArray()).readValue(column),
+                new Object[] {null, "value-2"});
+    }
+
+    @Test
+    public void testMapWithNullableValuesRoundTrip() throws Exception {
+        // Regression for #2721: a Nullable Map value must be prefixed with its null-marker byte.
+        ClickHouseColumn column = ClickHouseColumn.of("v", "Map(String, Nullable(String))");
+        Map<String, String> value = new LinkedHashMap<>();
+        value.put("k1", "v1");
+        value.put("k2", null);
+
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        SerializerUtils.serializeData(out, value, column);
+
+        Map<?, ?> result = newReader(out.toByteArray()).readValue(column);
+        Assert.assertEquals(result, value);
+    }
+
+    @Test
+    public void testTupleWithNullableFixedWidthElementsRoundTrip() throws Exception {
+        // #2721 is about the marker byte, not the element type: a fixed-width Nullable element
+        // (here Int32) must also get its marker so the following value bytes are not misaligned.
+        // Also covers an all-null tuple.
+        ClickHouseColumn column = ClickHouseColumn.of("v", "Tuple(Nullable(Int32), Nullable(String))");
+
+        ByteArrayOutputStream present = new ByteArrayOutputStream();
+        SerializerUtils.serializeData(present, Arrays.asList(42, "x"), column);
+        Assert.assertEquals((Object[]) newReader(present.toByteArray()).readValue(column),
+                new Object[] {42, "x"});
+
+        ByteArrayOutputStream allNull = new ByteArrayOutputStream();
+        SerializerUtils.serializeData(allNull, Arrays.asList(null, null), column);
+        Assert.assertEquals((Object[]) newReader(allNull.toByteArray()).readValue(column),
+                new Object[] {null, null});
+    }
+
+    @Test
+    public void testNestedContainerWithNullableRoundTrip() throws Exception {
+        // #2721: the marker handling must compose through nested containers — here a Map carrying a
+        // Nullable value sits inside a Tuple, exercising serializeTupleData -> serializeMapData.
+        ClickHouseColumn column = ClickHouseColumn.of("v", "Tuple(String, Map(String, Nullable(String)))");
+        Map<String, String> inner = new LinkedHashMap<>();
+        inner.put("k1", "v1");
+        inner.put("k2", null);
+
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        SerializerUtils.serializeData(out, Arrays.asList("id", inner), column);
+
+        Assert.assertEquals((Object[]) newReader(out.toByteArray()).readValue(column),
+                new Object[] {"id", inner});
+    }
+
     private void assertCustomGeoTypeTag(String typeName) throws Exception {
         ByteArrayOutputStream out = new ByteArrayOutputStream();
         SerializerUtils.writeDynamicTypeTag(out, ClickHouseColumn.of("v", typeName));