use std::ops::Range;
use std::sync::Arc;
use itertools::Itertools;
use vortex_array::ArrayRef;
use vortex_array::Columnar;
use vortex_array::VortexSessionExecute;
use vortex_array::dtype::DType;
use vortex_array::dtype::Field;
use vortex_array::dtype::FieldMask;
use vortex_array::dtype::FieldPath;
use vortex_array::dtype::FieldPathSet;
use vortex_array::expr::Expression;
use vortex_array::expr::pruning::checked_pruning_expr;
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 vortex_utils::aliases::hash_map::HashMap;
use crate::FileStatistics;
use crate::footer::Footer;
use crate::pruning::extract_relevant_file_stats_as_struct_row;
use crate::v2::FileStatsLayoutReader;
#[derive(Clone)]
pub struct VortexFile {
pub(crate) footer: Footer,
pub(crate) segment_source: Arc<dyn SegmentSource>,
pub(crate) session: VortexSession,
}
impl VortexFile {
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 layout_reader(&self) -> VortexResult<Arc<dyn LayoutReader>> {
let segment_source = self.segment_source();
self.footer
.layout()
.new_reader("".into(), segment_source, &self.session)
}
pub fn data_source(&self) -> VortexResult<DataSourceRef> {
let mut reader = self.layout_reader()?;
if let Some(stats) = self.file_stats().cloned() {
reader = Arc::new(FileStatsLayoutReader::new(
reader,
stats,
self.session.clone(),
));
}
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);
};
let set = FieldPathSet::from_iter(
fields
.names()
.iter()
.zip(stats.stats_sets().iter())
.flat_map(|(name, stats)| {
stats.iter().map(|(stat, _)| {
FieldPath::from_iter([
Field::Name(name.clone()),
Field::Name(stat.name().into()),
])
})
}),
);
let Some((predicate, required_stats)) = checked_pruning_expr(filter, &set) else {
return Ok(false);
};
let required_file_stats = HashMap::from_iter(
required_stats
.map()
.iter()
.map(|(path, stats)| (path.clone(), stats.clone())),
);
let Some(file_stats) = extract_relevant_file_stats_as_struct_row(
&required_file_stats,
stats.stats_sets(),
fields,
)?
else {
return Ok(false);
};
let mut ctx = self.session.create_execution_ctx();
Ok(
match file_stats
.apply(&predicate)?
.execute::<Columnar>(&mut ctx)?
{
Columnar::Constant(s) => s.scalar().as_bool().value() == Some(true),
Columnar::Canonical(_) => false,
},
)
}
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())
}
}