Don't rewrite single small file per partition during Optimize

[homar]

Description

Fixes: #18645

Additional context and related issues

Release notes

(x) This is not user-visible or is docs only, and no release notes are required.

[findinpath]

Reduce the dependency here.From reading the code, we need in DeltaLakeSplitManager to know whether we're in an OPTIMIZE situation from the DeltaLakeTableHandle. If yes, let's just add a boolean flag to point this out.
Do reflect this in equals() and hashCode() methods as well.

[findinpath]

Suggested change

	ImmutableList<String> partitionColumns = metadataEntry.getOriginalPartitionColumns().stream()
	.map(partitionColumnMapping::get).collect(toImmutableList());
	ImmutableList<String> partitionColumns = metadataEntry.getOriginalPartitionColumns().stream()
	.map(partitionColumnMapping::get)
	.collect(toImmutableList());

[findinpath]

Gentle reminder.

[findinpath]

maxScannedFileSizeInBytes.isPresent() && addAction.getSize() < maxScannedFileSizeInBytes.get()

If we have only one file in the partition, there is nothing to optimize, right? The above highlighted condition is not necessary.

[homar]

if it is bigger than maxScannedFileSizeInBytes then it should be optimzied and splitted into 2 files

[findinpath]

Instead of this rather ad-hoc mechanism, wouldn't it be enough to make use of io.trino.plugin.deltalake.transactionlog.AddFileEntry#getCanonicalPartitionValues as a key? I'm guessing that this map can be used for equality.

An alternative would be a list of deserialized partition values io.trino.plugin.deltalake.transactionlog.TransactionLogParser#deserializePartitionValue (probably not necessary though).

[homar]

I was kind of afraid that using map as a key in another map may not be reliable

[findinpath]

The map is just an object with a hashcode - i don't think it matters whether you use string or the canonicalPartitionValues as key in the map.

[homar]

so if it doesnt matter why do you insist? ;) i am kind of afraid of things like Set<Map<String, Optional<String>> i can do it but im not a big fun

[findinpath]

I'm insisting because I don't see the point of having a special logic for geting the stringified "partition key".

[findinpath]

IIUC we want to keep (in case we're doing OPTIMIZE) only the AddFileEntries with duplicate canonical partition values per partition/table (in case the table is not partitioned).

We don't necessarily need to do counting in order to keep only the entries which have duplicate canonical partition values.

Here's a chatgpt proof of concept

public class Main {
    public static void main(String[] args) {
        // Create a list of elements (strings)
        List<String> elements = Arrays.asList("apple", "banana", "cherry", "banana", "date", "banana", "apple", "fig");

        // Create a set to store elements seen before
        Set<String> seenElements = new HashSet<>();
        // Create a set to store elements that occur more than once
        Set<String> multipleOccurrenceElements = new HashSet<>();

        // Iterate through the list to find elements that occur more than once
        for (String element : elements) {
            if (!seenElements.add(element)) {
                // If the element was already in the set, it occurs more than once
                multipleOccurrenceElements.add(element);
            }
        }

        // Print elements that occur more than once
        for (String element : elements) {
            if (multipleOccurrenceElements.contains(element)) {
                System.out.println(element);
            }
        }
    }
}

[homar]

yep, great idea, thank you

[findinpath]

nit: isOptimize is redundant because the method name is forOptimize

[findinpath]

Did you remove the method parameter?

[findinpath]

Suggested change

	if (isOptimize && !partitionKeysWithMoreThanOneFile.contains(getPartitionKey(originalPartitionColumns, addAction)) && maxScannedFileSizeInBytes.isPresent() && addAction.getSize() < maxScannedFileSizeInBytes.get()) {
	if (isOptimize && partitionKeysWithSingleFile.contains(getPartitionKey(originalPartitionColumns, addAction)) && maxScannedFileSizeInBytes.isPresent() && addAction.getSize() < maxScannedFileSizeInBytes.get()) {

The positive tense (instead of !) is a bit easier to follow from maintainer perspective.
Scratch this in case it doesn't fit with the optimized logic for retrieving partitions with more than one file.

[findinpath]

Suggested change

	ImmutableSet.Builder<Map<String, Optional<String>>> partitionsWithAtMostOneFileBuilder = ImmutableSet.builder();
	if (isOptimize) {
	partitionsWithAtMostOneFileBuilder.addAll(findPartitionsWithAtMostOneFile(validDataFiles));
	}
	Set<Map<String, Optional<String>>> partitionsWithAtMostOneFile = partitionsWithAtMostOneFileBuilder.build();
	Set<Map<String, Optional<String>>> partitionsWithOneFileForOptimize = isOptimize ? findPartitionsWithAtMostOneFile(validDataFiles): Set.of();

[ebyhr]

/test-with-secrets sha=c65e54edd7972f08a28bd9d46a525cfe58b7895a

[github-actions]

The CI workflow run with tests that require additional secrets has been started: https://github.com/trinodb/trino/actions/runs/6234333370