perf: create a native plan's spill directories on its first spill by andygrove · Pull Request #7066 · apache/datafusion-comet

andygrove · 2026-10-16T20:01:42Z

Which issue does this PR close?

Closes #7060.

Rationale for this change

Every native plan builds its own RuntimeEnv, and with it a DiskManager in DiskManagerMode::Directories over the executor's local disks. DataFusion 55.2 creates that mode's datafusion-* directory in every configured directory when the DiskManager is built and removes it recursively when it is dropped, so each plan pays for it whether or not it ever spills. #7060 measured these syscalls at 4.7% of task and Tokio thread samples for a small-task workload on macOS.

On Linux (ext4 on NVMe) each local dir costs 11 syscalls and about 35 µs per plan, so 8 local dirs cost about 340 µs. The work is kernel CPU rather than device IO, but it serializes on the filesystem: with one local dir, 8 or 16 threads creating plans at once top out near 30k plans per second. It also queues behind other writers' journal commits: while another writer flushed about 0.7 GB/s to the same filesystem, 8 threads with 4 local dirs saw p99 per plan go from 1.1 ms to 22 ms. End to end, on Spark 4.1.1 local[8] with 500 small files (501 tasks), 8 local dirs made the query 12% slower in wall time and 9% in CPU. With 1 local dir, or at the default split size, the difference is in the noise.

What changes are included in this PR?

spill_dirs::plan_disk_manager gives each native plan a DiskManager with a TempFileFactory (DiskManagerMode::Custom). On the plan's first spill file, the factory builds the plan's Directories mode DiskManager, and it creates every spill file through it. A plan that never spills creates no directory. A plan that spills gets the same directories, the same spark.comet.maxTempDirectorySize limit and the same cleanup as before, so the limit still applies to each plan separately.
An unusable local dir now fails a plan at its first spill rather than when the plan is created, so plans that do not spill no longer fail on it.

The first version of this PR shared one DiskManager per executor instead, which made the limit executor-wide; review moved it to this approach. apache/datafusion#26446 proposes the same lazy creation in DataFusion, after which the factory can go.

How are these changes tested?

Unit tests in spill_dirs.rs: the directories are created on the first spill file, reused by later ones and removed once the plan and its files are dropped; each plan gets its own directories and size limit.
CometNativeSuite "Comet native should use spark local dir as temp dir" now checks that a plan that does not spill creates no temp dir, that a sort that spills creates one in a Spark local dir, and that none remain once the plans are released. It fails against the eager behavior.
The existing native sort, aggregate and shuffle writer spill tests pass through the new path.
Benchmarks on Linux, described above.

Closes apache#7060.

kazuyukitanimura · 2026-10-16T23:14:10Z

+    let key = (local_dirs, max_temp_directory_size);
+    if let Some(disk_manager) = managers.get(&key) {
+        return Ok(Arc::clone(disk_manager));
+    }


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

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.

I think keying on the limit is the safer of the two here. set_max_temp_directory_size stores into a shared AtomicU64, and FileSpillWriter::write reads max_temp_directory_size() again on every write. If the cache were keyed on the dirs only and each plan set its own limit on the cached manager, the last plan created would set the limit for every plan already spilling in that executor. A session running with a 500 GB limit could hit a 10 GB cap in the middle of a spill because another session just started a plan.

I'm also not sure this is limited to local-mode tests. The value comes from SQLConf.get, which on an executor reads the session confs that travel with the task. A Thrift server or Spark Connect app runs many sessions in one application, so different sessions can set different values and send tasks to the same executor. That is why I suggested in the comment on line 1016 that the docs say the limit is shared by plans that use the same value, rather than changing the keying.

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.

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.

kazuyukitanimura · 2026-10-16T23:14:51Z

        })
    }

+    #[test]


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.

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.

viirya

Thanks for tracking this down, and for documenting the change to spark.comet.maxTempDirectorySize instead of letting it slip in silently. I checked the DataFusion 55.2 side. RefCountedTempFile::drop subtracts a file's size on its last reference, so in the normal path a long-lived DiskManager does not accumulate usage. The native shuffle writer's spill also goes through runtime.disk_manager, so the docs are right that shuffle shares the limit.

A few things need to be addressed before this can go in.

The Spark 4.1 [exec] job fails on CometNativeSuite "Comet native should use spark local dir as temp dir". That test asserts the datafusion-* directory is gone after the job, and this PR intentionally keeps it. The map is a static, so Comet never removes the directory. Spark removes it along with the blockmgr dir when the executor stops. With the external shuffle service enabled, it stays until the application ends. Could the test be updated to the new contract, for example that the directory is created once per local dir and holds no spill files after the job? Its first assertion, that no datafusion-* directory exists yet, also becomes order-dependent once any earlier test in the same JVM has run a native plan.

The benefit and the alternative. The 4.7% in #7060 was measured on macOS with 1 MB splits, and the issue notes these syscalls are cheaper on Linux. Do you have numbers on Linux with a typical split size? This PR trades a user-visible semantic change and process-wide state for that gain. The issue's own preferred fix is to create the directories lazily in DataFusion, the way OsTmpDirectory already does. That keeps the per-plan limit and needs no global map. Has an upstream issue been filed? If that fix lands, would Comet switch back to a per-plan limit? That would change the semantics a second time for users.

viirya · 2026-10-17T00:00:44Z

+/// 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(


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?

The managers are per plan again, so drift from apache/datafusion#24230 goes away with the plan, as on main.

viirya · 2026-10-17T00:00:44Z

+    let mut managers = DISK_MANAGERS
+        .get_or_init(|| Mutex::new(HashMap::new()))
+        .lock();
+    let key = (local_dirs, max_temp_directory_size);


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.

No map now, so nothing outlives its plan, and each plan uses the value its own session sent.

viirya · 2026-10-17T00:00:44Z

-          "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 " +


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"?

Reverted to main's text, since the limit applies to each native plan separately again.

viirya · 2026-10-17T00:00:44Z

-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


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?

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.

Replace the executor-wide shared DiskManager with a per-plan one whose Directories mode DiskManager is built through DataFusion's TempFileFactory hook on the plan's first spill. A plan that never spills no longer creates or removes a datafusion-* directory on every local disk, and spark.comet.maxTempDirectorySize stays a per-plan limit, so the config and tuning docs go back to their previous text. Update the CometNativeSuite temp dir test: a plan that does not spill creates no temp dir, a sort that spills creates one in a Spark local dir, and both are gone once the plans are released.

andygrove · 2026-10-17T14:05:01Z

I switched to the alternative you pointed to, done in Comet rather than waiting for DataFusion. DataFusion 55.2 already lets an engine supply its own spill backend (DiskManagerMode::Custom with a TempFileFactory), so each plan's DiskManager now builds its Directories mode DiskManager on the plan's first spill file and creates every spill file through it (5514cce).

A plan that never spills creates no directory. A plan that spills gets the same datafusion-* directories, the same per-plan spark.comet.maxTempDirectorySize limit and the same cleanup as on main.
The global map, the shared limit and the doc changes are gone, so there is no semantic change for users, now or later: if DataFusion makes Directories mode lazy (DiskManager in Directories mode creates its temp directories when it is built, even if nothing spills datafusion#26446), Comet can drop the factory with nothing visible changing.
The CometNativeSuite test checks the new contract: a plan that does not spill creates no temp dir, a sort that spills creates one in a Spark local dir, and none remain once the plans are released. It fails against the eager behavior.

Linux numbers, from a Ryzen 7950X3D on Ubuntu 22.04 (kernel 6.8, ext4 on NVMe):

Per plan, the eager directories cost about 35 µs per local dir (11 syscalls each): 147 µs with 4 dirs and 339 µs with 8. With the factory, building and dropping a plan's disk manager takes 0.1 µs. The work is kernel CPU, and it serializes on the filesystem: with one local dir, 8 or 16 threads building plans at once top out near 30k plans/s, and the median per plan rises from 35 µs to about 175 µs. While another writer flushed about 0.7 GB/s to the same filesystem, 8 threads with 4 local dirs saw p99 per plan go from 1.1 ms to 22 ms.
End to end, Spark 4.1.1 local[8], SELECT count(*) FROM lineitem WHERE l_quantity > 49, 3 interleaved runs per arm:

Input	Tasks	Local dirs	Wall, main → PR	CPU, main → PR
SF10, default 128 MB splits	20	1 or 8	no change (±1%)	no change
SF1 as 500 files of 0.5 MB	501	1	177 → 176 ms	1710 → 1675 ms
SF1 as 500 files of 0.5 MB	501	8	197 → 173 ms (−12%)	1850 → 1680 ms (−9%)

With one local disk at a typical split size the gain is in the noise, as you expected. It shows up for many short tasks on executors with several local disks, which YARN and Kubernetes nodes often configure. On EBS-backed local dirs I'd expect the same CPU cost, since the loop above runs at 95% CPU in the kernel rather than waiting on the device, plus two costs that grow there: the waits behind other writers' journal commits get longer on a volume that flushes at gp3's 125 MB/s baseline, and the churn's metadata writes (about 150 small writes/s at 3k plans/s with 8 dirs here) come out of the volume's IOPS budget. I haven't measured on EBS. With no semantic change or process-wide state left, I think the factory is worth keeping for that case.

viirya

Thanks for reworking this. Going through the TempFileFactory keeps every behavior from main that I was worried about. I read the 55.2 path: the factory hands back the inner Directories manager's RefCountedTempFile, so the per-plan limit, the usage accounting on drop, and the directory cleanup are all the same as before. Nothing outlives the plan anymore. The Linux numbers are useful too. They make clear where the gain shows up, which is many short tasks on executors with several local disks.

The failed Spark SQL Tests (Spark 4.1) sql_core-2 job never ran a test. sbt could not download org.scala-sbt:sbt:1.11.7 after three attempts. A rerun should settle it, and that shard is the one that covers this path.

One small comment inline about the outer manager.

viirya · 2026-10-17T18:52:00Z

+            builder,
+            disk_manager: Mutex::new(None),
+        }))
+        .with_max_temp_directory_size(max_temp_directory_size)


The outer DiskManager only carries the factory now. Its own used_disk_space and spilling_progress() stay at 0, and its max_temp_directory_size is never checked, because the inner manager enforces the limit. Nothing reads those in 55.2 today, so this is correct. But a later change that calls RuntimeEnvBuilder::with_max_temp_directory_size, or reads spill usage from RuntimeEnv, would silently get the wrong answer. Could the doc comment on plan_disk_manager say that the inner manager holds the limit and the accounting? A link to apache/datafusion#26446 there would also tell the next reader when the factory can go.

kazuyukitanimura

The description says the existing shuffle writer spill tests pass through the new path. The Rust ones I found (spilling_every_partition_creates_one_file and the tests in spill.rs) build their own DiskManager, so I don't think they reach plan_disk_manager. Is there a Scala test where the native shuffle writer spills? If so, could you name it? If
not, could you add a small test in spill_dirs.rs or the shuffle crate that gets a spill file through plan_disk_manager the way ensure_spill_file_created does? It should check that path() is Some, that bytes written count against the plan's limit, and that the file is removed on drop. Shuffle spill is the one consumer here that keeps the file
open across batches, so I'd like it covered before merge.

kazuyukitanimura · 2026-10-19T17:59:17Z

+            builder,
+            disk_manager: Mutex::new(None),
+        }))
+        .with_max_temp_directory_size(max_temp_directory_size)


Thanks for reworking this. Going through TempFileFactory keeps the limit, the accounting and the cleanup exactly as on main. I agree with viirya's comment here. I'd add two sentences to this doc comment. One should say that the outer manager only carries the factory, so its used_disk_space and spilling_progress() stay at 0 and its
max_temp_directory_size is never checked. The second should say that the inner manager holds the limit and the accounting. A link to apache/datafusion#26446 would also tell the next reader when the factory can be removed. I checked 55.2 and nothing enforces on the outer manager today. The only readers I found are RuntimeEnv's reporting code at
runtime_env.rs:229 and :289. A later RuntimeEnvBuilder::with_max_temp_directory_size call would silently do nothing, though.

perf: share one DiskManager per executor across native plans

0843ac8

Closes apache#7060.

andygrove requested review from kazuyukitanimura and viirya October 16, 2026 20:03

github-actions Bot added enhancement New feature or request performance labels Oct 16, 2026

kazuyukitanimura reviewed Oct 16, 2026

View reviewed changes

viirya requested changes Oct 17, 2026

View reviewed changes

This was referenced Oct 17, 2026

DiskManager in Directories mode creates its temp directories when it is built, even if nothing spills apache/datafusion#26446

Closed

Spill limit error tells Comet users to raise datafusion.runtime.max_temp_directory_size #7091

Closed

andygrove changed the title ~~perf: share one DiskManager per executor across native plans~~ perf: create a native plan's spill directories on its first spill Oct 17, 2026

andygrove added the run-spark-4.1-tests Run the Spark 4.1 SQL tests on this pull request instead of waiting for the merge queue label Oct 17, 2026

viirya approved these changes Oct 17, 2026

View reviewed changes

viirya mentioned this pull request Oct 18, 2026

fix: point the spill limit error at spark.comet.maxTempDirectorySize #7098

Merged

Merge remote-tracking branch 'apache/main' into perf-shared-disk-manager

edfa673

kazuyukitanimura reviewed Oct 19, 2026

View reviewed changes

andygrove added this pull request to the merge queue Oct 20, 2026

Merged via the queue into apache:main with commit 53d8867 Oct 20, 2026
53 checks passed

andygrove mentioned this pull request Oct 23, 2026

Write blog post for 1.2.0 release #6889

Closed

                       })
                   }
+                  #[test]

Conversation

andygrove commented Oct 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

How are these changes tested?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

andygrove commented Oct 17, 2026

Uh oh!

viirya left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kazuyukitanimura left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

andygrove commented Oct 16, 2026 •

edited

Loading