Memory management

StreamFusion uses Flink's normal taskmanager.memory.task.off-heap.size setting as the single memory authority for one TaskManager process. Every StreamFusion task in that process shares the same pool, so spare capacity is available to whichever operator or connector needs it rather than being divided into fixed per-slot or per-operator slices.

The setting must be greater than zero. It is already part of Flink's TaskManager process-memory model, so increasing it also increases the total process memory Flink derives unless total process memory is configured explicitly. StreamFusion does not reserve Flink managed memory; the one consumer sized from the managed-memory share is the RocksDB backend's native store pool, which follows Flink's state.backend.rocksdb.memory.* semantics (see the RocksDB backend) but allocates natively outside the managed pool.

What is charged

The shared cap currently covers:

• native operator state and DataFusion working-memory reservations; and
• Arrow buffers allocated by the shared Arrow C Data Interface allocator.

Boundary workspaces are reserved before evaluation and released with their owner. DataFusion working-state reservations use its reported capacity estimates; grouped hash state reports growth after processing an input batch and handles a denial through its existing spill path. These are memory reservations, rather than a per-allocation RSS limit. Arrow allocations are charged and released at their actual allocation lifetime; operator and zero-copy batch handles also release ownership on normal close and task cancellation. Native DataFusion pools cache the Java authority class on the calling thread, so worker callbacks retain the correct application classloader. The internal bounded runtime additionally keeps its owner registration alive through the pool's final reference, including asynchronously cancelled workers; it does not reclaim their balance at Java operator close. Bounded sort copies borrowed Java input into compact native buffers under a reusable reservation before DataFusion retains it. Output uses the reverse handoff: native reserves the copy, Java imports/charges its Arrow buffers, then acknowledges that transfer so native can retire its copy reservation. Temporary overlap covers actual handoff allocations rather than charging every retained sort run through both allocators. Spillable sort state leaves headroom within the same process cap for the next input/copy/output; other consumers may use that headroom. See the bounded runtime for the lifetime and exhaustion controls. Verified fixed-width global batch aggregates consume imported input directly and retain only scalar accumulator values. The original Arrow allocation remains charged until consumption releases its C Data reference; an explicit expression workspace covers cast, literal, fixed decimal, TIME normalization and FILTER temporaries before evaluation, without a retained-sort ownership copy. Output uses the same compact, acknowledged handoff even when its aggregate schema differs from the input. Grouped aggregate projection space remains charged across every output/replay batch. After projection temporaries retire, a sibling MemoryReservation.split transfers that budget to the compact result/copy handoff. Acknowledgement retires the output charge and the next poll replenishes projection space. This shares the existing working budget instead of taking a second reservation while spill readers hold memory. DataFusion's grouped state and spill/replay use the same pool; no additional operator cap or overcommit allowance is introduced. Mini-batch cleanup skips native flushes when initialization has not acquired a state handle or completed bundle setup. This also applies after the handle is closed: cleanup must preserve the original initialization failure rather than enter JNI with an absent handle. Partially acquired handles and reservations are still released by normal close; initialized operators retain their normal end-of-input and checkpoint flushes. Before a row-to-Arrow conversion hands its batch to a consumer, the converter owns every allocated vector. A rejected allocation or a row conversion error closes those vectors, including the changelog sidecar, before propagating the original failure. Cleanup errors are attached as suppressed exceptions so they cannot hide the allocation or conversion error.

Stateful operators register reservations, native handles, predicate bindings and coalescers as they acquire them in a task-thread resource scope. Cleanup releases dependent resources first and attempts every release, even if another release fails. Later failures are suppressed on the first exception. Canonical readers close before native state, and the host operator closes even after failed native teardown. Fresh window output roots are released immediately if downstream emission fails before claiming them; roots already claimed downstream keep their transferred ownership. Runtime operator serialization identifiers remain frozen when bounded-output hooks are added. Calc/filter bindings and partially initialized state use the same ownership rule. These scopes are transient; they do not change deployed operator fields or state encodings. They do not dispose task-thread native handles from a concurrent cancellation callback.

Parquet sink writer allocations are not yet connected to this authority. Small allocations made inside native libraries, allocator metadata, thread stacks, loaded libraries, and other process overhead are also outside the reservation pool. The cap is therefore an allocation authority for the large StreamFusion-controlled consumers above, not an RSS or container hard limit. Leave process headroom for those costs and for the JVM.

Arrow allocation during startup

A connector can create its decoder schema before another task initializes the process memory budget. StreamFusion tracks those early Arrow allocations and adopts the live balance when the TaskManager budget becomes available. Buffers already released do not consume that budget. Initialization publishes the capacity only after the existing buffers are charged, so a source thread can release them safely while other tasks start.

If the configured cap cannot cover the live startup buffers, initialization fails with the normal memory-limit exception. Their ownership remains intact for cleanup or a later initialization with sufficient capacity. Startup schema and buffer overhead contributes to the same cap and high-water mark as later allocations; no separate configuration or per-operator allocator is introduced.

Arrow schema and buffer ownership

C Data import transfers the array release callback using the existing Comet-style ownership pattern. Imported arrays retain their input buffers after producer objects close. Field and root schema metadata survive both ordinary and cached-schema import; cached declarations remain immutable.

Flink sources can carry actual NULL slots beneath a NOT NULL logical hint. Import widens only those top-level physical Arrow fields before batch coalescing, shuffle or IPC validates the batch. It shares the same arrays and validity buffers, keeps the SQL logical type unchanged and leaves Flink's later sink constraints in place. No additional data copy, reservation or release callback is introduced. Calc column projections and JVM callback arguments also retain field metadata needed to reconstruct Flink's collection copy categories.

Exhaustion and spilling

A reservation that would cross the TaskManager-wide cap is denied with NativeMemoryLimitException; its message identifies the consumer and taskmanager.memory.task.off-heap.size. Arrow and fixed-reservation failures include the capacity, aggregate reserved bytes, and available bytes sampled when the request was rejected. Native JNI reservation denials log the same snapshot together with the owner id and its registered category/task name; the native exception retains that owner id. Optional caches may recover from a denial, so a denial log alone does not imply that the task failed. Other tasks can release memory immediately afterwards; later metric samples need not match the rejection snapshot.

With fine-grained resource scheduling, increase the affected task/vertex's off-heap resource request and let the scheduler derive the TaskManager capacity. Verify the actual runtime taskmanager.memory.task.off-heap.size against nativeOffHeapCapacityBytes; a job-level setting can be superseded by the scheduler's process specification. Increasing a JVM direct-buffer maximum or RocksDB's managed-memory budget alone does not enlarge this shared pool.

The in-memory state backend cannot spill, so a denied growth request fails the task and Flink's configured restart policy applies. Under the RocksDB backend, native state is written through to RocksDB at every batch, so an operator's resident working set stays bounded by its batch size; RocksDB's own memtables and block cache hold the rest, governed by Flink's state.backend.rocksdb.memory.* options rather than the task off-heap pool (see the RocksDB backend memory section).

Generic native RocksDB keyed stores own their decoded working-set reservations. Hydration reserves key and decode space before decoding persisted values, and bundle completion keeps live state and encoding workspace charged until writes finish and decoded values are released. A failed write retains the working-state charge until the store closes. Operators account mini-batch preimages and other staging separately, and nativeStateBytes includes the live store-owned state. This prevents an immediate bundle from allocating and clearing its decoded state between two operator-accounting samples. Raw codecs estimate their owned representation from their released encoding; Arrow-row codecs reserve overlapping input, decoded columns and owned values. These remain size estimates, not an allocator-level RSS limit.

Canonical scalar group imports keep the existing 1,024-row transfer frames but commit persistent arithmetic and typed headers in working sets of at most 64 rows. The complete frame is validated, including existing target keys, before any of its working sets is installed. Each commit releases decoded state and encoding workspace before the next working set, and the operator does not retain a second charge for store-owned imported groups. Wide keys or aggregate state can still exceed the pool; this is bounded staging, not a spill mechanism. The memory backend continues to charge all retained groups.

Read-through point access during an active bundle requires the key to have been hydrated by one of its input batches. A hydrated absent key returns no value; an unknown key fails the internal contract instead of silently masquerading as absent persisted state.

RocksDB Top-N reconciles its reservation with the current read-through working set, not the cumulative size of persisted partition keys. Append-only (including Top-1), retracting and update-fast rankers check the live working set before committing a bundle and return its charge after decoding the output and releasing that working set. An active mini-batch still charges hydrated rows and retained preimage payloads; payloads shared with the working set are counted conservatively in that estimate. The separately reserved winner cache is not included a second time in the completed operator-state charge. The memory backend retains its incremental charge for every resident partition. Checkpoint/state formats and key-group assignment are unchanged.

RocksTopNMemoryAccountingTest reproduces the shared-pool lifecycle through the production JNI entry points: 100 disjoint-key batches of 16 rows complete under a 128 KiB cap in both immediate and mini-batch modes. Run it with the repository's configured Flink/JDK build:

mvn -pl streamfusion-runtime -am -Dtest=RocksTopNMemoryAccountingTest \
  -Dmaven.javadoc.skip=true -Dsurefire.failIfNoSpecifiedTests=false test

Native topn::rocks_memory_tests additionally covers all three strategies, existing-key hydration, wide preimages, real budget denials, cache ownership and checkpoint/TTL recovery.

Constant append Top-1 additionally retains clean decoded winners under a 64 MiB FIFO limit. That cache uses a separate reservation against the same task off-heap pool as the operator. Cache bytes include retained row buffers, payload ownership, keys and container overhead; a value header or a sort-row header already held in its container is counted once. Cache growth reserves its estimated upper bound before retaining values; unused reservation is returned after eviction. Memory denial clears and disables the cache rather than failing the query; ordinary write-through state remains in RocksDB. Fresh unique keys bypass cache admission, and checkpoint/canonical scans clear it. The topn.cache.bytes gauge reports its accounted estimate. The large allocation authority remains separate from RocksDB's managed memtable/block-cache pool.

Sizing

Size taskmanager.memory.task.off-heap.size for the peak aggregate across every StreamFusion task running in one TaskManager:

• Add peak Arrow in-flight buffers plus native operator state and DataFusion working memory.

Flink's Kafka consumer and producer buffers are Java-client memory and use Flink's normal JVM memory model. Use the native high-water mark to size the Arrow and operator working set.

Metrics

StreamFusion registers these gauges with each native operator or connector metric group:

• nativeOffHeapCapacityBytes — configured TaskManager task off-heap capacity;
• nativeOffHeapReservedBytes — current aggregate reservations;
• nativeOffHeapAvailableBytes — remaining shared capacity;
• nativeOffHeapPeakBytes — process-wide reservation high-water mark;
• nativeOffHeapDeniedReservations — rejected growth requests;
• nativeArrowAllocatorBytes — current shared Arrow allocator footprint; and
• nativeStateBytes — the operator's sampled native state footprint.

See Configuration for the related settings and RocksDB backend for the RocksDB write path, memory pool, and checkpoint semantics.

Reused-JVM test isolation

Allocator and pool-limit tests save the process capacity before each test and restore it after releasing their reservations. They must not leave the pool unconfigured for subsequent operator harnesses, whose mock TaskManager configuration has no off-heap setting. Test-only resets reject live reservations rather than erasing their accounting, so restoration cannot hide a leak. The production requirement for a positive task off-heap setting is unchanged.

Bounded Parquet input

The optional native filesystem Parquet source shares this same authority. After reading the footer and before decoding a row group, it reserves the largest sum of compressed and uncompressed projected column chunks in a row group. It retains that decoder workspace reservation until the reader closes; imported Arrow output is separately accounted by the shared allocator until its final consumer closes it. Unprojected chunks are not read or reserved. The existing INT96 decoder's paired reader multiplier remains unchanged, although the initial filesystem source does not admit timestamp columns. Reader cancellation closes queued, unconsumed Arrow batches promptly. No source-specific memory budget is introduced.

Window final output

Production aligned-window, session and window-rank final output uses synchronous bounded receivers rather than one Arrow batch containing an entire watermark backlog. Decode frames and output assembly retain separate reservations until their consumers release them. Either can flush early when a larger reservation is denied; a single row that cannot fit still fails normally. The maximum batch has 1,024 rows and a 4 MiB estimated working-size target; one larger row is admitted when the pool can cover it. Released Java Arrow buffers remain charged independently after handoff.

RocksDB firing retains window-end metadata and key-group iterator indices, not all closed or future row payloads. The window-end inventory and the largest group still have to fit. Legacy single-result JNI helpers retain their compatibility behavior and can collect multiple emitted batches; the production fixed-window, session and local-partial operators use the bounded receiver entry points.

Session firing merges key-group scans by grouping key and session start. It keeps bounded key/end frames for calendar decisions and decodes only closed session payloads. Key-header, decode and final-output workspaces use the same shared cap and can flush independently under pressure; future payloads remain in RocksDB. Session input still hydrates a touched key's neighboring sessions for merging, so a single large key/session remains a working-memory floor.

Local partial transfer moves resident accumulators into bounded output frames and releases each group's state charge as ownership moves. Partial/decode workspace reservations cover the transfer until its receiver returns. The current resident window still requires key sorting, and the largest individual accumulator remains a necessary floor. Barrier drains emit every open slice without advancing the watermark; no pending transfer cursor becomes checkpoint state.

Window rank copies bounded serialized row frames from pinned RocksDB values before decoding; a large retained group does not force full-group hydration during firing. Small groups coalesce within a frame, while rank numbers continue across frames of one group. Negative and nonnegative window starts are scanned separately before merging so the existing signed timestamp order and persisted bytes remain compatible. The resident memory backend transfers retained row ownership; window inventory and sorted partition metadata remain working-memory floors.