def test_retry_with_multiple_match_fields(test_session):
    """Test retry functionality with multiple fields for matching."""

    def process_with_compound_key(
        category: str, id: int, content: str, attempt: int
    ) -> ProcessingResult:
        error = category == "A" and id % 2 == 1 and attempt == 1
        return ProcessingResult(
            processed_content="" if error else f"{category}-{content}".upper(),
            processed_at=datetime.now(tz=timezone.utc).isoformat(),
            error=f"Error for {category}-{id}" if error else "",
            attempt=attempt,
        )

    # Create dataset with compound keys
    categories = ["A", "A", "B", "B"]
    ids = [1, 2, 1, 2]
    contents = ["first", "second", "third", "fourth"]

    dc.read_values(
        category=categories, id=ids, content=contents, session=test_session
    ).save("compound_source")

    first_pass = (
        dc.read_dataset("compound_source", session=test_session)
        .setup(attempt=lambda: 1)
        .map(result=process_with_compound_key)
        .save("compound_result")
    )

    # Only A-1 should fail (category="A" and odd id)
    results = list(first_pass.filter(C("result.error") != "").collect("category", "id"))
    assert results == [("A", 1)]

    # Retry with multiple match fields
    (
        dc.read_dataset(
            "compound_source",
            session=test_session,
            delta=True,
            delta_on=["category", "id"],
            delta_retry="result.error",
        )
        .setup(attempt=lambda: 2)
        .map(result=process_with_compound_key)
        .save("compound_result")
    )

    final_result = dc.read_dataset("compound_result", session=test_session)
    results = list(
        final_result.collect("category", "id", "result.attempt", "result.error")
    )
    assert set(results) == {
        ("A", 2, 1, ""),
        ("B", 1, 1, ""),
        ("B", 2, 1, ""),
        ("A", 1, 2, ""),
    }
Test retry functionality with multiple fields for matching.
tests/func/test_retry.py