[Proposal] Exploring vectorized execution for Cloudberry as a pure extension (vec_executor)

[MisterRaindrop]

Proposers

@MisterRaindrop

Proposal Status

Under Discussion

Abstract

This is a personal exploration of one question: how far can Cloudberry get with vectorized execution if it is added only as an extension, with no change to the kernel?

I built vec_executor, an extension that hands the slices of a query plan it can run to a vectorized engine and leaves the rest to Cloudberry's executor, so every query still runs and returns the same rows. The engine is currently based on Apache DataFusion and plugs in through existing hooks.

First results are encouraging: on a single host, TPC-H at scale factor 100 runs much faster than on the same cluster without the extension, with identical results. These are basic tests only; tests with more data, more nodes and more detail are being prepared.

I am posting this to share the exploration and to hear the community's views: whether the direction is worth pursuing, what concerns people most, and where the kernel could help. The code is open source under the Apache License 2.0: https://github.com/MisterRaindrop/cloudberry-datafusion

Motivation

Cloudberry's executor follows PostgreSQL's volcano model and processes one row at a time. For analytical queries dominated by scans, aggregates and joins, much of the CPU time goes to per-row function calls, interpreted expressions, and forming and deforming tuples. Cloudberry already has a columnar storage format, PAX, but the executor still works row by row, so the benefit of columnar storage shows up mainly in I/O.

A vectorized engine processes batches of columns at a time (here in Arrow format), which removes much of the per-row overhead and can run on several threads within a segment.

Rewriting the executor inside the kernel is costly and slow. This exploration asks whether Cloudberry can use a mature vectorized engine through an extension, keeping Cloudberry's semantics and operational behaviour, and if so, where the limits of that approach lie.

Related work. #2065 describes pg_vexec, a vectorized executor built as an extension on a PostgreSQL 19 port of Cloudberry. vec_executor differs in two ways: it targets the current Cloudberry main as it is, without any kernel patch, and it delegates whole plan slices to a separate engine library behind a C interface rather than implementing vectorized nodes inside PostgreSQL. I would be glad to compare notes.

Implementation

Design principles

1. Extension only, no kernel changes. It plugs in through planner_hook, ExecutorStart_hook, ExecutorRun_hook and ExplainOneQuery_hook. Without the extension loaded, or with vec.mode = off (the default), Cloudberry behaves exactly as before.
2. Reuse the interconnect. Data between slices still travels through Cloudberry's Motions and interconnect; there is no separate network layer, and distributed execution keeps its semantics. When both ends of a Motion run in the engine, the Motion can carry Arrow batches instead of tuples (vec.motion_batches, experimental).
3. Memory and resources under Cloudberry's control. The engine's memory pool is bounded by the operator memory the coordinator assigns, and spills beyond it. Every byte the engine allocates is leased from the vmem tracker. Engine threads belong to the QE process and are subject to resource group CPU limits.

Architecture

The extension has two layers:

• The host (src/, C) runs in every backend and is the only part that calls PostgreSQL. It checks whether each slice can run in the engine, translates the slices that can into a plan IR (JSON), scans tables with the query's snapshot, sends and receives Motion data on the interconnect, leases the engine's memory from the vmem tracker, and turns engine errors into ereports.
• An engine (engines/<name>/) is a separate library, vec_engine_<name>.so, chosen by vec.engine and loaded on first use. It runs a slice's plan on its own threads. engines/datafusion is written in Rust on Apache DataFusion; engines/null is written in C, runs nothing, and is used for tests.

Three written contracts sit between them, so that a new engine can be built from them alone:

• include/vec_engine.h: the function table, the rules for threads, signals, errors and memory, and the capabilities an engine declares (VecCaps).
• docs/ir.md: the plan IR: nodes, expressions, types, and the PostgreSQL functions an engine must compute with PostgreSQL's semantics.
• docs/wire.md: the Arrow IPC streams between engine slices, including cdbhash routing, on which both ends must agree byte for byte.

Data passes between the host and the engine through the Arrow C Data Interface. Because the engine is replaceable, DataFusion is not a fixed choice; a C++ engine such as Velox could be added later.

For PAX tables there is an optional direct reader that reads PAX files column by column on the engine's threads, without forming tuples. It is built by patching a copy of the PAX sources; neither the source tree nor the installed pax.so is changed.

Current status

What runs in the engine:

• Scans of heap, AO row, AO column and PAX tables.
• Joins: hash joins (inner, outer, semi, anti, NOT IN), which spill; merge joins, run as hash joins; nested loop joins.
• Aggregates: plain, hashed and two-stage, DISTINCT aggregates, grouping sets, and the common aggregate functions.
• Sort, LIMIT, window functions, UNION ALL / INTERSECT / EXCEPT, and correlated aggregate subqueries.
• Types: integers, floats, numeric, bool, date and time types, text / varchar / char(n), with PostgreSQL's semantics, including the same errors and SQLSTATEs, the same numeric scales and rounding, and short-circuit AND / OR.

A slice runs in the engine only when everything in it can; otherwise the whole slice stays on PostgreSQL, and EXPLAIN says why.

Coverage (TPC-H and TPC-DS at SF1, both planners):

• All 44 TPC-H runs use the engine; in 41 of them every slice that computes anything runs in the engine. The rest are the top slices of init plans.
• All 206 TPC-DS runs use the engine; in 170 of them every slice that computes anything runs in the engine. The rest are mostly top slices of init plans, and a few slices that only pass rows on and are not worth handing over.

Tests: 24 regression test files, each query run with the extension on and off and the results compared; and an engine-level check that runs the same tests once with the DataFusion engine and once with the null engine and requires the same results.

Size, without comments, blank lines and tests: about 14,500 lines of C in the host and about 12,400 lines of Rust in the DataFusion engine.

Preliminary results

These are basic tests to check whether the approach is viable, not a full performance evaluation.

TPC-H SF100: one host, 3 primaries, PAX tables, compared with the same cluster without the extension. Each query ran 5 times; the table gives the sum of the 22 medians, warm cache.

Planner	Cloudberry	vec_executor	Speedup
Postgres planner	1089 s	116 s	9.4x
GPORCA	1101 s	173 s	6.4x

• All 22 queries are faster, and every configuration returned the same output for every query.
• Method and per-query figures: docs/benchmarks.md.
• Note: the host has 20 hardware threads. With 3 primaries, Cloudberry's executor uses 3 cores, while vec_executor runs several threads per segment. The comparison therefore favours vec_executor and should not be read as the performance gap between the two.

The bottleneck is now on the storage side. Profiling shows that most of the CPU time of scan-heavy queries goes into reading PAX data: about half for Q1 and about three quarters for Q19. That time is spent mainly on zstd decompression and on converting PAX's variable-length PostgreSQL numeric values into fixed-width decimals. Further work on the execution side gains little. A storage layer that offered a format and interface suited to vectorized reads (for example fixed-width numbers, lighter compression, and an API for reading columns in batches) would allow clear further gains. This concerns PAX itself, is beyond what an extension can do, and is one of the topics I would like to discuss.

Resource management and operations

To check that the engine stays under Cloudberry's control, I ran these basic tests on a single host:

• Resource group CPU limits (group-v2): effective. All engine threads are in the resource group's cgroup. In a group with cpu_max_percent=10 (2 cores) a query used exactly 2.0 cores, and its time went from 1.9 s to 7.3 s.
• Resource group memory quota: effective. memory_quota sets the engine's memory budget: a large quota avoids spilling, a small one spills, as without the extension. All of the engine's memory is counted in vmem. At the vmem limit, Cloudberry's runaway cleaner cancels queries with a clear error and nothing crashes.
• Cancellation and failures: clean. On query cancel, statement_timeout, terminating the connection, killing a QE, and killing the coordinator's backend, the engine stops within 0.1 to 0.4 s; threads, temporary files and CPU are released, and the cluster stays available.
• Off at any time. vec.mode is off by default and can be turned on per session or per database. When off, plans are neither rewritten nor executed by the engine.

Known differences and limits

I compared results with the extension on and off for about 200 queries on heap and PAX tables, under both planners. The problems this found have been fixed; the remaining differences are listed in the README. The main ones:

• Floating-point sum / avg can differ in the last digits, because values are added in a different order.
• With LIMIT and no ORDER BY, the engine may evaluate rows past the limit, so a query can fail where PostgreSQL does not.
• A WHERE condition that can fail may be skipped on rows a cheaper condition of the same WHERE already rejects, so a query that fails on PostgreSQL may return rows.
• EXPLAIN ANALYZE counts rows for a slice's top node only.
• Text sorts run in the engine only under a C collation.

Limits:

• Cloudberry's workfile limits do not count the engine's spill files. With gp_workfile_limit_per_query set, a native query over the limit fails, but the engine's spilling is not limited. This is the most visible operational gap today.
• For the same data, DataFusion's hash tables are larger than PostgreSQL's, so with a large quota the engine uses more of the memory it is allowed (about 3.9 GB against 3.1 GB in one test, within the budget).
• It depends on a fork of DataFusion that adds apache/datafusion#26264 (the final merge of a spilled sort leaves memory for other operators), which is under review upstream. Once it is merged and released, the extension will move back to the official release.
• Building the extension needs a Rust toolchain. The kernel's build is not affected.

Rollout/Adoption Plan

The extension is built and installed separately and is off by default, so it does not affect existing users.

Next steps on my side:

• More tests: single-host comparisons with more segments, so that Cloudberry's executor can use the whole machine; larger data (SF1000) and a full TPC-DS performance run; a multi-node cluster with mirrors, where Motions cross a real network and dozens of segments run together; concurrent workloads, long-running stability, and failures such as a full disk or running out of file handles.
• Follow fix: leave memory for others when a spilled sort merges its runs datafusion#26264 upstream and return to an official DataFusion release.
• Adjust the direction according to the feedback here.

Questions for the community:

1. Direction: what do you think of adding vectorized execution as an extension? Is it worth continuing?
2. Rust dependency: is a Rust toolchain acceptable for a separately built extension? If it were ever considered for an Apache project, what would building, releasing and license review require?
3. Kernel interfaces: the extension uses general executor hooks today. Would more explicit kernel interfaces be worthwhile, for example:
   • a stable PAX API for reading columns, so that external engines need not patch the PAX sources;
   • a way for an extension's spill files to count toward the workfile limits and appear in the gp_toolkit views?
4. Testing: if there is interest, how could the regression tests join the community's CI?

Links:

• Code: https://github.com/MisterRaindrop/cloudberry-datafusion
• Benchmarks: docs/benchmarks.md
• Implementation details and coverage: docs/internals.md
• Engine interfaces: docs/engines.md, docs/ir.md, docs/wire.md

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

[MisterRaindrop]

@leborchuk @Alena0704 @igor-suhorukov Would appreciate your thoughts on this when you have time.

@igor-suhorukov, pg_vexec (#2065) takes a different route to the same goal; I'd be glad to compare notes.

[tuhaihe]

Excellent proposal. We can advance this work through community-driven efforts. We need more thoughts on it.

[leborchuk]

Thank you, all looks promising!

Yes, I keep coming up with the idea of writing a vectorized engine, but I don't dare to start work here.

Thank you for sharing the result, it's very interesting, and I think something really interesting can be done here.

What kind of approach has developed in my mind. I may be wrong, because I haven't started experimenting or writing code yet.

We have a seq scan that can return data in arrow format - iceberg and pax. Then, instead of immediately converting them to PG format, we leave them in arrow and use vector operations to perform further transformations. We do this until we encounter an unsupported operation or the second statement does not contain data in PG format.

Visually, we sort of start performing vector-style operations from the very bottom of the execution plan and gradually move up, leaving the data in arrow format for as long as possible.

volcano - style is not a problem here, because it's more about transferring control between nodes. Yes, we start with root and move recursively through branches to get results, but this does not prevent Clickhouse from building a very efficient engine.

I really like velox. I would like to try this engine. Rust scares me a lot, to be honest. How I would like to use it is to learn how to perform individual execution plan operations using velox.

First of all, these are hash join, group by, and data transfer in arrow format between slices. Replacing not individual slices, but individual operations. Because I want more control over the engine and I want to be able to add my own operations or improve the implementation of the engine to meet the requirements of Cloudberry.

[yjhjstz]

I support this proposal. I believe this feature could help drive Cloudberry adoption by addressing a practical customer need. It would also help close a potential gap with Apache Doris, so customers evaluating both products would not see a significant disadvantage in this area.