diff --git a/pkg-py/src/commons/_backends.py b/pkg-py/src/commons/_backends.py
index d3192ae8..d5f234b8 100644
--- a/pkg-py/src/commons/_backends.py
+++ b/pkg-py/src/commons/_backends.py
@@ -65,8 +65,7 @@ def list_tables(self) -> list[str]:
         return [row[0] for row in rows]
 
     def quote(self, table_id: TableId) -> str:
-        parts = [table_id.schema, table_id.table] if table_id.schema else [table_id.table]
-        return ".".join(quote_identifier(part) for part in parts)
+        return ".".join(quote_identifier(part) for part in table_id.parts)
 
     def dialect(self) -> str:
         return "duckdb"
@@ -96,9 +95,16 @@ def list_tables(self) -> list[str]:
     def quote(self, table_id: TableId) -> str:
         preparer = self._engine.dialect.identifier_preparer
         quoted = preparer.quote(table_id.table)
-        if table_id.schema:
-            return f"{preparer.quote_schema(table_id.schema)}.{quoted}"
-        return quoted
+        if table_id.schema is None:
+            return quoted
+        # quote_schema() is given one component at a time: handed
+        # "ANALYTICS.PUBLIC" it produces one identifier containing a dot.
+        outer = ".".join(
+            preparer.quote_schema(part)
+            for part in (table_id.catalog, table_id.schema)
+            if part is not None
+        )
+        return f"{outer}.{quoted}"
 
     def dialect(self) -> str:
         return self._engine.dialect.name
@@ -107,6 +113,9 @@ def inspector(self) -> Callable[[TableId], bool] | None:
         inspector = sqlalchemy.inspect(self._engine)
 
         def exists(table_id: TableId) -> bool:
-            return inspector.has_table(table_id.table, schema=table_id.schema)
+            # SQLAlchemy takes every level above the table as one dotted
+            # `schema`, so a catalog is joined onto it rather than dropped.
+            outer = ".".join(table_id.parts[:-1])
+            return inspector.has_table(table_id.table, schema=outer or None)
 
         return exists
diff --git a/pkg-py/src/commons/_data_source.py b/pkg-py/src/commons/_data_source.py
index 102bca64..5d0583c2 100644
--- a/pkg-py/src/commons/_data_source.py
+++ b/pkg-py/src/commons/_data_source.py
@@ -41,15 +41,36 @@ def _fold(name: str) -> str:
 
 @dataclass(frozen=True)
 class TableId:
-    """A table's identity, schema-qualified where the backend has schemas."""
+    """A table's identity, qualified as far as the backend has levels.
+
+    Warehouses name a table `catalog.schema.table`, so the components are
+    kept apart rather than folded into one string. Folding them means quoting
+    `ANALYTICS.PUBLIC` as a single identifier, which names a schema with a dot
+    in it rather than a catalog and a schema.
+    """
 
     table: str
     schema: str | None = None
+    catalog: str | None = None
+
+    def __post_init__(self) -> None:
+        if self.catalog is not None and self.schema is None:
+            raise ValueError(
+                "A TableId with a catalog needs a schema: there is no level "
+                "between them to leave out."
+            )
+
+    @property
+    def parts(self) -> list[str]:
+        """The components, outermost first."""
+        return [
+            part for part in (self.catalog, self.schema, self.table) if part is not None
+        ]
 
     @property
     def label(self) -> str:
         """The name the agent uses for this table."""
-        return f"{self.schema}.{self.table}" if self.schema else self.table
+        return ".".join(self.parts)
 
 
 @dataclass
@@ -326,8 +347,9 @@ def _is_frame(value: Any) -> bool:
 def normalize_table_registry(tables: Any) -> dict[str, TableId]:
     """Turn a `tables` argument into label -> `TableId`.
 
-    Strings containing dots are read as schema-qualified. A literal table name
-    containing a dot is spelled as a `TableId`.
+    Strings containing dots are read as qualified names, at most three
+    parts: catalog.schema.table. A literal table name containing a dot is
+    spelled as a `TableId`.
     """
     if isinstance(tables, (str, TableId)):
         entries: list[Any] = [tables]
@@ -352,7 +374,7 @@ def normalize_table_registry(tables: Any) -> dict[str, TableId]:
 
 def _table_entry_id(entry: Any) -> TableId:
     if isinstance(entry, TableId):
-        if not entry.table or (entry.schema is not None and not entry.schema):
+        if not all(entry.parts):
             raise ValueError("TableId entries must not contain empty name components.")
         return entry
     if not isinstance(entry, str) or not entry:
@@ -366,9 +388,16 @@ def _table_entry_id(entry: Any) -> TableId:
             "Schema-qualified entries in tables must not contain empty name "
             f"components: {entry!r}."
         )
+    if len(parts) > 3:
+        raise ValueError(
+            "A table name has at most three parts, catalog.schema.table, got "
+            f"{entry!r}. Spell a name containing a dot as a TableId."
+        )
     if len(parts) == 1:
         return TableId(table=entry)
-    return TableId(table=parts[-1], schema=".".join(parts[:-1]))
+    if len(parts) == 2:
+        return TableId(table=parts[1], schema=parts[0])
+    return TableId(table=parts[2], schema=parts[1], catalog=parts[0])
 
 
 def _check_tables_exist(backend: Backend, registry: dict[str, TableId]) -> None:
diff --git a/pkg-py/tests/test_data_source_engine.py b/pkg-py/tests/test_data_source_engine.py
index d3114066..4f49a8ae 100644
--- a/pkg-py/tests/test_data_source_engine.py
+++ b/pkg-py/tests/test_data_source_engine.py
@@ -6,6 +6,7 @@
 import sqlalchemy
 
 from commons import data_source, list_tables
+from commons._backends import EngineBackend
 from commons._data_source import TableId, normalize_table_registry
 
 
@@ -41,11 +42,6 @@ def test_an_explicit_table_id_bypasses_dot_splitting() -> None:
     assert normalize_table_registry([literal]) == {"a.b": literal}
 
 
-def test_an_empty_name_component_is_rejected() -> None:
-    with pytest.raises(ValueError, match="empty name components"):
-        normalize_table_registry(["analytics."])
-
-
 def test_duplicate_labels_are_rejected() -> None:
     with pytest.raises(ValueError, match="duplicate labels"):
         normalize_table_registry(["sales", "sales"])
@@ -152,3 +148,67 @@ def broken_inspector(self: object) -> object:
 
     with pytest.raises(RuntimeError, match="connection reset"):
         data_source(engine, tables=["sales"])
+
+
+def test_the_inspector_looks_in_the_catalog_it_was_given(monkeypatch) -> None:
+    """SQLAlchemy takes every level above the table as one dotted `schema`.
+
+    Recording the argument is the only way to see this: no dialect available
+    to the test suite has three levels, so a real probe cannot distinguish
+    dropping the catalog from honouring it.
+    """
+    seen: list[str | None] = []
+
+    class _Recorder:
+        def has_table(self, table: str, schema: str | None = None) -> bool:
+            seen.append(schema)
+            return True
+
+    monkeypatch.setattr(sqlalchemy, "inspect", lambda _engine: _Recorder())
+    backend = EngineBackend(sqlalchemy.create_engine("sqlite://"))
+    exists = backend.inspector()
+    assert exists is not None
+    exists(TableId(table="ORDERS", schema="PUBLIC", catalog="ANALYTICS"))
+    assert seen == ["ANALYTICS.PUBLIC"]
+
+
+def test_a_three_part_name_quotes_each_component_separately() -> None:
+    import duckdb
+
+    from commons._backends import DuckDBBackend
+
+    table_id = normalize_table_registry("ANALYTICS.PUBLIC.ORDERS")[
+        "ANALYTICS.PUBLIC.ORDERS"
+    ]
+    quoted = DuckDBBackend(duckdb.connect()).quote(table_id)
+    assert quoted == '"ANALYTICS"."PUBLIC"."ORDERS"'
+
+
+def test_a_catalog_without_a_schema_is_refused() -> None:
+    # There is no level between them to leave out, so this is a typo rather
+    # than a shape to support.
+    with pytest.raises(ValueError, match="schema"):
+        TableId(table="t", catalog="c")
+
+
+def test_the_label_of_a_three_part_id_round_trips() -> None:
+    table_id = TableId(table="ORDERS", schema="PUBLIC", catalog="ANALYTICS")
+    assert table_id.label == "ANALYTICS.PUBLIC.ORDERS"
+    assert normalize_table_registry(table_id.label)[table_id.label] == table_id
+
+
+def test_the_engine_backend_quotes_each_component_separately() -> None:
+    # The preparer takes one component at a time; handed "ANALYTICS.PUBLIC" it
+    # produces a single identifier with a dot inside it.
+    backend = EngineBackend(sqlalchemy.create_engine("sqlite://"))
+    quoted = backend.quote(
+        TableId(table="ORDERS", schema="PUBLIC", catalog="ANALYTICS")
+    )
+    assert quoted == '"ANALYTICS"."PUBLIC"."ORDERS"'
+
+
+def test_the_engine_backend_still_quotes_a_two_part_name() -> None:
+    backend = EngineBackend(sqlalchemy.create_engine("sqlite://"))
+    assert backend.quote(TableId(table="sales", schema="analytics")) == (
+        "analytics.sales"
+    )
diff --git a/pkg-py/tests/test_table_names.py b/pkg-py/tests/test_table_names.py
new file mode 100644
index 00000000..e19e3e23
--- /dev/null
+++ b/pkg-py/tests/test_table_names.py
@@ -0,0 +1,55 @@
+"""Table-name parsing, driven by the shared fixture.
+
+How a `tables` string splits into catalog, schema, and table is a
+cross-language contract, so the cases live in
+``tests/shared/table-names.json`` and the R suite runs the same ones. Do not
+restate a case here; add it to the fixture.
+"""
+
+from typing import Any
+
+import pytest
+
+from commons._data_source import normalize_table_registry
+
+from ._shared import load_shared_fixture
+
+CASES: list[dict[str, Any]] = load_shared_fixture("table-names")[
+    "parse_table_name"
+]["cases"]
+
+# The fixture pins the refusal as a slug; the wording belongs to each
+# language.
+ERRORS: dict[str, tuple[type[Exception], str]] = {
+    "too_many_parts": (ValueError, "at most three parts"),
+    "empty_component": (ValueError, "empty name components"),
+    "not_a_name": (TypeError, "must be a table name"),
+}
+
+
+def test_shared_fixture_covers_every_shape() -> None:
+    # A truncated fixture would silently collect zero parametrized cases and
+    # the suite would still pass, so pin the coverage the table must have.
+    assert CASES
+    part_counts = {
+        len(case["expected"]) for case in CASES if "expected" in case
+    }
+    assert part_counts == {1, 2, 3}
+    assert {case["error"] for case in CASES if "error" in case} == set(ERRORS)
+
+
+@pytest.mark.parametrize("case", CASES, ids=lambda case: case["name"])
+def test_parse_matches_the_shared_fixture(case: dict[str, Any]) -> None:
+    if "error" in case:
+        error, match = ERRORS[case["error"]]
+        with pytest.raises(error, match=match):
+            normalize_table_registry(case["input"])
+        return
+
+    (table_id,) = normalize_table_registry(case["input"]).values()
+    expected = case["expected"]
+    assert table_id.catalog == expected.get("catalog")
+    assert table_id.schema == expected.get("schema")
+    assert table_id.table == expected["table"]
+    # The label is what the agent sees, so it round-trips the input.
+    assert table_id.label == case["input"]
diff --git a/pkg-r/R/data-source.R b/pkg-r/R/data-source.R
index 7d62dc66..d7b5839d 100644
--- a/pkg-r/R/data-source.R
+++ b/pkg-r/R/data-source.R
@@ -29,11 +29,11 @@
 #' @param tables Which tables to expose, used when a connection or a board is
 #'   supplied.
 #'
-#'   For a connection, a character vector of table names, schema-qualified
-#'   strings like `"schema.table"`, or `DBI::Id` objects. Defaults to every
-#'   table returned by [DBI::dbListTables()]. Strings containing dots are
-#'   interpreted as schema-qualified names; use `DBI::Id(table = "a.b")` for
-#'   literal table names containing dots. For Snowflake and Databricks
+#'   For a connection, a character vector of table names, qualified strings
+#'   like `"schema.table"` or `"catalog.schema.table"`, or `DBI::Id` objects.
+#'   Defaults to every table returned by [DBI::dbListTables()]. Strings
+#'   containing dots are interpreted as qualified names, at most three parts;
+#'   use `DBI::Id(table = "a.b")` for literal table names containing dots. For Snowflake and Databricks
 #'   connections, a `DBI::Id` ending in `catalog` or `schema` selects every
 #'   table and view in that namespace. Leaving `tables` unset selects the
 #'   current schema. A Databricks `hive_metastore` selection must include a
@@ -901,22 +901,33 @@ table_entry_id <- function(table, call = rlang::caller_env()) {
     )
   }
 
+  # strsplit() drops a trailing empty piece, so "orders." splits to "orders";
+  # check the trailing dot separately rather than accept it as a bare name.
   parts <- strsplit(table, ".", fixed = TRUE)[[1]]
-  if (any(parts == "")) {
+  if (any(parts == "") || endsWith(table, ".")) {
     cli::cli_abort(
       "Schema-qualified entries in {.arg tables} must not contain empty name components.",
       call = call
     )
   }
 
+  if (length(parts) > 3) {
+    cli::cli_abort(
+      c(
+        "A table name has at most three parts, catalog.schema.table: {.val {table}}.",
+        i = "Spell a name containing a dot as a {.cls DBI::Id}."
+      ),
+      call = call
+    )
+  }
   if (length(parts) == 1) {
     return(DBI::Id(table = table))
   }
+  if (length(parts) == 2) {
+    return(DBI::Id(schema = parts[[1]], table = parts[[2]]))
+  }
 
-  DBI::Id(
-    schema = paste(parts[-length(parts)], collapse = "."),
-    table = parts[[length(parts)]]
-  )
+  DBI::Id(catalog = parts[[1]], schema = parts[[2]], table = parts[[3]])
 }
 
 table_id_label <- function(id, call = rlang::caller_env()) {
diff --git a/pkg-r/man/data_source.Rd b/pkg-r/man/data_source.Rd
index b0469578..48cfbb6c 100644
--- a/pkg-r/man/data_source.Rd
+++ b/pkg-r/man/data_source.Rd
@@ -14,11 +14,11 @@ a table name the agent can query.}
 \item{tables}{Which tables to expose, used when a connection or a board is
 supplied.
 
-For a connection, a character vector of table names, schema-qualified
-strings like \code{"schema.table"}, or \code{DBI::Id} objects. Defaults to every
-table returned by \code{\link[DBI:dbListTables]{DBI::dbListTables()}}. Strings containing dots are
-interpreted as schema-qualified names; use \code{DBI::Id(table = "a.b")} for
-literal table names containing dots. For Snowflake and Databricks
+For a connection, a character vector of table names, qualified strings
+like \code{"schema.table"} or \code{"catalog.schema.table"}, or \code{DBI::Id} objects.
+Defaults to every table returned by \code{\link[DBI:dbListTables]{DBI::dbListTables()}}. Strings
+containing dots are interpreted as qualified names, at most three parts;
+use \code{DBI::Id(table = "a.b")} for literal table names containing dots. For Snowflake and Databricks
 connections, a \code{DBI::Id} ending in \code{catalog} or \code{schema} selects every
 table and view in that namespace. Leaving \code{tables} unset selects the
 current schema. A Databricks \code{hive_metastore} selection must include a
diff --git a/pkg-r/tests/testthat/fixtures/shared/table-names.json b/pkg-r/tests/testthat/fixtures/shared/table-names.json
new file mode 100644
index 00000000..b9409499
--- /dev/null
+++ b/pkg-r/tests/testthat/fixtures/shared/table-names.json
@@ -0,0 +1,51 @@
+{
+  "parse_table_name": {
+    "description": "How a `tables` string splits into name components. Warehouses name a table catalog.schema.table, so three parts are catalog, schema, and table; folding the outer parts into one schema would quote them as a single identifier containing a dot, which names a schema no warehouse has. More than three parts are refused because there is no fourth level to put them in. Errors are pinned as reason slugs; the wording belongs to each language.",
+    "cases": [
+      {
+        "name": "a bare name is just a table",
+        "input": "orders",
+        "expected": { "table": "orders" }
+      },
+      {
+        "name": "two parts are schema and table",
+        "input": "public.orders",
+        "expected": { "schema": "public", "table": "orders" }
+      },
+      {
+        "name": "three parts are catalog, schema, and table",
+        "input": "analytics.public.orders",
+        "expected": {
+          "catalog": "analytics",
+          "schema": "public",
+          "table": "orders"
+        }
+      },
+      {
+        "name": "four parts have no level to go in",
+        "input": "a.b.c.d",
+        "error": "too_many_parts"
+      },
+      {
+        "name": "an empty leading component",
+        "input": ".orders",
+        "error": "empty_component"
+      },
+      {
+        "name": "an empty middle component",
+        "input": "analytics..orders",
+        "error": "empty_component"
+      },
+      {
+        "name": "an empty trailing component",
+        "input": "orders.",
+        "error": "empty_component"
+      },
+      {
+        "name": "an empty string",
+        "input": "",
+        "error": "not_a_name"
+      }
+    ]
+  }
+}
diff --git a/pkg-r/tests/testthat/test-table-names.R b/pkg-r/tests/testthat/test-table-names.R
new file mode 100644
index 00000000..590da5e9
--- /dev/null
+++ b/pkg-r/tests/testthat/test-table-names.R
@@ -0,0 +1,52 @@
+# Table-name parsing, driven by the shared fixture.
+#
+# How a `tables` string splits into catalog, schema, and table is a
+# cross-language contract, so the cases live in tests/shared/table-names.json
+# and the Python suite runs the same ones. Do not restate a case here; add it
+# to the fixture.
+
+# The fixture pins the refusal as a slug; the wording belongs to each
+# language.
+table_name_error_patterns <- list(
+  too_many_parts = "at most three parts",
+  empty_component = "empty name components",
+  not_a_name = "must be a table name"
+)
+
+test_that("the shared fixture covers every shape", {
+  cases <- shared_fixture("table-names")$parse_table_name$cases
+  # A truncated fixture would silently collect zero cases and the suite
+  # would still pass, so pin the coverage the table must have.
+  expect_gt(length(cases), 0)
+
+  successes <- Filter(function(case) is.null(case$error), cases)
+  expect_setequal(
+    vapply(successes, function(case) length(case$expected), integer(1)),
+    c(1L, 2L, 3L)
+  )
+  errors <- Filter(function(case) !is.null(case$error), cases)
+  expect_setequal(
+    vapply(errors, function(case) case$error, character(1)),
+    names(table_name_error_patterns)
+  )
+})
+
+test_that("table_entry_id matches the shared fixture", {
+  cases <- shared_fixture("table-names")$parse_table_name$cases
+
+  for (case in cases) {
+    if (!is.null(case$error)) {
+      expect_error(
+        table_entry_id(case$input),
+        regexp = table_name_error_patterns[[case$error]],
+        fixed = TRUE
+      )
+      next
+    }
+
+    id <- table_entry_id(case$input)
+    expect_identical(as.list(id@name), case$expected, info = case$name)
+    # The label is what the agent sees, so it round-trips the input.
+    expect_identical(table_id_label(id), case$input, info = case$name)
+  }
+})
diff --git a/tests/shared/table-names.json b/tests/shared/table-names.json
new file mode 100644
index 00000000..b9409499
--- /dev/null
+++ b/tests/shared/table-names.json
@@ -0,0 +1,51 @@
+{
+  "parse_table_name": {
+    "description": "How a `tables` string splits into name components. Warehouses name a table catalog.schema.table, so three parts are catalog, schema, and table; folding the outer parts into one schema would quote them as a single identifier containing a dot, which names a schema no warehouse has. More than three parts are refused because there is no fourth level to put them in. Errors are pinned as reason slugs; the wording belongs to each language.",
+    "cases": [
+      {
+        "name": "a bare name is just a table",
+        "input": "orders",
+        "expected": { "table": "orders" }
+      },
+      {
+        "name": "two parts are schema and table",
+        "input": "public.orders",
+        "expected": { "schema": "public", "table": "orders" }
+      },
+      {
+        "name": "three parts are catalog, schema, and table",
+        "input": "analytics.public.orders",
+        "expected": {
+          "catalog": "analytics",
+          "schema": "public",
+          "table": "orders"
+        }
+      },
+      {
+        "name": "four parts have no level to go in",
+        "input": "a.b.c.d",
+        "error": "too_many_parts"
+      },
+      {
+        "name": "an empty leading component",
+        "input": ".orders",
+        "error": "empty_component"
+      },
+      {
+        "name": "an empty middle component",
+        "input": "analytics..orders",
+        "error": "empty_component"
+      },
+      {
+        "name": "an empty trailing component",
+        "input": "orders.",
+        "error": "empty_component"
+      },
+      {
+        "name": "an empty string",
+        "input": "",
+        "error": "not_a_name"
+      }
+    ]
+  }
+}