viirya commented on code in PR #7066:
URL: https://github.com/apache/datafusion-comet/pull/7066#discussion_r4300374973
##########
native/core/src/execution/jni_api.rs:
##########
@@ -990,6 +990,44 @@ fn configure_skip_partial_aggregation(config: &mut
SessionConfig, plan: &Operato
}
}
+/// The key of a shared [`DiskManager`]: the local directories and the temp
directory size limit.
+type DiskManagerKey = (Vec<String>, u64);
+
+static DISK_MANAGERS: OnceLock<Mutex<HashMap<DiskManagerKey,
Arc<DiskManager>>>> = OnceLock::new();
+
+/// Returns the [`DiskManager`] shared by every native plan in this process
that uses the same
+/// local directories and size limit.
+///
+/// A `DiskManager` in `Directories` mode creates a `datafusion-*` directory
in every local
+/// directory when it is built, and removes it recursively when it is dropped.
Building one per
+/// plan costs a `stat` and a `mkdir` per local disk when the plan is created,
and a recursive
+/// delete when it is released, whether or not the plan ever spills (#7060).
Sharing one makes
+/// that a once-per-executor cost. Spill files are still deleted when their
last reference is
+/// dropped, so a long-lived `DiskManager` does not accumulate them.
+///
+/// The size limit then covers all plans sharing the manager, not each plan
separately.
+fn shared_disk_manager(
Review Comment:
A shared, never-dropped manager also makes DataFusion's accounting leaks
permanent. In 55.2, `FileSpillWriter::write` adds `len` to `used_disk_space`
before `write_all`, and it does not roll that back when the write fails. The
per-file usage is not updated either, so `Drop` subtracts too little. This is
apache/datafusion#24230, which is still open. With a per-plan manager, the
leaked bytes went away with the plan. Here, every failed spill write
permanently lowers the executor's spill budget until restart. ENOSPC on a
shared local disk is exactly the case the docs say this limit protects against.
Should this wait for the upstream fix, or at least link it in this comment so
the next reader knows the limit can drift?
##########
docs/source/user-guide/latest/tuning/memory.md:
##########
@@ -276,13 +276,10 @@ flushes sorted spill files. A value larger than
`spark.comet.batchSize` is cappe
## Limiting Spill Disk Usage
Native operators that spill to disk (aggregate, sort, shuffle) are bounded by
-`spark.comet.maxTempDirectorySize` (default 100 GB). The operators of one
Comet native plan share
-the limit. A Spark task can run more than one native plan at a time, for
example the native
-operators on either side of a union or a coalesce, so an executor running `N`
concurrent tasks may
-use more than `N` times this value on shared local disks. If the limit is
reached, further spills
-fail and the query errors out. Raise this on workloads with large
sort/aggregate/shuffle spills, or
-lower it to protect executors on shared disks, remembering that the total
across an executor is a
-multiple of this value.
+`spark.comet.maxTempDirectorySize` (default 100 GB). All Comet native plans in
an executor share
Review Comment:
1.1 documents this limit as per plan, and the default stays at 100 GB. So an
executor whose concurrent tasks together spill more than 100 GB will fail a
query that succeeded on 1.1. When that happens, the error comes from DataFusion
and tells the user to raise `datafusion.runtime.max_temp_directory_size`, which
is not a Comet config. Would it make sense to call this out as an upgrade note,
and to point users at `spark.comet.maxTempDirectorySize` when they hit that
error? Was a larger default considered, now that the value covers the whole
executor?
##########
native/core/src/execution/jni_api.rs:
##########
@@ -990,6 +990,44 @@ fn configure_skip_partial_aggregation(config: &mut
SessionConfig, plan: &Operato
}
}
+/// The key of a shared [`DiskManager`]: the local directories and the temp
directory size limit.
+type DiskManagerKey = (Vec<String>, u64);
+
+static DISK_MANAGERS: OnceLock<Mutex<HashMap<DiskManagerKey,
Arc<DiskManager>>>> = OnceLock::new();
+
+/// Returns the [`DiskManager`] shared by every native plan in this process
that uses the same
+/// local directories and size limit.
+///
+/// A `DiskManager` in `Directories` mode creates a `datafusion-*` directory
in every local
+/// directory when it is built, and removes it recursively when it is dropped.
Building one per
+/// plan costs a `stat` and a `mkdir` per local disk when the plan is created,
and a recursive
+/// delete when it is released, whether or not the plan ever spills (#7060).
Sharing one makes
+/// that a once-per-executor cost. Spill files are still deleted when their
last reference is
+/// dropped, so a long-lived `DiskManager` does not accumulate them.
+///
+/// The size limit then covers all plans sharing the manager, not each plan
separately.
+fn shared_disk_manager(
+ local_dirs: Vec<String>,
+ max_temp_directory_size: u64,
+) -> CometResult<Arc<DiskManager>> {
+ let mut managers = DISK_MANAGERS
+ .get_or_init(|| Mutex::new(HashMap::new()))
+ .lock();
+ let key = (local_dirs, max_temp_directory_size);
Review Comment:
`spark.comet.maxTempDirectorySize` reaches here from `SQLConf.get`, so it is
a session-level value. Two sessions that set different values, for example on a
Thrift server, get separate managers, and each has its own full budget. Every
distinct value also leaves a manager, plus one `datafusion-*` directory per
local disk, in this map for the life of the executor. Keying on the limit is
the right call given the new semantics. But it means the executor-wide bound
only holds when every session uses the same value. Could the docs say that? See
the comment on `CometConf.scala`.
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -1363,11 +1363,9 @@ object CometConf extends ShimCometConf {
.category(CATEGORY_TUNING)
.doc(
"The maximum amount of data (in bytes) stored inside the temporary
directories " +
- "used by native operators when spilling. Applied to each Comet
native plan " +
- "separately, and a Spark task can run more than one native plan at a
time, so an " +
- "executor running N concurrent tasks may use more than N times this
value on shared " +
- "local disks. Once the limit is reached, further spills will fail
and the query will " +
- "error out.")
+ "used by native operators when spilling. Shared by all Comet native
plans running " +
Review Comment:
Because of the per-session keying in `shared_disk_manager`, "bounds the
spill files of the whole executor" is only true when every session uses the
same value. Maybe something like "Shared by all Comet native plans in an
executor that use the same value of this config"?
--
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]
