Improve Iceberg deletes when an entire file can be removed

[alexjo2144]

Description

If a delete would remove all rows from an individual file,
remove the whole file, rather than writing a position delete.

This does not include situations where a whole file is deleted
across multiple row-level passes. All rows must be deleted by
one delete operation.

Is this change a fix, improvement, new feature, refactoring, or other?

Improvement

Is this a change to the core query engine, a connector, client library, or the SPI interfaces? (be specific)

Iceberg connector

How would you describe this change to a non-technical end user or system administrator?

Improves performance when deleting all rows in a file.

Related issues, pull requests, and links

• Fixes Improve Iceberg Deletes/Update when an entire file is changed #12057

Documentation

(x) No documentation is needed.
( ) Sufficient documentation is included in this PR.
( ) Documentation PR is available with #prnumber.
( ) Documentation issue #issuenumber is filed, and can be handled later.

Release notes

( ) No release notes entries required.
(x) Release notes entries required with the following suggested text:

# Iceberg
* Avoid writing positional deletes when all rows in a file have been deleted. ({issue}`12057`)

[alexjo2144]

Hmm so that doesn't work, but we have a few options:

1. Change the beginDelete SPI method to include the Constraint being deleted
   This would let us filter out files that are fully delete in beginDelete and skip reading them

2. Change Split generation for DELETE operations against Iceberg to only generate one Split per file, and disable any RowGroup file level row filtering
   This would mean still reading the entire file, but we could recognize that a file was fully deleted in finishDelete. That has the benefit of working with complex predicates, and factoring in previous deletes against the file

[alexjo2144]

@findinpath @findepi I pushed a new approach here, accumulating the number of deleted rows during the writing of Position Delete files, and comparing that to the file's record count. PTAL

Piotr, we had talked offline about using a mechanism similar to IcebergSplitSource#getTableExecuteSplitsInfo but that would have required SPI changes just to pass the record count around. I can try it if you'd like, but for now the record count is just passed through the Splits and Fragments to finishDelete. Let me know if you'd rather make the SPI changes though.

[findepi]

why not propagate here?

[homar]

probably because we create delete files for this few lines below?
Tough I'd expect propagation too.

[alexjo2144]

Failing to delete a file that is not going to be committed didn't seem like enough of a problem to warrant failing the query.

If we have limited fs permissions and can't delete files, for example, we would still be able to write deletes.

This would eventually get picked up by a remove_orphan_files collection

[findepi]

@findinpath @homar please review this class

[homar]

Correct me if I'm wrong but here we do create position delete file pretty much the standard way but then instead of commiting it we delete entire data file if delete file contains all the rows from it(based on the row count) right ?
If so, this doesn't have a positive impact on performance of deletetion - more like negative tough not big, but have positive impact on future reads from this table - right ?

I know preparing benchmark env is in progress but do we plan to benchmark it without this change? I am just curious how big of an impact it has.

[alexjo2144]

@homar right, the delete itself is not any faster but read time is improved. Doing it this way means that the next read does not need to do any I/O for the deleted file vs the old way which would read the entire data-file AND the entire position-delete-file.

I don't have any benchmarks for this but I would be very surprised if it wasn't an improvement.

[alexjo2144]

AC Thanks for the reviews

[homar]

@homar right, the delete itself is not any faster but read time is improved. Doing it this way means that the next read does not need to do any I/O for the deleted file vs the old way which would read the entire data-file AND the entire position-delete-file.

I don't have any benchmarks for this but I would be very surprised if it wasn't an improvement.

I would also be very surprised. I am just wondering how much of an impact it could have :)

[findepi]

@alexjo2144 please rebase, there is a conflict. I will re-review after that

[alexjo2144]

Rebased, thanks

[findepi]

what for?

is 100 ok numher? we have 25 rows only

[alexjo2144]

It's 100 bytes, not rows. I'll add a comment but this ensures each ORC stripe gets a Split by itself

[findinpath]

private long getQuerySplits(QueryId queryId)
    {
        QueryStats stats = getDistributedQueryRunner().getCoordinator().getQueryManager().getFullQueryInfo(queryId).getQueryStats();
        long numberOfSplits = stats.getOperatorSummaries()
                .stream()
                .filter(summary -> summary.getOperatorType().equals("ScanFilterAndProjectOperator"))
                .mapToLong(OperatorStats::getTotalDrivers)
                .sum();
        return numberOfSplits;
    }

 ResultWithQueryId<MaterializedResult> deletionResult = getDistributedQueryRunner().executeWithQueryId(getSession(), "DELETE FROM " + tableName + " WHERE regionkey < 10");
        long deletionSplits = getQuerySplits(deletionResult.getQueryId());

I was hoping to see in the query stats that there are multiple splits for the file, but this wasn't the case.
I checked via debug and indeed there are actually ~ 62 splits of maximum 100B.

Any idea how we could retrieve the number of splits in the test case?

[alexjo2144]

There are some Delta tests that check the number of splits, but they are a bit finicky / often flaky so I didn't include one here.

[alexjo2144]

AC and rebased for conflicts, thanks @findepi

[findinpath]

Impressive work 👍

[findinpath]

finishWrite method is now ~ 150 lines long.
Please consider a refactoring of the method to smaller building blocks in order to ensure a good readability in the weeks/months to come.

[findepi]

follow-up

[findinpath]

private long getQuerySplits(QueryId queryId)
    {
        QueryStats stats = getDistributedQueryRunner().getCoordinator().getQueryManager().getFullQueryInfo(queryId).getQueryStats();
        long numberOfSplits = stats.getOperatorSummaries()
                .stream()
                .filter(summary -> summary.getOperatorType().equals("ScanFilterAndProjectOperator"))
                .mapToLong(OperatorStats::getTotalDrivers)
                .sum();
        return numberOfSplits;
    }

 ResultWithQueryId<MaterializedResult> deletionResult = getDistributedQueryRunner().executeWithQueryId(getSession(), "DELETE FROM " + tableName + " WHERE regionkey < 10");
        long deletionSplits = getQuerySplits(deletionResult.getQueryId());

I was hoping to see in the query stats that there are multiple splits for the file, but this wasn't the case.
I checked via debug and indeed there are actually ~ 62 splits of maximum 100B.

Any idea how we could retrieve the number of splits in the test case?

[alexjo2144]

Added some partitioned table tests. Thanks for the suggestion @findinpath