def _compare(  # noqa: C901, PLR0912
    left: "DataChain",
    right: "DataChain",
    on: Union[str, Sequence[str]],
    right_on: Optional[Union[str, Sequence[str]]] = None,
    compare: Optional[Union[str, Sequence[str]]] = None,
    right_compare: Optional[Union[str, Sequence[str]]] = None,
    added: bool = True,
    deleted: bool = True,
    modified: bool = True,
    same: bool = True,
    status_col: Optional[str] = None,
) -> "DataChain":
    """Comparing two chains by identifying rows that are added, deleted, modified
    or same"""
    rname = "right_"
    schema = left.signals_schema  # final chain must have schema from left chain

    def _to_list(obj: Optional[Union[str, Sequence[str]]]) -> Optional[list[str]]:
        if obj is None:
            return None
        return [obj] if isinstance(obj, str) else list(obj)

    on = _to_list(on)  # type: ignore[assignment]
    right_on = _to_list(right_on)
    compare = _to_list(compare)
    right_compare = _to_list(right_compare)

    if not any([added, deleted, modified, same]):
        raise ValueError(
            "At least one of added, deleted, modified, same flags must be set"
        )
    if on is None:
        raise ValueError("'on' must be specified")
    if right_on and len(on) != len(right_on):
        raise ValueError("'on' and 'right_on' must be have the same length")
    if right_compare and not compare:
        raise ValueError("'compare' must be defined if 'right_compare' is defined")
    if compare and right_compare and len(compare) != len(right_compare):
        raise ValueError("'compare' and 'right_compare' must have the same length")

    # all left and right columns
    cols = left.signals_schema.clone_without_sys_signals().db_signals()
    right_cols = right.signals_schema.clone_without_sys_signals().db_signals()
    cols_select = list(left.signals_schema.clone_without_sys_signals().values.keys())

    # getting correct on and right_on column names
    on_ = on
    on = left.signals_schema.resolve(*on).db_signals()  # type: ignore[assignment]
    right_on = right.signals_schema.resolve(*(right_on or on_)).db_signals()  # type: ignore[assignment]

    # getting correct compare and right_compare column names if they are defined
    if compare:
        compare_ = compare
        compare = left.signals_schema.resolve(*compare).db_signals()  # type: ignore[assignment]
        right_compare = right.signals_schema.resolve(
            *(right_compare or compare_)
        ).db_signals()  # type: ignore[assignment]
    elif not compare and len(cols) != len(right_cols):
        # here we will mark all rows that are not added or deleted as modified since
        # there was no explicit list of compare columns provided (meaning we need
        # to check all columns to determine if row is modified or same), but
        # the number of columns on left and right is not the same (one of the chains
        # have additional column)
        compare = None
        right_compare = None
    else:
        # we are checking all columns as explicit compare is not defined
        compare = right_compare = [c for c in cols if c in right_cols and c not in on]  # type: ignore[misc]

    # get diff column names
    diff_col = status_col or get_status_col_name()
    ldiff_col = get_status_col_name()
    rdiff_col = get_status_col_name()

    # adding helper diff columns, which will be removed after
    left = left.mutate(**{ldiff_col: 1})
    right = right.mutate(**{rdiff_col: 1})

    if not compare:
        modified_cond = True
    else:
        modified_cond = or_(  # type: ignore[assignment]
            *[
                C(c) != (C(f"{rname}{rc}") if c == rc else C(rc))
                for c, rc in zip(compare, right_compare)  # type: ignore[arg-type]
            ]
        )

    dc_diff = (
        left.merge(right, on=on, right_on=right_on, rname=rname, full=True)
        .mutate(
            **{
                diff_col: case(
                    (isnone(ldiff_col), CompareStatus.DELETED),
                    (isnone(rdiff_col), CompareStatus.ADDED),
                    (modified_cond, CompareStatus.MODIFIED),
                    else_=CompareStatus.SAME,
                )
            }
        )
        # when the row is deleted, we need to take column values from the right chain
        .mutate(
            **{
                f"{l_on}": ifelse(
                    C(diff_col) == CompareStatus.DELETED,
                    C(f"{rname + l_on if on == right_on else r_on}"),
                    C(l_on),
                )
                for l_on, r_on in zip(on, right_on)  # type: ignore[arg-type]
            }
        )
        .select_except(ldiff_col, rdiff_col)
    )

    if not added:
        dc_diff = dc_diff.filter(C(diff_col) != CompareStatus.ADDED)
    if not modified:
        dc_diff = dc_diff.filter(C(diff_col) != CompareStatus.MODIFIED)
    if not same:
        dc_diff = dc_diff.filter(C(diff_col) != CompareStatus.SAME)
    if not deleted:
        dc_diff = dc_diff.filter(C(diff_col) != CompareStatus.DELETED)

    if status_col:
        cols_select.append(diff_col)

    if not dc_diff._sys:
        # TODO workaround when sys signal is not available in diff
        dc_diff = dc_diff.settings(sys=True).select(*cols_select).settings(sys=False)
    else:
        dc_diff = dc_diff.select(*cols_select)

    # final schema is schema from the left chain with status column added if needed
    dc_diff.signals_schema = (
        schema if not status_col else SignalSchema({status_col: str}) | schema
    )

    return dc_diff
Comparing two chains by identifying rows that are added, deleted, modified
    or same
src/datachain/diff/__init__.py