pathwaycom/pathway · error · ValueError

timestamp_column {timestamp_column!r} is not present in the

Error message

timestamp_column {timestamp_column!r} is not present in the schema

What it means

Raised by pw.io.elasticsearch.read when the timestamp_column argument does not name a column of the provided schema. The connector uses that column to do incremental reads (polling for documents newer than the last seen timestamp), so it must exist in schema.

Source

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

    ...     doc_id: str
    ...     ts: int
    ...     message: str
    >>> table = pw.io.elasticsearch.read(
    ...     host="http://localhost:9200",
    ...     auth=pw.io.elasticsearch.ElasticSearchAuth.basic("admin", "admin"),
    ...     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")

View on GitHub (pinned to fa2f74a464)

Solutions

Set timestamp_column to an existing schema column: pw.io.elasticsearch.read(..., schema=LogSchema, timestamp_column="@timestamp") and add/fix that field in the schema class.
Verify with LogSchema.column_names() that the name matches exactly, including case.
If the index lacks a timestamp field, add one to the schema corresponding to the actual ES mapping field used for sequencing.

Example fix

# before
class LogSchema(pw.Schema):
    @timestamp: int
    level: str
pw.io.elasticsearch.read(..., schema=LogSchema, timestamp_column="ts")

# after
class LogSchema(pw.Schema):
    @timestamp: int
    level: str
pw.io.elasticsearch.read(..., schema=LogSchema, timestamp_column="@timestamp")

Defensive patterns

Strategy: validation

Validate before calling

column_names = schema.column_names()
assert timestamp_column in column_names, (
    f"timestamp_column {timestamp_column!r} not in schema columns {column_names}"
)

Prevention

Derive timestamp_column/id_column from the schema class itself (e.g. a constant next to the schema) instead of free strings.
Add a smoke test that constructs the schema and checks both column names before any ES call.

When it happens

Trigger: pw.io.elasticsearch.read(..., schema=LogSchema, timestamp_column="ts") where LogSchema has no 'ts' field — e.g. a renamed field, a typo, or the schema was written for a different index mapping.

Common situations: Index mappings evolve and the timestamp field is renamed (e.g. '@timestamp' vs 'ts'); schema class copy-pasted from another connector; casing mismatches ('Time' vs 'time').

Related errors

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