feat(ftindex): primary-key full-text and hybrid search read path (full scope)

crates/paimon/Cargo.toml

@@ -41,7 +41,7 @@ storage-all = [
     "storage-gcs",
     "storage-hdfs",
 ]
-fulltext = ["tantivy", "tempfile"]
+fulltext = ["tantivy", "tempfile", "dep:paimon-ftindex-core"]
 vortex = ["dep:vortex"]
 
 storage-memory = ["opendal/services-memory"]
@@ -104,6 +104,7 @@ uuid = { version = "1", features = ["v4"] }
 urlencoding = "2.1"
 tantivy = { version = "0.22", optional = true }
 tempfile = { version = "3", optional = true }
+paimon-ftindex-core = { git = "https://github.com/apache/paimon-full-text", tag = "v0.1.0-rc4", optional = true }
 paimon-mosaic-core = "0.2.0"
 paimon-vindex-core = "0.2.0"
 vortex = { version = "0.75.0", features = ["tokio"], optional = true }

crates/paimon/src/ftindex/mod.rs

@@ -0,0 +1,22 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Full-text index reader backed by the shared `paimon-ftindex-core` engine
+//! (the same native core the Java/Python bindings use), reading the v1
+//! archive format.
+
+pub mod reader;

crates/paimon/src/ftindex/reader.rs

@@ -0,0 +1,168 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Reader over the `paimon-ftindex-core` v1 archive format.
+
+use paimon_ftindex_core::io::SliceReader;
+use paimon_ftindex_core::{FullTextIndexReader, FullTextSearchResult};
+use roaring::RoaringTreemap;
+
+use crate::io::InputFile;
+
+/// Search hits from a full-text archive: parallel row-id and score arrays,
+/// ordered best-score-first (as returned by the engine).
+#[derive(Debug, Clone)]
+pub struct FullTextHits {
+    pub row_ids: Vec<i64>,
+    pub scores: Vec<f32>,
+}
+
+impl From<FullTextSearchResult> for FullTextHits {
+    fn from(r: FullTextSearchResult) -> Self {
+        Self {
+            row_ids: r.row_ids,
+            scores: r.scores,
+        }
+    }
+}
+
+/// Reader over a single full-text index archive, backed by the shared
+/// `paimon-ftindex-core` engine (v1 archive format).
+pub struct FullTextArchiveReader {
+    inner: FullTextIndexReader<SliceReader>,
+}
+
+impl FullTextArchiveReader {
+    /// Read the whole archive from `input` into memory and open the engine
+    /// reader over it. Archives are single index files, so a whole-read keeps
+    /// peak memory at one archive.
+    pub async fn from_input_file(input: &InputFile) -> crate::Result<Self> {
+        let bytes = input.read().await?;
+        let reader =
+            FullTextIndexReader::open(SliceReader::new(bytes.to_vec())).map_err(map_ft_err)?;
+        Ok(Self { inner: reader })
+    }
+
+    /// Search with a JSON DSL query (see the engine's query spec), returning up
+    /// to `limit` best-scoring hits.
+    pub fn search(&self, query_json: &str, limit: usize) -> crate::Result<FullTextHits> {
+        self.inner
+            .search(query_json, limit)
+            .map(Into::into)
+            .map_err(map_ft_err)
+    }
+
+    /// Like [`search`](Self::search) but restricts results to `include_row_ids`
+    /// (the live-row allow-list).
+    pub fn search_with_include(
+        &self,
+        query_json: &str,
+        limit: usize,
+        include_row_ids: RoaringTreemap,
+    ) -> crate::Result<FullTextHits> {
+        let mut filter_bytes = Vec::with_capacity(include_row_ids.serialized_size());
+        include_row_ids
+            .serialize_into(&mut filter_bytes)
+            .map_err(|e| crate::Error::UnexpectedError {
+                message: format!("failed to serialize row-id filter: {e}"),
+                source: None,
+            })?;
+        self.inner
+            .search_with_roaring_filter(query_json, limit, &filter_bytes)
+            .map(Into::into)
+            .map_err(map_ft_err)
+    }
+}
+
+fn map_ft_err(e: paimon_ftindex_core::FtIndexError) -> crate::Error {
+    crate::Error::UnexpectedError {
+        message: format!("full-text index engine error: {e}"),
+        source: None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::io::FileIOBuilder;
+    use bytes::Bytes;
+    use paimon_ftindex_core::io::PosWriter;
+    use paimon_ftindex_core::{FullTextIndexConfig, FullTextIndexWriter};
+
+    // Build a tiny archive in memory using the core writer, return its bytes.
+    fn build_archive(docs: &[(i64, &str)]) -> Vec<u8> {
+        let mut writer = FullTextIndexWriter::new(FullTextIndexConfig::new()).unwrap();
+        for (row_id, text) in docs {
+            writer.add_document(*row_id, (*text).to_string()).unwrap();
+        }
+        let mut out = PosWriter::new(Vec::<u8>::new());
+        writer.write(&mut out).unwrap();
+        out.into_inner()
+    }
+
+    async fn archive_input(bytes: Vec<u8>) -> crate::io::InputFile {
+        let file_io = FileIOBuilder::new("memory").build().unwrap();
+        let path = "memory:/ft-pr1-roundtrip.archive";
+        let output = file_io.new_output(path).unwrap();
+        output.write(Bytes::from(bytes)).await.unwrap();
+        output.to_input_file()
+    }
+
+    #[tokio::test]
+    async fn test_round_trip_search_returns_expected_rows() {
+        let bytes = build_archive(&[
+            (0, "alpha bravo charlie"),
+            (1, "bravo delta"),
+            (2, "echo foxtrot"),
+        ]);
+        let input = archive_input(bytes).await;
+        let reader = FullTextArchiveReader::from_input_file(&input)
+            .await
+            .unwrap();
+
+        let hits = reader.search(r#"{"match":{"query":"bravo"}}"#, 10).unwrap();
+        let mut ids = hits.row_ids.clone();
+        ids.sort_unstable();
+        assert_eq!(ids, vec![0, 1]);
+        assert_eq!(hits.scores.len(), hits.row_ids.len());
+    }
+
+    #[tokio::test]
+    async fn test_search_with_include_restricts_to_allow_list() {
+        let bytes = build_archive(&[
+            (0, "shared token here"),
+            (1, "shared token here"),
+            (2, "shared token here"),
+        ]);
+        let input = archive_input(bytes).await;
+        let reader = FullTextArchiveReader::from_input_file(&input)
+            .await
+            .unwrap();
+
+        // All three match, but restrict to row-ids {0, 2}.
+        let mut include = roaring::RoaringTreemap::new();
+        include.insert(0);
+        include.insert(2);
+
+        let hits = reader
+            .search_with_include(r#"{"match":{"query":"token"}}"#, 10, include)
+            .unwrap();
+        let mut ids = hits.row_ids.clone();
+        ids.sort_unstable();
+        assert_eq!(ids, vec![0, 2]);
+    }
+}

crates/paimon/src/lib.rs

@@ -30,6 +30,8 @@ pub mod btree;
 pub mod catalog;
 mod deletion_vector;
 pub mod file_index;
+#[cfg(feature = "fulltext")]
+pub mod ftindex;
 pub mod io;
 pub mod lumina;
 mod predicate_stats;

crates/paimon/src/spec/core_options.rs

@@ -121,6 +121,7 @@ pub(crate) const BLOB_DESCRIPTOR_FIELD_OPTION: &str = "blob-descriptor-field";
 pub(crate) const BLOB_VIEW_FIELD_OPTION: &str = "blob-view-field";
 pub const BLOB_VIEW_RESOLVE_ENABLED_OPTION: &str = "blob-view.resolve.enabled";
 const PK_VECTOR_INDEX_COLUMNS_OPTION: &str = "pk-vector.index.columns";
+const PK_FULL_TEXT_INDEX_COLUMNS_OPTION: &str = "pk-full-text.index.columns";
 
 /// Merge engine for primary-key tables.
 ///
@@ -1174,6 +1175,23 @@ impl<'a> CoreOptions<'a> {
         crate::vindex::pkvector::metric::VectorSearchMetric::parse(&raw)?;
         Ok(raw)
     }
+
+    /// True when the PK full-text index column option key is present (any value).
+    pub fn primary_key_full_text_index_enabled(&self) -> bool {
+        self.options.contains_key(PK_FULL_TEXT_INDEX_COLUMNS_OPTION)
+    }
+
+    /// Configured PK full-text index columns: split on ',' and trim each token,
+    /// mirroring Java `split(",",-1).map(trim)`. Blank tokens are PRESERVED (do NOT
+    /// filter them) so parsing matches Java exactly; `[]` only when the key is
+    /// absent. Never returns a `Result`, never errors (do not copy the fail-loud
+    /// shape of `primary_key_vector_index_columns`).
+    pub fn primary_key_full_text_index_columns(&self) -> Vec<String> {
+        match self.options.get(PK_FULL_TEXT_INDEX_COLUMNS_OPTION) {
+            None => Vec::new(),
+            Some(raw) => raw.split(',').map(|c| c.trim().to_string()).collect(),
+        }
+    }
 }
 
 /// Parse a memory size string to bytes using binary (1024-based) semantics.
@@ -2061,4 +2079,56 @@ mod tests {
             .primary_key_vector_index_type("e")
             .is_err());
     }
+
+    #[test]
+    fn test_pk_full_text_index_absent_is_disabled_and_empty() {
+        let opts = HashMap::new();
+        let co = CoreOptions::new(&opts);
+        assert!(!co.primary_key_full_text_index_enabled());
+        assert_eq!(
+            co.primary_key_full_text_index_columns(),
+            Vec::<String>::new()
+        );
+    }
+
+    #[test]
+    fn test_pk_full_text_index_columns_split_and_trim() {
+        let opts = HashMap::from([(
+            "pk-full-text.index.columns".to_string(),
+            "a, b ,c".to_string(),
+        )]);
+        let co = CoreOptions::new(&opts);
+        assert!(co.primary_key_full_text_index_enabled());
+        assert_eq!(
+            co.primary_key_full_text_index_columns(),
+            vec!["a".to_string(), "b".to_string(), "c".to_string()]
+        );
+    }
+
+    #[test]
+    fn test_pk_full_text_index_columns_preserve_blank_tokens() {
+        // Java `split(",",-1).map(trim)` keeps empty tokens, so "a,,b" yields
+        // three columns with a blank in the middle.
+        let opts = HashMap::from([("pk-full-text.index.columns".to_string(), "a,,b".to_string())]);
+        let co = CoreOptions::new(&opts);
+        assert!(co.primary_key_full_text_index_enabled());
+        assert_eq!(
+            co.primary_key_full_text_index_columns(),
+            vec!["a".to_string(), "".to_string(), "b".to_string()]
+        );
+    }
+
+    #[test]
+    fn test_pk_full_text_index_blank_value_is_enabled_but_empty_no_error() {
+        // key present but only blanks: enabled (key exists), NO error (lenient,
+        // unlike vector). Blank tokens are PRESERVED to match Java's
+        // `split(",",-1).map(trim)`, so " , " yields two empty tokens (NOT []).
+        let opts = HashMap::from([("pk-full-text.index.columns".to_string(), " , ".to_string())]);
+        let co = CoreOptions::new(&opts);
+        assert!(co.primary_key_full_text_index_enabled());
+        assert_eq!(
+            co.primary_key_full_text_index_columns(),
+            vec!["".to_string(), "".to_string()]
+        );
+    }
 }

crates/paimon/src/spec/mod.rs

@@ -64,8 +64,8 @@ pub use manifest_file_meta::*;
 mod index_file_meta;
 pub use index_file_meta::*;
 
-mod pk_vector_source;
-pub use pk_vector_source::*;
+mod pk_index_source;
+pub use pk_index_source::*;
 
 mod index_manifest;
 pub use index_manifest::{IndexManifest, IndexManifestEntry};