exstream

API · Graph

sortedJoin()

Merge-join exactly two pre-sorted streams without collecting both inputs.

Example

const left = exstream([
  { id: 1, tenant: 'eu', value: 'a' },
  { id: 1, tenant: 'eu', value: 'duplicate' },
  { id: 2, tenant: 'us', value: 'b' },
]).uniq(['tenant', 'id'])

const right = exstream([
  { id: 1, label: 'one' },
  { id: 2, label: 'two' },
])

await left
  .sortedJoin(right, {
    leftKey: 'id',
    rightKey: 'id',
    type: 'left',
    order: 'asc',
  })
  .mapAsync(async (row) => row, { concurrency: 2, ordered: false })
  .pipeTo(destination('joined', { speed: Infinity }))

Parameters

right

Type ExstreamRequired

The sorted right input. The stream on which sortedJoin() is called is the left input.

leftKey, rightKey

Type function | property keyRequired

Selectors for the two join keys. A string may also address a nested field.

type

Allowed 'inner' | 'left' | 'right'Default 'inner'

Controls which unmatched side is emitted with the other field set to null.

order

Type 'asc' | 'desc' | comparatorDefault 'asc'

Must match both inputs. A custom comparator follows the same numeric convention as Array.sort(): negative places the left key first, zero joins the keys, and positive places the right key first.

Input and output

Both inputs must already be sorted according to order; sortedJoin() does not sort them. Output is { key, left, right }. Repeated keys produce every matching pair. A left join may set right to null, a right join may set left to null, and an inner join omits unmatched rows. TypeScript reflects those three result shapes.

The two inputs are pulled incrementally with backpressure. The left side is grouped for inner and left joins; the right side is grouped for right joins. Memory is therefore proportional to one adjacent duplicate group plus one record from the other side, not to the complete inputs. Matching pairs are emitted one at a time as downstream requests them. Output context aggregates the contributing row contexts.

Errors

Invalid inputs and options throw when the join is constructed. Selector, comparator, and input record errors enter the result error protocol. Fatal failures and cancellation close the join and release both inputs. Incorrect input order is not detected and produces incorrect results.

Forms

sortedJoin() is available only on a live Exstream. It cannot be recorded in a reusable pipeline because it connects two specific stream instances:

left.sortedJoin(right, {
  leftKey: 'id',
  rightKey: 'parentId',
  type: 'left',
})

Signature

sortedJoin<Right>(
  right: Exstream<Right>,
  options: {
    leftKey: ((value: Left, context: object) => unknown) | keyof Left
    rightKey: ((value: Right, context: object) => unknown) | keyof Right
    type?: 'inner' | 'left' | 'right'
    order?: 'asc' | 'desc' | ((leftKey: unknown, rightKey: unknown) => number)
  },
): Exstream<SortedJoinResult<Key, Left, Right, Type>>

For an inner join both values are present. A left join returns Right | null; a right join returns Left | null.

sortedGroupBy(), merge(), sort()