Fix client-v2: reject null into non-nullable Array in RowBinary writer

CHANGELOG.md

@@ -103,6 +103,15 @@
   NPE instead of a clear error. It now throws `IllegalArgumentException` naming the column, consistent with
   the existing `IllegalArgumentException` for other unsupported enum values. Nullable enum columns are
   unaffected. (https://github.com/ClickHouse/clickhouse-java/issues/2931)
+- **[client-v2]** Fixed silent data corruption when serializing a Java `null` into a non-nullable
+  `Array(...)` column via `RowBinaryFormatWriter`. `RowBinaryFormatSerializer.writeValuePreamble`
+  special-cased `Array`, emitting a stray marker byte on top of the array length; the server read the
+  extra byte as a phantom extra row (single-column inserts) or as a column shift that failed the insert
+  with `CANNOT_READ_ALL_DATA` (multi-column inserts). A non-nullable `Array` cannot represent a `null`,
+  so it now throws `IllegalArgumentException` naming the column — consistent with every other non-nullable
+  type — in both the `RowBinary` and `RowBinaryWithDefaults` paths. Empty arrays (`[]`) still serialize
+  correctly, and `Dynamic` columns, which can hold a `null` as the implicit `Nothing` type, are
+  unaffected. (https://github.com/ClickHouse/clickhouse-java/issues/2938)
 - **[client-v2]** Fixed POJO insert error classification so transport write failures such as java.net.SocketException:
   Broken pipe (Write failed) are now surfaced as transfer/network errors instead of being wrapped as
   DataSerializationException. This only changes the exception type reported for request-body transport failures during

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

@@ -212,10 +212,9 @@ public static boolean writeValuePreamble(OutputStream out, boolean defaultsSuppo
                     SerializerUtils.writeNonNull(out);
                     SerializerUtils.writeNull(out);//Then we send null, write 1
                     return false;//And we're done
-                } else if (dataType == ClickHouseDataType.Array) {//If the column is an array
-                    SerializerUtils.writeNonNull(out);//Then we send nonNull
                 } else if (dataType == ClickHouseDataType.Dynamic) {
-                    SerializerUtils.writeNonNull(out);
+                    // A Dynamic column can hold a null: it is serialized below as the implicit `Nothing` type.
+                    SerializerUtils.writeNonNull(out);//Write 0 for no default
                 } else {
                     throw new IllegalArgumentException(String.format("An attempt to write null into not nullable column '%s'", column));
                 }
@@ -228,10 +227,8 @@ public static boolean writeValuePreamble(OutputStream out, boolean defaultsSuppo
                 }
                 SerializerUtils.writeNonNull(out);
             } else if (value == null) {
-                if (dataType == ClickHouseDataType.Array) {
-                    SerializerUtils.writeNonNull(out);
-                } else if (dataType == ClickHouseDataType.Dynamic) {
-                    // do nothing
+                if (dataType == ClickHouseDataType.Dynamic) {
+                    // A Dynamic column can hold a null: it is serialized below as the implicit `Nothing` type.
                 } else {
                     throw new IllegalArgumentException(String.format("An attempt to write null into not nullable column '%s'", column));
                 }

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

@@ -301,6 +301,96 @@ public static String tblCreateSQL(String table) {
         }
     }
 
+    @Test(groups = {"integration"}, dataProvider = "nonNullableArrayNullColumns")
+    public void testInsertNullIntoNonNullableArrayThrows(String columns) throws Exception {
+        final String table = "test_non_nullable_array_null";
+        client.execute("DROP TABLE IF EXISTS " + table).get();
+        client.execute(tableDefinition(table, columns)).get();
+
+        client.register(DTOForNonNullableArrayTests.class, client.getTableSchema(table));
+
+        Exception thrown = null;
+        try {
+            client.insert(table, Collections.singletonList(new DTOForNonNullableArrayTests(1, null, 7)))
+                    .get(EXECUTE_CMD_TIMEOUT, TimeUnit.SECONDS);
+        } catch (Exception e) {
+            thrown = e;
+        }
+
+        Assert.assertNotNull(thrown, "Expected the insert to fail for a null in a non-nullable Array column using " + columns);
+        boolean clearMessage = false;
+        for (Throwable t = thrown; t != null; t = t.getCause()) {
+            if (t.getMessage() != null && t.getMessage().contains("An attempt to write null into not nullable column")) {
+                clearMessage = true;
+                break;
+            }
+        }
+        Assert.assertTrue(clearMessage, "Expected a clear non-nullable column error using " + columns + ", but got: " + thrown);
+    }
+
+    @DataProvider(name = "nonNullableArrayNullColumns")
+    public static Object[][] nonNullableArrayNullColumns() {
+        return new Object[][] {
+                {"rowId Int32, arr Array(Int32), tail Int32"},
+                {"rowId Int32, arr Array(Int32), tail Int32 DEFAULT 99"},
+        };
+    }
+
+    @Test(groups = {"integration"}, dataProvider = "nonNullableArrayRoundTrip")
+    public void testInsertNonNullableArrayRoundTrips(List<Integer> arr, int expectedLength, String expectedConcat) throws Exception {
+        final String table = "test_non_nullable_array_round_trip";
+        client.execute("DROP TABLE IF EXISTS " + table).get();
+        client.execute(tableDefinition(table, "rowId Int32", "arr Array(Int32)", "tail Int32")).get();
+
+        client.register(DTOForNonNullableArrayTests.class, client.getTableSchema(table));
+        client.insert(table, Collections.singletonList(new DTOForNonNullableArrayTests(1, arr, 7)))
+                .get(EXECUTE_CMD_TIMEOUT, TimeUnit.SECONDS);
+
+        List<GenericRecord> records = client.queryAll(
+                "SELECT toInt32(length(arr)) AS alen, arrayStringConcat(arr, ',') AS acat, tail FROM " + table + " ORDER BY rowId");
+        Assert.assertEquals(records.size(), 1);
+        GenericRecord row = records.get(0);
+        Assert.assertEquals(row.getInteger("alen"), expectedLength);
+        Assert.assertEquals(row.getString("acat"), expectedConcat);
+        Assert.assertEquals(row.getInteger("tail"), 7);
+    }
+
+    @DataProvider(name = "nonNullableArrayRoundTrip")
+    public static Object[][] nonNullableArrayRoundTrip() {
+        return new Object[][] {
+                {new ArrayList<Integer>(), 0, ""},
+                {Arrays.asList(1, 2, 3), 3, "1,2,3"},
+        };
+    }
+
+    @Test(groups = {"integration"})
+    public void testInsertNullIntoDefaultedNonNullableArrayUsesDefault() throws Exception {
+        final String table = "test_non_nullable_array_null_default";
+        client.execute("DROP TABLE IF EXISTS " + table).get();
+        client.execute(tableDefinition(table, "rowId Int32", "arr Array(Int32) DEFAULT [1, 2]", "tail Int32")).get();
+
+        client.register(DTOForNonNullableArrayTests.class, client.getTableSchema(table));
+        client.insert(table, Collections.singletonList(new DTOForNonNullableArrayTests(1, null, 7)))
+                .get(EXECUTE_CMD_TIMEOUT, TimeUnit.SECONDS);
+
+        List<GenericRecord> records = client.queryAll(
+                "SELECT toInt32(length(arr)) AS alen, arrayStringConcat(arr, ',') AS acat, tail FROM " + table + " ORDER BY rowId");
+        Assert.assertEquals(records.size(), 1);
+        GenericRecord row = records.get(0);
+        Assert.assertEquals(row.getInteger("alen"), 2);
+        Assert.assertEquals(row.getString("acat"), "1,2");
+        Assert.assertEquals(row.getInteger("tail"), 7);
+    }
+
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    public static class DTOForNonNullableArrayTests {
+        private int rowId;
+        private List<Integer> arr;
+        private int tail;
+    }
+
     @Data
     @AllArgsConstructor
     @NoArgsConstructor

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

@@ -348,6 +348,111 @@
         return map;
     }
 
+    @DataProvider(name = "rowBinaryWriterFormats")
+    private Object[][] rowBinaryWriterFormats() {
+        return new Object[][] {
+                {ClickHouseFormat.RowBinary},
+                {ClickHouseFormat.RowBinaryWithDefaults},
+        };
+    }
+
+    @Test (groups = { "integration" }, dataProvider = "rowBinaryWriterFormats")
+    public void writeNullIntoNonNullableArrayThrowsTest(ClickHouseFormat format) throws Exception {
+        String tableName = "rowBinaryFormatWriterTest_nonNullableArrayNull_" + UUID.randomUUID().toString().replace('-', '_');
+        initTable(tableName,
+                "CREATE TABLE \"" + tableName + "\" (id Int32, arr Array(Int32), tail Int32) Engine = MergeTree ORDER BY id",
+                new CommandSettings());
+        TableSchema schema = client.getTableSchema(tableName);
+
+        Exception thrown = null;
+        try (InsertResponse response = client.insert(tableName, out -> {
+            RowBinaryFormatWriter w = new RowBinaryFormatWriter(out, schema, format);
+            w.setValue(schema.nameToColumnIndex("id"), 1);
+            w.setValue(schema.nameToColumnIndex("arr"), null);
+            w.setValue(schema.nameToColumnIndex("tail"), 7);
+            w.commitRow();
+        }, format, settings).get(EXECUTE_CMD_TIMEOUT, TimeUnit.SECONDS)) {
+        } catch (Exception e) {
+            thrown = e;
+        }
+
+        Assert.assertNotNull(thrown, "Expected the insert to fail for a null in a non-nullable Array column using " + format);
+        boolean clearMessage = false;
+        for (Throwable t = thrown; t != null; t = t.getCause()) {
+            if (t.getMessage() != null && t.getMessage().contains("An attempt to write null into not nullable column")) {
+                clearMessage = true;
+                break;
+            }
+        }
+        Assert.assertTrue(clearMessage, "Expected a clear non-nullable column error using " + format + ", but got: " + thrown);
+    }
+
+    @Test (groups = { "integration" }, dataProvider = "rowBinaryWriterFormats")
+    public void writeNonNullableArrayRoundTripsTest(ClickHouseFormat format) throws Exception {
+        String tableName = "rowBinaryFormatWriterTest_nonNullableArrayRoundTrip_" + UUID.randomUUID().toString().replace('-', '_');
+        initTable(tableName,
+                "CREATE TABLE \"" + tableName + "\" (id Int32, arr Array(Int32), tail Int32) Engine = MergeTree ORDER BY id",
+                new CommandSettings());
+        TableSchema schema = client.getTableSchema(tableName);
+
+        try (InsertResponse response = client.insert(tableName, out -> {
+            RowBinaryFormatWriter w = new RowBinaryFormatWriter(out, schema, format);
+            w.setValue(schema.nameToColumnIndex("id"), 1);
+            w.setValue(schema.nameToColumnIndex("arr"), new ArrayList<Integer>());
+            w.setValue(schema.nameToColumnIndex("tail"), 7);
+            w.commitRow();
+            w.setValue(schema.nameToColumnIndex("id"), 2);
+            w.setValue(schema.nameToColumnIndex("arr"), Arrays.asList(1, 2, 3));
+            w.setValue(schema.nameToColumnIndex("tail"), 8);
+            w.commitRow();
+        }, format, settings).get(EXECUTE_CMD_TIMEOUT, TimeUnit.SECONDS)) {
+            // inserted
+        }
+
+        List<GenericRecord> records = client.queryAll(
+                "SELECT id, toInt32(length(arr)) AS alen, arrayStringConcat(arr, ',') AS acat, tail FROM \""
+                        + tableName + "\" ORDER BY id");
+        Assert.assertEquals(records.size(), 2);
+
+        GenericRecord row1 = records.get(0);
+        Assert.assertEquals(row1.getInteger("alen"), 0);
+        Assert.assertEquals(row1.getString("acat"), "");
+        Assert.assertEquals(row1.getInteger("tail"), 7);
+
+        GenericRecord row2 = records.get(1);
+        Assert.assertEquals(row2.getInteger("alen"), 3);
+        Assert.assertEquals(row2.getString("acat"), "1,2,3");
+        Assert.assertEquals(row2.getInteger("tail"), 8);
+    }
+
+    @Test (groups = { "integration" })
+    public void writeNullIntoDefaultedArrayUsesDefaultTest() throws Exception {
+        String tableName = "rowBinaryFormatWriterTest_defaultedArrayNull_" + UUID.randomUUID().toString().replace('-', '_');
+        initTable(tableName,
+                "CREATE TABLE \"" + tableName + "\" (id Int32, arr Array(Int32) DEFAULT [1, 2], tail Int32) Engine = MergeTree ORDER BY id",
+                new CommandSettings());
+        TableSchema schema = client.getTableSchema(tableName);
+        ClickHouseFormat format = ClickHouseFormat.RowBinaryWithDefaults;
+
+        try (InsertResponse response = client.insert(tableName, out -> {
+            RowBinaryFormatWriter w = new RowBinaryFormatWriter(out, schema, format);
+            w.setValue(schema.nameToColumnIndex("id"), 1);
+            w.setValue(schema.nameToColumnIndex("arr"), null);
+            w.setValue(schema.nameToColumnIndex("tail"), 7);
+            w.commitRow();
+        }, format, settings).get(EXECUTE_CMD_TIMEOUT, TimeUnit.SECONDS)) {
+        }
+
+        List<GenericRecord> records = client.queryAll(
+                "SELECT toInt32(length(arr)) AS alen, arrayStringConcat(arr, ',') AS acat, tail FROM \""
+                        + tableName + "\" ORDER BY id");
+        Assert.assertEquals(records.size(), 1);
+        GenericRecord row = records.get(0);
+        Assert.assertEquals(row.getInteger("alen"), 2);
+        Assert.assertEquals(row.getString("acat"), "1,2");
+        Assert.assertEquals(row.getInteger("tail"), 7);
+    }
+
 
     @Test (groups = { "integration" })
     public void writeNumbersTest() throws Exception {

jdbc-v2/src/test/java/com/clickhouse/jdbc/WriterStatementImplTest.java

@@ -14,7 +14,12 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.sql.Types;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Properties;
+import java.util.UUID;
 
 @Test(groups = {"integration"})
 public class WriterStatementImplTest extends JdbcIntegrationTest {
@@ -36,6 +41,107 @@ public void testTargetTypeMethodThrowException() throws SQLException {
         }
     }
 
+    @DataProvider(name = "nonNullableArrayNullTables")
+    public static Object[][] nonNullableArrayNullTables() {
+        return new Object[][] {
+                {"id Int32, arr Array(Int32), tail Int32"},
+                {"id Int32, arr Array(Int32), tail Int32 DEFAULT 99"},
+        };
+    }
+
+    @Test(groups = {"integration"}, dataProvider = "nonNullableArrayNullTables")
+    public void testWriterNullIntoNonNullableArrayThrows(String columns) throws Exception {
+        String table = "writer_stmt_arr_null_" + UUID.randomUUID().toString().replace('-', '_');
+        runQuery("DROP TABLE IF EXISTS " + table);
+        runQuery("CREATE TABLE " + table + " (" + columns + ") Engine = MergeTree ORDER BY id");
+
+        Properties properties = new Properties();
+        properties.setProperty(DriverProperties.BETA_ROW_BINARY_WRITER.getKey(), "true");
+        try (Connection connection = getJdbcConnection(properties);
+             PreparedStatement stmt = connection.prepareStatement("INSERT INTO " + table + " VALUES (?, ?, ?)")) {
+            Assert.assertTrue(stmt instanceof WriterStatementImpl);
+            stmt.setInt(1, 1);
+            stmt.setNull(2, Types.ARRAY);
+            stmt.setInt(3, 7);
+
+            SQLException thrown = Assert.expectThrows(SQLException.class, stmt::execute);
+            Assert.assertTrue(hasNonNullableColumnMessage(thrown),
+                    "Expected a clear non-nullable column error using " + columns + ", but got: " + thrown);
+        }
+    }
+
+    @DataProvider(name = "nonNullableArrayRoundTrip")
+    public static Object[][] nonNullableArrayRoundTrip() {
+        return new Object[][] {
+                {new ArrayList<Integer>(), 0, ""},
+                {Arrays.asList(1, 2, 3), 3, "1,2,3"},
+        };
+    }
+
+    @Test(groups = {"integration"}, dataProvider = "nonNullableArrayRoundTrip")
+    public void testWriterNonNullableArrayRoundTrips(List<Integer> arr, int expectedLength, String expectedConcat) throws Exception {
+        String table = "writer_stmt_arr_round_trip_" + UUID.randomUUID().toString().replace('-', '_');
+        runQuery("DROP TABLE IF EXISTS " + table);
+        runQuery("CREATE TABLE " + table + " (id Int32, arr Array(Int32), tail Int32) Engine = MergeTree ORDER BY id");
+
+        Properties properties = new Properties();
+        properties.setProperty(DriverProperties.BETA_ROW_BINARY_WRITER.getKey(), "true");
+        try (Connection connection = getJdbcConnection(properties);
+             PreparedStatement stmt = connection.prepareStatement("INSERT INTO " + table + " VALUES (?, ?, ?)")) {
+            stmt.setInt(1, 1);
+            stmt.setObject(2, arr);
+            stmt.setInt(3, 7);
+            stmt.execute();
+        }
+
+        try (Connection connection = getJdbcConnection();
+             Statement stmt = connection.createStatement();
+             ResultSet rs = stmt.executeQuery("SELECT toInt32(length(arr)) AS alen, arrayStringConcat(arr, ',') AS acat, tail FROM "
+                     + table + " ORDER BY id")) {
+            Assert.assertTrue(rs.next());
+            Assert.assertEquals(rs.getInt("alen"), expectedLength);
+            Assert.assertEquals(rs.getString("acat"), expectedConcat);
+            Assert.assertEquals(rs.getInt("tail"), 7);
+            Assert.assertFalse(rs.next());
+        }
+    }
+
+    @Test(groups = {"integration"})
+    public void testWriterNullIntoDefaultedArrayUsesDefault() throws Exception {
+        String table = "writer_stmt_arr_null_default_" + UUID.randomUUID().toString().replace('-', '_');
+        runQuery("DROP TABLE IF EXISTS " + table);
+        runQuery("CREATE TABLE " + table + " (id Int32, arr Array(Int32) DEFAULT [1, 2], tail Int32) Engine = MergeTree ORDER BY id");
+
+        Properties properties = new Properties();
+        properties.setProperty(DriverProperties.BETA_ROW_BINARY_WRITER.getKey(), "true");
+        try (Connection connection = getJdbcConnection(properties);
+             PreparedStatement stmt = connection.prepareStatement("INSERT INTO " + table + " VALUES (?, ?, ?)")) {
+            stmt.setInt(1, 1);
+            stmt.setNull(2, Types.ARRAY);
+            stmt.setInt(3, 7);
+            stmt.execute();
+        }
+
+        try (Connection connection = getJdbcConnection();
+             Statement stmt = connection.createStatement();
+             ResultSet rs = stmt.executeQuery("SELECT toInt32(length(arr)) AS alen, arrayStringConcat(arr, ',') AS acat, tail FROM "
+                     + table + " ORDER BY id")) {
+            Assert.assertTrue(rs.next());
+            Assert.assertEquals(rs.getInt("alen"), 2);
+            Assert.assertEquals(rs.getString("acat"), "1,2");
+            Assert.assertEquals(rs.getInt("tail"), 7);
+        }
+    }
+
+    private static boolean hasNonNullableColumnMessage(Throwable thrown) {
+        for (Throwable t = thrown; t != null; t = t.getCause()) {
+            if (t.getMessage() != null && t.getMessage().contains("An attempt to write null into not nullable column")) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     @DataProvider(name = "insertColumnListForms")
     Object[][] insertColumnListForms() {
         return new Object[][]{