### Question:
ProjectNode extends PlanNode { @Override public SchemaKStream buildStream(final StreamsBuilder builder, final KsqlConfig ksqlConfig, final KafkaTopicClient kafkaTopicClient, final FunctionRegistry functionRegistry, final Map<String, Object> props, final SchemaRegistryClient schemaRegistryClient) { return getSource().buildStream(builder, ksqlConfig, kafkaTopicClient, functionRegistry, props, schemaRegistryClient) .select(getProjectNameExpressionPairList()); } @JsonCreator  ProjectNode(@JsonProperty("id") final PlanNodeId id,
                     @JsonProperty("source") final PlanNode source,
                     @JsonProperty("schema") final Schema schema,
                     @JsonProperty("projectExpressions")
                       final List<Expression> projectExpressions); @Override List<PlanNode> getSources(); @JsonProperty PlanNode getSource(); @Override Schema getSchema(); @Override Field getKeyField(); List<Pair<String, Expression>> getProjectNameExpressionPairList(); @Override R accept(PlanVisitor<C, R> visitor, C context); @Override SchemaKStream buildStream(final StreamsBuilder builder,
                                   final KsqlConfig ksqlConfig,
                                   final KafkaTopicClient kafkaTopicClient,
                                   final FunctionRegistry functionRegistry,
                                   final Map<String, Object> props,
                                   final SchemaRegistryClient schemaRegistryClient);  }### Answer:
@Test(expected = KsqlException.class) public void shouldThrowKsqlExcptionIfSchemaSizeDoesntMatchProjection() { mockSourceNode(); EasyMock.replay(source, stream); final ProjectNode node = new ProjectNode(new PlanNodeId("1"), source, SchemaBuilder.struct() .field("field1", Schema.STRING_SCHEMA) .field("field2", Schema.STRING_SCHEMA) .build(), Collections.singletonList(new BooleanLiteral("true"))); node.buildStream(builder, ksqlConfig, kafkaTopicClient, functionRegistry, props, new MockSchemaRegistryClient()); }
@Test public void shouldCreateProjectionWithFieldNameExpressionPairs() { mockSourceNode(); final BooleanLiteral trueExpression = new BooleanLiteral("true"); final BooleanLiteral falseExpression = new BooleanLiteral("false"); EasyMock.expect(stream.select( Arrays.asList(new Pair<>("field1", trueExpression), new Pair<>("field2", falseExpression)))) .andReturn(stream); EasyMock.replay(source, stream); final ProjectNode node = new ProjectNode(new PlanNodeId("1"), source, SchemaBuilder.struct() .field("field1", Schema.STRING_SCHEMA) .field("field2", Schema.STRING_SCHEMA) .build(), Arrays.asList(trueExpression, falseExpression)); node.buildStream(builder, ksqlConfig, kafkaTopicClient, functionRegistry, props, new MockSchemaRegistryClient()); EasyMock.verify(stream); }