16346697784

Committed 17 Jul 2025 01:39PM UTC coverage: 80.707% (+0.003%) from 80.704%

Build # 16346697784

Build Type

Pull #3901

github

Committed by

web-flow

Commit Message

Merge d01e093a5 into d53d06603

Pull Request Pull Request #3901: feat: duckdb global conversion cache

Run Details

40 of 56 new or added lines in 6 files covered. (71.43%)

5 existing lines in 3 files now uncovered.

41878 of 51889 relevant lines covered (80.71%)

157474.88 hits per line

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

85.94

/vortex-scan/src/multi_scan.rs

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright the Vortex contributors

use std::sync::Arc;

use crossbeam_queue::SegQueue;
use futures::executor::LocalPool;
use futures::future::BoxFuture;
use futures::stream::{FuturesUnordered, StreamExt};
use vortex_array::ArrayRef;
use vortex_error::VortexResult;

use crate::ScanBuilder;

type ArrayFuture<S> = BoxFuture<'static, VortexResult<Option<(ArrayRef, S)>>>;
type ScanBuilderFactory<S> =
    Arc<SegQueue<Box<(dyn FnOnce() -> ScanBuilder<(ArrayRef, S)> + Send + Sync)>>>;

/// Coordinator to orchestrate multiple scan operations.
///
/// `MultiScan` allows to queue multiple scan operations in order to execute
/// them in parallel. In particular, this enables scanning multiple files.
pub struct MultiScan<S> {
    scan_builder_factory: ScanBuilderFactory<S>,
}

impl<S> Default for MultiScan<S> {
    fn default() -> Self {
        Self::new()
    }
}

impl<S> MultiScan<S> {
    pub fn new() -> Self {
        Self {
            scan_builder_factory: Arc::new(SegQueue::new()),
        }
    }

    /// Add lazily constructed scan builders paired with their corresponding states.
    pub fn with_scan_builders<I, F>(self, closures: I) -> Self
    where
        F: FnOnce() -> ScanBuilder<(ArrayRef, S)> + 'static + Send + Sync,
        I: IntoIterator<Item = F>,
    {
        for closure in closures.into_iter() {
            self.scan_builder_factory.push(Box::new(closure));
        }

        self
    }

    /// Creates a new iterator to participate in the scan.
    ///
    /// The scan progresses when calling `next` on the iterator.
    pub fn new_scan_iterator(&self) -> MultiScanIterator<S> {
        MultiScanIterator {
            scan_builder_factory: self.scan_builder_factory.clone(),
            local_pool: LocalPool::new(),
            polled_tasks: FuturesUnordered::new(),
            task_queue: SegQueue::new(),
        }
    }
}

/// Scan iterator to participate in a `MultiScan`.
pub struct MultiScanIterator<S> {
    local_pool: LocalPool,
    polled_tasks: FuturesUnordered<ArrayFuture<S>>,

    /// Thread-safe queue of closures that lazily produce [`ScanBuilder`] instances.
    /// This queue is shared across all iterators being created with `new_scan_iterator`.
    scan_builder_factory: ScanBuilderFactory<S>,
    task_queue: SegQueue<ArrayFuture<S>>,
}

impl<S> MultiScanIterator<S> {
    fn pop_scan_task(&self) -> Option<VortexResult<ArrayFuture<S>>> {
        if let Some(task_with_state) = self.task_queue.pop() {
            return Some(Ok(task_with_state));
        }
        None
    }
}

impl<S: Send + Sync + 'static> Iterator for MultiScanIterator<S> {
    type Item = VortexResult<(ArrayRef, S)>;

    fn next(&mut self) -> Option<VortexResult<(ArrayRef, S)>> {
        loop {
            // Queue up tasks if the thread local queue is almost empty.
            if self.task_queue.len() <= 4 {
                if let Some(scan_builder_fn) = self.scan_builder_factory.pop() {
                    let split_tasks = scan_builder_fn().build().ok()?.1;
                    for task in split_tasks {
                        self.task_queue.push(Box::pin(task));
                    }
                }
                // TODO(Alex): worksteal tasks from other threads
            }

            if let Some(work_result) = self.pop_scan_task() {
                match work_result {
                    Ok(task) => {
                        self.polled_tasks.push(task);
                    }
                    Err(e) => return Some(Err(e)),
                }
            }

            if self.task_queue.is_empty() && self.polled_tasks.is_empty() {
                // All tasks have been fully processed.
                return None;
            }

            let result = self.local_pool.run_until(async {
                while let Some(result) = self.polled_tasks.next().await {
                    match result {
                        Ok(Some(array)) => return Some(Ok(array)),
                        Ok(None) => continue,
                        Err(e) => return Some(Err(e)),
                    }
                }
                None
            });

            match result {
                Some(Ok(array)) => {
                    return Some(Ok(array));
                }
                Some(Err(e)) => return Some(Err(e)),
                None => continue, // Try next batch of futures
            }
        }
    }
}

1	// SPDX-License-Identifier: Apache-2.0
2	// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4	use std::sync::Arc;
5
6	use crossbeam_queue::SegQueue;
7	use futures::executor::LocalPool;
8	use futures::future::BoxFuture;
9	use futures::stream::{FuturesUnordered, StreamExt};
10	use vortex_array::ArrayRef;
11	use vortex_error::VortexResult;
12
13	use crate::ScanBuilder;
14
15	type ArrayFuture<S> = BoxFuture<'static, VortexResult<Option<(ArrayRef, S)>>>;
16	type ScanBuilderFactory<S> =
17	Arc<SegQueue<Box<(dyn FnOnce() -> ScanBuilder<(ArrayRef, S)> + Send + Sync)>>>;
18
19	/// Coordinator to orchestrate multiple scan operations.
20	///
21	/// `MultiScan` allows to queue multiple scan operations in order to execute
22	/// them in parallel. In particular, this enables scanning multiple files.
23	pub struct MultiScan<S> {
24	scan_builder_factory: ScanBuilderFactory<S>,
25	}
26
27	impl<S> Default for MultiScan<S> {
NEW 28	fn default() -> Self {	×
NEW 29	Self::new()	×
NEW 30	}	×
31	}
32
33	impl<S> MultiScan<S> {
34	pub fn new() -> Self {	183✔
35	Self {	183✔
36	scan_builder_factory: Arc::new(SegQueue::new()),	183✔
37	}	183✔
38	}	183✔
39
40	/// Add lazily constructed scan builders paired with their corresponding states.
41	pub fn with_scan_builders<I, F>(self, closures: I) -> Self	183✔
42	where	183✔
43	F: FnOnce() -> ScanBuilder<(ArrayRef, S)> + 'static + Send + Sync,	183✔
44	I: IntoIterator<Item = F>,	183✔
45	{
46	for closure in closures.into_iter() {	184✔
47	self.scan_builder_factory.push(Box::new(closure));	184✔
48	}	184✔
49
50	self	183✔
51	}	183✔
52
53	/// Creates a new iterator to participate in the scan.
54	///
55	/// The scan progresses when calling `next` on the iterator.
56	pub fn new_scan_iterator(&self) -> MultiScanIterator<S> {	1,450✔
57	MultiScanIterator {	1,450✔
58	scan_builder_factory: self.scan_builder_factory.clone(),	1,450✔
59	local_pool: LocalPool::new(),	1,450✔
60	polled_tasks: FuturesUnordered::new(),	1,450✔
61	task_queue: SegQueue::new(),	1,450✔
62	}	1,450✔
63	}	1,450✔
64	}
65
66	/// Scan iterator to participate in a `MultiScan`.
67	pub struct MultiScanIterator<S> {
68	local_pool: LocalPool,
69	polled_tasks: FuturesUnordered<ArrayFuture<S>>,
70
71	/// Thread-safe queue of closures that lazily produce [`ScanBuilder`] instances.
72	/// This queue is shared across all iterators being created with `new_scan_iterator`.
73	scan_builder_factory: ScanBuilderFactory<S>,
74	task_queue: SegQueue<ArrayFuture<S>>,
75	}
76
77	impl<S> MultiScanIterator<S> {
78	fn pop_scan_task(&self) -> Option<VortexResult<ArrayFuture<S>>> {	2,034✔
79	if let Some(task_with_state) = self.task_queue.pop() {	2,034✔
80	return Some(Ok(task_with_state));	584✔
81	}	1,450✔
82	None	1,450✔
83	}	2,034✔
84	}
85
86	impl<S: Send + Sync + 'static> Iterator for MultiScanIterator<S> {
87	type Item = VortexResult<(ArrayRef, S)>;
88
89	fn next(&mut self) -> Option<VortexResult<(ArrayRef, S)>> {	2,034✔
90	loop {
91	// Queue up tasks if the thread local queue is almost empty.
92	if self.task_queue.len() <= 4 {	2,034✔
93	if let Some(scan_builder_fn) = self.scan_builder_factory.pop() {	1,846✔
94	let split_tasks = scan_builder_fn().build().ok()?.1;	184✔
95	for task in split_tasks {	768✔
96	self.task_queue.push(Box::pin(task));	584✔
97	}	584✔
98	}	1,662✔
99	// TODO(Alex): worksteal tasks from other threads
100	}	188✔
101
102	if let Some(work_result) = self.pop_scan_task() {	2,034✔
103	match work_result {	584✔
104	Ok(task) => {	584✔
105	self.polled_tasks.push(task);	584✔
106	}	584✔
UNCOV 107	Err(e) => return Some(Err(e)),	×
108	}
109	}	1,450✔
110
111	if self.task_queue.is_empty() && self.polled_tasks.is_empty() {	2,034✔
112	// All tasks have been fully processed.
113	return None;	1,450✔
114	}	584✔
115
116	let result = self.local_pool.run_until(async {	584✔
117	while let Some(result) = self.polled_tasks.next().await {	584✔
118	match result {	584✔
119	Ok(Some(array)) => return Some(Ok(array)),	584✔
120	Ok(None) => continue,	×
121	Err(e) => return Some(Err(e)),	×
122	}
123	}
124	None	×
125	});	584✔
126
127	match result {	584✔
128	Some(Ok(array)) => {	584✔
129	return Some(Ok(array));	584✔
130	}
131	Some(Err(e)) => return Some(Err(e)),	×
132	None => continue, // Try next batch of futures	×
133	}
134	}
135	}	2,034✔
136	}

vortex-data / vortex / 16346697784

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