[Feature][BufferManager] Support `BufferManager` in StreamingWindowBuild

[wangxinshuo-bolt]

Overview

• From Bolt's internal perspective, Bolt already has spill capabilities. However, the current source code still implements spill mainly through "operator-aware, container-aware, and format-aware" logic rather than through a unified memory lifecycle management layer. The BufferManager for Bolt discussed in this document aims to lift spill into a more fundamental infrastructure layer.
• From the Spark application perspective, reducing container specifications to the P99 of actual usage can increase the core-to-memory ratio by 10% to 20%. BufferManager is expected to allow Bolt to run under extremely low-memory conditions without OOM errors. Therefore, implementing BufferManager is also significant for reducing overall memory resource consumption.

What Is BufferManager?

The BufferManager idea comes from DuckDB. The following simplified example explains the BufferManager mechanism:

// Describes whether a BlockHandle payload currently resides in memory.
enum MemoryStatus {
    IN_MEMORY,
    SPILLED
};

struct BlockHandle {
    int id;
    MemoryStatus status;

    // The data is stored in payload.
    // After the block is spilled to disk, the memory held by payload can be released.
    unique_ptr<BlockMemory> payload;
};

// Simulates repeatedly accessing block data during database execution.
// A block can be of any size.
void Pipeline(BufferManager& buffer_manager,
              vector<shared_ptr<BlockHandle>>& blocks) {
    for (auto& block : blocks) {
        // Pin semantics:
        // Return an accessible BufferHandle.
        // If the block is already in memory, it can be used directly.
        // If the block has been spilled, BufferManager reads it back into memory.
        BufferHandle handle = buffer_manager.Pin(block);

        // The caller does not care whether the data was previously spilled.
        // It only cares that the data is now accessible through the handle.
        char* ptr = handle.Ptr();
        Consume(ptr);

        // The handle is destructed here.
        // RAII ends ownership of the payload access and makes it evictable again.
    }
}

Problems to Solve

This design aims to address the following issues:

1. Reduce the complexity of existing spill logic in the Bolt engine.
   1. Spill logic is still implemented mainly in an "operator-aware, container-aware, and format-aware" manner.
   2. New operators should not need to reinvent their own spill finite-state machine. The overall implementation should be more unified. Taking HashBuild as an example, spill requires 23 functions, covering various stages of spill: spill status/judgment, spill creation/recovery process, spill memory/trigger control, spill data processing, and spill statistics; there are 17 member variables related to spill, which will be used in various aspects such as spill configuration/control and spill execution.
2. Improve stability by covering long-tail cases that currently cannot be spilled at the right time.
   1. For example, many current Window OOM errors observed in production logs.
   2. For example, the large-string TableScan OOM case.

Performance Gains

This design is expected to bring the following benefits:

1. Less data copying: spill no longer requires data reproduction and can write directly from memory to disk. Even row-based spill still involves one memory copy.
2. Lower spill cost: taking RowContainer as an example, once BufferManager is implemented, partial and on-demand spill become possible, and full spill may no longer be necessary.
3. The current mechanism relies on the maybeReserve interface to trigger spill, so the program tends to request more memory than it needs on demand, resulting in a larger spill volume.
   1. maybeReserve tries to free as much memory as possible. If the requested reservation is too small and the spill opportunity is missed, spill may no longer be possible and the operation may fail due to insufficient memory.
4. Unified executor-level IO management can reduce competition and disk busyness.

Future Potential Benefits

After this design is implemented, it can also create the following opportunities:

1. Create room for IO and computation overlap.
   1. In the current design, an operator relies on reserve / mayBeReserve to trigger spill, and this operation is synchronous. In the future, spill could be performed concurrently with computation to overlap part of the cost.
2. Create the possibility of spill-file prefetching and heterogeneous spill storage.
   1. Perform block-granularity prefetching by combining block metadata with operator context.
   2. Differentiate cold and hot blocks and store them hierarchically on media with different speeds, such as SSD and HDD.
3. With support from BufferManager, Bolt is expected to run stably under extremely low memory. Performance may degrade, but execution should not fail. This is significant for improving the core-to-memory ratio of Spark tasks.

Prototype Selection

The first scenario to focus on is StreamingWindowBuild. The selection rationale is that the current Window computation implementation in Bolt and the current StreamingWindowBuild logic provide a representative and high-value target for validating this design.

Design Plan

Option A: DuckDB-Style BufferManager / BlockManager

BufferManager Design in DuckDB

DuckDB's BufferManager essentially separates "logical block identity" from "access rights to resident memory".

From DuckDB's interface definition, BufferHandle is a valid access handle that internally holds a BlockHandle and the actual FileBuffer pointer. It releases the handle upon destruction, prohibits copying, and supports moving. This is typical RAII access semantics.

In DuckDB, this mechanism can be understood through the following concepts:

Concept	Function
BlockManager	Responsible for the logical identity, external storage location, and read/write operations of a block.
BlockHandle	Represents the stable identity and state of a block. It is not equivalent to a directly accessible data pointer.
BlockMemory	The actual storage location of block data.
BufferManager	Responsible for resident memory budget, block pinning, eviction, temporary memory registration, and coordination when blocks are loaded from disk back into memory.
BufferHandle	A short-lived access right. The caller must obtain the handle to legally use the memory pointer, and the pin is automatically released when the handle is destructed.

Key implications:

• The logical identity of a block is stable, but its physical address is not a long-term stable commitment.
• The caller should not cache raw pointers for an extended period. It should obtain a handle when access is needed.
• Reclaiming resident memory does not require upper-level objects to remember all spill details themselves. The underlying manager can choose objects to evict based on pin count and memory pressure.
• External storage read/write and memory residency are handled by different layers, so responsibilities are clearer.

Combination with Bolt

The core idea of Option A in Bolt is to create a new BufferManagerRowContainer. The upper layer should no longer hold a stable char* row for a long time. Instead, the logical identity of a row should be represented as RowId, and "accessing data" should become a controlled, short-lived pin operation.

The object boundaries in Bolt can be defined similarly to DuckDB:

Concept / Class Name	Function
BlockManager	Responsible for block_id allocation, spill file layout, block write-out, and block read-back.
BufferManager	Responsible for resident block memory budget, pinning, eviction, restoration, and reclaim integration.
BlockHandle	Represents the identity of a block.
BufferHandle	Represents valid resident access. The system requires explicit Pin, while Unpin is completed automatically via RAII.
BufferManagerRowContainer	Organizes row data by block or page. Appending a row returns a RowId. Accessing a row uses RowId -> block_id + offset to find the corresponding block, then obtains a BufferHandle and parses the row.
BufferManagerWindowPartition	No longer stores folly::Range<char**>. Instead, it stores RowId sequences, row-range descriptors, or row lists segmented by block. It still exposes the ability to extract columns and compute peers / frames to Window, but internally accesses data through RowId.

The execution flow is as follows:

StreamingWindowBuild append row
  -> BufferManagerRowContainer
      -> return RowId
  -> store RowId in partition metadata
  -> build WindowPartition logical view
  -> Window execution requests rows
      -> locate block by RowId
      -> BufferManager Pin
      -> BufferHandle lifetime
      -> read / compare / extract row data
      -> handle leaves scope
      -> automatic Unpin

Combination with StreamingWindowBuild

In the StreamingWindowBuild scenario, this solution does not simply insert a spill component. Instead, it replaces the current RowContainer + char* access model.

The specific changes are:

• StreamingWindowBuild::addInput() no longer pushes char* into inputRows_ / sortedRows_ for a long time after data_->newRow(). Instead, it gets a RowId after append.
• The current partitionStartRows_ logic that indexes into a char* array needs to be changed to index into the sortedRows_ array or into a set of RowId row-range descriptors.
• WindowPartition can no longer directly call RowContainer::extractColumn(...) based on folly::Range<char**>. It needs to be changed to RowId-oriented extractColumn, compareRows, and computePeerBuffers.
• The WindowPartition construction phase only establishes a logical view and does not perform batch pinning. Actual pinning happens when window computation needs to read specific rows.

After RowId replaces char*, how does Window::callApplyForPartitionRows continue to compute?

The required changes mainly include:

1. Add BufferManagerWindowPartition: change partition_ from folly::Range<char**> to a RowId view.
2. Add RowId-oriented access interfaces for BufferManagerRowContainer, including extractColumn(RowId...), compare(RowId, RowId, ...), extractNulls(RowId...), and decodeRow(RowId).
3. Rewrite peer/frame-related logic. computePeerBuffers and searchFrameValue currently assume rows can be directly compared through char*. After migration, the corresponding block needs to be briefly pinned in the comparison path.
4. Evaluate pin granularity. If each comparison pins a single row separately, overhead will be high. It is more reasonable to pin in batches by block during extractColumn, peer comparison, and frame search, or to use a small-range cache.

Option B: Automatic Page Replacement Based on Linux User Page Faults

Introduction to the Linux User Page Fault Mechanism

Linux provides mechanisms that allow user mode to participate in page fault handling. The representative interface is userfaultfd. The basic idea is that when a page fault occurs in certain virtual memory regions, the kernel does not directly complete all recovery actions. Instead, it passes the fault event to a user-mode processing thread. The user-mode thread decides how to prepare the page data, copies the data back to the faulting page through a dedicated ioctl, and then lets the original thread continue running.

Reference: Linux ioctl_userfaultfd(2) manual.

From the perspective of system behavior, this mechanism can support at least three things:

1. Monitor page faults in a virtual address range.
2. Determine where the page data comes from in user mode, such as a zero page, remote memory, spill file, or compressed page.
3. Restore the page content at an appropriate time in user mode and then resume the original execution flow.

This mechanism allows applications to decide which pages to evict first, where to evict them, and how to restore them upon subsequent access, without having to include "restore" as part of upper-level operator code.

Minimal userfaultfd Example

The following code is not a complete runnable project. It is only a minimal demonstration of the typical process: create userfaultfd, negotiate the API, register the address range, and fill pages with UFFDIO_COPY after receiving a fault in the handler thread.

#include <fcntl.h>
#include <linux/userfaultfd.h>
#include <poll.h>
#include <pthread.h>
#include <string.h>
#include <sys/ioctl.h>
#include <sys/mman.h>
#include <sys/syscall.h>
#include <unistd.h>
#include <cstdint>
#include <cstdio>
#include <cstdlib>

constexpr size_t kPageSize = 4096;

struct HandlerArgs {
  int uffd;
};

void* faultHandler(void* arg) {
  auto* args = reinterpret_cast<HandlerArgs*>(arg);
  struct pollfd pfd;
  pfd.fd = args->uffd;
  pfd.events = POLLIN;

  for (;;) {
    // Wait with poll. This can be optimized with epoll.
    int nready = poll(&pfd, 1, -1);
    if (nready <= 0) continue;

    // Read the parameters passed by the kernel from the fd.
    struct uffd_msg msg;
    ssize_t nread = read(args->uffd, &msg, sizeof(msg));
    if (nread != sizeof(msg)) continue;
    if (msg.event != UFFD_EVENT_PAGEFAULT) continue;

    // Get the page-aligned address that triggered the page fault.
    void* fault_addr =
        reinterpret_cast<void*>(msg.arg.pagefault.address & ~(kPageSize - 1));

    // Allocate and initialize memory.
    void* page = mmap(nullptr, kPageSize, PROT_READ | PROT_WRITE,
                      MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);

    // Example: in a real scenario, this page can be read from a spill file.
    memset(page, 0x5A, kPageSize);

    // Pass the initialized memory page to the kernel. The kernel copies this
    // page data to the corresponding faulting page.
    struct uffdio_copy copy;
    memset(&copy, 0, sizeof(copy));
    copy.src = reinterpret_cast<unsigned long>(page);
    copy.dst = reinterpret_cast<unsigned long>(fault_addr);
    copy.len = kPageSize;

    ioctl(args->uffd, UFFDIO_COPY, &copy);
    munmap(page, kPageSize);
  }

  return nullptr;
}

int main() {
  void* region = mmap(nullptr, 16 * kPageSize, PROT_READ | PROT_WRITE,
                      MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);

  // Register the user page fault fd.
  int uffd = syscall(SYS_userfaultfd, O_CLOEXEC | O_NONBLOCK);
  struct uffdio_api api;
  memset(&api, 0, sizeof(api));
  api.api = UFFD_API;
  ioctl(uffd, UFFDIO_API, &api);

  // Register region with the system. Once a page fault occurs in region, the
  // user page fault fd becomes readable.
  struct uffdio_register reg;
  memset(&reg, 0, sizeof(reg));
  reg.range.start = reinterpret_cast<unsigned long>(region);
  reg.range.len = 16 * kPageSize;
  reg.mode = UFFDIO_REGISTER_MODE_MISSING;
  ioctl(uffd, UFFDIO_REGISTER, &reg);

  // Start the page fault handler thread.
  HandlerArgs args{uffd};
  pthread_t tid;
  pthread_create(&tid, nullptr, faultHandler, &args);

  // This is the first access to region. The kernel has not allocated a page
  // for it yet, so this access triggers a page fault.
  volatile uint8_t value = *reinterpret_cast<volatile uint8_t*>(region);
  printf("first byte = %u\n", value);

  pause();
  return 0;
}

Detailed Design in Bolt

The core idea of this solution is to avoid modifying the access protocols of upper-level objects such as StreamingWindowBuild, WindowPartition, and RowContainer as much as possible. Instead, it establishes a controlled virtual memory mapping for the pages carried by RowContainer at the bottom layer and implements automatic restore through the user-mode page fault mechanism.

Its basic mode of operation can be summarized as follows:

1. RowContainer still stores row data, and the upper layer still uses the existing char* row addresses and traversal logic.
2. The new BufferManager component maintains a set of controlled virtual pages and registers them with userfaultfd.
3. When memory pressure arrives, the new BufferManager component selects cold pages and writes them to disk. You can use madvise to clean up the memory after writing to disk, and then accessing that memory again will trigger a page fault.
4. When the upper layer accesses these addresses again, a page fault is triggered.
5. The user-mode fault handler restores page content from the backing store and then allows the original thread to continue execution.

The execution flow is as follows:

Existing RowContainer page
  -> if no memory pressure: keep resident
  -> if memory pressure:
       pager selects cold page
       -> write page to backing store
       -> mark page non-resident
  -> operator accesses row
  -> page fault
  -> user-space fault handler
  -> read page from spill store
  -> UFFDIO_COPY restore page
  -> continue existing access path

How to Integrate with StreamingWindowBuild

The most prominent feature of Option B is that, in principle, it does not introduce new data structures, does not modify upper-level application code, and does not require transforming WindowPartition into RowId views.

This means:

• StreamingWindowBuild::addInput() can continue writing char* newRow = data_->newRow().
• inputRows_, sortedRows_, and windowPartitions_ can theoretically continue to store char*.
• Window::callApplyForPartitionRows(), WindowPartition::extractColumn(...), and computePeerBuffers(...) can remain largely unchanged.

How does Option B trigger spill?

1. When MemoryArbitrator / MemoryPool determines that memory needs to be reclaimed, the pager selects victim pages from the pages where RowContainer resides.
2. Write the victim pages to the spill / backing store.
3. After a successful write-back, mark these pages as non-resident in the virtual address space.
4. Any subsequent access to these char* addresses by the upper layer naturally triggers a page fault and restore.

Horizontal Comparison

Option A: DuckDB-Style BufferManager

Advantages

• Compared with Option B, all operations are implemented in user mode. This is less affected by kernel behavior and does not rely on kernel-specific page fault mechanisms.

Disadvantages

• The required transformation work is relatively extensive. Almost all data structures and computational logic in the affected path need to be rewritten.
• Developers need to understand the new access model.

Option B: Automatic Page Replacement Based on Linux User Page Faults

Advantages

• Minimal changes to upper-level code. This is the core advantage of Option B.

Disadvantages

• Strong dependency on specific Linux mechanisms.
• Observability and debugging are more difficult.
• Compared with Option A, data restore involves one additional memory copy.

Summary

Option A is an engineering approach where Bolt explicitly takes over the block/page lifecycle. Its advantages are clear boundaries and easier unification of memory accounting and reclaim, but the cost is that RowContainer, WindowPartition, and the row access protocol must be refactored.

Option B is a system-level approach that maximally preserves the upper layer and moves automatic page replacement into the lower-level pager. Its advantage is minimal change to the existing StreamingWindowBuild and Window::callApplyForPartitionRows paths, but it depends heavily on mechanisms such as Linux userfaultfd, which increases lower-level implementation risk.

Q&A

Why Does BufferManager Reduce the Logical Complexity of Spill?

From the source code, current spill-related logic is scattered across multiple levels instead of converging in a unified buffer/page management layer:

• Window needs to decide by itself when to trigger sortSpill() or spill(). See the reclaim() path in Window.cpp.
• Spiller maintains spillRuns_, fillSpillRuns(), runSpill(), markAllPartitionsSpilled(), and a full runtime finite-state machine. See Spiller.cpp.
• RowContainer layout, serialization, and spill-size calculation also contain spill-specific logic. For example, only the data before rowSizeOffset participates in spill. See RowContainer.h.
• RowsStreamingWindowBuild differentiates between ordinary memory rows and SerializedRows, and also needs to handle buildNextInputOrPartitionFromSpill(), loadNextPartitionFromSpill(), and storeRows(). See RowsStreamingWindowBuild.cpp, RowsStreamingWindowBuild.cpp, and RowsStreamingWindowBuild.cpp.

These examples show that current spill behavior is not managed by a unified underlying mechanism. Instead, it is embedded separately in Window, Spiller, RowContainer, WindowPartition, and other objects. This is the direct basis for the high complexity.

This is also the basis for the expected engineering complexity and maintainability benefits. If a unified layer is not established, new operators will most likely continue to replicate the operator-managed spill model.

Why Can BufferManager Improve Spill Performance?

In the current implementation, the spill/restore path has significant overhead in format conversion, copying, and synchronous processing:

• RowsStreamingWindowBuild::storeRows() copies temporary sorted results back into a newly allocated contiguous memory block and then converts it to sortRows_. See RowsStreamingWindowBuild.cpp.
• RowContainer::sizeIncrement() states that "for spilling the practical minimum increment is a huge page", indicating that existing container growth and spill granularity are not naturally aligned. See RowContainer.cpp.
• Spiller's row-based spill first organizes rows in the container into spill runs and then executes runSpill(lastRun). This is an explicit batch-processing path. See Spiller.cpp.
• WindowPartition and Window read paths still fundamentally rely on char* row pointers and RowContainer::extractColumn(...) for column extraction. See Window.cpp and WindowPartition.cpp.

Why Can BufferManager Create Room for IO and Computation Overlap?

Currently, reclaim/spill semantics on the Window side are relatively synchronous. It is difficult for the system to naturally form a "background swap in/out + foreground computation continues" execution mode:

• Window::reclaim() directly calls windowBuild_->sortSpill() or windowBuild_->spill(), then calls pool()->release(). See Window.cpp.
• RowsStreamingWindowBuild::loadNextPartitionFromSpill() is an explicit path for reading the next partition back from the spill merge stream, where computation and recovery are serially alternated. See RowsStreamingWindowBuild.cpp.
• Window::callApplyForPartitionRows() does not have its own independent page residency protocol. It only calls currentPartition_->extractColumn(...) and computePeerAndFrameBuffers(...). See Window.cpp. This means that once the underlying data is not in memory, the current implementation lacks a natural on-demand pin / fault-restore layer to decouple IO from computation.

Why Implement BufferManager Instead of Letting Each Operator Implement Spill?

From the operator dimension, Bolt may have only 20+ operators, and it may seem sufficient for each operator to implement spill independently. In practice, however, the internal situations across operators repeat. For example, the oversized single-partition problem currently faced by Window was already encountered and solved by HashBuild, but Window now has to solve it again. Another example is RowBasedSpill, which was implemented once by Agg and once by HashJoin.

BufferManager is intended to extract this repeated lifecycle management into a shared layer, so new operators do not need to repeatedly implement their own spill state machine, row restore logic, and memory-pressure response path.