Fix client-v2: read SimpleAggregateFunction from a Dynamic column

CHANGELOG.md

@@ -67,6 +67,12 @@
 
 ### Bug Fixes 
 
+- **[client-v2]** Fixed reading a `SimpleAggregateFunction(func, T)` value held in a `Dynamic` column. The binary type
+  encoding of such a value (`0x2E <function_name> <parameters> <arguments> <argument_type_encodings>`) was not consumed
+  at all, so the read failed with `IndexOutOfBoundsException`, and the unconsumed encoding bytes would otherwise have
+  been interpreted as row data and desynchronized the rest of the `RowBinary` stream. The concrete type is now
+  reconstructed from the encoding and the value is read as its argument type `T`, so it reads exactly like the same
+  value in a plain `SimpleAggregateFunction` column. (https://github.com/ClickHouse/clickhouse-java/issues/3005)
 - **[jdbc-v2]** Fixed an `INSERT` whose values list holds a function call the bundled `ANTLR4` grammar cannot match -
   such as `hex(x'AB')`, valid ClickHouse the grammar has no hex string literal for - being reported to hold no function
   call when an `ANTLR4` parser backend is selected (`jdbc_sql_parser=ANTLR4` / `ANTLR4_PARAMS_PARSER`). Function calls in

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

@@ -263,7 +263,7 @@ private <T> T readValue(ClickHouseColumn column, Class<?> typeHint, boolean stri
                 case Nothing:
                     return null;
                 case SimpleAggregateFunction:
-                    return (T) readValue(column.getNestedColumns().get(0), typeHint, false);
+                    return (T) readValue(actualColumn.getNestedColumns().get(0), typeHint, false);
                 case AggregateFunction:
                     return (T) readBitmap( actualColumn);
                 case Variant:
@@ -1562,6 +1562,29 @@ private ClickHouseColumn readDynamicData() throws IOException {
                 int dimension = readVarInt(input);
                 return ClickHouseColumn.of("v", "QBit(" + elementColumn.getOriginalTypeName() + ", " + dimension + ")");
             }
+            case SimpleAggregateFunction: {
+                // 0x2E <function_name> <var_uint number_of_parameters><parameters>
+                //      <var_uint number_of_arguments><argument_type_encodings>
+                // The whole encoding MUST be consumed so a SimpleAggregateFunction nested in a
+                // Dynamic/Variant/JSON column does not desynchronize the stream.
+                String functionName = readString(input);
+                int numberOfParameters = readVarInt(input);
+                if (numberOfParameters > 0) {
+                    // Every function accepted by SimpleAggregateFunction is parameterless, so the
+                    // binary encoding of a parameter (a Field) never appears here. Fail loudly
+                    // instead of silently leaving the parameters in the stream.
+                    throw new ClientException("Parameterized SimpleAggregateFunction is not supported: "
+                            + functionName);
+                }
+                int numberOfArguments = readVarInt(input);
+                StringBuilder typeName = new StringBuilder(SB_INIT_SIZE);
+                typeName.append("SimpleAggregateFunction(").append(functionName);
+                for (int i = 0; i < numberOfArguments; i++) {
+                    typeName.append(", ").append(readDynamicData().getOriginalTypeName());
+                }
+                typeName.append(')');
+                return ClickHouseColumn.of("v", typeName.toString());
+            }
             case Time64: {
                 byte precision = readByte();
                 return ClickHouseColumn.of("v", "Time64(" + precision + ")");

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

@@ -1,11 +1,14 @@
 package com.clickhouse.client.api.data_formats.internal;
 
+import com.clickhouse.client.api.ClientException;
 import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseDataType;
 import com.clickhouse.data.format.BinaryStreamUtils;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.math.BigInteger;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.time.temporal.ChronoUnit;
@@ -192,6 +195,49 @@ public void testArrayValue() throws Exception {
         Assert.assertEquals(array1.length, array2.length);
     }
 
+    @Test
+    public void testDynamicSimpleAggregateFunctionConsumesWholeTypeEncoding() throws Exception {
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        baos.write(ClickHouseDataType.SimpleAggregateFunction.getBinTag());
+        BinaryStreamUtils.writeString(baos, "sum");
+        BinaryStreamUtils.writeVarInt(baos, 0);
+        BinaryStreamUtils.writeVarInt(baos, 1);
+        baos.write(ClickHouseDataType.UInt64.getBinTag());
+        BinaryStreamUtils.writeUnsignedInt64(baos, 42);
+        BinaryStreamUtils.writeInt32(baos, 4242);
+
+        BinaryStreamReader reader = new BinaryStreamReader(
+                new ByteArrayInputStream(baos.toByteArray()),
+                TimeZone.getTimeZone("UTC"),
+                null,
+                new BinaryStreamReader.CachingByteBufferAllocator(),
+                false,
+                null,
+                false);
+
+        Assert.assertEquals(reader.readValue(ClickHouseColumn.of("v", "Dynamic")), BigInteger.valueOf(42));
+        Assert.assertEquals(reader.readValue(ClickHouseColumn.of("guard", "Int32")), Integer.valueOf(4242));
+    }
+
+    @Test(expectedExceptions = ClientException.class)
+    public void testDynamicParameterizedSimpleAggregateFunctionRejected() throws Exception {
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        baos.write(ClickHouseDataType.SimpleAggregateFunction.getBinTag());
+        BinaryStreamUtils.writeString(baos, "sum");
+        BinaryStreamUtils.writeVarInt(baos, 1);
+
+        BinaryStreamReader reader = new BinaryStreamReader(
+                new ByteArrayInputStream(baos.toByteArray()),
+                TimeZone.getTimeZone("UTC"),
+                null,
+                new BinaryStreamReader.CachingByteBufferAllocator(),
+                false,
+                null,
+                false);
+
+        reader.readValue(ClickHouseColumn.of("v", "Dynamic"));
+    }
+
     @Test
     public void testReadNullVariantReturnsNull() throws Exception {
         ClickHouseColumn column = ClickHouseColumn.of("v", "Variant(Int32, String)");

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

@@ -253,6 +253,43 @@ public void testQBitInDynamicColumn() throws Exception {
         Assert.assertEquals(rows.get(0).getInteger("tail"), 42);
     }
 
+    @DataProvider(name = "simpleAggregateFunctionInDynamicColumn")
+    public static Object[][] simpleAggregateFunctionInDynamicColumn() {
+        return new Object[][]{
+                {"sum", "UInt64", "42", "42"},
+                {"max", "Int32", "-7", "-7"},
+                {"anyLast", "String", "'abc'", "abc"},
+                {"anyLast", "LowCardinality(String)", "'lc'", "lc"},
+                {"anyLast", "DateTime(\\'UTC\\')", "toDateTime(1700000000)", "2023-11-14T22:13:20Z[UTC]"},
+                {"groupArrayArray", "Array(String)", "['a', 'b']", "[a, b]"},
+                {"anyLast", "Map(String, UInt8)", "map('k', 1)", "{k=1}"},
+        };
+    }
+
+    @Test(groups = {"integration"}, dataProvider = "simpleAggregateFunctionInDynamicColumn")
+    public void testSimpleAggregateFunctionInDynamicColumn(String function, String argType, String valueSQL,
+                                                           String expected) throws Exception {
+        if (isVersionMatch("(,24.8]")) {
+            throw new SkipException("Dynamic requires ClickHouse 24.8+");
+        }
+
+        // A SimpleAggregateFunction held in a Dynamic column encodes its concrete type on the wire as
+        // 0x2E <function_name> <var_uint number_of_parameters> <var_uint number_of_arguments>
+        // <argument_type_encodings>. All of it must be consumed and the value must then be read as its
+        // argument type, otherwise the following column ("tail") misaligns. The trailing 42 is the
+        // desync guard; "plain" is the same value in a Dynamic column without the wrapper.
+        List<GenericRecord> rows = client.queryAll(
+                "SELECT CAST(CAST(" + valueSQL + ", 'SimpleAggregateFunction(" + function + ", " + argType + ")')"
+                        + " AS Dynamic) AS d, CAST(CAST(" + valueSQL + ", '" + argType + "') AS Dynamic) AS plain,"
+                        + " 42 AS tail SETTINGS allow_experimental_dynamic_type = 1");
+        Assert.assertEquals(rows.size(), 1);
+        GenericRecord row = rows.get(0);
+        Assert.assertEquals(row.getString("d"), expected);
+        Assert.assertEquals(row.getString("d"), row.getString("plain"));
+        Assert.assertEquals(row.getObject("d").getClass(), row.getObject("plain").getClass());
+        Assert.assertEquals(row.getInteger("tail"), 42);
+    }
+
     @Test(groups = {"integration"})
     public void testNestedDataTypes() throws Exception {
         final String table = "test_nested_types";