This is an automated email from the ASF dual-hosted git repository. jerry-024 pushed a commit to branch revert-13-feature/paimon-rust-fulltext-fixes in repository https://gitbox.apache.org/repos/asf/paimon-full-text.git
commit e391fcad4bf0c3a6358e89bdf854274dc4af577a Author: jerry <[email protected]> AuthorDate: Wed Jul 15 17:40:48 2026 +0800 Revert "fix(core): stream full-text index writes" --- core/src/index.rs | 164 +++++++-------------- core/src/storage.rs | 46 +----- core/tests/core_roundtrip.rs | 85 +---------- ffi/src/lib.rs | 4 - .../paimon/index/fulltext/FullTextIndexWriter.java | 6 - python/paimon_ftindex/writer.py | 5 - 6 files changed, 55 insertions(+), 255 deletions(-) diff --git a/core/src/index.rs b/core/src/index.rs index 007812b..0738c57 100644 --- a/core/src/index.rs +++ b/core/src/index.rs @@ -20,14 +20,14 @@ use crate::config::{FullTextIndexConfig, FullTextIndexMetadata}; use crate::error::{FtIndexError, Result}; use crate::io::{FullTextReadMetrics, ReadMetrics, ReadRequest, SeekRead, SeekWrite}; use crate::query::{BooleanOccur, MatchOperator, QuerySpec}; -use crate::storage::{read_header, write_envelope_from_paths, ArchiveFileEntry, IndexHeader}; +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, PathBuf}; +use std::path::Path; use std::sync::{Arc, Mutex}; use tantivy::collector::{FilterCollector, TopDocs}; use tantivy::query::{ @@ -40,14 +40,13 @@ use tantivy::tokenizer::{ SimpleTokenizer, Stemmer, StopWordFilter, TextAnalyzer, TokenStream, WhitespaceTokenizer, }; use tantivy::{ - DocId, DocSet, Index, IndexWriter, Score, SegmentReader, TantivyDocument, Term, TERMINATED, + DocId, DocSet, Index, Score, SegmentReader, SingleSegmentIndexWriter, TantivyDocument, Term, + TERMINATED, }; use tantivy_fst::Automaton; use tantivy_jieba::JiebaTokenizer; use tempfile::TempDir; -const INDEX_WRITER_MEMORY_BUDGET_BYTES: usize = 50_000_000; - #[derive(Clone, Debug, PartialEq)] pub struct FullTextSearchResult { pub row_ids: Vec<i64>, @@ -56,15 +55,7 @@ pub struct FullTextSearchResult { pub struct FullTextIndexWriter { config: FullTextIndexConfig, - state: Option<FullTextIndexWriterState>, - row_id_field: tantivy::schema::Field, - text_fields: HashMap<String, tantivy::schema::Field>, - document_count: u64, -} - -struct FullTextIndexWriterState { - index_writer: IndexWriter<TantivyDocument>, - temp_dir: TempDir, + documents: Vec<FullTextDocument>, } #[derive(Clone, Debug, PartialEq, Eq)] @@ -76,24 +67,9 @@ pub struct FullTextDocument { impl FullTextIndexWriter { pub fn new(config: FullTextIndexConfig) -> Result<Self> { config.validate()?; - let temp_dir = TempDir::new()?; - let schema = build_schema(&config); - let mut index = Index::create_in_dir(temp_dir.path(), schema.clone())?; - register_tokenizer(&mut index, &config.tokenizer)?; - let row_id_field = schema - .get_field(&config.row_id_field) - .map_err(|_| FtIndexError::InvalidStorage("missing row_id field".to_string()))?; - let text_fields = text_field_map(&schema, &config)?; - let index_writer = index.writer_with_num_threads(1, INDEX_WRITER_MEMORY_BUDGET_BYTES)?; Ok(Self { config, - state: Some(FullTextIndexWriterState { - index_writer, - temp_dir, - }), - row_id_field, - text_fields, - document_count: 0, + documents: Vec::new(), }) } @@ -113,79 +89,72 @@ impl FullTextIndexWriter { "row id must be non-negative, got {row_id}" ))); } - let state = self.state.as_ref().ok_or_else(|| { - FtIndexError::InvalidStorage("full-text index writer is already finalized".to_string()) - })?; - let next_document_count = self.document_count.checked_add(1).ok_or_else(|| { - FtIndexError::InvalidStorage("full-text document count overflow".to_string()) - })?; - let mut doc = TantivyDocument::new(); - doc.add_u64(self.row_id_field, row_id as u64); - let mut has_text_field = false; - for (name, text) in fields { - let name = name.into(); - let text = text.into(); - validate_indexed_field(&self.config, &name)?; - let text_field = self.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); - has_text_field = true; - } - if !has_text_field { + 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(), )); } - state.index_writer.add_document(doc)?; - self.document_count = next_document_count; + for (name, _) in &fields { + validate_indexed_field(&self.config, name)?; + } + self.documents.push(FullTextDocument { row_id, fields }); Ok(()) } - /// Finalizes this writer and streams the completed index archive to `output`. - /// - /// A write attempt is single-use regardless of whether it succeeds: the active Tantivy writer - /// is consumed before commit and serialization begin. After this method is called, subsequent - /// calls to `write`, `add_document`, or `add_document_fields` return an already-finalized - /// error. If `output` returns an error, it may contain a partial archive and must be discarded; - /// retrying requires a new writer and re-adding the documents. pub fn write<W: SeekWrite>(&mut self, output: &mut W) -> Result<()> { - let state = self.state.take().ok_or_else(|| { - FtIndexError::InvalidStorage("full-text index writer is already finalized".to_string()) - })?; - let FullTextIndexWriterState { - mut index_writer, - temp_dir, - } = state; - index_writer.commit()?; - index_writer.wait_merging_threads()?; + let temp_dir = TempDir::new()?; + let schema = build_schema(&self.config); + let mut index = Index::create_in_dir(temp_dir.path(), schema.clone())?; + register_tokenizer(&mut index, &self.config.tokenizer)?; + let row_id_field = schema + .get_field(&self.config.row_id_field) + .map_err(|_| FtIndexError::InvalidStorage("missing row_id field".to_string()))?; + let text_fields = text_field_map(&schema, &self.config)?; + + { + 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, 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.finalize()?; + } let files = collect_index_files(temp_dir.path())?; let mut offset = 0u64; let mut entries = Vec::with_capacity(files.len()); - for file in &files { + for (name, data) in &files { entries.push(ArchiveFileEntry { - name: file.name.clone(), + name: name.clone(), offset, - length: file.length, + length: data.len() as u64, }); - offset = offset.checked_add(file.length).ok_or_else(|| { - FtIndexError::InvalidStorage("full-text archive size overflow".to_string()) - })?; + offset += data.len() as u64; } let header = IndexHeader { metadata: FullTextIndexMetadata { config: self.config.clone(), - document_count: self.document_count, + document_count: self.documents.len() as u64, tantivy_version: tantivy::version().to_string(), }, files: entries, }; - let paths = files.iter().map(|file| &file.path).collect::<Vec<_>>(); - write_envelope_from_paths(output, &header, &paths) + write_envelope(output, &header, &files) } } @@ -487,13 +456,7 @@ fn validate_indexed_field(config: &FullTextIndexConfig, field: &str) -> Result<( } } -struct IndexFile { - name: String, - path: PathBuf, - length: u64, -} - -fn collect_index_files(path: &Path) -> Result<Vec<IndexFile>> { +fn collect_index_files(path: &Path) -> Result<Vec<(String, Vec<u8>)>> { let mut paths = Vec::new(); for entry in fs::read_dir(path)? { let entry = entry?; @@ -512,8 +475,7 @@ fn collect_index_files(path: &Path) -> Result<Vec<IndexFile>> { if name.ends_with(".lock") { continue; } - let length = fs::metadata(&path)?.len(); - files.push(IndexFile { name, path, length }); + files.push((name, fs::read(path)?)); } Ok(files) } @@ -1216,29 +1178,3 @@ impl Automaton for PrefixedDfaAutomaton { (dfa_state, prefix_state) } } - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn added_documents_are_not_retained_as_source_strings() -> Result<()> { - let mut writer = FullTextIndexWriter::new(FullTextIndexConfig::new())?; - - for row_id in 0..10_000 { - writer.add_document(row_id, format!("document {row_id} with unique source text"))?; - } - - // Keep this exhaustive: the writer state must not grow a source-document collection. - let FullTextIndexWriter { - config: _, - state, - row_id_field: _, - text_fields: _, - document_count, - } = writer; - assert!(state.is_some()); - assert_eq!(document_count, 10_000); - Ok(()) - } -} diff --git a/core/src/storage.rs b/core/src/storage.rs index eca06d4..e8322a0 100644 --- a/core/src/storage.rs +++ b/core/src/storage.rs @@ -20,8 +20,6 @@ use crate::error::{FtIndexError, Result}; use crate::io::{ReadRequest, SeekRead, SeekWrite}; use serde::{Deserialize, Serialize}; use std::collections::HashSet; -use std::fs::File; -use std::io::Read; use std::path::{Component, Path}; pub const FORMAT_MAGIC: &[u8; 8] = b"PFTIDX01"; @@ -29,7 +27,6 @@ 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; -const ARCHIVE_WRITE_BUFFER_BYTES: usize = 64 * 1024; #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] pub struct ArchiveFileEntry { @@ -49,45 +46,6 @@ pub fn write_envelope<W: SeekWrite>( header: &IndexHeader, files: &[(String, Vec<u8>)], ) -> Result<()> { - write_envelope_header(output, header)?; - for (_, data) in files { - output.write_all(data)?; - } - output.flush()?; - Ok(()) -} - -pub(crate) fn write_envelope_from_paths<W: SeekWrite>( - output: &mut W, - header: &IndexHeader, - paths: &[impl AsRef<Path>], -) -> Result<()> { - if header.files.len() != paths.len() { - return Err(FtIndexError::InvalidStorage(format!( - "archive header contains {} files but {} paths were provided", - header.files.len(), - paths.len() - ))); - } - - write_envelope_header(output, header)?; - let mut buffer = vec![0u8; ARCHIVE_WRITE_BUFFER_BYTES]; - for (entry, path) in header.files.iter().zip(paths) { - let mut file = File::open(path)?; - let mut remaining = entry.length; - while remaining > 0 { - let chunk_len = usize::try_from(remaining.min(buffer.len() as u64)) - .expect("chunk length is bounded by the archive write buffer"); - file.read_exact(&mut buffer[..chunk_len])?; - output.write_all(&buffer[..chunk_len])?; - remaining -= chunk_len as u64; - } - } - output.flush()?; - Ok(()) -} - -fn write_envelope_header<W: SeekWrite>(output: &mut W, header: &IndexHeader) -> Result<()> { let header_json = serde_json::to_vec(header)?; if header_json.len() > MAX_HEADER_BYTES { return Err(FtIndexError::InvalidStorage(format!( @@ -104,6 +62,10 @@ fn write_envelope_header<W: SeekWrite>(output: &mut W, header: &IndexHeader) -> .to_be_bytes(), )?; output.write_all(&header_json)?; + for (_, data) in files { + output.write_all(data)?; + } + output.flush()?; Ok(()) } diff --git a/core/tests/core_roundtrip.rs b/core/tests/core_roundtrip.rs index 1b71da9..2695191 100644 --- a/core/tests/core_roundtrip.rs +++ b/core/tests/core_roundtrip.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use paimon_ftindex_core::io::{PosWriter, ReadRequest, SeekRead, SeekWrite, SliceReader}; +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, FullTextIndexMetadata, FullTextIndexReader, FullTextIndexWriter, @@ -147,89 +147,6 @@ impl SeekRead for CountingSliceReader { } } -struct ChunkLimitedWriter { - data: Vec<u8>, - max_chunk_len: usize, -} - -impl ChunkLimitedWriter { - fn new() -> Self { - Self { - data: Vec::new(), - max_chunk_len: 0, - } - } -} - -impl SeekWrite for ChunkLimitedWriter { - fn write_all(&mut self, buf: &[u8]) -> io::Result<()> { - const MAX_CHUNK_LEN: usize = 64 * 1024; - if buf.len() > MAX_CHUNK_LEN { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - format!("archive write chunk exceeds {MAX_CHUNK_LEN} bytes"), - )); - } - self.max_chunk_len = self.max_chunk_len.max(buf.len()); - self.data.extend_from_slice(buf); - Ok(()) - } -} - -struct FailingWriter; - -impl SeekWrite for FailingWriter { - fn write_all(&mut self, _buf: &[u8]) -> io::Result<()> { - Err(io::Error::other("intentional output failure")) - } -} - -#[test] -fn failed_write_attempt_finalizes_writer() -> anyhow::Result<()> { - let mut writer = FullTextIndexWriter::new(FullTextIndexConfig::new())?; - writer.add_document(1, "Apache Paimon full text")?; - - let write_error = writer - .write(&mut FailingWriter) - .expect_err("the failing output should reject the first write attempt"); - assert!(write_error - .to_string() - .contains("intentional output failure")); - - let mut retry_output = ChunkLimitedWriter::new(); - let retry_error = writer - .write(&mut retry_output) - .expect_err("a failed write attempt should still finalize the writer"); - assert!(retry_error.to_string().contains("already finalized")); - - let add_error = writer - .add_document(2, "retry document") - .expect_err("a finalized writer should reject additional documents"); - assert!(add_error.to_string().contains("already finalized")); - Ok(()) -} - -#[test] -fn large_incremental_archive_is_streamed_and_searchable() -> anyhow::Result<()> { - let mut writer = FullTextIndexWriter::new(FullTextIndexConfig::new())?; - for row_id in 0..10_000 { - writer.add_document( - row_id, - format!("common archive text with unique marker{row_id}"), - )?; - } - - let mut output = ChunkLimitedWriter::new(); - writer.write(&mut output)?; - - assert!(output.max_chunk_len <= 64 * 1024); - let reader = FullTextIndexReader::open(SliceReader::new(output.data))?; - assert_eq!(reader.metadata().document_count, 10_000); - let result = reader.search(match_query("marker8191", "text"), 10)?; - assert_eq!(result.row_ids, vec![8191]); - Ok(()) -} - #[test] fn reader_open_does_not_load_all_archive_files() -> anyhow::Result<()> { let bytes = build_index()?; diff --git a/ffi/src/lib.rs b/ffi/src/lib.rs index 8908f82..60b7921 100644 --- a/ffi/src/lib.rs +++ b/ffi/src/lib.rs @@ -265,10 +265,6 @@ pub unsafe extern "C" fn paimon_ftindex_writer_add_document_fields( } #[no_mangle] -/// Finalizes the writer and writes its archive to `output`. -/// -/// The writer is finalized after any call, including calls that return a non-zero status. Callers -/// must discard a potentially partial output and create a new writer to retry. pub unsafe extern "C" fn paimon_ftindex_writer_write_index( writer: *mut PaimonFtindexWriterHandle, output: PaimonFtindexOutputFile, 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 337b996..776c851 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 @@ -62,12 +62,6 @@ public final class FullTextIndexWriter implements AutoCloseable { FullTextNative.addDocumentFields(requireOpen(), rowId, fieldNames, texts); } - /** - * Finalizes this writer and streams the index archive to the output. - * - * <p>Every write attempt finalizes the native writer, even when writing or flushing fails. - * Discard a potentially partial output and create a new writer to retry. - */ public void writeIndex(FullTextIndexOutput output) { if (output == null) { throw new NullPointerException("output"); diff --git a/python/paimon_ftindex/writer.py b/python/paimon_ftindex/writer.py index ff4c6e1..6f609a9 100644 --- a/python/paimon_ftindex/writer.py +++ b/python/paimon_ftindex/writer.py @@ -75,11 +75,6 @@ class FullTextIndexWriter: ) def write(self, output): - """Finalize this writer and stream the index archive to ``output``. - - Every write attempt finalizes the native writer, even when writing or flushing fails. - Discard a potentially partial output and create a new writer to retry. - """ if self._closed: raise RuntimeError("FullTextIndexWriter is closed")
