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 value columns - final projection can exclude them:
      final Map<ColumnName, Set<ColumnName>> seenKeyColumns = new HashMap<>();
      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;
          }

          if (parentSchema.isKeyColumn(columnName)) {
            seenKeyColumns.computeIfAbsent(columnName, k -> new HashSet<>()).add(se.getAlias());
            return true;
          }
        }
        return false;
      });

      for (final Entry<ColumnName, Set<ColumnName>> seenKey : seenKeyColumns.entrySet()) {
        if (seenKey.getValue().size() > 1) {
          final String keys = GrammaticalJoiner.and().join(
              seenKey.getValue().stream().map(Name::text).sorted());
          throw new KsqlException("The projection contains a key column (" + seenKey.getKey()
              + ") more than once, aliased as: "
              + keys + "."
              + System.lineSeparator()
              + "Each key column must only be in the projection once. "
              + "If you intended to copy the key into the value, then consider using the "
              + AsValue.NAME + " function to indicate which key reference should be copied."
          );
        }
      }
    }

    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);
  }
@Test
  public void shouldNotThrowOnSelectStructReferenceInProjection() {
    // Given:
    clearInvocations(source);

    selects = ImmutableList.of(new SingleColumn(STRUCT_COL1_REF, Optional.of(STRUCT_COL1)));

    // When:
    new FinalProjectNode(
        NODE_ID,
        source,
        selects,
        Optional.of(into),
        metaStore
    );

    // Then:
    verify(source).validateColumns(RequiredColumns.builder().add(STRUCT_COL1_REF).build());
  }