3. Ingesting CAE result files at scale

CAE post-processing pipelines rarely deal with a single tidy file. A typical crash or CFD campaign produces hundreds to thousands of result artifacts: per-iteration CSV exports from a solver's text writer, Parquet dumps from a co-simulation bridge, and Arrow IPC streams emitted by an in-memory solver plugin. The ingestion layer must accept all three formats, avoid loading gigabytes into RAM, and preserve enough schema fidelity that downstream aggregation (peak stress, drag coefficient, energy balance) stays numerically trustworthy. Polars provides three distinct entry points that map cleanly onto these formats: pl.scan_csv, pl.scan_parquet, and pl.scan_ipc, all of which return a LazyFrame rather than materialized data.

Why lazy scanning is the default, not an optimization

A LazyFrame is a query plan, not a dataset. When you call pl.scan_parquet("results/*.parquet"), Polars records the file glob, the projected columns, and any filter predicates, but reads nothing until you invoke .collect(). At collection time the optimizer pushes column selection and predicate filters down to the Parquet reader, so a query that touches 4 of 60 columns reads only those column chunks. For a 40 GB transient result set this is the difference between a 90-second scan and an out-of-memory crash.

The same principle applies to CSV, though with an important caveat: CSV has no columnar storage, so predicate pushdown cannot skip bytes. scan_csv still avoids materializing the full table, and it can stream row batches, but every column in the file must be parsed to evaluate a filter. This is why converting solver CSV output to Parquet once, then scanning the Parquet thereafter, is the single highest-leverage change in most CAE ingestion pipelines.

Reading solver CSV without choking on it

Solver CSV exports are frequently hostile: comment headers prefixed with #, inconsistent delimiters, and a metadata block before the real column names. Polars handles these with explicit arguments rather than guesswork.

import polars as pl

lf = pl.scan_csv(
    "run_042/forces_*.csv",
    separator=",",
    comment_prefix="#",
    has_header=True,
    schema_overrides={
        "time": pl.Float64,
        "cd": pl.Float32,
        "cl": pl.Float32,
        "cm": pl.Float32,
    },
    null_values=["NA", "-", ""],
    try_parse_dates=False,
)

peak = (
    lf.filter(pl.col("time") > 0.5)
      .group_by("time")
      .agg(pl.col("cd").mean().alias("cd_mean"))
      .collect(streaming=True)
)

Two arguments carry most of the weight. schema_overrides prevents Polars from inferring a column as Int64 when the first 1000 rows happen to be integral, then failing when a float appears at row 5000. comment_prefix strips the solver's provenance banner. The streaming=True flag on collect processes the query in chunks, keeping peak memory proportional to batch size rather than file size.

Parquet and IPC: columnar and streaming-native

Parquet is the natural interchange format for CAE results because it stores column statistics (min/max per row group) that the optimizer uses to skip entire row groups. A filter like pl.col("von_mises") > 250e6 on a Parquet file with 200 row groups may read only the 12 groups whose statistics overlap the threshold.

stress = (
    pl.scan_parquet("stress/*.parquet")
      .select(["node_id", "element_id", "von_mises", "time"])
      .filter(pl.col("von_mises") > 250e6)
      .group_by("element_id")
      .agg(pl.col("von_mises").max().alias("peak_vm"))
      .collect(streaming=True)
)

Arrow IPC is the right choice when the solver writes results incrementally during a run — for example, a Python co-simulation plugin that flushes a record batch every 100 timesteps. IPC files are memory-mappable and preserve the exact Arrow schema, so no type inference occurs at read time. pl.scan_ipc("live/*.arrow") reads them with the same lazy semantics as Parquet.

The table below summarizes when each format is appropriate and what the reader can and cannot skip.

Streaming large files without materializing them

When a single Parquet file exceeds available RAM, collect(streaming=True) is necessary but not always sufficient. The streaming engine processes the query plan in batches, but certain operations — global sorts, some joins, and group_by with high cardinality — still require buffering. For those cases, use sink_parquet to write results directly to disk without ever holding the full output in memory:

(
    pl.scan_parquet("transient/*.parquet")
      .filter(pl.col("time").is_between(0.0, 2.0))
      .with_columns((pl.col("pressure") * pl.col("area")).alias("force"))
      .sink_parquet("force_window.parquet")
)

For genuinely unbounded inputs — a solver still writing IPC batches while you read — combine scan_ipc with a polling loop that re-globs the directory and collects only new files. Polars does not watch directories natively, so the loop is application-level, but each iteration remains lazy and cheap.

Schema drift and failure symptoms

The most common ingestion failure in CAE pipelines is silent schema drift: a solver update renames von_mises to vonMises, or changes a column from Float32 to Float64. With CSV, this manifests as a ComputeError during collection or, worse, a column of nulls. With Parquet, a renamed column simply disappears from the projection and downstream aggregations return wrong results without error.

Guard against this by asserting the schema explicitly after the first scan:

expected = {"node_id": pl.Int64, "von_mises": pl.Float64, "time": pl.Float64}
actual = pl.scan_parquet("stress/*.parquet").collect_schema()
missing = set(expected) - set(actual.names())
if missing:
    raise ValueError(f"Schema drift detected, missing: {missing}")

Typical symptoms and their causes: a collect that succeeds but returns zero rows usually means a predicate was pushed down against a column with mismatched units (Pa vs MPa); a MemoryError during collect despite streaming=True usually indicates a high-cardinality group_by that the streaming engine cannot partition; and a scan that reads far more bytes than expected points to a missing column projection, forcing the reader to materialize all columns.

Putting it together: a reproducible ingestion check

To verify an ingestion pipeline against a real CAE result set, run the following sequence and inspect each observable output. Assume a directory run_042/ containing forces_*.csv (solver text export) and stress/*.parquet (post-processing archive).

1. Scan the CSV with explicit schema_overrides and comment_prefix="#". Collect only head(5) and confirm column names and dtypes match the solver's documented output.
2. Convert the CSV to Parquet once with sink_parquet. Compare file sizes: a 2 GB CSV typically becomes 300–500 MB Parquet with Snappy compression.
3. Scan the Parquet with a predicate on time and a projection of 3 columns. Use collect(streaming=True) and observe peak RSS — it should stay well below the file size.
4. Run collect_schema() on both the CSV-derived and Parquet-derived LazyFrames and assert they are identical. Any mismatch is schema drift.
5. Introduce a deliberate fault — rename von_mises to vonMises in one Parquet file — and confirm the schema guard raises before any aggregation runs.

The expected outcome is a pipeline that reads only the columns and row groups it needs, never materializes more than one batch at a time, and fails loudly on schema drift rather than silently returning wrong numbers. If step 3 shows memory growing linearly with file size, the streaming engine is not partitioning the query — check for a global sort or a group_by on a near-unique key such as node_id, and consider pre-aggregating or using a two-pass approach.