Add native Fluss log-table connector

.claude/todos/00-roadmap.md

@@ -114,10 +114,17 @@ here when the ticket is deleted.
   Flink's converters exactly (divergences/21). The residual tail lives in ticket 32 (CSV/JSON
   *file* sources and the smaller CDC follow-ups).
 - **Nexmark matrix vs Flink — running.** The full q0–q22 suite (q6 excluded, wontdos/39) runs
-  native-substituted vs stock Flink across four source rungs (generator, Kafka JSON/Avro/Protobuf);
+  native-substituted vs stock Flink across the source rungs (generator, Parquet, Kafka
+  JSON/Avro/Protobuf, and the opt-in Fluss rung below);
   per-query routed-fraction/fallback reasons via `NexmarkExplainTest`, throughput matrix in the readme.
   It is the standing prioritization + regression gate; the coverage/perf gaps it surfaces feed the
   backlog below.
+- **Native Fluss log-table source — done** (was ticket 36). Fluss (columnar streaming storage) is the
+  opt-in fourth source rung of the Nexmark matrix (`SF_MATRIX_FLUSS=true`): the native fluss-rs
+  log-table reader vs stock Flink-on-Fluss, same steelman perimeter — the columnar-on-the-wire source
+  the perimeter-transpose hypothesis called for. Requires the `fluss` cargo feature (a pinned
+  arrow-aligned fork of apache/fluss-rust until the bump is upstreamed — `native/Cargo.toml`).
+  Method + build line in `docs/benchmarks.md`; numbers pending a full run.
 
 ## Next, roughly in order (re-prioritized 2026-07-04 after the operator profiling round)
 
@@ -171,7 +178,8 @@ here when the ticket is deleted.
    the exchange split — OVER +121–162%, retracting Top-N +228%, exchange +208% on Criterion;
    session `update` batches gap-connected runs — dense shape 20× vs per-row; the `RowData → Arrow`
    transpose was made row-major + pre-sized, ~25% faster; a native decoder was investigated and
-   rejected on benchmark grounds — wontdos/28.)
+   rejected on benchmark grounds — wontdos/28.) Native Fluss follow-up: coalesce small fluss-rs
+   scanner batches before JNI export if profiles show per-batch overhead inside columnar pipelines.
 
 ## Production-readiness (not yet load-bearing)
 - **Memory accounting**: shipped for every stateful native operator (mini-batch local pre-aggregate
@@ -204,15 +212,6 @@ here when the ticket is deleted.
   decode; raw consume 1.21x the Java client). Remaining before the `kafkaSource` gate can default
   on: per-partition watermarks/idleness, specific-offsets / topic-pattern startup, key.format,
   SASL/SSL build features, Linux `mimalloc` link-alias verification, multi-broker measurement.
-- **Nexmark with Apache Fluss as the source** (ticket 36): add Fluss (columnar streaming storage) as a
-  fourth source in the Nexmark matrix; its columnar format may let the native island ingest Arrow with
-  little/no row transpose — the perimeter transpose is a visible share of the remaining stateful-query
-  cost, so Fluss is the source most likely to show the engine's largest end-to-end margin.
-- **Native Fluss log source** (ticket 44): the Kafka pattern mirrored onto Fluss — reuse its Flink
-  enumerator (dynamic partition discovery included) verbatim, swap the split reader for a JNI reader
-  over fluss-rust's Arrow batch scanner. Fluss's log is Arrow on the wire, so ingest can be
-  zero-decode/zero-transpose; feasibility confirmed 2026-07-05
-  (`.claude/research/fluss-native-source-findings.md`). Enables ticket 36's measurement.
 - **Native lookup join** (ticket 40): **coverage DONE** (2026-07-03). The native operators drive
   Flink's own generated lookup runners over each Arrow probe batch, so a processing-time lookup join
   (Nexmark q13) — INNER + LEFT, sync + async, field-ref *and constant* keys, pre-filter, dim-side

.claude/todos/11-arroyo-operator-coverage.md

@@ -86,8 +86,8 @@ is picked up. Operators are in `~/data/arroyo/crates/arroyo-worker/src/arrow/`.
 ## Connectors (later)
 - [x] Parquet source + sink (local `file:`), exactly-once sink.
 - [ ] Remote/columnar sources and sinks (Iceberg, `hdfs:`/`s3:`) — ticket 24.
-- [ ] Native columnar sources (Fluss / Iceberg-CDC) via Flink Source API +
-      availability futures — ticket 01.
+- [ ] Native columnar sources beyond Fluss (Iceberg-CDC) via Flink Source API +
+      availability futures — ticket 01. (The native Fluss log-table source is done.)
 
 ## Cross-cutting (do alongside, not after)
 - Acceleration config (allowIncompatible / master / per-operator flags) — done (`NativeConfig`)

.claude/todos/37-disaggregated-state-store.md

@@ -20,7 +20,8 @@ Flink-native client. An operator's keyed state maps cleanly onto a Fluss PK tabl
 `(operator, key)`. Crucially, the state we now hold is **already arrow-row bytes** (memcomparable key +
 value-encoded payload — from the Top-N/join/dedup/group-by refactors): that byte layout is exactly what
 a remote KV store wants on the wire, so the encode/decode is mostly already paid. Pairs naturally with
-ticket 36 (Fluss as a *source*) — one Fluss dependency serving both ingest and state.
+the native Fluss source (shipped — the Nexmark matrix's Fluss rung) — one Fluss dependency serving both
+ingest and state.
 
 **Sketch of the work (to be designed, not prescribed):**
 - A state-access abstraction over the operators' maps (get/put/delete/scan by key) so the in-memory map
@@ -35,5 +36,5 @@ ticket 36 (Fluss as a *source*) — one Fluss dependency serving both ingest and
 object-store + LSM (Hummock/ForSt-style) the better primitive with Fluss only for the changelog? Benchmark
 state-heavy Nexmark (q4/q9/q16 — large keyspaces) under disaggregated vs in-memory before committing.
 
-Relates to: ticket 36 (Fluss source), memory accounting (shipped — divergences/16), ticket 01 (mailbox/async),
+Relates to: the native Fluss source (shipped), memory accounting (shipped — divergences/16), ticket 01 (mailbox/async),
 and the changelog-operator byte-state refactors (the on-the-wire format is ready).

.github/workflows/ci.yml

@@ -64,3 +64,44 @@ jobs:
       # hosted runners' preinstalled Docker daemon serves directly.
       - name: Run Java test suite
         run: mvn -B -ntp test
+
+  # The jobs above build without `--features fluss`, so Native.flussFeatureBuilt() is false and
+  # every @EnabledIf-gated Fluss test silently skips there. This leg builds the native library
+  # with the fluss feature and actually runs them: the Rust unit tests under the feature, then
+  # the Fluss JUnit classes against a fluss-feature debug build (the Fluss harness uses an
+  # in-process test cluster, no Docker needed).
+  fluss-tests:
+    name: Fluss feature (Rust + Java tests)
+    runs-on: ubuntu-24.04
+    timeout-minutes: 150
+    steps:
+      - uses: actions/checkout@v6
+
+      # Same reason as java-tests: a cold debug build of the native library plus the
+      # Maven repository outgrows the runner's default free space.
+      - name: Free disk space
+        run: |
+          sudo rm -rf /usr/share/dotnet /usr/local/lib/android /opt/ghc /usr/local/.ghcup
+          df -h /
+
+      - uses: actions/setup-java@v4
+        with:
+          distribution: temurin
+          java-version: "17"
+          cache: maven
+
+      - uses: dtolnay/rust-toolchain@stable
+
+      - name: Install protoc
+        run: sudo apt-get update && sudo apt-get install -y protobuf-compiler
+
+      - uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: native
+
+      - name: Run native test suite (fluss feature)
+        working-directory: native
+        run: cargo test --features fluss
+
+      - name: Run Fluss Java tests (fluss-feature native build)
+        run: mvn -B -ntp -Dnative.cargo.args="build --features fluss" -Dtest='*Fluss*' test

docs/benchmarks.md

@@ -316,7 +316,8 @@ across the ladder — all vs stock Flink, same steelmanned perimeter. 500K event
 `SF_BENCHMARK=true mvn test -Pbench -Dnative.cargo.args="build --release --features mimalloc"
 -Dtest=NexmarkMatrixBenchmark` (Testcontainers Kafka; the `kafka` feature is a build default, and
 `mimalloc` — the recommended build — rebinds the library's allocator, divergences/19). Column
-toggles: `SF_MATRIX_GENERATOR` / `SF_MATRIX_PARQUET` / `SF_MATRIX_KAFKA` (`false` skips one).
+toggles: `SF_MATRIX_GENERATOR` / `SF_MATRIX_PARQUET` / `SF_MATRIX_KAFKA` (`false` skips one), plus
+`SF_MATRIX_FLUSS` (`true` *adds* the opt-in Fluss rung — off by default; see below).
 
 The matrix runs with the native managed-memory cap **in force**: the shared test cluster declares a
 deployment-like managed-memory size (flink-test-utils' default gave each slot ~10 MB, which the
@@ -490,6 +491,19 @@ q3/q14/q18/q21, whose JSON rows were below 1× on their old best rung and now si
 of the table is q16 and the changelog-bound q3/q19 — operator-bound queries where the consume saving
 is diluted, not reversed.
 
+**Fluss** — the opt-in fourth source rung (`SF_MATRIX_FLUSS=true`), the columnar-on-the-wire
+source: the same wide event row is preloaded into a local Fluss test cluster and read back by
+both engines in the identical default streaming runtime — stock Flink-on-Fluss vs the native
+fluss-rs log-table reader. Boundedness comes from a count-N-then-cancel collecting sink, so each
+cell measures time-to-Nth-row at `SF_ROWS` scale. The native reader requires the `fluss` cargo
+feature in the build, added alongside the recommended `mimalloc`: `SF_BENCHMARK=true
+SF_MATRIX_FLUSS=true mvn test -Pbench -Dnative.cargo.args="build --release --features
+mimalloc,fluss" -Dtest=NexmarkMatrixBenchmark`. Building the `fluss` feature currently needs
+`protoc` (`protobuf-compiler`) because fluss-rs generates its RPC protos at build time.
+
+_Fluss numbers are pending a full run; the rung's column will be reported here alongside the
+tables above once it lands._
+
 ### The tuned (mini-batch) matrix — the full suite
 
 Production Flink deployments routinely enable mini-batch for stateful queries, so the matrix has a