feat(streamer): align CSV columns by header name for append/upsert (UNTESTED on Composer)

[demenech]

⚠️ NOT TESTED end-to-end. The CSV-streaming logic was validated locally with stubbed
deps (reordered / added / omitted / sanitized-header cases), but it has not run through a
real dev-Composer → BigQuery load. Please validate on Composer before merging (steps below).
Branched off master (the line synced to the dev Composer bucket). ⚠️ master has since
advanced (e.g. the parquet-format commit); please confirm this rebases onto whatever is
actually deployed in the Composer DAG bucket, since aircan is bucket-synced, not pinned.

The issue

The CSV streamer preserves the source file's column order, and BigQuery then loads it
positionally against the schema (BQ maps CSV columns by ordinal, not header name). So an
append/upsert whose file has reordered, added, or omitted columns shifts data into the
wrong column — or fails the load. This blocks non-destructive column changes on append/upsert:

• a new column can't be added unless it's strictly last and the file keeps every existing
  column in its exact order;
• a reorder with type-compatible columns loads silently into the wrong columns (data corruption).

ALLOW_FIELD_ADDITION is already set on append loads, so BigQuery can add a new column — but
only if the streamed CSV is aligned to the schema by name first. That alignment is what's
missing.

The proposed fix

Add a field_order option to the streamer. When set, each output row is emitted in schema
field order, pulling each column from the source by sanitized header name:

• column missing from the source → emitted empty → NULL in BigQuery;
• extra source columns → dropped;
• reordered source columns → placed correctly by name.

date_formats stays valid because it's keyed by the descriptor field index, which equals the
output order. When field_order is None, the legacy positional behaviour is unchanged. The
DAG passes the descriptor's field order.

Header names are sanitized the same way the schema descriptor's field names are (BigQuery-safe),
so e.g. Price (GBP) in the file matches the Price_GBP schema field.

Files: aircan/dependencies/cloud/storage.py (_stream_csv), aircan/dags/pipeline_ckan_to_bigquery.py (passes field_order).

Why (frontend context)

The dx-helm-neso frontend now merges the uploaded file's schema with the stored one on
append/upsert (existing columns kept + locked, new columns appended). Today the FE blocks
reorder / mid-insert / missing because the deployed positional load can't handle them safely.
This change makes those layouts load correctly by name; once it's deployed, the FE flips its
capability flags and stops blocking them (no UI rewrite).

How to validate (dev Composer → BigQuery)

Against an existing datastore table, run append/upsert with files that:

1. Reorder existing columns → rows land in the correct columns (not shifted).
2. Add a new column (schema includes it) → column added, existing rows NULL for it.
3. Omit a column → that column NULL-filled for the appended rows, others intact.
4. Have sanitized-differing headers (Price (GBP) vs Price_GBP) → matched by name.
   Confirm row counts and per-column values in BigQuery after each.

Not covered here

The create-with-upload double-load is a separate issue (ckanext-aircan notify dedupe +
frontend create-collapse) — not this PR.

[coderabbitai]

Important

Review skipped

Draft detected.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 155d4a15-732a-495f-af04-2406b0caac26

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/align-csv-columns-by-name

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}