Prepare full-text index 0.1 API and storage
diff --git a/Cargo.lock b/Cargo.lock index b5b06ce..040ef91 100644 --- a/Cargo.lock +++ b/Cargo.lock
@@ -765,10 +765,13 @@ version = "0.1.0" dependencies = [ "anyhow", + "levenshtein_automata", "roaring", "serde", "serde_json", "tantivy", + "tantivy-common", + "tantivy-fst", "tantivy-jieba", "tempfile", "thiserror 1.0.69",
diff --git a/Cargo.toml b/Cargo.toml index 037ba19..79f1c14 100644 --- a/Cargo.toml +++ b/Cargo.toml
@@ -11,10 +11,13 @@ [workspace.dependencies] anyhow = "1" jni = "0.21" +levenshtein_automata = "0.2" serde = { version = "1", features = ["derive"] } serde_json = "1" roaring = "0.11" tantivy = "0.26" +tantivy-common = "0.11" +tantivy-fst = "0.5" tantivy-jieba = "0.20" tempfile = "3" thiserror = "1"
diff --git a/README.md b/README.md index 1b08b12..3931524 100644 --- a/README.md +++ b/README.md
@@ -9,7 +9,7 @@ - `java`: public Java API. - `python`: Python ctypes API over the C ABI. -The index file is self-describing. Readers only need positional `read_at` I/O +The index file is self-describing. Readers only need positional `pread` I/O and do not depend on Paimon manifest metadata. ## Current Status @@ -17,8 +17,12 @@ Implemented: - Rust writer, reader, v1 envelope, and search. -- C FFI writer/reader/search JSON, including serialized 64-bit Roaring row-id - filters. +- Single-field, multi-field, repeated-field string-array, and dotted-path text + fields. +- Match, fuzzy match (`fuzziness`, `auto`, `max_expansions`, + `prefix_length`), phrase, boolean, multi-match, and boost-demotion queries. +- C FFI writer/reader/search with query JSON strings, including serialized + 64-bit Roaring row-id filters. - Java API and JNI bridge. - Python ctypes package. - Cross-boundary round-trip tests for Rust core, FFI, Java/JNI, and Python. @@ -32,12 +36,13 @@ - `ngram` - `jieba` -Reserved for follow-up: +Default tokenizer behavior uses English full-text defaults: lower-case, +stemming, stop-word removal, ASCII folding, max token length 40, and no +positions unless `with-position=true` is set. Phrase search requires positions. -- stemming -- built-in and custom stop-word filters -- true seek-on-demand Tantivy directory instead of loading segment files into - memory at reader open +Readers expose archived Tantivy files through a seek-on-demand directory, so +opening an index reads the envelope and Tantivy metadata without loading all +segment files into memory. ## Build @@ -54,9 +59,7 @@ ```rust use paimon_ftindex_core::io::{PosWriter, SliceReader}; -use paimon_ftindex_core::{ - FullTextIndexConfig, FullTextIndexReader, FullTextIndexWriter, FullTextQuery, -}; +use paimon_ftindex_core::{FullTextIndexConfig, FullTextIndexReader, FullTextIndexWriter}; let mut writer = FullTextIndexWriter::new(FullTextIndexConfig::new())?; writer.add_document(1, "Apache Paimon full text search")?; @@ -64,16 +67,35 @@ let mut bytes = Vec::new(); writer.write(&mut PosWriter::new(&mut bytes))?; -let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; -let result = reader.search(FullTextQuery::match_query("paimon", "text"), 10)?; +let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; +reader.prewarm()?; +let result = reader.search(r#"{"match":{"query":"paimon","column":"text"}}"#, 10)?; ``` +Multi-field indexes can be configured with named fields: + +```rust +let config = FullTextIndexConfig::new().with_text_fields(["title", "body"]); +let mut writer = FullTextIndexWriter::new(config)?; +writer.add_document_fields( + 1, + [ + ("title".to_string(), "Apache Paimon".to_string()), + ("body".to_string(), "lake storage".to_string()), + ], +)?; +``` + +When a `match` query omits `column`, the reader searches all indexed text +fields. This lets a Paimon adapter populate extra fields internally without +requiring callers to build a `multi_match` query. + To restrict search to an upstream candidate set, pass a serialized `RoaringTreemap` of allowed row ids: ```rust let filtered = reader.search_with_roaring_filter( - FullTextQuery::match_query("paimon", "text"), + r#"{"match":{"query":"paimon","column":"text"}}"#, 10, roaring_filter_bytes, )?; @@ -83,11 +105,11 @@ ```python from io import BytesIO -from paimon_ftindex import FullTextIndexReader, FullTextIndexWriter, MatchQuery +from paimon_ftindex import FullTextIndexReader, FullTextIndexWriter out = BytesIO() -with FullTextIndexWriter() as writer: - writer.add_document(1, "Apache Paimon full text search") +with FullTextIndexWriter({"text-fields": "title,body"}) as writer: + writer.add_document_fields(1, {"title": "Apache Paimon", "body": "lake storage"}) writer.write(out) class Input: @@ -97,8 +119,19 @@ return self.data[pos:pos + length] with FullTextIndexReader(Input(out.getvalue())) as reader: - ids, scores = reader.search(MatchQuery("paimon"), limit=10) + reader.prewarm() + ids, scores = reader.search('{"match":{"query":"paimon"}}', limit=10) filtered_ids, filtered_scores = reader.search( - MatchQuery("paimon"), limit=10, filter_bytes=roaring_filter_bytes + '{"match":{"query":"paimno","column":"title","fuzziness":1}}', + limit=10, + filter_bytes=roaring_filter_bytes, ) + metrics = reader.read_metrics() ``` + +Search APIs accept the query DSL as a JSON string. + +`prewarm()` eagerly initializes the underlying search reader and archive cache +before a query burst. `read_metrics()` reports positional read calls/bytes and +archive cache hit/miss counters for tuning reader reuse and object-store access +patterns.
diff --git a/core/Cargo.toml b/core/Cargo.toml index 3bf1235..e498c71 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml
@@ -7,10 +7,13 @@ rust-version.workspace = true [dependencies] +levenshtein_automata.workspace = true roaring.workspace = true serde.workspace = true serde_json.workspace = true tantivy.workspace = true +tantivy-common.workspace = true +tantivy-fst.workspace = true tantivy-jieba.workspace = true tempfile.workspace = true thiserror.workspace = true
diff --git a/core/src/archive_directory.rs b/core/src/archive_directory.rs new file mode 100644 index 0000000..5da9da5 --- /dev/null +++ b/core/src/archive_directory.rs
@@ -0,0 +1,497 @@ +use crate::error::{FtIndexError, Result}; +use crate::io::{ReadMetrics, ReadRequest, SeekRead}; +use crate::storage::ArchiveFileEntry; +use std::collections::{HashMap, VecDeque}; +use std::fmt; +use std::io; +use std::ops::Range; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use tantivy::directory::error::{DeleteError, LockError, OpenReadError, OpenWriteError}; +use tantivy::directory::WritePtr; +use tantivy::directory::{ + Directory, DirectoryLock, FileHandle, Lock, OwnedBytes, WatchCallback, WatchHandle, META_LOCK, +}; +use tantivy_common::HasLen; + +const READ_CACHE_BLOCK_BYTES: usize = 64 * 1024; +const READ_CACHE_MAX_BLOCKS: usize = 256; + +pub(crate) struct ArchiveDirectory<R> { + input: Arc<R>, + files: Arc<HashMap<PathBuf, ArchivedFile>>, + cache: Arc<ArchiveReadCache>, +} + +impl<R> Clone for ArchiveDirectory<R> { + fn clone(&self) -> Self { + Self { + input: Arc::clone(&self.input), + files: Arc::clone(&self.files), + cache: Arc::clone(&self.cache), + } + } +} + +impl<R> ArchiveDirectory<R> { + #[cfg(test)] + pub(crate) fn new(input: Arc<R>, body_start: u64, files: &[ArchiveFileEntry]) -> Result<Self> { + Self::new_with_metrics(input, body_start, files, Arc::new(ReadMetrics::default())) + } + + pub(crate) fn new_with_metrics( + input: Arc<R>, + body_start: u64, + files: &[ArchiveFileEntry], + metrics: Arc<ReadMetrics>, + ) -> Result<Self> { + let mut mapped_files = HashMap::with_capacity(files.len()); + for file in files { + let start = body_start.checked_add(file.offset).ok_or_else(|| { + FtIndexError::InvalidStorage(format!( + "archive file '{}' offset overflow", + file.name + )) + })?; + let length = usize::try_from(file.length).map_err(|_| { + FtIndexError::InvalidStorage(format!("archive file '{}' is too large", file.name)) + })?; + mapped_files.insert(PathBuf::from(&file.name), ArchivedFile { start, length }); + } + Ok(Self { + input, + files: Arc::new(mapped_files), + cache: Arc::new(ArchiveReadCache::new(metrics)), + }) + } +} + +impl<R> fmt::Debug for ArchiveDirectory<R> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ArchiveDirectory") + .field("files", &self.files) + .finish_non_exhaustive() + } +} + +impl<R: SeekRead + 'static> Directory for ArchiveDirectory<R> { + fn get_file_handle( + &self, + path: &Path, + ) -> std::result::Result<Arc<dyn FileHandle>, OpenReadError> { + let file = self.file(path)?; + Ok(Arc::new(ArchiveFileHandle { + input: Arc::clone(&self.input), + cache: Arc::clone(&self.cache), + path: path.to_path_buf(), + start: file.start, + length: file.length, + })) + } + + fn delete(&self, path: &Path) -> std::result::Result<(), DeleteError> { + if self.files.contains_key(path) { + Err(DeleteError::IoError { + io_error: Arc::new(read_only_error()), + filepath: path.to_path_buf(), + }) + } else { + Err(DeleteError::FileDoesNotExist(path.to_path_buf())) + } + } + + fn exists(&self, path: &Path) -> std::result::Result<bool, OpenReadError> { + Ok(self.files.contains_key(path)) + } + + fn open_write(&self, path: &Path) -> std::result::Result<WritePtr, OpenWriteError> { + Err(OpenWriteError::wrap_io_error( + read_only_error(), + path.to_path_buf(), + )) + } + + fn atomic_read(&self, path: &Path) -> std::result::Result<Vec<u8>, OpenReadError> { + let bytes = self + .open_read(path)? + .read_bytes() + .map_err(|io_error| OpenReadError::wrap_io_error(io_error, path.to_path_buf()))?; + Ok(bytes.as_slice().to_vec()) + } + + fn atomic_write(&self, _path: &Path, _data: &[u8]) -> io::Result<()> { + Err(read_only_error()) + } + + fn sync_directory(&self) -> io::Result<()> { + Ok(()) + } + + fn acquire_lock(&self, lock: &Lock) -> std::result::Result<DirectoryLock, LockError> { + if lock.filepath == META_LOCK.filepath { + Ok(DirectoryLock::from(Box::new(()))) + } else { + Err(LockError::wrap_io_error(read_only_error())) + } + } + + fn watch(&self, _watch_callback: WatchCallback) -> tantivy::Result<WatchHandle> { + Ok(WatchHandle::empty()) + } +} + +impl<R> ArchiveDirectory<R> { + fn file(&self, path: &Path) -> std::result::Result<ArchivedFile, OpenReadError> { + self.files + .get(path) + .copied() + .ok_or_else(|| OpenReadError::FileDoesNotExist(path.to_path_buf())) + } +} + +#[derive(Clone, Copy, Debug)] +struct ArchivedFile { + start: u64, + length: usize, +} + +struct ArchiveFileHandle<R> { + input: Arc<R>, + cache: Arc<ArchiveReadCache>, + path: PathBuf, + start: u64, + length: usize, +} + +impl<R> fmt::Debug for ArchiveFileHandle<R> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ArchiveFileHandle") + .field("path", &self.path) + .field("start", &self.start) + .field("length", &self.length) + .finish_non_exhaustive() + } +} + +impl<R> HasLen for ArchiveFileHandle<R> { + fn len(&self) -> usize { + self.length + } +} + +impl<R: SeekRead + 'static> FileHandle for ArchiveFileHandle<R> { + fn read_bytes(&self, range: Range<usize>) -> io::Result<OwnedBytes> { + if range.start > range.end || range.end > self.length { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!( + "invalid range {}..{} for archive file '{}' with length {}", + range.start, + range.end, + self.path.display(), + self.length + ), + )); + } + if range.is_empty() { + return Ok(OwnedBytes::new(Vec::new())); + } + + if let Some(bytes) = self.read_cached_block(&range)? { + return Ok(bytes); + } + + let relative_start = u64::try_from(range.start) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "offset overflow"))?; + let pos = self + .start + .checked_add(relative_start) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "offset overflow"))?; + let data = vec![0u8; range.len()]; + self.read_direct(pos, data) + } +} + +impl<R: SeekRead + 'static> ArchiveFileHandle<R> { + fn read_cached_block(&self, range: &Range<usize>) -> io::Result<Option<OwnedBytes>> { + if range.len() > READ_CACHE_BLOCK_BYTES { + return Ok(None); + } + let block_start = (range.start / READ_CACHE_BLOCK_BYTES) * READ_CACHE_BLOCK_BYTES; + let block_end = self.length.min(block_start + READ_CACHE_BLOCK_BYTES); + if range.end > block_end { + return Ok(None); + } + + let relative_start = u64::try_from(block_start) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "offset overflow"))?; + let pos = self + .start + .checked_add(relative_start) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "offset overflow"))?; + let block = if let Some(block) = self.cache.get(pos)? { + block + } else { + self.cache.record_miss(); + let block_len = block_end - block_start; + let block = self.read_direct(pos, vec![0u8; block_len])?; + self.cache.insert(pos, block)? + }; + let start = range.start - block_start; + Ok(Some(block.slice(start..start + range.len()))) + } + + fn read_direct(&self, pos: u64, mut data: Vec<u8>) -> io::Result<OwnedBytes> { + let mut request = [ReadRequest::new(pos, data.as_mut_slice())]; + self.input.pread(&mut request)?; + Ok(OwnedBytes::new(data)) + } +} + +struct ArchiveReadCache { + inner: Mutex<ArchiveReadCacheInner>, + metrics: Arc<ReadMetrics>, +} + +#[derive(Default)] +struct ArchiveReadCacheInner { + blocks: HashMap<u64, OwnedBytes>, + order: VecDeque<u64>, +} + +impl ArchiveReadCache { + fn new(metrics: Arc<ReadMetrics>) -> Self { + Self { + inner: Mutex::new(ArchiveReadCacheInner::default()), + metrics, + } + } + + fn get(&self, pos: u64) -> io::Result<Option<OwnedBytes>> { + let inner = self + .inner + .lock() + .map_err(|_| io::Error::other("archive read cache lock poisoned"))?; + let block = inner.blocks.get(&pos).cloned(); + if block.is_some() { + self.metrics.record_cache_hit(); + } + Ok(block) + } + + fn record_miss(&self) { + self.metrics.record_cache_miss(); + } + + fn insert(&self, pos: u64, block: OwnedBytes) -> io::Result<OwnedBytes> { + let mut inner = self + .inner + .lock() + .map_err(|_| io::Error::other("archive read cache lock poisoned"))?; + if let Some(existing) = inner.blocks.get(&pos) { + return Ok(existing.clone()); + } + while inner.blocks.len() >= READ_CACHE_MAX_BLOCKS { + if let Some(evicted) = inner.order.pop_front() { + inner.blocks.remove(&evicted); + self.metrics.record_cache_eviction(); + } else { + break; + } + } + inner.order.push_back(pos); + inner.blocks.insert(pos, block.clone()); + self.metrics.record_cache_insert(); + Ok(block) + } +} + +fn read_only_error() -> io::Error { + io::Error::new( + io::ErrorKind::PermissionDenied, + "archive directory is read-only", + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Barrier; + use std::time::{Duration, Instant}; + + #[derive(Default)] + struct SyncCountingReader { + calls: AtomicUsize, + } + + impl SeekRead for SyncCountingReader { + fn pread(&self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> { + self.calls.fetch_add(1, Ordering::SeqCst); + for range in ranges { + range.buf.fill(0); + } + Ok(()) + } + } + + #[test] + fn archive_directory_accepts_shared_sync_reader_without_mutex() { + let input = Arc::new(SyncCountingReader::default()); + let directory = ArchiveDirectory::new( + Arc::clone(&input), + 0, + &[ArchiveFileEntry { + name: "meta.json".to_string(), + offset: 0, + length: 1, + }], + ) + .expect("archive directory"); + + let bytes = directory + .open_read(Path::new("meta.json")) + .expect("file") + .read_bytes() + .expect("bytes"); + + assert_eq!(bytes.as_slice(), &[0]); + assert_eq!(input.calls.load(Ordering::SeqCst), 1); + } + + struct ConcurrentReader { + data: Vec<u8>, + read_calls: AtomicUsize, + active_calls: AtomicUsize, + max_active_calls: AtomicUsize, + } + + impl SeekRead for ConcurrentReader { + fn pread(&self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> { + self.read_calls.fetch_add(1, Ordering::SeqCst); + let active = self.active_calls.fetch_add(1, Ordering::SeqCst) + 1; + self.max_active_calls.fetch_max(active, Ordering::SeqCst); + + let start = Instant::now(); + while self.max_active_calls.load(Ordering::SeqCst) < 2 + && start.elapsed() < Duration::from_millis(500) + { + std::thread::yield_now(); + } + + for range in ranges { + let start = usize::try_from(range.pos) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "offset overflow"))?; + let end = start + .checked_add(range.buf.len()) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "range overflow"))?; + range.buf.copy_from_slice(&self.data[start..end]); + } + self.active_calls.fetch_sub(1, Ordering::SeqCst); + Ok(()) + } + } + + struct CountingDataReader { + data: Vec<u8>, + read_calls: AtomicUsize, + } + + impl SeekRead for CountingDataReader { + fn pread(&self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> { + self.read_calls.fetch_add(1, Ordering::SeqCst); + for range in ranges { + let start = usize::try_from(range.pos) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "offset overflow"))?; + let end = start + .checked_add(range.buf.len()) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "range overflow"))?; + range.buf.copy_from_slice(&self.data[start..end]); + } + Ok(()) + } + } + + #[test] + fn archive_directory_allows_concurrent_pread_calls() { + let input = Arc::new(ConcurrentReader { + data: b"abcdef".to_vec(), + read_calls: AtomicUsize::new(0), + active_calls: AtomicUsize::new(0), + max_active_calls: AtomicUsize::new(0), + }); + let directory = ArchiveDirectory::new( + Arc::clone(&input), + 0, + &[ArchiveFileEntry { + name: "meta.json".to_string(), + offset: 0, + length: 6, + }], + ) + .expect("archive directory"); + + let start_reads = Arc::new(Barrier::new(3)); + let first_directory = directory.clone(); + let first_start = Arc::clone(&start_reads); + let second_directory = directory.clone(); + let second_start = Arc::clone(&start_reads); + let first = std::thread::spawn(move || { + first_start.wait(); + first_directory + .open_read(Path::new("meta.json")) + .expect("file") + .read_bytes() + .expect("bytes") + .as_slice() + .to_vec() + }); + let second = std::thread::spawn(move || { + second_start.wait(); + second_directory + .open_read(Path::new("meta.json")) + .expect("file") + .read_bytes() + .expect("bytes") + .as_slice() + .to_vec() + }); + + start_reads.wait(); + assert_eq!(first.join().expect("first read"), b"abcdef"); + assert_eq!(second.join().expect("second read"), b"abcdef"); + assert_eq!(input.max_active_calls.load(Ordering::SeqCst), 2); + } + + #[test] + fn archive_directory_caches_repeated_small_reads() { + let metrics = Arc::new(ReadMetrics::default()); + let input = Arc::new(CountingDataReader { + data: b"abcdefghijklmnopqrstuvwxyz".to_vec(), + read_calls: AtomicUsize::new(0), + }); + let directory = ArchiveDirectory::new_with_metrics( + Arc::clone(&input), + 0, + &[ArchiveFileEntry { + name: "meta.json".to_string(), + offset: 0, + length: 26, + }], + Arc::clone(&metrics), + ) + .expect("archive directory"); + let file = directory.open_read(Path::new("meta.json")).expect("file"); + + let first = file.read_bytes_slice(3..9).expect("first"); + let second = file.read_bytes_slice(3..9).expect("second"); + + assert_eq!(first.as_slice(), b"defghi"); + assert_eq!(second.as_slice(), b"defghi"); + assert_eq!(input.read_calls.load(Ordering::SeqCst), 1); + let metrics = metrics.snapshot(); + assert_eq!(metrics.cache_hits, 1); + assert_eq!(metrics.cache_misses, 1); + assert_eq!(metrics.cached_blocks, 1); + } +}
diff --git a/core/src/config.rs b/core/src/config.rs index d6cf964..295f1a7 100644 --- a/core/src/config.rs +++ b/core/src/config.rs
@@ -1,43 +1,192 @@ use crate::tokenizer::TokenizerConfig; +use crate::{FtIndexError, Result}; use serde::{Deserialize, Serialize}; +use std::collections::HashMap; #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] pub struct FullTextIndexConfig { + #[serde(default = "default_row_id_field")] pub row_id_field: String, - pub text_field: String, + #[serde(default = "default_text_fields")] + pub text_fields: Vec<String>, + #[serde(default)] pub tokenizer: TokenizerConfig, } impl Default for FullTextIndexConfig { fn default() -> Self { Self { - row_id_field: "row_id".to_string(), - text_field: "text".to_string(), + row_id_field: default_row_id_field(), + text_fields: default_text_fields(), tokenizer: TokenizerConfig::default(), } } } +fn default_row_id_field() -> String { + "row_id".to_string() +} + +fn default_text_fields() -> Vec<String> { + vec!["text".to_string()] +} + impl FullTextIndexConfig { pub fn new() -> Self { Self::default() } + pub fn from_options(options: &HashMap<String, String>) -> Result<Self> { + let tokenizer = TokenizerConfig::from_options(options)?; + let mut row_id_field = None; + let mut text_field = None; + let mut text_fields = None; + for (raw_key, value) in options { + let key = normalize_option_key(raw_key); + match key.as_str() { + "row-id-field" => row_id_field = Some(clean_name(value, &key)?), + "text-field" => text_field = Some(clean_name(value, &key)?), + "text-fields" | "columns" => text_fields = Some(split_text_fields(value)?), + _ => {} + } + } + + let mut fields = text_fields + .or_else(|| text_field.clone().map(|field| vec![field])) + .unwrap_or_else(default_text_fields); + dedup_preserve_order(&mut fields); + if fields.is_empty() { + return invalid("text-fields", "must contain at least one field"); + } + + let config = Self { + row_id_field: row_id_field.unwrap_or_else(default_row_id_field), + text_fields: fields, + tokenizer, + }; + config.validate()?; + Ok(config) + } + pub fn tokenizer(mut self, tokenizer: TokenizerConfig) -> Self { self.tokenizer = tokenizer; self } + pub fn with_text_fields<I, S>(mut self, fields: I) -> Self + where + I: IntoIterator<Item = S>, + S: Into<String>, + { + self.text_fields = fields.into_iter().map(Into::into).collect(); + self + } + pub fn with_positions(mut self, with_positions: bool) -> Self { self.tokenizer.with_position = with_positions; self } + + pub fn indexed_text_fields(&self) -> Vec<&str> { + let fields = self.text_fields.iter().map(String::as_str).collect(); + dedup_strs(fields) + } + + pub fn default_text_field(&self) -> &str { + self.text_fields + .first() + .map(String::as_str) + .unwrap_or("text") + } + + pub fn validate(&self) -> Result<()> { + if self.row_id_field.trim().is_empty() { + return invalid("row-id-field", "must not be empty"); + } + let mut fields = self.text_fields.clone(); + dedup_preserve_order(&mut fields); + if fields.is_empty() { + return invalid("text-fields", "must contain at least one field"); + } + for field in fields { + if field.trim().is_empty() { + return invalid("text-fields", "must not contain empty field names"); + } + if field == self.row_id_field { + return invalid("text-fields", "must not contain the row id field"); + } + } + self.tokenizer.validate() + } } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] pub struct FullTextIndexMetadata { - pub format_version: u32, + #[serde(default)] pub config: FullTextIndexConfig, + #[serde(default)] pub document_count: u64, + #[serde(default)] pub tantivy_version: String, } + +fn normalize_option_key(raw_key: &str) -> String { + raw_key + .strip_prefix("fulltext.") + .or_else(|| raw_key.strip_prefix("tantivy.")) + .unwrap_or(raw_key) + .trim() + .to_lowercase() + .replace('_', "-") +} + +fn clean_name(value: &str, key: &str) -> Result<String> { + let value = value.trim(); + if value.is_empty() { + invalid(key, "must not be empty") + } else { + Ok(value.to_string()) + } +} + +fn split_text_fields(value: &str) -> Result<Vec<String>> { + let mut fields = value + .split([',', ';']) + .map(str::trim) + .filter(|field| !field.is_empty()) + .map(ToOwned::to_owned) + .collect::<Vec<_>>(); + dedup_preserve_order(&mut fields); + if fields.is_empty() { + invalid("text-fields", "must contain at least one field") + } else { + Ok(fields) + } +} + +fn dedup_preserve_order(fields: &mut Vec<String>) { + let mut deduped = Vec::with_capacity(fields.len()); + for field in fields.drain(..) { + if !deduped.contains(&field) { + deduped.push(field); + } + } + *fields = deduped; +} + +fn dedup_strs(fields: Vec<&str>) -> Vec<&str> { + let mut deduped = Vec::with_capacity(fields.len()); + for field in fields { + if !deduped.contains(&field) { + deduped.push(field); + } + } + deduped +} + +fn invalid<T>(key: &str, message: &str) -> Result<T> { + Err(FtIndexError::InvalidOption { + key: key.to_string(), + message: message.to_string(), + }) +}
diff --git a/core/src/index.rs b/core/src/index.rs index e32773a..f026d50 100644 --- a/core/src/index.rs +++ b/core/src/index.rs
@@ -1,26 +1,32 @@ +use crate::archive_directory::ArchiveDirectory; use crate::config::{FullTextIndexConfig, FullTextIndexMetadata}; use crate::error::{FtIndexError, Result}; -use crate::io::{SeekRead, SeekWrite}; -use crate::query::{BooleanOccur, FullTextQuery, MatchOperator}; -use crate::storage::{ - read_archive_files_with, read_header, write_envelope, ArchiveFileEntry, IndexHeader, -}; +use crate::io::{FullTextReadMetrics, ReadMetrics, ReadRequest, SeekRead, SeekWrite}; +use crate::query::{BooleanOccur, MatchOperator, QuerySpec}; +use crate::storage::{read_header, write_envelope, ArchiveFileEntry, IndexHeader}; use crate::tokenizer::{TokenizerConfig, TokenizerKind}; +use levenshtein_automata::{Distance, LevenshteinAutomatonBuilder, DFA, SINK_STATE}; use roaring::RoaringTreemap; +use std::collections::HashMap; use std::fmt; use std::fs; use std::path::Path; +use std::sync::{Arc, Mutex}; use tantivy::collector::{FilterCollector, TopDocs}; -use tantivy::directory::{Directory, RamDirectory}; use tantivy::query::{ - BooleanQuery, BoostQuery, EnableScoring, Explanation, Occur, Query, QueryParser, Scorer, Weight, + BooleanQuery, BoostQuery, ConstScorer, EmptyQuery, EnableScoring, Explanation, Occur, Query, + QueryParser, Scorer, TermQuery, Weight, }; use tantivy::schema::{IndexRecordOption, NumericOptions, Schema, TextFieldIndexing, TextOptions}; use tantivy::tokenizer::{ - AsciiFoldingFilter, LowerCaser, NgramTokenizer, RawTokenizer, RemoveLongFilter, - SimpleTokenizer, TextAnalyzer, WhitespaceTokenizer, + AsciiFoldingFilter, Language, LowerCaser, NgramTokenizer, RawTokenizer, RemoveLongFilter, + SimpleTokenizer, Stemmer, StopWordFilter, TextAnalyzer, TokenStream, WhitespaceTokenizer, }; -use tantivy::{DocId, DocSet, Index, Score, SegmentReader, TantivyDocument, TERMINATED}; +use tantivy::{ + DocId, DocSet, Index, Score, SegmentReader, SingleSegmentIndexWriter, TantivyDocument, Term, + TERMINATED, +}; +use tantivy_fst::Automaton; use tantivy_jieba::JiebaTokenizer; use tempfile::TempDir; @@ -32,12 +38,18 @@ pub struct FullTextIndexWriter { config: FullTextIndexConfig, - documents: Vec<(i64, String)>, + documents: Vec<FullTextDocument>, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct FullTextDocument { + pub row_id: i64, + pub fields: Vec<(String, String)>, } impl FullTextIndexWriter { pub fn new(config: FullTextIndexConfig) -> Result<Self> { - config.tokenizer.validate()?; + config.validate()?; Ok(Self { config, documents: Vec::new(), @@ -45,12 +57,34 @@ } pub fn add_document(&mut self, row_id: i64, text: impl Into<String>) -> Result<()> { + let text_field = self.config.default_text_field().to_string(); + self.add_document_fields(row_id, [(text_field, text.into())]) + } + + pub fn add_document_fields<I, K, V>(&mut self, row_id: i64, fields: I) -> Result<()> + where + I: IntoIterator<Item = (K, V)>, + K: Into<String>, + V: Into<String>, + { if row_id < 0 { return Err(FtIndexError::InvalidStorage(format!( "row id must be non-negative, got {row_id}" ))); } - self.documents.push((row_id, text.into())); + let fields = fields + .into_iter() + .map(|(name, text)| (name.into(), text.into())) + .collect::<Vec<_>>(); + if fields.is_empty() { + return Err(FtIndexError::InvalidStorage( + "document must contain at least one text field".to_string(), + )); + } + for (name, _) in &fields { + validate_indexed_field(&self.config, name)?; + } + self.documents.push(FullTextDocument { row_id, fields }); Ok(()) } @@ -62,19 +96,25 @@ let row_id_field = schema .get_field(&self.config.row_id_field) .map_err(|_| FtIndexError::InvalidStorage("missing row_id field".to_string()))?; - let text_field = schema - .get_field(&self.config.text_field) - .map_err(|_| FtIndexError::InvalidStorage("missing text field".to_string()))?; + let text_fields = text_field_map(&schema, &self.config)?; { - let mut index_writer = index.writer(50_000_000)?; - for (row_id, text) in &self.documents { + let mut index_writer = + SingleSegmentIndexWriter::<TantivyDocument>::new(index, 50_000_000)?; + for document in &self.documents { let mut doc = TantivyDocument::new(); - doc.add_u64(row_id_field, *row_id as u64); - doc.add_text(text_field, text); + doc.add_u64(row_id_field, document.row_id as u64); + for (name, text) in &document.fields { + let text_field = text_fields.get(name).ok_or_else(|| { + FtIndexError::InvalidStorage(format!( + "document field '{name}' is not configured for this index" + )) + })?; + doc.add_text(*text_field, text); + } index_writer.add_document(doc)?; } - index_writer.commit()?; + index_writer.finalize()?; } let files = collect_index_files(temp_dir.path())?; @@ -91,7 +131,6 @@ let header = IndexHeader { metadata: FullTextIndexMetadata { - format_version: crate::storage::FORMAT_VERSION, config: self.config.clone(), document_count: self.documents.len() as u64, tantivy_version: tantivy::version().to_string(), @@ -103,43 +142,60 @@ } pub struct FullTextIndexReader<R> { - _input: R, + _input: Arc<MeteredSeekRead<R>>, index: Index, + reader: Mutex<Option<tantivy::IndexReader>>, + read_metrics: Arc<ReadMetrics>, metadata: FullTextIndexMetadata, } -impl<R: SeekRead> FullTextIndexReader<R> { - pub fn open(mut input: R) -> Result<Self> { - let (header, body_start) = read_header(&mut input)?; - let directory = RamDirectory::create(); - read_archive_files_with(&mut input, body_start, &header.files, |name, data| { - directory.atomic_write(Path::new(name), data)?; - Ok(()) - })?; +impl<R: SeekRead + 'static> FullTextIndexReader<R> { + pub fn open(input: R) -> Result<Self> { + let read_metrics = Arc::new(ReadMetrics::default()); + let input = MeteredSeekRead { + inner: input, + metrics: Arc::clone(&read_metrics), + }; + let (header, body_start) = read_header(&input)?; + validate_tantivy_version(&header.metadata)?; + let input = Arc::new(input); + let directory = ArchiveDirectory::new_with_metrics( + Arc::clone(&input), + body_start, + &header.files, + Arc::clone(&read_metrics), + )?; let mut index = Index::open(directory)?; register_tokenizer(&mut index, &header.metadata.config.tokenizer)?; Ok(Self { _input: input, index, + reader: Mutex::new(None), + read_metrics, metadata: header.metadata, }) } - pub fn optimize_for_search(&mut self) -> Result<()> { - Ok(()) - } - pub fn metadata(&self) -> &FullTextIndexMetadata { &self.metadata } - pub fn search(&mut self, query: FullTextQuery, limit: usize) -> Result<FullTextSearchResult> { + pub fn read_metrics(&self) -> FullTextReadMetrics { + self.read_metrics.snapshot() + } + + pub fn prewarm(&self) -> Result<()> { + let _ = self.searcher()?; + Ok(()) + } + + pub fn search<Q: AsRef<str>>(&self, query: Q, limit: usize) -> Result<FullTextSearchResult> { self.search_with_filter(query, limit, None) } - pub fn search_with_roaring_filter( - &mut self, - query: FullTextQuery, + pub fn search_with_roaring_filter<Q: AsRef<str>>( + &self, + query: Q, limit: usize, roaring_filter_bytes: &[u8], ) -> Result<FullTextSearchResult> { @@ -147,9 +203,9 @@ self.search_with_filter(query, limit, Some(filter)) } - fn search_with_filter( - &mut self, - query: FullTextQuery, + fn search_with_filter<Q: AsRef<str>>( + &self, + query: Q, limit: usize, filter: Option<RoaringTreemap>, ) -> Result<FullTextSearchResult> { @@ -158,8 +214,8 @@ "search limit must be positive".to_string(), )); } - let reader = self.index.reader()?; - let searcher = reader.searcher(); + let searcher = self.searcher()?; + let query = QuerySpec::from_json(query.as_ref())?; let tantivy_query = build_query(&self.index, &self.metadata.config, &query)?; let top_docs = if let Some(filter) = filter { let row_id_field = self.metadata.config.row_id_field.clone(); @@ -185,6 +241,39 @@ } Ok(FullTextSearchResult { row_ids, scores }) } + + fn searcher(&self) -> Result<tantivy::Searcher> { + let mut reader = self.reader.lock().map_err(|_| { + FtIndexError::InvalidStorage("Tantivy reader lock poisoned".to_string()) + })?; + if reader.is_none() { + *reader = Some(self.index.reader()?); + } + Ok(reader.as_ref().expect("reader initialized").searcher()) + } +} + +struct MeteredSeekRead<R> { + inner: R, + metrics: Arc<ReadMetrics>, +} + +impl<R: SeekRead> SeekRead for MeteredSeekRead<R> { + fn pread(&self, ranges: &mut [ReadRequest<'_>]) -> std::io::Result<()> { + self.metrics.record_pread(ranges); + self.inner.pread(ranges) + } +} + +fn validate_tantivy_version(metadata: &FullTextIndexMetadata) -> Result<()> { + let runtime_version = tantivy::version().to_string(); + if metadata.tantivy_version != runtime_version { + return Err(FtIndexError::InvalidStorage(format!( + "unsupported Tantivy index version {}, runtime uses {}", + metadata.tantivy_version, runtime_version + ))); + } + Ok(()) } fn decode_roaring_filter(bytes: &[u8]) -> Result<RoaringTreemap> { @@ -210,16 +299,17 @@ let indexing = TextFieldIndexing::default() .set_tokenizer(tokenizer_name) .set_index_option(index_option); - builder.add_text_field( - &config.text_field, - TextOptions::default().set_indexing_options(indexing), - ); + for field in config.indexed_text_fields() { + builder.add_text_field( + field, + TextOptions::default().set_indexing_options(indexing.clone()), + ); + } builder.build() } fn tokenizer_name(config: &TokenizerConfig) -> &'static str { match config.tokenizer { - TokenizerKind::Default if !needs_custom_default(config) => "default", TokenizerKind::Default | TokenizerKind::Simple => "paimon_custom", TokenizerKind::Whitespace => "paimon_custom", TokenizerKind::Raw => "paimon_custom", @@ -228,36 +318,15 @@ } } -fn needs_custom_default(config: &TokenizerConfig) -> bool { - !config.lower_case - || config.max_token_length != 40 - || config.ascii_folding - || config.stem - || config.remove_stop_words - || !config.stop_words.is_empty() -} - fn register_tokenizer(index: &mut Index, config: &TokenizerConfig) -> Result<()> { - match config.tokenizer { - TokenizerKind::Default if !needs_custom_default(config) => Ok(()), - _ => { - let analyzer = build_text_analyzer(config)?; - index - .tokenizers() - .register(tokenizer_name(config), analyzer); - Ok(()) - } - } + let analyzer = build_text_analyzer(config)?; + index + .tokenizers() + .register(tokenizer_name(config), analyzer); + Ok(()) } fn build_text_analyzer(config: &TokenizerConfig) -> Result<TextAnalyzer> { - if config.stem || config.remove_stop_words || !config.stop_words.is_empty() { - return Err(FtIndexError::InvalidOption { - key: "tokenizer filters".to_string(), - message: "stemming and stop-word filters are not enabled in this first implementation" - .to_string(), - }); - } let mut builder = match config.tokenizer { TokenizerKind::Default | TokenizerKind::Simple => { TextAnalyzer::builder(SimpleTokenizer::default()).dynamic() @@ -288,12 +357,88 @@ if config.lower_case { builder = builder.filter_dynamic(LowerCaser); } + if config.stem { + builder = builder.filter_dynamic(Stemmer::new(parse_language(&config.language)?)); + } + if config.remove_stop_words { + let language = parse_language(&config.language)?; + if let Some(filter) = StopWordFilter::new(language) { + builder = builder.filter_dynamic(filter); + } else if config.stop_words.is_empty() { + return Err(FtIndexError::InvalidOption { + key: "language".to_string(), + message: format!( + "removing stop words for language '{}' is not supported", + config.language + ), + }); + } + if !config.stop_words.is_empty() { + builder = builder.filter_dynamic(StopWordFilter::remove(config.stop_words.clone())); + } + } if config.ascii_folding { builder = builder.filter_dynamic(AsciiFoldingFilter); } Ok(builder.build()) } +fn parse_language(language: &str) -> Result<Language> { + match language.trim().to_lowercase().as_str() { + "arabic" => Ok(Language::Arabic), + "danish" => Ok(Language::Danish), + "dutch" => Ok(Language::Dutch), + "english" | "en" => Ok(Language::English), + "finnish" => Ok(Language::Finnish), + "french" | "fr" => Ok(Language::French), + "german" | "de" => Ok(Language::German), + "greek" => Ok(Language::Greek), + "hungarian" => Ok(Language::Hungarian), + "italian" | "it" => Ok(Language::Italian), + "norwegian" => Ok(Language::Norwegian), + "portuguese" | "pt" => Ok(Language::Portuguese), + "romanian" => Ok(Language::Romanian), + "russian" | "ru" => Ok(Language::Russian), + "spanish" | "es" => Ok(Language::Spanish), + "swedish" => Ok(Language::Swedish), + "tamil" => Ok(Language::Tamil), + "turkish" => Ok(Language::Turkish), + other => Err(FtIndexError::InvalidOption { + key: "language".to_string(), + message: format!("unsupported tokenizer language '{other}'"), + }), + } +} + +fn text_field_map( + schema: &Schema, + config: &FullTextIndexConfig, +) -> Result<HashMap<String, tantivy::schema::Field>> { + let mut fields = HashMap::new(); + for name in config.indexed_text_fields() { + let field = schema + .get_field(name) + .map_err(|_| FtIndexError::InvalidStorage(format!("missing text field '{name}'")))?; + fields.insert(name.to_string(), field); + } + Ok(fields) +} + +fn validate_indexed_field(config: &FullTextIndexConfig, field: &str) -> Result<()> { + if field.trim().is_empty() { + return Err(FtIndexError::InvalidStorage( + "document field name must not be empty".to_string(), + )); + } + if config.indexed_text_fields().contains(&field) { + Ok(()) + } else { + Err(FtIndexError::InvalidStorage(format!( + "document field '{field}' is not configured for this index" + ))) + } +} + fn collect_index_files(path: &Path) -> Result<Vec<(String, Vec<u8>)>> { let mut paths = Vec::new(); for entry in fs::read_dir(path)? { @@ -321,10 +466,10 @@ fn build_query( index: &Index, config: &FullTextIndexConfig, - query: &FullTextQuery, + query: &QuerySpec, ) -> Result<Box<dyn Query>> { match query { - FullTextQuery::Match { + QuerySpec::Match { column, terms, operator, @@ -333,40 +478,73 @@ max_expansions, prefix_length, } => { - validate_column(config, column)?; validate_match_options(*fuzziness, *max_expansions, *prefix_length)?; - let text_field = index - .schema() - .get_field(&config.text_field) - .map_err(|_| FtIndexError::InvalidQuery("missing text field".to_string()))?; - let mut parser = QueryParser::for_index(index, vec![text_field]); - if *operator == MatchOperator::And { - parser.set_conjunction_by_default(); + let fields = resolve_match_fields(index, config, column.as_deref())?; + let mut children = Vec::with_capacity(fields.len()); + let options = FieldMatchOptions { + operator: *operator, + boost: *boost, + fuzziness: *fuzziness, + max_expansions: *max_expansions, + prefix_length: *prefix_length, + }; + for field in fields { + children.push(( + Occur::Should, + build_field_match_query(index, config, field, terms, options)?, + )); } - let parsed = parser - .parse_query(terms) - .map_err(|e| FtIndexError::InvalidQuery(e.to_string()))?; - if (*boost - 1.0).abs() > f32::EPSILON { - Ok(Box::new(BoostQuery::new(parsed, *boost))) + if children.len() == 1 { + Ok(children.remove(0).1) } else { - Ok(parsed) + Ok(Box::new(BooleanQuery::new(children))) } } - FullTextQuery::MatchPhrase { + QuerySpec::MultiMatch { + terms, + columns, + boosts, + operator, + fuzziness, + max_expansions, + prefix_length, + } => { + validate_multi_match_options( + columns, + boosts, + *fuzziness, + *max_expansions, + *prefix_length, + )?; + let mut children = Vec::with_capacity(columns.len()); + for (idx, column) in columns.iter().enumerate() { + let boost = boosts.get(idx).copied().unwrap_or(1.0); + let field = resolve_text_field(index, config, Some(column))?; + let options = FieldMatchOptions { + operator: *operator, + boost, + fuzziness: *fuzziness, + max_expansions: *max_expansions, + prefix_length: *prefix_length, + }; + children.push(( + Occur::Should, + build_field_match_query(index, config, field, terms, options)?, + )); + } + Ok(Box::new(BooleanQuery::new(children))) + } + QuerySpec::MatchPhrase { column, terms, slop, } => { - validate_column(config, column)?; if !config.tokenizer.with_position { return Err(FtIndexError::InvalidQuery( "phrase query requires positions".to_string(), )); } - let text_field = index - .schema() - .get_field(&config.text_field) - .map_err(|_| FtIndexError::InvalidQuery("missing text field".to_string()))?; + let text_field = resolve_text_field(index, config, column.as_deref())?; let parser = QueryParser::for_index(index, vec![text_field]); let escaped = terms.replace('\\', "\\\\").replace('"', "\\\""); let query_text = if *slop == 0 { @@ -378,7 +556,7 @@ .parse_query(&query_text) .map_err(|e| FtIndexError::InvalidQuery(e.to_string())) } - FullTextQuery::Boolean { + QuerySpec::Boolean { should, must, must_not, @@ -389,6 +567,16 @@ "boolean query must contain at least one clause".to_string(), )); } + let has_positive_clause = !should.is_empty() + || !must.is_empty() + || queries + .iter() + .any(|(occur, _)| matches!(occur, BooleanOccur::Should | BooleanOccur::Must)); + if !has_positive_clause { + return Err(FtIndexError::InvalidQuery( + "boolean query must contain at least one should or must clause".to_string(), + )); + } let mut children = Vec::with_capacity(should.len() + must.len() + must_not.len() + queries.len()); for child in should { @@ -410,7 +598,7 @@ } Ok(Box::new(BooleanQuery::new(children))) } - FullTextQuery::Boost { + QuerySpec::Boost { positive, negative, negative_boost, @@ -430,24 +618,207 @@ max_expansions: usize, prefix_length: u32, ) -> Result<()> { - if fuzziness.unwrap_or(0) != 0 { + if fuzziness.unwrap_or(0) > 2 { return Err(FtIndexError::InvalidQuery( - "match query fuzziness is not supported by paimon-full-text yet".to_string(), + "match query fuzziness must be auto/null or a value in [0, 2]".to_string(), )); } - if max_expansions != 50 { + if max_expansions == 0 { return Err(FtIndexError::InvalidQuery( - "match query max_expansions is not supported by paimon-full-text yet".to_string(), + "match query max_expansions must be positive".to_string(), )); } - if prefix_length != 0 { + let _ = prefix_length; + Ok(()) +} + +fn validate_multi_match_options( + columns: &[String], + boosts: &[f32], + fuzziness: Option<u8>, + max_expansions: usize, + prefix_length: u32, +) -> Result<()> { + if columns.is_empty() { return Err(FtIndexError::InvalidQuery( - "match query prefix_length is not supported by paimon-full-text yet".to_string(), + "multi_match query must contain at least one column".to_string(), )); } + if !boosts.is_empty() && boosts.len() != columns.len() { + return Err(FtIndexError::InvalidQuery(format!( + "multi_match boosts length {} does not match columns length {}", + boosts.len(), + columns.len() + ))); + } + for boost in boosts { + validate_boost(*boost)?; + } + validate_match_options(fuzziness, max_expansions, prefix_length) +} + +fn validate_boost(boost: f32) -> Result<()> { + if !boost.is_finite() || boost <= 0.0 { + return Err(FtIndexError::InvalidQuery(format!( + "boost must be a finite positive value, got {boost}" + ))); + } Ok(()) } +fn resolve_text_field( + index: &Index, + config: &FullTextIndexConfig, + column: Option<&str>, +) -> Result<tantivy::schema::Field> { + let column = match column.map(str::trim).filter(|column| !column.is_empty()) { + Some(column) => column, + None => { + let fields = config.indexed_text_fields(); + if fields.len() == 1 { + fields[0] + } else { + return Err(FtIndexError::InvalidQuery( + "full-text query column must be set for multi-field indexes".to_string(), + )); + } + } + }; + if !config.indexed_text_fields().contains(&column) { + return Err(FtIndexError::InvalidQuery(format!( + "full-text query column '{column}' is not configured for this index" + ))); + } + index + .schema() + .get_field(column) + .map_err(|_| FtIndexError::InvalidQuery(format!("missing text field '{column}'"))) +} + +fn resolve_match_fields( + index: &Index, + config: &FullTextIndexConfig, + column: Option<&str>, +) -> Result<Vec<tantivy::schema::Field>> { + if column + .map(str::trim) + .filter(|column| !column.is_empty()) + .is_some() + { + return Ok(vec![resolve_text_field(index, config, column)?]); + } + config + .indexed_text_fields() + .into_iter() + .map(|field| resolve_text_field(index, config, Some(field))) + .collect() +} + +#[derive(Clone, Copy)] +struct FieldMatchOptions { + operator: MatchOperator, + boost: f32, + fuzziness: Option<u8>, + max_expansions: usize, + prefix_length: u32, +} + +fn build_field_match_query( + index: &Index, + config: &FullTextIndexConfig, + field: tantivy::schema::Field, + terms: &str, + options: FieldMatchOptions, +) -> Result<Box<dyn Query>> { + validate_boost(options.boost)?; + let tokens = analyze_terms(index, config, terms)?; + let mut query = if tokens.is_empty() { + Box::new(EmptyQuery) as Box<dyn Query> + } else if tokens.len() == 1 { + build_token_query( + field, + &tokens[0], + options.fuzziness, + options.max_expansions, + options.prefix_length, + )? + } else { + let occur = match options.operator { + MatchOperator::Or => Occur::Should, + MatchOperator::And => Occur::Must, + }; + let mut children = Vec::with_capacity(tokens.len()); + for token in tokens { + children.push(( + occur, + build_token_query( + field, + &token, + options.fuzziness, + options.max_expansions, + options.prefix_length, + )?, + )); + } + Box::new(BooleanQuery::new(children)) as Box<dyn Query> + }; + if (options.boost - 1.0).abs() > f32::EPSILON { + query = Box::new(BoostQuery::new(query, options.boost)); + } + Ok(query) +} + +fn analyze_terms(index: &Index, config: &FullTextIndexConfig, terms: &str) -> Result<Vec<String>> { + let tokenizer_name = tokenizer_name(&config.tokenizer); + let mut analyzer = index.tokenizers().get(tokenizer_name).ok_or_else(|| { + FtIndexError::InvalidQuery(format!("tokenizer '{tokenizer_name}' is not registered")) + })?; + let mut tokens = Vec::new(); + let mut token_stream = analyzer.token_stream(terms); + token_stream.process(&mut |token| tokens.push(token.text.clone())); + Ok(tokens) +} + +fn build_token_query( + field: tantivy::schema::Field, + token: &str, + fuzziness: Option<u8>, + max_expansions: usize, + prefix_length: u32, +) -> Result<Box<dyn Query>> { + let fuzziness = fuzziness.unwrap_or_else(|| auto_fuzziness(token)); + if fuzziness == 0 { + return Ok(Box::new(TermQuery::new( + Term::from_field_text(field, token), + IndexRecordOption::WithFreqs, + ))); + } + if fuzziness > 2 { + return Err(FtIndexError::InvalidQuery( + "match query fuzziness must be auto/null or a value in [0, 2]".to_string(), + )); + } + let term = Term::from_field_text(field, token); + let prefix = token + .chars() + .take(prefix_length as usize) + .collect::<String>(); + Ok(Box::new(CappedFuzzyTermQuery::new( + term, + fuzziness, + prefix, + max_expansions, + ))) +} + +fn auto_fuzziness(token: &str) -> u8 { + match token.chars().count() { + 0..=2 => 0, + 3..=5 => 1, + _ => 2, + } +} + fn validate_negative_boost(negative_boost: f32) -> Result<()> { if !negative_boost.is_finite() || !(0.0..=1.0).contains(&negative_boost) { return Err(FtIndexError::InvalidQuery(format!( @@ -602,12 +973,191 @@ } } -fn validate_column(_config: &FullTextIndexConfig, column: &str) -> Result<()> { - if !column.trim().is_empty() { - Ok(()) - } else { - Err(FtIndexError::InvalidQuery( - "full-text query column must not be empty".to_string(), - )) +#[derive(Clone)] +struct CappedFuzzyTermQuery { + term: Term, + distance: u8, + prefix: String, + max_expansions: usize, +} + +impl CappedFuzzyTermQuery { + fn new(term: Term, distance: u8, prefix: String, max_expansions: usize) -> Self { + Self { + term, + distance, + prefix, + max_expansions, + } + } +} + +impl fmt::Debug for CappedFuzzyTermQuery { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "CappedFuzzyTermQuery(term={:?}, distance={}, prefix={:?}, max_expansions={})", + self.term, self.distance, self.prefix, self.max_expansions + ) + } +} + +impl Query for CappedFuzzyTermQuery { + fn weight(&self, _enable_scoring: EnableScoring<'_>) -> tantivy::Result<Box<dyn Weight>> { + let term_value = self.term.value(); + let term_text = term_value.as_str().ok_or_else(|| { + tantivy::TantivyError::InvalidArgument("fuzzy query requires a string term".to_string()) + })?; + let builder = LevenshteinAutomatonBuilder::new(self.distance, true); + let automaton = PrefixedDfaAutomaton::new( + builder.build_dfa(term_text), + self.prefix.as_bytes().to_vec(), + ); + Ok(Box::new(CappedAutomatonWeight { + field: self.term.field(), + automaton, + max_expansions: self.max_expansions, + })) + } + + fn query_terms<'a>(&'a self, visitor: &mut dyn FnMut(&'a Term, bool)) { + visitor(&self.term, false); + } +} + +struct CappedAutomatonWeight<A> { + field: tantivy::schema::Field, + automaton: A, + max_expansions: usize, +} + +impl<A> Weight for CappedAutomatonWeight<A> +where + A: Automaton + Send + Sync + 'static, + A::State: Clone, +{ + fn scorer(&self, reader: &SegmentReader, boost: Score) -> tantivy::Result<Box<dyn Scorer>> { + let inverted_index = reader.inverted_index(self.field)?; + let term_dict = inverted_index.terms(); + let mut term_stream = term_dict.search(&self.automaton).into_stream()?; + let mut docs = Vec::new(); + let mut expansions = 0usize; + while expansions < self.max_expansions && term_stream.advance() { + expansions += 1; + let term_info = term_stream.value(); + let mut block_segment_postings = inverted_index + .read_block_postings_from_terminfo(term_info, IndexRecordOption::Basic)?; + loop { + let block_docs = block_segment_postings.docs(); + if block_docs.is_empty() { + break; + } + docs.extend_from_slice(block_docs); + block_segment_postings.advance(); + } + } + docs.sort_unstable(); + docs.dedup(); + Ok(Box::new(ConstScorer::new(SortedDocSet::new(docs), boost))) + } + + fn explain(&self, reader: &SegmentReader, doc: DocId) -> tantivy::Result<Explanation> { + let mut scorer = self.scorer(reader, 1.0)?; + if scorer.seek(doc) == doc { + Ok(Explanation::new("CappedAutomatonScorer", 1.0)) + } else { + Err(tantivy::TantivyError::InvalidArgument( + "Document does not match fuzzy query".to_string(), + )) + } + } +} + +struct SortedDocSet { + docs: Vec<DocId>, + cursor: usize, + doc: DocId, +} + +impl SortedDocSet { + fn new(docs: Vec<DocId>) -> Self { + let doc = docs.first().copied().unwrap_or(TERMINATED); + Self { + docs, + cursor: 0, + doc, + } + } +} + +impl DocSet for SortedDocSet { + fn advance(&mut self) -> DocId { + if self.doc == TERMINATED { + return TERMINATED; + } + self.cursor += 1; + self.doc = self.docs.get(self.cursor).copied().unwrap_or(TERMINATED); + self.doc + } + + fn seek(&mut self, target: DocId) -> DocId { + if self.doc >= target { + return self.doc; + } + let relative_cursor = self.docs[self.cursor..].partition_point(|doc| *doc < target); + self.cursor += relative_cursor; + self.doc = self.docs.get(self.cursor).copied().unwrap_or(TERMINATED); + self.doc + } + + fn doc(&self) -> DocId { + self.doc + } + + fn size_hint(&self) -> u32 { + if self.doc == TERMINATED { + 0 + } else { + (self.docs.len() - self.cursor) as u32 + } + } +} + +struct PrefixedDfaAutomaton { + dfa: DFA, + prefix: Vec<u8>, +} + +impl PrefixedDfaAutomaton { + fn new(dfa: DFA, prefix: Vec<u8>) -> Self { + Self { dfa, prefix } + } +} + +impl Automaton for PrefixedDfaAutomaton { + type State = (u32, Option<usize>); + + fn start(&self) -> Self::State { + (self.dfa.initial_state(), Some(0)) + } + + fn is_match(&self, state: &Self::State) -> bool { + matches!(self.dfa.distance(state.0), Distance::Exact(_)) + && matches!(state.1, Some(pos) if pos >= self.prefix.len()) + } + + fn can_match(&self, state: &Self::State) -> bool { + state.0 != SINK_STATE && state.1.is_some() + } + + fn accept(&self, state: &Self::State, byte: u8) -> Self::State { + let dfa_state = self.dfa.transition(state.0, byte); + let prefix_state = match state.1 { + None => None, + Some(pos) if pos >= self.prefix.len() => Some(pos), + Some(pos) if self.prefix[pos] == byte => Some(pos + 1), + Some(_) => None, + }; + (dfa_state, prefix_state) } }
diff --git a/core/src/io.rs b/core/src/io.rs index 0320cd9..e8abe11 100644 --- a/core/src/io.rs +++ b/core/src/io.rs
@@ -1,4 +1,5 @@ use std::io; +use std::sync::atomic::{AtomicU64, Ordering}; pub struct ReadRequest<'a> { pub pos: u64, @@ -11,8 +12,76 @@ } } -pub trait SeekRead: Send { - fn pread(&mut self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()>; +/// Positional reader shared by Tantivy file handles. +/// +/// Implementations must be safe to call concurrently. Each call may contain one +/// or more disjoint read requests, and callers may issue multiple `pread` calls +/// at the same time through shared references. +pub trait SeekRead: Send + Sync { + fn pread(&self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()>; +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct FullTextReadMetrics { + pub pread_calls: u64, + pub pread_ranges: u64, + pub pread_bytes: u64, + pub cache_hits: u64, + pub cache_misses: u64, + pub cache_evictions: u64, + pub cached_blocks: u64, +} + +#[derive(Default)] +pub(crate) struct ReadMetrics { + pread_calls: AtomicU64, + pread_ranges: AtomicU64, + pread_bytes: AtomicU64, + cache_hits: AtomicU64, + cache_misses: AtomicU64, + cache_evictions: AtomicU64, + cached_blocks: AtomicU64, +} + +impl ReadMetrics { + pub(crate) fn record_pread(&self, ranges: &[ReadRequest<'_>]) { + self.pread_calls.fetch_add(1, Ordering::Relaxed); + self.pread_ranges + .fetch_add(ranges.len() as u64, Ordering::Relaxed); + self.pread_bytes.fetch_add( + ranges.iter().map(|range| range.buf.len() as u64).sum(), + Ordering::Relaxed, + ); + } + + pub(crate) fn record_cache_hit(&self) { + self.cache_hits.fetch_add(1, Ordering::Relaxed); + } + + pub(crate) fn record_cache_miss(&self) { + self.cache_misses.fetch_add(1, Ordering::Relaxed); + } + + pub(crate) fn record_cache_insert(&self) { + self.cached_blocks.fetch_add(1, Ordering::Relaxed); + } + + pub(crate) fn record_cache_eviction(&self) { + self.cache_evictions.fetch_add(1, Ordering::Relaxed); + self.cached_blocks.fetch_sub(1, Ordering::Relaxed); + } + + pub(crate) fn snapshot(&self) -> FullTextReadMetrics { + FullTextReadMetrics { + pread_calls: self.pread_calls.load(Ordering::Relaxed), + pread_ranges: self.pread_ranges.load(Ordering::Relaxed), + pread_bytes: self.pread_bytes.load(Ordering::Relaxed), + cache_hits: self.cache_hits.load(Ordering::Relaxed), + cache_misses: self.cache_misses.load(Ordering::Relaxed), + cache_evictions: self.cache_evictions.load(Ordering::Relaxed), + cached_blocks: self.cached_blocks.load(Ordering::Relaxed), + } + } } pub trait SeekWrite: Send { @@ -66,7 +135,7 @@ } impl SeekRead for SliceReader { - fn pread(&mut self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> { + fn pread(&self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> { for range in ranges { let start = usize::try_from(range.pos) .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "offset overflow"))?;
diff --git a/core/src/lib.rs b/core/src/lib.rs index 4750845..15bc5f1 100644 --- a/core/src/lib.rs +++ b/core/src/lib.rs
@@ -1,13 +1,14 @@ +mod archive_directory; pub mod config; pub mod error; pub mod index; pub mod io; -pub mod query; +mod query; pub mod storage; pub mod tokenizer; pub use config::{FullTextIndexConfig, FullTextIndexMetadata}; pub use error::{FtIndexError, Result}; -pub use index::{FullTextIndexReader, FullTextIndexWriter, FullTextSearchResult}; -pub use query::{BooleanOccur, FullTextQuery, MatchOperator}; +pub use index::{FullTextDocument, FullTextIndexReader, FullTextIndexWriter, FullTextSearchResult}; +pub use io::FullTextReadMetrics; pub use tokenizer::{TokenizerConfig, TokenizerKind};
diff --git a/core/src/query.rs b/core/src/query.rs index 93ce35e..eb1f0f3 100644 --- a/core/src/query.rs +++ b/core/src/query.rs
@@ -3,7 +3,7 @@ use serde::{Deserialize, Deserializer, Serialize}; #[derive(Clone, Copy, Debug, Default, Serialize, PartialEq, Eq)] -pub enum MatchOperator { +pub(crate) enum MatchOperator { #[default] Or, And, @@ -26,7 +26,7 @@ } #[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq)] -pub enum BooleanOccur { +pub(crate) enum BooleanOccur { Should, Must, MustNot, @@ -51,9 +51,10 @@ #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] #[serde(rename_all = "snake_case")] -pub enum FullTextQuery { +pub(crate) enum QuerySpec { Match { - column: String, + #[serde(default)] + column: Option<String>, #[serde(alias = "query")] terms: String, #[serde(default)] @@ -70,9 +71,28 @@ #[serde(default, alias = "prefixLength")] prefix_length: u32, }, + MultiMatch { + #[serde(alias = "query")] + terms: String, + columns: Vec<String>, + #[serde(default, alias = "boost")] + boosts: Vec<f32>, + #[serde(default)] + operator: MatchOperator, + #[serde( + default = "default_fuzziness", + deserialize_with = "deserialize_fuzziness" + )] + fuzziness: Option<u8>, + #[serde(default = "default_max_expansions", alias = "maxExpansions")] + max_expansions: usize, + #[serde(default, alias = "prefixLength")] + prefix_length: u32, + }, #[serde(alias = "phrase")] MatchPhrase { - column: String, + #[serde(default)] + column: Option<String>, #[serde(alias = "query")] terms: String, #[serde(default)] @@ -80,17 +100,17 @@ }, Boolean { #[serde(default)] - should: Vec<FullTextQuery>, + should: Vec<QuerySpec>, #[serde(default)] - must: Vec<FullTextQuery>, + must: Vec<QuerySpec>, #[serde(default)] - must_not: Vec<FullTextQuery>, + must_not: Vec<QuerySpec>, #[serde(default)] - queries: Vec<(BooleanOccur, FullTextQuery)>, + queries: Vec<(BooleanOccur, QuerySpec)>, }, Boost { - positive: Box<FullTextQuery>, - negative: Box<FullTextQuery>, + positive: Box<QuerySpec>, + negative: Box<QuerySpec>, #[serde(default = "default_negative_boost")] negative_boost: f32, }, @@ -129,46 +149,8 @@ } } -impl FullTextQuery { - pub fn match_query(terms: impl Into<String>, column: impl Into<String>) -> Self { - Self::Match { - column: column.into(), - terms: terms.into(), - operator: MatchOperator::Or, - boost: 1.0, - fuzziness: Some(0), - max_expansions: 50, - prefix_length: 0, - } - } - - pub fn phrase(terms: impl Into<String>, column: impl Into<String>) -> Self { - Self::MatchPhrase { - column: column.into(), - terms: terms.into(), - slop: 0, - } - } - - pub fn operator_and(mut self) -> Self { - if let Self::Match { operator, .. } = &mut self { - *operator = MatchOperator::And; - } - self - } - - pub fn operator_or(mut self) -> Self { - if let Self::Match { operator, .. } = &mut self { - *operator = MatchOperator::Or; - } - self - } - - pub fn to_json(&self) -> Result<String> { - serde_json::to_string(self).map_err(FtIndexError::from) - } - - pub fn from_json(json: &str) -> Result<Self> { +impl QuerySpec { + pub(crate) fn from_json(json: &str) -> Result<Self> { serde_json::from_str(json).map_err(FtIndexError::from) } }
diff --git a/core/src/storage.rs b/core/src/storage.rs index 25f18a6..d5ca294 100644 --- a/core/src/storage.rs +++ b/core/src/storage.rs
@@ -2,9 +2,12 @@ use crate::error::{FtIndexError, Result}; use crate::io::{ReadRequest, SeekRead, SeekWrite}; use serde::{Deserialize, Serialize}; +use std::collections::HashSet; +use std::path::{Component, Path}; pub const FORMAT_MAGIC: &[u8; 8] = b"PFTIDX01"; pub const FORMAT_VERSION: u32 = 1; +const MAX_HEADER_BYTES: usize = 16 * 1024 * 1024; const MAX_ARCHIVE_READ_BATCH_BYTES: usize = 64 * 1024 * 1024; const MAX_ARCHIVE_READ_BATCH_RANGES: usize = 64; @@ -27,6 +30,13 @@ files: &[(String, Vec<u8>)], ) -> Result<()> { let header_json = serde_json::to_vec(header)?; + if header_json.len() > MAX_HEADER_BYTES { + return Err(FtIndexError::InvalidStorage(format!( + "header too large: {} bytes exceeds {} bytes", + header_json.len(), + MAX_HEADER_BYTES + ))); + } output.write_all(FORMAT_MAGIC)?; output.write_all(&FORMAT_VERSION.to_be_bytes())?; output.write_all( @@ -42,7 +52,7 @@ Ok(()) } -pub fn read_header<R: SeekRead>(input: &mut R) -> Result<(IndexHeader, u64)> { +pub fn read_header<R: SeekRead>(input: &R) -> Result<(IndexHeader, u64)> { let mut fixed = [0u8; 16]; read_exact_at(input, 0, &mut fixed)?; if &fixed[0..8] != FORMAT_MAGIC { @@ -55,14 +65,77 @@ ))); } let header_len = u32::from_be_bytes(fixed[12..16].try_into().expect("slice length")) as usize; + if header_len > MAX_HEADER_BYTES { + return Err(FtIndexError::InvalidStorage(format!( + "header too large: {header_len} bytes exceeds {MAX_HEADER_BYTES} bytes" + ))); + } let mut header_json = vec![0u8; header_len]; read_exact_at(input, 16, &mut header_json)?; let header = serde_json::from_slice(&header_json)?; + validate_header(&header)?; Ok((header, 16 + header_len as u64)) } +fn validate_header(header: &IndexHeader) -> Result<()> { + header.metadata.config.validate()?; + if header.metadata.tantivy_version.trim().is_empty() { + return Err(FtIndexError::InvalidStorage( + "missing Tantivy version in index metadata".to_string(), + )); + } + if header.files.is_empty() { + return Err(FtIndexError::InvalidStorage( + "archive file list must not be empty".to_string(), + )); + } + + let mut names = HashSet::with_capacity(header.files.len()); + let mut ranges = Vec::with_capacity(header.files.len()); + for file in &header.files { + validate_archive_file_name(&file.name)?; + if !names.insert(file.name.as_str()) { + return Err(FtIndexError::InvalidStorage(format!( + "duplicate archive file '{}'", + file.name + ))); + } + let end = file.offset.checked_add(file.length).ok_or_else(|| { + FtIndexError::InvalidStorage(format!("archive file '{}' range overflow", file.name)) + })?; + ranges.push((file.offset, end, file.name.as_str())); + } + + ranges.sort_by_key(|(start, _, _)| *start); + let mut previous_end = 0u64; + for (start, end, name) in ranges { + if start < previous_end { + return Err(FtIndexError::InvalidStorage(format!( + "archive file '{name}' overlaps a previous file" + ))); + } + previous_end = end; + } + Ok(()) +} + +fn validate_archive_file_name(name: &str) -> Result<()> { + if name.is_empty() { + return Err(FtIndexError::InvalidStorage( + "archive file name must not be empty".to_string(), + )); + } + let mut components = Path::new(name).components(); + match (components.next(), components.next()) { + (Some(Component::Normal(_)), None) => Ok(()), + _ => Err(FtIndexError::InvalidStorage(format!( + "archive file '{name}' must be a plain file name" + ))), + } +} + pub fn read_archive_files<R: SeekRead>( - input: &mut R, + input: &R, body_start: u64, files: &[ArchiveFileEntry], ) -> Result<Vec<(String, Vec<u8>)>> { @@ -72,7 +145,7 @@ } pub fn read_archive_files_with<R, F, T>( - input: &mut R, + input: &R, body_start: u64, files: &[ArchiveFileEntry], mut consume: F, @@ -123,7 +196,7 @@ } fn read_archive_file_batch<R, F, T>( - input: &mut R, + input: &R, pending: &mut Vec<PendingArchiveFile>, consume: &mut F, output: &mut Vec<T>, @@ -146,7 +219,7 @@ Ok(()) } -pub fn read_exact_at<R: SeekRead>(input: &mut R, pos: u64, buf: &mut [u8]) -> Result<()> { +pub fn read_exact_at<R: SeekRead>(input: &R, pos: u64, buf: &mut [u8]) -> Result<()> { let mut request = [ReadRequest::new(pos, buf)]; input.pread(&mut request)?; Ok(()) @@ -155,17 +228,21 @@ #[cfg(test)] mod tests { use super::*; + use crate::config::{FullTextIndexConfig, FullTextIndexMetadata}; + use crate::io::{PosWriter, SliceReader}; + use std::sync::atomic::{AtomicUsize, Ordering}; struct CountingReader { data: Vec<u8>, - pread_batches: usize, - max_ranges_per_batch: usize, + pread_batches: AtomicUsize, + max_ranges_per_batch: AtomicUsize, } impl SeekRead for CountingReader { - fn pread(&mut self, ranges: &mut [ReadRequest<'_>]) -> std::io::Result<()> { - self.pread_batches += 1; - self.max_ranges_per_batch = self.max_ranges_per_batch.max(ranges.len()); + fn pread(&self, ranges: &mut [ReadRequest<'_>]) -> std::io::Result<()> { + self.pread_batches.fetch_add(1, Ordering::SeqCst); + self.max_ranges_per_batch + .fetch_max(ranges.len(), Ordering::SeqCst); for range in ranges { let start = usize::try_from(range.pos).map_err(|_| { std::io::Error::new(std::io::ErrorKind::InvalidInput, "offset overflow") @@ -187,10 +264,10 @@ #[test] fn read_archive_files_uses_one_batched_pread() { - let mut reader = CountingReader { + let reader = CountingReader { data: b"headerabcdef".to_vec(), - pread_batches: 0, - max_ranges_per_batch: 0, + pread_batches: AtomicUsize::new(0), + max_ranges_per_batch: AtomicUsize::new(0), }; let files = vec![ ArchiveFileEntry { @@ -210,10 +287,10 @@ }, ]; - let files = read_archive_files(&mut reader, 6, &files).expect("archive files"); + let files = read_archive_files(&reader, 6, &files).expect("archive files"); - assert_eq!(reader.pread_batches, 1); - assert_eq!(reader.max_ranges_per_batch, 3); + assert_eq!(reader.pread_batches.load(Ordering::SeqCst), 1); + assert_eq!(reader.max_ranges_per_batch.load(Ordering::SeqCst), 3); assert_eq!( files, vec![ @@ -223,4 +300,114 @@ ] ); } + + #[test] + fn read_header_rejects_invalid_archive_metadata() { + let cases = [ + ( + vec![ + ArchiveFileEntry { + name: "meta.json".to_string(), + offset: 0, + length: 2, + }, + ArchiveFileEntry { + name: "meta.json".to_string(), + offset: 2, + length: 2, + }, + ], + "duplicate archive file", + ), + ( + vec![ArchiveFileEntry { + name: "../meta.json".to_string(), + offset: 0, + length: 2, + }], + "must be a plain file name", + ), + ( + vec![ + ArchiveFileEntry { + name: "a".to_string(), + offset: 0, + length: 3, + }, + ArchiveFileEntry { + name: "b".to_string(), + offset: 2, + length: 3, + }, + ], + "overlaps a previous file", + ), + ]; + + for (files, expected) in cases { + let bytes = encode_header(IndexHeader { + metadata: valid_metadata(), + files, + }); + let err = read_header(&SliceReader::new(bytes)).expect_err("invalid header"); + assert!( + err.to_string().contains(expected), + "expected '{expected}', got '{err}'" + ); + } + } + + #[test] + fn read_header_rejects_missing_tantivy_version() { + let bytes = encode_header(IndexHeader { + metadata: FullTextIndexMetadata { + tantivy_version: String::new(), + ..valid_metadata() + }, + files: vec![ArchiveFileEntry { + name: "meta.json".to_string(), + offset: 0, + length: 2, + }], + }); + + let err = read_header(&SliceReader::new(bytes)).expect_err("invalid header"); + + assert!(err + .to_string() + .contains("missing Tantivy version in index metadata")); + } + + #[test] + fn read_header_rejects_oversized_header_before_allocating() { + const OVERSIZED_HEADER_LEN: u32 = 16 * 1024 * 1024 + 1; + + let mut bytes = Vec::new(); + bytes.extend_from_slice(FORMAT_MAGIC); + bytes.extend_from_slice(&FORMAT_VERSION.to_be_bytes()); + bytes.extend_from_slice(&OVERSIZED_HEADER_LEN.to_be_bytes()); + + let err = read_header(&SliceReader::new(bytes)).expect_err("oversized header should fail"); + + assert!(err.to_string().contains("header too large")); + } + + fn valid_metadata() -> FullTextIndexMetadata { + FullTextIndexMetadata { + config: FullTextIndexConfig::new(), + document_count: 1, + tantivy_version: tantivy::version().to_string(), + } + } + + fn encode_header(header: IndexHeader) -> Vec<u8> { + let mut bytes = Vec::new(); + write_envelope( + &mut PosWriter::new(&mut bytes), + &header, + &[("meta.json".to_string(), b"ab".to_vec())], + ) + .expect("write envelope"); + bytes + } }
diff --git a/core/src/tokenizer.rs b/core/src/tokenizer.rs index 783877d..448bd75 100644 --- a/core/src/tokenizer.rs +++ b/core/src/tokenizer.rs
@@ -15,6 +15,7 @@ } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(default)] pub struct TokenizerConfig { pub tokenizer: TokenizerKind, pub ngram_min_gram: usize, @@ -36,19 +37,19 @@ fn default() -> Self { Self { tokenizer: TokenizerKind::Default, - ngram_min_gram: 2, - ngram_max_gram: 2, + ngram_min_gram: 3, + ngram_max_gram: 3, ngram_prefix_only: false, jieba_search_mode: true, jieba_ordinal_position: true, lower_case: true, max_token_length: 40, - ascii_folding: false, - stem: false, + ascii_folding: true, + stem: true, language: "english".to_string(), - remove_stop_words: false, + remove_stop_words: true, stop_words: Vec::new(), - with_position: true, + with_position: false, } } } @@ -61,7 +62,8 @@ .strip_prefix("fulltext.") .or_else(|| raw_key.strip_prefix("tantivy.")) .unwrap_or(raw_key.as_str()); - match key { + let key = normalize_option_key(key); + match key.as_str() { "tokenizer" => { config.tokenizer = parse_tokenizer(value)?; } @@ -129,12 +131,16 @@ if self.max_token_length == 0 { return invalid("max-token-length", "must be positive"); } + if !self.remove_stop_words && !self.stop_words.is_empty() { + return invalid("stop-words", "requires remove-stop-words=true"); + } Ok(()) } } fn parse_tokenizer(value: &str) -> Result<TokenizerKind> { - match value.trim().to_lowercase().as_str() { + let value = value.trim().to_lowercase(); + match value.as_str() { "default" => Ok(TokenizerKind::Default), "simple" => Ok(TokenizerKind::Simple), "whitespace" => Ok(TokenizerKind::Whitespace), @@ -148,22 +154,28 @@ } } -fn parse_usize(key: &str, value: &str) -> Result<usize> { +fn normalize_option_key(key: &str) -> String { + key.trim().to_lowercase() +} + +fn parse_usize(key: impl Into<String>, value: &str) -> Result<usize> { + let key = key.into(); value .trim() .parse() .map_err(|_| FtIndexError::InvalidOption { - key: key.to_string(), + key, message: format!("expected unsigned integer, got '{value}'"), }) } -fn parse_bool(key: &str, value: &str) -> Result<bool> { +fn parse_bool(key: impl Into<String>, value: &str) -> Result<bool> { + let key = key.into(); match value.trim().to_lowercase().as_str() { "true" => Ok(true), "false" => Ok(false), _ => Err(FtIndexError::InvalidOption { - key: key.to_string(), + key, message: format!("expected boolean, got '{value}'"), }), }
diff --git a/core/tests/core_roundtrip.rs b/core/tests/core_roundtrip.rs index 0ad7347..51e83be 100644 --- a/core/tests/core_roundtrip.rs +++ b/core/tests/core_roundtrip.rs
@@ -1,6 +1,7 @@ use paimon_ftindex_core::io::{PosWriter, ReadRequest, SeekRead, SliceReader}; +use paimon_ftindex_core::storage::{read_header, write_envelope, ArchiveFileEntry, IndexHeader}; use paimon_ftindex_core::{ - FullTextIndexConfig, FullTextIndexReader, FullTextIndexWriter, FullTextQuery, MatchOperator, + FullTextIndexConfig, FullTextIndexMetadata, FullTextIndexReader, FullTextIndexWriter, TokenizerConfig, TokenizerKind, }; use roaring::RoaringTreemap; @@ -9,7 +10,11 @@ use std::sync::{Arc, Mutex}; fn build_index() -> anyhow::Result<Vec<u8>> { - let mut writer = FullTextIndexWriter::new(FullTextIndexConfig::new())?; + build_index_with_config(FullTextIndexConfig::new()) +} + +fn build_index_with_config(config: FullTextIndexConfig) -> anyhow::Result<Vec<u8>> { + let mut writer = FullTextIndexWriter::new(config)?; writer.add_document(10, "Apache Paimon supports full text search")?; writer.add_document(11, "Tantivy is a Rust search engine")?; writer.add_document(12, "Paimon tables can use indexes")?; @@ -22,9 +27,70 @@ Ok(bytes) } +fn match_query(terms: &str, column: &str) -> String { + serde_json::json!({ + "match": { + "query": terms, + "column": column, + } + }) + .to_string() +} + +fn match_query_without_column(terms: &str) -> String { + serde_json::json!({ + "match": { + "query": terms, + } + }) + .to_string() +} + +fn match_query_and(terms: &str, column: &str) -> String { + serde_json::json!({ + "match": { + "query": terms, + "column": column, + "operator": "AND", + } + }) + .to_string() +} + +fn fuzzy_match_query( + terms: &str, + column: &str, + fuzziness: u8, + max_expansions: usize, + prefix_length: u32, +) -> String { + serde_json::json!({ + "match": { + "query": terms, + "column": column, + "fuzziness": fuzziness, + "max_expansions": max_expansions, + "prefix_length": prefix_length, + } + }) + .to_string() +} + +fn phrase_query(terms: &str, column: &str) -> String { + serde_json::json!({ + "match_phrase": { + "query": terms, + "column": column, + } + }) + .to_string() +} + #[derive(Default)] struct ReadStats { + pread_calls: usize, max_ranges_per_batch: usize, + total_bytes_read: usize, } struct CountingSliceReader { @@ -39,10 +105,12 @@ } impl SeekRead for CountingSliceReader { - fn pread(&mut self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> { + fn pread(&self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> { { let mut stats = self.stats.lock().unwrap(); + stats.pread_calls += 1; stats.max_ranges_per_batch = stats.max_ranges_per_batch.max(ranges.len()); + stats.total_bytes_read += ranges.iter().map(|range| range.buf.len()).sum::<usize>(); } for range in ranges { let start = usize::try_from(range.pos) @@ -63,27 +131,118 @@ } #[test] -fn reader_open_batches_archive_file_reads() -> anyhow::Result<()> { +fn reader_open_does_not_load_all_archive_files() -> anyhow::Result<()> { + let bytes = build_index()?; + let (header, body_start) = read_header(&SliceReader::new(bytes.clone()))?; + let body_len = header + .files + .iter() + .map(|file| file.offset + file.length) + .max() + .unwrap_or(0) as usize; + let stats = Arc::new(Mutex::new(ReadStats::default())); + let input = CountingSliceReader::new(bytes, Arc::clone(&stats)); + + let reader = FullTextIndexReader::open(input)?; + assert_eq!(reader.metadata().document_count, 3); + let bytes_read_at_open = stats.lock().unwrap().total_bytes_read; + let header_bytes = usize::try_from(body_start)?; + assert!( + bytes_read_at_open < header_bytes + body_len, + "reader open should not read the complete Tantivy archive body" + ); + let result = reader.search(match_query("paimon", "text"), 10)?; + assert_eq!(result.row_ids.len(), 2); + Ok(()) +} + +#[test] +fn repeated_search_reuses_tantivy_reader_io() -> anyhow::Result<()> { let bytes = build_index()?; let stats = Arc::new(Mutex::new(ReadStats::default())); let input = CountingSliceReader::new(bytes, Arc::clone(&stats)); - let mut reader = FullTextIndexReader::open(input)?; - let result = reader.search(FullTextQuery::match_query("paimon", "text"), 10)?; + let reader = FullTextIndexReader::open(input)?; + let calls_after_open = stats.lock().unwrap().pread_calls; + let result = reader.search(match_query("missing", "text"), 10)?; + assert!(result.row_ids.is_empty()); + let calls_after_first_search = stats.lock().unwrap().pread_calls; + let result = reader.search(match_query("missing", "text"), 10)?; + assert!(result.row_ids.is_empty()); + let calls_after_second_search = stats.lock().unwrap().pread_calls; - assert_eq!(result.row_ids.len(), 2); + let first_search_calls = calls_after_first_search - calls_after_open; + let second_search_calls = calls_after_second_search - calls_after_first_search; assert!( - stats.lock().unwrap().max_ranges_per_batch > 1, - "reader open should batch archive file reads into one pread call" + second_search_calls < first_search_calls, + "second search should reuse Tantivy reader metadata; first search used {first_search_calls} pread calls, second used {second_search_calls}" ); Ok(()) } #[test] +fn prewarm_initializes_tantivy_reader_before_first_search() -> anyhow::Result<()> { + let bytes = build_index()?; + + let cold_stats = Arc::new(Mutex::new(ReadStats::default())); + let cold_reader = FullTextIndexReader::open(CountingSliceReader::new( + bytes.clone(), + Arc::clone(&cold_stats), + ))?; + let cold_calls_after_open = cold_stats.lock().unwrap().pread_calls; + let cold_result = cold_reader.search(match_query("missing", "text"), 10)?; + assert!(cold_result.row_ids.is_empty()); + let cold_calls_after_search = cold_stats.lock().unwrap().pread_calls; + let cold_first_search_calls = cold_calls_after_search - cold_calls_after_open; + + let warm_stats = Arc::new(Mutex::new(ReadStats::default())); + let warm_reader = + FullTextIndexReader::open(CountingSliceReader::new(bytes, Arc::clone(&warm_stats)))?; + let warm_calls_after_open = warm_stats.lock().unwrap().pread_calls; + warm_reader.prewarm()?; + let warm_calls_after_prewarm = warm_stats.lock().unwrap().pread_calls; + assert!( + warm_calls_after_prewarm > warm_calls_after_open, + "prewarm should eagerly initialize Tantivy reader I/O" + ); + + let warm_result = warm_reader.search(match_query("missing", "text"), 10)?; + assert!(warm_result.row_ids.is_empty()); + let warm_calls_after_search = warm_stats.lock().unwrap().pread_calls; + let warm_first_search_calls = warm_calls_after_search - warm_calls_after_prewarm; + + assert!( + warm_first_search_calls < cold_first_search_calls, + "prewarm should move first-search reader I/O earlier; cold first search used {cold_first_search_calls} pread calls, warm first search used {warm_first_search_calls}" + ); + Ok(()) +} + +#[test] +fn read_metrics_report_pread_and_cache_activity() -> anyhow::Result<()> { + let bytes = build_index()?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + + let after_open = reader.read_metrics(); + assert!(after_open.pread_calls >= 2); + assert!(after_open.pread_bytes > 16); + + let result = reader.search(match_query("missing", "text"), 10)?; + assert!(result.row_ids.is_empty()); + let after_first_search = reader.read_metrics(); + assert!(after_first_search.pread_calls > after_open.pread_calls); + assert!(after_first_search.pread_bytes > after_open.pread_bytes); + assert!(after_first_search.cache_misses > after_open.cache_misses); + assert!(after_first_search.cached_blocks > 0); + + Ok(()) +} + +#[test] fn match_query_round_trip() -> anyhow::Result<()> { let bytes = build_index()?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; - let result = reader.search(FullTextQuery::match_query("paimon", "text"), 10)?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let result = reader.search(match_query("paimon", "text"), 10)?; assert_eq!(reader.metadata().document_count, 3); assert_eq!(result.row_ids.len(), 2); @@ -94,18 +253,115 @@ } #[test] +fn written_header_has_single_source_of_truth_for_format_and_fields() -> anyhow::Result<()> { + let bytes = build_index()?; + let header_len = u32::from_be_bytes(bytes[12..16].try_into()?) as usize; + let header: serde_json::Value = serde_json::from_slice(&bytes[16..16 + header_len])?; + + assert!(header.get("format_version").is_none()); + assert!(header["metadata"].get("format_version").is_none()); + assert!(header["metadata"]["config"].get("text_field").is_none()); + assert_eq!( + header["metadata"]["config"]["text_fields"], + serde_json::json!(["text"]) + ); + Ok(()) +} + +#[test] +fn storage_v1_envelope_matches_golden_file() -> anyhow::Result<()> { + let header = IndexHeader { + metadata: FullTextIndexMetadata { + config: FullTextIndexConfig::new(), + document_count: 2, + tantivy_version: "0.26.0".to_string(), + }, + files: vec![ + ArchiveFileEntry { + name: "meta.json".to_string(), + offset: 0, + length: 2, + }, + ArchiveFileEntry { + name: "segment.idx".to_string(), + offset: 2, + length: 4, + }, + ], + }; + let files = vec![ + ("meta.json".to_string(), b"ab".to_vec()), + ("segment.idx".to_string(), b"cdef".to_vec()), + ]; + + let mut bytes = Vec::new(); + write_envelope(&mut PosWriter::new(&mut bytes), &header, &files)?; + + let expected = decode_hex(include_str!("golden/storage_v1_envelope.hex"))?; + assert_eq!(bytes, expected); + + let (actual_header, body_start) = read_header(&SliceReader::new(bytes.clone()))?; + assert_eq!(actual_header, header); + assert_eq!(&bytes[body_start as usize..], b"abcdef"); + Ok(()) +} + +#[test] +fn reader_rejects_mismatched_tantivy_version() { + let mut bytes = Vec::new(); + let header = IndexHeader { + metadata: FullTextIndexMetadata { + config: FullTextIndexConfig::new(), + document_count: 0, + tantivy_version: "0.0.0".to_string(), + }, + files: vec![ArchiveFileEntry { + name: "meta.json".to_string(), + offset: 0, + length: 0, + }], + }; + write_envelope( + &mut PosWriter::new(&mut bytes), + &header, + &[("meta.json".to_string(), Vec::new())], + ) + .expect("write envelope"); + + let err = match FullTextIndexReader::open(SliceReader::new(bytes)) { + Ok(_) => panic!("Tantivy version mismatch should fail"), + Err(err) => err, + }; + + assert!(err + .to_string() + .contains("unsupported Tantivy index version 0.0.0")); +} + +fn decode_hex(hex: &str) -> anyhow::Result<Vec<u8>> { + let mut bytes = Vec::new(); + let mut high = None; + for ch in hex.chars().filter(|ch| !ch.is_whitespace()) { + let value = ch + .to_digit(16) + .ok_or_else(|| anyhow::anyhow!("invalid hex digit: {ch}"))? as u8; + if let Some(previous) = high.take() { + bytes.push((previous << 4) | value); + } else { + high = Some(value); + } + } + if high.is_some() { + anyhow::bail!("hex input has an odd number of digits"); + } + Ok(bytes) +} + +#[test] fn match_query_and_operator_filters_terms() -> anyhow::Result<()> { let bytes = build_index()?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; - let query = FullTextQuery::Match { - column: "text".to_string(), - terms: "paimon indexes".to_string(), - operator: MatchOperator::And, - boost: 1.0, - fuzziness: Some(0), - max_expansions: 50, - prefix_length: 0, - }; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let query = match_query_and("paimon indexes", "text"); let result = reader.search(query, 10)?; assert_eq!(result.row_ids, vec![12]); @@ -115,9 +371,9 @@ #[test] fn search_with_roaring_filter_limits_allowed_row_ids_before_top_docs() -> anyhow::Result<()> { let bytes = build_index()?; - let query = FullTextQuery::match_query("paimon", "text"); + let query = match_query("paimon", "text"); - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes.clone()))?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes.clone()))?; let unfiltered_top = reader.search(query.clone(), 1)?.row_ids[0]; let allowed_id = if unfiltered_top == 10 { 12 } else { 10 }; @@ -126,7 +382,7 @@ let mut filter_bytes = Vec::new(); allowed.serialize_into(&mut filter_bytes)?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; let result = reader.search_with_roaring_filter(query, 1, &filter_bytes)?; assert_eq!(result.row_ids, vec![allowed_id]); @@ -141,12 +397,9 @@ let mut filter_bytes = Vec::new(); empty.serialize_into(&mut filter_bytes)?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; - let result = reader.search_with_roaring_filter( - FullTextQuery::match_query("paimon", "text"), - 10, - &filter_bytes, - )?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let result = + reader.search_with_roaring_filter(match_query("paimon", "text"), 10, &filter_bytes)?; assert!(result.row_ids.is_empty()); assert!(result.scores.is_empty()); @@ -168,12 +421,9 @@ let mut filter_bytes = Vec::new(); allowed.serialize_into(&mut filter_bytes)?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; - let result = reader.search_with_roaring_filter( - FullTextQuery::match_query("paimon", "text"), - 10, - &filter_bytes, - )?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let result = + reader.search_with_roaring_filter(match_query("paimon", "text"), 10, &filter_bytes)?; assert_eq!(result.row_ids, vec![allowed_id]); Ok(()) @@ -182,13 +432,9 @@ #[test] fn search_rejects_invalid_roaring_filter_bytes() -> anyhow::Result<()> { let bytes = build_index()?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; let err = reader - .search_with_roaring_filter( - FullTextQuery::match_query("paimon", "text"), - 10, - b"not roaring", - ) + .search_with_roaring_filter(match_query("paimon", "text"), 10, b"not roaring") .expect_err("invalid filter bytes should fail"); assert!(err.to_string().contains("invalid RoaringTreemap filter")); @@ -197,9 +443,9 @@ #[test] fn phrase_query_uses_positions() -> anyhow::Result<()> { - let bytes = build_index()?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; - let result = reader.search(FullTextQuery::phrase("full text", "text"), 10)?; + let bytes = build_index_with_config(FullTextIndexConfig::new().with_positions(true))?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let result = reader.search(phrase_query("full text", "text"), 10)?; assert_eq!(result.row_ids, vec![10]); Ok(()) @@ -218,8 +464,8 @@ let mut bytes = Vec::new(); writer.write(&mut PosWriter::new(&mut bytes))?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; - let result = reader.search(FullTextQuery::match_query("中华", "text"), 10)?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let result = reader.search(match_query("中华", "text"), 10)?; assert_eq!(result.row_ids, vec![20]); Ok(()) @@ -230,6 +476,7 @@ let config = FullTextIndexConfig::new().tokenizer(TokenizerConfig { tokenizer: TokenizerKind::Jieba, jieba_ordinal_position: true, + with_position: true, ..TokenizerConfig::default() }); let mut writer = FullTextIndexWriter::new(config)?; @@ -239,8 +486,8 @@ let mut bytes = Vec::new(); writer.write(&mut PosWriter::new(&mut bytes))?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; - let result = reader.search(FullTextQuery::phrase("北京大学", "text"), 10)?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let result = reader.search(phrase_query("北京大学", "text"), 10)?; assert_eq!(result.row_ids, vec![30]); Ok(()) @@ -268,6 +515,40 @@ } #[test] +fn tokenizer_options_parse_ngram_and_stop_word_settings() -> anyhow::Result<()> { + let mut options = HashMap::new(); + options.insert("fulltext.tokenizer".to_string(), "ngram".to_string()); + options.insert("fulltext.ngram.min-gram".to_string(), "2".to_string()); + options.insert("fulltext.ngram.max-gram".to_string(), "4".to_string()); + options.insert("fulltext.ngram.prefix-only".to_string(), "true".to_string()); + options.insert( + "fulltext.stop-words".to_string(), + "apache;paimon".to_string(), + ); + + let config = TokenizerConfig::from_options(&options)?; + + assert_eq!(config.tokenizer, TokenizerKind::Ngram); + assert_eq!(config.ngram_min_gram, 2); + assert_eq!(config.ngram_max_gram, 4); + assert!(config.ngram_prefix_only); + assert_eq!(config.stop_words, vec!["apache", "paimon"]); + Ok(()) +} + +#[test] +fn stop_words_require_stop_word_removal() { + let mut options = HashMap::new(); + options.insert("remove-stop-words".to_string(), "false".to_string()); + options.insert("stop-words".to_string(), "apache".to_string()); + + let err = TokenizerConfig::from_options(&options) + .expect_err("stop words should require remove-stop-words=true"); + + assert!(err.to_string().contains("requires remove-stop-words=true")); +} + +#[test] fn boost_query_requires_positive_match() -> anyhow::Result<()> { let mut writer = FullTextIndexWriter::new(FullTextIndexConfig::new())?; writer.add_document(1, "apache paimon")?; @@ -276,12 +557,15 @@ let mut bytes = Vec::new(); writer.write(&mut PosWriter::new(&mut bytes))?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; - let query = FullTextQuery::Boost { - positive: Box::new(FullTextQuery::match_query("paimon", "text")), - negative: Box::new(FullTextQuery::match_query("tantivy", "text")), - negative_boost: 0.5, - }; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let query = serde_json::json!({ + "boost": { + "positive": {"match": {"query": "paimon", "column": "text"}}, + "negative": {"match": {"query": "tantivy", "column": "text"}}, + "negative_boost": 0.5, + } + }) + .to_string(); let result = reader.search(query, 10)?; assert_eq!(result.row_ids, vec![1]); @@ -297,12 +581,15 @@ let mut bytes = Vec::new(); writer.write(&mut PosWriter::new(&mut bytes))?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; - let query = FullTextQuery::Boost { - positive: Box::new(FullTextQuery::match_query("paimon", "text")), - negative: Box::new(FullTextQuery::match_query("bad", "text")), - negative_boost: 0.5, - }; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let query = serde_json::json!({ + "boost": { + "positive": {"match": {"query": "paimon", "column": "text"}}, + "negative": {"match": {"query": "bad", "column": "text"}}, + "negative_boost": 0.5, + } + }) + .to_string(); let result = reader.search(query, 10)?; assert_eq!(result.row_ids, vec![1, 2]); @@ -311,43 +598,23 @@ } #[test] -fn query_json_round_trip() -> anyhow::Result<()> { - let query = FullTextQuery::match_query("apache paimon", "text").operator_and(); - let json = query.to_json()?; - let parsed = FullTextQuery::from_json(&json)?; - - assert_eq!(parsed, query); - Ok(()) -} - -#[test] -fn parse_paimon_match_query_json_aliases() -> anyhow::Result<()> { - let query = FullTextQuery::from_json( - r#"{"match":{"column":"text","query":"paimon","operator":"AND","boost":2.0,"fuzziness":0,"maxExpansions":50,"prefixLength":0}}"#, - )?; - - assert_eq!( - query, - FullTextQuery::Match { - column: "text".to_string(), - terms: "paimon".to_string(), - operator: MatchOperator::And, - boost: 2.0, - fuzziness: Some(0), - max_expansions: 50, - prefix_length: 0, - } - ); - Ok(()) -} - -#[test] -fn search_accepts_paimon_logical_column_name() -> anyhow::Result<()> { +fn search_accepts_paimon_match_query_aliases() -> anyhow::Result<()> { let bytes = build_index()?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; - let query = FullTextQuery::from_json( - r#"{"match":{"column":"content","query":"paimon","operator":"OR"}}"#, + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let result = reader.search( + r#"{"match":{"column":"text","query":"paimon indexes","operator":"AND","boost":2.0,"fuzziness":0,"maxExpansions":50,"prefixLength":0}}"#, + 10, )?; + + assert_eq!(result.row_ids, vec![12]); + Ok(()) +} + +#[test] +fn match_query_can_omit_column_for_single_field_index() -> anyhow::Result<()> { + let bytes = build_index()?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let query = r#"{"match":{"query":"paimon","operator":"OR"}}"#; let result = reader.search(query, 10)?; assert_eq!(result.row_ids.len(), 2); @@ -357,12 +624,55 @@ } #[test] -fn parse_paimon_boolean_query_json_forms() -> anyhow::Result<()> { - let query = FullTextQuery::from_json( - r#"{"boolean":{"must":[{"match":{"column":"text","terms":"paimon"}}],"must_not":[{"phrase":{"column":"text","query":"legacy"}}]}}"#, +fn match_query_without_column_searches_all_indexed_fields() -> anyhow::Result<()> { + let config = FullTextIndexConfig::new().with_text_fields(["title", "body"]); + let mut writer = FullTextIndexWriter::new(config)?; + writer.add_document_fields( + 1, + [ + ("title".to_string(), "Apache Paimon".to_string()), + ("body".to_string(), "lake storage".to_string()), + ], )?; + writer.add_document_fields( + 2, + [ + ("title".to_string(), "Tantivy".to_string()), + ("body".to_string(), "Rust search engine".to_string()), + ], + )?; + + let mut bytes = Vec::new(); + writer.write(&mut PosWriter::new(&mut bytes))?; + + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let query = match_query_without_column("rust"); + let result = reader.search(query, 10)?; + + assert_eq!(result.row_ids, vec![2]); + Ok(()) +} + +#[test] +fn search_rejects_unknown_column_name() -> anyhow::Result<()> { let bytes = build_index()?; - let mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let query = r#"{"match":{"column":"content","query":"paimon","operator":"OR"}}"#; + let err = reader + .search(query, 10) + .expect_err("unknown column should fail"); + + assert!(err + .to_string() + .contains("column 'content' is not configured")); + Ok(()) +} + +#[test] +fn search_accepts_paimon_boolean_query_forms() -> anyhow::Result<()> { + let query = r#"{"boolean":{"must":[{"match":{"column":"text","terms":"paimon"}}],"must_not":[{"phrase":{"column":"text","query":"legacy"}}]}}"#; + let bytes = build_index_with_config(FullTextIndexConfig::new().with_positions(true))?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; let result = reader.search(query, 10)?; assert_eq!(result.row_ids.len(), 2); @@ -370,3 +680,187 @@ assert!(result.row_ids.contains(&12)); Ok(()) } + +#[test] +fn default_tokenizer_stems_removes_stop_words_and_folds_ascii() -> anyhow::Result<()> { + let mut writer = FullTextIndexWriter::new(FullTextIndexConfig::new())?; + writer.add_document(1, "The runners visited cafes")?; + writer.add_document(2, "plain unrelated text")?; + + let mut bytes = Vec::new(); + writer.write(&mut PosWriter::new(&mut bytes))?; + + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let result = reader.search(match_query_and("runner café", "text"), 10)?; + + assert_eq!(result.row_ids, vec![1]); + Ok(()) +} + +#[test] +fn fuzzy_match_query_matches_typos() -> anyhow::Result<()> { + let bytes = build_index()?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let query = fuzzy_match_query("paimno", "text", 1, 50, 0); + let result = reader.search(query, 10)?; + + assert_eq!(result.row_ids.len(), 2); + assert!(result.row_ids.contains(&10)); + assert!(result.row_ids.contains(&12)); + Ok(()) +} + +#[test] +fn fuzzy_match_query_supports_auto_fuzziness_json() -> anyhow::Result<()> { + let bytes = build_index()?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let query = r#"{"match":{"column":"text","query":"paimxx","fuzziness":"auto"}}"#; + let result = reader.search(query, 10)?; + + assert_eq!(result.row_ids.len(), 2); + Ok(()) +} + +#[test] +fn fuzzy_prefix_length_requires_exact_start() -> anyhow::Result<()> { + let mut writer = FullTextIndexWriter::new(FullTextIndexConfig::new())?; + writer.add_document(1, "paimon search")?; + writer.add_document(2, "xaimon search")?; + + let mut bytes = Vec::new(); + writer.write(&mut PosWriter::new(&mut bytes))?; + + let reader = FullTextIndexReader::open(SliceReader::new(bytes.clone()))?; + let query = fuzzy_match_query("paimno", "text", 1, 50, 3); + let result = reader.search(query, 10)?; + assert_eq!(result.row_ids, vec![1]); + + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let query = fuzzy_match_query("qaimon", "text", 1, 50, 1); + let result = reader.search(query, 10)?; + assert!(result.row_ids.is_empty()); + Ok(()) +} + +#[test] +fn fuzzy_match_query_honors_max_expansions() -> anyhow::Result<()> { + let mut writer = FullTextIndexWriter::new(FullTextIndexConfig::new())?; + writer.add_document(1, "aac")?; + writer.add_document(2, "abc")?; + writer.add_document(3, "acc")?; + writer.add_document(4, "bbc")?; + + let mut bytes = Vec::new(); + writer.write(&mut PosWriter::new(&mut bytes))?; + + let query = |max_expansions| fuzzy_match_query("abc", "text", 1, max_expansions, 0); + + let reader = FullTextIndexReader::open(SliceReader::new(bytes.clone()))?; + let capped = reader.search(query(1), 10)?; + assert_eq!(capped.row_ids.len(), 1); + + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let expanded = reader.search(query(50), 10)?; + assert!(expanded.row_ids.len() > capped.row_ids.len()); + assert_eq!(expanded.row_ids.len(), 4); + Ok(()) +} + +#[test] +fn multi_field_match_searches_named_fields() -> anyhow::Result<()> { + let config = FullTextIndexConfig::new().with_text_fields(["title", "body"]); + let mut writer = FullTextIndexWriter::new(config)?; + writer.add_document_fields( + 1, + [ + ("title".to_string(), "Apache Paimon".to_string()), + ("body".to_string(), "lake storage".to_string()), + ], + )?; + writer.add_document_fields( + 2, + [ + ("title".to_string(), "Tantivy".to_string()), + ("body".to_string(), "Rust search engine".to_string()), + ], + )?; + + let mut bytes = Vec::new(); + writer.write(&mut PosWriter::new(&mut bytes))?; + + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let result = reader.search(match_query("rust", "body"), 10)?; + + assert_eq!(result.row_ids, vec![2]); + Ok(()) +} + +#[test] +fn multi_match_searches_columns_and_applies_boosts() -> anyhow::Result<()> { + let config = FullTextIndexConfig::new().with_text_fields(["title", "body"]); + let mut writer = FullTextIndexWriter::new(config)?; + writer.add_document_fields( + 1, + [ + ("title".to_string(), "paimon".to_string()), + ("body".to_string(), "storage".to_string()), + ], + )?; + writer.add_document_fields( + 2, + [ + ("title".to_string(), "storage".to_string()), + ("body".to_string(), "paimon".to_string()), + ], + )?; + + let mut bytes = Vec::new(); + writer.write(&mut PosWriter::new(&mut bytes))?; + + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let query = + r#"{"multi_match":{"query":"paimon","columns":["title","body"],"boost":[3.0,1.0]}}"#; + let result = reader.search(query, 10)?; + + assert_eq!(result.row_ids, vec![1, 2]); + assert!(result.scores[0] > result.scores[1]); + Ok(()) +} + +#[test] +fn repeated_named_fields_support_string_arrays() -> anyhow::Result<()> { + let config = FullTextIndexConfig::new().with_text_fields(["tags"]); + let mut writer = FullTextIndexWriter::new(config)?; + writer.add_document_fields( + 1, + [ + ("tags".to_string(), "paimon".to_string()), + ("tags".to_string(), "lakehouse".to_string()), + ], + )?; + writer.add_document_fields(2, [("tags".to_string(), "tantivy".to_string())])?; + + let mut bytes = Vec::new(); + writer.write(&mut PosWriter::new(&mut bytes))?; + + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let result = reader.search(match_query("lakehouse", "tags"), 10)?; + + assert_eq!(result.row_ids, vec![1]); + Ok(()) +} + +#[test] +fn boolean_query_rejects_only_must_not() -> anyhow::Result<()> { + let bytes = build_index()?; + let reader = FullTextIndexReader::open(SliceReader::new(bytes))?; + let query = r#"{"boolean":{"must_not":[{"match":{"column":"text","query":"paimon"}}]}}"#; + let err = reader + .search(query, 10) + .expect_err("only must_not should fail"); + + assert!(err + .to_string() + .contains("at least one should or must clause")); + Ok(()) +}
diff --git a/core/tests/golden/storage_v1_envelope.hex b/core/tests/golden/storage_v1_envelope.hex new file mode 100644 index 0000000..0f835c5 --- /dev/null +++ b/core/tests/golden/storage_v1_envelope.hex
@@ -0,0 +1,18 @@ +504654494458303100000001000002107b226d65746164617461223a7b22636f +6e666967223a7b22726f775f69645f6669656c64223a22726f775f6964222c22 +746578745f6669656c6473223a5b2274657874225d2c22746f6b656e697a6572 +223a7b22746f6b656e697a6572223a2264656661756c74222c226e6772616d5f +6d696e5f6772616d223a332c226e6772616d5f6d61785f6772616d223a332c22 +6e6772616d5f7072656669785f6f6e6c79223a66616c73652c226a696562615f +7365617263685f6d6f6465223a747275652c226a696562615f6f7264696e616c +5f706f736974696f6e223a747275652c226c6f7765725f63617365223a747275 +652c226d61785f746f6b656e5f6c656e677468223a34302c2261736369695f66 +6f6c64696e67223a747275652c227374656d223a747275652c226c616e677561 +6765223a22656e676c697368222c2272656d6f76655f73746f705f776f726473 +223a747275652c2273746f705f776f726473223a5b5d2c22776974685f706f73 +6974696f6e223a66616c73657d7d2c22646f63756d656e745f636f756e74223a +322c2274616e746976795f76657273696f6e223a22302e32362e30227d2c2266 +696c6573223a5b7b226e616d65223a226d6574612e6a736f6e222c226f666673 +6574223a302c226c656e677468223a327d2c7b226e616d65223a227365676d65 +6e742e696478222c226f6666736574223a322c226c656e677468223a347d5d7d +616263646566
diff --git a/docs/designs/2026-07-04-paimon-full-text-index.md b/docs/designs/2026-07-04-paimon-full-text-index.md deleted file mode 100644 index ee22556..0000000 --- a/docs/designs/2026-07-04-paimon-full-text-index.md +++ /dev/null
@@ -1,467 +0,0 @@ -# Paimon Full Text Index Design - -## Problem Statement - -We want to split Paimon's Tantivy-based full-text search implementation into a -standalone repository, shaped like `apache/paimon-vector-index`, and expose a -consistent Rust, Java, and Python API. - -The upstream `paimon-tantivy` code is still pre-production for this use case, so -the standalone project does not need to read or write its existing archive -format. The new repository should use the cleanest v1 format and API. - -## Chosen Approach - -Use a Rust-first architecture: - -- `core`: owns Tantivy schema, tokenizer config, query semantics, storage - format, writer, reader, and tests. -- `ffi`: exposes a stable C ABI over `core`. -- `jni`: implements Java native calls by delegating to the Rust core. -- `java`: provides a small public Java API using Paimon-style input/output - callbacks. -- `python`: provides a ctypes binding over the C ABI. - -This matches the proven shape of `paimon-vector-index` and avoids divergent -Java and Python implementations. - -## Repository Layout - -```text -paimon-full-text/ - Cargo.toml - core/ - Cargo.toml - src/ - config.rs - document.rs - error.rs - index.rs - io.rs - query.rs - tokenizer.rs - storage.rs - lib.rs - tests/ - ffi/ - Cargo.toml - build.rs - cbindgen.toml - src/lib.rs - jni/ - Cargo.toml - src/lib.rs - include/ - paimon_ftindex.h - paimon_ftindex.hpp - java/ - pom.xml - src/main/java/org/apache/paimon/index/fulltext/ - python/ - pyproject.toml - setup.py - paimon_ftindex/ - c/ - CMakeLists.txt - test_ftindex.c - cpp/ - CMakeLists.txt - test_ftindex.cpp - docs/ - storage-format.md - java-api.md - python-api.md - paimon-integration.md -``` - -## Storage Format - -The v1 index file is self-describing. Readers must be able to open a file -without Paimon manifest metadata. - -```text -magic: 8 bytes "PFTIDX01" -version: u32 1 -header_len: u32 -header: JSON metadata and archive directory -body: concatenated Tantivy files -checksum: optional, future v2 -``` - -Header JSON: - -```json -{ - "format_version": 1, - "metadata": { - "format_version": 1, - "config": { - "row_id_field": "row_id", - "text_field": "text", - "tokenizer": { - "tokenizer": "default", - "lower_case": true, - "max_token_length": 40, - "with_position": true - } - }, - "document_count": 0, - "tantivy_version": "0.26.1" - }, - "files": [ - {"name": "meta.json", "offset": 0, "length": 1234} - ] -} -``` - -File offsets are relative to the body start. The archive directory lives in the -front header so C, Java, and Python readers only need positional `read_at` -callbacks and do not need to know total file length. - -## Rust API - -Rust is the source of truth. - -```rust -use paimon_ftindex_core::{ - FullTextIndexConfig, FullTextIndexReader, FullTextIndexWriter, - FullTextQuery, TokenizerConfig, -}; - -let config = FullTextIndexConfig::new() - .tokenizer(TokenizerConfig::default()) - .with_positions(true); - -let mut writer = FullTextIndexWriter::new(config)?; -writer.add_document(1, "Apache Paimon supports full text search")?; -writer.write(&mut output)?; - -let mut reader = FullTextIndexReader::open(input)?; -reader.optimize_for_search()?; -let result = reader.search( - FullTextQuery::match_query("paimon", "text").operator_or(), - 10, -)?; -``` - -Core concepts: - -- `FullTextIndexConfig`: schema and tokenizer config. -- `TokenizerConfig`: `default`, `simple`, `whitespace`, `raw`, `ngram`, - `jieba`, filters, stop words, stemming, positions. -- `FullTextDocument`: row id plus named text fields. -- `FullTextQuery`: structured query DSL. -- `FullTextSearchResult`: row ids plus BM25 scores. -- `SeekRead` and `SeekWrite`: callback-based I/O like vector-index. - -First release supports one logical text field named `text`. The file format -already stores field names so multi-field search can be added without changing -the public storage envelope. - -## Query DSL - -Keep the Paimon/LanceDB-style structured JSON, but implement parsing in Rust. - -Supported v1 queries: - -- `match`: tokenized term query with `OR` or `AND`. -- `match_phrase`: phrase query when positions are enabled. -- `boolean`: `should`, `must`, `must_not`. -- `boost`: positive query must match; matching the negative query multiplies - the positive score by `negative_boost` for demotion. - -Deferred: - -- `multi_match`: depends on multi-field index support. -- arbitrary Tantivy query strings: useful for debugging but too easy to make - unstable as a public API. - -Example: - -```json -{ - "match": { - "column": "text", - "terms": "apache paimon", - "operator": "And", - "boost": 1.0 - } -} -``` - -## C ABI - -The C ABI is the stable interop boundary. - -Naming: - -- library: `libpaimon_ftindex_ffi` -- header: `include/paimon_ftindex.h` -- prefix: `paimon_ftindex_` - -Main handles: - -- `PaimonFtindexConfigHandle` -- `PaimonFtindexWriterHandle` -- `PaimonFtindexReaderHandle` -- `PaimonFtindexSearchResult` - -I/O callbacks: - -```c -typedef struct { - void *ctx; - int (*write_fn)(void *ctx, const uint8_t *buf, size_t len); - int (*flush_fn)(void *ctx); - int64_t (*get_pos_fn)(void *ctx); -} PaimonFtindexOutputFile; - -typedef struct { - void *ctx; - int (*read_at_fn)(void *ctx, uint64_t pos, uint8_t *buf, size_t len); -} PaimonFtindexInputFile; -``` - -Core calls: - -```c -PaimonFtindexWriterHandle *paimon_ftindex_writer_open( - const char **keys, - const char **values, - size_t len); - -int paimon_ftindex_writer_add_document( - PaimonFtindexWriterHandle *writer, - int64_t row_id, - const char *text); - -int paimon_ftindex_writer_write_index( - PaimonFtindexWriterHandle *writer, - PaimonFtindexOutputFile output); - -PaimonFtindexReaderHandle *paimon_ftindex_reader_open( - PaimonFtindexInputFile input); - -int paimon_ftindex_reader_search_json( - PaimonFtindexReaderHandle *reader, - const char *query_json, - size_t limit, - int64_t *row_ids, - float *scores, - size_t capacity, - size_t *result_len); - -int paimon_ftindex_reader_search_json_with_roaring_filter( - PaimonFtindexReaderHandle *reader, - const char *query_json, - size_t limit, - const uint8_t *roaring_filter, - size_t roaring_filter_len, - int64_t *row_ids, - float *scores, - size_t capacity, - size_t *result_len); -``` - -All functions return `0` on success and `-1` on error. A thread-local -`paimon_ftindex_last_error()` exposes the error message. Native panics are -caught at the FFI boundary. - -## Java API - -Package: - -```text -org.apache.paimon.index.fulltext -``` - -Public classes: - -- `FullTextIndexInput` -- `FullTextIndexOutput` -- `FullTextIndexWriter` -- `FullTextIndexReader` -- `FullTextIndexMetadata` -- `FullTextQuery` -- `FullTextSearchResult` -- `FullTextNative` - -Usage: - -```java -Map<String, String> options = new HashMap<>(); -options.put("tokenizer", "ngram"); -options.put("ngram.min-gram", "2"); -options.put("ngram.max-gram", "2"); - -try (FullTextIndexWriter writer = FullTextIndexWriter.create(options)) { - writer.addDocument(1L, "Apache Paimon supports full text search"); - writer.writeIndex(output); -} - -try (FullTextIndexReader reader = new FullTextIndexReader(input)) { - FullTextSearchResult result = - reader.search(FullTextQuery.match("paimon", "text"), 10); - FullTextSearchResult filtered = - reader.search(FullTextQuery.match("paimon", "text"), 10, roaringFilterBytes); -} -``` - -Java should not expose Tantivy classes. It should use string options so Paimon -table and procedure options map directly into the standalone library. - -## Python API - -Package: - -```text -paimon_ftindex -``` - -Python uses ctypes over the C ABI and mirrors Java names in Python style. - -```python -from paimon_ftindex import FullTextIndexReader, FullTextIndexWriter, MatchQuery - -writer = FullTextIndexWriter({ - "tokenizer": "ngram", - "ngram.min-gram": "2", - "ngram.max-gram": "2", -}) -writer.add_document(1, "Apache Paimon supports full text search") -writer.write(output) - -reader = FullTextIndexReader(input) -ids, scores = reader.search(MatchQuery("paimon", column="text"), limit=10) -filtered_ids, filtered_scores = reader.search( - MatchQuery("paimon", column="text"), - limit=10, - filter_bytes=roaring_filter_bytes, -) -``` - -Python I/O protocol: - -- output object: `write(bytes)`, optional `flush()`, optional `tell()`. -- input object: `pread(pos: int, length: int) -> bytes`. - -The Python package should not depend on `tantivy-py`; Rust core provides the -search semantics. - -## Paimon Integration - -The Paimon core repository should eventually keep only a thin integration -module: - -- `paimon-fulltext` or `paimon-full-text` -- `GlobalIndexerFactory` identifier: `tantivy-fulltext` or `fulltext` -- index writer delegates to `FullTextIndexWriter` -- index reader delegates to `FullTextIndexReader` -- `FullTextQuery` and `FullTextSearch` in `paimon-common` can stay as the query - API and serialize to the JSON understood by this library - -Recommended Paimon option namespace: - -```text -fulltext.tokenizer -fulltext.ngram.min-gram -fulltext.ngram.max-gram -fulltext.ngram.prefix-only -fulltext.jieba.search-mode -fulltext.jieba.ordinal-position -fulltext.lower-case -fulltext.max-token-length -fulltext.ascii-folding -fulltext.stem -fulltext.language -fulltext.remove-stop-words -fulltext.stop-words -fulltext.with-position -``` - -The standalone library should accept both unprefixed keys and `fulltext.` -prefixed keys for caller convenience. - -## Tokenizer Policy - -Supported first implementation tokenizers: - -- `default` -- `simple` -- `whitespace` -- `raw` -- `ngram` -- `jieba` - -Reserved follow-up tokenizers and filters: - -- stemming -- built-in stop words -- custom stop words - -Supported first implementation filters: - -- lowercase -- max token length -- ASCII folding - -The `jieba` tokenizer uses Tantivy's current external tokenizer API via -`tantivy-jieba`. Search mode and ordinal token positions are enabled by -default for Chinese full-text recall and phrase query correctness. - -No dynamic Rust tokenizer plugins in v1. They complicate packaging and native -loading too much for an initial library. - -## Testing Strategy - -Core tests: - -- storage format round trip -- metadata round trip -- query JSON parsing -- tokenizer config validation -- match, phrase, boolean, boost query behavior -- seek-based input reads only requested byte ranges where possible - -Cross-language tests: - -- Rust writes, Java reads -- Java writes, Rust reads -- Rust writes, Python reads -- Python writes, Rust reads -- Java and Python return the same row ids and scores for fixed fixtures - -Fixture policy: - -- Commit small hex fixtures for v1 format. -- Add a test that fails on accidental format drift unless the version changes. - -## Release Strategy - -Publish artifacts independently: - -- Rust crates: `paimon-ftindex-core`, `paimon-ftindex-ffi`, - `paimon-ftindex-jni`. -- Java artifact: `org.apache.paimon:paimon-full-text-index`. -- Python package: `paimon-ftindex`. - -Linux native binaries should be built on the oldest supported glibc baseline, -matching the caution already present in upstream `paimon-tantivy-jni`. - -## Out Of Scope - -- Reading old upstream `paimon-tantivy` archive files. -- Multi-field indexing in the first release. -- Highlighting/snippets. -- Arbitrary query-string API as a stable surface. -- Distributed index merge and compaction policy inside this repository. -- Paimon SQL procedures and optimizer rules. Those remain Paimon repository - integration work. - -## Open Questions - -- Should the public index type be `fulltext` or keep `tantivy-fulltext` for - clarity? -- Should Java native loading copy prebuilt libraries from resources, or should - this repository initially require `PAIMON_FTINDEX_LIB_PATH` like the Python - package?
diff --git a/docs/java-api.md b/docs/java-api.md index de859ef..ecce6dd 100644 --- a/docs/java-api.md +++ b/docs/java-api.md
@@ -10,7 +10,6 @@ - `FullTextIndexWriter` - `FullTextIndexReader` -- `FullTextQuery` - `FullTextSearchResult` - `FullTextIndexInput` - `FullTextIndexOutput` @@ -18,27 +17,49 @@ Example: ```java -try (FullTextIndexWriter writer = FullTextIndexWriter.create(Collections.emptyMap())) { - writer.addDocument(1L, "Apache Paimon full text search"); +Map<String, String> options = Collections.singletonMap("text-fields", "title,body"); +try (FullTextIndexWriter writer = FullTextIndexWriter.create(options)) { + Map<String, String> fields = new LinkedHashMap<>(); + fields.put("title", "Apache Paimon"); + fields.put("body", "lake storage"); + writer.addDocument(1L, fields); writer.writeIndex(output); } try (FullTextIndexReader reader = new FullTextIndexReader(input)) { - FullTextSearchResult result = reader.search(FullTextQuery.match("paimon", "text"), 10); + reader.prewarm(); + FullTextSearchResult result = + reader.search("{\"match\":{\"query\":\"paimon\"}}", 10); FullTextSearchResult filtered = - reader.search(FullTextQuery.match("paimon", "text"), 10, roaringFilterBytes); + reader.search("{\"match\":{\"query\":\"paimno\",\"column\":\"title\",\"fuzziness\":1}}", + 10, + roaringFilterBytes); + FullTextReadMetrics metrics = reader.readMetrics(); } ``` +`search()` accepts the query DSL as a JSON string. `match` supports `column`, +`operator`, `boost`, `fuzziness`, `max_expansions`, and `prefix_length`. If +`column` is omitted, a multi-field index searches all indexed text fields. Use +`"fuzziness":"auto"` for auto fuzziness. Boolean and boost-demotion queries use +the same JSON DSL. + `roaringFilterBytes` must be a serialized 64-bit Roaring bitmap (`RoaringTreemap`) containing the allowed row ids. The filter is applied during Tantivy collection, before the top results are selected. +`prewarm()` eagerly initializes the underlying search reader and archive cache +before a query burst. `readMetrics()` returns a snapshot with `preadCalls`, +`preadRanges`, `preadBytes`, `cacheHits`, `cacheMisses`, `cacheEvictions`, and +`cachedBlocks`. + Input reads: -- Implement `FullTextIndexInput.readAt(...)` for compatibility. -- Override `FullTextIndexInput.pread(long[] positions, byte[][] buffers)` when - the storage client can batch, coalesce, or parallelize positional reads. +- Implement `FullTextIndexInput.pread(long position, byte[] buffer, int offset, + int length)` as a single positional read. Rust owns any batching or + parallelism above this callback. +- The implementation must be safe for concurrent calls. Synchronize inside + `pread` if the backing input keeps mutable state. Native loading:
diff --git a/docs/paimon-integration.md b/docs/paimon-integration.md index 4905292..a0aaaa2 100644 --- a/docs/paimon-integration.md +++ b/docs/paimon-integration.md
@@ -3,8 +3,7 @@ The standalone library is intentionally independent of Paimon core. Paimon integration should be a thin adapter: -- Keep `FullTextQuery` / `FullTextSearch` in Paimon common as query API. -- Serialize queries to the JSON accepted by this library. +- Pass the query DSL as a JSON string to the standalone reader. - Implement a Paimon `GlobalIndexerFactory` that delegates to Java `FullTextIndexWriter` and `FullTextIndexReader`. - Pass serialized 64-bit Roaring row-id filters to reader search when another @@ -21,6 +20,7 @@ ```text fulltext.tokenizer +fulltext.text-fields fulltext.ngram.min-gram fulltext.ngram.max-gram fulltext.ngram.prefix-only @@ -29,8 +29,19 @@ fulltext.lower-case fulltext.max-token-length fulltext.ascii-folding +fulltext.stem +fulltext.language +fulltext.remove-stop-words +fulltext.stop-words fulltext.with-position ``` The standalone library accepts both unprefixed keys and `fulltext.` prefixed keys. + +Query JSON supports `match`, `multi_match`, `match_phrase`, `boolean`, and +boost-demotion queries. `match` accepts `fuzziness`, `max_expansions`, and +`prefix_length`; use `fuzziness: "auto"` for automatic edit distance. If +`match.column` is omitted, the native reader searches every text field +configured in the index, so Paimon can derive native text fields from its +extra-fields mechanism without adding a user-facing `multi_match` requirement.
diff --git a/docs/plans/2026-07-04-paimon-full-text-index-implementation.md b/docs/plans/2026-07-04-paimon-full-text-index-implementation.md deleted file mode 100644 index 3ba184f..0000000 --- a/docs/plans/2026-07-04-paimon-full-text-index-implementation.md +++ /dev/null
@@ -1,65 +0,0 @@ -# Implementation Plan: Paimon Full Text Index - -## Prerequisites - -- [x] Confirm local Rust, Python, and Maven toolchains are available. -- [x] Confirm this repository is the standalone `paimon-full-text` workspace. - -## Tasks - -### Task 1: Create Rust workspace skeleton - -- **Files**: `Cargo.toml`, `core/Cargo.toml`, `ffi/Cargo.toml`, `jni/Cargo.toml` -- **Changes**: Define a top-level Cargo workspace with `core`, `ffi`, and `jni` - crates. Add Tantivy and serialization dependencies to `core`. -- **Verify**: `cargo metadata --no-deps` succeeds. -- **Dependencies**: None. - -### Task 2: Implement Rust core config, query, I/O, and storage - -- **Files**: `core/src/*.rs` -- **Changes**: Add tokenizer config parsing, structured query JSON parsing, - seek/read-write traits, v1 storage envelope, Tantivy writer, and reader. -- **Verify**: `cargo test -p paimon-ftindex-core`. -- **Dependencies**: Task 1. - -### Task 3: Add C FFI over Rust core - -- **Files**: `ffi/src/lib.rs`, `include/paimon_ftindex.h` -- **Changes**: Expose writer/reader handles, I/O callbacks, result buffers, - thread-local errors, and panic boundaries. -- **Verify**: `cargo test -p paimon-ftindex-ffi`. -- **Dependencies**: Task 2. - -### Task 4: Add Java API and JNI bridge - -- **Files**: `java/**`, `jni/src/lib.rs` -- **Changes**: Add Java wrapper classes and JNI methods backed by Rust core. -- **Verify**: `mvn test -pl java` where possible, and `cargo test -p - paimon-ftindex-jni`. -- **Dependencies**: Task 2. - -### Task 5: Add Python ctypes package - -- **Files**: `python/**` -- **Changes**: Add ctypes binding, Python query helpers, input/output adapter, - and unit tests using the local native library. -- **Verify**: `python3 -m pytest python/tests` after building FFI library. -- **Dependencies**: Task 3. - -### Task 6: Add repository docs and examples - -- **Files**: `README.md`, `docs/storage-format.md`, `docs/java-api.md`, - `docs/python-api.md`, `docs/paimon-integration.md` -- **Changes**: Document architecture, build commands, API examples, and Paimon - integration. -- **Verify**: `cargo test --workspace` and basic doc review. -- **Dependencies**: Tasks 2-5. - -## Post-Implementation - -- [ ] Run `cargo fmt --all`. -- [ ] Run `cargo test --workspace`. -- [ ] Run targeted Python tests if native library builds. -- [ ] Run targeted Java compilation/tests if JNI library builds. -- [ ] Review `git diff`.
diff --git a/docs/python-api.md b/docs/python-api.md index a5e0b65..bff973a 100644 --- a/docs/python-api.md +++ b/docs/python-api.md
@@ -9,23 +9,50 @@ Example: ```python -from paimon_ftindex import FullTextIndexReader, FullTextIndexWriter, MatchQuery +from paimon_ftindex import ( + FullTextIndexReader, + FullTextIndexWriter, +) -with FullTextIndexWriter({"tokenizer": "ngram"}) as writer: - writer.add_document(1, "Apache Paimon full text search") +with FullTextIndexWriter({"text-fields": "title,body"}) as writer: + writer.add_document_fields( + 1, + { + "title": "Apache Paimon", + "body": "lake storage", + }, + ) writer.write(output) with FullTextIndexReader(input_) as reader: - ids, scores = reader.search(MatchQuery("paimon"), limit=10) - filtered_ids, filtered_scores = reader.search( - MatchQuery("paimon"), limit=10, filter_bytes=roaring_filter_bytes + reader.prewarm() + ids, scores = reader.search( + '{"match":{"query":"paimon"}}', + limit=10, ) + filtered_ids, filtered_scores = reader.search( + '{"match":{"query":"paimno","column":"title","fuzziness":1}}', + limit=10, + filter_bytes=roaring_filter_bytes, + ) + metrics = reader.read_metrics() ``` +`search()` accepts the query DSL as a JSON string. `match` supports `column`, +`operator`, `boost`, `fuzziness`, `max_expansions`, and `prefix_length`. If +`column` is omitted, a multi-field index searches all indexed text fields. Use +`"fuzziness":"auto"` for auto fuzziness. `match_phrase` requires the index to +be created with `with-position=true`. + `filter_bytes` must be a serialized 64-bit Roaring bitmap (`RoaringTreemap`) containing the allowed row ids. The filter is applied during Tantivy collection, before the top results are selected. +`prewarm()` eagerly initializes the underlying search reader and archive cache +before a query burst. `read_metrics()` returns a snapshot with `pread_calls`, +`pread_ranges`, `pread_bytes`, `cache_hits`, `cache_misses`, `cache_evictions`, +and `cached_blocks`. + The output object must provide: - `write(bytes)` @@ -35,6 +62,9 @@ - `pread(pos: int, length: int) -> bytes` +`pread` must be safe for concurrent calls if the backing input keeps mutable +state. Rust owns batching and parallelism above this single-read callback. + Native loading: - Set `PAIMON_FTINDEX_LIB_PATH` to a library file or directory, or
diff --git a/docs/storage-format.md b/docs/storage-format.md index 0bc798d..8fc9130 100644 --- a/docs/storage-format.md +++ b/docs/storage-format.md
@@ -12,13 +12,16 @@ The header contains: -- `metadata.format_version` - `metadata.config` - `metadata.document_count` - `metadata.tantivy_version` - `files[]`: Tantivy file name, body-relative offset, and length -Readers first read the 16-byte fixed prefix, then the JSON header, then the -listed Tantivy files by positional reads. The current reader loads listed files -into Tantivy `RamDirectory`; a future reader can replace this with a custom -seek-on-demand `Directory` without changing the envelope. +Readers first read the 16-byte fixed prefix and JSON header. The Rust reader +then exposes listed Tantivy files through a read-only seek-on-demand directory, +so segment file bytes are fetched by positional reads only when Tantivy asks for +the corresponding byte range. Readers reject headers larger than 16 MiB before +allocating the header buffer. + +The body stores Tantivy segment files directly. Readers reject index files whose +recorded `metadata.tantivy_version` does not match the linked Tantivy runtime.
diff --git a/ffi/src/lib.rs b/ffi/src/lib.rs index 3a05525..0818b52 100644 --- a/ffi/src/lib.rs +++ b/ffi/src/lib.rs
@@ -2,7 +2,7 @@ use paimon_ftindex_core::io::{ReadRequest, SeekRead, SeekWrite}; use paimon_ftindex_core::{ - FullTextIndexConfig, FullTextIndexReader, FullTextIndexWriter, FullTextQuery, TokenizerConfig, + FullTextIndexConfig, FullTextIndexReader, FullTextIndexWriter, FullTextReadMetrics, }; use std::cell::RefCell; use std::collections::HashMap; @@ -116,7 +116,33 @@ #[repr(C)] pub struct PaimonFtindexInputFile { pub ctx: *mut c_void, - pub read_at_fn: Option<unsafe extern "C" fn(*mut c_void, u64, *mut u8, usize) -> c_int>, + pub pread_fn: Option<unsafe extern "C" fn(*mut c_void, u64, *mut u8, usize) -> c_int>, +} + +#[repr(C)] +#[derive(Clone, Copy, Debug, Default)] +pub struct PaimonFtindexReadMetrics { + pub pread_calls: u64, + pub pread_ranges: u64, + pub pread_bytes: u64, + pub cache_hits: u64, + pub cache_misses: u64, + pub cache_evictions: u64, + pub cached_blocks: u64, +} + +impl From<FullTextReadMetrics> for PaimonFtindexReadMetrics { + fn from(metrics: FullTextReadMetrics) -> Self { + Self { + pread_calls: metrics.pread_calls, + pread_ranges: metrics.pread_ranges, + pread_bytes: metrics.pread_bytes, + cache_hits: metrics.cache_hits, + cache_misses: metrics.cache_misses, + cache_evictions: metrics.cache_evictions, + cached_blocks: metrics.cached_blocks, + } + } } struct FfiInputFile { @@ -124,16 +150,17 @@ } unsafe impl Send for FfiInputFile {} +unsafe impl Sync for FfiInputFile {} impl SeekRead for FfiInputFile { - fn pread(&mut self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> { - let read_at_fn = self + fn pread(&self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> { + let pread_fn = self .raw - .read_at_fn - .ok_or_else(|| io::Error::other("read_at_fn is null"))?; + .pread_fn + .ok_or_else(|| io::Error::other("pread_fn is null"))?; for range in ranges { let status = unsafe { - read_at_fn( + pread_fn( self.raw.ctx, range.pos, range.buf.as_mut_ptr(), @@ -142,7 +169,7 @@ }; if status != 0 { return Err(io::Error::other(format!( - "read_at callback failed at offset {} length {}", + "pread callback failed at offset {} length {}", range.pos, range.buf.len() ))); @@ -160,8 +187,8 @@ inner: FullTextIndexReader<FfiInputFile>, } -struct SearchJsonRequest<'a> { - query_json: *const c_char, +struct SearchRequest<'a> { + query: *const c_char, limit: usize, roaring_filter: Option<&'a [u8]>, row_ids: *mut i64, @@ -178,8 +205,7 @@ ) -> *mut PaimonFtindexWriterHandle { ffi_ptr(|| { let options = options_from_raw(keys, values, len)?; - let tokenizer = TokenizerConfig::from_options(&options).map_err(|e| e.to_string())?; - let config = FullTextIndexConfig::new().tokenizer(tokenizer); + let config = FullTextIndexConfig::from_options(&options).map_err(|e| e.to_string())?; let writer = FullTextIndexWriter::new(config).map_err(|e| e.to_string())?; Ok(Box::into_raw(Box::new(PaimonFtindexWriterHandle { inner: writer, @@ -204,6 +230,24 @@ } #[no_mangle] +pub unsafe extern "C" fn paimon_ftindex_writer_add_document_fields( + writer: *mut PaimonFtindexWriterHandle, + row_id: i64, + field_names: *const *const c_char, + texts: *const *const c_char, + len: usize, +) -> c_int { + ffi_status(|| { + let writer = require_mut(writer, "writer")?; + let fields = fields_from_raw(field_names, texts, len)?; + writer + .inner + .add_document_fields(row_id, fields) + .map_err(|e| e.to_string()) + }) +} + +#[no_mangle] pub unsafe extern "C" fn paimon_ftindex_writer_write_index( writer: *mut PaimonFtindexWriterHandle, output: PaimonFtindexOutputFile, @@ -236,9 +280,9 @@ } #[no_mangle] -pub unsafe extern "C" fn paimon_ftindex_reader_search_json( +pub unsafe extern "C" fn paimon_ftindex_reader_search( reader: *mut PaimonFtindexReaderHandle, - query_json: *const c_char, + query: *const c_char, limit: usize, row_ids: *mut i64, scores: *mut f32, @@ -246,11 +290,11 @@ result_len: *mut usize, ) -> c_int { ffi_status(|| { - let reader = require_mut(reader, "reader")?; - search_json_impl( + let reader = require_ref(reader, "reader")?; + search_impl( reader, - SearchJsonRequest { - query_json, + SearchRequest { + query, limit, roaring_filter: None, row_ids, @@ -263,9 +307,9 @@ } #[no_mangle] -pub unsafe extern "C" fn paimon_ftindex_reader_search_json_with_roaring_filter( +pub unsafe extern "C" fn paimon_ftindex_reader_search_with_roaring_filter( reader: *mut PaimonFtindexReaderHandle, - query_json: *const c_char, + query: *const c_char, limit: usize, roaring_filter: *const u8, roaring_filter_len: usize, @@ -275,12 +319,12 @@ result_len: *mut usize, ) -> c_int { ffi_status(|| { - let reader = require_mut(reader, "reader")?; + let reader = require_ref(reader, "reader")?; let roaring_filter = const_slice(roaring_filter, roaring_filter_len, "roaring_filter")?; - search_json_impl( + search_impl( reader, - SearchJsonRequest { - query_json, + SearchRequest { + query, limit, roaring_filter: Some(roaring_filter), row_ids, @@ -293,6 +337,29 @@ } #[no_mangle] +pub unsafe extern "C" fn paimon_ftindex_reader_prewarm( + reader: *mut PaimonFtindexReaderHandle, +) -> c_int { + ffi_status(|| { + let reader = require_ref(reader, "reader")?; + reader.inner.prewarm().map_err(|e| e.to_string()) + }) +} + +#[no_mangle] +pub unsafe extern "C" fn paimon_ftindex_reader_read_metrics( + reader: *mut PaimonFtindexReaderHandle, + metrics: *mut PaimonFtindexReadMetrics, +) -> c_int { + ffi_status(|| { + let reader = require_ref(reader, "reader")?; + let metrics = require_mut(metrics, "metrics")?; + *metrics = reader.inner.read_metrics().into(); + Ok(()) + }) +} + +#[no_mangle] pub unsafe extern "C" fn paimon_ftindex_reader_free(reader: *mut PaimonFtindexReaderHandle) { if !reader.is_null() { drop(Box::from_raw(reader)); @@ -324,10 +391,39 @@ Ok(options) } +unsafe fn fields_from_raw( + field_names: *const *const c_char, + texts: *const *const c_char, + len: usize, +) -> Result<Vec<(String, String)>, String> { + if len == 0 { + return Err("document fields must not be empty".to_string()); + } + if field_names.is_null() { + return Err("field_names is null".to_string()); + } + if texts.is_null() { + return Err("texts is null".to_string()); + } + let field_names = slice::from_raw_parts(field_names, len); + let texts = slice::from_raw_parts(texts, len); + let mut fields = Vec::with_capacity(len); + for i in 0..len { + let field_name = cstr_to_string(field_names[i], "field name")?; + let text = cstr_to_string(texts[i], "field text")?; + fields.push((field_name, text)); + } + Ok(fields) +} + unsafe fn require_mut<'a, T>(ptr: *mut T, name: &str) -> Result<&'a mut T, String> { ptr.as_mut().ok_or_else(|| format!("{name} is null")) } +unsafe fn require_ref<'a, T>(ptr: *const T, name: &str) -> Result<&'a T, String> { + ptr.as_ref().ok_or_else(|| format!("{name} is null")) +} + unsafe fn const_slice<'a, T>(ptr: *const T, len: usize, name: &str) -> Result<&'a [T], String> { if len == 0 { Ok(&[]) @@ -348,24 +444,23 @@ .map_err(|e| format!("{name} is not valid UTF-8: {e}")) } -fn search_json_impl( - reader: &mut PaimonFtindexReaderHandle, - request: SearchJsonRequest<'_>, +fn search_impl( + reader: &PaimonFtindexReaderHandle, + request: SearchRequest<'_>, ) -> Result<(), String> { if request.result_len.is_null() { return Err("result_len is null".to_string()); } - let query_json = unsafe { cstr_to_string(request.query_json, "query_json") }?; - let query = FullTextQuery::from_json(&query_json).map_err(|e| e.to_string())?; + let query = unsafe { cstr_to_string(request.query, "query") }?; let result = if let Some(roaring_filter) = request.roaring_filter { reader .inner - .search_with_roaring_filter(query, request.limit, roaring_filter) + .search_with_roaring_filter(&query, request.limit, roaring_filter) .map_err(|e| e.to_string())? } else { reader .inner - .search(query, request.limit) + .search(&query, request.limit) .map_err(|e| e.to_string())? }; unsafe { @@ -421,7 +516,7 @@ } #[test] - fn ffi_round_trip_search_json() { + fn ffi_round_trip_search() { unsafe { let writer = paimon_ftindex_writer_open(ptr::null(), ptr::null(), 0); assert!(!writer.is_null()); @@ -443,22 +538,29 @@ let input = PaimonFtindexInputFile { ctx: &bytes as *const Vec<u8> as *mut c_void, - read_at_fn: Some(read_vec), + pread_fn: Some(read_vec), }; let reader = paimon_ftindex_reader_open(input); assert!(!reader.is_null()); - let query = CString::new( - paimon_ftindex_core::FullTextQuery::match_query("paimon", "text") - .to_json() - .unwrap(), - ) - .unwrap(); + let mut metrics = PaimonFtindexReadMetrics::default(); + assert_eq!(paimon_ftindex_reader_read_metrics(reader, &mut metrics), 0); + assert!(metrics.pread_calls >= 2); + assert!(metrics.pread_bytes > 16); + assert_eq!(paimon_ftindex_reader_prewarm(reader), 0); + let mut after_prewarm = PaimonFtindexReadMetrics::default(); + assert_eq!( + paimon_ftindex_reader_read_metrics(reader, &mut after_prewarm), + 0 + ); + assert!(after_prewarm.pread_calls > metrics.pread_calls); + + let query = CString::new(r#"{"match":{"query":"paimon","column":"text"}}"#).unwrap(); let mut row_ids = [0i64; 4]; let mut scores = [0f32; 4]; let mut result_len = 0usize; assert_eq!( - paimon_ftindex_reader_search_json( + paimon_ftindex_reader_search( reader, query.as_ptr(), 4, @@ -472,12 +574,19 @@ assert_eq!(result_len, 1); assert_eq!(row_ids[0], 7); assert!(scores[0] > 0.0); + let mut after_search = PaimonFtindexReadMetrics::default(); + assert_eq!( + paimon_ftindex_reader_read_metrics(reader, &mut after_search), + 0 + ); + assert!(after_search.pread_calls >= after_prewarm.pread_calls); + assert!(after_search.cache_misses >= metrics.cache_misses); paimon_ftindex_reader_free(reader); } } #[test] - fn ffi_round_trip_search_json_with_roaring_filter() { + fn ffi_round_trip_search_with_roaring_filter() { unsafe { let writer = paimon_ftindex_writer_open(ptr::null(), ptr::null(), 0); assert!(!writer.is_null()); @@ -504,17 +613,12 @@ let input = PaimonFtindexInputFile { ctx: &bytes as *const Vec<u8> as *mut c_void, - read_at_fn: Some(read_vec), + pread_fn: Some(read_vec), }; let reader = paimon_ftindex_reader_open(input); assert!(!reader.is_null()); - let query = CString::new( - paimon_ftindex_core::FullTextQuery::match_query("paimon", "text") - .to_json() - .unwrap(), - ) - .unwrap(); + let query = CString::new(r#"{"match":{"query":"paimon","column":"text"}}"#).unwrap(); let mut allowed = RoaringTreemap::new(); allowed.insert(9); let mut filter_bytes = Vec::new(); @@ -523,7 +627,7 @@ let mut scores = [0f32; 4]; let mut result_len = 0usize; assert_eq!( - paimon_ftindex_reader_search_json_with_roaring_filter( + paimon_ftindex_reader_search_with_roaring_filter( reader, query.as_ptr(), 4, @@ -542,4 +646,71 @@ paimon_ftindex_reader_free(reader); } } + + #[test] + fn ffi_add_document_fields_searches_multi_field_index() { + unsafe { + let key = CString::new("text-fields").unwrap(); + let value = CString::new("title,body").unwrap(); + let keys = [key.as_ptr()]; + let values = [value.as_ptr()]; + let writer = paimon_ftindex_writer_open(keys.as_ptr(), values.as_ptr(), keys.len()); + assert!(!writer.is_null()); + + let title = CString::new("title").unwrap(); + let body = CString::new("body").unwrap(); + let title_text = CString::new("Apache Paimon").unwrap(); + let body_text = CString::new("lake storage").unwrap(); + let field_names = [title.as_ptr(), body.as_ptr()]; + let texts = [title_text.as_ptr(), body_text.as_ptr()]; + assert_eq!( + paimon_ftindex_writer_add_document_fields( + writer, + 17, + field_names.as_ptr(), + texts.as_ptr(), + field_names.len(), + ), + 0 + ); + + let mut bytes = Vec::new(); + let output = PaimonFtindexOutputFile { + ctx: &mut bytes as *mut Vec<u8> as *mut c_void, + write_fn: Some(write_vec), + flush_fn: None, + }; + assert_eq!(paimon_ftindex_writer_write_index(writer, output), 0); + paimon_ftindex_writer_free(writer); + + let input = PaimonFtindexInputFile { + ctx: &bytes as *const Vec<u8> as *mut c_void, + pread_fn: Some(read_vec), + }; + let reader = paimon_ftindex_reader_open(input); + assert!(!reader.is_null()); + + let query = + CString::new(r#"{"multi_match":{"query":"paimon","columns":["title","body"]}}"#) + .unwrap(); + let mut row_ids = [0i64; 4]; + let mut scores = [0f32; 4]; + let mut result_len = 0usize; + assert_eq!( + paimon_ftindex_reader_search( + reader, + query.as_ptr(), + 4, + row_ids.as_mut_ptr(), + scores.as_mut_ptr(), + row_ids.len(), + &mut result_len, + ), + 0 + ); + assert_eq!(result_len, 1); + assert_eq!(row_ids[0], 17); + paimon_ftindex_reader_free(reader); + } + } }
diff --git a/include/paimon_ftindex.h b/include/paimon_ftindex.h index b4d2cea..1fd4cdd 100644 --- a/include/paimon_ftindex.h +++ b/include/paimon_ftindex.h
@@ -19,9 +19,20 @@ typedef struct { void *ctx; - int (*read_at_fn)(void *ctx, uint64_t pos, uint8_t *buf, size_t len); + /* Must be safe for concurrent calls. Rust owns batching and parallelism. */ + int (*pread_fn)(void *ctx, uint64_t pos, uint8_t *buf, size_t len); } PaimonFtindexInputFile; +typedef struct { + uint64_t pread_calls; + uint64_t pread_ranges; + uint64_t pread_bytes; + uint64_t cache_hits; + uint64_t cache_misses; + uint64_t cache_evictions; + uint64_t cached_blocks; +} PaimonFtindexReadMetrics; + const char *paimon_ftindex_last_error(void); PaimonFtindexWriterHandle *paimon_ftindex_writer_open( @@ -34,6 +45,13 @@ int64_t row_id, const char *text); +int paimon_ftindex_writer_add_document_fields( + PaimonFtindexWriterHandle *writer, + int64_t row_id, + const char **field_names, + const char **texts, + size_t len); + int paimon_ftindex_writer_write_index( PaimonFtindexWriterHandle *writer, PaimonFtindexOutputFile output); @@ -42,18 +60,18 @@ PaimonFtindexReaderHandle *paimon_ftindex_reader_open(PaimonFtindexInputFile input); -int paimon_ftindex_reader_search_json( +int paimon_ftindex_reader_search( PaimonFtindexReaderHandle *reader, - const char *query_json, + const char *query, size_t limit, int64_t *row_ids, float *scores, size_t capacity, size_t *result_len); -int paimon_ftindex_reader_search_json_with_roaring_filter( +int paimon_ftindex_reader_search_with_roaring_filter( PaimonFtindexReaderHandle *reader, - const char *query_json, + const char *query, size_t limit, const uint8_t *roaring_filter, size_t roaring_filter_len, @@ -62,6 +80,12 @@ size_t capacity, size_t *result_len); +int paimon_ftindex_reader_prewarm(PaimonFtindexReaderHandle *reader); + +int paimon_ftindex_reader_read_metrics( + PaimonFtindexReaderHandle *reader, + PaimonFtindexReadMetrics *metrics); + void paimon_ftindex_reader_free(PaimonFtindexReaderHandle *reader); #ifdef __cplusplus
diff --git a/java/src/main/java/org/apache/paimon/index/fulltext/FullTextIndexInput.java b/java/src/main/java/org/apache/paimon/index/fulltext/FullTextIndexInput.java index 5e37369..990870d 100644 --- a/java/src/main/java/org/apache/paimon/index/fulltext/FullTextIndexInput.java +++ b/java/src/main/java/org/apache/paimon/index/fulltext/FullTextIndexInput.java
@@ -4,26 +4,11 @@ public interface FullTextIndexInput { - void readAt(long position, byte[] buffer, int offset, int length) throws IOException; - - default void pread(long[] positions, byte[][] buffers) throws IOException { - if (positions == null) { - throw new NullPointerException("positions"); - } - if (buffers == null) { - throw new NullPointerException("buffers"); - } - if (positions.length != buffers.length) { - throw new IllegalArgumentException( - "positions length " + positions.length + " does not match buffers length " - + buffers.length); - } - for (int i = 0; i < positions.length; i++) { - byte[] buffer = buffers[i]; - if (buffer == null) { - throw new NullPointerException("buffers[" + i + "]"); - } - readAt(positions[i], buffer, 0, buffer.length); - } - } + /** + * Reads exactly {@code length} bytes at {@code position} into {@code buffer}. + * + * <p>Implementations must be safe for concurrent calls. The native reader owns batching and + * parallelism above this single-read callback. + */ + void pread(long position, byte[] buffer, int offset, int length) throws IOException; }
diff --git a/java/src/main/java/org/apache/paimon/index/fulltext/FullTextIndexReader.java b/java/src/main/java/org/apache/paimon/index/fulltext/FullTextIndexReader.java index 1520532..8137c18 100644 --- a/java/src/main/java/org/apache/paimon/index/fulltext/FullTextIndexReader.java +++ b/java/src/main/java/org/apache/paimon/index/fulltext/FullTextIndexReader.java
@@ -11,17 +11,17 @@ this.nativePtr = FullTextNative.openReader(input); } - public FullTextSearchResult search(FullTextQuery query, int limit) { + public FullTextSearchResult search(String query, int limit) { if (query == null) { throw new NullPointerException("query"); } if (limit <= 0) { throw new IllegalArgumentException("search limit must be positive"); } - return FullTextNative.searchJson(requireOpen(), query.toJson(), limit); + return FullTextNative.search(requireOpen(), query, limit); } - public FullTextSearchResult search(FullTextQuery query, int limit, byte[] roaringFilter) { + public FullTextSearchResult search(String query, int limit, byte[] roaringFilter) { if (query == null) { throw new NullPointerException("query"); } @@ -31,8 +31,15 @@ if (roaringFilter == null) { throw new NullPointerException("roaringFilter"); } - return FullTextNative.searchJsonWithRoaringFilter( - requireOpen(), query.toJson(), limit, roaringFilter); + return FullTextNative.searchWithRoaringFilter(requireOpen(), query, limit, roaringFilter); + } + + public FullTextReadMetrics readMetrics() { + return FullTextNative.readMetrics(requireOpen()); + } + + public void prewarm() { + FullTextNative.prewarm(requireOpen()); } @Override
diff --git a/java/src/main/java/org/apache/paimon/index/fulltext/FullTextIndexWriter.java b/java/src/main/java/org/apache/paimon/index/fulltext/FullTextIndexWriter.java index d0fd13d..22804d7 100644 --- a/java/src/main/java/org/apache/paimon/index/fulltext/FullTextIndexWriter.java +++ b/java/src/main/java/org/apache/paimon/index/fulltext/FullTextIndexWriter.java
@@ -28,6 +28,21 @@ FullTextNative.addDocument(requireOpen(), rowId, text); } + public void addDocument(long rowId, Map<String, String> fields) { + if (fields == null) { + throw new NullPointerException("fields"); + } + String[] fieldNames = new String[fields.size()]; + String[] texts = new String[fields.size()]; + int i = 0; + for (Map.Entry<String, String> entry : fields.entrySet()) { + fieldNames[i] = entry.getKey(); + texts[i] = entry.getValue(); + i++; + } + FullTextNative.addDocumentFields(requireOpen(), rowId, fieldNames, texts); + } + public void writeIndex(FullTextIndexOutput output) { if (output == null) { throw new NullPointerException("output");
diff --git a/java/src/main/java/org/apache/paimon/index/fulltext/FullTextNative.java b/java/src/main/java/org/apache/paimon/index/fulltext/FullTextNative.java index 5d2a07e..2016dcc 100644 --- a/java/src/main/java/org/apache/paimon/index/fulltext/FullTextNative.java +++ b/java/src/main/java/org/apache/paimon/index/fulltext/FullTextNative.java
@@ -17,16 +17,23 @@ static native void addDocument(long writerPtr, long rowId, String text); + static native void addDocumentFields( + long writerPtr, long rowId, String[] fieldNames, String[] texts); + static native void writeIndex(long writerPtr, FullTextIndexOutput output); static native void freeWriter(long writerPtr); static native long openReader(FullTextIndexInput input); - static native FullTextSearchResult searchJson(long readerPtr, String queryJson, int limit); + static native FullTextSearchResult search(long readerPtr, String query, int limit); - static native FullTextSearchResult searchJsonWithRoaringFilter( - long readerPtr, String queryJson, int limit, byte[] roaringFilter); + static native FullTextSearchResult searchWithRoaringFilter( + long readerPtr, String query, int limit, byte[] roaringFilter); + + static native void prewarm(long readerPtr); + + static native FullTextReadMetrics readMetrics(long readerPtr); static native void freeReader(long readerPtr); }
diff --git a/java/src/main/java/org/apache/paimon/index/fulltext/FullTextQuery.java b/java/src/main/java/org/apache/paimon/index/fulltext/FullTextQuery.java deleted file mode 100644 index 30e5108..0000000 --- a/java/src/main/java/org/apache/paimon/index/fulltext/FullTextQuery.java +++ /dev/null
@@ -1,75 +0,0 @@ -package org.apache.paimon.index.fulltext; - -public final class FullTextQuery { - - private final String json; - - private FullTextQuery(String json) { - this.json = json; - } - - public static FullTextQuery match(String terms, String column) { - return match(terms, column, "Or"); - } - - public static FullTextQuery match(String terms, String column, String operator) { - return new FullTextQuery( - "{\"match\":{\"column\":\"" - + escape(column) - + "\",\"terms\":\"" - + escape(terms) - + "\",\"operator\":\"" - + escape(operator) - + "\",\"boost\":1.0}}"); - } - - public static FullTextQuery phrase(String terms, String column) { - return new FullTextQuery( - "{\"match_phrase\":{\"column\":\"" - + escape(column) - + "\",\"terms\":\"" - + escape(terms) - + "\",\"slop\":0}}"); - } - - public static FullTextQuery json(String json) { - if (json == null) { - throw new NullPointerException("json"); - } - return new FullTextQuery(json); - } - - public String toJson() { - return json; - } - - private static String escape(String value) { - if (value == null) { - throw new NullPointerException("value"); - } - StringBuilder builder = new StringBuilder(value.length() + 8); - for (int i = 0; i < value.length(); i++) { - char c = value.charAt(i); - switch (c) { - case '\\': - builder.append("\\\\"); - break; - case '"': - builder.append("\\\""); - break; - case '\n': - builder.append("\\n"); - break; - case '\r': - builder.append("\\r"); - break; - case '\t': - builder.append("\\t"); - break; - default: - builder.append(c); - } - } - return builder.toString(); - } -}
diff --git a/java/src/main/java/org/apache/paimon/index/fulltext/FullTextReadMetrics.java b/java/src/main/java/org/apache/paimon/index/fulltext/FullTextReadMetrics.java new file mode 100644 index 0000000..79f909f --- /dev/null +++ b/java/src/main/java/org/apache/paimon/index/fulltext/FullTextReadMetrics.java
@@ -0,0 +1,57 @@ +package org.apache.paimon.index.fulltext; + +public final class FullTextReadMetrics { + + private final long preadCalls; + private final long preadRanges; + private final long preadBytes; + private final long cacheHits; + private final long cacheMisses; + private final long cacheEvictions; + private final long cachedBlocks; + + public FullTextReadMetrics( + long preadCalls, + long preadRanges, + long preadBytes, + long cacheHits, + long cacheMisses, + long cacheEvictions, + long cachedBlocks) { + this.preadCalls = preadCalls; + this.preadRanges = preadRanges; + this.preadBytes = preadBytes; + this.cacheHits = cacheHits; + this.cacheMisses = cacheMisses; + this.cacheEvictions = cacheEvictions; + this.cachedBlocks = cachedBlocks; + } + + public long preadCalls() { + return preadCalls; + } + + public long preadRanges() { + return preadRanges; + } + + public long preadBytes() { + return preadBytes; + } + + public long cacheHits() { + return cacheHits; + } + + public long cacheMisses() { + return cacheMisses; + } + + public long cacheEvictions() { + return cacheEvictions; + } + + public long cachedBlocks() { + return cachedBlocks; + } +}
diff --git a/java/src/test/java/org/apache/paimon/index/fulltext/FullTextNativeRoundTripTest.java b/java/src/test/java/org/apache/paimon/index/fulltext/FullTextNativeRoundTripTest.java index 77a441e..41701b9 100644 --- a/java/src/test/java/org/apache/paimon/index/fulltext/FullTextNativeRoundTripTest.java +++ b/java/src/test/java/org/apache/paimon/index/fulltext/FullTextNativeRoundTripTest.java
@@ -6,6 +6,8 @@ import java.io.File; import java.io.IOException; import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import static org.junit.Assert.assertEquals; @@ -29,32 +31,37 @@ } byte[] indexBytes = output.toByteArray(); - AtomicInteger maxBatchSize = new AtomicInteger(); + AtomicInteger bytesRead = new AtomicInteger(); FullTextIndexInput input = new FullTextIndexInput() { @Override - public void readAt(long position, byte[] buffer, int offset, int length) + public void pread(long position, byte[] buffer, int offset, int length) throws IOException { - readAtBytes(indexBytes, position, buffer, offset, length); - } - - @Override - public void pread(long[] positions, byte[][] buffers) throws IOException { - maxBatchSize.updateAndGet(current -> Math.max(current, positions.length)); - FullTextIndexInput.super.pread(positions, buffers); + bytesRead.addAndGet(length); + preadBytes(indexBytes, position, buffer, offset, length); } }; try (FullTextIndexReader reader = new FullTextIndexReader(input)) { - FullTextSearchResult result = reader.search(FullTextQuery.match("paimon", "text"), 10); + int bytesReadAtOpen = bytesRead.get(); + FullTextReadMetrics metrics = reader.readMetrics(); + assertTrue(metrics.preadCalls() >= 2); + assertTrue(metrics.preadBytes() > 16); + reader.prewarm(); + FullTextReadMetrics afterPrewarm = reader.readMetrics(); + assertTrue(afterPrewarm.preadCalls() > metrics.preadCalls()); + FullTextSearchResult result = reader.search(matchQuery("paimon", "text"), 10); + FullTextReadMetrics afterSearch = reader.readMetrics(); - assertTrue(maxBatchSize.get() > 1); + assertTrue(bytesReadAtOpen < indexBytes.length); assertEquals(1, result.size()); assertEquals(10L, result.rowIds()[0]); assertTrue(result.scores()[0] > 0.0f); + assertTrue(afterSearch.preadCalls() >= afterPrewarm.preadCalls()); + assertTrue(afterSearch.cacheMisses() >= metrics.cacheMisses()); try { - reader.search(FullTextQuery.match("paimon", "text"), 0); + reader.search(matchQuery("paimon", "text"), 0); fail("Expected non-positive search limit to fail"); } catch (IllegalArgumentException expected) { assertEquals("search limit must be positive", expected.getMessage()); @@ -62,12 +69,47 @@ } } + @Test + public void testJavaNativeMultiFieldRoundTrip() throws Exception { + assumeTrue( + "PAIMON_FTINDEX_JNI_LIB_PATH must point to the built JNI library", + nativeLibraryConfigured()); + + Map<String, String> options = Collections.singletonMap("text-fields", "title,body"); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + try (FullTextIndexWriter writer = FullTextIndexWriter.create(options)) { + Map<String, String> fields = new LinkedHashMap<>(); + fields.put("title", "Apache Paimon"); + fields.put("body", "lake storage"); + writer.addDocument(20L, fields); + writer.writeIndex(output::write); + } + + byte[] indexBytes = output.toByteArray(); + FullTextIndexInput input = + (position, buffer, offset, length) -> + preadBytes(indexBytes, position, buffer, offset, length); + + try (FullTextIndexReader reader = new FullTextIndexReader(input)) { + FullTextSearchResult result = + reader.search("{\"match\":{\"query\":\"paimon\"}}", 10); + + assertEquals(1, result.size()); + assertEquals(20L, result.rowIds()[0]); + assertTrue(result.scores()[0] > 0.0f); + } + } + private static boolean nativeLibraryConfigured() { String path = System.getenv("PAIMON_FTINDEX_JNI_LIB_PATH"); return path != null && !path.isEmpty() && new File(path).isFile(); } - private static void readAtBytes( + private static String matchQuery(String terms, String column) { + return "{\"match\":{\"query\":\"" + terms + "\",\"column\":\"" + column + "\"}}"; + } + + private static void preadBytes( byte[] source, long position, byte[] buffer, int offset, int length) throws IOException { long end = position + length; if (position < 0 || end > source.length || end < position) {
diff --git a/jni/src/lib.rs b/jni/src/lib.rs index ffbb0f5..03e71cd 100644 --- a/jni/src/lib.rs +++ b/jni/src/lib.rs
@@ -2,9 +2,7 @@ use jni::sys::{jint, jlong, jobject}; use jni::{JNIEnv, JavaVM}; use paimon_ftindex_core::io::{ReadRequest, SeekRead, SeekWrite}; -use paimon_ftindex_core::{ - FullTextIndexConfig, FullTextIndexReader, FullTextIndexWriter, FullTextQuery, TokenizerConfig, -}; +use paimon_ftindex_core::{FullTextIndexConfig, FullTextIndexReader, FullTextIndexWriter}; use std::collections::HashMap; use std::io; use std::ptr; @@ -60,82 +58,67 @@ } unsafe impl Send for JavaInput {} +unsafe impl Sync for JavaInput {} impl SeekRead for JavaInput { - fn pread(&mut self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> { - if ranges.is_empty() { - return Ok(()); - } - + fn pread(&self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> { let mut env = self .jvm .attach_current_thread() .map_err(|e| io::Error::other(format!("JNI attach failed: {e}")))?; - let positions = env - .new_long_array(ranges.len() as i32) - .map_err(|e| io::Error::other(format!("new_long_array failed: {e}")))?; - let position_values: Vec<i64> = ranges.iter().map(|range| range.pos as i64).collect(); - env.set_long_array_region(&positions, 0, &position_values) - .map_err(|e| io::Error::other(format!("set_long_array_region failed: {e}")))?; - - let byte_array_class = env - .find_class("[B") - .map_err(|e| io::Error::other(format!("find byte[] class failed: {e}")))?; - let buffers = env - .new_object_array(ranges.len() as i32, byte_array_class, JObject::null()) - .map_err(|e| io::Error::other(format!("new_object_array failed: {e}")))?; - for (idx, range) in ranges.iter().enumerate() { + for range in ranges { + let position = i64::try_from(range.pos) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "offset overflow"))?; + let length = i32::try_from(range.buf.len()) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "range too large"))?; let buffer = env - .new_byte_array(range.buf.len() as i32) + .new_byte_array(length) .map_err(|e| io::Error::other(format!("new_byte_array failed: {e}")))?; - env.set_object_array_element(&buffers, idx as i32, buffer) - .map_err(|e| io::Error::other(format!("set_object_array_element failed: {e}")))?; + let buffer_obj = JObject::from(buffer); + env.call_method( + self.input.as_obj(), + "pread", + "(J[BII)V", + &[ + JValue::Long(position), + JValue::Object(&buffer_obj), + JValue::Int(0), + JValue::Int(length), + ], + ) + .map_err(|e| io::Error::other(format!("Java input pread failed: {e}")))?; + copy_java_buffer(&mut env, JByteArray::from(buffer_obj), range.buf)?; } - - env.call_method( - self.input.as_obj(), - "pread", - "([J[[B)V", - &[JValue::Object(&positions), JValue::Object(&buffers)], - ) - .map_err(|e| io::Error::other(format!("Java input pread failed: {e}")))?; - - copy_java_buffers(&mut env, &buffers, ranges) + Ok(()) } } -fn copy_java_buffers( +fn copy_java_buffer( env: &mut JNIEnv<'_>, - buffers: &JObjectArray<'_>, - ranges: &mut [ReadRequest<'_>], + buffer: JByteArray<'_>, + output: &mut [u8], ) -> io::Result<()> { - for (idx, range) in ranges.iter_mut().enumerate() { - let object = env - .get_object_array_element(buffers, idx as i32) - .map_err(|e| io::Error::other(format!("get_object_array_element failed: {e}")))?; - let buffer = JByteArray::from(object); - let length = env - .get_array_length(&buffer) - .map_err(|e| io::Error::other(format!("get_array_length failed: {e}")))? - as usize; - if length != range.buf.len() { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!( - "Java pread returned buffer length {} != {}", - length, - range.buf.len() - ), - )); - } - if length > 0 { - let mut signed = vec![0i8; range.buf.len()]; - env.get_byte_array_region(&buffer, 0, &mut signed) - .map_err(|e| io::Error::other(format!("get_byte_array_region failed: {e}")))?; - for (dst, src) in range.buf.iter_mut().zip(signed) { - *dst = src as u8; - } + let length = env + .get_array_length(&buffer) + .map_err(|e| io::Error::other(format!("get_array_length failed: {e}")))? + as usize; + if length != output.len() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "Java pread returned buffer length {} != {}", + length, + output.len() + ), + )); + } + if length > 0 { + let mut signed = vec![0i8; output.len()]; + env.get_byte_array_region(&buffer, 0, &mut signed) + .map_err(|e| io::Error::other(format!("get_byte_array_region failed: {e}")))?; + for (dst, src) in output.iter_mut().zip(signed) { + *dst = src as u8; } } Ok(()) @@ -176,6 +159,20 @@ } #[no_mangle] +pub extern "system" fn Java_org_apache_paimon_index_fulltext_FullTextNative_addDocumentFields( + mut env: JNIEnv, + _class: JClass, + writer_ptr: jlong, + row_id: jlong, + field_names: JObjectArray, + texts: JObjectArray, +) { + if let Err(e) = add_document_fields(&mut env, writer_ptr, row_id, field_names, texts) { + throw(&mut env, &e); + } +} + +#[no_mangle] pub extern "system" fn Java_org_apache_paimon_index_fulltext_FullTextNative_writeIndex( mut env: JNIEnv, _class: JClass, @@ -213,35 +210,52 @@ } #[no_mangle] -pub extern "system" fn Java_org_apache_paimon_index_fulltext_FullTextNative_searchJson( +pub extern "system" fn Java_org_apache_paimon_index_fulltext_FullTextNative_search( mut env: JNIEnv, _class: JClass, reader_ptr: jlong, - query_json: JString, + query: JString, limit: jint, ) -> jobject { - match search_json(&mut env, reader_ptr, query_json, limit, None) { + match search(&mut env, reader_ptr, query, limit, None) { Ok(obj) => obj, Err(e) => throw_and_return(&mut env, &e, ptr::null_mut()), } } #[no_mangle] -pub extern "system" fn Java_org_apache_paimon_index_fulltext_FullTextNative_searchJsonWithRoaringFilter( +pub extern "system" fn Java_org_apache_paimon_index_fulltext_FullTextNative_searchWithRoaringFilter( mut env: JNIEnv, _class: JClass, reader_ptr: jlong, - query_json: JString, + query: JString, limit: jint, roaring_filter: JByteArray, ) -> jobject { - match search_json( - &mut env, - reader_ptr, - query_json, - limit, - Some(roaring_filter), - ) { + match search(&mut env, reader_ptr, query, limit, Some(roaring_filter)) { + Ok(obj) => obj, + Err(e) => throw_and_return(&mut env, &e, ptr::null_mut()), + } +} + +#[no_mangle] +pub extern "system" fn Java_org_apache_paimon_index_fulltext_FullTextNative_prewarm( + mut env: JNIEnv, + _class: JClass, + reader_ptr: jlong, +) { + if let Err(e) = prewarm(reader_ptr) { + throw(&mut env, &e); + } +} + +#[no_mangle] +pub extern "system" fn Java_org_apache_paimon_index_fulltext_FullTextNative_readMetrics( + mut env: JNIEnv, + _class: JClass, + reader_ptr: jlong, +) -> jobject { + match read_metrics(&mut env, reader_ptr) { Ok(obj) => obj, Err(e) => throw_and_return(&mut env, &e, ptr::null_mut()), } @@ -266,8 +280,7 @@ values: JObjectArray, ) -> Result<jlong, String> { let options = options_from_arrays(env, keys, values)?; - let tokenizer = TokenizerConfig::from_options(&options).map_err(|e| e.to_string())?; - let config = FullTextIndexConfig::new().tokenizer(tokenizer); + let config = FullTextIndexConfig::from_options(&options).map_err(|e| e.to_string())?; let writer = FullTextIndexWriter::new(config).map_err(|e| e.to_string())?; Ok(Box::into_raw(Box::new(WriterHandle { inner: writer })) as jlong) } @@ -289,6 +302,21 @@ .map_err(|e| e.to_string()) } +fn add_document_fields( + env: &mut JNIEnv, + writer_ptr: jlong, + row_id: jlong, + field_names: JObjectArray, + texts: JObjectArray, +) -> Result<(), String> { + let writer = handle_mut::<WriterHandle>(writer_ptr, "writer")?; + let fields = string_pair_arrays(env, field_names, texts)?; + writer + .inner + .add_document_fields(row_id, fields) + .map_err(|e| e.to_string()) +} + fn write_index(env: &mut JNIEnv, writer_ptr: jlong, output: JObject) -> Result<(), String> { let writer = handle_mut::<WriterHandle>(writer_ptr, "writer")?; let jvm = env.get_java_vm().map_err(|e| e.to_string())?; @@ -305,30 +333,29 @@ Ok(Box::into_raw(Box::new(ReaderHandle { inner: reader })) as jlong) } -fn search_json( +fn search( env: &mut JNIEnv, reader_ptr: jlong, - query_json: JString, + query: JString, limit: jint, roaring_filter: Option<JByteArray>, ) -> Result<jobject, String> { - let reader = handle_mut::<ReaderHandle>(reader_ptr, "reader")?; - let query_json: String = env - .get_string(&query_json) - .map_err(|e| format!("failed to read query json: {e}"))? + let reader = handle_ref::<ReaderHandle>(reader_ptr, "reader")?; + let query: String = env + .get_string(&query) + .map_err(|e| format!("failed to read query: {e}"))? .into(); - let query = FullTextQuery::from_json(&query_json).map_err(|e| e.to_string())?; let limit = validate_search_limit(limit)?; let result = if let Some(roaring_filter) = roaring_filter { let roaring_filter = read_byte_array(env, roaring_filter)?; reader .inner - .search_with_roaring_filter(query, limit, &roaring_filter) + .search_with_roaring_filter(&query, limit, &roaring_filter) .map_err(|e| e.to_string())? } else { reader .inner - .search(query, limit) + .search(&query, limit) .map_err(|e| e.to_string())? }; @@ -355,6 +382,32 @@ Ok(obj.into_raw()) } +fn prewarm(reader_ptr: jlong) -> Result<(), String> { + let reader = handle_ref::<ReaderHandle>(reader_ptr, "reader")?; + reader.inner.prewarm().map_err(|e| e.to_string()) +} + +fn read_metrics(env: &mut JNIEnv, reader_ptr: jlong) -> Result<jobject, String> { + let reader = handle_ref::<ReaderHandle>(reader_ptr, "reader")?; + let metrics = reader.inner.read_metrics(); + let obj = env + .new_object( + "org/apache/paimon/index/fulltext/FullTextReadMetrics", + "(JJJJJJJ)V", + &[ + JValue::Long(metrics.pread_calls as jlong), + JValue::Long(metrics.pread_ranges as jlong), + JValue::Long(metrics.pread_bytes as jlong), + JValue::Long(metrics.cache_hits as jlong), + JValue::Long(metrics.cache_misses as jlong), + JValue::Long(metrics.cache_evictions as jlong), + JValue::Long(metrics.cached_blocks as jlong), + ], + ) + .map_err(|e| e.to_string())?; + Ok(obj.into_raw()) +} + fn validate_search_limit(limit: jint) -> Result<usize, String> { if limit <= 0 { return Err("search limit must be positive".to_string()); @@ -404,6 +457,43 @@ Ok(options) } +fn string_pair_arrays( + env: &mut JNIEnv, + keys: JObjectArray, + values: JObjectArray, +) -> Result<Vec<(String, String)>, String> { + let key_len = env.get_array_length(&keys).map_err(|e| e.to_string())?; + let value_len = env.get_array_length(&values).map_err(|e| e.to_string())?; + if key_len != value_len { + return Err(format!( + "fieldNames length {} does not match texts length {}", + key_len, value_len + )); + } + if key_len == 0 { + return Err("document fields must not be empty".to_string()); + } + let mut fields = Vec::with_capacity(key_len as usize); + for i in 0..key_len { + let key = env + .get_object_array_element(&keys, i) + .map_err(|e| e.to_string())?; + let value = env + .get_object_array_element(&values, i) + .map_err(|e| e.to_string())?; + let key: String = env + .get_string(&JString::from(key)) + .map_err(|e| e.to_string())? + .into(); + let value: String = env + .get_string(&JString::from(value)) + .map_err(|e| e.to_string())? + .into(); + fields.push((key, value)); + } + Ok(fields) +} + fn handle_mut<'a, T>(ptr: jlong, name: &str) -> Result<&'a mut T, String> { if ptr == 0 { return Err(format!("{name} is closed")); @@ -415,6 +505,17 @@ } } +fn handle_ref<'a, T>(ptr: jlong, name: &str) -> Result<&'a T, String> { + if ptr == 0 { + return Err(format!("{name} is closed")); + } + unsafe { + (ptr as *const T) + .as_ref() + .ok_or_else(|| format!("{name} is null")) + } +} + fn throw(env: &mut JNIEnv, message: &str) { let _ = env.throw_new("java/lang/RuntimeException", message); }
diff --git a/python/paimon_ftindex/__init__.py b/python/paimon_ftindex/__init__.py index 485f1e4..980574a 100644 --- a/python/paimon_ftindex/__init__.py +++ b/python/paimon_ftindex/__init__.py
@@ -1,11 +1,8 @@ -from .query import FullTextQuery, MatchQuery, PhraseQuery -from .reader import FullTextIndexReader +from .reader import FullTextIndexReader, FullTextReadMetrics from .writer import FullTextIndexWriter __all__ = [ "FullTextIndexReader", + "FullTextReadMetrics", "FullTextIndexWriter", - "FullTextQuery", - "MatchQuery", - "PhraseQuery", ]
diff --git a/python/paimon_ftindex/_ffi.py b/python/paimon_ftindex/_ffi.py index af04a6c..931c972 100644 --- a/python/paimon_ftindex/_ffi.py +++ b/python/paimon_ftindex/_ffi.py
@@ -58,7 +58,7 @@ WRITE_FN = CFUNCTYPE(c_int, c_void_p, POINTER(c_uint8), c_size_t) FLUSH_FN = CFUNCTYPE(c_int, c_void_p) -READ_AT_FN = CFUNCTYPE(c_int, c_void_p, c_uint64, POINTER(c_uint8), c_size_t) +PREAD_FN = CFUNCTYPE(c_int, c_void_p, c_uint64, POINTER(c_uint8), c_size_t) class PaimonFtindexOutputFile(Structure): @@ -72,7 +72,19 @@ class PaimonFtindexInputFile(Structure): _fields_ = [ ("ctx", c_void_p), - ("read_at_fn", READ_AT_FN), + ("pread_fn", PREAD_FN), + ] + + +class PaimonFtindexReadMetrics(Structure): + _fields_ = [ + ("pread_calls", c_uint64), + ("pread_ranges", c_uint64), + ("pread_bytes", c_uint64), + ("cache_hits", c_uint64), + ("cache_misses", c_uint64), + ("cache_evictions", c_uint64), + ("cached_blocks", c_uint64), ] @@ -85,6 +97,15 @@ lib.paimon_ftindex_writer_add_document.argtypes = [c_void_p, c_int64, c_char_p] lib.paimon_ftindex_writer_add_document.restype = c_int +lib.paimon_ftindex_writer_add_document_fields.argtypes = [ + c_void_p, + c_int64, + POINTER(c_char_p), + POINTER(c_char_p), + c_size_t, +] +lib.paimon_ftindex_writer_add_document_fields.restype = c_int + lib.paimon_ftindex_writer_write_index.argtypes = [c_void_p, PaimonFtindexOutputFile] lib.paimon_ftindex_writer_write_index.restype = c_int @@ -94,7 +115,7 @@ lib.paimon_ftindex_reader_open.argtypes = [PaimonFtindexInputFile] lib.paimon_ftindex_reader_open.restype = c_void_p -lib.paimon_ftindex_reader_search_json.argtypes = [ +lib.paimon_ftindex_reader_search.argtypes = [ c_void_p, c_char_p, c_size_t, @@ -103,9 +124,9 @@ c_size_t, POINTER(c_size_t), ] -lib.paimon_ftindex_reader_search_json.restype = c_int +lib.paimon_ftindex_reader_search.restype = c_int -lib.paimon_ftindex_reader_search_json_with_roaring_filter.argtypes = [ +lib.paimon_ftindex_reader_search_with_roaring_filter.argtypes = [ c_void_p, c_char_p, c_size_t, @@ -116,7 +137,16 @@ c_size_t, POINTER(c_size_t), ] -lib.paimon_ftindex_reader_search_json_with_roaring_filter.restype = c_int +lib.paimon_ftindex_reader_search_with_roaring_filter.restype = c_int + +lib.paimon_ftindex_reader_prewarm.argtypes = [c_void_p] +lib.paimon_ftindex_reader_prewarm.restype = c_int + +lib.paimon_ftindex_reader_read_metrics.argtypes = [ + c_void_p, + POINTER(PaimonFtindexReadMetrics), +] +lib.paimon_ftindex_reader_read_metrics.restype = c_int lib.paimon_ftindex_reader_free.argtypes = [c_void_p] lib.paimon_ftindex_reader_free.restype = None
diff --git a/python/paimon_ftindex/query.py b/python/paimon_ftindex/query.py deleted file mode 100644 index 01185ae..0000000 --- a/python/paimon_ftindex/query.py +++ /dev/null
@@ -1,43 +0,0 @@ -import json - - -class FullTextQuery: - def to_dict(self): - raise NotImplementedError - - def to_json(self): - return json.dumps(self.to_dict(), separators=(",", ":")) - - -class MatchQuery(FullTextQuery): - def __init__(self, terms, column="text", operator="Or", boost=1.0): - self.terms = str(terms) - self.column = str(column) - self.operator = str(operator) - self.boost = float(boost) - - def to_dict(self): - return { - "match": { - "column": self.column, - "terms": self.terms, - "operator": self.operator, - "boost": self.boost, - } - } - - -class PhraseQuery(FullTextQuery): - def __init__(self, terms, column="text", slop=0): - self.terms = str(terms) - self.column = str(column) - self.slop = int(slop) - - def to_dict(self): - return { - "match_phrase": { - "column": self.column, - "terms": self.terms, - "slop": self.slop, - } - }
diff --git a/python/paimon_ftindex/reader.py b/python/paimon_ftindex/reader.py index 645d583..75f2080 100644 --- a/python/paimon_ftindex/reader.py +++ b/python/paimon_ftindex/reader.py
@@ -1,22 +1,35 @@ import ctypes +from dataclasses import dataclass from ctypes import c_float, c_int64, c_size_t, c_uint8, c_void_p from ._ffi import ( - READ_AT_FN, + PREAD_FN, PaimonFtindexInputFile, + PaimonFtindexReadMetrics, check_ptr, check_status, lib, ) +@dataclass(frozen=True) +class FullTextReadMetrics: + pread_calls: int + pread_ranges: int + pread_bytes: int + cache_hits: int + cache_misses: int + cache_evictions: int + cached_blocks: int + + class FullTextIndexReader: def __init__(self, input_): self._closed = False self._input = input_ - @READ_AT_FN - def read_at_fn(ctx, pos, buf, length): + @PREAD_FN + def pread_fn(ctx, pos, buf, length): try: data = input_.pread(int(pos), int(length)) if len(data) != length: @@ -26,22 +39,22 @@ except Exception: return -1 - self._read_at_fn = read_at_fn - native_input = PaimonFtindexInputFile(c_void_p(0), read_at_fn) + self._pread_fn = pread_fn + native_input = PaimonFtindexInputFile(c_void_p(0), pread_fn) self._ptr = check_ptr(lib.paimon_ftindex_reader_open(native_input)) def search(self, query, limit=10, filter_bytes=None): if self._closed: raise RuntimeError("FullTextIndexReader is closed") - query_json = query.to_json() if hasattr(query, "to_json") else str(query) + query = str(query) capacity = int(limit) row_ids = (c_int64 * capacity)() scores = (c_float * capacity)() result_len = c_size_t() if filter_bytes is None: - status = lib.paimon_ftindex_reader_search_json( + status = lib.paimon_ftindex_reader_search( self._ptr, - query_json.encode("utf-8"), + query.encode("utf-8"), capacity, row_ids, scores, @@ -51,9 +64,9 @@ else: filter_bytes = bytes(filter_bytes) filter_buf = (c_uint8 * len(filter_bytes)).from_buffer_copy(filter_bytes) - status = lib.paimon_ftindex_reader_search_json_with_roaring_filter( + status = lib.paimon_ftindex_reader_search_with_roaring_filter( self._ptr, - query_json.encode("utf-8"), + query.encode("utf-8"), capacity, filter_buf, len(filter_bytes), @@ -66,6 +79,28 @@ size = result_len.value return list(row_ids[:size]), list(scores[:size]) + def prewarm(self): + if self._closed: + raise RuntimeError("FullTextIndexReader is closed") + check_status(lib.paimon_ftindex_reader_prewarm(self._ptr)) + + def read_metrics(self): + if self._closed: + raise RuntimeError("FullTextIndexReader is closed") + native = PaimonFtindexReadMetrics() + check_status( + lib.paimon_ftindex_reader_read_metrics(self._ptr, ctypes.byref(native)) + ) + return FullTextReadMetrics( + pread_calls=native.pread_calls, + pread_ranges=native.pread_ranges, + pread_bytes=native.pread_bytes, + cache_hits=native.cache_hits, + cache_misses=native.cache_misses, + cache_evictions=native.cache_evictions, + cached_blocks=native.cached_blocks, + ) + def close(self): if not self._closed: self._closed = True
diff --git a/python/paimon_ftindex/writer.py b/python/paimon_ftindex/writer.py index 93dd42d..7d3bb97 100644 --- a/python/paimon_ftindex/writer.py +++ b/python/paimon_ftindex/writer.py
@@ -24,15 +24,39 @@ lib.paimon_ftindex_writer_open(key_array, value_array, len(keys)) ) - def add_document(self, row_id, text): + def add_document(self, row_id, text, column=None): if self._closed: raise RuntimeError("FullTextIndexWriter is closed") + if isinstance(text, dict): + if column is not None: + raise ValueError("column must not be set when text is a dict") + self.add_document_fields(row_id, text) + return + if column is not None: + self.add_document_fields(row_id, [(column, text)]) + return check_status( lib.paimon_ftindex_writer_add_document( self._ptr, int(row_id), str(text).encode("utf-8") ) ) + def add_document_fields(self, row_id, fields): + if self._closed: + raise RuntimeError("FullTextIndexWriter is closed") + items = list(fields.items()) if hasattr(fields, "items") else list(fields) + if not items: + raise ValueError("document fields must not be empty") + names = [str(name).encode("utf-8") for name, _ in items] + texts = [str(text).encode("utf-8") for _, text in items] + name_array = (c_char_p * len(names))(*names) + text_array = (c_char_p * len(texts))(*texts) + check_status( + lib.paimon_ftindex_writer_add_document_fields( + self._ptr, int(row_id), name_array, text_array, len(items) + ) + ) + def write(self, output): if self._closed: raise RuntimeError("FullTextIndexWriter is closed")
diff --git a/python/tests/test_roundtrip.py b/python/tests/test_roundtrip.py index b6a98ca..b54cc47 100644 --- a/python/tests/test_roundtrip.py +++ b/python/tests/test_roundtrip.py
@@ -1,7 +1,10 @@ from io import BytesIO import struct -from paimon_ftindex import FullTextIndexReader, FullTextIndexWriter, MatchQuery +from paimon_ftindex import ( + FullTextIndexReader, + FullTextIndexWriter, +) class BytesInput: @@ -20,10 +23,19 @@ writer.write(output) with FullTextIndexReader(BytesInput(output.getvalue())) as reader: - row_ids, scores = reader.search(MatchQuery("paimon"), limit=10) + metrics = reader.read_metrics() + assert metrics.pread_calls >= 2 + assert metrics.pread_bytes > 16 + reader.prewarm() + after_prewarm = reader.read_metrics() + assert after_prewarm.pread_calls > metrics.pread_calls + row_ids, scores = reader.search(match_query("paimon"), limit=10) + after_search = reader.read_metrics() assert row_ids == [1] assert scores[0] > 0 + assert after_search.pread_calls >= after_prewarm.pread_calls + assert after_search.cache_misses >= metrics.cache_misses def test_python_search_with_roaring_filter(): @@ -37,13 +49,39 @@ filter_bytes = _roaring_treemap_bytes([allowed_id]) with FullTextIndexReader(BytesInput(output.getvalue())) as reader: row_ids, scores = reader.search( - MatchQuery("paimon"), limit=10, filter_bytes=filter_bytes + match_query("paimon"), limit=10, filter_bytes=filter_bytes ) assert row_ids == [allowed_id] assert scores[0] > 0 +def test_python_multi_field_round_trip(): + output = BytesIO() + with FullTextIndexWriter({"text-fields": "title,body"}) as writer: + writer.add_document_fields( + 1, + { + "title": "Apache Paimon", + "body": "lake storage", + }, + ) + writer.add_document_fields( + 2, + { + "title": "Tantivy", + "body": "Rust search engine", + }, + ) + writer.write(output) + + with FullTextIndexReader(BytesInput(output.getvalue())) as reader: + row_ids, scores = reader.search(match_query("paimon"), limit=10) + + assert row_ids == [1] + assert scores[0] > 0 + + def _roaring_treemap_bytes(ids): bitmaps = {} for value in sorted(set(ids)): @@ -80,3 +118,7 @@ payload.extend(values_bytes) offset += len(values_bytes) return bytes(out + descriptions + offsets + payload) + + +def match_query(terms): + return '{"match":{"query":"' + terms + '"}}'