This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 6a9c019e2ebe00b5297b4f137553592fc521b17e Author: Jingsong Lee <[email protected]> AuthorDate: Thu Sep 17 16:28:39 2026 +0800 [core][flink][spark] Add API for top partitions (#9914) --- .../paimon/globalindex/DataEvolutionBatchScan.java | 5 + .../paimon/table/FallbackReadFileStoreTable.java | 10 ++ .../paimon/table/format/FormatTableScan.java | 56 +++++++++ .../table/source/AbstractBatchTableScan.java | 7 ++ .../paimon/table/source/PartitionTopNUtils.java | 127 +++++++++++++++++++++ .../org/apache/paimon/table/source/TableScan.java | 23 ++++ .../apache/paimon/table/system/AuditLogTable.java | 5 + .../paimon/table/format/FormatTableScanTest.java | 59 ++++++++++ .../table/source/TableScanListPartitionsTest.java | 47 ++++++++ .../flink/lookup/DynamicPartitionLevelLoader.java | 63 +--------- .../flink/lookup/DynamicPartitionNumberLoader.java | 14 +-- .../spark/catalog/functions/PaimonFunctions.scala | 2 +- .../catalyst/analysis/ReplacePaimonFunctions.scala | 55 +-------- .../paimon/spark/table/PaimonFormatTableTest.scala | 15 +-- 14 files changed, 356 insertions(+), 132 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java index 7a16a2e7ac..e412d8123f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java @@ -270,6 +270,11 @@ public class DataEvolutionBatchScan implements DataTableScan { return batchScan.listPartitionEntries(); } + @Override + public List<BinaryRow> topNPartitions(int num, int partitionFieldCount) { + return batchScan.topNPartitions(num, partitionFieldCount); + } + @Override public Plan plan() { RowRangeIndex rowRangeIndex = this.pushedRowRangeIndex; diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java index 940dd38b2b..cbdf01fd42 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java @@ -43,6 +43,7 @@ import org.apache.paimon.table.source.DataSplit; import org.apache.paimon.table.source.DataTableScan; import org.apache.paimon.table.source.InnerTableRead; import org.apache.paimon.table.source.InnerTableScan; +import org.apache.paimon.table.source.PartitionTopNUtils; import org.apache.paimon.table.source.Split; import org.apache.paimon.table.source.SplitSerializer; import org.apache.paimon.table.source.TableRead; @@ -626,6 +627,15 @@ public class FallbackReadFileStoreTable extends DelegatedFileStoreTable { return partitionEntries; } + @Override + public List<BinaryRow> topNPartitions(int num, int partitionFieldCount) { + return PartitionTopNUtils.topNFileStorePartitions( + listPartitionEntries(), + tableSchema.logicalPartitionType(), + num, + partitionFieldCount); + } + protected void setPartitionPredicate(PartitionPredicate predicate) { this.mainPartitionPredicate = predicate; this.fallbackPartitionPredicate = predicate; diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index 127535773a..32cd464315 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -39,12 +39,14 @@ import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.source.InnerTableScan; +import org.apache.paimon.table.source.PartitionTopNUtils; import org.apache.paimon.table.source.Split; import org.apache.paimon.table.source.TableScan; import org.apache.paimon.types.DataType; import org.apache.paimon.types.RowType; import org.apache.paimon.types.VarCharType; import org.apache.paimon.utils.InternalRowPartitionComputer; +import org.apache.paimon.utils.InternalRowUtils; import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.PartitionPathUtils; @@ -53,6 +55,7 @@ import javax.annotation.Nullable; import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; @@ -104,6 +107,59 @@ public class FormatTableScan implements InnerTableScan { return splitEnumerator.listPartitionEntries(partitionFilter); } + @Override + public List<BinaryRow> topNPartitions(int num, int partitionFieldCount) { + if (table.partitionKeys().isEmpty()) { + throw new UnsupportedOperationException( + "Cannot find top partitions for a non-partitioned table."); + } + + RowType partitionType = table.partitionType(); + PartitionTopNUtils.validateParameters(partitionType, num, partitionFieldCount); + List<List<BinaryRow>> candidateGroups = + PartitionTopNUtils.partitionGroupsDescending( + listPartitionEntries(), partitionType, partitionFieldCount); + List<BinaryRow> result = new ArrayList<>(); + PredicateBuilder builder = new PredicateBuilder(partitionType); + for (List<BinaryRow> candidateGroup : candidateGroups) { + BinaryRow candidate = candidateGroup.get(0); + List<Predicate> prefixPredicates = new ArrayList<>(); + for (int i = 0; i < partitionFieldCount; i++) { + Object value = InternalRowUtils.get(candidate, i, partitionType.getTypeAt(i)); + prefixPredicates.add(value == null ? builder.isNull(i) : builder.equal(i, value)); + } + PartitionPredicate candidateFilter = + PartitionPredicate.fromPredicate( + partitionType, PredicateBuilder.and(prefixPredicates)); + PartitionPredicate combinedFilter = + partitionFilter == null + ? candidateFilter + : PartitionPredicate.and( + Arrays.asList(partitionFilter, candidateFilter)); + try { + List<BinaryRow> nonEmptyPartitions = new ArrayList<>(); + for (Split split : splitEnumerator.plan(combinedFilter).splits()) { + BinaryRow partition = ((FormatDataSplit) split).partition(); + if (partition != null) { + nonEmptyPartitions.add(partition); + } + } + nonEmptyPartitions = + PartitionTopNUtils.distinctPartitionsDescending( + nonEmptyPartitions, partitionType); + if (!nonEmptyPartitions.isEmpty()) { + result.addAll(nonEmptyPartitions); + if (--num == 0) { + return result; + } + } + } catch (IOException e) { + throw new RuntimeException("Failed to scan files", e); + } + } + return result; + } + @Override public InnerTableScan withFilter(Predicate predicate) { throw new UnsupportedOperationException("Filter is not supported for FormatTable."); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java index 1abcfcea30..c95fb56a36 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java @@ -19,6 +19,7 @@ package org.apache.paimon.table.source; import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.manifest.PartitionEntry; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.SortValue; @@ -161,6 +162,12 @@ public abstract class AbstractBatchTableScan extends AbstractDataTableScan { return startingScanner.scanPartitions(snapshotReader); } + @Override + public List<BinaryRow> topNPartitions(int num, int partitionFieldCount) { + return PartitionTopNUtils.topNFileStorePartitions( + listPartitionEntries(), schema.logicalPartitionType(), num, partitionFieldCount); + } + private Optional<StartingScanner.Result> applyPushDownLimit() { // A read-time filter (WHERE or auth) drops rows after scanning, so only push the limit down // when neither is present. diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PartitionTopNUtils.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PartitionTopNUtils.java new file mode 100644 index 0000000000..6f690b5f94 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PartitionTopNUtils.java @@ -0,0 +1,127 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.source; + +import org.apache.paimon.codegen.CodeGenUtils; +import org.apache.paimon.codegen.RecordComparator; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.manifest.PartitionEntry; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Preconditions; + +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Collectors; + +/** Internal utilities for selecting top partition prefixes. */ +public final class PartitionTopNUtils { + + private PartitionTopNUtils() {} + + public static List<BinaryRow> topNFileStorePartitions( + List<PartitionEntry> partitionEntries, + RowType partitionType, + int num, + int partitionFieldCount) { + if (partitionType.getFieldCount() == 0) { + throw new UnsupportedOperationException( + "Cannot find top partitions for a non-partitioned table."); + } + validateParameters(partitionType, num, partitionFieldCount); + List<List<BinaryRow>> groups = + partitionGroupsDescending( + partitionEntries.stream() + .filter(entry -> entry.fileCount() > 0) + .collect(Collectors.toList()), + partitionType, + partitionFieldCount); + List<BinaryRow> result = new ArrayList<>(); + for (int i = 0; i < Math.min(num, groups.size()); i++) { + result.addAll(groups.get(i)); + } + return result; + } + + public static void validateParameters(RowType partitionType, int num, int partitionFieldCount) { + Preconditions.checkArgument(num > 0, "Number of top partitions must be positive."); + validatePartitionFieldCount(partitionType, partitionFieldCount); + } + + public static List<List<BinaryRow>> partitionGroupsDescending( + List<PartitionEntry> partitionEntries, RowType partitionType, int partitionFieldCount) { + return partitionGroupsDescendingFromRows( + partitionEntries.stream() + .map(PartitionEntry::partition) + .collect(Collectors.toList()), + partitionType, + partitionFieldCount); + } + + public static List<BinaryRow> distinctPartitionsDescending( + List<BinaryRow> partitions, RowType partitionType) { + List<List<BinaryRow>> groups = + partitionGroupsDescendingFromRows( + partitions, partitionType, partitionType.getFieldCount()); + List<BinaryRow> result = new ArrayList<>(); + groups.forEach(result::addAll); + return result; + } + + private static List<List<BinaryRow>> partitionGroupsDescendingFromRows( + List<BinaryRow> partitions, RowType partitionType, int partitionFieldCount) { + validatePartitionFieldCount(partitionType, partitionFieldCount); + + int[] prefixFields = new int[partitionFieldCount]; + for (int i = 0; i < partitionFieldCount; i++) { + prefixFields[i] = i; + } + RecordComparator prefixComparator = + CodeGenUtils.newRecordComparator(partitionType.getFieldTypes(), prefixFields); + RecordComparator fullComparator = + CodeGenUtils.newRecordComparator(partitionType.getFieldTypes()); + + List<BinaryRow> sorted = + partitions.stream().sorted(fullComparator.reversed()).collect(Collectors.toList()); + List<List<BinaryRow>> groups = new ArrayList<>(); + BinaryRow previousPartition = null; + BinaryRow previousPrefix = null; + for (BinaryRow partition : sorted) { + if (previousPartition != null + && fullComparator.compare(previousPartition, partition) == 0) { + continue; + } + if (previousPrefix == null + || prefixComparator.compare(previousPrefix, partition) != 0) { + groups.add(new ArrayList<>()); + previousPrefix = partition; + } + groups.get(groups.size() - 1).add(partition); + previousPartition = partition; + } + return groups; + } + + private static void validatePartitionFieldCount( + RowType partitionType, int partitionFieldCount) { + Preconditions.checkArgument( + partitionFieldCount > 0 && partitionFieldCount <= partitionType.getFieldCount(), + "Partition field count must be between 1 and %s.", + partitionType.getFieldCount()); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/TableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/TableScan.java index 31fa8c0f82..318f9c5c68 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/TableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/TableScan.java @@ -18,6 +18,7 @@ package org.apache.paimon.table.source; +import org.apache.paimon.annotation.Experimental; import org.apache.paimon.annotation.Public; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.globalindex.GlobalIndexResult; @@ -57,6 +58,28 @@ public interface TableScan { List<PartitionEntry> listPartitionEntries(); + /** + * Returns all non-empty partitions belonging to the top {@code num} distinct partition + * prefixes. + * + * <p>A prefix consists of the first {@code partitionFieldCount} partition fields. Prefixes and + * returned partitions are ordered descending by their logical field types. All returned rows + * contain the complete partition, so selecting a prefix may return more than {@code num} rows. + * Null values, including default partitions, are ordered below non-null values. The current + * scan's partition filters are respected. + * + * <p>This operation is only supported for batch scans of partitioned tables. {@code num} must + * be positive, and {@code partitionFieldCount} must be between one and the number of partition + * fields. + * + * @since 2.2.0 + */ + @Experimental + default List<BinaryRow> topNPartitions(int num, int partitionFieldCount) { + throw new UnsupportedOperationException( + "Finding top partitions is not supported by " + getClass().getName() + "."); + } + /** * Plan of scan. * diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java index 1e45ed1606..8f259502ff 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java @@ -633,6 +633,11 @@ public class AuditLogTable implements DataTable, ReadonlyTable { return batchScan.listPartitionEntries(); } + @Override + public List<BinaryRow> topNPartitions(int num, int partitionFieldCount) { + return batchScan.topNPartitions(num, partitionFieldCount); + } + @Override public DataTableScan withShard(int indexOfThisSubtask, int numberOfParallelSubtasks) { batchScan.withShard(indexOfThisSubtask, numberOfParallelSubtasks); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java index 53ba4cd067..53669467f4 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java @@ -1394,6 +1394,65 @@ public class FormatTableScanTest { .equals(entry.partition().getString(0)))); } + @TestTemplate + void testTopNPartitions() throws IOException { + Path tableLocation = new Path(tmpPath.toUri()); + LocalFileIO fileIO = LocalFileIO.create(); + FormatTable formatTable = createYearMonthFormatTable(fileIO, tableLocation); + String defaultPartName = formatTable.defaultPartName(); + + String nullPartition = + enablePartitionValueOnly + ? defaultPartName + "/1" + : "year=" + defaultPartName + "/month=1"; + writeTestFile(fileIO, new Path(tableLocation, nullPartition + "/data.csv"), 10); + + String empty2026 = enablePartitionValueOnly ? "2026/1" : "year=2026/month=1"; + fileIO.mkdirs(new Path(tableLocation, empty2026)); + assertThat( + yearMonthPartitions( + new FormatTableScan(formatTable, null, null).topNPartitions(1, 1))) + .containsExactly("null/1"); + + String data2025 = enablePartitionValueOnly ? "2025/1" : "year=2025/month=1"; + writeTestFile(fileIO, new Path(tableLocation, data2025 + "/data.csv"), 10); + assertThat( + yearMonthPartitions( + new FormatTableScan(formatTable, null, null).topNPartitions(1, 1))) + .containsExactly("2025/1"); + + String data2026 = enablePartitionValueOnly ? "2026/2" : "year=2026/month=2"; + writeTestFile(fileIO, new Path(tableLocation, data2026 + "/data.csv"), 10); + String data2026Month3 = enablePartitionValueOnly ? "2026/3" : "year=2026/month=3"; + writeTestFile(fileIO, new Path(tableLocation, data2026Month3 + "/data.csv"), 10); + FormatTableScan scan = new FormatTableScan(formatTable, null, null); + assertThat(yearMonthPartitions(scan.topNPartitions(1, 1))) + .containsExactly("2026/3", "2026/2"); + assertThat(yearMonthPartitions(scan.topNPartitions(2, 1))) + .containsExactly("2026/3", "2026/2", "2025/1"); + assertThat(yearMonthPartitions(scan.topNPartitions(1, 2))).containsExactly("2026/3"); + + PredicateBuilder builder = new PredicateBuilder(formatTable.partitionType()); + PartitionPredicate upTo2025 = + PartitionPredicate.fromPredicate( + formatTable.partitionType(), builder.lessOrEqual(0, 2025)); + assertThat( + yearMonthPartitions( + new FormatTableScan(formatTable, upTo2025, null) + .topNPartitions(1, 1))) + .containsExactly("2025/1"); + } + + private List<String> yearMonthPartitions(List<BinaryRow> partitions) { + return partitions.stream() + .map( + row -> + (row.isNullAt(0) ? "null" : row.getInt(0)) + + "/" + + (row.isNullAt(1) ? "null" : row.getInt(1))) + .collect(java.util.stream.Collectors.toList()); + } + private void createYearMonthPartitionDirs(LocalFileIO fileIO, Path tableLocation) throws IOException { for (int year = 2022; year <= 2026; year++) { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanListPartitionsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanListPartitionsTest.java index 19c2c0b3f2..75b1923012 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanListPartitionsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanListPartitionsTest.java @@ -29,11 +29,13 @@ import org.apache.paimon.table.source.snapshot.ScannerTestBase; import org.junit.jupiter.api.Test; import java.util.Arrays; +import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link TableScan} listPartitions. */ public class TableScanListPartitionsTest extends ScannerTestBase { @@ -62,4 +64,49 @@ public class TableScanListPartitionsTest extends ScannerTestBase { } commit.close(); } + + @Test + public void testTopNPartitions() throws Exception { + createAppendOnlyTable(); + + BatchTableWrite write = table.newWrite(commitUser); + write.write(GenericRow.of(null, 1, 1L)); + TableCommitImpl commit = table.newCommit(commitUser); + commit.commit(write.prepareCommit()); + write.close(); + commit.close(); + + assertThat(table.newReadBuilder().newScan().topNPartitions(1, 1)) + .singleElement() + .satisfies(partition -> assertThat(partition.isNullAt(0)).isTrue()); + + write = table.newWrite(commitUser); + write.write(GenericRow.of(9, 1, 1L)); + write.write(GenericRow.of(10, 1, 1L)); + write.write(GenericRow.of(2, 1, 1L)); + commit = table.newCommit(commitUser); + commit.commit(write.prepareCommit()); + write.close(); + commit.close(); + + assertThat(table.newReadBuilder().newScan().topNPartitions(2, 1)) + .extracting(row -> row.getInt(0)) + .containsExactly(10, 9); + + assertThat( + table.newReadBuilder() + .withPartitionFilter(Collections.singletonMap("pt", "9")) + .newScan() + .topNPartitions(1, 1)) + .extracting(row -> row.getInt(0)) + .containsExactly(9); + + TableScan scan = table.newReadBuilder().newScan(); + assertThatThrownBy(() -> scan.topNPartitions(0, 1)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> scan.topNPartitions(1, 0)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> scan.topNPartitions(1, 2)) + .isInstanceOf(IllegalArgumentException.class); + } } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/DynamicPartitionLevelLoader.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/DynamicPartitionLevelLoader.java index 88920c05fc..835a4afa6c 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/DynamicPartitionLevelLoader.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/DynamicPartitionLevelLoader.java @@ -19,23 +19,15 @@ package org.apache.paimon.flink.lookup; import org.apache.paimon.data.BinaryRow; -import org.apache.paimon.data.InternalRow; -import org.apache.paimon.options.Options; import org.apache.paimon.table.Table; -import org.apache.paimon.types.RowType; import org.apache.paimon.utils.Preconditions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.time.Duration; -import java.util.ArrayList; -import java.util.Arrays; import java.util.List; import java.util.Map; -import java.util.stream.Collectors; - -import static org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME; /** Dynamic partition loader which can specify the partition level to load for lookup. */ public class DynamicPartitionLevelLoader extends DynamicPartitionLoader { @@ -45,17 +37,12 @@ public class DynamicPartitionLevelLoader extends DynamicPartitionLoader { private static final long serialVersionUID = 1L; private final int maxPartitionLoadLevel; - private final List<InternalRow.FieldGetter> fieldGetters; - - private final String defaultPartitionName; DynamicPartitionLevelLoader( Table table, Duration refreshInterval, Map<String, String> partitionLoadConfig) { super(table, refreshInterval); maxPartitionLoadLevel = getMaxPartitionLoadLevel(partitionLoadConfig, table.partitionKeys()); - fieldGetters = createPartitionFieldGetters(); - defaultPartitionName = Options.fromMap(table.options()).get(PARTITION_DEFAULT_NAME); LOG.info( "Init DynamicPartitionLevelLoader(table={}),maxPartitionLoadLevel is {}", @@ -66,21 +53,7 @@ public class DynamicPartitionLevelLoader extends DynamicPartitionLoader { @Override public List<BinaryRow> getMaxPartitions() { List<BinaryRow> newPartitions = - table.newReadBuilder().newScan().listPartitions().stream() - .sorted(comparator.reversed()) - .collect(Collectors.toList()); - - if (maxPartitionLoadLevel == table.partitionKeys().size() - 1) { - // if maxPartitionLoadLevel is the max partition level, we only need to load the max - // partition - if (newPartitions.size() <= 1) { - return newPartitions; - } else { - return newPartitions.subList(0, 1); - } - } - - newPartitions = extractMaxPartitionsForFixedLevel(newPartitions, maxPartitionLoadLevel); + table.newReadBuilder().newScan().topNPartitions(1, maxPartitionLoadLevel + 1); if (LOG.isDebugEnabled()) { LOG.debug( "DynamicPartitionLevelLoader(currentPartitionLoadLevel={},table={}) finds new partitions: {}.", @@ -114,38 +87,4 @@ public class DynamicPartitionLevelLoader extends DynamicPartitionLoader { } return maxLoadLevel; } - - private List<InternalRow.FieldGetter> createPartitionFieldGetters() { - List<InternalRow.FieldGetter> fieldGetters = new ArrayList<>(); - - RowType partitionType = table.rowType().project(table.partitionKeys()); - - for (int i = 0; i < maxPartitionLoadLevel + 1; i++) { - fieldGetters.add(InternalRow.createFieldGetter(partitionType.getTypeAt(i), i)); - } - return fieldGetters; - } - - private List<BinaryRow> extractMaxPartitionsForFixedLevel( - List<BinaryRow> partitions, int level) { - int currentDistinct = 0; - Object[] lastFields = new Object[level + 1]; - for (int i = 0; i < partitions.size(); i++) { - BinaryRow partition = partitions.get(i); - Object[] newFields = new Object[level + 1]; - for (int j = 0; j <= level; j++) { - newFields[j] = fieldGetters.get(j).getFieldOrNull(partition); - if (newFields[j] == null) { - newFields[j] = defaultPartitionName; - } - } - if (!Arrays.equals(newFields, lastFields)) { - lastFields = newFields; - if (++currentDistinct > 1) { - return partitions.subList(0, i); - } - } - } - return partitions; - } } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/DynamicPartitionNumberLoader.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/DynamicPartitionNumberLoader.java index 7e210725de..bddf4b2616 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/DynamicPartitionNumberLoader.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/DynamicPartitionNumberLoader.java @@ -26,7 +26,6 @@ import org.slf4j.LoggerFactory; import java.time.Duration; import java.util.List; -import java.util.stream.Collectors; /** Dynamic partition loader which can specify the max partition number to load for lookup. */ public class DynamicPartitionNumberLoader extends DynamicPartitionLoader { @@ -48,15 +47,8 @@ public class DynamicPartitionNumberLoader extends DynamicPartitionLoader { @Override public List<BinaryRow> getMaxPartitions() { - List<BinaryRow> newPartitions = - table.newReadBuilder().newScan().listPartitions().stream() - .sorted(comparator.reversed()) - .collect(Collectors.toList()); - - if (newPartitions.size() <= maxPartitionNum) { - return newPartitions; - } else { - return newPartitions.subList(0, maxPartitionNum); - } + return table.newReadBuilder() + .newScan() + .topNPartitions(maxPartitionNum, table.partitionKeys().size()); } } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/PaimonFunctions.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/PaimonFunctions.scala index 0ac3b5f6bf..f212945fdf 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/PaimonFunctions.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalog/functions/PaimonFunctions.scala @@ -176,7 +176,7 @@ object BucketFunction { /** * For partitioned tables, this function returns the maximum value of the first level partition of - * the partitioned table, sorted alphabetically. Note, empty partitions will be skipped. For + * the partitioned table, ordered by its logical type. Note, empty partitions will be skipped. For * example, a partition created by `alter table ... add partition ...`. */ class MaxPtFunction extends UnboundFunction { diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/ReplacePaimonFunctions.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/ReplacePaimonFunctions.scala index d81560ffe6..2c981b23a5 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/ReplacePaimonFunctions.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/ReplacePaimonFunctions.scala @@ -18,15 +18,12 @@ package org.apache.paimon.spark.catalyst.analysis -import org.apache.paimon.partition.PartitionPredicate -import org.apache.paimon.predicate.PredicateBuilder import org.apache.paimon.spark.{BaseTable, DataConverter, SparkTable, SparkTypeUtils, SparkUtils} import org.apache.paimon.spark.catalog.SparkBaseCatalog import org.apache.paimon.spark.catalog.functions.PaimonFunctions import org.apache.paimon.spark.function.{BlobViewFieldIdSparkFunction, BlobViewSparkFunction, DescriptorToPresignedUrlFunction, ResolvedDescriptorToPresignedUrlFunction} import org.apache.paimon.spark.utils.CatalogUtils import org.apache.paimon.table.DataTable -import org.apache.paimon.table.FormatTable import org.apache.paimon.types.DataTypeRoot import org.apache.paimon.utils.{InternalRowUtils, TypeUtils} @@ -40,8 +37,6 @@ import org.apache.spark.sql.connector.catalog.PaimonCatalogImplicits._ import org.apache.spark.sql.types.{BinaryType, DataType, DayTimeIntervalType, NullType, StringType} import org.apache.spark.unsafe.types.UTF8String -import scala.collection.JavaConverters._ - object ReplacePaimonFunctions { def resolveDescriptorToPresignedUrl( @@ -167,55 +162,17 @@ case class ReplacePaimonFunctions(spark: SparkSession) extends Rule[LogicalPlan] val toplevelPartitionType = TypeUtils.project(paimonTable.rowType, paimonTable.partitionKeys()).getTypeAt(0) - val partitions = paimonTable match { - case formatTable: FormatTable => - val partitionType = TypeUtils.project(paimonTable.rowType, paimonTable.partitionKeys()) - val candidates = formatTable.newReadBuilder.newScan - .listPartitionEntries() - .asScala - .map(entry => InternalRowUtils.get(entry.partition(), 0, toplevelPartitionType)) - .filter(_ != null) - .distinct - .sortWith(InternalRowUtils.compare(_, _, toplevelPartitionType.getTypeRoot) > 0) - - // Catalog metadata uses zero both for an empty format table partition and for a - // partition whose file count has not been reported. Check candidates from largest to - // smallest and stop at the first value whose filtered scan produces data. This also avoids - // treating an empty filesystem directory as data without planning splits for the whole - // table. - candidates.find { - candidate => - val predicate = new PredicateBuilder(partitionType).equal(0, candidate) - val partitionFilter = - PartitionPredicate.fromPredicate(partitionType, predicate) - !formatTable.newReadBuilder - .withPartitionFilter(partitionFilter) - .newScan - .plan() - .splits() - .isEmpty - }.toSeq - case _ => - // FileStoreTable manifests carry an exact file count, so keep the cheaper metadata path. - paimonTable.newReadBuilder.newScan - .listPartitionEntries() - .asScala - .filter(_.fileCount() > 0) - .map(entry => InternalRowUtils.get(entry.partition(), 0, toplevelPartitionType)) - } - val partitionValues = partitions - // The default partition comes back as a real null, which InternalRowUtils.compare would - // dereference. It is also not a value anyone means by "the max partition". - .filter(_ != null) - .sortWith(InternalRowUtils.compare(_, _, toplevelPartitionType.getTypeRoot) < 0) - .map(DataConverter.fromPaimon(_, toplevelPartitionType)) - if (partitionValues.isEmpty) { + val topPartitions = paimonTable.newReadBuilder.newScan.topNPartitions(1, 1) + if (topPartitions.isEmpty) { throw new UnsupportedOperationException( s"$table has no partitions or none of the partitions have any data") } + val partitionValue = + InternalRowUtils.get(topPartitions.get(0), 0, toplevelPartitionType) val sparkType = SparkTypeUtils.fromPaimonType(toplevelPartitionType) - val literal = Literal(partitionValues.last, sparkType) + val literal = + Literal(DataConverter.fromPaimon(partitionValue, toplevelPartitionType), sparkType) Cast(literal, func.dataType) } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala index e554c72322..2668929b2f 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala @@ -278,18 +278,17 @@ class PaimonFormatTableTest extends PaimonSparkTestWithRestCatalogBase { } } - test("PaimonFormatTable: max_pt skips the default partition") { + test("PaimonFormatTable: max_pt ranks the default partition lowest") { val tableName = "max_pt_null" withTable(tableName) { sql( s"CREATE TABLE $tableName (f0 INT) USING CSV PARTITIONED BY (ds STRING) " + "TBLPROPERTIES ('format-table.implementation'='paimon', " + "'metastore.partitioned-table'='true')") - sql(s"INSERT INTO $tableName VALUES (1, '20240101')") - // A null partition value comes back as a real null, which the typed comparator would - // dereference; it is also not what anyone means by the max partition. sql(s"INSERT INTO $tableName VALUES (2, null)") + checkAnswer(sql(s"SELECT sys.max_pt('test_db.$tableName')"), Seq(Row(null))) + sql(s"INSERT INTO $tableName VALUES (1, '20240101')") checkAnswer(sql(s"SELECT sys.max_pt('test_db.$tableName')"), Seq(Row("20240101"))) } } @@ -324,16 +323,14 @@ class PaimonFormatTableTest extends PaimonSparkTestWithRestCatalogBase { } } - test("max_pt on a FileStoreTable skips a null partition instead of failing") { + test("max_pt on a FileStoreTable ranks a null partition lowest") { val tableName = "max_pt_fst_null" withTable(tableName) { sql(s"CREATE TABLE $tableName (id INT, ds STRING) USING paimon PARTITIONED BY (ds)") - sql(s"INSERT INTO $tableName VALUES (1, '20240101')") sql(s"INSERT INTO $tableName VALUES (2, null)") + checkAnswer(sql(s"SELECT sys.max_pt('test_db.$tableName')"), Seq(Row(null))) - // This is a behaviour change for FileStoreTable, not only for format tables: a null - // top-level partition value used to reach InternalRowUtils.compare and throw, and is now - // skipped. Pinned here so the change is visible rather than incidental. + sql(s"INSERT INTO $tableName VALUES (1, '20240101')") checkAnswer(sql(s"SELECT sys.max_pt('test_db.$tableName')"), Seq(Row("20240101"))) } }
