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

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

Why state lives as bytes, not as a {@code Pointer}. 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: + * + *

{@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
+ * }
+ * + *

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

Typical usage — per-vehicle running tbox union via + * {@code MeosOpsFreeCore.union_tbox_tbox} (stateless on its own; + * stateful when applied as a running fold): + * + *

{@code
+ * DataStream in = ...;          // (vehicleId, tbox)
+ * DataStream out = in
+ *     .keyBy(VehicleTbox::vehicleId)
+ *     .process(new MeosBoundedStateMap(
+ *         /* 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);
+ *         }));
+ * }
+ * + *

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

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

State serializer: 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 the key type ({@code keyBy} extractor return type) + * @param the input event type + * @param the output type emitted per event + */ +public final class MeosBoundedStateMap + extends KeyedProcessFunction { + + /** 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 extends Serializable { + MeosStep apply(Pointer prior, IN event) throws Exception; + } + + /** Tuple returned by the step lambda. */ + public static final class MeosStep 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 step; + + private transient ValueState handleState; + + public MeosBoundedStateMap(PointerSerialize serialize, + PointerDeserialize deserialize, + MeosStepFn step) { + this.serialize = serialize; + this.deserialize = deserialize; + this.step = step; + } + + @Override + public void open(Configuration parameters) throws Exception { + super.open(parameters); + MeosWiringRuntime.ensureInitializedOnThread(); + ValueStateDescriptor descriptor = new ValueStateDescriptor<>( + "meos-bounded-state", + PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO); + handleState = getRuntimeContext().getState(descriptor); + } + + @Override + public void processElement(IN event, + KeyedProcessFunction.Context ctx, + Collector out) throws Exception { + byte[] priorBytes = handleState.value(); + Pointer prior = (priorBytes == null) ? null : deserialize.fromBytes(priorBytes); + + MeosStep stepResult = step.apply(prior, event); + + byte[] newBytes = serialize.toBytes(stepResult.newState); + handleState.update(newBytes); + + if (stepResult.output != null) { + out.collect(stepResult.output); + } + } +} 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 ffcb437..e1ed310 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 @@ -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` per key) | next follow-up | +| `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 | | `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 | diff --git a/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosBoundedStateDemoJob.java b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosBoundedStateDemoJob.java new file mode 100644 index 0000000..c77b0e7 --- /dev/null +++ b/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. + * + *

Pipeline: + *

    + *
  1. Stream of {@code (vehicleId, eventTboxWKT)} for 2 vehicles, 3 + * events each.
  2. + *
  3. {@code keyBy(vehicleId)} so per-vehicle state isolates.
  4. + *
  5. 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.
  6. + *
  7. Emit {@code (vehicleId, runningUnionTboxWKT)} per event.
  8. + *
+ * + *

What the demo proves: + *

+ * + *

Run with: + * + *

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

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[] 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> events = + env.fromCollection(Arrays.asList(EVENTS)); + + KeyedStream, 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> runningUnion = keyed.process( + new MeosBoundedStateMap, Tuple2>( + /* 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 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>() {})); + + runningUnion.print("running-union"); + + env.execute("MeosWirings bounded-state tier demo"); + } +}