From 4726824e37773ed5d9ce624c35ae5119f27d0a80 Mon Sep 17 00:00:00 2001 From: Joshua Elliott Date: Fri, 7 Aug 2026 14:13:07 -0600 Subject: [PATCH 1/5] Apply the requested page when a workspace data frame read carries sql MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `repositories::workspaces::data_frames::query` dropped its `DFOpts` on the branch that runs caller-supplied SQL, passing `None` to `sql::query_df` where the other branch passes `Some(opts)`. `opts` is what carries page/page_size, so a read with a `sql` param returned the whole result set for every page. `prepare_sql` now composes the page onto the statement instead of appending to it, since caller SQL can carry its own ORDER BY, LIMIT, or OFFSET: - A statement that bounds its own extent is paged through a subquery, so the two bounds nest rather than collide, and no sort of ours reorders the rows it already picked. - Otherwise the statement's own ORDER BY wins over `opts.sort_by`. - A paginated statement left with no order at all gets `ORDER BY _oxen_row_id` wherever that column binds — the same stable order the non-sql branch builds, so an edited row stays on its page. Statements that aggregate, dedupe, or read a derived table can't bind it and page in whatever order they produce. The GET handler counts what the query selects rather than what the frame holds, so `total_pages` and `total_entries` describe the same row set the page came out of. `PyWorkspaceDataFrame::sql_query` still returns the full result, now by reading the pages. It stops on a short page, which is both the end of the result and what a server too old to paginate the query answers page 1 with. --- .../liboxen/src/core/db/data_frames/df_db.rs | 141 +++++++++++- .../liboxen/src/core/v_latest/data_frames.rs | 2 + .../repositories/workspaces/data_frames.rs | 216 +++++++++++++++++- crates/oxen-py/src/py_workspace_data_frame.rs | 74 ++++-- .../src/controllers/workspaces/data_frames.rs | 160 ++++++++++--- 5 files changed, 522 insertions(+), 71 deletions(-) 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..128f59fa8c 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,16 @@ pub fn count(conn: &duckdb::Connection, table_name: &str) -> Result Result { + let count = conn.query_row( + &format!("SELECT count(*) FROM ({sql}) AS _oxen_count"), + [], + |row| row.get(0), + )?; + Ok(count) +} + /// Query number of rows in a table. pub fn count_where( conn: &duckdb::Connection, @@ -565,6 +575,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, @@ -575,23 +599,118 @@ pub fn prepare_sql( let mut sql = add_special_columns(conn, stmt)?; - 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))); + // 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 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..7e1908d874 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,203 @@ 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 + } + + /// 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..511c650a72 100644 --- a/crates/oxen-py/src/py_workspace_data_frame.rs +++ b/crates/oxen-py/src/py_workspace_data_frame.rs @@ -183,30 +183,62 @@ impl PyWorkspaceDataFrame { Ok(result) } - /// Query the data frame using SQL + /// Query the data frame using SQL. Returns every row the query selects, + /// reading the paginated endpoint a page at a time. fn sql_query(&self, sql: String) -> Result { - let mut opts = DFOpts::empty(); - opts.sql = Some(sql); + // The whole result is held in memory either way, so this bounds only the + // size of a single response — keep it high enough that most queries come + // back in one request rather than a round trip per thousand rows. + const PAGE_SIZE: usize = 10_000; - match pyo3_async_runtimes::tokio::get_runtime().block_on(async { - 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) + let rows = pyo3_async_runtimes::tokio::get_runtime().block_on(async { + let mut opts = DFOpts::empty(); + opts.sql = Some(sql); + opts.page_size = Some(PAGE_SIZE); + + let mut rows = vec![]; + let mut page_num = 1; + loop { + opts.page = Some(page_num); + let response = api::client::workspaces::data_frames::get( + self.workspace.repo.repo()?, + &self.workspace.id, + &self.path, + &opts, + ) + .await + .map_err(|e| OxenError::basic_str(format!("Failed to query data frame: {e}")))?; + + // A page this function can't read is an error, not the end of the + // result: returning the rows collected so far would silently answer + // a query with part 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(page_rows) = view.data else { + return Err(OxenError::basic_str(format!( + "Expected a page of rows, got: {}", + view.data + ))); + }; + let page_len = page_rows.len(); + rows.extend(page_rows); + // A short page ends the result set. It is also what a server too + // old to paginate this query answers page 1 with, so stopping + // here keeps such a server from re-serving the same rows. + if page_len < PAGE_SIZE || page_num >= view.pagination.total_pages { + break; + } + page_num += 1; } - Err(e) => Err(OxenError::basic_str(format!("Failed to query data frame: {e}")).into()), - } + 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..24f5b8c77b 100644 --- a/crates/oxen-server/src/controllers/workspaces/data_frames.rs +++ b/crates/oxen-server/src/controllers/workspaces/data_frames.rs @@ -3,10 +3,12 @@ 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}; use liboxen::constants::{self, TABLE_NAME}; +use liboxen::core::db::data_frames::DataFrameError; use liboxen::core::db::data_frames::df_db::with_df_db_manager; use liboxen::core::db::data_frames::workspace_df_db::schema_without_oxen_cols; use liboxen::core::repo_locks; @@ -194,10 +196,22 @@ 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)?; + // Counting and reading the page are both DuckDB work, so they run as one unit + // off the request thread. + let (count, df) = { + let workspace = workspace.clone(); + let file_path = file_path.clone(); + let opts = opts.clone(); + tasks::spawn_blocking(move || { + let count = repositories::workspaces::data_frames::count_for_query( + &workspace, &file_path, &opts, + )?; + let df = repositories::workspaces::data_frames::query(&workspace, &file_path, &opts)?; + Ok::<_, DataFrameError>((count, df)) + }) + .await + .map_err(OxenError::from)?? + }; let Some(mut df_schema) = repositories::data_frames::schemas::get_by_path(&repo, &workspace.commit, &file_path)? @@ -713,7 +727,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 +1386,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 +1403,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 +1416,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 +1465,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 +1486,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 +1517,7 @@ mod tests { repo_name, &workspace_id, file_path, - 1, - 10, + page_query(1, 10, None), ) .await; assert!(!page_1.is_indexed); @@ -1499,8 +1534,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 +1547,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 +1563,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 +1574,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 +1611,7 @@ mod tests { repo_name, &workspace_id, file_path, - page, - 10, + page_query(page, 10, None), ) .await; assert!(indexed_page.is_indexed); @@ -1596,4 +1626,58 @@ mod tests { 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(()) + } } From 1f4d9ac546d43f6ba2f766941e22b08c4c18f5c7 Mon Sep 17 00:00:00 2001 From: Joshua Elliott Date: Wed, 12 Aug 2026 08:53:25 -0600 Subject: [PATCH 2/5] Strip a terminator or comment before composing a page onto SQL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `prepare_sql` and `count_sql` place a caller's statement somewhere other than the end of the text: before an appended `ORDER BY` or `LIMIT`, or inside a count's derived table. A trailing `;` or `--` comment is harmless at the end of a statement and changes what follows it in the middle — the terminator makes the composed statement a syntax error, and a comment swallows the bounds so the read returns the whole frame instead of one page. `add_special_columns` re-renders through the parser only when it injects `_oxen_id`, and returns the statement as written otherwise: for a DISTINCT, for a projection that isn't a subset of the source schema, and for one that already selects `_oxen_id`, which `SELECT *` does because the column is really there. So the text reaching composition is often the caller's own. Route both composition sites through `composable` to render a single parsed statement instead. Text that doesn't parse as exactly one statement passes through unchanged, for DuckDB to reject as it did before. --- .../liboxen/src/core/db/data_frames/df_db.rs | 21 ++++++++- .../repositories/workspaces/data_frames.rs | 45 +++++++++++++++++++ 2 files changed, 65 insertions(+), 1 deletion(-) 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 128f59fa8c..39ed9c0dd9 100644 --- a/crates/liboxen/src/core/db/data_frames/df_db.rs +++ b/crates/liboxen/src/core/db/data_frames/df_db.rs @@ -519,6 +519,7 @@ 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"), [], @@ -527,6 +528,21 @@ pub fn count_sql(conn: &duckdb::Connection, sql: &str) -> Result 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, @@ -597,7 +613,10 @@ 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() { diff --git a/crates/liboxen/src/repositories/workspaces/data_frames.rs b/crates/liboxen/src/repositories/workspaces/data_frames.rs index 7e1908d874..c30f81396c 100644 --- a/crates/liboxen/src/repositories/workspaces/data_frames.rs +++ b/crates/liboxen/src/repositories/workspaces/data_frames.rs @@ -2609,6 +2609,51 @@ mod tests { .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> { From 3e67d7daec12f0000b57cd846cbd878bc2f5e152 Mon Sep 17 00:00:00 2001 From: Joshua Elliott Date: Wed, 12 Aug 2026 08:53:42 -0600 Subject: [PATCH 3/5] Offload the page read and its count separately The workspace data frame read moved both DuckDB calls off the request thread in one closure. docs/async_policy.md puts the granularity at one offload per operation, and specifically not one bespoke closure per handler: sharing a closure means neither call can overlap with the other or be converted to an async API on its own. Give each its own offload. Both already open their own connection, so nothing was shared but the hop. --- .../src/controllers/workspaces/data_frames.rs | 24 ++++++++++++------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/crates/oxen-server/src/controllers/workspaces/data_frames.rs b/crates/oxen-server/src/controllers/workspaces/data_frames.rs index 24f5b8c77b..acba11a8c2 100644 --- a/crates/oxen-server/src/controllers/workspaces/data_frames.rs +++ b/crates/oxen-server/src/controllers/workspaces/data_frames.rs @@ -8,7 +8,6 @@ use crate::tasks; use actix_web::{HttpRequest, HttpResponse, web}; use liboxen::constants::{self, TABLE_NAME}; -use liboxen::core::db::data_frames::DataFrameError; use liboxen::core::db::data_frames::df_db::with_df_db_manager; use liboxen::core::db::data_frames::workspace_df_db::schema_without_oxen_cols; use liboxen::core::repo_locks; @@ -196,18 +195,25 @@ pub async fn get( log::debug!("querying data frame {file_path:?}"); log::debug!("opts: {opts:?}"); - // Counting and reading the page are both DuckDB work, so they run as one unit - // off the request thread. - let (count, df) = { + // 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 || { - let count = repositories::workspaces::data_frames::count_for_query( - &workspace, &file_path, &opts, - )?; - let df = repositories::workspaces::data_frames::query(&workspace, &file_path, &opts)?; - Ok::<_, DataFrameError>((count, df)) + 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)?? From b5bb2d879c7de9d4682f8d3c1350ec5ea0503f21 Mon Sep 17 00:00:00 2001 From: Joshua Elliott Date: Wed, 12 Aug 2026 09:32:55 -0600 Subject: [PATCH 4/5] Keep an unrepresentable page size from panicking the read The unindexed branch derives a slice from the requested page, and `slice_indices` reads those bounds back as i64. A page_size only usize can hold overflows that parse, and the parse panics rather than erroring, so a request naming a page wider than i64 takes the server down to a 500 instead of reading a page. Narrow page_size to what the bounds can carry when deriving the slice. The widest page a request can name then reads as the whole frame, which is what a client asking for one oversized page means by it. `slice_indices` still panics on bounds it cannot parse, reachable through `slice` directly; that expect is worth removing on its own terms. --- .../src/controllers/workspaces/data_frames.rs | 63 ++++++++++++++++++- 1 file changed, 61 insertions(+), 2 deletions(-) diff --git a/crates/oxen-server/src/controllers/workspaces/data_frames.rs b/crates/oxen-server/src/controllers/workspaces/data_frames.rs index acba11a8c2..cfbe22ffca 100644 --- a/crates/oxen-server/src/controllers/workspaces/data_frames.rs +++ b/crates/oxen-server/src/controllers/workspaces/data_frames.rs @@ -137,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}")); } @@ -1633,6 +1639,59 @@ mod tests { 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 From 737bf22f5fd8c4e211a420b3cd1ca2dbc03448b1 Mon Sep 17 00:00:00 2001 From: Joshua Elliott Date: Wed, 12 Aug 2026 09:33:13 -0600 Subject: [PATCH 5/5] Read a SQL query's whole result as one page MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `sql_query` returns every row a query selects, and read the paginated endpoint a page at a time to collect them. Pages of one query are not pages of one result: the read orders a query by `_oxen_row_id` only where that column resolves against it, so a query that groups or dedupes and carries no `ORDER BY` of its own need not return rows in the same order twice. Stitching its pages together repeats some rows and drops others, and answers the query with neither an error nor its result. Ask for the whole result as a single page instead, which is what this returns either way — the Python `query(sql=...)` ignores a page number, and `get_embeddings` reads every row a query matches. --- crates/oxen-py/src/py_workspace_data_frame.rs | 80 ++++++++----------- 1 file changed, 34 insertions(+), 46 deletions(-) diff --git a/crates/oxen-py/src/py_workspace_data_frame.rs b/crates/oxen-py/src/py_workspace_data_frame.rs index 511c650a72..573e30118c 100644 --- a/crates/oxen-py/src/py_workspace_data_frame.rs +++ b/crates/oxen-py/src/py_workspace_data_frame.rs @@ -183,56 +183,44 @@ impl PyWorkspaceDataFrame { Ok(result) } - /// Query the data frame using SQL. Returns every row the query selects, - /// reading the paginated endpoint a page at a time. + /// Query the data frame using SQL. Returns every row the query selects. fn sql_query(&self, sql: String) -> Result { - // The whole result is held in memory either way, so this bounds only the - // size of a single response — keep it high enough that most queries come - // back in one request rather than a round trip per thousand rows. - const PAGE_SIZE: usize = 10_000; - let rows = pyo3_async_runtimes::tokio::get_runtime().block_on(async { let mut opts = DFOpts::empty(); opts.sql = Some(sql); - opts.page_size = Some(PAGE_SIZE); - - let mut rows = vec![]; - let mut page_num = 1; - loop { - opts.page = Some(page_num); - let response = api::client::workspaces::data_frames::get( - self.workspace.repo.repo()?, - &self.workspace.id, - &self.path, - &opts, - ) - .await - .map_err(|e| OxenError::basic_str(format!("Failed to query data frame: {e}")))?; - - // A page this function can't read is an error, not the end of the - // result: returning the rows collected so far would silently answer - // a query with part 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(page_rows) = view.data else { - return Err(OxenError::basic_str(format!( - "Expected a page of rows, got: {}", - view.data - ))); - }; - let page_len = page_rows.len(); - rows.extend(page_rows); - // A short page ends the result set. It is also what a server too - // old to paginate this query answers page 1 with, so stopping - // here keeps such a server from re-serving the same rows. - if page_len < PAGE_SIZE || page_num >= view.pagination.total_pages { - break; - } - page_num += 1; - } + // 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); + + let response = api::client::workspaces::data_frames::get( + self.workspace.repo.repo()?, + &self.workspace.id, + &self.path, + &opts, + ) + .await + .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) })?;