Federated Joins at Scale: How We Join 70 Billion Rows We Can't Even Fetch

Security investigations and detections often require performing a large scale federated join between two different sources.

@OktaSystemLogs
| join kind=inner (@CrowdStrikeDetections) on user
| sort by timestamp desc
| take 100

Until recently we had the same limitation as many federated query engines: one side of the join had to be retrieved in one query and fit in memory. This post is about removing that requirement: how we join two very large tables that can't even be fully fetched, and the mistakes I made before getting to the final design.

How Federated Joins Work

Every engine that joins two tables does some version of the same thing: pick one side (the build side), collect its join keys, and use them against the other side (the probe side).

A classic hash join loads the build side into a hash map and streams the probe side through it. Federated engines add one trick on top: dynamic filtering. Instead of scanning the probe side in full and filtering in the engine, you take the keys from the build side and push them into the probe side's native query. Now the probe connector only ships rows that could match. At scale this is often the difference between reading gigabytes and reading megabytes.

Three problems show up when the join gets big on both sides:

1. 1.
   The filter must fit. The generated in (...) has to be possible to send to the probe connector: maximum query body sizes, maximum values per predicate, APIs that return creative errors past a few thousand list items. With enough distinct keys, the filter doesn't fit in on query.
   
2. 2.
   The build side must be fetched. All strategies assume you can fetch the build side in one query. Connectors cap the number of rows or bytes a single query may return. Microsoft Sentinel limits a query response body to 64mb or 500K rows (the lower of the two).
   
3. 3.
   The build side must fit. We hold the build side in memory, in a multi-tenant engine running many joins concurrently. We have to limit the memory we hold per query to avid OOM.

The Easy Wall: Filters That Don't Fit

We solved the first problem a while ago. When the build side produces more keys than the probe connector accepts in one query, we split the dynamic filter: the key set is rendered into N smaller filters, each sized to that connector's byte limits, sent as N queries whose results we union.

Everyone Assumes the build Side Fits

The other two problems are what this post is all about.

Say the build side is a month of authentication logs. 70 billion rows of nested JSON, a couple of kilobytes each, behind an API that throttles and caps every request.

For the memory problem, one popular escape in classical engines is spilling: when the hash table outgrows memory, write partitions to disk and process them in rounds. But this solution isn't good enough for federated query engines as it does nothing about the single query result cap: if the connector returns at most N rows per query, your disk-backed hash table receives at most N rows.

If the build side doesn't fit in one question, the answer is to ask more questions.

So we decided to page the build side.

The goal wasn't just to bypass these two limitations by fetching the entire build side with multiple requests- paging could solve a lot more. If pages had a global order, and every finished page extended a prefix of the final result, we could avoid holding the build side in memory *and* emit it to the client immediately, creating a stream of results.

Here is the loop we were aiming for:

1. 1.
   Fetch one page of the build side, not the entire row, just what we need for the next step.
2. 2.
   Collect the page's distinct join keys.
3. 3.
   Push them into the probe side as a dynamic filter, the same flow as before, only per page.
4. 4.
   Re-fetch the full build rows whose keys matched, join, and emit.

Every query in the loop fits within connector limits, and nothing bigger than a page is held in memory. The bigger advantage is that users see a stream of results unbelievably fast - queries matching billions of rows get results within seconds!

Going back to the implementation process, the next question was: what defines a page?

Window by Time

Every query already has a time range, so instead of fetching the entire build side at once, split its time range into windows. The idea felt reasonable but it failed for two reasons.

First, choosing the window size. Log volume isn't uniform, so we can't truly control the amount of data we fetch in each window. Even if we adapt the window size based on prior queries, we can never guarantee the next window is small enough. No matter the granularity, all the data could be concentrated in a single second, or a single nanosecond.

The second problem was global ordering. Time windows are only relevant to queries without a sort, or with exactly sort by timestamp desc. The page boundary can't be based on an arbitrary column, it must follow te query's sort order.

Page the Distinct Keys

Since the probe connector only needs distinct join keys, the cheapest page seemed to be a summarized one: page through the keys themselves, each carrying its position in the sort:

@OktaSystemLogs
| summarize timestamp=max(timestamp) by join_key
| sort by timestamp desc
| take 10000

But this has a global order problem of its own. Watch what happens with a page size of one:

If we page the distinct users with one row per page and take the first one, we retrieve the join key value ana. Then we re-fetch every row for ana, and we retrieve both the row at 12:00 and the row at 11:58. But the row at 11:58 belongs after the middle row at 11:59, whose join key hasn't been fetched yet. Deduplicating the keys before defining the page boundary has collapsed information required to preserve global row order.

The pager must operate on rows, not on summarized keys.

Keyset Pagination and Ties

Keyset pagination means you order the rows, take the top N, remember the last row, and ask the next query for rows strictly after it:

// page 2 of the build side, seeking past the last row of page 1
@OktaSystemLogs
| where timestamp < "12:00"
| sort by timestamp desc
| take 10000

So pages are rows in sort order, and the cursor is the last row's sort values. The problem: sort values aren't unique. Ten thousand logins can share a timestamp, the same second, the same batch. Look back at that seek predicate. You have two options, and both are wrong:

• timestamp < last skips every row and join key tied with the boundary row that didn't make the page, meaning missed rows and wrong join results.
• timestamp <= last re-reads the boundary rows on every page, and if a whole page shares one timestamp, the cursor never advances. You've built an infinite loop withan egress bill.
  

The standard fix is to make the cursor unique by appending a tiebreaker column. But we don't own these schemas. There is no rowid you can count on across Splunk, Sentinel, S3, and whatever gets connected to Vega next.

Since all we truly need from the page fetch is the distinct join keys within the cursor window, the join key is the perfect tiebreaker. Append it as the final cursor field and the cursor becomes a tuple, (query's sort columns..., join key), and te seek becomes a tuple comparison:

| where timestamp < "12:00"
     or (timestmp == "12:00" and user > "bob")

Rows tied on the sort get ordered by the join key, and the cursor advances through tie group instead of getting stuck in them.

Two Queries, Two Jobs

Pages should be cheap. As I mentioned earlier, the page scan doesn't need full rows, it needs specific columns:

• the cursor columns
• the join keys
• columns needed by engine-side operations that were not pushed down

I call this a thin page. It lets us fetch larger pages that still fit inside a connector's response byte caps: kilobytes per page instead of megabytes.

The thin page may cut in the middle of a tie group, and that's fine: its only job is to advance the cursor honestly and collect the distinct join keys in the window. The full build rows come from a second query, and that second query is where the tie problem is truly solved. The re-fetch asks for full rows by key inside a cursor window, with an inclusive upper bound and no top-N.

The window's inclusive edge covers the whole tie group, and with no row cap on the re-fetch, every twin comes back. The thin scan guarantees progress; the re-fetch guarantees completeness.

The Page Size

Page size shapes how the whole query behaves:

• A small page reduces latency: the thin scan and re-fetch have fewer rows to cover, and the probe side gets fewer keys to match on.
• A large page improves throughput: fewer connector round trips for the same data, and low match rates are processed more efficiently.

The best size depends on the connector, and on the data itself: the match rate, and how many distinct join keys hide in a page of rows. So the page size is dynamic. The first page is small, to get results in front of the user quickly, and the statistics of each page feed the size of the next one.

The Final Algorithm

As pseudocode:

cursor = start

loop:
    # 1. thin scan: cursor columns and join key only
    page = build.query(
        select cursor_columns,
        where cursor_columns not null and tuple > cursor,
        order by (sort_columns, join_key),
        take page_size,
    )
    if page is empty:
        stop

    keys = distinct join keys in page
    # the upper bound is inclusive
    window = (cursor, page.last_tuple]

    # 2. push the keys into the probe side, sliced to filter budgets
    matched = probe.query(where join_key in keys)

    # 3. re-fetch the matched rows at full width, no top-N
    rows = build.query(
        where tuple in window and join_key in matched,
    )

    emit(join(rows, matched))

    # a short page means the connector ran dry
    if page was short:
        stop
    cursor = page.last_tuple

That sums up the final paged join algorithm that allows us to treat large scale join as a stream. Queries that used to be capped by connector limits or the engine's memory or take hours to complete can now run at all and produce results fast.

If paging a database you do not own sounds like a cool problem to solve, we're hiring.