def test_retry_first_dataset_creation(test_session):
    """Test retry when dataset doesn't exist yet (first creation)."""
    source_ids = [1, 2]
    source_contents = ["first", "second"]

    dc.read_values(id=source_ids, content=source_contents, session=test_session).save(
        "source_data"
    )

    def simple_process(id: int, content: str) -> ProcessingResult:
        return ProcessingResult(
            processed_content=content.upper(),
            processed_at=datetime.now(tz=timezone.utc).isoformat(),
            error="",
            attempt=1,
        )

    # First run with retry enabled on non-existent dataset
    # Should process all records
    retry_chain = (
        dc.read_dataset(
            "source_data",
            session=test_session,
            delta=True,
            delta_on="id",
            delta_retry="result.error",
        )
        .map(result=simple_process)
        .save("new_dataset")
    )

    assert retry_chain.count() == 2
    assert retry_chain.filter(C("result.error") != "").count() == 0
Test retry when dataset doesn't exist yet (first creation).
tests/func/test_retry.py