feat(harness): python node write modes (append, upsert)

CHANGELOG.md

@@ -45,6 +45,14 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   `table_columns` reads `information_schema`. `ensure_table` accepts an
   upsert key (no constraint DDL) and rejects it with `format_version: 1`.
   `load` now shares the stage helpers (no behavior change).
+- The harness writes each node with its `mode_type`: `load` for `replace`,
+  `append`, or `upsert` with the node's `primary_key` (also passed to
+  `ensure_table`). An adapter without the needed capability fails with
+  `LoadError: <Adapter> does not support <mode>` before the script runs; an
+  `upsert` whose rows repeat a key fails with `ConformError` and writes
+  nothing. Before an `append` or `upsert` the live table's columns and
+  nullability are compared with `output_columns`; any difference is a
+  `LoadError` naming each one. See "Write modes" in the README.
 
 ### Changed
 

README.md

@@ -53,7 +53,8 @@ engine adapters, the validation runner, and the node harness. The former
 `continuo-validation` repository was merged in — there is no longer a separate
 validation-side port, adapter class, entry-point group, or image. One
 `WarehouseAdapter` per engine serves both the data plane (`fetch` /
-`ensure_table` / `load`) and validation (`ensure_schema` / `drop_schema` /
+`ensure_table` / `load`, plus the opt-in `table_columns` / `append` / `upsert`
+capabilities) and validation (`ensure_schema` / `drop_schema` /
 `build_empty_from_sql` / `build_empty_from_columns` / `clone_empty_from_prod` /
 `check_binds`), and one image per engine runs both roles.
 
@@ -175,6 +176,62 @@ produced, never how it is checked or written.
   script reads with `os.environ[...]`. The contract never holds a secret value
   or env var name; the Secret itself is created in the cluster by the operator.
 
+## Write modes
+
+`mode_type:` chooses how the harness writes a node's table on each run:
+
+| `mode_type` | Each run | Rows from earlier runs |
+| --- | --- | --- |
+| `replace` (default) | empties the table, inserts this run's rows | gone |
+| `append` | inserts this run's rows | kept |
+| `upsert` | inserts rows with a new key, overwrites rows whose key exists | kept unless overwritten; never deleted |
+
+```yaml
+- schema: fx
+  table: fx_rates
+  kind: python-api
+  mode_type: upsert
+  primary_key: [rate_date, base, quote]
+  output_columns:
+    - {name: rate_date, type: DATE,       nullable: false}
+    - {name: base,      type: VARCHAR(3), nullable: false}
+    - {name: quote,     type: VARCHAR(3), nullable: false}
+    - {name: rate,      type: DECIMAL(12,6)}
+```
+
+- `python-csv` nodes only `replace`.
+- `primary_key` is required for `upsert` and rejected otherwise: a non-empty
+  list of output columns declared `nullable: false` (a NULL key never
+  matches), each spelled exactly as in `output_columns`, names unique
+  case-insensitively. A Postgres `replace` table that
+  wants uniqueness uses `config.indexes` with `unique: true`.
+- Every write is atomic: all of the run's rows or none. An `append` or
+  `upsert` run with zero rows writes nothing (a `replace` run with zero rows
+  still empties the table).
+- `append` is a plain insert: a run that committed but was reported failed
+  and is retried stores its rows twice. `upsert` is idempotent.
+- An `upsert` run whose own rows repeat a key fails with `ConformError` and
+  writes nothing.
+- Before an `append` or `upsert` the harness compares the live table with
+  `output_columns` — names case-insensitively, nullability exactly; order and
+  types are not compared — and fails with `LoadError` naming every missing
+  column, unexpected column and nullability mismatch. Changing an `append` or
+  `upsert` table (columns, types, nullability, key or mode) needs a new table
+  name (`_v2`) under a new contract; continuo rejects the release otherwise.
+  Adding a nullable column is no exception: the existing rows cannot be
+  backfilled.
+- The installed engine must support the mode; all three shipped adapters do.
+  A bring-your-own adapter declares `AppendCapable` / `UpsertCapable` from
+  `continuo-engine-contract` (and can run its conformance suite); without them
+  an `append`/`upsert` node fails with `LoadError: <Adapter> does not support
+  <mode>` before its script runs.
+- Engines: on Postgres the key is the table's `PRIMARY KEY`, created with the
+  table; DuckLake and Trino/Iceberg have no constraints, so the key is
+  guaranteed by the write path only. On Trino an `upsert` table needs Iceberg
+  `format_version` 2 (the default).
+- `full_refresh` stays refused for every python node: a rebuild would destroy
+  an accumulating table.
+
 ## The script API
 
 A node script is a Python file with exactly one required entry point:
@@ -261,8 +318,8 @@ text.
 | `ContractError` | Contract missing or invalid, node not found for `NODE_ID`, or the declared script is missing/unreachable. | The contract yaml or the `script:` path. |
 | `ReadError` | `ctx.read()` was called with an undeclared name, or a declared read failed at the warehouse. | The `reads:` map, or the upstream query/warehouse access. |
 | `ScriptError` | `run()` raised, has no callable `run`, or returned a value that isn't Arrow-convertible. | The node script. |
-| `ConformError` | Structural mismatch (extra/missing/duplicate columns), a strict-cast failure, a not-null violation, or a VARCHAR/CHAR overflow. | The script's output shape, or the `output_columns` declaration. |
-| `LoadError` | Adapter construction failed, or the DDL/INSERT failed at the warehouse during the write. | Warehouse connectivity/permissions, or the target table. |
+| `ConformError` | Structural mismatch (extra/missing/duplicate columns), a strict-cast failure, a not-null violation, a VARCHAR/CHAR overflow, or an `upsert` run whose rows repeat a primary key. | The script's output shape, or the `output_columns` declaration. |
+| `LoadError` | Adapter construction failed, the installed engine does not support the node's `mode_type`, the live table's shape differs from `output_columns` (append/upsert), or the DDL/INSERT failed at the warehouse during the write. | Warehouse connectivity/permissions, the engine image, or the target table. |
 
 ## Engine selection
 

adapters/duckdb/tests/test_integration_duckdb_runtime.py

@@ -251,3 +251,81 @@ def test_non_ascii_names_that_differ_only_in_case_are_two_tables(adapter, schema
         "ORDER BY table_name"
     ).to_pylist()
     assert sorted(row["table_name"] for row in names) == ["É", "é"]
+
+
+# --- write modes end to end ----------------------------------------------------
+
+_FX_COLUMNS = [
+    {"name": "rate_date", "type": "DATE", "nullable": False},
+    {"name": "base", "type": "VARCHAR(3)", "nullable": False},
+    {"name": "rate", "type": "DECIMAL(12,6)", "nullable": True},
+]
+
+_FX_SCRIPT = (
+    "import datetime\n"
+    "from decimal import Decimal\n\n"
+    "import pyarrow as pa\n\n"
+    "ROWS = {rows!r}\n\n\n"
+    "def run(ctx):\n"
+    "    return pa.table({{\n"
+    "        'rate_date': pa.array([datetime.date.fromisoformat(r[0]) for r in ROWS], pa.date32()),\n"
+    "        'base': pa.array([r[1] for r in ROWS], pa.string()),\n"
+    "        'rate': pa.array([Decimal(r[2]) for r in ROWS], pa.decimal128(12, 6)),\n"
+    "    }})\n"
+)
+
+
+def _fx_run_env(repo, schema, run, rows, **mode):
+    """A python-api fx node returning *rows*, with a script file unique to *run*.
+
+    One file per run: rewriting the same file within a second at the same size
+    would let Python reuse its cached bytecode and return the previous rows.
+    """
+    (repo / "contracts").mkdir(exist_ok=True)
+    (repo / "scripts").mkdir(exist_ok=True)
+    script = f"scripts/fx_{run}.py"
+    (repo / script).write_text(_FX_SCRIPT.format(rows=rows))
+    node = {
+        "schema": schema, "table": "fx_rates", "owner": "m", "schedule": "hourly",
+        "criticality": "SECONDARY", "kind": "python-api", "script": script,
+        "output_columns": _FX_COLUMNS, **mode,
+    }
+    (repo / "contracts" / "fx.yml").write_text(yaml.safe_dump({"nodes": [node]}))
+    return {
+        "NODE_ID": f"python-api.svc.{schema}.fx_rates", "TABLE_NAME": "fx_rates",
+        "TARGET_SCHEMA": schema, "CONTRACT_DIR": str(repo / "contracts"), "APP_ROOT": str(repo),
+    }
+
+
+def test_run_node_append_keeps_the_rows_of_earlier_runs(adapter_factory, schema, scalar, tmp_path):
+    env = _fx_run_env(tmp_path, schema, "run1", [("2026-10-08", "EUR", "1.081200")], mode_type="append")
+    assert run_node(env, adapter=adapter_factory()) == 0
+    env = _fx_run_env(
+        tmp_path, schema, "run2",
+        [("2026-10-08", "EUR", "1.081200"), ("2026-10-09", "EUR", "1.090000")],
+        mode_type="append",
+    )
+    assert run_node(env, adapter=adapter_factory()) == 0
+    assert scalar(f'SELECT count(*) AS n FROM "{schema}"."fx_rates"') == 3
+
+
+def test_run_node_upsert_overwrites_by_key(adapter, adapter_factory, schema, tmp_path):
+    key = {"mode_type": "upsert", "primary_key": ["rate_date", "base"]}
+    env = _fx_run_env(
+        tmp_path, schema, "run1",
+        [("2026-10-08", "EUR", "1.081200"), ("2026-10-08", "GBP", "0.861000")], **key,
+    )
+    assert run_node(env, adapter=adapter_factory()) == 0
+    env = _fx_run_env(
+        tmp_path, schema, "run2",
+        [("2026-10-08", "EUR", "1.090000"), ("2026-10-09", "EUR", "1.100000")], **key,
+    )
+    assert run_node(env, adapter=adapter_factory()) == 0
+    rows = adapter.fetch(
+        f'SELECT rate_date, base, rate FROM "{schema}"."fx_rates" ORDER BY rate_date, base'
+    ).to_pylist()
+    assert rows == [
+        {"rate_date": date(2026, 10, 8), "base": "EUR", "rate": Decimal("1.090000")},
+        {"rate_date": date(2026, 10, 8), "base": "GBP", "rate": Decimal("0.861000")},
+        {"rate_date": date(2026, 10, 9), "base": "EUR", "rate": Decimal("1.100000")},
+    ]

adapters/postgres/tests/test_integration_runtime_postgres.py

@@ -3,6 +3,7 @@
 Run the stack first:
     docker compose -f tests/smoke/postgres-stack/docker-compose.yml up -d --wait
 """
+import datetime
 import decimal
 import os
 import uuid
@@ -382,3 +383,110 @@ def test_run_node_csv_kind_loads_minio_csv_into_postgres(clean_schema, csv_minio
         rows = cur.fetchall()
     conn.close()
     assert rows == [(1, 10.5), (2, 20.0), (3, 5.25)]
+
+
+# --- write modes end to end ----------------------------------------------------
+
+_FX_COLUMNS = [
+    {"name": "rate_date", "type": "DATE", "nullable": False},
+    {"name": "base", "type": "VARCHAR(3)", "nullable": False},
+    {"name": "rate", "type": "DECIMAL(12,6)", "nullable": True},
+]
+
+_FX_SCRIPT = (
+    "import datetime\n"
+    "from decimal import Decimal\n\n"
+    "import pyarrow as pa\n\n"
+    "ROWS = {rows!r}\n\n\n"
+    "def run(ctx):\n"
+    "    return pa.table({{\n"
+    "        'rate_date': pa.array([datetime.date.fromisoformat(r[0]) for r in ROWS], pa.date32()),\n"
+    "        'base': pa.array([r[1] for r in ROWS], pa.string()),\n"
+    "        'rate': pa.array([Decimal(r[2]) for r in ROWS], pa.decimal128(12, 6)),\n"
+    "    }})\n"
+)
+
+
+def _fx_run_env(repo, schema, run, rows, **mode):
+    """A python-api fx node returning *rows*, with a script file unique to *run*.
+
+    One file per run: rewriting the same file within a second at the same size
+    would let Python reuse its cached bytecode and return the previous rows.
+    """
+    (repo / "contracts").mkdir(exist_ok=True)
+    (repo / "scripts").mkdir(exist_ok=True)
+    script = f"scripts/fx_{run}.py"
+    (repo / script).write_text(_FX_SCRIPT.format(rows=rows))
+    node = {
+        "schema": schema, "table": "fx_rates", "owner": "m", "schedule": "hourly",
+        "criticality": "SECONDARY", "kind": "python-api", "script": script,
+        "output_columns": _FX_COLUMNS, **mode,
+    }
+    (repo / "contracts" / "fx.yml").write_text(yaml.safe_dump({"nodes": [node]}))
+    return {
+        "NODE_ID": f"python-api.svc.{schema}.fx_rates", "TABLE_NAME": "fx_rates",
+        "TARGET_SCHEMA": schema, "CONTRACT_DIR": str(repo / "contracts"), "APP_ROOT": str(repo),
+    }
+
+
+@pytest.mark.integration
+def test_run_node_append_keeps_the_rows_of_earlier_runs(clean_schema, tmp_path):
+    env = _fx_run_env(tmp_path, clean_schema, "run1", [("2026-10-08", "EUR", "1.081200")], mode_type="append")
+    assert run_node(env, adapter=_adapter()) == 0
+    env = _fx_run_env(
+        tmp_path, clean_schema, "run2",
+        [("2026-10-08", "EUR", "1.081200"), ("2026-10-09", "EUR", "1.090000")],
+        mode_type="append",
+    )
+    assert run_node(env, adapter=_adapter()) == 0
+    assert _count(clean_schema, "fx_rates") == 3
+
+
+@pytest.mark.integration
+def test_run_node_upsert_overwrites_by_key_and_keeps_the_constraint(clean_schema, tmp_path):
+    """Review Focus 4 and 5 through the whole harness."""
+    key = {"mode_type": "upsert", "primary_key": ["rate_date", "base"]}
+    env = _fx_run_env(
+        tmp_path, clean_schema, "run1",
+        [("2026-10-08", "EUR", "1.081200"), ("2026-10-08", "GBP", "0.861000")], **key,
+    )
+    assert run_node(env, adapter=_adapter()) == 0
+    env = _fx_run_env(
+        tmp_path, clean_schema, "run2",
+        [("2026-10-08", "EUR", "1.090000"), ("2026-10-09", "EUR", "1.100000")], **key,
+    )
+    assert run_node(env, adapter=_adapter()) == 0
+    conn = _conn()
+    with conn.cursor() as cur:
+        cur.execute(
+            f'SELECT rate_date, base, rate FROM "{clean_schema}"."fx_rates" ORDER BY rate_date, base'
+        )
+        rows = cur.fetchall()
+        cur.execute(
+            "SELECT kcu.column_name FROM information_schema.table_constraints tc "
+            "JOIN information_schema.key_column_usage kcu "
+            "ON kcu.constraint_name = tc.constraint_name AND kcu.table_schema = tc.table_schema "
+            "AND kcu.table_name = tc.table_name "
+            "WHERE tc.table_schema = %s AND tc.table_name = %s "
+            "AND tc.constraint_type = 'PRIMARY KEY' ORDER BY kcu.ordinal_position",
+            (clean_schema, "fx_rates"),
+        )
+        key_columns = [row[0] for row in cur.fetchall()]
+    conn.close()
+    assert rows == [
+        (datetime.date(2026, 10, 8), "EUR", decimal.Decimal("1.090000")),
+        (datetime.date(2026, 10, 8), "GBP", decimal.Decimal("0.861000")),
+        (datetime.date(2026, 10, 9), "EUR", decimal.Decimal("1.100000")),
+    ]
+    assert key_columns == ["rate_date", "base"]
+
+
+@pytest.mark.integration
+def test_run_node_append_into_a_drifted_table_fails_naming_the_column(clean_schema, tmp_path, capsys):
+    _seed(clean_schema, "fx_rates", "rate_date date NOT NULL, base varchar(3) NOT NULL", "")
+    env = _fx_run_env(tmp_path, clean_schema, "run1", [("2026-10-08", "EUR", "1.081200")], mode_type="append")
+    assert run_node(env, adapter=_adapter()) == 1
+    out = capsys.readouterr().out
+    assert '"message":"LoadError: table runtime_it.fx_rates does not match' in out
+    assert "missing column(s) ['rate']" in out
+    assert _count(clean_schema, "fx_rates") == 0