xai-org/x-algorithm · error · ValueError

Parquet file {path} does not contain min_kafka_timestamp_ms

Error message

Parquet file {path} does not contain min_kafka_timestamp_ms footer metadata. Time-range filtering requires this footer key on Hive-partitioned files. Error: {e}

What it means

The valid-batches metadata class reads the 'min_kafka_timestamp_ms' key from each parquet file's footer metadata to binary-search batch ids for a requested time range. If the footer lacks that key (KeyError), or the value is not an int-castable bytes entry (TypeError/ValueError), it raises ValueError explaining that time-range filtering requires this footer key on Hive-partitioned files.

Source

Thrown at phoenix/xrex/data/parquet_recsys_metadata.py:78

class _FooterTimestamps:
    def __init__(self, topic_dir: str, min_batch: int, max_batch: int):
        self._topic_dir = topic_dir
        self._min_batch = min_batch
        self._max_batch = max_batch

    def __len__(self) -> int:
        return self._max_batch - self._min_batch + 1

    def __getitem__(self, idx: int) -> int:
        bid = self._min_batch + idx
        path = batch_path(self._topic_dir, 0, bid)
        try:
            pf = pq.ParquetFile(path)
            meta = pf.metadata.metadata or {}
            return int(meta[b"min_kafka_timestamp_ms"])
        except (KeyError, TypeError, ValueError) as e:
            raise ValueError(
                f"Parquet file {path} does not contain min_kafka_timestamp_ms footer metadata. "
                f"Time-range filtering requires this footer key on "
                f"Hive-partitioned files. Error: {e}"
            ) from e
        except Exception as e:
            raise ValueError(f"Cannot read footer metadata from {path}: {e}") from e


def resolve_time_range(
    topic_dir: str,
    min_batch: int,
    max_batch: int,
    min_timestamp_ms: int | None,
    max_timestamp_ms: int | None,
) -> tuple[int, int]:
    ts = _FooterTimestamps(topic_dir, min_batch, max_batch)
    n = len(ts)

View on GitHub (pinned to 24c60942c5)

Solutions

Restrict the time range to partitions whose files contain the footer key, or regenerate footers by rewriting files with min_kafka_timestamp_ms set.
Upgrade the producer to the version that stamps kafka timestamps into footers for future data.
Use date_range filtering on partition paths instead of timestamp_ms ranges for legacy data.

Example fix

# before (file lacks footer key)
 meta = ds._meta[batch_id]  # ValueError: no min_kafka_timestamp_ms

# after: rewrite file with footer metadata
import pyarrow as pq
t = pq.read_table(p)
pq.write_table(t, p, use_compliant_nested_type=True,
  metadata={'min_kafka_timestamp_ms': b'1700000000000'})

Defensive patterns

Strategy: validation

Validate before calling

pf = pq.ParquetFile(path)
meta = pf.metadata.metadata or {}
has_key = b'min_kafka_timestamp_ms' in meta
if not has_key:
    mark_partition_legacy(path)  # exclude from timestamp filtering

Type guard

def has_kafka_footer(path) -> bool:
    m = pq.ParquetFile(path).metadata.metadata or {}
    return b'min_kafka_timestamp_ms' in m

Try / catch

try:
    bid = meta_index[i]
except ValueError as e:
    if 'min_kafka_timestamp_ms' in str(e):
        raise LegacyDataError('rewrite files with kafka footer metadata') from e
    raise

Prevention

Stamp min_kafka_timestamp_ms into footers at write time in the producer.
Avoid rewriting parquet with tools that strip KV metadata; verify after re-encoding.

When it happens

Trigger: Calling __getitem__ (via _resolve_time_range with min/max_timestamp_ms) on files written by a producer that did not set min_kafka_timestamp_ms in parquet key-value metadata; files copied/rewritten by a tool that strips footer metadata.

Common situations: Older data written before footer timestamps were added; third-party re-encoding (spark rewrite, parquet-tools copy) dropping KV metadata; mixed-version topics.

Related errors

AI-assisted analysis of xai-org/x-algorithm@24c60942c5 (2026-08-28). Data as JSON: /api/errors/ed6c886391303f73. Report an issue: GitHub.