feat(wirings): windowed tier DataStream wiring + demo (stacks on #7)

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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>State considerations</b>: 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.
+ *
+ * <p><b>Typical usage</b> — per-vehicle per-tumbling-window
+ * trajectory length via {@code MeosOpsTemporal.temporal_length} (tier
+ * = {@code windowed}):
+ *
+ * <pre>{@code
+ * DataStream<VehiclePoint> events = ...;     // (vehicleId, lon, lat, timestamp)
+ * DataStream<VehicleLength> lengths = events
+ *     .assignTimestampsAndWatermarks(...)
+ *     .keyBy(VehiclePoint::vehicleId)
+ *     .window(TumblingEventTimeWindows.of(Time.minutes(10)))
+ *     .process(new MeosWindowedAggregate<Integer, VehiclePoint, VehicleLength, TimeWindow>(
+ *         (window, events, ctx) -> {
+ *             Pointer trajectory = buildTrajectoryFromPoints(events);  // adopter helper
+ *             double length = MeosOpsTemporal.temporal_length(trajectory);
+ *             return new VehicleLength(ctx.getCurrentKey(), window.getStart(), length);
+ *         }));
+ * }</pre>
+ *
+ * <p>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.
+ *
+ * <p><b>Coverage</b>: 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 <K>   the key type
+ * @param <IN>  the input event type within the window
+ * @param <OUT> the per-window output type
+ * @param <W>   the window type ({@code TimeWindow}, {@code GlobalWindow}, etc.)
+ */
+public final class MeosWindowedAggregate<K, IN, OUT, W extends Window>
+        extends ProcessWindowFunction<IN, OUT, K, W> {
+
+    /**
+     * 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<K, IN, OUT, W extends Window> extends Serializable {
+        OUT aggregate(W window, Iterable<IN> events, ContextLike<K> 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> {
+        K getCurrentKey();
+        long getCurrentProcessingTime();
+        long getCurrentWatermark();
+    }
+
+    private final WindowFn<K, IN, OUT, W> windowFn;
+
+    public MeosWindowedAggregate(WindowFn<K, IN, OUT, W> windowFn) {
+        this.windowFn = windowFn;
+    }
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        MeosWiringRuntime.ensureInitializedOnThread();
+    }
+
+    @Override
+    public void process(K key,
+                         ProcessWindowFunction<IN, OUT, K, W>.Context context,
+                         Iterable<IN> elements,
+                         Collector<OUT> out) throws Exception {
+        ContextLike<K> ctx = new ContextLike<K>() {
+            @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);
+        }
+    }
+}

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<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` (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 |

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.
+ *
+ * <p>Pipeline:
+ * <ol>
+ *   <li>Stream of {@code (vehicleId, tboxWKT, eventTimeMs)} for 2
+ *       vehicles, 4 events each.</li>
+ *   <li>{@code assignTimestampsAndWatermarks} so event-time windows
+ *       fire on a bounded-out-of-orderness schedule.</li>
+ *   <li>{@code keyBy(vehicleId)} → 30-second tumbling event-time
+ *       window.</li>
+ *   <li>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)}.</li>
+ * </ol>
+ *
+ * <p>What the demo proves:
+ * <ul>
+ *   <li><b>Window-close timing</b> — events outside the window are
+ *       excluded; events within are aggregated together.</li>
+ *   <li><b>Per-key isolation</b> — vehicle 1's window aggregate does
+ *       not include vehicle 2's events, and vice versa.</li>
+ *   <li><b>Stateless aggregation</b> — unlike {@code bounded-state},
+ *       no MEOS handle persists across window boundaries; each window
+ *       builds its aggregate from scratch from the iterable.</li>
+ * </ul>
+ *
+ * <p>Run with:
+ *
+ * <pre>{@code
+ * mvn -q exec:java \
+ *     -Dexec.mainClass=org.mobilitydb.flink.meos.wirings.demo.MeosWindowedDemoJob \
+ *     -Dmobilityflink.meos.enabled=true
+ * }</pre>
+ *
+ * <p>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<Integer, String, Long>[] 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<Tuple3<Integer, String, Long>> events =
+                env.fromCollection(Arrays.asList(EVENTS))
+                   .assignTimestampsAndWatermarks(
+                           WatermarkStrategy
+                                   .<Tuple3<Integer, String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
+                                   .withTimestampAssigner((e, ts) -> e.f2));
+
+        DataStream<Tuple4<Integer, Long, Integer, String>> aggregates = events
+                .keyBy(t -> t.f0)
+                .window(TumblingEventTimeWindows.of(Time.seconds(30)))
+                .process(new MeosWindowedAggregate<
+                        Integer,                                        // K
+                        Tuple3<Integer, String, Long>,                  // IN
+                        Tuple4<Integer, Long, Integer, String>,         // OUT
+                        TimeWindow                                      // W
+                        >((window, inWindowEvents, ctx) -> {
+                            Pointer agg = null;
+                            int count = 0;
+                            for (Tuple3<Integer, String, Long> 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<Tuple4<Integer, Long, Integer, String>>() {}));
+
+        aggregates.print("window-aggregate");
+
+        env.execute("MeosWirings windowed tier demo");
+    }
+}