Adds duckdb-backed local adapter by will-langdale · Pull Request #625 · uktrade/matchbox

will-langdale · 2026-07-21T15:55:38Z

Adds a duckdb-backed local adapter.

Part of #560.

🛠️ Changes proposed in this pull request

Adds a local adapter ABC
Adds duckdb-backed adapter class
Adds duckdb ORM and tests. Note that autoincrement keys needed some interesting piping

👀 Guidance to review

I tried to factor out more stuff into the common SQL module. Do you agree with my decisions? Did I go too far, or not far enough? The principle I was trying to follow was to keep common subqueries separate to where they're hooked up to specific ORM tables or dialect mechanics. This is why inner functions of insert are abstracted rather than the whole thing.

Are you happy with how components in the two ORMs are shared? I was sceptical of the ORM as there's some repeated code in things like SourceConfigs, but given the alternatives (a different type of mixins, subclasses), this way seems smallest, seeing your ORM in one place is good, and I didn't want to add leaky abstractions.

Do you agree with the decision not to have a package duckdb singleton, so you can create lots of instances if you wish? We can't control what the user does and how they import like we do on the server.

Is it time to review how resolvers are stored locally, simplify query's logic, and cut it entirely from the server? Or is it simpler to mirror the logic here and pick it up later? Two facets: is it the right move AND will is swell the PR?

We implement _cascade_invalidate because duckdb (rightly) lacks delete cascade: it's just not made for that type of OLTP type work.

🤖 AI declaration

AI written but human reviewed throughout. Went through 2-3 iterations and spikes to get to this shape, including toys to validate duckdb mechanisms like sequences.

✅ Checklist:

This is the smallest, simplest solution to the problem
I've read our code standards and this code follows them
All new code is tested
I've updated all relevant documentation (select all that apply)
- API documentation (docstrings and indexes)
- Tutorials
- Developer docs

Signed-off-by: DBT pre-commit check

leo-mazzone · 2026-07-23T08:56:49Z

        backend.restore(snapshot=snapshot)
        _testkitdag_to_location(warehouse, dag_testkit)
+        if isinstance(backend, MatchboxLocalDBAdapter):
+            backend.bind(dag_testkit.dag)


note to self: what does binding do?

leo-mazzone · 2026-07-23T09:03:37Z

+@pytest.fixture(scope="function")
+def backend_instance(
+    request: pytest.FixtureRequest, backend: str
+) -> MatchboxClusterStoreAdapter:


note to self: what does the hierarchy of classes now look like?

leo-mazzone · 2026-07-23T09:05:18Z

+_NONEXISTENT_STEP = StepPath(
+    collection=CollectionName("local_test"), run=RunID(1), name="nonexistent"
+)


ignores them? What will the interface be for connecting to a DAG remotely?

The DAG object owns the collection and run ID -- that information will come from there.

leo-mazzone · 2026-07-23T09:12:39Z

+            assert self.backend.get_resolver_data(resolver.resolver.path).num_rows == 0
+
+    def test_clears_dependent_cache_only(self) -> None:
+        """Only cache entries depending on the dropped step are cleared."""


not an English sentence

"Only the caches of descendents of the dropped step are cleared"? In the test name too.

will-langdale · 2026-07-23T09:13:16Z

+            # 1) Bulk-insert new clusters: anti-join, ID omitted (sequence-assigned)
+            session.execute(
+                insert(tables.Clusters).from_select(
+                    ["cluster_hash"],
+                    select_new_cluster_hashes(resolver_hashes, hash_col="cluster_hash"),
+                )
+            )
+
+            # 2) Map parent_id to canonical cluster_id, now hashes exist
+            cluster_map = select_cluster_map(resolver_hashes)
+
+            # 3) ResolverClusters: distinct (step_id, cluster_id)
+            session.execute(
+                insert(tables.ResolverClusters).from_select(
+                    ["step_id", "cluster_id"],
+                    select_resolver_membership(step.step_id, cluster_map),
+                )
+            )
+
+            # 4) Contains: new (root, leaf) pairs, anti-joined against existing
+            candidate_contains = select_contains_pairs(
+                expanded_leaves, cluster_map
+            ).subquery()
+            session.execute(
+                insert(tables.Contains).from_select(
+                    ["root", "leaf"],
+                    select(candidate_contains.c.root, candidate_contains.c.leaf).where(
+                        ~exists(
+                            select(1).where(
+                                tables.Contains.c.root == candidate_contains.c.root,
+                                tables.Contains.c.leaf == candidate_contains.c.leaf,
+                            )
+                        )
+                    ),
+                )
+            )


Why does this follow a different order to PostgreSQL's insert_clusters()? Should something be factored out? Is exists going to work at scale?

leo-mazzone · 2026-07-23T09:13:51Z

+            assert self.backend.get_cached_query("independent") is not None
+
+    def test_raw_data_survives(self) -> None:
+        """drop_step_data doesn't touch RawData - it's canonical, not a cache."""


I actually don't like this behaviour. We might add or remove fields, want to run it again with fresh data. I think it should drop with the step.

leo-mazzone · 2026-07-23T09:14:41Z

@@ -0,0 +1,182 @@
+"""Tests for MatchboxLocalDuckDBLocalMixin: raw data, query cache, cascade."""


local/test_local_mixin.py

vs

local/data_mixin.py

?

Perhaps MatchboxLocalDuckDBLocalMixin is misnamed

Agree, will go for MatchboxLocalDuckDBCacheMixin.

will-langdale · 2026-07-23T09:15:17Z

+            # dump_tables only covers catalog rows - physical tables need
+            # dumping separately, same row format.
+            data["__dynamic_cache_tables__"] = {
+                name: _dump_dynamic_table(session, name) for name in physical_names
+            }


Could we solve this by just adding physical tables to the catalogue, and removing if steps change? Why handle separately?

No. We can reflect to dump, but not to restore, where we infer the schema from the JSON because the tables aren't defined. You need to create the table with the right columns... which is exactly what restore does, with the materialise helper.

leo-mazzone · 2026-07-23T09:18:37Z

@@ -0,0 +1,76 @@
+"""Tests for MatchboxLocalDuckDBDataMixin: replace-on-rerun.
+
+Local stores replace-on-rerun, the deliberate inversion of the server's


doesn't the server also replace on resync? I guess only if the signature or whatever is called changes?

I think we really need an idea of where we're going with regards to the server

Once you write a run, that's it, isn't it? Or you can overwrite it for a while then once it's default you need a new run.

leo-mazzone · 2026-07-23T09:23:24Z

+            assert (
+                self.backend.get_resolver_data(resolver.resolver.path).num_rows
+                == before
+            )


cascading isn't checked here?

TestDropStepData.test_cascades covers at a higher level.

will-langdale · 2026-07-23T09:26:23Z

+def compute_lineage(
+    graph: dict[StepName, list[StepName]],
+    resolver: StepName,
+    sources: list[StepName] | None = None,
+) -> list[StepName]:
+    """Ordered ancestor step names for `resolver`, closest first, self first.
+
+    When `sources` is given, ancestors are restricted to those on a path
+    to one of `sources`: kept if it is a source, or is itself downstream
+    of one.
+    """
+    ancestors = _ancestor_depths(graph, resolver)
+
+    if sources:
+        children: dict[StepName, list[StepName]] = {}
+        for name, parents in graph.items():
+            for parent in parents:
+                children.setdefault(parent, []).append(name)
+
+        allowed = set(sources)
+        for source in sources:
+            allowed.update(_ancestor_depths(children, source))
+        ancestors = {
+            name: depth for name, depth in ancestors.items() if name in allowed
+        }
+
+    ordered = sorted(ancestors.items(), key=lambda pair: (pair[1], pair[0]))
+    return [resolver, *(name for name, _ in ordered)]


Are we duplicating DAG._topological_sort() and the method that rely on it? How can we reduce this repetition?

Ideas:

bind() uses a whole DAG, and so can defer to its methods

An abstract common.graph module that centralises these algorithms and any coverage of them

Discussed with @leo-mazzone.

Backend belongs to the DAG

Like a Step, even though it belongs to the DAG, the backend can query the DAG

So topological algorithms belong to the DAG

Note: ensuring that DAG and backend are always one is crucial. If you delete a backend from a DAG, that backend shouldn't be able to call the DAG

will-langdale · 2026-07-23T09:28:12Z

+        """
+        self._graph_source = graph_source
+
+    def _graph(self) -> dict[StepName, list[StepName]]:


Is this a @property?

leo-mazzone · 2026-07-23T09:28:56Z

+        "result_id",
+        BigInteger,
+        primary_key=True,
+        autoincrement=False,


concerned about this autoincrement=False

joshwong-cs · 2026-07-23T09:30:58Z

+
+        for tbl in order:
+            ddl = str(
+                CreateTable(tbl, include_foreign_key_constraints=[]).compile(


CreateTable has an if_not_exists arg, do we want to toggle it on here?

leo-mazzone · 2026-07-23T09:31:18Z

+        prefixes = ("mb_raw_data__", "mb_query_cache__")
+        assert not any(n.startswith(prefixes) for n in names)


why is this so low level?

leo-mazzone · 2026-07-23T09:32:09Z

+def test_dump_restore_round_trips_dynamic_tables(
+    matchbox_local_duckdb: MatchboxLocalDuckDB, sqla_sqlite_warehouse: Engine
+) -> None:
+    """dump()/restore() carry RawData/QueryCache row content, not just pointers."""


leo-mazzone · 2026-07-23T10:29:48Z

+def _translate_lineage(
+    session: Session, lineage_names: list[str]
+) -> list[tuple[int, int | None]]:
+    """Translate step names from self.lineage() into (step_id, source_config_id).


self.lineage()? This ain't a class

joshwong-cs · 2026-07-23T10:32:46Z

+        self.clear(certain=True)
+
+        with self._session() as session:
+            restore_tables(


need to reset each sequence in SEQUENCE_COLUMNS to MAX(pk) + 1 since restore_tables() inserts explicit primary keys

leo-mazzone · 2026-07-23T10:34:22Z

@@ -0,0 +1,74 @@
+"""Ingest select builders, shared across relational backends.


more information needs to be given about

the structure of the incoming TableClause and the other two Subquerys

conceptually, what are these functions doing

otherwise it's really difficult to reuse functions from this file. Does this help my use case? Who knows

leo-mazzone · 2026-07-23T10:37:34Z

@@ -0,0 +1,74 @@
+"""Ingest select builders, shared across relational backends.
+


these functions assume a specific ORM structure, which I guess is fair for SQL adapters. But it seems we're also increasingly assuming all adapters will be relational?

leo-mazzone · 2026-07-23T10:39:42Z

+
+Handles only data movement: selecting every row and base64-encoding bytes
+columns for JSON-safety, or decoding and bulk-inserting them back.
+Engine-specific concerns stay with the caller.


for example? What's outside "data movement"? ID sequences?

…rity Signed-off-by: DBT pre-commit check

leo-mazzone · 2026-07-23T10:52:44Z

+        """Cache cleaned query data, keyed by query config plus upstream fingerprints.
+
+        Disposable: unlike insert_raw_data, this is just a cache, and can be
+        recomputed at any time.


what does "at any time" mean?

leo-mazzone · 2026-07-23T10:54:22Z

+    disposable query cache, and explicit cascade invalidation.
+
+    Local stores replace-on-rerun (with cascade invalidation of descendant
+    steps), unlike server backends, which are write-once per step per run.


like pointed out by many other comments elsewhere, this really does not clarify the invalidation model. I am confusion

leo-mazzone · 2026-07-23T10:56:30Z

+
+    _graph_source: HasGraph | None = None
+
+    def bind(self, graph_source: HasGraph) -> None:


bind_graph?

leo-mazzone · 2026-07-23T11:00:20Z

+
+
+class SourceConfigs(LocalBase):
+    """Degenerate one-row-per-source table.


degenerate?

leo-mazzone · 2026-07-23T11:01:40Z

+    Keeps the join shape identical to the server's, without needing the
+    rest of a real SourceConfig.


I don't understand what this means. Are these IDs the same you'll find on server?

leo-mazzone · 2026-07-23T11:05:15Z

+class Clusters(CountMixin, LocalBase):
+    """Table of indexed data and clusters that match it."""
+
+    __table__ = tables.Clusters.to_metadata(LOCAL_METADATA, schema=None)


is schema=None right? Where does the schema come from for these tables?

leo-mazzone · 2026-07-23T11:06:48Z

+LOCAL_TABLES = [
+    Steps.__table__,
+    SourceConfigs.__table__,
+    RawData.__table__,
+    QueryCache.__table__,
+    QueryCacheStep.__table__,
+]
+SHARED_TABLES = [
+    Clusters.__table__,
+    ClusterSourceKey.__table__,
+    Contains.__table__,
+    ModelEdges.__table__,
+    ResolverClusters.__table__,
+]


what are these two lists for?

…esitigial helper functions Signed-off-by: DBT pre-commit check

will-langdale changed the title ~~feat: adds duckdb local adapter~~ Adds duckdb-backed local adapter Jul 21, 2026

will-langdale force-pushed the feat/localmb-localadapter branch 2 times, most recently from 65c7dc0 to 8605086 Compare July 22, 2026 16:07

feat: adds duckdb local adapter

de3ab04

Signed-off-by: DBT pre-commit check

will-langdale force-pushed the feat/localmb-localadapter branch from 8605086 to de3ab04 Compare July 22, 2026 16:15

leo-mazzone reviewed Jul 23, 2026

View reviewed changes