GitHub user MisterRaindrop reopened a discussion: [Proposal] Exploring vectorized execution for Cloudberry as a pure extension (vec_executor)
### Proposers [@MisterRaindrop](https://github.com/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](https://github.com/apache/cloudberry/discussions/2065) describes [pg_vexec](https://github.com/igor-suhorukov/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 `ereport`s. - **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`](https://github.com/MisterRaindrop/cloudberry-datafusion/blob/main/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`](https://github.com/MisterRaindrop/cloudberry-datafusion/blob/main/docs/ir.md): the plan IR: nodes, expressions, types, and the PostgreSQL functions an engine must compute with PostgreSQL's semantics. - [`docs/wire.md`](https://github.com/MisterRaindrop/cloudberry-datafusion/blob/main/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`](https://github.com/MisterRaindrop/cloudberry-datafusion/blob/main/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](https://github.com/apache/datafusion/pull/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 apache/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`](https://github.com/MisterRaindrop/cloudberry-datafusion/blob/main/docs/benchmarks.md) - Implementation details and coverage: [`docs/internals.md`](https://github.com/MisterRaindrop/cloudberry-datafusion/blob/main/docs/internals.md) - Engine interfaces: [`docs/engines.md`](https://github.com/MisterRaindrop/cloudberry-datafusion/blob/main/docs/engines.md), [`docs/ir.md`](https://github.com/MisterRaindrop/cloudberry-datafusion/blob/main/docs/ir.md), [`docs/wire.md`](https://github.com/MisterRaindrop/cloudberry-datafusion/blob/main/docs/wire.md) ### Are you willing to submit a PR? - [X] Yes I am willing to submit a PR! GitHub link: https://github.com/apache/cloudberry/discussions/2267 ---- This is an automatically sent email for [email protected]. To unsubscribe, please send an email to: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
