andygrove commented on code in PR #7066:
URL: https://github.com/apache/datafusion-comet/pull/7066#discussion_r4302945878
##########
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);
+ if let Some(disk_manager) = managers.get(&key) {
+ return Ok(Arc::clone(disk_manager));
+ }
Review Comment:
There is no cached manager anymore. Each plan creates its own directory
again, now on its first spill rather than when it is built, so a directory that
a tmp cleaner removes only affects the plans that already spilled into it, as
on main.
##########
native/core/src/execution/jni_api.rs:
##########
@@ -3661,6 +3695,20 @@ mod tests {
})
}
+ #[test]
Review Comment:
The limit is per plan again, so the test is now
`each_plan_has_its_own_spill_directories_and_size_limit`: two plans with a
10-byte limit each write 8 bytes, a second 8-byte write in the first plan
fails, and the second plan's write still succeeds. It also checks that the two
plans get separate directories.
##########
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);
+ if let Some(disk_manager) = managers.get(&key) {
+ return Ok(Arc::clone(disk_manager));
+ }
Review Comment:
There is no cache to key now. Each plan builds its own `DiskManager` with
its session's limit on its first spill, so one session's limit can't reach
another session's plans.
##########
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:
The managers are per plan again, so drift from apache/datafusion#24230 goes
away with the plan, as on main.
##########
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:
No map now, so nothing outlives its plan, and each plan uses the value its
own session sent.
##########
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:
Reverted to main's text, since the limit applies to each native plan
separately again.
##########
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:
Reverted with the rest of the semantic change, so there is no upgrade note
to add and the default can stay. The DataFusion error that names
`datafusion.runtime.max_temp_directory_size` is misleading on main too; I
opened #7091 to point users at `spark.comet.maxTempDirectorySize`.
--
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]
