def test_retry_with_delta_functionality(test_session):
    """Test that retry and delta can work together."""

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

    dc.read_values(id=[1, 2], content=["first", "second"], session=test_session).save(
        "delta_source_v1"
    )

    result = _run_processing(1)
    assert set(result.collect("id", "result.error")) == {
        (1, "Processing error for item 1"),
        (2, ""),
    }

    # Add more data
    # Process again - should only process new records and retry errors
    extended_ids = [1, 2, 3]
    extended_contents = ["first", "second", "third"]
    dc.read_values(
        id=extended_ids, content=extended_contents, session=test_session
    ).save("delta_source_v1")

    result_v2 = _run_processing(2)
    assert set(result_v2.collect("id", "result.error", "result.attempt")) == {
        (1, "", 2),
        (2, "", 1),
        (3, "", 2),
    }
Test that retry and delta can work together.
tests/func/test_retry.py