From 2eaaa077d19221d2e8f0ea08f034ffbc65fa6d0d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Esteban=20Zim=C3=A1nyi?= Date: Sat, 10 Oct 2026 14:24:31 +0200 Subject: [PATCH] Check the window aggregates on the typed Flink surface GeneratedSqlSurfaceTest.windowAggregatesAnswerWhatMobilityDBAnswers checks that wCount, wMax and wAvg over two temporal integers and wSum over a discrete temporal float, with a window of one day, answer what MobilityDB answers, in a streaming job and in a batch job of two phases; aggregateCall reads the call of an aggregate of several arguments as aggregate reads one of one. Witness: with the generator of JMEOS ef197a02 the surface registers no window aggregate, the 30 overloads of wCount, wMin, wMax, wSum and wAvg left out as taking two arguments. Measured over MobilityDB 1f41894b35, the catalog of MEOS-API ff5f838173 and the generator of JMEOS #169 (26044b19): the 37 tests pass, 15 of them in GeneratedSqlSurfaceTest. Why: a window aggregate answers on Flink as it does in PostgreSQL. --- .../flink/sql/GeneratedSqlSurfaceTest.java | 35 ++++++++++++++++++- 1 file changed, 34 insertions(+), 1 deletion(-) 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 8137944..6f0f332 100644 --- a/binding/src/test/java/org/mobilitydb/flink/sql/GeneratedSqlSurfaceTest.java +++ b/binding/src/test/java/org/mobilitydb/flink/sql/GeneratedSqlSurfaceTest.java @@ -241,18 +241,51 @@ void valuesGroupAcrossAShuffle() throws Exception { */ private static String aggregate(TableEnvironment env, String agg, String in, String... values) throws Exception { + return aggregateCall(env, agg + "(v)", in, values); + } + + /** The text of the aggregate call over the values v, as #aggregate reads one of one argument. */ + private static String aggregateCall(TableEnvironment env, String call, 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 " + try (CloseableIterator it = env.executeSql("SELECT asText(" + call + ") 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 windowAggregatesAnswerWhatMobilityDBAnswers() throws Exception { + // What MobilityDB answers for the same window aggregate over the same values and a window + // of one day with the time zone set to UTC, in a streaming job and in a batch job of two + // phases, the twin of MobilitySpark's test. + TableEnvironment batch = TableEnvironment.create(EnvironmentSettings.inBatchMode()); + batch.getConfig().set("table.optimizer.agg-phase-strategy", "TWO_PHASE"); + MobilityFlinkSql.registerAll(batch); + String day = "(v, INTERVAL '1' DAY)"; + String[] tints = {"[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]"}; + for (TableEnvironment env : new TableEnvironment[] {tEnv, batch}) { + assertEquals("{[1@2001-01-01 00:00:00+00, 2@2001-01-02 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]}", + aggregateCall(env, "wCount" + day, "tintFromText", tints)); + assertEquals("{[1@2001-01-01 00:00:00+00, 3@2001-01-02 00:00:00+00, 3@2001-01-05 00:00:00+00]}", + aggregateCall(env, "wMax" + day, "tintFromText", tints)); + assertEquals("Interp=Step;{[1@2001-01-01 00:00:00+00, 2@2001-01-02 00:00:00+00, " + + "2@2001-01-04 00:00:00+00], (3@2001-01-04 00:00:00+00, 3@2001-01-05 00:00:00+00]}", + aggregateCall(env, "wAvg" + day, "tintFromText", tints)); + assertEquals("{[1@2001-01-01 00:00:00+00, 1@2001-01-02 00:00:00+00], " + + "[3@2001-01-03 00:00:00+00, 3@2001-01-04 00:00:00+00]}", + aggregateCall(env, "wSum" + day, "tfloatFromText", + "{1@2001-01-01 00:00:00+00, 3@2001-01-03 00:00:00+00}")); + } + } + @Test void aggregatesAnswerWhatMobilityDBAnswers() throws Exception { // What MobilityDB answers for the same aggregate over the same values with the time zone