feat(contract): mode_type and primary_key contract fields

CHANGELOG.md

@@ -5,6 +5,20 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Added
+
+- Contracts may set `mode_type` — `replace` (the default), `append` or
+  `upsert` — and, for `upsert` only, `primary_key`: a non-empty list of output
+  columns, each spelled exactly as in `output_columns` and declared
+  `nullable: false`, names unique case-insensitively.
+  `python-csv` nodes allow only `replace`. `continuo-runtime validate`, `merge`
+  and `run` enforce these rules; the harness's write path for the new modes
+  follows in a later entry.
+- The merged wire entry carries `mode_type` when it is not `replace` and
+  `primary_key` (authored order) for `upsert`; both are part of
+  `config_hash`. Entries of nodes that do not set them, and their hashes, are
+  unchanged.
+
 ### Changed
 
 - **Breaking for newly copied templates.** The `template/` release workflow

continuo_python_runtime/contract/kinds.py

@@ -1,8 +1,9 @@
 """Per-kind contract rules: which fields a node of each kind must carry.
 
 Each kind is one ``KindRules`` entry composed of a script policy, a reads rule,
-and whether it may name a Secret. Adding a kind means adding one entry to
-``RULES``; nothing that parses a contract branches on the kind's name.
+whether it may name a Secret, and which write modes it may declare. Adding a
+kind means adding one entry to ``RULES``; nothing that parses a contract
+branches on the kind's name.
 """
 
 from __future__ import annotations
@@ -14,7 +15,7 @@
 from continuo_engine_contract.sql import ensure_single_read  # type: ignore[import-untyped]
 from sqlglot.errors import TokenError
 
-from continuo_python_runtime.contract.model import KINDS
+from continuo_python_runtime.contract.model import DEFAULT_MODE_TYPE, KINDS, MODE_TYPES
 from continuo_python_runtime.csv_source import parse_csv_uri
 from continuo_python_runtime.errors import ContractError
 
@@ -108,12 +109,18 @@ class KindRules:
     script_required: bool
     reads: ReadsRule
     secret_allowed: bool = False
+    # The write modes a node of this kind may declare. A new kind gets replace
+    # only unless its entry opts into more.
+    mode_types: frozenset[str] = frozenset({DEFAULT_MODE_TYPE})
 
 
 RULES: dict[str, KindRules] = {
-    "python-node": KindRules(script_required=True, reads=SqlReads()),
+    "python-node": KindRules(script_required=True, reads=SqlReads(), mode_types=MODE_TYPES),
+    # A csv drop is a snapshot of its source: replace only (spec D2).
     "python-csv": KindRules(script_required=False, reads=CsvRead()),
-    "python-api": KindRules(script_required=True, reads=NoReads(), secret_allowed=True),
+    "python-api": KindRules(
+        script_required=True, reads=NoReads(), secret_allowed=True, mode_types=MODE_TYPES
+    ),
 }
 
 

continuo_python_runtime/contract/loader.py

@@ -17,7 +17,9 @@
 from continuo_python_runtime.contract.model import (
     CRITICALITIES,
     DEFAULT_KIND,
+    DEFAULT_MODE_TYPE,
     EXTRA_COLUMNS_POLICIES,
+    MODE_TYPES,
     Column,
     Node,
 )
@@ -40,6 +42,8 @@
     "config",
     "content_hash",
     "secret_ref",
+    "mode_type",
+    "primary_key",
 }
 
 _REQUIRED_STRING_FIELDS = ("schema", "table", "owner", "schedule")
@@ -140,6 +144,59 @@ def _check(value: Any, key: Any) -> None:
     return raw
 
 
+def _validate_primary_key(
+    raw: dict[str, Any], mode_type: str, columns: list[Column], label: str
+) -> tuple[str, ...]:
+    """Validate ``primary_key`` against ``mode_type`` and the declared columns.
+
+    Required for ``upsert`` and rejected otherwise: a key on ``append`` or
+    ``replace`` would look enforced when it is not. Names keep their order,
+    which is what the wire entry carries. Each must be spelled exactly as an
+    output column (postgres quotes identifiers, so ``Rate_Date`` is not
+    ``rate_date``) declared ``nullable: false``, because a NULL key never
+    matches an existing row. Repeats are refused case-insensitively too, so a
+    key cannot name two columns that differ only in case.
+    """
+    if mode_type != "upsert":
+        if "primary_key" in raw:
+            raise ContractError(
+                f"{label}: 'primary_key' is only allowed when mode_type is upsert "
+                f"(got {mode_type})"
+            )
+        return ()
+    if "primary_key" not in raw:
+        raise ContractError(f"{label}: 'primary_key' is required when mode_type is upsert")
+    raw_key = raw["primary_key"]
+    if not isinstance(raw_key, list) or not raw_key:
+        raise ContractError(
+            f"{label}: 'primary_key' must be a non-empty list of output column names"
+        )
+    by_name = {column.name: column for column in columns}
+    seen: set[str] = set()
+    for name in raw_key:
+        if not isinstance(name, str) or not name.strip():
+            raise ContractError(
+                f"{label}: 'primary_key' entry {name!r} must be a non-empty string"
+            )
+        folded = name.lower()  # same folding as continuo's parser
+        if folded in seen:
+            raise ContractError(
+                f"{label}: 'primary_key' names {name!r} more than once "
+                "(names are compared case-insensitively)"
+            )
+        seen.add(folded)
+        column = by_name.get(name)
+        if column is None:
+            raise ContractError(
+                f"{label}: 'primary_key' column {name!r} is not in output_columns"
+            )
+        if column.nullable:
+            raise ContractError(
+                f"{label}: 'primary_key' column {name!r} must be declared nullable: false"
+            )
+    return tuple(raw_key)
+
+
 def parse_node(
     raw: dict[str, Any],
     source: str,
@@ -158,7 +215,8 @@ def parse_node(
     ``check_reads=False`` skips *only* that :func:`ensure_single_read` call —
     the read-shape gate — leaving every other rule here (required fields,
     criticality, the ``reads`` map's own shape, output-column types and
-    uniqueness, ``config``, ``content_hash``) in force. It exists for
+    uniqueness, ``mode_type`` and ``primary_key``, ``config``,
+    ``content_hash``) in force. It exists for
     :func:`~continuo_python_runtime.harness.run_node`; see
     :func:`load_contract_dir` for why the runtime opts out.
     """
@@ -229,6 +287,17 @@ def parse_node(
             raise ContractError(f"{label}: 'secret_ref' is not allowed for kind {kind}")
         secret_ref = validate_secret_ref(raw["secret_ref"], label)
 
+    mode_type = raw.get("mode_type", DEFAULT_MODE_TYPE)
+    if not isinstance(mode_type, str) or mode_type not in MODE_TYPES:
+        raise ContractError(
+            f"{label}: 'mode_type' must be one of {sorted(MODE_TYPES)}, got {mode_type!r}"
+        )
+    if mode_type not in rules.mode_types:
+        raise ContractError(
+            f"{label}: 'mode_type' {mode_type} is not allowed for kind {kind} "
+            f"(allowed: {sorted(rules.mode_types)})"
+        )
+
     raw_columns = raw.get("output_columns")
     if not isinstance(raw_columns, list) or not raw_columns:
         raise ContractError(f"{label}: 'output_columns' must be a non-empty list")
@@ -269,6 +338,8 @@ def parse_node(
         seen_names.add(name)
         columns.append(Column(name=name, type=col_type, nullable=nullable))
 
+    primary_key = _validate_primary_key(raw, mode_type, columns, label)
+
     description = raw.get("description", "")
     if not isinstance(description, str):
         raise ContractError(f"{label}: 'description' must be a string")
@@ -294,6 +365,8 @@ def parse_node(
         content_hash=content_hash,
         kind=kind,
         secret_ref=secret_ref,
+        mode_type=mode_type,
+        primary_key=primary_key,
     )
 
 

continuo_python_runtime/contract/merge.py

@@ -7,7 +7,7 @@
 
 from continuo_python_runtime.closure import resolve_closure
 from continuo_python_runtime.contract.loader import load_contract_dir
-from continuo_python_runtime.contract.model import CONTRACT_VERSION, Node
+from continuo_python_runtime.contract.model import CONTRACT_VERSION, DEFAULT_MODE_TYPE, Node
 from continuo_python_runtime.contract.paths import resolve_script_path
 from continuo_python_runtime.errors import ContractError
 from continuo_python_runtime.hashing import hash_parts
@@ -20,7 +20,10 @@ def node_entry(node: Node) -> dict:
     Returns the node as a dict with all fields, output_columns as list of
     dicts. The entry carries no hash fields: build_wire_contract adds all
     four (source_hash, shared_code_hash, config_hash, content_hash). The
-    nullable field is always present in output_columns.
+    nullable field is always present in output_columns. ``secret_ref``,
+    ``mode_type`` and ``primary_key`` appear only when they differ from their
+    default, so a node that does not use them has the entry (and hashes) it
+    had before they existed.
     """
     entry = {
         "schema": node.schema,
@@ -47,6 +50,12 @@ def node_entry(node: Node) -> dict:
     # only when set keeps every other node's config_hash unchanged.
     if node.secret_ref:
         entry["secret_ref"] = node.secret_ref
+    # Same reasoning: every replace node's entry, config_hash and content_hash
+    # stay byte-identical, so contract_version stays 1.
+    if node.mode_type != DEFAULT_MODE_TYPE:
+        entry["mode_type"] = node.mode_type
+    if node.mode_type == "upsert":
+        entry["primary_key"] = list(node.primary_key)
     return entry
 
 

continuo_python_runtime/contract/model.py

@@ -8,6 +8,11 @@
 EXTRA_COLUMNS_POLICIES = frozenset({"raise", "warn"})
 KINDS = frozenset({"python-node", "python-csv", "python-api"})
 DEFAULT_KIND = "python-node"
+# How the harness writes a node's table on each run: replace empties it first,
+# append keeps the rows of earlier runs, upsert overwrites rows whose
+# primary_key exists and keeps the rest.
+MODE_TYPES = frozenset({"append", "upsert", "replace"})
+DEFAULT_MODE_TYPE = "replace"
 CONTRACT_VERSION = 1
 
 
@@ -38,6 +43,8 @@ class Node:
     content_hash: str | None = None
     kind: str = DEFAULT_KIND
     secret_ref: str = ""
+    mode_type: str = DEFAULT_MODE_TYPE
+    primary_key: tuple[str, ...] = ()
 
     @property
     def relation(self) -> str:

continuo_python_runtime/harness.py

@@ -29,7 +29,7 @@
 from continuo_python_runtime.conform import conform, to_arrow
 from continuo_python_runtime.context import RunContext
 from continuo_python_runtime.contract.loader import load_contract_dir
-from continuo_python_runtime.contract.model import Node
+from continuo_python_runtime.contract.model import DEFAULT_MODE_TYPE, Node
 from continuo_python_runtime.contract.paths import resolve_script_path
 from continuo_python_runtime.csv_loader import produce_csv
 from continuo_python_runtime.csv_source import CsvSourceReader
@@ -265,6 +265,16 @@ def run_node(
         nodes = load_contract_dir(contract_dir, check_reads=False)
         node = select_node(nodes, node_id)
 
+        # Interim: the harness writes every node with load() (replace) until the
+        # per-mode write dispatch lands. Until then, reject append/upsert before
+        # the script runs or anything is written, so an accepted contract cannot
+        # silently replace a table it meant to append to or upsert into.
+        if node.mode_type != DEFAULT_MODE_TYPE:
+            raise LoadError(
+                f"write mode {node.mode_type!r} is not supported by the harness yet "
+                f"(node {target_schema}.{table_name}); only {DEFAULT_MODE_TYPE!r} is available"
+            )
+
         if adapter is not None:
             active_adapter = adapter
         else:

docs/boundary-contract.md

@@ -61,6 +61,19 @@ s3://<bucket>/<service>/<release_id>/contract.yaml
   node's wire entry or `config_hash` changes), and it is part of
   `config_hash`, so `content_hash` changes when the referenced Secret name
   changes. `source_hash` does not.
+- **`mode_type`** and **`primary_key`** are additive and optional. `mode_type`
+  (`replace`, the default; `append`; `upsert`) says how the harness writes
+  the node's table on each run; `python-csv` allows only `replace`.
+  `primary_key` is required for `upsert` and rejected for every other mode: a
+  non-empty list of output column names, each spelled exactly as in
+  `output_columns` and declared `nullable: false`, unique case-insensitively. On the merged wire entry `mode_type` appears
+  only when it is not `replace`, and `primary_key` only for `upsert`, in the
+  authored order and spelling — so every existing node's wire entry,
+  `config_hash` and `content_hash` are unchanged and `contract_version` stays
+  1. Both are part of `config_hash` when present. A shared fixture of the
+  expected entries lives in this repo at
+  `tests/fixtures/wire_contract_write_modes.yml` and, byte-identical, in
+  continuo at `topology-controller/tests/fixtures/wire_contract_write_modes.yml`.
 - **Reads must be single-statement SELECTs with every table reference
   schema-qualified** (`analytics.table_a`, never `table_a`) — the resolver
   raises `UnqualifiedTableReference` and rejects the whole release otherwise

tests/contract/test_kinds.py

@@ -1,7 +1,7 @@
 import pytest
 
 from continuo_python_runtime.contract.kinds import RULES, CsvRead, NoReads, SqlReads, rules_for
-from continuo_python_runtime.contract.model import KINDS
+from continuo_python_runtime.contract.model import KINDS, MODE_TYPES
 from continuo_python_runtime.contract.secret_ref import validate_secret_ref
 from continuo_python_runtime.errors import ContractError
 
@@ -94,3 +94,18 @@ def test_secret_ref_accepts_valid_names(name):
 def test_secret_ref_rejects_invalid_names(name):
     with pytest.raises(ContractError, match="'secret_ref'"):
         validate_secret_ref(name, "L")
+
+
+def test_python_csv_only_replaces():
+    assert RULES["python-csv"].mode_types == frozenset({"replace"})
+
+
+@pytest.mark.parametrize("kind", ["python-node", "python-api"])
+def test_script_kinds_allow_every_mode(kind):
+    assert RULES[kind].mode_types == MODE_TYPES
+
+
+def test_every_kind_allows_replace_and_nothing_outside_the_vocabulary():
+    for rules in RULES.values():
+        assert "replace" in rules.mode_types
+        assert rules.mode_types <= MODE_TYPES