duckdb: user file path as footer cache

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<Scalar> {
         if self.value.is_none() && !target_dtype.is_nullable() {

vortex-duckdb/src/column_statistics.rs

@@ -24,17 +24,14 @@ pub struct ColumnStatistics {
 }
 
 impl ColumnStatistics {
-    pub fn try_from(stats: &ColumnStatisticsAggregate, dtype: DType) -> VortexResult<Self> {
-        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<Self> {
+        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

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_::<i64>()
                 .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_::<i32>();
-                    match days {
-                        Some(days) => Value::new_date(days),
-                        None => Value::null(&*ext_logical_type(self)?),
-                    }
-                }
+                TimeUnit::Days => match storage.as_::<i32>() {
+                    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 {

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<String> {
-    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<Self> {
-        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<Self> {
+        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<Option<Footer>> {
-    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::<MultiFileSession>().get_footer(&key);
+    let footer = SESSION
+        .get_opt::<MultiFileSession>()
+        .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<ColumnStat
     let dtype = fields.field_by_index(index)?;
     let stats = stats.stats_sets().get(index)?;
     let stats = ColumnStatisticsAggregate::new(stats);
-    match ColumnStatistics::try_from(&stats, dtype) {
+    match ColumnStatistics::try_from(stats, dtype) {
         Ok(stats) => Some(stats),
         Err(e) => vortex_panic!(e),
     }

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<Self> {
         let field_stats = fb.field_stats().unwrap_or_default();
-        let mut array_stats: Vec<ArrayStats> = 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 {

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<dyn VortexReadAt>,
-    fallback_key: &str,
     file_size: Option<u64>,
     open_options_fn: &(dyn Fn(VortexOpenOptions) -> VortexOpenOptions + Send + Sync),
 ) -> VortexResult<VortexFile> {
-    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.

vortex-jni/src/file.rs

@@ -129,7 +129,7 @@ fn read_metadata_segments(
     file_size: Option<u64>,
 ) -> VortexResult<Vec<(String, ByteBuffer)>> {
     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?;