From ebcd83e597783663001276d08d0878e0fcc9ae7e Mon Sep 17 00:00:00 2001 From: Esteban Zimanyi Date: Sun, 11 Oct 2026 19:57:21 +0200 Subject: [PATCH] Run the aggregates of the catalog over a base value on both typed SQL surfaces setUnion over an integer, a bigint, a float, a text, a date and a timestamp, extent over an integer, a bigint, a float, a date and a timestamp, and tCount over a timestamp fold a base value into a set, a span or a temporal integer, each role a public MEOS function: int_union_transfn, int_extent_transfn, timestamptz_tcount_transfn and their siblings, with the combine, final, serialize and deserialize of the same aggregate over the typed values. _catalog_aggregates of tools/codegen_jvm.py carries an aggregate whose first argument is a base value, read into the class the scalars of the surface take it as (Integer, Long, Double, String, LocalDate, Instant) and passed to the transition as they pass it, and _spark_encoder gives Spark the encoder of each of those classes. Flink chooses the accumulate of an overload from the class of the argument, as for the typed values. Witness: with the generator of JMEOS main e8c65dc4 both surfaces leave the 12 out as "argument or result of no value type", and MobilitySpark's test fails with "aggregate setUnionAgg(int) does not exist". Measured over MobilityDB b2dcdf7d84 and MEOS-API 9b337a00fe: both surfaces register 22 aggregates of 260 overloads against 22 of 248, the 12 added and nothing else changing; every aggregate of the catalog is carried. With this generator MobilitySpark passes 46 tests, 18 of them in GeneratedSqlSurfaceTest, and MobilityFlink 35 binding and 12 benchmark tests, each surface checking the 12 against MobilityDB's answers over the same values, with 0 warnings. Why: a set, a span or a count over the integers, dates or timestamps of a table answers on Spark and Flink as it does in PostgreSQL. --- tools/codegen_jvm.py | 22 ++++++++++++++++------ 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/tools/codegen_jvm.py b/tools/codegen_jvm.py index 5b079bcc1..bdb0c2608 100644 --- a/tools/codegen_jvm.py +++ b/tools/codegen_jvm.py @@ -2774,6 +2774,10 @@ def _spark_planning(arity): # both engines encode. _AGG_ARGUMENT = {'java.time.Duration', 'String', 'Double'} +# The class of the first argument of an aggregate over a base value, beside the value types: the +# integers, floats, texts, dates and timestamps setUnion, extent and tCount fold. +_AGG_BASE = {'Integer', 'Long', 'Double', 'String', 'java.time.LocalDate', 'java.time.Instant'} + # The MEOS function answering the start timestamp of a temporal value, the order an aggregate with # no combine folds its rows in: the instant of a temporal instant, the lower bound of a sequence. _ORDER_KEY = 'temporal_start_timestamptz' @@ -2783,7 +2787,8 @@ def _catalog_aggregates(m, cat, jmeos, scalars): """The aggregates of the catalog's `aggregates` section both typed surfaces carry: {SQL name: [overload]} and {reason: count} for what is left out. An overload is a dict of its argument and result classes, the transition's inputs as #_inputs passes them, and the Java of - its roles. An aggregate is carried when it takes a value of a type the surface holds, each + its roles. An aggregate is carried when it takes a value of a type the surface holds or a base + value (an integer, a float, a text, a date or a timestamp: setUnion, extent, tCount), each further argument a value, an interval (the window of wCount), a text (the interpolation of appendInstantAgg) or a float that the transition takes as the signature of the transition the aggregate names reads it (its boundArgs and sqlArgParams), states a transition, and its @@ -2821,8 +2826,8 @@ def _catalog_aggregates(m, cat, jmeos, scalars): elif stype not in m.value_class: left[f'state {stype}'] += 1 continue - if args[0] not in m.value_class or ret not in m.value_class: - left[f'argument {args[0]} or result {ret} of no value type'] += 1 + if ret not in m.value_class: + left[f'result {ret} of no value type'] += 1 continue roles = [meos(k) for k in ('transition', 'combine', 'final', 'serialize', 'deserialize')] if any(r and r not in jmeos for r in roles): @@ -2839,6 +2844,9 @@ def _catalog_aggregates(m, cat, jmeos, scalars): left[f'transition input {why or "of no value"}'] += 1 continue classes = [fc for fc, _ in got[0]] + if classes[0] not in _AGG_BASE and not classes[0].startswith(f'{m.pkg}.types.'): + left[f'argument {args[0]} of no value or base type'] += 1 + continue if any(c not in _AGG_ARGUMENT and not c.startswith(f'{m.pkg}.types.') for c in classes[1:]): left['argument of no value type or interval'] += 1 continue @@ -3309,10 +3317,12 @@ def _flink_aggregate_src(sqlname, cls, ovs): def _spark_encoder(classes): """The Spark encoder of the arguments of an aggregate overload: a value's bean encoder, the - Duration encoder of an interval, the String and Double encoders of a text and a float, and the - tuple of them for several arguments.""" + Duration encoder of an interval, the encoder of each base value (a text, a float, an integer, + a date, a timestamp), and the tuple of them for several arguments.""" scalar = {'java.time.Duration': 'Encoders.DURATION()', 'String': 'Encoders.STRING()', - 'Double': 'Encoders.DOUBLE()'} + 'Double': 'Encoders.DOUBLE()', 'Integer': 'Encoders.INT()', + 'Long': 'Encoders.LONG()', 'java.time.LocalDate': 'Encoders.LOCALDATE()', + 'java.time.Instant': 'Encoders.INSTANT()'} one = lambda c: scalar.get(c) or f'Encoders.bean({c}.class)' # noqa: E731 if len(classes) == 1: return one(classes[0])