@Override
  public LogicalSchema getSchema() {
    return getSource().getSchema();
  }
@Test
  public void shouldThrowOnMultiColKeySchema() {
    // Given:
    final LogicalSchema multiSchema = LogicalSchema.builder()
        .keyColumn(ColumnName.of("K"), SqlTypes.INTEGER)
        .keyColumn(ColumnName.of("K2"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of("C1"), SqlTypes.INTEGER)
        .build();
    final Expression expression = new ComparisonExpression(
        Type.EQUAL,
        new UnqualifiedColumnReferenceExp(ColumnName.of("K")),
        new ArithmeticUnaryExpression(Optional.empty(), Sign.MINUS, new IntegerLiteral(1))
    );
    when(source.getSchema()).thenReturn(multiSchema);

    // When:
    final KsqlException e = assertThrows(
        KsqlException.class,
        () -> new QueryFilterNode(
            NODE_ID,
            source,
            expression,
            metaStore,
            ksqlConfig,
            false,
            plannerOptions
        ));

    // Then:
    assertThat(e.getMessage(), containsString("Multi-column sources must specify every key in the WHERE clause. Specified: [`K`] Expected: [`K` INTEGER KEY, `K2` INTEGER KEY]."));
  }