dwsmith1983 commented on PR #5613: URL: https://github.com/apache/datafusion-comet/pull/5613#issuecomment-5793246686
> **1. CI on the final design.** ... I'd also like one off-heap query-level run with `fair_unified` (TPC-H is fine) before merging. c2d5a2836 is the final shape; could you apply `run-spark-4.1-tests`, since I cannot set labels here. TPC-H results are below: one run on head with the stock settings to show the design end to end, and a pressure run with 2g off-heap and eight cores per executor, base and head alternated, comparing spill counts and wall time. Stock settings, head jar, one pass of the 22 queries on TPC-H SF10 against Spark 4.1.3 on one 8-core executor: all 22 completed, results hashed identically to every other arm below, 33 seconds for the 22 queries, no spills, no failures. The executor log confirms `fair_unified` in effect with a 16g off-heap pool and a 2g per-task limit. > **2. The registry and `CometResult` changes no longer seem needed.** Reverted in c2d5a2836. `task_shared.rs` and `jni_api.rs` are back to main byte for byte, `create_memory_pool` returns the pool directly, and the only line left in `mod.rs` beyond main is a doc sentence saying that creating a pool makes no JVM call and the anchor byte is taken on the first grow. The three registry tests that covered the removed path are gone, the registry composition test now checks that a pool acquired through the registry takes its anchor on the first grow, and the four comments you listed are gone or rewritten with the revert. > **3. The early `used` rollback can cause spills that main would not.** The two windows get different answers. The second one, a shrink handing bytes back to `used` before Spark has them, is reordered in c2d5a2836: `shrink` does its bounds check under the lock, releases to Spark with the lock dropped, then takes the bytes off `used` and the JVM-held count in `settle_release`. The short-grant rollback follows the same rule, so the granted bytes stay charged until Spark has them back. If that rollback release itself fails, the bytes stay charged for the pool's life, since Spark holds them until the task ends, and the caller still gets the short-grant error so it can spill; a test with an injected release failure pins that. Two tests drive the interleaving through the release gate and fail on the old ordering: a grow racing a shrink is refused at the fair limit without a JVM call instead of reaching Spark ahead of the release, and a short-grant rollback keeps its bytes charged until the release lands. To be precise about what this buys: `used` never runs ahead of what Spark holds any more, an d a grow that only fits with the shrunk bytes is refused locally rather than sent to Spark for a short grant. A grow that fits without those bytes can still reach Spark ahead of the release and come back short at the task's share, so that window is narrowed, not closed; a test pins that shape too so it stays visible. Two smaller things came out of the same pass: a request whose total would overflow is refused at the fair limit instead of panicking, and the pool warns when it drops with bytes still charged. The first one, an in-flight grow counted in `used`, stays. Charging before the JVM call is what keeps two concurrent grows from jointly exceeding `pool_size / num`, and it can only refuse, never over-admit. A third test pins that direction. Its cost is the spill comparison from the pressure run: Pressure run: the same SF10 data, one 8-core executor, 2g off-heap so the per-task limit is 256 MB, `fair_unified`, three iterations of the 22 queries per arm, eight arms in two mirrored passes (base, head, head, base, then head, base, base, head). Base is the merge base with main, head is c2d5a2836, both built the same way. Spills first, since that is the question. Every full-suite arm spilled 360 times and 22.36 GB, all from the native sort, identical for base and head to within 16 bytes across all eight arms; no task, stage or job failed, no executor was lost, no out-of-memory. The two single-query arms behave the same way: q18 spills 80 times and 7.17 GB on both jars, q21 150 times and 4.76 GB on both. One caveat on what can be counted: the sort catches the pool's refusal and spills without logging it, so refusals and short grants are not visible in the logs; the spill counts are the observable, and they do not move. Wall time, sum of per-query medians per arm in run order: | pass | arm | seconds | |---|---|---| | 1 | base | 31.6 | | 1 | head | 32.6 | | 1 | head | 33.8 | | 1 | base | 30.5 | | 2 | head | 25.0 | | 2 | base | 24.8 | | 2 | base | 25.0 | | 2 | head | 24.0 | The first pass ran minutes after two release builds and data generation on a laptop and is 20 percent slower on both jars than the second, which ran after the machine had idled; within the settled second pass the head to base ratio on the sum of medians is 0.99, and no single query moves more than a few percent either way. Alone, q18 ran 2.94 s on head against 3.56 s on base pooled over ten samples per side, faster on head in both interleaved positions, and q21 5.07 s against 4.83 s, slower on head in both. Results hashed identically across every arm and iteration. > **4. Interaction with #5961.** The two comments now say only what the code does: the pool tracks one total across every consumer and checks the fair limit against that total rather than this reservation's own size. Nothing in them argues for the pool-wide cap or names a DataFusion version. > **5. Docs.** `memory_management.md` now has three paragraphs in the unified pools section: the anchor byte and why `NativeMemoryConsumer.getUsed` stays at least 1 from the first native grow until the pool drops, the rule that the pool mutex is never held across a JNI call and the parked-acquire reason behind it, and the two windows above with their conservative direction. Nothing else on the page changed meaning. > **6. Perf evidence (non-blocking).** Base and head ran in one binary in the earlier table, but that harness is no longer on disk, so I cannot say from it whether the rounds alternated; treat the single-thread row as unconfirmed and the contended rows as the claim. If the number matters for the decision, I will rebuild the harness and re-run with base and head alternated ten times, the first measurement of each dropped, medians and spread reported. On several threads growing one task pool at once: a task that shuffles runs its pre-shuffle operators and the shuffle writer as separate native blocks against one task-shared pool, and within a block DataFusion's operators reserve from tokio worker threads, so that is the normal shape rather than the exception. I did not profile it; no wall-clock profiler is set up on the machine the TPC-H comparison ran on. -- 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]
