Skip to content

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

Open
estebanzimanyi wants to merge 6 commits into
MobilityDB:mainfrom
estebanzimanyi:feat/flink-bounded-state-tier-wirings
Open

feat(wirings): bounded-state tier DataStream wiring + demo (stacks on #6)#7
estebanzimanyi wants to merge 6 commits into
MobilityDB:mainfrom
estebanzimanyi:feat/flink-bounded-state-tier-wirings

Conversation

@estebanzimanyi
Copy link
Copy Markdown
Member

Second of four planned tier-specific wiring follow-ups on PR #5's generated MEOS facade. Stacks on PR #6 (stateless wirings). Adds the bounded-state tier.

What's in this PR

File Purpose
MeosBoundedStateMap<K, IN, OUT> Generic KeyedProcessFunction with ValueState<byte[]> per key — wraps any bounded-state MeosOps call as a per-key MEOS-handle accumulator
demo/MeosBoundedStateDemoJob Runnable per-vehicle running-tbox-union pipeline (6 events × 2 vehicles); demonstrates per-key isolation, first-event-null correctness, and checkpoint-safe state
README.md bounded-state row marked ✅ shipped

Design — state crosses the boundary as bytes, not Pointer

A jnr.ffi.Pointer is a raw native-memory address. Storing one in Flink's ValueState would be a correctness time-bomb — pointers don't survive checkpoints, savepoints, or operator-instance rescaling. So MeosBoundedStateMap keeps state as byte[] (MEOS-WKB or MEOS-WKT, adopter's choice) with three adopter-supplied lambdas mediating the round-trip through MEOS:

byte[] state                  -- the per-key serialized MEOS value (in checkpoint)
    ↓ deserialize (bytes → 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 → bytes)
byte[] newState               -- the per-key serialized new MEOS value (back to checkpoint)

The first event for a key sees prior == null (no state yet); the wiring handles that case by skipping deserialize and letting the step seed state with the first event's result. Same serde discipline MobilityDuck's persistent state machines use.

Tier coverage now (PR #6 + this PR)

Tier Method count (v4 baseline) Wirable through
stateless 804 MeosStatelessMap / MeosStatelessFilter (PR #6)
bounded-state 797 MeosBoundedStateMap (this PR)
windowed 161 next follow-up (MeosWindowedAggregate)
cross-stream 140 next follow-up (MeosCrossStreamJoin)
io-meta 195 covered transitively by MeosStatelessMap (no state, no window)
sequence-only 14 n/a — inherently non-streamable

Cumulative: ~1,800 of PR #5's 2,097 generated methods (86%) are now wirable through 3 generic classes (Map + Filter + BoundedStateMap), without per-method registration.

Demo highlights

MeosBoundedStateDemoJob is a 6-event × 2-vehicle pipeline that computes a per-vehicle running tbox union via MeosOpsFreeCore.union_tbox_tbox. State holds the MEOS-WKT text of the current union; each event:

  1. Deserializes the prior state via MeosOpsTBox.tbox_in (or null on first event).
  2. Calls union_tbox_tbox(prior, eventTbox, 0) (skipping the call entirely when prior is null).
  3. Serializes the new state via MeosOpsTBox.tbox_out(newUnion, 6).
  4. Emits (vehicleId, runningUnionWKT).

Output (when run with libmeos available): 6 lines, monotonically growing per-vehicle union tboxes, vehicle 1's state independent of vehicle 2's.

Stacking

This PR stacks on feat/flink-stateless-tier-wirings (PR #6). Additive-only: 3 file changes (1 new wiring class + 1 new demo + 1 README row marked ✅). No existing file is touched beyond the README row.

Compile verification

Locally green: 135 .class files total (129 from PR #6 base + 6 new — 1 wiring class + 3 nested lambda interfaces + MeosStep tuple + 1 demo class).

…matrix on MobilityFlink

All nine BerlinMOD reference queries × three streaming forms each
(continuous, windowed, snapshot) on MobilityFlink — the complete 27-cell
stream-layers parity-matrix row, locally verified end-to-end with no
external dependencies (no Kafka, no Docker, no MEOS native lib, no
JMEOS call).

Queries:

  Q1  which vehicles have appeared in the stream
  Q2  where is vehicle X at time T
  Q3  which vehicles within d of P at time T
  Q4  which vehicles entered region R, and when
  Q5  pairs of vehicles meeting near point P
  Q6  cumulative distance per vehicle
  Q7  first passage of vehicles through POIs
  Q8  vehicles close to a road segment
  Q9  distance between vehicles X and Y at time T

Each query has three form classes (Q<N>{Continuous,Windowed,Snapshot}Function)
and a companion BerlinMODQ<N>LocalTest driver running the three forms
through a Flink mini-cluster against a hardcoded synthetic corpus.

Spatial predicates today are pure Java — Haversine distance for
point-to-point (Q3, Q5, Q6, Q9), point-in-box for region containment
(Q4), and a planar-projection point-to-line-segment distance (Q8). Each
spatial call site is marked TODO(meos) for migration to the JMEOS
bridge of the corresponding MEOS operator once the in-flight MEOS 1.4
bump signals settled (Q3 edwithin_tgeo_geo; Q4 STBox eintersects; Q5
NAD / edwithin_tgeo_tgeo; Q6 trajectory length; Q7 edwithin_tgeo_geo;
Q8 distance(tgeompoint, geometry(LINESTRING)); Q9 tdistance). Q1 and
Q2 have no spatial predicate.

State patterns exercised:
  - keyed simple flag (Q1)
  - keyed last-known position (Q2, Q8)
  - keyed transition + entry log (Q4)
  - keyed accumulator (Q6)
  - keyed first-passage map (Q7)
  - shared key-by-constant state (Q9 pair-wise, Q5 multi-pair MapState)

Verified output counts (see PR description for the exact-line excerpts):

  Q  | continuous | windowed | snapshot
  ---|------------|----------|---------
  Q1 |          3 |        2 |        9
  Q2 |          7 |        2 |        3
  Q3 |         21 |        2 |        6
  Q4 |          4 |        5 |        9
  Q5 |         14 |        2 |        3   (only pair (100,200) qualifies for our P + radii)
  Q6 |         21 |        6 |        9   (drift corpus; v100=601m, v200=300m, v300=1205m)
  Q7 |          3 |        6 |        9   (3 (vehicle, POI) first-passages; intra-window scope)
  Q8 |         21 |        2 |        6   (same shape as Q3 with segment-distance)
  Q9 |          7 |        2 |        3   (X=100, Y=200; distance 4124m = ~4.1km)

Build verification: mvn clean package green; all nine LocalTests run to
completion (Flink mini-cluster, parallelism=1) producing exactly the
expected output shapes.
… 1.4 MEOSBridge

Introduce MEOSBridge as the runtime spatial-predicate surface for all
BerlinMOD-9 × 3-form streaming cells. The bridge calls into MEOS via
JMEOS 1.4 (geog_dwithin over WGS84 geographies) when libmeos is loadable
and falls back to the pure-Java Haversine / SegmentDistance utilities
when it is not — the fallback path is what the BerlinMODQ*LocalTest
mini-cluster drivers exercise (system property mobilityflink.meos.enabled=false).

- New berlinmod/MEOSBridge.java with the dwithinMetres /
  dwithinSegmentMetres / distanceMetres surface and a fail-soft
  static init that flips MEOS_AVAILABLE to false on UnsatisfiedLinkError.
- All BerlinMOD-9 × 3-form spatial predicates rewritten to call
  MEOSBridge instead of Haversine / SegmentDistance directly. 27 cells,
  one bridge call surface, identical predicate semantics.
- JMEOS.jar updated to the JMEOS#15 regen branch artefact (478 305
  bytes); this is the JMEOS 1.4 regen build that exposes geog_dwithin /
  geom_in / geom_to_geog / edwithin_tgeo_geo / nad_tgeo_geo / tpoint_length.
- aisdata/Main.java and aisdata/TrajectoryWindowFunction.java adapted
  to the JMEOS 1.4 meos_initialize() / meos_initialize_timezone()
  split (the old two-arg meos_initialize(String, error_handler_fn)
  signature is gone in JMEOS#15).
- All nine BerlinMODQ*LocalTest mini-cluster drivers set
  mobilityflink.meos.enabled=false at main() entry so they remain
  green-CI without libmeos.so on the runtime path.
- target/ build artefacts gitignored.

The README's spatial-predicate paragraph is updated to describe the
MEOSBridge route as the production path; the TODO(meos) markers across
the BerlinMOD cells are gone.

Build: mvn clean package -DskipTests green.
Verify: BerlinMODQ{1,3,5,8}LocalTest all finish with FINISHED state
on the mini-cluster fallback path.
…s + extended types + utils.spatial)

Updates the bundled `flink-processor/jar/JMEOS.jar` to a combined build
of JMEOS PR #19 (regen against MEOS-API meos-idl.json, 2,699 methods
including extended types) AND PR #18 (utils.spatial.Haversine +
utils.spatial.PointToSegment wrappers that MEOSBridge.java imports).

Surface delta vs the previous bundled jar:
  - public static methods: 2 699 (was 1 685)
  - utils.spatial.Haversine.distance(lon1, lat1, lon2, lat2) → double
  - utils.spatial.PointToSegment.distance(pLon, pLat, s1Lon, s1Lat, s2Lon, s2Lat) → double
  - tnpoint_ methods: 50
  - tcbuffer / tpose / trgeo: now exposed
  - sha: a5895c9b94…  size: 1,210,863 B

Unblocks the MEOSBridge.java import path (line 116) — previously the
jar shipped PR #19's GeneratedFunctions but not PR #18's utils.spatial,
so base-branch mvn compile was RED.  Both PRs now coalesced into a
single jar built by:

    mvn -pl codegen,jmeos-core compile -Dmaven.test.skip=true
    cd jmeos-core/target/classes && jar cf JMEOS.jar .

Unblocks codegen/flink-meos-ops wedge stacked on this branch.
… surface

Add a generated, tier-aware Java facade over the MEOS public API,
organized as one Java class per MEOS object-model class plus one per
public-MEOS-header for free functions:

- 50 `MeosOps<Class>` classes (751 methods): one per MEOS object-model
  class (TFloat, TInt, TBool, TText, TGeomPoint, TGeogPoint, TCbuffer,
  TNpoint, TPose, TRGeometry, TBox, STBox, Set, Span, SpanSet, …).
- 6 `MeosOpsFree<Header>` classes (1,346 methods): one per public MEOS
  header for functions not assigned to any object-model class
  (MeosOpsFreeCore, MeosOpsFreeGeo, MeosOpsFreeCbuffer, MeosOpsFreeNpoint,
  MeosOpsFreePose, MeosOpsFreeRgeo).
- 1 shared `MeosOpsRuntime` (single `MEOS_AVAILABLE` static-init across
  all 56 facades).

Each emitted method forwards to `functions.GeneratedFunctions.<name>(...)`
after probing the shared `MeosOpsRuntime.MEOS_AVAILABLE` flag. Each
method carries a Javadoc tier marker (stateless / bounded-state /
windowed / cross-stream / io-meta) so consumers know the per-method
wiring shape.

Total emit: 2,097 of JMEOS PR #19's 2,699-method surface (77.7%);
remainder is the JMEOS-deliberately-omitted type-catalog helpers plus
the streaming-relevance-baseline ambiguous (59) and sequence-only (14)
buckets, both surfaced separately for design decisions before emit.

Two generators under flink-processor/tools/codegen/:
- codegen-oo.py: reads JMEOS jar signatures via javap-p +
  streaming-relevance baseline + MEOS object model → emits per-OO-class
  facades.
- codegen-free.py: same shape, but for functions not in the OO model →
  emits per-header facades.

Both are ~250 LOC, deterministic, audit-by-regeneration. Manifests
record provenance (JMEOS method total, baseline target count, emit
count, per-tier breakdown, per-class/per-header method count, sample
of functions absent from JMEOS).

Coexists with the existing berlinmod.MEOSBridge hand-written
BerlinMOD-scoped bridge (high-level, query-shaped); the generated
MeosOps* facades expose the raw MEOS surface tier-by-tier
(low-level, catalog-shaped). Both share the same MEOS_AVAILABLE
discipline and `functions.GeneratedFunctions` delegation.

Stacks on feat/jmeos-bridge-swap; additive-only; touches no existing
file. Locally compile-verified against the union of JMEOS PR #19's
jmeos-core + PR #18's utils.spatial (the latter needed by MEOSBridge,
separately tracked).
…OS facades

Adds the org.mobilitydb.flink.meos.wirings package — thin, generic
Flink-DataStream wrappers around the generated MeosOps* facades from
PR MobilityDB#5, organized per streaming tier.

This PR ships the stateless tier:
- MeosStatelessMap<IN, OUT>: generic MapFunction wrapping any stateless
  MeosOps* method (804 of the 2,097 generated methods qualify per the
  v4 baseline — 92 OO-classified + 712 free-fn)
- MeosStatelessFilter<IN>: generic FilterFunction wrapping any stateless
  boolean-returning MeosOps* method, plus a .fromIntPredicate(...)
  adapter for JMEOS' int-coded predicates
- demo/MeosWiringsDemoJob: runnable end-to-end DataStream pipeline
  parsing TBox WKT → filtering by overlap with a query box →
  serializing surviving boxes to hex-WKB, all through the generated
  facades wired via this package
- README documenting tier vocabulary, the wrap-once-use-everywhere
  pattern, the DataStream-API-only design choice (Table API as future
  follow-up), and coexistence with berlinmod.MEOSBridge

Future follow-ups (one PR per tier, mirroring this one's shape):
- MeosBoundedStateMap (generic KeyedProcessFunction with
  ValueState<Pointer> for MEOS handle per key — covers 797 of the
  generated methods)
- MeosWindowedAggregate (generic ProcessWindowFunction — 161 methods)
- MeosCrossStreamJoin (generic KeyedCoProcessFunction or interval-join
  — 140 methods)
- Optional: Table API sibling (MeosScalarUDF + MeosCatalogRegistrar)
  if the repo adopts Table API for other reasons

Stacks on codegen/flink-meos-ops (PR MobilityDB#5). Additive-only; touches no
existing file. Locally compile-verified: 129 .class files total (123
from the parent PR + 6 new from this package's classes + demo + their
nested lambdas).
Adds MeosBoundedStateMap<K, IN, OUT> — the second tier-wiring class
in the org.mobilitydb.flink.meos.wirings package, stacked on PR MobilityDB#6
(stateless wirings).

Bounded-state is the second-largest streaming tier in the v4 baseline
(797 of 2,097 emitted methods — 513 OO-classified + 284 free-fn). The
canonical pattern is per-key MEOS-handle accumulation: a running tbox
union, a running temporal value, a per-vehicle accumulator that keeps
the MEOS value alive across events.

## Design — state lives as bytes, not as Pointer

A jnr.ffi.Pointer is a raw native-memory address. It is not portable
across JVM restarts; Flink could not checkpoint, savepoint, or replay
state if the wiring stored raw pointers. MeosBoundedStateMap stores
state as byte[] (MEOS-WKB or MEOS-WKT, adopter's choice) with three
adopter-supplied lambdas mediating the round-trip:

- PointerSerialize:   Pointer → byte[]  (called after each step)
- PointerDeserialize: byte[] → Pointer  (called before each step)
- MeosStepFn:         (prior Pointer, event) → (new Pointer, output)

The first event for a key sees prior == null; the wiring handles that
case by skipping deserialize and seeding state with the first event's
result. Subsequent events re-hydrate, mutate, re-serialize.

Net effect: state crossing the operator boundary is always byte[];
checkpoints, savepoints, and rescaling all work correctly. This is
the same serde discipline MobilityDuck's persistent state machines
use.

## Files

- MeosBoundedStateMap.java — the generic wiring class
- demo/MeosBoundedStateDemoJob.java — runnable per-vehicle running
  tbox union pipeline (6 events × 2 vehicles; demonstrates per-key
  isolation, first-event-null correctness, and checkpoint-safe state)
- README — bounded-state row marked ✅ shipped

## Stacks on PR MobilityDB#6

Additive-only; touches no existing file. Locally compile-verified:
135 .class files total (129 from PR MobilityDB#6 base + 6 new — 1 wiring class
+ 3 nested lambda interfaces + MeosStep tuple + 1 demo class).
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant