You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by je...@apache.org on 2026/07/15 09:40:55 UTC
(paimon-full-text) branch revert-13-feature/paimon-rust-fulltext-fixes created (now e391fca)
This is an automated email from the ASF dual-hosted git repository.
jerry-024 pushed a change to branch revert-13-feature/paimon-rust-fulltext-fixes
in repository https://gitbox.apache.org/repos/asf/paimon-full-text.git
at e391fca Revert "fix(core): stream full-text index writes"
This branch includes the following new commits:
new e391fca Revert "fix(core): stream full-text index writes"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(paimon-full-text) 01/01: Revert "fix(core): stream full-text index writes"
Posted by je...@apache.org.
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 <ji...@gmail.com>
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")