grafeo_core/execution/parallel/
morsel.rs

1//! Morsel type for parallel execution units.
2//!
3//! A morsel represents a chunk of work (rows) to be processed by a worker thread.
4//! Morsels are larger than DataChunks (64K vs 2K rows) to amortize scheduling overhead.
5
6use grafeo_common::memory::buffer::PressureLevel;
7
8/// Default morsel size (64K rows).
9///
10/// This is larger than the typical DataChunk size to amortize scheduling overhead
11/// while still providing enough parallelism opportunities.
12pub const DEFAULT_MORSEL_SIZE: usize = 65536;
13
14/// Minimum morsel size under memory pressure.
15pub const MIN_MORSEL_SIZE: usize = 1024;
16
17/// Morsel size under moderate memory pressure.
18pub const MODERATE_PRESSURE_MORSEL_SIZE: usize = 32768;
19
20/// Morsel size under high memory pressure.
21pub const HIGH_PRESSURE_MORSEL_SIZE: usize = 16384;
22
23/// Morsel size under critical memory pressure.
24pub const CRITICAL_PRESSURE_MORSEL_SIZE: usize = MIN_MORSEL_SIZE;
25
26/// A morsel represents a unit of work for parallel execution.
27///
28/// Each morsel identifies a range of rows from a source to be processed
29/// by a single worker thread.
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
31pub struct Morsel {
32    /// Unique identifier for this morsel within a pipeline execution.
33    pub id: usize,
34    /// Source partition identifier (for multi-source queries).
35    pub source_id: usize,
36    /// Starting row index (inclusive).
37    pub start_row: usize,
38    /// Ending row index (exclusive).
39    pub end_row: usize,
40}
41
42impl Morsel {
43    /// Creates a new morsel.
44    #[must_use]
45    pub fn new(id: usize, source_id: usize, start_row: usize, end_row: usize) -> Self {
46        Self {
47            id,
48            source_id,
49            start_row,
50            end_row,
51        }
52    }
53
54    /// Returns the number of rows in this morsel.
55    #[must_use]
56    pub fn row_count(&self) -> usize {
57        self.end_row.saturating_sub(self.start_row)
58    }
59
60    /// Returns whether this morsel is empty.
61    #[must_use]
62    pub fn is_empty(&self) -> bool {
63        self.row_count() == 0
64    }
65
66    /// Splits this morsel into two at the given row offset.
67    ///
68    /// Returns `None` if the split point is outside the morsel range.
69    #[must_use]
70    pub fn split_at(&self, offset: usize) -> Option<(Morsel, Morsel)> {
71        let split_row = self.start_row + offset;
72        if split_row <= self.start_row || split_row >= self.end_row {
73            return None;
74        }
75
76        let first = Morsel {
77            id: self.id,
78            source_id: self.source_id,
79            start_row: self.start_row,
80            end_row: split_row,
81        };
82
83        let second = Morsel {
84            id: self.id + 1, // New ID for split morsel
85            source_id: self.source_id,
86            start_row: split_row,
87            end_row: self.end_row,
88        };
89
90        Some((first, second))
91    }
92}
93
94/// Computes the optimal morsel size based on memory pressure.
95///
96/// Under memory pressure, smaller morsels allow more fine-grained
97/// control over memory usage and enable earlier spilling.
98#[must_use]
99pub fn compute_morsel_size(pressure_level: PressureLevel) -> usize {
100    match pressure_level {
101        PressureLevel::Normal => DEFAULT_MORSEL_SIZE,
102        PressureLevel::Moderate => MODERATE_PRESSURE_MORSEL_SIZE,
103        PressureLevel::High => HIGH_PRESSURE_MORSEL_SIZE,
104        PressureLevel::Critical => CRITICAL_PRESSURE_MORSEL_SIZE,
105    }
106}
107
108/// Computes the optimal morsel size with a custom base size.
109#[must_use]
110pub fn compute_morsel_size_with_base(base_size: usize, pressure_level: PressureLevel) -> usize {
111    let factor = match pressure_level {
112        PressureLevel::Normal => 1.0,
113        PressureLevel::Moderate => 0.5,
114        PressureLevel::High => 0.25,
115        PressureLevel::Critical => MIN_MORSEL_SIZE as f64 / base_size as f64,
116    };
117
118    ((base_size as f64 * factor) as usize).max(MIN_MORSEL_SIZE)
119}
120
121/// Generates morsels for a given total row count.
122///
123/// Returns a vector of morsels that together cover all rows.
124#[must_use]
125pub fn generate_morsels(total_rows: usize, morsel_size: usize, source_id: usize) -> Vec<Morsel> {
126    if total_rows == 0 || morsel_size == 0 {
127        return Vec::new();
128    }
129
130    let num_morsels = (total_rows + morsel_size - 1) / morsel_size;
131    let mut morsels = Vec::with_capacity(num_morsels);
132
133    for (id, start) in (0..total_rows).step_by(morsel_size).enumerate() {
134        let end = (start + morsel_size).min(total_rows);
135        morsels.push(Morsel::new(id, source_id, start, end));
136    }
137
138    morsels
139}
140
141/// Generates morsels with adaptive sizing based on memory pressure.
142#[must_use]
143pub fn generate_adaptive_morsels(
144    total_rows: usize,
145    pressure_level: PressureLevel,
146    source_id: usize,
147) -> Vec<Morsel> {
148    let morsel_size = compute_morsel_size(pressure_level);
149    generate_morsels(total_rows, morsel_size, source_id)
150}
151
152#[cfg(test)]
153mod tests {
154    use super::*;
155
156    #[test]
157    fn test_morsel_creation() {
158        let morsel = Morsel::new(0, 1, 0, 1000);
159        assert_eq!(morsel.id, 0);
160        assert_eq!(morsel.source_id, 1);
161        assert_eq!(morsel.start_row, 0);
162        assert_eq!(morsel.end_row, 1000);
163        assert_eq!(morsel.row_count(), 1000);
164        assert!(!morsel.is_empty());
165    }
166
167    #[test]
168    fn test_morsel_empty() {
169        let morsel = Morsel::new(0, 0, 100, 100);
170        assert!(morsel.is_empty());
171        assert_eq!(morsel.row_count(), 0);
172    }
173
174    #[test]
175    fn test_morsel_split() {
176        let morsel = Morsel::new(0, 0, 0, 1000);
177
178        // Valid split
179        let (first, second) = morsel.split_at(400).unwrap();
180        assert_eq!(first.start_row, 0);
181        assert_eq!(first.end_row, 400);
182        assert_eq!(second.start_row, 400);
183        assert_eq!(second.end_row, 1000);
184
185        // Invalid splits
186        assert!(morsel.split_at(0).is_none());
187        assert!(morsel.split_at(1000).is_none());
188        assert!(morsel.split_at(1500).is_none());
189    }
190
191    #[test]
192    fn test_compute_morsel_size() {
193        assert_eq!(
194            compute_morsel_size(PressureLevel::Normal),
195            DEFAULT_MORSEL_SIZE
196        );
197        assert_eq!(
198            compute_morsel_size(PressureLevel::Moderate),
199            MODERATE_PRESSURE_MORSEL_SIZE
200        );
201        assert_eq!(
202            compute_morsel_size(PressureLevel::High),
203            HIGH_PRESSURE_MORSEL_SIZE
204        );
205        assert_eq!(
206            compute_morsel_size(PressureLevel::Critical),
207            CRITICAL_PRESSURE_MORSEL_SIZE
208        );
209    }
210
211    #[test]
212    fn test_compute_morsel_size_with_base() {
213        let base = 10000;
214
215        assert_eq!(
216            compute_morsel_size_with_base(base, PressureLevel::Normal),
217            10000
218        );
219        assert_eq!(
220            compute_morsel_size_with_base(base, PressureLevel::Moderate),
221            5000
222        );
223        assert_eq!(
224            compute_morsel_size_with_base(base, PressureLevel::High),
225            2500
226        );
227        assert_eq!(
228            compute_morsel_size_with_base(base, PressureLevel::Critical),
229            MIN_MORSEL_SIZE
230        );
231    }
232
233    #[test]
234    fn test_generate_morsels() {
235        let morsels = generate_morsels(1000, 300, 0);
236
237        assert_eq!(morsels.len(), 4);
238        assert_eq!(morsels[0].start_row, 0);
239        assert_eq!(morsels[0].end_row, 300);
240        assert_eq!(morsels[1].start_row, 300);
241        assert_eq!(morsels[1].end_row, 600);
242        assert_eq!(morsels[2].start_row, 600);
243        assert_eq!(morsels[2].end_row, 900);
244        assert_eq!(morsels[3].start_row, 900);
245        assert_eq!(morsels[3].end_row, 1000);
246    }
247
248    #[test]
249    fn test_generate_morsels_empty() {
250        assert!(generate_morsels(0, 100, 0).is_empty());
251        assert!(generate_morsels(100, 0, 0).is_empty());
252    }
253
254    #[test]
255    fn test_generate_morsels_exact_fit() {
256        let morsels = generate_morsels(1000, 250, 0);
257
258        assert_eq!(morsels.len(), 4);
259        for (i, morsel) in morsels.iter().enumerate() {
260            assert_eq!(morsel.row_count(), 250);
261            assert_eq!(morsel.id, i);
262        }
263    }
264
265    #[test]
266    fn test_generate_adaptive_morsels() {
267        let total = 100000;
268
269        let normal_morsels = generate_adaptive_morsels(total, PressureLevel::Normal, 0);
270        let high_morsels = generate_adaptive_morsels(total, PressureLevel::High, 0);
271
272        // More morsels under pressure (smaller size)
273        assert!(high_morsels.len() > normal_morsels.len());
274    }
275}
grafeo_core/execution/parallel/morsel.rs

grafeo_core/execution/parallel/
morsel.rs
