Skip to content

Remove the Fluss Nexmark rung's bounded-scan workarounds (blocked on upstream scan.bounded.mode) #10

Description

@jordepic

What

The Nexmark Fluss rung carries special measurement machinery no other source rung needs, and it should be removed:

  • the count-to-Nth-row cancel with a bounded-generator calibration (flussTargetRows),
  • the far-future watermark sentinel row that closes the final event-time windows,
  • the traced table + poison pair + marker sink for queries whose changelog cardinality is non-deterministic (q4/q9) or zero (q21),
  • the skip for q12 — a proctime window flushes early only at end-of-input, which no data row can fake, so the query can't be measured on this rung at all.

The end state: the Fluss rung measures exactly like the Kafka rung — both engines read bounded, the job ends by itself, the measurement is end-to-end time — and all 23 runnable queries measure uniformly, q12 included.

Blocker: an upstream Fluss change

All of that machinery exists for one reason: the Fluss Flink connector has no bounded read for log tables — no scan.bounded.mode, no stopping-offset surface (bounded reads exist only via the lake source in batch mode and the pushed-LIMIT RPC). So the fix is an upstream contribution to apache/fluss, not local work: Kafka-shaped scan.bounded.mode = 'latest-offset' (and plausibly timestamp) for log-table scans.

Every ingredient already exists upstream:

  • OffsetsInitializer.latest() already resolves per-bucket end offsets (it backs scan.startup.mode = 'latest') — a bounded read resolves the same offsets at enumerator start and stamps them as stopping offsets instead.
  • LogSplit already carries a stoppingOffset (used by the hybrid snapshot→log handoff), and the reader's finished-split machinery already terminates on it.
  • The enumerator already diffs/assigns per-bucket splits; boundedness is one more field at split construction plus Boundedness.BOUNDED on the source.

Our side needs nothing new: the native Fluss source reuses Fluss's enumerator verbatim (a bounded assignment flows through unchanged), and the native split reader already implements the stopping-offset path (immediate-finish for empty ranges, batch slicing at the stop, finished-split reporting) — currently exercised only by tests.

When the upstream change lands in a Fluss release

  • Re-point the benchmark's Fluss rung at the Kafka rung's bounded method and delete the machinery above, including the q12 skip reason.
  • The stock baseline needs the released connector version — do not ship a patched fluss-flink jar (unlike the fluss-rs cargo pin, a Maven fork is not a self-contained build).

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions