This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 58fa4eec30 [core][flink][spark] Add API for top partitions (#9914)
58fa4eec30 is described below
commit 58fa4eec30b6b3f453cbdcfa2cb9854bf0dd3652
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")))
}
}
