feat: define VGI 2.0 write result modes (#75)

Author: rustyconover

docs/protocol/vgi-protocol-proposed-changes.md

@@ -453,7 +453,34 @@ Triggers are different. Supporting a trigger in a virtual catalog would require
 
 The same principle applies to secure views, composable COPY, remote-plan pushdown, coordinate systems, and any catalog object VGI does not currently virtualize: they are not protocol requirements simply because DuckDB implements them. They should be gated through explicit catalog capabilities and fail early.
 
-### 6.1 New catalog kinds in DuckDB 2.0
+### 6.1 Writable result modes and OLD/NEW images
+
+VGI 2.0 replaces the four `TableInfo` booleans `supports_insert`,
+`supports_update`, `supports_delete`, and `supports_returning` with one required,
+non-null Arrow map:
+
+```text
+write_result_modes: map<utf8, utf8>
+```
+
+The only operation keys are `insert`, `update`, and `delete`; absence means the
+operation is unsupported. Each value is the maximum supported mode in the
+ordered lattice `count < rows < changes`, and therefore promises every lower
+mode. Duplicate keys and unknown operations or modes are protocol errors.
+
+The serialized `write_options` batch replaces `return_chunks: bool` with the
+required `result_mode: utf8`. `count` returns `(count int64 not null)`. `rows`
+returns NEW rows for INSERT/UPDATE and OLD rows for DELETE. `changes` returns
+two nullable structs, `old` and `new`, each shaped exactly like the table:
+INSERT populates only `new`, DELETE only `old`, and UPDATE both. The operation
+is known from the invoked write function and is not repeated in each result row.
+
+The DuckDB 1.5 adapter requests `count` for ordinary DML and `rows` for DML with
+RETURNING. It validates and preserves `changes` capability metadata but never
+requests that mode. A DuckDB 2.0 adapter can request `changes` when trigger
+planning needs OLD/NEW images without another protocol revision.
+
+### 6.2 New catalog kinds in DuckDB 2.0
 
 The `CatalogType` enum has exactly two additions relative to DuckDB 1.5.5:
 
@@ -464,7 +491,7 @@ The `CatalogType` enum has exactly two additions relative to DuckDB 1.5.5:
 
 Nested schemas are recursively contained `SCHEMA_ENTRY` objects, not another kind. Secure views remain `VIEW_ENTRY`; changes to aggregate-state representation belong to the logical type system rather than the catalog enum.
 
-### 6.2 How native DuckDB stores triggers
+### 6.3 How native DuckDB stores triggers
 
 Triggers are first-class `TriggerCatalogEntry` objects, but they do not live in the schema's ordinary catalog set. Each `DuckTableEntry` owns a separate transactional `CatalogSet`:
 
@@ -496,7 +523,7 @@ The dependency manager has special trigger identity/lookup handling because sche
 
 At bind time DuckDB scans the table's triggers and expands them into the DML plan. Statement triggers form a materialized CTE chain (BEFORE actions, base DML, AFTER actions), including materialized OLD/NEW transition tables when requested. Supported row triggers use `LogicalTrigger` and correlated `NEW.col`/`OLD.col` bindings. The current catalog set yields triggers in case-insensitive alphabetical name order.
 
-### 6.3 DuckDB changes needed by custom table implementations
+### 6.4 DuckDB changes needed by custom table implementations
 
 DuckDB 2.0 already exposes virtual `TableCatalogEntry::CreateTrigger`, `ScanTriggers`, and `GetTrigger` hooks, and its trigger binder consumes the generic scan hook. Two remaining native-table assumptions should be fixed upstream:
 
@@ -509,7 +536,7 @@ Upstream PR: [duckdb/duckdb#25539 — Support triggers on custom table catalog e
 
 The `ScanTriggers` contract also implies that returned catalog entries remain alive beyond the callback: both trigger selection and introspection retain references. VGI should materialize trigger entries in a table-owned metadata cache, invalidated through the existing catalog-version mechanism, rather than returning callback-local objects.
 
-### 6.4 Proposed VGI trigger protocol
+### 6.5 Proposed VGI trigger protocol
 
 The remote VGI catalog should be authoritative for persistent trigger definitions. Keeping definitions only in a local extension cache would lose them on detach and would give different clients inconsistent catalogs.
 
@@ -529,13 +556,22 @@ TriggerInfo
     for_each: statement | row
     referencing_new_table: optional<string>
     referencing_old_table: optional<string>
-    action_sql: string
+    definitions: map<string, string>
     dependencies: list<TriggerDependency>
     comment: optional<string>
     tags: map<string, string>
 ```
 
-Use structured fields plus DuckDB-dialect action SQL. Do not put DuckDB's binary `QueryNode` serialization on the VGI wire: it is an internal, version-coupled representation. The local extension can parse the action and create cached `TriggerCatalogEntry` objects. Each resolved dependency should carry its qualified object reference plus ownership, drop-blocking, and alter-blocking flags so the remote catalog can enforce DDL integrity across clients.
+Use structured fields plus engine-specific textual definitions keyed by stable
+lowercase engine identifiers such as `duckdb`, `datafusion`, `spark`, and
+`sqlite`. An engine executes only its matching definition; a missing entry means
+that trigger is unavailable on that engine, with no implicit SQL-dialect
+fallback. Do not put DuckDB's binary `QueryNode` serialization on the VGI wire:
+it is an internal, version-coupled representation. The DuckDB extension parses
+the `duckdb` definition and creates cached `TriggerCatalogEntry` objects. Each
+resolved dependency should carry its qualified object reference plus ownership,
+drop-blocking, and alter-blocking flags so the remote catalog can enforce DDL
+integrity across clients.
 
 Suggested methods:
 
@@ -545,7 +581,10 @@ Suggested methods:
 - `catalog_table_trigger_drop`
 - optional trigger comment/tag alteration
 
-Add explicit catalog/table capability metadata such as `supports_triggers`. A trigger-capable table must also support exact affected-row RETURNING, because DuckDB's statement-trigger expansion internally materializes base DML with `RETURNING *` even when the user did not write a RETURNING clause. UPDATE must additionally expose captured OLD columns before OLD transition tables can be enabled.
+Add explicit catalog/table capability metadata such as `supports_triggers`. A
+DuckDB-managed trigger-capable table must advertise `changes` for every event it
+uses, because DuckDB's statement-trigger expansion needs exact OLD/NEW images
+even when the user did not write a RETURNING clause.
 
 All base and trigger-body writes must use the same VGI transaction token and roll back atomically. Stable statement/sub-operation identifiers are recommended so transport retries cannot apply a trigger side effect twice. Trigger support should be rejected when the VGI catalog cannot provide transactional atomicity.
 
@@ -673,4 +712,11 @@ The first protocol-breaking phase is now implemented in Python:
 
 `CatalogAttachResult.default_schema` intentionally remains a scalar root-schema identifier for the reason described in section 2.
 
-This phase does **not** implement the other proposed protocol areas: `vgi.filters.v2`, logical-type identity annotations, trigger RPCs, or standalone window-function catalog entries. The sibling C++, C#, Go, Java, Rust, and TypeScript schema-path migrations were completed afterward. The vendored browser `vgi-client.js` remains generated from the TypeScript SDK and must never be edited manually in the Python repository.
+The writable-result phase replaces the legacy write booleans and
+`return_chunks` option with the `write_result_modes`/`result_mode` contract from
+section 6.1. Python implements and directly tests all three modes. The DuckDB
+1.5 extension consumes only `count` and `rows`; `changes` is reserved for the
+future DuckDB 2.0 trigger adapter. Trigger catalog metadata and RPCs remain
+unimplemented.
+
+This phase does **not** implement the other proposed protocol areas: logical-type identity annotations, trigger RPCs, or standalone window-function catalog entries. The sibling C++, C#, Go, Java, Rust, and TypeScript schema migrations are maintained in their corresponding repositories. The vendored browser `vgi-client.js` remains generated from the TypeScript SDK and must never be edited manually in the Python repository.

tests/catalog/test_writable_table.py

@@ -129,11 +129,11 @@ def test_delete_requires_scan_function(self) -> None:
             )
 
 
-class TestTableInfoWriteFlags:
-    """TableInfo serialization includes write support flags."""
+class TestTableInfoWriteResultModes:
+    """TableInfo serializes per-operation maximum result modes."""
 
-    def test_write_flags_set_when_functions_present(self) -> None:
-        """Write flags are True when corresponding functions are set."""
+    def test_count_modes_inferred_when_functions_present(self) -> None:
+        """Configured functions default to the count result mode."""
         table = Table(
             name="t",
             function=WritableTableScan,
@@ -142,35 +142,63 @@ def test_write_flags_set_when_functions_present(self) -> None:
             delete_function=WritableTableDelete,
         )
         info = table.to_table_info(["main"])
-        assert info.supports_insert is True
-        assert info.supports_update is True
-        assert info.supports_delete is True
+        assert info.write_result_modes == {"insert": "count", "update": "count", "delete": "count"}
 
-    def test_write_flags_false_by_default(self) -> None:
-        """Write flags are False when no write functions are set."""
+    def test_read_only_table_has_empty_modes(self) -> None:
+        """An empty map describes a read-only table."""
         table = Table(
             name="t",
             function=WritableTableScan,
         )
         info = table.to_table_info(["main"])
-        assert info.supports_insert is False
-        assert info.supports_update is False
-        assert info.supports_delete is False
+        assert info.write_result_modes == {}
 
-    def test_partial_write_flags(self) -> None:
-        """Only the flags for defined write functions are True."""
+    def test_partial_write_modes(self) -> None:
+        """Only operations with functions appear in the map."""
         table = Table(
             name="t",
             function=WritableTableScan,
             insert_function=WritableTableInsert,
         )
         info = table.to_table_info(["main"])
-        assert info.supports_insert is True
-        assert info.supports_update is False
-        assert info.supports_delete is False
+        assert info.write_result_modes == {"insert": "count"}
+
+    def test_explicit_maximum_modes(self) -> None:
+        """Authors can opt each configured operation into rows or changes."""
+        table = Table(
+            name="t",
+            function=WritableTableScan,
+            insert_function=WritableTableInsert,
+            update_function=WritableTableUpdate,
+            write_result_modes={"insert": "rows", "update": "changes"},
+        )
+        assert table.to_table_info(["main"]).write_result_modes == {
+            "insert": "rows",
+            "update": "changes",
+        }
+
+    @pytest.mark.parametrize("modes", [{"merge": "count"}, {"insert": "invalid"}])
+    def test_invalid_modes_rejected(self, modes: dict[str, str]) -> None:
+        """Unknown operations and modes fail at descriptor construction."""
+        with pytest.raises(ValueError):
+            Table(
+                name="t",
+                columns=pa.schema([("id", pa.int64())]),
+                insert_function=WritableTableInsert,
+                write_result_modes=modes,
+            )
+
+    def test_mode_without_function_rejected(self) -> None:
+        """Capability metadata cannot claim a missing implementation."""
+        with pytest.raises(ValueError, match="without a corresponding update_function"):
+            Table(
+                name="t",
+                columns=pa.schema([("id", pa.int64())]),
+                write_result_modes={"update": "rows"},
+            )
 
     def test_table_info_serialization_roundtrip(self) -> None:
-        """Write flags survive Arrow serialization roundtrip."""
+        """Write modes survive Arrow serialization roundtrip."""
         table = Table(
             name="t",
             function=WritableTableScan,
@@ -181,9 +209,7 @@ def test_table_info_serialization_roundtrip(self) -> None:
         info = table.to_table_info(["main"])
         data = info.serialize_to_bytes()
         restored = TableInfo.deserialize_from_bytes(data)
-        assert restored.supports_insert is True
-        assert restored.supports_update is True
-        assert restored.supports_delete is True
+        assert restored.write_result_modes == {"insert": "count", "update": "count", "delete": "count"}
 
 
 class TestReadOnlyCatalogInterfaceWriteMethods:

tests/test_write_results.py

@@ -0,0 +1,50 @@
+# Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+"""Tests for the VGI 2.0 writable result-mode contract."""
+
+import pyarrow as pa
+import pytest
+
+from vgi.schema_utils import schema
+from vgi.write_results import supports_write_result_mode, write_changes_batch, write_result_schema
+
+
+def test_ordered_result_modes() -> None:
+    """A maximum mode promises itself and every lower mode."""
+    assert supports_write_result_mode("count", "count")
+    assert not supports_write_result_mode("count", "rows")
+    assert supports_write_result_mode("rows", "count")
+    assert supports_write_result_mode("changes", "rows")
+    assert supports_write_result_mode("changes", "changes")
+
+
+def test_result_schemas() -> None:
+    """Every mode has the exact canonical Arrow schema."""
+    table = schema(id=pa.int64(), name=pa.string())
+    assert write_result_schema("count", table) == pa.schema([pa.field("count", pa.int64(), nullable=False)])
+    assert write_result_schema("rows", table) == table
+    changes = write_result_schema("changes", table)
+    assert changes.names == ["old", "new"]
+    assert changes.field("old").nullable
+    assert changes.field("old").type == pa.struct(list(table))
+
+
+@pytest.mark.parametrize(
+    ("old", "new"),
+    [
+        (None, {"id": 1, "name": "inserted"}),
+        ({"id": 1, "name": "old"}, {"id": 1, "name": "new"}),
+        ({"id": 1, "name": "deleted"}, None),
+    ],
+)
+def test_change_images(old: dict[str, object] | None, new: dict[str, object] | None) -> None:
+    """INSERT, UPDATE, and DELETE preserve the required OLD/NEW nullability."""
+    table = schema(id=pa.int64(), name=pa.string())
+    batch = write_changes_batch(table, [old], [new])
+    assert batch.to_pylist() == [{"old": old, "new": new}]
+
+
+def test_change_image_lengths_must_match() -> None:
+    """Every output row must contain one OLD/NEW pair."""
+    with pytest.raises(ValueError, match="equal length"):
+        write_changes_batch(pa.schema([("id", pa.int64())]), [], [{"id": 1}])

tests/transactor/test_transactor.py

@@ -24,7 +24,7 @@
 # Helpers
 # ============================================================================
 
-_COUNT_SCHEMA = schema(count=pa.int64())
+_COUNT_SCHEMA = pa.schema([pa.field("count", pa.int64(), nullable=False)])
 _FILTER_METADATA = {b"vgi_filter_version": b"1"}
 
 
@@ -66,7 +66,7 @@ def _insert_into(
     t: TransactorImpl, tx_id: bytes, table_name: str, rows: list[tuple[int, str]], *, returning: bool = False
 ) -> pa.RecordBatch:
     """Insert rows into an arbitrary table and return the response batch."""
-    stream = t.insert(tx_id=tx_id, table_name=table_name, returning=returning)
+    stream = t.insert(tx_id=tx_id, table_name=table_name, result_mode="rows" if returning else "count")
     batch = pa.record_batch(
         {"id": [r[0] for r in rows], "name": [r[1] for r in rows]},
         schema=schema(id=pa.int64(), name=pa.string()),
@@ -79,7 +79,7 @@ def _insert_into(
 
 def _insert(t: TransactorImpl, tx_id: bytes, rows: list[tuple[int, str]], *, returning: bool = False) -> pa.RecordBatch:
     """Insert rows via the transactor and return the response batch."""
-    stream = t.insert(tx_id=tx_id, table_name="data", returning=returning)
+    stream = t.insert(tx_id=tx_id, table_name="data", result_mode="rows" if returning else "count")
     batch = pa.record_batch(
         {"id": [r[0] for r in rows], "name": [r[1] for r in rows]},
         schema=schema(id=pa.int64(), name=pa.string()),
@@ -119,7 +119,7 @@ def _scan(
 
 def _delete(t: TransactorImpl, tx_id: bytes, rowids: list[int], *, returning: bool = False) -> pa.RecordBatch:
     """Delete rows by rowid, return the response batch."""
-    stream = t.delete(tx_id=tx_id, table_name="data", returning=returning)
+    stream = t.delete(tx_id=tx_id, table_name="data", result_mode="rows" if returning else "count")
     batch = pa.record_batch({"rowid": rowids}, schema=schema(rowid=pa.int64()))
     out = OutputCollector(stream.output_schema, producer_mode=False)
     assert isinstance(stream.state, ExchangeState)
@@ -136,7 +136,12 @@ def _update(
     t: TransactorImpl, tx_id: bytes, rowids: list[int], names: list[str], *, returning: bool = False
 ) -> pa.RecordBatch:
     """Update name column by rowid, return the response batch."""
-    stream = t.update(tx_id=tx_id, table_name="data", columns=["name"], returning=returning)
+    stream = t.update(
+        tx_id=tx_id,
+        table_name="data",
+        columns=["name"],
+        result_mode="rows" if returning else "count",
+    )
     batch = pa.record_batch(
         {"name": names, "rowid": rowids},
         schema=schema(name=pa.string(), rowid=pa.int64()),

vgi/__init__.py

@@ -129,10 +129,15 @@
     TableInOutGenerator,
 )
 from vgi.worker import Worker
+from vgi.write_results import WriteOperation, WriteResultMode, write_changes_batch, write_result_schema
 
 __all__ = [
     "AggregateBindParams",
     "AggregateFunction",
+    "WriteOperation",
+    "WriteResultMode",
+    "write_result_schema",
+    "write_changes_batch",
     "AnyArrow",
     "AnyArrowValue",
     "Arg",

vgi/_test_fixtures/schema_reconcile/worker.py

@@ -528,9 +528,7 @@ def _table_info(self, spec: TableSpec) -> TableInfo:
             not_null_constraints=[],
             unique_constraints=[],
             check_constraints=[],
-            supports_insert=True,
-            supports_update=True,
-            supports_delete=True,
+            write_result_modes={"insert": "changes", "update": "changes", "delete": "changes"},
         )
 
     def schemas(