@Override
  public LogicalSchema getSchema() {
    return getSource().getSchema();
  }
@Test
  public void shouldThrowKeyExpressionThatDoestCoverKey_multipleDisjuncts() {
    // Given:
    when(source.getSchema()).thenReturn(INPUT_SCHEMA);
    final Expression keyExp1 = new ComparisonExpression(
        Type.EQUAL,
        new UnqualifiedColumnReferenceExp(ColumnName.of("WINDOWSTART")),
        new IntegerLiteral(1)
    );
    final Expression keyExp2 = new ComparisonExpression(
        Type.EQUAL,
        new UnqualifiedColumnReferenceExp(ColumnName.of("K")),
        new IntegerLiteral(2)
    );
    final Expression expression = new LogicalBinaryExpression(
        LogicalBinaryExpression.Type.OR,
        keyExp1,
        keyExp2
    );

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

    // Then:
    assertThat(e.getMessage(), containsString("WHERE clause missing key column for disjunct: "
        + "(WINDOWSTART = 1)"));
  }