@Override
  public LogicalSchema getSchema() {
    return getSource().getSchema();
  }
@Test
  public void shouldThrowMultiKeyExpressionsThatDontCoverAllKeys() {
    // Given:
    when(source.getSchema()).thenReturn(MULTI_KEY_SCHEMA);
    final Expression expression = new ComparisonExpression(
        Type.EQUAL,
        new UnqualifiedColumnReferenceExp(ColumnName.of("K1")),
        new IntegerLiteral(1)
    );

    // 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: [`K1`] Expected: [`K1` INTEGER KEY, `K2` INTEGER KEY]"));
  }