Implementation:ArroyoSystems Arroyo Sink V2 Open File

Knowledge Sources	ArroyoSystems_Arroyo
Domains	Streaming, Connectors, File_Systems
Last Updated	2026-02-08 08:00 GMT

Overview

Implements the OpenFile state machine that manages the complete lifecycle of a single output file in the V2 filesystem sink, from initial writing through multipart upload orchestration, checkpointing, recovery, and finalization.

Description

The OpenFile<BBW> struct is the central file management abstraction in the V2 sink. It encapsulates a state machine (OpenFileState) that tracks the file through its entire lifecycle:

States:

• New - File has a writer but no multipart upload has been started. Data is buffered locally. Transitions to MultipartStarting when the buffer reaches minimum_multipart_size or target_part_size_bytes.
• MultipartStarting - A multipart upload initialization request has been sent. Optionally tracks then_close if the file needs to close before the init completes.
• MultipartStarted - Multipart upload is active. Parts are flushed when the buffer reaches target_part_size_bytes. Each part is tracked as a Part struct with index and optional content ID.
• Recovering - Restored from checkpoint with a known multipart ID, completed parts, and trailing bytes that still need uploading.
• ClosingMulti - Writer has been closed; all data converted to parts. Waiting for in-flight part uploads to complete.
• ClosingSingle - Writer has been closed and all data fits in a single file (no multipart). Bytes are held in memory.
• Finishing - Multipart finalization or single file upload has been initiated.
• Closed - File is fully written and finalized.
• Failed - Terminal error state.

Key Operations:

• add_batch - Writes a batch to the underlying BatchBufferingWriter and triggers maybe_flush to check if a new part upload should start.
• close - Transitions the file to a closing state. For multipart files, the writer's final bytes become the last upload parts. For small files, the bytes are held for single-file upload.
• handle_event - Processes asynchronous upload results (MultipartInitialized, PartFinished, SingleFileFinished, MultipartFinalized) and transitions the state machine accordingly.
• prepare_for_commit - Determines the commit strategy: Serializable (multipart files already uploaded), UploadThenSerialize (single files needing upload in PerOperator mode), or LocalCommit (single files in PerSubtask mode).
• finalize - Initiates the actual multipart close or single-file upload.
• as_checkpoint - Serializes the current file state for checkpoint recovery.
• from_checkpoint / from_commit - Reconstructs an OpenFile from checkpoint or commit data for recovery.

Supporting Types:

• PendingSingleFile - Minimal type holding a path, bytes, and metadata for a single-file upload that will be finalized during the commit phase.
• CommitPreparation - Enum describing the three possible ways a file can enter the commit pipeline.
• Part - Tracks a multipart upload part index and its completion status.

The module handles object store constraints such as same-part-size requirements (e.g., S3) by splitting oversized final parts using split_into_parts.

Usage

Used exclusively by FileSystemSinkV2 to manage individual output files. Each open partition maps to one OpenFile instance.

Code Reference

Source Location

• Repository: ArroyoSystems_Arroyo
• File: crates/arroyo-connectors/src/filesystem/sink/v2/open_file.rs
• Lines: 1-1081

Signature

pub struct OpenFile<BBW: BatchBufferingWriter + 'static> {
    pub path: Arc<Path>,
    pub stats: MultiPartWriterStats,
    pub logger: FsEventLogger,
    pub state: OpenFileState<BBW>,
    pub target_part_size_bytes: usize,
    pub minimum_multipart_size: usize,
    pub storage_provider: Arc<StorageProvider>,
}

pub enum OpenFileState<BBW: BatchBufferingWriter> {
    New { writer: BBW },
    MultipartStarting { writer: BBW, then_close: bool },
    MultipartStarted { writer: BBW, multipart_id: Arc<MultipartId>, parts: Vec<Part> },
    Recovering { multipart_id: Arc<MultipartId>, parts: Vec<Part>, trailing_bytes: Bytes, ... },
    ClosingMulti { multipart_id: Arc<MultipartId>, parts: Vec<Part>, ... },
    ClosingSingle { bytes: Bytes, iceberg_metadata: Option<IcebergFileMetadata> },
    Finishing { iceberg_metadata: Option<IcebergFileMetadata>, total_size: usize },
    Closed { iceberg_metadata: Option<IcebergFileMetadata>, total_size: usize },
    Failed,
}

pub enum CommitPreparation {
    Serializable(FileToCommit),
    UploadThenSerialize(BoxFuture<'static, DataflowResult<FileToCommit>>),
    LocalCommit(PendingSingleFile),
}

pub struct PendingSingleFile { ... }

impl<BBW: BatchBufferingWriter + 'static> OpenFile<BBW> {
    pub fn new(path: Arc<Path>, writer: BBW, ...) -> Self;
    pub fn from_checkpoint(file: InProgressFile, ...) -> DataflowResult<Self>;
    pub fn from_commit(file: FileToCommit, ...) -> DataflowResult<Self>;
    pub fn is_writable(&self) -> bool;
    pub fn ready_to_finalize(&self) -> bool;
    pub fn add_batch(&mut self, batch: &RecordBatch) -> DataflowResult<Option<UploadFuture>>;
    pub fn handle_event(&mut self, event: FsResponseData) -> DataflowResult<Vec<UploadFuture>>;
    pub fn close(&mut self) -> DataflowResult<Vec<UploadFuture>>;
    pub fn prepare_for_commit(self, strategy: CommitStrategy) -> DataflowResult<CommitPreparation>;
    pub fn finalize(&mut self) -> DataflowResult<UploadFuture>;
    pub fn as_checkpoint(&mut self) -> DataflowResult<InProgressFile>;
    pub fn metadata_for_closed(self) -> DataflowResult<(usize, Option<IcebergFileMetadata>)>;
}

Import

use arroyo_connectors::filesystem::sink::v2::open_file::{
    OpenFile, OpenFileState, CommitPreparation, PendingSingleFile,
};

I/O Contract

Inputs

Name	Type	Required	Description
path	Arc<Path>	Yes	Object store path for the file
writer	BBW (BatchBufferingWriter)	Yes	Format-specific buffered writer
batch	RecordBatch	Yes	Arrow record batch to write to the file
event	FsResponseData	Yes	Async upload event (init, part finish, file finish, finalize)
InProgressFile	InProgressFile	No	Checkpoint data for recovery
FileToCommit	FileToCommit	No	Pre-commit data for recovery during commit phase

Outputs

Name	Type	Description
UploadFuture	UploadFuture	Asynchronous upload operations (multipart init, part upload, finalize, single file put)
InProgressFile	InProgressFile	Serializable checkpoint state for the file
FileToCommit	FileToCommit	Serializable pre-commit data for the commit phase
CommitPreparation	CommitPreparation	Strategy for how this file should enter the commit pipeline

Usage Examples

// Create a new open file
let mut file = OpenFile::new(
    path,
    parquet_writer,
    logger,
    storage_provider,
    representative_timestamp,
    &config,
);

// Write batches
if let Some(upload_future) = file.add_batch(&batch)? {
    pending_uploads.push(upload_future);
}

// Close the file when rolling policy triggers
let close_futures = file.close()?;

// After all uploads complete, prepare for commit
let preparation = file.prepare_for_commit(CommitStrategy::PerOperator)?;

Related Pages

• Environment:ArroyoSystems_Arroyo_Rust_Runtime