p) {
+ return new MeosStatelessFilter<>(event -> p.test(event) != 0);
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ MeosWiringRuntime.ensureInitializedOnThread();
+ }
+
+ @Override
+ public boolean filter(IN event) throws Exception {
+ // When chained to a legacy source, records are processed on the source's
+ // emitter thread rather than the thread open() ran on; the ThreadLocal
+ // guard makes this a cheap no-op after the first call per thread.
+ MeosWiringRuntime.ensureInitializedOnThread();
+ return predicate.test(event);
+ }
+}
diff --git a/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessMap.java b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessMap.java
new file mode 100644
index 0000000..aadb4ac
--- /dev/null
+++ b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosStatelessMap.java
@@ -0,0 +1,93 @@
+package org.mobilitydb.flink.meos.wirings;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.configuration.Configuration;
+
+import java.io.Serializable;
+
+/**
+ * DataStream wiring for the {@code stateless} streaming tier of the
+ * generated {@code org.mobilitydb.flink.meos.MeosOps*} facades.
+ *
+ * Wraps any {@code MeosOps*.f(...)} call whose streaming tier is
+ * {@code stateless} (per {@code tools/codegen/meos-ops-manifest.json}
+ * or {@code meos-ops-free-manifest.json}) as a Flink
+ * {@link MapFunction}. No per-key state is allocated; each event is
+ * mapped independently. The wrapped call:
+ *
+ *
+ * - does no MEOS-handle state across events (per the
+ * {@code stateless} tier contract);
+ * - does not touch the time domain (no window required);
+ * - may delegate to MEOS via the bundled JMEOS jar when
+ * {@code MeosOpsRuntime.MEOS_AVAILABLE} — otherwise the
+ * generated facade throws {@code UnsupportedOperationException}.
+ *
+ *
+ * Typical usage: register a stateless MEOS predicate / arithmetic
+ * call as a per-event map step in a DataStream pipeline. Example with
+ * the generated {@code MeosOpsTBox.overlaps_tbox_tbox} (tier =
+ * {@code stateless}, per the codegen manifest):
+ *
+ *
{@code
+ * DataStream in = ...; // (tboxA, tboxB)
+ * DataStream overlap = in.map(
+ * new MeosStatelessMap<>(
+ * pair -> MeosOpsTBox.overlaps_tbox_tbox(pair.a, pair.b)));
+ * }
+ *
+ * Tier coverage: as of the codegen state on the parent PR,
+ * 804 of the 2,097 generated methods are {@code stateless} (92 OO-
+ * classified + 712 free-fn). Any of those can be wrapped through this
+ * single class without per-method boilerplate.
+ *
+ *
Coexistence with {@code berlinmod.MEOSBridge}: this is the
+ * low-level catalog-shaped wiring; {@code MEOSBridge} stays as
+ * the high-level query-shaped wiring for the BerlinMOD-9 suite.
+ * Both share the same {@code MeosOpsRuntime.MEOS_AVAILABLE} discipline.
+ *
+ * @param the input record type
+ * @param the output type returned by the wrapped MEOS call
+ */
+public final class MeosStatelessMap extends RichMapFunction {
+
+ /**
+ * Serializable per-event MEOS call. Implementations forward to a
+ * generated {@code MeosOps*.f(...)} static method, returning the
+ * Java type that the generated facade exposes.
+ */
+ @FunctionalInterface
+ public interface MeosCall extends Serializable {
+ OUT apply(IN event) throws Exception;
+ }
+
+ private final MeosCall call;
+
+ /**
+ * @param call serializable lambda forwarding to a stateless
+ * generated MEOS facade method. The lambda must be
+ * serializable (Java 8+ lambdas implementing a
+ * {@link Serializable} functional interface are).
+ */
+ public MeosStatelessMap(MeosCall call) {
+ this.call = call;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ // No per-key state in the stateless tier; the only per-operator
+ // concern is MEOS' per-thread session, initialized on this task thread.
+ MeosWiringRuntime.ensureInitializedOnThread();
+ }
+
+ @Override
+ public OUT map(IN event) throws Exception {
+ // When chained to a legacy source, records are processed on the source's
+ // emitter thread rather than the thread open() ran on; the ThreadLocal
+ // guard makes this a cheap no-op after the first call per thread.
+ MeosWiringRuntime.ensureInitializedOnThread();
+ return call.apply(event);
+ }
+}
diff --git a/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java
new file mode 100644
index 0000000..51840ae
--- /dev/null
+++ b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java
@@ -0,0 +1,37 @@
+package org.mobilitydb.flink.meos.wirings;
+
+import functions.GeneratedFunctions;
+
+/**
+ * Per-thread MEOS initialization for the {@code org.mobilitydb.flink.meos.wirings}
+ * operators.
+ *
+ * MEOS keeps its timezone / session state per OS thread. Each Flink
+ * subtask runs on its own task thread, so every wiring operator must
+ * initialize MEOS on that thread from its {@code open()} — the JVM-wide
+ * probe in {@code MeosOpsRuntime} only covers the thread that first
+ * touches a facade class (typically the job's main thread), not the task
+ * threads where the operators actually run.
+ *
+ *
{@link #ensureInitializedOnThread()} is idempotent per thread (guarded
+ * by a {@link ThreadLocal}), so it is safe to call from every operator's
+ * {@code open()} even when operators are chained onto the same thread. It
+ * installs a no-op error handler so a MEOS-side error surfaces as a thrown
+ * exception rather than terminating the JVM.
+ */
+public final class MeosWiringRuntime {
+
+ private static final ThreadLocal INITIALIZED =
+ ThreadLocal.withInitial(() -> Boolean.FALSE);
+
+ private MeosWiringRuntime() { /* utility */ }
+
+ /** Initialize MEOS on the calling thread exactly once. */
+ public static void ensureInitializedOnThread() {
+ if (!INITIALIZED.get()) {
+ GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { });
+ GeneratedFunctions.meos_initialize();
+ INITIALIZED.set(Boolean.TRUE);
+ }
+ }
+}
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
new file mode 100644
index 0000000..ffcb437
--- /dev/null
+++ b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/README.md
@@ -0,0 +1,89 @@
+# DataStream wirings for the generated MEOS facades
+
+This package supplies thin, generic Flink-DataStream wrappers around
+the generated `org.mobilitydb.flink.meos.MeosOps*` facades, organized
+per **streaming tier** (per
+`tools/codegen/meos-ops-manifest.json` + `tools/codegen/meos-ops-free-manifest.json`):
+
+| 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 |
+| `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 |
+| `sequence-only` | inherently non-streamable — no wiring | n/a |
+
+The wirings are **generic**: each takes a serializable lambda
+forwarding to whichever generated `MeosOps*.f(...)` method the adopter
+needs. No per-method boilerplate, no per-method registration —
+adopters wire the entire ~800-method `stateless` slice through
+`MeosStatelessMap` / `MeosStatelessFilter` without touching this
+package.
+
+## Why DataStream rather than Table API
+
+The repo's existing pipeline (`berlinmod/`, `aisdata/`) is
+DataStream-API only. Sticking to DataStream avoids adding the
+~50 MB `flink-table-planner` runtime dependency to the build matrix.
+A Table-API-shaped sibling
+(`MeosOpsTableCatalogRegistrar` / `MeosScalarUDF` / `MeosAggregateFunction`)
+is a clean follow-up if/when the repo adopts Table API for other
+reasons.
+
+## How a generated MEOS call becomes a Flink operator
+
+The pattern is the same across all four tiers:
+
+```java
+// 1. Pick the generated MeosOps method
+// (Javadoc tier marker tells you which wiring to use)
+boolean overlap = MeosOpsTBox.overlaps_tbox_tbox(boxA, boxB); // tier = stateless
+
+// 2. Wrap with the matching wiring
+MeosStatelessFilter filter = MeosStatelessFilter.fromIntPredicate(
+ pair -> MeosOpsTBox.overlaps_tbox_tbox(pair.a, pair.b));
+
+// 3. Apply to the DataStream
+DataStream overlapping = stream.filter(filter);
+```
+
+`MEOS_AVAILABLE` is probed once per JVM by `MeosOpsRuntime`'s static
+initializer (shared across all `MeosOps*` and `MeosOpsFree*`
+facades). When unavailable, every generated method throws
+`UnsupportedOperationException` with a clear message — the wiring
+layer doesn't have to handle that itself.
+
+## End-to-end runnable demo
+
+[`demo/MeosWiringsDemoJob.java`](demo/MeosWiringsDemoJob.java) walks
+through a 3-stage DataStream pipeline using two of the generated
+facades wired through `MeosStatelessMap` + `MeosStatelessFilter`:
+
+1. Parse a stream of TBox WKT strings via
+ `MeosOpsFreeCore.tbox_in` (io-meta, no state).
+2. Filter to those overlapping a fixed query box via
+ `MeosOpsTBox.overlaps_tbox_tbox` (stateless predicate).
+3. Serialize each survivor to hex-WKB via
+ `MeosOpsTBox.tbox_as_hexwkb` (io-meta, no state).
+
+Run with:
+
+```bash
+mvn -q exec:java \
+ -Dexec.mainClass=org.mobilitydb.flink.meos.wirings.demo.MeosWiringsDemoJob \
+ -Dmobilityflink.meos.enabled=true
+```
+
+Output (expected): two `overlapping-tbox-hex` lines (the two input
+boxes that overlap the query box), one disjoint box dropped, one
+`MeosWirings stateless tier demo` job completion line.
+
+## Coexistence with `berlinmod.MEOSBridge`
+
+`MEOSBridge.java` is the BerlinMOD-specific, hand-written bridge for
+the 9-query streaming-form parity matrix — high-level and
+query-shaped. The wirings here are low-level and catalog-shaped —
+applicable to any of the ~800 stateless or 800 bounded-state
+generated facade methods, not just the BerlinMOD-9 subset. Both
+share the same `MEOS_AVAILABLE` discipline (`MeosOpsRuntime`).
diff --git a/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosWiringsDemoJob.java b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosWiringsDemoJob.java
new file mode 100644
index 0000000..0597a86
--- /dev/null
+++ b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/demo/MeosWiringsDemoJob.java
@@ -0,0 +1,101 @@
+package org.mobilitydb.flink.meos.wirings.demo;
+
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.streaming.api.datastream.DataStream;
+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.MeosStatelessFilter;
+import org.mobilitydb.flink.meos.wirings.MeosStatelessMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+
+/**
+ * End-to-end runnable demo showing how the generated
+ * {@code org.mobilitydb.flink.meos.MeosOps*} facades wire into a Flink
+ * {@code DataStream} pipeline through the
+ * {@code org.mobilitydb.flink.meos.wirings} helpers.
+ *
+ * The demo:
+ *
+ * - Builds a small in-memory stream of TBox-WKT strings.
+ * - Parses each into a JMEOS {@code Pointer} via
+ * {@code MeosOpsTBox.tbox_in} (tier = {@code io-meta}).
+ * - Filters to those that overlap with a fixed query TBox via
+ * {@code MeosOpsTBox.overlaps_tbox_tbox} wrapped as a
+ * {@link MeosStatelessFilter} (tier = {@code stateless}).
+ * - Maps each surviving TBox to its serialized WKB hex via
+ * {@code MeosOpsTBox.tbox_as_hexwkb} wrapped as a
+ * {@link MeosStatelessMap} (tier = {@code io-meta} but no per-key
+ * state, so the {@code stateless} wiring works for it too).
+ *
+ *
+ * Run with:
+ *
+ *
{@code
+ * mvn -q exec:java \
+ * -Dexec.mainClass=org.mobilitydb.flink.meos.wirings.demo.MeosWiringsDemoJob \
+ * -Dmobilityflink.meos.enabled=true # require libmeos loadable
+ * }
+ *
+ * If libmeos is not loadable on the runtime (or
+ * {@code -Dmobilityflink.meos.enabled=false}), every wrapped MeosOps
+ * call throws {@code UnsupportedOperationException} with a clear
+ * message — the demo prints the throw shape and exits non-zero.
+ */
+public final class MeosWiringsDemoJob {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MeosWiringsDemoJob.class);
+
+ /** A small box covering (xmin=0, ymin=0, xmax=10, ymax=10). */
+ private static final String QUERY_TBOX_WKT = "TBOX XT([0,10],[2026-01-01,2026-01-02])";
+
+ /** Three input boxes — two overlap the query box, one doesn't. */
+ private static final String[] INPUT_TBOX_WKTS = {
+ "TBOX XT([5,15],[2026-01-01,2026-01-02])", // overlaps
+ "TBOX XT([20,30],[2026-01-01,2026-01-02])", // disjoint
+ "TBOX XT([3,8],[2026-01-01,2026-01-02])", // overlaps
+ };
+
+ public static void main(String[] args) throws Exception {
+ // Probe MEOS availability (the static initializer in MeosOpsRuntime
+ // fires the first time any MeosOps class is touched).
+ if (!MeosOpsTBox.MEOS_AVAILABLE) {
+ LOG.error("MEOS not available — the demo requires libmeos. "
+ + "Set -Dmobilityflink.meos.enabled=true and ensure libmeos is loadable.");
+ System.exit(1);
+ }
+
+ StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+
+ DataStream tboxWkts = env.fromCollection(Arrays.asList(INPUT_TBOX_WKTS));
+
+ // The records crossing operator boundaries are serialized MEOS values
+ // (WKT text) — never raw native pointers, which are process-local and
+ // not serializable across Flink tasks. Each operator parses to a
+ // transient MEOS handle, calls MEOS, and re-serializes.
+
+ // Stage 1: parse each WKT and re-serialize via tbox_out (stateless io-meta).
+ DataStream normalized = tboxWkts.map(
+ new MeosStatelessMap(
+ wkt -> MeosOpsTBox.tbox_out(MeosOpsTBox.tbox_in(wkt), 6)))
+ .returns(Types.STRING);
+
+ // Stage 2: filter to those overlapping the query box (stateless).
+ // The query box is the constant WKT operand, parsed inside the predicate;
+ // overlaps_tbox_tbox lives on MeosOpsFreeCore (free fn, not OO-classified).
+ DataStream overlapping = normalized.filter(
+ MeosStatelessFilter.fromIntPredicate(
+ wkt -> MeosOpsFreeCore.overlaps_tbox_tbox(
+ MeosOpsTBox.tbox_in(wkt),
+ MeosOpsTBox.tbox_in(QUERY_TBOX_WKT))));
+
+ overlapping.print("overlapping-tbox");
+
+ env.execute("MeosWirings stateless tier demo");
+ }
+}