diff --git a/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosCrossStreamJoin.java b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosCrossStreamJoin.java new file mode 100644 index 0000000..24c5ce1 --- /dev/null +++ b/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. + * + *

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}). + * + *

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. + * + *

Typical usage — 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}): + * + * 

{@code
+ * KeyedStream a = streamA.keyBy(VehiclePosition::regionId);
+ * KeyedStream b = streamB.keyBy(VehiclePosition::regionId);
+ *
+ * DataStream meetings = a
+ *     .intervalJoin(b)
+ *         .between(Time.minutes(-5), Time.minutes(5))
+ *         .process(new MeosCrossStreamJoin(
+ *             (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
+ *             }));
+ * }
+ * + *

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. + * + *

Slim adopter signature — 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. + * + *

Coverage: 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 the left-stream event type + * @param the right-stream event type + * @param the per-match output type + */ +public final class MeosCrossStreamJoin + extends ProcessJoinFunction { + + /** Serializable per-match MEOS pairwise call. */ + @FunctionalInterface + public interface JoinFn 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 joinFn; + + public MeosCrossStreamJoin(JoinFn 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.Context context, + Collector 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); + } + } +} diff --git a/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/README.md b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/README.md index 906ab8b..cf210d1 100644 --- a/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/README.md +++ b/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` 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 | diff --git a/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosCrossStreamDemoJob.java b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosCrossStreamDemoJob.java new file mode 100644 index 0000000..97c3223 --- /dev/null +++ b/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. + * + *

Pipeline: + *

    + *
  1. Two parallel streams, each carrying {@code (regionId, + * vehicleId, tboxWKT, eventTimeMs)}, sharing the {@code regionId} + * key so cross-stream pairing is per-region.
    2. + *
  2. {@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.
    3. + *
  3. {@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)}.
    4. + *
+ * + *

What the demo proves: + *

+ * + *

Run with: + * + * 

{@code
+ * mvn -q exec:java \
+ *     -Dexec.mainClass=org.mobilitydb.flink.meos.wirings.demo.MeosCrossStreamDemoJob \
+ *     -Dmobilityflink.meos.enabled=true
+ * }
+ */ +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[] 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[] 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> a = + env.fromCollection(Arrays.asList(EVENTS_A)) + .assignTimestampsAndWatermarks( + WatermarkStrategy + .>forBoundedOutOfOrderness(Duration.ofSeconds(1)) + .withTimestampAssigner((e, ts) -> e.f3)); + DataStream> b = + env.fromCollection(Arrays.asList(EVENTS_B)) + .assignTimestampsAndWatermarks( + WatermarkStrategy + .>forBoundedOutOfOrderness(Duration.ofSeconds(1)) + .withTimestampAssigner((e, ts) -> e.f3)); + + KeyedStream, Integer> aKeyed = a.keyBy(t -> t.f0); + KeyedStream, Integer> bKeyed = b.keyBy(t -> t.f0); + + // Interval-join: pair events in A with events in B within ±1 minute, same region key. + DataStream> overlaps = + aKeyed.intervalJoin(bKeyed) + .between(Time.minutes(-1), Time.minutes(1)) + .process(new MeosCrossStreamJoin< + Tuple4, // L + Tuple4, // R + Tuple5 // 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>() {})); + + overlaps.print("cross-stream-overlap"); + + env.execute("MeosWirings cross-stream tier demo"); + } +}