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:
+ *
+ * - Stream of {@code (vehicleId, eventTboxWKT)} for 2 vehicles, 3
+ * events each.
+ * - {@code keyBy(vehicleId)} so per-vehicle state isolates.
+ * - 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.
+ * - Emit {@code (vehicleId, runningUnionTboxWKT)} per event.
+ *
+ *
+ * What the demo proves:
+ *
+ * - Checkpoint-safe state — state crosses the operator
+ * boundary as {@code byte[]} (MEOS-WKT here, MEOS-WKB in
+ * production); no raw native pointers in checkpoints.
+ * - Per-key isolation — vehicle 1's running union does not
+ * leak into vehicle 2's, and vice versa.
+ * - First-event correctness — 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.
+ *
+ *
+ * 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");
+ }
+}