Push down DELETE for enforceable filters on Delta Lake

[findinpath]

Description

Perform DELETE only on the metadata of the Delta Lake
table when the delete filter does not touch any data
columns.

Sample queries affected by this enhancement are:

DELETE FROM table 
DELETE FROM table WHERE true
DELETE FROM partitioned_table WHERE part1=value1 AND part2=value2
DELETE FROM partitioned_table WHERE part2=value2

Fixes #18331

The operation will not be performed on the metadata layer when
there are add file entries in the transaction log of the table
without statistics containing number of records.

Additional context and related issues

Without DELETE pushdown

trino:default> create table delta.tiny.sforders100 as select * from tpch.sf100.orders;
CREATE TABLE: 150000000 rows

trino:default> delete from delta.tiny.sforders100 where true;
DELETE: 150000000 rows

Query 20230725_125258_00003_xfv9h, FINISHED, 1 node
http://localhost:8080/ui/query.html?20230725_125258_00003_xfv9h
Splits: 25 total, 25 done (100.00%)
CPU Time: 115.1s total,  1.3M rows/s, 11.2MB/s, 89% active
Per Node: 1.1 parallelism, 1.46M rows/s, 12.5MB/s
Parallelism: 1.1
Peak Memory: 87.3MB
1:43 [150M rows, 1.26GB] [1.46M rows/s, 12.5MB/s]

With DELETE pushdown

trino:default> delete from delta.tiny.sf100orders where true;
DELETE: 150000000 rows

Query 20230725_123039_00000_fk4kt, FINISHED, 1 node
http://localhost:8080/ui/query.html?20230725_123039_00000_fk4kt
Splits: 1 total, 1 done (100.00%)
CPU Time: 0.1s total,     0 rows/s,     0B/s, 83% active
Per Node: 0.1 parallelism,     0 rows/s,     0B/s
Parallelism: 0.1
Peak Memory: 112B
1.84 [0 rows, 0B] [0 rows/s, 0B/s]

tldr; before 103 s after 1.8 s

Release notes

(x) Release notes are required, with the following suggested text:

# Delta Lake
* Improve performance of `DELETE` statement when it deletes the whole table or the filters apply only to the partition columns. ({issue}`18332 `)

[findinpath]

This change is in sync with

trino/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMergeSink.java

Line 333 in 20e5779

Location targetLocation = sourceLocation.parentDirectory().appendPath(session.getQueryId() + "_" + randomUUID());

I'd benefit from a suggestion in which utility class could i extract the method for creating the file name.

[homar]

You can always add a new one ;)

[findinpath]

TestDeltaLakeDelete.testDeleteAllOssDeltaLake is failing because the number of deleted rows is not anymore returned by the DELETE query.

Should the accelerated DELETE (done only based on table metadata) be performed if and only if all the data files have the number of records present in the transaction log file?

As agreed with @findepi , it is rather unusual to deal with add entries in the transaction logs which don't contain the file stats.
In the unlikely case that we're dealing with such a table, simply don't return the number of rows affected for the DELETE operation if we can execute it faster.

[alexjo2144]

delete operation?

[findinpath]

I'm just following the same pattern as the one used for UPDATE/DELETE/MERGE statements in Delta Lake. See #15763

[findinpath]

Rebasing on master to address conflicts

[findinpath]

@pajaks & @alexjo2144 AC. Please review again the PR.

[findinpath]

Thank you everybody for the valuable feedback.
AC

[ebyhr]

Pushed small changes to fix CI failure.