You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2025/10/30 09:45:19 UTC
(paimon) branch release-1.3 updated (4f9bfd85fc -> d495257d4c)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a change to branch release-1.3
in repository https://gitbox.apache.org/repos/asf/paimon.git
from 4f9bfd85fc [Python] Move __version__ to setup.py (#6491)
new 3c213dc7ee [doc] fix MarkPartitionDoneProcedure doc. (#6473)
new d3a7451c16 [doc] add view doc (#6469)
new f170713ac8 [doc] Fix typo in flink/sql-ddl.md (#6476)
new b856f28363 [typo] fix typo in distinct (#6475)
new 5c90f445a0 [doc] Update batch partition mark done doc (#6478)
new de1f9c26b5 [core] support automatically clustering historical partition (#6472)
new 718f124d37 [core] Refactor HistoryPartitionCluster to load less history partitions
new 5625e47e48 [spark] Fix write non-pk dv table with external paths (#6487)
new d495257d4c [core] Fix endian spec for BloomFilter index reader (#6493)
The 9 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/content/concepts/spec/_index.md | 2 +-
docs/content/concepts/views.md | 107 ++++++++++++
docs/content/flink/sql-ddl.md | 2 +-
docs/content/maintenance/manage-partitions.md | 4 +-
docs/content/spark/procedures.md | 4 +-
.../shortcodes/generated/core_configuration.html | 12 ++
.../main/java/org/apache/paimon/CoreOptions.java | 24 +++
.../bloomfilter/BloomFilterFileIndex.java | 2 +-
.../append/cluster/HistoryPartitionCluster.java | 184 +++++++++++++++++++++
.../append/cluster/IncrementalClusterManager.java | 124 +++++++-------
.../compact/aggregate/FieldCollectAgg.java | 2 +-
.../paimon/operation/AbstractFileStoreScan.java | 6 +
.../org/apache/paimon/operation/FileStoreScan.java | 2 +
.../apache/paimon/operation/ManifestsReader.java | 10 ++
.../table/source/snapshot/SnapshotReader.java | 3 +
.../table/source/snapshot/SnapshotReaderImpl.java | 7 +
.../apache/paimon/table/system/AuditLogTable.java | 7 +
.../cluster/HistoryPartitionClusterTest.java | 135 +++++++++++++++
.../cluster/IncrementalClusterManagerTest.java | 90 +++++++++-
.../cluster/IncrementalClusterStrategyTest.java | 2 +-
.../action/IncrementalClusterActionITCase.java | 173 ++++++++++++++++++-
.../spark/commands/PaimonRowLevelCommand.scala | 31 +---
.../paimon/spark/commands/PaimonSparkWriter.scala | 5 +-
.../paimon/spark/commands/SparkDataFileMeta.scala | 13 +-
.../spark/commands/SparkDeletionVector.scala | 23 +--
.../spark/commands/UpdatePaimonTableCommand.scala | 7 +-
.../paimon/spark/sql/DeletionVectorTest.scala | 33 +++-
27 files changed, 887 insertions(+), 127 deletions(-)
create mode 100644 docs/content/concepts/views.md
create mode 100644 paimon-core/src/main/java/org/apache/paimon/append/cluster/HistoryPartitionCluster.java
create mode 100644 paimon-core/src/test/java/org/apache/paimon/append/cluster/HistoryPartitionClusterTest.java
(paimon) 06/09: [core] support automatically clustering historical partition (#6472)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.3
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit de1f9c26b51ba4c342d30a869e25c215ecd7407e
Author: LsomeYeah <94...@users.noreply.github.com>
AuthorDate: Tue Oct 28 21:09:42 2025 +0800
[core] support automatically clustering historical partition (#6472)
---
.../shortcodes/generated/core_configuration.html | 12 ++
.../main/java/org/apache/paimon/CoreOptions.java | 24 +++
.../append/cluster/HistoryPartitionCluster.java | 189 ++++++++++++++++++++
.../append/cluster/IncrementalClusterManager.java | 101 +++++++----
.../paimon/operation/AbstractFileStoreScan.java | 6 +
.../org/apache/paimon/operation/FileStoreScan.java | 2 +
.../apache/paimon/operation/ManifestsReader.java | 10 ++
.../table/source/snapshot/SnapshotReader.java | 3 +
.../table/source/snapshot/SnapshotReaderImpl.java | 6 +
.../apache/paimon/table/system/AuditLogTable.java | 8 +
.../cluster/HistoryPartitionClusterTest.java | 190 +++++++++++++++++++++
.../cluster/IncrementalClusterManagerTest.java | 88 +++++++++-
.../cluster/IncrementalClusterStrategyTest.java | 2 +-
.../action/IncrementalClusterActionITCase.java | 173 ++++++++++++++++++-
14 files changed, 777 insertions(+), 37 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
index 3411e2bd92..2d4abf9787 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -146,6 +146,18 @@ under the License.
<td>String</td>
<td>Specifies the column name(s) used for comparison during range partitioning, in the format 'columnName1,columnName2'. If not set or set to an empty string, it indicates that the range partitioning feature is not enabled. This option will be effective only for append table without primary keys and batch execution mode.</td>
</tr>
+ <tr>
+ <td><h5>clustering.history-partition.idle-to-full-sort</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>The duration after which a partition without new updates is considered a historical partition. Historical partitions will be automatically fully clustered during the cluster operation.This option takes effects when 'clustering.history-partition.auto.enabled' is true.</td>
+ </tr>
+ <tr>
+ <td><h5>clustering.history-partition.limit</h5></td>
+ <td style="word-wrap: break-word;">5</td>
+ <td>Integer</td>
+ <td>The limit of history partition number for automatically performing full clustering.</td>
+ </tr>
<tr>
<td><h5>clustering.incremental</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index d5209e54db..a3da8ef5f1 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1939,6 +1939,22 @@ public class CoreOptions implements Serializable {
.defaultValue(false)
.withDescription("Whether enable incremental clustering.");
+ public static final ConfigOption<Integer> CLUSTERING_HISTORY_PARTITION_LIMIT =
+ key("clustering.history-partition.limit")
+ .intType()
+ .defaultValue(5)
+ .withDescription(
+ "The limit of history partition number for automatically performing full clustering.");
+
+ public static final ConfigOption<Duration> CLUSTERING_HISTORY_PARTITION_IDLE_TIME =
+ key("clustering.history-partition.idle-to-full-sort")
+ .durationType()
+ .noDefaultValue()
+ .withDescription(
+ "The duration after which a partition without new updates is considered a historical partition. "
+ + "Historical partitions will be automatically fully clustered during the cluster operation."
+ + "This option takes effects when 'clustering.history-partition.auto.enabled' is true.");
+
public static final ConfigOption<Boolean> ROW_TRACKING_ENABLED =
key("row-tracking.enabled")
.booleanType()
@@ -3006,6 +3022,14 @@ public class CoreOptions implements Serializable {
return options.get(CLUSTERING_INCREMENTAL);
}
+ public Duration clusteringHistoryPartitionIdleTime() {
+ return options.get(CLUSTERING_HISTORY_PARTITION_IDLE_TIME);
+ }
+
+ public int clusteringHistoryPartitionLimit() {
+ return options.get(CLUSTERING_HISTORY_PARTITION_LIMIT);
+ }
+
public OrderType clusteringStrategy(int columnSize) {
return clusteringStrategy(options.get(CLUSTERING_STRATEGY), columnSize);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/cluster/HistoryPartitionCluster.java b/paimon-core/src/main/java/org/apache/paimon/append/cluster/HistoryPartitionCluster.java
new file mode 100644
index 0000000000..cccecdd2be
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/append/cluster/HistoryPartitionCluster.java
@@ -0,0 +1,189 @@
+/*
+ * 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.append.cluster;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.compact.CompactUnit;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.mergetree.LevelSortedRun;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.utils.BiFilter;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.append.cluster.IncrementalClusterManager.constructPartitionLevels;
+import static org.apache.paimon.append.cluster.IncrementalClusterManager.logForPartitionLevel;
+
+/** Handle historical partition for full clustering. */
+public class HistoryPartitionCluster {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HistoryPartitionCluster.class);
+ private final InternalRowPartitionComputer partitionComputer;
+
+ private final FileStoreTable table;
+ private final IncrementalClusterStrategy incrementalClusterStrategy;
+ private final int maxLevel;
+
+ private final int historyPartitionLimit;
+ @Nullable private final PartitionPredicate specifiedPartitions;
+ @Nullable private final Duration historyPartitionIdleTime;
+ @Nullable private final BiFilter<Integer, Integer> partitionLevelFilter;
+
+ public HistoryPartitionCluster(
+ FileStoreTable table,
+ IncrementalClusterStrategy incrementalClusterStrategy,
+ InternalRowPartitionComputer partitionComputer,
+ int maxLevel,
+ int historyPartitionLimit,
+ @Nullable PartitionPredicate specifiedPartitions,
+ @Nullable Duration historyPartitionIdleTime) {
+ this.table = table;
+ this.incrementalClusterStrategy = incrementalClusterStrategy;
+ this.partitionComputer = partitionComputer;
+ this.maxLevel = maxLevel;
+ this.historyPartitionLimit = historyPartitionLimit;
+ this.specifiedPartitions = specifiedPartitions;
+ this.historyPartitionIdleTime = historyPartitionIdleTime;
+ // (maxLevel + 1) / 2 is used to calculate the ceiling of maxLevel divided by 2
+ this.partitionLevelFilter =
+ (partitionMinLevel, partitionMaxLevel) -> partitionMinLevel < (maxLevel + 1) / 2;
+ }
+
+ public Map<BinaryRow, Optional<CompactUnit>> pickForHistoryPartitions() {
+ Map<BinaryRow, List<LevelSortedRun>> partitionLevels =
+ constructLevelsForHistoryPartitions();
+ logForPartitionLevel(partitionLevels, partitionComputer);
+
+ return partitionLevels.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry ->
+ incrementalClusterStrategy.pick(
+ maxLevel, entry.getValue(), true)));
+ }
+
+ @VisibleForTesting
+ public Map<BinaryRow, List<LevelSortedRun>> constructLevelsForHistoryPartitions() {
+ if (specifiedPartitions == null
+ || historyPartitionIdleTime == null
+ || historyPartitionLimit <= 0) {
+ return Collections.emptyMap();
+ }
+
+ long historyMilli =
+ LocalDateTime.now()
+ .minus(historyPartitionIdleTime)
+ .atZone(ZoneId.systemDefault())
+ .toInstant()
+ .toEpochMilli();
+ // read partitionEntries filter by partitionLevelFilter historyPartitionIdleTime
+ // sort partitionEntries by lastFileCreation time, and we will pick the oldest N partitions
+ List<BinaryRow> historyPartitions =
+ table.newSnapshotReader().withManifestLevelFilter(partitionLevelFilter)
+ .partitionEntries().stream()
+ .filter(entry -> entry.lastFileCreationTime() < historyMilli)
+ .sorted(Comparator.comparingLong(PartitionEntry::lastFileCreationTime))
+ .map(PartitionEntry::partition)
+ .collect(Collectors.toList());
+
+ // read dataFileMeta for history partitions
+ List<DataSplit> historyDataSplits =
+ table.newSnapshotReader()
+ .withPartitionFilter(historyPartitions)
+ .read()
+ .dataSplits();
+ Map<BinaryRow, List<DataFileMeta>> historyPartitionFiles = new HashMap<>();
+ for (DataSplit dataSplit : historyDataSplits) {
+ historyPartitionFiles
+ .computeIfAbsent(dataSplit.partition(), k -> new ArrayList<>())
+ .addAll(dataSplit.dataFiles());
+ }
+
+ // find history partitions which have low-level files
+ Set<BinaryRow> selectedHistoryPartitions =
+ findLowLevelPartitions(historyPartitions, historyPartitionFiles);
+ historyPartitionFiles =
+ historyPartitionFiles.entrySet().stream()
+ .filter(entry -> selectedHistoryPartitions.contains(entry.getKey()))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ return historyPartitionFiles.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry -> constructPartitionLevels(entry.getValue())));
+ }
+
+ @VisibleForTesting
+ protected Set<BinaryRow> findLowLevelPartitions(
+ List<BinaryRow> historyPartitions, Map<BinaryRow, List<DataFileMeta>> partitionFiles) {
+ Set<BinaryRow> partitions = new HashSet<>();
+ // 1. the partition is not specified in specifiedPartitions
+ // 2. the min file level in partition should be less than Math.ceil(maxLevel/2)
+ for (BinaryRow historyPartition : historyPartitions) {
+ if (specifiedPartitions != null && !specifiedPartitions.test(historyPartition)) {
+ List<DataFileMeta> files =
+ partitionFiles.getOrDefault(historyPartition, Collections.emptyList());
+ if (!files.isEmpty()) {
+ int partitionMinLevel = maxLevel + 1;
+ for (DataFileMeta file : files) {
+ partitionMinLevel = Math.min(partitionMinLevel, file.level());
+ }
+ if (partitionLevelFilter != null
+ && partitionLevelFilter.test(partitionMinLevel, maxLevel)) {
+ partitions.add(historyPartition);
+ if (partitions.size() >= historyPartitionLimit) {
+ break;
+ }
+ }
+ }
+ }
+ }
+ LOG.info(
+ "Find {} history partitions for full clustering, the history partitions are {}",
+ partitions.size(),
+ partitions.stream()
+ .map(partitionComputer::generatePartValues)
+ .collect(Collectors.toSet()));
+ return partitions;
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java b/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
index 1aa4d2b13e..8142f9dbf0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
@@ -30,6 +30,7 @@ import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.SplitGenerator;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,6 +52,7 @@ import static org.apache.paimon.utils.Preconditions.checkArgument;
public class IncrementalClusterManager {
private static final Logger LOG = LoggerFactory.getLogger(IncrementalClusterManager.class);
+ private final InternalRowPartitionComputer partitionComputer;
private final SnapshotReader snapshotReader;
@@ -58,6 +60,8 @@ public class IncrementalClusterManager {
private final CoreOptions.OrderType clusterCurve;
private final List<String> clusterKeys;
+ private final HistoryPartitionCluster historyPartitionCluster;
+
private int maxLevel;
public IncrementalClusterManager(FileStoreTable table) {
@@ -69,14 +73,20 @@ public class IncrementalClusterManager {
checkArgument(
table.bucketMode() == BucketMode.BUCKET_UNAWARE,
"only append unaware-bucket table support incremental clustering.");
- // drop stats to reduce memory usage
- this.snapshotReader =
- table.newSnapshotReader().withPartitionFilter(specifiedPartitions).dropStats();
CoreOptions options = table.coreOptions();
checkArgument(
options.clusteringIncrementalEnabled(),
"Only support incremental clustering when '%s' is true.",
CLUSTERING_INCREMENTAL.key());
+ this.maxLevel = options.numLevels();
+ this.partitionComputer =
+ new InternalRowPartitionComputer(
+ table.coreOptions().partitionDefaultName(),
+ table.store().partitionType(),
+ table.partitionKeys().toArray(new String[0]),
+ table.coreOptions().legacyPartitionName());
+ this.snapshotReader =
+ table.newSnapshotReader().dropStats().withPartitionFilter(specifiedPartitions);
this.incrementalClusterStrategy =
new IncrementalClusterStrategy(
table.schemaManager(),
@@ -86,31 +96,22 @@ public class IncrementalClusterManager {
options.numSortedRunCompactionTrigger());
this.clusterCurve = options.clusteringStrategy(options.clusteringColumns().size());
this.clusterKeys = options.clusteringColumns();
- this.maxLevel = options.numLevels();
+
+ this.historyPartitionCluster =
+ new HistoryPartitionCluster(
+ table,
+ incrementalClusterStrategy,
+ partitionComputer,
+ maxLevel,
+ options.clusteringHistoryPartitionLimit(),
+ specifiedPartitions,
+ options.clusteringHistoryPartitionIdleTime());
}
public Map<BinaryRow, CompactUnit> prepareForCluster(boolean fullCompaction) {
// 1. construct LSM structure for each partition
Map<BinaryRow, List<LevelSortedRun>> partitionLevels = constructLevels();
- if (LOG.isDebugEnabled()) {
- partitionLevels.forEach(
- (partition, levelSortedRuns) -> {
- String runsInfo =
- levelSortedRuns.stream()
- .map(
- lsr ->
- String.format(
- "level-%s:%s",
- lsr.level(),
- lsr.run().files().size()))
- .collect(Collectors.joining(","));
- LOG.debug(
- "Partition {} has {} runs: [{}]",
- partition,
- levelSortedRuns.size(),
- runsInfo);
- });
- }
+ logForPartitionLevel(partitionLevels, partitionComputer);
// 2. pick files to be clustered for each partition
Map<BinaryRow, Optional<CompactUnit>> units =
@@ -124,6 +125,10 @@ public class IncrementalClusterManager {
entry.getValue(),
fullCompaction)));
+ Map<BinaryRow, Optional<CompactUnit>> historyUnits =
+ historyPartitionCluster.pickForHistoryPartitions();
+ units.putAll(historyUnits);
+
// 3. filter out empty units
Map<BinaryRow, CompactUnit> filteredUnits =
units.entrySet().stream()
@@ -146,7 +151,7 @@ public class IncrementalClusterManager {
.collect(Collectors.joining(", "));
LOG.debug(
"Partition {}, outputLevel:{}, clustered with {} files: [{}]",
- partition,
+ partitionComputer.generatePartValues(partition),
compactUnit.outputLevel(),
compactUnit.files().size(),
filesInfo);
@@ -169,13 +174,7 @@ public class IncrementalClusterManager {
+ 1);
checkArgument(maxLevel > 1, "Number of levels must be at least 2.");
- Map<BinaryRow, List<DataFileMeta>> partitionFiles = new HashMap<>();
- for (DataSplit dataSplit : dataSplits) {
- partitionFiles
- .computeIfAbsent(dataSplit.partition(), k -> new ArrayList<>())
- .addAll(dataSplit.dataFiles());
- }
-
+ Map<BinaryRow, List<DataFileMeta>> partitionFiles = getPartitionFiles(dataSplits);
return partitionFiles.entrySet().stream()
.collect(
Collectors.toMap(
@@ -183,7 +182,7 @@ public class IncrementalClusterManager {
entry -> constructPartitionLevels(entry.getValue())));
}
- public List<LevelSortedRun> constructPartitionLevels(List<DataFileMeta> partitionFiles) {
+ public static List<LevelSortedRun> constructPartitionLevels(List<DataFileMeta> partitionFiles) {
List<LevelSortedRun> partitionLevels = new ArrayList<>();
Map<Integer, List<DataFileMeta>> levelMap =
partitionFiles.stream().collect(Collectors.groupingBy(DataFileMeta::level));
@@ -209,6 +208,16 @@ public class IncrementalClusterManager {
return partitionLevels;
}
+ private Map<BinaryRow, List<DataFileMeta>> getPartitionFiles(List<DataSplit> dataSplits) {
+ Map<BinaryRow, List<DataFileMeta>> partitionFiles = new HashMap<>();
+ for (DataSplit dataSplit : dataSplits) {
+ partitionFiles
+ .computeIfAbsent(dataSplit.partition(), k -> new ArrayList<>())
+ .addAll(dataSplit.dataFiles());
+ }
+ return partitionFiles;
+ }
+
public List<DataSplit> toSplits(BinaryRow partition, List<DataFileMeta> files) {
List<DataSplit> splits = new ArrayList<>();
@@ -242,6 +251,30 @@ public class IncrementalClusterManager {
.collect(Collectors.toList());
}
+ public static void logForPartitionLevel(
+ Map<BinaryRow, List<LevelSortedRun>> partitionLevels,
+ InternalRowPartitionComputer partitionComputer) {
+ if (LOG.isDebugEnabled()) {
+ partitionLevels.forEach(
+ (partition, levelSortedRuns) -> {
+ String runsInfo =
+ levelSortedRuns.stream()
+ .map(
+ lsr ->
+ String.format(
+ "level-%s:%s",
+ lsr.level(),
+ lsr.run().files().size()))
+ .collect(Collectors.joining(","));
+ LOG.debug(
+ "Partition {} has {} runs: [{}]",
+ partitionComputer.generatePartValues(partition),
+ levelSortedRuns.size(),
+ runsInfo);
+ });
+ }
+ }
+
public CoreOptions.OrderType clusterCurve() {
return clusterCurve;
}
@@ -249,4 +282,8 @@ public class IncrementalClusterManager {
public List<String> clusterKeys() {
return clusterKeys;
}
+
+ public HistoryPartitionCluster historyPartitionCluster() {
+ return historyPartitionCluster;
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
index b21c77db1d..6f06514ed3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
@@ -196,6 +196,12 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
return this;
}
+ @Override
+ public FileStoreScan withManifestLevelFilter(BiFilter<Integer, Integer> manifestLevelFilter) {
+ manifestsReader.withManifestLevelFilter(manifestLevelFilter);
+ return this;
+ }
+
@Override
public FileStoreScan enableValueFilter() {
return this;
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
index 5fa3127237..354c43e019 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
@@ -75,6 +75,8 @@ public interface FileStoreScan {
FileStoreScan withLevelFilter(Filter<Integer> levelFilter);
+ FileStoreScan withManifestLevelFilter(BiFilter<Integer, Integer> manifestLevelFilter);
+
FileStoreScan enableValueFilter();
FileStoreScan withManifestEntryFilter(Filter<ManifestEntry> filter);
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
index d58bb797e3..dbf5140583 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
@@ -27,6 +27,7 @@ import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.table.source.ScanMode;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.BiFilter;
import org.apache.paimon.utils.SnapshotManager;
import javax.annotation.Nullable;
@@ -52,6 +53,7 @@ public class ManifestsReader {
@Nullable private Integer specifiedBucket = null;
@Nullable private Integer specifiedLevel = null;
@Nullable private PartitionPredicate partitionFilter = null;
+ @Nullable private BiFilter<Integer, Integer> manifestLevelFilter = null;
public ManifestsReader(
RowType partitionType,
@@ -79,6 +81,11 @@ public class ManifestsReader {
return this;
}
+ public ManifestsReader withManifestLevelFilter(BiFilter<Integer, Integer> manifestLevelFilter) {
+ this.manifestLevelFilter = manifestLevelFilter;
+ return this;
+ }
+
public ManifestsReader withPartitionFilter(Predicate predicate) {
this.partitionFilter = PartitionPredicate.fromPredicate(partitionType, predicate);
return this;
@@ -160,6 +167,9 @@ public class ManifestsReader {
&& (specifiedLevel < minLevel || specifiedLevel > maxLevel)) {
return false;
}
+ if (manifestLevelFilter != null && !manifestLevelFilter.test(minLevel, maxLevel)) {
+ return false;
+ }
}
if (partitionFilter == null) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
index 82112f5fd4..91505ffcf9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
@@ -34,6 +34,7 @@ import org.apache.paimon.table.source.ScanMode;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.SplitGenerator;
import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.utils.BiFilter;
import org.apache.paimon.utils.ChangelogManager;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.Filter;
@@ -87,6 +88,8 @@ public interface SnapshotReader {
SnapshotReader withLevelFilter(Filter<Integer> levelFilter);
+ SnapshotReader withManifestLevelFilter(BiFilter<Integer, Integer> manifestLevelFilter);
+
SnapshotReader enableValueFilter();
SnapshotReader withManifestEntryFilter(Filter<ManifestEntry> filter);
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
index 81c015a9d4..7b1ca7b37e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
@@ -47,6 +47,7 @@ import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.PlanImpl;
import org.apache.paimon.table.source.ScanMode;
import org.apache.paimon.table.source.SplitGenerator;
+import org.apache.paimon.utils.BiFilter;
import org.apache.paimon.utils.ChangelogManager;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.Filter;
@@ -254,6 +255,11 @@ public class SnapshotReaderImpl implements SnapshotReader {
return this;
}
+ public SnapshotReader withManifestLevelFilter(BiFilter<Integer, Integer> manifestLevelFilter) {
+ scan.withManifestLevelFilter(manifestLevelFilter);
+ return this;
+ }
+
@Override
public SnapshotReader enableValueFilter() {
scan.enableValueFilter();
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 ab3b13a5ab..f71eba944d 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
@@ -59,6 +59,7 @@ import org.apache.paimon.table.source.snapshot.StartingContext;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.BiFilter;
import org.apache.paimon.utils.BranchManager;
import org.apache.paimon.utils.ChangelogManager;
import org.apache.paimon.utils.FileStorePathFactory;
@@ -359,6 +360,13 @@ public class AuditLogTable implements DataTable, ReadonlyTable {
return this;
}
+ @Override
+ public SnapshotReader withManifestLevelFilter(
+ BiFilter<Integer, Integer> manifestLevelFilter) {
+ wrapped.withManifestLevelFilter(manifestLevelFilter);
+ return this;
+ }
+
@Override
public SnapshotReader enableValueFilter() {
wrapped.enableValueFilter();
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/cluster/HistoryPartitionClusterTest.java b/paimon-core/src/test/java/org/apache/paimon/append/cluster/HistoryPartitionClusterTest.java
new file mode 100644
index 0000000000..4692b50fbb
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/append/cluster/HistoryPartitionClusterTest.java
@@ -0,0 +1,190 @@
+/*
+ * 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.append.cluster;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.mergetree.LevelSortedRun;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.assertj.core.util.Lists;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.append.cluster.IncrementalClusterManagerTest.writeOnce;
+import static org.apache.paimon.append.cluster.IncrementalClusterStrategyTest.createFile;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link HistoryPartitionCluster}. */
+public class HistoryPartitionClusterTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ public void testFindLowLevelPartitions() throws Exception {
+ FileStoreTable table = createTable(Collections.emptyMap(), Collections.emptyList());
+ long now = LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+ Map<BinaryRow, List<DataFileMeta>> partitionFiles = new HashMap<>();
+
+ // specified partition, has low-level files
+ BinaryRow partition1 = BinaryRow.singleColumn(1);
+ PartitionEntry partitionEntry1 = new PartitionEntry(partition1, 0, 0, 0, now);
+ partitionFiles.put(
+ partition1, Lists.newArrayList(createFile(100, 1, 3), createFile(100, 1, 5)));
+ // has no low-level files
+ BinaryRow partition2 = BinaryRow.singleColumn(2);
+ PartitionEntry partitionEntry2 = new PartitionEntry(partition2, 0, 0, 0, now);
+ partitionFiles.put(partition2, Lists.newArrayList(createFile(100, 1, 0)));
+ // has low-level files
+ BinaryRow partition3 = BinaryRow.singleColumn(3);
+ PartitionEntry partitionEntry3 = new PartitionEntry(partition3, 0, 0, 0, now);
+ partitionFiles.put(
+ partition3, Lists.newArrayList(createFile(100, 1, 0), createFile(100, 1, 2)));
+ // has no low-level files
+ BinaryRow partition4 = BinaryRow.singleColumn(4);
+ PartitionEntry partitionEntry4 = new PartitionEntry(partition3, 0, 0, 0, now);
+ partitionFiles.put(partition4, Lists.newArrayList(createFile(100, 1, 0)));
+
+ IncrementalClusterManager incrementalClusterManager =
+ new IncrementalClusterManager(
+ table,
+ PartitionPredicate.fromMultiple(
+ RowType.of(DataTypes.INT()), Lists.newArrayList(partition1)));
+ HistoryPartitionCluster historyPartitionCluster =
+ incrementalClusterManager.historyPartitionCluster();
+ Set<BinaryRow> selectedPartitions =
+ historyPartitionCluster.findLowLevelPartitions(
+ Lists.newArrayList(
+ partitionEntry1,
+ partitionEntry2,
+ partitionEntry3,
+ partitionEntry4)
+ .stream()
+ .map(PartitionEntry::partition)
+ .collect(Collectors.toList()),
+ partitionFiles);
+
+ assertThat(selectedPartitions).contains(partition2);
+ }
+
+ @Test
+ public void testHistoryPartitionAutoClustering() throws Exception {
+ FileStoreTable table = createTable(Collections.emptyMap(), Collections.singletonList("f2"));
+ writeOnce(
+ table,
+ GenericRow.of(
+ 1, 1, BinaryString.fromString("pt1"), BinaryString.fromString("test")));
+ writeOnce(
+ table,
+ GenericRow.of(
+ 1, 1, BinaryString.fromString("pt2"), BinaryString.fromString("test")));
+
+ Thread.sleep(2000);
+ writeOnce(
+ table,
+ GenericRow.of(
+ 1, 1, BinaryString.fromString("pt3"), BinaryString.fromString("test")));
+ writeOnce(
+ table,
+ GenericRow.of(
+ 1, 1, BinaryString.fromString("pt4"), BinaryString.fromString("test")));
+
+ // test specify history partition and enable history partition auto clustering
+ HistoryPartitionCluster historyPartitionCluster =
+ new IncrementalClusterManager(
+ table,
+ PartitionPredicate.fromMultiple(
+ RowType.of(DataTypes.INT()),
+ Lists.newArrayList(BinaryRow.singleColumn("pt1"))))
+ .historyPartitionCluster();
+ Map<BinaryRow, List<LevelSortedRun>> partitionLevels =
+ historyPartitionCluster.constructLevelsForHistoryPartitions();
+ assertThat(partitionLevels.size()).isEqualTo(1);
+ assertThat(partitionLevels.get(BinaryRow.singleColumn("pt2"))).isNotEmpty();
+
+ // test specify non-history partition and enable history partition auto clustering
+ historyPartitionCluster =
+ new IncrementalClusterManager(
+ table,
+ PartitionPredicate.fromMultiple(
+ RowType.of(DataTypes.INT()),
+ Lists.newArrayList(BinaryRow.singleColumn("pt3"))))
+ .historyPartitionCluster();
+ partitionLevels = historyPartitionCluster.constructLevelsForHistoryPartitions();
+ assertThat(partitionLevels.size()).isEqualTo(1);
+ assertThat(partitionLevels.get(BinaryRow.singleColumn("pt1"))).isNotEmpty();
+
+ // test not specify partition and disable history partition auto clustering
+ historyPartitionCluster = new IncrementalClusterManager(table).historyPartitionCluster();
+ partitionLevels = historyPartitionCluster.constructLevelsForHistoryPartitions();
+ assertThat(partitionLevels.isEmpty()).isTrue();
+ }
+
+ protected FileStoreTable createTable(
+ Map<String, String> customOptions, List<String> partitionKeys) throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "-1");
+ options.put(CoreOptions.CLUSTERING_COLUMNS.key(), "f0,f1");
+ options.put(CoreOptions.CLUSTERING_INCREMENTAL.key(), "true");
+ options.put(CoreOptions.CLUSTERING_HISTORY_PARTITION_IDLE_TIME.key(), "2s");
+ options.put(CoreOptions.CLUSTERING_HISTORY_PARTITION_LIMIT.key(), "1");
+ options.putAll(customOptions);
+
+ Schema schema =
+ new Schema(
+ RowType.of(
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.STRING(),
+ DataTypes.STRING())
+ .getFields(),
+ partitionKeys,
+ Collections.emptyList(),
+ options,
+ "");
+
+ SchemaManager schemaManager =
+ new SchemaManager(LocalFileIO.create(), new Path(tempDir.toString()));
+ return FileStoreTableFactory.create(
+ LocalFileIO.create(),
+ new Path(tempDir.toString()),
+ schemaManager.createTable(schema));
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
index 8b37aff0b8..37cc006673 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
@@ -19,18 +19,27 @@
package org.apache.paimon.append.cluster;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.compact.CompactUnit;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.mergetree.LevelSortedRun;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.assertj.core.util.Lists;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -143,7 +152,7 @@ public class IncrementalClusterManagerTest {
// Test upgrading to level 3
int outputLevel = 3;
List<DataFileMeta> upgradedFiles =
- incrementalClusterManager.upgrade(filesAfterCluster, outputLevel);
+ IncrementalClusterManager.upgrade(filesAfterCluster, outputLevel);
// Verify the results
assertThat(upgradedFiles).hasSize(3);
@@ -154,7 +163,67 @@ public class IncrementalClusterManagerTest {
}
}
- private FileStoreTable createTable(
+ @Test
+ public void testHistoryPartitionAutoClustering() throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.CLUSTERING_HISTORY_PARTITION_IDLE_TIME.key(), "2s");
+ options.put(CoreOptions.CLUSTERING_HISTORY_PARTITION_LIMIT.key(), "1");
+
+ FileStoreTable table = createTable(options, Collections.singletonList("f2"));
+ writeOnce(
+ table,
+ GenericRow.of(
+ 1, 1, BinaryString.fromString("pt1"), BinaryString.fromString("test")));
+ writeOnce(
+ table,
+ GenericRow.of(
+ 1, 1, BinaryString.fromString("pt2"), BinaryString.fromString("test")));
+
+ Thread.sleep(2000);
+ writeOnce(
+ table,
+ GenericRow.of(
+ 1, 1, BinaryString.fromString("pt3"), BinaryString.fromString("test")));
+ writeOnce(
+ table,
+ GenericRow.of(
+ 1, 1, BinaryString.fromString("pt4"), BinaryString.fromString("test")));
+
+ // test specify partition and enable history partition auto clustering
+ IncrementalClusterManager incrementalClusterManager =
+ new IncrementalClusterManager(
+ table,
+ PartitionPredicate.fromMultiple(
+ RowType.of(DataTypes.INT()),
+ Lists.newArrayList(BinaryRow.singleColumn("pt3"))));
+ Map<BinaryRow, CompactUnit> partitionLevels =
+ incrementalClusterManager.prepareForCluster(true);
+ assertThat(partitionLevels.size()).isEqualTo(2);
+ assertThat(partitionLevels.get(BinaryRow.singleColumn("pt1"))).isNotNull();
+ assertThat(partitionLevels.get(BinaryRow.singleColumn("pt3"))).isNotNull();
+
+ // test don't specify partition and enable history partition auto clustering
+ incrementalClusterManager = new IncrementalClusterManager(table);
+ partitionLevels = incrementalClusterManager.prepareForCluster(true);
+ assertThat(partitionLevels.size()).isEqualTo(4);
+
+ // test specify partition and disable history partition auto clustering
+ SchemaChange schemaChange =
+ SchemaChange.removeOption(CoreOptions.CLUSTERING_HISTORY_PARTITION_IDLE_TIME.key());
+ incrementalClusterManager =
+ new IncrementalClusterManager(
+ table.copy(
+ table.schemaManager()
+ .commitChanges(Collections.singletonList(schemaChange))),
+ PartitionPredicate.fromMultiple(
+ RowType.of(DataTypes.INT()),
+ Lists.newArrayList(BinaryRow.singleColumn("pt3"))));
+ partitionLevels = incrementalClusterManager.prepareForCluster(true);
+ assertThat(partitionLevels.size()).isEqualTo(1);
+ assertThat(partitionLevels.get(BinaryRow.singleColumn("pt3"))).isNotNull();
+ }
+
+ protected FileStoreTable createTable(
Map<String, String> customOptions, List<String> partitionKeys) throws Exception {
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.BUCKET.key(), "-1");
@@ -183,7 +252,20 @@ public class IncrementalClusterManagerTest {
schemaManager.createTable(schema));
}
- private static DataFileMeta createFile(long size, long schemaId, int level) {
+ protected static void writeOnce(FileStoreTable table, GenericRow... rows) {
+ String commitUser = "test_user";
+ try (BatchTableWrite write = table.newWrite(commitUser);
+ BatchTableCommit commit = table.newCommit(commitUser)) {
+ for (GenericRow row : rows) {
+ write.write(row);
+ }
+ commit.commit(write.prepareCommit());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected static DataFileMeta createFile(long size, long schemaId, int level) {
return DataFileMeta.create(
"",
size,
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterStrategyTest.java b/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterStrategyTest.java
index 1061c50c9e..a00f2442ae 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterStrategyTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterStrategyTest.java
@@ -199,7 +199,7 @@ public class IncrementalClusterStrategyTest {
SchemaChange.setOption(CoreOptions.CLUSTERING_COLUMNS.key(), "f2,f3"));
}
- private static DataFileMeta createFile(long size, long schemaId, int level) {
+ protected static DataFileMeta createFile(long size, long schemaId, int level) {
return DataFileMeta.create(
"",
size,
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/IncrementalClusterActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/IncrementalClusterActionITCase.java
index 7629ebf81d..48e0f1fa30 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/IncrementalClusterActionITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/IncrementalClusterActionITCase.java
@@ -18,10 +18,12 @@
package org.apache.paimon.flink.action;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.BatchTableCommit;
@@ -41,8 +43,11 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
@@ -361,6 +366,158 @@ public class IncrementalClusterActionITCase extends ActionITCaseBase {
}
}
+ @Test
+ public void testClusterHistoryPartition() throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.CLUSTERING_HISTORY_PARTITION_IDLE_TIME.key(), "3s");
+ FileStoreTable table = createTable("pt", 1, options);
+
+ BinaryString randomStr = BinaryString.fromString(randomString(150));
+ List<CommitMessage> messages = new ArrayList<>();
+
+ // first write
+ List<String> expected1 = new ArrayList<>();
+ for (int pt = 0; pt < 4; pt++) {
+ for (int i = 0; i < 3; i++) {
+ for (int j = 0; j < 3; j++) {
+ messages.addAll(write(GenericRow.of(i, j, randomStr, pt)));
+ expected1.add(String.format("+I[%s, %s, %s]", i, j, pt));
+ }
+ }
+ }
+ commit(messages);
+ ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {0, 1, 3});
+ List<String> result1 =
+ getResult(
+ readBuilder.newRead(),
+ readBuilder.newScan().plan().splits(),
+ readBuilder.readType());
+ assertThat(result1).containsExactlyElementsOf(expected1);
+
+ // first cluster, files in four partitions will be in top level
+ runAction(Collections.emptyList());
+ checkSnapshot(table);
+ List<Split> splits = readBuilder.newScan().plan().splits();
+ assertThat(splits.size()).isEqualTo(4);
+ assertThat(((DataSplit) splits.get(0)).dataFiles().size()).isEqualTo(1);
+ assertThat(((DataSplit) splits.get(0)).dataFiles().get(0).level()).isEqualTo(5);
+ List<String> result2 = getResult(readBuilder.newRead(), splits, readBuilder.readType());
+ List<String> expected2 = new ArrayList<>();
+ for (int pt = 0; pt < 4; pt++) {
+ expected2.add(String.format("+I[0, 0, %s]", pt));
+ expected2.add(String.format("+I[0, 1, %s]", pt));
+ expected2.add(String.format("+I[1, 0, %s]", pt));
+ expected2.add(String.format("+I[1, 1, %s]", pt));
+ expected2.add(String.format("+I[0, 2, %s]", pt));
+ expected2.add(String.format("+I[1, 2, %s]", pt));
+ expected2.add(String.format("+I[2, 0, %s]", pt));
+ expected2.add(String.format("+I[2, 1, %s]", pt));
+ expected2.add(String.format("+I[2, 2, %s]", pt));
+ }
+ assertThat(result2).containsExactlyElementsOf(expected2);
+
+ // second write
+ messages.clear();
+ for (int pt = 0; pt < 4; pt++) {
+ messages.addAll(
+ write(
+ GenericRow.of(0, 3, null, pt),
+ GenericRow.of(1, 3, null, pt),
+ GenericRow.of(2, 3, null, pt)));
+ messages.addAll(
+ write(
+ GenericRow.of(3, 0, null, pt),
+ GenericRow.of(3, 1, null, pt),
+ GenericRow.of(3, 2, null, pt),
+ GenericRow.of(3, 3, null, pt)));
+ // pt-0, pt-1 will be history partition
+ if (pt == 1) {
+ Thread.sleep(3000);
+ }
+ }
+ commit(messages);
+
+ List<String> result3 =
+ getResult(
+ readBuilder.newRead(),
+ readBuilder.newScan().plan().splits(),
+ readBuilder.readType());
+ List<String> expected3 = new ArrayList<>();
+ for (int pt = 0; pt < 4; pt++) {
+ expected3.addAll(expected2.subList(9 * pt, 9 * pt + 9));
+ expected3.add(String.format("+I[0, 3, %s]", pt));
+ expected3.add(String.format("+I[1, 3, %s]", pt));
+ expected3.add(String.format("+I[2, 3, %s]", pt));
+ expected3.add(String.format("+I[3, 0, %s]", pt));
+ expected3.add(String.format("+I[3, 1, %s]", pt));
+ expected3.add(String.format("+I[3, 2, %s]", pt));
+ expected3.add(String.format("+I[3, 3, %s]", pt));
+ }
+ assertThat(result3).containsExactlyElementsOf(expected3);
+
+ // second cluster
+ runAction(Lists.newArrayList("--partition", "pt=3"));
+ checkSnapshot(table);
+ splits = readBuilder.newScan().plan().splits();
+ List<String> result4 = getResult(readBuilder.newRead(), splits, readBuilder.readType());
+ List<String> expected4 = new ArrayList<>();
+ assertThat(splits.size()).isEqualTo(4);
+ // for pt-0 and pt-1: history partition, full clustering, all files will be
+ // picked for clustering, outputLevel is 5.
+ for (int pt = 0; pt <= 1; pt++) {
+ expected4.add(String.format("+I[0, 0, %s]", pt));
+ expected4.add(String.format("+I[0, 1, %s]", pt));
+ expected4.add(String.format("+I[1, 0, %s]", pt));
+ expected4.add(String.format("+I[1, 1, %s]", pt));
+ expected4.add(String.format("+I[0, 2, %s]", pt));
+ expected4.add(String.format("+I[0, 3, %s]", pt));
+ expected4.add(String.format("+I[1, 2, %s]", pt));
+ expected4.add(String.format("+I[1, 3, %s]", pt));
+ expected4.add(String.format("+I[2, 0, %s]", pt));
+ expected4.add(String.format("+I[2, 1, %s]", pt));
+ expected4.add(String.format("+I[3, 0, %s]", pt));
+ expected4.add(String.format("+I[3, 1, %s]", pt));
+ expected4.add(String.format("+I[2, 2, %s]", pt));
+ expected4.add(String.format("+I[2, 3, %s]", pt));
+ expected4.add(String.format("+I[3, 2, %s]", pt));
+ expected4.add(String.format("+I[3, 3, %s]", pt));
+ // the table has enabled 'scan.plan-sort-partition', so the splits has been sorted by
+ // partition
+ assertThat(((DataSplit) splits.get(pt)).dataFiles().size()).isEqualTo(1);
+ assertThat(((DataSplit) splits.get(pt)).dataFiles().get(0).level()).isEqualTo(5);
+ }
+ // for pt-2, non history partition, nor specified partition, nothing happened
+ expected4.addAll(expected3.subList(32, 48));
+ assertThat(((DataSplit) splits.get(2)).dataFiles().size()).isEqualTo(3);
+ // for pt-3: minor clustering, only file in level-0 will be picked for clustering,
+ // outputLevel is 4
+ expected4.add("+I[0, 0, 3]");
+ expected4.add("+I[0, 1, 3]");
+ expected4.add("+I[1, 0, 3]");
+ expected4.add("+I[1, 1, 3]");
+ expected4.add("+I[0, 2, 3]");
+ expected4.add("+I[1, 2, 3]");
+ expected4.add("+I[2, 0, 3]");
+ expected4.add("+I[2, 1, 3]");
+ expected4.add("+I[2, 2, 3]");
+ expected4.add("+I[0, 3, 3]");
+ expected4.add("+I[1, 3, 3]");
+ expected4.add("+I[3, 0, 3]");
+ expected4.add("+I[3, 1, 3]");
+ expected4.add("+I[2, 3, 3]");
+ expected4.add("+I[3, 2, 3]");
+ expected4.add("+I[3, 3, 3]");
+ assertThat(((DataSplit) splits.get(3)).dataFiles().size()).isEqualTo(2);
+ assertThat(
+ ((DataSplit) splits.get(3))
+ .dataFiles().stream()
+ .map(DataFileMeta::level)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(4, 5);
+
+ assertThat(result4).containsExactlyElementsOf(expected4);
+ }
+
@Test
public void testClusterOnEmptyData() throws Exception {
createTable("pt", 1);
@@ -410,8 +567,14 @@ public class IncrementalClusterActionITCase extends ActionITCaseBase {
protected FileStoreTable createTable(String partitionKeys, int sinkParallelism)
throws Exception {
+ return createTable(partitionKeys, sinkParallelism, Collections.emptyMap());
+ }
+
+ protected FileStoreTable createTable(
+ String partitionKeys, int sinkParallelism, Map<String, String> options)
+ throws Exception {
catalog.createDatabase(database, true);
- catalog.createTable(identifier(), schema(partitionKeys, sinkParallelism), true);
+ catalog.createTable(identifier(), schema(partitionKeys, sinkParallelism, options), true);
return (FileStoreTable) catalog.getTable(identifier());
}
@@ -440,6 +603,11 @@ public class IncrementalClusterActionITCase extends ActionITCaseBase {
}
private static Schema schema(String partitionKeys, int sinkParallelism) {
+ return schema(partitionKeys, sinkParallelism, Collections.emptyMap());
+ }
+
+ private static Schema schema(
+ String partitionKeys, int sinkParallelism, Map<String, String> options) {
Schema.Builder schemaBuilder = Schema.newBuilder();
schemaBuilder.column("a", DataTypes.INT());
schemaBuilder.column("b", DataTypes.INT());
@@ -454,6 +622,9 @@ public class IncrementalClusterActionITCase extends ActionITCaseBase {
schemaBuilder.option("clustering.incremental", "true");
schemaBuilder.option("scan.parallelism", "1");
schemaBuilder.option("sink.parallelism", String.valueOf(sinkParallelism));
+ for (String key : options.keySet()) {
+ schemaBuilder.option(key, options.get(key));
+ }
if (!StringUtils.isNullOrWhitespaceOnly(partitionKeys)) {
schemaBuilder.partitionKeys(partitionKeys);
}
(paimon) 04/09: [typo] fix typo in distinct (#6475)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.3
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit b856f283639ebd307bacbc3475eb8c5e484f799a
Author: tclxjunjie2-zhao <ju...@ly.com>
AuthorDate: Mon Oct 27 20:31:43 2025 +0800
[typo] fix typo in distinct (#6475)
---
.../org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
index d9e706f6e8..6fbd305085 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
@@ -86,7 +86,7 @@ public class FieldCollectAgg extends FieldAggregator {
public Object aggReversed(Object accumulator, Object inputField) {
// we don't need to actually do the reverse here for this agg
// because accumulator has been distinct, just let accumulator be accumulator will speed up
- // dinstinct process
+ // distinct process
return agg(accumulator, inputField);
}
(paimon) 01/09: [doc] fix MarkPartitionDoneProcedure doc. (#6473)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.3
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 3c213dc7eef852fd61d41c3d03ba52923b1813ba
Author: Jiajia Li <pl...@alibaba-inc.com>
AuthorDate: Mon Oct 27 15:26:21 2025 +0800
[doc] fix MarkPartitionDoneProcedure doc. (#6473)
---
docs/content/spark/procedures.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/docs/content/spark/procedures.md b/docs/content/spark/procedures.md
index f3cff70160..84da6821a4 100644
--- a/docs/content/spark/procedures.md
+++ b/docs/content/spark/procedures.md
@@ -370,9 +370,9 @@ This section introduce all available spark procedures about paimon.
</td>
<td>
-- mark single partition done<br/>
- CALL sys.mark_partition_done(table => 'default.T', parititions => 'day=2024-07-01')<br/><br/>
+ CALL sys.mark_partition_done(table => 'default.T', partitions => 'day=2024-07-01')<br/><br/>
-- mark multiple partitions done<br/>
- CALL sys.mark_partition_done(table => 'default.T', parititions => 'day=2024-07-01;day=2024-07-02')
+ CALL sys.mark_partition_done(table => 'default.T', partitions => 'day=2024-07-01;day=2024-07-02')
</td>
</tr>
<tr>
(paimon) 02/09: [doc] add view doc (#6469)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.3
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit d3a7451c162855df08db34d8d53a6f9d9081678a
Author: jerry <li...@alibaba-inc.com>
AuthorDate: Mon Oct 27 15:39:23 2025 +0800
[doc] add view doc (#6469)
---
docs/content/concepts/spec/_index.md | 2 +-
docs/content/concepts/views.md | 107 +++++++++++++++++++++++++++++++++++
2 files changed, 108 insertions(+), 1 deletion(-)
diff --git a/docs/content/concepts/spec/_index.md b/docs/content/concepts/spec/_index.md
index a1a1251c74..3cfdc14040 100644
--- a/docs/content/concepts/spec/_index.md
+++ b/docs/content/concepts/spec/_index.md
@@ -1,7 +1,7 @@
---
title: Specification
bookCollapseSection: true
-weight: 10
+weight: 11
---
<!--
Licensed to the Apache Software Foundation (ASF) under one
diff --git a/docs/content/concepts/views.md b/docs/content/concepts/views.md
new file mode 100644
index 0000000000..c89b5560f1
--- /dev/null
+++ b/docs/content/concepts/views.md
@@ -0,0 +1,107 @@
+---
+title: "Views"
+weight: 10
+type: docs
+aliases:
+- /concepts/views.html
+---
+<!--
+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.
+-->
+
+# Views
+A view is a logical table that encapsulates business logic and domain-specific semantics.
+While most compute engines support views natively, each engine stores view metadata in proprietary formats, creating interoperability challenges across different platforms.
+Paimon views abstracting engine-specific query dialects and establishing unified metadata standards.
+View metadata could enable centralized view management that facilitates cross-engine sharing and reduces maintenance complexity in heterogeneous computing environments.
+
+## Catalog support
+
+View metadata is persisted only when the catalog implementation supports it:
+
+- **Hive metastore catalog** – view metadata is stored together with table metadata inside the
+ metastore warehouse.
+- **REST catalog** – view metadata is kept in the REST backend and exposed through the catalog API.
+
+File-system catalogs do not currently support views because they lack persistent metadata storage.
+
+
+### Representation structure
+
+| Field | Type | Description |
+|-----------|------|-------------|
+| `query` | `string` | Canonical SQL `SELECT` statement that defines the view. |
+| `dialect` | `string` | SQL dialect identifier (for example, `spark` or `flink`). |
+
+Multiple representations can be stored for the same version so that different engines can access the
+view using their native dialect.
+
+## Operations
+
+### Create or replace view
+
+Use `CREATE VIEW` or `CREATE OR REPLACE VIEW` to register a view. Paimon assigns a UUID, writes the
+first metadata file, and records version `1`.
+
+```sql
+CREATE VIEW sales_view AS
+SELECT region, SUM(amount) AS total_amount
+FROM sales
+GROUP BY region;
+```
+
+### Alter view dialect via procedure
+
+Paimon provides the `sys.alter_view_dialect` procedure so that engines can manage multiple SQL
+representations for the same view version.
+
+#### Flink example
+
+```sql
+-- Add a Flink dialect
+CALL [catalog.]sys.alter_view_dialect('view_identifier', 'add', 'flink', 'SELECT ...');
+
+-- Update the stored Flink dialect
+CALL [catalog.]sys.alter_view_dialect('view_identifier', 'update', 'flink', 'SELECT ...');
+
+-- Drop the Flink dialect representation
+CALL [catalog.]sys.alter_view_dialect('view_identifier', 'drop', 'flink');
+```
+
+#### Spark example
+
+```sql
+-- Add a Spark dialect
+CALL sys.alter_view_dialect('view_identifier', 'add', 'spark', 'SELECT ...');
+
+-- Update the Spark dialect
+CALL sys.alter_view_dialect('view_identifier', 'update', 'spark', 'SELECT ...');
+
+-- Drop the Spark dialect
+CALL sys.alter_view_dialect('view_identifier', 'drop', 'spark');
+```
+
+### Drop view
+
+`DROP VIEW view_name;`
+
+## See also
+
+- [Spark SQL DDL – Views]({{< ref "spark/sql-ddl#view" >}})
+- [REST Catalog Overview]({{< ref "concepts/rest/overview" >}})
+- [REST Catalog View API]({{< ref "concepts/rest/rest-api/#tag/view" >}})
(paimon) 07/09: [core] Refactor HistoryPartitionCluster to load less history partitions
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.3
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 718f124d37189659dbebd3b7a3c4cd00d66341a3
Author: JingsongLi <ji...@gmail.com>
AuthorDate: Tue Oct 28 21:48:59 2025 +0800
[core] Refactor HistoryPartitionCluster to load less history partitions
---
.../append/cluster/HistoryPartitionCluster.java | 139 ++++++++++-----------
.../append/cluster/IncrementalClusterManager.java | 91 +++++---------
.../paimon/operation/AbstractFileStoreScan.java | 4 +-
.../org/apache/paimon/operation/FileStoreScan.java | 2 +-
.../apache/paimon/operation/ManifestsReader.java | 8 +-
.../table/source/snapshot/SnapshotReader.java | 2 +-
.../table/source/snapshot/SnapshotReaderImpl.java | 5 +-
.../apache/paimon/table/system/AuditLogTable.java | 5 +-
.../cluster/HistoryPartitionClusterTest.java | 57 +--------
.../cluster/IncrementalClusterManagerTest.java | 2 +-
10 files changed, 110 insertions(+), 205 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/cluster/HistoryPartitionCluster.java b/paimon-core/src/main/java/org/apache/paimon/append/cluster/HistoryPartitionCluster.java
index cccecdd2be..83f92aa9c7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/cluster/HistoryPartitionCluster.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/cluster/HistoryPartitionCluster.java
@@ -27,7 +27,6 @@ import org.apache.paimon.mergetree.LevelSortedRun;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.DataSplit;
-import org.apache.paimon.utils.BiFilter;
import org.apache.paimon.utils.InternalRowPartitionComputer;
import org.slf4j.Logger;
@@ -39,14 +38,11 @@ import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.paimon.append.cluster.IncrementalClusterManager.constructPartitionLevels;
@@ -56,70 +52,86 @@ import static org.apache.paimon.append.cluster.IncrementalClusterManager.logForP
public class HistoryPartitionCluster {
private static final Logger LOG = LoggerFactory.getLogger(HistoryPartitionCluster.class);
- private final InternalRowPartitionComputer partitionComputer;
private final FileStoreTable table;
private final IncrementalClusterStrategy incrementalClusterStrategy;
- private final int maxLevel;
-
+ private final InternalRowPartitionComputer partitionComputer;
+ private final PartitionPredicate specifiedPartitions;
+ private final Duration historyPartitionIdleTime;
private final int historyPartitionLimit;
- @Nullable private final PartitionPredicate specifiedPartitions;
- @Nullable private final Duration historyPartitionIdleTime;
- @Nullable private final BiFilter<Integer, Integer> partitionLevelFilter;
+ private final int maxLevel;
public HistoryPartitionCluster(
FileStoreTable table,
IncrementalClusterStrategy incrementalClusterStrategy,
InternalRowPartitionComputer partitionComputer,
- int maxLevel,
- int historyPartitionLimit,
- @Nullable PartitionPredicate specifiedPartitions,
- @Nullable Duration historyPartitionIdleTime) {
+ PartitionPredicate specifiedPartitions,
+ Duration historyPartitionIdleTime,
+ int historyPartitionLimit) {
this.table = table;
this.incrementalClusterStrategy = incrementalClusterStrategy;
this.partitionComputer = partitionComputer;
- this.maxLevel = maxLevel;
- this.historyPartitionLimit = historyPartitionLimit;
this.specifiedPartitions = specifiedPartitions;
this.historyPartitionIdleTime = historyPartitionIdleTime;
- // (maxLevel + 1) / 2 is used to calculate the ceiling of maxLevel divided by 2
- this.partitionLevelFilter =
- (partitionMinLevel, partitionMaxLevel) -> partitionMinLevel < (maxLevel + 1) / 2;
+ this.historyPartitionLimit = historyPartitionLimit;
+ this.maxLevel = table.coreOptions().numLevels() - 1;
+ }
+
+ @Nullable
+ public static HistoryPartitionCluster create(
+ FileStoreTable table,
+ IncrementalClusterStrategy incrementalClusterStrategy,
+ InternalRowPartitionComputer partitionComputer,
+ @Nullable PartitionPredicate specifiedPartitions) {
+ if (table.schema().partitionKeys().isEmpty()) {
+ return null;
+ }
+ if (specifiedPartitions == null) {
+ return null;
+ }
+
+ Duration idleTime = table.coreOptions().clusteringHistoryPartitionIdleTime();
+ if (idleTime == null) {
+ return null;
+ }
+
+ int limit = table.coreOptions().clusteringHistoryPartitionLimit();
+ return new HistoryPartitionCluster(
+ table,
+ incrementalClusterStrategy,
+ partitionComputer,
+ specifiedPartitions,
+ idleTime,
+ limit);
}
- public Map<BinaryRow, Optional<CompactUnit>> pickForHistoryPartitions() {
+ public Map<BinaryRow, CompactUnit> pickForHistoryPartitions() {
Map<BinaryRow, List<LevelSortedRun>> partitionLevels =
constructLevelsForHistoryPartitions();
logForPartitionLevel(partitionLevels, partitionComputer);
- return partitionLevels.entrySet().stream()
- .collect(
- Collectors.toMap(
- Map.Entry::getKey,
- entry ->
- incrementalClusterStrategy.pick(
- maxLevel, entry.getValue(), true)));
+ Map<BinaryRow, CompactUnit> units = new HashMap<>();
+ partitionLevels.forEach(
+ (k, v) -> {
+ Optional<CompactUnit> pick =
+ incrementalClusterStrategy.pick(maxLevel + 1, v, true);
+ pick.ifPresent(compactUnit -> units.put(k, compactUnit));
+ });
+ return units;
}
@VisibleForTesting
public Map<BinaryRow, List<LevelSortedRun>> constructLevelsForHistoryPartitions() {
- if (specifiedPartitions == null
- || historyPartitionIdleTime == null
- || historyPartitionLimit <= 0) {
- return Collections.emptyMap();
- }
-
long historyMilli =
LocalDateTime.now()
.minus(historyPartitionIdleTime)
.atZone(ZoneId.systemDefault())
.toInstant()
.toEpochMilli();
- // read partitionEntries filter by partitionLevelFilter historyPartitionIdleTime
- // sort partitionEntries by lastFileCreation time, and we will pick the oldest N partitions
+
List<BinaryRow> historyPartitions =
- table.newSnapshotReader().withManifestLevelFilter(partitionLevelFilter)
- .partitionEntries().stream()
+ table.newSnapshotReader().withLevelMinMaxFilter((min, max) -> min < maxLevel)
+ .withLevelFilter(level -> level < maxLevel).partitionEntries().stream()
.filter(entry -> entry.lastFileCreationTime() < historyMilli)
.sorted(Comparator.comparingLong(PartitionEntry::lastFileCreationTime))
.map(PartitionEntry::partition)
@@ -131,6 +143,7 @@ public class HistoryPartitionCluster {
.withPartitionFilter(historyPartitions)
.read()
.dataSplits();
+
Map<BinaryRow, List<DataFileMeta>> historyPartitionFiles = new HashMap<>();
for (DataSplit dataSplit : historyDataSplits) {
historyPartitionFiles
@@ -138,52 +151,34 @@ public class HistoryPartitionCluster {
.addAll(dataSplit.dataFiles());
}
- // find history partitions which have low-level files
- Set<BinaryRow> selectedHistoryPartitions =
- findLowLevelPartitions(historyPartitions, historyPartitionFiles);
- historyPartitionFiles =
- historyPartitionFiles.entrySet().stream()
- .filter(entry -> selectedHistoryPartitions.contains(entry.getKey()))
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-
- return historyPartitionFiles.entrySet().stream()
+ return filterPartitions(historyPartitionFiles).entrySet().stream()
.collect(
Collectors.toMap(
Map.Entry::getKey,
entry -> constructPartitionLevels(entry.getValue())));
}
- @VisibleForTesting
- protected Set<BinaryRow> findLowLevelPartitions(
- List<BinaryRow> historyPartitions, Map<BinaryRow, List<DataFileMeta>> partitionFiles) {
- Set<BinaryRow> partitions = new HashSet<>();
- // 1. the partition is not specified in specifiedPartitions
- // 2. the min file level in partition should be less than Math.ceil(maxLevel/2)
- for (BinaryRow historyPartition : historyPartitions) {
- if (specifiedPartitions != null && !specifiedPartitions.test(historyPartition)) {
- List<DataFileMeta> files =
- partitionFiles.getOrDefault(historyPartition, Collections.emptyList());
- if (!files.isEmpty()) {
- int partitionMinLevel = maxLevel + 1;
- for (DataFileMeta file : files) {
- partitionMinLevel = Math.min(partitionMinLevel, file.level());
+ private Map<BinaryRow, List<DataFileMeta>> filterPartitions(
+ Map<BinaryRow, List<DataFileMeta>> partitionFiles) {
+ Map<BinaryRow, List<DataFileMeta>> result = new HashMap<>();
+ partitionFiles.forEach(
+ (part, files) -> {
+ if (specifiedPartitions.test(part)) {
+ // already contain in specified partitions
+ return;
}
- if (partitionLevelFilter != null
- && partitionLevelFilter.test(partitionMinLevel, maxLevel)) {
- partitions.add(historyPartition);
- if (partitions.size() >= historyPartitionLimit) {
- break;
- }
+
+ if (result.size() < historyPartitionLimit) {
+ // in limit, can be picked
+ result.put(part, files);
}
- }
- }
- }
+ });
LOG.info(
"Find {} history partitions for full clustering, the history partitions are {}",
- partitions.size(),
- partitions.stream()
+ result.size(),
+ result.keySet().stream()
.map(partitionComputer::generatePartValues)
.collect(Collectors.toSet()));
- return partitions;
+ return result;
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java b/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
index 8142f9dbf0..6865b4b82e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
@@ -19,6 +19,7 @@
package org.apache.paimon.append.cluster;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.compact.CompactUnit;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.io.DataFileMeta;
@@ -52,17 +53,14 @@ import static org.apache.paimon.utils.Preconditions.checkArgument;
public class IncrementalClusterManager {
private static final Logger LOG = LoggerFactory.getLogger(IncrementalClusterManager.class);
- private final InternalRowPartitionComputer partitionComputer;
+ private final InternalRowPartitionComputer partitionComputer;
private final SnapshotReader snapshotReader;
-
private final IncrementalClusterStrategy incrementalClusterStrategy;
private final CoreOptions.OrderType clusterCurve;
private final List<String> clusterKeys;
-
- private final HistoryPartitionCluster historyPartitionCluster;
-
- private int maxLevel;
+ private final int numLevels;
+ private final @Nullable HistoryPartitionCluster historyPartitionCluster;
public IncrementalClusterManager(FileStoreTable table) {
this(table, null);
@@ -78,7 +76,7 @@ public class IncrementalClusterManager {
options.clusteringIncrementalEnabled(),
"Only support incremental clustering when '%s' is true.",
CLUSTERING_INCREMENTAL.key());
- this.maxLevel = options.numLevels();
+ this.numLevels = options.numLevels();
this.partitionComputer =
new InternalRowPartitionComputer(
table.coreOptions().partitionDefaultName(),
@@ -96,16 +94,9 @@ public class IncrementalClusterManager {
options.numSortedRunCompactionTrigger());
this.clusterCurve = options.clusteringStrategy(options.clusteringColumns().size());
this.clusterKeys = options.clusteringColumns();
-
this.historyPartitionCluster =
- new HistoryPartitionCluster(
- table,
- incrementalClusterStrategy,
- partitionComputer,
- maxLevel,
- options.clusteringHistoryPartitionLimit(),
- specifiedPartitions,
- options.clusteringHistoryPartitionIdleTime());
+ HistoryPartitionCluster.create(
+ table, incrementalClusterStrategy, partitionComputer, specifiedPartitions);
}
public Map<BinaryRow, CompactUnit> prepareForCluster(boolean fullCompaction) {
@@ -114,30 +105,20 @@ public class IncrementalClusterManager {
logForPartitionLevel(partitionLevels, partitionComputer);
// 2. pick files to be clustered for each partition
- Map<BinaryRow, Optional<CompactUnit>> units =
- partitionLevels.entrySet().stream()
- .collect(
- Collectors.toMap(
- Map.Entry::getKey,
- entry ->
- incrementalClusterStrategy.pick(
- maxLevel,
- entry.getValue(),
- fullCompaction)));
-
- Map<BinaryRow, Optional<CompactUnit>> historyUnits =
- historyPartitionCluster.pickForHistoryPartitions();
- units.putAll(historyUnits);
+ Map<BinaryRow, CompactUnit> units = new HashMap<>();
+ partitionLevels.forEach(
+ (k, v) -> {
+ Optional<CompactUnit> pick =
+ incrementalClusterStrategy.pick(numLevels, v, fullCompaction);
+ pick.ifPresent(compactUnit -> units.put(k, compactUnit));
+ });
+
+ if (historyPartitionCluster != null) {
+ units.putAll(historyPartitionCluster.pickForHistoryPartitions());
+ }
- // 3. filter out empty units
- Map<BinaryRow, CompactUnit> filteredUnits =
- units.entrySet().stream()
- .filter(entry -> entry.getValue().isPresent())
- .collect(
- Collectors.toMap(
- Map.Entry::getKey, entry -> entry.getValue().get()));
if (LOG.isDebugEnabled()) {
- filteredUnits.forEach(
+ units.forEach(
(partition, compactUnit) -> {
String filesInfo =
compactUnit.files().stream()
@@ -157,24 +138,17 @@ public class IncrementalClusterManager {
filesInfo);
});
}
- return filteredUnits;
+ return units;
}
public Map<BinaryRow, List<LevelSortedRun>> constructLevels() {
List<DataSplit> dataSplits = snapshotReader.read().dataSplits();
-
- maxLevel =
- Math.max(
- maxLevel,
- dataSplits.stream()
- .flatMap(split -> split.dataFiles().stream())
- .mapToInt(DataFileMeta::level)
- .max()
- .orElse(-1)
- + 1);
- checkArgument(maxLevel > 1, "Number of levels must be at least 2.");
-
- Map<BinaryRow, List<DataFileMeta>> partitionFiles = getPartitionFiles(dataSplits);
+ Map<BinaryRow, List<DataFileMeta>> partitionFiles = new HashMap<>();
+ for (DataSplit dataSplit : dataSplits) {
+ partitionFiles
+ .computeIfAbsent(dataSplit.partition(), k -> new ArrayList<>())
+ .addAll(dataSplit.dataFiles());
+ }
return partitionFiles.entrySet().stream()
.collect(
Collectors.toMap(
@@ -208,16 +182,6 @@ public class IncrementalClusterManager {
return partitionLevels;
}
- private Map<BinaryRow, List<DataFileMeta>> getPartitionFiles(List<DataSplit> dataSplits) {
- Map<BinaryRow, List<DataFileMeta>> partitionFiles = new HashMap<>();
- for (DataSplit dataSplit : dataSplits) {
- partitionFiles
- .computeIfAbsent(dataSplit.partition(), k -> new ArrayList<>())
- .addAll(dataSplit.dataFiles());
- }
- return partitionFiles;
- }
-
public List<DataSplit> toSplits(BinaryRow partition, List<DataFileMeta> files) {
List<DataSplit> splits = new ArrayList<>();
@@ -283,7 +247,8 @@ public class IncrementalClusterManager {
return clusterKeys;
}
- public HistoryPartitionCluster historyPartitionCluster() {
+ @VisibleForTesting
+ HistoryPartitionCluster historyPartitionCluster() {
return historyPartitionCluster;
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
index 6f06514ed3..7ed77406c7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
@@ -197,8 +197,8 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
}
@Override
- public FileStoreScan withManifestLevelFilter(BiFilter<Integer, Integer> manifestLevelFilter) {
- manifestsReader.withManifestLevelFilter(manifestLevelFilter);
+ public FileStoreScan withLevelMinMaxFilter(BiFilter<Integer, Integer> minMaxFilter) {
+ manifestsReader.withLevelMinMaxFilter(minMaxFilter);
return this;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
index 354c43e019..047b1c3f5d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
@@ -75,7 +75,7 @@ public interface FileStoreScan {
FileStoreScan withLevelFilter(Filter<Integer> levelFilter);
- FileStoreScan withManifestLevelFilter(BiFilter<Integer, Integer> manifestLevelFilter);
+ FileStoreScan withLevelMinMaxFilter(BiFilter<Integer, Integer> minMaxFilter);
FileStoreScan enableValueFilter();
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
index dbf5140583..b3b89e72aa 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
@@ -53,7 +53,7 @@ public class ManifestsReader {
@Nullable private Integer specifiedBucket = null;
@Nullable private Integer specifiedLevel = null;
@Nullable private PartitionPredicate partitionFilter = null;
- @Nullable private BiFilter<Integer, Integer> manifestLevelFilter = null;
+ @Nullable private BiFilter<Integer, Integer> levelMinMaxFilter = null;
public ManifestsReader(
RowType partitionType,
@@ -81,8 +81,8 @@ public class ManifestsReader {
return this;
}
- public ManifestsReader withManifestLevelFilter(BiFilter<Integer, Integer> manifestLevelFilter) {
- this.manifestLevelFilter = manifestLevelFilter;
+ public ManifestsReader withLevelMinMaxFilter(BiFilter<Integer, Integer> minMaxFilter) {
+ this.levelMinMaxFilter = minMaxFilter;
return this;
}
@@ -167,7 +167,7 @@ public class ManifestsReader {
&& (specifiedLevel < minLevel || specifiedLevel > maxLevel)) {
return false;
}
- if (manifestLevelFilter != null && !manifestLevelFilter.test(minLevel, maxLevel)) {
+ if (levelMinMaxFilter != null && !levelMinMaxFilter.test(minLevel, maxLevel)) {
return false;
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
index 91505ffcf9..ed80f3f92a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
@@ -88,7 +88,7 @@ public interface SnapshotReader {
SnapshotReader withLevelFilter(Filter<Integer> levelFilter);
- SnapshotReader withManifestLevelFilter(BiFilter<Integer, Integer> manifestLevelFilter);
+ SnapshotReader withLevelMinMaxFilter(BiFilter<Integer, Integer> minMaxFilter);
SnapshotReader enableValueFilter();
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
index 7b1ca7b37e..477ce34e93 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
@@ -255,8 +255,9 @@ public class SnapshotReaderImpl implements SnapshotReader {
return this;
}
- public SnapshotReader withManifestLevelFilter(BiFilter<Integer, Integer> manifestLevelFilter) {
- scan.withManifestLevelFilter(manifestLevelFilter);
+ @Override
+ public SnapshotReader withLevelMinMaxFilter(BiFilter<Integer, Integer> minMaxFilter) {
+ scan.withLevelMinMaxFilter(minMaxFilter);
return this;
}
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 f71eba944d..3fd6a8d8af 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
@@ -361,9 +361,8 @@ public class AuditLogTable implements DataTable, ReadonlyTable {
}
@Override
- public SnapshotReader withManifestLevelFilter(
- BiFilter<Integer, Integer> manifestLevelFilter) {
- wrapped.withManifestLevelFilter(manifestLevelFilter);
+ public SnapshotReader withLevelMinMaxFilter(BiFilter<Integer, Integer> minMaxFilter) {
+ wrapped.withLevelMinMaxFilter(minMaxFilter);
return this;
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/cluster/HistoryPartitionClusterTest.java b/paimon-core/src/test/java/org/apache/paimon/append/cluster/HistoryPartitionClusterTest.java
index 4692b50fbb..f1f93ca989 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/cluster/HistoryPartitionClusterTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/cluster/HistoryPartitionClusterTest.java
@@ -24,8 +24,6 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
-import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.mergetree.LevelSortedRun;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.schema.Schema;
@@ -39,17 +37,12 @@ import org.assertj.core.util.Lists;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
-import java.time.LocalDateTime;
-import java.time.ZoneId;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
-import java.util.stream.Collectors;
import static org.apache.paimon.append.cluster.IncrementalClusterManagerTest.writeOnce;
-import static org.apache.paimon.append.cluster.IncrementalClusterStrategyTest.createFile;
import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link HistoryPartitionCluster}. */
@@ -57,53 +50,6 @@ public class HistoryPartitionClusterTest {
@TempDir java.nio.file.Path tempDir;
- @Test
- public void testFindLowLevelPartitions() throws Exception {
- FileStoreTable table = createTable(Collections.emptyMap(), Collections.emptyList());
- long now = LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
- Map<BinaryRow, List<DataFileMeta>> partitionFiles = new HashMap<>();
-
- // specified partition, has low-level files
- BinaryRow partition1 = BinaryRow.singleColumn(1);
- PartitionEntry partitionEntry1 = new PartitionEntry(partition1, 0, 0, 0, now);
- partitionFiles.put(
- partition1, Lists.newArrayList(createFile(100, 1, 3), createFile(100, 1, 5)));
- // has no low-level files
- BinaryRow partition2 = BinaryRow.singleColumn(2);
- PartitionEntry partitionEntry2 = new PartitionEntry(partition2, 0, 0, 0, now);
- partitionFiles.put(partition2, Lists.newArrayList(createFile(100, 1, 0)));
- // has low-level files
- BinaryRow partition3 = BinaryRow.singleColumn(3);
- PartitionEntry partitionEntry3 = new PartitionEntry(partition3, 0, 0, 0, now);
- partitionFiles.put(
- partition3, Lists.newArrayList(createFile(100, 1, 0), createFile(100, 1, 2)));
- // has no low-level files
- BinaryRow partition4 = BinaryRow.singleColumn(4);
- PartitionEntry partitionEntry4 = new PartitionEntry(partition3, 0, 0, 0, now);
- partitionFiles.put(partition4, Lists.newArrayList(createFile(100, 1, 0)));
-
- IncrementalClusterManager incrementalClusterManager =
- new IncrementalClusterManager(
- table,
- PartitionPredicate.fromMultiple(
- RowType.of(DataTypes.INT()), Lists.newArrayList(partition1)));
- HistoryPartitionCluster historyPartitionCluster =
- incrementalClusterManager.historyPartitionCluster();
- Set<BinaryRow> selectedPartitions =
- historyPartitionCluster.findLowLevelPartitions(
- Lists.newArrayList(
- partitionEntry1,
- partitionEntry2,
- partitionEntry3,
- partitionEntry4)
- .stream()
- .map(PartitionEntry::partition)
- .collect(Collectors.toList()),
- partitionFiles);
-
- assertThat(selectedPartitions).contains(partition2);
- }
-
@Test
public void testHistoryPartitionAutoClustering() throws Exception {
FileStoreTable table = createTable(Collections.emptyMap(), Collections.singletonList("f2"));
@@ -153,8 +99,7 @@ public class HistoryPartitionClusterTest {
// test not specify partition and disable history partition auto clustering
historyPartitionCluster = new IncrementalClusterManager(table).historyPartitionCluster();
- partitionLevels = historyPartitionCluster.constructLevelsForHistoryPartitions();
- assertThat(partitionLevels.isEmpty()).isTrue();
+ assertThat(historyPartitionCluster).isNull();
}
protected FileStoreTable createTable(
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
index 37cc006673..4b62caab88 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
@@ -58,7 +58,7 @@ public class IncrementalClusterManagerTest {
@TempDir java.nio.file.Path tempDir;
@Test
- public void testNonUnAwareBucketTable() throws Exception {
+ public void testNonUnAwareBucketTable() {
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.BUCKET.key(), "1");
options.put(CoreOptions.BUCKET_KEY.key(), "f0");
(paimon) 08/09: [spark] Fix write non-pk dv table with external paths (#6487)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.3
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 5625e47e48dd6a26002c7b20d1dad4a26f97a7fb
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Wed Oct 29 10:26:11 2025 +0800
[spark] Fix write non-pk dv table with external paths (#6487)
---
.../spark/commands/PaimonRowLevelCommand.scala | 31 ++++++--------------
.../paimon/spark/commands/PaimonSparkWriter.scala | 5 ++--
.../paimon/spark/commands/SparkDataFileMeta.scala | 13 +++++----
.../spark/commands/SparkDeletionVector.scala | 23 ++++-----------
.../spark/commands/UpdatePaimonTableCommand.scala | 7 ++---
.../paimon/spark/sql/DeletionVectorTest.scala | 33 +++++++++++++++++++++-
6 files changed, 60 insertions(+), 52 deletions(-)
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonRowLevelCommand.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonRowLevelCommand.scala
index 300ca20ff8..11dd947009 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonRowLevelCommand.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonRowLevelCommand.scala
@@ -71,12 +71,6 @@ trait PaimonRowLevelCommand
}
}
- /** Gets a relative path against the table path. */
- protected def relativePath(absolutePath: String): String = {
- val location = table.location().toUri
- location.relativize(new URI(absolutePath)).toString
- }
-
protected def findCandidateDataSplits(
condition: Expression,
output: Seq[Attribute]): Seq[DataSplit] = {
@@ -121,7 +115,6 @@ trait PaimonRowLevelCommand
.distinct()
.as[String]
.collect()
- .map(relativePath)
}
protected def extractFilesAndCreateNewScan(
@@ -136,15 +129,14 @@ trait PaimonRowLevelCommand
(files, newRelation)
}
- /** Notice that, the key is a relative path, not just the file name. */
+ /** Notice that, the key is a file path, not just the file name. */
protected def candidateFileMap(
candidateDataSplits: Seq[DataSplit]): Map[String, SparkDataFileMeta] = {
val totalBuckets = coreOptions.bucket()
val candidateDataFiles = candidateDataSplits
.flatMap(dataSplit => convertToSparkDataFileMeta(dataSplit, totalBuckets))
- val fileStorePathFactory = fileStore.pathFactory()
candidateDataFiles
- .map(file => (file.relativePath(fileStorePathFactory), file))
+ .map(file => (file.filePath(), file))
.toMap
}
@@ -165,11 +157,12 @@ trait PaimonRowLevelCommand
dataset: Dataset[Row],
sparkSession: SparkSession): Dataset[SparkDeletionVector] = {
import sparkSession.implicits._
- val dataFileToPartitionAndBucket =
- dataFilePathToMeta.mapValues(meta => (meta.partition, meta.bucket)).toArray
+ // convert to a serializable map
+ val dataFileToPartitionAndBucket = dataFilePathToMeta.map {
+ case (k, v) => k -> (v.bucketPath, v.partition, v.bucket)
+ }
val my_table = table
- val location = my_table.location
val dvBitmap64 = my_table.coreOptions().deletionVectorBitmap64()
dataset
.select(DV_META_COLUMNS.map(col): _*)
@@ -183,18 +176,12 @@ trait PaimonRowLevelCommand
dv.delete(iter.next()._2)
}
- val relativeFilePath = location.toUri.relativize(new URI(filePath)).toString
- val (partition, bucket) = dataFileToPartitionAndBucket.toMap.apply(relativeFilePath)
- val pathFactory = my_table.store().pathFactory()
- val relativeBucketPath = pathFactory
- .relativeBucketPath(partition, bucket)
- .toString
-
+ val (bucketPath, partition, bucket) = dataFileToPartitionAndBucket.apply(filePath)
SparkDeletionVector(
- relativeBucketPath,
+ bucketPath,
SerializationUtils.serializeBinaryRow(partition),
bucket,
- new Path(filePath).getName,
+ filePath,
DeletionVector.serializeToBytes(dv)
)
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
index 6d0563b364..ee3104e9b3 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -25,6 +25,7 @@ import org.apache.paimon.crosspartition.{IndexBootstrap, KeyPartOrRow}
import org.apache.paimon.data.serializer.InternalSerializers
import org.apache.paimon.deletionvectors.DeletionVector
import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer
+import org.apache.paimon.fs.Path
import org.apache.paimon.index.{BucketAssigner, SimpleHashBucketAssigner}
import org.apache.paimon.io.{CompactIncrement, DataIncrement}
import org.apache.paimon.manifest.FileKind
@@ -310,7 +311,7 @@ case class PaimonSparkWriter(table: FileStoreTable, writeRowTracking: Boolean =
val sparkSession = deletionVectors.sparkSession
import sparkSession.implicits._
val serializedCommits = deletionVectors
- .groupByKey(_.partitionAndBucket)
+ .groupByKey(_.bucketPath)
.mapGroups {
(_, iter: Iterator[SparkDeletionVector]) =>
val indexHandler = table.store().newIndexFileHandler()
@@ -334,7 +335,7 @@ case class PaimonSparkWriter(table: FileStoreTable, writeRowTracking: Boolean =
}
dvIndexFileMaintainer.notifyNewDeletionVector(
- sdv.dataFileName,
+ new Path(sdv.dataFilePath).getName,
DeletionVector.deserializeFromBytes(sdv.deletionVector))
}
val indexEntries = dvIndexFileMaintainer.persist()
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SparkDataFileMeta.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SparkDataFileMeta.scala
index 569a84a74c..921d2e4735 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SparkDataFileMeta.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SparkDataFileMeta.scala
@@ -27,17 +27,19 @@ import org.apache.paimon.utils.FileStorePathFactory
import scala.collection.JavaConverters._
case class SparkDataFileMeta(
+ bucketPath: String,
partition: BinaryRow,
bucket: Int,
totalBuckets: Int,
dataFileMeta: DataFileMeta,
deletionFile: Option[DeletionFile] = None) {
- def relativePath(fileStorePathFactory: FileStorePathFactory): String = {
- fileStorePathFactory
- .relativeBucketPath(partition, bucket)
- .toUri
- .toString + "/" + dataFileMeta.fileName()
+ def filePath(): String = {
+ if (dataFileMeta.externalPath().isPresent) {
+ dataFileMeta.externalPath().get()
+ } else {
+ bucketPath + "/" + dataFileMeta.fileName()
+ }
}
}
@@ -52,6 +54,7 @@ object SparkDataFileMeta {
dataSplit.dataFiles().asScala.map {
file =>
SparkDataFileMeta(
+ dataSplit.bucketPath(),
dataSplit.partition,
dataSplit.bucket,
totalBuckets,
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SparkDeletionVector.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SparkDeletionVector.scala
index 9fc7fdadcb..53453af913 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SparkDeletionVector.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SparkDeletionVector.scala
@@ -18,9 +18,8 @@
package org.apache.paimon.spark.commands
-import org.apache.paimon.fs.Path
import org.apache.paimon.table.source.DataSplit
-import org.apache.paimon.utils.{FileStorePathFactory, SerializationUtils}
+import org.apache.paimon.utils.SerializationUtils
import scala.collection.JavaConverters._
@@ -29,31 +28,21 @@ import scala.collection.JavaConverters._
* or DeletionVector.
*/
case class SparkDeletionVector(
- partitionAndBucket: String,
+ bucketPath: String,
partition: Array[Byte],
bucket: Int,
- dataFileName: String,
+ dataFilePath: String,
deletionVector: Array[Byte]
-) {
- def relativePath(pathFactory: FileStorePathFactory): String = {
- val prefix = pathFactory
- .relativeBucketPath(SerializationUtils.deserializeBinaryRow(partition), bucket)
- .toUri
- .toString + "/"
- prefix + dataFileName
- }
-}
+)
object SparkDeletionVector {
def toDataSplit(
deletionVector: SparkDeletionVector,
- root: Path,
- pathFactory: FileStorePathFactory,
dataFilePathToMeta: Map[String, SparkDataFileMeta]): DataSplit = {
- val meta = dataFilePathToMeta(deletionVector.relativePath(pathFactory))
+ val meta = dataFilePathToMeta(deletionVector.dataFilePath)
DataSplit
.builder()
- .withBucketPath(root + "/" + deletionVector.partitionAndBucket)
+ .withBucketPath(deletionVector.bucketPath)
.withPartition(SerializationUtils.deserializeBinaryRow(deletionVector.partition))
.withBucket(deletionVector.bucket)
.withTotalBuckets(meta.totalBuckets)
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
index 8839d5c8ac..1258ebc449 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
@@ -81,8 +81,6 @@ case class UpdatePaimonTableCommand(
logDebug("No file need to rewrote. It's an empty Commit.")
Seq.empty[CommitMessage]
} else {
- val pathFactory = fileStore.pathFactory()
-
if (deletionVectorsEnabled) {
// Step2: collect all the deletion vectors that marks the deleted rows.
val deletionVectors = collectDeletionVectors(
@@ -95,9 +93,8 @@ case class UpdatePaimonTableCommand(
deletionVectors.cache()
try {
// Step3: write these updated data
- val touchedDataSplits = deletionVectors.collect().map {
- SparkDeletionVector.toDataSplit(_, root, pathFactory, dataFilePathToMeta)
- }
+ val touchedDataSplits =
+ deletionVectors.collect().map(SparkDeletionVector.toDataSplit(_, dataFilePathToMeta))
val addCommitMessage = writeOnlyUpdatedData(sparkSession, touchedDataSplits)
// Step4: write these deletion vectors.
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeletionVectorTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeletionVectorTest.scala
index 68e741fb13..e3a5896ab1 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeletionVectorTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeletionVectorTest.scala
@@ -660,7 +660,7 @@ class DeletionVectorTest extends PaimonSparkTestBase with AdaptiveSparkPlanHelpe
assert(dvMeta.cardinality() == 334)
}
- test("Paimon deletionVector: delete from non-pk table with data file path") {
+ test("Paimon deletionVector: delete from non-pk table with data file directory") {
sql(s"""
|CREATE TABLE T (id INT)
|TBLPROPERTIES (
@@ -677,6 +677,37 @@ class DeletionVectorTest extends PaimonSparkTestBase with AdaptiveSparkPlanHelpe
checkAnswer(sql("SELECT count(*) FROM T"), Row(49665))
}
+ test("Paimon deletionVector: delete from non-pk table with data file external paths") {
+ withTempDir {
+ tmpDir =>
+ {
+ sql(s"""
+ |CREATE TABLE T (id INT, v INT)
+ |TBLPROPERTIES (
+ | 'deletion-vectors.enabled' = 'true',
+ | 'deletion-vectors.bitmap64' = '${Random.nextBoolean()}',
+ | 'bucket-key' = 'id',
+ | 'bucket' = '1',
+ | 'data-file.external-paths' = 'file://${tmpDir.getCanonicalPath}',
+ | 'data-file.external-paths.strategy' = 'round-robin'
+ |)
+ |""".stripMargin)
+ sql("INSERT INTO T SELECT /*+ REPARTITION(1) */ id, id FROM range (1, 50000)")
+ sql("DELETE FROM T WHERE id >= 111 and id <= 444")
+ checkAnswer(sql("SELECT count(*) FROM T"), Row(49665))
+ checkAnswer(sql("SELECT sum(v) FROM T"), Row(1249882315L))
+
+ sql("UPDATE T SET v = v + 1 WHERE id >= 555 and id <= 666")
+ checkAnswer(sql("SELECT count(*) FROM T"), Row(49665))
+ checkAnswer(sql("SELECT sum(v) FROM T"), Row(1249882427L))
+
+ sql("UPDATE T SET v = v + 1 WHERE id >= 600 and id <= 800")
+ checkAnswer(sql("SELECT count(*) FROM T"), Row(49665))
+ checkAnswer(sql("SELECT sum(v) FROM T"), Row(1249882628L))
+ }
+ }
+ }
+
test("Paimon deletionVector: work v1 with v2") {
sql(s"""
|CREATE TABLE T (id INT)
(paimon) 05/09: [doc] Update batch partition mark done doc (#6478)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.3
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 5c90f445a05609f768959f5588922a0b8fa2012a
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Tue Oct 28 08:33:51 2025 +0800
[doc] Update batch partition mark done doc (#6478)
---
docs/content/maintenance/manage-partitions.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/docs/content/maintenance/manage-partitions.md b/docs/content/maintenance/manage-partitions.md
index 7b851fe055..42d6f4601b 100644
--- a/docs/content/maintenance/manage-partitions.md
+++ b/docs/content/maintenance/manage-partitions.md
@@ -166,5 +166,5 @@ enable this by setting `'partition.mark-done-action.mode' = 'watermark'`.
### Batch Mark Done
-For batch mode, you can trigger partition mark done when end input by setting `'partition.end-input-to-done'='true'`.
-
+For batch mode, you can trigger partition mark done when end input by setting `'partition.end-input-to-done'='true'`,
+and all partitions written in this batch will be marked done.
(paimon) 03/09: [doc] Fix typo in flink/sql-ddl.md (#6476)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.3
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit f170713ac837de551dff5f1d19f23a2f716d8ace
Author: YangJie <ya...@baidu.com>
AuthorDate: Mon Oct 27 20:31:13 2025 +0800
[doc] Fix typo in flink/sql-ddl.md (#6476)
---
docs/content/flink/sql-ddl.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/content/flink/sql-ddl.md b/docs/content/flink/sql-ddl.md
index 4626704270..21845a59be 100644
--- a/docs/content/flink/sql-ddl.md
+++ b/docs/content/flink/sql-ddl.md
@@ -213,7 +213,7 @@ note that partition fields and primary key fields can not be specified.
## Create Table As Select
-Table can be created and populated by the results of a query, for example, we have a sql like this: `CREATE TABLE table_b AS SELECT id, name FORM table_a`,
+Table can be created and populated by the results of a query, for example, we have a sql like this: `CREATE TABLE table_b AS SELECT id, name FROM table_a`,
The resulting table `table_b` will be equivalent to create the table and insert the data with the following statement:
`CREATE TABLE table_b (id INT, name STRING); INSERT INTO table_b SELECT id, name FROM table_a;`
(paimon) 09/09: [core] Fix endian spec for BloomFilter index reader (#6493)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.3
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit d495257d4c89be8aa526024e02cc43542c4d3215
Author: Yubin Li <li...@163.com>
AuthorDate: Wed Oct 29 19:37:04 2025 +0800
[core] Fix endian spec for BloomFilter index reader (#6493)
---
.../org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndex.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndex.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndex.java
index 3c9dcadba3..83d6b64b64 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndex.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndex.java
@@ -117,7 +117,7 @@ public class BloomFilterFileIndex implements FileIndexer {
private final FastHash hashFunction;
public Reader(DataType type, byte[] serializedBytes) {
- // little endian
+ // big endian
int numHashFunctions =
((serializedBytes[0] << 24)
+ (serializedBytes[1] << 16)