You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@spark.apache.org by Chang Chen <ba...@gmail.com> on 2026/05/20 02:47:16 UTC
[DISCUSS] Native memory cooperation: let BlockManager manage external-engine payloads
Hi all,
I would like to discuss a long-standing pain point for native execution
engines(Gluten) built on Spark: *the memory held by the native side cannot
participate in Spark's storage memory accounting as a first-class citizen*.
I will summarize what has been tried, what's still missing, and a concrete
mechanism I'd like the community's feedback on.
1. The problem
When a native engine wants to cache columnar payloads next to Spark —
native vectors, Arrow buffers, native file caches, etc. — the memory lives
outside the JVM heap and is allocated through a separate allocator. Two
things go wrong today:
*(a) The two pools don't share a budget.* The native allocator's budget is
configured independently from spark.memory.storage /
spark.memory.offHeap.size. As a result, the user has to statically
partition memory between Spark and the native engine, and neither side can
borrow from the other under pressure. Practical outcomes: the native cache
spills aggressively while Spark storage sits idle, or vice versa.
*(b) Even when df.persist() is used to cache native results, the payload is
copied through the JVM.* The columnar cache SPI today (CachedBatchSerializer)
requires the produced CachedBatch to live as JVM-owned bytes so that
MemoryStore can hold it as a DeserializedMemoryEntry / SerializedMemoryEntry.
For any native engine, this means serializing each batch from native memory
into a JVM byte[] on the write side and deserializing back into native
memory on the read side. Every cache hit pays the cost of two off-heap <->
on-heap copies (and any associated cross-boundary call overhead). The point
of a native engine — staying off-heap and zero-copy — is defeated at the
cache boundary.
2. Prior attempts and why they fell short
I want to credit two prior efforts that explored adjacent designs:
*SPARK-48694 <https://issues.apache.org/jira/browse/SPARK-48694> /
PR #47067 <https://github.com/apache/spark/pull/47067>* introduced an
ExternalMemoryStore parallel to MemoryStore, sharing the storage pool
budget, with three new configs (spark.memory.external.cache.enabled,
spark.memory.external.storageFraction,
spark.memory.storage.preferEvictExtCache). The eviction policy under
pressure is a *static global boolean* — always prefer one store over the
other — rather than unified per-block LRU. It also duplicates a fair amount
of MemoryStore infrastructure into a parallel surface, and is shaped around
the "file cache" use case rather than a general external-object lifecycle.
The PR was eventually closed as stale.
*SPARK-56918 <https://issues.apache.org/jira/browse/SPARK-56918> /
PR #55953 <https://github.com/apache/spark/pull/55953> (myself, a few days
ago)* went a different way: I added a ManagedConsumer SPI to
UnifiedMemoryManager, by analogy with UnmanagedConsumer. The native side
registers as a consumer, Spark calls back to shrink it. Two problems
surfaced in review:
1. *The accounting is not symmetric.* Consumers are an execution-memory
concept; they are asked to spill under execution pressure. Storage blocks
have their own LRU within the storage pool. Bolting a "managed consumer"
onto storage means the native side is always the first to be reclaimed
under any pressure — there is no fair LRU between native and JVM entries.
2. *It does not address the serdes problem.* Consumers register a
*budget*; they don't give Spark anything to evict at batch granularity.
df.persist() still has to go through the JVM byte path.
Neither attempt gives the native engine a way to say *"I have this opaque
payload, please count it against the storage pool and tell me when to
release it."*
3. What I'd like to propose
The core idea is small: *let BlockManager register and evict blocks whose
bytes live outside the JVM*, then build the cache SPI on top.
Layer 1 — a "native block" primitive in core
A new BlockManager entry point lets a caller register an executor-local
block with three pieces of information:
- a new NativeBlockId subtype, so the block is unambiguously
distinguishable from JVM-resident blocks;
- the number of bytes to reserve in the storage memory pool (on- or
off-heap, the caller's choice);
- a release() callback that Spark invokes when the block is evicted or
removed.
If storage pressure forces eviction, the existing
MemoryStore.evictBlocksToFreeSpace algorithm picks across *all* entries —
JVM-resident and native — using the same LRU policy. No preference flag, no
parallel store. When a native entry is chosen, release() is invoked exactly
once and the reserved bytes return to the pool.
Read paths (getLocalValues, getLocalBytes, getLocalBlockData, get,
replicateBlock) refuse native blocks with a clear error: the payload is
opaque to Spark and must be read through the owning engine's own API. The
master is never notified, because native payloads are inherently
executor-local and cannot be fetched over the network.
So BlockManager becomes a unified registry for "this many bytes of storage
are committed; call me back when you need them released," regardless of who
owns the actual bytes.
Layer 2 — a native variant of the columnar cache SPI
On top of layer 1, CachedBatchSerializer grows a small opt-in:
- a flag indicating that the serializer produces batches whose payload
is owned by an external engine;
- a NativeCachedBatch trait carrying release() plus a callback that
receives the NativeBlockId Spark assigns;
- a lightweight handle type (block id + row count + size + optional
column stats) that flows through the cached RDD in place of the original
payload.
When the flag is set, the cache builder takes a different fork. For each
batch produced by the serializer:
1. Spark reserves storage memory via the layer-1 entry point under a
deterministic id.
2. The serializer is informed of the id, so it can populate its own
executor-local lookup table mapping id back to the engine payload.
3. The Spark-visible CachedBatch is just the handle; the handle RDD is
persisted at MEMORY_ONLY (handles are tiny).
4. On scan, the serializer's convertCachedBatchTo* methods resolve the
handle's block id via the lookup table and hand back the native payload
directly.
Because the handle still carries column statistics in the existing
SimpleMetricsCachedBatch schema, partition pruning continues to work
without changes to the cache scan operator.
The cache-eviction story is now closed-loop: Spark LRU picks the oldest
entry (JVM or native), MemoryStore calls the registered release(), the
native engine frees its bytes, and the next scan recomputes the affected
partitions.
4. Why this is more general than the previous attempts
- *Unified LRU, not a preference flag.* Native and JVM blocks live in
the same entries map; the existing eviction algorithm handles them
uniformly. No new tunable for users to misconfigure.
- *No new configs.* It reuses
spark.memory.{offHeap.enabled,offHeap.size,fraction,storageFraction} exactly
as today.
- *No MemoryStore clone.* Only a new entry variant alongside the
existing two — small surface, same lifecycle hooks.
- *Not bound to a single use case.* "Native file cache" and "native
columnar cache" are both just consumers of the same core primitive.
Anything with an opaque, off-heap payload and a release() semantics fits.
- *Solves the serdes problem.* Native engines can opt into the SPI and
have df.persist(MEMORY_ONLY) stay zero-copy end to end. The columnar
cache SPI stops being a forced serialization boundary.
5. Trade-offs to discuss
A few things worth surfacing:
- *No replication, no spill, no remote fetch.* Native blocks are
executor-local and opaque to Spark; the SPI explicitly rejects storage
levels that request disk or replication > 1. Engines that want spill or
replication have to implement it themselves on top.
- *Dataset.cache() default is MEMORY_AND_DISK*, which the native path
would reject. Users must call persist(MEMORY_ONLY) explicitly. We could
relax this with a silent downgrade, or expose a new canonical storage-level
constant. I lean toward "be explicit" for the initial cut but happy to hear
opinions.
- *Cluster-mode unpersist cleanup is best-effort.* Today the cache
builder only touches the driver's local BlockManager; remote executors
will release their native blocks via LRU or shutdown. A driver -> executor
cleanup RPC is a natural follow-up.
6. What I have ready
I have a 3-commit stack on top of master that I'd like to put up for review
once a JIRA umbrella is filed:
1. [CORE] Add native block storage for opaque external-engine payloads —
the new BlockId subtype, the MemoryStore variant, the BlockManager entry
point, and tests covering reservation, eviction, release-on-evict, and
release-throws.
2. [CORE] Harden native block storage — type enforcement, duplicate-put
detection, replicate/master guards.
3. [SQL] Wire the SPI into the columnar cache builder, with end-to-end
tests using a fake native serializer that exercises the round trip through
df.persist(MEMORY_ONLY) and eviction.
7. What I'm asking
1. Is the community OK with BlockManager growing a "register external
payload" surface? I think this is the right home for it — the alternative
is yet another sidecar registry — but I want to confirm before pushing PRs.
2. Reviewers familiar with MemoryStore / UnifiedMemoryManager /
CachedBatchSerializer who can weigh in on the API shape: memory-mode
choice, lock semantics inside release(), the registration callback, and
how strict the disk / replication rejection should be.
3. Anything I'm missing from a prior discussion that already converged
on a different direction.
Thanks for reading. I'd love to hear what people think before I take this
further.
Best, Chang Chen
Re: [DISCUSS] Native memory cooperation: let BlockManager manage external-engine payloads
Posted by Kent Yao <ya...@apache.org>.
Hi Chang,
One concern on Layer 1 — I don't think the "unified LRU" property holds as stated.
MemoryStore.entries is an access-order LinkedHashMap. Per the LinkedHashMap contract, only get/put/compute/merge refresh a key's position; containsKey and iteration do not. Inside MemoryStore, the only methods that refresh position are getBytes and getValues, both called from the JVM read path in BlockManager.
The proposal makes those read paths refuse native blocks — the payload is opaque, only the owning engine can read it. So a native entry inserted at T0 has its access-order position fixed at T0. Any subsequent read of any JVM block moves that JVM block ahead of it. After a few read cycles, the native entry is permanently the oldest in iteration order, and evictBlocksToFreeSpace walks iteration order, so the native block gets picked first under any pressure.
That isn't unified LRU; it's "native blocks are always among the oldest entries." Fixing it requires exposing something like MemoryStore.touch(blockId) that the native side calls on every read — one synchronized lookup per read, which is non-trivial cost for a hot Velox/Gluten cache and deserves to be in the design conversation explicitly.
Either commit to the touch API and accept its cost, or describe what eviction policy native blocks actually get and why it's still acceptable. Both are legitimate paths, but the current proposal hangs on a property that doesn't follow from the mechanism alone.
Best,
Kent
On 2026/05/20 02:47:16 Chang Chen wrote:
> Hi all,
>
> I would like to discuss a long-standing pain point for native execution
> engines(Gluten) built on Spark: *the memory held by the native side cannot
> participate in Spark's storage memory accounting as a first-class citizen*.
> I will summarize what has been tried, what's still missing, and a concrete
> mechanism I'd like the community's feedback on.
> 1. The problem
>
> When a native engine wants to cache columnar payloads next to Spark —
> native vectors, Arrow buffers, native file caches, etc. — the memory lives
> outside the JVM heap and is allocated through a separate allocator. Two
> things go wrong today:
>
> *(a) The two pools don't share a budget.* The native allocator's budget is
> configured independently from spark.memory.storage /
> spark.memory.offHeap.size. As a result, the user has to statically
> partition memory between Spark and the native engine, and neither side can
> borrow from the other under pressure. Practical outcomes: the native cache
> spills aggressively while Spark storage sits idle, or vice versa.
>
> *(b) Even when df.persist() is used to cache native results, the payload is
> copied through the JVM.* The columnar cache SPI today (CachedBatchSerializer)
> requires the produced CachedBatch to live as JVM-owned bytes so that
> MemoryStore can hold it as a DeserializedMemoryEntry / SerializedMemoryEntry.
> For any native engine, this means serializing each batch from native memory
> into a JVM byte[] on the write side and deserializing back into native
> memory on the read side. Every cache hit pays the cost of two off-heap <->
> on-heap copies (and any associated cross-boundary call overhead). The point
> of a native engine — staying off-heap and zero-copy — is defeated at the
> cache boundary.
> 2. Prior attempts and why they fell short
>
> I want to credit two prior efforts that explored adjacent designs:
>
> *SPARK-48694 <https://issues.apache.org/jira/browse/SPARK-48694> /
> PR #47067 <https://github.com/apache/spark/pull/47067>* introduced an
> ExternalMemoryStore parallel to MemoryStore, sharing the storage pool
> budget, with three new configs (spark.memory.external.cache.enabled,
> spark.memory.external.storageFraction,
> spark.memory.storage.preferEvictExtCache). The eviction policy under
> pressure is a *static global boolean* — always prefer one store over the
> other — rather than unified per-block LRU. It also duplicates a fair amount
> of MemoryStore infrastructure into a parallel surface, and is shaped around
> the "file cache" use case rather than a general external-object lifecycle.
> The PR was eventually closed as stale.
>
> *SPARK-56918 <https://issues.apache.org/jira/browse/SPARK-56918> /
> PR #55953 <https://github.com/apache/spark/pull/55953> (myself, a few days
> ago)* went a different way: I added a ManagedConsumer SPI to
> UnifiedMemoryManager, by analogy with UnmanagedConsumer. The native side
> registers as a consumer, Spark calls back to shrink it. Two problems
> surfaced in review:
>
> 1. *The accounting is not symmetric.* Consumers are an execution-memory
> concept; they are asked to spill under execution pressure. Storage blocks
> have their own LRU within the storage pool. Bolting a "managed consumer"
> onto storage means the native side is always the first to be reclaimed
> under any pressure — there is no fair LRU between native and JVM entries.
> 2. *It does not address the serdes problem.* Consumers register a
> *budget*; they don't give Spark anything to evict at batch granularity.
> df.persist() still has to go through the JVM byte path.
>
> Neither attempt gives the native engine a way to say *"I have this opaque
> payload, please count it against the storage pool and tell me when to
> release it."*
> 3. What I'd like to propose
>
> The core idea is small: *let BlockManager register and evict blocks whose
> bytes live outside the JVM*, then build the cache SPI on top.
> Layer 1 — a "native block" primitive in core
>
> A new BlockManager entry point lets a caller register an executor-local
> block with three pieces of information:
>
> - a new NativeBlockId subtype, so the block is unambiguously
> distinguishable from JVM-resident blocks;
> - the number of bytes to reserve in the storage memory pool (on- or
> off-heap, the caller's choice);
> - a release() callback that Spark invokes when the block is evicted or
> removed.
>
> If storage pressure forces eviction, the existing
> MemoryStore.evictBlocksToFreeSpace algorithm picks across *all* entries —
> JVM-resident and native — using the same LRU policy. No preference flag, no
> parallel store. When a native entry is chosen, release() is invoked exactly
> once and the reserved bytes return to the pool.
>
> Read paths (getLocalValues, getLocalBytes, getLocalBlockData, get,
> replicateBlock) refuse native blocks with a clear error: the payload is
> opaque to Spark and must be read through the owning engine's own API. The
> master is never notified, because native payloads are inherently
> executor-local and cannot be fetched over the network.
>
> So BlockManager becomes a unified registry for "this many bytes of storage
> are committed; call me back when you need them released," regardless of who
> owns the actual bytes.
> Layer 2 — a native variant of the columnar cache SPI
>
> On top of layer 1, CachedBatchSerializer grows a small opt-in:
>
> - a flag indicating that the serializer produces batches whose payload
> is owned by an external engine;
> - a NativeCachedBatch trait carrying release() plus a callback that
> receives the NativeBlockId Spark assigns;
> - a lightweight handle type (block id + row count + size + optional
> column stats) that flows through the cached RDD in place of the original
> payload.
>
> When the flag is set, the cache builder takes a different fork. For each
> batch produced by the serializer:
>
> 1. Spark reserves storage memory via the layer-1 entry point under a
> deterministic id.
> 2. The serializer is informed of the id, so it can populate its own
> executor-local lookup table mapping id back to the engine payload.
> 3. The Spark-visible CachedBatch is just the handle; the handle RDD is
> persisted at MEMORY_ONLY (handles are tiny).
> 4. On scan, the serializer's convertCachedBatchTo* methods resolve the
> handle's block id via the lookup table and hand back the native payload
> directly.
>
> Because the handle still carries column statistics in the existing
> SimpleMetricsCachedBatch schema, partition pruning continues to work
> without changes to the cache scan operator.
>
> The cache-eviction story is now closed-loop: Spark LRU picks the oldest
> entry (JVM or native), MemoryStore calls the registered release(), the
> native engine frees its bytes, and the next scan recomputes the affected
> partitions.
> 4. Why this is more general than the previous attempts
>
> - *Unified LRU, not a preference flag.* Native and JVM blocks live in
> the same entries map; the existing eviction algorithm handles them
> uniformly. No new tunable for users to misconfigure.
> - *No new configs.* It reuses
> spark.memory.{offHeap.enabled,offHeap.size,fraction,storageFraction} exactly
> as today.
> - *No MemoryStore clone.* Only a new entry variant alongside the
> existing two — small surface, same lifecycle hooks.
> - *Not bound to a single use case.* "Native file cache" and "native
> columnar cache" are both just consumers of the same core primitive.
> Anything with an opaque, off-heap payload and a release() semantics fits.
> - *Solves the serdes problem.* Native engines can opt into the SPI and
> have df.persist(MEMORY_ONLY) stay zero-copy end to end. The columnar
> cache SPI stops being a forced serialization boundary.
>
> 5. Trade-offs to discuss
>
> A few things worth surfacing:
>
> - *No replication, no spill, no remote fetch.* Native blocks are
> executor-local and opaque to Spark; the SPI explicitly rejects storage
> levels that request disk or replication > 1. Engines that want spill or
> replication have to implement it themselves on top.
> - *Dataset.cache() default is MEMORY_AND_DISK*, which the native path
> would reject. Users must call persist(MEMORY_ONLY) explicitly. We could
> relax this with a silent downgrade, or expose a new canonical storage-level
> constant. I lean toward "be explicit" for the initial cut but happy to hear
> opinions.
> - *Cluster-mode unpersist cleanup is best-effort.* Today the cache
> builder only touches the driver's local BlockManager; remote executors
> will release their native blocks via LRU or shutdown. A driver -> executor
> cleanup RPC is a natural follow-up.
>
> 6. What I have ready
>
> I have a 3-commit stack on top of master that I'd like to put up for review
> once a JIRA umbrella is filed:
>
> 1. [CORE] Add native block storage for opaque external-engine payloads —
> the new BlockId subtype, the MemoryStore variant, the BlockManager entry
> point, and tests covering reservation, eviction, release-on-evict, and
> release-throws.
> 2. [CORE] Harden native block storage — type enforcement, duplicate-put
> detection, replicate/master guards.
> 3. [SQL] Wire the SPI into the columnar cache builder, with end-to-end
> tests using a fake native serializer that exercises the round trip through
> df.persist(MEMORY_ONLY) and eviction.
>
> 7. What I'm asking
>
> 1. Is the community OK with BlockManager growing a "register external
> payload" surface? I think this is the right home for it — the alternative
> is yet another sidecar registry — but I want to confirm before pushing PRs.
> 2. Reviewers familiar with MemoryStore / UnifiedMemoryManager /
> CachedBatchSerializer who can weigh in on the API shape: memory-mode
> choice, lock semantics inside release(), the registration callback, and
> how strict the disk / replication rejection should be.
> 3. Anything I'm missing from a prior discussion that already converged
> on a different direction.
>
> Thanks for reading. I'd love to hear what people think before I take this
> further.
>
> Best, Chang Chen
>
---------------------------------------------------------------------
To unsubscribe e-mail: dev-unsubscribe@spark.apache.org