def test_retry_with_error_records(test_session):
    """Test retry functionality with records that have errors."""

    def _run_processing(attempt: int) -> "DataChain":
        return (
            dc.read_dataset(
                "sample_data",
                delta=True,
                delta_on="id",
                delta_retry="result.error",
                session=test_session,
            )
            .setup(attempt=lambda: attempt)
            .map(result=_process_with_errors)
            .save("processed_data")
        )

    # First processing pass - some records will fail
    sample_ids = [1, 2, 3, 4]
    sample_contents = ["first item", "second item", "third item", "fourth item"]

    dc.read_values(id=sample_ids, content=sample_contents, session=test_session).save(
        "sample_data"
    )

    first_pass = _run_processing(1)

    # Check that some records failed
    error_count = first_pass.filter(C("result.error") != "").count()
    success_count = first_pass.filter(C("result.error") == "").count()

    assert error_count == 2
    assert success_count == 2

    # Retry processing - only failed records should be reprocessed
    _run_processing(2)

    final_result = dc.read_dataset("processed_data", session=test_session)
    results = list(final_result.collect("id", "result.attempt", "result.error"))
    assert set(results) == {(2, 1, ""), (4, 1, ""), (1, 2, ""), (3, 2, "")}
Test retry functionality with records that have errors.
tests/func/test_retry.py