From e5d9a2c0eda35cc1e94cf5a214d080ab50b83ae2 Mon Sep 17 00:00:00 2001 From: Nathan Stocks Date: Thu, 29 Oct 2026 11:09:57 -0600 Subject: [PATCH 1/5] S3 Perf (38.3.1.11): Fetch a stored file's first bytes and line counts for staging on the async runtime Staging a file on S3 kept a blocking thread waiting on S3 for the first bytes it inspects, for the first rows a CSV's schema is inferred from, and through the file's line-count download. These fetches now run on the async runtime, with only the parsing on blocking threads, and a data frame's schema is still read alongside its line count. When either fails, the other stops: a failed count fails the inspection with the store's error. Images, video, audio, Parquet, Arrow, and JSON files are still read through a blocking reader, since their parsers need random access. This does not affect local storage. --- crates/liboxen/src/core/df/tabular.rs | 333 ++++++++++-------- .../src/core/v_latest/workspaces/files.rs | 16 +- crates/liboxen/src/opts.rs | 2 - crates/liboxen/src/opts/count_lines_opts.rs | 14 - crates/liboxen/src/repositories/metadata.rs | 244 +++++++------ .../src/repositories/metadata/tabular.rs | 12 +- .../liboxen/src/repositories/metadata/text.rs | 22 +- crates/liboxen/src/storage/s3.rs | 34 +- crates/liboxen/src/util/fs.rs | 20 +- 9 files changed, 377 insertions(+), 320 deletions(-) delete mode 100644 crates/liboxen/src/opts/count_lines_opts.rs diff --git a/crates/liboxen/src/core/df/tabular.rs b/crates/liboxen/src/core/df/tabular.rs index a71284b1fc..bb14e52e7b 100644 --- a/crates/liboxen/src/core/df/tabular.rs +++ b/crates/liboxen/src/core/df/tabular.rs @@ -22,11 +22,11 @@ use crate::model::DataFrameSize; use crate::model::LocalRepository; use crate::model::data_frame::schema::DataType; use crate::model::merkle_tree::node::MerkleTreeNode; -use crate::opts::{CountLinesOpts, DFOpts}; +use crate::opts::DFOpts; use crate::repositories; use crate::repositories::metadata::{ByteSource, MIN_FETCH, StoredVersion}; use crate::storage::{VersionLocation, VersionStore}; -use crate::util::fs::{AtomicFile, read_first_n_bytes}; +use crate::util::fs::{AtomicFile, LineCounts, line_counts, read_first_n_bytes}; use crate::util::hasher; use std::sync::Arc; @@ -88,20 +88,16 @@ impl ByteSource<'_> { } } -fn base_lazy_csv_reader( - sources: ScanSources, - delimiter: u8, - quote_char: Option, -) -> LazyCsvReader { +fn base_lazy_csv_reader(sources: ScanSources, dialect: CsvDialect) -> LazyCsvReader { let reader = LazyCsvReader::new_with_sources(sources); reader .with_infer_schema_length(Some(CSV_INFER_SCHEMA_LENGTH)) .with_ignore_errors(true) .with_has_header(true) .with_truncate_ragged_lines(true) - .with_separator(delimiter) + .with_separator(dialect.delimiter) .with_eol_char(b'\n') - .with_quote_char(quote_char) + .with_quote_char(dialect.quote_char) .with_rechunk(true) .with_encoding(CsvEncoding::LossyUtf8) } @@ -110,15 +106,10 @@ pub fn new_df() -> DataFrame { DataFrame::empty() } -fn read_df_csv( - path: impl AsRef, - delimiter: u8, - quote_char: Option, -) -> Result { +fn read_df_csv(path: impl AsRef, dialect: CsvDialect) -> Result { let reader = base_lazy_csv_reader( ScanSource::Path(PlRefPath::try_from_path(path.as_ref())?).into_sources(), - delimiter, - quote_char, + dialect, ); reader .finish() @@ -192,43 +183,81 @@ pub fn take(df: LazyFrame, indices: Vec) -> Result { .map_err(|e| OxenError::basic_str(format!("{e:?}"))) } -/// A scan of `source`, a CSV. A stored version file's scan holds only its first rows, enough for its -/// schema or as many as fit in [`MAX_CSV_HEAD_LEN`], fetched into memory. +/// A scan of `source`, a CSV with `dialect`. A stored version file's CSV is scanned from its first +/// bytes in memory, which [`stored_schema`] fetches. fn scan_df_csv( source: &ByteSource, - delimiter: u8, - quote_char: Option, + dialect: CsvDialect, total_rows: usize, ) -> Result { - let sources = match source { - ByteSource::Stored(version) => { - ByteSource::Memory(csv_head(version, quote_char)?).scan_sources()? - } - ByteSource::Path(_) | ByteSource::Memory(_) => source.scan_sources()?, - }; - base_lazy_csv_reader(sources, delimiter, quote_char) + if let ByteSource::Stored(_) = source { + return Err(OxenError::internal_error(format!( + "{source:?}: a stored CSV is scanned from its first bytes in memory" + ))); + } + base_lazy_csv_reader(source.scan_sources()?, dialect) .with_n_rows(Some(total_rows)) .finish() .map_err(|_| OxenError::internal_error(format!("{READ_ERROR}: {source:?}"))) } +/// The schema of `version`, a stored data frame file with `extension` whose rows are counted from +/// its lines. A CSV's schema is read from its first bytes, fetched on the async runtime. Fails when +/// the store cannot give the bytes, and gives the read's error for content that cannot be parsed. +pub(crate) async fn stored_schema( + version: &StoredVersion, + extension: &str, +) -> Result, OxenError> { + let extension = extension.to_string(); + if !matches!(extension.as_str(), "csv" | "tsv" | "data") { + let version = version.clone(); + return Ok(task::spawn_blocking(move || { + scan_schema(&ByteSource::Stored(&version), Some(&extension)).map(|(_, schema)| schema) + }) + .await?); + } + let sample = version.fetch_head(CSV_SNIFF_SAMPLE_BYTES).await?; + let dialect = task::spawn_blocking(move || { + CsvDialect::sniff(&ByteSource::Memory(sample), &extension, &DFOpts::empty()) + }) + .await?; + let dialect = match dialect { + Ok(dialect) => dialect, + Err(err) => return Ok(Err(err)), + }; + let head = csv_head(version, dialect.quote_char).await?; + Ok(task::spawn_blocking(move || { + scan_df_csv( + &ByteSource::Memory(head), + dialect, + constants::DEFAULT_PAGE_SIZE, + )? + .collect_schema() + .map_err(|err| OxenError::basic_str(format!("{err:?}"))) + }) + .await?) +} + /// The first bytes of `version`, a CSV quoted with `quote_char`, holding every row its schema is /// inferred from or as many of them as fit in [`MAX_CSV_HEAD_LEN`], or all of it when it is shorter. -/// Blocks while it fetches them. -fn csv_head(version: &StoredVersion, quote_char: Option) -> Result { - let mut head = version.head(MIN_FETCH)?; +async fn csv_head(version: &StoredVersion, quote_char: Option) -> Result { + let mut head = version.fetch_head(MIN_FETCH).await?; while (head.len() as u64) < version.size().min(MAX_CSV_HEAD_LEN) { - let rows = count_rows_from_slice_par( - Buffer::from_owner(head.clone()), - quote_char, - None, - b'\n', - true, - 0, - 0, - 0, - false, - ); + let counted = head.clone(); + let rows = task::spawn_blocking(move || { + count_rows_from_slice_par( + Buffer::from_owner(counted), + quote_char, + None, + b'\n', + true, + 0, + 0, + 0, + false, + ) + }) + .await?; // The last row counted can be cut short, so a row past those inference reads leaves them // whole. let len = match rows { @@ -238,7 +267,9 @@ fn csv_head(version: &StoredVersion, quote_char: Option) -> Result 0, }; - head = version.head((len.max(2 * head.len()) as u64).min(MAX_CSV_HEAD_LEN))?; + head = version + .fetch_head((len.max(2 * head.len()) as u64).min(MAX_CSV_HEAD_LEN)) + .await?; } Ok(head) } @@ -1041,6 +1072,7 @@ pub fn df_hash_rows_on_cols( Ok(df) } +#[derive(Clone, Copy)] struct CsvDialect { delimiter: u8, quote_char: Option, @@ -1074,46 +1106,53 @@ fn parse_user_csv_opts(opts: &DFOpts) -> Result<(Option, Option), OxenEr Ok((user_delimiter, user_quote)) } -/// Combine sniffer output with the user overrides into a final dialect, falling back to RFC 4180 -/// defaults when sniffing failed. -fn build_csv_dialect( - sniffed: Result, - user_delimiter: Option, - user_quote: Option, -) -> CsvDialect { - match sniffed { - Ok(metadata) => { - log::debug!("Sniffed csv dialect: {:?}", metadata.dialect); - CsvDialect { - delimiter: user_delimiter.unwrap_or(metadata.dialect.delimiter), - quote_char: user_quote.or_else(|| Some(sniffed_quote(&metadata))), +impl CsvDialect { + /// The dialect of `source`, a CSV with `extension`, whose fields a TSV delimits with tabs. The + /// delimiter and quote character `opts` sets win over the sniffed ones. + fn sniff(source: &ByteSource, extension: &str, opts: &DFOpts) -> Result { + let (user_delimiter, user_quote) = parse_user_csv_opts(opts)?; + let dialect = match user_delimiter { + Some(delimiter) if user_quote.is_some() => Self { + delimiter, + quote_char: user_quote, + }, + _ => Self::from_sniffed(source.sniff(), user_delimiter, user_quote), + }; + Ok(match extension { + "tsv" => Self { + delimiter: b'\t', + ..dialect + }, + _ => dialect, + }) + } + + /// Combine sniffer output with the user overrides into a final dialect, falling back to RFC + /// 4180 defaults when sniffing failed. + fn from_sniffed( + sniffed: Result, + user_delimiter: Option, + user_quote: Option, + ) -> Self { + match sniffed { + Ok(metadata) => { + log::debug!("Sniffed csv dialect: {:?}", metadata.dialect); + Self { + delimiter: user_delimiter.unwrap_or(metadata.dialect.delimiter), + quote_char: user_quote.or_else(|| Some(sniffed_quote(&metadata))), + } } - } - Err(err) => { - log::warn!("Error sniffing csv -> {err:?}"); - CsvDialect { - delimiter: user_delimiter.unwrap_or(DEFAULT_DELIMITER), - quote_char: user_quote.or(Some(DEFAULT_QUOTE_CHAR)), + Err(err) => { + log::warn!("Error sniffing csv -> {err:?}"); + Self { + delimiter: user_delimiter.unwrap_or(DEFAULT_DELIMITER), + quote_char: user_quote.or(Some(DEFAULT_QUOTE_CHAR)), + } } } } } -fn sniff_csv_dialect(source: &ByteSource, opts: &DFOpts) -> Result { - let (user_delimiter, user_quote) = parse_user_csv_opts(opts)?; - if let (Some(delimiter), Some(_)) = (user_delimiter, user_quote) { - return Ok(CsvDialect { - delimiter, - quote_char: user_quote, - }); - } - Ok(build_csv_dialect( - source.sniff(), - user_delimiter, - user_quote, - )) -} - const DEFAULT_QUOTE_CHAR: u8 = b'"'; const DEFAULT_DELIMITER: u8 = b','; @@ -1174,13 +1213,9 @@ async fn _read_lazy_df_with_extension( match extension.as_str() { "ndjson" | "jsonl" => read_df_jsonl(&path), "json" => read_df_json(&path), - "csv" | "data" => { - let dialect = sniff_csv_dialect(&ByteSource::Path(&path), &opts_clone)?; - read_df_csv(&path, dialect.delimiter, dialect.quote_char) - } - "tsv" => { - let dialect = sniff_csv_dialect(&ByteSource::Path(&path), &opts_clone)?; - read_df_csv(&path, b'\t', dialect.quote_char) + "csv" | "data" | "tsv" => { + let dialect = CsvDialect::sniff(&ByteSource::Path(&path), &extension, &opts_clone)?; + read_df_csv(&path, dialect) } "parquet" => read_df_parquet(&path), "arrow" => { @@ -1308,10 +1343,7 @@ async fn read_s3_version_df( // `.tsv` pins the tab delimiter; only the quote character is sniffed. let force_delim = (extension == "tsv").then_some(b'\t'); let dialect = resolve_s3_csv_dialect(version_store, hash, opts, force_delim).await?; - task::spawn_blocking(move || { - read_s3_csv(&url, cloud_opts, dialect.delimiter, dialect.quote_char) - }) - .await?? + task::spawn_blocking(move || read_s3_csv(&url, cloud_opts, dialect)).await?? } "parquet" => task::spawn_blocking(move || read_s3_parquet(&url, cloud_opts)).await??, "arrow" => { @@ -1379,7 +1411,7 @@ async fn resolve_s3_csv_dialect( let sniffed = task::spawn_blocking(move || qsv_sniffer::Sniffer::new().sniff_reader(Cursor::new(sample))) .await?; - Ok(build_csv_dialect(sniffed, delimiter, user_quote)) + Ok(CsvDialect::from_sniffed(sniffed, delimiter, user_quote)) } fn read_s3_parquet(url: &str, cloud_opts: CloudOptions) -> Result { @@ -1419,13 +1451,11 @@ fn read_s3_jsonl(url: &str, cloud_opts: CloudOptions) -> Result, + dialect: CsvDialect, ) -> Result { base_lazy_csv_reader( ScanSource::Path(PlRefPath::new(url)).into_sources(), - delimiter, - quote_char, + dialect, ) .with_cloud_options(Some(cloud_opts)) .finish() @@ -1478,14 +1508,11 @@ fn scan_df_with_extension( "ndjson" => scan_df_jsonl(source, total_rows), "jsonl" => scan_df_jsonl(source, total_rows), "json" => read_json(source), - "csv" | "data" => { - let dialect = sniff_csv_dialect(source, opts)?; - scan_df_csv(source, dialect.delimiter, dialect.quote_char, total_rows) - } - "tsv" => { - let dialect = sniff_csv_dialect(source, opts)?; - scan_df_csv(source, b'\t', dialect.quote_char, total_rows) - } + "csv" | "data" | "tsv" => scan_df_csv( + source, + CsvDialect::sniff(source, extension, opts)?, + total_rows, + ), "parquet" => scan_df_parquet(source, total_rows), "arrow" => scan_df_arrow(source, total_rows), _ => Err(OxenError::basic_str(err)), @@ -1506,22 +1533,58 @@ pub(crate) fn get_size_and_schema_with_extension( source: &ByteSource, extension: Option<&str>, ) -> Result<(DataFrameSize, SchemaRef), OxenError> { + let (lazy_df, schema) = scan_schema(source, extension)?; + let height = match (header_lines(extension), extension) { + (Some(header), _) => height_from_lines(&line_counts(source.open()?, false)?, header), + (None, Some("parquet")) => ParquetReader::new(source.open()?).num_rows()?, + (None, Some("arrow")) => { + // With no columns projected, the reader counts the rows from the record batches' + // headers and reads none of their data. + let reader = IpcReader::new(source.open()?).with_projection(Some(vec![])); + reader.finish()?.height() + } + (None, Some("json")) => { + let df = lazy_df + .collect() + .map_err(|_| OxenError::basic_str("Could not collect json df"))?; + df.height() + } + _ => { + return Err(OxenError::basic_str(format!( + "Unknown file type get_size {source:?} {extension:?}" + ))); + } + }; + Ok(( + DataFrameSize { + width: schema.len(), + height, + }, + schema, + )) +} + +/// The lines before the first row of a data frame file with `extension` whose rows are counted from +/// its lines: 1 for CSV and TSV, and 0 for `.data` and JSON lines. `None` for any other format. +pub(crate) fn header_lines(extension: Option<&str>) -> Option { + match extension { + Some("csv" | "tsv") => Some(1), + Some("data" | "jsonl" | "ndjson") => Some(0), + _ => None, + } +} + +/// A scan of `source`, a data frame file with `extension`, and its schema. +fn scan_schema( + source: &ByteSource, + extension: Option<&str>, +) -> Result<(LazyFrame, SchemaRef), OxenError> { tracing::debug!( oxen.data_frame_source = ?source, file.extension = ?extension, - "Measuring a data frame's size" + "Reading a data frame's schema" ); - let mut opts = CountLinesOpts::empty(); - opts.remove_trailing_blank_line = true; - // A stored file downloads to count its lines while the scan reads its schema. - let line_count = match extension { - Some("csv" | "tsv" | "data" | "jsonl" | "ndjson") => { - Some(source.start_counting_lines(opts)) - } - _ => None, - }; - // Don't need that many rows to get the width let num_scan_rows = constants::DEFAULT_PAGE_SIZE; let mut lazy_df = scan_df_with_extension(source, extension, &DFOpts::empty(), num_scan_rows)?; @@ -1529,44 +1592,12 @@ pub(crate) fn get_size_and_schema_with_extension( source.record_scan_error(&err); OxenError::basic_str(format!("{err:?}")) })?; - let width = schema.len(); - - let err = format!("Unknown file type get_size {source:?} {extension:?}"); - - match (extension, line_count) { - (Some("csv" | "tsv"), Some(line_count)) => { - // Remove one line to account for CSV/TSV headers - let (mut height, _) = line_count.finish(source.open()?)?; - height -= 1; // Adjusting for header - - Ok((DataFrameSize { width, height }, schema)) - } - (Some("data" | "jsonl" | "ndjson"), Some(line_count)) => { - let (height, _) = line_count.finish(source.open()?)?; + Ok((lazy_df, schema)) +} - Ok((DataFrameSize { width, height }, schema)) - } - (Some("parquet"), _) => { - let mut reader = ParquetReader::new(source.open()?); - let height = reader.num_rows()?; - Ok((DataFrameSize { width, height }, schema)) - } - (Some("arrow"), _) => { - // With no columns projected, the reader counts the rows from the record batches' - // headers and reads none of their data. - let reader = IpcReader::new(source.open()?).with_projection(Some(vec![])); - let height = reader.finish()?.height(); - Ok((DataFrameSize { width, height }, schema)) - } - (Some("json"), _) => { - let df = lazy_df - .collect() - .map_err(|_| OxenError::basic_str("Could not collect json df"))?; - let height = df.height(); - Ok((DataFrameSize { width, height }, schema)) - } - _ => Err(OxenError::basic_str(err)), - } +/// The rows of a data frame file with `lines`, whose first `header` lines come before its rows. +pub(crate) fn height_from_lines(lines: &LineCounts, header: usize) -> usize { + lines.lines(true).0 - header } pub fn write_df_json>(df: &mut DataFrame, output: P) -> Result<(), OxenError> { @@ -2197,13 +2228,14 @@ mod tests { #[test] fn test_sniff_csv_dialect_detects_quote_char() -> Result<(), OxenError> { - let dialect = sniff_csv_dialect( + let dialect = CsvDialect::sniff( &ByteSource::Path( &TEST_DATA_DIR .join("test") .join("csvs") .join("quoted_fields.csv"), ), + "csv", &DFOpts::empty(), )?; assert_eq!(dialect.delimiter, b','); @@ -2213,13 +2245,14 @@ mod tests { #[test] fn test_reject_empty_quote() -> Result<(), OxenError> { - let r = sniff_csv_dialect( + let r = CsvDialect::sniff( &ByteSource::Path( &TEST_DATA_DIR .join("test") .join("csvs") .join("quoted_fields.csv"), ), + "csv", &{ let mut opts = DFOpts::empty(); opts.quote_char = Some("".to_string()); diff --git a/crates/liboxen/src/core/v_latest/workspaces/files.rs b/crates/liboxen/src/core/v_latest/workspaces/files.rs index 1fcab423fc..0c04d29f2f 100644 --- a/crates/liboxen/src/core/v_latest/workspaces/files.rs +++ b/crates/liboxen/src/core/v_latest/workspaces/files.rs @@ -124,13 +124,9 @@ pub(crate) async fn add_stored_version_file( url, ®ion, endpoint_url.as_deref(), - lines, ) .await?; - spawn_blocking(move || { - InspectedVersionFile::from_stored(dst_path_buf, file_hash, version) - }) - .await?? + InspectedVersionFile::from_stored(dst_path_buf, file_hash, version, lines).await? } }; let staged_db_manager = get_staged_db_manager(&workspace.workspace_repo)?; @@ -173,15 +169,17 @@ impl InspectedVersionFile { }) } - /// Inspect `version`, the version file `hash` in an S3 store, to stage at `dst_path`. Blocks - /// while it reads what the metadata needs from the store. - pub(crate) fn from_stored( + /// Inspect `version`, the version file `hash` in an S3 store, to stage at `dst_path`, reading + /// what the metadata needs from the store and its line counts from `lines` when they were taken + /// as it arrived. + pub(crate) async fn from_stored( dst_path: PathBuf, hash: MerkleHash, version: StoredVersion, + lines: Option, ) -> Result { let num_bytes = version.size(); - let (mime_type, data_type, metadata) = inspect_stored(&version, &dst_path)?; + let (mime_type, data_type, metadata) = inspect_stored(version, &dst_path, lines).await?; Ok(Self { dst_path, hash, diff --git a/crates/liboxen/src/opts.rs b/crates/liboxen/src/opts.rs index 970d08e177..a7a3b5cb06 100644 --- a/crates/liboxen/src/opts.rs +++ b/crates/liboxen/src/opts.rs @@ -4,7 +4,6 @@ pub mod add_opts; pub mod clean_opts; pub mod clone_opts; -pub mod count_lines_opts; pub mod df_opts; pub mod diff_opts; pub mod download_tree_opts; @@ -23,7 +22,6 @@ pub mod upload_opts; pub use crate::opts::add_opts::AddOpts; pub use crate::opts::clean_opts::CleanOpts; pub use crate::opts::clone_opts::CloneOpts; -pub use crate::opts::count_lines_opts::CountLinesOpts; pub use crate::opts::df_opts::DFOpts; pub use crate::opts::df_opts::SliceRange; pub use crate::opts::diff_opts::DiffOpts; diff --git a/crates/liboxen/src/opts/count_lines_opts.rs b/crates/liboxen/src/opts/count_lines_opts.rs deleted file mode 100644 index c11b2ff8f1..0000000000 --- a/crates/liboxen/src/opts/count_lines_opts.rs +++ /dev/null @@ -1,14 +0,0 @@ -#[derive(Clone, Debug)] -pub struct CountLinesOpts { - pub with_chars: bool, - pub remove_trailing_blank_line: bool, -} - -impl CountLinesOpts { - pub fn empty() -> CountLinesOpts { - CountLinesOpts { - with_chars: false, - remove_trailing_blank_line: false, - } - } -} diff --git a/crates/liboxen/src/repositories/metadata.rs b/crates/liboxen/src/repositories/metadata.rs index 3ab1b55106..6b07c41e51 100644 --- a/crates/liboxen/src/repositories/metadata.rs +++ b/crates/liboxen/src/repositories/metadata.rs @@ -1,7 +1,7 @@ //! Helper functions to get metadata from the local filesystem. //! -use crate::core::df::tabular::s3_cloud_options; +use crate::core::df::tabular::{header_lines, height_from_lines, s3_cloud_options, stored_schema}; use crate::error::OxenError; use crate::model::entry::entry_data_type::EntryDataType; use crate::model::entry::metadata_entry::CLIMetadataEntry; @@ -9,7 +9,6 @@ use crate::model::merkle_tree::node::{DirNode, FileNode}; use crate::model::metadata::generic_metadata::GenericMetadata; use crate::model::parsed_resource::ParsedResourceView; use crate::model::{Commit, LocalRepository, MerkleHash, MetadataEntry}; -use crate::opts::CountLinesOpts; use crate::repositories; use crate::storage::VersionStore; use crate::util; @@ -29,7 +28,7 @@ use std::io::{self, Cursor, Read, Seek, SeekFrom}; use std::path::{Path, PathBuf}; use std::sync::Arc; use tokio::runtime::Handle; -use tokio_util::task::AbortOnDropHandle; +use tokio::task::spawn_blocking; pub mod audio; pub mod image; @@ -168,33 +167,6 @@ impl ByteSource<'_> { } } - /// What [`util::fs::count_lines`] gives for `data`, a reader of these bytes, which is left - /// unread for a stored version file. Blocks while a stored file downloads to count its lines. - pub(crate) fn count_lines( - &self, - data: impl Read, - opts: CountLinesOpts, - ) -> Result<(usize, Option), OxenError> { - self.start_counting_lines(opts).finish(data) - } - - /// Starts counting the lines of these bytes. A stored version file whose lines were not counted - /// as it arrived downloads to count them while the caller goes on, until the count finishes or - /// is dropped. - pub(crate) fn start_counting_lines(&self, opts: CountLinesOpts) -> LineCount<'_> { - let counting = match self { - ByteSource::Stored(version) => match version.lines { - Some(lines) => Counting::Taken(lines), - None => Counting::Downloading( - &version.bytes, - version.bytes.download_line_counts(opts.with_chars), - ), - }, - ByteSource::Path(_) | ByteSource::Memory(_) => Counting::Unread, - }; - LineCount { opts, counting } - } - /// Fails a stored version file's inspection with `err`, from a Polars scan of the file, when /// the scan could not fetch the file's bytes. pub(crate) fn record_scan_error(&self, err: &PolarsError) { @@ -220,48 +192,12 @@ impl fmt::Debug for ByteSource<'_> { } } -/// A line count [`ByteSource::start_counting_lines`] started. -pub(crate) struct LineCount<'a> { - opts: CountLinesOpts, - counting: Counting<'a>, -} - -/// Where a [`LineCount`] gets the counts it finishes with. -enum Counting<'a> { - /// A reader of the bytes, read when the count finishes. - Unread, - /// The counts taken as a stored version file arrived. - Taken(LineCounts), - /// A download of a stored version file, which the task runs. - Downloading( - &'a StoredBytes, - AbortOnDropHandle>, - ), -} - -impl LineCount<'_> { - /// What [`util::fs::count_lines`] gives for `data`, a reader of the bytes, which is left unread - /// for a stored version file. Blocks until a stored file's download ends. - pub(crate) fn finish(self, data: impl Read) -> Result<(usize, Option), OxenError> { - let lines = match self.counting { - Counting::Unread => return util::fs::count_lines(data, self.opts), - Counting::Taken(lines) => lines, - Counting::Downloading(bytes, download) => bytes - .runtime - .block_on(download)? - .map_err(|err| bytes.record_error(err))?, - }; - let (line_count, chars) = lines.lines(self.opts.remove_trailing_blank_line); - Ok((line_count, chars.filter(|_| self.opts.with_chars))) - } -} - -/// A version file in an S3 version store, with its line counts when they were taken as it arrived. +/// A version file in an S3 version store. +#[derive(Clone)] pub(crate) struct StoredVersion { bytes: Arc, pub(crate) url: String, pub(crate) cloud_options: CloudOptions, - lines: Option, } impl StoredVersion { @@ -274,7 +210,6 @@ impl StoredVersion { url: String, region: &str, endpoint_url: Option<&str>, - lines: Option, ) -> Result { let cloud_options = s3_cloud_options(&url, region, endpoint_url)?; Ok(Self { @@ -288,7 +223,6 @@ impl StoredVersion { }), url, cloud_options, - lines, }) } @@ -307,10 +241,9 @@ impl StoredVersion { } /// The file's first `len` bytes, or all of it when it is shorter, running on past `len` when an - /// earlier read of its start fetched more. Fetches only what no earlier read of its start did, - /// and blocks while it does. - pub(crate) fn head(&self, len: u64) -> io::Result { - self.bytes.head(len) + /// earlier read of its start fetched more. Fetches only what no earlier read of its start did. + pub(crate) async fn fetch_head(&self, len: u64) -> Result { + self.bytes.fetch_head(len).await } } @@ -335,23 +268,30 @@ impl StoredBytes { .runtime .block_on(fetch) .map_err(|err| self.record_error(err))?; - if chunk.is_empty() { - return Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "the version store returned no bytes before the file's end", - )); - } - Ok(Bytes::from(chunk)) + nonempty(chunk) } - /// See [`StoredVersion::head`]. + /// What [`StoredVersion::fetch_head`] gives, blocking while it fetches. fn head(&self, len: u64) -> io::Result { + self.runtime + .block_on(self.fetch_head(len)) + .map_err(|err| self.record_error(err)) + } + + /// See [`StoredVersion::fetch_head`]. + async fn fetch_head(&self, len: u64) -> Result { let head = self.head.lock().clone(); let len = len.min(self.size); if head.len() as u64 >= len { return Ok(head); } - let rest = self.fetch(head.len() as u64, len - head.len() as u64)?; + let hash = self.hash.to_string(); + let offset = head.len() as u64; + let rest = nonempty( + self.store + .get_version_chunk(&hash, offset, len - offset) + .await?, + )?; let mut joined = BytesMut::with_capacity(head.len() + rest.len()); joined.extend_from_slice(&head); joined.extend_from_slice(&rest); @@ -360,24 +300,18 @@ impl StoredBytes { Ok(joined) } - /// Starts counting the file's lines, and its characters when `with_chars` is set, from a - /// download of it on the async runtime, which dropping the task stops. - fn download_line_counts( - self: &Arc, - with_chars: bool, - ) -> AbortOnDropHandle> { - let bytes = Arc::clone(self); - AbortOnDropHandle::new(self.runtime.spawn(async move { - let mut stream = bytes - .store - .get_version_stream(&bytes.hash.to_string()) - .await?; - let mut lines = LineCounts::new(with_chars); - while let Some(chunk) = stream.next().await { - lines.add(&chunk?); - } - Ok(lines) - })) + /// The file's line counts, and its character count when `with_chars` is set, from a download of + /// the whole file. + async fn download_line_counts(&self, with_chars: bool) -> Result { + let mut stream = self + .store + .get_version_stream(&self.hash.to_string()) + .await?; + let mut lines = LineCounts::new(with_chars); + while let Some(chunk) = stream.next().await { + lines.add(&chunk?); + } + Ok(lines) } /// Keeps `err` unless an earlier failure was kept, and returns an error to fail the read with. @@ -393,6 +327,17 @@ impl StoredBytes { } } +/// `chunk`, a range the store returned from before the file's end, which holds at least one byte. +fn nonempty(chunk: Vec) -> io::Result { + if chunk.is_empty() { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "the version store returned no bytes before the file's end", + )); + } + Ok(Bytes::from(chunk)) +} + /// The fewest bytes a [`VersionReader`] fetches at once, 64 KiB. pub(crate) const MIN_FETCH: u64 = 64 * 1024; /// The most bytes a [`VersionReader`] fetches at once to read ahead of a run of reads, 8 MiB. @@ -477,14 +422,80 @@ pub(crate) fn inspect_bytes( } /// Reads the MIME type, data type, and metadata of `version`, the bytes of a file named -/// `file_path`, fetching what it reads from the store. Blocks while it reads them. Fails when the -/// store cannot give it the bytes, and gives no metadata for content that cannot be parsed. -pub(crate) fn inspect_stored( - version: &StoredVersion, +/// `file_path`, fetching what it reads from the store. A text file's or line-counted data frame's +/// line counts come from `lines` when they were taken as it arrived, and otherwise from a download +/// of the file. Fails when the store cannot give it the bytes, and gives no metadata for content +/// that cannot be parsed. +pub(crate) async fn inspect_stored( + version: StoredVersion, file_path: &Path, + lines: Option, ) -> Result<(String, EntryDataType, Option), OxenError> { - let inspected = inspect(&ByteSource::Stored(version), file_path); - version.bytes.take_error().map_or(inspected, Err) + let head = read_head(&mut Cursor::new(version.fetch_head(MIN_FETCH).await?))?; + let mime_type = mime_type_from_head(&head, file_path); + let data_type = data_type_from_head(&head, file_path, &mime_type); + let extension = file_extension(file_path); + let metadata = match (&data_type, header_lines(Some(&extension))) { + (EntryDataType::Text, _) => { + let lines = match lines { + Some(lines) => lines, + None => version.bytes.download_line_counts(true).await?, + }; + Some(GenericMetadata::MetadataText(text::from_counts(&lines))) + } + // The schema is read while the lines are counted, and either failing stops the other. + (EntryDataType::Tabular, Some(header)) => { + let scanning = stored_schema(&version, &extension); + let counting = async { + match lines { + Some(lines) => Ok(lines), + None => version.bytes.download_line_counts(false).await, + } + }; + tokio::pin!(scanning, counting); + let scanned = tokio::select! { + scanned = &mut scanning => match scanned? { + Ok(schema) => Ok((schema, counting.await?)), + Err(err) => Err(err), + }, + lines = &mut counting => { + let lines = lines?; + scanning.await?.map(|schema| (schema, lines)) + } + }; + let tabular = scanned.map(|(schema, lines)| { + let height = height_from_lines(&lines, header); + GenericMetadata::MetadataTabular(tabular::from_schema(&schema, height)) + }); + settle( + tabular, + &ByteSource::Stored(&version), + file_path, + "Could not compute tabular metadata", + )? + } + ( + EntryDataType::Image + | EntryDataType::Video + | EntryDataType::Audio + | EntryDataType::Tabular, + _, + ) => { + let version = version.clone(); + let file_path = file_path.to_path_buf(); + let data_type = data_type.clone(); + spawn_blocking(move || { + let source = ByteSource::Stored(&version); + get_file_metadata(&mut version.reader(), &source, &file_path, &data_type) + }) + .await?? + } + (EntryDataType::Dir | EntryDataType::Binary, _) => None, + }; + version + .bytes + .take_error() + .map_or(Ok((mime_type, data_type, metadata)), Err) } fn inspect( @@ -503,9 +514,9 @@ fn inspect( } /// Returns the metadata of `data`, the bytes of a file of `data_type` named `file_path`. A tabular -/// file's metadata, and line counts taken as a stored file arrived, come from `source` instead, -/// which holds the same bytes as `data`. Gives no metadata for content that cannot be parsed, and -/// fails when the store cannot give a stored file's bytes. +/// file's metadata comes from `source` instead, which holds the same bytes as `data`. Gives no +/// metadata for content that cannot be parsed, and fails when the store cannot give a stored file's +/// bytes. fn get_file_metadata( data: &mut (impl Read + Seek), source: &ByteSource, @@ -516,7 +527,7 @@ fn get_file_metadata( data.rewind()?; let (metadata, failure) = match data_type { EntryDataType::Text => ( - text::get_metadata(data, source).map(GenericMetadata::MetadataText), + text::get_metadata(data).map(GenericMetadata::MetadataText), "Could not compute text metadata", ), EntryDataType::Image => ( @@ -538,6 +549,17 @@ fn get_file_metadata( ), _ => return Ok(None), }; + settle(metadata, source, file_path, failure) +} + +/// `metadata`, or `None` with `failure` logged when its content could not be parsed. Fails with the +/// store's error instead when a stored file's bytes could not be fetched. +fn settle( + metadata: Result, + source: &ByteSource, + file_path: &Path, + failure: &str, +) -> Result, OxenError> { match metadata { Ok(metadata) => Ok(Some(metadata)), Err(err) => { diff --git a/crates/liboxen/src/repositories/metadata/tabular.rs b/crates/liboxen/src/repositories/metadata/tabular.rs index 7465915b40..28ea111176 100644 --- a/crates/liboxen/src/repositories/metadata/tabular.rs +++ b/crates/liboxen/src/repositories/metadata/tabular.rs @@ -6,6 +6,7 @@ use crate::error::OxenError; use crate::model::Schema; use crate::model::metadata::MetadataTabular; use crate::repositories::metadata::ByteSource; +use polars::prelude::SchemaRef; use std::path::Path; @@ -23,11 +24,12 @@ pub(crate) fn get_metadata_with_extension( extension: &str, ) -> Result { let (size, schema) = tabular::get_size_and_schema_with_extension(source, Some(extension))?; - Ok(MetadataTabular::new( - size.width, - size.height, - Schema::from_polars(&schema), - )) + Ok(from_schema(&schema, size.height)) +} + +/// The tabular metadata of a data frame with `schema` and `height` rows. +pub(crate) fn from_schema(schema: &SchemaRef, height: usize) -> MetadataTabular { + MetadataTabular::new(schema.len(), height, Schema::from_polars(schema)) } #[cfg(test)] diff --git a/crates/liboxen/src/repositories/metadata/text.rs b/crates/liboxen/src/repositories/metadata/text.rs index a7c8b38036..e072ce4d6a 100644 --- a/crates/liboxen/src/repositories/metadata/text.rs +++ b/crates/liboxen/src/repositories/metadata/text.rs @@ -3,23 +3,19 @@ use crate::error::OxenError; use crate::model::metadata::MetadataText; -use crate::opts::CountLinesOpts; -use crate::repositories::metadata::ByteSource; +use crate::util::fs::{LineCounts, line_counts}; use std::io::Read; -/// Detects the text metadata of the file `data` reads, which holds `source`'s bytes. -pub(crate) fn get_metadata( - data: impl Read, - source: &ByteSource, -) -> Result { - let mut opts = CountLinesOpts::empty(); - opts.with_chars = true; - - let (lines_count, chars_count) = source.count_lines(data, opts)?; - let chars_count = chars_count.unwrap_or(0); +/// Detects the text metadata of the file `data` reads. +pub(crate) fn get_metadata(data: impl Read) -> Result { + Ok(from_counts(&line_counts(data, true)?)) +} - Ok(MetadataText::new(lines_count, chars_count)) +/// The text metadata of a file with `lines`, which counted its characters. +pub(crate) fn from_counts(lines: &LineCounts) -> MetadataText { + let (lines_count, chars_count) = lines.lines(false); + MetadataText::new(lines_count, chars_count.unwrap_or(0)) } #[cfg(test)] diff --git a/crates/liboxen/src/storage/s3.rs b/crates/liboxen/src/storage/s3.rs index 0c07049810..9bf1ec1ab7 100644 --- a/crates/liboxen/src/storage/s3.rs +++ b/crates/liboxen/src/storage/s3.rs @@ -2611,14 +2611,11 @@ pub(crate) mod tests { url, ®ion, endpoint_url.as_deref(), - None, ) .await .unwrap(); - let file_path = path.to_path_buf(); - tokio::task::spawn_blocking(move || inspect_stored(&version, &file_path)) + inspect_stored(version, path, None) .await - .unwrap() .unwrap_or_else(|e| panic!("{path:?}: {e}")) } @@ -2809,6 +2806,25 @@ pub(crate) mod tests { .await?, UploadStart::AlreadyStored )); + // A data frame's height comes from the line counts taken as its parts arrive. + let csv = Bytes::from_static(b"a,b\n1,2\n3,4\n5,6\n"); + let csv_hash = hash_buffer(&csv); + let csv_destination = WorkspaceDestination { + workspace_id: s3_workspace.id.clone(), + path: PathBuf::from("uploaded.csv"), + }; + let UploadStart::AwaitingParts(upload) = Upload::begin( + &s3_workspace.base_repo, + &csv_hash, + csv.len() as u64, + Some(csv_destination), + ) + .await? + else { + panic!("the store does not have {csv_hash} yet"); + }; + let receiving = upload.receive_part(1)?.expect("part 1 is not received yet"); + assert_eq!(receiving.store(csv.clone()).await?, PartStored::Completed); let staged_db = get_staged_db_manager(&s3_workspace.workspace_repo)?; for path in ["uploaded.txt", "again.txt"] { let staged = staged_db @@ -2822,6 +2838,16 @@ pub(crate) mod tests { assert_eq!(num_lines, 3, "{path} is staged with its line count"); assert_eq!(file.num_bytes(), data.len() as u64, "{path}"); } + let staged = staged_db + .read_from_staged_db("uploaded.csv")? + .expect("uploaded.csv is staged"); + match staged.node.file()?.metadata() { + Some(GenericMetadata::MetadataTabular(tabular)) => { + assert_eq!(tabular.tabular.height, 3, "the header and trailing newline"); + assert_eq!(tabular.tabular.width, 2); + } + other => panic!("uploaded.csv is staged as a data frame: {other:?}"), + } drop(staged_db); Ok(()) diff --git a/crates/liboxen/src/util/fs.rs b/crates/liboxen/src/util/fs.rs index 2adec83b3b..f8008fa505 100644 --- a/crates/liboxen/src/util/fs.rs +++ b/crates/liboxen/src/util/fs.rs @@ -36,7 +36,6 @@ use crate::model::merkle_tree::node::FileNode; use crate::model::metadata::metadata_image::ImgResize; use crate::model::metadata::metadata_video::VideoThumbnail; use crate::model::{EntryDataType, LocalRepository}; -use crate::opts::CountLinesOpts; use crate::storage::version_store::VersionStore; use crate::view::health::DiskUsage; use crate::{constants, repositories, util}; @@ -230,12 +229,10 @@ pub fn append_to_file(path: &Path, value: &str) -> Result<(), OxenError> { } } -pub(crate) fn count_lines( - data: impl Read, - opts: CountLinesOpts, -) -> Result<(usize, Option), OxenError> { +/// The line counts of `data`, with its character count when `with_chars` is set. +pub(crate) fn line_counts(data: impl Read, with_chars: bool) -> Result { let mut reader = BufReader::with_capacity(1024 * 32, data); - let mut counts = LineCounts::new(opts.with_chars); + let mut counts = LineCounts::new(with_chars); loop { let len = { let buf = reader.fill_buf()?; @@ -247,11 +244,11 @@ pub(crate) fn count_lines( }; reader.consume(len); } - Ok(counts.lines(opts.remove_trailing_blank_line)) + Ok(counts) } -/// The newlines, and optionally the UTF-8 characters, in bytes added in order, from which -/// [`count_lines`]'s answer follows however the bytes were split. +/// The newlines, and optionally the UTF-8 characters, in bytes added in order, which come out the +/// same however the bytes were split. #[derive(Clone, Copy, Debug)] pub(crate) struct LineCounts { newlines: usize, @@ -278,9 +275,8 @@ impl LineCounts { } } - /// What [`count_lines`] returns for the bytes added so far: the line count, less a trailing - /// newline's empty line when `remove_trailing_blank_line` is set, and the character count when - /// these counts take one. + /// The line count of the bytes added so far, less a trailing newline's empty line when + /// `remove_trailing_blank_line` is set, and the character count when these counts take one. pub(crate) fn lines(&self, remove_trailing_blank_line: bool) -> (usize, Option) { let trailing = remove_trailing_blank_line && self.last_byte == Some(b'\n'); (1 + self.newlines - usize::from(trailing), self.chars) From d3b37285f39c66ee2513e0ac7b37623bf3a93bbb Mon Sep 17 00:00:00 2001 From: Nathan Stocks Date: Thu, 29 Oct 2026 11:41:04 -0600 Subject: [PATCH 2/5] Sniff a stored CSV's dialect for a data frame view the way staging does A read given both the delimiter and the quote character fetches the file's first 64 KiB as well. ENG-2397 --- crates/liboxen/src/core/df/tabular.rs | 48 ++++++++------------------- 1 file changed, 13 insertions(+), 35 deletions(-) diff --git a/crates/liboxen/src/core/df/tabular.rs b/crates/liboxen/src/core/df/tabular.rs index bb14e52e7b..2be9c23cae 100644 --- a/crates/liboxen/src/core/df/tabular.rs +++ b/crates/liboxen/src/core/df/tabular.rs @@ -1340,10 +1340,19 @@ async fn read_s3_version_df( task::spawn_blocking(move || read_json(&ByteSource::Memory(bytes))).await?? } "csv" | "data" | "tsv" => { - // `.tsv` pins the tab delimiter; only the quote character is sniffed. - let force_delim = (extension == "tsv").then_some(b'\t'); - let dialect = resolve_s3_csv_dialect(version_store, hash, opts, force_delim).await?; - task::spawn_blocking(move || read_s3_csv(&url, cloud_opts, dialect)).await?? + // The dialect is sniffed from the file's first bytes, as for a file on disk. + let sample = Bytes::from( + version_store + .get_version_chunk(hash, 0, CSV_SNIFF_SAMPLE_BYTES) + .await?, + ); + let extension = extension.to_string(); + let opts = opts.clone(); + task::spawn_blocking(move || { + let dialect = CsvDialect::sniff(&ByteSource::Memory(sample), &extension, &opts)?; + read_s3_csv(&url, cloud_opts, dialect) + }) + .await?? } "parquet" => task::spawn_blocking(move || read_s3_parquet(&url, cloud_opts)).await??, "arrow" => { @@ -1383,37 +1392,6 @@ pub(crate) fn s3_cloud_options( )?) } -/// Resolve the CSV/TSV dialect for an S3 version file: honor explicit `DFOpts` overrides, otherwise -/// sniff a head sample fetched through the store's `get_version_chunk` (full parity with the local -/// path, which also samples only the head). `force_delim` pins the delimiter for `.tsv` while still -/// sniffing the quote character. -async fn resolve_s3_csv_dialect( - version_store: &Arc, - hash: &str, - opts: &DFOpts, - force_delim: Option, -) -> Result { - let (user_delimiter, user_quote) = parse_user_csv_opts(opts)?; - let delimiter = force_delim.or(user_delimiter); - if let (Some(delimiter), Some(_)) = (delimiter, user_quote) { - return Ok(CsvDialect { - delimiter, - quote_char: user_quote, - }); - } - // Fixed-size head sample. `sniff_reader` needs Read+Seek and rewinds per pass, so it can't - // consume the forward-only `get_version_stream` directly; a buffer (hence a byte cap) is - // required. If huge first lines sniff wrong, raise this cap. - let sample = version_store - .get_version_chunk(hash, 0, CSV_SNIFF_SAMPLE_BYTES) - .await?; - // Sniffing parses the sample (CPU-bound); keep it off the async runtime. - let sniffed = - task::spawn_blocking(move || qsv_sniffer::Sniffer::new().sniff_reader(Cursor::new(sample))) - .await?; - Ok(CsvDialect::from_sniffed(sniffed, delimiter, user_quote)) -} - fn read_s3_parquet(url: &str, cloud_opts: CloudOptions) -> Result { let args = ScanArgsParquet { cloud_options: Some(cloud_opts), From ae18acab1033dfb549e9cf95584aa7abc7c92682 Mon Sep 17 00:00:00 2001 From: Nathan Stocks Date: Thu, 29 Oct 2026 11:41:23 -0600 Subject: [PATCH 3/5] Fail staging when the store returns no bytes partway through a file ENG-2397 --- crates/liboxen/src/repositories/metadata.rs | 40 +++++++++------------ 1 file changed, 17 insertions(+), 23 deletions(-) diff --git a/crates/liboxen/src/repositories/metadata.rs b/crates/liboxen/src/repositories/metadata.rs index 6b07c41e51..1b7912931f 100644 --- a/crates/liboxen/src/repositories/metadata.rs +++ b/crates/liboxen/src/repositories/metadata.rs @@ -262,13 +262,23 @@ impl StoredBytes { /// The `len` bytes at `offset`, which must lie before the file's end, fetched from the store. /// Blocks. fn fetch(&self, offset: u64, len: u64) -> io::Result { - let hash = self.hash.to_string(); - let fetch = self.store.get_version_chunk(&hash, offset, len); + self.runtime + .block_on(self.fetch_chunk(offset, len)) + .map_err(|err| self.record_error(err)) + } + + /// What [`StoredBytes::fetch`] gives, without blocking. + async fn fetch_chunk(&self, offset: u64, len: u64) -> Result { let chunk = self - .runtime - .block_on(fetch) - .map_err(|err| self.record_error(err))?; - nonempty(chunk) + .store + .get_version_chunk(&self.hash.to_string(), offset, len) + .await?; + if chunk.is_empty() { + return Err(OxenError::internal_error( + "the version store returned no bytes before the file's end", + )); + } + Ok(Bytes::from(chunk)) } /// What [`StoredVersion::fetch_head`] gives, blocking while it fetches. @@ -285,13 +295,8 @@ impl StoredBytes { if head.len() as u64 >= len { return Ok(head); } - let hash = self.hash.to_string(); let offset = head.len() as u64; - let rest = nonempty( - self.store - .get_version_chunk(&hash, offset, len - offset) - .await?, - )?; + let rest = self.fetch_chunk(offset, len - offset).await?; let mut joined = BytesMut::with_capacity(head.len() + rest.len()); joined.extend_from_slice(&head); joined.extend_from_slice(&rest); @@ -327,17 +332,6 @@ impl StoredBytes { } } -/// `chunk`, a range the store returned from before the file's end, which holds at least one byte. -fn nonempty(chunk: Vec) -> io::Result { - if chunk.is_empty() { - return Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "the version store returned no bytes before the file's end", - )); - } - Ok(Bytes::from(chunk)) -} - /// The fewest bytes a [`VersionReader`] fetches at once, 64 KiB. pub(crate) const MIN_FETCH: u64 = 64 * 1024; /// The most bytes a [`VersionReader`] fetches at once to read ahead of a run of reads, 8 MiB. From bcfea6e08ea03e817e4e14b2f1cf71259dddbff2 Mon Sep 17 00:00:00 2001 From: Nathan Stocks Date: Thu, 29 Oct 2026 11:41:33 -0600 Subject: [PATCH 4/5] Fetch a stored JSON data frame whole on the async runtime Polars reads a JSON file to its end, so staging one on S3 kept a blocking thread waiting for the whole download. ENG-2397 --- crates/liboxen/src/repositories/metadata.rs | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/crates/liboxen/src/repositories/metadata.rs b/crates/liboxen/src/repositories/metadata.rs index 1b7912931f..5a2b0ee3c1 100644 --- a/crates/liboxen/src/repositories/metadata.rs +++ b/crates/liboxen/src/repositories/metadata.rs @@ -468,6 +468,27 @@ pub(crate) async fn inspect_stored( "Could not compute tabular metadata", )? } + // A JSON data frame is parsed whole, so its bytes are fetched on the runtime. + (EntryDataType::Tabular, None) if extension == "json" => { + let data = Bytes::from( + version + .bytes + .store + .get_version(&version.bytes.hash.to_string()) + .await?, + ); + let tabular = spawn_blocking(move || { + tabular::get_metadata_with_extension(&ByteSource::Memory(data), "json") + }) + .await? + .map(GenericMetadata::MetadataTabular); + settle( + tabular, + &ByteSource::Stored(&version), + file_path, + "Could not compute tabular metadata", + )? + } ( EntryDataType::Image | EntryDataType::Video From a906e887b621599077498e897f65f27d82a11846 Mon Sep 17 00:00:00 2001 From: Nathan Stocks Date: Thu, 29 Oct 2026 12:26:12 -0600 Subject: [PATCH 5/5] Join a stored CSV's growing head off the async runtime --- crates/liboxen/src/repositories/metadata.rs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/crates/liboxen/src/repositories/metadata.rs b/crates/liboxen/src/repositories/metadata.rs index 5a2b0ee3c1..4605c02991 100644 --- a/crates/liboxen/src/repositories/metadata.rs +++ b/crates/liboxen/src/repositories/metadata.rs @@ -297,10 +297,14 @@ impl StoredBytes { } let offset = head.len() as u64; let rest = self.fetch_chunk(offset, len - offset).await?; - let mut joined = BytesMut::with_capacity(head.len() + rest.len()); - joined.extend_from_slice(&head); - joined.extend_from_slice(&rest); - let joined = joined.freeze(); + // A CSV's head grows toward 64 MiB, too large to copy on the runtime. + let joined = spawn_blocking(move || { + let mut joined = BytesMut::with_capacity(head.len() + rest.len()); + joined.extend_from_slice(&head); + joined.extend_from_slice(&rest); + joined.freeze() + }) + .await?; *self.head.lock() = joined.clone(); Ok(joined) }