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 cebc5cb..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<>(); @@ -129,9 +120,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()); + } + } +} 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