Function squash_updates

python/pathway/internals/api.py:197–226

(
    updates: CapturedStream, *, terminate_on_error: bool = True
)

def squash_updates(
    updates: CapturedStream, *, terminate_on_error: bool = True
) -> CapturedTable:
    state: CapturedTable = {}
    updates.sort(key=lambda row: (row.time, row.diff))

    def handle_error(row: DataRow, msg: str):
        if terminate_on_error:
            raise KeyError(msg)
        else:
            warnings.warn(msg)
            t: tuple[Value, ...] = (ERROR,) * len(row.values)
            state[row.key] = t

    for row in updates:
        if row.diff == 1:
            if row.key in state:
                handle_error(row, f"duplicated entries for key {row.key}")
                continue
            state[row.key] = tuple(row.values)
        elif row.diff == -1:
            if state[row.key] != tuple(row.values):
                handle_error(row, f"deleting non-existing entry {row.values}")
                continue
            del state[row.key]
        else:
            handle_error(row, f"invalid diff value: {row.diff}")
            continue

    return state


class PyObjectWrapperSerializerProtocol(Protocol):