Function topKWithIndex

packages/db-ivm/src/operators/topK.ts:59–91

(
  comparator: (a: V1Type, b: V1Type) => number,
  options?: TopKOptions,
)

 * @returns A piped operator that orders the elements and limits the number of results
 */
export function topKWithIndex<
  KType extends T extends KeyValue<infer K, infer _V> ? K : never,
  V1Type extends T extends KeyValue<KType, infer V> ? V : never,
  T,
>(
  comparator: (a: V1Type, b: V1Type) => number,
  options?: TopKOptions,
): PipedOperator<T, KeyValue<KType, [V1Type, number]>> {
  const limit = options?.limit ?? Infinity
  const offset = options?.offset ?? 0

  return (
    stream: IStreamBuilder<T>,
  ): IStreamBuilder<KeyValue<KType, [V1Type, number]>> => {
    const reduced = stream.pipe(
      reduce<KType, V1Type, [V1Type, number], T>((values) => {
        // `values` is a list of tuples, first element is the value, second is the multiplicity
        const consolidated = new MultiSet(values).consolidate()
        let i = offset
        const sortedValues = consolidated
          .getInner()
          .sort((a, b) => comparator(a[0], b[0]))
          .slice(offset, offset + limit)
          .map(([value, multiplicity]): [[V1Type, number], number] => [
            [value, i++],
            multiplicity,
          ])
        return sortedValues
      }),
    )
    return reduced
  }
}