diff --git a/vortex-array/src/scalar/typed_view/extension/mod.rs b/vortex-array/src/scalar/typed_view/extension/mod.rs index b9f524c7077..ba33c2ce91f 100644 --- a/vortex-array/src/scalar/typed_view/extension/mod.rs +++ b/vortex-array/src/scalar/typed_view/extension/mod.rs @@ -79,6 +79,11 @@ impl<'a> ExtScalar<'a> { .vortex_expect("ExtScalar is invalid") } + /// Returns a reference to the underlying value + pub fn value(&self) -> Option<&ScalarValue> { + self.value + } + /// Casts this scalar to the given `dtype`. pub(crate) fn cast(&self, target_dtype: &DType) -> VortexResult { if self.value.is_none() && !target_dtype.is_nullable() { diff --git a/vortex-duckdb/src/column_statistics.rs b/vortex-duckdb/src/column_statistics.rs index 2d6aeb979b8..9be4ec2fefa 100644 --- a/vortex-duckdb/src/column_statistics.rs +++ b/vortex-duckdb/src/column_statistics.rs @@ -24,17 +24,14 @@ pub struct ColumnStatistics { } impl ColumnStatistics { - pub fn try_from(stats: &ColumnStatisticsAggregate, dtype: DType) -> VortexResult { - let min = stats.min.as_ref().and_then(|value| { - Scalar::try_new(dtype.clone(), Some(value.clone())) + pub fn try_from(stats: ColumnStatisticsAggregate, dtype: DType) -> VortexResult { + let to_value = |value: ScalarValue| { + Scalar::try_new(dtype.clone(), Some(value)) .and_then(|scalar| scalar.try_to_duckdb_scalar()) .ok() - }); - let max = stats.max.as_ref().and_then(|value| { - Scalar::try_new(dtype.clone(), Some(value.clone())) - .and_then(|scalar| scalar.try_to_duckdb_scalar()) - .ok() - }); + }; + let min = stats.min.and_then(to_value); + let max = stats.max.and_then(to_value); let max_string_length = stats .max_string_length diff --git a/vortex-duckdb/src/convert/scalar.rs b/vortex-duckdb/src/convert/scalar.rs index 4f11f349253..cffd5ef4538 100644 --- a/vortex-duckdb/src/convert/scalar.rs +++ b/vortex-duckdb/src/convert/scalar.rs @@ -196,12 +196,9 @@ impl ToDuckDBScalar for ExtScalar<'_> { vortex_bail!("Cannot convert non-temporal extension scalar to duckdb value"); }; + let storage = PrimitiveScalar::try_new(self.ext_dtype().storage_dtype(), self.value())?; let value = || { - self.to_storage_scalar() - .as_primitive_opt() - .ok_or_else(|| { - vortex_err!("Cannot have a temporal time type not packed by a primitive scalar") - })? + storage .as_::() .ok_or_else(|| vortex_err!("temporal types must be convertible to i64")) }; @@ -227,19 +224,10 @@ impl ToDuckDBScalar for ExtScalar<'_> { } } TemporalMetadata::Date(unit) => match unit { - TimeUnit::Days => { - let days = self - .to_storage_scalar() - .as_primitive_opt() - .ok_or_else(|| { - vortex_err!("temporal types must be backed by primitive scalars") - })? - .as_::(); - match days { - Some(days) => Value::new_date(days), - None => Value::null(&*ext_logical_type(self)?), - } - } + TimeUnit::Days => match storage.as_::() { + Some(days) => Value::new_date(days), + None => Value::null(&*ext_logical_type(self)?), + }, _ => vortex_bail!("cannot have TimeUnit {unit}, so represent a day"), }, TemporalMetadata::Time(unit) => match unit { diff --git a/vortex-duckdb/src/file_reader.rs b/vortex-duckdb/src/file_reader.rs index ed62739e712..5048aa8a025 100644 --- a/vortex-duckdb/src/file_reader.rs +++ b/vortex-duckdb/src/file_reader.rs @@ -22,7 +22,6 @@ use vortex::file::v2::FileStatsLayoutReader; use vortex::io::compat::Compat; use vortex::io::filesystem::FileSystemRef; use vortex::io::object_store::ObjectStoreFileSystem; -use vortex::io::object_store::object_path_from_literal; use vortex::io::runtime::BlockingRuntime as _; use vortex::layout::LayoutReaderRef; use vortex::layout::scan::scan_builder::ScanBuilder; @@ -97,14 +96,6 @@ fn resolve_filesystem(url: &Url) -> VortexResult<(FileSystemRef, String)> { )) } -/// Same as resolve_filesystem but doesn't create filesystem object -fn resolve_path(url: &Url) -> VortexResult { - if url.scheme() == "file" { - return Ok(url.path().to_string()); - } - Ok(REGISTRY.resolve(url)?.1.to_string()) -} - pub struct OpenFileReader { pub reader: LayoutReaderRef, /// File splits stored in inverse order @@ -114,11 +105,10 @@ pub struct OpenFileReader { } impl OpenFileReader { - async fn open(file_path: String) -> VortexResult { - let url = parse_uri_or_path(&file_path)?; - let (fs, path) = resolve_filesystem(&url)?; - let file = fs.open_read(&path).await?; - let file = open_cached(&SESSION, file, &path, None, &|options| options).await?; + async fn open(path: String) -> VortexResult { + let (fs, fs_path) = resolve_filesystem(&parse_uri_or_path(&path)?)?; + let source = fs.open_read(&fs_path).await?; + let file = open_cached(&SESSION, Some(&path), source, None, &|options| options).await?; Ok(OpenFileReader { reader: file.layout_reader()?, cache: ConversionCache::default(), @@ -181,18 +171,14 @@ pub fn reader_initialize(file: &mut OpenFileReader, global: &GlobalState) -> Vor // Getting splits is non-trivial work so we prefer doing it here under file // lock and not in reader_try_initialize_scan under global lock. - let ordered = global.file_row_number_column_pos.is_some(); let reader = Arc::clone(&file.reader); let filter = &global.filter; - let mut builder = ScanBuilder::new(SESSION.clone(), reader) + let builder = ScanBuilder::new(SESSION.clone(), reader) .with_projection(global.projection.clone()) - .with_ordered(ordered) .with_some_filter(filter.filter.clone()) .with_selection(filter.row_selection.clone()); - if let Some(row_range) = filter.row_range.as_ref() { - builder = builder.with_row_range(row_range.clone()); - } - let mut splits = builder.build()?; + let scan = builder.prepare()?; + let mut splits = scan.execute(filter.row_range.clone())?; // threads take last element of file.splits so we need to reverse splits.reverse(); @@ -296,7 +282,7 @@ pub fn reader_get_statistics( let dtype = fields.field_by_index(index)?; let stats = ColumnStatisticsAggregate::new(stats_sets.get(index)?); - match ColumnStatistics::try_from(&stats, dtype) { + match ColumnStatistics::try_from(stats, dtype) { Ok(stats) => Some(stats), Err(e) => vortex_panic!(e), } @@ -328,10 +314,10 @@ pub fn can_get_partition_stats(bind: &BindState) -> bool { /// If any footer is not present, it sets a flag in BindState so we won't try /// again. pub fn footer_get_cached(bind: &mut BindState, path: &str) -> VortexResult> { - let url = parse_uri_or_path(path)?; - let path = resolve_path(&url)?; - let key = object_path_from_literal(&path).to_string(); - let footer = SESSION.get::().get_footer(&key); + let footer = SESSION + .get_opt::() + .vortex_expect("MultiFileSession not found") + .get_footer(path); bind.no_footer_caches |= footer.is_none(); Ok(footer) } @@ -345,7 +331,7 @@ pub fn footer_get_statistics(footer: &Footer, index: usize) -> Option Some(stats), Err(e) => vortex_panic!(e), } diff --git a/vortex-file/src/footer/file_statistics.rs b/vortex-file/src/footer/file_statistics.rs index 4fac3ad8482..62cdf199afd 100644 --- a/vortex-file/src/footer/file_statistics.rs +++ b/vortex-file/src/footer/file_statistics.rs @@ -13,12 +13,10 @@ use flatbuffers::WIPOffset; use itertools::Itertools; use vortex_array::dtype::DType; use vortex_array::stats::StatsSet; -use vortex_error::VortexExpect; use vortex_error::VortexResult; use vortex_error::vortex_ensure_eq; use vortex_flatbuffers::FlatBufferRoot; use vortex_flatbuffers::WriteFlatBuffer; -use vortex_flatbuffers::array::ArrayStats; use vortex_flatbuffers::footer as fb; use vortex_session::VortexSession; @@ -94,12 +92,11 @@ impl FileStatistics { session: &VortexSession, ) -> VortexResult { let field_stats = fb.field_stats().unwrap_or_default(); - let mut array_stats: Vec = field_stats.iter().collect(); if let DType::Struct(struct_fields, _) = file_dtype { - vortex_ensure_eq!(array_stats.len(), struct_fields.nfields()); + vortex_ensure_eq!(field_stats.len(), struct_fields.nfields()); - let stats_sets: Arc<[StatsSet]> = array_stats + let stats_sets: Arc<[StatsSet]> = field_stats .into_iter() .zip(struct_fields.fields()) .map(|(array_stat, field_dtype)| { @@ -114,11 +111,9 @@ impl FileStatistics { dtypes, }) } else { - vortex_ensure_eq!(array_stats.len(), 1); + vortex_ensure_eq!(field_stats.len(), 1); - let array_stat = array_stats - .pop() - .vortex_expect("we just checked that there was 1 field"); + let array_stat = field_stats.get(0); let stats_set = StatsSet::from_flatbuffer(&array_stat, file_dtype, session)?; Ok(Self { diff --git a/vortex-file/src/multi/mod.rs b/vortex-file/src/multi/mod.rs index b8f371d05b4..57af2974980 100644 --- a/vortex-file/src/multi/mod.rs +++ b/vortex-file/src/multi/mod.rs @@ -234,51 +234,44 @@ async fn open_file( tracing::trace!(path = %file.path, "opening vortex file"); let source = fs.open_read(&file.path).await?; - open_cached(session, source, &file.path, file.size, open_options_fn).await + let key = source.uri().is_none().then_some(file.path.as_str()); + open_cached(session, key, source, file.size, open_options_fn).await } -/// Open a single Vortex file through the session's footer cache, so that a later open of the -/// same file skips the footer read. +/// Open a Vortex file and cache its footer on the session. +/// Subsequent calls to this function will reuse the footer from cache. /// -/// The cache is keyed by the source's [`uri`](vortex_io::VortexReadAt::uri) where it reports one, -/// since that includes the full path (with any filesystem prefix) and so stays unique even when -/// different filesystems strip paths to the same relative name. `fallback_key` identifies the file -/// for sources that report no URI, and must be stable and unique within the session — two -/// different files sharing a key would read each other's footer. -/// -/// Caching the footer is independent of [`VortexOpenOptions::include_metadata`]: the footer holds -/// only metadata *locators*, and each open resolves the segments it was asked for. +/// "key" is the optional cache key provided by user. If it's not found, +/// source.uri() is probed. If there's no uri(), open_cached errors. pub async fn open_cached( session: &VortexSession, + mut key: Option<&str>, source: Arc, - fallback_key: &str, file_size: Option, open_options_fn: &(dyn Fn(VortexOpenOptions) -> VortexOpenOptions + Send + Sync), ) -> VortexResult { - let cache_key = source - .uri() - .map_or_else(|| fallback_key.to_owned(), |uri| uri.to_string()); - - // Build open options. The cache guard from multi_file() must not live across an await, - // so we scope the cache lookup in a block. - let options = { - let mut options = open_options_fn(session.open_options()); - if let Some(size) = file_size { - options = options.with_file_size(size); - } - if let Some(footer) = session.multi_file().get_footer(&cache_key) { - options = options.with_footer(footer); - } - options + let uri = source.uri().cloned(); + if key.is_none() { + key = uri.as_deref(); + } + let Some(key) = key else { + vortex_bail!("Missing cache key"); }; - let vortex_file = options.open(source).await?; + let mut options = open_options_fn(session.open_options()); + if let Some(size) = file_size { + options = options.with_file_size(size); + } + + { + if let Some(footer) = session.multi_file().get_footer(key) { + options = options.with_footer(footer); + } + } - // Store footer in cache (scoped to avoid holding the guard across subsequent code). - session - .multi_file() - .put_footer(&cache_key, vortex_file.footer().clone()); - Ok(vortex_file) + let file = options.open(source).await?; + session.multi_file().put_footer(key, file.footer().clone()); + Ok(file) } /// A [`LayoutReaderFactory`] that lazily opens a single Vortex file and returns its layout reader. diff --git a/vortex-jni/src/file.rs b/vortex-jni/src/file.rs index 5369f9b965b..552d36675c7 100644 --- a/vortex-jni/src/file.rs +++ b/vortex-jni/src/file.rs @@ -129,7 +129,7 @@ fn read_metadata_segments( file_size: Option, ) -> VortexResult> { RUNTIME.block_on(async move { - let file = open_cached(session, source, cache_key, file_size, &|options| { + let file = open_cached(session, Some(cache_key), source, file_size, &|options| { options.include_metadata() }) .await?;