pathwaycom/pathway · error · ValueError

The sort_by column {column} doesn't belong to the table pass

Error message

The sort_by column {column} doesn't belong to the table passed to pw.io.kafka.write.

What it means

Pathway's Kafka writer can order rows within a batch via sort_by, but only using columns that belong either to the table passed to pw.io.kafka.write or to the internally built output table. This ValueError is raised when a sort_by ColumnReference comes from a different table (e.g. a filtered, joined, or retyped intermediate). Pathway rejects it early because it cannot map that column onto the data actually being written.

Source

Thrown at python/pathway/io/kafka/__init__.py:769

            sort_by=remapped_sort_by,
        )
    )


def _remap_sort_by(
    sort_by: Iterable[ColumnReference] | None,
    original_table: Table,
    output_table: Table,
) -> list[ColumnReference] | None:
    if sort_by is None:
        return None
    remapped: list[ColumnReference] = []
    for column in sort_by:
        if column._table is output_table:
            remapped.append(column)
            continue
        if column._table is not original_table:
            raise ValueError(
                f"The sort_by column {column} doesn't belong to the table "
                "passed to pw.io.kafka.write."
            )
        if column.name not in output_table._columns:
            raise ValueError(
                f"The sort_by column {column.name!r} is not part of the "
                "data being written. For 'raw' or 'plaintext' format, only "
                "the 'value', 'key', 'topic_name' and 'headers' columns "
                "are forwarded."
            )
        remapped.append(output_table[column.name])
    return remapped


__all__ = [
    "SchemaRegistryHeader",
    "SchemaRegistrySettings",
    "read",

View on GitHub (pinned to fa2f74a464)

Solutions

Pass sort_by references taken from the very same table object you pass as the first argument to pw.io.kafka.write
Assign the final transformed table to a variable (e.g. output = table.filter(...)) and use both output and output.col in the write call
If you need a column from another table, join/rename it into the written table first so all sort_by columns share one table

Example fix

// before
filtered = table.filter(t.value > 0)
pw.io.kafka.write(table, topic, sort_by=[filtered.ts])

// after
filtered = table.filter(t.value > 0)
pw.io.kafka.write(filtered, topic, sort_by=[filtered.ts])

Defensive patterns

Strategy: validation

Validate before calling

def check_sort_by(table, sort_by):
    for c in sort_by:
        assert c._table is table, f"sort_by column {c} is not from the written table"
    return sort_by

Type guard

def columns_belong_to(table: pw.Table, cols: list[pw.ColumnReference]) -> bool:
    return all(c._table is table for c in cols)

Try / catch

try:
    pw.io.kafka.write(table, topic, sort_by=cols)
except ValueError as e:
    if "doesn't belong to the table" in str(e):
        raise  # re-raise with context after logging table/col ids
    raise

Prevention

Keep one variable for the final table and derive both the table argument and all sort_by references from it
Never pass column references captured before a transform alongside the post-transform table
Wrap connector setup in a small helper that asserts col._table is table for every column argument

When it happens

Trigger: Calling pw.io.kafka.write(table, ..., sort_by=[other_table.col]) where other_table is not the exact table object passed as the first argument (nor the internal output table). Typical with pw.Table.filter()/join() results: the connector receives the original table but sort_by references a column of a derived table.

Common situations: Building an output table, then passing an older table variable to write() while reusing column references captured from a later transform; mixing references captured before and after a with_columns() or filter() step.

Related errors

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