Apache Doris 4.1 Spill to Disk: Running Memory-Intensive Queries Without OOM

Picture an analytics team running its end-of-quarter close. The ETL job joins three years of transaction history against a customer dimension and a product hierarchy, then aggregates by region for the executive deck. Three hours in, the database server runs out of memory, and the query dies with an OOM error. The team retries with a smaller date range, splits the job in half, and ships the deck the next morning instead of that evening. If the spill-to-disk feature had been enabled, that job would have finished, and there would be no morning panic.

Apache Doris 4.1 release matured this feature for the Doris community.

TL'DR of what changed in 4.1:

• Full operator coverage: Hash Join, Aggregation, and Sort all spill. These three operators account for almost all memory pressure in analytical queries.
• Recursive repartitioning: When a recovered partition is still too large for memory, the engine repartitions it again. This handles heavy data skew, not just average-case workloads.
• Proactive triggering: Doris detects memory pressure early and starts spilling before hitting a hard limit, rather than waiting for a query to fail and then trying to recover.

The rest of this post covers what Spill to Disk is, which workloads benefit most from it, how Apache Doris implements it end-to-end, the configuration and observability surface, and the full TPC-DS benchmark.

1. Memory-Intensive Operators and Why They Need Spill to Disk

In Apache Doris, three operators sit at the top of the OOM-risk list. They share one trait: each must accumulate state over the entire input before producing results. Memory usage scales with input size, cardinality, or skew, so the worst case sits far above the average. Spill to Disk targets exactly these three. The following operators most likely to hit memory limits are those that need to accumulate state:

These three share two characteristics that make them candidates for spilling:

• Memory usage scales with data size: Input volume, cardinality, and data skew all affect how much memory they consume.

• State is serializable: The state these operators maintain doesn't need to stay in memory permanently. It can be written to disk and read back on demand.

The core concept behind spill to disk is to convert revocable memory into disk state, then read it back in smaller streaming batches during the recovery phase.

2. How Apache Doris Implemented Spill to Disk

Doris does not implement spill inside each operator in isolation. A control layer and an operator layer coordinate the work, and share a common infrastructure layer beneath them.

The four layers:

• Control layer: PipelineTask handles the reserve entry point and memory pressure detection. WorkloadGroupMgr manages paused queries and decides whether to spill or cancel. QueryTaskController selects specific spill targets and creates SpillContext.

• Operator layer: Three core operators: Partitioned HashJoin, Partitioned Aggregation, and Spill Sort, each implement revoke_memory() to handle their own data persistence and recovery.

• Infrastructure layer: SpillFileManager manages disk directories and garbage collection. SpillFile, SpillFileWriter, and SpillFileReader provide block-level read/write abstractions. SpillRepartitioner handles multi-level repartitioning.

• Memory management layer: Three-level memory checks (Query Limit → Workload Group Limit → Process Limit) form a layered memory defense.

3. The Trigger Flow: Reserve, Pause, Spill, Resume

The entire Spill to Disk process follows a pause → persist → resume cycle.

Overview

Overall trigger flow: reserve, pause, spill, resume

Reserve Memory: Three-Level Validation

Before each execution round, PipelineTask estimates how much memory the operator will need for the next step, then calls try_reserve() to reserve that amount. This isn't a real malloc, it's a pre-check to determine whether execution should continue before entering operator logic.

try_reserve() validates three constraints simultaneously:

Three-level reserve validation

Key design note: When enable_reserve_memory is on, QueryContext will disable the query tracker's immediate hard check. Apache Doris prefers that queries enter a "pausable and spillable" path rather than failing immediately when a limit is hit.

Proactive Spill Trigger

Beyond the passive trigger (reserve failure), PipelineTask also checks proactively whether spill should be initiated early. Spill triggers when all three conditions are met simultaneously:

The goal of the proactive spill trigger is to avoid waiting until memory allocation fails. It proactively converts revocable memory to disk state when it detects memory under high pressure.

Paused Query State

When a reserve fails or proactive revoking triggers, the query enters a Paused state:

• Records the pause reason (QUERY_MEMORY_EXCEEDED / WG_MEMORY_EXCEEDED / PROCESS_MEMORY_EXCEEDED)
• Enables low memory mode
• Blocks on memory_sufficient_dependency

The query stops executing and waits for WorkloadGroupMgr to handle it.

WorkloadGroupMgr Decision Logic

WorkloadGroupMgr follows different branches depending on the pause reason.

WorkloadGroupMgr decision flow

Two key constraints to keep in mind:

• Only Paused Queries are eligible for spill triggering by WorkloadGroupMgr.
• QueryTaskController requires that there are no running tasks in a fragment before starting revocation. Apache Doris doesn't allow uncoordinated memory reclamation of a query's state while it's actively executing.

QueryTaskController: Selecting Spill Targets

QueryTaskController::revoke_memory() executes the following logic:

1. Collect all revocable tasks across all fragments
2. Sort by revocable size in descending order
3. Select a batch of tasks: target recovery amount = current actual memory consumption × 20% (uses actual consumption, not the limit. For example, if the limit is 1.6 GB but only 1 GB is in use, the target recovery amound is 200 MB, not 300 MB)
4. Create a SpillContext for the selected tasks (with a completion callback)
5. Call each task's revoke_memory(), conditional on revocable_mem_size >= MIN_SPILL_WRITE_BATCH_MEM (512 KB)
6. When all tasks complete, the SpillContext callback fires → memory_sufficient = true → query resumes scheduling

4. Shared Infrastructure

Although each operator implements spill differently, they share a common infrastructure layer.

SpillFileManager

Role: Global management of spill storage directories and disk resources.

                     SpillFileManager
                          │
          ┌───────────────┼───────────────┐
          │               │               │
    ┌─────▼─────┐  ┌─────▼─────┐  ┌─────▼─────┐
    │ SpillData │  │ SpillData │  │ SpillData │
    │ Dir (SSD) │  │ Dir (SSD) │  │ Dir (HDD) │
    └───────────┘  └───────────┘  └───────────┘

Key behaviors:

• Disk selection: Prefers SSD. Selects among available disks in ascending order of usage rate.
• Background GC: Periodically scans and cleans expired spill directories.
• Capacity management: Checks remaining disk space via reach_capacity_limit().
• Read/write metrics: Maintains global counters for spill read and write byte totals.

SpillFile, SpillFileWriter, and SpillFileReader

SpillFile is a logical file abstraction that can automatically split into multiple parts under the hood:

SpillFile internals

This design lets upper-level operators focus on "write block / read block" without managing file splitting or metadata.

Fault tolerance: SpillFile's destructor automatically calls gc() to delete the spill directory and release the disk quota. Even if the writer doesn't close normally, byte counts continue to be tracked, ensuring GC has accurate disk usage statistics.

SpillRepartitioner

SpillRepartitioner is the core building block for multi-level spill in Apache Doris, primarily used by Hash Join and Aggregation:

SpillRepartitioner

Key characteristics:

• Two modes: Join mode (uses PartitionerBase for hashing) and Aggregation mode (uses key column index for CRC32 hashing).
• Incremental processing: Processes approximately 32 MB at a time, then yields CPU back to the scheduler to avoid long blocking.
• Persistent Reader: Maintains read position across yields, avoiding redundant re-reads.

5. How Spill to Disk Works with Each Operator

Hash Join:

Hash Join is the most complex part of Apache Doris's spill implementation because it requires coordination between the build and probe sides, and recovery may require recursive repartitioning.

Normal Path vs. Spill Path

First Spill: Build Side Switches from Full Hash Table to Partitioned Writes

• First spill (is_spilled == false): Extract build data from the internal hash table → partition by join key → write each partition to its own SpillFile → disable Runtime Filter (the build side is no longer fully resident in memory)
• Subsequent spills (is_spilled == true): Continue flushing the current in-memory partition blocks to their corresponding spill files

Why the Probe Side Also Spills

Once a build-side partition has been written to disk, the probe side cannot use the current hash table for that partition, because the table no longer contains the complete build data. The probe operator partitions using the same join key, writes probe rows for spilled partitions to disk, and continues joining non-spilled partitions against the in-memory hash table.

Recovery Phase: Processing Partitions One at a Time

When input is exhausted, the probe side assembles, builds, and probe spill files into a unified Partition Queue (_spill_partition_queue). Each element is a JoinSpillPartitionInfo containing both a build_file and a probe_file. Each partition is processed in four steps:

1. Recover build blocks from the build spill file into memory
2. Build the internal hash table from those build blocks
3. Recover probe blocks from the probe spill file
4. Complete the join for that partition and output results

Recursive Repartitioning: Handling Data Skew

If a recovered partition's build data is still too large for available memory, SpillRepartitioner splits the current build/probe partition into next-level sub-partitions (level + 1), which are placed back into _spill_partition_queue. This repeats up to spill_repartition_max_depth levels.

This means Apache Doris' Hash Join spill is not a simple write to disk, read back operation, it's partitioned execution with recursive partitioned recovery as needed.

Aggregation:

This section covers the Partitioned Aggregation spill. Streaming Aggregation does not support spill; related parameters only limit its memory usage.

Sink Side: Serialize Group State and Partition to Disk

When revoke_memory() triggers: iterate the current hash table → partition by group key hash → serialize key columns and aggregate state into blocks → write each partition to a SpillFile → reset the hash table.

Key point: Aggregation spill writes intermediate aggregation state, not final results. The same group may be spilled multiple times, and recovery requires re-merging.

Source Side: Recover by Partition, Merge, Output

1. Read a batch of blocks from the spill file
2. merge_with_serialized_key: merge the serialized key/state back into the in-memory hash table
3. Repeat until the current partition is fully read
4. Produce final results from the hash table
5. Reset the hash table and continue to the next partition

If the hash table grows large again during recovery, SpillRepartitioner creates sub-partitions, which are placed back into _partition_queue.

Aggregation recovery doesn't pipe spill file data directly to downstream operators. It first reconstructs the hash table, then produces output normally from that table.

Order By/Sort:

Spill for Sort is different from Join or Aggregation. It follows a classic External Merge Sort model, where the core unit is a sorted run rather than a partition.

Basic Approach

External Merge Sort flow

Sink Side

Each revoke_memory() trigger:

1. Mark shared state as spilled, record limit/offset
2. Create a new SpillFile
3. Call the internal Sorter's prepare_for_spill() to complete sort preparation
4. Loop reading sorted blocks and write them as one sorted run to disk
5. Reset the sorter and continue receiving new input

Source Side: Multi-Way Merge

• Fewer runs: Use VSortedRunMerger directly for the final merge.
• Too many runs: Select a batch of runs for an intermediate merge to generate a temporary spill file, then repeat until the run count is small enough.

The number of files that can be merged in parallel = spill_sort_merge_mem_limit_bytes ÷ spill_buffer_size_bytes (minimum 8). The user query's limit/offset is applied only in the final merge round.

Operator Comparison Summary

6. Key Configuration Parameters

The most important parameters in the current implementation:

Key Workload Group watermark parameters:

• memory_low_watermark
• memory_high_watermark
• slot_memory_policy

Note: low_watermark and high_watermark don't directly mean "start spilling." They signal high memory pressure to the system, which then influences how paused queries are handled.

7. Observability

When a spill occurs, these tools are available for monitoring and debugging:

• Query Profile: Spilled operators show "Spilled": "true" in their profile, along with timers like SpillWriteSerializeBlockTime and SpillMergeSortTime. These help identify where disk write latency is concentrated.
• Logs: Key spill events are logged at INFO level (e.g., "all spill tasks done, resume it"). More detailed partition-level information is available by enabling VLOG_DEBUG.
• Force spill testing (enable_force_spill): Setting this to true bypasses reserve and Workload Group watermark checks and forces all spill-capable operators through the disk path. Useful for verifying spill correctness in testing without needing to reproduce real memory pressure.

8. Benchmark: TPC-DS 10TB on a Single Node

Test Environment and Configuration

Test environment:

• 1 FE (8C16G) + 1 BE (8C16G, 500GB SSD)
• Labeled as 1BE 10T TPC-DS test

Session variables used:

set global enable_profile = true;
set global spill_hash_join_partition_count = 8;
set global broadcast_row_count_limit = 3000000.0;
set global query_timeout = 28800;
set global profile_level = 2;
set global enable_spill = true;
set global spill_join_build_sink_mem_limit_bytes = 1073741824;
set global spill_aggregation_partition_count = 8;
set global spill_aggregation_sink_mem_limit_bytes = 1073741824;
set global exec_mem_limit = 8589934592;
set global max_file_scanners_concurrency = 8;

Workload Group configuration:

alter workload group normal properties(
    'slot_memory_policy' = 'dynamic'
);

BE configuration:

disable_storage_page_cache = true
spill_storage_limit = 95%
spill_storage_root_path = /mnt/disk1/storage
mem_limit = 14G
max_sys_mem_available_low_water_mark_bytes = 268435456
soft_mem_limit_frac = 0.95
disable_segment_cache = true
max_external_file_meta_cache_num = 64

Schema Setup

Data was prepared using an external catalog:

CREATE CATALOG `dlf` PROPERTIES (
    "type" = "hms",
    "ipc.client.fallback-to-simple-auth-allowed" = "true",
    "hive.metastore.type" = "dlf",
    "dlf.uid" = "217316283625971977",
    "dlf.secret_key" = "xxxx",
    "dlf.region" = "cn-beijing",
    "dlf.proxy.mode" = "DLF_ONLY",
    "dlf.endpoint" = "dlf-vpc.cn-beijing.aliyuncs.com",
    "dlf.catalog.id" = "emr_dev",
    "dlf.access_key" = "xxxxx"
);

Query Results

Important limitation: The Intersect/Except operator does not yet support spill. SQL involving that logic was rewritten using equivalent Join syntax for this test.

This reflects the current coverage boundary: spill mainly covers Join, Agg, and Sort. Set operators are not yet fully supported.

System Resource Usage

Memory usage:

CPU usage:

Representative Test Cases

Three representative SQL types cover the most important spill scenarios:

• q14 / q14_1: Centered on cross_items and avg_sales CTEs, combining multi-table joins, Union All, aggregation, Having, Rollup, Order By, and Limit in a complex execution path.
• q38: Joins three customer sets from store_sales, catalog_sales, and web_sales, validating execution under complex join combinations.
• q87: Uses Left Join + IS NULL to express semantics that would normally require Intersect/Except, a workaround for the current Set spill limitation.

These cases validate that Apache Doris can stably combine and execute multiple operator spill mechanisms in real TPC-DS-style SQL.

Summary

To summarize the Apache Doris Spill to Disk mechanism in one sentence: Use Reserve Memory to detect risk early; WorkloadManager pauses the query; specific operators convert their revocable state into SpillFiles; and during recovery, that state is read back in smaller-granularity batches.

The overall flow:

PipelineTask::execute()
  ├── get_reserve_mem_size()  ──→  estimate memory needs for this round
  ├── _should_trigger_revoking()  ──→  check for high memory pressure
  │     ├── reserve × parallelism > query_limit / 5 ?
  │     ├── query_used ≥ 90% or WG touches watermark?
  │     └── revocable × parallelism ≥ query_limit × 20%?
  ├── [high pressure] add_paused_query(QUERY_MEMORY_EXCEEDED)
  │     └── WorkloadGroupMgr::handle_paused_queries()
  │           ├── already recovered → resume
  │           ├── has revocable → QueryTaskController::revoke_memory()
  │           │     ├── sort tasks by revocable size
  │           │     ├── target recovery = consumption × 20%
  │           │     └── PipelineTask::do_revoke_memory()
  │           │           └── per-operator revoke_memory()
  │           └── no revocable → disable reserve, may cancel
  └── [normal] try_reserve() → continue execution

The three core operators differ in approach:

• Join: Partition by join key, coordinate spill across both build and probe sides, recursively repartition on recovery as needed.
• Aggregation: Serialize intermediate aggregation state and partition by key. During recovery, merge back into the hash table before producing output.
• Sort: Each spill produces one sorted run. Final output uses multi-way merge.

In addition, SpillIcebergTableSinkOperatorX (be/src/exec/operator/spill_iceberg_table_sink_operator.cpp) also implements revoke_memory(), supporting spill for Iceberg table write scenarios.

Operators that currently do not support spill include Window Functions / Analytic Functions and Set operators (Union/Intersect/Except). When these operators run out of memory, queries still terminate with an error.

The most important thing to understand about Apache Doris's Spill to Disk design: it's deeply embedded in Doris' execution engine, tightly coupled with PipelineTask, MemTracker, WorkloadGroup, and Operator State, making it a core scheduling mechanism in the execution engine itself.

Join the Apache Doris community on Slack and connect with Doris experts and users. If you're looking for a fully managed Apache Doris cloud service, contact the VeloDB team.