pathwaycom/pathway · error · ValueError

Defining a primary key in the schema is not supported for pw

Error message

Defining a primary key in the schema is not supported for pw.io.elasticsearch.read. The connector keys the resulting table by id_column. If you need to reindex the table by a different column, use pw.Table.with_id_from() after reading.

What it means

Raised by pw.io.elasticsearch.read when the provided pw.Schema defines a primary key. The connector itself keys the resulting table by id_column; a schema-level primary key conflicts with that and is explicitly unsupported. To re-key the table, use pw.Table.with_id_from() after reading.

Source

Thrown at python/pathway/io/elasticsearch/__init__.py:318

    ...     index_name="logs",
    ...     schema=LogSchema,
    ...     timestamp_column="ts",
    ...     id_column="doc_id",
    ...     max_transaction_duration=datetime.timedelta(minutes=5),
    ... )
    """

    _check_entitlements("elasticsearch")

    column_names = schema.column_names()
    if timestamp_column not in column_names:
        raise ValueError(
            f"timestamp_column {timestamp_column!r} is not present in the schema"
        )
    if id_column not in column_names:
        raise ValueError(f"id_column {id_column!r} is not present in the schema")
    if schema.primary_key_columns():
        raise ValueError(
            "Defining a primary key in the schema is not supported for "
            "pw.io.elasticsearch.read. The connector keys the resulting table by "
            "id_column. If you need to reindex the table by a different column, use "
            "pw.Table.with_id_from() after reading."
        )

    max_transaction_duration_ms = round(
        as_duration_seconds(max_transaction_duration, "max_transaction_duration") * 1000
    )
    poll_interval_ms = round(as_duration_seconds(poll_interval, "poll_interval") * 1000)
    if read_batch_size <= 0:
        raise ValueError("read_batch_size must be positive")

    data_storage = api.DataStorage(
        storage_type="elasticsearch",
        mode=internal_connector_mode(mode),
        elasticsearch_params=api.ElasticSearchParams(
            host=host,

View on GitHub (pinned to fa2f74a464)

Solutions

1. Remove primary_key=True from the schema class passed to pw.io.elasticsearch.read; let the connector key the table by id_column.
2. If you need a different key, read first and then re-key: table = table.with_id_from(table.some_col).

Example fix

# before
class DocSchema(pw.Schema):
    doc_id: str = pw.column_definition(primary_key=True)
    text: str
pw.io.elasticsearch.read(..., schema=DocSchema, id_column="doc_id")

# after
class DocSchema(pw.Schema):
    doc_id: str
    text: str
table = pw.io.elasticsearch.read(..., schema=DocSchema, id_column="doc_id")

assert not schema.primary_key_columns(), (
    "elasticsearch schemas must not define primary keys; use with_id_from() after read"
)

When it happens

Trigger: pw.io.elasticsearch.read(..., schema=class S(pw.Schema): doc_id: str = pw.column_definition(primary_key=True)) — any schema passed with primary_key=True on one or more columns.

Common situations: Reusing a schema class from pw.io.csv.read or pw.io.json.read where primary_key=True is idiomatic; habitually marking id columns as primary keys in every schema.

Related errors

• timestamp_column {timestamp_column!r} is not present in the
• id_column {id_column!r} is not present in the schema
• 'json_field_paths' references field {field_name!r} which is
• pw.Schema has column names that differ only in case ({case_c
• Column(s) {collisions} collide with the 'time' and 'diff' me

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