public static Optional<Expression> convert(
      org.apache.flink.table.expressions.Expression flinkExpression) {
    if (!(flinkExpression instanceof CallExpression)) {
      return Optional.empty();
    }

    CallExpression call = (CallExpression) flinkExpression;
    Operation op = FILTERS.get(call.getFunctionDefinition());
    if (op != null) {
      switch (op) {
        case IS_NULL:
          return onlyChildAs(call, FieldReferenceExpression.class)
              .map(FieldReferenceExpression::getName)
              .map(Expressions::isNull);

        case NOT_NULL:
          return onlyChildAs(call, FieldReferenceExpression.class)
              .map(FieldReferenceExpression::getName)
              .map(Expressions::notNull);

        case LT:
          return convertFieldAndLiteral(Expressions::lessThan, Expressions::greaterThan, call);

        case LT_EQ:
          return convertFieldAndLiteral(
              Expressions::lessThanOrEqual, Expressions::greaterThanOrEqual, call);

        case GT:
          return convertFieldAndLiteral(Expressions::greaterThan, Expressions::lessThan, call);

        case GT_EQ:
          return convertFieldAndLiteral(
              Expressions::greaterThanOrEqual, Expressions::lessThanOrEqual, call);

        case EQ:
          return convertFieldAndLiteral(
              (ref, lit) -> {
                if (NaNUtil.isNaN(lit)) {
                  return Expressions.isNaN(ref);
                } else {
                  return Expressions.equal(ref, lit);
                }
              },
              call);

        case NOT_EQ:
          return convertFieldAndLiteral(
              (ref, lit) -> {
                if (NaNUtil.isNaN(lit)) {
                  return Expressions.notNaN(ref);
                } else {
                  return Expressions.notEqual(ref, lit);
                }
              },
              call);

        case NOT:
          return onlyChildAs(call, CallExpression.class)
              .flatMap(FlinkFilters::convert)
              .map(Expressions::not);

        case AND:
          return convertLogicExpression(Expressions::and, call);

        case OR:
          return convertLogicExpression(Expressions::or, call);

        case STARTS_WITH:
          return convertLike(call);
      }
    }

    return Optional.empty();
  }
@Test
  public void testIsNotNull() {
    Expression expr = resolve(Expressions.$("field1").isNotNull());
    Optional<org.apache.iceberg.expressions.Expression> actual = FlinkFilters.convert(expr);
    assertThat(actual).isPresent();
    UnboundPredicate<Object> expected =
        org.apache.iceberg.expressions.Expressions.notNull("field1");
    assertPredicatesMatch(expected, actual.get());
  }