{"record":{"id":"ed6c886391303f73","repo":"xai-org/x-algorithm","slug":"parquet-file-path-does-not-contain-min-kafka-tim","errorCode":null,"errorMessage":"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}","messagePattern":"Parquet file (.+?) does not contain min_kafka_timestamp_ms footer metadata\\. Time-range filtering requires this footer key on Hive-partitioned files\\. Error: (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"phoenix/xrex/data/parquet_recsys_metadata.py","lineNumber":78,"sourceCode":"\nclass _FooterTimestamps:\n    def __init__(self, topic_dir: str, min_batch: int, max_batch: int):\n        self._topic_dir = topic_dir\n        self._min_batch = min_batch\n        self._max_batch = max_batch\n\n    def __len__(self) -> int:\n        return self._max_batch - self._min_batch + 1\n\n    def __getitem__(self, idx: int) -> int:\n        bid = self._min_batch + idx\n        path = batch_path(self._topic_dir, 0, bid)\n        try:\n            pf = pq.ParquetFile(path)\n            meta = pf.metadata.metadata or {}\n            return int(meta[b\"min_kafka_timestamp_ms\"])\n        except (KeyError, TypeError, ValueError) as e:\n            raise ValueError(\n                f\"Parquet file {path} does not contain min_kafka_timestamp_ms footer metadata. \"\n                f\"Time-range filtering requires this footer key on \"\n                f\"Hive-partitioned files. Error: {e}\"\n            ) from e\n        except Exception as e:\n            raise ValueError(f\"Cannot read footer metadata from {path}: {e}\") from e\n\n\ndef resolve_time_range(\n    topic_dir: str,\n    min_batch: int,\n    max_batch: int,\n    min_timestamp_ms: int | None,\n    max_timestamp_ms: int | None,\n) -> tuple[int, int]:\n    ts = _FooterTimestamps(topic_dir, min_batch, max_batch)\n    n = len(ts)\n","sourceCodeStart":60,"sourceCodeEnd":96,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/phoenix/xrex/data/parquet_recsys_metadata.py#L60-L96","documentation":"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.","triggerScenarios":"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.","commonSituations":"Older data written before footer timestamps were added; third-party re-encoding (spark rewrite, parquet-tools copy) dropping KV metadata; mixed-version topics.","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."],"exampleFix":"# before (file lacks footer key)\n meta = ds._meta[batch_id]  # ValueError: no min_kafka_timestamp_ms\n\n# after: rewrite file with footer metadata\nimport pyarrow as pq\nt = pq.read_table(p)\npq.write_table(t, p, use_compliant_nested_type=True,\n  metadata={'min_kafka_timestamp_ms': b'1700000000000'})","handlingStrategy":"validation","validationCode":"pf = pq.ParquetFile(path)\nmeta = pf.metadata.metadata or {}\nhas_key = b'min_kafka_timestamp_ms' in meta\nif not has_key:\n    mark_partition_legacy(path)  # exclude from timestamp filtering","typeGuard":"def has_kafka_footer(path) -> bool:\n    m = pq.ParquetFile(path).metadata.metadata or {}\n    return b'min_kafka_timestamp_ms' in m","tryCatchPattern":"try:\n    bid = meta_index[i]\nexcept ValueError as e:\n    if 'min_kafka_timestamp_ms' in str(e):\n        raise LegacyDataError('rewrite files with kafka footer metadata') from e\n    raise","preventionTips":["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."],"tags":["parquet","footer-metadata","time-range","schema-validation"],"backgroundTag":"missing-metadata-key","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}