Pure-Java fallback for {@link MEOSBridge#dwithinMetres} and - * {@link MEOSBridge#distanceMetres}, used by the BerlinMOD-9 × 3-form - * streaming scaffold when libmeos is not loadable on the runtime path - * (e.g. the mini-cluster local tests in {@code BerlinMODQ*LocalTest}). The - * primary spatial-predicate surface is {@link MEOSBridge}; this class is a - * fallback only. - */ -public final class Haversine { - - private static final double EARTH_RADIUS_METRES = 6_371_000.0; - - private Haversine() { - // utility - } - - /** - * @return great-circle distance in metres between (lon1, lat1) and (lon2, lat2) - */ - public static double distanceMetres(double lon1, double lat1, double lon2, double lat2) { - double phi1 = Math.toRadians(lat1); - double phi2 = Math.toRadians(lat2); - double dPhi = Math.toRadians(lat2 - lat1); - double dLambda = Math.toRadians(lon2 - lon1); - - double a = Math.sin(dPhi / 2) * Math.sin(dPhi / 2) - + Math.cos(phi1) * Math.cos(phi2) - * Math.sin(dLambda / 2) * Math.sin(dLambda / 2); - double c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a)); - return EARTH_RADIUS_METRES * c; - } - - /** - * @return true if the great-circle distance from (lon, lat) to (pLon, pLat) - * is ≤ {@code radiusMetres} - */ - public static boolean withinMetres(double lon, double lat, double pLon, double pLat, double radiusMetres) { - return distanceMetres(lon, lat, pLon, pLat) <= radiusMetres; - } -} diff --git a/flink-processor/src/main/java/berlinmod/MEOSBridge.java b/flink-processor/src/main/java/berlinmod/MEOSBridge.java index 12570e0..6b29952 100644 --- a/flink-processor/src/main/java/berlinmod/MEOSBridge.java +++ b/flink-processor/src/main/java/berlinmod/MEOSBridge.java @@ -1,158 +1,133 @@ package berlinmod; -import functions.functions; +import functions.GeneratedFunctions; import jnr.ffi.Pointer; -import utils.spatial.PointToSegment; +import org.mobilitydb.flink.meos.wirings.MeosWiringRuntime; /** - * Runtime bridge from MobilityFlink BerlinMOD streaming-form predicates to - * MEOS via JMEOS. + * Thin wiring from the BerlinMOD streaming-form predicates to MEOS via JMEOS. * - *
All spatial predicates exercised by the BerlinMOD-9 × 3-form scaffold - * flow through this class. When the JMEOS native libmeos shared object is - * present and loadable, each predicate evaluates through MEOS' WGS84 - * geography surface ({@code geom_to_geog} + {@code geog_dwithin}). When - * libmeos is not available, each predicate falls back to the corresponding - * pure-Java implementation in {@link Haversine} or {@link SegmentDistance} - * so the BerlinMOD mini-cluster local tests stay runnable on systems - * without a MEOS install. + *
Every spatial predicate exercised by the BerlinMOD-9 × 3-form scaffold + * flows through this class, and every predicate evaluates through MEOS. The + * within-distance predicate is the canonical temporal operator + * {@code edwithin_tgeo_geo} — ever-within between the vehicle's {@code tgeogpoint} + * instant and the query geography, in metres on the WGS84 spheroid — the same + * MEOS operator the streaming-form parity contract names. Distances are + * {@code geog_distance} over WGS84 geographies. This class holds no spatial + * mathematics of its own: it constructs the MEOS inputs and delegates the + * computation to libmeos. * - *
The fallback is gated by the {@link #MEOS_AVAILABLE} static flag, set - * once at class-load time: - *
{@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. */ public final class MEOSBridge { - /** - * {@code true} iff MEOS is available on this runtime and the bridge - * routes through it; {@code false} iff the bridge will use the pure-Java - * fallbacks. - */ - public static final boolean MEOS_AVAILABLE; - - static { - boolean enabled = - Boolean.parseBoolean(System.getProperty("mobilityflink.meos.enabled", "true")); - boolean ok = false; - if (enabled) { - try { - functions.meos_initialize(); - ok = true; - } catch (Throwable t) { - // libmeos shared object not loadable on this runtime — fall back. - ok = false; - } - } - MEOS_AVAILABLE = ok; - } - private MEOSBridge() { // utility } // ---------------------------------------------------------------------- - // Public bridge surface — same shape as Haversine + SegmentDistance. + // Public predicate surface — all evaluation delegated to MEOS. // ---------------------------------------------------------------------- /** - * @return {@code true} if the great-circle distance from {@code (lon1, lat1)} - * to {@code (lon2, lat2)} on the WGS84 spheroid is at most - * {@code radiusMetres}. MEOS-backed via {@code geog_dwithin} when - * available, else pure-Java {@link Haversine#withinMetres}. + * @return {@code true} iff the WGS84 spheroidal distance from + * {@code (lon1, lat1)} to {@code (lon2, lat2)} is at most + * {@code radiusMetres}, via MEOS {@code edwithin_tgeo_geo} between + * the {@code (lon1, lat1)} {@code tgeogpoint} instant and the + * {@code (lon2, lat2)} point geography. */ public static boolean dwithinMetres(double lon1, double lat1, double lon2, double lat2, double radiusMetres) { - if (!MEOS_AVAILABLE) { - return Haversine.withinMetres(lon1, lat1, lon2, lat2, radiusMetres); - } - Pointer g1 = pointGeog(lon1, lat1); - Pointer g2 = pointGeog(lon2, lat2); - if (g1 == null || g2 == null) { - return Haversine.withinMetres(lon1, lat1, lon2, lat2, radiusMetres); - } - return functions.geog_dwithin(g1, g2, radiusMetres, true); + MeosWiringRuntime.ensureInitializedOnThread(); + return GeneratedFunctions.edwithin_tgeo_geo( + tgeogInst(lon1, lat1), pointGeog(lon2, lat2), radiusMetres) == 1; } /** - * @return {@code true} if the spheroidal distance from {@code (pLon, pLat)} - * to the LineString {@code (s1, s2)} is at most {@code radiusMetres}. - * MEOS-backed via {@code geog_dwithin} on geographies built from - * the point and line WKTs, else pure-Java - * {@link SegmentDistance#withinMetres}. + * @return {@code true} iff the WGS84 spheroidal distance from + * {@code (pLon, pLat)} to the LineString {@code (s1, s2)} is at most + * {@code radiusMetres}, via MEOS {@code edwithin_tgeo_geo} between + * the point {@code tgeogpoint} instant and the line geography. */ public static boolean dwithinSegmentMetres(double pLon, double pLat, double s1Lon, double s1Lat, double s2Lon, double s2Lat, double radiusMetres) { - if (!MEOS_AVAILABLE) { - return SegmentDistance.withinMetres(pLon, pLat, s1Lon, s1Lat, s2Lon, s2Lat, radiusMetres); - } - Pointer pg = pointGeog(pLon, pLat); - Pointer lg = lineGeog(s1Lon, s1Lat, s2Lon, s2Lat); - if (pg == null || lg == null) { - return SegmentDistance.withinMetres(pLon, pLat, s1Lon, s1Lat, s2Lon, s2Lat, radiusMetres); - } - return functions.geog_dwithin(pg, lg, radiusMetres, true); + MeosWiringRuntime.ensureInitializedOnThread(); + return GeneratedFunctions.edwithin_tgeo_geo( + tgeogInst(pLon, pLat), + lineGeog(s1Lon, s1Lat, s2Lon, s2Lat), radiusMetres) == 1; } /** - * @return the spheroidal distance in metres between two WGS84 points. - * MEOS-backed via {@code utils.spatial.Haversine.distance} - * (which calls MEOS' {@code geog_distance} over two POINT - * geographies) when libmeos is loadable, else pure-Java - * {@link Haversine#distanceMetres}. + * @return {@code true} iff {@code (lon, lat)} lies in the axis-aligned box + * {@code [xmin, xmax] × [ymin, ymax]}, via MEOS + * {@code eintersects_tgeo_geo} between the point's {@code tgeompoint} + * instant and the box polygon (planar, SRID 4326). + */ + 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; + } + + /** + * @return the WGS84 spheroidal distance in metres between two points, via + * MEOS {@code geog_distance}. */ public static double distanceMetres(double lon1, double lat1, double lon2, double lat2) { - if (!MEOS_AVAILABLE) { - return Haversine.distanceMetres(lon1, lat1, lon2, lat2); - } - return utils.spatial.Haversine.distance(lon1, lat1, lon2, lat2); + MeosWiringRuntime.ensureInitializedOnThread(); + return GeneratedFunctions.geog_distance(pointGeog(lon1, lat1), pointGeog(lon2, lat2)); } /** - * @return the spheroidal distance in metres from {@code (pLon, pLat)} to - * the LineString {@code (s1, s2)}. MEOS-backed via - * {@code utils.spatial.PointToSegment.distance} when libmeos is - * loadable, else pure-Java - * {@link SegmentDistance#distanceMetres}. + * @return the WGS84 spheroidal distance in metres from {@code (pLon, pLat)} + * to the LineString {@code (s1, s2)}, via MEOS {@code geog_distance}. */ public static double distanceSegmentMetres(double pLon, double pLat, double s1Lon, double s1Lat, double s2Lon, double s2Lat) { - if (!MEOS_AVAILABLE) { - return SegmentDistance.distanceMetres(pLon, pLat, s1Lon, s1Lat, s2Lon, s2Lat); - } - return PointToSegment.distance(pLon, pLat, s1Lon, s1Lat, s2Lon, s2Lat); + MeosWiringRuntime.ensureInitializedOnThread(); + return GeneratedFunctions.geog_distance( + pointGeog(pLon, pLat), lineGeog(s1Lon, s1Lat, s2Lon, s2Lat)); } // ---------------------------------------------------------------------- - // Internal helpers — WKT → geometry → geography in one MEOS-side step. + // Internal helpers — construct the MEOS temporal / geography inputs. // ---------------------------------------------------------------------- + private static Pointer tgeogInst(double lon, double lat) { + return GeneratedFunctions.tgeogpoint_in( + String.format("SRID=4326;Point(%.7f %.7f)@2000-01-01", lon, lat)); + } + + private static Pointer tgeomInst(double lon, double lat) { + return GeneratedFunctions.tgeompoint_in( + String.format("SRID=4326;Point(%.7f %.7f)@2000-01-01", lon, lat)); + } + + private static Pointer boxPolygon(double xmin, double ymin, + double xmax, double ymax) { + return GeneratedFunctions.geom_in(String.format( + "SRID=4326;Polygon((%.7f %.7f, %.7f %.7f, %.7f %.7f, %.7f %.7f, %.7f %.7f))", + xmin, ymin, xmax, ymin, xmax, ymax, xmin, ymax, xmin, ymin), -1); + } + private static Pointer pointGeog(double lon, double lat) { - String wkt = String.format("SRID=4326;Point(%.7f %.7f)", lon, lat); - Pointer g = functions.geom_in(wkt, -1); - if (g == null) { - return null; - } - return functions.geom_to_geog(g); + return GeneratedFunctions.geom_to_geog( + GeneratedFunctions.geom_in(String.format("SRID=4326;Point(%.7f %.7f)", lon, lat), -1)); } private static Pointer lineGeog(double s1Lon, double s1Lat, double s2Lon, double s2Lat) { - String wkt = String.format("SRID=4326;LineString(%.7f %.7f, %.7f %.7f)", - s1Lon, s1Lat, s2Lon, s2Lat); - Pointer g = functions.geom_in(wkt, -1); - if (g == null) { - return null; - } - return functions.geom_to_geog(g); + return GeneratedFunctions.geom_to_geog( + GeneratedFunctions.geom_in(String.format( + "SRID=4326;LineString(%.7f %.7f, %.7f %.7f)", s1Lon, s1Lat, s2Lon, s2Lat), -1)); } } diff --git a/flink-processor/src/main/java/berlinmod/Q3ContinuousFunction.java b/flink-processor/src/main/java/berlinmod/Q3ContinuousFunction.java index 6628d05..e28f825 100644 --- a/flink-processor/src/main/java/berlinmod/Q3ContinuousFunction.java +++ b/flink-processor/src/main/java/berlinmod/Q3ContinuousFunction.java @@ -16,9 +16,8 @@ * predicate and emit {@code (vehicleId, eventTimeMillis, isNear)} per event. * No windowing — output updates per event, watermark-independent. * - *
Predicate: {@link MEOSBridge#dwithinMetres} — MEOS' {@code geog_dwithin} - * over WGS84 geographies when libmeos is loadable, with a pure-Java great-circle - * fallback ({@link Haversine}) for runtimes without MEOS. + *
Predicate: {@link MEOSBridge#dwithinMetres} — MEOS
+ * {@code edwithin_tgeo_geo} over WGS84 geographies.
*/
public class Q3ContinuousFunction extends ProcessFunction Predicate: {@link MEOSBridge#dwithinMetres} — MEOS {@code geog_dwithin}
- * when libmeos is loadable, with {@link Haversine} fallback otherwise. The
- * snapshot-form output at watermark T is equal to the batch BerlinMOD-Q3
- * result up to T regardless of which path is active.
+ * Predicate: {@link MEOSBridge#dwithinMetres} — MEOS
+ * {@code edwithin_tgeo_geo} over WGS84 geographies. The snapshot-form output
+ * at watermark T is equal to the batch BerlinMOD-Q3 result up to T.
*/
public class Q3SnapshotFunction
extends KeyedProcessFunction Predicate: {@link MEOSBridge#dwithinMetres} — MEOS {@code geog_dwithin}
- * over WGS84 geographies when libmeos is loadable, with {@link Haversine}
- * fallback otherwise.
+ * Predicate: {@link MEOSBridge#dwithinMetres} — MEOS
+ * {@code edwithin_tgeo_geo} over WGS84 geographies.
*/
public class Q3WindowedFunction
extends ProcessAllWindowFunction Predicate: pure-Java axis-aligned point-in-box. The rectangular region
- * is degenerate as a geographic predicate (no projection needed); a generic
- * polygon-R variant would route through {@link MEOSBridge} for MEOS
- * {@code eintersects_tgeo_geo}.
+ * Predicate: {@link MEOSBridge#intersectsBox} — MEOS
+ * {@code eintersects_tgeo_geo} between the point's {@code tgeompoint} instant
+ * and the region polygon.
*/
public class Q4ContinuousFunction
extends KeyedProcessFunction Predicate: {@link MEOSBridge#dwithinMetres} for the near-P filter and
- * for the pairwise meeting predicate — MEOS {@code geog_dwithin} when
- * libmeos is loadable, with {@link Haversine} fallback otherwise.
+ * Predicate: {@link MEOSBridge#dwithinMetres} (MEOS
+ * {@code edwithin_tgeo_geo}) for the near-P filter and
+ * {@link MEOSBridge#distanceMetres} (MEOS {@code geog_distance}) for the
+ * pairwise meeting distance.
*/
public class Q5ContinuousFunction
extends KeyedProcessFunction "What is each vehicle's cumulative distance travelled so far?"
*
- * Keyed by vehicleId. For each event, computes the great-circle distance
- * from the previous-known position (or 0 if first event), adds it to the
- * cumulative total, and emits {@code (vehicleId, t, cumulativeMetres)}.
+ * Keyed by vehicleId. For each event, computes the distance from the
+ * previous-known position (or 0 if first event), adds it to the cumulative
+ * total, and emits {@code (vehicleId, t, cumulativeMetres)}.
*
- * Predicate today: pure-Java great-circle distance (see {@link Haversine}).
- * Same MEOS-side analogue as Q3 — a future JMEOS bridge would replace the
- * Java accumulator with a MEOS {@code length} call over the per-vehicle
- * trajectory.
+ * Distance: {@link MEOSBridge#distanceMetres} — MEOS {@code geog_distance}
+ * between consecutive WGS84 positions.
*/
public class Q6ContinuousFunction
extends KeyedProcessFunction Predicate: {@link MEOSBridge#dwithinSegmentMetres} — MEOS
- * {@code geog_dwithin} against a LineString geography when libmeos is
- * loadable, with {@link SegmentDistance} fallback otherwise.
+ * {@code edwithin_tgeo_geo} against a LineString geography.
*/
public class Q8ContinuousFunction extends ProcessFunction Pure-Java fallback for {@link MEOSBridge#dwithinSegmentMetres}, used
- * by the BerlinMOD-Q8 streaming scaffold when libmeos is not loadable on
- * the runtime path. The primary point-to-line spatial predicate is
- * {@link MEOSBridge#dwithinSegmentMetres}, which routes through MEOS'
- * {@code geog_dwithin} on a LineString geography when available.
- */
-public final class SegmentDistance {
-
- private static final double EARTH_RADIUS_METRES = 6_371_000.0;
-
- private SegmentDistance() {
- // utility
- }
-
- /**
- * @return distance in metres from point (pLon, pLat) to the line segment
- * from (s1Lon, s1Lat) to (s2Lon, s2Lat)
- */
- public static double distanceMetres(
- double pLon, double pLat,
- double s1Lon, double s1Lat,
- double s2Lon, double s2Lat) {
- // Local equirectangular projection centred on the segment midpoint
- double midLat = (s1Lat + s2Lat) / 2.0;
- double mPerDegLat = Math.toRadians(1.0) * EARTH_RADIUS_METRES;
- double mPerDegLon = mPerDegLat * Math.cos(Math.toRadians(midLat));
-
- double px = pLon * mPerDegLon;
- double py = pLat * mPerDegLat;
- double s1x = s1Lon * mPerDegLon;
- double s1y = s1Lat * mPerDegLat;
- double s2x = s2Lon * mPerDegLon;
- double s2y = s2Lat * mPerDegLat;
-
- double dx = s2x - s1x;
- double dy = s2y - s1y;
- double lenSq = dx * dx + dy * dy;
- if (lenSq == 0.0) {
- // Degenerate segment — point to endpoint distance
- return Math.hypot(px - s1x, py - s1y);
- }
- double t = ((px - s1x) * dx + (py - s1y) * dy) / lenSq;
- if (t < 0.0) {
- t = 0.0;
- } else if (t > 1.0) {
- t = 1.0;
- }
- double cx = s1x + t * dx;
- double cy = s1y + t * dy;
- return Math.hypot(px - cx, py - cy);
- }
-
- /**
- * @return true if the distance from (pLon, pLat) to the segment
- * (s1, s2) is ≤ {@code radiusMetres}
- */
- public static boolean withinMetres(
- double pLon, double pLat,
- double s1Lon, double s1Lat,
- double s2Lon, double s2Lat,
- double radiusMetres) {
- return distanceMetres(pLon, pLat, s1Lon, s1Lat, s2Lon, s2Lat) <= radiusMetres;
- }
-}