kazuyukitanimura commented on code in PR #7066:
URL: https://github.com/apache/datafusion-comet/pull/7066#discussion_r4300153227
##########
native/core/src/execution/jni_api.rs:
##########
@@ -3661,6 +3695,20 @@ mod tests {
})
}
+ #[test]
Review Comment:
The test confirms the manager and its directory are shared. Could we also
add one that shows the limit is shared? For example, create spill files through
two create_tmp_file calls on the same manager, write past a small limit across
both, and check the second write fails and used_disk_space() drops back after
the files are dropped. That is the behavior change users will see, so it
deserves a test of its own.
##########
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:
Thanks for tracking this down, the profile in #7060 is convincing. One thing
I'd like to understand is what happens if the datafusion-* directory disappears
while the executor is running, for example because a tmp cleaner removes it. On
1.1 every plan recreated its own directory, so the next plan healed. Here the
TempDir is created once, so I think every later create_tmp_file fails with
ENOENT until the executor restarts. Could we check that the directory still
exists when handing out the cached manager and rebuild it if not? Or is there a
reason that can't happen? Even a small test that removes the directory and then
requests a manager would settle it
##########
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:
DiskManager has set_max_temp_directory_size(&self, ..). Is there a reason to
key the cache on the limit instead of keying on the dirs and setting the limit
on the cached manager? As written, two sessions with different limits in one
JVM get separate counters and separate directories. In production that can't
happen, but it does in local-mode tests.
--
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]
