You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/07/05 07:43:47 UTC
(paimon-full-text) branch codex/paimon-full-text-dsl updated: Prepare full-text index 0.1 API and storage
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch codex/paimon-full-text-dsl
in repository https://gitbox.apache.org/repos/asf/paimon-full-text.git
The following commit(s) were added to refs/heads/codex/paimon-full-text-dsl by this push:
new e8ab5b7 Prepare full-text index 0.1 API and storage
e8ab5b7 is described below
commit e8ab5b7d3118c6e7ab6d975c20a5ac0f45bb1e9f
Author: JingsongLi <ji...@gmail.com>
AuthorDate: Sun Jul 5 15:43:24 2026 +0800
Prepare full-text index 0.1 API and storage
---
Cargo.lock | 3 +
Cargo.toml | 3 +
README.md | 71 +-
core/Cargo.toml | 3 +
core/src/archive_directory.rs | 497 +++++++++++++
core/src/config.rs | 157 ++++-
core/src/index.rs | 774 ++++++++++++++++++---
core/src/io.rs | 75 +-
core/src/lib.rs | 7 +-
core/src/query.rs | 84 +--
core/src/storage.rs | 219 +++++-
core/src/tokenizer.rs | 36 +-
core/tests/core_roundtrip.rs | 676 +++++++++++++++---
core/tests/golden/storage_v1_envelope.hex | 18 +
docs/designs/2026-07-04-paimon-full-text-index.md | 467 -------------
docs/java-api.md | 37 +-
docs/paimon-integration.md | 15 +-
...-07-04-paimon-full-text-index-implementation.md | 65 --
docs/python-api.md | 40 +-
docs/storage-format.md | 13 +-
ffi/src/lib.rs | 269 +++++--
include/paimon_ftindex.h | 34 +-
.../paimon/index/fulltext/FullTextIndexInput.java | 29 +-
.../paimon/index/fulltext/FullTextIndexReader.java | 17 +-
.../paimon/index/fulltext/FullTextIndexWriter.java | 15 +
.../paimon/index/fulltext/FullTextNative.java | 13 +-
.../paimon/index/fulltext/FullTextQuery.java | 75 --
.../paimon/index/fulltext/FullTextReadMetrics.java | 57 ++
.../fulltext/FullTextNativeRoundTripTest.java | 68 +-
jni/src/lib.rs | 275 +++++---
python/paimon_ftindex/__init__.py | 7 +-
python/paimon_ftindex/_ffi.py | 42 +-
python/paimon_ftindex/query.py | 43 --
python/paimon_ftindex/reader.py | 55 +-
python/paimon_ftindex/writer.py | 26 +-
python/tests/test_roundtrip.py | 48 +-
36 files changed, 3143 insertions(+), 1190 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index b5b06ce..040ef91 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -765,10 +765,13 @@ name = "paimon-ftindex-core"
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 @@ rust-version = "1.86"
[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 @@ lake storage. The project follows the same shape as `paimon-vector-index`:
- `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 @@ and do not depend on Paimon manifest metadata.
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 @@ Supported tokenizers in this first implementation:
- `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 @@ PYTHONPATH=python python3 -m pytest -q python/tests
```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 @@ writer.add_document(1, "Apache Paimon full text search")?;
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 @@ let filtered = reader.search_with_roaring_filter(
```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 @@ class Input:
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 @@ repository.workspace = true
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, SingleSegmentIndexWriter, TantivyDocument, Term,
+ TERMINATED,
};
-use tantivy::{DocId, DocSet, Index, Score, SegmentReader, TantivyDocument, TERMINATED};
+use tantivy_fst::Automaton;
use tantivy_jieba::JiebaTokenizer;
use tempfile::TempDir;
@@ -32,12 +38,18 @@ pub struct FullTextSearchResult {
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 @@ impl FullTextIndexWriter {
}
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 @@ impl FullTextIndexWriter {
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 @@ impl FullTextIndexWriter {
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 @@ impl FullTextIndexWriter {
}
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 @@ impl<R: SeekRead> FullTextIndexReader<R> {
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 @@ impl<R: SeekRead> FullTextIndexReader<R> {
"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 @@ impl<R: SeekRead> FullTextIndexReader<R> {
}
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 @@ fn build_schema(config: &FullTextIndexConfig) -> Schema {
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 tokenizer_name(config: &TokenizerConfig) -> &'static str {
}
}
-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 @@ fn build_text_analyzer(config: &TokenizerConfig) -> Result<TextAnalyzer> {
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 collect_index_files(path: &Path) -> Result<Vec<(String, Vec<u8>)>> {
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 @@ fn build_query(
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)))
+ }
+ }
+ 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)))
}
- FullTextQuery::MatchPhrase {
+ 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 @@ fn build_query(
.parse_query(&query_text)
.map_err(|e| FtIndexError::InvalidQuery(e.to_string()))
}
- FullTextQuery::Boolean {
+ QuerySpec::Boolean {
should,
must,
must_not,
@@ -389,6 +567,16 @@ fn build_query(
"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 @@ fn build_query(
}
Ok(Box::new(BooleanQuery::new(children)))
}
- FullTextQuery::Boost {
+ QuerySpec::Boost {
positive,
negative,
negative_boost,
@@ -430,24 +618,207 @@ fn validate_match_options(
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 matches_doc(docset: &mut dyn DocSet, doc: DocId) -> bool {
}
}
-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 @@ impl<'a> ReadRequest<'a> {
}
}
-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 SliceReader {
}
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::de::Error as DeError;
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 @@ impl<'de> Deserialize<'de> for MatchOperator {
}
#[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq)]
-pub enum BooleanOccur {
+pub(crate) enum BooleanOccur {
Should,
Must,
MustNot,
@@ -51,9 +51,10 @@ impl<'de> Deserialize<'de> for BooleanOccur {
#[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 @@ pub enum FullTextQuery {
#[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 @@ pub enum FullTextQuery {
},
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 @@ where
}
}
-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::config::FullTextIndexMetadata;
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 @@ pub fn write_envelope<W: SeekWrite>(
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 @@ pub fn write_envelope<W: SeekWrite>(
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 @@ pub fn read_header<R: SeekRead>(input: &mut R) -> Result<(IndexHeader, u64)> {
)));
}
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<R: SeekRead>(
}
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 @@ struct PendingArchiveFile {
}
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 @@ where
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 @@ pub fn read_exact_at<R: SeekRead>(input: &mut R, pos: u64, buf: &mut [u8]) -> Re
#[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 @@ mod tests {
#[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 @@ mod tests {
},
];
- 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 @@ mod tests {
]
);
}
+
+ #[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 @@ pub enum TokenizerKind {
}
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(default)]
pub struct TokenizerConfig {
pub tokenizer: TokenizerKind,
pub ngram_min_gram: usize,
@@ -36,19 +37,19 @@ impl Default for TokenizerConfig {
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 @@ impl TokenizerConfig {
.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 @@ impl TokenizerConfig {
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_tokenizer(value: &str) -> Result<TokenizerKind> {
}
}
-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::io;
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 @@ fn build_index() -> anyhow::Result<Vec<u8>> {
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 CountingSliceReader {
}
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 @@ impl SeekRead for CountingSliceReader {
}
#[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 mut reader = FullTextIndexReader::open(input)?;
- let result = reader.search(FullTextQuery::match_query("paimon", "text"), 10)?;
-
+ 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 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;
+
+ 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 @@ fn match_query_round_trip() -> anyhow::Result<()> {
}
#[test]
-fn match_query_and_operator_filters_terms() -> anyhow::Result<()> {
+fn written_header_has_single_source_of_truth_for_format_and_fields() -> 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 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 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 @@ fn match_query_and_operator_filters_terms() -> anyhow::Result<()> {
#[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 @@ fn search_with_roaring_filter_limits_allowed_row_ids_before_top_docs() -> anyhow
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 @@ fn search_with_empty_roaring_filter_returns_empty_results() -> anyhow::Result<()
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 @@ fn search_with_roaring_filter_supports_64_bit_row_ids() -> anyhow::Result<()> {
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 @@ fn search_with_roaring_filter_supports_64_bit_row_ids() -> anyhow::Result<()> {
#[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 @@ fn search_rejects_invalid_roaring_filter_bytes() -> anyhow::Result<()> {
#[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 @@ fn jieba_tokenizer_searches_chinese_terms() -> anyhow::Result<()> {
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 @@ fn jieba_tokenizer_supports_chinese_phrase_queries() -> anyhow::Result<()> {
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 @@ fn jieba_tokenizer_supports_chinese_phrase_queries() -> anyhow::Result<()> {
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(())
@@ -267,6 +514,40 @@ fn tokenizer_options_parse_jieba_settings() -> anyhow::Result<()> {
Ok(())
}
+#[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())?;
@@ -276,12 +557,15 @@ fn boost_query_requires_positive_match() -> anyhow::Result<()> {
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 @@ fn boost_query_demotes_negative_matches() -> anyhow::Result<()> {
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,81 @@ fn boost_query_demotes_negative_matches() -> anyhow::Result<()> {
}
#[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)?;
+fn search_accepts_paimon_match_query_aliases() -> anyhow::Result<()> {
+ let bytes = build_index()?;
+ 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!(parsed, query);
+ assert_eq!(result.row_ids.len(), 2);
+ assert!(result.row_ids.contains(&10));
+ assert!(result.row_ids.contains(&12));
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}}"#,
+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()),
+ ],
)?;
- 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,
- }
- );
+ 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_accepts_paimon_logical_column_name() -> anyhow::Result<()> {
+fn search_rejects_unknown_column_name() -> 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 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);
@@ -357,12 +682,26 @@ fn search_accepts_paimon_logical_column_name() -> anyhow::Result<()> {
}
#[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 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 mut reader = FullTextIndexReader::open(SliceReader::new(bytes))?;
+ 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);
@@ -370,3 +709,158 @@ fn parse_paimon_boolean_query_json_forms() -> anyhow::Result<()> {
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 @@ Main classes:
- `FullTextIndexWriter`
- `FullTextIndexReader`
-- `FullTextQuery`
- `FullTextSearchResult`
- `FullTextIndexInput`
- `FullTextIndexOutput`
@@ -18,27 +17,49 @@ Main classes:
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 @@ Suggested option namespace:
```text
fulltext.tokenizer
+fulltext.text-fields
fulltext.ngram.min-gram
fulltext.ngram.max-gram
fulltext.ngram.prefix-only
@@ -29,8 +29,19 @@ 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 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 @@ paimon_ftindex
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)
+ 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()` 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 @@ The input object must provide:
- `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 @@ body: concatenated Tantivy files
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 @@ impl SeekWrite for FfiOutputFile {
#[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 @@ struct FfiInputFile {
}
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 @@ impl SeekRead for FfiInputFile {
};
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 @@ pub struct PaimonFtindexReaderHandle {
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 @@ pub unsafe extern "C" fn paimon_ftindex_writer_open(
) -> *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,
@@ -203,6 +229,24 @@ pub unsafe extern "C" fn paimon_ftindex_writer_add_document(
})
}
+#[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,
@@ -236,9 +280,9 @@ pub unsafe extern "C" fn paimon_ftindex_reader_open(
}
#[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 @@ pub unsafe extern "C" fn paimon_ftindex_reader_search_json(
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 @@ pub unsafe extern "C" fn paimon_ftindex_reader_search_json(
}
#[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 @@ pub unsafe extern "C" fn paimon_ftindex_reader_search_json_with_roaring_filter(
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,
@@ -292,6 +336,29 @@ pub unsafe extern "C" fn paimon_ftindex_reader_search_json_with_roaring_filter(
})
}
+#[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() {
@@ -324,10 +391,39 @@ unsafe fn options_from_raw(
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 @@ unsafe fn cstr_to_string(ptr: *const c_char, name: &str) -> Result<String, Strin
.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 @@ mod tests {
}
#[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 @@ mod tests {
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 @@ mod tests {
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 @@ mod tests {
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 @@ mod tests {
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 @@ mod tests {
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 {
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 @@ int paimon_ftindex_writer_add_document(
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 @@ void paimon_ftindex_writer_free(PaimonFtindexWriterHandle *writer);
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 @@ int paimon_ftindex_reader_search_json_with_roaring_filter(
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 @@ import java.io.IOException;
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 @@ public final class FullTextIndexReader implements AutoCloseable {
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 @@ public final class FullTextIndexReader implements AutoCloseable {
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 @@ public final class FullTextIndexWriter implements AutoCloseable {
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 @@ final class FullTextNative {
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 searchWithRoaringFilter(
+ long readerPtr, String query, int limit, byte[] roaringFilter);
+
+ static native void prewarm(long readerPtr);
- static native FullTextSearchResult searchJsonWithRoaringFilter(
- long readerPtr, String queryJson, int limit, byte[] roaringFilter);
+ 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.ByteArrayOutputStream;
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 @@ public class FullTextNativeRoundTripTest {
}
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 @@ public class FullTextNativeRoundTripTest {
}
}
+ @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::objects::{GlobalRef, JByteArray, JClass, JObject, JObjectArray, JString
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 @@ struct JavaInput {
}
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(())
@@ -175,6 +158,20 @@ pub extern "system" fn Java_org_apache_paimon_index_fulltext_FullTextNative_addD
}
}
+#[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,
@@ -213,35 +210,52 @@ pub extern "system" fn Java_org_apache_paimon_index_fulltext_FullTextNative_open
}
#[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 @@ fn create_writer(
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 @@ fn add_document(
.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 @@ fn open_reader(env: &mut JNIEnv, input: JObject) -> Result<jlong, String> {
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 @@ fn search_json(
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 @@ fn options_from_arrays(
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_mut<'a, T>(ptr: jlong, name: &str) -> Result<&'a mut T, String> {
}
}
+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 @@ lib = _load_library()
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 PaimonFtindexOutputFile(Structure):
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_open.restype = c_void_p
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_writer_free.restype = None
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 @@ lib.paimon_ftindex_reader_search_json.argtypes = [
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 @@ lib.paimon_ftindex_reader_search_json_with_roaring_filter.argtypes = [
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 @@ class FullTextIndexReader:
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 @@ class FullTextIndexReader:
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 @@ class FullTextIndexReader:
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 @@ class FullTextIndexWriter:
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 @@ def test_python_round_trip():
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 @@ def test_python_search_with_roaring_filter():
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 @@ def _roaring_bitmap_bytes(values):
payload.extend(values_bytes)
offset += len(values_bytes)
return bytes(out + descriptions + offsets + payload)
+
+
+def match_query(terms):
+ return '{"match":{"query":"' + terms + '"}}'