[python] Introduce Full Text Search and Tantivy index in Python

Author: JingsongLi

.github/workflows/paimon-python-checks.yml

@@ -79,6 +79,20 @@ jobs:
           java -version
           mvn -version
 
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Build Tantivy native library
+        run: |
+          cd paimon-tantivy/paimon-tantivy-jni/rust
+          cargo build --release
+
+      - name: Copy Tantivy native library to resources
+        run: |
+          RESOURCE_DIR=paimon-tantivy/paimon-tantivy-jni/src/main/resources/native/linux-amd64
+          mkdir -p ${RESOURCE_DIR}
+          cp paimon-tantivy/paimon-tantivy-jni/rust/target/release/libtantivy_jni.so ${RESOURCE_DIR}/
+
       - name: Verify Python version
         run: python --version
 

docs/content/append-table/global-index.md

@@ -211,4 +211,25 @@ try (RecordReader<InternalRow> reader = readBuilder.newRead().createReader(plan)
 ```
 {{< /tab >}}
 
+{{< tab "Python SDK" >}}
+```python
+table = catalog.get_table('db.my_table')
+
+# Step 1: Build full-text search
+builder = table.new_full_text_search_builder()
+builder.with_text_column('content')
+builder.with_query_text('paimon lake format')
+builder.with_limit(10)
+result = builder.execute_local()
+
+# Step 2: Read matching rows using the search result
+read_builder = table.new_read_builder()
+scan = read_builder.new_scan().with_global_index_result(result)
+plan = scan.plan()
+table_read = read_builder.new_read()
+pa_table = table_read.to_arrow(plan.splits())
+print(pa_table)
+```
+{{< /tab >}}
+
 {{< /tabs >}}

paimon-python/dev/run_mixed_tests.sh

@@ -244,6 +244,30 @@ run_compressed_text_test() {
     fi
 }
 
+# Function to run Tantivy full-text index test (Java write index, Python read and search)
+run_tantivy_fulltext_test() {
+    echo -e "${YELLOW}=== Step 8: Running Tantivy Full-Text Index Test (Java Write, Python Read) ===${NC}"
+
+    cd "$PROJECT_ROOT"
+
+    echo "Running Maven test for JavaPyTantivyE2ETest.testTantivyFullTextIndexWrite..."
+    if mvn test -Dtest=org.apache.paimon.tantivy.index.JavaPyTantivyE2ETest#testTantivyFullTextIndexWrite -pl paimon-tantivy/paimon-tantivy-index -q -Drun.e2e.tests=true; then
+        echo -e "${GREEN}✓ Java test completed successfully${NC}"
+    else
+        echo -e "${RED}✗ Java test failed${NC}"
+        return 1
+    fi
+    cd "$PAIMON_PYTHON_DIR"
+    echo "Running Python test for JavaPyReadWriteTest.test_read_tantivy_full_text_index..."
+    if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest::test_read_tantivy_full_text_index -v; then
+        echo -e "${GREEN}✓ Python test completed successfully${NC}"
+        return 0
+    else
+        echo -e "${RED}✗ Python test failed${NC}"
+        return 1
+    fi
+}
+
 # Main execution
 main() {
     local java_write_result=0
@@ -253,6 +277,7 @@ main() {
     local pk_dv_result=0
     local btree_index_result=0
     local compressed_text_result=0
+    local tantivy_fulltext_result=0
 
     echo -e "${YELLOW}Starting mixed language test execution...${NC}"
     echo ""
@@ -311,6 +336,13 @@ main() {
 
     echo ""
 
+    # Run Tantivy full-text index test (Java write index, Python read and search)
+    if ! run_tantivy_fulltext_test; then
+        tantivy_fulltext_result=1
+    fi
+
+    echo ""
+
     echo -e "${YELLOW}=== Test Results Summary ===${NC}"
 
     if [[ $java_write_result -eq 0 ]]; then
@@ -355,12 +387,18 @@ main() {
         echo -e "${RED}✗ Compressed Text Test (Java Write, Python Read): FAILED${NC}"
     fi
 
+    if [[ $tantivy_fulltext_result -eq 0 ]]; then
+        echo -e "${GREEN}✓ Tantivy Full-Text Index Test (Java Write, Python Read): PASSED${NC}"
+    else
+        echo -e "${RED}✗ Tantivy Full-Text Index Test (Java Write, Python Read): FAILED${NC}"
+    fi
+
     echo ""
 
     # Clean up warehouse directory after all tests
     cleanup_warehouse
 
-    if [[ $java_write_result -eq 0 && $python_read_result -eq 0 && $python_write_result -eq 0 && $java_read_result -eq 0 && $pk_dv_result -eq 0 && $btree_index_result -eq 0 && $compressed_text_result -eq 0 ]]; then
+    if [[ $java_write_result -eq 0 && $python_read_result -eq 0 && $python_write_result -eq 0 && $java_read_result -eq 0 && $pk_dv_result -eq 0 && $btree_index_result -eq 0 && $compressed_text_result -eq 0 && $tantivy_fulltext_result -eq 0 ]]; then
         echo -e "${GREEN}🎉 All tests passed! Java-Python interoperability verified.${NC}"
         return 0
     else

paimon-python/pypaimon/globalindex/__init__.py

@@ -19,6 +19,7 @@
 from pypaimon.globalindex.global_index_result import GlobalIndexResult
 from pypaimon.globalindex.global_index_reader import GlobalIndexReader, FieldRef
 from pypaimon.globalindex.vector_search import VectorSearch
+from pypaimon.globalindex.full_text_search import FullTextSearch
 from pypaimon.globalindex.vector_search_result import (
     ScoredGlobalIndexResult,
     DictBasedScoredIndexResult,
@@ -27,19 +28,22 @@
 from pypaimon.globalindex.global_index_meta import GlobalIndexMeta, GlobalIndexIOMeta
 from pypaimon.globalindex.global_index_evaluator import GlobalIndexEvaluator
 from pypaimon.globalindex.global_index_scanner import GlobalIndexScanner
+from pypaimon.globalindex.offset_global_index_reader import OffsetGlobalIndexReader
 from pypaimon.utils.range import Range
 
 __all__ = [
     'GlobalIndexResult',
     'GlobalIndexReader',
     'FieldRef',
     'VectorSearch',
+    'FullTextSearch',
     'ScoredGlobalIndexResult',
     'DictBasedScoredIndexResult',
     'ScoreGetter',
     'GlobalIndexMeta',
     'GlobalIndexIOMeta',
     'GlobalIndexEvaluator',
     'GlobalIndexScanner',
+    'OffsetGlobalIndexReader',
     'Range',
 ]

paimon-python/pypaimon/globalindex/full_text_search.py

@@ -0,0 +1,53 @@
+################################################################################
+#  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.
+################################################################################
+
+"""FullTextSearch for performing full-text search on a text column."""
+
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass
+class FullTextSearch:
+    """
+    FullTextSearch to perform full-text search on a text column.
+
+    Attributes:
+        query_text: The query text to search
+        limit: Maximum number of results to return
+        field_name: Name of the text field to search
+    """
+
+    query_text: str
+    limit: int
+    field_name: str
+
+    def __post_init__(self):
+        if not self.query_text:
+            raise ValueError("Query text cannot be None or empty")
+        if self.limit <= 0:
+            raise ValueError(f"Limit must be positive, got: {self.limit}")
+        if not self.field_name:
+            raise ValueError("Field name cannot be null or empty")
+
+    def visit(self, visitor: 'GlobalIndexReader') -> Optional['ScoredGlobalIndexResult']:
+        """Visit the global index reader with this full-text search."""
+        return visitor.visit_full_text_search(self)
+
+    def __repr__(self) -> str:
+        return f"FullTextSearch(field={self.field_name}, query='{self.query_text}', limit={self.limit})"

paimon-python/pypaimon/globalindex/global_index_reader.py

@@ -42,6 +42,10 @@ def visit_vector_search(self, vector_search: 'VectorSearch') -> Optional['Global
         """Visit a vector search query."""
         raise NotImplementedError("Vector search not supported by this reader")
 
+    def visit_full_text_search(self, full_text_search: 'FullTextSearch') -> Optional['GlobalIndexResult']:
+        """Visit a full-text search query."""
+        raise NotImplementedError("Full-text search not supported by this reader")
+
     def visit_equal(self, field_ref: FieldRef, literal: object) -> Optional['GlobalIndexResult']:
         """Visit an equality predicate."""
         return None

paimon-python/pypaimon/globalindex/offset_global_index_reader.py

@@ -0,0 +1,102 @@
+################################################################################
+#  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.
+################################################################################
+
+"""A GlobalIndexReader that wraps another reader and applies an offset to all row IDs."""
+
+from typing import List, Optional
+
+from pypaimon.globalindex.global_index_reader import GlobalIndexReader, FieldRef
+from pypaimon.globalindex.global_index_result import GlobalIndexResult
+
+
+class OffsetGlobalIndexReader(GlobalIndexReader):
+    """
+    A GlobalIndexReader that wraps another reader and applies an offset
+    to all row IDs in the results.
+    """
+
+    def __init__(self, wrapped: GlobalIndexReader, offset: int, to: int):
+        self._wrapped = wrapped
+        self._offset = offset
+        self._to = to
+
+    def visit_vector_search(self, vector_search) -> Optional[GlobalIndexResult]:
+        result = self._wrapped.visit_vector_search(
+            vector_search.offset_range(self._offset, self._to))
+        if result is not None:
+            return result.offset(self._offset)
+        return None
+
+    def visit_full_text_search(self, full_text_search) -> Optional[GlobalIndexResult]:
+        result = self._wrapped.visit_full_text_search(full_text_search)
+        if result is not None:
+            return result.offset(self._offset)
+        return None
+
+    def visit_equal(self, field_ref: FieldRef, literal: object) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_equal(field_ref, literal))
+
+    def visit_not_equal(self, field_ref: FieldRef, literal: object) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_not_equal(field_ref, literal))
+
+    def visit_less_than(self, field_ref: FieldRef, literal: object) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_less_than(field_ref, literal))
+
+    def visit_less_or_equal(self, field_ref: FieldRef, literal: object) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_less_or_equal(field_ref, literal))
+
+    def visit_greater_than(self, field_ref: FieldRef, literal: object) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_greater_than(field_ref, literal))
+
+    def visit_greater_or_equal(self, field_ref: FieldRef, literal: object) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_greater_or_equal(field_ref, literal))
+
+    def visit_is_null(self, field_ref: FieldRef) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_is_null(field_ref))
+
+    def visit_is_not_null(self, field_ref: FieldRef) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_is_not_null(field_ref))
+
+    def visit_in(self, field_ref: FieldRef, literals: List[object]) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_in(field_ref, literals))
+
+    def visit_not_in(self, field_ref: FieldRef, literals: List[object]) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_not_in(field_ref, literals))
+
+    def visit_starts_with(self, field_ref: FieldRef, literal: object) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_starts_with(field_ref, literal))
+
+    def visit_ends_with(self, field_ref: FieldRef, literal: object) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_ends_with(field_ref, literal))
+
+    def visit_contains(self, field_ref: FieldRef, literal: object) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_contains(field_ref, literal))
+
+    def visit_like(self, field_ref: FieldRef, literal: object) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_like(field_ref, literal))
+
+    def visit_between(self, field_ref: FieldRef, min_v: object, max_v: object) -> Optional[GlobalIndexResult]:
+        return self._apply_offset(self._wrapped.visit_between(field_ref, min_v, max_v))
+
+    def _apply_offset(self, result: Optional[GlobalIndexResult]) -> Optional[GlobalIndexResult]:
+        if result is not None:
+            return result.offset(self._offset)
+        return None
+
+    def close(self) -> None:
+        self._wrapped.close()

paimon-python/pypaimon/globalindex/tantivy/__init__.py

@@ -0,0 +1,29 @@
+################################################################################
+#  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.
+################################################################################
+
+from pypaimon.globalindex.tantivy.tantivy_full_text_global_index_reader import (
+    TantivyFullTextGlobalIndexReader,
+)
+from pypaimon.globalindex.tantivy.tantivy_full_text_global_indexer import (
+    TANTIVY_FULLTEXT_IDENTIFIER,
+)
+
+__all__ = [
+    'TantivyFullTextGlobalIndexReader',
+    'TANTIVY_FULLTEXT_IDENTIFIER',
+]