This is an automated email from the ASF dual-hosted git repository.
XiaoHongbo-Hope pushed a commit to branch release-0.1.0
in repository https://gitbox.apache.org/repos/asf/paimon-full-text.git
The following commit(s) were added to refs/heads/release-0.1.0 by this push:
new ba268f6 fix(core): stream full-text index writes #13
ba268f6 is described below
commit ba268f6287fd179651ebd2f1c689454defb3218b
Author: jerry <[email protected]>
AuthorDate: Wed Jul 15 17:40:18 2026 +0800
fix(core): stream full-text index writes #13
(cherry picked from commit 87cb32dd8f4991f3b3943c864b57aac99c06f4f0)
---
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, 255 insertions(+), 55 deletions(-)
diff --git a/core/src/index.rs b/core/src/index.rs
index 0738c57..007812b 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, ArchiveFileEntry,
IndexHeader};
+use crate::storage::{read_header, write_envelope_from_paths, 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::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use tantivy::collector::{FilterCollector, TopDocs};
use tantivy::query::{
@@ -40,13 +40,14 @@ use tantivy::tokenizer::{
SimpleTokenizer, Stemmer, StopWordFilter, TextAnalyzer, TokenStream,
WhitespaceTokenizer,
};
use tantivy::{
- DocId, DocSet, Index, Score, SegmentReader, SingleSegmentIndexWriter,
TantivyDocument, Term,
- TERMINATED,
+ DocId, DocSet, Index, IndexWriter, Score, SegmentReader, 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>,
@@ -55,7 +56,15 @@ pub struct FullTextSearchResult {
pub struct FullTextIndexWriter {
config: FullTextIndexConfig,
- documents: Vec<FullTextDocument>,
+ 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,
}
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -67,9 +76,24 @@ 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,
- documents: Vec::new(),
+ state: Some(FullTextIndexWriterState {
+ index_writer,
+ temp_dir,
+ }),
+ row_id_field,
+ text_fields,
+ document_count: 0,
})
}
@@ -89,72 +113,79 @@ impl FullTextIndexWriter {
"row id must be non-negative, got {row_id}"
)));
}
- let fields = fields
- .into_iter()
- .map(|(name, text)| (name.into(), text.into()))
- .collect::<Vec<_>>();
- if fields.is_empty() {
+ 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 {
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 });
+ state.index_writer.add_document(doc)?;
+ self.document_count = next_document_count;
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 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 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 files = collect_index_files(temp_dir.path())?;
let mut offset = 0u64;
let mut entries = Vec::with_capacity(files.len());
- for (name, data) in &files {
+ for file in &files {
entries.push(ArchiveFileEntry {
- name: name.clone(),
+ name: file.name.clone(),
offset,
- length: data.len() as u64,
+ length: file.length,
});
- offset += data.len() as u64;
+ offset = offset.checked_add(file.length).ok_or_else(|| {
+ FtIndexError::InvalidStorage("full-text archive size
overflow".to_string())
+ })?;
}
let header = IndexHeader {
metadata: FullTextIndexMetadata {
config: self.config.clone(),
- document_count: self.documents.len() as u64,
+ document_count: self.document_count,
tantivy_version: tantivy::version().to_string(),
},
files: entries,
};
- write_envelope(output, &header, &files)
+ let paths = files.iter().map(|file| &file.path).collect::<Vec<_>>();
+ write_envelope_from_paths(output, &header, &paths)
}
}
@@ -456,7 +487,13 @@ fn validate_indexed_field(config: &FullTextIndexConfig,
field: &str) -> Result<(
}
}
-fn collect_index_files(path: &Path) -> Result<Vec<(String, Vec<u8>)>> {
+struct IndexFile {
+ name: String,
+ path: PathBuf,
+ length: u64,
+}
+
+fn collect_index_files(path: &Path) -> Result<Vec<IndexFile>> {
let mut paths = Vec::new();
for entry in fs::read_dir(path)? {
let entry = entry?;
@@ -475,7 +512,8 @@ fn collect_index_files(path: &Path) -> Result<Vec<(String,
Vec<u8>)>> {
if name.ends_with(".lock") {
continue;
}
- files.push((name, fs::read(path)?));
+ let length = fs::metadata(&path)?.len();
+ files.push(IndexFile { name, path, length });
}
Ok(files)
}
@@ -1178,3 +1216,29 @@ 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 e8322a0..eca06d4 100644
--- a/core/src/storage.rs
+++ b/core/src/storage.rs
@@ -20,6 +20,8 @@ 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";
@@ -27,6 +29,7 @@ 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 {
@@ -46,6 +49,45 @@ 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!(
@@ -62,10 +104,6 @@ pub fn write_envelope<W: SeekWrite>(
.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 2695191..1b71da9 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, SliceReader};
+use paimon_ftindex_core::io::{PosWriter, ReadRequest, SeekRead, SeekWrite,
SliceReader};
use paimon_ftindex_core::storage::{read_header, write_envelope,
ArchiveFileEntry, IndexHeader};
use paimon_ftindex_core::{
FullTextIndexConfig, FullTextIndexMetadata, FullTextIndexReader,
FullTextIndexWriter,
@@ -147,6 +147,89 @@ 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 60b7921..8908f82 100644
--- a/ffi/src/lib.rs
+++ b/ffi/src/lib.rs
@@ -265,6 +265,10 @@ 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 776c851..337b996 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,6 +62,12 @@ 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 6f609a9..ff4c6e1 100644
--- a/python/paimon_ftindex/writer.py
+++ b/python/paimon_ftindex/writer.py
@@ -75,6 +75,11 @@ 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")
