Check the aggregates of every state the catalog states on the typed surface

GENERATION.md

@@ -63,6 +63,17 @@ length of the form and its bytes, so a value of any size crosses. A SQL array ar
 signature whose types the surface cannot carry (`Datum`, aggregate state, an array element with
 no codec) is skipped and counted in the generator's report.
 
+An aggregate of the catalog's `aggregates` section is one Flink `AggregateFunction` per SQL name,
+each overload an `accumulate` method, run from the public MEOS function of each role the
+aggregate states: the transition folds a value into the state, the combine joins two partial
+states, and the final answers the result. The accumulator, `MeosAggState`, holds the bytes of the
+state, those the serialize function writes and the deserialize function reads, or the form the
+codec of the state's type writes (the box of `extent`), never a native pointer. An aggregate
+taking one value with a combine is generated; the others (`appendInstant`, the `w*` window
+aggregates) are counted in the generator's report. A name a scalar function carries goes to the
+scalar, the aggregate taking the `Agg` name the catalog states (`mergeAgg`, `tMinAgg`) or the
+suffix (`tAndAgg`, `setUnionAgg`).
+
 `org.mobilitydb.flink.sql.MobilityFlinkSql.registerAll(tEnv)` registers every function as a
 catalog function under its MobilityDB SQL name. A Flink built-in of the same name (`lower`,
 `round`, `abs`, …) resolves before a catalog function, and a name Flink's parser reserves

binding/src/test/java/org/mobilitydb/flink/sql/GeneratedSqlSurfaceTest.java

@@ -38,6 +38,7 @@
 import org.apache.flink.util.CloseableIterator;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.mobilitydb.flink.sql.functions.TAvg;
 import org.mobilitydb.flink.sql.types.TFloat;
 import org.mobilitydb.flink.sql.types.TGeomPoint;
 import org.mobilitydb.flink.sql.types.TInt;
@@ -233,6 +234,81 @@ void valuesGroupAcrossAShuffle() throws Exception {
         assertEquals(2L, last.getField(1));
     }
 
+    /**
+     * The text of an aggregate over the values, the last row a grouped query emits in env, read
+     * as #rows reads the rows of tEnv; the twin of MobilitySpark's GeneratedSqlSurfaceTest
+     * aggregate.
+     */
+    private static String aggregate(TableEnvironment env, String agg, String in, String... values)
+            throws Exception {
+        StringBuilder rows = new StringBuilder();
+        for (String v : values) {
+            rows.append(rows.length() == 0 ? "" : ", ").append("('").append(v).append("')");
+        }
+        List<Row> out = new ArrayList<>();
+        try (CloseableIterator<Row> it = env.executeSql("SELECT asText(" + agg + "(v)) FROM (SELECT "
+                + in + "(s) AS v FROM (VALUES " + rows + ") AS t(s))").collect()) {
+            it.forEachRemaining(out::add);
+        }
+        return (String) out.get(out.size() - 1).getField(0);
+    }
+
+    @Test
+    void aggregatesAnswerWhatMobilityDBAnswers() throws Exception {
+        // What MobilityDB answers for the same aggregate over the same values with the time zone
+        // set to UTC, in a streaming job folding each value into the bytes of the state, and in
+        // a batch job of two phases, whose global phase combines the partial states the local
+        // one writes: the skip lists of tCount, tAvg and tCentroid as taggstate_serialize writes
+        // them, the set and span set of the unions as setstate_serialize and
+        // spansetstate_serialize write them, and the box of extent as its WKB.
+        TableEnvironment batch = TableEnvironment.create(EnvironmentSettings.inBatchMode());
+        batch.getConfig().set("table.optimizer.agg-phase-strategy", "TWO_PHASE");
+        MobilityFlinkSql.registerAll(batch);
+        for (TableEnvironment env : new TableEnvironment[] {tEnv, batch}) {
+            assertEquals("{[1@2001-01-01 00:00:00+00, 2@2001-01-02 00:00:00+00, "
+                + "3@2001-01-03 00:00:00+00], (2@2001-01-03 00:00:00+00, 2@2001-01-04 00:00:00+00], "
+                + "(1@2001-01-04 00:00:00+00, 1@2001-01-05 00:00:00+00]}",
+                aggregate(env, "tCount", "tintFromText",
+                          "[1@2001-01-01 00:00:00+00, 2@2001-01-03 00:00:00+00]",
+                          "[3@2001-01-02 00:00:00+00, 4@2001-01-04 00:00:00+00]",
+                          "[5@2001-01-03 00:00:00+00, 5@2001-01-05 00:00:00+00]"));
+            assertEquals("{[1@2001-01-01 00:00:00+00, 2@2001-01-02 00:00:00+00), "
+                + "[2.5@2001-01-02 00:00:00+00, 3.5@2001-01-03 00:00:00+00], "
+                + "(4@2001-01-03 00:00:00+00, 5@2001-01-04 00:00:00+00]}",
+                aggregate(env, "tAvg", "tfloatFromText",
+                          "[1@2001-01-01 00:00:00+00, 3@2001-01-03 00:00:00+00]",
+                          "[3@2001-01-02 00:00:00+00, 5@2001-01-04 00:00:00+00]"));
+            assertEquals("{[POINT(1 0)@2001-01-01 00:00:00+00, POINT(3 2)@2001-01-03 00:00:00+00]}",
+                aggregate(env, "tCentroid", "tgeompointFromText",
+                          "[Point(0 0)@2001-01-01 00:00:00+00, Point(2 2)@2001-01-03 00:00:00+00]",
+                          "[Point(2 0)@2001-01-01 00:00:00+00, Point(4 2)@2001-01-03 00:00:00+00]"));
+            assertEquals("{1, 2, 3, 5}",
+                aggregate(env, "setUnionAgg", "intsetFromText", "{1, 3}", "{2, 3, 5}"));
+            assertEquals("{[1, 5), [7, 8)}",
+                aggregate(env, "spanUnionAgg", "intspanFromText", "[1, 3)", "[2, 5)", "[7, 8)"));
+            assertEquals("STBOX X((0,1),(2,4))",
+                aggregate(env, "extent", "stboxFromText", "STBOX X((1,1),(2,2))", "STBOX X((0,3),(1,4))"));
+        }
+        // The batch job's one local partition hands its global phase one partial state, so two
+        // partial states are joined here as a global phase joins those of two partitions, the
+        // second read back from the bytes its serializer writes for the exchange.
+        TAvg avg = new TAvg();
+        MeosAggState a = avg.createAccumulator();
+        MeosAggState b = avg.createAccumulator();
+        avg.accumulate(a, TFloat.encode(GeneratedFunctions.tfloat_in(
+                "[1@2001-01-01 00:00:00+00, 3@2001-01-03 00:00:00+00]")));
+        avg.accumulate(b, TFloat.encode(GeneratedFunctions.tfloat_in(
+                "[3@2001-01-02 00:00:00+00, 5@2001-01-04 00:00:00+00]")));
+        MeosAggState.Serializer ser = new MeosAggState.Serializer();
+        DataOutputSerializer out = new DataOutputSerializer(64);
+        ser.serialize(b, out);
+        avg.merge(a, List.of(ser.deserialize(new DataInputDeserializer(out.getCopyOfBuffer()))));
+        assertEquals("{[1@2001-01-01 00:00:00+00, 2@2001-01-02 00:00:00+00), "
+            + "[2.5@2001-01-02 00:00:00+00, 3.5@2001-01-03 00:00:00+00], "
+            + "(4@2001-01-03 00:00:00+00, 5@2001-01-04 00:00:00+00]}",
+            GeneratedFunctions.tfloat_out(((TFloat) avg.getValue(a)).decode(), 15));
+    }
+
     @Test
     void aValueOfAnySizeCrossesTheSerializer() throws Exception {
         // 20,000 instants, a trajectory of an ordinary day: its WKB is over 64 KB, the most