[BUG] Filter + JoinExternal creates key errors

[bschifferer]

Describe the bug
If I create a Filter op followed with join_external, which uses columns_ext as a parameter, the workflow produces a key error

KeyError: 'ext_col3'

Steps/Code to reproduce bug

import pandas as pd
import cudf

import nvtabular as nvt

df = cudf.DataFrame({
    'col1': [0,0,0,0,0,0],
    'col2': [1,2,3,4,5,6],
    'col3': [2,2,2,2,2,2]
})

df_ext = pd.DataFrame({
    'col1': [0,0,0,0,0,0],
    'ext_col2': [1,1,1,1,1,1],
    'ext_col3': [2,2,2,2,2,2]
})

df.to_parquet('df.parquet')

out = ['col1', 'col2', 'col3'] >> nvt.ops.Filter(lambda df: df['col2'].isin([1,2])) >> nvt.ops.JoinExternal(df_ext=df_ext,on='col1', columns_ext=['col1', 'ext_col2'])

dataset = nvt.Dataset('df.parquet')

workflow = nvt.Workflow(out)
workflow.transform(dataset).to_parquet('./test/')

Additional context
If I do not use the filter op, it works
If I dont use the columns_ext parameter, it works too

Aha! Link: https://nvaiinfra.aha.io/features/MERLIN-507

[karlhigley]

This is probably related to recent changes to the Workflow graph, so I'm assigning the two of us who worked on that to triage and figure out what's going on here.

[karlhigley]

This works on the main branch after merging #1194