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

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

@@ -4,13 +4,22 @@
 import com.clickhouse.data.ClickHouseDataType;
 import com.clickhouse.data.value.ClickHouseGeoPolygonValue;
 import org.testng.Assert;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.net.InetAddress;
+import java.time.LocalDate;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.TimeZone;
+import java.util.UUID;
 
 @Test(groups = {"unit"})
 public class SerializerUtilsTest {
@@ -134,6 +143,115 @@
                         ClickHouseColumn.of("v", "Geometry")));
     }
 
+    @Test(dataProvider = "nestedNullableData")
+    public void testNestedNullableRoundTrip(String typeName, Object value) throws Exception {
+        ClickHouseColumn column = ClickHouseColumn.of("v", typeName);
+
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        SerializerUtils.serializeData(out, value, column);
+
+        Object actual = newReader(out.toByteArray()).readValue(column);
+        Assert.assertEquals(normalize(actual), normalize(value));
+    }
+
+    @DataProvider(name = "nestedNullableData")
+    private Object[][] nestedNullableData() throws Exception {
+        UUID uuid = UUID.fromString("61f0c404-5cb3-11e7-907b-a6006ad3dba0");
+        InetAddress ipv4 = InetAddress.getByName("1.2.3.4");
+        return new Object[][] {
+                // Each present Nullable element sits in the MIDDLE of the schema: a non-nullable
+                // leading column, the Nullable, then a trailing non-nullable Float64. If the
+                // present-marker byte is dropped, every following byte shifts and the trailing
+                // Float64 reads a wrong value, so a faulty serialization is detected positionally
+                // rather than only by running out of bytes. The assertion compares the whole row.
+                {"Tuple(Int32, Nullable(String), Float64)", Arrays.asList(7, "opt", 9.5d)},
+                {"Tuple(Int32, Nullable(FixedString(3)), Float64)", Arrays.asList(7, "abc", 9.5d)},
+                {"Tuple(Int32, Nullable(Int8), Float64)", Arrays.asList(7, (byte) -5, 9.5d)},
+                {"Tuple(Int32, Nullable(UInt8), Float64)", Arrays.asList(7, (short) 200, 9.5d)},
+                {"Tuple(Int32, Nullable(Int16), Float64)", Arrays.asList(7, (short) -1600, 9.5d)},
+                {"Tuple(Int32, Nullable(UInt16), Float64)", Arrays.asList(7, 40000, 9.5d)},
+                {"Tuple(Int32, Nullable(Int32), Float64)", Arrays.asList(7, 42, 9.5d)},
+                {"Tuple(Int32, Nullable(UInt32), Float64)", Arrays.asList(7, 4_000_000_000L, 9.5d)},
+                {"Tuple(Int32, Nullable(Int64), Float64)", Arrays.asList(7, -64L, 9.5d)},
+                {"Tuple(Int32, Nullable(UInt64), Float64)", Arrays.asList(7, BigInteger.valueOf(64), 9.5d)},
+                {"Tuple(Int32, Nullable(Float32), Float64)", Arrays.asList(7, 1.5f, 9.5d)},
+                {"Tuple(Int32, Nullable(Float64), Float64)", Arrays.asList(7, 2.5d, 9.5d)},
+                {"Tuple(Int32, Nullable(Bool), Float64)", Arrays.asList(7, true, 9.5d)},
+                {"Tuple(Int32, Nullable(UUID), Float64)", Arrays.asList(7, uuid, 9.5d)},
+                {"Tuple(Int32, Nullable(Date), Float64)", Arrays.asList(7, LocalDate.of(2021, 2, 3), 9.5d)},
+                {"Tuple(Int32, Nullable(Decimal64(4)), Float64)", Arrays.asList(7, new BigDecimal("1.2345"), 9.5d)},
+                {"Tuple(Int32, Nullable(IPv4), Float64)", Arrays.asList(7, ipv4, 9.5d)},
+
+                // A Tuple value given as a Java array (not a List) takes the other branch of
+                // serializeTupleData, which is routed through the same nested-marker path, for
+                // both a present value and a null.
+                {"Tuple(Int32, Nullable(String), Float64)", new Object[] {7, "opt", 9.5d}},
+                {"Tuple(Int32, Nullable(String), Float64)", new Object[] {7, null, 9.5d}},
+
+                // The Map value path: the Nullable map value sits between the key and a trailing
+                // Float64, so a dropped value-marker misaligns the float.
+                {"Tuple(Int32, Map(String, Nullable(String)), Float64)", Arrays.asList(7, newMap("k", "v"), 9.5d)},
+                {"Tuple(Int32, Map(String, Nullable(Int32)), Float64)", Arrays.asList(7, newMap("k", 32), 9.5d)},
+                {"Tuple(Int32, Map(String, Nullable(Float64)), Float64)", Arrays.asList(7, newMap("k", 2.5d), 9.5d)},
+                {"Tuple(Int32, Map(String, Nullable(UUID)), Float64)", Arrays.asList(7, newMap("k", uuid), 9.5d)},
+
+                // Null elements/values still serialize a single null-marker byte; the trailing
+                // Float64 confirms the following data stays aligned.
+                {"Tuple(Int32, Nullable(String), Float64)", Arrays.asList(7, null, 9.5d)},
+                {"Tuple(Int32, Nullable(Int32), Nullable(String), Float64)", Arrays.asList(7, null, null, 9.5d)},
+                {"Tuple(Int32, Map(String, Nullable(String)), Float64)", Arrays.asList(7, newMap("k", null), 9.5d)},
+
+                // Containers compose: marker handling threads through nested Tuple/Map/Array,
+                // including Array(Tuple(Nullable)) which is how Nested columns are encoded. A
+                // trailing Float64 after each nested container detects misalignment.
+                {"Array(Tuple(Int32, Nullable(String), Float64))",
+                        Arrays.asList(Arrays.asList(7, "a", 9.5d), Arrays.asList(7, null, 8.5d))},
+                {"Tuple(String, Map(String, Nullable(Int32)), Float64)",
+                        Arrays.asList("id", newMap("k1", 7, "k2", null), 9.5d)},
+                {"Tuple(Array(Nullable(Int32)), Float64)", Arrays.asList(Arrays.asList(1, null, 3), 9.5d)},
+
+                // Contrast: non-nullable nested elements must keep serializing without a marker,
+                // so these rows round-trip identically with or without the fix.
+                {"Tuple(Int32, String, Float64)", Arrays.asList(7, "tail", 9.5d)},
+                {"Tuple(Int32, Map(String, String), Float64)", Arrays.asList(7, newMap("k", "v"), 9.5d)},
+        };
+    }
+
+    // Normalizes Tuple (Object[]) and Array (ArrayValue / List) results to nested Lists so
+    // round-tripped values compare structurally regardless of the container representation the
+    // reader returns.
+    @SuppressWarnings("unchecked")
+    private static Object normalize(Object value) {
+        if (value instanceof BinaryStreamReader.ArrayValue) {
+            return normalizeList(((BinaryStreamReader.ArrayValue) value).asList());
+        } else if (value instanceof Object[]) {
+            return normalizeList(Arrays.asList((Object[]) value));
+        } else if (value instanceof List) {
+            return normalizeList((List<Object>) value);
+        } else if (value instanceof Map) {
+            Map<Object, Object> result = new LinkedHashMap<>();
+            ((Map<Object, Object>) value).forEach((k, v) -> result.put(k, normalize(v)));
+            return result;
+        }
+        return value;
+    }
+
+    private static List<Object> normalizeList(List<Object> values) {
+        List<Object> result = new ArrayList<>(values.size());
+        for (Object v : values) {
+            result.add(normalize(v));
+        }
+        return result;
+    }
+
+    private static Map<Object, Object> newMap(Object... kv) {
+        Map<Object, Object> map = new LinkedHashMap<>();
+        for (int i = 0; i < kv.length; i += 2) {
+            map.put(kv[i], kv[i + 1]);
+        }
+        return map;
+    }
+
     private void assertCustomGeoTypeTag(String typeName) throws Exception {
         ByteArrayOutputStream out = new ByteArrayOutputStream();
         SerializerUtils.writeDynamicTypeTag(out, ClickHouseColumn.of("v", typeName));