public void validateFilterExpression(final Expression exp) {
    final SqlType type = getExpressionReturnType(exp);
    if (!SqlTypes.BOOLEAN.equals(type)) {
      throw new KsqlStatementException(
          "Type error in " + filterType.name() + " expression: "
          + "Should evaluate to boolean but is"
          + " (" + type.toString(FormatOptions.none()) + ") instead.",
          "Type error in " + filterType.name() + " expression: "
          + "Should evaluate to boolean but is " + exp.toString()
          + " (" + type.toString(FormatOptions.none()) + ") instead.",
          exp.toString()
      );
    }
  }
@Test
  public void shouldThrowOnBadTypeComparison() {
    // Given:
    final Expression left = new UnqualifiedColumnReferenceExp(COLUMN1);
    final Expression right = new IntegerLiteral(10);

    final Expression comparision = new ComparisonExpression(Type.EQUAL, left, right);

    when(schema.findValueColumn(any()))
        .thenReturn(Optional.of(Column.of(COLUMN1, STRING, VALUE, 10)));

    // When:
    assertThrows("Error in WHERE expression: "
        + "Cannot compare col1 (STRING) to 10 (INTEGER) with EQUAL.",
        KsqlException.class,
        () -> validator.validateFilterExpression(comparision));
  }