diff --git a/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWindowedAggregate.java b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWindowedAggregate.java new file mode 100644 index 0000000..7196708 --- /dev/null +++ b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWindowedAggregate.java @@ -0,0 +1,123 @@ +package org.mobilitydb.flink.meos.wirings; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction; +import org.apache.flink.streaming.api.windowing.windows.Window; +import org.apache.flink.util.Collector; + +import java.io.Serializable; + +/** + * DataStream wiring for the {@code windowed} streaming tier of the + * generated {@code org.mobilitydb.flink.meos.MeosOps*} facades. + * + *

The {@code windowed} tier is "output cardinality changes; needs a + * window". The canonical examples are + * {@code temporal_length(tgeo)} (one length per trajectory window), + * {@code temporal_twavg(tnumber)} (one time-weighted average per + * window), and the per-class {@code _trajectory} / {@code _time} / + * {@code _timespan} accessors that reduce a full sequence to a single + * derived value. + * + *

Wraps any windowed MeosOps call as a Flink + * {@link ProcessWindowFunction}: per-window, the adopter receives the + * full iterable of events in the window, applies whatever MEOS + * sequence-derived operation is appropriate, and emits a single + * per-window output. The wiring handles the + * {@code ProcessWindowFunction} boilerplate (context, collector) so + * adopters write a single serializable lambda. + * + *

State considerations: unlike + * {@link MeosBoundedStateMap}, the {@code windowed} tier does not + * keep MEOS handles across event boundaries — each window's MEOS + * value is built fresh from the iterable on window close (or + * watermark trigger), used to compute the output, and discarded. The + * iterable's events are Flink-side data; MEOS handles are short-lived + * per-window. + * + *

Typical usage — per-vehicle per-tumbling-window + * trajectory length via {@code MeosOpsTemporal.temporal_length} (tier + * = {@code windowed}): + * + * 

{@code
+ * DataStream events = ...;     // (vehicleId, lon, lat, timestamp)
+ * DataStream lengths = events
+ *     .assignTimestampsAndWatermarks(...)
+ *     .keyBy(VehiclePoint::vehicleId)
+ *     .window(TumblingEventTimeWindows.of(Time.minutes(10)))
+ *     .process(new MeosWindowedAggregate(
+ *         (window, events, ctx) -> {
+ *             Pointer trajectory = buildTrajectoryFromPoints(events);  // adopter helper
+ *             double length = MeosOpsTemporal.temporal_length(trajectory);
+ *             return new VehicleLength(ctx.getCurrentKey(), window.getStart(), length);
+ *         }));
+ * }
+ * + *

The window-close path is event-time-aware: when Flink determines + * the window is complete (via watermark), it invokes the lambda once + * with the full iterable, the window metadata, and a context giving + * access to the key. The adopter returns a single output value. + * + *

Coverage: 161 of the 2,097 emitted methods (~8%) qualify + * as {@code windowed} per the v4 baseline — all of them wrappable + * through this single class. + * + * @param the key type + * @param the input event type within the window + * @param the per-window output type + * @param the window type ({@code TimeWindow}, {@code GlobalWindow}, etc.) + */ +public final class MeosWindowedAggregate + extends ProcessWindowFunction { + + /** + * Serializable per-window MEOS aggregate. The lambda receives the + * window metadata, the full iterable of in-window events, and a + * context (for key access). It returns a single per-window output + * value, or {@code null} to emit nothing. + */ + @FunctionalInterface + public interface WindowFn extends Serializable { + OUT aggregate(W window, Iterable events, ContextLike ctx) throws Exception; + } + + /** + * Slimmer alternative to Flink's {@code ProcessWindowFunction.Context} + * — exposes only the bits a MEOS aggregate typically needs (key + + * current processing time + current watermark). Keeps the wiring + * lambda free of Flink internals. + */ + public interface ContextLike { + K getCurrentKey(); + long getCurrentProcessingTime(); + long getCurrentWatermark(); + } + + private final WindowFn windowFn; + + public MeosWindowedAggregate(WindowFn windowFn) { + this.windowFn = windowFn; + } + + @Override + public void open(Configuration parameters) throws Exception { + super.open(parameters); + MeosWiringRuntime.ensureInitializedOnThread(); + } + + @Override + public void process(K key, + ProcessWindowFunction.Context context, + Iterable elements, + Collector out) throws Exception { + ContextLike ctx = new ContextLike() { + @Override public K getCurrentKey() { return key; } + @Override public long getCurrentProcessingTime() { return context.currentProcessingTime(); } + @Override public long getCurrentWatermark() { return context.currentWatermark(); } + }; + OUT result = windowFn.aggregate(context.window(), elements, 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 e1ed310..906ab8b 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 @@ -9,7 +9,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` (generic `ProcessWindowFunction`) | next follow-up | +| `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 | | `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/MeosWindowedDemoJob.java b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosWindowedDemoJob.java new file mode 100644 index 0000000..8d5e1b6 --- /dev/null +++ b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosWindowedDemoJob.java @@ -0,0 +1,132 @@ +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.Tuple3; +import org.apache.flink.api.java.tuple.Tuple4; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; +import org.apache.flink.streaming.api.windowing.time.Time; +import org.apache.flink.streaming.api.windowing.windows.TimeWindow; +import org.mobilitydb.flink.meos.MeosOpsFreeCore; +import org.mobilitydb.flink.meos.MeosOpsTBox; +import org.mobilitydb.flink.meos.wirings.MeosWindowedAggregate; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.util.Arrays; + +/** + * End-to-end runnable demo of the {@code windowed} tier wiring. + * + *

Pipeline: + *

    + *
  1. Stream of {@code (vehicleId, tboxWKT, eventTimeMs)} for 2 + * vehicles, 4 events each.
    2. + *
  2. {@code assignTimestampsAndWatermarks} so event-time windows + * fire on a bounded-out-of-orderness schedule.
    3. + *
  3. {@code keyBy(vehicleId)} → 30-second tumbling event-time + * window.
    4. + *
  4. Per-window {@link MeosWindowedAggregate}: union all in-window + * event tboxes into a single per-window aggregate tbox via + * repeated {@code MeosOpsFreeCore.union_tbox_tbox}, emit + * {@code (vehicleId, windowStart, eventCount, aggregateTboxWKT)}.
    5. + *
+ * + *

What the demo proves: + *

+ * + *

Run with: + * + * 

{@code
+ * mvn -q exec:java \
+ *     -Dexec.mainClass=org.mobilitydb.flink.meos.wirings.demo.MeosWindowedDemoJob \
+ *     -Dmobilityflink.meos.enabled=true
+ * }
+ * + *

Expected output: 4 lines (2 windows × 2 vehicles), each showing + * the aggregate tbox spanning that window's events for that vehicle. + */ +public final class MeosWindowedDemoJob { + + private static final Logger LOG = LoggerFactory.getLogger(MeosWindowedDemoJob.class); + + /** 8 events across 2 vehicles, two 30s windows each. */ + private static final Tuple3[] EVENTS = new Tuple3[]{ + // window 1: [0s, 30s) + Tuple3.of(1, "TBOX XT([0,2],[2026-01-01,2026-01-01 00:00:10])", ts("00:00:00")), + Tuple3.of(2, "TBOX XT([10,12],[2026-01-01,2026-01-01 00:00:10])", ts("00:00:05")), + Tuple3.of(1, "TBOX XT([3,5],[2026-01-01 00:00:10,2026-01-01 00:00:20])", ts("00:00:10")), + Tuple3.of(2, "TBOX XT([13,15],[2026-01-01 00:00:10,2026-01-01 00:00:20])", ts("00:00:15")), + // window 2: [30s, 60s) + Tuple3.of(1, "TBOX XT([1,4],[2026-01-01 00:00:30,2026-01-01 00:00:40])", ts("00:00:30")), + Tuple3.of(2, "TBOX XT([11,14],[2026-01-01 00:00:30,2026-01-01 00:00:40])", ts("00:00:35")), + Tuple3.of(1, "TBOX XT([2,3],[2026-01-01 00:00:40,2026-01-01 00:00:50])", ts("00:00:40")), + Tuple3.of(2, "TBOX XT([12,13],[2026-01-01 00:00:40,2026-01-01 00:00:50])", ts("00:00:45")), + }; + + /** Convert "HH:MM:SS" relative to 2026-01-01T00:00:00 into epoch milliseconds. */ + 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> events = + env.fromCollection(Arrays.asList(EVENTS)) + .assignTimestampsAndWatermarks( + WatermarkStrategy + .>forBoundedOutOfOrderness(Duration.ofSeconds(1)) + .withTimestampAssigner((e, ts) -> e.f2)); + + DataStream> aggregates = events + .keyBy(t -> t.f0) + .window(TumblingEventTimeWindows.of(Time.seconds(30))) + .process(new MeosWindowedAggregate< + Integer, // K + Tuple3, // IN + Tuple4, // OUT + TimeWindow // W + >((window, inWindowEvents, ctx) -> { + Pointer agg = null; + int count = 0; + for (Tuple3 evt : inWindowEvents) { + Pointer evtTbox = MeosOpsTBox.tbox_in(evt.f1); + agg = (agg == null) + ? evtTbox + : MeosOpsFreeCore.union_tbox_tbox(agg, evtTbox, /*strict=*/0); + count++; + } + String aggWkt = (agg == null) ? "(empty)" : MeosOpsTBox.tbox_out(agg, 6); + return Tuple4.of(ctx.getCurrentKey(), window.getStart(), count, aggWkt); + })) + .returns(org.apache.flink.api.common.typeinfo.TypeInformation.of( + new org.apache.flink.api.common.typeinfo.TypeHint>() {})); + + aggregates.print("window-aggregate"); + + env.execute("MeosWirings windowed tier demo"); + } +}