From 1de04621125e48def3f12d141f8f8d0c731aea47 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Esteban=20Zim=C3=A1nyi?= Date: Mon, 5 Oct 2026 10:14:44 +0200 Subject: [PATCH 1/2] Call the topological and position functions by their class-prefixed names, unquoted The topological and position functions of the generated SQL surface take the prefix of the class whose values or bounding boxes they compare, the names the portable dialect chapter of the MobilityDB manual gives: set, span, spanset, tbox, stbox and tpcbox, and the base type for same (cbufferSame, npointSame, poseSame, posechainSame). Flink's parser takes each unquoted. GeneratedSqlSurfaceTest.theClassPrefixedNameIsCalledUnquoted calls spanOverlaps over two float spans built with floatspanFromText. TopologicalPositionSurfaceTest checks that the 31 topological names are registered, calls every overload of the two families the surface registers, over values read by their readers (FromText, FromHexWKB for a point cloud value, FromHexEWKB for a geometry and a geography, a value with Z for the operators reading the third dimension, and the box of the value beside it for a box argument), and checks that each answers a Boolean; it checks the known answer of each operator over two spans, two time spans and two spatiotemporal boxes. Witness: since MobilityDB 762b3501cc the engines call the overlap of two spans spanOverlaps, so the generated surface registers no function overlaps and the test's call of `overlaps`(floatspan, floatspan) fails validation: No match found for function signature overlaps(FloatSpan, FloatSpan). Measured over MobilityDB master 5e7da7e8d9 with the commits #2980, #2981 and #2984 merge, the catalog of MEOS-API 18a1e6d53c, and JMEOS main with the commits #153, #154 and #155 merge: the surface registers 999 SQL functions and 7709 overloads, the two families 95 names and 1809 overloads; the binding module passes its 21 tests and the benchmark module its 12, and the build prints no warning. Why: a binding calls a function by the name the portable dialect gives it, and every overload of the surface answers. --- .../flink/sql/GeneratedSqlSurfaceTest.java | 6 +- .../sql/TopologicalPositionSurfaceTest.java | 399 ++++++++++++++++++ 2 files changed, 402 insertions(+), 3 deletions(-) create mode 100644 binding/src/test/java/org/mobilitydb/flink/sql/TopologicalPositionSurfaceTest.java 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 cebc5cb..7a7e415 100644 --- a/binding/src/test/java/org/mobilitydb/flink/sql/GeneratedSqlSurfaceTest.java +++ b/binding/src/test/java/org/mobilitydb/flink/sql/GeneratedSqlSurfaceTest.java @@ -129,9 +129,9 @@ void textConstructorsAndOutput() throws Exception { } @Test - void aReservedNameIsCalledQuoted() throws Exception { - assertEquals(true, - scalar("SELECT `overlaps`(floatspan_in('[1, 3]'), floatspan_in('[2, 4]'))")); + void theClassPrefixedNameIsCalledUnquoted() throws Exception { + assertEquals(true, scalar( + "SELECT spanOverlaps(floatspanFromText('[1, 3]'), floatspanFromText('[2, 4]'))")); } @Test diff --git a/binding/src/test/java/org/mobilitydb/flink/sql/TopologicalPositionSurfaceTest.java b/binding/src/test/java/org/mobilitydb/flink/sql/TopologicalPositionSurfaceTest.java new file mode 100644 index 0000000..01b9239 --- /dev/null +++ b/binding/src/test/java/org/mobilitydb/flink/sql/TopologicalPositionSurfaceTest.java @@ -0,0 +1,399 @@ +/***************************************************************************** + * + * This MobilityDB code is provided under The PostgreSQL License. + * Copyright (c) 2020-2026, Université libre de Bruxelles and MobilityDB + * contributors + * + * Permission to use, copy, modify, and distribute this software and its + * documentation for any purpose, without fee, and without a written + * agreement is hereby granted, provided that the above copyright notice and + * this paragraph and the following two paragraphs appear in all copies. + * + * IN NO EVENT SHALL UNIVERSITE LIBRE DE BRUXELLES BE LIABLE TO ANY PARTY FOR + * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING + * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, + * EVEN IF UNIVERSITE LIBRE DE BRUXELLES HAS BEEN ADVISED OF THE POSSIBILITY + * OF SUCH DAMAGE. + * + * UNIVERSITE LIBRE DE BRUXELLES SPECIFICALLY DISCLAIMS ANY WARRANTIES, + * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY + * AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS ON + * AN "AS IS" BASIS, AND UNIVERSITE LIBRE DE BRUXELLES HAS NO OBLIGATIONS TO + * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS. + * + *****************************************************************************/ + +package org.mobilitydb.flink.sql; + +import functions.GeneratedFunctions; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Calls every topological and position function of the generated Flink SQL surface by its + * class-prefixed name, unquoted, on the mini-cluster and libmeos #GeneratedSqlSurfaceTest runs + * the surface on. The topological functions take the prefix of the class whose values or + * bounding boxes they compare, as the position functions do, the names the portable dialect + * chapter of the MobilityDB manual gives: set, span, spanset, tbox, stbox and tpcbox, and a + * base type for same. Each overload the surface registers is called once over values read + * from their text, a box argument being the box of the value beside it, and answers a + * Boolean; each operator answers its known value over two spans, two time spans and two + * spatiotemporal boxes. + */ +class TopologicalPositionSurfaceTest { + + private static TableEnvironment tEnv; + + /** Each name the surface registers, with its function class. */ + private static final Map> REGISTERED = new LinkedHashMap<>(); + + /** The topological names, the operators each prefix takes. */ + private static final Map> TOPOLOGICAL = new LinkedHashMap<>(); + + static { + List all = List.of("Overlaps", "Contains", "Contained", "Same", "Adjacent"); + TOPOLOGICAL.put("stbox", all); + TOPOLOGICAL.put("tbox", all); + TOPOLOGICAL.put("span", all); + TOPOLOGICAL.put("tpcbox", all); + TOPOLOGICAL.put("set", List.of("Overlaps", "Contains", "Contained")); + TOPOLOGICAL.put("spanset", List.of("Overlaps", "Contains", "Contained", "Adjacent")); + for (String base : List.of("cbuffer", "npoint", "pose", "posechain")) { + TOPOLOGICAL.put(base, List.of("Same")); + } + } + + private static final String OPS = "Overlaps|Contains|Contained|Same|Adjacent" + + "|Left|Overleft|Right|Overright|Below|Overbelow|Above|Overabove" + + "|Front|Overfront|Back|Overback|Before|Overbefore|After|Overafter"; + + private static final Pattern FAMILY = Pattern.compile( + "(set|span|spanset|tbox|stbox|tpcbox|cbuffer|npoint|pose|posechain)(" + OPS + ")"); + + /** The position operators reading the third dimension, called over values having one. */ + private static final Set Z_OPS = Set.of("Front", "Overfront", "Back", "Overback"); + + private static final String PCPOINT = "010100000064000000C80000002C010000"; + private static final String PCPATCH = + "01010000000000000002000000640000006400000064000000C8000000C8000000C8000000"; + + /** The text of a value of each SQL type, as the reader of #READER reads it. */ + private static final Map TEXT = new LinkedHashMap<>(); + + /** + * The reader of a SQL type whose text is not read by its FromText: the hex Well-Known + * Binary of a point cloud value and the hex extended one of a geometry and a geography. + */ + private static final Map READER = Map.of( + "pcpoint", "pcpointFromHexWKB", "pcpatch", "pcpatchFromHexWKB", + "geometry", "geometryFromHexEWKB", "geography", "geographyFromHexEWKB"); + + /** The text of a spatial value with a third dimension, where it differs. */ + private static final Map TEXT_Z = new LinkedHashMap<>(); + + static { + TEXT.put("intset", "{1, 2}"); + TEXT.put("bigintset", "{1, 2}"); + TEXT.put("floatset", "{1.5, 2.5}"); + TEXT.put("textset", "{\"a\", \"b\"}"); + TEXT.put("dateset", "{2020-01-01, 2020-01-02}"); + TEXT.put("tstzset", "{2020-01-01, 2020-01-02}"); + TEXT.put("geomset", "{Point(1 1), Point(2 2)}"); + TEXT.put("geogset", "{Point(1 1), Point(2 2)}"); + TEXT.put("cbufferset", "{\"Cbuffer(Point(1 1),0.5)\", \"Cbuffer(Point(2 2),0.5)\"}"); + TEXT.put("npointset", "{\"Npoint(1,0.5)\", \"Npoint(2,0.5)\"}"); + TEXT.put("poseset", "{\"Pose(Point(1 1),0.5)\", \"Pose(Point(2 2),0.5)\"}"); + TEXT.put("posechainset", "{\"PoseChain(Pose(Point(0 0), 0), Pose(Point(10 0), 0))\"}"); + TEXT.put("jsonbset", "{\"{\\\"a\\\": 1}\"}"); + TEXT.put("h3indexset", "{8a2a1072b59ffff}"); + TEXT.put("quadbinset", "{4813ffffffffffff, 4817ffffffffffff}"); + TEXT.put("s2cellset", "{47c3c444c000000c, 47c3c444c0000004}"); + TEXT.put("pcpointset", "{\"" + PCPOINT + "\"}"); + TEXT.put("pcpatchset", "{\"" + PCPATCH + "\"}"); + TEXT.put("intspan", "[1, 3]"); + TEXT.put("bigintspan", "[1, 3]"); + TEXT.put("floatspan", "[1.5, 3.5]"); + TEXT.put("datespan", "[2020-01-01, 2020-01-03]"); + TEXT.put("tstzspan", "[2020-01-01, 2020-01-03]"); + TEXT.put("intspanset", "{[1, 3]}"); + TEXT.put("bigintspanset", "{[1, 3]}"); + TEXT.put("floatspanset", "{[1.5, 3.5]}"); + TEXT.put("datespanset", "{[2020-01-01, 2020-01-03]}"); + TEXT.put("tstzspanset", "{[2020-01-01, 2020-01-03]}"); + TEXT.put("tbox", "TBOXFLOAT XT([1, 3],[2020-01-01, 2020-01-03])"); + TEXT.put("stbox", "STBOX XT(((1,1),(3,3)),[2020-01-01, 2020-01-03])"); + TEXT.put("tpcbox", "TPCBOX(XT(((0,0),(10,10)),[2020-01-01, 2020-01-31]), 1)"); + TEXT.put("cbuffer", "Cbuffer(Point(1 1), 0.5)"); + TEXT.put("npoint", "Npoint(1, 0.5)"); + TEXT.put("nsegment", "NSegment(1, 0.2, 0.6)"); + TEXT.put("pose", "Pose(Point(1 1), 0.5)"); + TEXT.put("posechain", "PoseChain(Pose(Point(0 0), 0), Pose(Point(10 0), 0))"); + TEXT.put("tint", "[1@2020-01-01, 2@2020-01-02]"); + TEXT.put("tbigint", "[1@2020-01-01, 2@2020-01-02]"); + TEXT.put("tfloat", "[1.5@2020-01-01, 2.5@2020-01-02]"); + TEXT.put("tbool", "[t@2020-01-01, f@2020-01-02]"); + TEXT.put("ttext", "[\"a\"@2020-01-01, \"b\"@2020-01-02]"); + TEXT.put("tjsonb", "\"{\\\"a\\\": 1}\"@2020-01-01"); + TEXT.put("tgeompoint", "[Point(1 1)@2020-01-01, Point(2 2)@2020-01-02]"); + TEXT.put("tgeogpoint", "[Point(1 1)@2020-01-01, Point(2 2)@2020-01-02]"); + TEXT.put("tgeometry", "Point(1 1)@2020-01-01"); + TEXT.put("tgeography", "Point(1 1)@2020-01-01"); + TEXT.put("tcbuffer", "Cbuffer(Point(1 1), 0.5)@2020-01-01"); + TEXT.put("tnpoint", "Npoint(1, 0.5)@2020-01-01"); + TEXT.put("tpose", "Pose(Point(1 1), 0.1)@2020-01-01"); + TEXT.put("tposechain", "PoseChain(Pose(Point(1 2), 0.1))@2020-01-01"); + TEXT.put("trgeometry", "Polygon((0 0, 1 0, 1 1, 0 1, 0 0));Pose(Point(5 5),0)@2020-01-01"); + TEXT.put("th3index", "831c02fffffffff@2020-01-01"); + TEXT.put("tquadbin", "480fffffffffffff@2020-01-01"); + TEXT.put("ts2cell", "47c3c3@2020-01-01"); + TEXT.put("tpcpoint", PCPOINT + "@2020-01-01"); + TEXT.put("tpcpatch", PCPATCH + "@2020-01-01"); + TEXT.put("geometry", "0101000000000000000000F03F000000000000F03F"); + TEXT.put("geography", "0101000020E6100000000000000000F03F000000000000F03F"); + TEXT.put("jsonb", "{\"a\": 1}"); + TEXT.put("h3index", "8a2a1072b59ffff"); + TEXT.put("quadbin", "4813ffffffffffff"); + TEXT.put("s2cell", "47c3c444c000000c"); + TEXT.put("pcpoint", PCPOINT); + TEXT.put("pcpatch", PCPATCH); + + TEXT_Z.put("stbox", "STBOX ZT(((1,1,1),(3,3,3)),[2020-01-01, 2020-01-03])"); + TEXT_Z.put("tpcbox", "TPCBOX(ZT(((0,0,0),(10,10,10)),[2020-01-01, 2020-01-31]), 1)"); + TEXT_Z.put("geomset", "{Point(1 1 1), Point(2 2 2)}"); + TEXT_Z.put("geogset", "{Point(1 1 1), Point(2 2 2)}"); + TEXT_Z.put("tgeompoint", "[Point(1 1 1)@2020-01-01, Point(2 2 2)@2020-01-02]"); + TEXT_Z.put("tgeogpoint", "[Point(1 1 1)@2020-01-01, Point(2 2 2)@2020-01-02]"); + TEXT_Z.put("tgeometry", "Point(1 1 1)@2020-01-01"); + TEXT_Z.put("tgeography", "Point(1 1 1)@2020-01-01"); + TEXT_Z.put("tpose", "Pose(Point Z(1 1 1), 1, 0, 0, 0)@2020-01-01"); + TEXT_Z.put("tposechain", "PoseChain(Pose(Point Z(1 1 1), 1, 0, 0, 0))@2020-01-01"); + TEXT_Z.put("trgeometry", + "Polygon Z((0 0 0,2 0 0,2 1 0,0 1 0,0 0 0));Pose(Point Z(1 1 0), 1, 0, 0, 0)@2020-01-01"); + } + + /** + * States the schema of pcid 1 and registers the surface. MEOS is set up by the generated + * wrappers themselves, which install the handler that turns a MEOS error into the + * exception each call below expects, so this test installs none of its own. + */ + @BeforeAll + static void init() { + // The point cloud values name pcid 1, whose schema the libmeos installation ships + // beside its library directory, the one surefire puts on the load path. + Path lib = Paths.get(System.getenv("LD_LIBRARY_PATH").split(":")[0]); + GeneratedFunctions.meos_set_pointcloud_schemas_xml( + lib.resolveSibling("share").resolve("pointcloud_schemas.xml").toString()); + tEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode()); + // #registerAll registers through the environment it is given; a proxy forwarding every + // call records each name with its function class, as the registrar states them + TableEnvironment recording = (TableEnvironment) Proxy.newProxyInstance( + TableEnvironment.class.getClassLoader(), new Class[] {TableEnvironment.class}, + (proxy, method, args) -> { + if (method.getName().equals("createTemporaryFunction") && args.length == 2 + && args[0] instanceof String && args[1] instanceof Class) { + REGISTERED.put((String) args[0], (Class) args[1]); + } + try { + return method.invoke(tEnv, args); + } catch (InvocationTargetException e) { + throw e.getCause(); + } + }); + MobilityFlinkSql.registerAll(recording); + } + + /** The first field of the first row, as #GeneratedSqlSurfaceTest reads an answer. */ + private static Object scalar(String sql) throws Exception { + try (CloseableIterator it = tEnv.executeSql(sql).collect()) { + return it.next().getField(0); + } + } + + /** The SQL type a generated value class carries, its class name in lower case. */ + private static String sqlType(Class c) { + return c.getSimpleName().toLowerCase(); + } + + /** True when the surface converts a value of class {@code from} to the box class {@code box}. */ + private static boolean converts(Class box, Class from) { + String type = sqlType(box); + try { + Class fn = Class.forName("org.mobilitydb.flink.sql.functions." + + type.substring(0, 1).toUpperCase() + type.substring(1)); + return Arrays.stream(fn.getMethods()).anyMatch(m -> m.getName().equals("eval") + && m.getParameterCount() == 1 && m.getParameterTypes()[0] == from); + } catch (ClassNotFoundException e) { + return false; + } + } + + /** The SQL expression of an argument of class {@code c} beside one of class {@code other}. */ + private static String arg(Class c, Class other, boolean z) { + if (c == Integer.class) { + return "CAST(2 AS INT)"; + } + if (c == Long.class) { + return "CAST(2 AS BIGINT)"; + } + if (c == Double.class) { + return "CAST(2.5 AS DOUBLE)"; + } + if (c == java.time.Instant.class) { + return "TO_TIMESTAMP_LTZ(1577923200000, 3)"; + } + if (c == java.time.LocalDate.class) { + return "DATE '2020-01-02'"; + } + if (c == String.class) { + return "'a'"; + } + if (c == Boolean.class) { + return "TRUE"; + } + String type = sqlType(c); + if ((type.equals("stbox") || type.equals("tbox")) && other != null && converts(c, other)) { + return type + "(" + arg(other, null, z) + ")"; + } + String text = z && TEXT_Z.containsKey(type) ? TEXT_Z.get(type) : TEXT.get(type); + if (text == null) { + throw new IllegalStateException("no text stated for the SQL type " + type); + } + return READER.getOrDefault(type, type + "FromText") + "('" + text + "')"; + } + + /** The registered names of the two families, as #registerAll registers them. */ + private static Set familyNames() { + Set names = new TreeSet<>(); + for (String n : REGISTERED.keySet()) { + if (FAMILY.matcher(n).matches()) { + names.add(n); + } + } + return names; + } + + @Test + void everyTopologicalNameIsRegistered() { + List missing = new ArrayList<>(); + for (Map.Entry> e : TOPOLOGICAL.entrySet()) { + for (String op : e.getValue()) { + if (!REGISTERED.containsKey(e.getKey() + op)) { + missing.add(e.getKey() + op); + } + } + } + assertTrue(missing.isEmpty(), "topological names not registered: " + missing); + } + + @Test + void everyOverloadAnswersUnquoted() throws Exception { + Set names = familyNames(); + int calls = 0; + List failures = new ArrayList<>(); + for (String name : names) { + Matcher m = FAMILY.matcher(name); + assertTrue(m.matches()); + boolean z = Z_OPS.contains(m.group(2)); + Class fn = REGISTERED.get(name); + for (Method eval : fn.getMethods()) { + if (!eval.getName().equals("eval") || eval.getParameterCount() != 2) { + continue; + } + Class[] p = eval.getParameterTypes(); + String sql = name + Arrays.toString(p); + try { + sql = "SELECT " + name + "(" + arg(p[0], p[1], z) + ", " + + arg(p[1], p[0], z) + ")"; + Object answer = scalar(sql); + if (!(answer instanceof Boolean)) { + failures.add(sql + " -> " + answer); + } + } catch (Exception e) { + Throwable root = e; + while (root.getCause() != null) { + root = root.getCause(); + } + failures.add(sql + " -> " + String.valueOf(root.getMessage()).split("\n")[0]); + } + calls++; + } + } + assertTrue(calls > 0, "no overload of the two families is registered"); + assertTrue(failures.isEmpty(), failures.size() + " of " + calls + + " overloads of " + names.size() + " names failed:\n" + + String.join("\n", failures)); + } + + @Test + void eachOperatorAnswersItsValue() throws Exception { + // Two spans that overlap, the first ending before the second does + String a = "floatspanFromText('[1, 3]')"; + String b = "floatspanFromText('[2, 4]')"; + Map span = new LinkedHashMap<>(); + span.put("spanOverlaps", true); + span.put("spanContains", false); + span.put("spanContained", false); + span.put("spanAdjacent", false); + span.put("spanLeft", false); + span.put("spanOverleft", true); + span.put("spanRight", false); + span.put("spanOverright", false); + for (Map.Entry e : span.entrySet()) { + assertEquals(e.getValue(), scalar("SELECT " + e.getKey() + "(" + a + ", " + b + ")"), + e.getKey()); + } + // A time span ending before the other starts + String t1 = "tstzspanFromText('[2020-01-01, 2020-01-02]')"; + String t2 = "tstzspanFromText('[2020-01-03, 2020-01-04]')"; + assertEquals(true, scalar("SELECT spanBefore(" + t1 + ", " + t2 + ")")); + assertEquals(true, scalar("SELECT spanOverbefore(" + t1 + ", " + t2 + ")")); + assertEquals(false, scalar("SELECT spanAfter(" + t1 + ", " + t2 + ")")); + assertEquals(false, scalar("SELECT spanOverafter(" + t1 + ", " + t2 + ")")); + // A temporal Boolean whose time span is the time span it is compared with + assertEquals(true, scalar("SELECT spanSame(" + t1 + + ", tboolFromText('[t@2020-01-01, f@2020-01-02]'))")); + // A box left of, below, in front of and before the other + String s1 = "stboxFromText('STBOX ZT(((1,1,1),(2,2,2)),[2020-01-01, 2020-01-02])')"; + String s2 = "stboxFromText('STBOX ZT(((3,3,3),(4,4,4)),[2020-01-03, 2020-01-04])')"; + Map box = new LinkedHashMap<>(); + box.put("stboxLeft", true); + box.put("stboxRight", false); + box.put("stboxBelow", true); + box.put("stboxOverbelow", true); + box.put("stboxAbove", false); + box.put("stboxOverabove", false); + box.put("stboxFront", true); + box.put("stboxOverfront", true); + box.put("stboxBack", false); + box.put("stboxOverback", false); + box.put("stboxBefore", true); + box.put("stboxAfter", false); + box.put("stboxOverlaps", false); + box.put("stboxSame", false); + for (Map.Entry e : box.entrySet()) { + assertEquals(e.getValue(), scalar("SELECT " + e.getKey() + "(" + s1 + ", " + s2 + ")"), + e.getKey()); + } + } +} From cd8658ce78744d508012a58fe2a261d95c5971e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Esteban=20Zim=C3=A1nyi?= Date: Mon, 5 Oct 2026 09:02:45 +0200 Subject: [PATCH 2/2] Report a MEOS error as the exception JMEOS raises for it Every call into MEOS goes through a GeneratedFunctions wrapper, which runs GeneratedFunctions.ensureReady() first: that initialises MEOS once per process, installs MeosErrorHandler as the error handler, and sets up the timezone and collation of the calling thread. After the call the wrapper reads MeosErrorHandler.checkError(), which rethrows the error MEOS reported as a typed MeosException carrying MEOS's own code and message. The wirings, the benchmark programs and the tests therefore install no handler of their own, and the tests neither initialise nor finalise MEOS: a handler installed over MeosErrorHandler records nothing, so checkError() finds no error and a failed call answers false or NULL, and meos_initialize() puts back MEOS's default handler, which ends the process. The wirings call the operator functions directly, as the UDFs of MobilitySpark do. The callback MEOS makes into the JVM to report an error runs through jffi's closures, and JMEOS builds against jnr-ffi 2.2.14, so the binding and the benchmark use that version. Witness: with jnr-ffi 2.1.10 (jffi 1.2.19), GeneratedFunctions.tfloat_in("garbage") ends the JVM with SIGSEGV in the jffi closure stub; with jnr-ffi 2.2.14 it throws MeosTextInputError (22): Missing delimeter character '@': garbage, and back_stbox_stbox over two boxes without Z throws MeosInvalidArgValueError (12): The stbox must have Z dimension. Measured over MobilityDB master 5e7da7e8d9 with the commits #2980, #2981 and #2984 merge, the catalog of MEOS-API 18a1e6d53c and JMEOS main with the commits #153, #154 and #155 merge, on top of #70: the binding module passes its 21 tests, TopologicalPositionSurfaceTest among them, and the benchmark module its 12, and the build prints no warning. Why: MEOS reports an error through the handler JMEOS installs, so the binding leaves that handler in place and receives the error as the exception it raises. --- benchmark/pom.xml | 2 +- benchmark/src/main/java/aisdata/Main.java | 2 - .../src/main/java/aisdata/Query1_Main.java | 6 -- .../src/main/java/aisdata/Query2_Main.java | 6 -- .../src/main/java/aisdata/Query3_Main.java | 6 -- .../src/main/java/aisdata/Query4_Main.java | 6 -- .../src/main/java/aisdata/Query5_Main.java | 6 -- .../src/main/java/aisdata/Query6_Main.java | 5 - .../src/main/java/aisdata/Query7_Main.java | 5 - .../src/main/java/aisdata/Query8_Main.java | 3 - .../src/main/java/aisdata/Query8_V2_Main.java | 6 -- .../src/main/java/aisdata/Query9_Main.java | 5 - .../aisdata/TrajectoryWindowFunction.java | 3 - .../main/java/berlinmod/BerlinMODCorpus.java | 2 - .../main/java/berlinmod/BerlinMODParity.java | 5 - .../src/main/java/berlinmod/MEOSBridge.java | 11 +-- .../meos/wirings/MeosBoundedStateMap.java | 1 - .../meos/wirings/MeosCrossStreamJoin.java | 7 -- .../meos/wirings/MeosStatelessFilter.java | 11 --- .../flink/meos/wirings/MeosStatelessMap.java | 13 --- .../meos/wirings/MeosWindowedAggregate.java | 7 -- .../flink/meos/wirings/MeosWiringRuntime.java | 62 ------------ .../src/main/java/sncbdata/Query1_Main.java | 6 -- .../src/main/java/sncbdata/Query2_Main.java | 6 -- .../src/main/java/sncbdata/Query3_Main.java | 9 -- .../src/main/java/sncbdata/Query4_Main.java | 9 -- .../src/main/java/sncbdata/Query5_Main.java | 9 -- .../src/main/java/sncbdata/Query6_Main.java | 6 -- .../src/main/java/sncbdata/Query7_Main.java | 5 - .../src/main/java/sncbdata/Query8_Main.java | 3 - .../main/java/sncbdata/Query8_V2_Main.java | 6 -- .../src/main/java/sncbdata/Query9_Main.java | 6 -- .../berlinmod/BerlinMODSetSetJoinTest.java | 8 -- .../flink/meos/MeosCbufferSmokeTest.java | 14 --- .../flink/meos/MeosFacadeSmokeTest.java | 15 --- .../flink/meos/MeosNpointSmokeTest.java | 14 --- .../flink/meos/MeosPoseSmokeTest.java | 14 --- .../meos/wirings/MeosBoundedStateMap.java | 1 - .../meos/wirings/MeosCrossStreamJoin.java | 7 -- .../meos/wirings/MeosStatelessFilter.java | 11 --- .../flink/meos/wirings/MeosStatelessMap.java | 13 --- .../meos/wirings/MeosWindowedAggregate.java | 7 -- .../flink/meos/wirings/MeosWiringRuntime.java | 97 ------------------- .../flink/meos/MeosCbufferSmokeTest.java | 14 --- .../flink/meos/MeosFacadeSmokeTest.java | 15 --- .../flink/meos/MeosNpointSmokeTest.java | 14 --- .../flink/meos/MeosPoseSmokeTest.java | 14 --- .../flink/sql/GeneratedSqlSurfaceTest.java | 9 -- pom.xml | 2 +- 49 files changed, 4 insertions(+), 510 deletions(-) delete mode 100644 benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java delete mode 100644 binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java diff --git a/benchmark/pom.xml b/benchmark/pom.xml index bf4a42d..b9c40f7 100644 --- a/benchmark/pom.xml +++ b/benchmark/pom.xml @@ -152,7 +152,7 @@ com.github.jnr jnr-ffi - 2.1.10 + ${jnr.version} com.github.jnr diff --git a/benchmark/src/main/java/aisdata/Main.java b/benchmark/src/main/java/aisdata/Main.java index b5f1b88..7c53e78 100644 --- a/benchmark/src/main/java/aisdata/Main.java +++ b/benchmark/src/main/java/aisdata/Main.java @@ -51,7 +51,6 @@ public class Main { private static final Logger logger = LoggerFactory.getLogger(Main.class); - static error_handler_fn errorHandler = new error_handler(); private STBox stbx; public static void main(String[] args) throws Exception { @@ -63,7 +62,6 @@ public static void main(String[] args) throws Exception { try { logger.info("Initializing MEOS library"); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); diff --git a/benchmark/src/main/java/aisdata/Query1_Main.java b/benchmark/src/main/java/aisdata/Query1_Main.java index 0325d30..0047c6c 100644 --- a/benchmark/src/main/java/aisdata/Query1_Main.java +++ b/benchmark/src/main/java/aisdata/Query1_Main.java @@ -24,8 +24,6 @@ import jnr.ffi.Pointer; import functions.GeneratedFunctions; -import functions.error_handler; -import functions.error_handler_fn; /** * Query 1 - High-Risk Zone Proximity Monitoring @@ -166,7 +164,6 @@ public static void main(String[] args) throws Exception { try { logger.info("Initializing MEOS library"); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -260,7 +257,6 @@ public static class HighRiskZoneWindowFunction private transient Pointer[] hazardZones; - private transient error_handler_fn errorHandler; private static final DateTimeFormatter TIMESTAMP_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ssxxx"); @@ -274,9 +270,7 @@ public HighRiskZoneWindowFunction(String[] zoneWkt, double distanceMeters) { @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); // Parse the INPolygons only once per worker. // geog_in(wkt, -1) creates a geography type (SRID=4326), so // edwithin_tgeo_geo computes distances geodetically in metres - consistent diff --git a/benchmark/src/main/java/aisdata/Query2_Main.java b/benchmark/src/main/java/aisdata/Query2_Main.java index fdb29de..af54a56 100644 --- a/benchmark/src/main/java/aisdata/Query2_Main.java +++ b/benchmark/src/main/java/aisdata/Query2_Main.java @@ -26,8 +26,6 @@ import jnr.ffi.Pointer; import functions.GeneratedFunctions; -import functions.error_handler; -import functions.error_handler_fn; /** * Query 2 - Brake System Monitoring @@ -169,7 +167,6 @@ public static void main(String[] args) throws Exception { try { logger.info("Initializing MEOS library"); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -289,7 +286,6 @@ public static class BrakeMonitoringWindowFunction private transient Pointer[] maintenanceZones; - private transient error_handler_fn errorHandler; private static final DateTimeFormatter TIMESTAMP_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ssxxx"); @@ -307,9 +303,7 @@ public BrakeMonitoringWindowFunction( @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); // Parse the maintenance area polygons only once per worker. // geog_in(wkt, -1) creates a geography type (SRID=4326 implicit), consistent // with the tgeogpoint created by tgeogpoint_in below. diff --git a/benchmark/src/main/java/aisdata/Query3_Main.java b/benchmark/src/main/java/aisdata/Query3_Main.java index 6f0c575..1c47e81 100644 --- a/benchmark/src/main/java/aisdata/Query3_Main.java +++ b/benchmark/src/main/java/aisdata/Query3_Main.java @@ -27,8 +27,6 @@ import jnr.ffi.Pointer; import functions.GeneratedFunctions; -import functions.error_handler; -import functions.error_handler_fn; /** * Query 3 - Trajectory Creation @@ -99,7 +97,6 @@ public static void main(String[] args) throws Exception { try { logger.info("Initializing MEOS library"); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -181,7 +178,6 @@ public static class TrajectoryCreationWindowFunction private static final Logger log = LoggerFactory.getLogger(TrajectoryCreationWindowFunction.class); - private transient error_handler_fn errorHandler; private static final DateTimeFormatter TIMESTAMP_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ssxxx"); @@ -190,9 +186,7 @@ public static class TrajectoryCreationWindowFunction @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); log.info("MEOS initialized in TrajectoryCreationWindowFunction.open()"); } diff --git a/benchmark/src/main/java/aisdata/Query4_Main.java b/benchmark/src/main/java/aisdata/Query4_Main.java index 5a4d00f..492e387 100644 --- a/benchmark/src/main/java/aisdata/Query4_Main.java +++ b/benchmark/src/main/java/aisdata/Query4_Main.java @@ -26,8 +26,6 @@ import jnr.ffi.Pointer; import functions.GeneratedFunctions; -import functions.error_handler; -import functions.error_handler_fn; /** * Query 4 - Trajectory Creation in a Restricted Space @@ -150,7 +148,6 @@ public static void main(String[] args) throws Exception { try { logger.info("Initializing MEOS library"); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -243,7 +240,6 @@ public static class RestrictedTrajectoryWindowFunction private transient Pointer stbox; /** Initialised in {@link #open}: Pointer is not serialisable. */ - private transient error_handler_fn errorHandler; private static final DateTimeFormatter TIMESTAMP_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ssxxx"); @@ -267,9 +263,7 @@ public RestrictedTrajectoryWindowFunction( @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); // stbox_make parameters: // hasx=true → include spatial (XY) dimensions diff --git a/benchmark/src/main/java/aisdata/Query5_Main.java b/benchmark/src/main/java/aisdata/Query5_Main.java index 082dafe..c152107 100644 --- a/benchmark/src/main/java/aisdata/Query5_Main.java +++ b/benchmark/src/main/java/aisdata/Query5_Main.java @@ -26,8 +26,6 @@ import jnr.ffi.Pointer; import functions.GeneratedFunctions; -import functions.error_handler; -import functions.error_handler_fn; /** * Query 5 - Trajectory Creation and High-Speed Alert @@ -169,7 +167,6 @@ public static void main(String[] args) throws Exception { try { logger.info("Initializing MEOS library"); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -269,7 +266,6 @@ public static class HighSpeedAlertWindowFunction private final double minSpeedThresholdMs; /** Initialised in {@link #open} since Pointer is not serialisable. */ - private transient error_handler_fn errorHandler; /** * Geofence polygon pointer, parsed once in {@link #open} for reuse across all windows. @@ -298,9 +294,7 @@ public HighSpeedAlertWindowFunction( @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); // geog_in(wkt, -1) → SRID=4326 geography, consistent with tgeogpoint_in. geofence = GeneratedFunctions.geog_in(geofenceWkt, -1); diff --git a/benchmark/src/main/java/aisdata/Query6_Main.java b/benchmark/src/main/java/aisdata/Query6_Main.java index 8ad1d4b..2a0f95d 100644 --- a/benchmark/src/main/java/aisdata/Query6_Main.java +++ b/benchmark/src/main/java/aisdata/Query6_Main.java @@ -23,7 +23,6 @@ import jnr.ffi.Pointer; import functions.GeneratedFunctions; -import functions.error_handler; /** * Query 6 - Positional Divergence for a Device @@ -109,7 +108,6 @@ public static void main(String[] args) throws Exception { try { logger.info("Initializing MEOS library"); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -205,14 +203,11 @@ public static class NearestApproachJoinFunction extends RichJoinFunction, TGeomPointSeq, Integer, TimeWindow> { private static final Logger logger = LoggerFactory.getLogger(TrajectoryWindowFunction.class); - private transient error_handler_fn errorHandler; // Non-static and transient // private int count = 0; // count variable seems unused, can be removed if not needed @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); // Initialize error handler here GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); logger.info("MEOS initialized in TrajectoryWindowFunction.open()"); } diff --git a/benchmark/src/main/java/berlinmod/BerlinMODCorpus.java b/benchmark/src/main/java/berlinmod/BerlinMODCorpus.java index 8848724..a6ae406 100644 --- a/benchmark/src/main/java/berlinmod/BerlinMODCorpus.java +++ b/benchmark/src/main/java/berlinmod/BerlinMODCorpus.java @@ -27,7 +27,6 @@ import functions.GeneratedFunctions; import jnr.ffi.Pointer; -import org.mobilitydb.flink.meos.wirings.MeosWiringRuntime; import java.nio.file.Files; import java.nio.file.Paths; @@ -92,7 +91,6 @@ public static final class Params { * (columns {@code tripid,vehid,day,seqno,geom,t}), reprojected 3857→4326 * through MEOS, sorted by timestamp. {@code maxRows <= 0} loads all rows. */ public static List fromInstantsCsv(String path, int maxRows) throws Exception { - MeosWiringRuntime.ensureInitializedOnThread(); List events = new ArrayList<>(); try (Stream lines = Files.lines(Paths.get(path))) { java.util.Iterator it = lines.iterator(); diff --git a/benchmark/src/main/java/berlinmod/BerlinMODParity.java b/benchmark/src/main/java/berlinmod/BerlinMODParity.java index 6dee710..9605233 100644 --- a/benchmark/src/main/java/berlinmod/BerlinMODParity.java +++ b/benchmark/src/main/java/berlinmod/BerlinMODParity.java @@ -86,7 +86,6 @@ public static void main(String[] args) throws Exception { csv != null ? "real BerlinMOD instants" : "synthetic", corpus.size(), p.pLon, p.pLat, p.radiusMetres); - MeosWiringInit(); BiPredicate q3 = (lon, lat) -> MEOSBridge.dwithinMetres(lon, lat, p.pLon, p.pLat, p.radiusMetres); BiPredicate q8 = (lon, lat) -> @@ -143,8 +142,4 @@ private static String[] check(String query, List corpus, return new String[]{query, String.valueOf(corpus.size()), String.valueOf(streamingTrue), String.valueOf(batchTrue), String.valueOf(mismatches), parity ? "exact" : "MISMATCH"}; } - - private static void MeosWiringInit() { - org.mobilitydb.flink.meos.wirings.MeosWiringRuntime.ensureInitializedOnThread(); - } } diff --git a/benchmark/src/main/java/berlinmod/MEOSBridge.java b/benchmark/src/main/java/berlinmod/MEOSBridge.java index c74480f..d438f60 100644 --- a/benchmark/src/main/java/berlinmod/MEOSBridge.java +++ b/benchmark/src/main/java/berlinmod/MEOSBridge.java @@ -27,7 +27,6 @@ import functions.GeneratedFunctions; import jnr.ffi.Pointer; -import org.mobilitydb.flink.meos.wirings.MeosWiringRuntime; /** * Thin wiring from the BerlinMOD streaming-form predicates to MEOS via JMEOS. @@ -42,9 +41,8 @@ * mathematics of its own: it constructs the MEOS inputs and delegates the * computation to libmeos. * - *

{@link MeosWiringRuntime#ensureInitializedOnThread()} initialises MEOS on - * the calling task thread (idempotent per thread) before the first call, since - * MEOS keeps its session state per OS thread. + *

Each {@code GeneratedFunctions} call initialises MEOS on the calling task + * thread on its first use, since MEOS keeps its session state per OS thread. */ public final class MEOSBridge { @@ -66,7 +64,6 @@ private MEOSBridge() { public static boolean dwithinMetres(double lon1, double lat1, double lon2, double lat2, double radiusMetres) { - MeosWiringRuntime.ensureInitializedOnThread(); return GeneratedFunctions.edwithin_tgeo_geo( tgeogInst(lon1, lat1), pointGeog(lon2, lat2), radiusMetres) == 1; } @@ -81,7 +78,6 @@ public static boolean dwithinSegmentMetres(double pLon, double pLat, double s1Lon, double s1Lat, double s2Lon, double s2Lat, double radiusMetres) { - MeosWiringRuntime.ensureInitializedOnThread(); return GeneratedFunctions.edwithin_tgeo_geo( tgeogInst(pLon, pLat), lineGeog(s1Lon, s1Lat, s2Lon, s2Lat), radiusMetres) == 1; @@ -96,7 +92,6 @@ public static boolean dwithinSegmentMetres(double pLon, double pLat, public static boolean intersectsBox(double lon, double lat, double xmin, double ymin, double xmax, double ymax) { - MeosWiringRuntime.ensureInitializedOnThread(); return GeneratedFunctions.eintersects_tgeo_geo( tgeomInst(lon, lat), boxPolygon(xmin, ymin, xmax, ymax)) == 1; } @@ -107,7 +102,6 @@ public static boolean intersectsBox(double lon, double lat, */ public static double distanceMetres(double lon1, double lat1, double lon2, double lat2) { - MeosWiringRuntime.ensureInitializedOnThread(); return GeneratedFunctions.geog_distance(pointGeog(lon1, lat1), pointGeog(lon2, lat2)); } @@ -118,7 +112,6 @@ public static double distanceMetres(double lon1, double lat1, public static double distanceSegmentMetres(double pLon, double pLat, double s1Lon, double s1Lat, double s2Lon, double s2Lat) { - MeosWiringRuntime.ensureInitializedOnThread(); return GeneratedFunctions.geog_distance( pointGeog(pLon, pLat), lineGeog(s1Lon, s1Lat, s2Lon, s2Lat)); } diff --git a/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosBoundedStateMap.java b/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosBoundedStateMap.java index e32e71f..a3ae335 100644 --- a/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosBoundedStateMap.java +++ b/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosBoundedStateMap.java @@ -157,7 +157,6 @@ public MeosBoundedStateMap(PointerSerialize serialize, @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - MeosWiringRuntime.ensureInitializedOnThread(); ValueStateDescriptor descriptor = new ValueStateDescriptor<>( "meos-bounded-state", PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO); diff --git a/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosCrossStreamJoin.java b/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosCrossStreamJoin.java index dc1e4cc..f11c8b8 100644 --- a/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosCrossStreamJoin.java +++ b/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosCrossStreamJoin.java @@ -25,7 +25,6 @@ package org.mobilitydb.flink.meos.wirings; -import org.apache.flink.api.common.functions.OpenContext; import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction; import org.apache.flink.util.Collector; @@ -118,12 +117,6 @@ public MeosCrossStreamJoin(JoinFn joinFn) { this.joinFn = joinFn; } - @Override - public void open(OpenContext parameters) throws Exception { - super.open(parameters); - MeosWiringRuntime.ensureInitializedOnThread(); - } - @Override public void processElement(L left, R right, ProcessJoinFunction.Context context, diff --git a/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessFilter.java b/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessFilter.java index 0685443..13a2bab 100644 --- a/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessFilter.java +++ b/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessFilter.java @@ -27,7 +27,6 @@ import org.apache.flink.api.common.functions.FilterFunction; import org.apache.flink.api.common.functions.RichFilterFunction; -import org.apache.flink.api.common.functions.OpenContext; import java.io.Serializable; @@ -95,18 +94,8 @@ public static MeosStatelessFilter fromIntPredicate(MeosIntPredicate return new MeosStatelessFilter<>(event -> p.test(event) != 0); } - @Override - public void open(OpenContext parameters) throws Exception { - super.open(parameters); - MeosWiringRuntime.ensureInitializedOnThread(); - } - @Override public boolean filter(IN event) throws Exception { - // When chained to a legacy source, records are processed on the source's - // emitter thread rather than the thread open() ran on; the ThreadLocal - // guard makes this a cheap no-op after the first call per thread. - MeosWiringRuntime.ensureInitializedOnThread(); return predicate.test(event); } } diff --git a/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessMap.java b/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessMap.java index dafd21c..ab0ed8d 100644 --- a/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessMap.java +++ b/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessMap.java @@ -27,7 +27,6 @@ import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.RichMapFunction; -import org.apache.flink.api.common.functions.OpenContext; import java.io.Serializable; @@ -99,20 +98,8 @@ public MeosStatelessMap(MeosCall call) { this.call = call; } - @Override - public void open(OpenContext parameters) throws Exception { - super.open(parameters); - // No per-key state in the stateless tier; the only per-operator - // concern is MEOS' per-thread session, initialized on this task thread. - MeosWiringRuntime.ensureInitializedOnThread(); - } - @Override public OUT map(IN event) throws Exception { - // When chained to a legacy source, records are processed on the source's - // emitter thread rather than the thread open() ran on; the ThreadLocal - // guard makes this a cheap no-op after the first call per thread. - MeosWiringRuntime.ensureInitializedOnThread(); return call.apply(event); } } diff --git a/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWindowedAggregate.java b/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWindowedAggregate.java index 6921c9c..b018a1b 100644 --- a/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWindowedAggregate.java +++ b/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWindowedAggregate.java @@ -25,7 +25,6 @@ package org.mobilitydb.flink.meos.wirings; -import org.apache.flink.api.common.functions.OpenContext; import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction; import org.apache.flink.streaming.api.windowing.windows.Window; import org.apache.flink.util.Collector; @@ -124,12 +123,6 @@ public MeosWindowedAggregate(WindowFn windowFn) { this.windowFn = windowFn; } - @Override - public void open(OpenContext parameters) throws Exception { - super.open(parameters); - MeosWiringRuntime.ensureInitializedOnThread(); - } - @Override public void process(K key, ProcessWindowFunction.Context context, diff --git a/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java b/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java deleted file mode 100644 index c99bf16..0000000 --- a/benchmark/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java +++ /dev/null @@ -1,62 +0,0 @@ -/***************************************************************************** - * - * This MobilityDB code is provided under The PostgreSQL License. - * Copyright (c) 2020-2026, Université libre de Bruxelles and MobilityDB - * contributors - * - * Permission to use, copy, modify, and distribute this software and its - * documentation for any purpose, without fee, and without a written - * agreement is hereby granted, provided that the above copyright notice and - * this paragraph and the following two paragraphs appear in all copies. - * - * IN NO EVENT SHALL UNIVERSITE LIBRE DE BRUXELLES BE LIABLE TO ANY PARTY FOR - * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING - * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, - * EVEN IF UNIVERSITE LIBRE DE BRUXELLES HAS BEEN ADVISED OF THE POSSIBILITY - * OF SUCH DAMAGE. - * - * UNIVERSITE LIBRE DE BRUXELLES SPECIFICALLY DISCLAIMS ANY WARRANTIES, - * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY - * AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS ON - * AN "AS IS" BASIS, AND UNIVERSITE LIBRE DE BRUXELLES HAS NO OBLIGATIONS TO - * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS. - * - *****************************************************************************/ - -package org.mobilitydb.flink.meos.wirings; - -import functions.GeneratedFunctions; - -/** - * Per-thread MEOS initialization for the {@code org.mobilitydb.flink.meos.wirings} - * operators. - * - *

MEOS keeps its timezone / session state per OS thread. Each Flink - * subtask runs on its own task thread, so every wiring operator must - * initialize MEOS on that thread from its {@code open()} — the JVM-wide - * probe in {@code MeosOpsRuntime} only covers the thread that first - * touches a facade class (typically the job's main thread), not the task - * threads where the operators actually run. - * - *

{@link #ensureInitializedOnThread()} is idempotent per thread (guarded - * by a {@link ThreadLocal}), so it is safe to call from every operator's - * {@code open()} even when operators are chained onto the same thread. It - * installs a no-op error handler so a MEOS-side error surfaces as a thrown - * exception rather than terminating the JVM. - */ -public final class MeosWiringRuntime { - - private static final ThreadLocal INITIALIZED = - ThreadLocal.withInitial(() -> Boolean.FALSE); - - private MeosWiringRuntime() { /* utility */ } - - /** Initialize MEOS on the calling thread exactly once. */ - public static void ensureInitializedOnThread() { - if (!INITIALIZED.get()) { - GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); - GeneratedFunctions.meos_initialize(); - INITIALIZED.set(Boolean.TRUE); - } - } -} diff --git a/benchmark/src/main/java/sncbdata/Query1_Main.java b/benchmark/src/main/java/sncbdata/Query1_Main.java index bd48350..a5bc64f 100644 --- a/benchmark/src/main/java/sncbdata/Query1_Main.java +++ b/benchmark/src/main/java/sncbdata/Query1_Main.java @@ -17,8 +17,6 @@ import org.slf4j.LoggerFactory; import functions.GeneratedFunctions; -import functions.MeosErrorHandler; -import functions.error_handler_fn; import java.time.Duration; import java.time.Instant; @@ -71,7 +69,6 @@ public static void main(String[] args) throws Exception { try { logger.info("Initializing MEOS library"); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new MeosErrorHandler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -132,7 +129,6 @@ public static class HighRiskZoneWindowFunction private final double distanceMeters; private transient Pointer[] hazardZones; - private transient error_handler_fn error_handler; private static final DateTimeFormatter TIMESTAMP_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ssxxx"); @@ -145,9 +141,7 @@ public HighRiskZoneWindowFunction(String[] zoneWkt, double distanceMeters) { @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - error_handler = new MeosErrorHandler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(error_handler); // Parse INPolygons once per worker: not per window call. this.hazardZones = new Pointer[zoneWkt.length]; for (int i = 0; i < zoneWkt.length; i++) { diff --git a/benchmark/src/main/java/sncbdata/Query2_Main.java b/benchmark/src/main/java/sncbdata/Query2_Main.java index 696b722..c674d61 100644 --- a/benchmark/src/main/java/sncbdata/Query2_Main.java +++ b/benchmark/src/main/java/sncbdata/Query2_Main.java @@ -16,8 +16,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import functions.GeneratedFunctions; -import functions.MeosErrorHandler; -import functions.error_handler_fn; import java.time.Duration; import java.time.Instant; import java.time.OffsetDateTime; @@ -67,7 +65,6 @@ public static void main(String[] args) throws Exception { try { logger.info("Initializing MEOS library"); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new MeosErrorHandler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -129,7 +126,6 @@ public static class BrakeMonitoringWindowFunction private final double varFfThreshold; private transient Pointer[] maintenanceZones; - private transient error_handler_fn errorHandler; private static final DateTimeFormatter TIMESTAMP_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ssxxx"); @@ -143,9 +139,7 @@ public BrakeMonitoringWindowFunction(String[] maintenanceAreasWkt, double varFaT @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new MeosErrorHandler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); // Parse maintenance area polygons once per worker maintenanceZones = new Pointer[maintenanceAreasWkt.length]; for (int i = 0; i < maintenanceAreasWkt.length; i++) { diff --git a/benchmark/src/main/java/sncbdata/Query3_Main.java b/benchmark/src/main/java/sncbdata/Query3_Main.java index eb31115..b041647 100644 --- a/benchmark/src/main/java/sncbdata/Query3_Main.java +++ b/benchmark/src/main/java/sncbdata/Query3_Main.java @@ -26,8 +26,6 @@ import org.slf4j.LoggerFactory; import functions.GeneratedFunctions; -import functions.error_handler; -import functions.error_handler_fn; import types.temporal.TInterpolation; /** @@ -66,7 +64,6 @@ public static void main(String[] args) throws Exception { try { GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -120,14 +117,11 @@ public static class TrajectoryCreationV1_WKT private static final Logger log = LoggerFactory.getLogger(TrajectoryCreationV1_WKT.class); private static final DateTimeFormatter TIMESTAMP_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ssxxx"); - private transient error_handler_fn errorHandler; @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); } @Override @@ -179,14 +173,11 @@ public static class TrajectoryCreationV2_Expand private static final Logger log = LoggerFactory.getLogger(TrajectoryCreationV2_Expand.class); private static final DateTimeFormatter TIMESTAMP_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ssxxx"); - private transient error_handler_fn errorHandler; @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); } @Override diff --git a/benchmark/src/main/java/sncbdata/Query4_Main.java b/benchmark/src/main/java/sncbdata/Query4_Main.java index c73bbe3..f114958 100644 --- a/benchmark/src/main/java/sncbdata/Query4_Main.java +++ b/benchmark/src/main/java/sncbdata/Query4_Main.java @@ -25,8 +25,6 @@ import org.slf4j.LoggerFactory; import functions.GeneratedFunctions; -import functions.error_handler; -import functions.error_handler_fn; import types.temporal.TInterpolation; /** @@ -75,7 +73,6 @@ public static void main(String[] args) throws Exception { try { GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -136,7 +133,6 @@ public static class RestrictedTrajectoryV1_WKT private final String tspanLiteral; private transient Pointer stbox; - private transient error_handler_fn errorHandler; public RestrictedTrajectoryV1_WKT( double xmin, double xmax, double ymin, double ymax, String tspanLiteral) { @@ -147,9 +143,7 @@ public RestrictedTrajectoryV1_WKT( @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); Pointer tspan = GeneratedFunctions.tstzspan_in(tspanLiteral); if (tspan == null) { log.error("tstzspan_in returned null for: {}", tspanLiteral); return; } stbox = GeneratedFunctions.stbox_make(true, false, true, 4326, xmin, xmax, ymin, ymax, 0, 0, tspan); @@ -219,7 +213,6 @@ public static class RestrictedTrajectoryV2_Expand private final String tspanLiteral; private transient Pointer stbox; - private transient error_handler_fn errorHandler; public RestrictedTrajectoryV2_Expand( double xmin, double xmax, double ymin, double ymax, String tspanLiteral) { @@ -230,9 +223,7 @@ public RestrictedTrajectoryV2_Expand( @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); Pointer tspan = GeneratedFunctions.tstzspan_in(tspanLiteral); if (tspan == null) { log.error("tstzspan_in returned null for: {}", tspanLiteral); return; } stbox = GeneratedFunctions.stbox_make(true, false, true, 4326, xmin, xmax, ymin, ymax, 0, 0, tspan); diff --git a/benchmark/src/main/java/sncbdata/Query5_Main.java b/benchmark/src/main/java/sncbdata/Query5_Main.java index 7c2ddd2..ce11f41 100644 --- a/benchmark/src/main/java/sncbdata/Query5_Main.java +++ b/benchmark/src/main/java/sncbdata/Query5_Main.java @@ -25,8 +25,6 @@ import org.slf4j.LoggerFactory; import functions.GeneratedFunctions; -import functions.error_handler; -import functions.error_handler_fn; import types.temporal.TInterpolation; /** @@ -85,7 +83,6 @@ public static void main(String[] args) throws Exception { try { GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -149,7 +146,6 @@ public static class HighSpeedAlertV1_WKT private final double avgSpeedThresholdMs; private final double minSpeedThresholdMs; - private transient error_handler_fn errorHandler; private transient Pointer geofence; public HighSpeedAlertV1_WKT(String geofenceWkt, double geofenceDistMeters, @@ -163,9 +159,7 @@ public HighSpeedAlertV1_WKT(String geofenceWkt, double geofenceDistMeters, @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); geofence = GeneratedFunctions.geog_in(geofenceWkt, -1); if (geofence == null) log.error("[V1] geog_in returned null for geofence: {}", geofenceWkt); } @@ -246,7 +240,6 @@ public static class HighSpeedAlertV2_Expand private final double avgSpeedThresholdMs; private final double minSpeedThresholdMs; - private transient error_handler_fn errorHandler; private transient Pointer geofence; public HighSpeedAlertV2_Expand(String geofenceWkt, double geofenceDistMeters, @@ -260,9 +253,7 @@ public HighSpeedAlertV2_Expand(String geofenceWkt, double geofenceDistMeters, @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); geofence = GeneratedFunctions.geog_in(geofenceWkt, -1); if (geofence == null) log.error("[V2] geog_in returned null for geofence: {}", geofenceWkt); } diff --git a/benchmark/src/main/java/sncbdata/Query6_Main.java b/benchmark/src/main/java/sncbdata/Query6_Main.java index 3f68264..c1ce913 100644 --- a/benchmark/src/main/java/sncbdata/Query6_Main.java +++ b/benchmark/src/main/java/sncbdata/Query6_Main.java @@ -13,8 +13,6 @@ import org.slf4j.LoggerFactory; import functions.GeneratedFunctions; -import functions.error_handler_fn; -import functions.MeosErrorHandler; import java.time.Duration; import java.time.Instant; @@ -31,7 +29,6 @@ public static void main(String[] args) throws Exception { try { logger.info("Initializing MEOS"); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new MeosErrorHandler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -103,14 +100,11 @@ public static class NearestApproachJoinFunction DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ssxxx"); // Initialized once per worker in open(). - private transient error_handler_fn errorHandler; @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new MeosErrorHandler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); } @Override diff --git a/benchmark/src/main/java/sncbdata/Query7_Main.java b/benchmark/src/main/java/sncbdata/Query7_Main.java index 4726dc3..a0b0d61 100644 --- a/benchmark/src/main/java/sncbdata/Query7_Main.java +++ b/benchmark/src/main/java/sncbdata/Query7_Main.java @@ -24,7 +24,6 @@ import jnr.ffi.Pointer; import functions.GeneratedFunctions; -import functions.error_handler; /** * Query 7 - Global Closest Device Pairs (Top-k) @@ -59,7 +58,6 @@ public static void main(String[] args) throws Exception { try { GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -118,16 +116,13 @@ public static class ClosestPairsCoGroupFunction private final int topK; // Initialized once per worker in open(). - private transient error_handler errorHandler; public ClosestPairsCoGroupFunction(int topK) { this.topK = topK; } @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); } @Override diff --git a/benchmark/src/main/java/sncbdata/Query8_Main.java b/benchmark/src/main/java/sncbdata/Query8_Main.java index f282539..23aa23a 100644 --- a/benchmark/src/main/java/sncbdata/Query8_Main.java +++ b/benchmark/src/main/java/sncbdata/Query8_Main.java @@ -32,7 +32,6 @@ import jnr.ffi.Pointer; import functions.GeneratedFunctions; -import functions.error_handler; /** * Query 8 - Trajectory Denoising with Extended Kalman Filter (SNCB dataset) @@ -68,7 +67,6 @@ public static void main(String[] args) throws Exception { try { GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -130,7 +128,6 @@ public EKFWindowFunction(double gate, double q, double variance, boolean dropOut public void open(OpenContext parameters) throws Exception { super.open(parameters); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); } @Override diff --git a/benchmark/src/main/java/sncbdata/Query8_V2_Main.java b/benchmark/src/main/java/sncbdata/Query8_V2_Main.java index 73444db..b20950b 100644 --- a/benchmark/src/main/java/sncbdata/Query8_V2_Main.java +++ b/benchmark/src/main/java/sncbdata/Query8_V2_Main.java @@ -25,8 +25,6 @@ import org.slf4j.LoggerFactory; import functions.GeneratedFunctions; -import functions.error_handler; -import functions.error_handler_fn; import types.temporal.TInterpolation; /** @@ -82,7 +80,6 @@ public static void main(String[] args) throws Exception { try { GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -134,7 +131,6 @@ public static class MeosEkfWindowFunction private final double r; private final boolean dropOutliers; - private transient error_handler_fn errorHandler; public MeosEkfWindowFunction(double gate, double q, double r, boolean dropOutliers) { this.gate = gate; @@ -146,9 +142,7 @@ public MeosEkfWindowFunction(double gate, double q, double r, boolean dropOutlie @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); } @Override diff --git a/benchmark/src/main/java/sncbdata/Query9_Main.java b/benchmark/src/main/java/sncbdata/Query9_Main.java index a95ed04..e61940c 100644 --- a/benchmark/src/main/java/sncbdata/Query9_Main.java +++ b/benchmark/src/main/java/sncbdata/Query9_Main.java @@ -24,7 +24,6 @@ import jnr.ffi.Pointer; import functions.GeneratedFunctions; -import functions.error_handler; /** * Query 9 - Windowed Per-Device kNN Join (SNCB dataset) @@ -61,7 +60,6 @@ public static void main(String[] args) throws Exception { try { GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(new error_handler()); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -118,18 +116,14 @@ public static class KnnCoGroupFunction private final int k; - // transient: error_handler wraps a native C callback — not serializable by Flink. // Initialized once per worker in open(). - private transient error_handler errorHandler; public KnnCoGroupFunction(int k) { this.k = k; } @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - errorHandler = new error_handler(); GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_error_handler(errorHandler); } @Override diff --git a/benchmark/src/test/java/berlinmod/BerlinMODSetSetJoinTest.java b/benchmark/src/test/java/berlinmod/BerlinMODSetSetJoinTest.java index c334be2..11f770f 100644 --- a/benchmark/src/test/java/berlinmod/BerlinMODSetSetJoinTest.java +++ b/benchmark/src/test/java/berlinmod/BerlinMODSetSetJoinTest.java @@ -26,7 +26,6 @@ import functions.GeneratedFunctions; import jnr.ffi.Pointer; -import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.mobilitydb.meos.MeosSetSetJoin; @@ -60,17 +59,10 @@ class BerlinMODSetSetJoinTest { @BeforeAll static void init() { - GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); - GeneratedFunctions.meos_initialize(); trips = new Pointer[TRIPS.length]; for (int i = 0; i < TRIPS.length; i++) trips[i] = GeneratedFunctions.tgeompoint_in(TRIPS[i]); } - @AfterAll - static void fini() { - GeneratedFunctions.meos_finalize(); - } - private static Set pairSet(int[][] pairs) { Set s = new HashSet<>(); for (int[] p : pairs) s.add(((long) p[0] << 32) | (p[1] & 0xffffffffL)); diff --git a/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosCbufferSmokeTest.java b/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosCbufferSmokeTest.java index 6406df5..e0cf0fc 100644 --- a/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosCbufferSmokeTest.java +++ b/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosCbufferSmokeTest.java @@ -27,10 +27,7 @@ import org.mobilitydb.meos.*; -import functions.GeneratedFunctions; import jnr.ffi.Pointer; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -44,17 +41,6 @@ */ class MeosCbufferSmokeTest { - @BeforeAll - static void init() { - GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); - GeneratedFunctions.meos_initialize(); - } - - @AfterAll - static void finalizeMeos() { - GeneratedFunctions.meos_finalize(); - } - @Test void cbuffer() { Pointer cb = MeosOpsCbuffer.cbuffer_make(MeosOpsGeometry.geom_in("POINT(1 1)", 0), 0.5); diff --git a/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosFacadeSmokeTest.java b/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosFacadeSmokeTest.java index be423b1..3fb2b40 100644 --- a/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosFacadeSmokeTest.java +++ b/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosFacadeSmokeTest.java @@ -27,10 +27,7 @@ import org.mobilitydb.meos.*; -import functions.GeneratedFunctions; import jnr.ffi.Pointer; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -46,18 +43,6 @@ */ class MeosFacadeSmokeTest { - @BeforeAll - static void init() { - // No-op error handler so a parse error returns rather than terminating the JVM. - GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); - GeneratedFunctions.meos_initialize(); - } - - @AfterAll - static void finalizeMeos() { - GeneratedFunctions.meos_finalize(); - } - @Test void coreTbox() { Pointer tbox = MeosOpsTBox.tbox_in("TBOX X([1, 2])"); diff --git a/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosNpointSmokeTest.java b/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosNpointSmokeTest.java index b6a300f..57a1bb2 100644 --- a/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosNpointSmokeTest.java +++ b/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosNpointSmokeTest.java @@ -27,10 +27,7 @@ import org.mobilitydb.meos.*; -import functions.GeneratedFunctions; import jnr.ffi.Pointer; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -43,17 +40,6 @@ */ class MeosNpointSmokeTest { - @BeforeAll - static void init() { - GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); - GeneratedFunctions.meos_initialize(); - } - - @AfterAll - static void finalizeMeos() { - GeneratedFunctions.meos_finalize(); - } - @Test void npoint() { Pointer np = MeosOpsNpoint.npoint_make(1, 0.5); diff --git a/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosPoseSmokeTest.java b/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosPoseSmokeTest.java index 0bbfb57..874f6ec 100644 --- a/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosPoseSmokeTest.java +++ b/benchmark/src/test/java/org/mobilitydb/flink/meos/MeosPoseSmokeTest.java @@ -27,10 +27,7 @@ import org.mobilitydb.meos.*; -import functions.GeneratedFunctions; import jnr.ffi.Pointer; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -44,17 +41,6 @@ */ class MeosPoseSmokeTest { - @BeforeAll - static void init() { - GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); - GeneratedFunctions.meos_initialize(); - } - - @AfterAll - static void finalizeMeos() { - GeneratedFunctions.meos_finalize(); - } - @Test void pose() { Pointer pose = MeosOpsPose.pose_in("Pose(Point(1 1), 0.5)"); diff --git a/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosBoundedStateMap.java b/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosBoundedStateMap.java index e32e71f..a3ae335 100644 --- a/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosBoundedStateMap.java +++ b/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosBoundedStateMap.java @@ -157,7 +157,6 @@ public MeosBoundedStateMap(PointerSerialize serialize, @Override public void open(OpenContext parameters) throws Exception { super.open(parameters); - MeosWiringRuntime.ensureInitializedOnThread(); ValueStateDescriptor descriptor = new ValueStateDescriptor<>( "meos-bounded-state", PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO); diff --git a/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosCrossStreamJoin.java b/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosCrossStreamJoin.java index dc1e4cc..f11c8b8 100644 --- a/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosCrossStreamJoin.java +++ b/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosCrossStreamJoin.java @@ -25,7 +25,6 @@ package org.mobilitydb.flink.meos.wirings; -import org.apache.flink.api.common.functions.OpenContext; import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction; import org.apache.flink.util.Collector; @@ -118,12 +117,6 @@ public MeosCrossStreamJoin(JoinFn joinFn) { this.joinFn = joinFn; } - @Override - public void open(OpenContext parameters) throws Exception { - super.open(parameters); - MeosWiringRuntime.ensureInitializedOnThread(); - } - @Override public void processElement(L left, R right, ProcessJoinFunction.Context context, diff --git a/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessFilter.java b/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessFilter.java index 0685443..13a2bab 100644 --- a/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessFilter.java +++ b/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessFilter.java @@ -27,7 +27,6 @@ import org.apache.flink.api.common.functions.FilterFunction; import org.apache.flink.api.common.functions.RichFilterFunction; -import org.apache.flink.api.common.functions.OpenContext; import java.io.Serializable; @@ -95,18 +94,8 @@ public static MeosStatelessFilter fromIntPredicate(MeosIntPredicate return new MeosStatelessFilter<>(event -> p.test(event) != 0); } - @Override - public void open(OpenContext parameters) throws Exception { - super.open(parameters); - MeosWiringRuntime.ensureInitializedOnThread(); - } - @Override public boolean filter(IN event) throws Exception { - // When chained to a legacy source, records are processed on the source's - // emitter thread rather than the thread open() ran on; the ThreadLocal - // guard makes this a cheap no-op after the first call per thread. - MeosWiringRuntime.ensureInitializedOnThread(); return predicate.test(event); } } diff --git a/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessMap.java b/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessMap.java index dafd21c..ab0ed8d 100644 --- a/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessMap.java +++ b/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessMap.java @@ -27,7 +27,6 @@ import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.RichMapFunction; -import org.apache.flink.api.common.functions.OpenContext; import java.io.Serializable; @@ -99,20 +98,8 @@ public MeosStatelessMap(MeosCall call) { this.call = call; } - @Override - public void open(OpenContext parameters) throws Exception { - super.open(parameters); - // No per-key state in the stateless tier; the only per-operator - // concern is MEOS' per-thread session, initialized on this task thread. - MeosWiringRuntime.ensureInitializedOnThread(); - } - @Override public OUT map(IN event) throws Exception { - // When chained to a legacy source, records are processed on the source's - // emitter thread rather than the thread open() ran on; the ThreadLocal - // guard makes this a cheap no-op after the first call per thread. - MeosWiringRuntime.ensureInitializedOnThread(); return call.apply(event); } } diff --git a/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWindowedAggregate.java b/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWindowedAggregate.java index 6921c9c..b018a1b 100644 --- a/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWindowedAggregate.java +++ b/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWindowedAggregate.java @@ -25,7 +25,6 @@ package org.mobilitydb.flink.meos.wirings; -import org.apache.flink.api.common.functions.OpenContext; import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction; import org.apache.flink.streaming.api.windowing.windows.Window; import org.apache.flink.util.Collector; @@ -124,12 +123,6 @@ public MeosWindowedAggregate(WindowFn windowFn) { this.windowFn = windowFn; } - @Override - public void open(OpenContext parameters) throws Exception { - super.open(parameters); - MeosWiringRuntime.ensureInitializedOnThread(); - } - @Override public void process(K key, ProcessWindowFunction.Context context, diff --git a/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java b/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java deleted file mode 100644 index 6614177..0000000 --- a/binding/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java +++ /dev/null @@ -1,97 +0,0 @@ -/***************************************************************************** - * - * This MobilityDB code is provided under The PostgreSQL License. - * Copyright (c) 2020-2026, Université libre de Bruxelles and MobilityDB - * contributors - * - * Permission to use, copy, modify, and distribute this software and its - * documentation for any purpose, without fee, and without a written - * agreement is hereby granted, provided that the above copyright notice and - * this paragraph and the following two paragraphs appear in all copies. - * - * IN NO EVENT SHALL UNIVERSITE LIBRE DE BRUXELLES BE LIABLE TO ANY PARTY FOR - * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING - * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, - * EVEN IF UNIVERSITE LIBRE DE BRUXELLES HAS BEEN ADVISED OF THE POSSIBILITY - * OF SUCH DAMAGE. - * - * UNIVERSITE LIBRE DE BRUXELLES SPECIFICALLY DISCLAIMS ANY WARRANTIES, - * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY - * AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS ON - * AN "AS IS" BASIS, AND UNIVERSITE LIBRE DE BRUXELLES HAS NO OBLIGATIONS TO - * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS. - * - *****************************************************************************/ - -package org.mobilitydb.flink.meos.wirings; - -import functions.GeneratedFunctions; -import functions.error_handler_fn; - -/** - * MEOS initialization for the {@code org.mobilitydb.flink.meos.wirings} operators. - * - *

MEOS setup has two lifetimes: process-global state (the allocator and the - * error handler) is installed once per JVM, while thread-local state (the - * timezone and collation caches; the PROJ, GEOS and GSL contexts) belongs to - * each task thread. Each Flink subtask runs on its own task thread, so every - * wiring operator calls {@link #ensureInitializedOnThread()} from its - * {@code open()} to install the thread-local part on that thread. - * - *

{@link #ensureInitializedOnThread()} is idempotent per thread (guarded by a - * {@link ThreadLocal}), so it is safe to call from every operator's - * {@code open()} even when operators are chained onto the same thread. The - * process-global no-exit error handler makes a MEOS-side error surface as a - * thrown exception rather than terminating the JVM. - */ -public final class MeosWiringRuntime { - - /** - * No-exit MEOS error handler. MEOS's default handler calls exit(EXIT_FAILURE) - * on an ERROR, which would tear the JVM down if a MEOS error fired inside a - * task thread; this handler returns instead, and the error still surfaces - * because MEOS sets meos_errno. Held as a static field so JNR keeps the - * native callback alive for the process lifetime. - */ - private static final error_handler_fn NOEXIT_ERROR_HANDLER = - (level, code, message) -> { /* do not exit the JVM */ }; - - /** - * Process-global MEOS setup, installed exactly once per JVM: the allocator - * and the error handler are process-global, not thread-local. The holder's - * class initializer runs under the JVM class-initialization lock. meos_initialize() - * runs first — it installs MEOS's exiting default handler — and the no-exit - * handler then replaces it, so a thread reaching its per-thread setup always - * sees the no-exit handler and the exiting default is never observable. - */ - private static final class ProcessInit { - static { - GeneratedFunctions.meos_initialize(); - GeneratedFunctions.meos_initialize_error_handler(NOEXIT_ERROR_HANDLER); - } - /** Invoking this forces the class initializer above to run once. */ - static void ensure() { /* side effect: class initialization */ } - } - - /** - * Per-thread MEOS setup, run once per task thread: only the thread-local - * caches. The timezone and collation caches are thread-local and set - * explicitly per thread; the PROJ, GEOS and GSL contexts are thread-local too - * and created lazily by MEOS on first use. Full meos_initialize() is NOT run - * per thread — it re-installs the exiting default error handler that every - * other thread relies on being the no-exit one. - */ - private static final ThreadLocal INITIALIZED = ThreadLocal.withInitial(() -> { - ProcessInit.ensure(); - GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_collation(); - return Boolean.TRUE; - }); - - private MeosWiringRuntime() { /* utility */ } - - /** Initialize MEOS on the calling thread exactly once. */ - public static void ensureInitializedOnThread() { - INITIALIZED.get(); - } -} diff --git a/binding/src/test/java/org/mobilitydb/flink/meos/MeosCbufferSmokeTest.java b/binding/src/test/java/org/mobilitydb/flink/meos/MeosCbufferSmokeTest.java index 6406df5..e0cf0fc 100644 --- a/binding/src/test/java/org/mobilitydb/flink/meos/MeosCbufferSmokeTest.java +++ b/binding/src/test/java/org/mobilitydb/flink/meos/MeosCbufferSmokeTest.java @@ -27,10 +27,7 @@ import org.mobilitydb.meos.*; -import functions.GeneratedFunctions; import jnr.ffi.Pointer; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -44,17 +41,6 @@ */ class MeosCbufferSmokeTest { - @BeforeAll - static void init() { - GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); - GeneratedFunctions.meos_initialize(); - } - - @AfterAll - static void finalizeMeos() { - GeneratedFunctions.meos_finalize(); - } - @Test void cbuffer() { Pointer cb = MeosOpsCbuffer.cbuffer_make(MeosOpsGeometry.geom_in("POINT(1 1)", 0), 0.5); diff --git a/binding/src/test/java/org/mobilitydb/flink/meos/MeosFacadeSmokeTest.java b/binding/src/test/java/org/mobilitydb/flink/meos/MeosFacadeSmokeTest.java index be423b1..3fb2b40 100644 --- a/binding/src/test/java/org/mobilitydb/flink/meos/MeosFacadeSmokeTest.java +++ b/binding/src/test/java/org/mobilitydb/flink/meos/MeosFacadeSmokeTest.java @@ -27,10 +27,7 @@ import org.mobilitydb.meos.*; -import functions.GeneratedFunctions; import jnr.ffi.Pointer; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -46,18 +43,6 @@ */ class MeosFacadeSmokeTest { - @BeforeAll - static void init() { - // No-op error handler so a parse error returns rather than terminating the JVM. - GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); - GeneratedFunctions.meos_initialize(); - } - - @AfterAll - static void finalizeMeos() { - GeneratedFunctions.meos_finalize(); - } - @Test void coreTbox() { Pointer tbox = MeosOpsTBox.tbox_in("TBOX X([1, 2])"); diff --git a/binding/src/test/java/org/mobilitydb/flink/meos/MeosNpointSmokeTest.java b/binding/src/test/java/org/mobilitydb/flink/meos/MeosNpointSmokeTest.java index b6a300f..57a1bb2 100644 --- a/binding/src/test/java/org/mobilitydb/flink/meos/MeosNpointSmokeTest.java +++ b/binding/src/test/java/org/mobilitydb/flink/meos/MeosNpointSmokeTest.java @@ -27,10 +27,7 @@ import org.mobilitydb.meos.*; -import functions.GeneratedFunctions; import jnr.ffi.Pointer; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -43,17 +40,6 @@ */ class MeosNpointSmokeTest { - @BeforeAll - static void init() { - GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); - GeneratedFunctions.meos_initialize(); - } - - @AfterAll - static void finalizeMeos() { - GeneratedFunctions.meos_finalize(); - } - @Test void npoint() { Pointer np = MeosOpsNpoint.npoint_make(1, 0.5); diff --git a/binding/src/test/java/org/mobilitydb/flink/meos/MeosPoseSmokeTest.java b/binding/src/test/java/org/mobilitydb/flink/meos/MeosPoseSmokeTest.java index 0bbfb57..874f6ec 100644 --- a/binding/src/test/java/org/mobilitydb/flink/meos/MeosPoseSmokeTest.java +++ b/binding/src/test/java/org/mobilitydb/flink/meos/MeosPoseSmokeTest.java @@ -27,10 +27,7 @@ import org.mobilitydb.meos.*; -import functions.GeneratedFunctions; import jnr.ffi.Pointer; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -44,17 +41,6 @@ */ class MeosPoseSmokeTest { - @BeforeAll - static void init() { - GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); - GeneratedFunctions.meos_initialize(); - } - - @AfterAll - static void finalizeMeos() { - GeneratedFunctions.meos_finalize(); - } - @Test void pose() { Pointer pose = MeosOpsPose.pose_in("Pose(Point(1 1), 0.5)"); 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 7a7e415..996170e 100644 --- a/binding/src/test/java/org/mobilitydb/flink/sql/GeneratedSqlSurfaceTest.java +++ b/binding/src/test/java/org/mobilitydb/flink/sql/GeneratedSqlSurfaceTest.java @@ -36,7 +36,6 @@ import org.apache.flink.table.api.TableEnvironment; import org.apache.flink.types.Row; import org.apache.flink.util.CloseableIterator; -import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.mobilitydb.flink.sql.types.TFloat; @@ -61,9 +60,6 @@ class GeneratedSqlSurfaceTest { @BeforeAll static void init() { - // No-op error handler so a parse error returns rather than terminating the JVM. - GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); - GeneratedFunctions.meos_initialize(); String hex = TFloat.encode(GeneratedFunctions.tfloat_in( "[1@2020-01-01 00:00:00+00, 3@2020-01-03 00:00:00+00]")).toString(); hexLiteral = "'" + hex + "'"; @@ -72,11 +68,6 @@ static void init() { MobilityFlinkSql.registerAll(tEnv); } - @AfterAll - static void finalizeMeos() { - GeneratedFunctions.meos_finalize(); - } - /** Every row the query emits; a grouped query's last row is its final answer. */ private static List rows(String sql) throws Exception { List out = new ArrayList<>(); diff --git a/pom.xml b/pom.xml index 0032fe7..43c0c15 100644 --- a/pom.xml +++ b/pom.xml @@ -27,7 +27,7 @@ 3.2.0 2.17.2 1.0 - 2.1.10 + 2.2.14