Bug in python replication logic?

[ChuckHastings]

def _mg_call_plc_replicate(
    client: dask.distributed.client.Client,
    sID: bytes,
    dask_object: dict,
    input_type: str,
    col_names: list,
) -> Union[dask_cudf.DataFrame, dask_cudf.Series]:
    if input_type == "dataframe":
        result = [
            client.submit(
                _call_plc_replicate_dataframe,
                sID,
                edata,
                workers=[w],
                allow_other_workers=False,
                pure=False,
            )
            for w, edata in dask_object.items()
        ]
    elif input_type == "dataframe":
        result = [
            client.submit(
                _call_plc_replicate_series,
                sID,
                edata,
                workers=[w],
                allow_other_workers=False,
                pure=False,
            )
            for w, edata in dask_object.items()
        ]

Not directly related to this PR, but this code looks suspicious as well.

    elif input_type == "dataframe":

Shouldn't "dataframe" here be "series"?

Originally posted by @seunghwak in #5628 (comment)