def _get_retry_chain(
    name: str,
    latest_version: str,
    source_ds_name: str,
    source_ds_latest_version: str,
    on: Union[str, Sequence[str]],
    right_on: Optional[Union[str, Sequence[str]]],
    delta_retry: Optional[Union[bool, str]],
) -> Optional["DataChain"]:
    """Get retry chain for processing error records and missing records."""
    # Import here to avoid circular import
    from datachain.lib.dc import C

    retry_chain = None

    # Read the latest version of the result dataset for retry logic
    result_dataset = datachain.read_dataset(name, latest_version)
    source_dc_latest = datachain.read_dataset(source_ds_name, source_ds_latest_version)

    # Handle error records if delta_retry is a string (column name)
    if isinstance(delta_retry, str):
        error_records = result_dataset.filter(C(delta_retry) != "")
        error_source_records = source_dc_latest.merge(
            error_records, on=on, right_on=right_on, inner=True
        ).select(*list(source_dc_latest.signals_schema.values))
        retry_chain = error_source_records

    # Handle missing records if delta_retry is True
    elif delta_retry is True:
        missing_records = source_dc_latest.subtract(
            result_dataset, on=on, right_on=right_on
        )
        retry_chain = missing_records

    return retry_chain
Get retry chain for processing error records and missing records.
src/datachain/delta.py