class FinalProjectNode extends ProjectNode implements VerifiableNode {

  private Pair<LogicalSchema, List<SelectExpression>> build(final MetaStore metaStore) {
    final LogicalSchema parentSchema = getSource().getSchema();
    final Optional<LogicalSchema> targetSchema = getTargetSchema(metaStore);

    final List<SelectExpression> selectExpressions = SelectionUtil
        .buildSelectExpressions(getSource(), projection.selectItems(), targetSchema);

    final LogicalSchema schema =
        SelectionUtil.buildProjectionSchema(parentSchema, selectExpressions, metaStore);

    if (into.isPresent()) {
      // Persistent queries have key columns as key columns - so final projection can exclude them:
      selectExpressions.removeIf(se -> {
        if (se.getExpression() instanceof UnqualifiedColumnReferenceExp) {
          final ColumnName columnName = ((UnqualifiedColumnReferenceExp) se.getExpression())
              .getColumnName();

          // Window bounds columns are currently removed if not aliased:
          if (SystemColumns.isWindowBound(columnName) && se.getAlias().equals(columnName)) {
            return true;
          }

          return parentSchema.isKeyColumn(columnName);
        }
        return false;
      });
    }

    final LogicalSchema nodeSchema;
    if (into.isPresent()) {
      nodeSchema = schema.withoutPseudoAndKeyColsInValue();
    } else {
      // Transient queries return key columns in the value, so the projection includes them, and
      // the schema needs to include them too:
      final Builder builder = LogicalSchema.builder();

      builder.keyColumns(parentSchema.key());

      schema.columns()
          .forEach(builder::valueColumn);

      nodeSchema = builder.build();
    }

    return Pair.of(nodeSchema, selectExpressions);
  }

  public  FinalProjectNode(
      final PlanNodeId id,
      final PlanNode source,
      final List<SelectItem> selectItems,
      final Optional<Analysis.Into> into,
      final MetaStore metaStore
  );

}