class TableFunctionCollectSource implements CollectSource {

    private final NodeContext nodeCtx;
    private final InputFactory inputFactory;

    @Inject
    public TableFunctionCollectSource(NodeContext nodeCtx) {
        this.nodeCtx = nodeCtx;
        inputFactory = new InputFactory(nodeCtx);
    }

    @Override
    @SuppressWarnings("unchecked")
    public CompletableFuture<BatchIterator<Row>> getIterator(TransactionContext txnCtx,
                                                             CollectPhase collectPhase,
                                                             CollectTask collectTask,
                                                             boolean supportMoveToStart) {<FILL_FUNCTION_BODY>}
}

        TableFunctionCollectPhase phase = (TableFunctionCollectPhase) collectPhase;
        TableFunctionImplementation<?> functionImplementation = phase.functionImplementation();
        RowType rowType = functionImplementation.returnType();

        //noinspection unchecked  Only literals can be passed to table functions. Anything else is invalid SQL
        List<Input<?>> inputs = (List<Input<?>>) (List<?>) phase.functionArguments();

        List<Input<?>> topLevelInputs = new ArrayList<>(phase.toCollect().size());
        List<String> columns = rowType.fieldNames();
        InputFactory.Context<RowCollectExpression> ctx = inputFactory.ctxForRefs(
            txnCtx,
            ref -> {
                for (int i = 0; i < columns.size(); i++) {
                    String column = columns.get(i);
                    if (ref.column().isRoot() && ref.column().name().equals(column)) {
                        return new RowCollectExpression(i);
                    }
                }
                throw new IllegalStateException(
                    "Column `" + ref + "` not found in " + functionImplementation.signature().getName().displayName());
            });
        for (Symbol symbol : phase.toCollect()) {
            topLevelInputs.add(ctx.add(symbol));
        }

        var inputRow = new ValueAndInputRow<>(topLevelInputs, ctx.expressions());
        Input<Boolean> condition = (Input<Boolean>) ctx.add(phase.where());
        Iterable<Row> rows = () -> StreamSupport
            .stream(functionImplementation.evaluate(txnCtx, nodeCtx, inputs.toArray(new Input[0])).spliterator(), false)
            .map(inputRow)
            .filter(InputCondition.asPredicate(condition))
            .iterator();
        return CompletableFuture.completedFuture(
            InMemoryBatchIterator.of(rows, SentinelRow.SENTINEL, functionImplementation.hasLazyResultSet())
        );
    
<no_super_class>