def delta_retry_update(
    dc: "DataChain",
    name: str,
    on: Union[str, Sequence[str]],
    right_on: Optional[Union[str, Sequence[str]]] = None,
    compare: Optional[Union[str, Sequence[str]]] = None,
    delta_retry: Optional[Union[bool, str]] = None,
) -> tuple[Optional["DataChain"], Optional[list[DatasetDependency]], bool]:
    """
    Creates new chain that consists of the last version of current delta dataset
    plus diff from the source with all needed modifications.
    This way we don't need to re-calculate the whole chain from the source again
    (apply all the DataChain methods like filters, mappers, generators etc.)
    but just the diff part which is very important for performance.

    Note that currently delta update works only if there is only one direct
    dependency.

    Additionally supports retry functionality to filter records that either:
    1. Have a non-None value in the field specified by delta_retry (when it's a string)
    2. Exist in the source dataset but are missing in the result dataset
       (when delta_retry=True)

    Parameters:
        dc: The DataChain to filter for records that need reprocessing
        name: Name of the destination dataset
        on: Field(s) in source dataset that uniquely identify records
        right_on: Corresponding field(s) in result dataset if they differ from
                  source
        compare: Field(s) used to check if the same row has been modified
        delta_retry: If string, field in result dataset that indicates an error
                    when not None. If True, include records missing from result dataset.
                    If False/None, no retry functionality.

    Returns:
        A tuple containing (filtered chain for delta/retry processing,
                          dependencies, found records flag)
    """

    catalog = dc.session.catalog
    dc._query.apply_listing_pre_step()

    # Check if dataset exists
    try:
        dataset = catalog.get_dataset(name)
        latest_version = dataset.latest_version
    except DatasetNotFoundError:
        # First creation of result dataset
        return None, None, True

    # Initialize variables
    diff_chain = None
    dependencies = None
    retry_chain = None
    processing_chain = None

    source_ds_name, source_ds_version, source_ds_latest_version, dependencies = (
        _get_source_info(name, latest_version, catalog)
    )

    # If source_ds_name is None, starting dataset was removed
    if source_ds_name is None:
        return None, None, True

    assert source_ds_version
    assert source_ds_latest_version

    diff_chain = _get_delta_chain(
        source_ds_name, source_ds_version, source_ds_latest_version, on, compare
    )

    # Filter out removed dep
    if dependencies:
        dependencies = copy(dependencies)
        dependencies = [d for d in dependencies if d is not None]
        # Update to latest version
        dependencies[0].version = source_ds_latest_version  # type: ignore[union-attr]

    # Handle retry functionality if enabled
    if delta_retry:
        retry_chain = _get_retry_chain(
            name,
            latest_version,
            source_ds_name,
            source_ds_latest_version,
            on,
            right_on,
            delta_retry,
        )

    # Combine delta and retry chains
    if retry_chain is not None:
        processing_chain = diff_chain.union(retry_chain)
    else:
        processing_chain = diff_chain

    # Apply all the steps from the original chain to processing_chain
    processing_chain = _append_steps(processing_chain, dc).persist()

    # Check if chain becomes empty after applying steps
    if processing_chain is None or (processing_chain and processing_chain.empty):
        return None, None, False

    latest_dataset = datachain.read_dataset(name, latest_version)
    compared_chain = latest_dataset.compare(
        processing_chain,
        on=right_on or on,
        added=True,
        modified=False,
        deleted=False,
    )
    result_chain = compared_chain.union(processing_chain)
    return result_chain, dependencies, True

    Creates new chain that consists of the last version of current delta dataset
    plus diff from the source with all needed modifications.
    This way we don't need to re-calculate the whole chain from the source again
    (apply all the DataChain methods like filters, mappers, generators etc.)
    but just the diff part which is very important for performance.

    Note that currently delta update works only if there is only one direct
    dependency.

    Additionally supports retry functionality to filter records that either:
    1. Have a non-None value in the field specified by delta_retry (when it's a string)
    2. Exist in the source dataset but are missing in the result dataset
       (when delta_retry=True)

    Parameters:
        dc: The DataChain to filter for records that need reprocessing
        name: Name of the destination dataset
        on: Field(s) in source dataset that uniquely identify records
        right_on: Corresponding field(s) in result dataset if they differ from
                  source
        compare: Field(s) used to check if the same row has been modified
        delta_retry: If string, field in result dataset that indicates an error
                    when not None. If True, include records missing from result dataset.
                    If False/None, no retry functionality.

    Returns:
        A tuple containing (filtered chain for delta/retry processing,
                          dependencies, found records flag)
    
src/datachain/delta.py