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 a6e6647c5b [core] Support incremental clustering for Data Evolution
tables (#10680)
a6e6647c5b is described below
commit a6e6647c5b8417130ea11bc139c498394522986c
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 20 12:44:42 2026 +0800
[core] Support incremental clustering for Data Evolution tables (#10680)
---
docs/docs/append-table/incremental-clustering.mdx | 9 +-
.../data-evolution-maintenance.mdx | 42 +
docs/docs/multimodal-table/data-evolution.mdx | 4 +-
.../DataEvolutionIncrementalClusterRewriter.java | 155 ++++
.../append/cluster/IncrementalClusterManager.java | 6 +
.../append/cluster/IncrementalClusterRewriter.java | 4 +-
.../paimon/operation/FileStoreCommitImpl.java | 2 +-
.../org/apache/paimon/schema/SchemaValidation.java | 13 +-
.../cluster/IncrementalClusterManagerTest.java | 27 +
.../dataevolution/DataEvolutionClusterTest.java | 882 +++++++++++++++++++++
.../apache/paimon/schema/SchemaValidationTest.java | 87 ++
.../apache/paimon/flink/action/CompactAction.java | 9 +-
.../flink/compact/IncrementalClusterCompact.java | 31 +-
.../DataEvolutionCommitPreparationOperator.java | 31 +-
.../paimon/flink/DataEvolutionDeleteSqlITCase.java | 337 +++++++-
.../action/IncrementalClusterActionITCase.java | 13 +
.../paimon/spark/procedure/CompactProcedure.java | 190 +++--
.../spark/procedure/CompactProcedureTestBase.scala | 334 +++++++-
18 files changed, 2089 insertions(+), 87 deletions(-)
diff --git a/docs/docs/append-table/incremental-clustering.mdx
b/docs/docs/append-table/incremental-clustering.mdx
index 25249a8f16..49f1763712 100644
--- a/docs/docs/append-table/incremental-clustering.mdx
+++ b/docs/docs/append-table/incremental-clustering.mdx
@@ -48,8 +48,10 @@ by a query, and does not replace SQL `ORDER BY`.
| Global/local sort mode | Configurable for batch clustering jobs. |
Clustering is performed within each partition and bucket; the global/local
option does not select this path. |
| Historical-partition auto-clustering | Supported. |
`clustering.history-partition.*` does not apply. |
-Data Evolution tables cannot enable incremental clustering. If streaming
consumers require ordered append reads from a
-bucketed table, keep that ordering and do not enable clustering.
+Data Evolution tables support incremental clustering without dedicated BLOB or
VECTOR files; see
+[Data Evolution
clustering](../multimodal-table/data-evolution-maintenance#incremental-clustering)
for its execution
+and row-ID behavior. If streaming consumers require ordered append reads from
a bucketed table, keep that ordering
+and do not enable clustering.
## Enable Incremental Clustering
@@ -134,6 +136,9 @@ For unaware-bucket tables, the mode controls how the
**selected files in each pa
Here, “global” refers to the selected clustering work within a partition. It
does not imply that all existing files or
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.
+
## Run Incremental Clustering
Run explicit compact jobs in batch mode. Table options supply the clustering
columns and strategy. The examples below
diff --git a/docs/docs/multimodal-table/data-evolution-maintenance.mdx
b/docs/docs/multimodal-table/data-evolution-maintenance.mdx
index 7666899d42..682334e99b 100644
--- a/docs/docs/multimodal-table/data-evolution-maintenance.mdx
+++ b/docs/docs/multimodal-table/data-evolution-maintenance.mdx
@@ -33,6 +33,7 @@ have different effects on row IDs and dependent indexes.
| Reduce column-file fragmentation | [Ordinary
compaction](#ordinary-compaction) | Preserved |
| Resize large normal files | [Split during
compaction](#splitting-large-files-during-compaction) | Preserved |
| Remove logically deleted positions | [Materialize deletion
vectors](#materialize-deletion-vectors) | Reassigned in rewritten ranges |
+| Sort ordinary columns for pruning | [Incremental
clustering](#incremental-clustering) | Reassigned in rewritten groups |
For updates and logical deletes, see [Data Evolution](./data-evolution).
For the normal/dedicated range contract, see [File Layout and
Reads](./data-evolution-file-layout).
@@ -169,6 +170,46 @@ expiration to remove the files. The legacy
`data-evolution.compaction.rewrite-row-ids = true` setting is rejected; use the
materialization procedure instead.
+### Incremental Clustering
+
+Data Evolution tables can enable `clustering.incremental = true` with nonempty
+`clustering.columns` when they have no dedicated BLOB or VECTOR files. BLOB
+descriptor and view fields stored inline, and VECTOR columns stored in normal
+files, do not require dedicated files and remain supported.
+
+The existing [clustering strategies and file
selection](../append-table/incremental-clustering)
+choose the work. Selection includes every column version in each selected
+row-ID group. Readers merge the latest column values and apply logical deletes
+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. `clustering.incremental.mode` selects the same distributed global
sort
+or task-local sort used by ordinary append-table clustering.
+
+For example, enable clustering with Spark SQL and run an explicit batch
compact:
+
+```sql
+ALTER TABLE default.target_table SET TBLPROPERTIES (
+ 'clustering.incremental' = 'true',
+ 'clustering.columns' = 'id'
+);
+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 and drops global indexes in
+affected partitions because their row IDs no longer identify the same rows.
+Rebuild those indexes after clustering, or use `scalar-index.search-mode =
full`
+for complete scalar query results while only part of the table remains indexed.
+Concurrent updates or deletes that overlap the selected row-ID groups cause
+commit to fail, preventing stale rewrites from replacing newer values.
+Historical snapshots and tags retain
+the previous files until expiration.
+
### Row-ID Lifetime
| Operation | Effect on row IDs |
@@ -178,6 +219,7 @@ materialization procedure instead.
| 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. |
| 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/docs/docs/multimodal-table/data-evolution.mdx
b/docs/docs/multimodal-table/data-evolution.mdx
index c57b95a065..fbe2d4ac88 100644
--- a/docs/docs/multimodal-table/data-evolution.mdx
+++ b/docs/docs/multimodal-table/data-evolution.mdx
@@ -50,7 +50,9 @@ read work until compaction consolidates them.
Set both `row-tracking.enabled` and `data-evolution.enabled` when creating the
table. These options are immutable. Data Evolution requires an append table
with no primary key and `bucket = -1` (the default for an append table).
-`clustering.incremental` cannot be enabled with Data Evolution.
+[Incremental clustering](./data-evolution-maintenance#incremental-clustering)
is
+supported for tables without dedicated BLOB or VECTOR files. Clustering
rewrites
+the selected groups and assigns new row IDs.
The following examples assume a configured Paimon catalog and the `default`
database. Deletion vectors are enabled so the same table can also run the
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
new file mode 100644
index 0000000000..238c4057ea
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/DataEvolutionIncrementalClusterRewriter.java
@@ -0,0 +1,155 @@
+/*
+ * 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.DataEvolutionCompactDeletionVectorRewriter;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.deletionvectors.append.AppendDeleteFileMaintainer;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.mergetree.LevelSortedRun;
+import org.apache.paimon.mergetree.SortedRun;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.SplitGenerator;
+import org.apache.paimon.utils.RangeHelper;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+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;
+
+/** Preserves complete row-id groups and materializes deletion vectors for
clustering. */
+class DataEvolutionIncrementalClusterRewriter extends
IncrementalClusterRewriter {
+
+ private final FileStoreTable table;
+ private final Snapshot snapshot;
+
+ DataEvolutionIncrementalClusterRewriter(FileStoreTable table, Snapshot
snapshot) {
+ this.table = table;
+ this.snapshot = snapshot;
+ }
+
+ @Override
+ List<LevelSortedRun> constructPartitionLevels(List<DataFileMeta> files) {
+ checkArgument(
+ files.stream()
+ .noneMatch(
+ file ->
+ isBlobFile(file.fileName())
+ ||
isVectorStoreFile(file.fileName())),
+ "Data evolution incremental clustering does not support
dedicated BLOB/VECTOR files.");
+ RangeHelper<DataFileMeta> helper = new
RangeHelper<>(DataFileMeta::nonNullRowIdRange);
+ List<LevelSortedRun> runs = new ArrayList<>();
+ Map<Integer, List<DataFileMeta>> higherLevels = new HashMap<>();
+ for (List<DataFileMeta> group : helper.mergeOverlappingRanges(files)) {
+ checkArgument(
+ helper.areAllRangesSame(group), "Normal files must have
equal row-id ranges.");
+ // A partial update makes the complete group eligible again,
including its base.
+ int level =
group.stream().mapToInt(DataFileMeta::level).min().getAsInt();
+ if (level == 0) {
+ runs.add(new LevelSortedRun(0, SortedRun.fromSorted(group)));
+ } else {
+ higherLevels.computeIfAbsent(level, ignored -> new
ArrayList<>()).addAll(group);
+ }
+ }
+ higherLevels.forEach(
+ (level, group) -> runs.add(new LevelSortedRun(level,
SortedRun.fromSorted(group))));
+ runs.sort(Comparator.comparingInt(LevelSortedRun::level));
+ return runs;
+ }
+
+ @Override
+ void configureSplit(
+ DataSplit.Builder builder,
+ SplitGenerator.SplitGroup splitGroup,
+ @Nullable AppendDeleteFileMaintainer dvMaintainer) {
+ builder.rawConvertible(false);
+ if (dvMaintainer != null) {
+ builder.withDataDeletionFiles(
+ splitGroup.files.stream()
+ .map(file ->
dvMaintainer.getDeletionFile(file.fileName()))
+ .collect(Collectors.toList()));
+ }
+ }
+
+ @Override
+ @Nullable
+ CommitMessage finishPartition(
+ @Nullable AppendDeleteFileMaintainer dvMaintainer, int
totalBuckets) {
+ // Keep source deletion vectors intact until the rewrite commits.
+ return null;
+ }
+
+ @Override
+ public List<CommitMessage> prepareCommitMessages(List<CommitMessage>
messages) {
+ Map<BinaryRow, CompactIncrement> increments = new HashMap<>();
+ 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 merged =
+ increments.computeIfAbsent(
+ message.partition(),
+ ignored ->
+ new CompactIncrement(
+ new ArrayList<>(),
+ new ArrayList<>(),
+ 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(
+ !increment.compactBefore().isEmpty(),
+ "Data evolution clustering output must include its
source files.");
+ prepared.add(
+ new CommitMessageImpl(
+ partition, 0, null,
DataIncrement.emptyIncrement(), increment));
+ });
+ // The committer invalidates global indexes against the latest
snapshot on every attempt.
+ prepared.addAll(
+ new DataEvolutionCompactDeletionVectorRewriter(table, snapshot)
+ .rewriteDeletionVectors(prepared));
+ return prepared.stream()
+ .map(message -> ((CommitMessageImpl)
message).withCheckFromSnapshot(snapshot.id()))
+ .collect(Collectors.toList());
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
index 2fa240e028..6c196c3b04 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
@@ -43,6 +43,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -115,6 +116,10 @@ public class IncrementalClusterManager {
}
public Map<BinaryRow, CompactUnit> createCompactUnits(boolean
fullCompaction) {
+ if (snapshot == null) {
+ return Collections.emptyMap();
+ }
+
// 1. construct LSM structure for each partition
Map<BinaryRow, List<LevelSortedRun>> partitionLevels =
constructLevels();
logForPartitionLevel(partitionLevels, partitionComputer);
@@ -186,6 +191,7 @@ public class IncrementalClusterManager {
DataSplit.Builder builder =
DataSplit.builder()
+ .withSnapshot(snapshot.id())
.withPartition(partition)
.withBucket(0)
.withTotalBuckets(1)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterRewriter.java
index 91fd178567..d021e2c815 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterRewriter.java
@@ -47,7 +47,9 @@ import java.util.stream.Collectors;
public class IncrementalClusterRewriter {
public static IncrementalClusterRewriter create(FileStoreTable table,
Snapshot snapshot) {
- return new IncrementalClusterRewriter();
+ return table.coreOptions().dataEvolutionEnabled()
+ ? new DataEvolutionIncrementalClusterRewriter(table, snapshot)
+ : new IncrementalClusterRewriter();
}
List<LevelSortedRun> constructPartitionLevels(List<DataFileMeta>
partitionFiles) {
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 ffe79cf4bb..d1573b9412 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
@@ -836,7 +836,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
}
private Set<Pair<BinaryRow, Integer>>
materializedBuckets(List<CommitMessage> commitMessages) {
- if (!options.dataEvolutionEnabled() ||
!options.deletionVectorsEnabled()) {
+ if (!options.dataEvolutionEnabled()) {
return Collections.emptySet();
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index c98600f7ee..9aaca5f6d9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -1628,9 +1628,6 @@ public class SchemaValidation {
checkArgument(
rowTrackingEnabled,
"Data evolution config must enabled with
row-tracking.enabled");
- checkArgument(
- !options.clusteringIncrementalEnabled(),
- "Data evolution config must disabled with
clustering.incremental");
}
if (options.dataEvolutionNestedFieldEnabled()) {
@@ -1925,6 +1922,16 @@ public class SchemaValidation {
schema.primaryKeys().isEmpty(),
"Cannot define %s for incremental clustering table.",
PRIMARY_KEY.key());
+ if (options.dataEvolutionEnabled()) {
+ checkArgument(
+ fieldNamesInBlobFile(schema.logicalRowType(),
options.blobInlineField())
+ .isEmpty(),
+ "Data evolution incremental clustering does not
support dedicated BLOB files.");
+ checkArgument(
+ vectorFileFormat(options) == null
+ || fieldsInVectorFile(schema.logicalRowType(),
true).isEmpty(),
+ "Data evolution incremental clustering does not
support dedicated VECTOR files.");
+ }
if (options.bucket() != -1) {
checkArgument(
!options.bucketAppendOrdered(),
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
index 2997ec5a9e..484a206afd 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
@@ -43,6 +43,8 @@ import org.apache.paimon.types.RowType;
import org.assertj.core.util.Lists;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
import java.util.Collections;
@@ -73,6 +75,31 @@ public class IncrementalClusterManagerTest {
"Only support incremental clustering when
'clustering.incremental' is true.");
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void
testEmptyPlanningSnapshotDoesNotReadConcurrentFirstAppend(boolean dataEvolution)
+ throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.WRITE_ONLY.key(), "true");
+ options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
String.valueOf(dataEvolution));
+ options.put(CoreOptions.ROW_TRACKING_ENABLED.key(),
String.valueOf(dataEvolution));
+ FileStoreTable table = createTable(options, Collections.emptyList());
+ assertThat(table.snapshotManager().latestSnapshot()).isNull();
+ IncrementalClusterManager manager = new
IncrementalClusterManager(table, null, null);
+
+ writeOnce(
+ table,
+ GenericRow.of(1, 2, BinaryString.fromString("a"),
BinaryString.fromString("b")));
+
+ assertThat(manager.createCompactUnits(true)).isEmpty();
+ assertThat(manager.createCompactUnits(false)).isEmpty();
+ assertThat(
+ new IncrementalClusterManager(
+ table, null,
table.snapshotManager().latestSnapshot())
+ .createCompactUnits(true))
+ .isNotEmpty();
+ }
+
@Test
public void testConstructPartitionLevels() {
// Create test files with different levels
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
new file mode 100644
index 0000000000..51e79a1c31
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
@@ -0,0 +1,882 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.append.cluster.IncrementalClusterManager;
+import org.apache.paimon.append.cluster.IncrementalClusterRewriter;
+import org.apache.paimon.append.cluster.Sorter;
+import org.apache.paimon.catalog.RenamingSnapshotCommit;
+import org.apache.paimon.catalog.SnapshotCommit;
+import org.apache.paimon.compact.CompactUnit;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.deletionvectors.BitmapDeletionVector;
+import org.apache.paimon.deletionvectors.DeletionVector;
+import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.operation.commit.RowIdExistenceConflictException;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.partition.PartitionStatistics;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.reader.RecordReaderIterator;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.CatalogEnvironment;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.MutableObjectIterator;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+import static
org.apache.paimon.errors.ErrorMessages.DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE;
+import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET;
+import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Behavioral tests for sorting data-evolution rows into fresh row-ID ranges.
*/
+class DataEvolutionClusterTest extends TableTestBase {
+
+ @Test
+ void testClusterSortsLatestValuesAndAppliesDeletionVectors() throws
Exception {
+ FileStoreTable table = createTable(true);
+ append(table, "a", 30, 10, 20);
+ append(table, "a", 60, 40, 50);
+ updatePayload(table, "a", 0L, 30, 10, 20);
+ delete(table, "a", new Range(0, 2), 2L);
+
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareCluster(table, snapshot, null,
true);
+ assertThat(messages).isNotEmpty();
+ commitCluster(table, snapshot, messages);
+
+ assertThat(readRows(table))
+ .containsExactly(
+ "a|10|updated-10|6",
+ "a|30|updated-30|7",
+ "a|40|base-40|8",
+ "a|50|base-50|9",
+ "a|60|base-60|10");
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(11L);
+ assertThat(
+
table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry ->
entry.indexFile().globalIndexMeta() == null))
+ .isEmpty();
+ }
+
+ @Test
+ void testFullClusterSkipsAlreadyClusteredPartition() throws Exception {
+ FileStoreTable table = createTable(false);
+ append(table, "a", 30, 10, 20);
+ Snapshot snapshot = table.latestSnapshot().get();
+ commitCluster(table, snapshot, prepareCluster(table, snapshot, null,
true));
+ List<String> clusteredRows = readRows(table);
+ List<String> clusteredFiles = fileNames(table, "a");
+
+ Snapshot clusteredSnapshot = table.latestSnapshot().get();
+ assertThat(prepareCluster(table, clusteredSnapshot, null,
true)).isEmpty();
+ assertThat(readRows(table)).isEqualTo(clusteredRows);
+ assertThat(fileNames(table, "a")).isEqualTo(clusteredFiles);
+ }
+
+ @Test
+ void testClusterRemovesFullyDeletedGroup() throws Exception {
+ FileStoreTable table = createTable(true);
+ append(table, "a", 30, 10, 20);
+ delete(table, "a", new Range(0, 2), 0L, 1L, 2L);
+
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareCluster(table, snapshot, null,
true);
+ assertThat(messages).isNotEmpty();
+ commitCluster(table, snapshot, messages);
+
+ assertThat(readRows(table)).isEmpty();
+ assertThat(fileNames(table, "a")).isEmpty();
+
assertThat(table.store().newIndexFileHandler().scanEntries()).isEmpty();
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(3L);
+ }
+
+ @Test
+ void testClusterHonorsOutputFileRowLimit() throws Exception {
+ FileStoreTable table = createTable(false);
+ append(table, "a", 70, 10, 50, 30, 60, 20, 40);
+ table =
table.copy(Collections.singletonMap(CoreOptions.TARGET_FILE_ROW_NUM.key(),
"2"));
+
+ Snapshot snapshot = table.latestSnapshot().get();
+ commitCluster(table, snapshot, prepareCluster(table, snapshot, null,
true));
+
+ List<DataFileMeta> files =
+
table.store().newScan().withSnapshot(table.latestSnapshot().get()).plan().files()
+ .stream()
+ .map(entry -> entry.file())
+ .collect(Collectors.toList());
+ assertThat(files).hasSize(4);
+ assertThat(files).allSatisfy(file ->
assertThat(file.rowCount()).isBetween(1L, 2L));
+ assertThat(
+ files.stream()
+ .map(DataFileMeta::nonNullRowIdRange)
+ .sorted(Comparator.comparingLong(range ->
range.from)))
+ .containsExactly(
+ new Range(7, 8), new Range(9, 10), new Range(11, 12),
new Range(13, 13));
+ assertThat(readRows(table))
+ .containsExactly(
+ "a|10|base-10|7",
+ "a|20|base-20|8",
+ "a|30|base-30|9",
+ "a|40|base-40|10",
+ "a|50|base-50|11",
+ "a|60|base-60|12",
+ "a|70|base-70|13");
+ }
+
+ @Test
+ void testPartialSortKeyUpdateDirtiesClusteredGroup() throws Exception {
+ FileStoreTable table = createTable(false);
+ append(table, "a", 30, 10, 20);
+ Snapshot snapshot = table.latestSnapshot().get();
+ commitCluster(table, snapshot, prepareCluster(table, snapshot, null,
true));
+
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ RowType writeType = table.rowType().project(Arrays.asList("pt", "id"));
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType);
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int id : new int[] {50, 10, 30}) {
+ write.write(GenericRow.of(BinaryString.fromString("a"), id));
+ }
+ List<CommitMessage> updates = write.prepareCommit();
+ assignFirstRowId(updates, 3L);
+ commit.commit(updates);
+ }
+
+ Snapshot updated = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareCluster(table, updated, null,
true);
+ assertThat(messages).isNotEmpty();
+ commitCluster(table, updated, messages);
+
+ assertThat(readRows(table))
+ .containsExactly("a|10|base-20|6", "a|30|base-30|7",
"a|50|base-10|8");
+ }
+
+ @Test
+ void testIncrementalClusterWaitsForCompactionThreshold() throws Exception {
+ FileStoreTable table = createTable(false);
+ append(table, "a", 30, 10, 20);
+ assertThat(prepareCluster(table, table.latestSnapshot().get(), null,
false)).isEmpty();
+
+ append(table, "a", 60, 40, 50);
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareCluster(table, snapshot, null,
false);
+ assertThat(messages).isNotEmpty();
+ commitCluster(table, snapshot, messages);
+
+ assertThat(readRows(table))
+ .containsExactly(
+ "a|10|base-10|6",
+ "a|20|base-20|7",
+ "a|30|base-30|8",
+ "a|40|base-40|9",
+ "a|50|base-50|10",
+ "a|60|base-60|11");
+ }
+
+ @Test
+ void testMinorClusterSelectsCompleteDirtyGroupAndRetainsUntouchedGroup()
throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.TARGET_FILE_ROW_NUM.key(), "3");
+ options.put(CoreOptions.FILE_COMPRESSION.key(), "none");
+
options.put(CoreOptions.COMPACTION_MAX_SIZE_AMPLIFICATION_PERCENT.key(),
"10000");
+ options.put(CoreOptions.COMPACTION_SIZE_RATIO.key(), "0");
+ FileStoreTable table = createTable(false).copy(options);
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int id : new int[] {60, 10, 50, 30, 20, 40}) {
+ Random random = new Random(id);
+ int length = id <= 30 ? 8192 : 131072;
+ StringBuilder payload = new StringBuilder(length);
+ for (int i = 0; i < length; i++) {
+ payload.append((char) ('a' + random.nextInt(26)));
+ }
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString("a"),
+ id,
+ BinaryString.fromString(payload.toString())));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ Snapshot initial = table.latestSnapshot().get();
+ commitCluster(table, initial, prepareCluster(table, initial, null,
true));
+
+ List<DataFileMeta> clusteredFiles =
+
table.store().newScan().withSnapshot(table.latestSnapshot().get()).plan().files()
+ .stream()
+ .map(entry -> entry.file())
+
.sorted(Comparator.comparingLong(DataFileMeta::nonNullFirstRowId))
+ .collect(Collectors.toList());
+ assertThat(clusteredFiles).hasSize(2);
+ DataFileMeta dirtyBase = clusteredFiles.get(0);
+ DataFileMeta untouchedBase = clusteredFiles.get(1);
+ assertThat(dirtyBase.nonNullRowIdRange()).isEqualTo(new Range(6, 8));
+ assertThat(untouchedBase.nonNullRowIdRange()).isEqualTo(new Range(9,
11));
+
assertThat(untouchedBase.fileSize()).isGreaterThan(dirtyBase.fileSize() * 8);
+ List<String> expected = new ArrayList<>(readRows(table).subList(3, 6));
+
+ updatePayload(table, "a", 6L, 10, 20, 30);
+ append(table, "a", 90, 70, 80);
+ Snapshot snapshot = table.latestSnapshot().get();
+ IncrementalClusterManager manager = new
IncrementalClusterManager(table, null, snapshot);
+ Map<BinaryRow, CompactUnit> units = manager.createCompactUnits(false);
+ assertThat(units).hasSize(1);
+ CompactUnit unit = units.get(partition(table, "a"));
+ assertThat(unit.files())
+ .extracting(DataFileMeta::fileName)
+ .contains(dirtyBase.fileName())
+ .doesNotContain(untouchedBase.fileName());
+ assertThat(unit.files())
+ .filteredOn(file -> file.nonNullRowIdRange().equals(new
Range(6, 8)))
+ .hasSize(2);
+ commitCluster(table, snapshot, rewriteClusterUnits(table, manager,
units));
+
+ expected.addAll(
+ Arrays.asList(
+ "a|10|updated-10|15",
+ "a|20|updated-20|16",
+ "a|30|updated-30|17",
+ "a|70|base-70|18",
+ "a|80|base-80|19",
+ "a|90|base-90|20"));
+ assertThat(readRows(table)).containsExactlyElementsOf(expected);
+ assertThat(fileNames(table, "a"))
+ .contains(untouchedBase.fileName())
+ .doesNotContain(dirtyBase.fileName());
+ }
+
+ @Test
+ void testConcurrentPartialUpdateRejectsStaleCluster() throws Exception {
+ FileStoreTable table = createTable(true);
+ append(table, "a", 30, 10, 20);
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareCluster(table, snapshot, null,
true);
+
+ updatePayload(table, "a", 0L, 30, 10, 20);
+ long updateSnapshot = 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(updateSnapshot);
+ assertThat(readRows(table))
+ .containsExactly("a|30|updated-30|0", "a|10|updated-10|1",
"a|20|updated-20|2");
+ }
+
+ @Test
+ void testConcurrentDeleteRejectsStaleCluster() throws Exception {
+ FileStoreTable table = createTable(true);
+ append(table, "a", 30, 10, 20);
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareCluster(table, snapshot, null,
true);
+
+ delete(table, "a", new Range(0, 2), 1L);
+ long deleteSnapshot = table.latestSnapshot().get().id();
+ // The delete changes the anchor's DV metadata, so normal file
conflict detection rejects
+ // the stale removal before the additional row-ID conflict check is
reached.
+ assertThatThrownBy(() -> commitCluster(table, snapshot, messages))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("File deletion conflicts detected!");
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(deleteSnapshot);
+ assertThat(readRows(table)).containsExactly("a|30|base-30|0",
"a|20|base-20|2");
+ }
+
+ @Test
+ void testConcurrentIndependentAppendIsPreserved() throws Exception {
+ FileStoreTable table = createTable(true);
+ append(table, "a", 30, 10, 20);
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareCluster(table, snapshot, null,
true);
+
+ append(table, "a", 60, 40, 50);
+ commitCluster(table, snapshot, messages);
+
+ assertThat(readRows(table))
+ .containsExactly(
+ "a|60|base-60|3",
+ "a|40|base-40|4",
+ "a|50|base-50|5",
+ "a|10|base-10|6",
+ "a|20|base-20|7",
+ "a|30|base-30|8");
+ }
+
+ @Test
+ void testPartitionClusterDropsOnlyAffectedGlobalIndexes() throws Exception
{
+ FileStoreTable table = createTable(true);
+ append(table, "a", 30, 10, 20);
+ append(table, "b", 60, 40, 50);
+ addGlobalIndex(table, "a", "index-a", new Range(0, 2), "id");
+ addGlobalIndex(table, "b", "index-b", new Range(3, 5), "id");
+ List<String> untouchedFiles = fileNames(table, "b");
+
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages =
+ prepareCluster(table, snapshot, partitionPredicate(table,
"a"), true);
+ addGlobalIndex(table, "a", "concurrent-index-a", new Range(0, 2),
"payload");
+ assertThat(globalIndexNames(table))
+ .containsExactly("concurrent-index-a", "index-a", "index-b");
+ commitCluster(table, snapshot, messages);
+
+ assertThat(globalIndexNames(table)).containsExactly("index-b");
+ assertThat(fileNames(table, "b")).isEqualTo(untouchedFiles);
+ assertThat(readRows(table))
+ .containsExactly(
+ "b|60|base-60|3",
+ "b|40|base-40|4",
+ "b|50|base-50|5",
+ "a|10|base-10|6",
+ "a|20|base-20|7",
+ "a|30|base-30|8");
+ }
+
+ @Test
+ void testWithoutDeletionVectorsDropsIndexCommittedAfterPreparation()
throws Exception {
+ FileStoreTable table = createTable(false);
+ append(table, "a", 30, 10, 20);
+ addGlobalIndex(table, "a", "index-before-plan", new Range(0, 2), "id");
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareCluster(table, snapshot, null,
true);
+ messages =
+ IncrementalClusterRewriter.create(table,
snapshot).prepareCommitMessages(messages);
+
+ addGlobalIndex(table, "a", "index-after-preparation", new Range(0, 2),
"payload");
+ assertThat(globalIndexNames(table))
+ .containsExactly("index-after-preparation",
"index-before-plan");
+ commitChecked(table, messages);
+
+ assertThat(globalIndexNames(table)).isEmpty();
+ assertThat(readRows(table))
+ .containsExactly("a|10|base-10|3", "a|20|base-20|4",
"a|30|base-30|5");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testClusterRetryDropsIndexCommittedBeforeSnapshotPublication(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 =
createTable(deletionVectors).copy(retryOptions);
+ append(original, "a", 30, 10, 20);
+ if (deletionVectors) {
+ delete(original, "a", new Range(0, 2), 1L);
+ }
+ commit(original, prepareGlobalIndex(original,
original.latestSnapshot().get(), "id"));
+ Snapshot scanSnapshot = original.latestSnapshot().get();
+ List<CommitMessage> messages = prepareCluster(original, scanSnapshot,
null, true);
+ 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 {
+ assertThat(snapshot.commitKind())
+
.isEqualTo(Snapshot.CommitKind.COMPACT);
+ int attempt = attempts.incrementAndGet();
+ // Index cleanup has already been prepared for
this candidate.
+ assertThat(
+ original.store()
+
.indexManifestFileFactory()
+ .create()
+
.read(snapshot.indexManifest()))
+ .isEmpty();
+ if (attempt == 1) {
+
assertThat(original.latestSnapshot().get().id())
+ .isEqualTo(scanSnapshot.id());
+ DataEvolutionClusterTest.this.commit(
+ original,
+ prepareGlobalIndex(original,
scanSnapshot, "payload"));
+
assertThat(globalIndexNames(original)).hasSize(2);
+ }
+ 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() +
2);
+ assertThat(original.latestSnapshot().get().nextRowId())
+ .isEqualTo(deletionVectors ? 5L : 6L);
+ assertThat(globalIndexNames(original)).isEmpty();
+ assertThat(readRows(original))
+ .containsExactlyElementsOf(
+ deletionVectors
+ ? Arrays.asList("a|20|base-20|3",
"a|30|base-30|4")
+ : Arrays.asList(
+ "a|10|base-10|3", "a|20|base-20|4",
"a|30|base-30|5"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testClusterRejectsPreviouslyPreparedPartialUpdate(boolean
deletionVectors)
+ throws Exception {
+ FileStoreTable table = createTable(deletionVectors);
+ append(table, "a", 30, 10, 20);
+ Snapshot scanSnapshot = table.latestSnapshot().get();
+ List<CommitMessage> updates =
+ preparePayloadUpdate(table, "a", 0L, 30, 10, 20).stream()
+ .map(
+ message ->
+ ((CommitMessageImpl) message)
+
.withCheckFromSnapshot(scanSnapshot.id()))
+ .collect(Collectors.toList());
+ commitCluster(table, scanSnapshot, prepareCluster(table, scanSnapshot,
null, true));
+ commit(table, prepareGlobalIndex(table, table.latestSnapshot().get(),
"id"));
+
+ assertStaleCommitPreservesCluster(
+ table, updates, RowIdExistenceConflictException.class, "Row ID
existence conflict");
+ }
+
+ @Test
+ void testClusterRejectsPreviouslyPreparedDelete() throws Exception {
+ FileStoreTable table = createTable(true);
+ append(table, "a", 30, 10, 20);
+ Snapshot scanSnapshot = table.latestSnapshot().get();
+ CommitMessage deletion =
+ ((CommitMessageImpl) prepareDelete(table, "a", new Range(0,
2), 1L))
+ .withCheckFromSnapshot(scanSnapshot.id());
+ commitCluster(table, scanSnapshot, prepareCluster(table, scanSnapshot,
null, true));
+ commit(table, prepareGlobalIndex(table, table.latestSnapshot().get(),
"id"));
+
+ assertStaleCommitPreservesCluster(
+ table,
+ Collections.singletonList(deletion),
+ RuntimeException.class,
+ "File deletion conflicts detected!");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testClusterRejectsPreviouslyBuiltGlobalIndex(boolean deletionVectors)
throws Exception {
+ FileStoreTable table = createTable(deletionVectors);
+ append(table, "a", 30, 10, 20);
+ Snapshot scanSnapshot = table.latestSnapshot().get();
+ List<CommitMessage> indexes = prepareGlobalIndex(table, scanSnapshot,
"id");
+ commitCluster(table, scanSnapshot, prepareCluster(table, scanSnapshot,
null, true));
+ commit(table, prepareGlobalIndex(table, table.latestSnapshot().get(),
"id"));
+
+ assertStaleCommitPreservesCluster(
+ table, indexes, RuntimeException.class, "Global index row ID
existence conflict");
+ }
+
+ private void assertStaleCommitPreservesCluster(
+ FileStoreTable table,
+ List<CommitMessage> staleMessages,
+ Class<? extends RuntimeException> exceptionClass,
+ String conflictMessage)
+ throws Exception {
+ Snapshot clustered = table.latestSnapshot().get();
+ List<String> files = fileNames(table, "a");
+ List<String> indexes = globalIndexNames(table);
+ assertThat(indexes).isNotEmpty();
+ assertThatThrownBy(() -> commit(table, staleMessages))
+ .isInstanceOf(exceptionClass)
+ .hasMessageContaining(conflictMessage);
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(clustered.id());
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(6L);
+ assertThat(fileNames(table, "a")).isEqualTo(files);
+ assertThat(globalIndexNames(table)).isEqualTo(indexes);
+ assertThat(readRows(table))
+ .containsExactly("a|10|base-10|3", "a|20|base-20|4",
"a|30|base-30|5");
+ }
+
+ private FileStoreTable createTable(boolean deletionVectors) throws
Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .column("pt", DataTypes.STRING())
+ .column("id", DataTypes.INT())
+ .column("payload", DataTypes.STRING())
+ .partitionKeys("pt")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(
+ CoreOptions.DELETION_VECTORS_ENABLED.key(),
+ String.valueOf(deletionVectors))
+ .option(CoreOptions.CLUSTERING_COLUMNS.key(), "id")
+ .option(CoreOptions.CLUSTERING_STRATEGY.key(), "order")
+ .option(CoreOptions.CLUSTERING_INCREMENTAL.key(),
"true")
+
.option(CoreOptions.NUM_SORTED_RUNS_COMPACTION_TRIGGER.key(), "2")
+ .build();
+ catalog.createTable(identifier(), schema, false);
+ return getTableDefault();
+ }
+
+ private void append(FileStoreTable table, String partition, int... values)
throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int value : values) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString(partition),
+ value,
+ BinaryString.fromString("base-" + value)));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private void updatePayload(
+ FileStoreTable table, String partition, long firstRowId, int...
values)
+ throws Exception {
+ commit(table, preparePayloadUpdate(table, partition, firstRowId,
values));
+ }
+
+ private List<CommitMessage> preparePayloadUpdate(
+ FileStoreTable table, String partition, long firstRowId, int...
values)
+ throws Exception {
+ RowType writeType = table.rowType().project(Arrays.asList("pt",
"payload"));
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType)) {
+ for (int value : values) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString(partition),
+ BinaryString.fromString("updated-" + value)));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ assignFirstRowId(messages, firstRowId);
+ return messages;
+ }
+ }
+
+ private void assignFirstRowId(List<CommitMessage> messages, long
firstRowId) {
+ for (CommitMessage message : messages) {
+ List<DataFileMeta> files = ((CommitMessageImpl)
message).newFilesIncrement().newFiles();
+ List<DataFileMeta> assigned =
+ files.stream()
+ .map(file -> file.assignFirstRowId(firstRowId))
+ .collect(Collectors.toList());
+ files.clear();
+ files.addAll(assigned);
+ }
+ }
+
+ private List<CommitMessage> prepareCluster(
+ FileStoreTable table,
+ Snapshot snapshot,
+ PartitionPredicate partitionPredicate,
+ boolean fullCompaction)
+ throws Exception {
+ IncrementalClusterManager manager =
+ new IncrementalClusterManager(table, partitionPredicate,
snapshot);
+ return rewriteClusterUnits(table, manager,
manager.createCompactUnits(fullCompaction));
+ }
+
+ // Run the engine-independent read/sort/write stages in-process. The
source selection and
+ // split planning are the same shared manager used by the distributed
engine implementations.
+ private List<CommitMessage> rewriteClusterUnits(
+ FileStoreTable table,
+ IncrementalClusterManager manager,
+ Map<BinaryRow, CompactUnit> units)
+ throws Exception {
+ Map<BinaryRow, Pair<List<DataSplit>, CommitMessage>> planned =
+ manager.toSplitsAndRewriteDvFiles(units);
+ List<CommitMessage> messages = new ArrayList<>();
+ FileStoreTable writeTable =
+
table.copy(Collections.singletonMap(CoreOptions.WRITE_ONLY.key(), "true"));
+ for (Map.Entry<BinaryRow, CompactUnit> entry : units.entrySet()) {
+ Pair<List<DataSplit>, CommitMessage> partitionPlan =
planned.get(entry.getKey());
+ assertThat(partitionPlan.getRight()).isNull();
+ List<DataFileMeta> after = new ArrayList<>();
+ try (IOManager ioManager = IOManager.create(tempPath.toString());
+ RecordReaderIterator<InternalRow> reader =
+ new RecordReaderIterator<>(
+ table.newReadBuilder()
+ .newRead()
+ .createReader(
+ new
ArrayList<>(partitionPlan.getLeft())));
+ BatchTableWrite write =
+ writeTable
+ .newBatchWriteBuilder()
+ .newWrite()
+ .withWriteType(table.rowType())) {
+ if (reader.hasNext()) {
+ Sorter sorter =
+ Sorter.getSorter(
+ reader, ioManager, table.rowType(),
table.coreOptions());
+ try {
+ MutableObjectIterator<BinaryRow> sorted =
sorter.sort();
+ BinaryRow row = new BinaryRow(sorter.arity());
+ while ((row = sorted.next(row)) != null) {
+ write.write(sorter.removeSortKey(row));
+ }
+ } finally {
+ sorter.close();
+ }
+ }
+ for (CommitMessage output : write.prepareCommit()) {
+ after.addAll(((CommitMessageImpl)
output).newFilesIncrement().newFiles());
+ }
+ }
+ messages.add(
+ new CommitMessageImpl(
+ entry.getKey(),
+ 0,
+ null,
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ entry.getValue().files(),
+ IncrementalClusterManager.upgrade(
+ after,
entry.getValue().outputLevel()),
+ Collections.emptyList())));
+ }
+ return messages;
+ }
+
+ private void commitCluster(
+ FileStoreTable table, Snapshot snapshot, List<CommitMessage>
messages)
+ throws Exception {
+ commitChecked(
+ table,
+ IncrementalClusterRewriter.create(table,
snapshot).prepareCommitMessages(messages));
+ }
+
+ private void commitChecked(FileStoreTable table, List<CommitMessage>
messages)
+ throws Exception {
+ try (TableCommitImpl commit = table.newCommit("test-cluster")) {
+ commit.materializeDvRowIdCheck().commit(messages);
+ }
+ }
+
+ private void delete(FileStoreTable table, String value, Range range,
long... rowIds)
+ throws Exception {
+ commit(table, prepareDelete(table, value, range, rowIds));
+ }
+
+ private CommitMessage prepareDelete(
+ FileStoreTable table, String value, Range range, long... rowIds)
throws Exception {
+ BinaryRow partition = partition(table, value);
+ List<DataFileMeta> group =
+
table.store().newScan().withSnapshot(table.latestSnapshot().get()).plan().files()
+ .stream()
+ .filter(entry -> entry.partition().equals(partition))
+ .map(entry -> entry.file())
+ .filter(file -> file.nonNullRowIdRange().equals(range))
+ .collect(Collectors.toList());
+ DataFileMeta anchor = retrieveAnchorFile(group, file -> file);
+ BaseAppendDeleteFileMaintainer maintainer =
+ BaseAppendDeleteFileMaintainer.forUnawareAppend(
+ table.store().newIndexFileHandler(),
+ table.latestSnapshot().get(),
+ partition);
+ DeletionVector vector = new BitmapDeletionVector();
+ for (long rowId : rowIds) {
+ vector.delete(rowId - range.from);
+ }
+ maintainer.notifyNewDeletionVector(anchor.fileName(), vector);
+ List<IndexFileMeta> additions = new ArrayList<>();
+ List<IndexFileMeta> removals = new ArrayList<>();
+ for (IndexManifestEntry entry : maintainer.persist()) {
+ (entry.kind() == FileKind.ADD ? additions :
removals).add(entry.indexFile());
+ }
+ return new CommitMessageImpl(
+ partition,
+ UNAWARE_BUCKET,
+ null,
+ new DataIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ additions,
+ removals),
+ CompactIncrement.emptyIncrement());
+ }
+
+ private void addGlobalIndex(
+ FileStoreTable table, String value, String fileName, Range range,
String column)
+ throws Exception {
+ IndexFileMeta index =
+ new IndexFileMeta(
+ "test-global-index",
+ fileName,
+ 1,
+ range.count(),
+ new GlobalIndexMeta(
+ range.from,
+ range.to,
+ table.rowType().getField(column).id(),
+ null,
+ null),
+ null);
+ commit(
+ table,
+ new CommitMessageImpl(
+ partition(table, value),
+ UNAWARE_BUCKET,
+ null,
+
DataIncrement.indexIncrement(Collections.singletonList(index)),
+ CompactIncrement.emptyIncrement()));
+ }
+
+ private void commit(FileStoreTable table, CommitMessage message) throws
Exception {
+ commit(table, Collections.singletonList(message));
+ }
+
+ private void commit(FileStoreTable table, List<CommitMessage> messages)
throws Exception {
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
+ private List<CommitMessage> prepareGlobalIndex(
+ FileStoreTable table, Snapshot snapshot, String column) throws
Exception {
+ List<CommitMessage> messages = new ArrayList<>();
+ for (DataSplit split :
+ GlobalIndexBuilderUtils.splitByContiguousRowRange(
+ new SortedGlobalIndexScanner(table, "btree")
+
.withIndexFields(Collections.singletonList(column))
+ .withSnapshot(snapshot)
+ .scan()
+ .get()
+ .entries())) {
+ messages.addAll(
+ SortedGlobalIndexTestUtils.buildIndex(
+ table, "btree", column, split, snapshot.id()));
+ }
+ assertThat(messages).isNotEmpty();
+ return messages;
+ }
+
+ private List<String> readRows(FileStoreTable table) throws Exception {
+ ReadBuilder builder =
+ table.newReadBuilder()
+
.withReadType(SpecialFields.rowTypeWithRowId(table.rowType()));
+ List<String> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ builder.newRead().createReader(builder.newScan().plan())) {
+ reader.forEachRemaining(
+ row ->
+ rows.add(
+ row.getString(0)
+ + "|"
+ + row.getInt(1)
+ + "|"
+ + row.getString(2)
+ + "|"
+ + row.getLong(3)));
+ }
+ rows.sort(
+ Comparator.comparingLong(
+ row ->
Long.parseLong(row.substring(row.lastIndexOf('|') + 1))));
+ return rows;
+ }
+
+ private List<String> fileNames(FileStoreTable table, String value) {
+ BinaryRow partition = partition(table, value);
+ return
table.store().newScan().withSnapshot(table.latestSnapshot().get()).plan().files()
+ .stream()
+ .filter(entry -> entry.partition().equals(partition))
+ .map(entry -> entry.file().fileName())
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ private List<String> globalIndexNames(FileStoreTable table) {
+ return table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry -> entry.indexFile().globalIndexMeta() != null)
+ .map(entry -> entry.indexFile().fileName())
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ private BinaryRow partition(FileStoreTable table, String value) {
+ return new InternalRowSerializer(table.schema().logicalPartitionType())
+ .toBinaryRow(GenericRow.of(BinaryString.fromString(value)));
+ }
+
+ private PartitionPredicate partitionPredicate(FileStoreTable table, String
value) {
+ return PartitionPredicate.fromMaps(
+ table.schema().logicalPartitionType(),
+ Collections.singletonList(Collections.singletonMap("pt",
value)),
+ table.coreOptions().partitionDefaultName());
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index 54a7088950..bcd43a5565 100644
---
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -1563,6 +1563,93 @@ class SchemaValidationTest {
assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
}
+ @Test
+ void testDataEvolutionIncrementalClustering() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "value", DataTypes.STRING()));
+ TableSchema schema =
+ new TableSchema(
+ 1,
+ fields,
+ 1,
+ emptyList(),
+ emptyList(),
+ dataEvolutionClusteringOptions(),
+ "");
+
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"blob-descriptor-field", "blob-view-field"})
+ void testDataEvolutionIncrementalClusteringAllowsInlineBlob(String
inlineOption) {
+ Map<String, String> options = dataEvolutionClusteringOptions();
+ options.put(inlineOption, "payload");
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "payload", DataTypes.BLOB()));
+ TableSchema schema = new TableSchema(1, fields, 1, emptyList(),
emptyList(), options, "");
+
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testDataEvolutionIncrementalClusteringRejectsDedicatedBlob() {
+ for (DataType blobType :
+ Arrays.asList(
+ DataTypes.BLOB(),
+ DataTypes.ARRAY(DataTypes.BLOB()),
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.BLOB()))) {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "payload", blobType));
+ TableSchema schema =
+ new TableSchema(
+ 1,
+ fields,
+ 1,
+ emptyList(),
+ emptyList(),
+ dataEvolutionClusteringOptions(),
+ "");
+
+ assertThatThrownBy(() -> validateTableSchema(schema))
+ .hasMessage(
+ "Data evolution incremental clustering does not
support dedicated BLOB files.");
+ }
+ }
+
+ @Test
+ void testDataEvolutionIncrementalClusteringVectorStorage() {
+ Map<String, String> options = dataEvolutionClusteringOptions();
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "embedding", DataTypes.VECTOR(3,
DataTypes.FLOAT())));
+ TableSchema schema = new TableSchema(1, fields, 1, emptyList(),
emptyList(), options, "");
+
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+
+ options.put(VECTOR_FILE_FORMAT.key(), "json");
+ assertThatThrownBy(() -> validateTableSchema(schema))
+ .hasMessage(
+ "Data evolution incremental clustering does not
support dedicated VECTOR files.");
+ }
+
+ private Map<String, String> dataEvolutionClusteringOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "-1");
+ options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ options.put(DATA_EVOLUTION_ENABLED.key(), "true");
+ options.put(CoreOptions.CLUSTERING_INCREMENTAL.key(), "true");
+ options.put(CoreOptions.CLUSTERING_COLUMNS.key(), "id");
+ return options;
+ }
+
@Test
public void testFileIndexColumns() {
List<String> keys =
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
index 1d01ba1f2e..94480deb1b 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
@@ -154,12 +154,15 @@ public class CompactAction extends TableActionBase {
if (fileStoreTable.coreOptions().bucket() ==
BucketMode.POSTPONE_BUCKET) {
buildForPostponeBucketCompaction(env, fileStoreTable, isStreaming);
} else if (fileStoreTable.bucketMode() == BucketMode.BUCKET_UNAWARE) {
- if (fileStoreTable.coreOptions().dataEvolutionEnabled()) {
- buildForDataEvolutionTableCompact(env, fileStoreTable,
isStreaming);
- } else if
(fileStoreTable.coreOptions().clusteringIncrementalEnabled()) {
+ if (fileStoreTable.coreOptions().clusteringIncrementalEnabled()) {
+ checkArgument(
+ !isStreaming ||
!fileStoreTable.coreOptions().dataEvolutionEnabled(),
+ "Data evolution incremental clustering only supports
batch mode yet.");
new IncrementalClusterCompact(
env, fileStoreTable, partitionPredicate,
fullCompaction)
.build();
+ } else if (fileStoreTable.coreOptions().dataEvolutionEnabled()) {
+ buildForDataEvolutionTableCompact(env, fileStoreTable,
isStreaming);
} else {
buildForAppendTableCompact(env, fileStoreTable, isStreaming);
}
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 0367adce8c..ef5a58632b 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
@@ -19,6 +19,7 @@
package org.apache.paimon.flink.compact;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.append.cluster.IncrementalClusterManager;
import org.apache.paimon.codegen.CodeGenUtils;
import org.apache.paimon.compact.CompactUnit;
@@ -31,6 +32,8 @@ import
org.apache.paimon.flink.cluster.IncrementalClusterSplitSource;
import
org.apache.paimon.flink.cluster.RewriteIncrementalClusterCommittableOperator;
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.DataEvolutionDeletionVectorMaterializeSink;
import org.apache.paimon.flink.sink.FlinkSinkBuilder;
import org.apache.paimon.flink.sink.RowAppendTableSink;
import org.apache.paimon.flink.sorter.SortOperator;
@@ -59,12 +62,15 @@ import org.apache.flink.table.data.RowData;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import static org.apache.paimon.flink.FlinkConnectorOptions.PRECOMMIT_COMPACT;
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.table.PrimaryKeyTableUtils.addKeyNamePrefix;
/** Compact for incremental clustering. */
@@ -87,6 +93,15 @@ public class IncrementalClusterCompact {
PartitionPredicate partitionPredicate,
@Nullable Boolean fullCompaction) {
this.env = env;
+ if (table.coreOptions().dataEvolutionEnabled()) {
+ Map<String, String> rewriteOptions = new HashMap<>();
+ // Rewrite stored references, including views whose upstream value
is currently null.
+ rewriteOptions.put(CoreOptions.BLOB_VIEW_RESOLVE_ENABLED.key(),
"false");
+ // Sorted outputs must reach the clustering committer without
another rewrite or commit.
+ rewriteOptions.put(PRECOMMIT_COMPACT.key(), "false");
+ rewriteOptions.put(SINK_COORDINATOR_COMMIT_ENABLED.key(), "false");
+ table = table.copy(rewriteOptions);
+ }
this.table = table;
this.clusterManager =
new IncrementalClusterManager(
@@ -331,6 +346,20 @@ public class IncrementalClusterCompact {
for (int i = 1; i < dataStreams.size(); i++) {
dataStream = dataStream.union(dataStreams.get(i));
}
- sink.doCommit(dataStream, commitUser);
+ if (table.coreOptions().dataEvolutionEnabled()) {
+ Snapshot snapshot = clusterManager.snapshot();
+ dataStream =
+ dataStream
+ .transform(
+ "Data Evolution Cluster Commit Preparation
: " + table.name(),
+ new CommittableTypeInfo(),
+
DataEvolutionCommitPreparationOperator.Factory.forClustering(
+ table, snapshot))
+ .forceNonParallel();
+ new DataEvolutionDeletionVectorMaterializeSink(table, snapshot)
+ .doCommit(dataStream, commitUser);
+ } else {
+ sink.doCommit(dataStream, commitUser);
+ }
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionCommitPreparationOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionCommitPreparationOperator.java
index f2c0aa7bff..40c262af08 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionCommitPreparationOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionCommitPreparationOperator.java
@@ -19,6 +19,7 @@
package org.apache.paimon.flink.sink;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.append.cluster.IncrementalClusterRewriter;
import
org.apache.paimon.append.dataevolution.DataEvolutionCompactionCommitPreparation;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
@@ -41,17 +42,20 @@ public class DataEvolutionCommitPreparationOperator
private final FileStoreTable table;
private final Snapshot snapshot;
private final boolean materializeDvRowIdCheck;
+ private final boolean clustering;
private final List<Committable> committables;
private DataEvolutionCommitPreparationOperator(
StreamOperatorParameters<Committable> parameters,
FileStoreTable table,
Snapshot snapshot,
- boolean materializeDvRowIdCheck) {
+ boolean materializeDvRowIdCheck,
+ boolean clustering) {
super(parameters, Options.fromMap(table.options()));
this.table = table;
this.snapshot = snapshot;
this.materializeDvRowIdCheck = materializeDvRowIdCheck;
+ this.clustering = clustering;
this.committables = new ArrayList<>();
}
@@ -73,6 +77,15 @@ public class DataEvolutionCommitPreparationOperator
for (Committable committable : toCommit) {
messages.add(committable.commitMessage());
}
+ if (clustering) {
+ List<Committable> prepared = new ArrayList<>();
+ for (CommitMessage message :
+ IncrementalClusterRewriter.create(table, snapshot)
+ .prepareCommitMessages(messages)) {
+ prepared.add(new Committable(checkpointId, message));
+ }
+ return prepared;
+ }
for (CommitMessage message :
new DataEvolutionCompactionCommitPreparation(table,
snapshot).prepare(messages)) {
toCommit.add(new Committable(toCommit.get(0).checkpointId(),
message));
@@ -97,16 +110,30 @@ public class DataEvolutionCommitPreparationOperator
private final FileStoreTable table;
private final Snapshot snapshot;
private final boolean materializeDvRowIdCheck;
+ private final boolean clustering;
public Factory(FileStoreTable table, Snapshot snapshot) {
this(table, snapshot, false);
}
public Factory(FileStoreTable table, Snapshot snapshot, boolean
materializeDvRowIdCheck) {
+ this(table, snapshot, materializeDvRowIdCheck, false);
+ }
+
+ public static Factory forClustering(FileStoreTable table, Snapshot
snapshot) {
+ return new Factory(table, snapshot, true, true);
+ }
+
+ private Factory(
+ FileStoreTable table,
+ Snapshot snapshot,
+ boolean materializeDvRowIdCheck,
+ boolean clustering) {
super(Options.fromMap(table.options()));
this.table = table;
this.snapshot = snapshot;
this.materializeDvRowIdCheck = materializeDvRowIdCheck;
+ this.clustering = clustering;
}
@Override
@@ -115,7 +142,7 @@ public class DataEvolutionCommitPreparationOperator
StreamOperatorParameters<Committable> parameters) {
return (T)
new DataEvolutionCommitPreparationOperator(
- parameters, table, snapshot,
materializeDvRowIdCheck);
+ parameters, table, snapshot,
materializeDvRowIdCheck, clustering);
}
@Override
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 2115105394..e0504d20bd 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
@@ -19,14 +19,36 @@
package org.apache.paimon.flink;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.index.DeletionVectorMeta;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.operation.FileStoreWrite;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
import org.apache.flink.table.api.config.TableConfigOptions;
import org.apache.flink.types.Row;
import org.junit.jupiter.api.Test;
-
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
@@ -87,6 +109,237 @@ public class DataEvolutionDeleteSqlITCase extends
CatalogITCaseBase {
Row.of(6, "six", "B"));
}
+ @ParameterizedTest
+ @CsvSource({
+ "global-sort,",
+ "local-sort,",
+ "global-sort,precommit-compact=true",
+ "local-sort,sink.coordinator-commit.enabled=true"
+ })
+ public void
testIncrementalClusteringProcedurePreservesUnselectedPartitionIndexes(
+ String mode, String ingestionOption) throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ createTable();
+ sql(
+ "ALTER TABLE T SET ("
+ + "'clustering.incremental' = 'true', "
+ + "'clustering.strategy' = 'order', "
+ + "'clustering.columns' = 'id', "
+ + "'clustering.incremental.mode' = '"
+ + mode
+ + "', "
+ + "'scan.parallelism' = '2', "
+ + "'source.split.target-size' = '1 b', "
+ + "'sink.parallelism' = '1', "
+ + "'num-sorted-run.compaction-trigger' = '10')");
+ // Keep more distinct rows than shuffle ranges so duplicate maximum
sample boundaries do
+ // not randomly send every row to one writer. Both readers still see
overlapping ranges.
+ List<String> oddValues = new ArrayList<>();
+ List<String> evenValues = new ArrayList<>();
+ for (int id = 39; id >= 1; id -= 2) {
+ oddValues.add("(" + id + ", 'value-" + id + "', 'A')");
+ evenValues.add("(" + (id + 1) + ", 'value-" + (id + 1) + "',
'A')");
+ }
+ sql("INSERT INTO T VALUES " + String.join(", ", oddValues));
+ updateFirstGroupNames(paimonTable("T"));
+ sql("INSERT INTO T VALUES " + String.join(", ", evenValues));
+ sql("INSERT INTO T VALUES (41, 'value-41', 'B')");
+ sql("DELETE FROM T WHERE id = 2");
+ createGlobalIndex();
+
+ FileStoreTable table = paimonTable("T");
+ long nextRowId = table.latestSnapshot().get().nextRowId();
+ long beforeSnapshot = table.latestSnapshot().get().id();
+ assertThat(globalIndexPartitions(table)).containsExactly("A", "B");
+
+ sql("CALL sys.compact(`table` => 'default.T', partitions => 'dt=A')");
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(beforeSnapshot);
+ sql(
+ "CALL sys.compact(`table` => 'default.T', partitions =>
'dt=A', "
+ + "compact_strategy => 'full'"
+ + (ingestionOption == null ? "" : ", options => '" +
ingestionOption + "'")
+ + ")");
+
+ List<Row> expectedRows = new ArrayList<>();
+ List<Row> expectedRowIds = new ArrayList<>();
+ for (int id = 1; id <= 40; id++) {
+ if (id != 2) {
+ expectedRows.add(Row.of(id, (id % 2 == 1 ? "updated-" :
"value-") + id, "A"));
+ expectedRowIds.add(Row.of(nextRowId + expectedRowIds.size()));
+ }
+ }
+ expectedRows.add(Row.of(41, "value-41", "B"));
+ assertThat(sql("SELECT id, name, dt FROM T ORDER BY id"))
+ .containsExactlyElementsOf(expectedRows);
+ assertThat(sql("SELECT _ROW_ID FROM T$row_tracking " + "WHERE dt = 'A'
ORDER BY _ROW_ID"))
+ .containsExactlyElementsOf(expectedRowIds);
+ assertClusteredFileOrder(table, mode);
+ assertThat(deletionVectorCardinality(table)).isZero();
+ assertThat(globalIndexPartitions(table)).containsExactly("B");
+ sql("ALTER TABLE T SET ('scalar-index.search-mode' = 'full')");
+ assertThat(sql("SELECT id FROM T WHERE name =
'updated-3'")).containsExactly(Row.of(3));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"global-sort", "local-sort"})
+ public void testIncrementalClusteringPreservesInlineBlobViews(String mode)
throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql(
+ "CREATE TABLE upstream_blob (id INT, picture BYTES) WITH ("
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'blob-field' = 'picture', 'sink.parallelism' =
'1')");
+ sql("INSERT INTO upstream_blob VALUES (3, X'03'), (1, X'01'), (5,
X'05')");
+ sql("INSERT INTO upstream_blob VALUES (6, X'06'), (4, X'04'), (2,
CAST(NULL AS BYTES))");
+ sql(
+ "CREATE TABLE blob_views (id INT, image_ref BYTES) WITH ("
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'blob-view-field' = 'image_ref', "
+ + "'clustering.incremental' = 'true', "
+ + "'clustering.strategy' = 'order',
'clustering.columns' = 'id', "
+ + "'clustering.incremental.mode' = '"
+ + mode
+ + "', 'scan.parallelism' = '2',
'source.split.target-size' = '1 b', "
+ + "'sink.parallelism' = '1')");
+ for (String ids : new String[] {"1, 3, 5", "2, 4, 6"}) {
+ sql(
+ "INSERT INTO blob_views SELECT id, "
+ + "sys.blob_view('default.upstream_blob',
'picture', _ROW_ID) "
+ + "FROM `upstream_blob$row_tracking` WHERE id IN ("
+ + ids
+ + ")");
+ }
+ String rawReferences =
+ "SELECT id, image_ref FROM blob_views "
+ + "/*+ OPTIONS('blob-view.resolve.enabled'='false') */
ORDER BY id";
+ List<Row> references = sql(rawReferences);
+ assertThat(references).hasSize(6);
+ assertThat(references).allSatisfy(row ->
assertThat(row.getField(1)).isNotNull());
+ FileStoreTable table = paimonTable("blob_views");
+ long nextRowId = table.latestSnapshot().get().nextRowId();
+
+ sql("CALL sys.compact(`table` => 'default.blob_views',
compact_strategy => 'full')");
+
+ // A currently null upstream value must not erase the non-null stored
reference.
+ assertThat(sql(rawReferences)).containsExactlyElementsOf(references);
+ assertThat(sql("SELECT id, image_ref FROM blob_views ORDER BY id"))
+ .containsExactly(
+ Row.of(1, new byte[] {1}),
+ Row.of(2, null),
+ Row.of(3, new byte[] {3}),
+ Row.of(4, new byte[] {4}),
+ Row.of(5, new byte[] {5}),
+ Row.of(6, new byte[] {6}));
+ assertThat(sql("SELECT _ROW_ID FROM blob_views$row_tracking ORDER BY
_ROW_ID"))
+ .containsExactly(
+ Row.of(nextRowId),
+ Row.of(nextRowId + 1),
+ Row.of(nextRowId + 2),
+ Row.of(nextRowId + 3),
+ Row.of(nextRowId + 4),
+ Row.of(nextRowId + 5));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"global-sort,parquet", "local-sort,parquet",
"global-sort,row", "local-sort,row"})
+ public void testIncrementalClusteringPreservesInlineVectors(String mode,
String format)
+ throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql(
+ "CREATE TABLE vectors (id INT, embedding ARRAY<FLOAT>) WITH ("
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'file.format' = '"
+ + format
+ + "', "
+ + "'vector-field' = 'embedding',
'field.embedding.vector-dim' = '3', "
+ + "'clustering.incremental' = 'true', "
+ + "'clustering.strategy' = 'order',
'clustering.columns' = 'id', "
+ + "'clustering.incremental.mode' = '"
+ + mode
+ + "', 'scan.parallelism' = '2',
'source.split.target-size' = '1 b', "
+ + "'sink.parallelism' = '1')");
+ sql(
+ "INSERT INTO vectors VALUES "
+ + "(3, CAST(ARRAY[0, 3, 4] AS ARRAY<FLOAT>)), "
+ + "(1, CAST(ARRAY[0, 1, 2] AS ARRAY<FLOAT>)), "
+ + "(5, CAST(ARRAY[0, 5, 6] AS ARRAY<FLOAT>))");
+ sql(
+ "INSERT INTO vectors VALUES "
+ + "(6, CAST(ARRAY[0, 6, 7] AS ARRAY<FLOAT>)), "
+ + "(4, CAST(ARRAY[0, 4, 5] AS ARRAY<FLOAT>)), "
+ + "(2, CAST(NULL AS ARRAY<FLOAT>))");
+ FileStoreTable table = paimonTable("vectors");
+ long nextRowId = table.latestSnapshot().get().nextRowId();
+
+ sql("CALL sys.compact(`table` => 'default.vectors', compact_strategy
=> 'full')");
+
+ List<String> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ table.newRead().createReader(table.newScan().plan())) {
+ reader.forEachRemaining(
+ row ->
+ rows.add(
+ row.getInt(0)
+ + "|"
+ + (row.isNullAt(1)
+ ? "null"
+ : Arrays.toString(
+
row.getVector(1).toFloatArray()))));
+ }
+ Collections.sort(rows);
+ assertThat(rows)
+ .containsExactly(
+ "1|[0.0, 1.0, 2.0]",
+ "2|null",
+ "3|[0.0, 3.0, 4.0]",
+ "4|[0.0, 4.0, 5.0]",
+ "5|[0.0, 5.0, 6.0]",
+ "6|[0.0, 6.0, 7.0]");
+ assertThat(sql("SELECT _ROW_ID FROM vectors$row_tracking ORDER BY
_ROW_ID"))
+ .containsExactly(
+ Row.of(nextRowId),
+ Row.of(nextRowId + 1),
+ Row.of(nextRowId + 2),
+ Row.of(nextRowId + 3),
+ Row.of(nextRowId + 4),
+ Row.of(nextRowId + 5));
+ }
+
+ @Test
+ public void testIncrementalClusteringRemovesFullyDeletedPartition() throws
Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ createTable();
+ sql(
+ "ALTER TABLE T SET ("
+ + "'clustering.incremental' = 'true', "
+ + "'clustering.strategy' = 'order', "
+ + "'clustering.columns' = 'id', "
+ + "'scan.parallelism' = '2')");
+ sql("INSERT INTO T VALUES (3, 'three', 'A'), (1, 'one', 'A'), (2,
'two', 'A')");
+ sql("INSERT INTO T VALUES (4, 'four', 'B')");
+ createGlobalIndex();
+ sql("DELETE FROM T WHERE id IN (1, 2, 3)");
+ FileStoreTable table = paimonTable("T");
+ assertThat(deletionVectorCardinality(table)).isEqualTo(3L);
+ long nextRowId = table.latestSnapshot().get().nextRowId();
+
+ sql(
+ "CALL sys.compact(`table` => 'default.T', partitions =>
'dt=A', "
+ + "compact_strategy => 'full')");
+
+ assertThat(sql("SELECT id, name, dt FROM
T")).containsExactly(Row.of(4, "four", "B"));
+ assertThat(table.newScan().plan().splits())
+ .allSatisfy(
+ split ->
+ assertThat(((DataSplit)
split).partition().getString(0).toString())
+ .isEqualTo("B"));
+ assertThat(deletionVectorCardinality(table)).isZero();
+ assertThat(globalIndexPartitions(table)).containsExactly("B");
+
assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(nextRowId);
+ }
+
@Test
public void testMaterializeDeletionVectorsForEmptyTable() throws Exception
{
tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
@@ -238,6 +491,88 @@ public class DataEvolutionDeleteSqlITCase extends
CatalogITCaseBase {
assertThat(sql("SELECT id, name, dt FROM
T")).containsExactly(Row.of(1, "old", "A"));
}
+ @SuppressWarnings("unchecked")
+ private static void updateFirstGroupNames(FileStoreTable table) throws
Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ BinaryRow partition = ((DataSplit)
table.newScan().plan().splits().get(0)).partition();
+ try (BatchTableWrite write =
+ builder.newWrite()
+ .withWriteType(
+
table.rowType().project(Arrays.asList("name", "dt")));
+ BatchTableCommit commit = builder.newCommit()) {
+ // Partial rows do not contain the partition field at its
full-schema position.
+ FileStoreWrite<InternalRow> fileStoreWrite =
+ ((TableWriteImpl<InternalRow>) write).getWrite();
+ for (int id = 39; id >= 1; id -= 2) {
+ fileStoreWrite.write(
+ partition,
+ 0,
+ GenericRow.of(
+ BinaryString.fromString("updated-" + id),
+ BinaryString.fromString("A")));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ for (CommitMessage message : messages) {
+ List<DataFileMeta> files =
+ ((CommitMessageImpl)
message).newFilesIncrement().newFiles();
+ List<DataFileMeta> assigned =
+ files.stream()
+ .map(file -> file.assignFirstRowId(0L))
+ .collect(Collectors.toList());
+ files.clear();
+ files.addAll(assigned);
+ }
+ commit.commit(messages);
+ }
+ }
+
+ private static void assertClusteredFileOrder(FileStoreTable table, String
mode)
+ throws Exception {
+ List<DataFileMeta> files = new ArrayList<>();
+ List<List<Integer>> fileIds = new ArrayList<>();
+ for (Split split : table.newScan().plan().splits()) {
+ DataSplit dataSplit = (DataSplit) split;
+ if (!dataSplit.partition().getString(0).toString().equals("A")) {
+ continue;
+ }
+ for (DataFileMeta file : dataSplit.dataFiles()) {
+ files.add(file);
+ DataSplit fileSplit =
+ DataSplit.builder()
+ .withSnapshot(dataSplit.snapshotId())
+ .withPartition(dataSplit.partition())
+ .withBucket(dataSplit.bucket())
+ .withBucketPath(dataSplit.bucketPath())
+ .withDataFiles(Collections.singletonList(file))
+ .build();
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
table.newRead().createReader(fileSplit)) {
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ }
+ assertThat(ids).isNotEmpty().isSorted();
+ fileIds.add(ids);
+ }
+ }
+ // Verify ordering across actual output files, as well as within each
file.
+ assertThat(files).hasSizeGreaterThanOrEqualTo(2);
+ assertThat(files).allSatisfy(file ->
assertThat(file.level()).isGreaterThan(0));
+ fileIds.sort(Comparator.comparingInt(ids -> ids.get(0)));
+ boolean overlappingRanges = false;
+ for (int i = 1; i < fileIds.size(); i++) {
+ List<Integer> previous = fileIds.get(i - 1);
+ overlappingRanges |= previous.get(previous.size() - 1) >=
fileIds.get(i).get(0);
+ }
+ if (mode.equals("global-sort")) {
+ assertThat(overlappingRanges)
+ .as("global-sort output file ranges must not overlap")
+ .isFalse();
+ } else {
+ assertThat(overlappingRanges)
+ .as("local-sort preserves the interleaved input ranges
across readers")
+ .isTrue();
+ }
+ }
+
private void createGlobalIndex() {
sql(
"CALL sys.create_global_index(`table` => 'default.T', "
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/IncrementalClusterActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/IncrementalClusterActionITCase.java
index 9914bc68d3..a9da8ef0f3 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/IncrementalClusterActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/IncrementalClusterActionITCase.java
@@ -64,10 +64,23 @@ import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** IT cases for incremental clustering action. */
public class IncrementalClusterActionITCase extends ActionITCaseBase {
+ @Test
+ public void testStreamingDataEvolutionClusteringIsRejected() throws
Exception {
+ Map<String, String> options = commonOptions();
+ options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ createTable(null, options);
+
+ assertThatThrownBy(() -> runAction(true, Collections.emptyList()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Data evolution incremental clustering only
supports batch mode yet.");
+ }
+
@Test
public void testClusterUnpartitionedTable() throws Exception {
FileStoreTable table = createTable(null);
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 7acca09d85..ada67b342b 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
@@ -37,6 +37,7 @@ import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.operation.BaseAppendFileStoreWrite;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.partition.PartitionValuesTimeExpireStrategy;
+import org.apache.paimon.spark.SparkTable;
import org.apache.paimon.spark.SparkUtils;
import
org.apache.paimon.spark.commands.DataEvolutionCompactMergeConflictRewriter;
import org.apache.paimon.spark.commands.PaimonSparkWriter;
@@ -75,6 +76,7 @@ import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation;
+import org.apache.spark.sql.paimon.shims.SparkShimLoader;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
@@ -104,7 +106,6 @@ import java.util.function.Predicate;
import java.util.stream.Collectors;
import scala.collection.JavaConverters;
-import scala.collection.Seq;
import static org.apache.paimon.CoreOptions.createCommitUser;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
@@ -328,16 +329,20 @@ public class CompactProcedure extends BaseProcedure {
javaSparkContext);
break;
case BUCKET_UNAWARE:
- if (table.coreOptions().dataEvolutionEnabled()) {
+ if (clusterIncrementalEnabled) {
+ clusterIncrementalUnAwareBucketTable(
+ table,
+ partitionPredicate,
+ partitionIdleTime,
+ fullCompact,
+ relation);
+ } else if (table.coreOptions().dataEvolutionEnabled()) {
compactDataEvolutionTable(
table,
relation,
partitionPredicate,
partitionIdleTime,
javaSparkContext);
- } else if (clusterIncrementalEnabled) {
- clusterIncrementalUnAwareBucketTable(
- table, partitionPredicate, fullCompact,
relation);
} else {
compactUnAwareBucketTable(
table, partitionPredicate, partitionIdleTime,
javaSparkContext);
@@ -841,22 +846,57 @@ public class CompactProcedure extends BaseProcedure {
private void clusterIncrementalUnAwareBucketTable(
FileStoreTable table,
@Nullable PartitionPredicate partitionPredicate,
+ @Nullable Duration partitionIdleTime,
boolean fullCompaction,
DataSourceV2Relation relation) {
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ if (snapshot == null) {
+ return;
+ }
IncrementalClusterManager incrementalClusterManager =
- new IncrementalClusterManager(
- table, partitionPredicate,
table.snapshotManager().latestSnapshot());
+ new IncrementalClusterManager(table, partitionPredicate,
snapshot);
Map<BinaryRow, CompactUnit> compactUnits =
incrementalClusterManager.createCompactUnits(fullCompaction);
+ if (partitionIdleTime != null) {
+ Set<BinaryRow> idlePartitions =
+ getPartitionsToCompact(
+ table.newSnapshotReader().withSnapshot(snapshot),
partitionIdleTime);
+ compactUnits.keySet().retainAll(idlePartitions);
+ }
LOG.info(
"Planned {} compact units to incrementally cluster for table
{}, full compaction {}.",
compactUnits.size(),
table.fullName(),
fullCompaction);
+ if (compactUnits.isEmpty()) {
+ return;
+ }
+
Map<BinaryRow, Pair<List<DataSplit>, CommitMessage>> partitionSplits =
incrementalClusterManager.toSplitsAndRewriteDvFiles(compactUnits);
+ final FileStoreTable clusterTable;
+ final DataSourceV2Relation clusterRelation;
+ if (table.coreOptions().dataEvolutionEnabled()) {
+ Map<String, String> clusterOptions = new HashMap<>();
+ // Preserve inline references and the layout already produced by
the sorter.
+ clusterOptions.put(CoreOptions.BLOB_AS_DESCRIPTOR.key(), "true");
+ clusterOptions.put(CoreOptions.BLOB_VIEW_RESOLVE_ENABLED.key(),
"false");
+ clusterOptions.put(CoreOptions.PARTITION_SINK_STRATEGY.key(),
"none");
+
clusterOptions.put(CoreOptions.CLUSTERING_INCREMENTAL_OPTIMIZE_WRITE.key(),
"false");
+ clusterTable = table.copy(clusterOptions);
+ clusterRelation =
+ SparkShimLoader.shim()
+ .copyDataSourceV2Relation(
+ relation,
+ ((SparkTable)
relation.table()).copy(clusterTable),
+ relation.output());
+ } else {
+ clusterTable = table;
+ clusterRelation = relation;
+ }
+
// sort in partition
TableSorter sorter =
TableSorter.getSorter(
@@ -876,6 +916,7 @@ public class CompactProcedure extends BaseProcedure {
Dataset<Row> datasetForWrite =
partitionSplits.values().stream()
.map(Pair::getKey)
+ .filter(splits -> !splits.isEmpty())
.map(
splits -> {
Dataset<Row> dataset =
@@ -883,7 +924,7 @@ public class CompactProcedure extends BaseProcedure {
spark(),
ScanPlanHelper$.MODULE$.createNewScanPlan(
splits.toArray(new
DataSplit[0]),
- relation));
+ clusterRelation));
// Use sortLocal() for LOCAL_SORT, sort()
for GLOBAL_SORT
if (mode ==
CoreOptions.ClusteringIncrementalMode.LOCAL_SORT) {
return sorter.sortLocal(dataset);
@@ -893,73 +934,82 @@ public class CompactProcedure extends BaseProcedure {
})
.reduce(Dataset::union)
.orElse(null);
- if (datasetForWrite != null) {
- // set to write only to prevent invoking compaction
- // do not use overwrite, we don't need to overwrite the whole
partition
- PaimonSparkWriter writer =
PaimonSparkWriter.apply(table).writeOnly();
- Seq<CommitMessage> commitMessages = writer.write(datasetForWrite);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Commit messages after writing:{}", commitMessages);
- }
+ // set to write only to prevent invoking compaction
+ // do not use overwrite, we don't need to overwrite the whole partition
+ PaimonSparkWriter writer =
PaimonSparkWriter.apply(clusterTable).writeOnly();
+ List<CommitMessage> commitMessages =
+ datasetForWrite == null
+ ? Collections.emptyList()
+ :
JavaConverters.seqAsJavaList(writer.write(datasetForWrite));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Commit messages after writing:{}", commitMessages);
+ }
- // re-organize the commit messages to generate the compact messages
- Map<BinaryRow, List<DataFileMeta>> partitionClustered = new
HashMap<>();
- for (CommitMessage commitMessage :
JavaConverters.seqAsJavaList(commitMessages)) {
- checkArgument(commitMessage.bucket() == 0);
- partitionClustered
- .computeIfAbsent(commitMessage.partition(), k -> new
ArrayList<>())
- .addAll(((CommitMessageImpl)
commitMessage).newFilesIncrement().newFiles());
- }
+ // re-organize the commit messages to generate the compact messages
+ Map<BinaryRow, List<DataFileMeta>> partitionClustered = new
HashMap<>();
+ for (CommitMessage commitMessage : commitMessages) {
+ checkArgument(commitMessage.bucket() == 0);
+ partitionClustered
+ .computeIfAbsent(commitMessage.partition(), k -> new
ArrayList<>())
+ .addAll(((CommitMessageImpl)
commitMessage).newFilesIncrement().newFiles());
+ }
- List<CommitMessage> clusterMessages = new ArrayList<>();
- for (Map.Entry<BinaryRow, List<DataFileMeta>> entry :
partitionClustered.entrySet()) {
- BinaryRow partition = entry.getKey();
- CommitMessageImpl dvCommitMessage =
- (CommitMessageImpl)
partitionSplits.get(partition).getValue();
- List<DataFileMeta> clusterBefore =
compactUnits.get(partition).files();
- // upgrade the clustered file to outputLevel
- List<DataFileMeta> clusterAfter =
- IncrementalClusterManager.upgrade(
- entry.getValue(),
compactUnits.get(partition).outputLevel());
- LOG.info(
- "Partition {}: upgrade file level to {}",
- partition,
- compactUnits.get(partition).outputLevel());
-
- List<IndexFileMeta> newIndexFiles = new ArrayList<>();
- List<IndexFileMeta> deletedIndexFiles = new ArrayList<>();
- if (dvCommitMessage != null) {
- newIndexFiles =
dvCommitMessage.compactIncrement().newIndexFiles();
- deletedIndexFiles =
dvCommitMessage.compactIncrement().deletedIndexFiles();
- }
-
- // get the dv index messages
- CompactIncrement compactIncrement =
- new CompactIncrement(
- clusterBefore,
- clusterAfter,
- Collections.emptyList(),
- newIndexFiles,
- deletedIndexFiles);
- clusterMessages.add(
- new CommitMessageImpl(
- partition,
- // bucket 0 is bucket for unaware-bucket table
- // for compatibility with the old design
- 0,
- table.coreOptions().bucket(),
- DataIncrement.emptyIncrement(),
- compactIncrement));
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("Commit messages after reorganizing:{}",
clusterMessages);
+ List<CommitMessage> clusterMessages = new ArrayList<>();
+ for (Map.Entry<BinaryRow, CompactUnit> entry :
compactUnits.entrySet()) {
+ BinaryRow partition = entry.getKey();
+ CommitMessageImpl dvCommitMessage =
+ (CommitMessageImpl)
partitionSplits.get(partition).getValue();
+ List<DataFileMeta> clusterBefore = entry.getValue().files();
+ // upgrade the clustered file to outputLevel
+ List<DataFileMeta> clusterAfter =
+ IncrementalClusterManager.upgrade(
+ partitionClustered.getOrDefault(partition,
Collections.emptyList()),
+ entry.getValue().outputLevel());
+ LOG.info(
+ "Partition {}: upgrade file level to {}",
+ partition,
+ compactUnits.get(partition).outputLevel());
+
+ List<IndexFileMeta> newIndexFiles = new ArrayList<>();
+ List<IndexFileMeta> deletedIndexFiles = new ArrayList<>();
+ if (dvCommitMessage != null) {
+ newIndexFiles =
dvCommitMessage.compactIncrement().newIndexFiles();
+ deletedIndexFiles =
dvCommitMessage.compactIncrement().deletedIndexFiles();
}
-
writer.commit(JavaConverters.asScalaBuffer(clusterMessages).toSeq());
+ // get the dv index messages
+ CompactIncrement compactIncrement =
+ new CompactIncrement(
+ clusterBefore,
+ clusterAfter,
+ Collections.emptyList(),
+ newIndexFiles,
+ deletedIndexFiles);
+ clusterMessages.add(
+ new CommitMessageImpl(
+ partition,
+ // bucket 0 is bucket for unaware-bucket table
+ // for compatibility with the old design
+ 0,
+ table.coreOptions().bucket(),
+ DataIncrement.emptyIncrement(),
+ compactIncrement));
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Commit messages after reorganizing:{}",
clusterMessages);
+ }
+
+ if (table.coreOptions().dataEvolutionEnabled()) {
+ List<CommitMessage> preparedMessages =
+
incrementalClusterManager.prepareCommitMessages(clusterMessages);
+ String commitUser =
createCommitUser(table.coreOptions().toConfiguration());
+ try (TableCommitImpl commit =
table.newCommit(commitUser).materializeDvRowIdCheck()) {
+ commit.commit(preparedMessages);
+ } catch (Exception e) {
+ throw new RuntimeException("Commit data evolution clustering
failed.", e);
+ }
} else {
- LOG.info(
- "No data split to incrementally cluster for table {}, skip
this compact job.",
- table.fullName());
+
writer.commit(JavaConverters.asScalaBuffer(clusterMessages).toSeq());
}
}
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 9364ad824b..2df8a5936c 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
@@ -18,10 +18,10 @@
package org.apache.paimon.spark.procedure
-import org.apache.paimon.Snapshot
+import org.apache.paimon.{CoreOptions, Snapshot}
import org.apache.paimon.Snapshot.CommitKind
import org.apache.paimon.append.dataevolution.{DataEvolutionCompactTask,
DataEvolutionNormalCompactTask}
-import org.apache.paimon.data.BinaryRow
+import org.apache.paimon.data.{BinaryRow, BlobDescriptor, BlobViewStruct}
import
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX
import org.apache.paimon.format.blob.BlobFileFormat
import org.apache.paimon.fs.Path
@@ -2678,6 +2678,319 @@ abstract class CompactProcedureTestBase extends
PaimonSparkTestBase with StreamT
}
}
+ test("Paimon Procedure: data evolution clustering rewrites partial updates
and deletions") {
+ withTable("T") {
+ sql("""
+ |CREATE TABLE T (id INT, value INT)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'write-only' = 'true',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'clustering.incremental' = 'true',
+ | 'clustering.strategy' = 'order',
+ | 'clustering.columns' = 'value',
+ | 'num-sorted-run.compaction-trigger' = '2')
+ |""".stripMargin)
+ sql("INSERT INTO T SELECT /*+ REPARTITION(1) */ * FROM VALUES (3, 30),
(1, 10)")
+ sql("INSERT INTO T VALUES (2, 20), (4, 40)")
+ sql("UPDATE T SET value = 5 WHERE id = 2")
+ sql("DELETE FROM T WHERE id = 4")
+ val table = loadTable("T")
+ val nextRowId = table.latestSnapshot().get().nextRowId().longValue()
+
+ checkAnswer(sql("CALL sys.compact(table => 'T')"), Row(true) :: Nil)
+
+ checkAnswer(sql("SELECT id, value FROM T"), Seq(Row(2, 5), Row(1, 10),
Row(3, 30)))
+ assertDataEvolutionFileOrder("T", "value")
+ assert(
+ sql("SELECT _ROW_ID FROM T ORDER BY _ROW_ID")
+ .collect()
+ .map(_.getLong(0))
+ .sameElements(nextRowId until nextRowId + 3))
+ Assertions.assertThat(deletionVectorCardinality(table)).isZero
+ assert(normalDataFiles(table).forall(_.level() > 0))
+
+ // New partial updates must use the row IDs assigned by clustering.
+ sql("UPDATE T SET value = 15 WHERE id = 2")
+ checkAnswer(sql("SELECT id, value FROM T"), Seq(Row(1, 10), Row(2, 15),
Row(3, 30)))
+ sql("INSERT INTO T VALUES (5, 0)")
+ checkAnswer(sql("CALL sys.compact(table => 'T')"), Row(true) :: Nil)
+ checkAnswer(
+ sql("SELECT id, value FROM T"),
+ Seq(Row(5, 0), Row(1, 10), Row(2, 15), Row(3, 30)))
+ assertDataEvolutionFileOrder("T", "value")
+ }
+ }
+
+ test("Paimon Procedure: data evolution clustering preserves inline blob
descriptors") {
+ withTable("cluster_descriptor") {
+ val blobData = "descriptor content".getBytes("UTF-8")
+ val externalFile = new java.io.File(tempDBDir,
"clustering_descriptor_blob")
+ java.nio.file.Files.write(externalFile.toPath, blobData)
+ val descriptor = new BlobDescriptor(externalFile.toURI.toString, 0,
blobData.length)
+ val descriptorHex = descriptor.serialize().map(value => f"${value &
0xff}%02X").mkString
+ sql("""
+ |CREATE TABLE cluster_descriptor (id INT, picture BINARY)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'blob-descriptor-field' = 'picture',
+ | 'clustering.incremental' = 'true',
+ | 'clustering.columns' = 'id')
+ |""".stripMargin)
+ sql(s"INSERT INTO cluster_descriptor VALUES (3, X'$descriptorHex'), (1,
NULL)")
+ val nextRowId =
loadTable("cluster_descriptor").latestSnapshot().get().nextRowId().longValue()
+
+ sql("CALL sys.compact(table => 'cluster_descriptor', compact_strategy =>
'full')")
+
+ checkAnswer(
+ sql("SELECT id, picture FROM cluster_descriptor"),
+ Seq(Row(1, null), Row(3, blobData)))
+ val descriptorOption =
+
s"spark.paimon.paimon.$dbName0.cluster_descriptor.${CoreOptions.BLOB_AS_DESCRIPTOR.key()}"
+ withSparkSQLConf(descriptorOption -> "true") {
+ val rows = sql("SELECT picture FROM cluster_descriptor ORDER BY
id").collect()
+ assert(rows(0).isNullAt(0))
+ assert(util.Arrays.equals(rows(1).getAs[Array[Byte]](0),
descriptor.serialize()))
+ }
+ assert(
+ sql("SELECT _ROW_ID FROM
cluster_descriptor").collect().forall(_.getLong(0) >= nextRowId))
+ assertDataEvolutionFileOrder("cluster_descriptor", "id")
+ assert(normalDataFiles(loadTable("cluster_descriptor")).forall(_.level()
> 0))
+ }
+ }
+
+ test("Paimon Procedure: data evolution clustering preserves inline blob
views") {
+ withTable("cluster_view_source", "cluster_view") {
+ sql("""
+ |CREATE TABLE cluster_view_source (id INT, picture BINARY)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'blob-field' = 'picture')
+ |""".stripMargin)
+ sql("INSERT INTO cluster_view_source VALUES (3, X'48656C6C6F'), (1,
NULL)")
+ sql("""
+ |CREATE TABLE cluster_view (id INT, picture BINARY)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'blob-view-field' = 'picture',
+ | 'clustering.incremental' = 'true',
+ | 'clustering.columns' = 'id')
+ |""".stripMargin)
+ sql(
+ s"INSERT INTO cluster_view SELECT id, " +
+ s"sys.blob_view('$dbName0.cluster_view_source', 'picture', _ROW_ID)
" +
+ "FROM `cluster_view_source$row_tracking`")
+ sql("INSERT INTO cluster_view VALUES (2, NULL)")
+ val preserveViewOption =
+
s"spark.paimon.paimon.$dbName0.cluster_view.${CoreOptions.BLOB_VIEW_RESOLVE_ENABLED.key()}"
+ var originalReferences = Array.empty[Row]
+ withSparkSQLConf(preserveViewOption -> "false") {
+ originalReferences = sql("SELECT picture FROM cluster_view ORDER BY
id").collect()
+ }
+ // A non-null view of an upstream NULL must remain a reference after the
rewrite.
+ assert(!originalReferences(0).isNullAt(0))
+ assert(originalReferences(1).isNullAt(0))
+ assert(
+ BlobViewStruct
+ .deserialize(originalReferences(0).getAs[Array[Byte]](0))
+ .identifier()
+ .getFullName == s"$dbName0.cluster_view_source")
+ val nextRowId =
loadTable("cluster_view").latestSnapshot().get().nextRowId().longValue()
+
+ sql("CALL sys.compact(table => 'cluster_view', compact_strategy =>
'full')")
+
+ checkAnswer(
+ sql("SELECT id, picture FROM cluster_view"),
+ Seq(Row(1, null), Row(2, null), Row(3, "Hello".getBytes("UTF-8"))))
+ withSparkSQLConf(preserveViewOption -> "false") {
+ val rewrittenReferences = sql("SELECT picture FROM cluster_view ORDER
BY id").collect()
+ assert(rewrittenReferences.length == originalReferences.length)
+ rewrittenReferences.zip(originalReferences).foreach {
+ case (rewritten, original) =>
+ assert(
+ util.Arrays.equals(rewritten.getAs[Array[Byte]](0),
original.getAs[Array[Byte]](0)))
+ }
+ }
+ assert(sql("SELECT _ROW_ID FROM
cluster_view").collect().forall(_.getLong(0) >= nextRowId))
+ assertDataEvolutionFileOrder("cluster_view", "id")
+ assert(normalDataFiles(loadTable("cluster_view")).forall(_.level() > 0))
+ }
+ }
+
+ test("Paimon Procedure: data evolution historical clustering applies
partition idle filter") {
+ withTable("T") {
+ sql("""
+ |CREATE TABLE T (id INT, pt STRING)
+ |PARTITIONED BY (pt)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'write-only' = 'true',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'clustering.incremental' = 'true',
+ | 'clustering.strategy' = 'order',
+ | 'clustering.columns' = 'id',
+ | 'clustering.history-partition.idle-to-full-sort' = '0 s')
+ |""".stripMargin)
+ sql("INSERT INTO T SELECT /*+ REPARTITION(1) */ * FROM VALUES (3,
'old'), (1, 'old')")
+ sql("INSERT INTO T SELECT /*+ REPARTITION(1) */ * FROM VALUES (4,
'new'), (2, 'new')")
+ val table = loadTable("T")
+ val nextRowId = table.latestSnapshot().get().nextRowId().longValue()
+
+ checkAnswer(
+ sql(
+ "CALL sys.compact(table => 'T', partitions => 'pt=new', " +
+ "compact_strategy => 'full', partition_idle_time => '0 s')"),
+ Row(true) :: Nil)
+
+ checkAnswer(sql("SELECT id FROM T WHERE pt = 'old'"), Seq(Row(1),
Row(3)))
+ checkAnswer(sql("SELECT id FROM T WHERE pt = 'new'"), Seq(Row(2),
Row(4)))
+ assertDataEvolutionFileOrder("T", "id")
+ assert(sql("SELECT _ROW_ID FROM T").collect().forall(_.getLong(0) >=
nextRowId))
+ assert(normalDataFiles(table).forall(_.level() ==
table.coreOptions().numLevels() - 1))
+ }
+ }
+
+ test("Paimon Procedure: data evolution clustering respects minor and full
strategies") {
+ withTable("T") {
+ sql("""
+ |CREATE TABLE T (id INT)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'write-only' = 'true',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'clustering.incremental' = 'true',
+ | 'clustering.strategy' = 'order',
+ | 'clustering.columns' = 'id',
+ | 'num-sorted-run.compaction-trigger' = '10')
+ |""".stripMargin)
+ sql("INSERT INTO T SELECT /*+ REPARTITION(1) */ * FROM VALUES (3), (1),
(2)")
+ val table = loadTable("T")
+ val before = lastSnapshotId(table)
+ checkAnswer(sql("CALL sys.compact(table => 'T')"), Row(true) :: Nil)
+ Assertions.assertThat(lastSnapshotId(table)).isEqualTo(before)
+
+ checkAnswer(
+ sql("CALL sys.compact(table => 'T', compact_strategy => 'full')"),
+ Row(true) :: Nil)
+ Assertions.assertThat(lastSnapshotId(table)).isEqualTo(before + 1)
+ checkAnswer(sql("SELECT id FROM T"), Seq(Row(1), Row(2), Row(3)))
+ assertDataEvolutionFileOrder("T", "id")
+ assert(normalDataFiles(table).forall(_.level() ==
table.coreOptions().numLevels() - 1))
+ }
+ }
+
+ Seq("local-sort", "global-sort").foreach {
+ mode =>
+ Seq("", "partition.sink-strategy=hash",
"clustering.incremental.optimize-write=true")
+ .foreach {
+ writeOption =>
+ val suffix = if (writeOption.isEmpty) "" else s" with $writeOption"
+ test(s"Paimon Procedure: data evolution clustering uses
distributed $mode$suffix") {
+ withSparkSQLConf(
+ "spark.sql.shuffle.partitions" -> "2",
+ "spark.sql.adaptive.enabled" -> "false") {
+ 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.strategy' = 'order',
+ | 'clustering.columns' = 'value',
+ | 'source.split.target-size' = '1 b')
+ |""".stripMargin)
+ (0 until 4).foreach {
+ batch =>
+ sql(s"""
+ |INSERT INTO T
+ |SELECT /*+ REPARTITION(1) */
+ | CAST(59 - (id * 4 + $batch) AS INT),
+ | CAST((59 - (id * 4 + $batch)) * 10 AS INT), 'A'
+ |FROM range(15)
+ |""".stripMargin)
+ }
+ sql("UPDATE T SET value = -2 WHERE id = 2")
+ sql("DELETE FROM T WHERE id = 4")
+ val table = loadTable("T")
+ val nextRowId =
table.latestSnapshot().get().nextRowId().longValue()
+
+ checkAnswer(
+ sql("CALL sys.compact(table => 'T', compact_strategy =>
'full', " +
+ s"options => '$writeOption')"),
+ Row(true) :: Nil)
+
+ val expected =
+ (0 until 60).filter(_ != 4).map(id => Row(id, if (id == 2)
-2 else id * 10))
+ checkAnswer(sql("SELECT id, value FROM T"), expected)
+ val fileValues = assertDataEvolutionFileOrder("T", "value")
+ assert(fileValues.size > 1)
+ val ranges = fileValues.map(values => (values.head,
values.last)).sortBy(_._1)
+ if (mode == "global-sort") {
+ assert(fileValues.size == 2)
+ assert(ranges.sliding(2).forall(pair => pair.head._2 <
pair.last._1))
+ } else {
+ assert(ranges.sliding(2).exists(pair => pair.head._2 >
pair.last._1))
+ }
+ assert(sql("SELECT _ROW_ID FROM
T").collect().forall(_.getLong(0) >= nextRowId))
+ assert(
+ normalDataFiles(table).forall(_.level() ==
table.coreOptions().numLevels() - 1))
+
Assertions.assertThat(deletionVectorCardinality(table)).isZero
+
+ sql("UPDATE T SET value = 12 WHERE id = 2")
+ checkAnswer(sql("SELECT value FROM T WHERE id = 2"), Row(12)
:: Nil)
+ }
+ }
+ }
+ }
+ }
+
+ test("Paimon Procedure: data evolution clustering removes fully deleted
partitions") {
+ withTable("T") {
+ sql("""
+ |CREATE TABLE T (id 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.strategy' = 'order',
+ | 'clustering.columns' = 'id')
+ |""".stripMargin)
+ sql("INSERT INTO T VALUES (3, 'dead'), (1, 'dead'), (4, 'live'), (2,
'live')")
+ sql("DELETE FROM T WHERE id IN (1, 3)")
+ val table = loadTable("T")
+ val beforeFiles = normalDataFiles(table).map(_.fileName()).toSet
+ Assertions.assertThat(deletionVectorCardinality(table)).isEqualTo(2)
+
+ checkAnswer(
+ sql("CALL sys.compact(table => 'T', compact_strategy => 'full')"),
+ Row(true) :: Nil)
+
+ checkAnswer(sql("SELECT id, pt FROM T"), Seq(Row(2, "live"), Row(4,
"live")))
+ assert(normalDataFiles(table).forall(file =>
!beforeFiles.contains(file.fileName())))
+ Assertions.assertThat(deletionVectorCardinality(table)).isZero
+ assertDataEvolutionFileOrder("T", "id")
+ }
+ }
+
test("Paimon Procedure: materialize deletion vectors enables snapshot
expiration") {
withTable("T") {
sql("""
@@ -2822,6 +3135,23 @@ abstract class CompactProcedureTestBase extends
PaimonSparkTestBase with StreamT
.get
}
+ private def assertDataEvolutionFileOrder(tableName: String, columnName:
String): Seq[Seq[Int]] = {
+ val rows = sql(s"SELECT _ROW_ID, `$columnName` FROM $tableName ORDER BY
_ROW_ID")
+ .collect()
+ .map(row => (row.getLong(0), row.getInt(1)))
+ normalDataFiles(loadTable(tableName)).map {
+ file =>
+ val firstRowId = file.nonNullFirstRowId()
+ val values = rows
+ .filter { case (rowId, _) => rowId >= firstRowId && rowId <
firstRowId + file.rowCount() }
+ .map(_._2)
+ .toSeq
+ assert(values.size.toLong == file.rowCount())
+ assert(values == values.sorted)
+ values
+ }
+ }
+
private def normalDataFiles(table: FileStoreTable): Seq[DataFileMeta] = {
table
.newSnapshotReader()
