pathwaycom/pathway · error · KeyError

duplicated entries for key {row.key}

Error message

duplicated entries for key {row.key}

What it means

squash_updates folds a captured update stream into a table state; the semantics are that a key can be inserted (diff=1) at most once without a preceding deletion (diff=-1). When a second insert for a key that is already present arrives, the stream contradicts itself. By default terminate_on_error=True and this raises KeyError; with terminate_on_error=False the error is only warned about and the key's values become ERROR sentinels.

Source

Thrown at python/pathway/internals/api.py:205

            raise ValueError(f"Column {DIFF_PSEUDOCOLUMN} can only contain 1 and -1.")
        shard = data[SHARD_PSEUDOCOLUMN][i] if SHARD_PSEUDOCOLUMN in data else None
        input_row = DataRow(
            key, values, time=time, diff=diff, shard=shard, dtypes=dtypes
        )
        input_data.append(input_row)

    return scope.static_table(input_data, connector_properties)


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}")

View on GitHub (pinned to fa2f74a464)

Solutions

1. Deduplicate inserts by key before replay: keep only the first insert per (key, time) or convert later inserts into delete/insert pairs.
2. If updates are encoded as new inserts, emit diff=-1 for the old row before diff=1 for the new one.
3. For exploratory analysis where you accept corrupted keys, pass terminate_on_error=False and handle ERROR values downstream.

Example fix

# before (duplicate inserts for key 'a')
rows = [
    ("a", 1, 1),
    ("a", 2, 1),  # duplicated entries for key 'a'
]

# after (delete then re-insert)
rows = [
    ("a", 1, 1),
    ("a", 1, -1),
    ("a", 2, 1),
]

seen: set = set()
for row in updates:
    key, diff = row.key, row.diff
    if diff == 1:
        if key in seen:
            raise ValueError(f"duplicate insert for key {key}; add a diff=-1 row first")
        seen.add(key)
    elif diff == -1:
        seen.discard(key)

try:
    state = pw.internals.api.squash_updates(updates)
except KeyError as e:
    # duplicated insert for a key: fix the generator, don't continue with ERROR sentinels
    raise RuntimeError(f"update stream is inconsistent: {e}") from e

When it happens

Trigger: Calling squash_updates (used internally when replaying captured/static data with updates) on a stream where the same key appears twice with diff=1 and no intervening diff=-1; generating test dataframes with duplicated primary keys and __diff__=1 on both rows; replaying a CDC feed where an UPDATE was encoded as two inserts instead of insert+delete.

Common situations: Synthetic test data generated with np.random so keys collide; CDC feeds where updates should be delete+insert pairs but were emitted as inserts; replaying deduplicated Kafka topics by primary key where duplicates slipped through.

Related errors

• Column {DIFF_PSEUDOCOLUMN} can only contain 1 and -1.
• Incorrect S3 path: {s3_path}
• SchemaRegistryHeader.key must be a str, got {type(self.key).
• Can't use 'id' as a column name
• schema does not have columns {missing_columns}

AI-assisted analysis of pathwaycom/pathway@fa2f74a464 (2026-08-15). Data as JSON: /api/errors/64fce60edf941164. Report an issue: GitHub.