Implementation:ArroyoSystems Arroyo Record Partitioning

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

Overview

Partitioner splits Arrow RecordBatch data into groups based on partition key expressions, enabling Hive-style partitioned file output in the Arroyo filesystem sink.

Description

The Partitioner struct compiles DataFusion Expr expressions against an Arrow Schema into PhysicalExpr instances using the DefaultPhysicalPlanner. It supports two modes via PartitionerMode: FileConfig (for file system connectors with Hive-style paths) and Iceberg (for Iceberg partitioning). The partition method evaluates expressions against a RecordBatch to produce key arrays, converts them to OwnedRow instances via a RowConverter, groups rows by partition key while preserving insertion order, and returns a vector of (OwnedRow, RecordBatch) tuples. The hive_path method converts a partition row back to a Hive-style path string (e.g., year=2025/month=10/day=07). The module includes unit tests validating column-based partitioning, scalar key collapse, and timestamp-based partitioning patterns using strftime-compatible format strings.

Usage

Use Partitioner when implementing file system sinks that need to partition output files by column values or time-based expressions in Hive-compatible directory structures.

Code Reference

Source Location

• Repository: ArroyoSystems_Arroyo
• File: crates/arroyo-connectors/src/filesystem/sink/partitioning.rs

Signature

#[derive(Clone, Debug)]
pub enum PartitionerMode {
    FileConfig(PartitioningConfig),
    Iceberg(IcebergPartitioning),
}

pub struct Partitioner {
    mode: PartitionerMode,
    exprs: Vec<Arc<dyn PhysicalExpr>>,
    row_converter: RowConverter,
}

impl Partitioner {
    pub fn new(mode: PartitionerMode, schema: &Schema) -> DataflowResult<Self>;
    pub fn is_partitioned(&self) -> bool;
    pub fn partition(&self, batch: &RecordBatch)
        -> Result<Vec<(OwnedRow, RecordBatch)>, DataFusionError>;
    pub fn hive_path(&self, partition: &OwnedRow) -> Option<String>;
}

fn compile_expression(expr: &Expr, schema: &Schema)
    -> Result<Arc<dyn PhysicalExpr>, DataFusionError>;

Import

use arroyo_connectors::filesystem::sink::partitioning::{Partitioner, PartitionerMode};

I/O Contract

Inputs

Name	Type	Required	Description
mode	PartitionerMode	Yes	FileConfig or Iceberg partitioning mode
schema	&Schema	Yes	Arrow schema for the record batches to partition
batch	&RecordBatch	Yes	Arrow record batch to partition (passed to partition())

Outputs

Name	Type	Description
partitions	Vec<(OwnedRow, RecordBatch)>	List of partition key and corresponding record batch pairs
hive_path	Option<String>	Hive-style directory path string for a given partition key

Usage Examples

use arroyo_connectors::filesystem::sink::partitioning::{Partitioner, PartitionerMode};
use crate::filesystem::config::PartitioningConfig;

let config = PartitioningConfig {
    time_pattern: Some("%Y/%m/%d/%H".to_string()),
    ..Default::default()
};
let partitioner = Partitioner::new(
    PartitionerMode::FileConfig(config),
    &schema,
)?;
let groups = partitioner.partition(&batch)?;
for (key, sub_batch) in groups {
    if let Some(path) = partitioner.hive_path(&key) {
        // write sub_batch to path like "2025/10/07/21"
    }
}

Related Pages

• Environment:ArroyoSystems_Arroyo_Rust_Runtime