@Override
    public Result applyFilters(List<ResolvedExpression> flinkFilters) {
        List<ResolvedExpression> acceptedFilters = Lists.newArrayList();
        List<Expression> expressions = Lists.newArrayList();

        for (ResolvedExpression resolvedExpression : flinkFilters) {
            Optional<Expression> icebergExpression = FlinkFilters.convert(resolvedExpression);
            if (icebergExpression.isPresent()) {
                expressions.add(icebergExpression.get());
                acceptedFilters.add(resolvedExpression);
            }
        }

        this.filters = expressions;
        return Result.of(acceptedFilters, flinkFilters);
    }
Flink Iceberg table source.
Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/IcebergTableSource.java