Make predicates of delete only initialize once

[shidayang]

DeleteFilter initialize 'delete rows' multiple times when call DeleteFilter#filter method multiple times, This cause performance of MOR is very low in trino.
In my case, It spend 8 minutes when I query a table, after optimize it only spend 20 seconds.

[chenjunjiedada]

Do we need to change this?

[shidayang]

Yes,The function to avoid initialize 'delete row' of eqDeleteRows is the same as eqDeletePredicates.

[chenjunjiedada]

I just found out that we don't use this function right now. How about directly using the original function like this:

private CloseableIterable<T> applyEqDeletes(CloseableIterable<T> records) {
  Filter<T> remainingRowsFilter = new Filter<T>() {
    @Override
    protected boolean shouldKeep(T item) {
      return eqDeletedRowFilter().test(item);
    }
  };

  return remainingRowsFilter.filter(records);
}

[rdblue]

I don't think this needs to change. The unified predicate and the list of predicates can coexist.

[shidayang]

I get you

[chenjunjiedada]

The DeleteFilter#filter is called once for each scan task, it is created as a new object. How do you call it multiple times with the same DeleteFilter object?

[shidayang]

@chenjunjiedada In trino query engine every "page" will call DeleteFilter#filter, The "The DeleteFilter#filter is called once for each scan task" is not guaranteed。

[chenjunjiedada]

LGTM

[lhofhansl]

Note that the same can be done in Trino without any API change in Iceberg. See trinodb/trino#13112.
(But I do not care where we fix it.)

Always caching delete filters might be detrimental to Spark, which does not require the filters to be cached.

[amogh-jahagirdar]

@lhofhansl More for my understanding, could you elaborate on the cases where caching delete filters might be detrimental to Spark? Do you mean in terms of retaining the filters in memory?

[lhofhansl]

could you elaborate on the cases where caching delete filters might be detrimental to Spark? Do you mean in terms of retaining the filters in memory?

Yep. We'd be losing out on the opportunity to not materialize the complete filter as a set in memory.
(see:

iceberg/data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

Line 231 in 6cc4a19

Deletes.streamingFilter(records, this::pos, Deletes.deletePositions(filePath, deletes));

)

An alternative is to add the same API I added to TrinoDeleteFilter (which extends DeleteFilter<TrinoRow>) in the Trino PR, namely a filterCached method (could call it cacheAndFilter as well). That way we can leave the Spark code in place and Trino can use the caching method.

[rdblue]

Thanks, @shidayang!

I think this is a good idea for cases where the delete filter is reused. In Spark, this shouldn't affect how long deletes are held in memory because both the filter iterator and the filter are held by the reader. Since there's no negative effect, but there's a very positive effect if this is reused, I think it's a good idea.

[rdblue]

Merged. Thanks for catching this, @shidayang!

[shidayang]

Thanks for your review, @rdblue