private Operator constructFilterOperator(
      Expression predicate,
      Operator inputOperator,
      Expression[] projectExpressions,
      List<TSDataType> inputDataTypes,
      Map<String, List<InputLocation>> inputLocations,
      boolean isKeepNull,
      PlanNodeId planNodeId,
      Ordering scanOrder,
      LocalExecutionPlanContext context) {
    final Map<NodeRef<Expression>, TSDataType> expressionTypes = new HashMap<>();
    if (context.isBuildPlanUseTemplate()) {
      ExpressionTypeAnalyzer.analyzeExpression(
          expressionTypes,
          predicate,
          new ExpressionTypeAnalyzer.TemplateTypeProvider(
              context.getTypeProvider().getTemplatedInfo().getSchemaMap()));
    } else {
      ExpressionTypeAnalyzer.analyzeExpression(
          expressionTypes,
          predicate,
          new ExpressionTypeAnalyzer.TypeProviderWrapper(
              context.getTypeProvider().getTreeModelTypeMap()));
    }

    // check whether predicate contains Non-Mappable UDF
    if (!predicate.isMappable(expressionTypes)) {
      throw new UnsupportedOperationException("Filter can not contain Non-Mappable UDF");
    }

    final List<TSDataType> filterOutputDataTypes = new ArrayList<>(inputDataTypes);

    for (Expression projectExpression : projectExpressions) {
      if (context.isBuildPlanUseTemplate()) {
        ExpressionTypeAnalyzer.analyzeExpression(
            expressionTypes,
            projectExpression,
            new ExpressionTypeAnalyzer.TemplateTypeProvider(
                context.getTypeProvider().getTemplatedInfo().getSchemaMap()));
      } else {
        ExpressionTypeAnalyzer.analyzeExpression(
            expressionTypes,
            projectExpression,
            new ExpressionTypeAnalyzer.TypeProviderWrapper(
                context.getTypeProvider().getTreeModelTypeMap()));
      }
    }

    boolean hasNonMappableUdf = false;
    for (Expression expression : projectExpressions) {
      if (!expression.isMappable(expressionTypes)) {
        hasNonMappableUdf = true;
        break;
      }
    }

    // init UDTFContext
    UDTFContext filterContext = new UDTFContext(context.getZoneId());
    filterContext.constructUdfExecutors(new Expression[] {predicate});

    // records LeafColumnTransformer of filter
    List<LeafColumnTransformer> filterLeafColumnTransformerList = new ArrayList<>();

    // records common ColumnTransformer between filter and project expressions
    List<ColumnTransformer> commonTransformerList = new ArrayList<>();

    // records LeafColumnTransformer of project expressions
    List<LeafColumnTransformer> projectLeafColumnTransformerList = new ArrayList<>();

    // records subexpression -> ColumnTransformer for filter
    Map<Expression, ColumnTransformer> filterExpressionColumnTransformerMap = new HashMap<>();

    ColumnTransformerVisitor visitor = new ColumnTransformerVisitor();

    ColumnTransformerVisitor.ColumnTransformerVisitorContext filterColumnTransformerContext =
        new ColumnTransformerVisitor.ColumnTransformerVisitorContext(
            filterContext,
            expressionTypes,
            filterLeafColumnTransformerList,
            inputLocations,
            filterExpressionColumnTransformerMap,
            ImmutableMap.of(),
            ImmutableList.of(),
            ImmutableList.of(),
            0,
            null);

    ColumnTransformer filterOutputTransformer =
        visitor.process(predicate, filterColumnTransformerContext);

    List<ColumnTransformer> projectOutputTransformerList = new ArrayList<>();

    Map<Expression, ColumnTransformer> projectExpressionColumnTransformerMap = new HashMap<>();

    // init project transformer when project expressions are all mappable
    if (!hasNonMappableUdf) {
      // init project UDTFContext
      UDTFContext projectContext = new UDTFContext(context.getZoneId());
      projectContext.constructUdfExecutors(projectExpressions);

      ColumnTransformerVisitor.ColumnTransformerVisitorContext projectColumnTransformerContext =
          new ColumnTransformerVisitor.ColumnTransformerVisitorContext(
              projectContext,
              expressionTypes,
              projectLeafColumnTransformerList,
              inputLocations,
              projectExpressionColumnTransformerMap,
              filterExpressionColumnTransformerMap,
              commonTransformerList,
              filterOutputDataTypes,
              inputLocations.size() - 1,
              null);

      for (Expression expression : projectExpressions) {
        projectOutputTransformerList.add(
            visitor.process(expression, projectColumnTransformerContext));
      }
    }

    final OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                planNodeId,
                FilterAndProjectOperator.class.getSimpleName());
    Operator filter =
        new FilterAndProjectOperator(
            operatorContext,
            inputOperator,
            filterOutputDataTypes,
            filterLeafColumnTransformerList,
            filterOutputTransformer,
            commonTransformerList,
            projectLeafColumnTransformerList,
            projectOutputTransformerList,
            hasNonMappableUdf,
            true);

    // Project expressions don't contain Non-Mappable UDF, TransformOperator is not needed
    if (!hasNonMappableUdf) {
      return filter;
    }

    // has Non-Mappable UDF, we wrap a TransformOperator for further calculation
    try {
      final OperatorContext transformContext =
          context
              .getDriverContext()
              .addOperatorContext(
                  context.getNextOperatorId(), planNodeId, TransformOperator.class.getSimpleName());
      return new TransformOperator(
          transformContext,
          filter,
          inputDataTypes,
          inputLocations,
          projectExpressions,
          isKeepNull,
          context.getZoneId(),
          expressionTypes,
          scanOrder == ASC);
    } catch (QueryProcessException e) {
      throw new RuntimeException(e);
    }
  }
This Visitor is responsible for transferring PlanNode Tree to Operator Tree.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java