def fix_overlap(ddf, overlap):
    """Ensures that the upper bound on each partition of ddf (except the last) is exclusive"""
    name = "fix-overlap-" + tokenize(ddf, overlap)
    n = len(ddf.divisions) - 1
    dsk = {(name, i): (ddf._name, i) for i in range(n)}

    frames = []
    for i in overlap:

        # `frames` is a list of data from previous partitions that we may want to
        # move to partition i.  Here, we add "overlap" from the previous partition
        # (i-1) to this list.
        frames.append((get_overlap, (ddf._name, i - 1), ddf.divisions[i]))

        # Make sure that any data added from partition i-1 to `frames` is removed
        # from partition i-1.
        dsk[(name, i - 1)] = (drop_overlap, dsk[(name, i - 1)], ddf.divisions[i])

        # We do not want to move "overlap" from the previous partition (i-1) into
        # this partition (i) if the data from this partition will need to be moved
        # to the next partition (i+1) anyway.  If we concatenate data too early,
        # we may lose rows (https://github.com/dask/dask/issues/6972).
        if i == ddf.npartitions - 2 or ddf.divisions[i] != ddf.divisions[i + 1]:
            frames.append((ddf._name, i))
            dsk[(name, i)] = (methods.concat, frames)
            frames = []

    graph = HighLevelGraph.from_collections(name, dsk, dependencies=[ddf])
    return new_dd_object(graph, name, ddf._meta, ddf.divisions)