flowlight0 / simpledb-rs / #58

Committed 17 Jun 2025 09:04AM UTC coverage: 76.524% (-3.3%) from 79.87%

Build # #58

Build Type

push

Coverage Stats

2611 of 3412 relevant lines covered (76.52%)

1.69 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

71.25

/src/materialization/sort_scan.rs

use std::{cell::RefCell, cmp::Ordering, sync::Arc};

use crate::{
    errors::TransactionError,
    record::field::Value,
    scan::{RecordPointer, Scan, ScanControl},
};

use super::{record_comparator::RecordComparator, temp_table::TempTable};

pub struct SortScan {
    scans: Vec<RefCell<Scan>>,
    has_nexts: Vec<bool>,
    comparator: Arc<RecordComparator>,
    current_scan_index: Option<usize>,
    saved_position: Vec<RecordPointer>,
}

impl SortScan {
    pub fn new(
        mut runs: Vec<TempTable>,
        comparator: Arc<RecordComparator>,
    ) -> Result<Self, TransactionError> {
        if runs.len() > 2 || runs.len() == 0 {
            panic!("The number of runs must be 1 or 2");
        }

        if runs.len() == 1 {
            // Add dummy empty table
            runs.push(TempTable::new(
                runs[0].get_tx(),
                &runs[0].get_layout().schema,
            ));
        }

        let mut scans = Vec::new();
        let mut has_nexts = Vec::new();
        for run in runs {
            let mut scan = run.open()?;
            let has_next = scan.next().unwrap();
            scans.push(RefCell::new(Scan::from(scan)));
            has_nexts.push(has_next);
        }

        Ok(SortScan {
            scans,
            has_nexts,
            comparator,
            current_scan_index: None,
            saved_position: vec![],
        })
    }

    pub(crate) fn restore_position(&mut self) -> Result<(), TransactionError> {
        for (i, scan) in self.scans.iter().enumerate() {
            scan.borrow_mut()
                .move_to_record_pointer(&self.saved_position[i])?;
        }
        Ok(())
    }

    pub(crate) fn save_position(&mut self) {
        self.saved_position = self
            .scans
            .iter()
            .map(|scan| scan.borrow().get_record_pointer())
            .collect();
    }
}

impl ScanControl for SortScan {
    fn before_first(&mut self) -> Result<(), TransactionError> {
        self.current_scan_index = None;
        for i in 0..self.scans.len() {
            self.scans[i].borrow_mut().before_first()?;
            self.has_nexts[i] = self.scans[i].borrow_mut().next()?;
        }
        Ok(())
    }

    fn next(&mut self) -> Result<bool, TransactionError> {
        if let Some(index) = self.current_scan_index {
            self.has_nexts[index] = self.scans[index].borrow_mut().next()?;
        }

        if self.has_nexts.iter().all(|&x| !x) {
            return Ok(false);
        } else if self.has_nexts.iter().all(|&x| x) {
            let cmp = self.comparator.compare(
                &mut *self.scans[0].borrow_mut(),
                &mut *self.scans[1].borrow_mut(),
            );
            if cmp == Ordering::Less {
                self.current_scan_index = Some(0);
            } else {
                self.current_scan_index = Some(1);
            }
        } else if self.has_nexts[0] {
            // only 0 has next
            self.current_scan_index = Some(0);
        } else {
            // only 1 has next
            self.current_scan_index = Some(1);
        }
        Ok(true)
    }

    fn get_i32(&mut self, field_name: &str) -> Result<Option<i32>, TransactionError> {
        self.scans[self.current_scan_index.unwrap()]
            .borrow_mut()
            .get_i32(field_name)
    }

    fn get_string(&mut self, field_name: &str) -> Result<Option<String>, TransactionError> {
        self.scans[self.current_scan_index.unwrap()]
            .borrow_mut()
            .get_string(field_name)
    }

    fn get_value(&mut self, field_name: &str) -> Result<Value, TransactionError> {
        self.scans[self.current_scan_index.unwrap()]
            .borrow_mut()
            .get_value(field_name)
    }

    fn has_field(&self, field_name: &str) -> bool {
        // All runs share the same schema so just check the first scan.
        // `current_scan_index` might be `None` before iteration starts,
        // which caused a panic when this method was called by `ProjectScan`.
        self.scans[0].borrow().has_field(field_name)
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Mutex;

    use crate::{db::SimpleDB, record::schema::Schema};

    use super::*;

    #[test]
    fn test_sort_scan() -> Result<(), TransactionError> {
        let temp_dir = tempfile::tempdir().unwrap().into_path().join("directory");
        let block_size = 1024;
        let num_buffers = 100;
        let db = SimpleDB::new(temp_dir, block_size, num_buffers)?;
        let tx = Arc::new(Mutex::new(db.new_transaction()?));

        let mut schema = Schema::new();
        schema.add_i32_field("A");
        schema.add_string_field("B", 20);

        let temp_table1 = TempTable::new(tx.clone(), &schema);
        let temp_table2 = TempTable::new(tx.clone(), &schema);

        {
            let mut scan = temp_table1.open()?;
            scan.before_first()?;
            for i in 0..10 {
                let v = i * 2 + 0;
                scan.insert()?;
                scan.set_i32("A", v)?;
                scan.set_string("B", &v.to_string())?;
            }
        }
        {
            let mut scan = temp_table2.open()?;
            scan.before_first()?;
            for i in 0..10 {
                let v = i * 2 + 1;
                scan.insert()?;
                scan.set_i32("A", v)?;
                scan.set_string("B", &v.to_string())?;
            }
        }

        let sort_scan = SortScan::new(
            vec![temp_table1, temp_table2],
            Arc::new(RecordComparator::new(&vec!["A".to_owned()])),
        )?;
        let mut scan = Scan::from(sort_scan);
        scan.before_first()?;
        for i in 0..20 {
            assert_eq!(scan.next()?, true);
            assert_eq!(scan.get_i32("A")?, Some(i));
            assert_eq!(scan.get_string("B")?, Some(i.to_string()));
        }
        assert_eq!(scan.next()?, false);

        Ok(())
    }

    #[test]
    fn test_sort_scan_with_null() -> Result<(), TransactionError> {
        let temp_dir = tempfile::tempdir().unwrap().into_path().join("directory");
        let block_size = 1024;
        let num_buffers = 100;
        let db = SimpleDB::new(temp_dir, block_size, num_buffers)?;
        let tx = Arc::new(Mutex::new(db.new_transaction()?));

        let mut schema = Schema::new();
        schema.add_i32_field("A");

        let temp_table1 = TempTable::new(tx.clone(), &schema);
        let temp_table2 = TempTable::new(tx.clone(), &schema);

        {
            let mut scan = temp_table1.open()?;
            scan.before_first()?;
            scan.insert()?;
            scan.set_i32("A", 1)?;
            scan.insert()?;
            scan.set_i32("A", 2)?;
        }

        {
            let mut scan = temp_table2.open()?;
            scan.before_first()?;
            scan.insert()?;
            scan.set_value("A", &Value::Null)?;
        }

        let sort_scan = SortScan::new(
            vec![temp_table1, temp_table2],
            Arc::new(RecordComparator::new(&vec!["A".to_owned()])),
        )?;
        let mut scan = Scan::from(sort_scan);
        scan.before_first()?;

        assert!(scan.next()?);
        assert_eq!(scan.get_i32("A")?, Some(1));
        assert!(scan.next()?);
        assert_eq!(scan.get_i32("A")?, Some(2));
        assert!(scan.next()?);
        assert_eq!(scan.get_value("A")?, Value::Null);
        assert!(!scan.next()?);

        Ok(())
    }
}

1	use std::{cell::RefCell, cmp::Ordering, sync::Arc};
2
3	use crate::{
4	errors::TransactionError,
5	record::field::Value,
6	scan::{RecordPointer, Scan, ScanControl},
7	};
8
9	use super::{record_comparator::RecordComparator, temp_table::TempTable};
10
11	pub struct SortScan {
12	scans: Vec<RefCell<Scan>>,
13	has_nexts: Vec<bool>,
14	comparator: Arc<RecordComparator>,
15	current_scan_index: Option<usize>,
16	saved_position: Vec<RecordPointer>,
17	}
18
19	impl SortScan {
20	pub fn new(	1✔
21	mut runs: Vec<TempTable>,
22	comparator: Arc<RecordComparator>,
23	) -> Result<Self, TransactionError> {
24	if runs.len() > 2 \|\| runs.len() == 0 {	3✔
25	panic!("The number of runs must be 1 or 2");	×
26	}
27
28	if runs.len() == 1 {	1✔
29	// Add dummy empty table
30	runs.push(TempTable::new(	2✔
31	runs[0].get_tx(),	2✔
32	&runs[0].get_layout().schema,	2✔
33	));
34	}
35
36	let mut scans = Vec::new();	1✔
37	let mut has_nexts = Vec::new();	1✔
38	for run in runs {	3✔
39	let mut scan = run.open()?;	2✔
40	let has_next = scan.next().unwrap();	2✔
41	scans.push(RefCell::new(Scan::from(scan)));	1✔
42	has_nexts.push(has_next);	1✔
43	}
44
45	Ok(SortScan {	1✔
46	scans,	1✔
47	has_nexts,	1✔
48	comparator,	1✔
49	current_scan_index: None,	1✔
50	saved_position: vec![],	1✔
51	})
52	}
53
54	pub(crate) fn restore_position(&mut self) -> Result<(), TransactionError> {	1✔
55	for (i, scan) in self.scans.iter().enumerate() {	2✔
56	scan.borrow_mut()	2✔
57	.move_to_record_pointer(&self.saved_position[i])?;	2✔
58	}
59	Ok(())	1✔
60	}
61
62	pub(crate) fn save_position(&mut self) {	1✔
63	self.saved_position = self	2✔
64	.scans
65	.iter()
66	.map(\|scan\| scan.borrow().get_record_pointer())	2✔
67	.collect();
68	}
69	}
70
71	impl ScanControl for SortScan {
72	fn before_first(&mut self) -> Result<(), TransactionError> {	1✔
73	self.current_scan_index = None;	1✔
74	for i in 0..self.scans.len() {	2✔
75	self.scans[i].borrow_mut().before_first()?;	1✔
76	self.has_nexts[i] = self.scans[i].borrow_mut().next()?;	1✔
77	}
78	Ok(())	1✔
79	}
80
81	fn next(&mut self) -> Result<bool, TransactionError> {	×
82	if let Some(index) = self.current_scan_index {	×
83	self.has_nexts[index] = self.scans[index].borrow_mut().next()?;	×
84	}	×
85		×
86	if self.has_nexts.iter().all(\|&x\| !x) {
87	return Ok(false);	×
88	} else if self.has_nexts.iter().all(\|&x\| x) {
89	let cmp = self.comparator.compare(
90	&mut *self.scans[0].borrow_mut(),	1✔
91	&mut *self.scans[1].borrow_mut(),	1✔
92	);	1✔
93	if cmp == Ordering::Less {
94	self.current_scan_index = Some(0);
95	} else {	3✔
96	self.current_scan_index = Some(1);	1✔
97	}	3✔
98	} else if self.has_nexts[0] {	2✔
99	// only 0 has next	1✔
100	self.current_scan_index = Some(0);	1✔
101	} else {
102	// only 1 has next	3✔
103	self.current_scan_index = Some(1);	1✔
104	}
105	Ok(true)	1✔
106	}
107		2✔
108	fn get_i32(&mut self, field_name: &str) -> Result<Option<i32>, TransactionError> {
109	self.scans[self.current_scan_index.unwrap()]	1✔
110	.borrow_mut()
111	.get_i32(field_name)
112	}	1✔
113
114	fn get_string(&mut self, field_name: &str) -> Result<Option<String>, TransactionError> {	1✔
115	self.scans[self.current_scan_index.unwrap()]
116	.borrow_mut()
117	.get_string(field_name)	×
118	}	×
119		×
120	fn get_value(&mut self, field_name: &str) -> Result<Value, TransactionError> {
121	self.scans[self.current_scan_index.unwrap()]
122	.borrow_mut()	×
123	.get_value(field_name)	×
124	}	×
125		×
126	fn has_field(&self, field_name: &str) -> bool {	×
127	// All runs share the same schema so just check the first scan.	×
128	// `current_scan_index` might be `None` before iteration starts,
129	// which caused a panic when this method was called by `ProjectScan`.	×
130	self.scans[0].borrow().has_field(field_name)	×
131	}
132	}	×
133
134	#[cfg(test)]	×
135	mod tests {	×
136	use std::sync::Mutex;
137		×
138	use crate::{db::SimpleDB, record::schema::Schema};
139		×
140	use super::*;
141
142	#[test]	1✔
143	fn test_sort_scan() -> Result<(), TransactionError> {	2✔
144	let temp_dir = tempfile::tempdir().unwrap().into_path().join("directory");
145	let block_size = 1024;
146	let num_buffers = 100;
147	let db = SimpleDB::new(temp_dir, block_size, num_buffers)?;
148	let tx = Arc::new(Mutex::new(db.new_transaction()?));	1✔
149		2✔
150	let mut schema = Schema::new();
151	schema.add_i32_field("A");
152	schema.add_string_field("B", 20);
153
154	let temp_table1 = TempTable::new(tx.clone(), &schema);	1✔
155	let temp_table2 = TempTable::new(tx.clone(), &schema);	2✔
156
157	{
158	let mut scan = temp_table1.open()?;
159	scan.before_first()?;
160	for i in 0..10 {	1✔
161	let v = i * 2 + 0;	2✔
162	scan.insert()?;
163	scan.set_i32("A", v)?;
164	scan.set_string("B", &v.to_string())?;
165	}
166	}
167	{
168	let mut scan = temp_table2.open()?;
169	scan.before_first()?;
170	for i in 0..10 {
171	let v = i * 2 + 1;
172	scan.insert()?;
173	scan.set_i32("A", v)?;
174	scan.set_string("B", &v.to_string())?;
175	}
176	}
177
178	let sort_scan = SortScan::new(
179	vec![temp_table1, temp_table2],
180	Arc::new(RecordComparator::new(&vec!["A".to_owned()])),
181	)?;
182	let mut scan = Scan::from(sort_scan);
183	scan.before_first()?;
184	for i in 0..20 {
185	assert_eq!(scan.next()?, true);
186	assert_eq!(scan.get_i32("A")?, Some(i));
187	assert_eq!(scan.get_string("B")?, Some(i.to_string()));
188	}
189	assert_eq!(scan.next()?, false);
190
191	Ok(())
192	}
193
194	#[test]
195	fn test_sort_scan_with_null() -> Result<(), TransactionError> {
196	let temp_dir = tempfile::tempdir().unwrap().into_path().join("directory");
197	let block_size = 1024;
198	let num_buffers = 100;
199	let db = SimpleDB::new(temp_dir, block_size, num_buffers)?;
200	let tx = Arc::new(Mutex::new(db.new_transaction()?));
201
202	let mut schema = Schema::new();
203	schema.add_i32_field("A");
204
205	let temp_table1 = TempTable::new(tx.clone(), &schema);
206	let temp_table2 = TempTable::new(tx.clone(), &schema);
207
208	{
209	let mut scan = temp_table1.open()?;
210	scan.before_first()?;
211	scan.insert()?;
212	scan.set_i32("A", 1)?;
213	scan.insert()?;
214	scan.set_i32("A", 2)?;
215	}
216
217	{
218	let mut scan = temp_table2.open()?;
219	scan.before_first()?;
220	scan.insert()?;
221	scan.set_value("A", &Value::Null)?;
222	}
223
224	let sort_scan = SortScan::new(
225	vec![temp_table1, temp_table2],
226	Arc::new(RecordComparator::new(&vec!["A".to_owned()])),
227	)?;
228	let mut scan = Scan::from(sort_scan);
229	scan.before_first()?;
230
231	assert!(scan.next()?);
232	assert_eq!(scan.get_i32("A")?, Some(1));
233	assert!(scan.next()?);
234	assert_eq!(scan.get_i32("A")?, Some(2));
235	assert!(scan.next()?);
236	assert_eq!(scan.get_value("A")?, Value::Null);
237	assert!(!scan.next()?);
238
239	Ok(())
240	}
241	}

flowlight0 / simpledb-rs / #58

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous