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        _ => CRITICAL_PRESSURE_MORSEL_SIZE,
106    }
107}
108
109/// Computes the optimal morsel size with a custom base size.
110#[must_use]
111pub fn compute_morsel_size_with_base(base_size: usize, pressure_level: PressureLevel) -> usize {
112    let factor = match pressure_level {
113        PressureLevel::Normal => 1.0,
114        PressureLevel::Moderate => 0.5,
115        PressureLevel::High => 0.25,
116        PressureLevel::Critical => MIN_MORSEL_SIZE as f64 / base_size as f64,
117        _ => MIN_MORSEL_SIZE as f64 / base_size as f64,
118    };
119
120    ((base_size as f64 * factor) as usize).max(MIN_MORSEL_SIZE)
121}
122
123/// Generates morsels for a given total row count.
124///
125/// Returns a vector of morsels that together cover all rows.
126#[must_use]
127pub fn generate_morsels(total_rows: usize, morsel_size: usize, source_id: usize) -> Vec<Morsel> {
128    if total_rows == 0 || morsel_size == 0 {
129        return Vec::new();
130    }
131
132    let num_morsels = (total_rows + morsel_size - 1) / morsel_size;
133    let mut morsels = Vec::with_capacity(num_morsels);
134
135    for (id, start) in (0..total_rows).step_by(morsel_size).enumerate() {
136        let end = (start + morsel_size).min(total_rows);
137        morsels.push(Morsel::new(id, source_id, start, end));
138    }
139
140    morsels
141}
142
143/// Generates morsels with adaptive sizing based on memory pressure.
144#[must_use]
145pub fn generate_adaptive_morsels(
146    total_rows: usize,
147    pressure_level: PressureLevel,
148    source_id: usize,
149) -> Vec<Morsel> {
150    let morsel_size = compute_morsel_size(pressure_level);
151    generate_morsels(total_rows, morsel_size, source_id)
152}
153
154#[cfg(test)]
155mod tests {
156    use super::*;
157
158    #[test]
159    fn test_morsel_creation() {
160        let morsel = Morsel::new(0, 1, 0, 1000);
161        assert_eq!(morsel.id, 0);
162        assert_eq!(morsel.source_id, 1);
163        assert_eq!(morsel.start_row, 0);
164        assert_eq!(morsel.end_row, 1000);
165        assert_eq!(morsel.row_count(), 1000);
166        assert!(!morsel.is_empty());
167    }
168
169    #[test]
170    fn test_morsel_empty() {
171        let morsel = Morsel::new(0, 0, 100, 100);
172        assert!(morsel.is_empty());
173        assert_eq!(morsel.row_count(), 0);
174    }
175
176    #[test]
177    fn test_morsel_split() {
178        let morsel = Morsel::new(0, 0, 0, 1000);
179
180        // Valid split
181        let (first, second) = morsel.split_at(400).unwrap();
182        assert_eq!(first.start_row, 0);
183        assert_eq!(first.end_row, 400);
184        assert_eq!(second.start_row, 400);
185        assert_eq!(second.end_row, 1000);
186
187        // Invalid splits
188        assert!(morsel.split_at(0).is_none());
189        assert!(morsel.split_at(1000).is_none());
190        assert!(morsel.split_at(1500).is_none());
191    }
192
193    #[test]
194    fn test_compute_morsel_size() {
195        assert_eq!(
196            compute_morsel_size(PressureLevel::Normal),
197            DEFAULT_MORSEL_SIZE
198        );
199        assert_eq!(
200            compute_morsel_size(PressureLevel::Moderate),
201            MODERATE_PRESSURE_MORSEL_SIZE
202        );
203        assert_eq!(
204            compute_morsel_size(PressureLevel::High),
205            HIGH_PRESSURE_MORSEL_SIZE
206        );
207        assert_eq!(
208            compute_morsel_size(PressureLevel::Critical),
209            CRITICAL_PRESSURE_MORSEL_SIZE
210        );
211    }
212
213    #[test]
214    fn test_compute_morsel_size_with_base() {
215        let base = 10000;
216
217        assert_eq!(
218            compute_morsel_size_with_base(base, PressureLevel::Normal),
219            10000
220        );
221        assert_eq!(
222            compute_morsel_size_with_base(base, PressureLevel::Moderate),
223            5000
224        );
225        assert_eq!(
226            compute_morsel_size_with_base(base, PressureLevel::High),
227            2500
228        );
229        assert_eq!(
230            compute_morsel_size_with_base(base, PressureLevel::Critical),
231            MIN_MORSEL_SIZE
232        );
233    }
234
235    #[test]
236    fn test_generate_morsels() {
237        let morsels = generate_morsels(1000, 300, 0);
238
239        assert_eq!(morsels.len(), 4);
240        assert_eq!(morsels[0].start_row, 0);
241        assert_eq!(morsels[0].end_row, 300);
242        assert_eq!(morsels[1].start_row, 300);
243        assert_eq!(morsels[1].end_row, 600);
244        assert_eq!(morsels[2].start_row, 600);
245        assert_eq!(morsels[2].end_row, 900);
246        assert_eq!(morsels[3].start_row, 900);
247        assert_eq!(morsels[3].end_row, 1000);
248    }
249
250    #[test]
251    fn test_generate_morsels_empty() {
252        assert!(generate_morsels(0, 100, 0).is_empty());
253        assert!(generate_morsels(100, 0, 0).is_empty());
254    }
255
256    #[test]
257    fn test_generate_morsels_exact_fit() {
258        let morsels = generate_morsels(1000, 250, 0);
259
260        assert_eq!(morsels.len(), 4);
261        for (i, morsel) in morsels.iter().enumerate() {
262            assert_eq!(morsel.row_count(), 250);
263            assert_eq!(morsel.id, i);
264        }
265    }
266
267    #[test]
268    fn test_generate_adaptive_morsels() {
269        let total = 100000;
270
271        let normal_morsels = generate_adaptive_morsels(total, PressureLevel::Normal, 0);
272        let high_morsels = generate_adaptive_morsels(total, PressureLevel::High, 0);
273
274        // More morsels under pressure (smaller size)
275        assert!(high_morsels.len() > normal_morsels.len());
276    }
277}
grafeo_core/execution/parallel/morsel.rs

grafeo_core/execution/parallel/
morsel.rs
