Implement Cluster 2 spark.write fidelity

README.md

@@ -82,6 +82,30 @@ if country != "ALL":
 df.write.mode("overwrite").saveAsTable("silver.customers_enriched")
 ```
 
+## Spark write modes and schema options
+
+Table writes (`saveAsTable` / `insertInto`) stay CSV-backed. File writes (`csv` / `parquet` / `json` / `save`) use native Spark under `base_path`. `format("delta").save(path)` is stored as parquet (no Delta log).
+
+| Write | Missing table | Existing table |
+|---|---|---|
+| default / `overwrite` | create | replace rows |
+| `append` | create | append rows (exact column set, unless `mergeSchema`) |
+| `error` / `errorIfExists` | create | raise `AnalysisException` |
+| `ignore` | create | no-op (file and temp view unchanged) |
+| `insertInto` | raise `AnalysisException` | append, or replace when `overwrite=True` / `mode("overwrite")` |
+
+Schema flags:
+
+| Option | Effect |
+|---|---|
+| `overwriteSchema=true` + overwrite | replace the CSV even when columns change |
+| `overwriteSchema=false` (default) + overwrite | raise `SchemaMismatchError` on incompatible schema change |
+| `mergeSchema=true` + append | union missing columns with nulls |
+| `mergeSchema=false` (default) + append | raise `SchemaMismatchError` if columns differ |
+| same column, incompatible types | always raise `SchemaMismatchError` (merge only adds columns) |
+
+`partitionBy` columns must exist on the DataFrame. `option("replaceWhere", "<predicate>")` with `mode("overwrite")` deletes matching stored rows then appends the new frame. `bucketBy` / `sortBy` are accepted no-ops (bucketing is not simulated). `df.writeTo(table).using(...).create()` / `.replace()` / `.append()` maps onto the same table writer; `createOrReplace` and `overwritePartitions` raise `NotImplementedError` with a `saveAsTable` hint.
+
 ## Key Modules
 1. `SparkProxy` - A Spark proxy that manipulates incoming Delta table reads and writes and redirects them to interactions with CSV files stored locally
 2. `LocalWorkflowRunner` - A notebook orchestrator that takes the notebook .py files as defined in a Databricks Workflow JSON file and executes them as per the DAG definition. Databricks comment magics `%run` and `%sh` (`# %sh` / `# MAGIC %sh`, including `%sh -e`) work in those notebooks.

docs/superpowers/specs/2026-09-03-spark-write-lakeflow-roadmap-design.md

@@ -42,14 +42,14 @@ Storage stays `{base_path}/{schema}/{table}.csv` + temp views `{schema}_{table}`
 
 ### Cluster 2 — `spark.write` fidelity (in order)
 
-- [ ] **W1. Mode fidelity: `error` / `errorIfExists` + `ignore`** — `saveAsTable` with `error`/`errorIfExists` raises when CSV exists; `ignore` silently skips the write (no file touch, no view refresh). Few lines in `save_dataframe`; completes the Spark save-mode truth table alongside existing overwrite/append/default-overwrite.
-- [ ] **W2. `insertInto(table, overwrite=False)`** — DataFrame API twin of append/overwrite `saveAsTable`; honor writer `mode` (`append` vs `overwrite`); require the table (CSV) to exist and raise a Spark-like AnalysisException message when missing; accept the `overwrite=True` kwarg for full-refresh semantics.
-- [ ] **W3. `partitionBy` validation + `replaceWhere` dynamic overwrite** — validate `partitionBy` cols exist in the DataFrame (raise, don't silently ignore typos); support `.option("replaceWhere", "<predicate>")` with `mode("overwrite")` as overwrite-where-predicate: delete matching rows from stored CSV via pandas query, append the new frame. Full partition-overwrite layout stays out of scope (still one CSV).
-- [ ] **W4. CSV write options** — honor `delimiter`/`sep`, `quote`, `escape`, `nullValue`, `dateFormat`, `timestampFormat` on both `saveAsTable` (persist options per table for round-trip reads) and `csv(path)` passthrough; store the effective options so `read.table` round-trips without callers repeating them.
-- [ ] **W5. File-write dispatch: `parquet` / `json` / `save` + `format().save()`** — route `df.write.parquet(path)` / `.json(path)` / `.save(path)` and `.format("parquet"|"json"|"csv"|"delta").save(path)` under `base_path` using native Spark writers; `format("delta").save(path)` maps to parquet-on-disk (documented, no Delta log). Table APIs stay CSV; file APIs use real formats.
-- [ ] **W6. `overwriteSchema` / `mergeSchema` truth table** — `overwriteSchema=true` + overwrite = replace file even on schema change (already true, add tests); `mergeSchema=true` + append = union missing columns with nulls via pandas instead of raising `SchemaMismatchError`; `overwriteSchema=false` + incompatible change = raise. Document the matrix in README.
-- [ ] **W7. `bucketBy` / `sortBy` accepted no-ops** — accept and ignore (like `partitionBy` today) with an explicit log/docstring that bucketing/sorting is not simulated; prevents AttributeError on production chains that call `.bucketBy(n, col).sortBy(col)`.
-- [ ] **W8. `writeTo` (DataFrameWriterV2) decision** — either a minimal `writeTo(table).using(...).partitionedBy(...).option(...).create()/replace()/append()` façade over `save_dataframe`, or an explicit `NotImplementedError` with a migration hint to `saveAsTable`. Decide once real notebooks show which V2 verbs appear; do not build full V2 (overwritePartitions, createOrReplace) preemptively.
+- [x] **W1. Mode fidelity: `error` / `errorIfExists` + `ignore`** — `saveAsTable` with `error`/`errorIfExists` raises when CSV exists; `ignore` silently skips the write (no file touch, no view refresh). Few lines in `save_dataframe`; completes the Spark save-mode truth table alongside existing overwrite/append/default-overwrite.
+- [x] **W2. `insertInto(table, overwrite=False)`** — DataFrame API twin of append/overwrite `saveAsTable`; honor writer `mode` (`append` vs `overwrite`); require the table (CSV) to exist and raise a Spark-like AnalysisException message when missing; accept the `overwrite=True` kwarg for full-refresh semantics.
+- [x] **W3. `partitionBy` validation + `replaceWhere` dynamic overwrite** — validate `partitionBy` cols exist in the DataFrame (raise, don't silently ignore typos); support `.option("replaceWhere", "<predicate>")` with `mode("overwrite")` as overwrite-where-predicate: delete matching rows from stored CSV via pandas query, append the new frame. Full partition-overwrite layout stays out of scope (still one CSV).
+- [x] **W4. CSV write options** — honor `delimiter`/`sep`, `quote`, `escape`, `nullValue`, `dateFormat`, `timestampFormat` on both `saveAsTable` (persist options per table for round-trip reads) and `csv(path)` passthrough; store the effective options so `read.table` round-trips without callers repeating them.
+- [x] **W5. File-write dispatch: `parquet` / `json` / `save` + `format().save()`** — route `df.write.parquet(path)` / `.json(path)` / `.save(path)` and `.format("parquet"|"json"|"csv"|"delta").save(path)` under `base_path` using native Spark writers; `format("delta").save(path)` maps to parquet-on-disk (documented, no Delta log). Table APIs stay CSV; file APIs use real formats.
+- [x] **W6. `overwriteSchema` / `mergeSchema` truth table** — `overwriteSchema=true` + overwrite = replace file even on schema change (already true, add tests); `mergeSchema=true` + append = union missing columns with nulls via pandas instead of raising `SchemaMismatchError`; `overwriteSchema=false` + incompatible change = raise. Document the matrix in README.
+- [x] **W7. `bucketBy` / `sortBy` accepted no-ops** — accept and ignore (like `partitionBy` today) with an explicit log/docstring that bucketing/sorting is not simulated; prevents AttributeError on production chains that call `.bucketBy(n, col).sortBy(col)`.
+- [x] **W8. `writeTo` (DataFrameWriterV2) decision** — either a minimal `writeTo(table).using(...).partitionedBy(...).option(...).create()/replace()/append()` façade over `save_dataframe`, or an explicit `NotImplementedError` with a migration hint to `saveAsTable`. Decide once real notebooks show which V2 verbs appear; do not build full V2 (overwritePartitions, createOrReplace) preemptively.
 
 ### Cluster 3 — missing `dbutils` APIs (in order)