fix(ingestion): write the raw zone as parquet

automated_ingestion/sources/san_acacia/dlt_pipeline.py

@@ -59,6 +59,17 @@
 than SO-0125.
 """
 
+LOADER_FILE_FORMAT = "parquet"
+"""Raw-zone file format.
+
+dlt writes gzipped JSONL unless told otherwise, and the first live run landed
+that way. Parquet is what Mode B replay assumes: replay reads the raw zone
+filtered on event time, and a columnar format with real types lets that read a
+window without decompressing and parsing every record. It also preserves the
+distinction between a null and a missing field, which JSONL round-trips less
+reliably.
+"""
+
 INITIAL_START = "2015-01-01T00:00:00+00:00"
 """Floor for a point that has never been ingested.
 

automated_ingestion/sources/san_acacia/ingest.py

@@ -43,6 +43,7 @@ def _client() -> DiverHubClient:
 def raw_san_acacia_locations(context: AssetExecutionContext) -> Output[int]:
     """Land the point roster in the raw zone."""
     from automated_ingestion.sources.san_acacia.dlt_pipeline import (
+        LOADER_FILE_FORMAT,
         PROJECT_ID,
         build_pipeline,
         vanessen_locations,
@@ -51,7 +52,7 @@ def raw_san_acacia_locations(context: AssetExecutionContext) -> Output[int]:
     client = _client()
     points = list(client.monitoring_points(PROJECT_ID))
     pipeline = build_pipeline()
-    pipeline.run(vanessen_locations(client))
+    pipeline.run(vanessen_locations(client), loader_file_format=LOADER_FILE_FORMAT)
 
     context.log.info("landed %s monitoring points", len(points))
     return Output(
@@ -72,6 +73,7 @@ def raw_san_acacia_locations(context: AssetExecutionContext) -> Output[int]:
 def raw_san_acacia_readings(context: AssetExecutionContext) -> Output[int]:
     """Land water levels for every point, isolating per-point failure."""
     from automated_ingestion.sources.san_acacia.dlt_pipeline import (
+        LOADER_FILE_FORMAT,
         PROJECT_ID,
         build_pipeline,
         vanessen_readings,
@@ -86,7 +88,10 @@ def raw_san_acacia_readings(context: AssetExecutionContext) -> Output[int]:
 
     pipeline = build_pipeline()
     failures: list[dict[str, Any]] = []
-    info = pipeline.run(vanessen_readings(client, points, end, failures))
+    info = pipeline.run(
+        vanessen_readings(client, points, end, failures),
+        loader_file_format=LOADER_FILE_FORMAT,
+    )
     rows = _row_count(info)
 
     if failures:

docs/sources/san_acacia.md

@@ -77,6 +77,22 @@ halve-on-500 recovery is **not** a routine path for `WaterLevels`. Do not
 assume a 500 there means "too much data" without re-measuring; on `DiverData`
 that assumption is provably wrong.
 
+## Raw zone format
+
+Parquet, date-partitioned:
+
+```
+raw_sanacaciareach/vanessen_readings/year=2026/month=08/day=19/<load_id>.<file_id>.parquet
+```
+
+dlt writes gzipped JSONL unless told otherwise, and the first live run landed
+that way before this was set. Parquet is what Mode B replay assumes: replay
+reads the raw zone filtered on event time, and a columnar format with real types
+lets it read a window without decompressing and parsing every record.
+
+Objects written before this change are `.jsonl.gz`. dlt reads both, so they do
+not need migrating, but a replay spanning that boundary reads two formats.
+
 ## Field mapping
 
 ### Water levels — the ingested series