Help Needed : Best practices for client-side metadata filtering in custom checkpoint implementations?

OSS Product Help LangGraph
,
1

Hi LangGraph team,

I’m implementing a custom BaseCheckpointSaver for a database that doesn’t support JSON path queries on large text columns (similar to how older database versions lack JSON support).

Context:

  • My database stores metadata as a large text field (JSON string)
  • Unlike SQLite’s json_extract() or PostgreSQL’s JSONB operators, there’s no SQL function to query JSON fields inside this column type
  • This means I can’t do server-side filtering like the built-in checkpointers do

So I need to:

  1. Fetch all checkpoints for a thread (filtered by thread_id and checkpoint_ns only)
  2. Apply the filter parameter client-side in Python
  3. Then apply limit after filtering

My current implementation:

def list(self, config, *, filter=None, before=None, limit=None):
rows = self._fetch_checkpoints(thread_id, checkpoint_ns, before_id)

  matched = 0
  for row in rows:
      checkpoint_tuple = self._row_to_tuple(row)
      
      # Client-side metadata filter
      if filter and not self._matches_metadata_filter(
          checkpoint_tuple.metadata, filter
      ):
          continue

      matched += 1
      yield checkpoint_tuple

      # Apply limit AFTER filtering
      if limit is not None and matched >= limit:
          break

def _matches_metadata_filter(self, metadata, filter):
“”“Check if metadata matches all filter conditions.”“”
return all(metadata.get(k) == v for k, v in filter.items())

Questions:

  1. Is this approach acceptable? The conformance tests pass, but I want to ensure I’m not missing any edge cases or violating expectations about the filter parameter.
  2. Performance considerations: Most queries will fetch ~100 checkpoints per thread before filtering. Is there a recommended threshold where client-side filtering becomes problematic?
  3. Filter semantics: Should I support only exact equality (metadata.get(k) == v), or are there other operators I should handle for compatibility?
  4. Limit behavior: I’m applying limit AFTER metadata filtering (so if filter matches 10 rows, limit=5 returns 5 rows). Is this the expected behavior, or should limit be applied before
    filtering?
  5. Empty filter: When filter=None or filter={}, I skip filtering entirely. Correct?
  6. Nested keys: Should I handle nested JSON paths like {“user.role”: “admin”} or only top-level keys?

Any guidance on client-side filtering best practices would be appreciated!

Additional info:

  • All 9 conformance tests pass with this implementation
  • Typical use case: ~10-50 checkpoints per thread
  • Most metadata filters are simple equality checks on 1-2 keys

Thanks!

2

Your approach is fine. Client side filtering is the expected fallback when the storage layer can’t push the predicate down, and nothing in BaseCheckpointSaver requires filtering to happen in SQL. A few notes on the specifics, based on how InMemorySaver, the SQLite saver, and the Postgres saver actually behave.

4. Limit after filtering. You have this right.
InMemorySaver decrements its limit counter only after the metadata check passes, so limit=5 with 10 matches returns 5 matches, not 5 rows scanned. Same for before, which is exclusive (checkpoint_id < before_id).

5. Empty filter. Also right.
Every built in saver guards with if filter:, which is falsy for both None and {}, so skipping entirely is the correct behavior.

3. Filter semantics. Exact equality on top level keys is all you need.
InMemorySaver does literally query_value == metadata.get(query_key). Postgres uses JSONB containment (metadata @> filter), which is slightly looser for nested objects, but for scalar values on top level keys the three backends agree. I wouldn’t invent operators. Nothing in LangGraph emits them today, and if the interface ever formalizes them you’d be stuck with conflicting semantics.

6. Nested keys. Top level only.
InMemorySaver does metadata.get("user.role"), a literal key lookup, and Postgres containment treats it the same way. SQLite is the odd one out because it interpolates the key into json_extract(metadata, '$.user.role'), so dotted keys accidentally resolve as nested paths there. That’s an inconsistency between backends, not a contract. Don’t implement it.

1. The edge cases I’d actually worry about are in your list signature, not your filter logic:

rows = self._fetch_checkpoints(thread_id, checkpoint_ns, before_id)

This assumes all three are present. In the reference savers:

thread_ids = (config["configurable"]["thread_id"],) if config else self.storage
  1. config=None means list across all threads. Used by aget_state_history style sweeps and some tooling.
  2. config without checkpoint_ns means all namespaces for that thread, not just "". Note the check is is not None, so an explicit "" filters to the root namespace but a missing key does not filter at all.
  3. config with a checkpoint_id means return only that checkpoint.

Also make sure you sort ORDER BY checkpoint_id DESC. Newest first is assumed by callers and by before based pagination, and it isn’t stated in the docstring.

2. Performance.
10 to 50 rows per thread is nothing, don’t optimize it. The threshold that matters isn’t the per thread count, it’s the unbounded case: list(None) on a busy database, or one long lived thread with thousands of checkpoints. Rather than picking a row count threshold, I’d fetch in pages and stop early:

def list(self, config, *, filter=None, before=None, limit=None):
    matched = 0
    for row in self._fetch_paged(..., page_size=500):
        tup = self._row_to_tuple(row)
        if filter and not self._matches(tup.metadata, filter):
            continue
        yield tup
        matched += 1
        if limit is not None and matched >= limit:
            return

Because you’re already a generator and limit short circuits, a paged fetch means the common case (small limit, match found early) never reads the whole thread. Keyset pagination on checkpoint_id works cleanly here since the IDs sort lexicographically.

One cheap optimization if your text column supports LIKE: for scalar filters you can prefilter server side with something like metadata LIKE '%"source":"loop"%' to cut the candidate set, then confirm in Python. It’s a lossy prefilter, so the Python check stays authoritative, but it can drop the row count by an order of magnitude.

Happy to look at the paging helper if you want to post it.

4

Thanks for the detailed response! This is super helpful.

I want to clarify my implementation approach and use case:

Use Case:

I mentioned “~10-50 checkpoints per thread” as typical, but I should have been clearer - we do have **long-running tasks** where threads can accumulate **hundreds or thousands of checkpoints** over time (multi-hour agent runs with frequent state saves).

Current Implementation:

My approach differs slightly from the paged fetching pattern you described, because I’m using an **HTTP-based checkpoint saver** backed by an OData service. Here’s what I’m doing:

1. Keyset pagination - I use the `before` parameter to implement cursor-based pagination:

```python

# Filter: snapshotId < before_id (lexicographic ordering)

params[“$filter”] = f"threadId eq ‘{thread_id}’ and snapshotId lt ‘{before_id}’"

```

2. Server-side limit - The `limit` parameter translates to OData `$top`:

```python

params[“$top”] = str(limit) # Database applies limit

```

3. Client-side metadata filtering - Since my database stores metadata as a large text column (no JSON path query support):

```python

r = client.get(url, params=params)

rows = r.json().get(“value”, [])

# Filter metadata on the client side

for row in rows:

if filter and not self._matches_metadata_filter(row[“metadata”], filter):

continue

yield self._row_to_tuple(row, write_rows)

```

4. Single HTTP request per page - Not row-by-row streaming:

- I fetch a page (up to `$top` rows) in one HTTP request

- Then apply metadata filtering in Python

- If `limit=10` but only 5 match the metadata filter after fetching, the caller gets 5 (not 10)

Question:

Is this approach acceptable for the `list()` contract? The key difference from your suggested pattern:

- I’m doing **page-based HTTP fetching** (via `before` + `limit` cursor pagination)

- But NOT **database row-by-row streaming** with in-Python filtering as you described

The caller would need to handle pagination explicitly if they want exactly N matches:

```python

before = None

matches = []

while len(matches) < desired_count:

batch = list(config, filter={...}, before=before, limit=100)

if not batch:

break

matches.extend(batch)

before = batch\[-1\].config  # Next page cursor

```

Does this HTTP-based pagination approach fit within the expected `list()` semantics, or should I implement additional logic to fetch more pages automatically when metadata filtering reduces the result count below the requested `limit`?

Thanks again!

5

The pagination itself is fine. But pushing limit into $top before the metadata filter runs is a correctness bug, not a stylistic difference from what I described. That’s the one thing I’d change.

Why it breaks. limit in the list() contract is a cap on matches returned, not on rows scanned. Your flow is $top=10 then filter, so a caller asking for 10 matches gets however many of the newest 10 rows happen to match. If the filter is at all selective, that’s usually zero. And the caller can’t tell the difference between “only 3 matches exist” and “only 3 matches were in the page I looked at,” because both come back as a short list with no signal.

The worst case is limit=1, which is a natural way to ask “the most recent checkpoint where source is loop.” Your implementation answers “the newest checkpoint, but only if it happens to be a loop one,” which is a silently wrong answer.

The caller side loop won’t save you. I checked get_state_history in pregel/main.py:

# eagerly consume list() to avoid holding up the db cursor
for checkpoint_tuple in list(
    checkpointer.list(config, before=before, limit=limit, filter=filter)
):

filter and limit go straight through in a single call and the generator is drained immediately. There’s no pagination wrapper, and aget_state_history does the same. So the primary caller of your saver is LangGraph itself, and it will never run the while len(matches) < desired_count loop from your example. Any pagination has to live inside list().

The fix is to stop conflating page size with limit. $top becomes your page size; limit stays a match counter:

def list(self, config, *, filter=None, before=None, limit=None):
    # No metadata filter means every row is a match, so the server
    # can apply the limit directly. This is the fast path.
    page_size = max((limit or 0) * 4, 200) if filter else (limit or 500)

    cursor = get_checkpoint_id(before) if before else None
    matched = 0

    while True:
        rows, next_link = self._fetch_page(config, cursor=cursor, top=page_size)
        if not rows:
            return

        for row in rows:
            if filter and not self._matches_metadata_filter(row["metadata"], filter):
                continue
            yield self._row_to_tuple(row, self._fetch_writes(row))
            matched += 1
            if limit is not None and matched >= limit:
                return

        cursor = rows[-1]["snapshotId"]
        if next_link is None and len(rows) < page_size:
            return

For an unfiltered request this is exactly one HTTP call, same as today. For a filtered one it keeps pulling pages until it satisfies limit or the thread runs out, which is what callers expect.

One OData specific trap. Don’t terminate on len(rows) < page_size alone. Many OData services do server driven paging and silently cap $top, so a full page can come back shorter than you asked for and you’d stop early with results still available. Follow @odata.nextLink when the service returns it and treat the short page check as a fallback only.

Given thousands of checkpoints per thread, push a lossy prefilter down. OData v4 has contains(), which works on your text column:

if filter:
    for k, v in filter.items():
        if isinstance(v, (str, int, bool)):
            needle = json.dumps({k: v})[1:-1]  # '"source":"loop"'
            clauses.append(f"contains(metadata,'{_odata_str(needle)}')")

Keep _matches_metadata_filter as the authority since contains can match substrings inside nested values, but it will usually cut the candidate set by an order of magnitude. Only enable it if you control metadata serialization end to end, since it depends on separator whitespace and key ordering. If you serialize with separators=(",", ":") you’re fine; otherwise gate it behind a flag.

Small thing I noticed in your $filter construction:

params["$filter"] = f"threadId eq '{thread_id}' and snapshotId lt '{before_id}'"

OData escapes a single quote by doubling it, so a thread_id containing ' will either break the query or inject into it. Worth a value.replace("'", "''") helper on every interpolated literal.

To answer your direct question: page based fetching is completely fine and is the right shape for an HTTP backend. You just need the page loop to be internal so that limit still means what callers think it means.

6

Thanks for sharing such a detailed write-up and the follow-up testing. I’m interested in seeing how this performs under a sustained production workload rather than short benchmarks. If you end up collecting metrics like request latency, cache hit rate, or memory usage after running it for a while, I’d be curious to see those results. Real-world numbers would make it much easier to compare this approach with the other checkpoint strategies.