Streaming MEOS-parity harness and NebulaStream adapter

.github/workflows/streaming_parity_gate.yml

@@ -0,0 +1,28 @@
+# Streaming-parity gate — the measured-not-guessed guard (Path-to-100 step 6).
+# Pure-Python, path-filtered to the parity tooling: it does NOT build NebulaStream
+# and never runs on a normal C++ PR. It checks the committed feed for an L3
+# callability regression and an over-claim (a "100%" callability statement while
+# the gap list is non-empty). Re-measuring the full surface needs the JMEOS jar +
+# libmeos and is run out-of-band (see tools/streaming_parity/feeds/README.md).
+name: streaming-parity gate
+
+on:
+  pull_request:
+    paths:
+      - 'tools/streaming_parity/**'
+      - 'doc/methodology/streaming_parity*'
+  push:
+    paths:
+      - 'tools/streaming_parity/**'
+      - 'doc/methodology/streaming_parity*'
+
+jobs:
+  gate:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.x'
+      - name: streaming-parity gate (no L3 regression, no over-claim)
+        run: python3 tools/streaming_parity/ci_gate.py

CMakeLists.txt

@@ -29,6 +29,16 @@ option(NES_LOG_WITH_STACKTRACE "Log exceptions with stacktrace" ON)
 option(ENABLE_LARGE_TESTS "Runs testcases with larger input data" OFF)
 option(NES_DEBUG_TUPLE_BUFFER_LEAKS "Heavyweight instrumentation for Tuplebuffer debugging" OFF)
 
+# MEOS type-family compile flags. Mirror MobilityDB/MEOS selective compilation
+# (CBUFFER/NPOINT/POSE/RGEO): a family OFF drops its generated MEOS operators from
+# the build entirely (per-family add_subdirectory gated below) so the binary links
+# cleanly against a libmeos built with the same family disabled. Defaults match the
+# families enabled in the libmeos that the dev image links.
+option(CBUFFER "Include circular-buffer (tcbuffer) MEOS operators" ON)
+option(NPOINT "Include network-point (tnpoint) MEOS operators" ON)
+option(POSE "Include geopose (tpose) MEOS operators" ON)
+CMAKE_DEPENDENT_OPTION(RGEO "Include rigid-geometry (trgeometry) MEOS operators" ON "POSE" OFF)
+
 
 if (CMAKE_BUILD_TYPE STREQUAL "Benchmark")
     add_compile_definitions(NO_ASSERT)
@@ -62,6 +72,21 @@ set(CMAKE_THREAD_LIBS_INIT "-lpthread")
 include(cmake/Sanitizers.cmake)
 include(cmake/ImportDependencies.cmake)
 project(NES LANGUAGES C CXX)
+
+# MEOS family flags as a reusable define list (NAME=1|0), exactly as MobilityDB/MEOS
+# uses -DCBUFFER=1|0 etc. Applied with file/target scope where an #if <FAMILY> guard
+# actually appears (currently only the windowed-aggregation lowering) rather than
+# globally: a global add_compile_definitions would change every TU's compile command
+# and blow the whole ccache on any flag change. Keeping it scoped means toggling a
+# family recompiles only the guard file, and unrelated TUs keep hitting the cache.
+set(MEOS_FAMILY_COMPILE_DEFS "")
+foreach(_meos_family CBUFFER NPOINT POSE RGEO)
+  if(${_meos_family})
+    list(APPEND MEOS_FAMILY_COMPILE_DEFS ${_meos_family}=1)
+  else()
+    list(APPEND MEOS_FAMILY_COMPILE_DEFS ${_meos_family}=0)
+  endif()
+endforeach()
 set(VCPKG_POLICY_ALLOW_RESTRICTED_HEADERS enabled)
 message(STATUS "Going to use ${CMAKE_CXX_COMPILER}")
 

Dockerfile.paho

@@ -0,0 +1,60 @@
+# Adds Eclipse Paho MQTT (C + C++) to the NebulaStream dev image so that
+# find_package(PahoMqttCpp CONFIG) succeeds and the optional MQTT source/sink
+# plugins build. The base dev image's vcpkg ships no upstream ports tree and its
+# build attempted a non-existent port (paho-mqttcpp), so paho was never present.
+#
+# Paho is built with the SAME toolchain NebulaStream uses (clang-19, libc++,
+# c++23, static, -fPIC) so the C++ wrapper is ABI-compatible with the rest of
+# the build and the MQTT plugin links against PahoMqttCpp::paho-mqttpp3-static.
+#
+#   docker build -f Dockerfile.paho -t localhost/nes-development:mobilitynebula-v5 .
+ARG BASE_IMAGE=localhost/nes-development:mobilitynebula-v4
+FROM ${BASE_IMAGE}
+
+ARG PAHO_C_TAG=v1.3.14
+ARG PAHO_CPP_TAG=v1.4.1
+ENV CC=clang-19 CXX=clang++-19
+
+# 1) Paho MQTT C client (static, SSL, PIC) — pure C, no libc++ concern.
+RUN set -eux; \
+    cd /tmp; \
+    git clone --depth 1 --branch ${PAHO_C_TAG} https://github.com/eclipse-paho/paho.mqtt.c.git; \
+    cmake -S paho.mqtt.c -B paho.mqtt.c/build -G Ninja \
+        -DCMAKE_BUILD_TYPE=Release \
+        -DPAHO_BUILD_STATIC=ON -DPAHO_BUILD_SHARED=OFF \
+        -DPAHO_WITH_SSL=ON -DPAHO_HIGH_PERFORMANCE=ON \
+        -DPAHO_ENABLE_TESTING=OFF \
+        -DCMAKE_POSITION_INDEPENDENT_CODE=ON \
+        -DCMAKE_INSTALL_PREFIX=/usr/local; \
+    cmake --build paho.mqtt.c/build --target install; \
+    rm -rf /tmp/paho.mqtt.c
+
+# libc++ char_traits<unsigned char> shim (see header for rationale). Installed
+# where find_package(PahoMqttCpp) adds it to the include path, and pulled into
+# the installed mqtt/types.h so the NebulaStream MQTT plugin compiles too.
+COPY docker/paho_uchar_char_traits.h /usr/local/include/paho_uchar_char_traits.h
+
+# 2) Paho MQTT C++ wrapper, built with the project's clang/libc++/c++23 toolchain
+#    so it is ABI-compatible. libc++ provides no std::char_traits<unsigned char>,
+#    which Paho's std::basic_string<unsigned char> buffers require; pull the
+#    char_traits shim into the installed mqtt/types.h (so the NebulaStream MQTT
+#    plugin gets it too) and force-include it, and pin Paho to c++23 so it
+#    matches the project's libc++ standard library.
+RUN set -eux; \
+    cd /tmp; \
+    git clone --depth 1 --branch ${PAHO_CPP_TAG} https://github.com/eclipse-paho/paho.mqtt.cpp.git; \
+    sed -i '1i #include <paho_uchar_char_traits.h>' paho.mqtt.cpp/include/mqtt/types.h; \
+    find paho.mqtt.cpp -name CMakeLists.txt -exec sed -i 's/CXX_STANDARD 11/CXX_STANDARD 23/g' {} +; \
+    cmake -S paho.mqtt.cpp -B paho.mqtt.cpp/build -G Ninja \
+        -DCMAKE_BUILD_TYPE=Release \
+        -DPAHO_BUILD_STATIC=ON -DPAHO_BUILD_SHARED=OFF \
+        -DPAHO_WITH_SSL=ON \
+        -DCMAKE_POSITION_INDEPENDENT_CODE=ON \
+        -DCMAKE_CXX_COMPILER=clang++-19 -DCMAKE_C_COMPILER=clang-19 \
+        -DCMAKE_CXX_STANDARD=23 -DCMAKE_CXX_STANDARD_REQUIRED=ON \
+        -DCMAKE_CXX_FLAGS="-stdlib=libc++ -fPIC -include paho_uchar_char_traits.h" \
+        -DCMAKE_INSTALL_PREFIX=/usr/local; \
+    cmake --build paho.mqtt.cpp/build --target install; \
+    test -f /usr/local/lib/cmake/PahoMqttCpp/PahoMqttCppConfig.cmake; \
+    ldconfig; \
+    rm -rf /tmp/paho.mqtt.cpp

Input/input_berlinmod.csv

@@ -0,0 +1,21 @@
+1735711200,100,4.3517,50.8503
+1735711200,300,4.2000,50.7500
+1735711201,200,4.3060,50.8270
+1735711202,100,4.3517,50.8503
+1735711202,300,4.2000,50.7500
+1735711203,200,4.3060,50.8270
+1735711204,100,4.3517,50.8503
+1735711204,300,4.2000,50.7500
+1735711205,200,4.3060,50.8270
+1735711206,100,4.3517,50.8503
+1735711206,300,4.2000,50.7500
+1735711207,200,4.3060,50.8270
+1735711208,100,4.3517,50.8503
+1735711208,300,4.2000,50.7500
+1735711209,200,4.3060,50.8270
+1735711210,100,4.3517,50.8503
+1735711210,300,4.2000,50.7500
+1735711211,200,4.3060,50.8270
+1735711212,100,4.3517,50.8503
+1735711212,300,4.2000,50.7500
+1735711213,200,4.3060,50.8270

Queries/berlinmod/q1_continuous.yaml

@@ -0,0 +1,47 @@
+# BerlinMOD-Q1 — continuous form
+# "Which vehicles have appeared in the stream?"
+# Per 1-second sliding bucket: emit (start, end, vehicle_id, event-count-in-bucket).
+# Reading N rows over consecutive buckets enumerates the distinct-vehicles-seen set.
+
+query: |
+  SELECT start,
+         end,
+         vehicle_id,
+         COUNT(time_utc) AS events
+  FROM berlinmod_stream
+  GROUP BY vehicle_id
+  WINDOW SLIDING(time_utc, SIZE 1 SEC, ADVANCE BY 1 SEC)
+  INTO file_sink;
+
+sinks:
+  - name: FILE_SINK
+    type: File
+    schema:
+      - { name: BERLINMOD_STREAM$START,      type: UINT64 }
+      - { name: BERLINMOD_STREAM$END,        type: UINT64 }
+      - { name: BERLINMOD_STREAM$VEHICLE_ID, type: UINT64 }
+      - { name: BERLINMOD_STREAM$EVENTS,     type: UINT64 }
+    config:
+      file_path: "/workspace/Output/output_berlinmod_q1_continuous.csv"
+      input_format: CSV
+
+logical:
+  - name: BERLINMOD_STREAM
+    schema:
+      - { name: TIME_UTC,   type: UINT64 }
+      - { name: VEHICLE_ID, type: UINT64 }
+      - { name: GPS_LON,    type: FLOAT64 }
+      - { name: GPS_LAT,    type: FLOAT64 }
+
+physical:
+  - logical: BERLINMOD_STREAM
+    type: TCP
+    parser_config:
+      type: CSV
+      field_delimiter: ","
+      tuple_delimiter: "\n"
+    source_config:
+      socket_host: "host.docker.internal"
+      socket_port: "32325"
+      socket_type: "SOCK_STREAM"
+      socket_domain: "AF_INET"

Queries/berlinmod/q1_snapshot.yaml

@@ -0,0 +1,46 @@
+# BerlinMOD-Q1 — snapshot form
+# "At each 5-second tick, list of distinct vehicles seen in the tick window."
+# Streaming approximation of the batch BerlinMOD-Q1 snapshot at time T.
+
+query: |
+  SELECT start,
+         end,
+         vehicle_id,
+         COUNT(time_utc) AS events
+  FROM berlinmod_stream
+  GROUP BY vehicle_id
+  WINDOW TUMBLING(time_utc, SIZE 5 SEC)
+  INTO file_sink;
+
+sinks:
+  - name: FILE_SINK
+    type: File
+    schema:
+      - { name: BERLINMOD_STREAM$START,      type: UINT64 }
+      - { name: BERLINMOD_STREAM$END,        type: UINT64 }
+      - { name: BERLINMOD_STREAM$VEHICLE_ID, type: UINT64 }
+      - { name: BERLINMOD_STREAM$EVENTS,     type: UINT64 }
+    config:
+      file_path: "/workspace/Output/output_berlinmod_q1_snapshot.csv"
+      input_format: CSV
+
+logical:
+  - name: BERLINMOD_STREAM
+    schema:
+      - { name: TIME_UTC,   type: UINT64 }
+      - { name: VEHICLE_ID, type: UINT64 }
+      - { name: GPS_LON,    type: FLOAT64 }
+      - { name: GPS_LAT,    type: FLOAT64 }
+
+physical:
+  - logical: BERLINMOD_STREAM
+    type: TCP
+    parser_config:
+      type: CSV
+      field_delimiter: ","
+      tuple_delimiter: "\n"
+    source_config:
+      socket_host: "host.docker.internal"
+      socket_port: "32325"
+      socket_type: "SOCK_STREAM"
+      socket_domain: "AF_INET"

Queries/berlinmod/q1_windowed.yaml

@@ -0,0 +1,46 @@
+# BerlinMOD-Q1 — windowed form
+# "Per 10-second tumbling window, distinct vehicles seen."
+# Emits one row per (window, vehicle) seen; reading N rows per window = distinctCount.
+
+query: |
+  SELECT start,
+         end,
+         vehicle_id,
+         COUNT(time_utc) AS events
+  FROM berlinmod_stream
+  GROUP BY vehicle_id
+  WINDOW TUMBLING(time_utc, SIZE 10 SEC)
+  INTO file_sink;
+
+sinks:
+  - name: FILE_SINK
+    type: File
+    schema:
+      - { name: BERLINMOD_STREAM$START,      type: UINT64 }
+      - { name: BERLINMOD_STREAM$END,        type: UINT64 }
+      - { name: BERLINMOD_STREAM$VEHICLE_ID, type: UINT64 }
+      - { name: BERLINMOD_STREAM$EVENTS,     type: UINT64 }
+    config:
+      file_path: "/workspace/Output/output_berlinmod_q1_windowed.csv"
+      input_format: CSV
+
+logical:
+  - name: BERLINMOD_STREAM
+    schema:
+      - { name: TIME_UTC,   type: UINT64 }
+      - { name: VEHICLE_ID, type: UINT64 }
+      - { name: GPS_LON,    type: FLOAT64 }
+      - { name: GPS_LAT,    type: FLOAT64 }
+
+physical:
+  - logical: BERLINMOD_STREAM
+    type: TCP
+    parser_config:
+      type: CSV
+      field_delimiter: ","
+      tuple_delimiter: "\n"
+    source_config:
+      socket_host: "host.docker.internal"
+      socket_port: "32325"
+      socket_type: "SOCK_STREAM"
+      socket_domain: "AF_INET"

Queries/berlinmod/q2_continuous.yaml

@@ -0,0 +1,44 @@
+# BerlinMOD-Q2 — continuous form
+# "Where is vehicle X (= 200) right now?"
+# Per 1-second sliding bucket, emit a trajectory snippet for vehicle X.
+
+query: |
+  SELECT start,
+         end,
+         TEMPORAL_SEQUENCE(gps_lon, gps_lat, time_utc) AS trajectory
+  FROM berlinmod_stream
+  WHERE vehicle_id = UINT64(200)
+  WINDOW SLIDING(time_utc, SIZE 1 SEC, ADVANCE BY 1 SEC)
+  INTO file_sink;
+
+sinks:
+  - name: FILE_SINK
+    type: File
+    schema:
+      - { name: BERLINMOD_STREAM$START,      type: UINT64 }
+      - { name: BERLINMOD_STREAM$END,        type: UINT64 }
+      - { name: BERLINMOD_STREAM$TRAJECTORY, type: VARSIZED }
+    config:
+      file_path: "/workspace/Output/output_berlinmod_q2_continuous.csv"
+      input_format: CSV
+
+logical:
+  - name: BERLINMOD_STREAM
+    schema:
+      - { name: TIME_UTC,   type: UINT64 }
+      - { name: VEHICLE_ID, type: UINT64 }
+      - { name: GPS_LON,    type: FLOAT64 }
+      - { name: GPS_LAT,    type: FLOAT64 }
+
+physical:
+  - logical: BERLINMOD_STREAM
+    type: TCP
+    parser_config:
+      type: CSV
+      field_delimiter: ","
+      tuple_delimiter: "\n"
+    source_config:
+      socket_host: "host.docker.internal"
+      socket_port: "32325"
+      socket_type: "SOCK_STREAM"
+      socket_domain: "AF_INET"

Queries/berlinmod/q2_snapshot.yaml

@@ -0,0 +1,43 @@
+# BerlinMOD-Q2 — snapshot form
+# "At each 5-second tick, snapshot of vehicle X's (= 200) trajectory in the tick."
+
+query: |
+  SELECT start,
+         end,
+         TEMPORAL_SEQUENCE(gps_lon, gps_lat, time_utc) AS trajectory
+  FROM berlinmod_stream
+  WHERE vehicle_id = UINT64(200)
+  WINDOW TUMBLING(time_utc, SIZE 5 SEC)
+  INTO file_sink;
+
+sinks:
+  - name: FILE_SINK
+    type: File
+    schema:
+      - { name: BERLINMOD_STREAM$START,      type: UINT64 }
+      - { name: BERLINMOD_STREAM$END,        type: UINT64 }
+      - { name: BERLINMOD_STREAM$TRAJECTORY, type: VARSIZED }
+    config:
+      file_path: "/workspace/Output/output_berlinmod_q2_snapshot.csv"
+      input_format: CSV
+
+logical:
+  - name: BERLINMOD_STREAM
+    schema:
+      - { name: TIME_UTC,   type: UINT64 }
+      - { name: VEHICLE_ID, type: UINT64 }
+      - { name: GPS_LON,    type: FLOAT64 }
+      - { name: GPS_LAT,    type: FLOAT64 }
+
+physical:
+  - logical: BERLINMOD_STREAM
+    type: TCP
+    parser_config:
+      type: CSV
+      field_delimiter: ","
+      tuple_delimiter: "\n"
+    source_config:
+      socket_host: "host.docker.internal"
+      socket_port: "32325"
+      socket_type: "SOCK_STREAM"
+      socket_domain: "AF_INET"