feat(wirings): bounded-state tier DataStream wiring + demo (stacks on #6)

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

@@ -0,0 +1,158 @@
+package org.mobilitydb.flink.meos.wirings;
+
+import jnr.ffi.Pointer;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
+import org.apache.flink.util.Collector;
+
+import java.io.Serializable;
+
+/**
+ * DataStream wiring for the {@code bounded-state} streaming tier of
+ * the generated {@code org.mobilitydb.flink.meos.MeosOps*} facades.
+ *
+ * <p>The {@code bounded-state} tier is "per-event with bounded per-key
+ * state, the state IS a MEOS handle". The canonical example is a
+ * per-key accumulator that keeps the running MEOS value alive across
+ * events (e.g. per-vehicle running trajectory, per-key tbox union).
+ *
+ * <p><b>Why state lives as bytes, not as a {@code Pointer}.</b> A
+ * {@code jnr.ffi.Pointer} is a raw native-memory address. It is not
+ * portable across JVM restarts; Flink would not be able to checkpoint
+ * or replay state. The wiring stores the state as a {@code byte[]}
+ * (typically the MEOS-WKB serialization of the temporal value), with
+ * adopter-supplied serialize / deserialize / step lambdas mediating
+ * the round-trip through MEOS:
+ *
+ * <pre>{@code
+ *  byte[] state               -- the per-key serialized MEOS value
+ *      ↓ deserialize (MEOS-WKB → Pointer)
+ *  Pointer prev               -- the in-flight MEOS handle
+ *      ↓ step(prev, event) → (newPointer, output)
+ *  Pointer next, OUT out      -- the new in-flight handle + per-event output
+ *      ↓ serialize (Pointer → MEOS-WKB)
+ *  byte[] newState            -- the per-key serialized new MEOS value
+ * }</pre>
+ *
+ * <p>This serde discipline is the same one MobilityDuck's persistent
+ * state machines use; it survives Flink savepoint / checkpoint /
+ * rescaling correctly because state crossing the operator boundary is
+ * always the MEOS-WKB bytes — never a raw pointer.
+ *
+ * <p><b>Typical usage</b> — per-vehicle running tbox union via
+ * {@code MeosOpsFreeCore.union_tbox_tbox} (stateless on its own;
+ * stateful when applied as a running fold):
+ *
+ * <pre>{@code
+ * DataStream<VehicleTbox> in = ...;          // (vehicleId, tbox)
+ * DataStream<RunningTbox> out = in
+ *     .keyBy(VehicleTbox::vehicleId)
+ *     .process(new MeosBoundedStateMap<Integer, VehicleTbox, RunningTbox>(
+ *         /* serialize *(/   ptr -> MeosOpsTBox.tbox_as_wkb(ptr, (byte) 4).array(),
+ *         /* deserialize *(/ bytes -> MeosOpsTBox.tbox_from_wkb(Pointer.wrap(...), bytes.length),
+ *         /* step *(/        (prev, evt) -> {
+ *             Pointer eventTbox = evt.toMeosTbox();
+ *             Pointer merged = (prev == null) ? eventTbox
+ *                                              : MeosOpsFreeCore.union_tbox_tbox(prev, eventTbox);
+ *             RunningTbox result = new RunningTbox(evt.vehicleId(), MeosOpsTBox.tbox_as_hexwkb(merged, (byte) 4, null));
+ *             return new MeosStep<>(merged, result);
+ *         }));
+ * }</pre>
+ *
+ * <p>The first event for a key sees {@code prev == null} (no prior
+ * state); the wiring handles that case by skipping the
+ * {@code deserialize} call. On subsequent events, the state is
+ * re-hydrated, mutated, re-serialized.
+ *
+ * <p><b>Coverage</b>: bounded-state is the second-largest tier in the
+ * v4 baseline (797 of 2,097 emitted methods — 513 OO-classified + 284
+ * free-fn). Any of them can be wrapped through this single class —
+ * adopters provide the three lambdas, the wiring handles all of the
+ * Flink state plumbing.
+ *
+ * <p><b>State serializer</b>: this implementation uses Flink's built-in
+ * {@code byte[]} primitive-array serializer (no custom Kryo / Avro / Pojo
+ * registration needed). The state size per key is bounded by the
+ * MEOS-WKB size of the running value — sub-KB for typical
+ * accumulator scenarios.
+ *
+ * @param <K>   the key type ({@code keyBy} extractor return type)
+ * @param <IN>  the input event type
+ * @param <OUT> the output type emitted per event
+ */
+public final class MeosBoundedStateMap<K, IN, OUT>
+        extends KeyedProcessFunction<K, IN, OUT> {
+
+    /** Serializable Pointer → bytes serializer (typically MEOS-WKB). */
+    @FunctionalInterface
+    public interface PointerSerialize extends Serializable {
+        byte[] toBytes(Pointer pointer) throws Exception;
+    }
+
+    /** Serializable bytes → Pointer deserializer (typically MEOS-WKB). */
+    @FunctionalInterface
+    public interface PointerDeserialize extends Serializable {
+        Pointer fromBytes(byte[] bytes) throws Exception;
+    }
+
+    /** Per-event step: (prior MEOS handle, event) → (new handle, output). */
+    @FunctionalInterface
+    public interface MeosStepFn<IN, OUT> extends Serializable {
+        MeosStep<OUT> apply(Pointer prior, IN event) throws Exception;
+    }
+
+    /** Tuple returned by the step lambda. */
+    public static final class MeosStep<OUT> implements Serializable {
+        private static final long serialVersionUID = 1L;
+        public final Pointer newState;
+        public final OUT output;
+        public MeosStep(Pointer newState, OUT output) {
+            this.newState = newState;
+            this.output = output;
+        }
+    }
+
+    private final PointerSerialize serialize;
+    private final PointerDeserialize deserialize;
+    private final MeosStepFn<IN, OUT> step;
+
+    private transient ValueState<byte[]> handleState;
+
+    public MeosBoundedStateMap(PointerSerialize serialize,
+                               PointerDeserialize deserialize,
+                               MeosStepFn<IN, OUT> step) {
+        this.serialize = serialize;
+        this.deserialize = deserialize;
+        this.step = step;
+    }
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        MeosWiringRuntime.ensureInitializedOnThread();
+        ValueStateDescriptor<byte[]> descriptor = new ValueStateDescriptor<>(
+                "meos-bounded-state",
+                PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO);
+        handleState = getRuntimeContext().getState(descriptor);
+    }
+
+    @Override
+    public void processElement(IN event,
+                                KeyedProcessFunction<K, IN, OUT>.Context ctx,
+                                Collector<OUT> out) throws Exception {
+        byte[] priorBytes = handleState.value();
+        Pointer prior = (priorBytes == null) ? null : deserialize.fromBytes(priorBytes);
+
+        MeosStep<OUT> stepResult = step.apply(prior, event);
+
+        byte[] newBytes = serialize.toBytes(stepResult.newState);
+        handleState.update(newBytes);
+
+        if (stepResult.output != null) {
+            out.collect(stepResult.output);
+        }
+    }
+}

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

@@ -8,7 +8,7 @@ per **streaming tier** (per
 | Tier | Wiring class(es) here | Status in this package |
 |---|---|---|
 | `stateless` | [`MeosStatelessMap`](MeosStatelessMap.java) (generic `MapFunction`) · [`MeosStatelessFilter`](MeosStatelessFilter.java) (generic `FilterFunction`) | ✅ shipped |
-| `bounded-state` | `MeosBoundedStateMap` (generic `KeyedProcessFunction` with `ValueState<Pointer>` per key) | next follow-up |
+| `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 |
 | `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 |

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

@@ -0,0 +1,114 @@
+package org.mobilitydb.flink.meos.wirings.demo;
+
+import jnr.ffi.Pointer;
+import org.apache.flink.api.java.tuple.Tuple2;
+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.mobilitydb.flink.meos.MeosOpsFreeCore;
+import org.mobilitydb.flink.meos.MeosOpsTBox;
+import org.mobilitydb.flink.meos.wirings.MeosBoundedStateMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+/**
+ * End-to-end runnable demo of the {@code bounded-state} tier wiring.
+ *
+ * <p>Pipeline:
+ * <ol>
+ *   <li>Stream of {@code (vehicleId, eventTboxWKT)} for 2 vehicles, 3
+ *       events each.</li>
+ *   <li>{@code keyBy(vehicleId)} so per-vehicle state isolates.</li>
+ *   <li>Per-vehicle running tbox union via
+ *       {@link MeosBoundedStateMap}: state holds the MEOS-WKT text of
+ *       the current union; on each event, deserialize → call
+ *       {@code MeosOpsFreeCore.union_tbox_tbox} → re-serialize.</li>
+ *   <li>Emit {@code (vehicleId, runningUnionTboxWKT)} per event.</li>
+ * </ol>
+ *
+ * <p>What the demo proves:
+ * <ul>
+ *   <li><b>Checkpoint-safe state</b> — state crosses the operator
+ *       boundary as {@code byte[]} (MEOS-WKT here, MEOS-WKB in
+ *       production); no raw native pointers in checkpoints.</li>
+ *   <li><b>Per-key isolation</b> — vehicle 1's running union does not
+ *       leak into vehicle 2's, and vice versa.</li>
+ *   <li><b>First-event correctness</b> — the wiring handles
+ *       {@code prior == null} on the first event for each key by
+ *       skipping deserialize and seeding state with the first event's
+ *       tbox.</li>
+ * </ul>
+ *
+ * <p>Run with:
+ *
+ * <pre>{@code
+ * mvn -q exec:java \
+ *     -Dexec.mainClass=org.mobilitydb.flink.meos.wirings.demo.MeosBoundedStateDemoJob \
+ *     -Dmobilityflink.meos.enabled=true
+ * }</pre>
+ *
+ * <p>Expected output: 6 lines (3 per vehicle), each showing the growing
+ * union tbox after that event.
+ */
+public final class MeosBoundedStateDemoJob {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MeosBoundedStateDemoJob.class);
+
+    /** 6 events across 2 vehicles — the running union grows monotonically per key. */
+    private static final Tuple2<Integer, String>[] EVENTS = new Tuple2[]{
+            Tuple2.of(1, "TBOX XT([0,2],[2026-01-01,2026-01-01 01:00])"),
+            Tuple2.of(2, "TBOX XT([10,12],[2026-01-01,2026-01-01 01:00])"),
+            Tuple2.of(1, "TBOX XT([3,5],[2026-01-01 01:00,2026-01-01 02:00])"),
+            Tuple2.of(2, "TBOX XT([13,15],[2026-01-01 01:00,2026-01-01 02:00])"),
+            Tuple2.of(1, "TBOX XT([1,4],[2026-01-01 02:00,2026-01-01 03:00])"),
+            Tuple2.of(2, "TBOX XT([11,14],[2026-01-01 02:00,2026-01-01 03:00])"),
+    };
+
+    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<Tuple2<Integer, String>> events =
+                env.fromCollection(Arrays.asList(EVENTS));
+
+        KeyedStream<Tuple2<Integer, String>, Integer> keyed =
+                events.keyBy(t -> t.f0);
+
+        // Wire the per-vehicle running union via MeosBoundedStateMap.
+        // State is the MEOS-WKT text of the current union (byte[] form);
+        // each event deserializes, unions with the event tbox, re-serializes.
+        DataStream<Tuple2<Integer, String>> runningUnion = keyed.process(
+                new MeosBoundedStateMap<Integer, Tuple2<Integer, String>, Tuple2<Integer, String>>(
+                        /* serialize:   Pointer → byte[] */
+                        ptr -> MeosOpsTBox.tbox_out(ptr, 6).getBytes(StandardCharsets.UTF_8),
+                        /* deserialize: byte[] → Pointer */
+                        bytes -> MeosOpsTBox.tbox_in(new String(bytes, StandardCharsets.UTF_8)),
+                        /* step: (prior union, this event) → (new union, output) */
+                        (prior, evt) -> {
+                            Pointer eventTbox = MeosOpsTBox.tbox_in(evt.f1);
+                            // First event for a key: prior is null — seed with the event's tbox.
+                            // Subsequent events: union prior with the new event's tbox.
+                            Pointer newUnion = (prior == null)
+                                    ? eventTbox
+                                    : MeosOpsFreeCore.union_tbox_tbox(prior, eventTbox, /*strict=*/0);
+                            Tuple2<Integer, String> output =
+                                    Tuple2.of(evt.f0, MeosOpsTBox.tbox_out(newUnion, 6));
+                            return new MeosBoundedStateMap.MeosStep<>(newUnion, output);
+                        }))
+                .returns(org.apache.flink.api.common.typeinfo.TypeInformation.of(
+                        new org.apache.flink.api.common.typeinfo.TypeHint<Tuple2<Integer, String>>() {}));
+
+        runningUnion.print("running-union");
+
+        env.execute("MeosWirings bounded-state tier demo");
+    }
+}