[python] Introduce Full Text Search and Tantivy index in Python (#7560)

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
 
@@ -118,6 +132,17 @@ jobs:
             fi
           fi
           df -h
+
+      - name: Build and install tantivy-py from source
+        if: matrix.python-version != '3.6.15'
+        shell: bash
+        run: |
+          pip install maturin
+          git clone -b support_directory https://github.com/JingsongLi/tantivy-py.git /tmp/tantivy-py
+          cd /tmp/tantivy-py
+          maturin build --release
+          pip install target/wheels/tantivy-*.whl
+
       - name: Run lint-python.sh
         shell: bash
         run: |

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 >}}

docs/content/pypaimon/cli.md

@@ -362,6 +362,48 @@ Table 'mydb.old_name' renamed to 'mydb.new_name' successfully.
 
 **Note:** Both filesystem and REST catalogs support table rename. For filesystem catalogs, the rename is performed by renaming the underlying table directory.
 
+### Table Full-Text Search
+
+Perform full-text search on a Paimon table with a Tantivy full-text index and display matching rows.
+
+```shell
+paimon table full-text-search mydb.articles --column content --query "paimon lake"
+```
+
+**Options:**
+
+- `--column, -c`: Text column to search on - **Required**
+- `--query, -q`: Query text to search for - **Required**
+- `--limit, -l`: Maximum number of results to return (default: 10)
+- `--select, -s`: Select specific columns to display (comma-separated)
+- `--format, -f`: Output format: `table` (default) or `json`
+
+**Examples:**
+
+```shell
+# Basic full-text search
+paimon table full-text-search mydb.articles -c content -q "paimon lake"
+
+# Search with limit
+paimon table full-text-search mydb.articles -c content -q "streaming data" -l 20
+
+# Search with column projection
+paimon table full-text-search mydb.articles -c content -q "paimon" -s "id,title,content"
+
+# Output as JSON
+paimon table full-text-search mydb.articles -c content -q "paimon" -f json
+```
+
+Output:
+```
+ id                                            content
+  0  Apache Paimon is a streaming data lake platform
+  2  Paimon supports real-time data ingestion and...
+  4  Data lake platforms like Paimon handle large-...
+```
+
+**Note:** The table must have a Tantivy full-text index built on the target column. See [Global Index]({{< ref "append-table/global-index" >}}) for how to create full-text indexes.
+
 ### Table Drop
 
 Drop a table from the catalog. This will permanently delete the table and all its data.

paimon-python/dev/run_mixed_tests.sh

@@ -138,9 +138,6 @@ run_java_read_test() {
 
     cd "$PROJECT_ROOT"
 
-    PYTHON_VERSION=$(python -c "import sys; print(f'{sys.version_info.major}.{sys.version_info.minor}')" 2>/dev/null || echo "unknown")
-    echo "Detected Python version: $PYTHON_VERSION"
-
     # Run Java test for Parquet/Orc/Avro format in paimon-core
     echo "Running Maven test for JavaPyE2ETest.testReadPkTable (Java Read Parquet/Orc/Avro)..."
     echo "Note: Maven may download dependencies on first run, this may take a while..."
@@ -171,6 +168,7 @@ run_java_read_test() {
         return 1
     fi
 }
+
 run_pk_dv_test() {
     echo -e "${YELLOW}=== Step 5: Running Primary Key & Deletion Vector Test (testPKDeletionVectorWriteRead) ===${NC}"
 
@@ -244,6 +242,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 +275,12 @@ main() {
     local pk_dv_result=0
     local btree_index_result=0
     local compressed_text_result=0
+    local tantivy_fulltext_result=0
+
+    # Detect Python version
+    PYTHON_VERSION=$(python -c "import sys; print(f'{sys.version_info.major}.{sys.version_info.minor}')" 2>/dev/null || echo "unknown")
+    PYTHON_MINOR=$(python -c "import sys; print(sys.version_info.minor)" 2>/dev/null || echo "0")
+    echo "Detected Python version: $PYTHON_VERSION"
 
     echo -e "${YELLOW}Starting mixed language test execution...${NC}"
     echo ""
@@ -311,6 +339,18 @@ main() {
 
     echo ""
 
+    # Run Tantivy full-text index test (requires Python >= 3.10)
+    if [[ "$PYTHON_MINOR" -ge 10 ]]; then
+        if ! run_tantivy_fulltext_test; then
+            tantivy_fulltext_result=1
+        fi
+    else
+        echo -e "${YELLOW}⏭ Skipping Tantivy Full-Text Index Test (requires Python >= 3.10, current: $PYTHON_VERSION)${NC}"
+        tantivy_fulltext_result=0
+    fi
+
+    echo ""
+
     echo -e "${YELLOW}=== Test Results Summary ===${NC}"
 
     if [[ $java_write_result -eq 0 ]]; then
@@ -355,12 +395,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/cli/cli_table.py

@@ -147,6 +147,83 @@ def cmd_table_read(args):
         print(df.to_string(index=False))
 
 
+def cmd_table_full_text_search(args):
+    """
+    Execute the 'table full-text-search' command.
+
+    Performs full-text search on a Paimon table and displays matching rows.
+
+    Args:
+        args: Parsed command line arguments.
+    """
+    from pypaimon.cli.cli import load_catalog_config, create_catalog
+
+    config_path = args.config
+    config = load_catalog_config(config_path)
+    catalog = create_catalog(config)
+
+    table_identifier = args.table
+    parts = table_identifier.split('.')
+    if len(parts) != 2:
+        print(f"Error: Invalid table identifier '{table_identifier}'. "
+              f"Expected format: 'database.table'", file=sys.stderr)
+        sys.exit(1)
+
+    database_name, table_name = parts
+
+    try:
+        table = catalog.get_table(f"{database_name}.{table_name}")
+    except Exception as e:
+        print(f"Error: Failed to get table '{table_identifier}': {e}", file=sys.stderr)
+        sys.exit(1)
+
+    # Build full-text search
+    text_column = args.column
+    query_text = args.query
+    limit = args.limit
+
+    try:
+        builder = table.new_full_text_search_builder()
+        builder.with_text_column(text_column)
+        builder.with_query_text(query_text)
+        builder.with_limit(limit)
+        result = builder.execute_local()
+    except Exception as e:
+        print(f"Error: Full-text search failed: {e}", file=sys.stderr)
+        sys.exit(1)
+
+    if result.is_empty():
+        print("No matching rows found.")
+        return
+
+    # Read matching rows using global index result
+    read_builder = table.new_read_builder()
+
+    select_columns = args.select
+    if select_columns:
+        projection = [col.strip() for col in select_columns.split(',')]
+        available_fields = set(field.name for field in table.table_schema.fields)
+        invalid_columns = [col for col in projection if col not in available_fields]
+        if invalid_columns:
+            print(f"Error: Column(s) {invalid_columns} do not exist in table '{table_identifier}'.",
+                  file=sys.stderr)
+            sys.exit(1)
+        read_builder = read_builder.with_projection(projection)
+
+    scan = read_builder.new_scan().with_global_index_result(result)
+    plan = scan.plan()
+    splits = plan.splits()
+    read = read_builder.new_read()
+    df = read.to_pandas(splits)
+
+    output_format = getattr(args, 'format', 'table')
+    if output_format == 'json':
+        import json
+        print(json.dumps(df.to_dict(orient='records'), ensure_ascii=False))
+    else:
+        print(df.to_string(index=False))
+
+
 def cmd_table_get(args):
     """
     Execute the 'table get' command.
@@ -773,6 +850,43 @@ def add_table_subcommands(table_parser):
 
     # table rename command
     rename_parser = table_subparsers.add_parser('rename', help='Rename a table')
+
+    # table full-text-search command
+    fts_parser = table_subparsers.add_parser('full-text-search', help='Full-text search on a table')
+    fts_parser.add_argument(
+        'table',
+        help='Table identifier in format: database.table'
+    )
+    fts_parser.add_argument(
+        '--column', '-c',
+        required=True,
+        help='Text column to search on'
+    )
+    fts_parser.add_argument(
+        '--query', '-q',
+        required=True,
+        help='Query text to search for'
+    )
+    fts_parser.add_argument(
+        '--limit', '-l',
+        type=int,
+        default=10,
+        help='Maximum number of results to return (default: 10)'
+    )
+    fts_parser.add_argument(
+        '--select', '-s',
+        type=str,
+        default=None,
+        help='Select specific columns to display (comma-separated, e.g., "id,name,content")'
+    )
+    fts_parser.add_argument(
+        '--format', '-f',
+        type=str,
+        choices=['table', 'json'],
+        default='table',
+        help='Output format: table (default) or json'
+    )
+    fts_parser.set_defaults(func=cmd_table_full_text_search)
     rename_parser.add_argument(
         'table',
         help='Source table identifier in format: database.table'

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',
 ]