diff --git a/benchmarks/full-text/README.md b/benchmarks/full-text/README.md new file mode 100644 index 000000000..390d1d4e9 --- /dev/null +++ b/benchmarks/full-text/README.md @@ -0,0 +1,59 @@ + + +# Full-text Top-K benchmark + +Run the production full-text scan/read path against a temporary local filesystem +catalog. Each mode uses the same data, index archives and snapshot. The program +checks exact row IDs and scores on every run, warms both modes, alternates their +execution order, and emits JSON with every sample plus p50 and p95 latency. + +```bash +cargo run --release --locked -p paimon --features internal-api \ + --example full_text_top_k_benchmark -- \ + 10000 10 10 2500 +``` + +Arguments are rows, measured iterations per mode, K, and rows per index shard. +All must be positive. Increase rows and iterations for a representative workload. +The `skewed` corpus gives each shard distinct high scores followed by many common +low-score matches. The `tied` corpus has identical text in every row and exercises +the exhaustive fallback required by Paimon's row-ID tie-break. + +The benchmark uses the `internal-api` feature's exhaustive reference reader for +the baseline. Production searches select the optimization automatically without +a table option. The fast path is limited to FAST-mode global full-text +reads over disjoint index row ranges, with no raw-data sources, deletion vectors +or later deletion exclusions. Native search still evaluates the complete query +DSL and applies its allow-list before selecting candidates. A shard requests +K+1 hits and retains K only if the score boundary is strict and the selected row +IDs are distinct. Boundary ties fall back to the original full candidate limit. +FULL/DETAIL, overlapping ranges, and deletion/raw-fallback paths remain exhaustive. +Primary-key and raw temporary-index searches keep their existing behavior. + +This is a synthetic, warm-cache local benchmark, not an object-store, cold-cache, +Lance, or whole-system memory comparison. It reports no cross-shard BM25 corpus +normalization change. Record the revision, compiler, CPU, RAM, OS and command with +results. The JSON field `optimized_build` must be true for performance reports. + +## Local sample + +A [sample run with raw timings](sample-results.md) records 100,000 matching rows, +four shards, K=10 and 20 iterations on an Apple M4 Pro. It illustrates this +synthetic workload only; use the protocol above to evaluate other workloads. diff --git a/benchmarks/full-text/sample-results.md b/benchmarks/full-text/sample-results.md new file mode 100644 index 000000000..50c8e91e9 --- /dev/null +++ b/benchmarks/full-text/sample-results.md @@ -0,0 +1,173 @@ + + +# Local full-text Top-K sample + +- Date: 2026-10-28. +- Source: `87dc89ca`, before moving the exhaustive comparison to the internal + benchmark reader. The raw timings below describe that revision. +- Compiler: Rust 1.98.0, Cargo release profile. +- Machine: Apple M4 Pro, 48 GiB RAM, macOS 27.0.1. +- Storage: temporary local filesystem; both modes warmed before measurement. +- Workload: 100,000 matching documents, four shards, K=10; 20 samples per mode + with alternating order. Every run asserts identical IDs and scores. + +```bash +cargo run --release --locked -p paimon --features internal-api \ + --example full_text_top_k_benchmark -- \ + 100000 20 10 25000 +``` + +The command above uses the current internal benchmark entry; the original +measurement selected the same exhaustive and certified collectors via a table option. + +| Corpus | Exhaustive p50 | Certified p50 | Exhaustive p95 | Certified p95 | +| --- | ---: | ---: | ---: | ---: | +| skewed | 65.680 ms | 1.300 ms | 71.194 ms | 1.584 ms | +| tied | 61.498 ms | 59.544 ms | 65.693 ms | 65.988 ms | + +The skewed corpus has a strict per-shard score cutoff. The tied corpus requires +exhaustive fallback to preserve row-ID ordering. These are warm-cache synthetic +local results, not a general speedup claim or a comparison with Lance. The +benchmark does not measure peak RSS, cloud requests, or cold-cache behavior. + +Raw JSON output (milliseconds): + +```json +{ + "cache": "warm; alternating path order", + "certified_ms": [ + 1.493625, + 1.3470419999999999, + 1.584334, + 1.265542, + 1.442334, + 1.35375, + 1.527584, + 1.574292, + 1.428708, + 1.104042, + 1.428958, + 0.9946670000000001, + 1.1383750000000001, + 1.020333, + 2.98675, + 1.178459, + 1.2998750000000001, + 1.14975, + 1.281458, + 1.223833 + ], + "certified_p50_ms": 1.2998750000000001, + "certified_p95_ms": 1.584334, + "corpus": "skewed", + "exhaustive_ms": [ + 63.123042000000005, + 64.412542, + 65.869167, + 61.668333, + 73.404917, + 67.536125, + 62.193084000000006, + 67.91274999999999, + 65.6795, + 66.166958, + 65.237833, + 62.372875, + 71.194334, + 64.104334, + 65.31225, + 65.99804099999999, + 68.3735, + 68.70008299999999, + 63.100249999999996, + 70.09375 + ], + "exhaustive_p50_ms": 65.6795, + "exhaustive_p95_ms": 71.194334, + "iterations": 20, + "k": 10, + "optimized_build": true, + "rows": 100000, + "shards": 4, + "storage": "local temporary filesystem" +} +``` + +```json +{ + "cache": "warm; alternating path order", + "certified_ms": [ + 60.786959, + 65.987917, + 63.890249999999995, + 66.674334, + 64.062834, + 58.520042000000004, + 64.140084, + 59.344916999999995, + 59.02675, + 59.543541, + 56.94875, + 59.583791999999995, + 58.512415999999995, + 58.018041, + 59.953208000000004, + 56.848166, + 61.324375, + 60.756958, + 56.141957999999995, + 55.769375000000004 + ], + "certified_p50_ms": 59.543541, + "certified_p95_ms": 65.987917, + "corpus": "tied", + "exhaustive_ms": [ + 61.498125, + 61.992708, + 64.92604200000001, + 63.23775, + 60.250625, + 65.69324999999999, + 63.07212500000001, + 69.43341699999999, + 62.332499999999996, + 59.03325, + 59.881334, + 56.568625, + 61.359958, + 58.073375000000006, + 61.913416999999995, + 61.050541, + 62.432625, + 61.058542, + 57.116375, + 63.045041 + ], + "exhaustive_p50_ms": 61.498125, + "exhaustive_p95_ms": 65.69324999999999, + "iterations": 20, + "k": 10, + "optimized_build": true, + "rows": 100000, + "shards": 4, + "storage": "local temporary filesystem" +} +``` + diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml index 74169546b..b655d2607 100644 --- a/crates/paimon/Cargo.toml +++ b/crates/paimon/Cargo.toml @@ -172,3 +172,7 @@ cc = "1" [dev-dependencies] axum = { version = "0.7", features = ["macros", "tokio", "http1", "http2"] } libc = "0.2" + +[[example]] +name = "full_text_top_k_benchmark" +required-features = ["internal-api"] diff --git a/crates/paimon/examples/full_text_top_k_benchmark.rs b/crates/paimon/examples/full_text_top_k_benchmark.rs new file mode 100644 index 000000000..1194180c6 --- /dev/null +++ b/crates/paimon/examples/full_text_top_k_benchmark.rs @@ -0,0 +1,162 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Compare certified and exhaustive FTS on the same local table and snapshot. +//! cargo run --release -p paimon --features internal-api --example full_text_top_k_benchmark -- [rows] [iterations] [k] [rows-per-shard] + +use arrow_array::{RecordBatch, StringArray}; +use arrow_schema::{DataType as ArrowType, Field, Schema as ArrowSchema}; +use paimon::catalog::Identifier; +use paimon::spec::{DataType, Schema, VarCharType}; +use paimon::table::Table; +use paimon::vector_search::ScoredRowIds; +use paimon::{CatalogFactory, Options}; +use std::collections::HashMap; +use std::error::Error; +use std::sync::Arc; +use std::time::Instant; + +async fn search(table: &Table, enabled: bool, k: usize) -> paimon::Result { + let mut builder = table.new_full_text_search_builder(); + builder + .with_text_column("body") + .with_query_text("paimon") + .with_limit(k); + let reader = builder.new_read()?; + let plan = builder.new_scan()?.scan().await?; + let result = if enabled { + reader.read(plan).await? + } else { + paimon::table::internal::read_full_text_exhaustive(&reader, plan).await? + }; + result.into_row_ids() +} + +fn percentile(samples: &mut [f64], percentile: usize) -> f64 { + samples.sort_by(f64::total_cmp); + samples[(samples.len() * percentile).div_ceil(100).saturating_sub(1)] +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let defaults = [20_000, 10, 10, 5_000]; + let mut args = std::env::args().skip(1); + let mut config = defaults; + for (value, default) in config.iter_mut().zip(defaults) { + *value = args + .next() + .map(|s| s.parse::()) + .transpose()? + .unwrap_or(default); + if *value == 0 { + return Err("all arguments must be positive".into()); + } + } + if args.next().is_some() { + return Err("expected rows iterations k rows-per-shard".into()); + } + let [rows, iterations, k, shard_rows] = config; + let directory = tempfile::tempdir()?; + let catalog = CatalogFactory::create(Options::from_map(HashMap::from([( + "warehouse".into(), + directory.path().to_string_lossy().into_owned(), + )]))) + .await?; + catalog + .create_database("bench", false, HashMap::new()) + .await?; + for tied in [false, true] { + let name = if tied { "tied" } else { "skewed" }; + let identifier = Identifier::new("bench", name); + let schema = Schema::builder() + .column("body", DataType::VarChar(VarCharType::string_type())) + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .option("global-index.enabled", "true") + .option("global-index.row-count-per-shard", shard_rows.to_string()) + .build()?; + catalog.create_table(&identifier, schema, false).await?; + let table = catalog.get_table(&identifier).await?; + let builder = table.new_write_builder(); + let mut writer = builder.new_write()?; + let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new( + "body", + ArrowType::Utf8, + true, + )])); + for start in (0..rows).step_by(4096) { + let texts = (start..(start + 4096).min(rows)) + .map(|row| { + let position = row % shard_rows; + if tied { + "paimon".to_string() + } else if position < 64 { + "paimon ".repeat(64 - position) + } else { + "paimon padding padding padding padding padding padding padding".to_string() + } + }) + .collect::>(); + writer + .write_arrow_batch(&RecordBatch::try_new( + arrow_schema.clone(), + vec![Arc::new(StringArray::from(texts))], + )?) + .await?; + } + builder + .try_new_commit()? + .commit(writer.prepare_commit().await?) + .await?; + let shards = table + .new_global_index_build_builder() + .with_index_type("full-text") + .with_index_column("body") + .execute() + .await?; + // Warm both paths and assert IDs and scores before collecting timings. + let expected = search(&table, false, k).await?; + let actual = search(&table, true, k).await?; + assert_eq!(actual.row_ids, expected.row_ids); + assert_eq!(actual.scores, expected.scores); + let mut timings = [Vec::new(), Vec::new()]; + for iteration in 0..iterations { + for mode in [iteration % 2, 1 - iteration % 2] { + let started = Instant::now(); + let actual = search(&table, mode == 1, k).await?; + let elapsed = started.elapsed().as_secs_f64() * 1000.0; + assert_eq!(actual.row_ids, expected.row_ids); + assert_eq!(actual.scores, expected.scores); + timings[mode].push(elapsed); + } + } + println!( + "{}", + serde_json::json!({ + "corpus": name, "rows": rows, "k": k, "shards": shards, + "iterations": iterations, "optimized_build": !cfg!(debug_assertions), + "cache": "warm; alternating path order", "storage": "local temporary filesystem", + "exhaustive_ms": timings[0], "certified_ms": timings[1], + "exhaustive_p50_ms": percentile(&mut timings[0], 50), + "exhaustive_p95_ms": percentile(&mut timings[0], 95), + "certified_p50_ms": percentile(&mut timings[1], 50), + "certified_p95_ms": percentile(&mut timings[1], 95), + }) + ); + } + Ok(()) +} diff --git a/crates/paimon/src/spec/core_options.rs b/crates/paimon/src/spec/core_options.rs index 6d2d20131..89071ed7c 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -1195,20 +1195,21 @@ impl<'a> CoreOptions<'a> { Ok(value) } - /// Maximum retained serialized sort keys, excluding the active Arrow batch, - /// format-writer buffers and the two keys used by a disk merge. - pub fn sorted_index_sort_memory(&self) -> crate::Result { + /// Write buffer budget, matching Java `CoreOptions.WRITE_BUFFER_SIZE`. + /// Sorted global-index builds use it for retained serialized sort keys, + /// excluding the active Arrow batch, writer buffers and disk-merge keys. + pub fn write_buffer_size(&self) -> crate::Result { let raw = self .options - .get("sorted-index.sort-memory") + .get("write-buffer-size") .map(String::as_str) - .unwrap_or("64mb"); + .unwrap_or("256mb"); let bytes = parse_memory_size(raw) .and_then(|size| usize::try_from(size).ok()) .filter(|size| *size > 0); bytes.ok_or_else(|| crate::Error::ConfigInvalid { message: format!( - "Option 'sorted-index.sort-memory' must be a positive memory size, got: {raw}" + "Option 'write-buffer-size' must be a positive memory size, got: {raw}" ), }) } diff --git a/crates/paimon/src/table/full_text_index_adapter.rs b/crates/paimon/src/table/full_text_index_adapter.rs index 680d1c3da..42ae23406 100644 --- a/crates/paimon/src/table/full_text_index_adapter.rs +++ b/crates/paimon/src/table/full_text_index_adapter.rs @@ -28,12 +28,26 @@ use std::sync::{mpsc, Arc, Mutex}; const FULL_TEXT_RANGE_READ_CONCURRENCY: usize = 8; +#[cfg(test)] pub(crate) async fn search_full_text_index( reader: Box, file_name: String, query: String, limit: usize, include_row_ids: Option, +) -> crate::Result { + search_full_text_index_top_k(reader, file_name, query, limit, include_row_ids, None).await +} + +/// The caller may request bounded candidates only when no later stage removes, +/// replaces or deduplicates hits across shards. Ties at the boundary fall back. +pub(crate) async fn search_full_text_index_top_k( + reader: Box, + file_name: String, + query: String, + limit: usize, + include_row_ids: Option, + top_k: Option, ) -> crate::Result { let handle = tokio::runtime::Handle::try_current().map_err(|error| crate::Error::UnexpectedError { @@ -48,6 +62,7 @@ pub(crate) async fn search_full_text_index( &query, limit, include_row_ids, + top_k, ) }) .await @@ -72,6 +87,7 @@ pub(crate) async fn search_full_text_file( &query, limit, include_row_ids, + None, ) }) .await @@ -87,20 +103,56 @@ fn search_native_index( query: &str, limit: usize, include_row_ids: Option, + top_k: Option, ) -> crate::Result { let reader = FullTextArchiveReader::from_seek_read(input) .map_err(|error| native_error(file_name, error))?; - let hits = match include_row_ids { - Some(include) => reader.search_with_include(query, limit, &include), - None => reader.search(query, limit), - } - .map_err(|error| native_error(file_name, error))?; + let search = |limit| { + match &include_row_ids { + Some(include) => reader.search_with_include(query, limit, include), + None => reader.search(query, limit), + } + .map_err(|error| native_error(file_name, error)) + }; + let hits = if let Some(k) = top_k.filter(|k| *k > 0 && *k < limit) { + let candidate_count = k.saturating_add(1).min(limit); + let mut hits = search(candidate_count)?; + if certify_top_k(&mut hits, k) { + hits + } else { + search(limit)? + } + } else { + search(limit)? + }; Ok(FullTextSearchResult { row_ids: hits.row_ids, scores: hits.scores, }) } +fn certify_top_k(hits: &mut crate::ftindex::reader::FullTextHits, k: usize) -> bool { + if hits.row_ids.len() <= k { + return true; + } + // Tantivy's tie-break uses document addresses, while Paimon uses row IDs. + // Fetch one extra hit to prove that no excluded document can tie the cutoff. + // Duplicate row IDs also require the exhaustive path before deduplication. + let distinct = hits.row_ids[..k] + .iter() + .collect::>(); + if distinct.len() != k + || !hits.scores[k - 1].is_finite() + || !hits.scores[k].is_finite() + || hits.scores[k - 1] <= hits.scores[k] + { + return false; + } + hits.row_ids.truncate(k); + hits.scores.truncate(k); + true +} + fn native_error(file_name: &str, error: impl std::fmt::Display) -> crate::Error { crate::Error::UnexpectedError { message: format!("Failed to read full-text index '{file_name}': {error}"), @@ -254,6 +306,81 @@ mod tests { bytes } + #[test] + fn certificate_requires_strict_scores_and_distinct_row_ids() { + use crate::ftindex::reader::FullTextHits; + let mut strict = FullTextHits { + row_ids: vec![9, 3, 1], + scores: vec![3.0, 2.0, 1.0], + }; + assert!(certify_top_k(&mut strict, 2)); + assert_eq!(strict.row_ids, [9, 3]); + for (row_ids, scores) in [ + (vec![9, 3, 1], vec![3.0, 2.0, 2.0]), + (vec![9, 9, 1], vec![3.0, 2.0, 1.0]), + (vec![9, 3, 1], vec![3.0, f32::NAN, 1.0]), + ] { + let mut hits = FullTextHits { + row_ids: row_ids.clone(), + scores, + }; + assert!(!certify_top_k(&mut hits, 2)); + assert_eq!(hits.row_ids, row_ids); + } + } + + #[tokio::test] + async fn bounded_candidates_match_exhaustive_and_ties_fall_back() { + for tied in [false, true] { + let mut writer = FullTextIndexWriter::new(FullTextIndexConfig::new()).unwrap(); + // Deliberately reverse row IDs so document-address ties cannot stand + // in for the public row-ID tie-break. + for (id, text) in [ + (9, "paimon"), + (3, "paimon padding"), + (1, "paimon padding padding padding"), + ] { + writer + .add_document(id, if tied { "paimon" } else { text }) + .unwrap(); + } + let mut data = Vec::new(); + writer.write(&mut PosWriter::new(&mut data)).unwrap(); + let query = r#"{"match":{"query":"paimon"}}"#.to_string(); + let search = |top_k| { + search_full_text_index_top_k( + Box::new(BytesFileRead { + data: Bytes::from(data.clone()), + ranges: Default::default(), + }), + "bounded.index".into(), + query.clone(), + 3, + None, + top_k, + ) + }; + let exhaustive = search(None).await.unwrap(); + let bounded = search(Some(1)).await.unwrap(); + assert_eq!(bounded.row_ids.len(), if tied { 3 } else { 1 }); + let expected = crate::full_text::SearchResult::new( + exhaustive.row_ids.into_iter().map(|id| id as u64).collect(), + exhaustive.scores, + ) + .top_k(1); + let actual = crate::full_text::SearchResult::new( + bounded.row_ids.into_iter().map(|id| id as u64).collect(), + bounded.scores, + ) + .top_k(1); + assert_eq!(actual.row_ids, expected.row_ids); + assert_eq!(actual.scores, expected.scores); + if tied { + assert_eq!(actual.row_ids, [1]); + } + } + } + #[test] fn test_range_backed_search_works_on_current_thread_runtime() { let runtime = tokio::runtime::Builder::new_current_thread() diff --git a/crates/paimon/src/table/full_text_search_builder.rs b/crates/paimon/src/table/full_text_search_builder.rs index dfebafaba..0f573240b 100644 --- a/crates/paimon/src/table/full_text_search_builder.rs +++ b/crates/paimon/src/table/full_text_search_builder.rs @@ -26,7 +26,7 @@ use crate::spec::{ IndexManifestEntry, Predicate, ROW_ID_FIELD_NAME, }; use crate::table::data_file_reader::DataFileReader; -use crate::table::full_text_index_adapter::{search_full_text_file, search_full_text_index}; +use crate::table::full_text_index_adapter::{search_full_text_file, search_full_text_index_top_k}; use crate::table::global_index_scanner::{ deleted_row_ranges_for_data_evolution_dvs, unindexed_ranges_for_global_index_entries, RowRangeIndex, @@ -259,6 +259,7 @@ struct FullTextSearchEvaluation<'a> { partition_filter: Option<&'a Predicate>, row_filter: Option<&'a Predicate>, source_splits: Option<&'a [crate::table::SearchScanSplit]>, + allow_top_k: bool, } #[cfg(test)] @@ -362,12 +363,25 @@ async fn evaluate_full_text_search_with_executor( fulltext_entries, indexed_search.include_row_ids.as_ref(), )?; + // Raw fallback can replace indexed hits, and deletion vectors are applied + // after the shard searches. Those paths require exhaustive candidates. + let top_k = (evaluation.allow_top_k + && search_mode == GlobalIndexSearchMode::Fast + && deleted_row_index.is_none() + && !core_options.deletion_vectors_enabled() + && evaluation.source_splits.is_none_or(|splits| { + splits + .iter() + .all(|split| !matches!(split, crate::table::SearchScanSplit::Raw { .. })) + }) + && disjoint_full_text_ranges(&plans)) + .then_some(search.limit); let results = if let Some(executor) = executor { let table = evaluation.table.ok_or_else(|| crate::Error::DataInvalid { message: "Scheduled full-text search requires table context".into(), source: None, })?; - execution::search_indexes(table, plans, &search.query_text, executor).await? + execution::search_indexes(table, plans, &search.query_text, top_k, executor).await? } else { let results = plans .into_iter() @@ -383,18 +397,18 @@ async fn evaluate_full_text_search_with_executor( let local_filter = plan.local_filter; let row_range_start = global_meta.row_range_start; let row_range_end = global_meta.row_range_end; - // Compound query leaves must see every candidate in the shard - // before Boolean/boost composition and the final global Top-K. + // The native engine evaluates the complete query before selecting hits. let limit = candidate_limit(row_range_start, row_range_end)?; Ok::<_, crate::Error>(async move { let input = evaluation.file_io.new_input(&path)?; let reader = Box::new(input.reader().await?); - let result = search_full_text_index( + let result = search_full_text_index_top_k( reader, file_name, query_text, limit, local_filter, + top_k, ) .await?; Ok::<_, crate::Error>( @@ -498,6 +512,16 @@ struct FullTextIndexSearchPlan { local_filter: Option, } +fn disjoint_full_text_ranges(plans: &[FullTextIndexSearchPlan]) -> bool { + let mut ranges = plans + .iter() + .filter_map(|plan| plan.entry.index_file.global_index_meta.as_ref()) + .map(|meta| (meta.row_range_start, meta.row_range_end)) + .collect::>(); + ranges.sort_unstable(); + ranges.windows(2).all(|pair| pair[0].1 < pair[1].0) +} + fn plan_full_text_index_searches( entries: Vec<&IndexManifestEntry>, include_row_ids: Option<&RoaringTreemap>, @@ -1016,6 +1040,7 @@ mod tests { partition_filter: None, row_filter: None, source_splits: None, + allow_top_k: true, }, &[], &search, @@ -1172,6 +1197,7 @@ mod tests { partition_filter: None, row_filter: None, source_splits: None, + allow_top_k: true, }, &[entry], &search, @@ -1259,6 +1285,7 @@ mod tests { partition_filter: None, row_filter: None, source_splits: None, + allow_top_k: true, }, &[entry, missing_empty_filter_entry], &search, @@ -1372,6 +1399,22 @@ mod tests { bytes } + #[test] + fn top_k_requires_disjoint_inclusive_shard_ranges() { + let entries = [ + full_text_index_entry("a", 0, 9), + full_text_index_entry("b", 10, 19), + ]; + let plans = plan_full_text_index_searches(entries.iter().collect(), None).unwrap(); + assert!(disjoint_full_text_ranges(&plans)); + let entries = [ + full_text_index_entry("a", 0, 9), + full_text_index_entry("b", 9, 19), + ]; + let plans = plan_full_text_index_searches(entries.iter().collect(), None).unwrap(); + assert!(!disjoint_full_text_ranges(&plans)); + } + #[test] fn test_plan_full_text_index_searches_partitions_once_and_preserves_order() { let entries = [ diff --git a/crates/paimon/src/table/full_text_search_builder/execution.rs b/crates/paimon/src/table/full_text_search_builder/execution.rs index 262608985..53c6b9d85 100644 --- a/crates/paimon/src/table/full_text_search_builder/execution.rs +++ b/crates/paimon/src/table/full_text_search_builder/execution.rs @@ -30,6 +30,8 @@ const ROWS: &str = "fulltext-rows"; struct IndexWork { entry: IndexManifestEntry, query: String, + #[serde(default)] + top_k: Option, local_filter: Option>, } @@ -101,6 +103,7 @@ pub(super) async fn search_indexes( table: &Table, plans: Vec, query: &str, + top_k: Option, executor: &dyn SearchTaskExecutor, ) -> crate::Result> { let mut plans = plans.into_iter(); @@ -118,6 +121,7 @@ pub(super) async fn search_indexes( encode(IndexWork { entry: plan.entry, query: query.to_string(), + top_k, local_filter: plan.local_filter.as_ref().map(encode_bitmap).transpose()?, })?, ) @@ -193,12 +197,13 @@ pub(crate) async fn execute_search_task( ); let input = table.file_io().new_input(&path)?; let reader = Box::new(input.reader().await?); - let result = search_full_text_index( + let result = search_full_text_index_top_k( reader, work.entry.index_file.file_name.clone(), work.query, limit, work.local_filter.map(decode_bitmap).transpose()?, + work.top_k, ) .await?; encode(Hits::from_search( diff --git a/crates/paimon/src/table/full_text_search_builder/parity_tests.rs b/crates/paimon/src/table/full_text_search_builder/parity_tests.rs index f2f37752a..6abae9120 100644 --- a/crates/paimon/src/table/full_text_search_builder/parity_tests.rs +++ b/crates/paimon/src/table/full_text_search_builder/parity_tests.rs @@ -1095,3 +1095,126 @@ async fn index_only_data_plan_applies_reader_partitions_across_commits() { assert_eq!(user_ids(&table, result).await, expected); } } + +#[tokio::test] +async fn certified_top_k_preserves_local_and_scheduled_filtered_results() { + let table = new_table(false).copy_with_options(HashMap::from([( + "full-text-index.search-mode".into(), + "fast".into(), + )])); + let schema = Arc::new(ArrowSchema::new(vec![ + Field::new("id", ArrowType::Int32, true), + Field::new("text", ArrowType::Utf8, true), + Field::new("label", ArrowType::Utf8, true), + Field::new("pt", ArrowType::Int32, true), + ])); + let mut writer = TableWrite::new(&table, "distinct-scores".into()).unwrap(); + writer + .write_arrow_batch( + &RecordBatch::try_new( + schema, + vec![ + Arc::new(Int32Array::from((0..8).collect::>())) as ArrayRef, + Arc::new(StringArray::from( + (0..8) + .map(|id| "paimon ".repeat(8 - id % 4)) + .collect::>(), + )), + Arc::new(StringArray::from(vec![ + "drop", "keep", "keep", "keep", "drop", "keep", "keep", "keep", + ])), + Arc::new(Int32Array::from(vec![0; 8])), + ], + ) + .unwrap(), + ) + .await + .unwrap(); + TableCommit::new(table.clone(), "distinct-scores".into()) + .commit(writer.prepare_commit().await.unwrap()) + .await + .unwrap(); + index(&table, "full-text", "text").await; + let filter = PredicateBuilder::new(table.schema().fields()) + .equal("label", Datum::String("keep".into())) + .unwrap(); + let mut builder = query(&table); + builder.with_filter(filter).with_limit(2); + let read = builder.new_read().unwrap(); + let plan = builder.new_scan().unwrap().scan().await.unwrap(); + let exhaustive = CandidateCountExecutor { + inner: WireExecutor { + table: table.clone(), + work: Default::default(), + batch_sizes: Default::default(), + corrupt_scores: false, + }, + expected_candidates: 3, + }; + let expected = read + .read_exhaustive(plan.clone(), Some(&exhaustive)) + .await + .unwrap() + .into_row_ids() + .unwrap(); + assert!(exhaustive + .inner + .work + .lock() + .unwrap() + .iter() + .all(|task| task["payload"]["top_k"].is_null())); + let local = read + .read(plan.clone()) + .await + .unwrap() + .into_row_ids() + .unwrap(); + let executor = WireExecutor { + table: table.clone(), + work: Default::default(), + batch_sizes: Default::default(), + corrupt_scores: false, + }; + let checked = CandidateCountExecutor { + inner: executor, + expected_candidates: 2, + }; + let remote = read + .read_with_executor(plan, &checked) + .await + .unwrap() + .into_row_ids() + .unwrap(); + assert_eq!(local.row_ids, expected.row_ids); + assert_eq!(local.scores, expected.scores); + assert_eq!(remote.row_ids, expected.row_ids); + assert_eq!(remote.scores, expected.scores); + let work = checked.inner.work.lock().unwrap(); + assert!(work.iter().all(|task| task["payload"]["top_k"] == 2)); +} + +struct CandidateCountExecutor { + inner: WireExecutor, + expected_candidates: usize, +} + +#[async_trait::async_trait] +impl crate::table::SearchTaskExecutor for CandidateCountExecutor { + async fn execute( + &self, + tasks: Vec, + ) -> crate::Result> { + let results = self.inner.execute(tasks).await?; + assert!(!results.is_empty()); + for result in &results { + let value: serde_json::Value = serde_json::from_slice(&result.serialize()?).unwrap(); + assert_eq!( + value["payload"]["row_ids"].as_array().unwrap().len(), + self.expected_candidates, + "worker must return the expected candidate count" + ); + } + Ok(results) + } +} diff --git a/crates/paimon/src/table/full_text_search_builder/read.rs b/crates/paimon/src/table/full_text_search_builder/read.rs index 81502982f..b12bd5eac 100644 --- a/crates/paimon/src/table/full_text_search_builder/read.rs +++ b/crates/paimon/src/table/full_text_search_builder/read.rs @@ -86,7 +86,7 @@ impl FullTextRead { &self, plan: FullTextScanPlan, ) -> crate::Result { - self.read_internal(plan, None).await + self.read_internal(plan, None, true).await } /// Execute shard work through a scheduler while retaining the plan's @@ -96,13 +96,24 @@ impl FullTextRead { plan: FullTextScanPlan, executor: &dyn crate::table::SearchTaskExecutor, ) -> crate::Result { - self.read_internal(plan, Some(executor)).await + self.read_internal(plan, Some(executor), true).await + } + + /// Exhaustive reference path for internal benchmarks and parity tests. + #[cfg(any(feature = "internal-api", test))] + pub(crate) async fn read_exhaustive( + &self, + plan: FullTextScanPlan, + executor: Option<&dyn crate::table::SearchTaskExecutor>, + ) -> crate::Result { + self.read_internal(plan, executor, false).await } async fn read_internal( &self, plan: FullTextScanPlan, executor: Option<&dyn crate::table::SearchTaskExecutor>, + allow_top_k: bool, ) -> crate::Result { CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?; if executor.is_some() && !self.table.schema().core_options().data_evolution_enabled() { @@ -268,6 +279,7 @@ impl FullTextRead { partition_filter: partition_filter.as_ref(), row_filter: row_filter.as_ref(), source_splits: Some(plan.splits()), + allow_top_k, }, &entries, &search, diff --git a/crates/paimon/src/table/internal.rs b/crates/paimon/src/table/internal.rs index b840f113b..497b57d5f 100644 --- a/crates/paimon/src/table/internal.rs +++ b/crates/paimon/src/table/internal.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -//! Unstable execution bridge for the repository's language bindings. +//! Unstable execution bridge for the repository's language bindings and benchmarks. //! //! Source plans are the public input to search readers. Candidate and scoring //! requests below are internal execution stages: Core owns their ordering and @@ -60,6 +60,14 @@ pub async fn read_full_text_with_executor( reader.read_with_executor(plan, executor).await } +/// Run the exhaustive reference collector for benchmarks without a table option. +pub async fn read_full_text_exhaustive( + reader: &super::FullTextRead, + plan: super::FullTextScanPlan, +) -> crate::Result { + reader.read_exhaustive(plan, None).await +} + pub async fn execute_hybrid_with_executor( builder: &HybridSearchBuilder<'_>, executor: &dyn SearchTaskExecutor, diff --git a/crates/paimon/src/table/sorted_global_index_build_builder.rs b/crates/paimon/src/table/sorted_global_index_build_builder.rs index 238f02dca..8d6be1858 100644 --- a/crates/paimon/src/table/sorted_global_index_build_builder.rs +++ b/crates/paimon/src/table/sorted_global_index_build_builder.rs @@ -151,7 +151,7 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> { super::global_index_build_common::validate_build_table(self.table)?; let records_per_range = core_options.sorted_index_records_per_range()?; if index_type != FM_GLOBAL_INDEX_TYPE { - core_options.sorted_index_sort_memory()?; + core_options.write_buffer_size()?; } let write_options = if index_type == FM_GLOBAL_INDEX_TYPE { GlobalIndexWriteOptions::FM(FMOptions::from_options(&resolved_options)?.write) diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/sort.rs b/crates/paimon/src/table/sorted_global_index_build_builder/sort.rs index 68e1ecd2e..dee452556 100644 --- a/crates/paimon/src/table/sorted_global_index_build_builder/sort.rs +++ b/crates/paimon/src/table/sorted_global_index_build_builder/sort.rs @@ -58,7 +58,10 @@ impl IndexSorter { .ok_or_else(|| io_error(std::io::Error::other("index key size overflow")))?; if bytes > self.budget { return Err(Error::ResourceExhausted { - message: format!("Sorted index key requires {bytes} bytes, exceeding sorted-index.sort-memory={}", self.budget), + message: format!( + "Sorted index key requires {bytes} bytes, exceeding write-buffer-size={}", + self.budget + ), }); } if bytes > self.budget - self.bytes { diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs index d393a3219..a47de3da0 100644 --- a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs +++ b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs @@ -844,7 +844,7 @@ async fn assert_btree_build_version(version: u32, budget: &str) { let mut options = table_options("10"); options.insert("read.batch-size".into(), "1".into()); options.insert("btree-index.file-version".into(), version.to_string()); - options.insert("sorted-index.sort-memory".into(), budget.to_string()); + options.insert("write-buffer-size".into(), budget.to_string()); options.insert( "btree-index.bloom-filter.enabled".to_string(), "true".to_string(), @@ -1637,7 +1637,7 @@ async fn test_detail_mode_defers_manifest_pruning_for_unindexed_ranges() { async fn test_execute_writes_bitmap_index_manifest_and_java_file() { let table_path = "memory:/test_bitmap_global_index_builder_e2e"; let mut options = table_options("10"); - options.insert("sorted-index.sort-memory".into(), "128b".into()); + options.insert("write-buffer-size".into(), "128b".into()); let table = test_table_with_path(table_path, options); setup_dirs(&table).await; @@ -1742,7 +1742,7 @@ async fn test_execute_writes_bitmap_index_manifest_and_java_file() { async fn test_execute_multivalue_index_and_array_queries_end_to_end() { let table_path = "memory:/test_multivalue_global_index_builder_e2e"; let table = multivalue_table(table_path).copy_with_options(HashMap::from([ - ("sorted-index.sort-memory".into(), "128b".into()), + ("write-buffer-size".into(), "128b".into()), ("read.batch-size".into(), "1".into()), ])); setup_dirs(&table).await; @@ -2806,7 +2806,7 @@ async fn composite_ranges( async fn test_composite_btree_build_query_coverage_and_drop() { for version in [1, 2] { let mut options = table_options("2"); - options.insert("sorted-index.sort-memory".into(), "128b".into()); + options.insert("write-buffer-size".into(), "128b".into()); options.insert("btree-index.file-version".into(), version.to_string()); options.insert("btree-index.bloom-filter.enabled".into(), "true".into()); let table = @@ -3639,10 +3639,10 @@ async fn test_composite_btree_preserves_exact_column_names() { } #[tokio::test] -async fn invalid_sort_budget_is_rejected_before_an_empty_build() { +async fn invalid_write_buffer_size_is_rejected_before_an_empty_build() { for value in ["0", "-1mb", "invalid"] { let mut options = table_options("10"); - options.insert("sorted-index.sort-memory".into(), value.into()); + options.insert("write-buffer-size".into(), value.into()); let table = test_table(options); let error = table .new_global_index_build_builder() @@ -3651,6 +3651,62 @@ async fn invalid_sort_budget_is_rejected_before_an_empty_build() { .await .unwrap_err(); assert!(matches!(error, Error::ConfigInvalid { .. })); - assert!(error.to_string().contains("sorted-index.sort-memory")); + assert!(error.to_string().contains("write-buffer-size")); } } + +#[tokio::test] +async fn write_buffer_size_limits_keys_and_builder_options_override_table() { + let table = test_table_with_path( + "memory:/index_build_write_buffer_override", + table_options("10"), + ); + setup_dirs(&table).await; + let builder = table.new_write_builder(); + let mut writer = builder.new_write().unwrap(); + writer + .write_arrow_batch(&data_batch(vec![1, 2, 3], vec!["alice", "bob", "alice"])) + .await + .unwrap(); + builder + .new_commit() + .commit(writer.prepare_commit().await.unwrap()) + .await + .unwrap(); + let table = table.copy_with_options(HashMap::from([("write-buffer-size".into(), "1b".into())])); + let before = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(); + let error = table + .new_global_index_build_builder() + .with_index_column("name") + .build() + .await + .unwrap_err(); + assert!(matches!(error, Error::ResourceExhausted { .. })); + assert!(error.to_string().contains("write-buffer-size")); + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + before + ); + assert_eq!( + table + .new_global_index_build_builder() + .with_index_column("name") + .with_options(HashMap::from([("write-buffer-size".into(), "128b".into())])) + .execute() + .await + .unwrap(), + 1 + ); + let predicate = PredicateBuilder::new(table.schema().fields()) + .equal("name", Datum::String("alice".into())) + .unwrap(); + assert_eq!(scan_ids(&table, predicate).await, vec![1, 3]); +} diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs index 1523e91dd..83ec21c1c 100644 --- a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs +++ b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs @@ -81,7 +81,7 @@ impl SortedGlobalIndexBuildBuilder<'_> { } else { let mut options = self.table.schema().options().clone(); options.extend(self.options.clone()); - let budget = crate::spec::CoreOptions::new(&options).sorted_index_sort_memory()?; + let budget = crate::spec::CoreOptions::new(&options).write_buffer_size()?; extract_index_rows( self.table, shard, diff --git a/docs/src/benchmark.md b/docs/src/benchmark.md index 9c433ea8e..59f64813a 100644 --- a/docs/src/benchmark.md +++ b/docs/src/benchmark.md @@ -303,3 +303,16 @@ useful as a reproducible engineering checkpoint, not a universal performance ranking. A broader study should repeat complete benchmark processes, separate cold and warm cache states, and include object storage and multi-node environments. + +## Full-text Top-K + +The repository also includes a synthetic full-text benchmark comparing certified +per-shard Top-K with exhaustive candidate collection on the same local table and +snapshot. It checks exact row IDs and scores, alternates execution order, and +reports latency for distinct-score and boundary-tie corpora. See the +[benchmark protocol](https://github.com/apache/paimon-rust/tree/main/benchmarks/full-text). + +The optimization is selected automatically for FAST-mode global searches with +disjoint index ranges and no raw sources or deletion exclusions. Other paths and boundary ties retain +exhaustive evaluation. The benchmark uses an internal reference reader to compare +against exhaustive candidate collection without adding a table option. diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md index d64108bd3..5bcce1a87 100644 --- a/docs/src/getting-started.md +++ b/docs/src/getting-started.md @@ -252,8 +252,8 @@ and additional index types are not supported. ## Scalar Global Index Build Memory BTree, Bitmap, and Multivalue global index builds stream projected Arrow batches -and sort serialized keys within `sorted-index.sort-memory` (default `64mb`). -This is a Rust build setting and does not change index files or shard boundaries. +and sort serialized keys within `write-buffer-size` (default `256mb`), matching +Java's sort buffer configuration. This does not change index files or shard boundaries. Set it as a table property or pass it through the index builder's `with_options`. When keys exceed the sort buffer, the build writes temporary runs to the system