升级 lance-ray 至 0.5.0 并固定 pylance 9.0.0

AGENTS.md

@@ -28,7 +28,7 @@ tests/                 # pytest；按 audio/image/storage/workflow 分目录
 ## 硬性架构约定
 
 1. **媒体隔离**：audio 和 image 各自拥有独立的 package 和 workflow 入口，**不要**把共享 Stage 参数化成"通用管线"。可共享的只有 `storage/` 层 infra 和 `config.py` 的辅助函数。
-2. **数据 API 优先级**：所有数据操作优先用 **Daft**；Daft 做不到再用 **lance-ray**；**pylance 直接调用是最后手段**（目前锁在 <8，blob compaction 被禁用）。
+2. **数据 API 优先级**：所有数据操作优先用 **Daft**；Daft 做不到再用 **lance-ray**；**pylance 直接调用是最后手段**。当前固定为 lance-ray 0.5.0 与 pylance 9.0.0；Blob v2 表删除后由 lance-ray 执行分布式 compaction，涉及该路径的改动必须覆盖本地和对象存储回归验证。
 3. **行不丢弃**：analyze 阶段 manifest 里每个条目对应输出一行，失败用 `status` 列标记（`download_failed` / `decode_failed` / ...），分数结论置 null。合规场景必须能区分"内容有问题"和"根本没处理"。
 4. **配置只走环境变量**，经 `config.py` 的 `env_int`/`env_bool`/`env_choice` 等辅助函数读取。注意：Daft 原生只读少数 `DAFT_*` 环境变量，新增 Daft 执行参数必须在 `storage/io.py::configure_daft_runner()` 里显式传给 `daft.set_execution_config()`，否则是死配置；S3 参数同理要传进 `daft_io_config()` 的 `S3Config`。
 5. **写出受控**：`write_lance` 必须传 `LANCE_MAX_ROWS_PER_FILE` / `LANCE_MAX_BYTES_PER_FILE`；JSONL 写出前用 `coalesce_for_write()` 收敛分区，避免小文件。
@@ -43,7 +43,7 @@ USE_RAY=1 ...                      # 切换 Ray runner（默认 native）
 ```
 
 - 本地依赖 MinIO（默认 `http://127.0.0.1:9000`，minioadmin/minioadmin），配置见 `.env` / README Setup 小节。
-- venv 里的 lance-ray 是本地 dev checkout（版本号 0.4.2 与 PyPI 同名但内容不同），不要"顺手升级"它。
+- lance-ray 与 pylance 必须按 `pyproject.toml` 和 `uv.lock` 固定的配套版本安装，不能只升级其中一个。
 
 ## 提交与 PR
 

README.md

@@ -74,9 +74,9 @@ Manifest (parquet / jsonl / csv)
 | 引擎           | 用于                                                                        | 原因                                                        |
 |----------------|-----------------------------------------------------------------------------|-------------------------------------------------------------|
 | **Daft**       | manifest 读取、S3 下载、ASR/LLM 流水线、Lance 写入（Stage 1 & 2）、标量与 ANN 查询 | 主引擎；API 稳定                                            |
-| **lance_ray**  | IVF_PQ 向量索引创建、`compact_files`                                        | Lance 表管理的首选；可用分布式 Ray worker                   |
+| **lance_ray**  | IVF_PQ 向量索引创建、`compact_files`                                        | 分布式索引与删除后的表维护                                   |
 | **pylance**    | ZONEMAP 标量索引、行删除、`cleanup_old_versions`                            | ZONEMAP：lance_ray 依赖未发布代码；delete：只有这一个 API    |
-| **daft_lance** | lance_ray 不可用时 `compact_files` 的兜底                                   | 不用于建索引；Daft 优先原则只适用于数据处理                 |
+| **daft_lance** | `read_lance`、`write_lance` 等 Daft 数据操作                                | 不用于建索引；Daft 优先原则只适用于数据处理                 |
 
 ## 环境准备
 
@@ -410,8 +410,8 @@ python -m multimodal_toolkit.workflow.manage \
 |------|------|------|
 | Daft | 0.7.15 | 主执行引擎 |
 | daft-lance | 0.4.0 | `read_lance`、`write_lance`、`take_blobs`、`create_scalar_index`、`compact_files` |
-| pylance | 7.0.0 | Lance dataset、blob v2、ANN scanner、delete、cleanup |
-| lance-ray | 0.4.2 | 向量索引创建；写回路径暂缓 |
+| pylance | 9.0.0 | Lance dataset、blob v2、ANN scanner、delete、cleanup |
+| lance-ray | 0.5.0 | 分布式向量索引创建与分布式 compaction API |
 | Ray | 2.55.1 | 由 lance-ray 引入；除非 `USE_RAY=1`，否则 Daft 走 native runner |
 
 Daft 默认 runner 是 `native`（本地多线程）。设置 `USE_RAY=1` 可把 Daft 相关步骤切到 Ray。
@@ -455,8 +455,11 @@ confidence 与 reason 字段在本地行中为 null。VLM 调用失败时仍保
 **本地 Lance URI 已做端到端验证。**
 S3 上的 Lance 表读写由底层库覆盖，但在本 POC 中应作为一个独立的验证项对待。
 
-**blob v2 资产表的 compaction 暂时禁用。**
-当前 `lance-ray 0.4.2` 的向量索引创建依赖 pylance 7.x，而 pylance 8.0.0 虽然修复了 blob v2
-compaction，却改动了 `lance-ray 0.4.2` 仍在调用的分布式索引 commit API。项目暂时锁定
-`pylance<8.0.0`，Stage 5 在删除行之后跳过 compaction。等升级到 pylance 8.x 并配套兼容的
-lance-ray 版本后再重新开启 blob v2 compaction。
+**分布式向量索引失败会直接报错。**
+Stage 3 固定使用 lance-ray 0.5.0 的分布式提交链路，不再回退到 pylance 单机建索引。这样调用方
+能明确获知 Ray、对象存储凭据或索引提交失败，避免大表在驱动进程静默退化为单机执行。
+
+**删除后会执行 Blob v2 compaction。**
+Stage 5 先提交删除，再使用 lance-ray 0.5.0 的默认 compaction 选项压实文件和删除向量，最后清理旧版本。
+0.4.x 中为规避 `compaction_options=None` 缺陷而传入空字典的 workaround 已移除。对象存储部署仍应在
+发布前验证 worker 凭据、共享 Lance URI 和 Blob v2 读取。

docs/daft-tuning.md

@@ -112,7 +112,8 @@ CPU/网络 在下载下一批图片
 | `LANCE_MAX_BYTES_PER_FILE` | `512MB` | 单个 Lance 数据文件最大字节数。 |
 | `DAFT_JSON_TARGET_FILESIZE` | `128MB` | JSONL 单文件目标大小。 |
 
-为什么要管：碎文件会拖慢后续所有读取，而我们的 pylance 7 禁用了 blob compaction，写坏了没有事后补救，只能在写入侧控制。默认值一般不用动。
+为什么要管：碎文件会拖慢后续所有读取。Stage 5 会在删除后压实 Blob v2 表，但写入侧控制仍能
+降低 compaction 的资源消耗和文件膨胀；对象存储环境还应验证 worker 的共享 URI 与凭据。默认值一般不用动。
 
 ### 3.7 观测（长任务建议必开）
 

multimodal_toolkit/workflow/index.py

@@ -42,20 +42,9 @@ def build_embedding_index(
     )
     if index_type in ("IVF_PQ", "IVF_HNSW_PQ"):
         kwargs["num_sub_vectors"] = num_sub_vectors
-    try:
-        lance_ray.create_index(lance_uri, **kwargs)
-    except Exception as exc:
-        print(f"[warn] lance_ray index build failed; falling back to pylance: {exc}")
-        fallback_kwargs: dict = dict(
-            index_type=index_type,
-            replace=True,
-            num_partitions=num_partitions,
-            sample_rate=sample_rate,
-            storage_options=storage_options,
-        )
-        if index_type in ("IVF_PQ", "IVF_HNSW_PQ"):
-            fallback_kwargs["num_sub_vectors"] = num_sub_vectors
-        ds.create_index(column, **fallback_kwargs)
+    # 向量索引必须由 lance-ray 的分布式提交链路完成。不能静默回退到 pylance
+    # 单机建索引，否则大表会在驱动进程失去并行能力且调用方无法感知执行模式变化。
+    lance_ray.create_index(lance_uri, **kwargs)
     print(f"[ok] built {index_type} index on {column} ({num_partitions} partitions): {lance_uri}")
 
 

multimodal_toolkit/workflow/manage.py

@@ -1,21 +1,22 @@
-"""Stage 5: manage the lance asset table — delete rows by ingest_time range.
+"""Stage 5：按 ingest_time 范围删除 Lance 资产表中的行。
 
-API priority (lance_ray preferred for table management):
-  delete        → pylance ds.delete()              (no Daft/lance-ray equivalent)
-  compact       → disabled for blob v2 tables until pylance 8 is supported
-  cleanup       → pylance ds.cleanup_old_versions() (no alternative)
+表管理仍遵循项目的数据 API 优先级：
+  delete        → pylance ds.delete()               （Daft/lance-ray 没有等价 API）
+  compact       → lance_ray.compact_files()        （删除后回收小文件与删除向量）
+  cleanup       → pylance ds.cleanup_old_versions() （没有替代 API）
 
-  --before DATE   delete rows where ingest_time < DATE
-  --after  DATE   delete rows where ingest_time > DATE
+  --before DATE   删除 ingest_time < DATE 的行
+  --after  DATE   删除 ingest_time > DATE 的行
 
-DATE format: ISO 8601, e.g. 2025-01-01 or 2025-01-01T00:00:00
-At least one bound must be provided; both can be combined for a date range.
+DATE 使用 ISO 8601 格式，例如 2025-01-01 或 2025-01-01T00:00:00。
+至少传一个边界；两个边界可以组合成日期范围。
 """
 from __future__ import annotations
 
 import argparse
 
 import lance
+import lance_ray
 
 from ..storage.io import lance_storage_options
 
@@ -35,16 +36,24 @@ def delete_by_date(
         clauses.append(f"ingest_time > timestamp '{after}'")
     filter_str = " AND ".join(clauses)
 
-    # delete: pylance only (no Daft/lance-ray equivalent)
+    # 删除目前只有 pylance 提供 API；这里保留直接调用，不能为了统一入口而绕过
+    # Lance 自身的事务提交语义。
     ds = lance.dataset(lance_uri, storage_options=lance_storage_options(lance_uri))
     ds.delete(filter_str)
 
     print(f"[ok] deleted rows where: {filter_str}")
 
-    # Blob v2 tables cannot be compacted safely on pylance 7.x
-    # (lance-format/lance#7071). Keep compaction disabled until the project
-    # can move to pylance 8.x together with a compatible lance-ray release.
-    print("[warn] compaction skipped; enable after upgrading to pylance 8.x")
+    # lance-ray 0.5.0 在未传 compaction_options 时会构造默认选项；不再需要 0.4.x
+    # 为绕过 None 传参缺陷而显式传入空字典。压实必须在删除提交之后执行，才能把删除
+    # 向量和小文件合并成新的 fragment。
+    metrics = lance_ray.compact_files(
+        lance_uri,
+        storage_options=lance_storage_options(lance_uri),
+    )
+    if metrics is None:
+        print(f"[ok] no compaction needed: {lance_uri}")
+    else:
+        print(f"[ok] compacted: {lance_uri}")
     ds.cleanup_old_versions()
 
 

pyproject.toml

@@ -10,7 +10,7 @@ dependencies = [
   "getdaft",
   "httpx",
   "insightface",
-  "lance-ray",
+  "lance-ray==0.5.0",
   "modelscope",
   "numpy",
   "onnxruntime",
@@ -19,7 +19,7 @@ dependencies = [
   "pillow",
   "psutil",
   "pyarrow",
-  "pylance<8.0.0",
+  "pylance==9.0.0",
   "python-dotenv",
   "ray",
   "soundfile",
@@ -28,6 +28,11 @@ dependencies = [
   "transformers",
 ]
 
+[dependency-groups]
+dev = [
+  "pytest>=9.1.1",
+]
+
 [tool.pytest.ini_options]
 markers = [
   "ray: hard dependency on a local Ray cluster; excluded by default, run with `pytest -m ray`",

tests/workflow/test_index_manage.py

@@ -1,9 +1,8 @@
-"""Tests for workflow/index.py and workflow/manage.py — local lance tables.
+"""在本地 Lance 表上验证 workflow/index.py 和 workflow/manage.py。
 
-build_embedding_index is marked `ray` and excluded from the default run (see
-pyproject) — run it explicitly with `pytest -m ray`. delete_by_date does not
-compact while the project is pinned to pylance 7.x, so those tests stay in the
-default suite.
+build_embedding_index 的真实 Ray 路径不属于默认测试；需要时用 `pytest -m ray`
+显式执行。delete_by_date 会执行分布式 compaction，因此删除测试使用共享的 local_ray
+fixture，并同时覆盖普通表和 Blob v2 表。
 """
 from __future__ import annotations
 
@@ -50,6 +49,37 @@ def lance_uri() -> str:
     return uri
 
 
+@pytest.fixture()
+def lance_uri_blob() -> str:
+    """按资产表的写入方式创建多 fragment 的 Blob v2 表。
+
+    每批 100 行分开 append，确保删除后 compaction 至少有多个 fragment 可以合并；
+    这能覆盖 pylance 9 与 lance-ray 0.5 的真实 Blob v2 表维护链路。
+    """
+    import daft
+    from daft import col
+
+    tmp = tempfile.mkdtemp()
+    uri = str(pathlib.Path(tmp) / "table_blob.lance")
+    for batch in range(3):
+        times = [datetime(2024, 1, batch + 1, tzinfo=timezone.utc)] * 100
+        df = daft.from_pydict(
+            {
+                "doc_id": [f"blob_{batch}_{row:04d}" for row in range(100)],
+                "ingest_time": times,
+                "blob": [b"x" * 100] * 100,
+            }
+        ).with_column(
+            "ingest_time", col("ingest_time").cast(daft.DataType.timestamp("us", "UTC"))
+        )
+        df.write_lance(
+            uri,
+            mode="create" if batch == 0 else "append",
+            blob_columns=["blob"],
+        )
+    return uri
+
+
 @pytest.fixture()
 def lance_uri_no_embedding() -> str:
     tmp = tempfile.mkdtemp()
@@ -92,41 +122,14 @@ def fake_create_index(uri, **kwargs):
     ]
 
 
-def test_build_embedding_index_falls_back_to_pylance(monkeypatch):
-    calls = []
-
-    class FakeSchema:
-        names = ["audio_embedding"]
-
-    class FakeDataset:
-        schema = FakeSchema()
-
-        def create_index(self, column, **kwargs):
-            calls.append((column, kwargs))
-
+def test_build_embedding_index_propagates_lance_ray_failure(monkeypatch, lance_uri):
     def fake_create_index(uri, **kwargs):
-        raise RuntimeError("ray worker failed")
+        raise RuntimeError("distributed index failed")
 
-    monkeypatch.setattr(
-        "multimodal_toolkit.workflow.index.lance.dataset",
-        lambda *args, **kwargs: FakeDataset(),
-    )
     monkeypatch.setattr("multimodal_toolkit.workflow.index.lance_ray.create_index", fake_create_index)
 
-    build_embedding_index("table.lance", num_partitions=1, sample_rate=2, index_type="IVF_FLAT")
-
-    assert calls == [
-        (
-            "audio_embedding",
-            {
-                "index_type": "IVF_FLAT",
-                "replace": True,
-                "num_partitions": 1,
-                "sample_rate": 2,
-                "storage_options": None,
-            },
-        )
-    ]
+    with pytest.raises(RuntimeError, match="distributed index failed"):
+        build_embedding_index(lance_uri, num_partitions=1, sample_rate=2, index_type="IVF_FLAT")
 
 
 def test_build_embedding_index_missing_column(lance_uri_no_embedding):
@@ -148,18 +151,35 @@ def test_delete_requires_a_bound(lance_uri):
         delete_by_date(lance_uri)
 
 
-def test_delete_before(lance_uri):
+def test_delete_before(lance_uri, local_ray):
     delete_by_date(lance_uri, before="2024-03-01")
     assert lance.dataset(lance_uri).count_rows() == 200  # 2024-01-01 rows gone
 
 
-def test_delete_after(lance_uri):
+def test_delete_after(lance_uri, local_ray):
     delete_by_date(lance_uri, after="2024-09-01")
     assert lance.dataset(lance_uri).count_rows() == 200  # 2024-12-01 rows gone
 
 
-def test_delete_window(lance_uri):
+def test_delete_window(lance_uri, local_ray):
     # Outside 2024-03-01 .. 2024-09-01 survives: keeps Jan and Dec rows.
     delete_by_date(lance_uri, before="2024-09-01", after="2024-03-01")
     remaining = lance.dataset(lance_uri).count_rows()
     assert remaining == 200
+
+
+def test_delete_and_compact_blob_v2_table(lance_uri_blob, local_ray):
+    """删除后必须能压实 Blob v2 表，并保持剩余 blob 可读。"""
+    from multimodal_toolkit.storage.blob import validate_blob_v2
+
+    # Daft 的执行配置会影响每次 append 实际写出的 fragment 数；这里只约束
+    # compaction 的核心语义：压实前确实有多个 fragment，压实后数量必须减少。
+    fragments_before = len(lance.dataset(lance_uri_blob).get_fragments())
+    assert fragments_before > 1
+    delete_by_date(lance_uri_blob, before="2024-01-02")
+
+    dataset = lance.dataset(lance_uri_blob)
+    assert dataset.count_rows() == 200
+    assert len(dataset.get_fragments()) < fragments_before
+    assert len(dataset.take_blobs("blob", indices=[0])[0].read()) == 100
+    validate_blob_v2(lance_uri_blob, "blob")