Is this the correct behavior?

I think the reason why we didn't have bulk conversion was so that the caller could handle predicates that couldn't be converted individually. In most cases, we want to push as many predicates as possible. Failing the entire conversion with an exception because one couldn't be converted isn't a good option because we then can't push the predicates that can be converted and let Spark or other engines handle the rest.

I kept the behavior of SparkFilters. You are right. It should not raise an exception.

Okay, I think that the difference is that there are limited uses of SparkFilters and those cases would reject the operation if any filter wasn't convertible (e.g. when deleting or in an overwrite).

I think this is okay since it's following what was done elsewhere.

Even if it's following what was done elsewhere this means if there was non-iceberg UDF predicate that got pushed down it would fail to push down the iceberg expressions? Looking at the code this only seems to be used (currently) in the deleteWhere, but I think for the deleteWhere codepath we should not throw.

I checked the doc and implementation of canDeleteWhere/deleteWhere :

  /**
   * Checks whether it is possible to delete data from a data source table that matches filter
   * expressions.
   * <p>
   * Rows should be deleted from the data source iff all of the filter expressions match.
   * That is, the expressions must be interpreted as a set of filters that are ANDed together.
   * <p>
   * Spark will call this method at planning time to check whether {@link #deleteWhere(Predicate[])}
   * would reject the delete operation because it requires significant effort. If this method
   * returns false, Spark will not call {@link #deleteWhere(Predicate[])} and will try to rewrite
   * the delete operation and produce row-level changes if the data source table supports deleting
   * individual records.
   *
   * @param predicates V2 filter expressions, used to select rows to delete when all expressions
   *                  match
   * @return true if the delete operation can be performed
   *
   * @since 3.4.0
   */
  default boolean canDeleteWhere(Predicate[] predicates) {
    return true;
  }

  /**
   * Delete data from a data source table that matches filter expressions. Note that this method
   * will be invoked only if {@link #canDeleteWhere(Predicate[])} returns true.
   * <p>
   * Rows are deleted from the data source iff all of the filter expressions match. That is, the
   * expressions must be interpreted as a set of filters that are ANDed together.
   * <p>
   * Implementations may reject a delete operation if the delete isn't possible without significant
   * effort. For example, partitioned data sources may reject deletes that do not filter by
   * partition columns because the filter may require rewriting files without deleted records.
   * To reject a delete implementations should throw {@link IllegalArgumentException} with a clear
   * error message that identifies which expression was rejected.
   *
   * @param predicates predicate expressions, used to select rows to delete when all expressions
   *                  match
   * @throws IllegalArgumentException If the delete is rejected due to required effort
   */
  void deleteWhere(Predicate[] predicates);

And the implementation of canDelete:

  @Override
  public boolean canDeleteWhere(Predicate[] predicates) {
    Preconditions.checkArgument(
        snapshotId == null, "Cannot delete from table at a specific snapshot: %s", snapshotId);

    Expression deleteExpr = Expressions.alwaysTrue();

    for (Predicate predicate : predicates) {
      Expression expr = SparkV2Filters.convert(predicate);
      if (expr != null) {
        deleteExpr = Expressions.and(deleteExpr, expr);
      } else {
        return false;
      }
    }

    return canDeleteUsingMetadata(deleteExpr);
  }

So the Expression deleteExpr = SparkV2Filters.convert(predicates); should only be called when all the predicate could be translated into iceberg expressions.

null is filtered out, so it will not fail or return null when null is in the value list.

I think this isn't correct. If there is a null here then at least one literal could not be converted right? That means we can't convert the entire expression.

Yeah, looking into this more I don't think this is a bug introduced by this PR, but a null should not be ignored. Any null value must return null for the entire IN expression. Any IN or NOT IN expression with a null in the set MUST be handled by Spark.

Here the operations same as SparkFilters. I think the original comments are here: #749 (comment). Should we change it?

Yes, this is absolutely a bug. Iceberg should never accept an IN or NOT IN predicate containing a null value.

I will do a follow-up for this.

Do we need this string check? I feel like all of the functions should be listed in the list on 369?

There could be possible some other UDF functions like 'catalog.other.years'.

I think the problem is that this is using the UDF's canonicalName but the other check uses the function's name. Those can differ: BucketFunction.name() return bucket, what the user would call, but the canonical function name identifies the exact bound function no matter how it is loaded so BucketInt.canonicalName() returns iceberg.bucket(int).

If we want to limit to just the Iceberg-defined functions, then this is necessary. It may be better to have a set of supported functions that uses just the canonicalName.

It may be better to have a set of supported functions that uses just the canonicalName.

There are some canonicalName that are not constants, for example:

    @Override
    public String canonicalName() {
      return String.format("iceberg.bucket(%s)", sqlType.catalogString());
    }

While that's not a constant, there are a limited number of values that sqlType.catalogString() might return. I think it would be better to enumerate them, rather than just looking for iceberg and mapping the name.

I have tried to do that, however TruncateDecimal needs precision and scale which are runtime values.

    @Override
    public String canonicalName() {
      return String.format("iceberg.truncate(decimal(%d,%d))", precision, scale);
    }

You can similarly enumerate them, but this isn't a big deal.

It would be really nice if Spark passed the BoundFunction in through the expression!