From 6bc974bacd640a8455f8d2773766742377e91de1 Mon Sep 17 00:00:00 2001
From: Esteban Zimanyi <estebanzimanyi@gmail.com>
Date: Sat, 10 Oct 2026 10:53:47 +0200
Subject: [PATCH] Check the aggregates of every state the catalog states on
 both Spark surfaces

GeneratedSqlSurfaceTest.aggregatesOfEveryStateAnswerWhatMobilityDBAnswers checks that tAvg and
tCentroid, whose skip lists carry sums, counts and coordinates, setUnionAgg and spanUnionAgg,
whose states are a set and a span set, and extent over boxes and over a temporal float, whose
state is a box, answer what MobilityDB answers over values in partitions of their own, so every
partial state crosses the shuffle as the bytes its serialize function or its type's codec
writes. GeneratedSurfaceTest.temporal_aggregates_answer_what_mobilitydb_answers checks tAvg over
binary values on the UDF surface.

Witness: with the generator of JMEOS main both tests fail on "Cannot resolve routine `tAvg`".

Measured over MobilityDB bdc4e41234, the catalog of MEOS-API d259ddbfea and the generator of
JMEOS codegen/catalog-aggregates: the 38 tests pass.

Why: an aggregate answers on Spark as it does in PostgreSQL, whatever state it keeps.
---
 .../spark/GeneratedSqlSurfaceTest.java        | 26 +++++++++++++++++++
 .../spark/GeneratedSurfaceTest.java           |  7 +++++
 2 files changed, 33 insertions(+)

diff --git a/src/test/java/org/mobilitydb/spark/GeneratedSqlSurfaceTest.java b/src/test/java/org/mobilitydb/spark/GeneratedSqlSurfaceTest.java
index d20fd64c..666f2c74 100644
--- a/src/test/java/org/mobilitydb/spark/GeneratedSqlSurfaceTest.java
+++ b/src/test/java/org/mobilitydb/spark/GeneratedSqlSurfaceTest.java
@@ -280,6 +280,32 @@ void temporalAggregatesAnswerWhatMobilityDBAnswers() {
             + "tintFromText(s) AS v FROM VALUES ('" + String.join("'), ('", tints) + "') AS t(s))"));
     }
 
+    @Test
+    void aggregatesOfEveryStateAnswerWhatMobilityDBAnswers() {
+        // What MobilityDB answers for the same aggregate over the same values with the time zone
+        // set to UTC, each value in a partition of its own, so every partial state crosses the
+        // shuffle: the skip list of tAvg and tCentroid as the bytes taggstate_serialize writes,
+        // the set and span set of the unions as setstate_serialize and spansetstate_serialize
+        // write them, and the box of extent as its WKB.
+        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("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("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("setUnionAgg", "intsetFromText", "{1, 3}", "{2, 3, 5}"));
+        assertEquals("{[1, 5), [7, 8)}",
+            aggregate("spanUnionAgg", "intspanFromText", "[1, 3)", "[2, 5)", "[7, 8)"));
+        assertEquals("STBOX X((0,1),(2,4))",
+            aggregate("extent", "stboxFromText", "STBOX X((1,1),(2,2))", "STBOX X((0,3),(1,4))"));
+        assertEquals("TBOXFLOAT XT([1, 5],[2001-01-01 00:00:00+00, 2001-01-04 00:00:00+00])",
+            aggregate("extent", "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]"));
+    }
+
     @Test
     void peakCountOverMergedPieces() {
         // The pieces of one trip are merged first, so the count counts trips, not pieces: two
diff --git a/src/test/java/org/mobilitydb/spark/GeneratedSurfaceTest.java b/src/test/java/org/mobilitydb/spark/GeneratedSurfaceTest.java
index e50387ac..80bdefe4 100644
--- a/src/test/java/org/mobilitydb/spark/GeneratedSurfaceTest.java
+++ b/src/test/java/org/mobilitydb/spark/GeneratedSurfaceTest.java
@@ -505,6 +505,13 @@ void temporal_aggregates_answer_what_mobilitydb_answers() {
             aggregate("tAndAgg", "tbool_in", "tbool_out", tbools));
         assertEquals("{[t@2001-01-01 00:00:00+00, t@2001-01-04 00:00:00+00]}",
             aggregate("tOrAgg", "tbool_in", "tbool_out", tbools));
+        // The skip list of tAvg carries the sum and the count of each instant, which the bytes
+        // taggstate_serialize writes carry across the shuffle.
+        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("tAvg", "tfloat_in", "asText", "[1@2001-01-01, 3@2001-01-03]",
+                      "[3@2001-01-02, 5@2001-01-04]"));
         // A temporal integer and a temporal float in two partitions are values of two
         // types, which a typed aggregate refuses as MobilityDB's typed aggregates do.
         Exception e = assertThrows(Exception.class, () -> scalar("SELECT tSum(v) FROM (SELECT "