Add support for partition pruning in Delta checkpoint iterator

[ebyhr]

Release notes

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

# Delta Lake
* Improve performance when reading large checkpoint files on partitioned tables. ({issue}`issuenumber`)

[findinpath]

Shouldn't we have here only 1 entry?
Probably this relates to https://github.com/trinodb/trino/pull/19588/files/7c9ac692875bdb08827aa1dc9f7beac63a9874d4#r1383331077
We should have also the check to see that a reduced amount of entries has been actually read from the parquet file

        assertThat(checkpointEntryIterator.getCompletedPositions().orElseThrow()).isEqualTo(....);

[findinpath]

When doing buildAddEntry check whether the partitionValues / partitionValues_parsed match the partitionConstraint and return null if not matching.

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

Lines 212 to 213 in 10fdc95

	Map<DeltaLakeColumnHandle, Domain> enforcedDomains = enforcedPartitionConstraint.getDomains().orElseThrow();
	if (!partitionMatchesPredicate(addAction.getCanonicalPartitionValues(), enforcedDomains)) {

[ebyhr]

Just rebased on master.

[findepi]

perhaps remove && !partitionConstraint.isAll()

i think the new code path should eventually replace the old cache-based approach, so we can use isCheckpointPartitionFilterEnabled as a algorithm-selecting toggle

[ebyhr]

CI hit #19602

[raunaqmorarka]

Do we need to check this for every position ? Seems like we should know this per file based on parquet file metadata (maybe it's possible to use io.trino.plugin.hive.ReaderPageSource#getReaderColumns).

[ebyhr]

Agree with using Parquet metadata though getReaderColumns returns an empty list in this case. Sent another PR #19727

[raunaqmorarka]

While this may help in reducing the number of DeltaLakeTransactionLogEntry, doing the filtering after materialising all channels on each position of a page means that we can't benefit from lazy loading of blocks.
Ideally we should filter directly on the relevant block channels and skip to next position without decoding the remaining channels when the predicate does not match. But this can be looked at as a follow-up.

[findinpath]

Correct.
The partition matching check should be done directly in io.trino.plugin.deltalake.transactionlog.checkpoint.CheckpointEntryIterator#buildAddEntry

If we know that we have the field partitionValues_parsed (see https://github.com/trinodb/trino/pull/19588/files#r1389691135) , maybe we should do this check right away after doing

trino/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/checkpoint/CheckpointEntryIterator.java

Lines 410 to 413 in 65686fa

	log.debug("Building add entry from %s pagePosition %d", block, pagePosition);
	if (block.isNull(pagePosition)) {
	return null;
	}

optional: One word concerning using entry.getAdd().getCanonicalPartitionValues().
We have at hand the partitionValues_parsed. We could avoid deserializing the stringified partition values and use the "parsed" values directly. OTOH, we don't actually use the parsed partition values otherwise anywhere else. Did you intentionally restrain from reading the parsed partition values in favor of the stringified partition values?

[findinpath]

Correct.
The partition matching check should be done directly in io.trino.plugin.deltalake.transactionlog.checkpoint.CheckpointEntryIterator#buildAddEntry

If we know that we have the field partitionValues_parsed (see https://github.com/trinodb/trino/pull/19588/files#r1389691135) , maybe we should do this check right away after doing

trino/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/checkpoint/CheckpointEntryIterator.java

Lines 410 to 413 in 65686fa

	log.debug("Building add entry from %s pagePosition %d", block, pagePosition);
	if (block.isNull(pagePosition)) {
	return null;
	}

optional: One word concerning using entry.getAdd().getCanonicalPartitionValues().
We have at hand the partitionValues_parsed. We could avoid deserializing the stringified partition values and use the "parsed" values directly. OTOH, we don't actually use the parsed partition values otherwise anywhere else. Did you intentionally restrain from reading the parsed partition values in favor of the stringified partition values?

[findepi]

Why TODO? why not do it right away?

[ebyhr]

I just wanted to focus on SELECT path in this PR. Going to handle in this PR.

[findepi]

The callers (eg split source) will likely repeat this work, so it's partially wasted.
Still useful because this allows us to materialize a shorter list.

I think this wouldn't be needed here if we could return a Stream/Iterator instead of a List.

[findepi]

require... or use... ?

we wnt to use use partitionvalues_parsed field if it is present, but we don't require that it exists (we don't fail when it doesn't), right?

[findepi]

The set of partitioning columns may change in the meantime probably only through the CREATE OR REPLACE TABLE operation. In such case, we shouldn't need to read the old checkpoint file at all, but I don't know whether this is the case.

[findepi]

( #19586 )

[findinpath]

Could you pls test coverage into TestDeltaLakeFileOperations with checkpoint_filtering_enabled session property enabled to add more transparence in regards to the consequences coming with this change?