freshdata stream output breaks when chunk dtypes change: misaligned CSV, partial Parquet #248

Summary

freshdata stream writes each cleaned chunk as it arrives, and assumes every chunk has the first chunk's columns and dtypes. When a later chunk differs, the output is wrong in two ways:

• CSV output, time-series anomaly mode. A chunk where a numeric column holds a stray string emits no <col>_anomaly column. The CSV is appended positionally under batch 1's header, so the remaining flags shift under the wrong column. The command still exits 0.
• Parquet output. A chunk where an integer column gains its first missing value (int64 to double) makes ParquetWriter raise mid-stream. The command exits 1 and leaves a valid Parquet file holding only the rows written so far.

Reproduction

from freshdata.enterprise.cli import main   # the `freshdata` console script

# CSV: column 'a' has a stray string in the second chunk
open("in.csv", "w").write("ts,a,b\n2024-01-01,0,0\n2024-01-02,1,10\n2024-01-03,2,20\n"
                          "2024-01-04,oops,30\n2024-01-05,4,40\n2024-01-06,5,50\n")
main(["stream", "in.csv", "-o", "out.csv", "--batch-size", "3", "--quiet",
      "--timestamp", "ts", "--anomaly", "mad"])
print(open("out.csv").read())

# Parquet: 'v' gains a missing value in the second chunk
open("in2.csv", "w").write("id,v\n1,1\n2,2\n3,\n4,4\n")
rc = main(["stream", "in2.csv", "-o", "out.parquet", "--batch-size", "2", "--quiet"])
import pandas as pd; print(rc, len(pd.read_parquet("out.parquet")))

The equivalent shell commands are freshdata stream in.csv -o out.csv --batch-size 3 --timestamp ts --anomaly mad and freshdata stream in2.csv -o out.parquet --batch-size 2.

Expected

• CSV: every row has a value under each header, and b's flags stay under b_anomaly.
• Parquet: exit 0 with all 4 rows, or no output file left behind on failure.

Actual

ts,a,b,a_anomaly,b_anomaly
2024-01-01,0,0,False,False
2024-01-02,1,10,False,False
2024-01-03,2,20,False,False
2024-01-04,oops,30,False
2024-01-05,4,40,False
2024-01-06,5,50,False

Exit code 0. Rows 4-6 carry b's flags under a_anomaly, and b_anomaly is empty.

freshdata: error: Table schema does not match schema used to create file:
table:
id: int64
v: double
...
file:
id: int64
v: int64

Exit code 1. out.parquet exists with 2 of 4 rows.

Environment

freshdata 2.0.0 @ 55a8044. Reproduces on Python 3.9.6 / pandas 1.5.3 / numpy 1.26.4 and on Python 3.12.14 / pandas 2.3.3 / numpy 2.5.3.
Re-checked on main c87efbd (2026-09-15): still reproduces.

Where

src/freshdata/streaming/_cli.py:53-62 (_BatchWriter.write):

table = pa.Table.from_pandas(df, preserve_index=False)
if self._pq_writer is None:
    self._pq_writer = pq.ParquetWriter(self.path, table.schema)
self._pq_writer.write_table(table)
...
df.to_csv(self.path, mode="w" if self._csv_header else "a",
          header=self._csv_header, index=False)

The set of anomaly target columns is re-derived per batch in src/freshdata/streaming/_timeseries.py:596 (if not pd.api.types.is_numeric_dtype(df[c]): continue).

Suggested fix

• Record the first batch's column list. Reindex later batches to it (missing flag columns become empty) and fail loudly on new, unexpected columns.
• For Parquet, cast each table to a unified schema (e.g. promote int64 to double, or declare nullable types up front). Write to a temporary path and rename on success, and close the writer in a finally.
• Decide anomaly target columns once, from locked roles, instead of per batch.