feat(berlinmod): route the streaming-form spatial predicates through MEOS

Every BerlinMOD-9 × 3-form spatial predicate evaluates through MEOS via the
thin MEOSBridge wiring rather than in-binding spatial mathematics: the
within-distance predicate builds the vehicle position as a tgeogpoint instant
and calls edwithin_tgeo_geo against the query geography in metres on the WGS84
spheroid (the operator MobilityNebula/Queries/Query1.yaml names); region
containment calls eintersects_tgeo_geo between the point's tgeompoint instant
and the region polygon; pairwise and cumulative distances call geog_distance.
MEOSBridge holds no spatial mathematics of its own — it constructs the MEOS
temporal instants and geographies and delegates the computation to libmeos,
initialising MEOS on the calling task thread. The pure-Java Haversine and
SegmentDistance classes are removed.

Author: estebanzimanyi

doc/berlinmod-q3-streaming-forms.md

@@ -89,13 +89,13 @@ The `BerlinMODQ3Main` entry point uses:
 
 ## Predicate implementation
 
-The scaffold today uses a pure-Java great-circle (Haversine) distance check in
-[`Haversine`](../flink-processor/src/main/java/berlinmod/Haversine.java). This
-matches the predicate semantics of the MEOS `edwithin_tgeo_geo` operator (the
-same call used by `MobilityNebula/Queries/Query1.yaml`), so swapping the
-predicate body for a JMEOS-bridged `edwithin_tgeo_geo` call is a one-line
-change once the JMEOS surface for that operator is verified — it is marked
-`TODO(meos)` in each form's class.
+The within-distance predicate evaluates through the MEOS `edwithin_tgeo_geo`
+operator — the same call used by `MobilityNebula/Queries/Query1.yaml`. The
+vehicle position is built as a `tgeogpoint` instant and tested against the
+query geography in metres on the WGS84 spheroid. All spatial predicates route
+through [`MEOSBridge`](../flink-processor/src/main/java/berlinmod/MEOSBridge.java),
+which holds no spatial mathematics of its own: it constructs the MEOS inputs
+(temporal instants and geographies) and delegates the computation to libmeos.
 
 ## Companion producer
 

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

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

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

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

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

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

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

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

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