Fix client-v2: serialize SimpleAggregateFunction columns in RowBinary writer

SerializerUtils.serializeData had no case for SimpleAggregateFunction, so
inserting into such a column threw UnsupportedOperationException ("Unsupported
data type: SimpleAggregateFunction") even though BinaryStreamReader already
reads these columns. Delegate to the underlying type via serializeNestedData
so a Nullable underlying still gets its RowBinary null-marker byte, mirroring
the read path.

Related to: #2477

Author: polyglotAI-bot

CHANGELOG.md

@@ -4,6 +4,12 @@
 
 ### 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,12 @@ 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; delegate through serializeNestedData so a Nullable underlying still gets
+                // its RowBinary null-marker byte, mirroring BinaryStreamReader's read path.
+                serializeNestedData(stream, value, column.getNestedColumns().get(0));
+                break;
             case AggregateFunction:
                 serializeAggregateFunction(stream, value, column);
                 break;

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

@@ -217,6 +217,47 @@ private Object[][] nestedNullableData() throws Exception {
         };
     }
 
+    @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),
                 }
         };