Fix client-v2: serialize SimpleAggregateFunction columns in RowBinary writer

CHANGELOG.md

@@ -4,6 +4,13 @@
 
 ### Bug Fixes 
 
+- **[client-v2]** Fixed the `RowBinary` writer throwing
+  `UnsupportedOperationException: Unsupported data type: SimpleAggregateFunction` when inserting into a
+  `SimpleAggregateFunction(func, T)` column (the reader already supported these columns). The value is now
+  serialized identically to its underlying type `T`, writing the `Nullable` null-marker byte when the
+  underlying type is nullable (e.g. `SimpleAggregateFunction(anyLast, Nullable(String))`), mirroring the
+  read path. (https://github.com/ClickHouse/clickhouse-java/issues/2477)
+
 - **[client-v2]** Fixed binary array decoding for nullable element types so `Array(Nullable(Float64))` and similar columns now return boxed arrays such as `Double[]` instead of `Object[]`. This keeps null-supporting arrays aligned with their element type while preserving the existing `Object[]` fallback for Variant/Dynamic/Geometry arrays. (https://github.com/ClickHouse/clickhouse-java/issues/2846)
 
 - **[client-v2]** Fixed `Float32`/`Float64` columns throwing `ClassCastException` when a value of a

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

@@ -72,6 +72,15 @@ public static void serializeData(OutputStream stream, Object value, ClickHouseCo
             case Map:
                 serializeMapData(stream, value, column);
                 break;
+            case SimpleAggregateFunction:
+                // A SimpleAggregateFunction(func, T) value serializes identically to its underlying
+                // type T. This is a deliberate exception to serializeNestedData's "nested elements
+                // only" contract: the SAF wrapper is itself non-nullable, so writeValuePreamble emits
+                // no null-marker for a top-level SAF column, yet a Nullable underlying still needs one.
+                // serializeNestedData writes that marker iff the underlying is nullable, mirroring
+                // BinaryStreamReader's read path.
+                serializeNestedData(stream, value, column.getNestedColumns().get(0));
+                break;
             case AggregateFunction:
                 serializeAggregateFunction(stream, value, column);
                 break;
@@ -117,7 +126,10 @@ public static void serializeData(OutputStream stream, Object value, ClickHouseCo
      * {@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.
+     * so this helper is normally used only for nested elements. The one deliberate exception is a
+     * top-level {@code SimpleAggregateFunction} column: its wrapper is itself non-nullable (so
+     * {@code writeValuePreamble} writes no marker), but its underlying type may be {@code Nullable}
+     * and still needs the marker, which this helper supplies.
      */
     private static void serializeNestedData(OutputStream stream, Object value, ClickHouseColumn column) throws IOException {
         if (column.isNullable()) {

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

@@ -217,6 +217,47 @@
         };
     }
 
+    @Test(dataProvider = "simpleAggregateFunctionData")
+    public void testSimpleAggregateFunctionRoundTrip(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 = "simpleAggregateFunctionData")
+    private Object[][] simpleAggregateFunctionData() {
+        return new Object[][] {
+                // Top-level SAF columns - the exact shape reported in the bug, reached directly
+                // through the serializeData switch's SimpleAggregateFunction case.
+                {"SimpleAggregateFunction(sum, UInt64)", BigInteger.valueOf(42)},
+                {"SimpleAggregateFunction(anyLast, Nullable(String))", "present"},
+
+                // A SimpleAggregateFunction(func, T) value serializes byte-identically to its
+                // underlying type T. Each SAF below sits in the MIDDLE of the schema between a
+                // leading Int32 and a trailing Float64, so a dropped or extra byte (such as a
+                // wrongly written null-marker) shifts the trailing Float64 and is detected
+                // positionally. The assertion compares the whole row.
+
+                // Non-nullable fixed-width underlying: no null-marker byte precedes the value.
+                {"Tuple(Int32, SimpleAggregateFunction(sum, UInt64), Float64)",
+                        Arrays.asList(7, BigInteger.valueOf(42), 9.5d)},
+                // Non-nullable variable-length underlying: still no marker. This is the contrast
+                // case - it would misalign if the SAF branch unconditionally wrote a marker.
+                {"Tuple(Int32, SimpleAggregateFunction(anyLast, String), Float64)",
+                        Arrays.asList(7, "kept", 9.5d)},
+                // Nullable underlying, value present: a single present-marker (0x00) precedes it.
+                {"Tuple(Int32, SimpleAggregateFunction(anyLast, Nullable(String)), Float64)",
+                        Arrays.asList(7, "opt", 9.5d)},
+                // Nullable underlying, value null: a single null-marker (0x01) and no value.
+                {"Tuple(Int32, SimpleAggregateFunction(anyLast, Nullable(String)), Float64)",
+                        Arrays.asList(7, null, 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.

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

@@ -611,19 +611,23 @@ public void writeAggregateFunctionTests() throws Exception {
     }
 
 
-    //TODO: Do we support this?
-    @Test (groups = { "integration" }, enabled = false)
+    @Test (groups = { "integration" })
     public void writeSimpleAggregateFunctionTests() throws Exception {
         String tableName = "rowBinaryFormatWriterTest_writeSimpleAggregateFunctionTests_" + UUID.randomUUID().toString().replace('-', '_');
         String tableCreate = "CREATE TABLE \"" + tableName + "\" " +
                 " (id Int32, " +
-                "  simple_aggregate_function SimpleAggregateFunction(count, Int8), " +
-                "  ) Engine = MergeTree ORDER BY id";
+                "  saf_sum SimpleAggregateFunction(sum, UInt64), " +
+                "  saf_str SimpleAggregateFunction(anyLast, Nullable(String)), " +
+                "  tail Int32 " +
+                "  ) Engine = AggregatingMergeTree ORDER BY id";
 
-        // Insert random (valid) values
+        // The SimpleAggregateFunction columns sit between id and a trailing Int32 so a byte
+        // dropped/added while writing them misaligns "tail" and is detected.
         Field[][] rows = new Field[][] {{
-                    new Field("id", 1), //Row ID
-                    new Field("simple_aggregate_function", Arrays.asList((byte) 1)).set(Arrays.asList((byte) 1)), //SimpleAggregateFunction
+                    new Field("id", 1),
+                    new Field("saf_sum", BigInteger.valueOf(42)),
+                    new Field("saf_str", "hello"),
+                    new Field("tail", 7),
                 }
         };