From c5aaa86f96b053c7deb2d6bbe8c0ee1ab195a94c Mon Sep 17 00:00:00 2001 From: mgrani Date: Wed, 15 Jul 2026 18:14:20 +0200 Subject: [PATCH] fix: query slice produced header-only index.ciff.gz Each source dataset's CIFF was filtered directly onto the same target path, so the last-processed source (often containing none of the sliced documents) clobbered the result with an empty index. Source CIFFs are now filtered into temp files and the non-empty results merged via ciff_toolkit merge_ciff_files, with a console warning when the sliced index ends up empty. Also ports core/ciff.py to the ciff-toolkit 0.2.x writer API and bumps the pin to >=0.2.2 (0.1.x-style write_message/raw write no longer exist, and DEFAULT_CHUNK_SIZE was removed from ciff_toolkit.read); drop_parallel now returns (kept_docs, postings_lists). Fixes an UnboundLocalError when --search-terms is combined with CIFF slicing, normalizes parquet ids to str, unskips the merge-after-drop test (upstream assert fixed in 0.2.2), and adds regression tests including an index doc-count assertion in the slice integration test. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_0196bRzaYxjC3Mfr14UXUkBA --- owilix/core/ciff.py | 130 +++++++-------- owilix/core/tasks/query.py | 144 ++++++++++------- pyproject.toml | 2 +- .../cli/test_query_slice_integration.py | 18 +++ tests/owilix/core/test_drop_parallel.py | 42 ++++- tests/owilix/core/test_slice_ciff.py | 150 ++++++++++++++++++ uv.lock | 8 +- 7 files changed, 356 insertions(+), 138 deletions(-) create mode 100644 tests/owilix/core/test_slice_ciff.py diff --git a/owilix/core/ciff.py b/owilix/core/ciff.py index 4dd8d7c..f6065b6 100644 --- a/owilix/core/ciff.py +++ b/owilix/core/ciff.py @@ -1,6 +1,5 @@ from __future__ import annotations -import tempfile from pathlib import Path from typing import Callable, Optional from concurrent.futures import ThreadPoolExecutor @@ -15,7 +14,7 @@ def drop_parallel( progress_callback: Optional[Callable[[str], None]] = None, keep: bool = True, threads: int = 4, -): +) -> tuple[int, int]: """ Keep (default) or drop documents with specified collection_docids from a CIFF file. @@ -26,13 +25,17 @@ def drop_parallel( Write (canonical): - Output order: HEADER -> POSTINGS -> DOCUMENTS - Header is fully recomputed for the filtered file (num_*, total_*, average_doclength). + + Requires ciff-toolkit >= 0.2.2 (CiffWriter assembles canonical section order on close). + + :return: (kept_doc_count, surviving_postings_list_count) """ def report(msg: str): if progress_callback: progress_callback(f"{msg} ({new_ciff_file})") from ciff_toolkit.ciff_pb2 import Posting, DocRecord - from ciff_toolkit.read import CiffReader, DEFAULT_CHUNK_SIZE + from ciff_toolkit.read import CiffReader from ciff_toolkit.write import CiffWriter report("Phase 1: Build doc mapping (header->postings->documents)") @@ -64,8 +67,10 @@ def drop_parallel( ) with CiffWriter(new_ciff_file) as final_writer: final_writer.write_header(final_header) + final_writer.write_postings_lists([]) + final_writer.write_documents([]) report("Complete: 0 docs, 0 postings lists") - return + return 0, 0 # Preallocate mapping helpers drop_mask = np.zeros(num_docs_total, dtype=bool) @@ -167,58 +172,47 @@ def drop_parallel( pl.cf = int(np.sum(tfs_sorted)) return pl - report("Phase 2: Stream/process postings (parallel) → tmp_postings; docs → tmp_docs") - - # ---- PASS 2: Write to two temp files to preserve canonical order later ---- - with tempfile.TemporaryFile() as tmp_postings_file, tempfile.TemporaryFile() as tmp_docs_file: - tmp_post_writer = CiffWriter(tmp_postings_file) - tmp_docs_writer = CiffWriter(tmp_docs_file) - - # Write DOCUMENTS block to its own temp now (they fit in memory) - tmp_docs_writer.write_documents(kept_docs) + report("Phase 2: Stream/process postings (parallel) → output") + + # ---- PASS 2: Parallel postings pipeline, yielded in original order ---- + postings_queue: "queue.Queue[tuple[object,int]]" = queue.Queue(maxsize=max(1, threads * 2)) + results_queue: "queue.Queue[tuple[object,int]]" = queue.Queue() + + def producer(): + from ciff_toolkit.read import CiffReader # local import inside thread + with CiffReader(ciff_file) as r2: + r2.read_header() + for idx, pl in enumerate(r2.read_postings_lists()): + postings_queue.put((pl, idx)) + for _ in range(threads): + postings_queue.put((None, -1)) + + def worker(): + while True: + item, idx = postings_queue.get() + if item is None: + results_queue.put((None, -1)) + postings_queue.task_done() + break + processed = process_postings_list_vectorized(item) + results_queue.put((processed, idx)) + postings_queue.task_done() - # Parallel postings pipeline → write to tmp_postings_file - postings_queue: "queue.Queue[tuple[object,int]]" = queue.Queue(maxsize=max(1, threads * 2)) - results_queue: "queue.Queue[tuple[object,int]]" = queue.Queue() + filtered_postings_count = 0 + total_terms_in_collection = 0 # sum of cf over surviving lists - def producer(): - from ciff_toolkit.read import CiffReader # local import inside thread - with CiffReader(ciff_file) as r2: - r2.read_header() - for idx, pl in enumerate(r2.read_postings_lists()): - postings_queue.put((pl, idx)) - for _ in range(threads): - postings_queue.put((None, -1)) - - def worker(): - while True: - item, idx = postings_queue.get() - if item is None: - results_queue.put((None, -1)) - postings_queue.task_done() - break - processed = process_postings_list_vectorized(item) - results_queue.put((processed, idx)) - postings_queue.task_done() + def surviving_postings_lists(): + nonlocal filtered_postings_count, total_terms_in_collection producer_thread = threading.Thread(target=producer, daemon=True) producer_thread.start() - filtered_postings_count = 0 - total_terms_in_collection = 0 # sum of cf over surviving lists - pending: dict[int, object] = {} dropped_indices: set[int] = set() next_expected_idx = 0 completed_workers = 0 processed_count = 0 - def write_pl(pl_obj): - nonlocal filtered_postings_count, total_terms_in_collection - tmp_post_writer.write_message(pl_obj) - filtered_postings_count += 1 - total_terms_in_collection += int(pl_obj.cf) - with ThreadPoolExecutor(max_workers=threads) as pool: for _ in range(threads): pool.submit(worker) @@ -234,38 +228,41 @@ def drop_parallel( report(f"Postings processed: {processed_count}") if result is None: - if idx == next_expected_idx: - next_expected_idx += 1 - while next_expected_idx in pending: - write_pl(pending.pop(next_expected_idx)) - next_expected_idx += 1 - while next_expected_idx in dropped_indices: - dropped_indices.remove(next_expected_idx) - next_expected_idx += 1 - else: - dropped_indices.add(idx) - continue + dropped_indices.add(idx) + else: + pending[idx] = result - pending[idx] = result while True: if next_expected_idx in dropped_indices: dropped_indices.remove(next_expected_idx) next_expected_idx += 1 continue if next_expected_idx in pending: - write_pl(pending.pop(next_expected_idx)) + pl_obj = pending.pop(next_expected_idx) next_expected_idx += 1 + filtered_postings_count += 1 + total_terms_in_collection += int(pl_obj.cf) + yield pl_obj continue break producer_thread.join() while next_expected_idx in pending: - write_pl(pending.pop(next_expected_idx)) + pl_obj = pending.pop(next_expected_idx) next_expected_idx += 1 + filtered_postings_count += 1 + total_terms_in_collection += int(pl_obj.cf) + yield pl_obj + + # ---- Phase 3: Write output; CiffWriter buffers out-of-order sections and + # assembles the canonical HEADER -> POSTINGS -> DOCUMENTS order on close. + # The header is written last because its stats depend on the postings pass. + with CiffWriter(new_ciff_file) as final_writer: + final_writer.write_postings_lists(surviving_postings_lists()) + final_writer.write_documents(kept_docs) report(f"Phase 2 done: wrote {filtered_postings_count} postings lists") - # ---- Phase 3: Header & finalize in canonical order ---- final_header = original_header.__class__() final_header.CopyFrom(original_header) @@ -284,18 +281,7 @@ def drop_parallel( else: final_header.description = addendum - # Write final file in canonical order: HEADER -> POSTINGS -> DOCUMENTS - with CiffWriter(new_ciff_file) as final_writer: - final_writer.write_header(final_header) - - # Copy POSTINGS block first - tmp_postings_file.seek(0) - while chunk := tmp_postings_file.read(DEFAULT_CHUNK_SIZE): - final_writer.write(chunk) - - # Then copy DOCUMENTS block - tmp_docs_file.seek(0) - while chunk := tmp_docs_file.read(DEFAULT_CHUNK_SIZE): - final_writer.write(chunk) + final_writer.write_header(final_header) report(f"Complete: {kept_count} docs, {filtered_postings_count} postings lists") + return kept_count, filtered_postings_count diff --git a/owilix/core/tasks/query.py b/owilix/core/tasks/query.py index 0eaa053..5445e4b 100644 --- a/owilix/core/tasks/query.py +++ b/owilix/core/tasks/query.py @@ -436,6 +436,8 @@ def slice_query( if not collection_name or collection_name == "main": return CommandResult.error("Collection name must be specified and cannot be 'main'") + md_file_pattern = paths + "*.parquet" + if search_terms: from owilix.core.tasks.search import resolve_search_to_files @@ -455,8 +457,6 @@ def slice_query( search_where = f"id IN ({id_list})" where = f"({search_where}) AND ({where})" if where else search_where else: - md_file_pattern = paths + "*.parquet" - try: all_ds_md_files = get_all_files( owi, md_file_pattern, local_specifier, remote_specifier, @@ -539,25 +539,14 @@ def slice_query( owi, paths + "*ciff*.gz", local_specifier, remote_specifier, console=console, print_it=False ) - # Need to filter out empty entries if any - ciff_files = {k: v for k, v in ciff_files.items() if v} - # Also convert ciff_files structure? - # get_all_files returns Dict[Dataset, List[tuple(path, root, id)]] - # _process_ciff_files expects source_ciff as string? - # Let's adjust helper. - - # Helper expects: ciff_files: Dict[Dataset, List[str]] - # But get_all_files returns List[tuple]. - # I must clean this up. - - ciff_files_clean = {} - for d, flist in ciff_files.items(): - ciff_files_clean[d] = [f[0] for f in flist] - + # get_all_files returns Dict[Dataset, List[tuple(path, root, id)]]; + # keep only the path of each entry + ciff_files = {d: [f[0] for f in flist] for d, flist in ciff_files.items() if flist} + if ciff_files and local_files: _process_ciff_files( - target_ds, ciff_files_clean, local_files, - md_file_pattern, num_threads, console + target_ds, ciff_files, local_files, + num_threads, console ) except Exception as e: @@ -597,62 +586,107 @@ def slice_query( ) -def _process_ciff_files(target_ds, ciff_files, local_files, md_pattern, num_threads, console): - import os +def _process_ciff_files(target_ds, ciff_files, local_files, num_threads, console): local_groups = set([f.rsplit("/", 1)[0] for f in local_files]) errors = [] - + for group in local_groups: + group_label = group or "/" + group_pq_files = [f[len(group):] for f in local_files if f.startswith(group)] + + sources = [] for source_ds, source_files in ciff_files.items(): for source_ciff in source_files: - try: - rel_file = source_ciff[len(source_ds.path):] - if rel_file.startswith(group): - _copy_and_extract_ciff( - source_ds, source_ciff, group, - [f[len(group):] for f in local_files if f.startswith(group)], - target_ds, None, num_threads - ) - except Exception as e: - errors.append(str(e)) - if console: - console.print(f"[red]CIFF error: {str(e)[:100]}[/red]") - + rel_file = source_ciff[len(source_ds.path):] + if rel_file.startswith(group): + sources.append((source_ds, source_ciff)) + + if not sources: + if console: + console.print( + f"[yellow]Warning: no source CIFF index matched '{group_label}'; " + f"the sliced dataset will have no index there[/yellow]" + ) + continue + + try: + kept = _slice_and_merge_ciffs( + sources, group, group_pq_files, target_ds, None, num_threads + ) + if kept == 0 and console: + console.print( + f"[yellow]Warning: sliced index for '{group_label}' contains 0 documents — " + f"no parquet 'id' matched any 'collection_docid' in the source indexes[/yellow]" + ) + except Exception as e: + errors.append(str(e)) + if console: + console.print(f"[red]CIFF error: {str(e)[:100]}[/red]") + return len(errors) == 0 -def _copy_and_extract_ciff(source_ds, source_ciff, pq_sub_dir, pq_files, target_ds, callback, num_threads): +def _slice_and_merge_ciffs(sources, pq_sub_dir, pq_files, target_ds, callback, num_threads): + """ + Filter each source CIFF down to the ids of the sliced parquet files, then merge + the non-empty results into a single index for the target group. Filtering each + source into its own temp file (instead of the target path) is what keeps one + source's empty result from clobbering another's documents. + + :param sources: list of (source_dataset, source_ciff_path) tuples + :return: total number of documents kept across the filtered source indexes + """ import tempfile import os import shutil - + + from ciff_toolkit.merge import merge_ciff_files + def _rel_path(p, sep="/"): return p[1:] if p.startswith(sep) else p - - tmp_dir = None + + target_fs = target_ds.repository.fs + tmp_dir = tempfile.mkdtemp() try: - tmp_dir = tempfile.mkdtemp() - source_fs = source_ds.repository.fs - target_fs = target_ds.repository.fs - - _, ciff_filename = source_ciff.rsplit(source_fs.sep, 1) - tmp_ciff = os.path.join(tmp_dir, ciff_filename) - source_fs.get(source_ciff, tmp_ciff) - ids = set() for file_path in pq_files: full_path = os.path.join(target_ds.path, _rel_path(pq_sub_dir), _rel_path(file_path)) with target_fs.open(full_path, "rb") as f: - ids.update(pq.read_table(f, columns=["id"]).to_pandas()["id"].values) - + for v in pq.read_table(f, columns=["id"]).column("id").to_pylist(): + ids.add(v.decode() if isinstance(v, bytes) else str(v)) + + filtered = [] + ciff_filename = None + for i, (source_ds, source_ciff) in enumerate(sources): + source_fs = source_ds.repository.fs + _, ciff_filename = source_ciff.rsplit(source_fs.sep, 1) + tmp_src = os.path.join(tmp_dir, f"src_{i}_{ciff_filename}") + source_fs.get(source_ciff, tmp_src) + + tmp_out = os.path.join(tmp_dir, f"filtered_{i}_{ciff_filename}") + if callback: callback(f"Filtering {source_ciff}") + kept, _ = drop_parallel(tmp_src, tmp_out, ids, progress_callback=callback, threads=num_threads) + filtered.append((kept, tmp_out)) + os.remove(tmp_src) + + non_empty = [path for kept, path in filtered if kept > 0] + total_kept = sum(kept for kept, _ in filtered) + + if len(non_empty) > 1: + result = os.path.join(tmp_dir, f"merged_{ciff_filename}") + merge_ciff_files(inputs=non_empty, output=result, show_progress=False) + elif non_empty: + result = non_empty[0] + else: + result = filtered[0][1] # a well-formed, empty index + new_ciff = os.path.join(target_ds.path, _rel_path(pq_sub_dir), _rel_path(ciff_filename)) - if callback: callback(f"Processing {new_ciff}") - - drop_parallel(tmp_ciff, new_ciff, ids, progress_callback=callback, threads=num_threads) - + if callback: callback(f"Writing {new_ciff}") + target_fs.put(result, new_ciff) + + return total_kept finally: - if tmp_dir and os.path.exists(tmp_dir): - shutil.rmtree(tmp_dir, ignore_errors=True) + shutil.rmtree(tmp_dir, ignore_errors=True) def query_aggregate( diff --git a/pyproject.toml b/pyproject.toml index d23ad35..b5a1d8a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,7 +36,7 @@ dependencies = [ "python-dateutil>=2.9.0.post0", "s3fs>=2024.12.0", "url-normalize>=2.2.1", - "ciff-toolkit>=0.1.1", + "ciff-toolkit>=0.2.2", "tldextract>=5.0.0", "python-http-irods-client @ git+https://opencode.it4i.eu/lexis-platform/data/python-http-irods-client.git@1.1.4", ] diff --git a/tests/owilix/cli/test_query_slice_integration.py b/tests/owilix/cli/test_query_slice_integration.py index a2c84b1..d531047 100644 --- a/tests/owilix/cli/test_query_slice_integration.py +++ b/tests/owilix/cli/test_query_slice_integration.py @@ -95,6 +95,24 @@ class TestQuerySliceIntegration: print(f"StartDate: {start_date}") # Not asserting date yet, just checking output + print("\n--- Step 2b: Verify Sliced CIFF Index ---") + # Regression check: the sliced index.ciff.gz must contain the sliced + # documents, not just a header (previously each source CIFF overwrote + # the same target file, so the last empty one clobbered the result). + from ciff_toolkit.read import CiffReader + + ciff_paths = [] + for root, _dirs, files in os.walk(path_1): + ciff_paths += [os.path.join(root, f) for f in files if f.endswith("ciff.gz")] + + for ciff_path in ciff_paths: + with CiffReader(ciff_path) as reader: + header = reader.read_header() + print(f"CIFF {ciff_path}: num_docs={header.num_docs}") + assert header.num_docs > 0, ( + f"Sliced index {ciff_path} contains no documents (header-only file)" + ) + print("\n--- Step 3: Second Slice (COM) - Append ---") # Slice 'com' into the EXISTING dataset diff --git a/tests/owilix/core/test_drop_parallel.py b/tests/owilix/core/test_drop_parallel.py index 36fe8f9..8a4820d 100644 --- a/tests/owilix/core/test_drop_parallel.py +++ b/tests/owilix/core/test_drop_parallel.py @@ -71,8 +71,7 @@ def _write_ciff_original(path: Path, *, docs_spec, pls_spec, description: str): with CiffWriter(path) as w: w.write_header(hdr) - for pl in pls: - w.write_message(pl) # postings first + w.write_postings_lists(pls) # postings first w.write_documents(docs) # documents last return path @@ -142,13 +141,15 @@ def test_drop_parallel_keep_subset_and_dump_canonical(tmp_path: Path, capsys: py ) out_path = tmp_path / "out_keep_AC.ciff" - drop_parallel( + kept_count, postings_count = drop_parallel( ciff_file=in_path, new_ciff_file=out_path, collection_ids={"A", "C"}, # keep A & C keep=True, threads=2, ) + assert kept_count == 2 + assert postings_count == 2 # Must be canonical; fail with a clear message if not hdr, pls, docs = _assert_canonical_order(out_path) @@ -190,10 +191,39 @@ def test_drop_parallel_keep_subset_and_dump_canonical(tmp_path: Path, capsys: py assert "Doc 1 (C), length=5" in printed +# ---------- 1b) DROP with zero matches ---------- +def test_drop_parallel_zero_matches_writes_wellformed_empty_ciff(tmp_path: Path): + """ + When no collection_id matches, the output must still be a well-formed CIFF + (readable header, postings and documents sections) with zero documents, + and the returned stats must reflect the empty result. + """ + in_path = tmp_path / "in.ciff" + _write_ciff_original( + in_path, + docs_spec=[("A", 3), ("B", 2)], + pls_spec=[("alpha", [0, 1], [2, 1])], + description="Synthetic input for zero-match drop", + ) + + out_path = tmp_path / "out_empty.ciff" + kept_count, postings_count = drop_parallel( + ciff_file=in_path, + new_ciff_file=out_path, + collection_ids={"ZZZ"}, + keep=True, + threads=2, + ) + assert kept_count == 0 + assert postings_count == 0 + + hdr, pls, docs = _read_original_order(out_path) + assert hdr.num_docs == 0 + assert hdr.num_postings_lists == 0 + assert pls == [] and docs == [] + + # ---------- 2) MERGE after drop ---------- -@pytest.mark.skip(reason="Pre-existing failure in third-party ciff_toolkit.merge: " - "asserts equal average_doclength across inputs, which is " - "violated by design when merging after disjoint drops.") def test_merge_after_two_drops_canonical_and_correct(tmp_path: Path, capsys: pytest.CaptureFixture): """ Drop two disjoint subsets from the same balanced input and merge them. diff --git a/tests/owilix/core/test_slice_ciff.py b/tests/owilix/core/test_slice_ciff.py new file mode 100644 index 0000000..55036c7 --- /dev/null +++ b/tests/owilix/core/test_slice_ciff.py @@ -0,0 +1,150 @@ +# Regression tests for the CIFF slicing step of `query slice` +# (owilix.core.tasks.query._process_ciff_files / _slice_and_merge_ciffs). +# +# Bug history: every source dataset's index.ciff.gz used to be filtered +# directly onto the SAME target path, so the last-processed source (often one +# containing none of the sliced documents) clobbered the result with a +# header-only index. +from pathlib import Path +from types import SimpleNamespace + +import pyarrow as pa +import pyarrow.parquet as pq +from fsspec.implementations.local import LocalFileSystem + +from ciff_toolkit.ciff_pb2 import DocRecord, Header, Posting, PostingsList +from ciff_toolkit.read import CiffReader +from ciff_toolkit.write import CiffWriter + +from owilix.core.tasks.query import _process_ciff_files + + +def _write_ciff(path: Path, docs_spec, pls_spec): + """docs_spec: [(collection_docid, doclength)]; pls_spec: [(term, [docids], [tfs])].""" + docs = [] + for i, (coll_id, dlen) in enumerate(docs_spec): + dr = DocRecord() + dr.docid = i + dr.collection_docid = coll_id + dr.doclength = int(dlen) + docs.append(dr) + + pls = [] + for term, docids, tfs in pls_spec: + pl = PostingsList() + pl.term = term + pl.df = len(docids) + pl.cf = int(sum(tfs)) + prev = 0 + for d, tf in zip(docids, tfs): + p = Posting() + p.docid = int(d - prev) + p.tf = int(tf) + pl.postings.append(p) + prev = d + pls.append(pl) + + hdr = Header() + hdr.version = 1 + hdr.num_docs = len(docs) + hdr.num_postings_lists = len(pls) + hdr.total_docs = len(docs) + hdr.total_postings_lists = len(pls) + hdr.total_terms_in_collection = sum(pl.cf for pl in pls) + hdr.average_doclength = ( + sum(d.doclength for d in docs) / len(docs) if docs else 0.0 + ) + hdr.description = "test fixture" + + with CiffWriter(path) as w: + w.write_header(hdr) + w.write_postings_lists(pls) + w.write_documents(docs) + + +class _FakeDataset: + """Just enough of Dataset for _process_ciff_files: .path and .repository.fs.""" + + def __init__(self, path: Path): + self.path = str(path) + self.repository = SimpleNamespace(fs=LocalFileSystem()) + + +def _make_dataset(path: Path): + path.mkdir(parents=True, exist_ok=True) + return _FakeDataset(path) + + +def _read_index(path: Path): + with CiffReader(path) as r: + header = r.read_header() + postings = list(r.read_postings_lists()) + docs = list(r.read_documents()) + return header, postings, docs + + +def _setup(tmp_path: Path, sliced_ids): + """Two source datasets with their own index; a flat target with sliced parquet.""" + src1 = _make_dataset(tmp_path / "src1") + _write_ciff( + Path(src1.path) / "index.ciff.gz", + docs_spec=[("A", 3), ("B", 2), ("C", 5)], + pls_spec=[("alpha", [0, 2], [2, 1]), ("bravo", [1], [2])], + ) + + src2 = _make_dataset(tmp_path / "src2") + _write_ciff( + Path(src2.path) / "index.ciff.gz", + docs_spec=[("E", 4), ("F", 1)], + pls_spec=[("alpha", [0], [3]), ("delta", [1], [1])], + ) + + target = _make_dataset(tmp_path / "target") + pq.write_table(pa.table({"id": list(sliced_ids)}), Path(target.path) / "part_0.parquet") + + ciff_files = { + src1: [str(Path(src1.path) / "index.ciff.gz")], + src2: [str(Path(src2.path) / "index.ciff.gz")], + } + return target, ciff_files + + +def test_slice_ciff_survives_nonmatching_source_processed_last(tmp_path: Path): + """Docs come from src1 only; src2 (processed last) must not clobber the index.""" + target, ciff_files = _setup(tmp_path, sliced_ids=["A", "C"]) + + ok = _process_ciff_files(target, ciff_files, ["/part_0.parquet"], num_threads=2, console=None) + assert ok + + header, postings, docs = _read_index(Path(target.path) / "index.ciff.gz") + assert header.num_docs == 2 + assert sorted(d.collection_docid for d in docs) == ["A", "C"] + assert {pl.term for pl in postings} == {"alpha"} + + +def test_slice_ciff_merges_documents_from_all_sources(tmp_path: Path): + """Docs from both sources must end up in one merged index.""" + target, ciff_files = _setup(tmp_path, sliced_ids=["A", "E", "F"]) + + ok = _process_ciff_files(target, ciff_files, ["/part_0.parquet"], num_threads=2, console=None) + assert ok + + header, postings, docs = _read_index(Path(target.path) / "index.ciff.gz") + assert header.num_docs == 3 + assert sorted(d.collection_docid for d in docs) == ["A", "E", "F"] + # "alpha" occurs in both sources and must be merged into one postings list + terms = [pl.term for pl in postings] + assert terms.count("alpha") == 1 + assert "delta" in terms + + +def test_slice_ciff_zero_matches_writes_wellformed_empty_index(tmp_path: Path): + """No id matches at all: the index must exist, be readable, and have 0 docs.""" + target, ciff_files = _setup(tmp_path, sliced_ids=["ZZZ"]) + + ok = _process_ciff_files(target, ciff_files, ["/part_0.parquet"], num_threads=2, console=None) + assert ok + + header, postings, docs = _read_index(Path(target.path) / "index.ciff.gz") + assert header.num_docs == 0 + assert postings == [] and docs == [] diff --git a/uv.lock b/uv.lock index 35683d5..0a61983 100644 --- a/uv.lock +++ b/uv.lock @@ -306,15 +306,15 @@ wheels = [ [[package]] name = "ciff-toolkit" -version = "0.1.1" +version = "0.2.2" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "protobuf" }, { name = "tqdm" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/ae/e5/fa32c9b820229dab4082ffb3a5e94607d86ef2af2c41c3ad1915f26c81b4/ciff-toolkit-0.1.1.tar.gz", hash = "sha256:361444935f3524d03fb1ca80dc234539dfdf897db6a057cdf60ac75b2a1a3f91", size = 11159, upload-time = "2023-06-22T12:27:13.546Z" } +sdist = { url = "https://files.pythonhosted.org/packages/af/dc/20b961d6ac14dc8387d0be10bf2c2497d20ac57dd7b8b7bff316a3b9af14/ciff_toolkit-0.2.2.tar.gz", hash = "sha256:4e832c2baf16b48eaa3f1290e71b2b8c7d768f0a84e11d40ed173d4e826c1666", size = 10056, upload-time = "2026-02-12T10:04:45.543Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/bb/6c/2564ac35844265106a121bf0cdc65b224ad81a309f45f10e1c0fd4b47cdd/ciff_toolkit-0.1.1-py3-none-any.whl", hash = "sha256:701d48028783ae9618a45d1a1a6dfb8d4cbaa3cb268c8bd3e09db5d932fde3c7", size = 12464, upload-time = "2023-06-22T12:27:15.187Z" }, + { url = "https://files.pythonhosted.org/packages/3d/62/e238ef9fc445f5a823b93b4c54ad1a2e130d42609f686fd805b7b5ba9465/ciff_toolkit-0.2.2-py3-none-any.whl", hash = "sha256:6324b33dcb255ef558106840082d676bf1352f02a5596cd5436909dee70b8de7", size = 13073, upload-time = "2026-02-12T10:04:44.521Z" }, ] [[package]] @@ -1177,7 +1177,7 @@ warc = [ [package.metadata] requires-dist = [ - { name = "ciff-toolkit", specifier = ">=0.1.1" }, + { name = "ciff-toolkit", specifier = ">=0.2.2" }, { name = "click", specifier = ">8.1.6" }, { name = "duckdb", specifier = ">=1.5.0" }, { name = "fastapi", marker = "extra == 'http'", specifier = ">=0.115,<1.0" }, -- 2.51.2