chore: instrument JNI traffic in the unified memory pools - #5384
Draft
andygrove wants to merge 1 commit into
Draft
chore: instrument JNI traffic in the unified memory pools#5384andygrove wants to merge 1 commit into
andygrove wants to merge 1 commit into
Conversation
Both `CometUnifiedMemoryPool` and `CometFairMemoryPool` carried their own copy of the JNI acquire/release helpers. Extract them into a single `SparkMemoryClient` that records how many calls each pool makes, how many bytes were requested and granted, how many grants came back short, and how long was spent inside the JNI round-trip. The summary is logged per task attempt when the pool is dropped. This is the measurement step for apache#5383: batching acquisitions and releasing with hysteresis both need a before-and-after call count to justify them. The JNI call is reached through a `SparkMemoryBackend` trait, so the pools can now be exercised without a JVM. That gives the unified pool its first unit tests, covering grow/shrink accounting and the short-grant path, and confines the `unsafe impl Send`/`Sync` to the type that actually holds the JNI global reference. No behavioural change: the pools make exactly the same calls as before.
This was referenced Aug 17, 2026
Open
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Part of #5383 (which is in turn positions 17 and 18 of #5212). This is the measurement step; it does not change how much memory is acquired or when.
Rationale for this change
Every DataFusion
try_growandshrinkagainstCometUnifiedMemoryPoolorCometFairMemoryPoolis a JNI round-trip intoCometTaskMemoryManager, which then takes Spark's executor-widesynchronizedmemory manager lock. There is no batching and no hysteresis: memory is acquired in exactly the amount requested and released the moment a reservation shrinks, however small.Before adding chunked acquisition and slack retention I want a before-and-after number. There is currently no way to tell how many round-trips a query makes, so there is no way to show that the added complexity buys anything, or to spot a pool whose traffic is dominated by tiny grow/shrink churn.
What changes are included in this PR?
Both pools carried their own copy of the JNI acquire/release helpers. Those move into a single
SparkMemoryClient(native/core/src/execution/memory_pools/spark_client.rs) which records, per task attempt:The counters are relaxed atomics and are always collected. The summary is logged when the pool is dropped — for the task-shared unified pools, when the last native plan for the task is released:
That is one line per task attempt, so it logs at debug level. To collect it without turning on debug logging for every module, a
log4rs.yaml(viaCOMET_CONF_DIRor thecomet.log.file.pathsystem property) can raise the level forcomet::execution::memory_poolsalone; the doc comment onlog_statsspells this out.The JNI call itself is reached through a small
SparkMemoryBackendtrait. Two things follow from that:CometUnifiedMemoryPoolgets its first unit tests.unsafe impl Send/Syncmoves off both pools and ontoJniMemoryBackend, the only type that actually holds the JNI global reference. The pools now satisfy the auto traits on their own.No behavioural change — the same calls are made in the same order with the same arguments.
How are these changes tested?
Three new unit tests for the unified pool, over an in-process backend that mimics Spark's "grant as much as is available" behaviour:
grow_and_shrink_track_spark_grants— grow/shrink accounting stays in step with what the backend has handed outshort_grant_is_returned_to_spark_and_reported_as_an_error— a partial grant is released rather than retained, and the error names the requested sizeevery_grow_and_shrink_reaches_spark— 10 grow/shrink cycles produce 10 acquires and 10 releases, pinning the current no-batching behaviour that Reduce JNI round-trips in the unified memory pools (batching, hysteresis, cheaper call path) #5383 aims to reducePlus three tests for the counters themselves in
spark_client.rs.cargo test -p datafusion-comet --libpasses (163 tests), andcargo clippy --all-targetsandcargo fmt --checkare clean.CometJoinSuitepasses (29 tests) as an end-to-end check of the JNI path, which every Comet suite exercises sinceCometTestBaseenables off-heap mode.I deliberately left
CometFairMemoryPoolwithout unit tests here: any assertion about its limit would encode thepool_size / num_consumersbug from position 1 of #5212. Those tests belong with that fix.