use std::ops::Range;
use std::sync::Arc;
use std::sync::OnceLock;
use itertools::Itertools;
use vortex_array::ArrayRef;
use vortex_array::dtype::DType;
use vortex_array::dtype::FieldMask;
use vortex_array::expr::Expression;
use vortex_error::VortexResult;
use vortex_layout::LayoutReader;
use vortex_layout::scan::layout::LayoutReaderDataSource;
use vortex_layout::scan::scan_builder::ScanBuilder;
use vortex_layout::scan::split_by::SplitBy;
use vortex_layout::segments::SegmentSource;
use vortex_scan::DataSourceRef;
use vortex_session::VortexSession;
use crate::FileStatistics;
use crate::footer::Footer;
use crate::pruning::can_prune_file_stats;
use crate::v2::FileStatsLayoutReader;
#[derive(Clone)]
pub struct VortexFile {
footer: Footer,
segment_source: Arc<dyn SegmentSource>,
session: VortexSession,
layout_reader_cache: Option<OnceLock<Arc<dyn LayoutReader>>>,
}
fn layout_reader(
segment_source: Arc<dyn SegmentSource>,
footer: &Footer,
session: &VortexSession,
) -> VortexResult<Arc<dyn LayoutReader>> {
let root_reader = footer
.layout()
.new_reader("".into(), segment_source, session, &Default::default())?;
Ok(if let Some(stats) = footer.statistics().cloned() {
Arc::new(FileStatsLayoutReader::new(
root_reader,
stats,
session.clone(),
))
} else {
root_reader
})
}
impl VortexFile {
pub fn new(
footer: Footer,
segment_source: Arc<dyn SegmentSource>,
session: VortexSession,
) -> Self {
Self {
footer,
segment_source,
session,
layout_reader_cache: None,
}
}
pub fn with_caching(self) -> Self {
Self {
footer: self.footer,
segment_source: self.segment_source,
session: self.session,
layout_reader_cache: Some(OnceLock::new()),
}
}
pub fn footer(&self) -> &Footer {
&self.footer
}
pub fn row_count(&self) -> u64 {
self.footer.row_count()
}
pub fn dtype(&self) -> &DType {
self.footer.dtype()
}
pub fn file_stats(&self) -> Option<&FileStatistics> {
self.footer.statistics()
}
pub fn segment_source(&self) -> Arc<dyn SegmentSource> {
Arc::clone(&self.segment_source)
}
pub fn session(&self) -> &VortexSession {
&self.session
}
pub fn layout_reader(&self) -> VortexResult<Arc<dyn LayoutReader>> {
match &self.layout_reader_cache {
None => layout_reader(
Arc::clone(&self.segment_source),
&self.footer,
&self.session,
),
Some(reader) => {
if let Some(val) = reader.get() {
Ok(Arc::clone(val))
} else {
let inner = layout_reader(
Arc::clone(&self.segment_source),
&self.footer,
&self.session,
)?;
Ok(if let Err(val) = reader.set(Arc::clone(&inner)) {
val
} else {
inner
})
}
}
}
}
pub fn data_source(&self) -> VortexResult<DataSourceRef> {
let reader = self.layout_reader()?;
Ok(Arc::new(LayoutReaderDataSource::new(
reader,
self.session.clone(),
)))
}
pub fn scan(&self) -> VortexResult<ScanBuilder<ArrayRef>> {
Ok(ScanBuilder::new(
self.session.clone(),
self.layout_reader()?,
))
}
pub fn can_prune(&self, filter: &Expression) -> VortexResult<bool> {
let Some((stats, fields)) = self
.footer
.statistics()
.zip(self.footer.dtype().as_struct_fields_opt())
else {
return Ok(false);
};
can_prune_file_stats(
filter,
self.footer.dtype(),
self.footer.row_count(),
stats,
fields,
&self.session,
)
}
pub fn splits(&self) -> VortexResult<Vec<Range<u64>>> {
let reader = self.layout_reader()?;
Ok(SplitBy::Layout
.splits(reader.as_ref(), &(0..reader.row_count()), &[FieldMask::All])?
.into_iter()
.tuple_windows()
.map(|(start, end)| start..end)
.collect())
}
}