leann integration (#10107)

GitOrigin-RevId: 6da4c020ced767ba7bddc968aa7217907bba0210

Author: zxqfd555

CHANGELOG.md

@@ -11,6 +11,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - `pw.io.milvus.write` connector, which writes a Pathway table to a Milvus collection. Row additions are sent as upserts and row deletions are sent as deletes keyed on the configured primary key column. Requires a Pathway Scale license.
 - `pathway spawn` now supports the `--addresses` and `--process-id` flags for multi-machine deployments. Pass a comma-separated list of `host:port` addresses for all processes and the index of the local process; Pathway will connect the cluster over TCP without requiring all processes to run on the same machine.
 - `pw.xpacks.llm.parsers.AudioParser`, audio transcription parser based on OpenAI Whisper API. Accepts raw audio bytes and returns transcribed text, following the same interface as other Pathway document parsers.
+- `pw.io.leann.write` connector for writing Pathway tables to LEANN vector indices. LEANN uses graph-based selective recomputation to achieve 97% storage reduction compared to traditional vector databases.
 
 ## [0.30.0] - 2026-03-24
 

integration_tests/db_connectors/test_leann.py

@@ -0,0 +1,171 @@
+# Copyright © 2026 Pathway
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from leann import LeannSearcher
+
+import pathway as pw
+
+
+class DocumentSchema(pw.Schema):
+    text: str
+
+
+class DocumentWithMetadataSchema(pw.Schema):
+    text: str
+    title: str
+    category: str
+
+
+def test_basic_write(tmp_path: Path):
+    """Test basic writing to a LEANN index."""
+    index_path = tmp_path / "test_basic.leann"
+
+    table = pw.debug.table_from_rows(
+        schema=DocumentSchema,
+        rows=[
+            ("LEANN is a vector database",),
+            ("Pathway enables streaming",),
+            ("Vector search is powerful",),
+        ],
+    )
+
+    pw.io.leann.write(
+        table,
+        index_path=str(index_path),
+        text_column=table.text,
+        backend_name="hnsw",
+    )
+
+    pw.run(monitoring_level=pw.MonitoringLevel.NONE)
+
+    # Verify the index was created (LEANN creates multiple files with base name)
+    # Check for the metadata file which is always created
+    meta_file = Path(str(index_path) + ".meta.json")
+    assert meta_file.exists(), f"Expected {meta_file} to exist"
+
+    searcher = LeannSearcher(str(index_path))
+    results = searcher.search("vector database", top_k=1)
+    assert len(results) > 0
+
+
+def test_write_with_metadata_columns(tmp_path: Path):
+    """Test writing with metadata columns."""
+    index_path = tmp_path / "test_metadata.leann"
+
+    table = pw.debug.table_from_rows(
+        schema=DocumentWithMetadataSchema,
+        rows=[
+            ("Introduction to LEANN", "LEANN Guide", "tutorial"),
+            ("Advanced vector search", "Search Tips", "advanced"),
+        ],
+    )
+
+    pw.io.leann.write(
+        table,
+        index_path=str(index_path),
+        text_column=table.text,
+        metadata_columns=[table.title, table.category],
+        backend_name="hnsw",
+    )
+
+    pw.run(monitoring_level=pw.MonitoringLevel.NONE)
+    assert Path(str(index_path) + ".meta.json").exists()
+
+
+def test_write_with_custom_text_column(tmp_path: Path):
+    """Test writing with a custom text column name."""
+    index_path = tmp_path / "test_custom_column.leann"
+
+    class ContentSchema(pw.Schema):
+        content: str
+        id: int
+
+    table = pw.debug.table_from_rows(
+        schema=ContentSchema,
+        rows=[
+            ("First document content", 1),
+            ("Second document content", 2),
+        ],
+    )
+
+    pw.io.leann.write(
+        table,
+        index_path=str(index_path),
+        text_column=table.content,
+        backend_name="hnsw",
+    )
+    pw.run(monitoring_level=pw.MonitoringLevel.NONE)
+    assert Path(str(index_path) + ".meta.json").exists()
+
+
+def test_write_creates_parent_directories(tmp_path: Path):
+    """Test that write() creates parent directories if they don't exist."""
+    index_path = tmp_path / "nested" / "dir" / "test.leann"
+
+    table = pw.debug.table_from_rows(
+        schema=DocumentSchema,
+        rows=[("Test document",)],
+    )
+
+    pw.io.leann.write(
+        table,
+        index_path=str(index_path),
+        text_column=table.text,
+        backend_name="hnsw",
+    )
+
+    pw.run(monitoring_level=pw.MonitoringLevel.NONE)
+    assert Path(str(index_path) + ".meta.json").exists()
+
+
+def test_empty_table(tmp_path: Path):
+    """Test behavior with an empty table."""
+    index_path = tmp_path / "test_empty.leann"
+
+    table = pw.debug.table_from_rows(
+        schema=DocumentSchema,
+        rows=[],
+    )
+
+    pw.io.leann.write(
+        table,
+        index_path=str(index_path),
+        text_column=table.text,
+        backend_name="hnsw",
+    )
+
+    pw.run(monitoring_level=pw.MonitoringLevel.NONE)
+
+    # Index should not be created for empty table
+    assert not Path(str(index_path) + ".meta.json").exists()
+
+
+def test_write_with_diskann_backend(tmp_path: Path):
+    pass
+    # TODO: uncomment if diskann needs to be tested. Currently we keep it this way, since
+    # diskann installation requires additional steps in Dockerfile and will lead to the
+    # increase in the time required to build the base integration tests image.
+    #
+    #    """Test writing with DiskANN backend."""
+    #    index_path = tmp_path / "test_diskann.leann"
+    #
+    #    table = pw.debug.table_from_rows(
+    #        schema=DocumentSchema,
+    #        rows=[
+    #            ("Document one for DiskANN test",),
+    #            ("Document two for DiskANN test",),
+    #        ],
+    #    )
+    #
+    #    pw.io.leann.write(
+    #        table,
+    #        index_path=str(index_path),
+    #        text_column=table.text,
+    #        backend_name="diskann",
+    #    )
+    #
+    #    pw.run(monitoring_level=pw.MonitoringLevel.NONE)
+    #    assert Path(str(index_path) + ".meta.json").exists()

integration_tests/license/test_license.py

@@ -46,6 +46,7 @@
     "xpack-llm-mcp",
     "worker-count-scaling",
     "multiple-machines",
+    "leann",
 ]
 
 
@@ -125,6 +126,7 @@ def test_license_default_policy(caplog):
     _check_entitlements("mongodb-oplog-reader")
     _check_entitlements("multiple-machines")
     _check_entitlements("mssql")
+    _check_entitlements("leann")
 
     run_all()
     assert "Telemetry enabled" in caplog.text

python/pathway/io/__init__.py

@@ -15,6 +15,7 @@
     jsonlines,
     kafka,
     kinesis,
+    leann,
     logstash,
     milvus,
     minio,
@@ -88,4 +89,5 @@
     "mssql",
     "mysql",
     "SynchronizedColumn",
+    "leann",
 ]