Improve performance of reading iceberg table with many equality delete files

[Heltman]

Description

Fixes #17114

Additional context and related issues

If a split with many delete file RowPredicate.and will create a deep stack, this pr compact all StructLikeSet to a collection to reduce stack depth.

The stack depth is only a hidden danger. The real problem is that multiple StructLikeSet of a split are not merged according to the ids, resulting in too many StructLikeSet being generated, which makes the filtering efficiency very lowly.

The main content is to classify delete files according to id, and only use the same StructLikeSet to collect deletion data.

Release notes

( ) This is not user-visible or docs only and no release notes are required.
( ) Release notes are required, please propose a release note for me.
(x) Release notes are required, with the following suggested text:

# Section
* Improve performance of reading iceberg table with many equality delete files. ({issue}`17114`)

[alexjo2144]

The core change here, reducing the number of StructLikeSets created seems like a good idea to me. Just a couple code simplification questions.

Just for reference here's a thread on this from the original PR: #13219 (comment)
cc: @findepi

[xiacongling]

Great job on a quick fix of the issue, @Heltman @alexjo2144 ! It may significantly improve Trino's performance on Iceberg delete file filtering.

IMHO, the patch made by @alexjo2144 makes less code changes and seems easier to understand, but some problems may not completely resolved:

1. EqualityDeleteSets of delete files can be grouped by thier schemas. Since the equality field IDs come from the table schema, for two delete files with the same fields in different order, the projection structs will be identical. Using a set is preferred and it is what Iceberg does in its org.apache.iceberg.data.DeleteFilter.
2. chaining RowPredicts with RowPredict.and will lead to recursive method calls, it will impact the performance and have the risk of stack overflow when too many delete files present. Delete file grouping can significantly reduce the number of delete filters, this improvement may not seem as important. Since delete filter will be applied to each row scanned from data files, even the smallest enhancement may improve the efficiency of the query. Would you consider adding this change as well, @alexjo2144 ? Is there any stats can be provided to prove the performance improvement, @Heltman?

[alexjo2144]

Looks like using a Set is safe there in the Iceberg Spark implementation because they do an additional projection step to ensure that the readers return the Schema stored in the Set there, even if it doesn't match the file schema. We don't have that additional projection here yet, so it will cause problems in Trino.

It needs some clean-up but here's a test case illustrating the problem: https://gist.github.com/alexjo2144/8a80ff5146ab3c82fa0c5fc5b4f33e66

So we can either need to add the additional projection, or use an ordered Collection like a List

[Heltman]

@alexjo2144 Schema mismatch is indeed a problem, but fortunately, the additional projection step implemented by spark is also easy to implement in trino. We only need to refer to the implementation of spark and organize the fields in order when reading the delete files. Trino's ParquetReader has additional projection(same as spark). Please check the new commit, I fix this problem and add your test case.

[findinpath]

The stack depth is only a hidden danger. The real problem is that multiple StructLikeSet of a split are not merged according to the ids, resulting in too many StructLikeSet being generated, which makes the filtering efficiency very lowly.

Can you pls provide a test case in your PR which can be used to showcase this situation?

I was trying with the code from master and see only one StructLikeSet with 7 elements (same as in your submission).

@Test
public void testMultipleDeletes()
            throws Exception
    {
        String tableName = "test_equality_delets_different_schemas_" + randomNameSuffix();
        assertUpdate("CREATE TABLE " + tableName + " AS SELECT * FROM tpch.tiny.nation", 25);
        Table icebergTable = updateTableToV2(tableName);
        Assertions.assertThat(icebergTable.currentSnapshot().summary().get("total-equality-deletes")).isEqualTo("0");
        Path metadataDir = new Path(metastoreDir.toURI());
        TrinoFileSystem fs = HDFS_FILE_SYSTEM_FACTORY.create(SESSION);

        String deleteFile1 = "delete_file_" + UUID.randomUUID();
        List<String> firstDeleteFileColumns = ImmutableList.of("regionkey");
        Schema deleteRowSchema = icebergTable.schema().select(firstDeleteFileColumns);
        List<Integer> equalityFieldIds = firstDeleteFileColumns.stream()
                .map(name -> deleteRowSchema.findField(name).fieldId())
                .collect(toImmutableList());
        Parquet.DeleteWriteBuilder writerBuilder = Parquet.writeDeletes(new ForwardingFileIo(fs).newOutputFile(new Path(metadataDir, deleteFile1).toString()))
                .forTable(icebergTable)
                .rowSchema(deleteRowSchema)
                .createWriterFunc(GenericParquetWriter::buildWriter)
                .equalityFieldIds(equalityFieldIds)
                .overwrite();
        EqualityDeleteWriter<Record> writer = writerBuilder.buildEqualityWriter();

        Record dataDelete = GenericRecord.create(deleteRowSchema);
        try (Closeable ignored = writer) {
            writer.write(dataDelete.copy(ImmutableMap.of("regionkey", 1L)));
            writer.write(dataDelete.copy(ImmutableMap.of("regionkey", 2L)));
            writer.write(dataDelete.copy(ImmutableMap.of("regionkey", 3L)));
            writer.write(dataDelete.copy(ImmutableMap.of("regionkey", 4L)));
            writer.write(dataDelete.copy(ImmutableMap.of("regionkey", 5L)));
            writer.write(dataDelete.copy(ImmutableMap.of("regionkey", 6L)));
            writer.write(dataDelete.copy(ImmutableMap.of("regionkey", 7L)));
        }
        icebergTable.newRowDelta().addDeletes(writer.toDeleteFile()).commit();


        assertQuery("SELECT * FROM " + tableName, "SELECT * FROM nation WHERE  regionkey > 7");
        assertUpdate("DROP TABLE " + tableName);
    }

[Heltman]

Looks like using a Set is safe there in the Iceberg Spark implementation because they do an additional projection step to ensure that the readers return the Schema stored in the Set there, even if it doesn't match the file schema. We don't have that additional projection here yet, so it will cause problems in Trino.

It needs some clean-up but here's a test case illustrating the problem: https://gist.github.com/alexjo2144/8a80ff5146ab3c82fa0c5fc5b4f33e66

So we can either need to add the additional projection, or use an ordered Collection like a List

@findinpath just check below.

[Heltman]

The stack depth is only a hidden danger. The real problem is that multiple StructLikeSet of a split are not merged according to the ids, resulting in too many StructLikeSet being generated, which makes the filtering efficiency very lowly.

Can you pls provide a test case in your PR which can be used to showcase this situation?

It is difficult to simulate this performance problem, because it needs to have more deletefiles, and it has been updated many times according to same columns.

We only need to imagine that if there are two deletefiles, each with 10,000 lines, 9,000 of which are the same. Originally, we need to filter a line of data from 20,000 lines, but now we only need to match 11,000 lines.

[Heltman]

@findinpath , add testMultipleEqualityDeletes, delete file has compact:

[findinpath]

Overall there are some cosmetics improvements needed and testing coverage for nested deletes is missing.

Great work ! 👍

[alexjo2144]

Pending the last couple comments from Marius, this looks good to me.
It does need a rebase though.

[findepi]

the call site looks like new EqualityDeleteSet(deleteSchema, schemaFromHandles(readColumns))
are the constructor arg names right? is the call site right?

[Heltman]

I suggest public EqualityDeleteSet(Schema deleteSchema, Schema fileSchema). Because first one is schema from iceberg metadata, second schema from read file (parquet, orc, etc.)

[findinpath]

@Heltman yes, public EqualityDeleteSet(Schema deleteSchema, Schema fileSchema) should be fine. Let's go forward with this suggestion

[Heltman]

Finally we reached a consensus, public EqualityDeleteSet(Schema deleteSchema, Schema dataSchema) is good idea. deleteSchema come from iceberg metadata, dataSchema come from equality delete file.

[findepi]

Merged, thanks!