xai-org/x-algorithm · error · ValueError

Error processing file {file}: {e}

Error message

Error processing file {file}: {e}

What it means

While building the batch pool, _impl() constructs an inner dataset for each file; any exception other than a plain missing-file ('No such file or directory', which is skipped with a warning) is re-raised as ValueError wrapping the original exception with the file name. This is a generic wrapper so the failing file is identifiable in multi-file loads.

Source

Thrown at phoenix/xrex/data/parquet_recsys.py:505

    def _open_file(self, file: str, pool: ThreadPoolExecutor, active: deque) -> None:
        rank_logger.info(f"Worker {self._shard_index}/{self._num_shards} opening file {file}")
        path = _resolve_file_path(self._path, file)

        def _impl():
            try:
                pf = ParquetFile(path)
                return LazyRecordBatchIterator(
                    pf,
                    self._batch_size,
                    path,
                    self._conversion_delay_columns,
                    self._include_action_delay_columns,
                )
            except Exception as e:
                if "No such file or directory" in str(e):
                    rank_logger.warning(f"Skipping missing file {file}")
                    return None
                raise ValueError(f"Error processing file {file}: {e}") from e

        active.append(pool.submit(_impl))

    @staticmethod
    def _safe_read(
        holder: "LazyRecordBatchIterator",
    ) -> pa.RecordBatch | None:
        try:
            return holder.read()
        except StopIteration:
            return None

    def _drain_files(
        self,
        files: list[str],
        *,
        pool: ThreadPoolExecutor,
        prefetched: deque[Future[LazyRecordBatchIterator | None]] | None = None,

View on GitHub (pinned to 24c60942c5)

Solutions

1. Read the wrapped exception text after 'Error processing file X' to identify the root cause, then inspect/repair that specific file.
2. Remove the bad file from the index/metadata and retry if the data is expendable.
3. If it is a transient storage error, retry the job after the storage issue clears.

Example fix

# before: opaque failure mid-epoch
Error processing file /data/date=.../batch-042.parquet: ...

# after: triage
pq.ParquetFile('/data/date=.../batch-042.parquet').metadata  # reproduce, then repair or drop from index

for f in files:
    if not os.path.isfile(f):
        continue
    pq.ParquetFile(f).metadata  # cheap footer probe; surfaces corrupt files early

try:
    batch = read_one(file)
except ValueError as e:
    if 'Error processing file' in str(e):
        logger.error('quarantining %s', file); dead_letters.append(file); return None
    raise

When it happens

Trigger: Any parquet read error inside per-file dataset construction: corrupted footer, post_sid invariant violation (391), sidecar mismatch, permission error, or bad schema — anything not matching the missing-file message.

Common situations: One corrupted file among thousands in an index; Opaque OSError text from storage layers that hides the file context without this wrapper.

Related errors

• no cached cls parquet under {split_dir}
• not a parquet path: {batch_file_path}
• sidecar {sidecar_path} missing column(s) {missing}; availabl
• post_sid schema invariant violated in {file_path}: expected
• Parquet file {path} does not contain min_kafka_timestamp_ms

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