feat(wirings): cross-stream tier DataStream wiring + demo — completes the 4-tier matrix (stacks on #8)

flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosCrossStreamJoin.java

@@ -0,0 +1,115 @@
+package org.mobilitydb.flink.meos.wirings;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
+import org.apache.flink.util.Collector;
+
+import java.io.Serializable;
+
+/**
+ * DataStream wiring for the {@code cross-stream} streaming tier of
+ * the generated {@code org.mobilitydb.flink.meos.MeosOps*} facades.
+ *
+ * <p>The {@code cross-stream} tier is "pairwise across two streams,
+ * pre-keyed by the same K, time-bounded match window". Canonical
+ * examples are spatial-relations between two trajectories
+ * ({@code edwithin_tgeo_tgeo}, {@code eintersects_tgeo_tgeo}) and
+ * distance functions on two temporals
+ * ({@code nad_tgeo_tgeo}, {@code mindistance_tgeo_tgeo}).
+ *
+ * <p>Wraps any cross-stream MeosOps call as a Flink
+ * {@link ProcessJoinFunction} (the operator backing
+ * {@code KeyedStream.intervalJoin(other)}). The wiring receives one
+ * left event and one right event per match, both already paired by
+ * Flink's interval-join machinery, and the adopter's lambda computes
+ * the pairwise output via the matching MeosOps call.
+ *
+ * <p><b>Typical usage</b> — per-vehicle-pair "did they come within
+ * 100m of each other in the last 5 minutes?" via
+ * {@code MeosOpsTGeo.edwithin_tgeo_tgeo} (tier = {@code cross-stream}):
+ *
+ * <pre>{@code
+ * KeyedStream<VehiclePosition, Integer> a = streamA.keyBy(VehiclePosition::regionId);
+ * KeyedStream<VehiclePosition, Integer> b = streamB.keyBy(VehiclePosition::regionId);
+ *
+ * DataStream<MeetingEvent> meetings = a
+ *     .intervalJoin(b)
+ *         .between(Time.minutes(-5), Time.minutes(5))
+ *         .process(new MeosCrossStreamJoin<VehiclePosition, VehiclePosition, MeetingEvent>(
+ *             (left, right, ctx) -> {
+ *                 Pointer leftT  = left.toTGeoPointer();
+ *                 Pointer rightT = right.toTGeoPointer();
+ *                 if (MeosOpsTGeo.edwithin_tgeo_tgeo(leftT, rightT, 100.0) != 0) {
+ *                     return new MeetingEvent(left.id(), right.id(), ctx.getLeftTimestamp());
+ *                 }
+ *                 return null;  // no output for non-matches
+ *             }));
+ * }</pre>
+ *
+ * <p>The interval-join is keyed (both streams must be pre-keyed by
+ * the same K, and only events sharing a key are considered for
+ * pairing). The match window is time-bounded
+ * ({@code .between(lowerBound, upperBound)}) and event-time aware —
+ * watermarks drive when matches are emitted.
+ *
+ * <p><b>Slim adopter signature</b> — same {@code ContextLike}-style
+ * pattern as {@link MeosWindowedAggregate}: the lambda receives the
+ * matched left + right events and a slim context exposing the
+ * left/right timestamps (the bits a MEOS cross-stream call typically
+ * needs), keeping the wiring lambda free of Flink internals.
+ *
+ * <p><b>Coverage</b>: 140 of the 2,097 emitted methods (~7%) qualify
+ * as {@code cross-stream} per the v4 baseline — all of them wrappable
+ * through this single class. With this PR, every streamable tier in
+ * the baseline has a generic wiring class; 1,957 of 2,097 (93%) of
+ * the generated MeosOps* methods are wirable through 4 classes
+ * without per-method registration.
+ *
+ * @param <L>   the left-stream event type
+ * @param <R>   the right-stream event type
+ * @param <OUT> the per-match output type
+ */
+public final class MeosCrossStreamJoin<L, R, OUT>
+        extends ProcessJoinFunction<L, R, OUT> {
+
+    /** Serializable per-match MEOS pairwise call. */
+    @FunctionalInterface
+    public interface JoinFn<L, R, OUT> extends Serializable {
+        OUT join(L left, R right, ContextLike ctx) throws Exception;
+    }
+
+    /**
+     * Slimmer alternative to Flink's {@code ProcessJoinFunction.Context}
+     * — exposes only the bits a MEOS pairwise call typically needs.
+     */
+    public interface ContextLike {
+        long getLeftTimestamp();
+        long getRightTimestamp();
+    }
+
+    private final JoinFn<L, R, OUT> joinFn;
+
+    public MeosCrossStreamJoin(JoinFn<L, R, OUT> joinFn) {
+        this.joinFn = joinFn;
+    }
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        MeosWiringRuntime.ensureInitializedOnThread();
+    }
+
+    @Override
+    public void processElement(L left, R right,
+                                ProcessJoinFunction<L, R, OUT>.Context context,
+                                Collector<OUT> out) throws Exception {
+        ContextLike ctx = new ContextLike() {
+            @Override public long getLeftTimestamp() { return context.getLeftTimestamp(); }
+            @Override public long getRightTimestamp() { return context.getRightTimestamp(); }
+        };
+        OUT result = joinFn.join(left, right, ctx);
+        if (result != null) {
+            out.collect(result);
+        }
+    }
+}

flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/README.md

@@ -10,7 +10,7 @@ per **streaming tier** (per
 | `stateless` | [`MeosStatelessMap`](MeosStatelessMap.java) (generic `MapFunction`) · [`MeosStatelessFilter`](MeosStatelessFilter.java) (generic `FilterFunction`) | ✅ shipped |
 | `bounded-state` | [`MeosBoundedStateMap`](MeosBoundedStateMap.java) (generic `KeyedProcessFunction` with `ValueState<byte[]>` per key — state crosses the operator boundary as MEOS-WKB/WKT bytes so checkpoints/rescaling/savepoints are safe; raw `Pointer` never leaves the JVM-local operator instance) | ✅ shipped |
 | `windowed` | [`MeosWindowedAggregate`](MeosWindowedAggregate.java) (generic `ProcessWindowFunction`; window-close-only aggregation; no MEOS handles persist across window boundaries) | ✅ shipped |
-| `cross-stream` | `MeosCrossStreamJoin` (generic `KeyedCoProcessFunction` or interval-join) | next follow-up |
+| `cross-stream` | [`MeosCrossStreamJoin`](MeosCrossStreamJoin.java) (generic `ProcessJoinFunction` over `KeyedStream.intervalJoin(other)`; time-bounded match window; same-key pairing) | ✅ shipped |
 | `io-meta` | covered transitively by the stateless wirings (no state, no window) | n/a |
 | `sequence-only` | inherently non-streamable — no wiring | n/a |
 

flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosCrossStreamDemoJob.java

@@ -0,0 +1,135 @@
+package org.mobilitydb.flink.meos.wirings.demo;
+
+import jnr.ffi.Pointer;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.tuple.Tuple5;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.KeyedStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.mobilitydb.flink.meos.MeosOpsFreeCore;
+import org.mobilitydb.flink.meos.MeosOpsTBox;
+import org.mobilitydb.flink.meos.wirings.MeosCrossStreamJoin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Arrays;
+
+/**
+ * End-to-end runnable demo of the {@code cross-stream} tier wiring.
+ *
+ * <p>Pipeline:
+ * <ol>
+ *   <li>Two parallel streams, each carrying {@code (regionId,
+ *       vehicleId, tboxWKT, eventTimeMs)}, sharing the {@code regionId}
+ *       key so cross-stream pairing is per-region.</li>
+ *   <li>{@code keyBy(regionId)} on both, then
+ *       {@code .intervalJoin().between(-1m, +1m)} so each event in
+ *       stream A is matched with events in stream B within ±1 minute
+ *       in the same region.</li>
+ *   <li>{@link MeosCrossStreamJoin}: for each matched pair, test
+ *       whether the two tboxes overlap via
+ *       {@code MeosOpsFreeCore.overlaps_tbox_tbox}; if yes, emit
+ *       {@code (regionId, vehAId, vehBId, leftTs, rightTs)}.</li>
+ * </ol>
+ *
+ * <p>What the demo proves:
+ * <ul>
+ *   <li><b>Interval-join semantics</b> — only pairs within the time
+ *       bound are matched; outside-window events are skipped.</li>
+ *   <li><b>Per-key isolation</b> — events in region 1 don't match
+ *       events in region 2, even if their timestamps overlap.</li>
+ *   <li><b>Pairwise MEOS call</b> — the wiring lambda receives both
+ *       matched events; the adopter calls any cross-stream MeosOps
+ *       method on the pair (here {@code overlaps_tbox_tbox}, which
+ *       is technically stateless on box pairs but the join-pairing
+ *       is what makes it cross-stream).</li>
+ * </ul>
+ *
+ * <p>Run with:
+ *
+ * <pre>{@code
+ * mvn -q exec:java \
+ *     -Dexec.mainClass=org.mobilitydb.flink.meos.wirings.demo.MeosCrossStreamDemoJob \
+ *     -Dmobilityflink.meos.enabled=true
+ * }</pre>
+ */
+public final class MeosCrossStreamDemoJob {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MeosCrossStreamDemoJob.class);
+
+    /** Stream A — vehicle events, 3 per region across 2 regions. */
+    private static final Tuple4<Integer, Integer, String, Long>[] EVENTS_A = new Tuple4[]{
+            Tuple4.of(1, 10, "TBOX XT([0,5],[2026-01-01,2026-01-01 00:00:30])",   ts("00:00:00")),
+            Tuple4.of(2, 20, "TBOX XT([100,105],[2026-01-01,2026-01-01 00:00:30])", ts("00:00:05")),
+            Tuple4.of(1, 11, "TBOX XT([10,15],[2026-01-01 00:00:30,2026-01-01 00:01:00])", ts("00:00:30")),
+    };
+
+    /** Stream B — different vehicles, 3 per region across 2 regions. */
+    private static final Tuple4<Integer, Integer, String, Long>[] EVENTS_B = new Tuple4[]{
+            Tuple4.of(1, 30, "TBOX XT([3,8],[2026-01-01,2026-01-01 00:00:30])",   ts("00:00:10")), // overlaps with A:(1,10)
+            Tuple4.of(2, 40, "TBOX XT([200,205],[2026-01-01,2026-01-01 00:00:30])", ts("00:00:15")), // disjoint from A:(2,20)
+            Tuple4.of(1, 31, "TBOX XT([12,17],[2026-01-01 00:00:30,2026-01-01 00:01:00])", ts("00:00:40")), // overlaps with A:(1,11)
+    };
+
+    private static long ts(String hms) {
+        String[] parts = hms.split(":");
+        long secs = Integer.parseInt(parts[0]) * 3600L
+                  + Integer.parseInt(parts[1]) * 60L
+                  + Integer.parseInt(parts[2]);
+        return 1767225600000L + secs * 1000L;   // 2026-01-01T00:00:00 UTC in ms
+    }
+
+    public static void main(String[] args) throws Exception {
+        if (!MeosOpsTBox.MEOS_AVAILABLE) {
+            LOG.error("MEOS not available — the demo requires libmeos.");
+            System.exit(1);
+        }
+
+        StreamExecutionEnvironment env =
+                StreamExecutionEnvironment.getExecutionEnvironment();
+        env.setParallelism(1);
+
+        DataStream<Tuple4<Integer, Integer, String, Long>> a =
+                env.fromCollection(Arrays.asList(EVENTS_A))
+                   .assignTimestampsAndWatermarks(
+                           WatermarkStrategy
+                                   .<Tuple4<Integer, Integer, String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
+                                   .withTimestampAssigner((e, ts) -> e.f3));
+        DataStream<Tuple4<Integer, Integer, String, Long>> b =
+                env.fromCollection(Arrays.asList(EVENTS_B))
+                   .assignTimestampsAndWatermarks(
+                           WatermarkStrategy
+                                   .<Tuple4<Integer, Integer, String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
+                                   .withTimestampAssigner((e, ts) -> e.f3));
+
+        KeyedStream<Tuple4<Integer, Integer, String, Long>, Integer> aKeyed = a.keyBy(t -> t.f0);
+        KeyedStream<Tuple4<Integer, Integer, String, Long>, Integer> bKeyed = b.keyBy(t -> t.f0);
+
+        // Interval-join: pair events in A with events in B within ±1 minute, same region key.
+        DataStream<Tuple5<Integer, Integer, Integer, Long, Long>> overlaps =
+                aKeyed.intervalJoin(bKeyed)
+                      .between(Time.minutes(-1), Time.minutes(1))
+                      .process(new MeosCrossStreamJoin<
+                              Tuple4<Integer, Integer, String, Long>,  // L
+                              Tuple4<Integer, Integer, String, Long>,  // R
+                              Tuple5<Integer, Integer, Integer, Long, Long>  // OUT: (region, vehA, vehB, lts, rts)
+                              >((left, right, ctx) -> {
+                                  Pointer leftTbox  = MeosOpsTBox.tbox_in(left.f2);
+                                  Pointer rightTbox = MeosOpsTBox.tbox_in(right.f2);
+                                  if (MeosOpsFreeCore.overlaps_tbox_tbox(leftTbox, rightTbox) != 0) {
+                                      return Tuple5.of(left.f0, left.f1, right.f1,
+                                                       ctx.getLeftTimestamp(), ctx.getRightTimestamp());
+                                  }
+                                  return null;
+                              }))
+                      .returns(org.apache.flink.api.common.typeinfo.TypeInformation.of(
+                              new org.apache.flink.api.common.typeinfo.TypeHint<Tuple5<Integer, Integer, Integer, Long, Long>>() {}));
+
+        overlaps.print("cross-stream-overlap");
+
+        env.execute("MeosWirings cross-stream tier demo");
+    }
+}