Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
357 changes: 205 additions & 152 deletions contrib/delta/native/Cargo.lock

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion contrib/delta/native/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -47,4 +47,4 @@ crate-type = ["rlib"]
datafusion-comet-proto = { path = "../../../native/proto" }
# For `ExecutionPlan` (return type), `SchemaRef` (the shim hands these in), and `DataFusionError`
# (the not-implemented error). Versions/features mirror core so the shared lockfile resolves cleanly.
datafusion = { version = "55.2.0", default-features = false, features = ["parquet"] }
datafusion = { git = "https://github.com/apache/datafusion", rev = "1635ed94742c1c060852bc34dc6ebb2302241c7b", default-features = false, features = ["parquet"] }
2 changes: 1 addition & 1 deletion contrib/lance/native/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -28,5 +28,5 @@ license = "Apache-2.0"
crate-type = ["rlib"]

[dependencies]
datafusion = { version = "55.0.0", default-features = false }
datafusion = { git = "https://github.com/apache/datafusion", rev = "1635ed94742c1c060852bc34dc6ebb2302241c7b", default-features = false }
datafusion-comet-proto = { path = "../../../native/proto" }
25 changes: 5 additions & 20 deletions docs/source/contributor-guide/memory_management.md
Original file line number Diff line number Diff line change
Expand Up @@ -300,15 +300,12 @@ the inside out, a Comet plan in the default configuration sees:
```text
[LoggingMemoryPool] <- only when spark.comet.debug.memory=true
[TaskSharedMemoryPool] <- RAII handle for the per-task registry
[SpillReplayPool] <- lets a spilled final aggregate read its spill files back
[TrackConsumersPool] <- DataFusion; names the top 10 consumers in error messages
[CometFairMemoryPool] <- delegates acquire/release to Spark over JNI
[TrackConsumersPool] <- DataFusion; names the top 10 consumers in error messages
[CometFairMemoryPool] <- delegates acquire/release to Spark over JNI
```

Each decorator forwards every `MemoryPool` method to its inner pool, so `reserved()` at any level
reports the base pool's number. `SpillReplayPool` also turns one kind of refused `try_grow` into a
`grow`; see
[Final aggregates reading their spill files back](#final-aggregates-reading-their-spill-files-back).
reports the base pool's number.

### The unified pools

Expand Down Expand Up @@ -394,8 +391,8 @@ A consumer that cannot spill takes what Spark grants. The pool refuses it only w
total plus the request would exceed `pool_size`, which Spark's own limit on the task is at least as
tight as unless the deprecated fraction is below `1 / N`.

The pool goes by what a consumer declares, with one exception. DataFusion 55.2 registers
`OrderedFinalAggregateStream` as able to spill even when its input is sorted on every grouping key,
The pool goes by what a consumer declares, with one exception. DataFusion registers
`ClusteredFinalAggregateStream` (`OrderedFinalAggregateStream` before DataFusion 56) as able to spill even when its input is sorted on every grouping key,
as a workaround for [apache/datafusion#17334](https://github.com/apache/datafusion/issues/17334), but
builds a spill path only for partially sorted input. Fully sorted, as in a `CometSortAggregateExec`
over a native sort, a refusal fails the query, so the pool checks its requests only against the
Expand Down Expand Up @@ -463,18 +460,6 @@ hands every native plan in a task the same one and drops it when the task comple
covers the whole task, so `CometExecIterator.close()` warns about memory still in use only when
the task's last open native plan closes.

### Final aggregates reading their spill files back

`SpillReplayPool` (`spill_replay.rs`) wraps both Comet pools to work around
[issue #6254](https://github.com/apache/datafusion-comet/issues/6254). Once one of DataFusion 55's
final aggregates has spilled, it reads its spill files back through an aggregate that cannot spill,
so a refused `try_grow` there fails the task. When the pool refuses such a request,
`SpillReplayPool` records it with the pool's `grow` instead, which skips `CometFairMemoryPool`'s
local checks and carries what Spark does not grant as overcommit. Every other refusal is passed on
unchanged. `spill_replay.rs` describes how the wrapper recognizes these requests and what that relies
on in DataFusion. [Issue #6583](https://github.com/apache/datafusion-comet/issues/6583) tracks
removing it.

## How DataFusion consumes the pool

Native operators reserve through DataFusion's `MemoryConsumer` / `MemoryReservation` API:
Expand Down
Loading
Loading