Daft is a computing engine designed for DATA + AI multimodal data processing and analysis scenarios. It supports both single-node and distributed operation modes, with its core implemented in Rust, and provides two interaction methods: SQL and Python DataFrame.

Considering the characteristics of multimodal data processing in DATA + AI scenarios, Daft adopts a Pipeline execution model for its overall architecture design. Applications written by users using Daft DataFrame or SQL API are first represented internally in Daft as an abstract syntax tree (AST: Abstract Syntax Tree). After a series of optimization and transformation rules are applied, this abstract syntax tree is converted into a Pipeline tree that can be executed by the underlying execution engine, and is ultimately submitted to either the single-node or distributed execution engine for execution.

Architecture layering #

As shown in the diagram above, Daft's overall architecture can be divided into three layers: the API layer, the Plan layer, and the Execution layer:

• API layer: Provides two access APIs, SQL and Python DataFrame. By abstracting distributed data into tables and splitting it into multiple partitions along the row dimension, the partitions are distributed across nodes in the cluster for distributed processing, enabling parallel computation of data. Daft partitions data by default using heuristic strategies (for example, by inheriting the default partitioning strategy of files), and also supports manual repartitioning.
• Plan layer: Based on user input in SQL or constructed DataFrame objects, a logical execution plan (LogicalPlan) is built. The built-in LogicalPlan optimizer is used to optimize it (such as eliminating CrossJoin, Subquery, Repartition, predicate pushdown, column pruning, Join reorder, and so on) to obtain an optimized LogicalPlan. The optimized LogicalPlan is further converted into a PhysicalPlan, and the PhysicalPlan optimizer is used for further optimization (such as eliminating unnecessary repartitioning).
• Execution layer: Daft supports both single-node and distributed operation modes. The PhysicalPlan constructed by the Plan layer needs to be further split into a series of executable tasks according to the specific operation mode. The Scheduler dispatches these tasks to the single-node Swordfish engine or the distributed Flotilla engine for execution, while also managing the execution status of the tasks.

Plan example #

Let us take the following DataFrame as an example and explain, from the perspective of the overall execution pipeline, how Daft starts from this DataFrame and ultimately generates a PhysicalPlan through optimization and transformation.

df = daft.read_parquet("/opt/workspace/daft/data")

df = (
    df
    .select("id", "name")
    .sort(by="id", desc=True)
    .filter(col("id") < 127)
    .offset(100)
    .limit(10)
)

df.to_pydict()

This is a relatively simple DataFrame example, used to read a parquet dataset and select 10 records from it as required. Internally, Daft represents the DataFrame by converting it into a LogicalPlan according to the construction order:

* Limit: 10
|
* Offset: 100
|
* Filter: col(id) < lit(127)
|
* Sort: Sort by = (col(id), descending, nulls first)
|
* Project: col(id), col(name)
|
* GlobScanOperator
|   Glob paths = [/opt/workspace/daft/data]
|   Coerce int96 timestamp unit = Nanoseconds
|   IO config = ...
|   Use multithreading = true
|   File schema = id#Int64, name#Utf8, email#Utf8
|   Partitioning keys = []
|   Output schema = id#Int64, name#Utf8, email#Utf8

Considering that the DataFrame or SQL written by the user may not be optimal, to achieve maximum execution efficiency, Daft's internal LogicalPlan optimizer traverses and applies a series of optimization rules to the above LogicalPlan, ultimately producing the following optimized LogicalPlan:

* TopN: Sort by = (col(id), descending, nulls first), Num Rows = 10, Offset = 100
|   Stats = { Approx num rows = 110, Approx size bytes = 3.01 KiB, Accumulated
|     selectivity = 0.01 }
|
* Num Scan Tasks = 112
|   File schema = id#Int64, name#Utf8, email#Utf8
|   Partitioning keys = []
|   Projection pushdown = [id, name]
|   Filter pushdown = col(id) < lit(127)
|   Output schema = id#Int64, name#Utf8
|   Stats = { Approx num rows = 3,279, Approx size bytes = 92.09 KiB, Accumulated
|     selectivity = 0.30 }

The optimized LogicalPlan is much more streamlined compared to the original, resulting in improved execution efficiency. From the optimized LogicalPlan above, we can also infer that Daft has roughly applied the following optimization rules:

• Predicate pushdown: Operations such as .filter(col("id") < 127) and .select("id", "name") are pushed down to the data scan stage, greatly reducing the amount of data scanned.
• Limit Offset rewriting: The semantics of .offset(100).limit(10) are implemented by rewriting it as global_limit(10, 100) and local_limit(0, 110) to achieve offset...limit.
• Sort Limit rewriting: The operation .sort(by="id", desc=True).limit(10, 100) is rewritten as a TopN operator to implement sorting and truncation semantics.

The optimized LogicalPlan is further converted into a PhysicalPlan, and additional optimization rules specific to the PhysicalPlan are applied. For example, the optimized LogicalPlan above would be converted into the following PhysicalPlan:

* TopN: Sort by = (col(0: id), descending, nulls first), Num Rows = 10, Offset
|     = 100
|   Stats = { Approx num rows = 110, Approx size bytes = 3.01 KiB, Accumulated
|     selectivity = 0.01 }
|
* ScanTaskSource:
|   Num Scan Tasks = 112
|   Estimated Scan Bytes = 180568
|   Pushdowns: {projection: [id, name], filter: col(id) < lit(127)}
|   Schema: {id#Int64, name#Utf8, email#Utf8}
|   Scan Tasks: [ ... ]
|   Stats = { Approx num rows = 3,279, Approx size bytes = 92.09 KiB, Accumulated
|     selectivity = 0.30 }

Currently, mainstream data computation engines in the DATA + AI domain (such as Pandas and Polars) by default only provide single-node execution mode. The distributed version of Polars is only available in the commercial edition, while Dask, although considered a distributed version of Pandas, has relatively low execution efficiency. Daft provides both single-node and distributed execution modes in its open-source version, and allows switching freely between these two modes.

Single-node execution engine #

By default, Daft executes user-input SQL or constructed DataFrames in single-node mode. Its single-node execution engine, Swordfish, is implemented in Rust and is designed based on a pipeline computation model, leveraging multithreading to enable parallel data processing.

As shown in the figure above, the PhysicalPlan on the left is parsed by the Swordfish engine into a Pipeline tree. The engine traverses the tree in a depth-first manner and sequentially starts and runs each node. Swordfish PipelineNode includes four types of nodes:

Data transmission between nodes relies on Kanal Channel.

Distributed execution engine #

Daft supports not only running in standalone mode, but also running in distributed mode based on Ray. Users only need to simply configure the connection information of the Ray cluster to switch between standalone and distributed modes, without the need to adapt the business logic layer.
Flotilla v1.5 is the latest distributed execution engine of Daft. Its design goal is to provide better scalability, performance, and observability for multimodal data processing, supporting:

1. Robust handling of highly scalable multimodal pure mapping pipelines (reading URLs, downloading UDFs, and write modes).
2. Reliable reshuffling of 10TB of data for data exchange operations such as groupby, join, and sort.
3. Providing easy-to-understand and interactive observability tools.
4. Allowing Daft developers to quickly debug and develop efficiently.

In terms of design philosophy, Flotilla v1.5 leverages Ray to distribute multiple Swordfish standalone execution engines, and data shuffling between nodes is achieved via the built-in Flight Shuffle Server.

The implementation of the Flotilla distributed execution engine includes the following core components:

• Ray Worker Manager: Responsible for managing Swordfish Worker nodes on the Ray cluster, ensuring that each Ray Worker node runs a RaySwordfishActor. It also triggers Ray cluster autoscaling as needed, submits Swordfish Tasks to target Worker nodes for execution, and other related tasks.
• Plan Runner: Responsible for executing the PhysicalPlan by splitting the StagePlans corresponding to the PhysicalPlan into a series of SwordfishTasks, which are scheduled by the Scheduler for execution on the Ray cluster.
• Scheduler: Responsible for continuously monitoring the task queue to check for pending SwordfishTasks and distributing tasks to the Ray cluster for execution via the Dispatcher. Additionally, the Scheduler is responsible for triggering Ray cluster autoscaling upon receiving an autoscale request.
• Dispatcher: Responsible for organizing SwordfishTasks by Ray Worker node and submitting tasks in batches to the target Worker nodes in the Ray cluster for execution.

Each Ray Worker node runs a resident RaySwordfishActor, which is responsible for receiving the LocalPhysicalPlan contained within a SwordfishTask that can be executed on a standalone Swordfish engine. When an execution request is received, the RaySwordfishActor submits the LocalPhysicalPlan to the local standalone Swordfish engine for execution. For details on the execution process, refer to Section 1 above.
The Flotilla distributed execution engine has now evolved to version v1.5, which offers significant performance improvements over v1.0. However, its feature completeness is still limited, so when unsupported operations are encountered, it will fall back to v1.0 to ensure that the job can run properly. Unlike v1.5, which uses the Actor model in conjunction with the standalone Swordfish execution engine to run Pipeline operators, v1.0 splits each Stage into multiple Ray Remote Tasks and submits them in batches to the Ray cluster for execution.

As shown in the left part of the figure above, the core execution process of PhysicalPlan in v1.0 is as follows:

1. Based on the PhysicalPlanScheduler, the PhysicalPlan is divided into a series of Tasks, and Stages are split according to their dependencies. A Task mainly consists of three attributes: input, a set of instructions, and a resource description.
2. Tasks are batched according to their dependencies and the number of tasks that the current Ray cluster can execute (measured by available CPU), and are submitted in batches for execution in Ray Remote Task mode.

3. The Scheduler internally maintains a Queue to record the final execution results of tasks and provides an iterator interface for upstream access.

The main reason for redesigning the Flotilla distributed execution engine is that v1.0 has the following two core issues compared to v1.5:

1. Low execution performance: For moderately complex tasks, the system typically splits them into multiple Stages, each containing a large number of Tasks. Daft, during scheduling, prioritizes submitting the Tasks of child Stages to the Ray cluster. When the number of Tasks is large, the execution mode often degrades from Pipeline to Stage by Stage. For heterogeneous computing scenarios such as multimodal data processing, Stage by Stage execution leads to underutilization of resources, resulting in inefficient processing performance.
2. High memory pressure: Ray clusters use Object Memory Store to facilitate data exchange between tasks. The stage-by-stage execution model requires all upstream stage tasks to be completed before downstream stage tasks can begin, so all intermediate data must be stored in Object Memory Store. This places significant memory pressure on Ray and can easily lead to OOM errors.