diff --git a/GENERATION.md b/GENERATION.md index 64077f1..e4ea905 100644 --- a/GENERATION.md +++ b/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 diff --git a/binding/src/test/java/org/mobilitydb/flink/sql/GeneratedSqlSurfaceTest.java b/binding/src/test/java/org/mobilitydb/flink/sql/GeneratedSqlSurfaceTest.java index 2ba45f9..8137944 100644 --- a/binding/src/test/java/org/mobilitydb/flink/sql/GeneratedSqlSurfaceTest.java +++ b/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 out = new ArrayList<>(); + try (CloseableIterator 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