[feat] SID: add offline codebook collision-prevention tool

[WhiteSwan1]

Summary

This PR adds TorchEasyRec-native tools for evaluating Semantic ID (SID) quality and resolving over-capacity SID buckets offline:

• tzrec.tools.sid.evaluate_sid_quality computes exact global and per-layer quality metrics. It can compare original and resolved fields on the same rows and emit before/after/delta reports.
• tzrec.tools.sid.resolve_sid_collisions performs deterministic, best-effort collision resolution within each SID prefix band. It supports model-provided candidates and the legacy deterministic random strategy.
• SID prediction exposes each item's candidate SIDs in the flat topk * n_layers wire format consumed by the resolver.
• Reusable NumPy logic lives in tzrec/utils/sid/quality.py and tzrec/utils/sid/collision.py; tool modules contain repository I/O and CLI orchestration.
• Shared local/ODPS path-overlap detection lives in tzrec/utils/path_util.py and reports both conflicting groups and their normalized overlapping location.

Both tools use the repository create_reader / create_writer interfaces and support CSV, Parquet, and ODPS. Collision resolution is single-process and CPU-only.

Input and output contract

The resolver consumes fixed-width tzrec.predict output:

• item_id, which is expected to be unique upstream;
• codes, a list<int> of width n_layers;
• for --strategy candidate, candidate_codes, a flat list<int> of width topk * n_layers.

It writes:

• an item map containing item_id, origin_codebook, codebook, and the one-based index within the final SID bucket;
• a required resolved grouping containing codebook and itemids;
• an optional original grouping with the same schema.

--rate_only computes statistics without writing outputs. With unique item IDs, results are independent of source-row order and I/O batch boundaries. Rare duplicate IDs are preserved as independent rows on a best-effort basis; they should still be corrected upstream.

Resolution is intentionally best-effort. An overflow item takes its first free last-layer candidate while retaining its prefix. If its finite candidate list has no free destination, it keeps its original SID. The bucket and its index may therefore remain above --max_items_per_codebook; this is reported through unresolved_count and final collision statistics.

Design decisions

• Fixed-width SID validation follows the supported tzrec.predict producer contract. Arbitrary malformed external Arrow tables are not a supported input format in this PR.
• Item-ID uniqueness remains an upstream responsibility. Best-effort duplicate tolerance avoids stopping a very large job for a few invalid rows but does not claim to repair or deduplicate them.
• The first-fit loop remains sequential because occupancy changes after every placement. Chunking limits temporary objects without changing candidate order or bucket state.

Test plan

The five focused modules run 145 tests covering golden resolution and quality results, CSV/Parquet/mocked ODPS I/O, before/after comparison, duplicate-ID tolerance, no-overflow and unresolved fallback paths, local/ODPS path aliases, cross-batch candidate alignment, Arrow offset-aware writes, progress reporting, and equivalence across first-fit chunk boundaries. All focused tests and pre-commit run -a pass. The local Pyre wrapper reports no critical type errors; the installed Pyre CLI also emits a version-option compatibility warning.

[tiankongdeguiji]

Please restructure the SID tooling as follows:

tzrec/tools/sid/
├── resolve_sid_collisions.py
└── evaluate_sid_quality.py

tzrec/utils/sid/
├── collision.py
├── quality.py
└── catalog.py       # Future; do not add in this PR

Specifically:

• Rename collision_prevention.py to resolve_sid_collisions.py.
• Move reusable logic from collision_resolution.py into utils/sid/collision.py.
• Rename sid_quality_report.py to evaluate_sid_quality.py and move reusable metrics into utils/sid/quality.py.

Please preserve the current two-pass processing design:

plan = prepare_collision_plan(item_ids, origin_codes, config)

# Only required by KnnCollisionResolver.
candidate_codes = load_candidate_codes(
    candidate_input_path,
    plan.overflow_item_ids,
)

result = resolver.resolve(plan, candidate_codes)

The first pass should build a compact CollisionPlan. For KNN resolution, the second pass should load candidates only for plan.overflow_item_ids. The random strategy should skip candidate loading.

utils/sid/collision.py should provide:

CollisionPlan
prepare_collision_plan()
CollisionResolver
KnnCollisionResolver
RandomCollisionResolver

The resolver interface should operate on the prepared plan:

class CollisionResolver(ABC):
    @abstractmethod
    def resolve(
        self,
        plan: CollisionPlan,
        candidate_codes: np.ndarray | None = None,
    ) - CollisionResolutionResult:
        """Resolve collisions for a prepared overflow plan."""

evaluate_sid_quality.py should support both raw assignments and the output of resolve_sid_collisions.py. When both origin_codes and resolved codes are present, it should report before/after metrics and their deltas.

Please also document that the current resolvers are best-effort: a bucket may remain over capacity when all candidates are exhausted.

[tiankongdeguiji]

Make original_sid_groups_output_path optional. The original grouping is primarily a diagnostic/audit artifact, while the item map and resolved grouping are the operational outputs consumed downstream.

[github-actions]

Static review only; no tests or builds were run.

I left seven inline comments on the remaining noteworthy issues:

• silent row-boundary corruption for balanced ragged SID/candidate lists;
• order-dependent candidate selection for conflicting duplicate item IDs;
• input data loss through multi-source or multi-partition path-alias gaps;
• quality outputs overwriting the input or each other;
• three memory/throughput blockers for the stated hundred-million-row scale.

The PR description should also be updated: it currently calls the output capacity-safe, while the implementation intentionally keeps unplaceable items in over-capacity origin buckets. The module and CLI already describe that best-effort behavior accurately.

[WhiteSwan1]

thanks for github-actions's comment. Changes are included in d8e1e00.

Accepted

• R3/R4: Added shared local/ODPS path-conflict detection. Both SID tools validate all input/output groups before I/O and return the conflicting locations.
• R7: Bounded .tolist() temporary objects with _ROW_CHUNK_SIZE while preserving greedy order and results.
• R8: Updated the PR description with best-effort behavior, fixed-width candidates, the unique-ID assumption, and optional original grouping.

Rebutted

• R1: Ragged rows cannot be produced by the supported fixed-width tzrec.predict pipeline. Arbitrary malformed external tables are outside the current contract.
• R2: Item-ID uniqueness belongs upstream. Best-effort duplicate tolerance prevents a few invalid rows from blocking large jobs; it does not claim to repair duplicates.

Follow-up

• R5: batch_size only controls I/O. Chunk lists are now released after concatenation; out-of-core planning requires production memory profiling.
• R6: Candidate memory scales with overflow rows. Random streaming and KNN spill should be separate, profile-driven performance changes.

Summary

All correctness and low-risk memory findings were addressed. Contract-only cases were documented, while architectural performance work remains scoped to future PRs.