How to properly add a child generator (sub-flow) to an Observable on RxJS?

[pedrokehl]

I'm using RxJS 7, and I would like to have a child generator (another Observable) emitting values based on the parent data.

Constraints:

• Emitted values from the Parent needs to be available to child generator;
• Parent needs to know when child flow is done;
• Child Observable can have many operators;
• Efficient!

I was already able to achieve this, in two different implementations, but I believe I'm not using here the full potential RxJS has.
The reason I'm looking for a more efficient implementation is because it's intended for a data intensive system which can have millions of items flowing, so every save in CPU usage and time has a massive impact :)

Implementation 1:
It's not efficient because it needs to build a new RxJS pipeline for every parent item, which can be expensive in my system where we can have a good number of "steps", also the "pipeline builder" does more than just call RxJS, it also has some logging structure to create and all that.

const { from, mergeMap, reduce, lastValueFrom } = rxjs

function run() {
  const parentData = [{ parentId: 1 }, { parentId: 2 }, { parentId: 3 }]

  from(parentData)
    .pipe(mergeMap((parent) => lastValueFrom(getChildFlow(parent))))
    .subscribe((parent) => console.log(parent))
}

function getChildFlow(parent) {
  return from(childGenerator(parent))
    .pipe(reduce((acc, value) => {
      acc.inner.push(value)
      return acc
    }, { inner: [] }))
}

async function* childGenerator(parentData) {
  for await (const index of [1, 2, 3]) {
    yield { childId: index, ...parentData }
  }
}

run()

Implementation 2:
The child pipeline is reused, but the efficiency is even worst when I do some benchmarking because of all the manipulations we are doing with the groupBy, reduce and zip.

const { from, mergeMap, tap, zip, map, groupBy, reduce } = rxjs

function run() {
  const parent$ = from([{ parentId: 1 }, { parentId: 2 }, { parentId: 3 }])
    .pipe(tap(doWhatever))

  const reducer = reduce(accumulator, [])

  const child$ = parent$
    .pipe(mergeMap(childGenerator))
    .pipe(tap(doWhatever))
    .pipe(groupBy((p) => p.parentId))
    .pipe(mergeMap((group$) => group$.pipe(reducer)))

  zip([parent$, child$])
    .pipe(map((results) => ({ ...results[0], inner: results[1] })))
    .pipe(tap(doWhatever))
    .subscribe(console.log)
}

function accumulator(acc, cur) {
  return [...acc, cur]
}

function doWhatever() {}

async function* childGenerator(parentData) {
  for await (const index of [1, 2, 3]) {
    yield { childId: index, ...parentData }
  }
}

run()

My question is:
Can I have the FIRST implementation reuse the pipeline somehow and just change the from behavior for each Parent Item received?