From f8141d2cf7161e21bee38c44f0b9256302876728 Mon Sep 17 00:00:00 2001 From: Esteban Zimanyi Date: Thu, 28 May 2026 02:31:13 +0200 Subject: [PATCH 1/2] docs(parity): record the JMEOS-jar / MEOS-library resolution skew Section 5 of parity-status.md reports facade-method resolution against a built MEOS library (2277 of 2286) and splits the 9 unresolved into the seven symbols declared in the public headers that the build does not export and the two functions the JMEOS jar carries ahead of the current public headers. (cherry picked from commit 8972cafe961bf9cbb155c1b967227a9850c382ab) --- flink-processor/docs/parity-status.md | 16 ++++++-------- flink-processor/tools/parity/parity_audit.py | 22 +++++++++++++++----- 2 files changed, 23 insertions(+), 15 deletions(-) diff --git a/flink-processor/docs/parity-status.md b/flink-processor/docs/parity-status.md index 570226f..e26eba2 100644 --- a/flink-processor/docs/parity-status.md +++ b/flink-processor/docs/parity-status.md @@ -45,14 +45,10 @@ The facade is also matched against the underlying MEOS C symbol of each addressa ## 5. Runtime symbol resolution -Every facade method delegates to a libmeos symbol of the same name. Against a libmeos built with the extended modules (`-DCBUFFER=ON -DNPOINT=ON -DPOSE=ON -DRGEO=ON`), **2278 of 2286** facade methods resolve to an exported symbol. The following require a libmeos built from current MEOS sources: - -- `geog_from_binary` -- `nad_stbox_trgeo` -- `tcbuffer_from_mfjson` -- `tfloat_avg_value` -- `tnpoint_from_mfjson` -- `trgeo_points` -- `trgeo_rotation` -- `trgeo_segments` +Every facade method delegates to a libmeos symbol of the same name. Against a MEOS shared library built with the extended modules (`-DCBUFFER=ON -DNPOINT=ON -DPOSE=ON -DRGEO=ON`), **2277 of 2286** facade methods resolve to an exported symbol. + +The remaining 9 are present in the JMEOS jar but not exported by the MEOS shared library (a JMEOS-jar / library version skew): + +- declared in the public headers, not exported by this build (7): `geog_from_binary`, `nad_stbox_trgeo`, `tfloat_avg_value`, `trgeo_points`, `trgeo_rotation`, `trgeo_segments`, `trgeo_traversed_area` +- not declared in the current public headers, JMEOS jar ahead of the library (2): `tcbuffer_from_mfjson`, `tnpoint_from_mfjson` diff --git a/flink-processor/tools/parity/parity_audit.py b/flink-processor/tools/parity/parity_audit.py index 8e6b6c7..263d507 100644 --- a/flink-processor/tools/parity/parity_audit.py +++ b/flink-processor/tools/parity/parity_audit.py @@ -195,12 +195,24 @@ def main(): unresolved = sorted(fa_all - libsyms) L.append("## 5. Runtime symbol resolution\n") L.append("Every facade method delegates to a libmeos symbol of the same name. Against a " - "libmeos built with the extended modules (`-DCBUFFER=ON -DNPOINT=ON -DPOSE=ON " - "-DRGEO=ON`), " + "MEOS shared library built with the extended modules (`-DCBUFFER=ON -DNPOINT=ON " + "-DPOSE=ON -DRGEO=ON`), " f"**{len(resolved)} of {len(fa_all)}** facade methods resolve to an exported " - "symbol. The following require a libmeos built from current MEOS sources:\n") - L.append(("\n".join(f"- `{n}`" for n in unresolved) if unresolved - else "- (none — all facade methods resolve)") + "\n") + "symbol.\n") + if unresolved: + hdr_only = sorted(n for n in unresolved if n in pub) + jmeos_only = sorted(n for n in unresolved if n not in pub) + L.append(f"The remaining {len(unresolved)} are present in the JMEOS jar but not " + "exported by the MEOS shared library (a JMEOS-jar / library version skew):\n") + if hdr_only: + L.append(f"- declared in the public headers, not exported by this build " + f"({len(hdr_only)}): " + ", ".join(f"`{n}`" for n in hdr_only)) + if jmeos_only: + L.append(f"- not declared in the current public headers, JMEOS jar ahead of the " + f"library ({len(jmeos_only)}): " + ", ".join(f"`{n}`" for n in jmeos_only)) + L.append("") + else: + L.append("All facade methods resolve.\n") print(f"libmeos resolution: {len(resolved)}/{len(fa_all)} " f"({len(unresolved)} unresolved)") From 02524dbd622638eae2d14625e6e4be3c89c324e0 Mon Sep 17 00:00:00 2001 From: Esteban Zimanyi Date: Sat, 6 Jun 2026 09:23:51 +0200 Subject: [PATCH 2/2] Remove the pure-Java great-circle fallback; BerlinMOD predicates are MEOS-only The berlinmod Haversine and SegmentDistance classes computed a planar/great-circle distance that diverges from MEOS geog_distance (4124.39 vs 4132.97 m on the parity corpus) and were selected silently when libmeos failed to load, so the parity-oracle suite validated against a non-canonical distance and a missing libmeos produced a wrong number rather than an error. MEOSBridge now evaluates every spatial predicate through MEOS geog_dwithin / geog_distance with no fallback and fails loudly at load when libmeos is absent, matching the MeosOps surface. The BerlinMOD local drivers run on MEOS. The distance cells (Q5/Q6/Q9) now equal the MEOS value, matching the Kafka Streams parity row. --- .../java/berlinmod/BerlinMODQ1LocalTest.java | 1 - .../java/berlinmod/BerlinMODQ2LocalTest.java | 1 - .../java/berlinmod/BerlinMODQ3LocalTest.java | 1 - .../java/berlinmod/BerlinMODQ4LocalTest.java | 1 - .../java/berlinmod/BerlinMODQ5LocalTest.java | 1 - .../java/berlinmod/BerlinMODQ6LocalTest.java | 1 - .../java/berlinmod/BerlinMODQ7LocalTest.java | 1 - .../java/berlinmod/BerlinMODQ8LocalTest.java | 1 - .../java/berlinmod/BerlinMODQ9LocalTest.java | 1 - .../src/main/java/berlinmod/Haversine.java | 44 ------- .../src/main/java/berlinmod/MEOSBridge.java | 121 +++++------------- .../java/berlinmod/Q3ContinuousFunction.java | 2 +- .../java/berlinmod/Q3SnapshotFunction.java | 2 +- .../java/berlinmod/Q3WindowedFunction.java | 2 +- .../java/berlinmod/Q5ContinuousFunction.java | 2 +- .../java/berlinmod/Q6ContinuousFunction.java | 2 +- .../java/berlinmod/Q8ContinuousFunction.java | 2 +- .../main/java/berlinmod/SegmentDistance.java | 70 ---------- 18 files changed, 38 insertions(+), 218 deletions(-) delete mode 100644 flink-processor/src/main/java/berlinmod/Haversine.java delete mode 100644 flink-processor/src/main/java/berlinmod/SegmentDistance.java diff --git a/flink-processor/src/main/java/berlinmod/BerlinMODQ1LocalTest.java b/flink-processor/src/main/java/berlinmod/BerlinMODQ1LocalTest.java index 7b609b0..9218e16 100644 --- a/flink-processor/src/main/java/berlinmod/BerlinMODQ1LocalTest.java +++ b/flink-processor/src/main/java/berlinmod/BerlinMODQ1LocalTest.java @@ -38,7 +38,6 @@ public class BerlinMODQ1LocalTest { private static final long T0 = 1_735_711_200_000L; public static void main(String[] args) throws Exception { - System.setProperty("mobilityflink.meos.enabled", "false"); LOG.info("BerlinMODQ1LocalTest starting; window={}s tick={}ms", WINDOW_SIZE_SECONDS, SNAPSHOT_TICK_MILLIS); diff --git a/flink-processor/src/main/java/berlinmod/BerlinMODQ2LocalTest.java b/flink-processor/src/main/java/berlinmod/BerlinMODQ2LocalTest.java index 9446c30..84950c9 100644 --- a/flink-processor/src/main/java/berlinmod/BerlinMODQ2LocalTest.java +++ b/flink-processor/src/main/java/berlinmod/BerlinMODQ2LocalTest.java @@ -44,7 +44,6 @@ public class BerlinMODQ2LocalTest { private static final long T0 = 1_735_711_200_000L; // 2025-01-01 06:00:00 UTC public static void main(String[] args) throws Exception { - System.setProperty("mobilityflink.meos.enabled", "false"); LOG.info("BerlinMODQ2LocalTest starting; X={} window={}s tick={}ms", TARGET_VEHICLE_ID, WINDOW_SIZE_SECONDS, SNAPSHOT_TICK_MILLIS); diff --git a/flink-processor/src/main/java/berlinmod/BerlinMODQ3LocalTest.java b/flink-processor/src/main/java/berlinmod/BerlinMODQ3LocalTest.java index e1b7128..69e2022 100644 --- a/flink-processor/src/main/java/berlinmod/BerlinMODQ3LocalTest.java +++ b/flink-processor/src/main/java/berlinmod/BerlinMODQ3LocalTest.java @@ -55,7 +55,6 @@ public class BerlinMODQ3LocalTest { private static final long T0 = 1_735_711_200_000L; // 2025-01-01 06:00:00 UTC public static void main(String[] args) throws Exception { - System.setProperty("mobilityflink.meos.enabled", "false"); LOG.info("BerlinMODQ3LocalTest starting; P=({}, {}) radius={}m window={}s tick={}ms", P_LON, P_LAT, RADIUS_METRES, WINDOW_SIZE_SECONDS, SNAPSHOT_TICK_MILLIS); diff --git a/flink-processor/src/main/java/berlinmod/BerlinMODQ4LocalTest.java b/flink-processor/src/main/java/berlinmod/BerlinMODQ4LocalTest.java index 1f302a5..428e7fc 100644 --- a/flink-processor/src/main/java/berlinmod/BerlinMODQ4LocalTest.java +++ b/flink-processor/src/main/java/berlinmod/BerlinMODQ4LocalTest.java @@ -60,7 +60,6 @@ public class BerlinMODQ4LocalTest { private static final long T0 = 1_735_711_200_000L; public static void main(String[] args) throws Exception { - System.setProperty("mobilityflink.meos.enabled", "false"); LOG.info("BerlinMODQ4LocalTest starting; R=({},{},{},{}) window={}s tick={}ms", XMIN, YMIN, XMAX, YMAX, WINDOW_SIZE_SECONDS, SNAPSHOT_TICK_MILLIS); diff --git a/flink-processor/src/main/java/berlinmod/BerlinMODQ5LocalTest.java b/flink-processor/src/main/java/berlinmod/BerlinMODQ5LocalTest.java index f54e102..394932d 100644 --- a/flink-processor/src/main/java/berlinmod/BerlinMODQ5LocalTest.java +++ b/flink-processor/src/main/java/berlinmod/BerlinMODQ5LocalTest.java @@ -52,7 +52,6 @@ public class BerlinMODQ5LocalTest { private static final long T0 = 1_735_711_200_000L; public static void main(String[] args) throws Exception { - System.setProperty("mobilityflink.meos.enabled", "false"); LOG.info("BerlinMODQ5LocalTest starting; P=({}, {}) dP={}m dMeet={}m", P_LON, P_LAT, D_P_METRES, D_MEET_METRES); diff --git a/flink-processor/src/main/java/berlinmod/BerlinMODQ6LocalTest.java b/flink-processor/src/main/java/berlinmod/BerlinMODQ6LocalTest.java index 4c18eb5..6a6fd2b 100644 --- a/flink-processor/src/main/java/berlinmod/BerlinMODQ6LocalTest.java +++ b/flink-processor/src/main/java/berlinmod/BerlinMODQ6LocalTest.java @@ -60,7 +60,6 @@ public class BerlinMODQ6LocalTest { private static final double V300_DLON = -200.0 / (111_000.0 * Math.cos(Math.toRadians(50.85))); public static void main(String[] args) throws Exception { - System.setProperty("mobilityflink.meos.enabled", "false"); LOG.info("BerlinMODQ6LocalTest starting; window={}s tick={}ms", WINDOW_SIZE_SECONDS, SNAPSHOT_TICK_MILLIS); diff --git a/flink-processor/src/main/java/berlinmod/BerlinMODQ7LocalTest.java b/flink-processor/src/main/java/berlinmod/BerlinMODQ7LocalTest.java index fe9f075..e7560c8 100644 --- a/flink-processor/src/main/java/berlinmod/BerlinMODQ7LocalTest.java +++ b/flink-processor/src/main/java/berlinmod/BerlinMODQ7LocalTest.java @@ -59,7 +59,6 @@ public class BerlinMODQ7LocalTest { new PointOfInterest(3, 4.2100, 50.7600, 2_000.0)); public static void main(String[] args) throws Exception { - System.setProperty("mobilityflink.meos.enabled", "false"); LOG.info("BerlinMODQ7LocalTest starting; #POIs={} window={}s tick={}ms", POIS.size(), WINDOW_SIZE_SECONDS, SNAPSHOT_TICK_MILLIS); diff --git a/flink-processor/src/main/java/berlinmod/BerlinMODQ8LocalTest.java b/flink-processor/src/main/java/berlinmod/BerlinMODQ8LocalTest.java index c644860..9dc6709 100644 --- a/flink-processor/src/main/java/berlinmod/BerlinMODQ8LocalTest.java +++ b/flink-processor/src/main/java/berlinmod/BerlinMODQ8LocalTest.java @@ -51,7 +51,6 @@ public class BerlinMODQ8LocalTest { private static final long T0 = 1_735_711_200_000L; public static void main(String[] args) throws Exception { - System.setProperty("mobilityflink.meos.enabled", "false"); LOG.info("BerlinMODQ8LocalTest starting; segment=({},{}) → ({},{}) d={}m", S1_LON, S1_LAT, S2_LON, S2_LAT, RADIUS_METRES); diff --git a/flink-processor/src/main/java/berlinmod/BerlinMODQ9LocalTest.java b/flink-processor/src/main/java/berlinmod/BerlinMODQ9LocalTest.java index 6468ba7..f990031 100644 --- a/flink-processor/src/main/java/berlinmod/BerlinMODQ9LocalTest.java +++ b/flink-processor/src/main/java/berlinmod/BerlinMODQ9LocalTest.java @@ -45,7 +45,6 @@ public class BerlinMODQ9LocalTest { private static final long T0 = 1_735_711_200_000L; public static void main(String[] args) throws Exception { - System.setProperty("mobilityflink.meos.enabled", "false"); LOG.info("BerlinMODQ9LocalTest starting; X={} Y={} window={}s tick={}ms", X_VEHICLE_ID, Y_VEHICLE_ID, WINDOW_SIZE_SECONDS, SNAPSHOT_TICK_MILLIS); diff --git a/flink-processor/src/main/java/berlinmod/Haversine.java b/flink-processor/src/main/java/berlinmod/Haversine.java deleted file mode 100644 index cb6f888..0000000 --- a/flink-processor/src/main/java/berlinmod/Haversine.java +++ /dev/null @@ -1,44 +0,0 @@ -package berlinmod; - -/** - * Great-circle distance in metres between two WGS84 (lon, lat) points. - * - *

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..4c7f3ce 100644 --- a/flink-processor/src/main/java/berlinmod/MEOSBridge.java +++ b/flink-processor/src/main/java/berlinmod/MEOSBridge.java @@ -8,47 +8,25 @@ * Runtime bridge from MobilityFlink 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. - * - *

The fallback is gated by the {@link #MEOS_AVAILABLE} static flag, set - * once at class-load time: - *

+ *

All spatial predicates exercised by the BerlinMOD-9 × 3-form scaffold flow + * through this class and evaluate through MEOS' WGS84 geography surface + * ({@code geom_to_geog} + {@code geog_dwithin}/{@code geog_distance}). There is + * no pure-Java approximation: a hand-rolled great-circle would diverge from + * MEOS' spheroidal result and could silently mask a missing libmeos, so the + * bridge requires MEOS and fails loudly when it is absent — exactly as the + * {@code MeosOps*} surface does ({@code "requires libmeos — set + * -Dmobilityflink.meos.enabled=true"}). */ 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; - } + try { + functions.meos_initialize(); + } catch (Throwable t) { + throw new IllegalStateException( + "MEOSBridge requires libmeos: its spatial predicates are MEOS geog_dwithin/" + + "geog_distance with no pure-Java fallback. Put libmeos on java.library.path.", t); } - MEOS_AVAILABLE = ok; } private MEOSBridge() { @@ -56,79 +34,52 @@ private MEOSBridge() { } // ---------------------------------------------------------------------- - // Public bridge surface — same shape as Haversine + SegmentDistance. + // Public bridge surface — MEOS geography predicates, no fallback. // ---------------------------------------------------------------------- /** - * @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} if the WGS84 spheroidal distance from + * {@code (lon1, lat1)} to {@code (lon2, lat2)} is at most + * {@code radiusMetres}, via MEOS {@code geog_dwithin}. */ 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); + return functions.geog_dwithin(pointGeog(lon1, lat1), pointGeog(lon2, lat2), radiusMetres, true); } /** * @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}. + * to the LineString {@code (s1, s2)} is at most {@code radiusMetres}, + * via MEOS {@code geog_dwithin} on geographies built from the point + * and line WKTs. */ 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); + return functions.geog_dwithin( + pointGeog(pLon, pLat), lineGeog(s1Lon, s1Lat, s2Lon, s2Lat), radiusMetres, true); } /** - * @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 the WGS84 spheroidal distance in metres between two points, via + * {@code utils.spatial.Haversine.distance} (MEOS {@code geog_distance} + * over two POINT geographies). */ 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); } /** - * @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 spheroidal distance in metres from {@code (pLon, pLat)} to the + * LineString {@code (s1, s2)}, via {@code utils.spatial.PointToSegment.distance} + * (MEOS {@code geog_distance} over POINT/LINESTRING geographies). */ 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); } @@ -138,21 +89,13 @@ public static double distanceSegmentMetres(double pLon, double pLat, 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 functions.geom_to_geog(functions.geom_in(wkt, -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 functions.geom_to_geog(functions.geom_in(wkt, -1)); } } diff --git a/flink-processor/src/main/java/berlinmod/Q3ContinuousFunction.java b/flink-processor/src/main/java/berlinmod/Q3ContinuousFunction.java index 6628d05..9b3c12a 100644 --- a/flink-processor/src/main/java/berlinmod/Q3ContinuousFunction.java +++ b/flink-processor/src/main/java/berlinmod/Q3ContinuousFunction.java @@ -18,7 +18,7 @@ * *

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. + * fallback (MEOS geog_dwithin/geog_distance) for runtimes without MEOS. */ public class Q3ContinuousFunction extends ProcessFunction> { diff --git a/flink-processor/src/main/java/berlinmod/Q3SnapshotFunction.java b/flink-processor/src/main/java/berlinmod/Q3SnapshotFunction.java index 479854d..8c600b7 100644 --- a/flink-processor/src/main/java/berlinmod/Q3SnapshotFunction.java +++ b/flink-processor/src/main/java/berlinmod/Q3SnapshotFunction.java @@ -28,7 +28,7 @@ * snapshot. * *

Predicate: {@link MEOSBridge#dwithinMetres} — MEOS {@code geog_dwithin} - * when libmeos is loadable, with {@link Haversine} fallback otherwise. The + * when libmeos is loadable, with MEOS geog_dwithin/geog_distance 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. */ diff --git a/flink-processor/src/main/java/berlinmod/Q3WindowedFunction.java b/flink-processor/src/main/java/berlinmod/Q3WindowedFunction.java index c6dfa98..8c5e984 100644 --- a/flink-processor/src/main/java/berlinmod/Q3WindowedFunction.java +++ b/flink-processor/src/main/java/berlinmod/Q3WindowedFunction.java @@ -22,7 +22,7 @@ * {@code (windowStart, windowEnd, distinctCount)}. * *

Predicate: {@link MEOSBridge#dwithinMetres} — MEOS {@code geog_dwithin} - * over WGS84 geographies when libmeos is loadable, with {@link Haversine} + * over WGS84 geographies when libmeos is loadable, with MEOS geog_dwithin/geog_distance * fallback otherwise. */ public class Q3WindowedFunction diff --git a/flink-processor/src/main/java/berlinmod/Q5ContinuousFunction.java b/flink-processor/src/main/java/berlinmod/Q5ContinuousFunction.java index d14f272..40e3b52 100644 --- a/flink-processor/src/main/java/berlinmod/Q5ContinuousFunction.java +++ b/flink-processor/src/main/java/berlinmod/Q5ContinuousFunction.java @@ -33,7 +33,7 @@ * *

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. + * libmeos is loadable, with MEOS geog_dwithin/geog_distance fallback otherwise. */ public class Q5ContinuousFunction extends KeyedProcessFunction> { diff --git a/flink-processor/src/main/java/berlinmod/Q6ContinuousFunction.java b/flink-processor/src/main/java/berlinmod/Q6ContinuousFunction.java index 85204f3..77efd4a 100644 --- a/flink-processor/src/main/java/berlinmod/Q6ContinuousFunction.java +++ b/flink-processor/src/main/java/berlinmod/Q6ContinuousFunction.java @@ -19,7 +19,7 @@ * 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}). + *

Predicate today: pure-Java great-circle distance (see MEOS geog_dwithin/geog_distance). * 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. diff --git a/flink-processor/src/main/java/berlinmod/Q8ContinuousFunction.java b/flink-processor/src/main/java/berlinmod/Q8ContinuousFunction.java index fac4f1a..6c510d0 100644 --- a/flink-processor/src/main/java/berlinmod/Q8ContinuousFunction.java +++ b/flink-processor/src/main/java/berlinmod/Q8ContinuousFunction.java @@ -17,7 +17,7 @@ * *

Predicate: {@link MEOSBridge#dwithinSegmentMetres} — MEOS * {@code geog_dwithin} against a LineString geography when libmeos is - * loadable, with {@link SegmentDistance} fallback otherwise. + * loadable, with MEOS geog_dwithin/geog_distance fallback otherwise. */ public class Q8ContinuousFunction extends ProcessFunction> { diff --git a/flink-processor/src/main/java/berlinmod/SegmentDistance.java b/flink-processor/src/main/java/berlinmod/SegmentDistance.java deleted file mode 100644 index 62db56b..0000000 --- a/flink-processor/src/main/java/berlinmod/SegmentDistance.java +++ /dev/null @@ -1,70 +0,0 @@ -package berlinmod; - -/** - * Distance from a (lon, lat) point to a (lon, lat) line segment, in metres, - * via a local equirectangular projection centred on the segment midpoint. - * - *

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; - } -}