fix: refuse layout for a table not being created, and read it back in one request

Five review points, all taken.

create_managed_database now raises ValueError when partition_by or sorted_by
names a table that is not in `tables`. Previously a typo -- {"fils": ...} against
tables=["files"] -- was dropped in silence and the intended table created flat,
which is permanent because a layout is fixed at creation with no alter path. The
asymmetry with `keys` is deliberate: a wrong key is recoverable, since
load_managed_table takes `key=` per call.

managed_table_layout now filters server-side (var_schema=/table=/limit=1) instead
of paging iter_tables until it finds a match, which cost several round trips for
a table sorting late in the listing. Same KeyError contract.

The absent-vs-unpartitioned test shared one response object across both calls, so
its second assertion inspected a consumed result rather than the case named in
the docstring. Each call now gets a fresh response, and the fake answers by table
name so the "listing has files but not missing" scenario is the one exercised.

TableLayout gains to_dict(), matching every other public dataclass here. Not
asdict(): the key lists hold pydantic models, which asdict copies through
untouched, so each key is mapped through its own to_dict(). The reviewer assumed
the omission was deliberate -- it was not.

CONTRACT.md lists managed_table_layout among the methods accepting an
already-resolved ManagedDatabase, which matters because a create-scoped key that
cannot read /databases is exactly the caller needing this read-back.

Re-sabotaged after the rewrite -- dropping the layout, collapsing missing-table
into an empty layout, and removing the unknown-table guard are each caught.
138 -> 139 passed.

Author: anoop-narang

CONTRACT.md

@@ -69,7 +69,7 @@ Adapters should import from `hotdata_framework` and treat this surface as the st
 - `load_managed_table(database, table, schema=..., upload_id=..., file=...)` publishes parquet data into a declared managed table.
 - `delete_managed_table(database, table, schema=...)` deletes a managed table.
 - `create_index(database, table, schema=..., columns=..., index_type=..., index_name=...)` builds a `"sorted"`, `"bm25"`, or `"vector"` index on a managed table and returns a `CreateIndexResult`. It is the framework-side equivalent of the CLI's `hotdata indexes create`; indexing a table on a plain (non-managed) connection is out of scope. `index_name` defaults to `{table}_{columns}_{index_type}`, matching the CLI's derivation when `--name` is omitted. `index_type` is required rather than defaulting to the API's `"sorted"`. The build runs as a background job; the call polls it to a terminal state and raises `RuntimeError` with the job's `error_message` when it fails, because the submit call reports success regardless. `wait=False` returns as soon as the job is accepted, with `status="pending"` and a `job_id` for the caller to poll. For `index_type="vector"`, omitting `embedding_provider_id` indexes an existing vector column and `metric` (`"l2"`, `"cosine"`, `"dot"`) selects the distance function the index accelerates — a query using a different function silently falls back to a full scan; setting `embedding_provider_id` indexes a source *text* column instead, and the returned `source_column` names the column to pass to `vector_distance`. Argument combinations the server would silently ignore raise `ValueError` before any request is sent.
-- The `database` argument of `list_managed_tables`, `load_managed_table`, `add_managed_table`, `delete_managed_table`, `delete_managed_database`, `create_index`, and `execute_sql` accepts a name/id **or** an already-resolved `ManagedDatabase`. Passing a `ManagedDatabase` skips the name/id read probe, so a create-scoped key that cannot read `/databases` can load into a database it just created.
+- The `database` argument of `list_managed_tables`, `load_managed_table`, `add_managed_table`, `delete_managed_table`, `delete_managed_database`, `create_index`, `managed_table_layout`, and `execute_sql` accepts a name/id **or** an already-resolved `ManagedDatabase`. Passing a `ManagedDatabase` skips the name/id read probe, so a create-scoped key that cannot read `/databases` can load into a database it just created.
 
 ### `QueryResult`
 

hotdata_framework/client.py

@@ -306,6 +306,16 @@ def create_managed_database(
         keys = keys or {}
         partition_by = partition_by or {}
         sorted_by = sorted_by or {}
+        # A layout aimed at a table that is not being created would otherwise be
+        # dropped in silence, and the table it was meant for created flat — which
+        # is permanent, since a layout is fixed at creation with no alter path. A
+        # typo'd `keys` entry costs nothing by comparison: load_managed_table
+        # takes `key=` per call, so it can be corrected later.
+        unknown = (set(partition_by) | set(sorted_by)) - set(tables or ())
+        if unknown:
+            raise ValueError(
+                f"layout given for tables not being created: {', '.join(sorted(unknown))}"
+            )
         schemas = None
         if tables:
             schemas = [
@@ -502,14 +512,23 @@ def managed_table_layout(
         two are very different for a caller deciding whether to load.
         """
         db = self._as_managed_database(database)
-        for info in self.iter_tables(connection_id=db.default_connection_id):
-            if info.table == table and info.var_schema == schema:
-                return TableLayout(
-                    schema_name=schema,
-                    table_name=table,
-                    partition_by=list(info.partition_by or []),
-                    sorted_by=list(info.sorted_by or []),
-                )
+        # Filtered server-side rather than paging iter_tables: this answers a
+        # single-table question, and a table sorting late in the listing would
+        # otherwise cost several round trips. include_columns is left off — the
+        # layout lives on the table row, not the columns.
+        resp = self._information_schema().information_schema(
+            connection_id=db.default_connection_id,
+            var_schema=schema,
+            table=table,
+            limit=1,
+        )
+        for info in resp.tables:
+            return TableLayout(
+                schema_name=schema,
+                table_name=table,
+                partition_by=list(info.partition_by or []),
+                sorted_by=list(info.sorted_by or []),
+            )
         raise KeyError(f"{schema}.{table} is not declared on database {db.id}")
 
     def delete_managed_table(

hotdata_framework/databases.py

@@ -57,6 +57,20 @@ class TableLayout:
     partition_by: list[TablePartitionKey]
     sorted_by: list[TableSortKey]
 
+    def to_dict(self) -> dict[str, Any]:
+        """Plain dict, like the other dataclasses here.
+
+        Not `asdict()`: the key lists hold pydantic models, which `asdict` copies
+        through untouched, so the result would not be a plain dict. Each key is
+        mapped through its own `to_dict()` instead.
+        """
+        return {
+            "schema_name": self.schema_name,
+            "table_name": self.table_name,
+            "partition_by": [k.to_dict() for k in self.partition_by],
+            "sorted_by": [k.to_dict() for k in self.sorted_by],
+        }
+
     @property
     def is_partitioned(self) -> bool:
         return bool(self.partition_by)

tests/test_databases.py

@@ -489,15 +489,19 @@ def create_database(self, request):
 
 
 def test_managed_table_layout_reads_it_back():
-
     parts, sorts = _layout()
     client = _client()
     info = SimpleNamespace(
         table="files", var_schema="public", partition_by=parts, sorted_by=sorts
     )
-    other = SimpleNamespace(table="elsewhere", var_schema="public", partition_by=[], sorted_by=[])
+    captured = {}
+
+    class FakeInfoSchema:
+        def information_schema(self, **kwargs):
+            captured.update(kwargs)
+            return SimpleNamespace(tables=[info])
 
-    with patch.object(client, "iter_tables", return_value=iter([other, info])), patch.object(
+    with patch.object(client, "_information_schema", return_value=FakeInfoSchema()), patch.object(
         client, "_as_managed_database", return_value=_MANAGED_DB
     ):
         layout = client.managed_table_layout("db_1", "files")
@@ -507,21 +511,53 @@ def test_managed_table_layout_reads_it_back():
     assert [s.column for s in layout.sorted_by] == ["event_time"]
     assert layout.is_partitioned is True
     assert layout.is_sorted is True
+    # Filtered server-side, not by paging the whole listing.
+    assert captured["var_schema"] == "public"
+    assert captured["table"] == "files"
+    assert captured["connection_id"] == _MANAGED_DB.default_connection_id
+    # to_dict flattens the pydantic keys rather than copying them through.
+    assert layout.to_dict()["partition_by"] == [
+        {"column": "event_date", "transform": "identity"}
+    ]
 
 
 def test_managed_table_layout_distinguishes_absent_from_unpartitioned():
-    """KeyError for a missing table, empty lists for one declared without a
-    layout. Collapsing the two would let a caller treat "table isn't there" as
-    "confirmed no layout" and load into something it never verified."""
-
+    """KeyError for a table the server does not report, empty lists for one
+    declared without a layout. Collapsing the two would let a caller read "table
+    isn't there" as "confirmed no layout" and load into something unverified.
+
+    Each call gets a FRESH response via side_effect. Sharing one would leave the
+    second assertion inspecting a consumed result rather than the case named
+    here.
+    """
     client = _client()
     plain = SimpleNamespace(table="files", var_schema="public", partition_by=[], sorted_by=[])
 
-    with patch.object(client, "iter_tables", return_value=iter([plain])), patch.object(
+    def respond(**kwargs):
+        found = [plain] if kwargs.get("table") == "files" else []
+        return SimpleNamespace(tables=found)
+
+    api = SimpleNamespace(information_schema=respond)
+    with patch.object(client, "_information_schema", return_value=api), patch.object(
         client, "_as_managed_database", return_value=_MANAGED_DB
     ):
         layout = client.managed_table_layout("db_1", "files")
         assert layout.is_partitioned is False and layout.is_sorted is False
 
         with pytest.raises(KeyError, match="missing"):
             client.managed_table_layout("db_1", "missing")
+
+
+def test_create_managed_database_refuses_layout_for_a_table_it_is_not_creating():
+    """A typo'd table name in the layout mapping would otherwise be dropped, and
+    the table it was meant for created flat — permanently, since a layout cannot
+    be added later."""
+    parts, _ = _layout()
+    client = _client()
+
+    with patch.object(client, "_databases_api") as dbs:
+        with pytest.raises(ValueError, match="fils"):
+            client.create_managed_database(
+                "demo", tables=["files"], partition_by={"fils": parts}
+            )
+    dbs.return_value.create_database.assert_not_called()