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..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 @@ -85,6 +85,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; @@ -124,6 +126,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 +1692,14 @@ public Optional> applyFilter(C newUnenforcedConstraint = TupleDomain.all(); } else { - Set enforceableColumnIds = enforceablePartitionColumnIds(handle); + Map partitionColumns = enforceablePartitionColumns(handle); 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(handle, column, partitionColumn))) { enforceableDomains.put(column, entry.getValue()); } else { @@ -1730,32 +1735,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 +1771,40 @@ 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 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 boolean everyPartitionValueDecidesTheColumn( + DuckLakeTableHandle handle, + DuckLakeColumnHandle column, + DuckLakePartitionColumn partitionColumn) + { + List transforms = ImmutableList.of(partitionColumn); + 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/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..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 @@ -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,16 @@ 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)); + // 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