diff --git a/crates/liboxen/src/core/db/data_frames/df_db.rs b/crates/liboxen/src/core/db/data_frames/df_db.rs index 0ed679229b..39ed9c0dd9 100644 --- a/crates/liboxen/src/core/db/data_frames/df_db.rs +++ b/crates/liboxen/src/core/db/data_frames/df_db.rs @@ -517,6 +517,32 @@ pub fn count(conn: &duckdb::Connection, table_name: &str) -> Result Result { + let sql = composable(sql)?; + let count = conn.query_row( + &format!("SELECT count(*) FROM ({sql}) AS _oxen_count"), + [], + |row| row.get(0), + )?; + Ok(count) +} + +/// Render `sql` as one statement that can be composed onto, dropping the trailing +/// terminator and any comments a caller may have written. +/// +/// Composition puts a statement somewhere other than the end of the text — before +/// an appended `ORDER BY` or `LIMIT`, or inside a derived table — where a `;` and a +/// line comment both change what follows them rather than being ignored. Text that +/// does not parse as exactly one statement is returned as written, for the database +/// to reject on its own terms. +fn composable(sql: &str) -> Result { + match Parser::parse_sql(&DIALECT, sql)?.as_slice() { + [stmt] => Ok(stmt.to_string()), + _ => Ok(sql.to_string()), + } +} + /// Query number of rows in a table. pub fn count_where( conn: &duckdb::Connection, @@ -565,6 +591,20 @@ pub fn export( Ok(()) } +/// Compose `opts`' sort and pagination onto `stmt`. +/// +/// `stmt` may be caller-supplied SQL that already sorts or bounds itself, so the +/// two are composed rather than concatenated: +/// +/// - A statement that bounds its own extent is a finished result set: the page +/// is read out of it through a subquery, so the two bounds nest instead of +/// colliding, and `opts`' sort is left off rather than reordering rows the +/// statement already picked. +/// - Otherwise the statement's own `ORDER BY` wins, and `opts.sort_by` applies +/// only to a statement that does not sort itself. +/// - A paginated statement left with no order at all is ordered by +/// `_oxen_row_id` where that column resolves against it, so a page names the +/// same rows on every read. See [`SqlShape`]. pub fn prepare_sql( conn: &duckdb::Connection, stmt: &str, @@ -573,25 +613,123 @@ pub fn prepare_sql( let empty_opts = DFOpts::empty(); let opts = opts.unwrap_or(&empty_opts); - let mut sql = add_special_columns(conn, stmt)?; + // Normalize before composing: `add_special_columns` returns the statement as + // written whenever it has no `_oxen_id` to inject, so a terminator would + // otherwise survive into the middle of the composed statement. + let mut sql = add_special_columns(conn, &composable(stmt)?)?; + + // Nothing to compose means nothing to inspect the statement for. + if opts.page.is_some() || opts.sort_by.is_some() { + let shape = sql_shape(&sql)?; + + if !shape.bounds_itself && !shape.orders_itself { + if let Some(sort_by) = &opts.sort_by { + sql.push_str(&format!(" ORDER BY {}", quote_ident(sort_by))); + } else if opts.page.is_some() && shape.resolves_row_id { + sql.push_str(&format!(" ORDER BY {OXEN_ROW_ID_COL}")); + } + } - if opts.sort_by.is_some() { - let sort_by: String = opts.sort_by.clone().unwrap_or_default(); - sql.push_str(&format!(" ORDER BY {}", quote_ident(&sort_by))); + if let Some(page) = opts.page { + let page = if page == 0 { 1 } else { page }; + let requested_size = opts.page_size.unwrap_or(DEFAULT_PAGE_SIZE); + // LIMIT and OFFSET are BIGINTs, so both have to land inside i64 as well + // as usize. Saturating there rather than overflowing means an absurd + // page number reads past the end and yields an empty page, the same + // outcome a non-indexed read gives it. + let page_size = bigint(requested_size); + let offset = bigint(requested_size.saturating_mul(page - 1)); + sql = if shape.bounds_itself { + format!("SELECT * FROM ({sql}) AS _oxen_page LIMIT {page_size} OFFSET {offset}") + } else { + format!("{sql} LIMIT {page_size} OFFSET {offset}") + }; + } } - let pagination_clause = if let Some(page) = opts.page { - let page = if page == 0 { 1 } else { page }; - let page_size = opts.page_size.unwrap_or(DEFAULT_PAGE_SIZE); - format!(" LIMIT {} OFFSET {}", page_size, (page - 1) * page_size) - } else { - "".to_string() - }; - sql.push_str(&pagination_clause); log::debug!("select_str() running sql: {sql}"); Ok(sql) } +/// Clamp a row count to the range DuckDB's `BIGINT` accepts, so a value only +/// `usize` can hold reads as the largest DuckDB can rather than failing to cast. +fn bigint(rows: usize) -> u64 { + (rows as u64).min(i64::MAX as u64) +} + +/// What a statement already says about its own ordering and extent, which +/// decides how [`prepare_sql`] composes `opts` onto it. A statement whose shape +/// this can't read is left alone: every field defaults to false. +#[derive(Default)] +struct SqlShape { + /// Carries its own `ORDER BY`. + orders_itself: bool, + /// Bounds its own extent, with `LIMIT`, `OFFSET`, or `FETCH`. + bounds_itself: bool, + /// `ORDER BY _oxen_row_id` is both valid against it and meaningful: it + /// selects rows of the staged table directly, rather than aggregating, + /// deduping, or reading a derived table that need not carry the column + /// through. DuckDB `UPDATE`s physically relocate rows, so a statement that + /// resolves the column can be given a page order that survives an edit. + resolves_row_id: bool, +} + +fn sql_shape(sql: &str) -> Result { + let ast = Parser::parse_sql(&DIALECT, sql)?; + let Some(Statement::Query(query)) = ast.first() else { + return Ok(SqlShape::default()); + }; + + let resolves_row_id = match &*query.body { + ast::SetExpr::Select(select) => { + let group_by_is_empty = matches!( + &select.group_by, + ast::GroupByExpr::Expressions(exprs, modifiers) + if exprs.is_empty() && modifiers.is_empty() + ); + let reads_table_directly = match select.from.as_slice() { + // Compare the parsed identifier rather than the rendered name so + // a quoted `"df"` reads as the same table. + [table] if table.joins.is_empty() => matches!( + &table.relation, + ast::TableFactor::Table { name, .. } + if matches!(name.0.as_slice(), [ident] if ident.value == TABLE_NAME) + ), + _ => false, + }; + let projects_columns_only = select.projection.iter().all(|item| { + matches!( + item, + SelectItem::Wildcard(_) + | SelectItem::QualifiedWildcard(_, _) + | SelectItem::UnnamedExpr(SqlExpr::Identifier(_)) + | SelectItem::UnnamedExpr(SqlExpr::CompoundIdentifier(_)) + | SelectItem::ExprWithAlias { + expr: SqlExpr::Identifier(_) | SqlExpr::CompoundIdentifier(_), + .. + } + ) + }); + select.distinct.is_none() + && select.having.is_none() + && select.qualify.is_none() + && group_by_is_empty + && reads_table_directly + && projects_columns_only + } + _ => false, + }; + + Ok(SqlShape { + orders_itself: query.order_by.is_some(), + bounds_itself: query.limit.is_some() + || query.offset.is_some() + || query.fetch.is_some() + || !query.limit_by.is_empty(), + resolves_row_id, + }) +} + /// Use this for DuckDB: the sqlparser dialect for all SQL destined for it. pub(crate) const DIALECT: PostgreSqlDialect = PostgreSqlDialect {}; diff --git a/crates/liboxen/src/core/v_latest/data_frames.rs b/crates/liboxen/src/core/v_latest/data_frames.rs index cf8ba20546..67839358cc 100644 --- a/crates/liboxen/src/core/v_latest/data_frames.rs +++ b/crates/liboxen/src/core/v_latest/data_frames.rs @@ -162,6 +162,8 @@ async fn handle_sql_querying( )?; let db_path = repositories::workspaces::data_frames::duckdb_path(&workspace, &query_path); + // No opts: `collect_with_opts` below paginates this path in polars. Passing + // them here would apply the page twice, in SQL and then again on the page. let df = with_hardened_query_conn(&db_path, |conn| sql::query_df(conn, sql, None))?; Ok((workspace, df)) }) diff --git a/crates/liboxen/src/repositories/workspaces/data_frames.rs b/crates/liboxen/src/repositories/workspaces/data_frames.rs index 660707bdc5..c30f81396c 100644 --- a/crates/liboxen/src/repositories/workspaces/data_frames.rs +++ b/crates/liboxen/src/repositories/workspaces/data_frames.rs @@ -120,6 +120,23 @@ pub fn count(workspace: &Workspace, path: &Path) -> Result Result { + let Some(sql) = &opts.sql else { + return count(workspace, path); + }; + + let db_path = repositories::workspaces::data_frames::duckdb_path(workspace, path); + with_hardened_query_conn(&db_path, |conn| df_db::count_sql(conn, sql)) +} + pub fn query( workspace: &Workspace, path: &Path, @@ -147,7 +164,7 @@ pub fn query( with_hardened_query_conn(&db_path, |conn| { if let Some(sql) = &opts.sql { log::debug!("querying sql: {sql:?}"); - sql::query_df(conn, sql.clone(), None) + sql::query_df(conn, sql.clone(), Some(opts)) } else { let mut select = Select::new().select("*").from(TABLE_NAME); // Deterministic page order: DuckDB UPDATEs physically relocate @@ -2422,6 +2439,248 @@ mod tests { .join("bounding_box.csv") } + /// Values of `name` in the frame's current row order, so a page can be + /// compared against the page before it. + fn column_values(df: &DataFrame, name: &str) -> Result, OxenError> { + let column = df.column(name)?; + (0..df.height()) + .map(|i| Ok(column.get(i)?.to_string().trim_matches('"').to_string())) + .collect() + } + + fn page_opts(page: usize, page_size: usize, sql: Option<&str>) -> DFOpts { + let mut opts = DFOpts::empty(); + opts.page = Some(page); + opts.page_size = Some(page_size); + opts.sql = sql.map(str::to_string); + opts + } + + /// A `sql` read returns the page it was asked for. Pages are disjoint, they + /// run out, and they count against the query's own row set rather than the + /// whole frame. + #[tokio::test] + async fn test_query_with_sql_returns_one_page_at_a_time() -> Result<(), OxenError> { + if std::env::consts::OS == "windows" { + return Ok(()); + } + test::run_bounding_box_csv_repo_test_fully_committed_async(|repo| async move { + let commit = repositories::commits::head_commit(&repo)?; + let workspace_id = UserConfig::identifier()?; + let workspace = repositories::workspaces::create(&repo, &commit, workspace_id, true)?; + let path = list_typed_add_row_test_paths(); + workspaces::data_frames::index(&repo, &workspace, &path).await?; + + // 4 of the fixture's 6 rows are labelled dog. + let dogs = format!("SELECT * FROM {TABLE_NAME} WHERE label = 'dog'"); + let sql_page = |page| page_opts(page, 2, Some(&dogs)); + + let first = workspaces::data_frames::query(&workspace, &path, &sql_page(1))?; + assert_eq!(first.height(), 2, "a page must hold at most page_size rows"); + + let second = workspaces::data_frames::query(&workspace, &path, &sql_page(2))?; + assert_eq!(second.height(), 2); + assert_eq!( + column_values(&first, "min_x")?, + vec!["101.5", "102.5"], + "an unsorted query pages in row order" + ); + assert_eq!(column_values(&second, "min_x")?, vec!["7.0", "19.0"]); + + let third = workspaces::data_frames::query(&workspace, &path, &sql_page(3))?; + assert_eq!(third.height(), 0, "pages past the result set are empty"); + + // Pagination values too large to represent must clamp rather than + // overflow the offset or exceed what the LIMIT/OFFSET types accept: the + // furthest page reads past the end, the widest page holds everything. + let furthest = page_opts(usize::MAX, 2, Some(&dogs)); + assert_eq!( + workspaces::data_frames::query(&workspace, &path, &furthest)?.height(), + 0 + ); + let widest = page_opts(1, usize::MAX, Some(&dogs)); + assert_eq!( + workspaces::data_frames::query(&workspace, &path, &widest)?.height(), + 4 + ); + + assert_eq!( + workspaces::data_frames::count_for_query(&workspace, &path, &sql_page(1))?, + 4, + "a query's count is what it selects, not what the frame holds" + ); + assert_eq!( + workspaces::data_frames::count_for_query( + &workspace, + &path, + &page_opts(1, 2, None) + )?, + 6, + "without sql the count is the whole frame" + ); + + Ok(()) + }) + .await + } + + /// A `sql` read pages within whatever the caller's SQL bounds and orders + /// itself to, instead of colliding with its LIMIT or overriding its sort. + #[tokio::test] + async fn test_query_with_sql_pages_inside_callers_own_limit() -> Result<(), OxenError> { + if std::env::consts::OS == "windows" { + return Ok(()); + } + test::run_bounding_box_csv_repo_test_fully_committed_async(|repo| async move { + let commit = repositories::commits::head_commit(&repo)?; + let workspace_id = UserConfig::identifier()?; + let workspace = repositories::workspaces::create(&repo, &commit, workspace_id, true)?; + let path = list_typed_add_row_test_paths(); + workspaces::data_frames::index(&repo, &workspace, &path).await?; + + // The 3 smallest min_x of the 6 rows: 7.0, 19.0, 30.5. + let smallest = format!("SELECT * FROM {TABLE_NAME} ORDER BY min_x LIMIT 3"); + let sql_page = |page| page_opts(page, 2, Some(&smallest)); + + let first = workspaces::data_frames::query(&workspace, &path, &sql_page(1))?; + assert_eq!( + column_values(&first, "min_x")?, + vec!["7.0", "19.0"], + "the caller's sort survives pagination" + ); + + let second = workspaces::data_frames::query(&workspace, &path, &sql_page(2))?; + assert_eq!( + column_values(&second, "min_x")?, + vec!["30.5"], + "the last page is short because the caller's LIMIT bounds it" + ); + + let third = workspaces::data_frames::query(&workspace, &path, &sql_page(3))?; + assert_eq!(third.height(), 0); + + assert_eq!( + workspaces::data_frames::count_for_query(&workspace, &path, &sql_page(1))?, + 3, + "the count must respect the caller's LIMIT too" + ); + + // A single row addressed by offset: the caller bounds the result to + // one row and the page has to leave that row intact. + let offset_row = format!("SELECT * FROM {TABLE_NAME} LIMIT 1 OFFSET 3"); + let opts = page_opts(1, 100, Some(&offset_row)); + let one = workspaces::data_frames::query(&workspace, &path, &opts)?; + assert_eq!(column_values(&one, "min_x")?, vec!["19.0"]); + + Ok(()) + }) + .await + } + + /// A query whose shape can't resolve `_oxen_row_id` — it aggregates or + /// dedupes — still paginates rather than failing to bind an added sort. + #[tokio::test] + async fn test_query_with_aggregate_sql_paginates() -> Result<(), OxenError> { + if std::env::consts::OS == "windows" { + return Ok(()); + } + test::run_bounding_box_csv_repo_test_fully_committed_async(|repo| async move { + let commit = repositories::commits::head_commit(&repo)?; + let workspace_id = UserConfig::identifier()?; + let workspace = repositories::workspaces::create(&repo, &commit, workspace_id, true)?; + let path = list_typed_add_row_test_paths(); + workspaces::data_frames::index(&repo, &workspace, &path).await?; + + let grouped_sql = format!( + "SELECT label, COUNT(*) AS n FROM {TABLE_NAME} GROUP BY label ORDER BY label" + ); + let opts = page_opts(1, 10, Some(&grouped_sql)); + let grouped = workspaces::data_frames::query(&workspace, &path, &opts)?; + assert_eq!(column_values(&grouped, "label")?, vec!["cat", "dog"]); + assert_eq!(column_values(&grouped, "n")?, vec!["2", "4"]); + + let distinct_sql = format!("SELECT DISTINCT label FROM {TABLE_NAME}"); + let opts = page_opts(1, 1, Some(&distinct_sql)); + let distinct = workspaces::data_frames::query(&workspace, &path, &opts)?; + assert_eq!(distinct.height(), 1, "a DISTINCT query paginates as well"); + + Ok(()) + }) + .await + } + + /// A statement written the way it would be typed at a prompt — terminated by + /// `;`, or trailing a comment — pages and counts like any other. Both are + /// harmless at the end of the text and change its meaning in the middle, so + /// they have to come off before a page's bounds are composed on. + #[tokio::test] + async fn test_query_with_terminated_sql_paginates() -> Result<(), OxenError> { + if std::env::consts::OS == "windows" { + return Ok(()); + } + test::run_bounding_box_csv_repo_test_fully_committed_async(|repo| async move { + let commit = repositories::commits::head_commit(&repo)?; + let workspace_id = UserConfig::identifier()?; + let workspace = repositories::workspaces::create(&repo, &commit, workspace_id, true)?; + let path = list_typed_add_row_test_paths(); + workspaces::data_frames::index(&repo, &workspace, &path).await?; + + // Each of these reaches pagination by a different route: a plain + // projection is rewritten to carry `_oxen_id`, while an aggregate and a + // DISTINCT are passed through as written. + let selected = format!("SELECT * FROM {TABLE_NAME};"); + let aggregated = + format!("SELECT label, COUNT(*) AS n FROM {TABLE_NAME} GROUP BY label;"); + let deduped = format!("SELECT DISTINCT label FROM {TABLE_NAME};"); + // A comment would otherwise swallow the composed bounds and quietly + // return the whole frame rather than failing. + let commented = format!("SELECT * FROM {TABLE_NAME} -- every row"); + + for (sql, rows, of) in [ + (&selected, 2, 6), + (&aggregated, 1, 2), + (&deduped, 1, 2), + (&commented, 2, 6), + ] { + let opts = page_opts(1, rows, Some(sql)); + let df = workspaces::data_frames::query(&workspace, &path, &opts)?; + assert_eq!(df.height(), rows, "{sql} should page to {rows} rows"); + let count = workspaces::data_frames::count_for_query(&workspace, &path, &opts)?; + assert_eq!(count, of, "{sql} should count {of} rows"); + } + + Ok(()) + }) + .await + } + + /// The read with no `sql` is unchanged: row-ordered pages of page_size. + #[tokio::test] + async fn test_query_without_sql_pages_in_row_order() -> Result<(), OxenError> { + if std::env::consts::OS == "windows" { + return Ok(()); + } + test::run_bounding_box_csv_repo_test_fully_committed_async(|repo| async move { + let commit = repositories::commits::head_commit(&repo)?; + let workspace_id = UserConfig::identifier()?; + let workspace = repositories::workspaces::create(&repo, &commit, workspace_id, true)?; + let path = list_typed_add_row_test_paths(); + workspaces::data_frames::index(&repo, &workspace, &path).await?; + + let first = workspaces::data_frames::query(&workspace, &path, &page_opts(1, 4, None))?; + assert_eq!( + column_values(&first, "min_x")?, + vec!["101.5", "102.5", "7.0", "19.0"] + ); + + let second = workspaces::data_frames::query(&workspace, &path, &page_opts(2, 4, None))?; + assert_eq!(column_values(&second, "min_x")?, vec!["57.0", "30.5"]); + + Ok(()) + }) + .await + } + #[tokio::test] async fn test_export_with_sql_cannot_read_host_files() -> Result<(), OxenError> { if std::env::consts::OS == "windows" { diff --git a/crates/oxen-py/src/py_workspace_data_frame.rs b/crates/oxen-py/src/py_workspace_data_frame.rs index 55a6f21ea2..573e30118c 100644 --- a/crates/oxen-py/src/py_workspace_data_frame.rs +++ b/crates/oxen-py/src/py_workspace_data_frame.rs @@ -183,30 +183,50 @@ impl PyWorkspaceDataFrame { Ok(result) } - /// Query the data frame using SQL + /// Query the data frame using SQL. Returns every row the query selects. fn sql_query(&self, sql: String) -> Result { - let mut opts = DFOpts::empty(); - opts.sql = Some(sql); + let rows = pyo3_async_runtimes::tokio::get_runtime().block_on(async { + let mut opts = DFOpts::empty(); + opts.sql = Some(sql); + // The read is paginated, and this returns the whole result, so ask for it + // as a single page. Stitching several pages together would join separate + // queries, and a query the read cannot order — one that groups or dedupes + // and carries no `ORDER BY` of its own — need not return rows in the same + // order twice, so its pages would repeat some rows and drop others. + opts.page = Some(1); + // The widest page the read can describe: pagination bounds travel as i64, + // so a larger number is not a larger page, only one the server has to + // narrow before it can use it. + opts.page_size = Some(i64::MAX as usize); - match pyo3_async_runtimes::tokio::get_runtime().block_on(async { - api::client::workspaces::data_frames::get( + let response = api::client::workspaces::data_frames::get( self.workspace.repo.repo()?, &self.workspace.id, &self.path, &opts, ) .await - }) { - Ok(data) => { - // Extract the serde_json::Value from the JsonDataFrameView - let view = data.data_frame.unwrap().view.data; - - // convert json to String - let result: String = serde_json::to_string(&view).unwrap(); - Ok(result) - } - Err(e) => Err(OxenError::basic_str(format!("Failed to query data frame: {e}")).into()), - } + .map_err(|e| OxenError::basic_str(format!("Failed to query data frame: {e}")))?; + + // A result this function can't read is an error, not an empty one: + // returning no rows would silently answer a query with none of its result. + let Some(view) = response.data_frame.map(|df| df.view) else { + return Err(OxenError::basic_str( + "Query returned no data frame. Index the data frame before querying.", + )); + }; + let serde_json::Value::Array(rows) = view.data else { + return Err(OxenError::basic_str(format!( + "Expected an array of rows, got: {}", + view.data + ))); + }; + Ok::<_, OxenError>(rows) + })?; + + let result = serde_json::to_string(&rows) + .map_err(|e| OxenError::basic_str(format!("Could not convert view to json: {e}")))?; + Ok(result) } fn is_nearest_neighbors_enabled(&self, column: String) -> Result { diff --git a/crates/oxen-server/src/controllers/workspaces/data_frames.rs b/crates/oxen-server/src/controllers/workspaces/data_frames.rs index a310464814..cfbe22ffca 100644 --- a/crates/oxen-server/src/controllers/workspaces/data_frames.rs +++ b/crates/oxen-server/src/controllers/workspaces/data_frames.rs @@ -3,6 +3,7 @@ use std::path::{Path, PathBuf}; use crate::errors::OxenHttpError; use crate::helpers::get_repo; use crate::params::{DFOptsQuery, PageNumQuery, app_data, df_opts_query, path_param, query_param}; +use crate::tasks; use actix_web::{HttpRequest, HttpResponse, web}; @@ -136,13 +137,19 @@ pub async fn get( // read returns the whole frame regardless of the page. Skip if slice/row is set. if opts.slice_indices().is_none() { let page = opts.page.unwrap_or(constants::DEFAULT_PAGE_NUM); - let page_size = opts.page_size.unwrap_or(constants::DEFAULT_PAGE_SIZE); + // Both bounds are read back through `slice_indices`, which parses them as + // i64, so a page_size only usize can hold has to come down to a width that + // survives the round trip. + let page_size = opts + .page_size + .unwrap_or(constants::DEFAULT_PAGE_SIZE) + .min(i64::MAX as usize); // page/page_size are clamped to >= 1 when read. Cap start so end = start + page_size // can't overflow and stays strictly greater, preserving slice()'s start < end invariant // even for an absurd page number (which then just yields an empty page). let start = page_size .saturating_mul(page - 1) - .min(usize::MAX - page_size); + .min(i64::MAX as usize - page_size); let end = start + page_size; opts.slice = Some(format!("{start}..{end}")); } @@ -194,10 +201,29 @@ pub async fn get( log::debug!("querying data frame {file_path:?}"); log::debug!("opts: {opts:?}"); - let count = repositories::workspaces::data_frames::count(&workspace, &file_path)?; - - // Query the data frame - let df = repositories::workspaces::data_frames::query(&workspace, &file_path, &opts)?; + // Each read is DuckDB work, and each gets its own offload from the request + // thread rather than sharing one: per docs/async_policy.md the granularity is + // one offload per operation, so the two stay independently convertible. + let count = { + let workspace = workspace.clone(); + let file_path = file_path.clone(); + let opts = opts.clone(); + tasks::spawn_blocking(move || { + repositories::workspaces::data_frames::count_for_query(&workspace, &file_path, &opts) + }) + .await + .map_err(OxenError::from)?? + }; + let df = { + let workspace = workspace.clone(); + let file_path = file_path.clone(); + let opts = opts.clone(); + tasks::spawn_blocking(move || { + repositories::workspaces::data_frames::query(&workspace, &file_path, &opts) + }) + .await + .map_err(OxenError::from)?? + }; let Some(mut df_schema) = repositories::data_frames::schemas::get_by_path(&repo, &workspace.commit, &file_path)? @@ -713,7 +739,8 @@ mod tests { use actix_web::{App, web}; use liboxen::core::db::data_frames::df_db::with_df_db_manager; use liboxen::error::OxenError; - use liboxen::model::Schema; + use liboxen::model::{LocalRepository, Schema, Workspace}; + use liboxen::opts::DFOpts; use liboxen::repositories; use liboxen::util; use liboxen::view::json_data_frame_view::WorkspaceJsonDataFrameViewResponse; @@ -1371,6 +1398,16 @@ mod tests { Ok(()) } + /// The query string a page request carries, built the way the client builds + /// it so the handler parses what it parses in production. + fn page_query(page: usize, page_size: usize, sql: Option<&str>) -> DFOpts { + let mut opts = DFOpts::empty(); + opts.page = Some(page); + opts.page_size = Some(page_size); + opts.sql = sql.map(str::to_string); + opts + } + /// GET a page of a workspace data frame through the `get` handler. async fn get_data_frame_page( sync_dir: &std::path::Path, @@ -1378,8 +1415,7 @@ mod tests { repo_name: &str, workspace_id: &str, file_path: &str, - page: usize, - page_size: usize, + query: DFOpts, ) -> WorkspaceJsonDataFrameViewResponse { let app = actix_web::test::init_service( App::new() @@ -1392,7 +1428,8 @@ mod tests { .await; let uri = format!( - "/oxen/{namespace}/{repo_name}/workspaces/{workspace_id}/data_frames/resource/{file_path}?page={page}&page_size={page_size}" + "/oxen/{namespace}/{repo_name}/workspaces/{workspace_id}/data_frames/resource/{file_path}?{}", + query.to_http_query_params() ); let req = actix_web::test::TestRequest::get().uri(&uri).to_request(); let resp = actix_web::test::call_service(&app, req).await; @@ -1440,20 +1477,16 @@ mod tests { .collect() } - /// An unindexed workspace data frame larger than the requested `page_size` must paginate: - /// each page returns at most `page_size` rows and an out-of-range page returns zero rows so - /// pagination terminates. Regression test for the unindexed branch returning the full frame on - /// every page. Also asserts the indexed branch paginates identically so the two paths stay in - /// lockstep. - #[actix_web::test] - async fn test_get_unindexed_data_frame_paginates() -> Result<(), OxenError> { - liboxen::test::init_test_env(); - let sync_dir = test::get_sync_dir()?; - let namespace = "Testing-Namespace"; - let repo_name = "Testing-Name"; - let repo = test::create_local_repo(&sync_dir, namespace, repo_name)?; + /// A committed `data/history.csv` and a workspace on that commit, not yet + /// indexed. The CSV holds 25 rows — larger than the page_size of 10 the + /// pagination tests request, so their pages are real pages. + async fn commit_history_csv_workspace( + sync_dir: &std::path::Path, + namespace: &str, + repo_name: &str, + ) -> Result<(LocalRepository, Workspace, String), OxenError> { + let repo = test::create_local_repo(sync_dir, namespace, repo_name)?; - // Commit a CSV with 25 rows — larger than the page_size of 10 we request below. let csv_dir = repo.path.join("data"); util::fs::create_dir_all(&csv_dir)?; let csv_path = csv_dir.join("history.csv"); @@ -1465,13 +1498,28 @@ mod tests { repositories::add(&repo, &csv_path).await?; let commit = repositories::commit(&repo, "Add 25-row CSV")?; - let file_path = "data/history.csv"; - - // A workspace created from the commit but not yet indexed, so the GET handler takes the - // unindexed read path. We index this same workspace later to exercise the indexed branch - // (a commit can only have one non-editable workspace, so we reuse it). let workspace_id = uuid::Uuid::new_v4().to_string(); let workspace = repositories::workspaces::create(&repo, &commit, &workspace_id, false)?; + Ok((repo, workspace, workspace_id)) + } + + /// An unindexed workspace data frame larger than the requested `page_size` must paginate: + /// each page returns at most `page_size` rows and an out-of-range page returns zero rows so + /// pagination terminates. Regression test for the unindexed branch returning the full frame on + /// every page. Also asserts the indexed branch paginates identically so the two paths stay in + /// lockstep. + #[actix_web::test] + async fn test_get_unindexed_data_frame_paginates() -> Result<(), OxenError> { + liboxen::test::init_test_env(); + let sync_dir = test::get_sync_dir()?; + let namespace = "Testing-Namespace"; + let repo_name = "Testing-Name"; + // The workspace starts unindexed, so the GET handler takes the unindexed read path. We + // index this same workspace later to exercise the indexed branch (a commit can only have + // one non-editable workspace, so we reuse it). + let (repo, workspace, workspace_id) = + commit_history_csv_workspace(&sync_dir, namespace, repo_name).await?; + let file_path = "data/history.csv"; // Unindexed branch: page 1 of 10 → exactly 10 rows, and total_pages/total_entries // reflect the full frame. @@ -1481,8 +1529,7 @@ mod tests { repo_name, &workspace_id, file_path, - 1, - 10, + page_query(1, 10, None), ) .await; assert!(!page_1.is_indexed); @@ -1499,8 +1546,7 @@ mod tests { repo_name, &workspace_id, file_path, - 3, - 10, + page_query(3, 10, None), ) .await; assert_eq!(page_row_count(&page_3), 5); @@ -1513,8 +1559,7 @@ mod tests { repo_name, &workspace_id, file_path, - 4, - 10, + page_query(4, 10, None), ) .await; assert_eq!(page_row_count(&page_4), 0); @@ -1530,8 +1575,7 @@ mod tests { repo_name, &workspace_id, file_path, - 0, - 10, + page_query(0, 10, None), ) .await; assert_eq!(page_row_count(&page_zero), 10); @@ -1542,8 +1586,7 @@ mod tests { repo_name, &workspace_id, file_path, - 1, - 0, + page_query(1, 0, None), ) .await; assert_eq!(page_row_count(&zero_page_size), 1); @@ -1580,8 +1623,7 @@ mod tests { repo_name, &workspace_id, file_path, - page, - 10, + page_query(page, 10, None), ) .await; assert!(indexed_page.is_indexed); @@ -1596,4 +1638,111 @@ mod tests { test::cleanup_repo_and_sync_dir(repo, &sync_dir)?; Ok(()) } + + /// A client that wants a whole result asks for it as one maximal page. The + /// widest page a request can name has to read as "all of them" on either + /// branch, rather than overflowing the slice or exceeding what a SQL `LIMIT` + /// accepts. + #[actix_web::test] + async fn test_get_data_frame_page_holding_every_row() -> Result<(), OxenError> { + liboxen::test::init_test_env(); + let sync_dir = test::get_sync_dir()?; + let namespace = "Testing-Namespace"; + let repo_name = "Testing-Widest-Page"; + let (repo, workspace, workspace_id) = + commit_history_csv_workspace(&sync_dir, namespace, repo_name).await?; + let file_path = "data/history.csv"; + + let widest = || page_query(1, usize::MAX, Some("SELECT * FROM df")); + let response = get_data_frame_page( + &sync_dir, + namespace, + repo_name, + &workspace_id, + file_path, + widest(), + ) + .await; + assert!( + !response.is_indexed, + "sql is ignored until the frame is indexed" + ); + assert_eq!(page_row_count(&response), 25); + + repositories::workspaces::data_frames::index( + &repo, + &workspace, + std::path::Path::new(file_path), + ) + .await?; + let response = get_data_frame_page( + &sync_dir, + namespace, + repo_name, + &workspace_id, + file_path, + widest(), + ) + .await; + assert!(response.is_indexed); + assert_eq!(page_row_count(&response), 25); + + drop(workspace); + test::cleanup_repo_and_sync_dir(repo, &sync_dir)?; + Ok(()) + } + + /// A `sql` query is paginated like any other read, and the pagination the + /// response reports describes the rows the query selects rather than the + /// whole frame — otherwise a client paging by `total_pages` walks past the + /// end of the result. + #[actix_web::test] + async fn test_get_data_frame_with_sql_paginates() -> Result<(), OxenError> { + liboxen::test::init_test_env(); + let sync_dir = test::get_sync_dir()?; + let namespace = "Testing-Namespace"; + let repo_name = "Testing-SQL-Pagination"; + let (repo, workspace, workspace_id) = + commit_history_csv_workspace(&sync_dir, namespace, repo_name).await?; + let file_path = "data/history.csv"; + repositories::workspaces::data_frames::index( + &repo, + &workspace, + std::path::Path::new(file_path), + ) + .await?; + + // 15 of the 25 rows match, so 2 pages of 10 and nothing after them. + let sql = "SELECT * FROM df WHERE id >= 10"; + let expectations: [(usize, Vec); 3] = [ + (1, (10..20).collect()), + (2, (20..25).collect()), + (3, Vec::new()), + ]; + for (page, expected_ids) in expectations { + let response = get_data_frame_page( + &sync_dir, + namespace, + repo_name, + &workspace_id, + file_path, + page_query(page, 10, Some(sql)), + ) + .await; + assert!(response.is_indexed); + assert_eq!( + page_row_count(&response), + expected_ids.len(), + "page {page} must hold only its own rows" + ); + assert_eq!(page_ids(&response), expected_ids); + let pagination = &response.data_frame.as_ref().unwrap().view.pagination; + assert_eq!(pagination.total_entries, 15); + assert_eq!(pagination.total_pages, 2); + } + + drop(workspace); + test::cleanup_repo_and_sync_dir(repo, &sync_dir)?; + Ok(()) + } }