From d769c3382a430e11e36bdccc97459cb39cae69f4 Mon Sep 17 00:00:00 2001 From: James Greenhill Date: Fri, 4 Sep 2026 00:23:33 +0000 Subject: [PATCH 1/2] Enforce partition predicates on temporal keys isEnforceableType excluded DATE and every timestamp type, so a DELETE covering whole partitions of a daily key read and rewrote every row of those partitions instead of ending their files. A pipeline that replaces day intervals paid for a full scan on every run. Identity partition keys of type DATE, TIMESTAMP(p) and TIMESTAMP(p) WITH TIME ZONE are now enforced, with a fail-open parse. A value the connector cannot read back is never guessed at in either direction: the file is not pruned, and it is not counted as matching. DuckDB writes four such values: infinity, -infinity, a BC date, and a year of five or more digits. The connector may not prune such a file, and it cannot apply a predicate to the rows of a file on its own. So one of these values in a table leaves the whole predicate with the engine, which keeps filtering. Enforcement is therefore decided per table: applyFilter enforces a temporal key only when the partition value of every visible data file decides the column. It checks that over the same files, and through the same code, that the split manager prunes by. The parse itself does not change. StatsValueParser already read these values as unknown. Its unit tests now pin that, along with the offset forms a writer outside UTC records. Transform partitions are unaffected. year, month, day and hour are not identity keys, so predicates on them stay unenforced and their pruning stays fail-open. Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe --- docs/src/main/sphinx/connector/ducklake.md | 9 ++ .../plugin/ducklake/DuckLakeMetadata.java | 78 ++++++++--- .../ducklake/TestDuckLakeWriteRoundTrips.java | 131 +++++++++++++++--- .../ducklake/util/TestStatsValueParser.java | 23 +++ 4 files changed, 205 insertions(+), 36 deletions(-) diff --git a/docs/src/main/sphinx/connector/ducklake.md b/docs/src/main/sphinx/connector/ducklake.md index 2f500fe16c95..ac2227966298 100644 --- a/docs/src/main/sphinx/connector/ducklake.md +++ b/docs/src/main/sphinx/connector/ducklake.md @@ -233,6 +233,15 @@ The connector skips data files that cannot match query predicates: `file_statistics_pruning_enabled` [catalog session property](/sql/set-session) disable this behavior. +A predicate on an `identity` partition key is enforced by the connector, not +only used to prune. A `DELETE` that covers whole partitions of such a key ends +those data files in the catalog, without reading a row of them. Keys of type +`DATE`, `TIMESTAMP`, and `TIMESTAMP WITH TIME ZONE` are enforced along with the +integer, boolean and `VARCHAR` keys. DuckDB writes some temporal values in a +form the connector cannot read back — `infinity`, `-infinity`, BC dates, and +years of five or more digits — and a table that holds one leaves the predicate +to the engine, so the file is read and filtered instead of pruned. + Before scheduling a data file, the coordinator reads its Parquet footer once and groups adjacent row groups into splits. `ducklake.max-split-size` is a soft target for the sum of the row groups' compressed column sizes. A row group is diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java index 512e6727d94b..b0f7885bcc43 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java @@ -13,6 +13,7 @@ */ package io.trino.plugin.ducklake; +import com.google.common.base.Suppliers; import com.google.common.collect.ArrayListMultimap; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; @@ -85,6 +86,8 @@ import io.trino.spi.statistics.DoubleRange; import io.trino.spi.statistics.Estimate; import io.trino.spi.statistics.TableStatistics; +import io.trino.spi.type.TimestampType; +import io.trino.spi.type.TimestampWithTimeZoneType; import io.trino.spi.type.Type; import io.trino.spi.type.VarcharType; import it.unimi.dsi.fastutil.longs.LongOpenHashSet; @@ -102,6 +105,7 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Supplier; import java.util.function.UnaryOperator; import static com.google.common.collect.ImmutableList.toImmutableList; @@ -124,6 +128,7 @@ import static io.trino.spi.connector.RowChangeParadigm.DELETE_ROW_AND_INSERT_ROW; import static io.trino.spi.type.BigintType.BIGINT; import static io.trino.spi.type.BooleanType.BOOLEAN; +import static io.trino.spi.type.DateType.DATE; import static io.trino.spi.type.DoubleType.DOUBLE; import static io.trino.spi.type.IntegerType.INTEGER; import static io.trino.spi.type.RealType.REAL; @@ -1689,12 +1694,17 @@ public Optional> applyFilter(C newUnenforcedConstraint = TupleDomain.all(); } else { - Set enforceableColumnIds = enforceablePartitionColumnIds(handle); + Map partitionColumns = enforceablePartitionColumns(handle); + // the files are only listed for a column whose values have to be checked one by one, + // and then only once however many such columns the predicate constrains + Supplier> dataFiles = Suppliers.memoize(() -> metastore.dataFiles(handle.snapshotId(), handle.tableId())); ImmutableMap.Builder enforceableDomains = ImmutableMap.builder(); ImmutableMap.Builder unenforceableDomains = ImmutableMap.builder(); for (Map.Entry entry : predicate.getDomains().orElseThrow().entrySet()) { DuckLakeColumnHandle column = entry.getKey(); - if (enforceableColumnIds.contains(column.columnId()) && isEnforceableType(column.type())) { + DuckLakePartitionColumn partitionColumn = partitionColumns.get(column.columnId()); + if (partitionColumn != null && isEnforceableType(column.type()) + && (!isTemporalType(column.type()) || everyPartitionValueDecidesTheColumn(column, partitionColumn, dataFiles.get()))) { enforceableDomains.put(column, entry.getValue()); } else { @@ -1730,32 +1740,34 @@ public Optional> applyFilter(C } /** - * Returns the ids of partition columns whose predicates the connector can fully enforce by - * pruning files: identity-transformed columns where every visible data file was written with - * the current partitioning scheme, so every row of a kept file carries the partition value. + * Returns the partition columns whose predicates the connector can fully enforce by pruning + * files, by column id: identity-transformed columns where every visible data file was written + * with the current partitioning scheme, so every row of a kept file carries the partition + * value. */ - private Set enforceablePartitionColumnIds(DuckLakeTableHandle handle) + private Map enforceablePartitionColumns(DuckLakeTableHandle handle) { Optional partitionInfo = metastore.partitionInfo(handle.snapshotId(), handle.tableId()); if (partitionInfo.isEmpty()) { - return ImmutableSet.of(); + return ImmutableMap.of(); } if (!metastore.allDataFilesUsePartition(handle.snapshotId(), handle.tableId(), partitionInfo.get().partitionId())) { - return ImmutableSet.of(); + return ImmutableMap.of(); } return partitionInfo.get().columns().stream() .filter(column -> column.transform().equalsIgnoreCase(IDENTITY_TRANSFORM)) - .map(DuckLakePartitionColumn::columnId) - .collect(toImmutableSet()); + // a column filed under two identity keys is read from the first of them, which is + // the one PartitionTransforms picks as well + .collect(toImmutableMap(DuckLakePartitionColumn::columnId, column -> column, (first, _) -> first)); } /** - * Types whose identity partition values round-trip exactly through the string - * representation in {@code ducklake_file_partition_value}, so pruning on them can be - * used to enforce a predicate. {@code DATE} is excluded because DuckDB writes values - * such as {@code infinity}, {@code -infinity}, BC dates and years with five or more - * digits that cannot be parsed back reliably; predicates on it stay unenforced and - * pruning remains fail-open. + * Types whose identity partition values round-trip exactly through the string representation + * in {@code ducklake_file_partition_value}, so pruning on them can be used to enforce a + * predicate. A temporal type is admitted here only as a candidate: DuckDB writes values such + * as {@code infinity}, {@code -infinity}, BC dates and years with five or more digits, which + * do not parse back, so {@link #isTemporalType} sends its columns through + * {@link #everyPartitionValueDecidesTheColumn} first. */ private static boolean isEnforceableType(Type type) { @@ -1764,7 +1776,39 @@ private static boolean isEnforceableType(Type type) || type.equals(INTEGER) || type.equals(BIGINT) || type.equals(BOOLEAN) - || type instanceof VarcharType varcharType && varcharType.isUnbounded(); + || type instanceof VarcharType varcharType && varcharType.isUnbounded() + || isTemporalType(type); + } + + /** + * Types a partition value of which DuckDB may write in a form that does not parse back, so + * that the values of the table decide whether a predicate on the column can be enforced. + */ + private static boolean isTemporalType(Type type) + { + return type.equals(DATE) + || type instanceof TimestampType + || type instanceof TimestampWithTimeZoneType; + } + + /** + * Whether the partition value of every visible data file says which value of the column the + * rows of that file hold. A value the connector cannot read back leaves the file unprunable, + * and a predicate the connector cannot apply to one file it is not allowed to prune is one it + * cannot enforce for the table, so the engine keeps filtering the rows instead. + *

+ * The check runs over the same files and through the same code as + * {@link DuckLakeSplitManager#getSplits}, which reads the same snapshot, so a predicate + * enforced here is a predicate that splits can be pruned by there. + */ + private static boolean everyPartitionValueDecidesTheColumn( + DuckLakeColumnHandle column, + DuckLakePartitionColumn partitionColumn, + List dataFiles) + { + List transforms = ImmutableList.of(partitionColumn); + return dataFiles.stream() + .allMatch(dataFile -> PartitionTransforms.partitionDomain(column, transforms, dataFile.partitionValues()).isPresent()); } /** diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWriteRoundTrips.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWriteRoundTrips.java index 9c42de63a3bd..d2d2042b0ce2 100644 --- a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWriteRoundTrips.java +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWriteRoundTrips.java @@ -13,6 +13,7 @@ */ package io.trino.plugin.ducklake; +import io.trino.Session; import io.trino.testing.AbstractTestQueryFramework; import io.trino.testing.QueryRunner; import org.intellij.lang.annotations.Language; @@ -20,6 +21,7 @@ import java.sql.SQLException; import java.util.List; +import java.util.Set; import static io.trino.testing.TestingNames.randomNameSuffix; import static org.assertj.core.api.Assertions.assertThat; @@ -106,14 +108,12 @@ void testInsertIntoTableDuckDbPartitionedByAColumn() * The shape a daily pipeline writes: DuckDB files the table by a {@code DATE} column and the * pipeline replaces whole days. *

- * The rows end up right, but the days are not dropped from the catalog the way an equivalent - * {@code VARCHAR} or {@code INTEGER} key is. {@code DuckLakeMetadata.isEnforceableType} - * excludes {@code DATE}, so the connector never enforces a predicate on such a column and the - * delete cannot be decided from the partition values. It falls back to reading every row of - * the matching days, and the files are dropped only because nothing is left in them. + * The days are dropped from the catalog the same way an equivalent {@code VARCHAR} or + * {@code INTEGER} key is: the partition value of each file decides the delete, so the files + * end without a row of them being read. */ @Test - void testDeletingWholeDaysOfATableDuckDbPartitionedByDateReadsEveryRow() + void testDeletingWholeDaysOfATableDuckDbPartitionedByDateEndsTheFiles() throws SQLException { String table = "part_day_" + randomNameSuffix(); @@ -132,18 +132,19 @@ void testDeletingWholeDaysOfATableDuckDbPartitionedByDateReadsEveryRow() assertThat(activeDataFileCount(table)).isEqualTo(3); assertThat(partitions(table)).isEqualTo(List.of("2026-01-01", "2026-01-02", "2026-01-03")); - // the two days are removed, but every row of them is read to find out that they are + // the two days are ended by the partition values alone, without reading a row assertQueryStats( getSession(), "DELETE FROM %s WHERE day BETWEEN DATE '2026-01-01' AND DATE '2026-01-02'".formatted(table), - stats -> assertThat(stats.getProcessedInputPositions()).isEqualTo(3), + stats -> assertThat(stats.getProcessedInputPositions()).isEqualTo(0), _ -> {}); assertThat(activeDataFileCount(table)).isEqualTo(1); assertThat(activeDeleteFileCount(table)).isEqualTo(0); assertQuery("SELECT id FROM " + table + " ORDER BY id", "VALUES 4, 5"); assertThat(catalog.rows("SELECT id::VARCHAR FROM " + table + " ORDER BY id")).isEqualTo(List.of("4", "5")); - // part of a day is answered the same way, and leaves the file behind a delete file + // a predicate that reaches beyond the partitioning removes rows one by one, which + // leaves the file in place behind a delete file assertUpdate("DELETE FROM %s WHERE day = DATE '2026-01-03' AND id = 4".formatted(table), 1); assertThat(activeDataFileCount(table)).isEqualTo(1); assertThat(activeDeleteFileCount(table)).isEqualTo(1); @@ -161,18 +162,16 @@ void testDeletingWholeDaysOfATableDuckDbPartitionedByDateReadsEveryRow() } /** - * The same daily shape as {@link #testDeletingWholeDaysOfATableDuckDbPartitionedByDateReadsEveryRow}, + * The same daily shape as {@link #testDeletingWholeDaysOfATableDuckDbPartitionedByDateEndsTheFiles}, * with the day held as a {@code TIMESTAMPTZ} truncated to midnight rather than as a * {@code DATE}. That is the column a pipeline materializing * {@code DATE_TRUNC('day', ts)::TIMESTAMPTZ AS day} files its tables by. *

- * It costs the same. {@code DuckLakeMetadata.isEnforceableType} admits only {@code TINYINT}, - * {@code SMALLINT}, {@code INTEGER}, {@code BIGINT}, {@code BOOLEAN} and unbounded - * {@code VARCHAR}, so a predicate on a column of any timestamp type is never enforced and the - * delete cannot be decided from the partition values. + * It costs the same: nothing. The value is recorded in UTC and read back as the instant it + * names, so the delete is decided from the catalog for this type too. */ @Test - void testDeletingWholeDaysOfATableDuckDbPartitionedByTimestampWithTimeZoneReadsEveryRow() + void testDeletingWholeDaysOfATableDuckDbPartitionedByTimestampWithTimeZoneEndsTheFiles() throws SQLException { String table = "part_tstz_day_" + randomNameSuffix(); @@ -196,13 +195,11 @@ void testDeletingWholeDaysOfATableDuckDbPartitionedByTimestampWithTimeZoneReadsE "2026-01-02 00:00:00+00", "2026-01-03 00:00:00+00")); - // Every row of the two days is read to decide a delete the partition values already - // answer. This is the assertion that flips to 0 once temporal partition predicates - // become enforceable (plan item 0.6b); until then it counts the rows of those days. + // the partition values answer the delete, so the two days end without a row being read assertQueryStats( getSession(), "DELETE FROM %s WHERE day BETWEEN TIMESTAMP '2026-01-01 00:00:00 UTC' AND TIMESTAMP '2026-01-02 00:00:00 UTC'".formatted(table), - stats -> assertThat(stats.getProcessedInputPositions()).isEqualTo(3), + stats -> assertThat(stats.getProcessedInputPositions()).isEqualTo(0), _ -> {}); assertThat(activeDataFileCount(table)).isEqualTo(1); assertThat(activeDeleteFileCount(table)).isEqualTo(0); @@ -220,6 +217,102 @@ void testDeletingWholeDaysOfATableDuckDbPartitionedByTimestampWithTimeZoneReadsE } } + /** + * A day partition holding {@code DATE 'infinity'}, which DuckDB records as the string + * {@code infinity} and no date parser reads back. + *

+ * The value is not guessed at in either direction: the file is never pruned, and it is never + * counted as matching. A predicate the connector cannot decide for one file is one it cannot + * enforce for the table, so the delete is answered row by row and the engine keeps filtering. + * The days it covers still end, because nothing is left in them, and the infinite day is left + * where it is. + */ + @Test + void testDeletingWholeDaysBesideAnInfiniteDateLeavesItAlone() + throws SQLException + { + String table = "part_inf_day_" + randomNameSuffix(); + catalog.executeInDuckDb( + "CREATE TABLE %s (id INTEGER, day DATE, v VARCHAR)".formatted(table), + "ALTER TABLE %s SET PARTITIONED BY (day)".formatted(table)); + try { + assertUpdate( + """ + INSERT INTO %s VALUES + (1, DATE '2026-01-01', 'a'), (2, DATE '2026-01-01', 'b'), + (3, DATE '2026-01-02', 'c'), + (4, DATE '2026-01-03', 'd'), (5, DATE '2026-01-03', 'e')""".formatted(table), + 5); + catalog.executeInDuckDb("INSERT INTO %s VALUES (9, DATE 'infinity', 'inf')".formatted(table)); + assertThat(partitions(table)).isEqualTo(List.of("2026-01-01", "2026-01-02", "2026-01-03", "infinity")); + + // the three rows of the two days are read to place the delete. The infinite day is + // read too, and its file gives up no row, because the predicate excludes the value + // its Parquet statistics report even though its partition value says nothing + assertQueryStats( + getSession(), + "DELETE FROM %s WHERE day BETWEEN DATE '2026-01-01' AND DATE '2026-01-02'".formatted(table), + stats -> assertThat(stats.getProcessedInputPositions()).isEqualTo(3), + _ -> {}); + + // exactly the two days are gone, and no delete file was written for either + assertThat(activeDataFileCount(table)).isEqualTo(2); + assertThat(activeDeleteFileCount(table)).isEqualTo(0); + assertThat(partitions(table)).isEqualTo(List.of("2026-01-03", "infinity")); + + // and the infinite day still reads as the row DuckDB wrote, through both engines + assertQuery("SELECT id FROM " + table + " ORDER BY id", "VALUES 4, 5, 9"); + assertThat(catalog.rows("SELECT day::VARCHAR FROM %s WHERE id = 9".formatted(table))).isEqualTo(List.of("infinity")); + } + finally { + assertUpdate("DROP TABLE " + table); + } + } + + /** + * A read of the same table: the days the connector can read back are pruned by their partition + * value, and the one it cannot is read and filtered rather than dropped. + */ + @Test + void testReadingADayBesideAnInfiniteDatePrunesTheOtherDays() + throws SQLException + { + String table = "read_inf_day_" + randomNameSuffix(); + catalog.executeInDuckDb( + "CREATE TABLE %s (id INTEGER, day DATE, v VARCHAR)".formatted(table), + "ALTER TABLE %s SET PARTITIONED BY (day)".formatted(table)); + try { + assertUpdate( + """ + INSERT INTO %s VALUES + (1, DATE '2026-01-01', 'a'), (2, DATE '2026-01-01', 'b'), + (3, DATE '2026-01-02', 'c'), + (4, DATE '2026-01-03', 'd'), (5, DATE '2026-01-03', 'e')""".formatted(table), + 5); + catalog.executeInDuckDb("INSERT INTO %s VALUES (9, DATE 'infinity', 'inf')".formatted(table)); + + // With the file statistics of the catalog and of the reader both switched off, the + // partition values are the only thing left to prune by. The two days that cannot + // match are not opened; the day asked for and the day whose value does not parse are + // read, and the engine drops the row of the second. + Session partitionPruningOnly = Session.builder(getSession()) + .setCatalogSessionProperty("ducklake", "file_statistics_pruning_enabled", "false") + .setCatalogSessionProperty("ducklake", "parquet_ignore_statistics", "true") + .build(); + assertQueryStats( + partitionPruningOnly, + "SELECT id FROM %s WHERE day = DATE '2026-01-02'".formatted(table), + stats -> assertThat(stats.getProcessedInputPositions()).isEqualTo(2), + result -> assertThat(result.getOnlyColumnAsSet()).isEqualTo(Set.of(3))); + + // the row of the infinite day is not lost: a predicate that covers it returns it + assertQuery("SELECT id FROM %s WHERE day > DATE '2026-01-02' ORDER BY id".formatted(table), "VALUES 4, 5, 9"); + } + finally { + assertUpdate("DROP TABLE " + table); + } + } + @Test void testInsertIntoTableDuckDbPartitionedByTemporalTransforms() throws SQLException diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/util/TestStatsValueParser.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/util/TestStatsValueParser.java index 8f4412b65eb8..406ed1b6a906 100644 --- a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/util/TestStatsValueParser.java +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/util/TestStatsValueParser.java @@ -90,6 +90,26 @@ void testDate() { assertThat(StatsValueParser.parse(DATE, "2024-05-01")).contains(19844L); assertThat(StatsValueParser.parse(DATE, "not a date")).isEmpty(); + // a year past four digits is signed as this connector writes it, and bare as DuckDB does + assertThat(StatsValueParser.parse(DATE, "+10000-01-01")).contains(2932897L); + assertThat(StatsValueParser.parse(DATE, "10000-01-01")).isEmpty(); + } + + /** + * The forms DuckDB writes for a date or a timestamp that names no instant. Reading one back + * would be a guess, so it parses as unknown and leaves the file it partitions unprunable. + */ + @Test + void testTemporalValuesDuckDbWritesThatNameNoInstant() + { + assertThat(StatsValueParser.parse(DATE, "infinity")).isEmpty(); + assertThat(StatsValueParser.parse(DATE, "-infinity")).isEmpty(); + assertThat(StatsValueParser.parse(DATE, "0001-01-01 (BC)")).isEmpty(); + assertThat(StatsValueParser.parse(TIMESTAMP_MICROS, "infinity")).isEmpty(); + assertThat(StatsValueParser.parse(TIMESTAMP_MICROS, "-infinity")).isEmpty(); + assertThat(StatsValueParser.parse(TIMESTAMP_MICROS, "0001-01-01 00:00:00 (BC)")).isEmpty(); + assertThat(StatsValueParser.parse(TIMESTAMP_TZ_MICROS, "infinity")).isEmpty(); + assertThat(StatsValueParser.parse(TIMESTAMP_TZ_MICROS, "-infinity")).isEmpty(); } @Test @@ -109,6 +129,9 @@ void testTimestampWithTimeZone() .contains(LongTimestampWithTimeZone.fromEpochMillisAndFraction(1714559696789L, 12_000_000, UTC_KEY)); assertThat(StatsValueParser.parse(TIMESTAMP_TZ_MICROS, "2024-05-01 12:34:56.789012+02")) .contains(LongTimestampWithTimeZone.fromEpochMillisAndFraction(1714559696789L, 12_000_000, UTC_KEY)); + // a writer in a zone offset by half an hour records the offset with its minutes + assertThat(StatsValueParser.parse(TIMESTAMP_TZ_MICROS, "2024-05-01 16:04:56.789012+05:30")) + .contains(LongTimestampWithTimeZone.fromEpochMillisAndFraction(1714559696789L, 12_000_000, UTC_KEY)); } @Test From 86a249bce08157d943a4b26164eb58df6e8829d9 Mon Sep 17 00:00:00 2001 From: James Greenhill Date: Fri, 4 Sep 2026 00:28:34 +0000 Subject: [PATCH 2/2] Bound the temporal partition value check The check that every visible file's value parses listed the data files of the table. That is one row per file on a table Portola files by day and fills with thousands of files, paid at plan time, on every query with a predicate on the day. JdbcDuckLakeMetastore.distinctPartitionValues asks for the distinct values of one partition key instead: one row per day, not one per file. It returns each value in the shape DuckLakeDataFileEntry carries, so PartitionTransforms still reads it, and a file recording nothing under the key still comes back as the empty map that says so. It is a separate query from the one the split manager prunes by, because applyFilter decides enforcement before any file list is loaded. It runs only for a temporal key that a predicate constrains, and it is the third query enforcement already costs there. The parser tests now pin the day key a pipeline writes: midnight with no fractional seconds parses to the value TIMESTAMP '2026-01-01 00:00:00 UTC' carries at precision 6, and the same instant written with trailing zeros parses to the same value. A possible follow-up, deliberately not taken here: map infinity and -infinity to the exact sentinels DuckDB encodes them as, INT32_MAX days and INT64_MAX microseconds. Trino reads those same numbers back, so such a file would become prunable and a whole-partition DELETE beside one would stay on the metadata path. It would tie the connector to DuckDB's sentinel encoding, so it needs its own decision. Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe --- .../plugin/ducklake/DuckLakeMetadata.java | 24 +++++------ .../metastore/JdbcDuckLakeMetastore.java | 40 +++++++++++++++++++ .../ducklake/util/TestStatsValueParser.java | 7 ++++ 3 files changed, 57 insertions(+), 14 deletions(-) diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java index b0f7885bcc43..ee215be6c4a6 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java @@ -13,7 +13,6 @@ */ package io.trino.plugin.ducklake; -import com.google.common.base.Suppliers; import com.google.common.collect.ArrayListMultimap; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; @@ -105,7 +104,6 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; -import java.util.function.Supplier; import java.util.function.UnaryOperator; import static com.google.common.collect.ImmutableList.toImmutableList; @@ -1695,16 +1693,13 @@ public Optional> applyFilter(C } else { Map partitionColumns = enforceablePartitionColumns(handle); - // the files are only listed for a column whose values have to be checked one by one, - // and then only once however many such columns the predicate constrains - Supplier> dataFiles = Suppliers.memoize(() -> metastore.dataFiles(handle.snapshotId(), handle.tableId())); ImmutableMap.Builder enforceableDomains = ImmutableMap.builder(); ImmutableMap.Builder unenforceableDomains = ImmutableMap.builder(); for (Map.Entry entry : predicate.getDomains().orElseThrow().entrySet()) { DuckLakeColumnHandle column = entry.getKey(); DuckLakePartitionColumn partitionColumn = partitionColumns.get(column.columnId()); if (partitionColumn != null && isEnforceableType(column.type()) - && (!isTemporalType(column.type()) || everyPartitionValueDecidesTheColumn(column, partitionColumn, dataFiles.get()))) { + && (!isTemporalType(column.type()) || everyPartitionValueDecidesTheColumn(handle, column, partitionColumn))) { enforceableDomains.put(column, entry.getValue()); } else { @@ -1797,18 +1792,19 @@ private static boolean isTemporalType(Type type) * and a predicate the connector cannot apply to one file it is not allowed to prune is one it * cannot enforce for the table, so the engine keeps filtering the rows instead. *

- * The check runs over the same files and through the same code as - * {@link DuckLakeSplitManager#getSplits}, which reads the same snapshot, so a predicate - * enforced here is a predicate that splits can be pruned by there. + * The values are read through the same code as {@link DuckLakeSplitManager#getSplits}, at the + * same snapshot, so a predicate enforced here is a predicate splits can be pruned by there. + * The query behind it costs one row per distinct value of the key, not one per data file, so + * a table of many files filed by few days is answered by few rows. */ - private static boolean everyPartitionValueDecidesTheColumn( + private boolean everyPartitionValueDecidesTheColumn( + DuckLakeTableHandle handle, DuckLakeColumnHandle column, - DuckLakePartitionColumn partitionColumn, - List dataFiles) + DuckLakePartitionColumn partitionColumn) { List transforms = ImmutableList.of(partitionColumn); - return dataFiles.stream() - .allMatch(dataFile -> PartitionTransforms.partitionDomain(column, transforms, dataFile.partitionValues()).isPresent()); + return metastore.distinctPartitionValues(handle.snapshotId(), handle.tableId(), partitionColumn.partitionKeyIndex()).stream() + .allMatch(partitionValues -> PartitionTransforms.partitionDomain(column, transforms, partitionValues).isPresent()); } /** diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/JdbcDuckLakeMetastore.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/JdbcDuckLakeMetastore.java index ef2cc17624ee..1c1563d81ec7 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/JdbcDuckLakeMetastore.java +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/JdbcDuckLakeMetastore.java @@ -646,6 +646,46 @@ SELECT count(*) } } + /** + * Returns the distinct partition values one key of the partitioning scheme takes across the + * data files of the table visible at the snapshot, each in the shape + * {@link DuckLakeDataFileEntry#partitionValues()} carries so that both are read the same way. + * A file that records no value under the key contributes the empty map. + *

+ * The result holds one entry per distinct value rather than one per file, which is what makes + * it affordable to ask about a table of many files. + */ + public List>> distinctPartitionValues(long snapshotId, long tableId, int partitionKeyIndex) + { + try (Handle handle = jdbi.open()) { + return handle.createQuery( + """ + SELECT DISTINCT v.partition_key_index, v.partition_value + FROM %s f + LEFT JOIN %s v ON f.data_file_id = v.data_file_id AND f.table_id = v.table_id + AND v.partition_key_index = :partitionKeyIndex + WHERE f.table_id = :tableId AND %s""".formatted( + table("ducklake_data_file"), + table("ducklake_file_partition_value"), + visible("f"))) + .bind("snapshot", snapshotId) + .bind("tableId", tableId) + .bind("partitionKeyIndex", partitionKeyIndex) + .map((rs, _) -> { + int keyIndex = rs.getInt("partition_key_index"); + if (rs.wasNull()) { + // a file that records nothing under the key says nothing about its rows + return Map.>of(); + } + return Map.of(keyIndex, Optional.ofNullable(rs.getString("partition_value"))); + }) + .list(); + } + catch (JdbiException e) { + throw metastoreError(e); + } + } + public List fileColumnStats(long tableId, Set columnIds) { if (columnIds.isEmpty()) { diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/util/TestStatsValueParser.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/util/TestStatsValueParser.java index 406ed1b6a906..fad19ed67d0a 100644 --- a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/util/TestStatsValueParser.java +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/util/TestStatsValueParser.java @@ -132,6 +132,13 @@ void testTimestampWithTimeZone() // a writer in a zone offset by half an hour records the offset with its minutes assertThat(StatsValueParser.parse(TIMESTAMP_TZ_MICROS, "2024-05-01 16:04:56.789012+05:30")) .contains(LongTimestampWithTimeZone.fromEpochMillisAndFraction(1714559696789L, 12_000_000, UTC_KEY)); + // A day key holds midnight, which is recorded without a fraction. It parses to the value + // TIMESTAMP '2026-01-01 00:00:00 UTC' carries at precision 6, so a predicate on the day + // matches the file filed under it, with or without the trailing zeros. + assertThat(StatsValueParser.parse(TIMESTAMP_TZ_MICROS, "2026-01-01 00:00:00+00")) + .contains(LongTimestampWithTimeZone.fromEpochMillisAndFraction(1767225600000L, 0, UTC_KEY)); + assertThat(StatsValueParser.parse(TIMESTAMP_TZ_MICROS, "2026-01-01 00:00:00.000000+00")) + .isEqualTo(StatsValueParser.parse(TIMESTAMP_TZ_MICROS, "2026-01-01 00:00:00+00")); } @Test