Introduce morsel-driven Parquet scan by Dandandan · Pull Request #20481 · apache/datafusion

Dandandan · 2026-02-22T17:48:59Z

Which issue does this PR close?

Closes Introduce morsel-driven Parquet scan #20529

Rationale for this change

Current parelllization of Parquet scan is bounded by the thread that has the most data / is the slowest to execute, which means in the case of data skew (driven by either larger partitions or less selective filters during pruning / filter pushdown..., variable object store latency), the parallelism will be significantly limited.

We can change the strategy by morsel-driven parallelism like described in https://db.in.tum.de/~leis/papers/morsels.pdf.

Doing so is faster for a lot of queries, when there is an amount of skew (such as clickbench) and we have enough row filters to spread out the work.
For clickbench_partitioned / clickbench_pushdown it seems up to ~2x as fast for some queries, on a 10 core machine.

It seems to have almost no regressions, perhaps some due to different file scanning order(?) - so different statistics that can be used to prune and thus some minor variation.

Morsel-Driven Execution Architecture (partly claude-generated)

This branch implements a morsel-driven execution model for Parquet scans, based on the concept
from the Morsel-Driven Parallelism paper (Leis et al.). The core idea: instead of statically
assigning files to partitions, all work is pooled in a shared queue that all partition streams pull
from dynamically.

The Problem It Solves

In the traditional model, partition 0 might get a 1 GB file while partition 1 gets nothing --
partition 1 idles while 0 is busy. Currently we already try to statically spread out work to n partitions / threads based on stats (which works very well on perfectly distributed scans on SSDs (e.g. TPCH running locally), this doesn't work well when there is any data skew caused by any of those:

filters being more selective on part of the data
high variation in object store response times
Morsel-driven execution prevents this by sharing work dynamically.

Key Types

ParquetMorsel -- datafusion/datasource-parquet/src/opener.rs:129

A morsel = one row group of a Parquet file. Stored as an extension on PartitionedFile.
```
pub struct ParquetMorsel {
    pub metadata: Arc<ParquetMetaData>,   // cached, shared across morsels from same file
    pub access_plan: ParquetAccessPlan,   // which row groups to read
}
```


  **`WorkQueue`** -- `datafusion/datasource/src/file_stream.rs:410`

  The shared, thread-safe queue. Each partition stream calls `pull()` which returns:

  - `Work(file)` -- here's a file/morsel to process
  - `Wait` -- queue is empty but workers are still morselizing (wait for notification)
  - `Done` -- all work consumed

  **`MorselState`** -- `datafusion/datasource/src/source.rs:240`

  Tracks the shared queue lifecycle. A new queue is created once per execution cycle when all
  partition streams have opened.

  ```rust
  struct MorselState {
      queue: Option<Arc<WorkQueue>>,
      streams_opened: usize,
      expected_streams: usize,
  }

MorselizingGuard -- datafusion/datasource/src/file_stream.rs:49

RAII wrapper that atomically decrements morselizing_count when a worker finishes -- enabling
WorkStatus::Wait vs Done decisions.

`FileOpener` Trait Extension -- `datafusion/datasource/src/file_stream.rs:498`

A new morselize() method is added to FileOpener. The default implementation is a no-op
(returns the file as-is). ParquetOpener overrides it to split files by row group.

pub trait FileOpener {
    fn open(&self, file: PartitionedFile) -> Result<FileOpenFuture>;

    // NEW: split a file into morsels (default = no-op)
    fn morselize(&self, file: PartitionedFile)
        -> BoxFuture<'static, Result<Vec<PartitionedFile>>> { ... }
}

ParquetOpener::morselize() at opener.rs:232:

Loads Parquet metadata once (shared via Arc across all resulting morsels)
Applies file-level and row-group-level statistics pruning
Returns one PartitionedFile per surviving row group, each carrying a ParquetMorsel extension

`FileStream` State Machine -- `datafusion/datasource/src/file_stream.rs:141`

The morsel-driven path adds two new states (Morselizing and Waiting):

  Idle
   +-- morsel_driven=true  -> queue.pull()
   |    +-- Work(file) -> Morselizing (run morselize())
   |    +-- Wait       -> Waiting (yield, wake on notify)
   |    +-- Done       -> return None
   +-- morsel_driven=false -> traditional local file_iter

  Morselizing
   +-- morsels > 1 -> push_many() back to queue -> Idle
   +-- morsel = 1  -> Open

  Waiting
   +-- notified -> Idle

Configuration

datafusion.execution.parquet.allow_morsel_driven -- datafusion/common/src/config.rs:748

Default: true. Can be disabled per-session.

FileScanConfig::morsel_driven -- datafusion/datasource/src/file_scan_config.rs:211

Automatically disabled when:

partitioned_by_file_group = true (breaks hash-partitioning guarantees)
preserve_order = true (breaks SortPreservingMerge guarantees)

Benchmark results

Details


--------------------
Benchmark clickbench_extended.json
--------------------
┏━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━┓
┃ Query    ┃        main ┃ parquet-morsel-driven-execution-237164415184908839 ┃          Change ┃
┡━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━┩
│ QQuery 0 │   888.03 ms │                                         1005.51 ms │    1.13x slower │
│ QQuery 1 │   298.74 ms │                                          253.19 ms │   +1.18x faster │
│ QQuery 2 │   669.67 ms │                                          577.42 ms │   +1.16x faster │
│ QQuery 3 │   500.88 ms │                                          415.63 ms │   +1.21x faster │
│ QQuery 4 │   750.81 ms │                                          815.64 ms │    1.09x slower │
│ QQuery 5 │ 10765.06 ms │                                        10392.77 ms │       no change │
│ QQuery 6 │  1468.42 ms │                                            2.07 ms │ +710.78x faster │
│ QQuery 7 │  1146.72 ms │                                         1072.55 ms │   +1.07x faster │
└──────────┴─────────────┴────────────────────────────────────────────────────┴─────────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary                                                 ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (main)                                                 │ 16488.33ms │
│ Total Time (parquet-morsel-driven-execution-237164415184908839)   │ 14534.77ms │
│ Average Time (main)                                               │  2061.04ms │
│ Average Time (parquet-morsel-driven-execution-237164415184908839) │  1816.85ms │
│ Queries Faster                                                    │          5 │
│ Queries Slower                                                    │          2 │
│ Queries with No Change                                            │          1 │
│ Queries with Failure                                              │          0 │
└───────────────────────────────────────────────────────────────────┴────────────┘
--------------------
Benchmark clickbench_partitioned.json
--------------------
┏━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query     ┃        main ┃ parquet-morsel-driven-execution-237164415184908839 ┃        Change ┃
┡━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 0  │     1.04 ms │                                            1.04 ms │     no change │
│ QQuery 1  │    14.08 ms │                                           16.24 ms │  1.15x slower │
│ QQuery 2  │    48.33 ms │                                           46.40 ms │     no change │
│ QQuery 3  │    47.06 ms │                                           45.26 ms │     no change │
│ QQuery 4  │   366.76 ms │                                          369.05 ms │     no change │
│ QQuery 5  │   535.43 ms │                                          450.32 ms │ +1.19x faster │
│ QQuery 6  │     2.96 ms │                                            4.96 ms │  1.68x slower │
│ QQuery 7  │    15.86 ms │                                           17.48 ms │  1.10x slower │
│ QQuery 8  │   488.23 ms │                                          482.34 ms │     no change │
│ QQuery 9  │   692.94 ms │                                          631.48 ms │ +1.10x faster │
│ QQuery 10 │   115.64 ms │                                          109.67 ms │ +1.05x faster │
│ QQuery 11 │   133.16 ms │                                          123.68 ms │ +1.08x faster │
│ QQuery 12 │   508.25 ms │                                          430.19 ms │ +1.18x faster │
│ QQuery 13 │   680.40 ms │                                          615.79 ms │ +1.10x faster │
│ QQuery 14 │   481.06 ms │                                          430.01 ms │ +1.12x faster │
│ QQuery 15 │   432.32 ms │                                          436.09 ms │     no change │
│ QQuery 16 │   990.45 ms │                                         1005.22 ms │     no change │
│ QQuery 17 │   962.77 ms │                                          978.92 ms │     no change │
│ QQuery 18 │  2050.47 ms │                                         2095.02 ms │     no change │
│ QQuery 19 │    40.32 ms │                                           37.51 ms │ +1.08x faster │
│ QQuery 20 │   933.54 ms │                                          726.31 ms │ +1.29x faster │
│ QQuery 21 │  1027.66 ms │                                          858.70 ms │ +1.20x faster │
│ QQuery 22 │  1530.58 ms │                                         1478.12 ms │     no change │
│ QQuery 23 │  4663.56 ms │                                         4815.10 ms │     no change │
│ QQuery 24 │    93.25 ms │                                           52.85 ms │ +1.76x faster │
│ QQuery 25 │   203.85 ms │                                          165.61 ms │ +1.23x faster │
│ QQuery 26 │    94.02 ms │                                           54.23 ms │ +1.73x faster │
│ QQuery 27 │  1126.73 ms │                                          964.02 ms │ +1.17x faster │
│ QQuery 28 │ 10180.43 ms │                                         9036.37 ms │ +1.13x faster │
│ QQuery 29 │   364.42 ms │                                          320.42 ms │ +1.14x faster │
│ QQuery 30 │   474.22 ms │                                          425.94 ms │ +1.11x faster │
│ QQuery 31 │   458.37 ms │                                          414.97 ms │ +1.10x faster │
│ QQuery 32 │  2584.13 ms │                                         2360.70 ms │ +1.09x faster │
│ QQuery 33 │  2233.45 ms │                                         2202.39 ms │     no change │
│ QQuery 34 │  2229.80 ms │                                         2521.80 ms │  1.13x slower │
│ QQuery 35 │   706.55 ms │                                          696.50 ms │     no change │
│ QQuery 36 │   106.00 ms │                                           59.85 ms │ +1.77x faster │
│ QQuery 37 │    43.51 ms │                                           32.26 ms │ +1.35x faster │
│ QQuery 38 │    74.87 ms │                                           38.75 ms │ +1.93x faster │
│ QQuery 39 │   199.26 ms │                                          119.58 ms │ +1.67x faster │
│ QQuery 40 │    18.67 ms │                                           11.94 ms │ +1.56x faster │
│ QQuery 41 │    16.78 ms │                                           11.80 ms │ +1.42x faster │
│ QQuery 42 │    13.52 ms │                                            9.78 ms │ +1.38x faster │
└───────────┴─────────────┴────────────────────────────────────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary                                                 ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (main)                                                 │ 37984.70ms │
│ Total Time (parquet-morsel-driven-execution-237164415184908839)   │ 35704.65ms │
│ Average Time (main)                                               │   883.37ms │
│ Average Time (parquet-morsel-driven-execution-237164415184908839) │   830.34ms │
│ Queries Faster                                                    │         26 │
│ Queries Slower                                                    │          4 │
│ Queries with No Change                                            │         13 │
│ Queries with Failure                                              │          0 │
└───────────────────────────────────────────────────────────────────┴────────────┘
--------------------
Benchmark clickbench_pushdown.json
--------------------
┏━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query     ┃        main ┃ parquet-morsel-driven-execution-237164415184908839 ┃        Change ┃
┡━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 0  │     0.94 ms │                                            0.89 ms │     no change │
│ QQuery 1  │    15.46 ms │                                           17.03 ms │  1.10x slower │
│ QQuery 2  │    47.74 ms │                                           47.21 ms │     no change │
│ QQuery 3  │    47.91 ms │                                           46.42 ms │     no change │
│ QQuery 4  │   377.94 ms │                                          380.81 ms │     no change │
│ QQuery 5  │   531.75 ms │                                          463.25 ms │ +1.15x faster │
│ QQuery 6  │     5.65 ms │                                            5.99 ms │  1.06x slower │
│ QQuery 7  │    21.40 ms │                                           21.74 ms │     no change │
│ QQuery 8  │   485.04 ms │                                          489.23 ms │     no change │
│ QQuery 9  │   685.29 ms │                                          628.46 ms │ +1.09x faster │
│ QQuery 10 │   153.55 ms │                                          143.76 ms │ +1.07x faster │
│ QQuery 11 │   167.04 ms │                                          158.46 ms │ +1.05x faster │
│ QQuery 12 │   540.31 ms │                                          463.87 ms │ +1.16x faster │
│ QQuery 13 │   689.88 ms │                                          668.03 ms │     no change │
│ QQuery 14 │   525.73 ms │                                          458.53 ms │ +1.15x faster │
│ QQuery 15 │   451.15 ms │                                          429.07 ms │     no change │
│ QQuery 16 │  1024.55 ms │                                          972.48 ms │ +1.05x faster │
│ QQuery 17 │   992.24 ms │                                          950.12 ms │     no change │
│ QQuery 18 │  1998.18 ms │                                         2153.69 ms │  1.08x slower │
│ QQuery 19 │    44.15 ms │                                           40.87 ms │ +1.08x faster │
│ QQuery 20 │   944.79 ms │                                          717.18 ms │ +1.32x faster │
│ QQuery 21 │   977.02 ms │                                          843.59 ms │ +1.16x faster │
│ QQuery 22 │  1514.11 ms │                                         1449.66 ms │     no change │
│ QQuery 23 │   381.34 ms │                                          170.80 ms │ +2.23x faster │
│ QQuery 24 │   113.30 ms │                                           51.19 ms │ +2.21x faster │
│ QQuery 25 │   262.47 ms │                                          223.74 ms │ +1.17x faster │
│ QQuery 26 │   161.43 ms │                                           74.77 ms │ +2.16x faster │
│ QQuery 27 │  1236.38 ms │                                         1045.83 ms │ +1.18x faster │
│ QQuery 28 │ 10319.06 ms │                                         9246.37 ms │ +1.12x faster │
│ QQuery 29 │   358.66 ms │                                          314.35 ms │ +1.14x faster │
│ QQuery 30 │   473.48 ms │                                          427.07 ms │ +1.11x faster │
│ QQuery 31 │   455.18 ms │                                          424.80 ms │ +1.07x faster │
│ QQuery 32 │  2087.10 ms │                                         2558.84 ms │  1.23x slower │
│ QQuery 33 │  2146.63 ms │                                         2317.84 ms │  1.08x slower │
│ QQuery 34 │  2292.89 ms │                                         2766.27 ms │  1.21x slower │
│ QQuery 35 │   718.07 ms │                                          706.21 ms │     no change │
│ QQuery 36 │    91.43 ms │                                           61.60 ms │ +1.48x faster │
│ QQuery 37 │    51.60 ms │                                           37.37 ms │ +1.38x faster │
│ QQuery 38 │    52.58 ms │                                           35.61 ms │ +1.48x faster │
│ QQuery 39 │   160.05 ms │                                          104.88 ms │ +1.53x faster │
│ QQuery 40 │    29.39 ms │                                           19.89 ms │ +1.48x faster │
│ QQuery 41 │    26.49 ms │                                           17.91 ms │ +1.48x faster │
│ QQuery 42 │    18.22 ms │                                           12.23 ms │ +1.49x faster │
└───────────┴─────────────┴────────────────────────────────────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary                                                 ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (main)                                                 │ 33677.56ms │
│ Total Time (parquet-morsel-driven-execution-237164415184908839)   │ 32167.88ms │
│ Average Time (main)                                               │   783.20ms │
│ Average Time (parquet-morsel-driven-execution-237164415184908839) │   748.09ms │
│ Queries Faster                                                    │         26 │
│ Queries Slower                                                    │          6 │
│ Queries with No Change                                            │         11 │
│ Queries with Failure                                              │          0 │
└───────────────────────────────────────────────────────────────────┴────────────┘
--------------------
Benchmark tpch_sf1.json
--------------------
┏━━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┓
┃ Query     ┃      main ┃ parquet-morsel-driven-execution-237164415184908839 ┃    Change ┃
┡━━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━┩
│ QQuery 1  │  77.57 ms │                                           77.57 ms │ no change │
│ QQuery 2  │  41.27 ms │                                           41.13 ms │ no change │
│ QQuery 3  │  49.50 ms │                                           49.79 ms │ no change │
│ QQuery 4  │  36.79 ms │                                           36.42 ms │ no change │
│ QQuery 5  │  64.85 ms │                                           66.16 ms │ no change │
│ QQuery 6  │  24.49 ms │                                           24.84 ms │ no change │
│ QQuery 7  │  74.28 ms │                                           74.31 ms │ no change │
│ QQuery 8  │  69.07 ms │                                           70.09 ms │ no change │
│ QQuery 9  │  89.52 ms │                                           89.43 ms │ no change │
│ QQuery 10 │  84.54 ms │                                           84.90 ms │ no change │
│ QQuery 11 │  26.99 ms │                                           26.79 ms │ no change │
│ QQuery 12 │  51.52 ms │                                           52.86 ms │ no change │
│ QQuery 13 │ 119.33 ms │                                          120.11 ms │ no change │
│ QQuery 14 │  35.97 ms │                                           36.81 ms │ no change │
│ QQuery 15 │  40.64 ms │                                           41.74 ms │ no change │
│ QQuery 16 │  28.08 ms │                                           28.32 ms │ no change │
│ QQuery 17 │  97.26 ms │                                           98.76 ms │ no change │
│ QQuery 18 │ 117.66 ms │                                          119.44 ms │ no change │
│ QQuery 19 │  55.89 ms │                                           56.63 ms │ no change │
│ QQuery 20 │  52.98 ms │                                           52.69 ms │ no change │
│ QQuery 21 │  89.56 ms │                                           91.00 ms │ no change │
│ QQuery 22 │  18.42 ms │                                           18.60 ms │ no change │
└───────────┴───────────┴────────────────────────────────────────────────────┴───────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┓
┃ Benchmark Summary                                                 ┃           ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━┩
│ Total Time (main)                                                 │ 1346.17ms │
│ Total Time (parquet-morsel-driven-execution-237164415184908839)   │ 1358.38ms │
│ Average Time (main)                                               │   61.19ms │
│ Average Time (parquet-morsel-driven-execution-237164415184908839) │   61.74ms │
│ Queries Faster                                                    │         0 │
│ Queries Slower                                                    │         0 │
│ Queries with No Change                                            │        22 │
│ Queries with Failure                                              │         0 │
└───────────────────────────────────────────────────────────────────┴───────────┘
--------------------
Benchmark tpch_sf10.json
--------------------
┏━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━┓
┃ Query     ┃       main ┃ parquet-morsel-driven-execution-237164415184908839 ┃       Change ┃
┡━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━┩
│ QQuery 1  │  530.11 ms │                                          547.41 ms │    no change │
│ QQuery 2  │  120.45 ms │                                          121.96 ms │    no change │
│ QQuery 3  │  359.06 ms │                                          378.49 ms │ 1.05x slower │
│ QQuery 4  │  186.42 ms │                                          191.28 ms │    no change │
│ QQuery 5  │  529.28 ms │                                          546.25 ms │    no change │
│ QQuery 6  │  176.91 ms │                                          181.17 ms │    no change │
│ QQuery 7  │  715.92 ms │                                          738.24 ms │    no change │
│ QQuery 8  │  551.52 ms │                                          571.76 ms │    no change │
│ QQuery 9  │  840.61 ms │                                          854.68 ms │    no change │
│ QQuery 10 │  429.98 ms │                                          435.94 ms │    no change │
│ QQuery 11 │   96.37 ms │                                           97.49 ms │    no change │
│ QQuery 12 │  268.06 ms │                                          280.03 ms │    no change │
│ QQuery 13 │  352.76 ms │                                          396.81 ms │ 1.12x slower │
│ QQuery 14 │  230.51 ms │                                          240.39 ms │    no change │
│ QQuery 15 │  340.44 ms │                                          342.36 ms │    no change │
│ QQuery 16 │   79.96 ms │                                           81.67 ms │    no change │
│ QQuery 17 │  952.83 ms │                                          989.26 ms │    no change │
│ QQuery 18 │ 1097.49 ms │                                         1110.96 ms │    no change │
│ QQuery 19 │  368.82 ms │                                          367.51 ms │    no change │
│ QQuery 20 │  335.05 ms │                                          334.54 ms │    no change │
│ QQuery 21 │  973.47 ms │                                          983.11 ms │    no change │
│ QQuery 22 │   93.95 ms │                                          100.63 ms │ 1.07x slower │
└───────────┴────────────┴────────────────────────────────────────────────────┴──────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┓
┃ Benchmark Summary                                                 ┃           ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━┩
│ Total Time (main)                                                 │ 9629.97ms │
│ Total Time (parquet-morsel-driven-execution-237164415184908839)   │ 9891.95ms │
│ Average Time (main)                                               │  437.73ms │
│ Average Time (parquet-morsel-driven-execution-237164415184908839) │  449.63ms │
│ Queries Faster                                                    │         0 │
│ Queries Slower                                                    │         3 │
│ Queries with No Change                                            │        19 │
│ Queries with Failure                                              │         0 │
└───────────────────────────────────────────────────────────────────┴───────────┘

What changes are included in this PR?

Changes Parquet to create morsels when "opening" files based on rowgroups (this strategy could be changed, e.g. by splitting large rowgroups based on size / page index - this is TBD as current benchmarks probably don't benefit). The current design allows for this.

The morsels are shared in a workqueue, each partition will take some morsel from the queue, "morselize" (e.g. split into individual morsels) it or just process the data.

Are these changes tested?

Are there any user-facing changes?

Yes, FileStream::new got a new shared_queue parameter.

pub fn new(
        config: &FileScanConfig,
        partition: usize,
        file_opener: Arc<dyn FileOpener>,
        metrics: &ExecutionPlanMetricsSet,
        shared_queue: Option<Arc<WorkQueue>>,
    )
    
    ```


## Acknowledgements

I heavily used AI (Jules / Claude) for this PR, but reviewed

This PR implements morsel-driven execution for Parquet files in DataFusion, enabling row-group level work sharing across partitions to mitigate data skew. Key changes: - Introduced `WorkQueue` in `datafusion/datasource/src/file_stream.rs` for shared pool of work. - Added `morselize` method to `FileOpener` trait to allow dynamic splitting of files into morsels. - Implemented `morselize` for `ParquetOpener` to split files into individual row groups. - Cached `ParquetMetaData` in `ParquetMorsel` extensions to avoid redundant I/O. - Modified `FileStream` to support work stealing from the shared queue. - Implemented `Weak` pointer pattern for `WorkQueue` in `FileScanConfig` to support plan re-executability. - Added `MorselizingGuard` to ensure shared state consistency on cancellation. - Added `allow_morsel_driven` configuration option (enabled by default for Parquet). - Implemented row-group pruning during the morselization phase for better efficiency. Tests: - Added `parquet_morsel_driven_execution` test to verify work distribution and re-executability. - Added `parquet_morsel_driven_enabled_by_default` to verify the default configuration. Co-authored-by: Dandandan <163737+Dandandan@users.noreply.github.com>

…en-execution-237164415184908839

Dandandan · 2026-02-23T10:14:24Z

run benchmarks

alamb-ghbot · 2026-02-23T16:59:51Z

🤖: Benchmark completed

Details

Comparing HEAD and parquet-morsel-driven-execution-237164415184908839
--------------------
Benchmark clickbench_extended.json
--------------------

Dandandan · 2026-02-23T17:01:35Z

run benchmarks

alamb-ghbot · 2026-02-23T17:41:09Z

🤖 ./gh_compare_branch.sh gh_compare_branch.sh Running
Linux aal-dev 6.14.0-1018-gcp #19~24.04.1-Ubuntu SMP Wed Sep 24 23:23:09 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing parquet-morsel-driven-execution-237164415184908839 (d2df36b) to 9660c98 diff using: tpch_mem clickbench_partitioned clickbench_extended
Results will be posted here when complete

alamb-ghbot · 2026-02-23T18:19:44Z

🤖: Benchmark completed

Details

Comparing HEAD and parquet-morsel-driven-execution-237164415184908839
--------------------
Benchmark clickbench_extended.json
--------------------

Dandandan · 2026-02-23T18:22:54Z

🤖: Benchmark completed

Huh?

Dandandan · 2026-02-24T05:23:17Z

run benchmarks

Dandandan · 2026-02-24T06:43:49Z

Show benchmark queue

alamb-ghbot · 2026-02-24T06:43:57Z

🤖 Hi @Dandandan, you asked to view the benchmark queue (#20481 (comment)).

Job	User	Benchmarks	Comment
`20363_3947413716.sh`	adriangb	tpcds (env: DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS=true DATAFUSION_EXECUTION_PARQUET_REORDER_FILTERS=true)	`https://github.com/apache/datafusion/pull/20363#issuecomment-3947413716`
`20481_3949158428.sh`	Dandandan	default	`https://github.com/apache/datafusion/pull/20481#issuecomment-3949158428`

This reverts commit c859d6a.

This reverts commit 24b95fb.

Dandandan · 2026-02-24T16:23:35Z

@alamb this is now mostly ready
@adriangb @AdamGS you might be interested as well.

I will update the issue with my latest benchmark results (it's slightly better than in the issue description, especially for filter pushdown) and also run Q6 of clickbench_extended as a ~160x improvement seems a bit too good to be true

google-labs-jules bot and others added 2 commits February 22, 2026 13:12

Proto

a67f9ac

github-actions bot added core Core DataFusion crate common Related to common crate proto Related to proto crate datasource Changes to the datasource crate labels Feb 22, 2026

Dandandan mentioned this pull request Feb 22, 2026

PoC: introduce morsel-driven parquet scan #20477

Closed

Dandandan changed the title ~~PoC: Parquet morsel driven execution~~ PoC: introduce morsel-driven parquet scan Feb 22, 2026

Dandandan added 9 commits February 22, 2026 18:55

Proto

d0da5da

Fmt

32eec3c

Merge remote-tracking branch 'upstream/main' into parquet-morsel-driv…

5dc895c

…en-execution-237164415184908839

Proto

cc73788

Fix

d517b5d

Fix

de1606d

Clippy

950f6db

Refactor

7f57317

WIP

fd6d7fd

Dandandan changed the title ~~PoC: introduce morsel-driven parquet scan~~ Introduce morsel-driven parquet scan Feb 23, 2026

WIP

37126bf

Dandandan changed the title ~~Introduce morsel-driven parquet scan~~ Introduce morsel-driven Parquet scan Feb 23, 2026

Dandandan added 5 commits February 23, 2026 09:31

WIP

2d3c33e

WIP

98f0ea9

WIP

a389b02

Update

4065448

Update

415315d

Config

13b4977

github-actions bot added the documentation Improvements or additions to documentation label Feb 23, 2026

Dandandan added 2 commits February 23, 2026 12:53

Test

a30c3f8

Refactor

8b32ca8

Update test

d2df36b

Autofix

869b7d3

Dandandan added 3 commits February 23, 2026 19:24

Prune files

67ea9ab

Update test

e845675

Update test

6885981

github-actions bot added physical-expr Changes to the physical-expr crates sqllogictest SQL Logic Tests (.slt) labels Feb 23, 2026

Update morsel_driven

3384b8f

Dandandan force-pushed the parquet-morsel-driven-execution-237164415184908839 branch from 9453b05 to 3384b8f Compare February 23, 2026 20:04

Dandandan added 3 commits February 23, 2026 21:09

Update morsel_driven

211d4fc

fmt

2db61f1

move pruning

c859d6a

Dandandan added 7 commits February 24, 2026 08:08

Revert "move pruning"

24b95fb

This reverts commit c859d6a.

Reapply "move pruning"

80fa1ec

This reverts commit 24b95fb.

Autofix

1dcd401

Autofix

04b08a6

Autofix

9799b96

Autofix

de29e40

CLippy

aa27a43

Dandandan marked this pull request as ready for review February 24, 2026 16:10

Undo submodule

9a4aa84

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Comments

Introduce morsel-driven Parquet scan#20481

Introduce morsel-driven Parquet scan#20481
Dandandan wants to merge 38 commits intoapache:mainfrom
Dandandan:parquet-morsel-driven-execution-237164415184908839

Dandandan commented Feb 22, 2026 •

edited

Loading

Uh oh!

Dandandan commented Feb 23, 2026

Uh oh!

alamb-ghbot commented Feb 23, 2026

Uh oh!

Dandandan commented Feb 23, 2026

Uh oh!

alamb-ghbot commented Feb 23, 2026

Uh oh!

alamb-ghbot commented Feb 23, 2026

Uh oh!

Dandandan commented Feb 23, 2026

Uh oh!

Dandandan commented Feb 24, 2026

Uh oh!

Dandandan commented Feb 24, 2026

Uh oh!

alamb-ghbot commented Feb 24, 2026

Uh oh!

Dandandan commented Feb 24, 2026 •

edited

Loading

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Comments

Conversation

Dandandan commented Feb 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

Morsel-Driven Execution Architecture (partly claude-generated)

The Problem It Solves

Key Types

FileOpener Trait Extension -- datafusion/datasource/src/file_stream.rs:498

FileStream State Machine -- datafusion/datasource/src/file_stream.rs:141

Benchmark results

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

Dandandan commented Feb 23, 2026

Uh oh!

alamb-ghbot commented Feb 23, 2026

Uh oh!

Dandandan commented Feb 23, 2026

Uh oh!

alamb-ghbot commented Feb 23, 2026

Uh oh!

alamb-ghbot commented Feb 23, 2026

Uh oh!

Dandandan commented Feb 23, 2026

Uh oh!

Dandandan commented Feb 24, 2026

Uh oh!

Dandandan commented Feb 24, 2026

Uh oh!

alamb-ghbot commented Feb 24, 2026

Uh oh!

Dandandan commented Feb 24, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Dandandan commented Feb 22, 2026 •

edited

Loading

`FileOpener` Trait Extension -- `datafusion/datasource/src/file_stream.rs:498`

`FileStream` State Machine -- `datafusion/datasource/src/file_stream.rs:141`

Dandandan commented Feb 24, 2026 •

edited

Loading