KudtfFlatMapper pushdown

[agavra]

When we build a logical plan, the first thing that we do is to apply the flat map operation on any UDTFs:

  public OutputNode buildPlan() {
    PlanNode currentNode = buildSourceNode();

    if (!analysis.getTableFunctions().isEmpty()) {
      currentNode = buildFlatMapNode(currentNode);
    }

    if (analysis.getWhereExpression().isPresent()) {
      currentNode = buildFilterNode(currentNode, analysis.getWhereExpression().get());
    }

    if (analysis.getGroupByExpressions().isEmpty()) {
      currentNode = buildProjectNode(currentNode);
    } else {
      currentNode = buildAggregateNode(currentNode);
    }

    return buildOutputNode(currentNode);
  }

This means that we will be applying all of the following nodes on the output of the flat map operation. This has a few issues:

• we may be evaluating complex expressions n-times more than we need to, where n is the number of outputs of the udtf. Take SELECT explode(COL1), udtfWithNetworkCall(COL2) FROM FOO; as an example. We should only evaluate udtfWithNetworkCall(COL2) once per input row since it is a static udf. Instead, with the current implementation, we call it once for each output of explode(COL1).
• we need to "special case" evaluation of complex expressions that are inputs to UDTFs (e.g. SELECT explode(my_udf(COL1)) (see feat: Implement complex expressions for table functions #3683 (comment)). Instead, I feel that this should be done as its own logical node step so that KudtfFlatMapper does not need to duplicate the expression execution logic.