RocksDB state backend

StreamFusion's persistent backend follows Flink's RocksDB lifecycle while keeping native operator state in a RocksDB instance owned and accessed directly by Rust. Java coordinates Flink checkpoint handles and file upload, but native state reads, writes, flushes, and compactions do not cross JNI. Host-side fallback state and timers use Flink's EmbeddedRocksDBStateBackend delegate.

Backend teardown attempts every independent release even when snapshot, stream or delegate cleanup fails. Canonical readers, working directories and shared native-memory leases still receive their cleanup attempt; later failures remain suppressed on the first failure. Operator-owned native handles close on the task thread before their shared-memory leases are released.

Native builds enable rocksdb-state by default. Tests that transfer window partials between stores exercise memory-to-memory and all RocksDB combinations in that build. A --no-default-features build exercises only the memory-store combination; it has no RocksDB implementation or JNI entries. Use the default feature build for RocksDB recovery validation rather than treating a memory-only test run as persistent-backend evidence.

Select the backend with the normal Flink setting:

state.backend.type=tech.streamfusion.state.RocksDBNativeStateBackendFactory

On Flink 1.18, native SQL planning also selects this backend automatically for state.backend.type=rocksdb, the stock embedded factory, and programmatic EmbeddedRocksDBStateBackend or legacy RocksDBStateBackend instances. The configured host backend is retained for ordinary Flink operators. Incremental checkpointing, local paths, predefined options, previously configured RocksDB options and programmatic memory settings keep Flink's precedence; legacy checkpoint storage keeps its precedence too. Custom backend subclasses, options factories and memory factories retain the stock backend and cause keyed SQL to fall back. Configuration supplied through EnvironmentSettings participates in this selection, including RocksDB options in the table's root configuration. After successful selection, the application table configuration's backend selector is consumed so Flink's later pipeline reconfiguration retains the resolved native backend and its configured host delegate. Custom root configurations that cannot be copied retain the stock backend and cause keyed SQL to fall back. Disabled native acceleration leaves the original backend selection alone. Flink 2.2 still requires the explicit StreamFusion factory above to select the Rust-owned RocksDB backend.

The backend translates Flink's state.backend.rocksdb.* options into the Rust RocksDB instance, including local directories, incremental checkpoints, compaction style, level and target-file sizes, write-buffer settings, compression, log level, and TTL compaction query cadence. Incremental checkpoints reuse immutable SST handles; full checkpoints upload the complete live file set.

Write path and memory

The write path is Flink's: state writes go through to the RocksDB memtable (WAL off), amortized to one write per touched key per batch where the operator layout permits. RocksDB's background threads own memtable flushing and compaction. There is no StreamFusion-side write buffer above RocksDB and no forced flush — the checkpoint barrier commits the current batch's residue and takes RocksDB's native hard-link checkpoint, which flushes live memtables itself. Most typed-store values are compact arrow-row bytes; their state-TTL values carry an 8-byte last-write prefix for the compaction filter. Sink candidate TTL instead lives in whole-key metadata and uses bounded key cleanup.

Batch-oriented point reads use RocksDB's optimized MultiGet API on the default column family. The Rust binding borrows the batch's key bytes and returns pinned values, which remain valid through the batch decode and are then released. Operator state owns its decoded values; it does not retain RocksDB cache pins across bundles. Generic keyed stores reserve their decoded working set and conversion workspace against the shared task off-heap pool before hydration and through bundle completion. Operator mini-batch staging has a separate reservation; state metrics include both owners. Checkpoint and canonical-scan completion release the decoded working-set charge as soon as those values are cleared. Persisted keys, row encodings, checkpoint handles and key-group routing are unchanged. Constant AppendFast Top-1 can retain clean, owned decoded winners in a bounded cache governed by task off-heap reservations; the Top-N page describes its admission, metrics and checkpoint behavior. This changes the read path without changing persisted keys, value encodings, or checkpoint files. Range-oriented state continues to use iterators.

The group-aggregate codec builds primitive Arrow columns directly from its accumulators and restores groups from decoded columns, avoiding temporary scalar rows. Arrow-row conversion still allocates and copies between columnar buffers and row bytes; it is a batched conversion, not a zero-copy view. Decimal and string state retain the general scalar-to-array conversion.

Native store memory follows Flink's RocksDB memory control (state.backend.rocksdb.memory.*): one shared block cache and write-buffer manager per slot — sized by memory.fixed-per-slot if set, else the slot's managed-memory share (memory.managed, on by default), else memory.fixed-per-tm at TaskManager scope — with memtables charged against the cache under Flink's memory.write-buffer-ratio split. With none of these configured, each store falls back to its own per-instance write buffers and block cache from the translated options. A shared store sizes its memtable arena quantum from one eighth of the smaller of the configured write buffer and the shared write-buffer-manager capacity, rounded to RocksDB's 4 KiB minimum alignment. This prevents an oversized configured buffer from allocating an entire small slot budget at once and evicting cached reads. The configured write-buffer size and total shared memory budget are unchanged.

Each native operator also holds a lease on that pool until its native handles close. Flink may close a keyed backend while a canceled task is still initializing or restoring its native store. Releasing the backend's slot lease therefore stops new acquisitions but does not free a cache or write-buffer manager still used by a native operator. Native handles close before their operator lease is released. This preserves the existing memory budget and checkpoint formats while preventing recovery from dereferencing a freed pool.

Aligned same-range recovery opens a local copy of the checkpoint database. Rescaling reads only the key groups assigned to the recovering subtask and writes a new local database. This preserves Flink's max-parallelism and key-group redistribution semantics.

Stateful native operators use this backend in tests and production. Operators with a typed direct state codec access RocksDB per key; other native operators persist their existing key-group snapshot payloads in the same Rust-owned RocksDB lifecycle. Doris sink upsert materialization uses record-level candidate, linked-node and matching-index tables. Its point mutations hydrate only affected records; native checkpoints keep the database files, and legacy TTL cleanup seeks past candidate payloads. See sink materialization.

Compatibility boundary

The compatibility target is Flink's SQL state and recovery semantics, not a byte-for-byte clone of EmbeddedRocksDBStateBackend internals. Both backends disable the RocksDB WAL, drain the configured write batch at the checkpoint barrier, create the immutable local snapshot with RocksDB's native Checkpoint API, upload its files asynchronously, and reuse completed-checkpoint SST handles when incremental checkpointing is enabled.

There are also deliberate implementation differences:

Immediate inner JOIN uses record-level point updates and bounded opposite-side probes; other families and mini-batched joins retain complete buckets. The access pattern matters: an earlier storage-only experiment retained full hydration and regressed from 0.270 s to 0.655 s despite reducing hot-key logical writes from 262 MiB to 67 KiB per 64 bundles. That rejected design remains documented in .claude/wontdos/61-record-join-full-hydration.md. The selective implementation migrates old buckets lazily and preserves the logical canonical savepoint format; the matched Q23 test measured 2.80× over Flink. See record-state execution. The remaining join families and broader storage/memory performance validation remain tracked in #244.

• the group aggregate, changelog normalize, keep-last deduplicate, updating join, the three Top-N rankers, the event-time window join, and the aligned event-time window aggregates (tumble/hop/cumulate, including the global two-phase half) perform per-key RocksDB reads and writes (a two-sided operator's tables share one database under table-prefixed keys; window buffers append sequence-keyed entries and windowed aggregate state keys by window end, so watermarks fire with per-key-group range scans); session aggregates join them with key-major session lists that hydrate by prefix scan for merging, and over aggregates across every admitted shape — the unbounded folds and ranking window functions keep one fold row per key; bounded ROWS/RANGE frames keep their per-key sliding buffers as [key][rowtime][arrival]-ordered frame rows (a firing prefix-scans exactly the fired keys, recomputes on the resident buffer, then writes the appended rows and deletes the evicted ones) with per-key deadline stamp rows; DISTINCT aggregates keep insert-only seen-sets as per-element companion rows, point-probed per batch; and the proctime shapes run the same layouts eagerly, ordered by a persisted arrival counter — event-time window rank, the event-time interval join (append-mostly sequence-keyed buffers carrying each row's matched flag), the temporal join (versioned build rows in byte-comparable version order), keep-first deduplicate, and the temporal sort buffer. Group aggregates with MIN/MAX retraction or DISTINCT keep their per-key multisets as companion element tables: a bundle point-reads only the elements its batches name, the running count/sum/extreme rides the main row, changes write through as element-level deltas, and a retraction that removes the current extreme reseeks the ordered element table (a numeric MIN/MAX over an insert-only input needs no multiset at all and runs as a plain running value). Fixed/session COUNT DISTINCT stores likewise separate scalar headers from per-value set/count records. Canonical capture pins those records and converts bounded frames asynchronously; execution can reconstruct the touched window/session. Previous snapshot-backed checkpoints migrate one key-group blob at a time when opened, preserving processing deadlines. Typed stores persist an operator's processing-time timer deadline under a reserved key, so proctime windows, sessions, rank, and the window and interval joins run direct too. Every native operator now reads and writes RocksDB per key; the generic snapshot path remains only as the fallback for a shape whose state has no fixed-type native codec (a DISTINCT over a value type with no faithful type code), which keeps its memory-resident state and replaces key-group snapshot payloads in RocksDB at each checkpoint;
• the native stores' shared cache and write-buffer manager live in StreamFusion's own RocksDB library, sized by the same Flink options and formulas but leased separately from the delegate backend's pool (C++ objects cannot cross the two RocksDB libraries), and the binding exposes no high-priority cache pool (memory.high-priority-pool-ratio is not applied);
• RocksDB options factories are unsupported; and
• Flink's local-recovery snapshot and restore-tuning options are not implemented for native state.

The normal RocksDB compaction, compression, write-buffer, block-cache, logging, TTL-filter cadence, local-directory, and incremental-checkpoint settings listed above are translated. The direct SQL and operator recovery suites cover result, checkpoint, incremental reuse, restore, and TTL semantics; they do not imply identical local I/O behavior for the differences above.

Savepoints and backend changes

Native-format savepoints retain the incremental RocksDB/SST representation. Canonical savepoints materialize the backend-independent StreamFusion canonical state format and use Flink's full-snapshot writer, producing KeyGroupsSavepointStateHandle state. They can restore onto either this backend or the memory backend. A direct store scans and decodes its logical rows for this operation, so canonical savepoints are intentionally full and more expensive than ordinary incremental checkpoints. Qualified stock-interchange adapters instead project live state into Flink's original descriptors and serializers; see canonical state. The frozen stock-canonical retracting MIN/MAX subset (integral/date/time/boolean values) uses direct native RocksDB records, including independent view-entry TTL ages. Wider accelerated aggregate shapes without a stock-canonical contract may still use snapshot-backed execution.

Restoring a canonical savepoint (or legacy raw keyed state) onto this backend decodes the blob key groups once at open and bulk-writes them — rows, TTL stamps, watermarks, sequence high-water marks, and timer deadlines — through the operator's typed store, so a memory-to-RocksDB transition continues on the direct per-key path and its next checkpoint is an ordinary incremental RocksDB handle. A multiset group aggregate's import also spreads each blob's side batches into its companion element tables, and an over aggregate's import fans its buffer rows into the frames table and its seen-sets into the distinct element tables. Only the snapshot-path fallback shapes above restore such state into the generic snapshot store instead.

The two-component timestamp layout changes native row bytes and increments the snapshot metadata version. Older snapshots fail with a format-version error; see the timestamp layout upgrade before upgrading a stateful job.

Flink 1.18 canonical projection

Flink 1.18 and 2.2 use the same disk-backed stock projection for bound canonical contracts. At a canonical-savepoint barrier, direct native RocksDB operators flush pending bundles and pin an independent read view. Flink's async writer converts bounded records with the released stock serializers. Complex map/list/timer layouts reuse the shared disk spool and a temporary stock RocksDB backend; direct scalar layouts stream without that intermediate projection. Neither path loads the whole operator into heap. The largest encoded native or stock value remains a working-memory floor. Overlapping captures retain independent views through completion, cancellation and live-operator closure. Ordinary native checkpoints keep the incremental SST path.

The synthetic-key projection used for private native-format compatibility is distinct from a bound stock contract. Release adapters retain independent serialized keys/key groups where that format needs them. Use the StreamFusion native backend, selected automatically for supported stock RocksDB configurations on 1.18 as described above; 2.2 requires the explicit factory. Unverified custom backends still decline keyed native operators during planning. See canonical state for the frozen operator scope and Flink line compatibility for release-specific behavior.

Failed and aborted checkpoints

If the native snapshot fails before returning its manifest, the backend removes its partial hard-link directory. If an upload or shared-file reuse notification fails, newly uploaded files are discarded and the attempt cannot become an incremental reuse base. Previously confirmed shared files stay owned by Flink's checkpoint coordinator. An aborted checkpoint is removed from reuse bookkeeping; discarding its returned handle releases its files. A subsequent checkpoint can succeed without reopening the operator.

Regression tests inject failures after the first upload, during native snapshot preparation and in the storage factory's reuse callback, and check both file cleanup and the next checkpoint.

Changelog state backend

Flink's changelog state wrapper is unsupported for native keyed state on both 1.18 and 2.2. The wrapper cannot pass the canonical native-state context handling, and log replay recomputes key groups from key hashes rather than preserving our explicit partition/group pairing. It also requires an abstract Flink keyed backend, which the Rust-owned backend does not implement. Query admission alone therefore does not establish checkpoint/recovery support.

The planner declines keyed SQL with a changelog-specific reason and preserves the stock backend and changelog setting. Stateless native SQL remains eligible. Use state.changelog.enabled=false with StreamFusion's native state backends. This is unrelated to SQL insert/update/delete changelog rows, which supported native SQL operators handle normally.

The controlled record-layout and probe-reuse measurements show that bounded reuse is a memory safeguard, not a universal speedup. Unique-key inputs favored the old bucket store in the state-level diagnostic; groups larger than the former 8 MiB allowance paid repeated lazy-scan costs. That fixed ceiling has been removed; probe reuse is governed by reservations from the shared native memory pool. Immediate inner joins still persist only changed records, independent of whether probe reuse fits.

Linux io_uring reads

Linux builds can opt into RocksDB's io_uring SST MultiRead implementation with the Cargo feature streamfusion/rocksdb-io-uring. Install the system liburing development package and pkg-config first (for example, apt-get install liburing-dev pkg-config on Ubuntu), then build from native/:

cargo build --release --workspace --features mimalloc,streamfusion/rocksdb-io-uring

For Maven builds, pass -Dnative.cargo.args="build --release --features mimalloc,streamfusion/rocksdb-io-uring" in addition to the usual -Pbench or -Prelease profile. Feature-enabled Linux libraries also require the liburing runtime library on each deployment host (for example liburing2 on Ubuntu); missing runtime libraries prevent loading the native library before any read fallback applies. Feature-enabled Linux libraries attempt io_uring automatically. Leave SF_ROCKSDB_IO_URING unset or set it to true to enable it; set it to false in each TaskManager's environment before starting its JVM to disable it. Other values also disable the integration. The setting is cached on the first use in each process; change it by restarting the process, not by rebuilding. Builds without the feature keep their existing behavior and do not need liburing. On non-Linux targets the feature does not activate the Linux hook or require liburing.

This selects upstream RocksDB's concurrent SST batch reads. State access remains synchronous to the operator, and WriteBatch, compaction, checkpoints, and disk-cache flushes keep their existing write paths. Successful use also requires a kernel and sandbox that allow io_uring_setup; if RocksDB cannot initialize a ring, it falls back to ordinary reads. Building with the feature alone is not evidence that the runtime used io_uring.

The integration uses linker wrapping of the released binding's always-enabled hook, retaining the registry dependency without patching or forking it. The feature stays opt-in: a 2M-event, native-RocksDB generated-row/blackhole comparison showed a modest +3.93% best-of-three geometric-mean throughput gain, with a +0.80% median query gain and several regressions. See method, limits and all trials.

Bounded OVER floating SUM0 frame values retain the running accumulator needed for Flink's add/retract rounding. Extended layouts carry a version prefix before Arrow row encoding. The reader accepts old unmarked frame values using the previous layout and maps new fields to typed NULL until the operator reconstructs them from retained rows. The same dual decoder serves normal reads and canonical savepoint export. Existing frame keys and unrelated state layouts remain unchanged. See OVER checkpoint semantics for the limit on reconstructing arithmetic history discarded by old writers.

Canonical conversion reservations

A captured read context carries the operator's memory authority onto Flink's asynchronous writer. Shared scan batches reserve copied keys/values, vector capacity and the continuation key before allocation. Refused frames retain their prior cursor position and can retry after memory becomes available. Frames split before crossing their byte limit, except for an individual oversized record. Direct scalar cursors include a decoding estimate. Window-rank buckets and DISTINCT membership keys are borrowed from pinned iterators; their decoded transfer frames reserve work before construction. The encoded RocksDB value still has an intrinsic memory floor, and estimates for other conversion stages remain part of the accounting audit.

Retained nested rows

Buffered event-time deduplication, event-time sort, window rank/deduplication, OVER retained input and interval/temporal join payloads share a retained-row codec for scalar, ROW, ARRAY and MAP payloads. It reuses Arrow's row encoding, treating map payloads as their equivalent list/struct buffers without copying the columns or changing entry order. Decode restores the declared child fields and metadata. Scalar row bytes remain unchanged. These layouts use pinned native RocksDB canonical capture instead of the former memory capture with snapshot-backed checkpoints. Legacy databases migrate one opaque key-group partition at a time; this compatibility path retains a partition-sized memory floor. Keys and sort semantics retain their existing independent admission rules.

Scalar window keys

Direct fixed-window and session stores decode newly admitted TINYINT/SMALLINT, TIME, BINARY/VARBINARY and both interval key families using their natural Arrow types. TIME uses millisecond Time32 in the shared row codec; month intervals use Int32 and day-time intervals use Int64. Existing INT window keys retain their widened Int64 encoding. No previously admitted type changes its persisted encoding in this extension. Two-restore SQL comparisons cover nullable composite keys on memory and direct RocksDB for fixed/session windows, bounded ROWS OVER, interval joins and window joins on both supported Flink releases. See scalar key coverage.

FLOAT/DOUBLE partition keys also use their natural Float32/Float64 carriage in direct fixed-window and session state. The floating extension's two-restore comparisons regenerate distinct NaN payloads after source deserialization and project their raw bits after the stateful operation; source/sink canonicalization cannot collapse the oracle. Signed zero, infinities, subnormals and NULL are covered alongside ordinary finite values.

Hidden OFFSET row-kind mutations

General retracting Top-N with OFFSET and no projected rank follows released Flink's RocksDB MapState copy semantics. Emitting a row kind mutates a transient deserialized row; it changes stored state only when the owning list is explicitly put. Insertions write their source list after emission, while retractions write it before shifting subsequent sort groups. The native whole-partition codec preserves those logical per-list write boundaries even when it flushes one partition value at a physical batch boundary. Heap state keeps its shared-reference behavior. Consequently, the raw changelogs can differ between host backends; parity uses the same configured backend on both engines.

No type code, checkpoint field or persisted byte layout changes. Existing stored row-kind history is retained, including history written by the preceding native implementation's heap-style behavior on RocksDB. That past stock-RocksDB state cannot be inferred afterward. Fresh jobs and checkpoints made with the corrected behavior match the selected backend in sixteen SQL scenarios per released Flink line, including ties and two failure/restores. See Top-N for the affected shape and historical-state limitation.

Retained map and multiset payloads

Top-N and regular updating joins export declared MAP/MULTISET schemas through the existing tagged List-of-entry-struct view. This changes schema interpretation while borrowing the original Arrow buffers at the JNI boundary. The row codec still owns its encoded retained payloads. Canonical IPC and direct RocksDB restoration use the same physical schema, recursively through ARRAY/ROW/maps. Previously admitted scalar schemas and bytes are unchanged; these MAP-bearing shapes previously fell back. Updating Top-N/LIMIT and MAP-bearing updating join inputs still use stock execution until the retained-row contract in #437 is implemented. Join/partition/order/unique-key contracts remain separately checked.

Grouped multiset companion tables reserve identifier zero for the main row and encode aggregate positions 0–254 as identifiers 1–255. The codec rejects larger companion positions before opening a database; it does not truncate an index into a different table. Scalar-only wide rows do not need companion IDs. Canonical DISTINCT planning falls back to the captured stock owner for an unencodable view layout; see grouped-state admission. Wider native companion encoding remains pending in #569.

Captured JVM UDF aggregate accumulators and data views remain Flink managed state. With the RocksDB configuration they use the released embedded RocksDB delegate, rather than the native grouped-table codec. The managed UDF canonical scope keeps this distinction explicit; native grouped-table coverage does not qualify user accumulator state or its full restore matrix.

Captured one-phase mini-batch UDFs retain this same managed RocksDB delegate and released serializers. Their transient bundles and optional cached accumulators drain before canonical barriers; the one-phase qualification includes both stores, backend changes, TTL ages and ordinary recovery. This is separate from the native grouped-table codec and from the global owner's separately qualified two-stage contract.

Captured local/global UDFs use the same released managed backend at the global owner. Local partials are transient Arrow events serialized with the original stock local serializer and drain before canonical barriers; they do not use the native grouped-table codec. The two-stage qualification includes heap/RocksDB, backend changes, original TTL ages, checkpoint returns, local/global UDF fault recovery and 56 complete-job handoffs. Wider user/key/type and failure matrices remain open.

Captured managed UDF key qualification additionally compares the original canonical key and view bytes through heap/RocksDB backend changes for wide decimal, nanosecond timestamp and selected composite/nested/container keys. These operators use the released managed backend and key serializer, not a new native RocksDB key format. The combined component passes 1,190 focused checks and 168 full-job handoffs, including selected heap/RocksDB changes, no-input returns, native checkpoint recovery and 1→3→2 key redistribution. Original nested binary-row key payloads survive the managed Arrow boundaries. See the exact key scope and remaining requirements. Benchmarking remains deferred.

The all-operator backend-transition controls require supported window DISTINCT sets to use direct native RocksDB, including ordinary checkpoint and memory/disk transitions. Their original continued-output assertions remain in place; the former snapshot-backed route expectation is obsolete for these already-admitted layouts.

Optimized Linux builds resolve native database paths into a caller-owned realpath buffer before copying the result into the DSO's allocator. The snapshot registry still uses resolved filesystem identities, including symlinks. This avoids returning a host library's allocated path to a different allocator during ordinary RocksDB initialization or asynchronous snapshot lookup. The release image SQL smoke exercises this path with the production allocator enabled.

Ordered native scans merge pinned iterator heads using a heap of indices. A scan comparator requires each input range to already be monotonic under that comparison; layouts with an unbiased signed field split its negative and nonnegative spans into separate ranges. This preserves logical ordering without copying an entire key or payload backlog into the merge.