Add native Fluss log-table connector

[michaelmitchell-bit]

Summary

This PR adds the native Fluss log-table source path and the Fluss Nexmark baseline scaffold.

The shape is the same pattern as the native Kafka source: keep Fluss/Flink coordination on the JVM side, reuse Fluss' enumerator for startup offsets, partition discovery, and split assignment, then pass the assigned log splits into Rust so fluss-rs can read Arrow RecordBatches directly.

What's In Here

• Adds an optional SF_MATRIX_FLUSS=true Nexmark matrix rung using a local Fluss test cluster.
• Writes the same wide Nexmark event row into Fluss, then reads it back through the stock Flink connector and the native Fluss path.
• Updates the matrix output so Fluss reports stock Flink-on-Fluss vs native Fluss when the native library is built with the fluss feature.
• Adds the Java source wrapper, planner node, split/config translation, and JNI declarations for the native Fluss source.
• Adds the Rust reader behind the fluss cargo feature: open table scan, subscribe assigned buckets, poll ScanBatches, honor stopping offsets, and export Arrow through C Data.
• Supports append-only log-table partition discovery through Fluss' normal Java enumerator. Newly discovered partitions arrive as normal LogSplits and get handed to Rust with table/partition/bucket/offset info.
• Handles Fluss' partition-removal round trip: PartitionsRemovedEvent -> native split removal/unsubscribe -> PartitionBucketsUnsubscribedEvent.
• Covers projection pushdown through the native Fluss path.
• Hardens the draft path so native failures become Java IOExceptions instead of Rust panics crossing JNI.

Benchmarks

Fresh post-switch Fluss Nexmark q0 run, using the streaming/count-N path:

• Stock Flink-on-Fluss: 0.124s (808,251 ev/s)
• Native Fluss source: 0.089s (1,129,921 ev/s)
• Speedup: 1.40x

This was Fluss-only q0 at SF_ROWS=100000, best-of-2, with a release native build using --features mimalloc,fluss. I have not rerun the full Fluss matrix yet.

Testing

Ran locally with Java 17 and the native fluss feature built:

• mvn -Dnative.cargo.args="build --features fluss" -Dtest='io.github.jordepic.streamfusion.fluss.*Test' test
  • 13 tests passing
  • includes E2E native reads, projection pruning, static partitioned log tables, and dynamic partition discovery while the read job is running
• cargo check --manifest-path native/Cargo.toml --features fluss --keep-going

[jordepic]

Questions:

1. "local fluss test cluster" -> docker images/testingcontainers?
2. Can you add a headline for the benchmarks, and whether they seem "reasonable" i.e. around parquet? from first principles that's what we want
3. What did you end up doing about the arrow version?

"Partitioned tables with partitioned discovery enabled" -> I wonder if you can tell me more about these? Do we want to be supporting partitioned logs from the start? Otherwise this feels like more of just a benchmarking feature

[jordepic]

Additionally - how was this change tested? do we have round trip tests to ensure config is right? And that we're reading the right fluss messages?

[jordepic]

Thanks for taking this on — the overall shape is exactly what the design (ticket 44 / .claude/research/fluss-native-source-findings.md) called for, and several of the hard parts landed cleanly. Full review below, ranked by severity.

What's done well

• The reuse seam is right: NativeFlussSource delegates enumerator, split serializers, checkpoint state, and restore to Fluss's own classes, so partition discovery, FLIP-288 offset semantics, and checkpoint compatibility come for free. NativeFlussLogSplit stays a JNI-boundary translation, not a checkpoint type.
• Split IDs travel Java→Rust at assign time and come back verbatim on drain — no fragile splitId reconstruction on the native side.
• The class-loading guard in PhysicalPlanScan.isFlussTableSource (string class-name comparison, LinkageError caught) keeps deployments without the Fluss jar from crashing at plan time.
• Zero-copy Arrow export across the C Data Interface; catch_unwind → IOException matches the SplitReader.fetch contract; the emitter's offset math is correct (base + rows ≡ Fluss's logOffset + 1).
• The harness tests run a real embedded Fluss cluster, assert substitutions() > 0 so fallback can't masquerade as coverage, and genuinely exercise mid-job partition discovery (100 ms interval, partition added while the job runs) and a mid-job drop. The Rust reader even handles the race where a poll hits PartitionNotExists before the enumerator notices.

Must fix — correctness

1. Dropped partitions never leave the reader's checkpoint state. Fluss's own split reader accumulates removed split IDs and reports them as finished splits through the next fetch() (FlinkSourceSplitReader.java:152 upstream) — that is the only mechanism that removes SourceReaderBase's internal split states, because the fetcher manager's splitFinishedHook is a no-op both in Fluss's wiring and in this PR's super(splitReaderSupplier, config). NativeFlussSplitReader.removeSplits cleans its own maps but never reports the splits finished, so after a partition drop every subsequent checkpoint still carries the dead splits, and a restore re-subscribes to a dropped partition. With Fluss's auto-partitioned tables (TTL drops partitions on a schedule) this is a routine production path, and recovery then leans entirely on the brittle string-parse fallback in (4). Fix: mirror Fluss — remember removed IDs, purge the native pending queue for them (only remove_missing_partition does this today; unassign_splits doesn't), and addFinishedSplit them on the next fetch. The drop test can't see this; please add a drop → checkpoint → restore scenario (or a test asserting snapshotState no longer contains the dropped split).

2. No type vetting anywhere. FlussTables.plan never checks column types, and every test uses only BIGINT/STRING/INT. fluss-rs's Arrow mapping (Timestamp-LTZ with UTC zone, Time32/64, Decimal128, FixedSizeBinary, nested List/Map/Struct) has to line up with what the Arrow→RowData transpose expects per Flink type — currently unverified, and a mismatch is a runtime crash or silent wrong results on a table Flink runs fine. Same class of gap: an INDEXED-log-format table or a non-default ZSTD compression level routes natively and then fails at runtime where Flink succeeds — both violate identical-or-fall-back. Fix: gate types in the matcher (the csvColumnsSupported pattern) and gate log-format/compression from table metadata; add a parity test that writes the full supported type surface through the stock connector and compares native vs stock output row-for-row (the referee pattern the Kafka decode work established).

3. The benchmark compares different engines running different plans. The stock rung runs in batch mode via Fluss's limit-read path (capped at 2,048 rows, limit pushed into a dedicated RPC); the native rung runs in streaming with the LIMIT becoming a native Top-N operator. Different execution mode, different plan shape, and ≤2,048 rows measures job startup, not throughput — this can't support any ratio claim, and ticket 36 explicitly requires stock-vs-native at the same steelman perimeter. Every existing rung in the matrix (generator, Kafka, Parquet) runs both engines in the identical default streaming runtime with boundedness coming from the source; the Fluss rung should too. Fix: both engines in streaming mode with no LIMIT and a count-N-then-cancel sink (the harness's CollectingSink pattern), measuring time-to-Nth-row at the normal SF_ROWS scale.

Should fix

4. parse_missing_partition_id string-parses an error message (fluss.rs:236) — pinned to fluss-rs's current wording; if it changes, a routine TTL partition drop becomes a poll failure → restart loop. Prefer re-listing subscribed partitions against metadata on PartitionNotExists (or upstreaming a structured partition ID on the error), and make an unparseable message degrade to waiting for the enumerator's removal event rather than failing the job.
5. The stopping-offset path is half-implemented dead code. It's unreachable today (streaming-only; batch is lake-gated), but as written an empty split never finishes on the Java side (no batch ⇒ positions never reaches the stop), and after the stop is reached, stopping_offsets is removed while split_ids remains, so a still-in-flight batch would emit rows past the stop for an already-finished split. Either finish it (immediate-finish at assign when start >= stop, keep the guard until unassign) or replace it with an assert-and-gate so bounded support isn't silently half-shipped.
6. FlussConfigTranslator silently ignores unknown client.* options — the silent-divergence trap the Kafka translator exists to refuse. It also passes a non-sasl security.protocol through to fluss-rs (runtime failure where Flink works — gate it), and translates writer/lookup options a source never uses (dropping them shrinks the parity surface).
7. The fluss-rs dependency is a git pin to a personal fork whose only real change is the arrow 57→58 bump (clean and minimal — I diffed it against upstream). That bump should be PR'd to apache/fluss-rust; until it lands, the Cargo.toml entry needs a comment saying exactly what the fork changes (matching the file's existing comment style), and the fork should live somewhere the project controls. Also, the "Arrow 59" commit message and the ticket-36 note about "Arrow 59 alignment" don't match the lockfile (which unifies on arrow 58.3.0) — please clean those up.
8. CI never runs any of this: fluss isn't a default cargo feature and no workflow was added, so all five harness tests skip via @EnabledIf. Add a CI job building --features fluss (or make it a default feature if the build cost is acceptable).
9. No fallback reasons: every FlussTables decline (PK table, lake, reflection failure, untranslatable config) returns bare null with nothing recorded in fallbackReasons. The coverage doctrine is that a non-accelerating query can always explain itself; the reflection-failure case especially (a Fluss upgrade renaming a private field would silently de-accelerate every Fluss table) must at least be visible.

Minor

10. assign_splits does one block_on subscribe per split — fluss-rs has subscribe_buckets/subscribe_partition_buckets batch calls (one lock, one await).
11. Batch granularity: one ScanBatch = one producer wire batch, so a table written with small writer batches floods the island with tiny Arrow batches. Consider per-bucket coalescing to a target row count before emitting — worth measuring, since per-batch overhead is what the columnar operators are sensitive to.
12. NativeFlussSource.createReader passes new Configuration() instead of the context's configuration — user-set SourceReaderOptions (element-queue capacity etc.) are ignored; Fluss passes context.getConfiguration().
13. All tests use bucket.num = 1 — a reader multiplexing several splits (interleaved buckets, per-split position bookkeeping) is never exercised; one multi-bucket test would cover it.
14. Bookkeeping on rebase: the branch predates master's two newest commits, so it will conflict with the ticket-44/roadmap changes. On rebase: delete tickets 44 and 36 (board convention is delete-on-ship — state moves to the docs), add the Fluss section to docs/coverage-and-fallbacks.md (native path, every gate from Add Fluss baseline to Nexmark matrix #2/Wire native Fluss source planning #6, discovery/removal semantics), put real numbers + method in docs/benchmarks.md once Add native Fluss log-table connector #3 is fixed, and update the roadmap index.

[jordepic]

Re-reviewed the new patchset (d0978c2..b5db52e). Strong revision — 11 of the 14 points addressed properly, and I verified the branch compiles and the 12 JVM-only unit tests pass locally. But the patchset introduces one new critical inconsistency that has to be resolved before the benchmark/roadmap claims hold.

New critical issue: the type gate and the benchmark contradict each other

Commit d875b05 (correctly) gates TIMESTAMP, TIMESTAMP_LTZ, and nested ROW/ARRAY/MAP out of the native path, with an honest reason — fluss-rs exports per-precision timestamp units (UTC-zoned for LTZ) while the vendored Arrow conversion pins Timestamp(ns, null). But the benchmark's Fluss table is still created from PARQUET_SCHEMA — three nested ROW columns plus dateTime TIMESTAMP(3) — so the matcher can never substitute it, and the reworked benchmark (correctly!) throws "native island did not engage" rather than measure stock-vs-stock. Every native rung of the Fluss benchmark fails by construction. Meanwhile the roadmap says "Native Fluss log-table source — done … numbers pending a full run" and docs/benchmarks.md promises numbers — neither is currently achievable.

The two right fixes compose: land timestamp (and ideally struct) support across the boundary — a post-decode unit/zone cast in the Rust reader is cheap and stays columnar — or, if deferring that, benchmark an honestly-labeled flattened-schema variant and reword the roadmap/docs to say the Nexmark rung is blocked on boundary type support. Realistically, TIMESTAMP is table stakes for a streaming source; without it the zero-transpose story doesn't reach any real workload, so the roadmap "done" wording should be softened either way — today this is a flat-scalar-table source.

Related precision issue: the gate checks the full catalog schema (FilesystemTables.physicalRowType), not the projected read columns — SELECT id FROM t falls back because t has an unused MAP column somewhere. The projection is already pushed into the scan; gating on the projected fields would recover those queries.

What was fixed well

• Partition removal (Add Fluss baseline to Nexmark matrix #1) — the exact Fluss pattern: removed splits are reported through finishedSplits on the next fetch so SourceReaderBase drops them from checkpoint state; the native side purges pending batches on unassign; empty bounded splits finish immediately at assign. Best of all, the new integration test drives the real NativeFlussSplitReader against a live cluster, drops a partition, and asserts the finished-split report — it pins the mechanism, not just the symptom.
• Types (Add Fluss baseline to Nexmark matrix #2) — whitelist plus NativeFlussTypeParityTest, a genuine referee (edge values, extremes, empty strings, NULLs, unicode, leap-day) comparing stock vs native row-for-row.
• Benchmark shape (Add native Fluss log-table connector #3) — identical SQL, identical default streaming environment, count-N-then-cancel sink with the target calibrated per query against the bounded generator, and a loud failure if substitution didn't happen. Methodologically right now; it just can't run yet (above).
• The rest: unparseable partition-not-exists errors degrade to the enumerator's removal path instead of failing the job (Add Fluss config and split translation seam #4); the stopping-offset path got the immediate-finish and both-maps retirement fixes (Add native Fluss source shell #5); the translator gates unknown client.* and unsupported security protocols while documenting why writer/lookup options are ignored (Wire native Fluss source planning #6); the fork pin is documented in Cargo.toml with the replace-when-upstreamed note (Native async scalar UDF (StreamPhysicalAsyncCalc): within-batch concurrency, not the AsyncWaitOperator mailbox bridge #7); a fluss-tests CI job runs the Rust tests and the Java suites under the feature (Native OVER aggregation: support AVG, COUNT(*), and decimal value columns #8); every matcher decline records a precise fallback reason, including reflection failures (Route legacy group windows (GROUP BY TUMBLE/HOP) onto the native window operators #9); subscribes are batched (Remove the Fluss Nexmark rung's bounded-scan workarounds (blocked on upstream scan.bounded.mode) #10); the reader honors the context configuration (Disaggregated state store: remote/tiered operator state with a local working-set cache #12); there's a multi-bucket test (Columnar endpoints beyond local Parquet: Iceberg, and remote filesystems for the source #13); ticket 36 was properly deleted with roadmap/coverage/benchmarks docs updated (Perf backlog: native timing hook, last scalar-keyed maps, Fluss aggregate levers #14).

Still open

1. CI has never actually run on this PR — zero check runs on the head commit (fork PRs need workflow approval). Until a run is approved, the feature-gated integration tests and the new CI job itself are unverified. Local verification covered compile + the 12 JVM-only unit tests; the cluster-backed tests need that CI run.
2. Rebase still pending — the branch is based on 8cde095, before ticket 44 and the roadmap/ticket-36 edits on master. The rebase must delete ticket 44 (this PR is its implementation) and reconcile the roadmap entry.
3. Batch coalescing (prior Align the Fluss partition-removal plumbing with the connector's own shape #11) wasn't addressed — fine as a perf follow-up, but worth a line in the optimizations backlog / a ticket so it isn't lost, since per-batch overhead is what the columnar operators are sensitive to.

[jordepic]

Took another pass over the latest patchset (7f4fa8f + cfebbab, rebased onto current master). This round resolves everything that was blocking — nice work — and the first real CI run of the new fluss-tests job surfaced one environmental gap to close.

The blockers are properly fixed

• The type-gate-vs-benchmark contradiction is fixed from both ends. The Rust reader normalizes timestamps to Timestamp(ns, null) recursively (including inside Structs) as a columnar per-batch cast, and the gate whitelists plain TIMESTAMP and nested ROWs of supported leaves. Combined with the gate now checking only the projected columns instead of the full catalog schema, the Nexmark event schema routes natively and the benchmark can engage.
• Both fixes come with the right tests. The parity test covers TIMESTAMP and nested-ROW-with-timestamp — including 2262-04-11 23:47:16.854, the exact nanosecond-i64 ceiling — and the harness gained an e2e where a table with an unused TIMESTAMP_LTZ column still routes when only supported columns are projected, plus a nested-row read in the Nexmark shape.
• Bookkeeping is clean. Rebase done; tickets 36 and 44 both deleted per board convention; the batch-coalescing follow-up is captured on the roadmap; the coverage doc's Fluss section lists every gate with its recorded reason.

CI: the new job failed on a missing system dependency (its first run — it did its job)

fluss-tests failed before any test ran: fluss-rs's build script generates its RPC protos via prost-build and needs protoc, which the runner doesn't have:

error: failed to run custom build command for `fluss-rs v0.2.0 (…fluss-rust.git…)`
  Error: Could not find `protoc`. … To install it on Debian, run `apt-get install protobuf-compiler`.

Two levels of fix:

1. Immediate: add an apt-get install -y protobuf-compiler step (or arduino/setup-protoc) to the fluss-tests job before the cargo build.
2. The deeper issue: the fluss feature build now has an undocumented machine-specific prerequisite. This project deliberately avoided that for Kafka (bundled-static librdkafka — a build that needs no system install). At minimum, the build lines in docs/benchmarks.md and the coverage doc should note that protoc is required for --features fluss; the better fix is upstream — fluss-rust switching prost-build to a vendored protoc (e.g. protoc-bin-vendored) would make the feature self-contained on every contributor's machine, and could ride along with the arrow-bump PR to apache/fluss-rust.

Two remaining one-liners (non-blocking)

1. Dangling pointer: .claude/research/fluss-native-source-findings.md:96 still references the now-deleted 44-native-fluss-log-source.md — board convention is to remove every pointer when a ticket dies.
2. Overflow semantics on the new timestamp cast: arrow::compute::cast defaults to safe, so a millisecond value beyond the ns range (year 2262+) becomes NULL silently where Flink reads it. This isn't a regression — the engine-wide transpose (TimestampWriter.java:88) does a plain millis * 1_000_000 that silently wraps past the same ceiling — but the project doctrine is identical-or-loud, so CastOptions { safe: false } would fail the job instead of silently nulling. The engine-wide ns ceiling deserves a short divergences/ note as a follow-up either way.

Bottom line: mergeable once fluss-tests goes green with the protoc step, ideally with the two one-liners above. Remaining tail after merge: the real -Pbench benchmark run for the docs, and upstreaming the arrow bump (+ vendored protoc) to apache/fluss-rust.