pathwaycom/pathway · error · ValueError

primary_key column ' ' is declared nullable; primary_key…

Error message

primary_key column '{pkey_field.name}' is declared nullable; primary_key columns must be non-nullable in snapshot mode. Either remove the Optional wrapper in the schema or filter out nulls upstream via .filter(t.pkey.is_not_none()).

What it means

Raised by pw.io.postgres.write in snapshot mode (output_table_type='snapshot' or the write_snapshot path) when a primary_key column is typed Optional. Snapshot mode writes PRIMARY KEY ... NOT NULL on create_if_not_exists/replace (so NULL rows fail at INSERT), and against a pre-existing table with a nullable PK the DELETE ... WHERE pkey=$1 retraction never matches because SQL = NULL is always false. Both outcomes silently lose data, so Pathway refuses the configuration up front.

Solutions

Remove the Optional wrapper on the key column in the schema class (id: int instead of id: Optional[int]).
If nulls genuinely occur, filter them out upstream before writing: t = t.filter(t.id.is_not_none()), which also narrows the type to non-optional.
If the data cannot be trusted, cast with .unwrap() or apply a default so the column is non-nullable by the time it reaches write().

Example fix

# before
class InputSchema(pw.Schema):
    id: Optional[int]
    value: str

pw.io.postgres.write(t, conn, "t", primary_key=[t.id], output_table_type="snapshot")

# after
class InputSchema(pw.Schema):
    id: int
    value: str

pw.io.postgres.write(t, conn, "t", primary_key=[t.id], output_table_type="snapshot")

Defensive patterns

Strategy: validation

Validate before calling

key_col = "id"
assert not t.schema.columns()[key_col].dtype.is_optional, \
    f"primary key column {key_col!r} must not be Optional"
pw.io.postgres.write(t, conn, "t", primary_key=[t.id], output_table_type="snapshot")

Type guard

def key_is_not_optional(t: "pw.Table", col: str) -> bool:
    return not t.schema.columns()[col].dtype.is_optional

Prevention

Keep primary-key schema fields non-Optional from the start; only value columns may be Optional.
Filter nulls early: t = t.filter(t.id.is_not_none()) — it fixes the type and cleans the data.
Treat Optional keys as a schema smell in code review.

When it happens

Trigger: pw.io.postgres.write(t, conn, 't', primary_key=[t.id], output_table_type='snapshot') where the schema declares id: Optional[int] (or any Optional type), or pw.io.postgres.write_snapshot with a nullable key column.

Common situations: Schemas generated from JSON/CSV inputs where every column defaults to Optional; ingesting raw data whose key column can be None; migrating an existing pipeline to snapshot mode without tightening the schema.

Related errors

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

Appendix: source

Thrown at python/pathway/io/postgres/__init__.py:944

        # ``primary_key_fields`` index-swap used on DELETE. Reject it
        # here with a clear message instead.
        duplicates = sorted(
            {name for name in key_field_names if key_field_names.count(name) > 1}
        )
        if duplicates:
            raise ValueError(f"primary_key contains duplicate column(s) {duplicates}")
        # A nullable primary-key column in snapshot mode is silently
        # broken: either the `NOT NULL` PRIMARY KEY we emit on
        # create_if_not_exists / replace rejects the NULL row at
        # insert time, or (against a pre-existing table that allows
        # NULL in the PK) the retraction ``DELETE ... WHERE pkey=$1``
        # never matches anything because SQL ``= NULL`` is always
        # false. Both are data-loss footguns, so we refuse the setup
        # here with an actionable message.
        if is_snapshot_mode:
            for pkey_field in primary_key:
                if isinstance(pkey_field._column.dtype, dtype.Optional):
                    raise ValueError(
                        f"primary_key column '{pkey_field.name}' is "
                        "declared nullable; primary_key columns must be "
                        "non-nullable in snapshot mode. Either remove "
                        "the Optional wrapper in the schema or filter "
                        "out nulls upstream via "
                        ".filter(t.pkey.is_not_none())."
                    )
    data_format = api.DataFormat(
        format_type="identity",
        key_field_names=key_field_names,
        value_fields=_format_output_value_fields(table),
        table_name=table_name,
        external_diff_column_index=external_diff_column_index,
    )

    datasink_type = "snapshot" if is_snapshot_mode else "sink"
    table.to(
        datasink.GenericDataSink(

View on GitHub (pinned to fa2f74a464)