Method run

packages/db-ivm/src/operators/join.ts:84–128

()

  }

  run(): void {
    // Build deltas from input messages
    const deltaA = Index.fromMultiSets<K, V1>(
      this.inputAMessages() as Array<MultiSet<[K, V1]>>,
    )
    const deltaB = Index.fromMultiSets<K, V2>(
      this.inputBMessages() as Array<MultiSet<[K, V2]>>,
    )

    // Early-out if nothing changed
    if (deltaA.size === 0 && deltaB.size === 0) return

    const results = new MultiSet<any>()

    // Emit inner results (all modes except anti)
    if (this.#mode !== `anti`) {
      this.emitInnerResults(deltaA, deltaB, results)
    }

    // Emit left outer/anti results
    if (
      this.#mode === `left` ||
      this.#mode === `full` ||
      this.#mode === `anti`
    ) {
      this.emitLeftOuterResults(deltaA, deltaB, results)
    }

    // Emit right outer results
    if (this.#mode === `right` || this.#mode === `full`) {
      this.emitRightOuterResults(deltaA, deltaB, results)
    }

    // Update state and send results
    // IMPORTANT: All emissions use pre-append snapshots of indexA/indexB.
    // Now append ALL deltas to indices - this happens unconditionally for every key,
    // regardless of whether presence flipped. Consolidated multiplicity tracking is automatic.
    this.#indexA.append(deltaA)
    this.#indexB.append(deltaB)

    // Send results
    if (results.getInner().length > 0) {
      this.output.sendData(results)
    }
  }

  private emitInnerResults(
    deltaA: Index<K, V1>,