sunchao opened a new pull request, #58763: URL: https://github.com/apache/spark/pull/58763
### Why are the changes needed? JIRA: [SPARK-59461](https://issues.apache.org/jira/browse/SPARK-59461). Speculative work, such as downloading the next selected Parquet row group while decoding the current one, can hide I/O latency. Its unused buffers should not prevent a query from obtaining memory for work it needs to complete. For example, consider a task with 400 bytes of disposable read-ahead in a 1,000-byte execution pool. Its next ordinary allocation needs 700 bytes. Admitting the read-ahead only when memory was free is insufficient: unless those 400 bytes can be returned, the later allocation sees just 600 bytes. Other tasks' optional reservations can also reduce the requester's available capacity or fair share. `TaskMemoryManager` can ask consumers in its own task to spill, while `UnifiedMemoryManager` can evict borrowed storage memory. Neither provides an executor-wide release mechanism for disposable execution buffers owned by other tasks. Unmanaged-memory reporting reduces effective capacity, but does not provide an ownership or reclamation protocol. This is an internal memory-management enhancement, not a fix for an existing built-in reader's correctness bug. [SPARK-56918](https://issues.apache.org/jira/browse/SPARK-56918) is related work on shrinking executor-wide external storage caches; this change instead retains task attribution and execution-pool accounting. ### What changes were proposed in this pull request? Add an internal optional-memory policy to the existing execution pool. An optional request receives all its bytes from currently free execution memory or receives zero, without borrowing storage, evicting blocks, spilling, or waiting for capacity. Successful reservations use existing task accounting and release methods. Ordinary allocations retain optional buffers when the full request fits immediately under Spark's existing capacity and fair-share rules. Otherwise they first ask registered owners to release unused optional work, then retry the existing allocator. A shared admission gate prevents fresh optional reservations during reclamation. Callbacks run outside the memory-manager and registration monitors and must release only their own unused resources; they must not allocate memory, wait for I/O, or acquire a task-manager monitor. Some `MemoryStore` operations already hold the memory-manager monitor across nested allocation or eviction. Those operations establish the reclamation boundary before taking that monitor, so callbacks cannot create a lock-order inversion. Their outer boundaries conservatively drain optional owners. Focused cleanup changes preserve block contents/metadata consistency, unroll credits, and consumed-buffer ownership if a callback fails. The admission preflights respect upstream unmanaged-memory accounting. A successful preflight and immediate grant use the same unmanaged-memory sample; allocation paths that can wait retain normal fresh sampling. Public task/consumer APIs, task-completion integration, and reader-specific prefetch scheduling are intentionally separate follow-ups. This PR does not enable read-ahead or change file selection, page skipping, or the general spilling/fairness policy. ### Does this PR introduce _any_ user-facing change? No new public API, SQL behavior, configuration, or reader feature. There is no production optional-memory consumer in this change. Existing allocation and storage paths gain internal coordination for future consumers; that bookkeeping is not claimed to be free or to improve existing workloads by itself. ### How was this patch tested? Built Spark core and its test sources against upstream master `5d388249c3cf28f5ba11cdd5275d1fe5ae9cc1c1`, then ran the following on JDK 17 / Apple Silicon: ```sh SPARK_LOCAL_IP=127.0.0.1 SERIAL_SBT_TESTS=1 build/sbt -batch \ 'core/testOnly org.apache.spark.memory.UnifiedMemoryManagerSuite org.apache.spark.memory.TestMemoryManagerSuite org.apache.spark.storage.MemoryStoreSuite org.apache.spark.util.io.ChunkedByteBufferOutputStreamSuite' \ 'core/testOnly org.apache.spark.storage.BlockManagerSuite -- -z "optional reclaimer failure" -z "cache unroll preserves resource ownership"' \ 'core/Compile/scalastyle' 'core/Test/scalastyle' ``` At `9d965e4be4be060c7c99bd68569cc44d7616d7cd`: **73 tests passed, zero failures**. One existing unmanaged-memory test is canceled on Apple Silicon. Both production and test scalastyle checks passed with zero errors and warnings. Source hashes were unchanged after validation. Full upstream CI has not run. An isolated negative control removed the shared unmanaged-memory sample from a copy of the current implementation. It compiled, and the polling-race regression failed at the expected storage-admission assertion. The tested source checkout was not modified by this control. Coverage includes full/denied optional admission; no-owner and ample-capacity paths; same-task and cross-task pressure; on/off-heap isolation; callback lock ordering and failures; real `MemoryStore`/`BlockManager` cleanup; consumed-buffer disposal; and upstream unmanaged-memory limits and polling races. No end-to-end reader performance improvement or current no-owner allocator-overhead result is claimed by this PR. Those measurements are separate from correctness validation and require an adopting consumer where applicable. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Codex -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
