This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 6528cb5d3a [core] Compact Data Evolution column versions alongside
clustering (#10684)
6528cb5d3a is described below
commit 6528cb5d3a9577507d1f8a14aee5acfdd5f515f3
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 20 12:56:01 2026 +0800
[core] Compact Data Evolution column versions alongside clustering (#10684)
---
docs/docs/append-table/incremental-clustering.mdx | 2 +
.../data-evolution-maintenance.mdx | 31 +-
.../cluster/DataEvolutionClusterCompaction.java | 112 ++++++
.../DataEvolutionIncrementalClusterRewriter.java | 18 +-
.../dataevolution/ClusteringIndexRewriter.java | 48 ++-
.../DataEvolutionCompactTaskSerializer.java | 8 +-
.../DataEvolutionNormalCompactTask.java | 27 +-
.../paimon/operation/FileStoreCommitImpl.java | 26 +-
.../commit/DataEvolutionConflictDetection.java | 19 +-
.../commit/RowIdRangeConflictChecker.java | 20 +-
.../apache/paimon/utils/DataEvolutionUtils.java | 66 ++++
.../DataEvolutionClusterCompactionTest.java | 277 ++++++++++++++
.../dataevolution/DataEvolutionClusterTest.java | 414 +++++++++++++++++++++
.../DataEvolutionCompactTaskSerializerTest.java | 100 +++++
.../DataEvolutionNormalCompactTaskTest.java | 80 ++++
.../paimon/utils/DataEvolutionUtilsTest.java | 100 +++++
.../flink/compact/IncrementalClusterCompact.java | 32 +-
.../paimon/flink/DataEvolutionDeleteSqlITCase.java | 207 +++++++++++
.../paimon/spark/procedure/CompactProcedure.java | 18 +-
.../procedure/DataEvolutionRewriteExecutor.java | 111 +++---
.../spark/procedure/CompactProcedureTestBase.scala | 72 ++++
21 files changed, 1666 insertions(+), 122 deletions(-)
diff --git a/docs/docs/append-table/incremental-clustering.mdx
b/docs/docs/append-table/incremental-clustering.mdx
index 49f1763712..d0152cde9e 100644
--- a/docs/docs/append-table/incremental-clustering.mdx
+++ b/docs/docs/append-table/incremental-clustering.mdx
@@ -138,6 +138,8 @@ all table partitions become globally ordered after an
incremental run.
Data Evolution uses the same sort modes. Its readers merge the latest column
values and apply deletion vectors
for complete selected file groups before sorting. The rewritten rows receive
new row IDs at commit time.
+Already-clustered groups with accumulated non-clustering column versions can
instead be compacted in the same job,
+preserving their row IDs and logical deletes; see [Data Evolution
clustering](../multimodal-table/data-evolution-maintenance#incremental-clustering).
## Run Incremental Clustering
diff --git a/docs/docs/multimodal-table/data-evolution-maintenance.mdx
b/docs/docs/multimodal-table/data-evolution-maintenance.mdx
index 25646225fe..64d6bedb77 100644
--- a/docs/docs/multimodal-table/data-evolution-maintenance.mdx
+++ b/docs/docs/multimodal-table/data-evolution-maintenance.mdx
@@ -184,8 +184,15 @@ before the existing Spark or Flink clustering pipeline
sorts the surviving rows
and writes new normal files with new row IDs. Unselected groups keep their row
IDs.
The output records the clustering columns and level for subsequent incremental
selection. Updating only non-clustering columns does not make an already
clustered
-group eligible again: its row order remains valid, so even a full clustering
call
-can skip it and preserve its row IDs and indexes on unchanged columns.
+group require sorting again: its row order remains valid. Unselected groups
with
+at least `compaction.min.file-num` normal files (default 5) are instead
compacted
+within the same job and snapshot. This merges their latest column values while
+preserving row IDs, clustering level, logical deletes, and indexes on unchanged
+columns. If minor selection skips a group whose sorting order is no longer
+valid, column-version compaction keeps it at level 0 for later clustering.
+Each such task handles one complete row-ID group; adjacent groups are
+not combined because row-ID order alone does not prove clustering-key order.
+Groups below the threshold can be skipped even by a full clustering call.
Updating a clustering column
makes the group eligible again. When a group is selected, all its column
versions
are rewritten together, including newer values of non-clustering columns.
@@ -203,11 +210,15 @@ CALL sys.compact(table => 'default.target_table',
compact_strategy => 'minor');
```
Use the corresponding Flink `ALTER TABLE ... SET` syntax and run the compact
job
-in batch mode. Explicit compact jobs use clustering while this option is
enabled.
-To consolidate column versions while preserving row IDs, disable clustering for
-the compact invocation with `options => 'clustering.incremental=false'`.
-
-Clustering removes the applied deletion vectors. Flink clustering also rewrites
+in batch mode. Explicit compact jobs combine clustering and column-version
+compaction while this option is enabled. A group selected for clustering is
+rewritten only once. To run ordinary Data Evolution compaction across adjacent
+row-ID groups, disable clustering for the invocation with
+`options => 'clustering.incremental=false'`.
+
+The sorting path removes the applied deletion vectors; column-version
compaction
+retains logical deletes and moves their vectors to the replacement files.
+Flink clustering also rewrites
B-tree and bitmap index files with the new row IDs, reusing their stored keys.
Indexes outside the rewritten row-ID ranges are preserved, including those in
the same partition. Partially indexed data retains its exact coverage after
sorting.
@@ -219,8 +230,8 @@ The job records the old row IDs alongside each output file
so that concurrent
index creation and optimistic commit retries use the final assigned row IDs.
Indexes with stale keys after indexed-column updates, and affected index types
-without RowID rewrite support, are dropped. Spark clustering currently drops
-global indexes in the affected partitions. Rebuild dropped indexes, or use
+without RowID rewrite support, are dropped. Spark clustering drops index files
+whose coverage intersects the row-ID ranges selected for sorting. Rebuild
dropped indexes, or use
`scalar-index.search-mode = full` for complete scalar query results while part
of the table remains unindexed.
Concurrent updates or deletes that overlap the selected row-ID groups cause
@@ -237,7 +248,7 @@ the previous files until expiration.
| Logical delete | Hides IDs without shifting the remaining rows. |
| Ordinary Data Evolution compaction | Preserves IDs, but may change
file-group boundaries. |
| Deletion-vector materialization | Assigns new IDs to surviving rows in the
rewritten ranges. |
-| Data Evolution incremental clustering | Assigns new IDs to surviving rows in
the selected groups. |
+| Data Evolution incremental clustering | Assigns new IDs to surviving rows in
groups selected for sorting; column-version compaction preserves IDs. |
| Explicit `reassign_row_id` maintenance | Changes IDs by remapping file
metadata. |
| Overwrite or partition removal | Removes the replaced rows from the current
table state; do not assume their IDs remain usable. |
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/DataEvolutionClusterCompaction.java
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/DataEvolutionClusterCompaction.java
new file mode 100644
index 0000000000..58ee72a1ff
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/DataEvolutionClusterCompaction.java
@@ -0,0 +1,112 @@
+/*
+ * 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.Snapshot;
+import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
+import org.apache.paimon.append.dataevolution.DataEvolutionNormalCompactTask;
+import org.apache.paimon.compact.CompactUnit;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.utils.RangeHelper;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+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.format.blob.BlobFileFormat.isBlobFile;
+import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Compacts column versions left outside the incremental clustering
selection. */
+public final class DataEvolutionClusterCompaction {
+
+ private DataEvolutionClusterCompaction() {}
+
+ public static List<DataEvolutionCompactTask> plan(
+ FileStoreTable table,
+ Snapshot snapshot,
+ @Nullable PartitionPredicate predicate,
+ Map<BinaryRow, CompactUnit> clusteringUnits) {
+ if (!table.coreOptions().dataEvolutionEnabled()) {
+ return Collections.emptyList();
+ }
+ Map<BinaryRow, List<DataFileMeta>> partitions = new HashMap<>();
+ for (DataSplit split :
+ table.newSnapshotReader()
+ .dropStats()
+ .withSnapshot(snapshot)
+ .withPartitionFilter(predicate)
+ .read()
+ .dataSplits()) {
+ partitions
+ .computeIfAbsent(split.partition(), ignored -> new
ArrayList<>())
+ .addAll(split.dataFiles());
+ }
+ DataEvolutionIncrementalClusterRewriter rewriter =
+ new DataEvolutionIncrementalClusterRewriter(table, snapshot);
+ RangeHelper<DataFileMeta> helper = new
RangeHelper<>(DataFileMeta::nonNullRowIdRange);
+ List<DataEvolutionCompactTask> tasks = new ArrayList<>();
+ partitions.forEach(
+ (partition, files) -> {
+ CompactUnit clustering = clusteringUnits.get(partition);
+ Set<String> selected =
+ clustering == null
+ ? Collections.emptySet()
+ : clustering.files().stream()
+ .map(DataFileMeta::fileName)
+ .collect(Collectors.toSet());
+ List<DataFileMeta> normalFiles =
+ files.stream()
+ .filter(
+ file ->
+
!isBlobFile(file.fileName())
+ &&
!isVectorStoreFile(file.fileName()))
+ .collect(Collectors.toList());
+ for (List<DataFileMeta> group :
helper.mergeOverlappingRanges(normalFiles)) {
+ checkArgument(
+ helper.areAllRangesSame(group),
+ "Normal files must have equal row-id ranges.");
+ if (group.size() < Math.max(2,
table.coreOptions().compactionMinFileNum())
+ || group.stream()
+ .anyMatch(file ->
selected.contains(file.fileName()))) {
+ continue;
+ }
+ // A single group preserves even LOCAL_SORT order when
its level is trusted.
+ // Dirty groups still compact at L0 without claiming
that they are sorted.
+ tasks.add(
+ new DataEvolutionNormalCompactTask(
+ partition,
+ group,
+ Collections.emptyList(),
+ rewriter.groupLevel(group)));
+ }
+ });
+ return tasks;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/DataEvolutionIncrementalClusterRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/DataEvolutionIncrementalClusterRewriter.java
index f50acb29a0..90269cd4bb 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/DataEvolutionIncrementalClusterRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/DataEvolutionIncrementalClusterRewriter.java
@@ -107,6 +107,10 @@ class DataEvolutionIncrementalClusterRewriter extends
IncrementalClusterRewriter
return runs;
}
+ int groupLevel(List<DataFileMeta> group) {
+ return groupLevel(group, retrieveAnchorFile(group, file -> file));
+ }
+
private int groupLevel(List<DataFileMeta> group, DataFileMeta anchor) {
int level = anchor.level();
CoreOptions anchorOptions =
CoreOptions.fromMap(schema(anchor.schemaId()).options());
@@ -162,11 +166,21 @@ class DataEvolutionIncrementalClusterRewriter extends
IncrementalClusterRewriter
@Override
public List<CommitMessage> prepareCommitMessages(List<CommitMessage>
messages) {
Map<BinaryRow, CompactIncrement> increments = new HashMap<>();
+ List<CommitMessage> prepared = new ArrayList<>();
for (CommitMessage message : messages) {
CommitMessageImpl impl = (CommitMessageImpl) message;
checkArgument(
impl.bucket() == 0 && impl.newFilesIncrement().isEmpty(),
"Data evolution clustering requires compact messages for
unaware bucket 0.");
+ CompactIncrement increment = impl.compactIncrement();
+ if (!increment.compactAfter().isEmpty()
+ && increment.compactAfter().stream()
+ .allMatch(file -> file.firstRowId() != null)) {
+ // Keep row-preserving tasks separate: their DVs move to the
replacement anchor,
+ // whereas clustering outputs have already applied DVs and
need fresh row IDs.
+ prepared.add(message);
+ continue;
+ }
CompactIncrement merged =
increments.computeIfAbsent(
message.partition(),
@@ -177,14 +191,12 @@ class DataEvolutionIncrementalClusterRewriter extends
IncrementalClusterRewriter
new ArrayList<>(),
new ArrayList<>(),
new ArrayList<>()));
- CompactIncrement increment = impl.compactIncrement();
merged.compactBefore().addAll(increment.compactBefore());
merged.compactAfter().addAll(increment.compactAfter());
merged.changelogFiles().addAll(increment.changelogFiles());
merged.newIndexFiles().addAll(increment.newIndexFiles());
merged.deletedIndexFiles().addAll(increment.deletedIndexFiles());
}
- List<CommitMessage> prepared = new ArrayList<>();
increments.forEach(
(partition, increment) -> {
checkArgument(
@@ -194,7 +206,7 @@ class DataEvolutionIncrementalClusterRewriter extends
IncrementalClusterRewriter
new CommitMessageImpl(
partition, 0, null,
DataIncrement.emptyIncrement(), increment));
});
- // The committer invalidates global indexes against the latest
snapshot on every attempt.
+ // The committer maintains global indexes against the latest snapshot
on every attempt.
prepared.addAll(
new DataEvolutionCompactDeletionVectorRewriter(table, snapshot)
.rewriteDeletionVectors(prepared));
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringIndexRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringIndexRewriter.java
index 3cf1f2f687..36d360dac4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringIndexRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringIndexRewriter.java
@@ -68,6 +68,7 @@ import java.util.Set;
import java.util.TreeMap;
import java.util.stream.Collectors;
+import static
org.apache.paimon.utils.DataEvolutionUtils.materializedRowIdRanges;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Rewrites scalar index postings after clustering output files receive their
final row IDs. */
@@ -88,6 +89,7 @@ public final class ClusteringIndexRewriter {
List<ManifestEntry> dataFiles,
List<IndexManifestEntry> indexFiles,
Set<String> mappedClusteringOutputs,
+ Set<String> rowIdPreservingOutputs,
long newSnapshotId,
List<Path> createdFiles)
throws IOException {
@@ -121,13 +123,6 @@ public final class ClusteringIndexRewriter {
.filter(e -> e.kind() == FileKind.ADD)
.map(ManifestEntry::file)
.collect(Collectors.toList());
- // Other materialization callers do not supply a permutation and
retain their existing
- // invalidation behavior. A mixed mapping must never silently
retain a stale posting.
- if (after.stream()
- .anyMatch(file ->
!mappedClusteringOutputs.contains(file.fileName()))) {
- result.addAll(group.getValue());
- continue;
- }
List<ManifestEntry> before =
files.stream()
.filter(e -> e.kind() == FileKind.DELETE)
@@ -144,16 +139,41 @@ public final class ClusteringIndexRewriter {
result.addAll(group.getValue());
continue;
}
+ List<Range> removed =
+ materializedRowIdRanges(
+
before.stream().map(ManifestEntry::file).collect(Collectors.toList()),
+ after.stream()
+ .filter(
+ file ->
+
rowIdPreservingOutputs.contains(
+ file.fileName()))
+ .collect(Collectors.toList()));
+ if (removed.isEmpty()) {
+ // Preserve explicitly requested index deletions for ordinary
compaction callers.
+ result.addAll(group.getValue());
+ continue;
+ }
+ after.removeIf(file ->
rowIdPreservingOutputs.contains(file.fileName()));
+ // Engines without a permutation still invalidate affected
indexes. Independently
+ // compacted ranges in the same partition retain their original
index files.
+ if (after.stream()
+ .anyMatch(file ->
!mappedClusteringOutputs.contains(file.fileName()))) {
+ group.getValue().stream()
+ .filter(
+ entry ->
+ !Range.and(
+
Collections.singletonList(
+
entry.indexFile()
+
.globalIndexMeta()
+
.rowRange()),
+ removed)
+ .isEmpty())
+ .forEach(result::add);
+ continue;
+ }
after.sort(Comparator.comparingLong(DataFileMeta::nonNullFirstRowId));
long firstAssignedRowId =
after.isEmpty() ? Long.MAX_VALUE :
after.get(0).nonNullFirstRowId();
- List<Range> removed =
- Range.sortAndMergeOverlap(
- before.stream()
- .map(ManifestEntry::file)
- .map(DataFileMeta::nonNullRowIdRange)
- .collect(Collectors.toList()),
- true);
DataFilePathFactory dataPath =
pathFactory.createDataFilePathFactory(
group.getKey().getLeft(),
group.getKey().getRight());
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializer.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializer.java
index 5b4e463dcb..cd6129a170 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializer.java
@@ -43,7 +43,7 @@ import static
org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
public class DataEvolutionCompactTaskSerializer
implements VersionedSerializer<DataEvolutionCompactTask> {
- private static final int CURRENT_VERSION = 4;
+ private static final int CURRENT_VERSION = 5;
private final DataFileMetaSerializer dataFileSerializer;
@@ -83,6 +83,7 @@ public class DataEvolutionCompactTaskSerializer
view.writeLong(range.from);
view.writeLong(range.to);
}
+ view.writeInt(((DataEvolutionNormalCompactTask)
task).outputLevel());
} else if (task.type() ==
DataEvolutionCompactTask.TaskType.MATERIALIZE_DELETION) {
DeletionFile.serializeList(
view, ((DataEvolutionMaterializeDeletionCompactTask)
task).deletionFiles());
@@ -108,7 +109,7 @@ public class DataEvolutionCompactTaskSerializer
}
private void checkVersion(int version) {
- if (version != CURRENT_VERSION) {
+ if (version != 4 && version != CURRENT_VERSION) {
throw new UnsupportedOperationException(
"Expecting DataEvolutionCompactTask version to be "
+ CURRENT_VERSION
@@ -132,7 +133,8 @@ public class DataEvolutionCompactTaskSerializer
for (int i = 0; i < rangeCount; i++) {
ranges.add(new Range(view.readLong(), view.readLong()));
}
- return new DataEvolutionNormalCompactTask(partition, files,
ranges);
+ int outputLevel = version >= 5 ? view.readInt() : 0;
+ return new DataEvolutionNormalCompactTask(partition, files,
ranges, outputLevel);
case BLOB:
return new DataEvolutionBlobCompactTask(partition, files);
case MATERIALIZE_DELETION:
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
index 7065a37bc1..2191e5026b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
@@ -69,6 +69,7 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
private static final Logger LOG =
LoggerFactory.getLogger(DataEvolutionNormalCompactTask.class);
private final List<Range> protectedRanges;
+ private final int outputLevel;
public DataEvolutionNormalCompactTask(BinaryRow partition,
List<DataFileMeta> files) {
this(partition, files, Collections.emptyList());
@@ -76,8 +77,19 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
public DataEvolutionNormalCompactTask(
BinaryRow partition, List<DataFileMeta> files, List<Range>
protectedRanges) {
+ this(partition, files, protectedRanges, 0);
+ }
+
+ /** Preserves a clustering level only when the caller has verified the
complete input group. */
+ public DataEvolutionNormalCompactTask(
+ BinaryRow partition,
+ List<DataFileMeta> files,
+ List<Range> protectedRanges,
+ int outputLevel) {
super(partition, files);
+ checkArgument(outputLevel >= 0, "Output level must not be negative.");
checkContiguousRowRange(files);
+ this.outputLevel = outputLevel;
this.protectedRanges =
Collections.unmodifiableList(Range.sortAndMergeOverlap(protectedRanges));
}
@@ -86,15 +98,20 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
return protectedRanges;
}
+ public int outputLevel() {
+ return outputLevel;
+ }
+
@Override
public boolean equals(Object other) {
return super.equals(other)
- && protectedRanges.equals(((DataEvolutionNormalCompactTask)
other).protectedRanges);
+ && protectedRanges.equals(((DataEvolutionNormalCompactTask)
other).protectedRanges)
+ && outputLevel == ((DataEvolutionNormalCompactTask)
other).outputLevel;
}
@Override
public int hashCode() {
- return Objects.hash(super.hashCode(), protectedRanges);
+ return Objects.hash(super.hashCode(), protectedRanges, outputLevel);
}
@Override
@@ -178,7 +195,8 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
long maxSequenceNumber = maxSequenceId(compactBefore);
long nextRowId = firstRowId;
long[] writeColsSequences =
- options.ignoreIndexColumnUpdate() && !writeResult.isEmpty()
+ (options.ignoreIndexColumnUpdate() ||
options.clusteringIncrementalEnabled())
+ && !writeResult.isEmpty()
? compactedWriteColsSequences(
table,
writeResult
@@ -192,6 +210,9 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
if (writeColsSequences != null) {
dataFileMeta =
dataFileMeta.withWriteColsSequences(writeColsSequences);
}
+ if (outputLevel > 0) {
+ dataFileMeta = dataFileMeta.upgrade(outputLevel);
+ }
compactAfter.add(dataFileMeta);
nextRowId += dataFileMeta.rowCount();
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 5ff13ba2f1..70a4f2ac95 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -111,7 +111,6 @@ import java.util.stream.Collectors;
import static java.util.Collections.emptyList;
import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
-import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
import static org.apache.paimon.manifest.ManifestEntry.nullableRecordCount;
import static org.apache.paimon.manifest.ManifestEntry.recordCountAdd;
import static org.apache.paimon.manifest.ManifestEntry.recordCountDelete;
@@ -119,7 +118,7 @@ import static
org.apache.paimon.operation.commit.ManifestEntryChanges.changedPar
import static
org.apache.paimon.operation.commit.RowTrackingCommitUtils.assignRowTracking;
import static
org.apache.paimon.partition.PartitionPredicate.createBinaryPartitions;
import static
org.apache.paimon.partition.PartitionPredicate.createPartitionPredicate;
-import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static
org.apache.paimon.utils.DataEvolutionUtils.materializedRowIdRanges;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -845,16 +844,12 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
Set<Pair<BinaryRow, Integer>> result = new HashSet<>();
for (CommitMessage message : commitMessages) {
CommitMessageImpl commitMessage = (CommitMessageImpl) message;
- if (commitMessage.compactIncrement().compactBefore().stream()
- .noneMatch(
- file ->
- !isBlobFile(file.fileName())
- &&
!isVectorStoreFile(file.fileName()))
- || commitMessage.compactIncrement().compactAfter().stream()
- .anyMatch(file -> file.firstRowId() != null)) {
- continue;
+ if (!materializedRowIdRanges(
+ commitMessage.compactIncrement().compactBefore(),
+ commitMessage.compactIncrement().compactAfter())
+ .isEmpty()) {
+ result.add(Pair.of(commitMessage.partition(),
commitMessage.bucket()));
}
- result.add(Pair.of(commitMessage.partition(),
commitMessage.bucket()));
}
return result;
}
@@ -1172,6 +1167,14 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
.map(entry -> entry.file().fileName())
.collect(Collectors.toSet())
: Collections.emptySet();
+ Set<String> rowIdPreservingOutputs =
+ commitKind == CommitKind.COMPACT &&
options.dataEvolutionEnabled()
+ ? deltaFiles.stream()
+ .filter(entry -> entry.kind() == FileKind.ADD)
+ .filter(entry -> entry.file().firstRowId() !=
null)
+ .map(entry -> entry.file().fileName())
+ .collect(Collectors.toSet())
+ : Collections.emptySet();
Set<Long> rebasedReassignments =
changesProvider.rebasedReassignments();
long newSnapshotId = Snapshot.FIRST_SNAPSHOT_ID;
@@ -1359,6 +1362,7 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
deltaFiles,
indexFiles,
mappedClusteringOutputs,
+ rowIdPreservingOutputs,
newSnapshotId,
clusteringIndexOutputs);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
index b3a7c82fae..344c9e3892 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
@@ -63,6 +63,7 @@ import java.util.stream.Collectors;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
import static
org.apache.paimon.operation.commit.ManifestEntryChanges.changedPartitions;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static
org.apache.paimon.utils.DataEvolutionUtils.materializedRowIdRanges;
import static org.apache.paimon.utils.Preconditions.checkState;
/** Conflict detection for Data Evolution tables. */
@@ -566,19 +567,19 @@ public class DataEvolutionConflictDetection extends
ConflictDetection {
SchemaManager schemaManager,
List<ManifestEntry> deltaFiles,
boolean nestedFieldEnabled) {
- // Materializing deletion vectors rewrites complete row ranges. A
concurrent ADD in a
- // deleted normal-file range can otherwise restore logically
deleted rows.
- List<DataFileMeta> deletedNormalFiles =
+ // Only materialized ranges require exclusive access.
Row-ID-preserving compaction
+ // keeps column sequences and can coexist with later
partial-column writes.
+ List<DataFileMeta> before =
deltaFiles.stream()
.filter(entry -> entry.kind() == FileKind.DELETE)
.map(ManifestEntry::file)
- .filter(file -> file.firstRowId() != null)
- .filter(
- file ->
- !isBlobFile(file.fileName())
- &&
!isVectorStoreFile(file.fileName()))
.collect(Collectors.toList());
- return RowIdRangeConflictChecker.fromDataFiles(deletedNormalFiles);
+ List<DataFileMeta> after =
+ deltaFiles.stream()
+ .filter(entry -> entry.kind() == FileKind.ADD)
+ .map(ManifestEntry::file)
+ .collect(Collectors.toList());
+ return
RowIdRangeConflictChecker.fromRanges(materializedRowIdRanges(before, after));
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdRangeConflictChecker.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdRangeConflictChecker.java
index cf6bea0a51..d0c65aaa5c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdRangeConflictChecker.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdRangeConflictChecker.java
@@ -19,6 +19,7 @@
package org.apache.paimon.operation.commit;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RowRangeIndex;
import java.util.List;
@@ -29,17 +30,20 @@ public class RowIdRangeConflictChecker implements
RowIdConflictChecker {
private final RowRangeIndex rowRangeIndex;
- private RowIdRangeConflictChecker(List<DataFileMeta> deltaFiles) {
- this.rowRangeIndex =
- RowRangeIndex.create(
- deltaFiles.stream()
- .filter(file -> file.firstRowId() != null)
- .map(DataFileMeta::nonNullRowIdRange)
- .collect(Collectors.toList()));
+ private RowIdRangeConflictChecker(List<Range> ranges) {
+ this.rowRangeIndex = RowRangeIndex.create(ranges);
}
public static RowIdRangeConflictChecker fromDataFiles(List<DataFileMeta>
deltaFiles) {
- return new RowIdRangeConflictChecker(deltaFiles);
+ return fromRanges(
+ deltaFiles.stream()
+ .filter(file -> file.firstRowId() != null)
+ .map(DataFileMeta::nonNullRowIdRange)
+ .collect(Collectors.toList()));
+ }
+
+ public static RowIdRangeConflictChecker fromRanges(List<Range> ranges) {
+ return new RowIdRangeConflictChecker(ranges);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
index f8a2112eb6..d38bccc754 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
@@ -331,6 +331,72 @@ public class DataEvolutionUtils {
return anchor;
}
+ /** Returns normal-file row ranges removed without row-ID-preserving
replacement files. */
+ public static List<Range> materializedRowIdRanges(
+ List<DataFileMeta> before, List<DataFileMeta> after) {
+ List<Range> retained =
+ Range.sortAndMergeOverlap(
+ after.stream()
+ .filter(file -> file.firstRowId() != null)
+ .filter(
+ file ->
+ !isBlobFile(file.fileName())
+ &&
!isVectorStoreFile(file.fileName()))
+ .map(DataFileMeta::nonNullRowIdRange)
+ .collect(Collectors.toList()));
+ List<Range> removed =
+ Range.sortAndMergeOverlap(
+ before.stream()
+ .filter(file -> file.firstRowId() != null)
+ .filter(
+ file ->
+ !isBlobFile(file.fileName())
+ &&
!isVectorStoreFile(file.fileName()))
+ .map(DataFileMeta::nonNullRowIdRange)
+ .collect(Collectors.toList()));
+ List<Range> result = new ArrayList<>();
+ int retainedIndex = 0;
+ for (Range range : removed) {
+ long cursor = range.from;
+ boolean fullyRetained = false;
+ while (retainedIndex < retained.size()) {
+ Range output = retained.get(retainedIndex);
+ if (output.to < cursor) {
+ retainedIndex++;
+ continue;
+ }
+ if (output.from > range.to) {
+ break;
+ }
+ if (output.from > cursor) {
+ addMaterializedRange(result, cursor, output.from - 1);
+ }
+ if (output.to >= range.to) {
+ fullyRetained = true;
+ break;
+ }
+ // The preceding comparison guarantees this cannot overflow at
Long.MAX_VALUE.
+ cursor = output.to + 1;
+ retainedIndex++;
+ }
+ if (!fullyRetained) {
+ addMaterializedRange(result, cursor, range.to);
+ }
+ }
+ return result;
+ }
+
+ private static void addMaterializedRange(List<Range> ranges, long from,
long to) {
+ if (!ranges.isEmpty()) {
+ Range previous = ranges.get(ranges.size() - 1);
+ if (previous.to != Long.MAX_VALUE && previous.to + 1 == from) {
+ ranges.set(ranges.size() - 1, new Range(previous.from, to));
+ return;
+ }
+ }
+ ranges.add(new Range(from, to));
+ }
+
/** Check files row ranges. */
public static Range checkContiguousRowRange(List<DataFileMeta> files) {
checkArgument(!files.isEmpty(), "%s should not be empty.", "Data
evolution compact files");
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/cluster/DataEvolutionClusterCompactionTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/cluster/DataEvolutionClusterCompactionTest.java
new file mode 100644
index 0000000000..5129266344
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/append/cluster/DataEvolutionClusterCompactionTest.java
@@ -0,0 +1,277 @@
+/*
+ * 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.Snapshot;
+import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
+import org.apache.paimon.append.dataevolution.DataEvolutionNormalCompactTask;
+import org.apache.paimon.compact.CompactUnit;
+import org.apache.paimon.data.BinaryRow;
+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.partition.PartitionPredicate;
+import org.apache.paimon.schema.FileSystemSchemaManager;
+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.source.DataSplit;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.stats.SimpleStats.EMPTY_STATS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Compaction fallback for accumulated data-evolution column versions. */
+class DataEvolutionClusterCompactionTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testThresholdIncludesEveryVersionAndPreservesEachGroup() throws
Exception {
+ FileStoreTable table = createTable();
+ List<DataFileMeta> files = versions(table, 0, 5);
+ files.addAll(versions(table, 3, 4));
+ List<DataEvolutionCompactTask> tasks = plan(table, files,
Collections.emptyMap());
+ assertThat(tasks).hasSize(1);
+ assertThat(tasks.get(0).compactBefore())
+ .containsExactlyInAnyOrderElementsOf(files.subList(0, 5));
+ assertThat(((DataEvolutionNormalCompactTask)
tasks.get(0)).outputLevel()).isEqualTo(2);
+
+ files.add(file(table, "fifth-second-group", 3, 30, 0, "payload"));
+ tasks = plan(table, files, Collections.emptyMap());
+ assertThat(tasks).hasSize(2);
+ assertThat(tasks)
+ .allSatisfy(
+ task -> {
+ assertThat(task.compactBefore()).hasSize(5);
+ assertThat(task.compactBefore())
+
.extracting(DataFileMeta::nonNullRowIdRange)
+
.containsOnly(task.compactBefore().get(0).nonNullRowIdRange());
+ });
+ }
+
+ @Test
+ void testSingleFileIsNotRewrittenEvenWithLowThreshold() throws Exception {
+ FileStoreTable table =
+ createTable()
+ .copy(
+ Collections.singletonMap(
+
CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "1"));
+ assertThat(plan(table, versions(table, 0, 1),
Collections.emptyMap())).isEmpty();
+ }
+
+ @Test
+ void testClusteringSelectedGroupIsNotCompactedAgain() throws Exception {
+ FileStoreTable table = createTable();
+ List<DataFileMeta> files = versions(table, 0, 5);
+ List<DataFileMeta> secondGroup = versions(table, 3, 5);
+ files.addAll(secondGroup);
+ Map<BinaryRow, CompactUnit> clustering =
+ Collections.singletonMap(
+ BinaryRow.EMPTY_ROW, CompactUnit.fromFiles(2,
files.subList(0, 5), true));
+ List<DataEvolutionCompactTask> tasks = plan(table, files, clustering);
+ assertThat(tasks).hasSize(1);
+
assertThat(tasks.get(0).compactBefore()).containsExactlyInAnyOrderElementsOf(secondGroup);
+ }
+
+ @Test
+ void testKeyUpdatesAndChangedDefinitionsDoNotInheritLevel() throws
Exception {
+ FileStoreTable table = createTable();
+ List<DataFileMeta> keyUpdated = versions(table, 0, 5);
+ keyUpdated.add(file(table, "key-update", 0, 30, 0, "id"));
+ assertUnsortedCompaction(plan(table, keyUpdated,
Collections.emptyMap()), keyUpdated);
+
+ List<DataFileMeta> versions = versions(table, 0, 5);
+ for (String[] definition : new String[][] {{"payload", "order"},
{"id", "zorder"}}) {
+ FileStoreTable changed =
+ table.copy(
+ table.schemaManager()
+ .commitChanges(
+ Arrays.asList(
+ SchemaChange.setOption(
+
CoreOptions.CLUSTERING_COLUMNS.key(),
+ definition[0]),
+ SchemaChange.setOption(
+
CoreOptions.CLUSTERING_STRATEGY.key(),
+ definition[1]))));
+ assertUnsortedCompaction(plan(changed, versions,
Collections.emptyMap()), versions);
+ }
+ FileStoreTable newKey =
+ table.copy(
+ table.schemaManager()
+ .commitChanges(
+ Arrays.asList(
+
SchemaChange.addColumn("new_key", DataTypes.INT()),
+ SchemaChange.setOption(
+
CoreOptions.CLUSTERING_COLUMNS.key(),
+ "new_key"))));
+ assertUnsortedCompaction(plan(newKey, versions,
Collections.emptyMap()), versions);
+ }
+
+ @Test
+ void testUnsortedOrIncompleteAnchorDoesNotInheritLevel() throws Exception {
+ FileStoreTable table = createTable();
+ List<DataFileMeta> unsorted = versions(table, 0, 5);
+ unsorted.set(0, file(table, "unsorted-anchor", 0, 10, 0, null));
+ assertUnsortedCompaction(plan(table, unsorted,
Collections.emptyMap()), unsorted);
+
+ List<DataFileMeta> incomplete = versions(table, 0, 5);
+ incomplete.set(0, file(table, "partial-anchor", 0, 10, 2, "payload"));
+ assertUnsortedCompaction(plan(table, incomplete,
Collections.emptyMap()), incomplete);
+ }
+
+ @Test
+ void testSingleDirtyRunSkippedByMinorStillCompactsColumnVersions() throws
Exception {
+ FileStoreTable table = createTable();
+ List<DataFileMeta> files = versions(table, 0, 5);
+ files.set(4, file(table, "key-update", 0, 30, 0, "id"));
+ IncrementalClusterStrategy strategy =
+ new IncrementalClusterStrategy(
+ table.schemaManager(),
table.coreOptions().clusteringColumns(), 25, 1, 2);
+ assertThat(
+ strategy.pick(
+ 3,
+ new
DataEvolutionIncrementalClusterRewriter(table, null)
+ .constructPartitionLevels(files),
+ false))
+ .isEmpty();
+
+ assertUnsortedCompaction(plan(table, files, Collections.emptyMap()),
files);
+ }
+
+ private static void assertUnsortedCompaction(
+ List<DataEvolutionCompactTask> tasks, List<DataFileMeta> input) {
+ assertThat(tasks).hasSize(1);
+
assertThat(tasks.get(0).compactBefore()).containsExactlyInAnyOrderElementsOf(input);
+ assertThat(((DataEvolutionNormalCompactTask)
tasks.get(0)).outputLevel()).isZero();
+ }
+
+ private List<DataEvolutionCompactTask> plan(
+ FileStoreTable table,
+ List<DataFileMeta> files,
+ Map<BinaryRow, CompactUnit> clustering) {
+ FileStoreTable input = spy(table);
+ Snapshot snapshot = mock(Snapshot.class);
+ PartitionPredicate predicate = mock(PartitionPredicate.class);
+ SnapshotReader reader = mock(SnapshotReader.class);
+ SnapshotReader.Plan plan = mock(SnapshotReader.Plan.class);
+ doReturn(reader).when(input).newSnapshotReader();
+ when(reader.dropStats()).thenReturn(reader);
+ when(reader.withSnapshot(snapshot)).thenReturn(reader);
+ when(reader.withPartitionFilter(predicate)).thenReturn(reader);
+ when(reader.read()).thenReturn(plan);
+ // Scan splits may cut between physical column versions. Planning must
regroup them.
+ List<DataSplit> splits = new ArrayList<>();
+ for (DataFileMeta file : files) {
+ splits.add(
+ DataSplit.builder()
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath("bucket-0")
+ .withDataFiles(Collections.singletonList(file))
+ .build());
+ }
+ when(plan.dataSplits()).thenReturn(splits);
+ List<DataEvolutionCompactTask> result =
+ DataEvolutionClusterCompaction.plan(input, snapshot,
predicate, clustering);
+ verify(reader).withSnapshot(snapshot);
+ verify(reader).withPartitionFilter(predicate);
+ return result;
+ }
+
+ private FileStoreTable createTable() throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("payload", DataTypes.STRING())
+ .option(CoreOptions.BUCKET.key(), "-1")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.CLUSTERING_INCREMENTAL.key(),
"true")
+ .option(CoreOptions.CLUSTERING_COLUMNS.key(), "id")
+ .option(CoreOptions.NUM_LEVELS.key(), "3")
+ .option(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "5")
+ .build();
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(tempDir.toString());
+ SchemaManager schemaManager = new FileSystemSchemaManager(fileIO,
path);
+ return FileStoreTableFactory.create(fileIO, path,
schemaManager.createTable(schema));
+ }
+
+ private static List<DataFileMeta> versions(FileStoreTable table, long
firstRowId, int count) {
+ List<DataFileMeta> files = new ArrayList<>();
+ files.add(file(table, "base-" + firstRowId, firstRowId, 10, 2, null));
+ for (int i = 1; i < count; i++) {
+ files.add(
+ file(
+ table,
+ "update-" + firstRowId + "-" + i,
+ firstRowId,
+ 10 + i,
+ 0,
+ "payload"));
+ }
+ return files;
+ }
+
+ private static DataFileMeta file(
+ FileStoreTable table,
+ String name,
+ long firstRowId,
+ long sequence,
+ int level,
+ String writeColumn) {
+ DataFileMeta file =
+ DataFileMeta.forAppend(
+ name + ".parquet",
+ 100,
+ 3,
+ EMPTY_STATS,
+ sequence,
+ sequence,
+ table.schema().id(),
+ Collections.emptyList(),
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ firstRowId,
+ writeColumn == null ? null :
Collections.singletonList(writeColumn));
+ return level == 0 ? file : file.upgrade(level);
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
index 1a0d1da601..adb355dfc9 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
@@ -1179,6 +1179,420 @@ class DataEvolutionClusterTest extends TableTestBase {
.containsExactly("a|10|base-10|3", "a|20|base-20|4",
"a|30|base-30|5");
}
+ @ParameterizedTest
+ @CsvSource({"btree,false", "btree,true", "bitmap,false", "bitmap,true"})
+ void testMixedClusteringAndCompactionPreservesUnchangedIndexSegments(
+ String indexType, boolean deletionVectors) throws Exception {
+ FileStoreTable table = createMixedMaintenanceTable(deletionVectors);
+ commit(table, prepareMixedIndexes(table, table.latestSnapshot().get(),
"id", indexType));
+ List<String> retainedIndexes = globalIndexNames(table, new Range(3,
5));
+ List<String> remappedIndexes = globalIndexNames(table, new Range(6,
8));
+ assertThat(retainedIndexes).isNotEmpty();
+ assertThat(remappedIndexes).isNotEmpty();
+ if (deletionVectors) {
+ delete(table, "a", new Range(3, 5), 4L);
+ delete(table, "a", new Range(6, 8), 7L);
+ }
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareMixedMaintenance(table,
snapshot);
+ commitCluster(table, snapshot, messages);
+
+ assertThat(globalIndexNames(table))
+ .containsAll(retainedIndexes)
+ .doesNotContainAnyElementsOf(remappedIndexes);
+ assertThat(globalIndexNames(table, new Range(3,
5))).isEqualTo(retainedIndexes);
+ assertThat(filesInRange(table, new Range(3, 5))).hasSize(1);
+ assertThat(readRows(table))
+ .containsExactlyElementsOf(
+ deletionVectors
+ ? Arrays.asList(
+ "a|40|updated-40|3",
+ "a|60|updated-60|5",
+ "a|20|base-20|9",
+ "a|30|base-30|10")
+ : Arrays.asList(
+ "a|40|updated-40|3",
+ "a|50|updated-50|4",
+ "a|60|updated-60|5",
+ "a|10|base-10|9",
+ "a|20|base-20|10",
+ "a|30|base-30|11"));
+ int id = table.rowType().getFieldIndex("id");
+ PredicateBuilder predicates = new PredicateBuilder(table.rowType());
+ assertIndexRows(table, predicates.equal(id, 40), new Range(3, 3));
+ assertIndexRows(
+ table,
+ predicates.equal(id, 20),
+ new Range(deletionVectors ? 9 : 10, deletionVectors ? 9 : 10));
+
assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(deletionVectors
? 11L : 12L);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"btree", "bitmap"})
+ void
testFullyDeletedClusteringAndCompactionRetainsOtherIndexSegments(String
indexType)
+ throws Exception {
+ FileStoreTable table = createMixedMaintenanceTable(true);
+ commit(table, prepareMixedIndexes(table, table.latestSnapshot().get(),
"id", indexType));
+ List<String> retainedIndexes = globalIndexNames(table, new Range(3,
5));
+ delete(table, "a", new Range(3, 5), 4L);
+ delete(table, "a", new Range(6, 8), 6L, 7L, 8L);
+ Snapshot snapshot = table.latestSnapshot().get();
+ commitCluster(table, snapshot, prepareMixedMaintenance(table,
snapshot));
+
+ assertThat(globalIndexNames(table)).isEqualTo(retainedIndexes);
+ assertThat(readRows(table)).containsExactly("a|40|updated-40|3",
"a|60|updated-60|5");
+ assertThat(filesInRange(table, new Range(3, 5))).hasSize(1);
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(9L);
+ PredicateBuilder predicates = new PredicateBuilder(table.rowType());
+ assertIndexRows(
+ table, predicates.equal(table.rowType().getFieldIndex("id"),
40), new Range(3, 3));
+ assertIndexRows(table,
predicates.equal(table.rowType().getFieldIndex("id"), 20));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testNormalCompactionDoesNotInvalidateIndexes(boolean deletionVectors)
throws Exception {
+ FileStoreTable table = createMixedMaintenanceTable(deletionVectors);
+ if (deletionVectors) {
+ delete(table, "a", new Range(3, 5), 4L);
+ }
+ commit(table, prepareGlobalIndex(table, table.latestSnapshot().get(),
"id"));
+ List<String> indexes = globalIndexNames(table);
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<String> rows = readRows(table);
+ commitCluster(
+ table,
+ snapshot,
+ Collections.singletonList(prepareNormalCompaction(table, new
Range(3, 5))));
+
+ assertThat(globalIndexNames(table)).isEqualTo(indexes);
+ assertThat(readRows(table)).isEqualTo(rows);
+ assertThat(filesInRange(table, new Range(3, 5))).hasSize(1);
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(9L);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"btree", "bitmap"})
+ void
testMixedMaintenanceSplitsIndexSpanningRetainedAndReassignedRows(String
indexType)
+ throws Exception {
+ FileStoreTable table = createMixedMaintenanceTable(false);
+ commit(
+ table,
+ prepareGlobalIndex(table, table.latestSnapshot().get(), "id",
indexType, null));
+ assertThat(globalIndexNames(table, new Range(3, 8))).isNotEmpty();
+ Snapshot snapshot = table.latestSnapshot().get();
+ commitCluster(table, snapshot, prepareMixedMaintenance(table,
snapshot));
+
+ assertThat(
+
table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry ->
entry.indexFile().globalIndexMeta() != null)
+ .map(entry ->
entry.indexFile().globalIndexMeta().rowRange()))
+ .containsExactlyInAnyOrder(new Range(3, 5), new Range(9, 11));
+ PredicateBuilder predicates = new PredicateBuilder(table.rowType());
+ int id = table.rowType().getFieldIndex("id");
+ assertIndexRows(table, predicates.equal(id, 40), new Range(3, 3));
+ assertIndexRows(table, predicates.equal(id, 20), new Range(10, 10));
+ assertIndexRows(table, predicates.notEqual(id, 40), new Range(4, 5),
new Range(9, 11));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"btree,false", "btree,true", "bitmap,false", "bitmap,true"})
+ void testMixedMaintenanceRetryRemapsConcurrentIndexAfterAppend(
+ String indexType, boolean deletionVectors) throws Exception {
+ Map<String, String> retryOptions = new HashMap<>();
+ retryOptions.put(CoreOptions.COMMIT_MIN_RETRY_WAIT.key(), "0ms");
+ retryOptions.put(CoreOptions.COMMIT_MAX_RETRY_WAIT.key(), "0ms");
+ FileStoreTable original =
createMixedMaintenanceTable(deletionVectors).copy(retryOptions);
+ if (deletionVectors) {
+ delete(original, "a", new Range(3, 5), 4L);
+ delete(original, "a", new Range(6, 8), 7L);
+ }
+ commit(
+ original,
+ prepareMixedIndexes(original, original.latestSnapshot().get(),
"id", indexType));
+ Snapshot scanSnapshot = original.latestSnapshot().get();
+ List<CommitMessage> messages = prepareMixedMaintenance(original,
scanSnapshot);
+ List<String> retainedIndexes = new
ArrayList<>(globalIndexNames(original, new Range(3, 5)));
+ AtomicInteger attempts = new AtomicInteger();
+ CatalogEnvironment environment =
+ new CatalogEnvironment(null, null, null, null, null, null,
false, false) {
+ @Override
+ public SnapshotCommit snapshotCommit(
+ SnapshotManager manager, CoreOptions options,
String commitUser) {
+ SnapshotCommit delegate = new
RenamingSnapshotCommit(manager, Lock.empty());
+ return new SnapshotCommit(Lock.empty()) {
+ @Override
+ protected boolean commitImpl(
+ String baseUuid,
+ Snapshot snapshot,
+ String branch,
+ List<PartitionStatistics> statistics)
+ throws Exception {
+ int attempt = attempts.incrementAndGet();
+ if (attempt == 1) {
+ DataEvolutionClusterTest.this.commit(
+ original,
+ prepareMixedIndexes(
+ original, scanSnapshot,
"payload", indexType));
+ retainedIndexes.clear();
+ retainedIndexes.addAll(
+ globalIndexNames(original, new
Range(3, 5)));
+ append(original, "a", 90, 70, 80);
+ }
+ boolean committed =
+ delegate.commit(baseUuid, snapshot,
branch, statistics);
+ assertThat(committed).isEqualTo(attempt == 2);
+ return committed;
+ }
+
+ @Override
+ protected void closeImpl() throws Exception {
+ delegate.close();
+ }
+ };
+ }
+ };
+ FileStoreTable committingTable =
+ FileStoreTableFactory.create(
+ original.fileIO(), original.location(),
original.schema(), environment);
+ commitCluster(committingTable, scanSnapshot, messages);
+
+ assertThat(attempts).hasValue(2);
+
assertThat(original.latestSnapshot().get().id()).isEqualTo(scanSnapshot.id() +
3);
+ assertThat(globalIndexNames(original, new Range(3,
5))).isEqualTo(retainedIndexes);
+ assertThat(original.latestSnapshot().get().nextRowId())
+ .isEqualTo(deletionVectors ? 14L : 15L);
+ PredicateBuilder predicates = new PredicateBuilder(original.rowType());
+ int id = original.rowType().getFieldIndex("id");
+ assertIndexRows(original, predicates.equal(id, 40), new Range(3, 3));
+ assertIndexRows(
+ original,
+ predicates.equal(id, 20),
+ new Range(deletionVectors ? 12 : 13, deletionVectors ? 12 :
13));
+ assertIndexRows(
+ original,
+ predicates.equal(
+ original.rowType().getFieldIndex("payload"),
+ BinaryString.fromString("updated-40")),
+ new Range(3, 3));
+ assertIndexRows(
+ original,
+ predicates.equal(
+ original.rowType().getFieldIndex("payload"),
+ BinaryString.fromString("base-20")),
+ new Range(deletionVectors ? 12 : 13, deletionVectors ? 12 :
13));
+ assertThat(readRows(original))
+ .contains("a|90|base-90|9", "a|70|base-70|10",
"a|80|base-80|11");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void
testMixedMaintenanceAllowsConcurrentPartialUpdateOfRetainedRows(boolean
deletionVectors)
+ throws Exception {
+ FileStoreTable table = createMixedMaintenanceTable(deletionVectors);
+ commit(table, prepareGlobalIndex(table, table.latestSnapshot().get(),
"id"));
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareMixedMaintenance(table,
snapshot);
+ updatePayload(table, "a", 3L, 400, 500, 600);
+ commitCluster(table, snapshot, messages);
+
+ assertThat(readRows(table))
+ .containsExactly(
+ "a|40|updated-400|3",
+ "a|50|updated-500|4",
+ "a|60|updated-600|5",
+ "a|10|base-10|9",
+ "a|20|base-20|10",
+ "a|30|base-30|11");
+ assertThat(filesInRange(table, new Range(3, 5))).hasSize(2);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void
testMixedMaintenanceRejectsConcurrentPartialUpdateOfReassignedRows(boolean
deletionVectors)
+ throws Exception {
+ FileStoreTable table = createMixedMaintenanceTable(deletionVectors);
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareMixedMaintenance(table,
snapshot);
+ updatePayload(table, "a", 6L, 300, 100, 200);
+ long updatedSnapshot = table.latestSnapshot().get().id();
+ assertThatThrownBy(() -> commitCluster(table, snapshot, messages))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining(DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE);
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(updatedSnapshot);
+ assertThat(readRows(table))
+ .contains("a|30|updated-300|6", "a|10|updated-100|7",
"a|20|updated-200|8");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testMixedMaintenanceAllowsOldIndexOnlyForRetainedRows(boolean
deletionVectors)
+ throws Exception {
+ FileStoreTable table = createMixedMaintenanceTable(deletionVectors);
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> retainedIndex =
+ prepareGlobalIndex(table, snapshot, "id", "bitmap", new
Range(3, 5));
+ List<CommitMessage> reassignedIndex =
+ prepareGlobalIndex(table, snapshot, "id", "bitmap", new
Range(6, 8));
+ commitCluster(table, snapshot, prepareMixedMaintenance(table,
snapshot));
+ commit(table, retainedIndex);
+ List<String> indexes = globalIndexNames(table);
+ long committedSnapshot = table.latestSnapshot().get().id();
+ assertThatThrownBy(() -> commit(table, reassignedIndex))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Global index row ID existence
conflict");
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(committedSnapshot);
+ assertThat(globalIndexNames(table)).isEqualTo(indexes);
+ assertIndexRows(
+ table,
+ new PredicateBuilder(table.rowType())
+ .equal(table.rowType().getFieldIndex("id"), 40),
+ new Range(3, 3));
+ }
+
+ @ParameterizedTest
+ @ValueSource(longs = {3L, 6L})
+ void testMixedMaintenanceRejectsConcurrentDeletionVectorChange(long
firstRowId)
+ throws Exception {
+ FileStoreTable table = createMixedMaintenanceTable(true);
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareMixedMaintenance(table,
snapshot);
+ delete(table, "a", new Range(firstRowId, firstRowId + 2), firstRowId +
1);
+ List<String> deletedRows = readRows(table);
+ long deletedSnapshot = table.latestSnapshot().get().id();
+ assertThatThrownBy(() -> commitCluster(table, snapshot, messages))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("File deletion conflicts detected!");
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(deletedSnapshot);
+ assertThat(readRows(table)).isEqualTo(deletedRows);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testMixedMaintenanceAllowsPreviouslyPreparedUpdateOnlyForRetainedRows(
+ boolean deletionVectors) throws Exception {
+ FileStoreTable table = createMixedMaintenanceTable(deletionVectors);
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> retainedUpdate =
+ preparePayloadUpdate(table, "a", 3L, 400, 500, 600).stream()
+ .map(
+ message ->
+ ((CommitMessageImpl) message)
+
.withCheckFromSnapshot(snapshot.id()))
+ .collect(Collectors.toList());
+ List<CommitMessage> reassignedUpdate =
+ preparePayloadUpdate(table, "a", 6L, 300, 100, 200).stream()
+ .map(
+ message ->
+ ((CommitMessageImpl) message)
+
.withCheckFromSnapshot(snapshot.id()))
+ .collect(Collectors.toList());
+ commitCluster(table, snapshot, prepareMixedMaintenance(table,
snapshot));
+ commit(table, retainedUpdate);
+ long committedSnapshot = table.latestSnapshot().get().id();
+ assertThatThrownBy(() -> commit(table, reassignedUpdate))
+ .isInstanceOf(RowIdExistenceConflictException.class)
+ .hasMessageContaining("Row ID existence conflict");
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(committedSnapshot);
+ assertThat(readRows(table))
+ .contains("a|40|updated-400|3", "a|50|updated-500|4",
"a|60|updated-600|5");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"btree", "bitmap"})
+ void
testMixedMaintenanceWithoutMappingOnlyInvalidatesReassignedIndexRange(String
indexType)
+ throws Exception {
+ FileStoreTable table = createMixedMaintenanceTable(false);
+ commit(table, prepareMixedIndexes(table, table.latestSnapshot().get(),
"id", indexType));
+ List<String> retainedIndexes = globalIndexNames(table, new Range(3,
5));
+ Snapshot snapshot = table.latestSnapshot().get();
+ commitCluster(table, snapshot, prepareMixedMaintenance(table,
snapshot, false));
+
+ assertThat(globalIndexNames(table)).isEqualTo(retainedIndexes);
+ assertThat(readRows(table))
+ .containsExactly(
+ "a|40|updated-40|3",
+ "a|50|updated-50|4",
+ "a|60|updated-60|5",
+ "a|10|base-10|9",
+ "a|20|base-20|10",
+ "a|30|base-30|11");
+ assertIndexRows(
+ table,
+ new PredicateBuilder(table.rowType())
+ .equal(table.rowType().getFieldIndex("id"), 40),
+ new Range(3, 3));
+ }
+
+ private FileStoreTable createMixedMaintenanceTable(boolean
deletionVectors) throws Exception {
+ FileStoreTable table = createTable(deletionVectors);
+ append(table, "a", 60, 40, 50);
+ Snapshot initial = table.latestSnapshot().get();
+ commitCluster(table, initial, prepareMappedCluster(table, initial,
true));
+ for (int i = 0; i < 4; i++) {
+ updatePayload(table, "a", 3L, 40, 50, 60);
+ }
+ append(table, "a", 30, 10, 20);
+ assertThat(filesInRange(table, new Range(3, 5))).hasSize(5);
+ return table;
+ }
+
+ private List<CommitMessage> prepareMixedMaintenance(FileStoreTable table,
Snapshot snapshot)
+ throws Exception {
+ return prepareMixedMaintenance(table, snapshot, true);
+ }
+
+ private List<CommitMessage> prepareMixedMaintenance(
+ FileStoreTable table, Snapshot snapshot, boolean preserveIndexes)
throws Exception {
+ List<DataFileMeta> selected =
+
table.store().newScan().withSnapshot(snapshot).plan().files().stream()
+ .map(entry -> entry.file())
+ .filter(file -> file.nonNullRowIdRange().equals(new
Range(6, 8)))
+ .collect(Collectors.toList());
+ assertThat(selected).isNotEmpty();
+ IncrementalClusterManager manager = new
IncrementalClusterManager(table, null, snapshot);
+ List<CommitMessage> messages =
+ rewriteClusterUnits(
+ table,
+ manager,
+ Collections.singletonMap(
+ partition(table, "a"),
CompactUnit.fromFiles(3, selected, true)),
+ preserveIndexes);
+ messages.add(prepareNormalCompaction(table, new Range(3, 5)));
+ return messages;
+ }
+
+ private CommitMessage prepareNormalCompaction(FileStoreTable table, Range
range)
+ throws Exception {
+ List<DataFileMeta> files =
+
table.store().newScan().withSnapshot(table.latestSnapshot().get()).plan().files()
+ .stream()
+ .map(entry -> entry.file())
+ .filter(file -> file.nonNullRowIdRange().equals(range))
+ .collect(Collectors.toList());
+ return new DataEvolutionNormalCompactTask(partition(table, "a"), files)
+ .doCompact(table, "test-normal-compact");
+ }
+
+ private List<String> globalIndexNames(FileStoreTable table, Range range) {
+ return table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry -> entry.indexFile().globalIndexMeta() != null)
+ .filter(entry ->
entry.indexFile().globalIndexMeta().rowRange().equals(range))
+ .map(entry -> entry.indexFile().fileName())
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ private List<CommitMessage> prepareMixedIndexes(
+ FileStoreTable table, Snapshot snapshot, String column, String
indexType)
+ throws Exception {
+ List<CommitMessage> indexes =
+ new ArrayList<>(
+ prepareGlobalIndex(table, snapshot, column, indexType,
new Range(3, 5)));
+ indexes.addAll(prepareGlobalIndex(table, snapshot, column, indexType,
new Range(6, 8)));
+ return indexes;
+ }
+
@ParameterizedTest
@ValueSource(booleans = {false, true})
void testClusterRetryDropsIndexCommittedBeforeSnapshotPublication(boolean
deletionVectors)
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializerTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializerTest.java
new file mode 100644
index 0000000000..98f273969f
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializerTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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.dataevolution;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFileMetaSerializer;
+import org.apache.paimon.io.DataInputDeserializer;
+import org.apache.paimon.io.DataOutputSerializer;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.utils.Range;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.stats.SimpleStats.EMPTY_STATS;
+import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Task serialization keeps clustering provenance without changing ordinary
compaction defaults. */
+class DataEvolutionCompactTaskSerializerTest {
+
+ @Test
+ void testOutputLevelsSurviveMixedTaskList() throws Exception {
+ DataEvolutionCompactTaskSerializer serializer = new
DataEvolutionCompactTaskSerializer();
+ List<DataEvolutionCompactTask> tasks =
+ Arrays.asList(
+ new DataEvolutionNormalCompactTask(
+ BinaryRow.EMPTY_ROW,
+ files(),
+ Collections.singletonList(new Range(0, 2)),
+ 2),
+ new
DataEvolutionNormalCompactTask(BinaryRow.EMPTY_ROW, files()),
+ new DataEvolutionBlobCompactTask(BinaryRow.EMPTY_ROW,
files()));
+ DataOutputSerializer out = new DataOutputSerializer(128);
+ serializer.serializeList(tasks, out);
+ List<DataEvolutionCompactTask> restored =
+ serializer.deserializeList(
+ serializer.getVersion(), new
DataInputDeserializer(out.getCopyOfBuffer()));
+ assertThat(restored).containsExactlyElementsOf(tasks);
+ assertThat(((DataEvolutionNormalCompactTask)
restored.get(0)).outputLevel()).isEqualTo(2);
+ assertThat(((DataEvolutionNormalCompactTask)
restored.get(1)).outputLevel()).isZero();
+ }
+
+ @Test
+ void testVersionFourDefaultsToUnsortedOutput() throws Exception {
+ DataEvolutionCompactTaskSerializer serializer = new
DataEvolutionCompactTaskSerializer();
+ DataOutputSerializer out = new DataOutputSerializer(128);
+ List<DataFileMeta> files = files();
+ serializeBinaryRow(BinaryRow.EMPTY_ROW, out);
+ new DataFileMetaSerializer().serializeList(files, out);
+ out.writeInt(DataEvolutionCompactTask.TaskType.NORMAL.code());
+ out.writeInt(1);
+ out.writeLong(0);
+ out.writeLong(2);
+ DataEvolutionNormalCompactTask restored =
+ (DataEvolutionNormalCompactTask) serializer.deserialize(4,
out.getCopyOfBuffer());
+ assertThat(restored.outputLevel()).isZero();
+ assertThat(restored.protectedRanges()).containsExactly(new Range(0,
2));
+ assertThat(restored.compactBefore()).containsExactlyElementsOf(files);
+ }
+
+ private static List<DataFileMeta> files() {
+ return Collections.singletonList(
+ DataFileMeta.forAppend(
+ "normal.parquet",
+ 100,
+ 3,
+ EMPTY_STATS,
+ 10,
+ 10,
+ 0,
+ Collections.emptyList(),
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ 0L,
+ null));
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTaskTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTaskTest.java
index 2464cbf892..22395b9d33 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTaskTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTaskTest.java
@@ -81,6 +81,57 @@ public class DataEvolutionNormalCompactTaskTest extends
TableTestBase {
.build();
}
+ @ParameterizedTest
+ @ValueSource(ints = {0, 2})
+ public void testExplicitOutputLevelPreservesRowsAndRowIds(int outputLevel)
throws Exception {
+ write();
+ FileStoreTable table = getTableDefault();
+ List<DataFileMeta> input =
+ table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .collect(Collectors.toList());
+ if (outputLevel == 0) {
+ // Ordinary compaction must still reset an already clustered input
to level zero.
+ DataEvolutionNormalCompactTask seed =
+ new DataEvolutionNormalCompactTask(
+
table.store().newScan().plan().files().get(0).partition(),
+ input,
+ Collections.emptyList(),
+ 2);
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(Collections.singletonList(seed.doCompact(table,
"seed-level")));
+ }
+ input = seed.compactAfter();
+ }
+ DataEvolutionNormalCompactTask task =
+ outputLevel == 0
+ ? new DataEvolutionNormalCompactTask(
+
table.store().newScan().plan().files().get(0).partition(), input)
+ : new DataEvolutionNormalCompactTask(
+
table.store().newScan().plan().files().get(0).partition(),
+ input,
+ Collections.emptyList(),
+ outputLevel);
+ DataEvolutionCompactTaskSerializer serializer = new
DataEvolutionCompactTaskSerializer();
+ task =
+ (DataEvolutionNormalCompactTask)
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(task));
+ CommitMessage message = task.doCompact(table,
"preserve-clustering-level");
+ assertThat(task.compactAfter()).hasSize(1);
+ assertThat(task.compactAfter().get(0).level()).isEqualTo(outputLevel);
+ assertThat(task.compactAfter().get(0).nonNullRowIdRange())
+ .isEqualTo(input.get(0).nonNullRowIdRange());
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(Collections.singletonList(message));
+ }
+ List<InternalRow> rows = read(table);
+ assertThat(rows).hasSize(ROW_COUNT);
+ for (int i = 0; i < ROW_COUNT; i++) {
+ assertThat(rows.get(i).getInt(1)).isEqualTo(i);
+ assertThat(rows.get(i).getString(2).toString()).isEqualTo("f1_" +
i);
+ }
+ }
+
@Test
public void testPropagateWriteColsSequencesAcrossCompactions() throws
Exception {
write();
@@ -130,6 +181,22 @@ public class DataEvolutionNormalCompactTaskTest extends
TableTestBase {
.isLessThan(partialUpdate.maxSequenceNumber());
}
+ @Test
+ public void testLevelZeroClusteringCompactionPreservesFieldSequences()
throws Exception {
+ write();
+ DataFileMeta compacted =
+ updateColumnsAndCompact(
+ Collections.singletonList("f1"),
+ 1,
+ CoreOptions.GlobalIndexColumnUpdateAction.THROW_ERROR,
+ true);
+ assertThat(compacted.level()).isZero();
+ assertThat(writeColSequence(compacted,
getTableDefault().rowType().getField("f0").id()))
+ .isLessThan(compacted.maxSequenceNumber());
+ assertThat(writeColSequence(compacted,
getTableDefault().rowType().getField("f1").id()))
+ .isEqualTo(compacted.maxSequenceNumber());
+ }
+
@Test
public void testOmitWriteColsSequencesUnlessUpdatesAreIgnored() throws
Exception {
write();
@@ -740,9 +807,22 @@ public class DataEvolutionNormalCompactTaskTest extends
TableTestBase {
int updateRound,
CoreOptions.GlobalIndexColumnUpdateAction updateAction)
throws Exception {
+ return updateColumnsAndCompact(columns, updateRound, updateAction,
false);
+ }
+
+ private DataFileMeta updateColumnsAndCompact(
+ List<String> columns,
+ int updateRound,
+ CoreOptions.GlobalIndexColumnUpdateAction updateAction,
+ boolean incrementalClustering)
+ throws Exception {
Map<String, String> writeOptions = new HashMap<>();
writeOptions.put(
CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION.key(),
updateAction.toString());
+ if (incrementalClustering) {
+ writeOptions.put(CoreOptions.CLUSTERING_INCREMENTAL.key(), "true");
+ writeOptions.put(CoreOptions.CLUSTERING_COLUMNS.key(), "f0");
+ }
FileStoreTable table = getTableDefault().copy(writeOptions);
List<String> writeColumns = new ArrayList<>();
writeColumns.add("dt");
diff --git
a/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
index 028c12ee44..066bace5a6 100644
---
a/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
@@ -426,6 +426,106 @@ public class DataEvolutionUtilsTest {
.isSameAs(smallerFileName);
}
+ @Test
+ public void testMaterializedRowIdRangesExcludePreservedCompactionOutputs()
{
+ List<DataFileMeta> before =
+ Arrays.asList(
+ rangedDataFile("first.parquet", 0L, 5L),
+ rangedDataFile("first-update.parquet", 0L, 5L),
+ rangedDataFile("second.parquet", 8L, 3L));
+ List<DataFileMeta> after =
+ Arrays.asList(
+ rangedDataFile("retained-first.parquet", 1L, 2L),
+ rangedDataFile("retained-second.parquet", 3L, 1L),
+ rangedDataFile("fresh.parquet", null, 4L),
+ rangedDataFile("blob-output.blob", 0L, 11L),
+ rangedDataFile("vector-output.vector.lance", 0L, 11L));
+
+ assertThat(DataEvolutionUtils.materializedRowIdRanges(before, after))
+ .containsExactly(new Range(0, 0), new Range(4, 4), new
Range(8, 10));
+ assertThat(DataEvolutionUtils.materializedRowIdRanges(before,
before)).isEmpty();
+ assertThat(DataEvolutionUtils.materializedRowIdRanges(before,
Collections.emptyList()))
+ .containsExactly(new Range(0, 4), new Range(8, 10));
+ assertThat(
+ DataEvolutionUtils.materializedRowIdRanges(
+ Arrays.asList(
+ rangedDataFile("input.blob", 0L, 5L),
+ rangedDataFile("input.vector.lance",
0L, 5L)),
+ Collections.emptyList()))
+ .isEmpty();
+ }
+
+ @Test
+ public void testMaterializedRowIdRangesWalkInterleavedGroupsAndOutputs() {
+ List<DataFileMeta> before =
+ Arrays.asList(
+ rangedDataFile("last.parquet", 20L, 5L),
+ rangedDataFile("first.parquet", 0L, 5L),
+ rangedDataFile("middle.parquet", 10L, 5L),
+ rangedDataFile("adjacent.parquet", 5L, 2L));
+ List<DataFileMeta> after =
+ Arrays.asList(
+ rangedDataFile("middle-cut.parquet", 12L, 1L),
+ rangedDataFile("spanning-gaps.parquet", 3L, 9L),
+ rangedDataFile("outside.parquet", 30L, 3L),
+ rangedDataFile("prefix.parquet", 0L, 1L),
+ rangedDataFile("suffix.parquet", 24L, 3L));
+
+ assertThat(DataEvolutionUtils.materializedRowIdRanges(before, after))
+ .containsExactly(new Range(1, 2), new Range(13, 14), new
Range(20, 23));
+ assertThat(DataEvolutionUtils.materializedRowIdRanges(before,
Collections.emptyList()))
+ .containsExactly(new Range(0, 6), new Range(10, 14), new
Range(20, 24));
+ assertThat(
+ DataEvolutionUtils.materializedRowIdRanges(
+ before,
+
Collections.singletonList(rangedDataFile("all.parquet", 0L, 100L))))
+ .isEmpty();
+
assertThat(DataEvolutionUtils.materializedRowIdRanges(Collections.emptyList(),
after))
+ .isEmpty();
+ }
+
+ @Test
+ public void testMaterializedRowIdRangesHandleMaximumRowIdWithoutOverflow()
{
+ List<DataFileMeta> before =
+ Arrays.asList(
+ rangedDataFile("last.parquet", Long.MAX_VALUE - 2, 3L),
+ rangedDataFile("last-update.parquet", Long.MAX_VALUE -
2, 3L),
+ rangedDataFile("adjacent.parquet", Long.MAX_VALUE - 3,
1L),
+ rangedDataFile("first.parquet", 0L, 1L));
+ assertThat(
+ DataEvolutionUtils.materializedRowIdRanges(
+ before,
+ Collections.singletonList(
+ rangedDataFile("suffix.parquet",
Long.MAX_VALUE - 1, 2L))))
+ .containsExactly(
+ new Range(0, 0), new Range(Long.MAX_VALUE - 3,
Long.MAX_VALUE - 2));
+ assertThat(
+ DataEvolutionUtils.materializedRowIdRanges(
+ before,
+ Collections.singletonList(
+ rangedDataFile("prefix.parquet",
Long.MAX_VALUE - 3, 3L))))
+ .containsExactly(new Range(0, 0), new Range(Long.MAX_VALUE,
Long.MAX_VALUE));
+ assertThat(DataEvolutionUtils.materializedRowIdRanges(before,
before)).isEmpty();
+ }
+
+ private static DataFileMeta rangedDataFile(String fileName, Long
firstRowId, long rowCount) {
+ return DataFileMeta.forAppend(
+ fileName,
+ 1L,
+ rowCount,
+ SimpleStats.EMPTY_STATS,
+ 1L,
+ 1L,
+ 1L,
+ Collections.emptyList(),
+ null,
+ null,
+ null,
+ null,
+ firstRowId,
+ null);
+ }
+
private static DataFileMeta dataFile(String fileName, long
maxSequenceNumber) {
return dataFile(fileName, maxSequenceNumber, Collections.emptyList());
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/IncrementalClusterCompact.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/IncrementalClusterCompact.java
index b848f2467e..6fa55b1c4a 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/IncrementalClusterCompact.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/IncrementalClusterCompact.java
@@ -20,7 +20,9 @@ package org.apache.paimon.flink.compact;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.append.cluster.DataEvolutionClusterCompaction;
import org.apache.paimon.append.cluster.IncrementalClusterManager;
+import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
import org.apache.paimon.codegen.CodeGenUtils;
import org.apache.paimon.compact.CompactUnit;
import org.apache.paimon.data.BinaryRow;
@@ -34,6 +36,8 @@ import
org.apache.paimon.flink.cluster.RewriteIncrementalClusterCommittableOpera
import org.apache.paimon.flink.sink.Committable;
import org.apache.paimon.flink.sink.CommittableTypeInfo;
import org.apache.paimon.flink.sink.DataEvolutionCommitPreparationOperator;
+import org.apache.paimon.flink.sink.DataEvolutionCompactionTaskTypeInfo;
+import org.apache.paimon.flink.sink.DataEvolutionCompactionWorkerOperator;
import org.apache.paimon.flink.sink.DataEvolutionDeletionVectorMaterializeSink;
import org.apache.paimon.flink.sink.FlinkSinkBuilder;
import org.apache.paimon.flink.sink.RowAppendTableSink;
@@ -64,6 +68,7 @@ import org.apache.flink.table.data.RowData;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
@@ -75,6 +80,7 @@ import static
org.apache.paimon.flink.FlinkConnectorOptions.SCAN_PARALLELISM;
import static
org.apache.paimon.flink.FlinkConnectorOptions.SINK_COORDINATOR_COMMIT_ENABLED;
import static
org.apache.paimon.flink.FlinkConnectorOptions.SINK_MANAGED_WRITER_BUFFER_MEMORY;
import static
org.apache.paimon.flink.FlinkConnectorOptions.SINK_USE_MANAGED_MEMORY;
+import static
org.apache.paimon.flink.FlinkConnectorOptions.UNAWARE_BUCKET_COMPACTION_PARALLELISM;
import static
org.apache.paimon.flink.utils.ManagedMemoryUtils.declareManagedMemory;
import static org.apache.paimon.table.PrimaryKeyTableUtils.addKeyNamePrefix;
@@ -91,6 +97,7 @@ public class IncrementalClusterCompact {
protected final int localSampleMagnification;
protected final Map<BinaryRow, CompactUnit> compactUnits;
protected final Map<BinaryRow, Pair<List<DataSplit>, CommitMessage>>
compactSplits;
+ private final List<DataEvolutionCompactTask> normalCompactTasks;
public IncrementalClusterCompact(
StreamExecutionEnvironment env,
@@ -126,11 +133,16 @@ public class IncrementalClusterCompact {
// non-full strategy as default for incremental clustering
this.compactUnits =
clusterManager.createCompactUnits(fullCompaction != null &&
fullCompaction);
+ this.normalCompactTasks =
+ clusterManager.snapshot() == null
+ ? Collections.emptyList()
+ : DataEvolutionClusterCompaction.plan(
+ table, clusterManager.snapshot(),
partitionPredicate, compactUnits);
this.compactSplits =
clusterManager.toSplitsAndRewriteDvFiles(compactUnits);
}
public void build() throws Exception {
- if (compactUnits.isEmpty()) {
+ if (compactUnits.isEmpty() && normalCompactTasks.isEmpty()) {
env.fromSequence(0, 0).name("Nothing to Cluster
Source").sinkTo(new DiscardingSink<>());
return;
}
@@ -145,6 +157,24 @@ public class IncrementalClusterCompact {
entry.getValue().getRight(),
parallelism));
}
+ if (!normalCompactTasks.isEmpty()) {
+ Integer workerParallelism =
+
Options.fromMap(table.options()).get(UNAWARE_BUCKET_COMPACTION_PARALLELISM);
+ dataStreams.add(
+ env.fromCollection(
+ normalCompactTasks, new
DataEvolutionCompactionTaskTypeInfo())
+ .name("Data Evolution Cluster Compaction Tasks")
+ .rebalance()
+ .transform(
+ "Compact data evolution column versions",
+ new CommittableTypeInfo(),
+ new
DataEvolutionCompactionWorkerOperator.Factory(
+ table, commitUser))
+ .setParallelism(
+ workerParallelism == null
+ ? env.getParallelism()
+ : workerParallelism));
+ }
buildCommitOperator(dataStreams);
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DataEvolutionDeleteSqlITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DataEvolutionDeleteSqlITCase.java
index 01e17a2bd9..2ba7e87e8a 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DataEvolutionDeleteSqlITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DataEvolutionDeleteSqlITCase.java
@@ -50,6 +50,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
+import java.util.Random;
import java.util.stream.Collectors;
import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
@@ -349,6 +350,204 @@ public class DataEvolutionDeleteSqlITCase extends
CatalogITCaseBase {
assertThat(scalarIndexFiles(table, indexType, "A")).isEmpty();
}
+ @ParameterizedTest
+ @CsvSource({"btree,global-sort", "bitmap,local-sort"})
+ public void testIncrementalClusteringCompactsRepeatedColumnUpdates(
+ String indexType, String mode) throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ createTable();
+ sql(
+ "ALTER TABLE T SET ('write-only' = 'true', "
+ + "'clustering.incremental' = 'true', "
+ + "'clustering.strategy' = 'order',
'clustering.columns' = 'id', "
+ + "'clustering.incremental.mode' = '"
+ + mode
+ + "', 'compaction.min.file-num' = '5', "
+ + "'scan.parallelism' = '1', 'sink.parallelism' =
'1')");
+ sql("INSERT INTO T VALUES (3, 'three', 'A'), (1, 'one', 'A'), (2,
'two', 'A')");
+ sql("CALL sys.compact(`table` => 'default.T', compact_strategy =>
'full')");
+ createIdIndex(indexType);
+ sql("DELETE FROM T WHERE id = 2");
+
+ FileStoreTable table = paimonTable("T");
+ DataFileMeta base =
table.store().newScan().plan().files().get(0).file();
+ List<String> indexes = scalarIndexFiles(table, indexType, "A");
+ List<Row> originalIds = sql("SELECT id, _ROW_ID FROM T$row_tracking
ORDER BY id");
+ for (int version = 0; version < 4; version++) {
+ writePartialGroup(
+ table,
+ "name",
+ base.nonNullFirstRowId(),
+ "one-" + version,
+ "two-" + version,
+ "three-" + version);
+ }
+ Snapshot before = table.latestSnapshot().get();
+ List<String> beforeFiles = plannedFiles(table);
+ assertThat(beforeFiles).hasSize(5);
+
+ sql("CALL sys.compact(`table` => 'default.T', compact_strategy =>
'full')");
+
+ Snapshot compacted = table.latestSnapshot().get();
+ assertThat(compacted.id()).isEqualTo(before.id() + 1);
+
assertThat(compacted.commitKind()).isEqualTo(Snapshot.CommitKind.COMPACT);
+ assertThat(compacted.nextRowId()).isEqualTo(before.nextRowId());
+ List<DataFileMeta> afterFiles =
+ table.store().newScan().plan().files().stream()
+ .map(entry -> entry.file())
+ .collect(Collectors.toList());
+ assertThat(afterFiles)
+ .singleElement()
+ .satisfies(
+ file -> {
+ assertThat(file.nonNullRowIdRange())
+ .isEqualTo(base.nonNullRowIdRange());
+ assertThat(file.level()).isEqualTo(base.level());
+ assertThat(file.rowCount()).isEqualTo(3);
+ });
+
assertThat(plannedFiles(table)).doesNotContainAnyElementsOf(beforeFiles);
+ assertThat(scalarIndexFiles(table, indexType, "A")).isEqualTo(indexes);
+ assertThat(deletionVectorCardinality(table)).isEqualTo(1L);
+ assertThat(sql("SELECT id, _ROW_ID FROM T$row_tracking ORDER BY id"))
+ .containsExactlyElementsOf(originalIds);
+ sql("ALTER TABLE T SET ('scalar-index.search-mode' = 'fast')");
+ assertThat(sql("SELECT id, name FROM T WHERE id IN (1, 2, 3) ORDER BY
id"))
+ .containsExactly(Row.of(1, "one-3"), Row.of(3, "three-3"));
+
+ sql("CALL sys.compact(`table` => 'default.T', compact_strategy =>
'full')");
+
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(compacted.id());
+
assertThat(plannedFiles(table)).containsExactly(afterFiles.get(0).fileName());
+ assertThat(scalarIndexFiles(table, indexType, "A")).isEqualTo(indexes);
+
+ // The compacted file contains all columns, but its id values still
predate the index.
+ // Retaining per-column sequences lets later clustering reuse that
valid index.
+ sql("INSERT INTO T VALUES (4, 'four', 'A')");
+ Snapshot beforeClustering = table.latestSnapshot().get();
+
+ sql("CALL sys.compact(`table` => 'default.T', compact_strategy =>
'full')");
+
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(beforeClustering.id() +
1);
+ assertThat(table.latestSnapshot().get().nextRowId())
+ .isEqualTo(beforeClustering.nextRowId() + 3);
+ assertThat(scalarIndexFiles(table, indexType, "A"))
+ .isNotEmpty()
+ .doesNotContainAnyElementsOf(indexes);
+ assertThat(sql("SELECT id, name FROM T WHERE id IN (1, 2, 3) ORDER BY
id"))
+ .containsExactly(Row.of(1, "one-3"), Row.of(3, "three-3"));
+ assertThat(sql("SELECT id, _ROW_ID FROM T$row_tracking ORDER BY id"))
+ .containsExactly(
+ Row.of(1, beforeClustering.nextRowId()),
+ Row.of(3, beforeClustering.nextRowId() + 1),
+ Row.of(4, beforeClustering.nextRowId() + 2));
+ assertThat(deletionVectorCardinality(table)).isZero();
+ }
+
+ @Test
+ public void testIncrementalClusteringCombinesCompactionInSamePartition()
throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ createTable();
+ sql(
+ "ALTER TABLE T SET ('write-only' = 'true', "
+ + "'clustering.incremental' = 'true', "
+ + "'clustering.strategy' = 'order',
'clustering.columns' = 'id', "
+ + "'clustering.incremental.mode' = 'global-sort', "
+ + "'compaction.min.file-num' = '5', "
+ + "'num-sorted-run.compaction-trigger' = '2', "
+ + "'compaction.max-size-amplification-percent' =
'10000', "
+ + "'compaction.size-ratio' = '0', 'file.compression' =
'none', "
+ + "'target-file-row-num' = '3', "
+ + "'scan.parallelism' = '1', 'sink.parallelism' =
'1')");
+ FileStoreTable table = paimonTable("T");
+ // A large clustered run must stay outside the minor clustering
candidate. Its column
+ // versions still require normal compaction in the same partition and
commit.
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int id : new int[] {3, 1, 2}) {
+ Random random = new Random(id);
+ StringBuilder name = new StringBuilder(65536);
+ for (int i = 0; i < 65536; i++) {
+ name.append((char) ('a' + random.nextInt(26)));
+ }
+ write.write(
+ GenericRow.of(
+ id,
+ BinaryString.fromString(name.toString()),
+ BinaryString.fromString("A")));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ sql("CALL sys.compact(`table` => 'default.T', compact_strategy =>
'full')");
+ createIdIndex("btree");
+ sql("DELETE FROM T WHERE id = 2");
+ DataFileMeta base =
table.store().newScan().plan().files().get(0).file();
+ List<String> retainedIndexes = scalarIndexFiles(table, "btree", "A");
+ List<Row> originalIds = sql("SELECT id, _ROW_ID FROM T$row_tracking
ORDER BY id");
+ for (int version = 0; version < 4; version++) {
+ writePartialGroup(
+ table,
+ "name",
+ base.nonNullFirstRowId(),
+ "one-" + version,
+ "two-" + version,
+ "three-" + version);
+ }
+ sql("INSERT INTO T VALUES (6, 'six', 'A'), (4, 'four', 'A'), (5,
'five', 'A')");
+ sql("INSERT INTO T VALUES (9, 'nine', 'A'), (7, 'seven', 'A'), (8,
'eight', 'A')");
+ createIdIndex("btree");
+ List<String> allIndexes = scalarIndexFiles(table, "btree", "A");
+ assertThat(allIndexes)
+ .containsAll(retainedIndexes)
+ .hasSizeGreaterThan(retainedIndexes.size());
+ List<String> rewrittenIndexes = new ArrayList<>(allIndexes);
+ rewrittenIndexes.removeAll(retainedIndexes);
+ Snapshot before = table.latestSnapshot().get();
+ List<String> beforeFiles = plannedFiles(table);
+ assertThat(beforeFiles).hasSize(7);
+
+ sql("CALL sys.compact(`table` => 'default.T', compact_strategy =>
'minor')");
+
+ Snapshot compacted = table.latestSnapshot().get();
+ assertThat(compacted.id()).isEqualTo(before.id() + 1);
+
assertThat(compacted.commitKind()).isEqualTo(Snapshot.CommitKind.COMPACT);
+ assertThat(compacted.nextRowId()).isEqualTo(before.nextRowId() + 6);
+ List<DataFileMeta> afterFiles =
+ table.store().newScan().plan().files().stream()
+ .map(entry -> entry.file())
+ .collect(Collectors.toList());
+ assertThat(afterFiles).hasSize(3);
+ assertThat(afterFiles)
+ .filteredOn(file -> file.nonNullFirstRowId() ==
base.nonNullFirstRowId())
+ .singleElement()
+ .satisfies(
+ file -> {
+ assertThat(file.level()).isEqualTo(base.level());
+ assertThat(file.nonNullRowIdRange())
+ .isEqualTo(base.nonNullRowIdRange());
+ });
+
assertThat(plannedFiles(table)).doesNotContainAnyElementsOf(beforeFiles);
+ assertThat(scalarIndexFiles(table, "btree", "A"))
+ .containsAll(retainedIndexes)
+ .doesNotContainAnyElementsOf(rewrittenIndexes);
+ assertThat(sql("SELECT id, _ROW_ID FROM T$row_tracking WHERE id <= 3
ORDER BY id"))
+ .containsExactlyElementsOf(originalIds);
+ assertThat(sql("SELECT _ROW_ID FROM T$row_tracking WHERE id >= 4"))
+ .hasSize(6)
+ .allSatisfy(
+ row ->
+ assertThat((Long) row.getField(0))
+
.isGreaterThanOrEqualTo(before.nextRowId()));
+ assertThat(deletionVectorCardinality(table)).isEqualTo(1L);
+ sql("ALTER TABLE T SET ('scalar-index.search-mode' = 'fast')");
+ assertThat(sql("SELECT id, name FROM T WHERE id IN (1, 2, 3, 4, 9)
ORDER BY id"))
+ .containsExactly(
+ Row.of(1, "one-3"),
+ Row.of(3, "three-3"),
+ Row.of(4, "four"),
+ Row.of(9, "nine"));
+ }
+
@ParameterizedTest
@ValueSource(strings = {"global-sort", "local-sort"})
public void testIncrementalClusteringPreservesInlineBlobViews(String mode)
throws Exception {
@@ -760,6 +959,14 @@ public class DataEvolutionDeleteSqlITCase extends
CatalogITCaseBase {
+ "index_column => 'name', index_type => 'btree')");
}
+ private void createIdIndex(String indexType) {
+ sql(
+ "CALL sys.create_global_index(`table` => 'default.T', "
+ + "index_column => 'id', index_type => '"
+ + indexType
+ + "')");
+ }
+
private void createTable() {
sql(
"CREATE TABLE T (id INT, name STRING, dt STRING) PARTITIONED
BY (dt) WITH ("
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
index ada67b342b..5cdac96c4b 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
@@ -23,6 +23,7 @@ import org.apache.paimon.CoreOptions.OrderType;
import org.apache.paimon.Snapshot;
import org.apache.paimon.append.AppendCompactCoordinator;
import org.apache.paimon.append.AppendCompactTask;
+import org.apache.paimon.append.cluster.DataEvolutionClusterCompaction;
import org.apache.paimon.append.cluster.IncrementalClusterManager;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
@@ -857,11 +858,15 @@ public class CompactProcedure extends BaseProcedure {
new IncrementalClusterManager(table, partitionPredicate,
snapshot);
Map<BinaryRow, CompactUnit> compactUnits =
incrementalClusterManager.createCompactUnits(fullCompaction);
+ List<DataEvolutionCompactTask> compactionTasks =
+ DataEvolutionClusterCompaction.plan(
+ table, snapshot, partitionPredicate, compactUnits);
if (partitionIdleTime != null) {
Set<BinaryRow> idlePartitions =
getPartitionsToCompact(
table.newSnapshotReader().withSnapshot(snapshot),
partitionIdleTime);
compactUnits.keySet().retainAll(idlePartitions);
+ compactionTasks.removeIf(task ->
!idlePartitions.contains(task.partition()));
}
LOG.info(
"Planned {} compact units to incrementally cluster for table
{}, full compaction {}.",
@@ -869,7 +874,7 @@ public class CompactProcedure extends BaseProcedure {
table.fullName(),
fullCompaction);
- if (compactUnits.isEmpty()) {
+ if (compactUnits.isEmpty() && compactionTasks.isEmpty()) {
return;
}
@@ -1000,10 +1005,17 @@ public class CompactProcedure extends BaseProcedure {
}
if (table.coreOptions().dataEvolutionEnabled()) {
- List<CommitMessage> preparedMessages =
-
incrementalClusterManager.prepareCommitMessages(clusterMessages);
String commitUser =
createCommitUser(table.coreOptions().toConfiguration());
try (TableCommitImpl commit =
table.newCommit(commitUser).materializeDvRowIdCheck()) {
+ clusterMessages.addAll(
+ DataEvolutionRewriteExecutor.executeTasks(
+ clusterTable,
+ compactionTasks,
+ new JavaSparkContext(spark().sparkContext()),
+ spark(),
+ commitUser));
+ List<CommitMessage> preparedMessages =
+
incrementalClusterManager.prepareCommitMessages(clusterMessages);
commit.commit(preparedMessages);
} catch (Exception e) {
throw new RuntimeException("Commit data evolution clustering
failed.", e);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DataEvolutionRewriteExecutor.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DataEvolutionRewriteExecutor.java
index a6610bfff8..cd942f9ab2 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DataEvolutionRewriteExecutor.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DataEvolutionRewriteExecutor.java
@@ -105,7 +105,6 @@ final class DataEvolutionRewriteExecutor {
CommitConfigurer commitConfigurer,
@Nullable CommitMessageRewriter commitMessageRewriter,
@Nullable Consumer<List<CommitMessage>> commitObserver) {
- CommitMessageSerializer messageSerializer = new
CommitMessageSerializer();
String commitUser =
createCommitUser(table.coreOptions().toConfiguration());
Snapshot preparationSnapshot = initialSnapshot;
int round = 0;
@@ -122,64 +121,14 @@ final class DataEvolutionRewriteExecutor {
table.fullName());
continue;
}
- boolean containsMaterializeDeletion =
- compactionTasks.stream()
- .anyMatch(
- task ->
- task.type()
- ==
DataEvolutionCompactTask.TaskType
-
.MATERIALIZE_DELETION);
-
- DataEvolutionCompactTaskSerializer serializer =
- new DataEvolutionCompactTaskSerializer();
- List<byte[]> serializedTasks = new ArrayList<>();
- try {
- for (DataEvolutionCompactTask compactionTask :
compactionTasks) {
-
serializedTasks.add(serializer.serialize(compactionTask));
- }
- } catch (IOException e) {
- throw new RuntimeException("Serialize data evolution
rewrite task failed.", e);
- }
-
- int readParallelism = readParallelism(serializedTasks,
sparkSession);
- LOG.info(
- "Starting to execute {} data evolution rewrite tasks
of table {} in round {} "
- + "with read parallelism {}, contains
materialize deletion {}.",
- serializedTasks.size(),
- table.fullName(),
- round,
- readParallelism,
- containsMaterializeDeletion);
- JavaRDD<byte[]> commitMessageJavaRDD =
- javaSparkContext
- .parallelize(serializedTasks, readParallelism)
- .mapPartitions(
- (FlatMapFunction<Iterator<byte[]>,
byte[]>)
- taskIterator -> {
-
DataEvolutionCompactTaskSerializer ser =
- new
DataEvolutionCompactTaskSerializer();
- List<byte[]> messagesBytes
= new ArrayList<>();
- CommitMessageSerializer
messageSer =
- new
CommitMessageSerializer();
- while
(taskIterator.hasNext()) {
-
DataEvolutionCompactTask task =
-
ser.deserialize(
-
ser.getVersion(),
-
taskIterator.next());
- messagesBytes.add(
-
messageSer.serialize(
-
task.doCompact(
-
table,
-
commitUser)));
- }
- return
messagesBytes.iterator();
- });
-
- List<byte[]> serializedMessages = new
ArrayList<>(commitMessageJavaRDD.collect());
try {
List<CommitMessage> compactMessages =
- deserializeCommitMessagesAndReleaseSerializedBytes(
- messageSerializer, serializedMessages);
+ executeTasks(
+ table,
+ compactionTasks,
+ javaSparkContext,
+ sparkSession,
+ commitUser);
Snapshot committedSnapshot =
commitWithMergeConflictRetry(
table,
@@ -209,6 +158,54 @@ final class DataEvolutionRewriteExecutor {
}
}
+ /** Executes one task batch without publishing it, so callers can share a
commit. */
+ static List<CommitMessage> executeTasks(
+ FileStoreTable table,
+ List<DataEvolutionCompactTask> tasks,
+ JavaSparkContext javaSparkContext,
+ SparkSession sparkSession,
+ String commitUser)
+ throws IOException {
+ if (tasks.isEmpty()) {
+ return Collections.emptyList();
+ }
+ DataEvolutionCompactTaskSerializer serializer = new
DataEvolutionCompactTaskSerializer();
+ List<byte[]> serializedTasks = new ArrayList<>(tasks.size());
+ for (DataEvolutionCompactTask task : tasks) {
+ serializedTasks.add(serializer.serialize(task));
+ }
+ int parallelism = readParallelism(serializedTasks, sparkSession);
+ LOG.info(
+ "Executing {} data evolution rewrite tasks for table {} with
parallelism {}.",
+ tasks.size(),
+ table.fullName(),
+ parallelism);
+ JavaRDD<byte[]> serializedMessages =
+ javaSparkContext
+ .parallelize(serializedTasks, parallelism)
+ .mapPartitions(
+ (FlatMapFunction<Iterator<byte[]>, byte[]>)
+ taskIterator -> {
+ DataEvolutionCompactTaskSerializer
taskSerializer =
+ new
DataEvolutionCompactTaskSerializer();
+ CommitMessageSerializer
messageSerializer =
+ new
CommitMessageSerializer();
+ List<byte[]> messages = new
ArrayList<>();
+ while (taskIterator.hasNext()) {
+ DataEvolutionCompactTask task =
+
taskSerializer.deserialize(
+
taskSerializer.getVersion(),
+
taskIterator.next());
+ messages.add(
+
messageSerializer.serialize(
+
task.doCompact(table, commitUser)));
+ }
+ return messages.iterator();
+ });
+ return deserializeCommitMessagesAndReleaseSerializedBytes(
+ new CommitMessageSerializer(), new
ArrayList<>(serializedMessages.collect()));
+ }
+
private static Snapshot commitWithMergeConflictRetry(
FileStoreTable table,
Snapshot taskSnapshot,
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
index 2df8a5936c..1cd9ed2ca6 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
@@ -2724,6 +2724,78 @@ abstract class CompactProcedureTestBase extends
PaimonSparkTestBase with StreamT
}
}
+ Seq("local-sort", "global-sort").foreach {
+ mode =>
+ test(s"Paimon Procedure: data evolution clustering compacts column
versions with $mode") {
+ withTable("T") {
+ sql(s"""
+ |CREATE TABLE T (id INT, value INT, pt STRING)
+ |PARTITIONED BY (pt)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'write-only' = 'true',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'clustering.incremental' = 'true',
+ | 'clustering.incremental.mode' = '$mode',
+ | 'clustering.columns' = 'id')
+ |""".stripMargin)
+ sql(
+ "INSERT INTO T SELECT /*+ REPARTITION(1) */ * FROM VALUES " +
+ "(3, 30, 'a'), (1, 10, 'a'), (2, 20, 'a')")
+ sql("CALL sys.compact(table => 'T', compact_strategy => 'full')")
+ val table = loadTable("T")
+ val maxLevel = table.coreOptions().numLevels() - 1
+ val originalIds = sql("SELECT id, _ROW_ID FROM T ORDER BY
id").collect().toSeq
+ (1 to 3).foreach(i => sql(s"UPDATE T SET value = ${100 + i} WHERE pt
= 'a'"))
+ val belowThreshold = lastSnapshotId(table)
+ sql("CALL sys.compact(table => 'T')")
+ assert(lastSnapshotId(table) == belowThreshold)
+ assert(normalDataFiles(table).size == 4)
+
+ sql("UPDATE T SET value = 104 WHERE pt = 'a'")
+ sql("DELETE FROM T WHERE id = 2")
+ val before = table.latestSnapshot().get()
+ assert(normalDataFiles(table).size == 5)
+ sql(
+ "CALL sys.compact(table => 'T', compact_strategy => 'full', " +
+ "partitions => 'pt=a', partition_idle_time => '0 s')")
+
+ assert(lastSnapshotId(table) == before.id() + 1)
+ assert(table.latestSnapshot().get().nextRowId() ==
before.nextRowId())
+ assert(normalDataFiles(table).size == 1)
+ assert(normalDataFiles(table).head.level() == maxLevel)
+ assert(normalDataFiles(table).head.rowCount() == 3)
+ assert(deletionVectorCardinality(table) == 1)
+ checkAnswer(sql("SELECT id, value FROM T"), Seq(Row(1, 104), Row(3,
104)))
+ checkAnswer(sql("SELECT id, _ROW_ID FROM T"),
originalIds.filter(_.getInt(0) != 2))
+ val compactedSnapshot = lastSnapshotId(table)
+ sql("CALL sys.compact(table => 'T', compact_strategy => 'full')")
+ assert(lastSnapshotId(table) == compactedSnapshot)
+
+ // The same invocation also clusters another partition, with one
shared commit.
+ (1 to 4).foreach(i => sql(s"UPDATE T SET value = ${200 + i} WHERE pt
= 'a'"))
+ sql(
+ "INSERT INTO T SELECT /*+ REPARTITION(1) */ * FROM VALUES " +
+ "(8, 80, 'b'), (6, 60, 'b')")
+ val mixed = table.latestSnapshot().get()
+ sql("CALL sys.compact(table => 'T', compact_strategy => 'full')")
+ assert(lastSnapshotId(table) == mixed.id() + 1)
+ assert(table.latestSnapshot().get().nextRowId() == mixed.nextRowId()
+ 2)
+ assert(normalDataFiles(table).size == 2)
+ assert(normalDataFiles(table).forall(_.level() == maxLevel))
+ assert(deletionVectorCardinality(table) == 1)
+ checkAnswer(
+ sql("SELECT id, _ROW_ID FROM T WHERE pt = 'a'"),
+ originalIds.filter(_.getInt(0) != 2))
+ checkAnswer(
+ sql("SELECT id, value FROM T"),
+ Seq(Row(1, 204), Row(3, 204), Row(6, 60), Row(8, 80)))
+ }
+ }
+ }
+
test("Paimon Procedure: data evolution clustering preserves inline blob
descriptors") {
withTable("cluster_descriptor") {
val blobData = "descriptor content".getBytes("UTF-8")
