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 a8010b4ca5 [core] Preserve clustering levels for non-key Data
Evolution updates (#10683)
a8010b4ca5 is described below
commit a8010b4ca576f94491f49cc59c58ec0692421e97
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 20 12:51:01 2026 +0800
[core] Preserve clustering levels for non-key Data Evolution updates
(#10683)
---
.../data-evolution-maintenance.mdx | 8 +-
.../DataEvolutionIncrementalClusterRewriter.java | 54 +++-
...ataEvolutionIncrementalClusterRewriterTest.java | 282 +++++++++++++++++++++
.../dataevolution/DataEvolutionClusterTest.java | 92 ++++++-
.../paimon/flink/DataEvolutionDeleteSqlITCase.java | 99 +++++++-
5 files changed, 526 insertions(+), 9 deletions(-)
diff --git a/docs/docs/multimodal-table/data-evolution-maintenance.mdx
b/docs/docs/multimodal-table/data-evolution-maintenance.mdx
index 9c71595773..25646225fe 100644
--- a/docs/docs/multimodal-table/data-evolution-maintenance.mdx
+++ b/docs/docs/multimodal-table/data-evolution-maintenance.mdx
@@ -183,7 +183,13 @@ 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
+selection. Updating only non-clustering columns does not make an already
clustered
+group eligible again: its row order remains valid, so even a full clustering
call
+can skip it and preserve its row IDs and indexes on unchanged columns.
+Updating a clustering column
+makes the group eligible again. When a group is selected, all its column
versions
+are rewritten together, including newer values of non-clustering columns.
+`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:
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/DataEvolutionIncrementalClusterRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/DataEvolutionIncrementalClusterRewriter.java
index 238c4057ea..f50acb29a0 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/DataEvolutionIncrementalClusterRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/DataEvolutionIncrementalClusterRewriter.java
@@ -18,6 +18,7 @@
package org.apache.paimon.append.cluster;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import
org.apache.paimon.append.dataevolution.DataEvolutionCompactDeletionVectorRewriter;
import org.apache.paimon.data.BinaryRow;
@@ -27,24 +28,30 @@ 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.schema.TableSchema;
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.Pair;
import org.apache.paimon.utils.RangeHelper;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static org.apache.paimon.utils.DataEvolutionUtils.fileFieldIds;
+import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Preserves complete row-id groups and materializes deletion vectors for
clustering. */
@@ -52,10 +59,21 @@ class DataEvolutionIncrementalClusterRewriter extends
IncrementalClusterRewriter
private final FileStoreTable table;
private final Snapshot snapshot;
+ private final List<String> clusterKeys;
+ private final Set<Integer> clusterFieldIds;
+ private final Map<Long, TableSchema> schemas = new HashMap<>();
+ private final Map<Pair<Long, List<String>>, Set<Integer>> writtenFieldIds
= new HashMap<>();
DataEvolutionIncrementalClusterRewriter(FileStoreTable table, Snapshot
snapshot) {
this.table = table;
this.snapshot = snapshot;
+ this.clusterKeys = table.coreOptions().clusteringColumns();
+ this.clusterFieldIds =
+ clusterKeys.stream()
+ // A partition key is constant within a group,
including partial writes.
+ .filter(key -> !table.partitionKeys().contains(key))
+ .map(key -> table.rowType().getField(key).id())
+ .collect(Collectors.toSet());
}
@Override
@@ -73,8 +91,10 @@ class DataEvolutionIncrementalClusterRewriter extends
IncrementalClusterRewriter
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();
+ DataFileMeta anchor = retrieveAnchorFile(group, file -> file);
+ // The shared full planner reads the first file's clustering
schema.
+ Collections.swap(group, 0, group.indexOf(anchor));
+ int level = groupLevel(group, anchor);
if (level == 0) {
runs.add(new LevelSortedRun(0, SortedRun.fromSorted(group)));
} else {
@@ -87,6 +107,36 @@ class DataEvolutionIncrementalClusterRewriter extends
IncrementalClusterRewriter
return runs;
}
+ private int groupLevel(List<DataFileMeta> group, DataFileMeta anchor) {
+ int level = anchor.level();
+ CoreOptions anchorOptions =
CoreOptions.fromMap(schema(anchor.schemaId()).options());
+ if (!fieldIds(anchor).containsAll(clusterFieldIds)
+ || !anchorOptions.clusteringColumns().equals(clusterKeys)
+ || anchorOptions.clusteringStrategy(clusterKeys.size())
+ !=
table.coreOptions().clusteringStrategy(clusterKeys.size())) {
+ // The shared full planner checks only the first file's schema. Do
not let a newer
+ // non-key update or another group hide an incompatible clustering
definition.
+ return 0;
+ }
+ for (DataFileMeta file : group) {
+ if (file.level() < level && !Collections.disjoint(fieldIds(file),
clusterFieldIds)) {
+ // Keep every column version in the run, but only key writes
lower its level.
+ level = file.level();
+ }
+ }
+ return level;
+ }
+
+ private Set<Integer> fieldIds(DataFileMeta file) {
+ return writtenFieldIds.computeIfAbsent(
+ Pair.of(file.schemaId(), file.writeCols()),
+ ignored -> fileFieldIds(schema(file.schemaId()), file));
+ }
+
+ private TableSchema schema(long schemaId) {
+ return schemas.computeIfAbsent(schemaId, id ->
table.schemaManager().schema(id));
+ }
+
@Override
void configureSplit(
DataSplit.Builder builder,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/cluster/DataEvolutionIncrementalClusterRewriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/cluster/DataEvolutionIncrementalClusterRewriterTest.java
new file mode 100644
index 0000000000..41e06cb993
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/append/cluster/DataEvolutionIncrementalClusterRewriterTest.java
@@ -0,0 +1,282 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.append.cluster;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.compact.CompactUnit;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.schema.FileSystemSchemaManager;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.types.DataTypes;
+
+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.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.paimon.stats.SimpleStats.EMPTY_STATS;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Planner behavior for data-evolution column versions. */
+class DataEvolutionIncrementalClusterRewriterTest {
+
+ private static final int NUM_LEVELS = 3;
+ private static final int MAX_LEVEL = NUM_LEVELS - 1;
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testNonKeyVersionsStayClusteredUntilKeyIsWritten(boolean
fullSchemaUpdate)
+ throws Exception {
+ FileStoreTable original =
+ createTable(
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("payload", DataTypes.STRING()),
+ "id",
+ "auto");
+ FileStoreTable table =
+ evolve(
+ original,
+
SchemaChange.setOption(CoreOptions.CLUSTERING_STRATEGY.key(), "order"));
+ List<DataFileMeta> files = new ArrayList<>();
+ for (long firstRowId : new long[] {0, 3}) {
+ files.add(file(original, "base-" + firstRowId, firstRowId, 10,
MAX_LEVEL, null));
+ files.add(
+ file(
+ table,
+ "payload-" + firstRowId,
+ firstRowId,
+ 20,
+ 0,
+ Collections.singletonList("payload"))
+ .withWriteColsSequences(new long[] {20}));
+ }
+
+ // "auto" and "order" resolve to the same strategy for a single key.
Non-key versions
+ // neither create extra runs nor make a full cluster rewrite already
sorted rows.
+ assertThat(pick(table, files, true)).isEmpty();
+ assertThat(pick(table, files, false)).isEmpty();
+
+ files.add(
+ file(
+ table,
+ "key-update",
+ 0,
+ 30,
+ 0,
+ fullSchemaUpdate ? null :
Collections.singletonList("id")));
+ assertSelectedFiles(pick(table, files, true), files);
+ assertSelectedFiles(pick(table, files, false), files);
+ }
+
+ @Test
+ void testAnchorMissingKeyRequiresClustering() throws Exception {
+ FileStoreTable table =
+ createTable(
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("payload", DataTypes.STRING()),
+ "id",
+ "order");
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file(
+ table,
+ "partial-anchor",
+ 0,
+ 10,
+ MAX_LEVEL,
+ Collections.singletonList("payload")),
+ file(
+ table,
+ "later-key",
+ 0,
+ 20,
+ MAX_LEVEL,
+ Collections.singletonList("id")));
+
+ // Neither file has a lower level, but the initial partial rows were
not sorted by id.
+ assertSelectedFiles(pick(table, files, true), files);
+ }
+
+ @Test
+ void testOlderSchemaCompactFileDoesNotWriteNewKey() throws Exception {
+ FileStoreTable original =
+ createTable(
+ Schema.newBuilder().column("payload",
DataTypes.STRING()),
+ "payload",
+ "order");
+ FileStoreTable table =
+ evolve(
+ original,
+ SchemaChange.addColumn("id", DataTypes.INT()),
+
SchemaChange.setOption(CoreOptions.CLUSTERING_COLUMNS.key(), "id"));
+ DataFileMeta anchor = file(table, "new-key-anchor", 0, 10, MAX_LEVEL,
null);
+ DataFileMeta compactedPayload =
+ file(original, "old-schema-compact", 0, 20, 0, null)
+ .assignFileSource(FileSource.COMPACT)
+ .withWriteColsSequences(new long[] {20});
+
+ // Null write-cols means this file's physical schema, not the current
schema. A compact
+ // output from an older schema cannot write a key which was introduced
later.
+ assertThat(pick(table, Arrays.asList(compactedPayload, anchor),
true)).isEmpty();
+ }
+
+ @ParameterizedTest
+ @CsvSource({"payload, order", "id, zorder"})
+ void testFirstCurrentGroupCannotHideDifferentClusteringDefinition(
+ String oldKey, String oldStrategy) throws Exception {
+ FileStoreTable original =
+ createTable(
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("payload", DataTypes.STRING()),
+ oldKey,
+ oldStrategy);
+ FileStoreTable table =
+ evolve(
+ original,
+
SchemaChange.setOption(CoreOptions.CLUSTERING_COLUMNS.key(), "id"),
+
SchemaChange.setOption(CoreOptions.CLUSTERING_STRATEGY.key(), "order"));
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file(table, "current-definition", 0, 10, MAX_LEVEL,
null),
+ file(original, "old-definition", 3, 10, MAX_LEVEL,
null));
+
+ // The shared full planner checks the first file's schema for a single
highest-level run.
+ // Each group's provenance must survive grouping even when a
current-schema group is first.
+ assertSelectedFiles(pick(table, files, true), files);
+ }
+
+ @Test
+ void testPartitionOnlyKeyDoesNotRequirePhysicalColumn() throws Exception {
+ FileStoreTable table =
+ createTable(
+ Schema.newBuilder()
+ .column("pt", DataTypes.STRING())
+ .column("payload", DataTypes.STRING())
+ .partitionKeys("pt"),
+ "pt",
+ "order");
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file(
+ table,
+ "partition-anchor",
+ 0,
+ 10,
+ MAX_LEVEL,
+ Collections.singletonList("payload")),
+ file(
+ table,
+ "payload-update",
+ 0,
+ 20,
+ 0,
+ Collections.singletonList("payload")));
+
+ // A partition key is constant and need not be stored in either column
version.
+ assertThat(pick(table, files, true)).isEmpty();
+ }
+
+ private FileStoreTable createTable(
+ Schema.Builder builder, String clusterColumns, String strategy)
throws Exception {
+ Schema schema =
+ builder.option(CoreOptions.BUCKET.key(), "-1")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.CLUSTERING_INCREMENTAL.key(),
"true")
+ .option(CoreOptions.CLUSTERING_COLUMNS.key(),
clusterColumns)
+ .option(CoreOptions.CLUSTERING_STRATEGY.key(),
strategy)
+ .option(CoreOptions.NUM_LEVELS.key(),
String.valueOf(NUM_LEVELS))
+ .build();
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(tempDir.toString());
+ SchemaManager schemaManager = new FileSystemSchemaManager(fileIO,
path);
+ return FileStoreTableFactory.create(fileIO, path,
schemaManager.createTable(schema));
+ }
+
+ private static FileStoreTable evolve(FileStoreTable table, SchemaChange...
changes)
+ throws Exception {
+ return
table.copy(table.schemaManager().commitChanges(Arrays.asList(changes)));
+ }
+
+ private static DataFileMeta file(
+ FileStoreTable table,
+ String name,
+ long firstRowId,
+ long sequenceNumber,
+ int level,
+ @Nullable List<String> writeCols) {
+ DataFileMeta file =
+ DataFileMeta.forAppend(
+ name + ".parquet",
+ 100,
+ 3,
+ EMPTY_STATS,
+ sequenceNumber,
+ sequenceNumber,
+ table.schema().id(),
+ Collections.emptyList(),
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ firstRowId,
+ writeCols);
+ return level == 0 ? file : file.upgrade(level);
+ }
+
+ private static Optional<CompactUnit> pick(
+ FileStoreTable table, List<DataFileMeta> files, boolean full) {
+ IncrementalClusterStrategy strategy =
+ new IncrementalClusterStrategy(
+ table.schemaManager(),
table.coreOptions().clusteringColumns(), 25, 1, 2);
+ return strategy.pick(
+ NUM_LEVELS,
+ new DataEvolutionIncrementalClusterRewriter(table, null)
+ .constructPartitionLevels(files),
+ full);
+ }
+
+ private static void assertSelectedFiles(
+ Optional<CompactUnit> selected, List<DataFileMeta> expected) {
+ assertThat(selected).isPresent();
+ assertThat(selected.get().outputLevel()).isEqualTo(MAX_LEVEL);
+
assertThat(selected.get().files()).containsExactlyInAnyOrderElementsOf(expected);
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
index 48b60d3ab8..1a0d1da601 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
@@ -226,6 +226,76 @@ class DataEvolutionClusterTest extends TableTestBase {
.containsExactly("a|10|base-20|6", "a|30|base-30|7",
"a|50|base-10|8");
}
+ @Test
+ void testNonClusteringUpdatesDoNotDirtyClusteredGroups() throws Exception {
+ FileStoreTable table = createTable(false);
+ append(table, "a", 60, 10, 50, 30, 20, 40);
+ table =
table.copy(Collections.singletonMap(CoreOptions.TARGET_FILE_ROW_NUM.key(),
"3"));
+ Snapshot initial = table.latestSnapshot().get();
+ commitCluster(table, initial, prepareMappedCluster(table, initial,
true));
+ commit(table, prepareGlobalIndex(table, table.latestSnapshot().get(),
"id"));
+ List<String> indexes = globalIndexNames(table);
+ assertThat(indexes).isNotEmpty();
+
+ updatePayload(table, "a", 6L, 10, 20, 30);
+ updatePayload(table, "a", 9L, 40, 50, 60);
+ Snapshot updated = table.latestSnapshot().get();
+ List<String> files = fileNames(table, "a");
+ assertThat(files).hasSize(4);
+ List<String> latestRows =
+ Arrays.asList(
+ "a|10|updated-10|6",
+ "a|20|updated-20|7",
+ "a|30|updated-30|8",
+ "a|40|updated-40|9",
+ "a|50|updated-50|10",
+ "a|60|updated-60|11");
+ for (boolean fullCompaction : new boolean[] {false, true}) {
+ List<CommitMessage> messages = prepareMappedCluster(table,
updated, fullCompaction);
+ assertThat(messages).isEmpty();
+ commitCluster(table, updated, messages);
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(updated.id());
+
assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(12L);
+ assertThat(fileNames(table, "a")).isEqualTo(files);
+ assertThat(globalIndexNames(table)).isEqualTo(indexes);
+ assertThat(readRows(table)).containsExactlyElementsOf(latestRows);
+ }
+ assertIndexRows(
+ table,
+ new PredicateBuilder(table.rowType())
+ .equal(table.rowType().getFieldIndex("id"), 20),
+ new Range(7, 7));
+
+ append(table, "a", 90, 70, 80);
+ Snapshot appended = table.latestSnapshot().get();
+ IncrementalClusterManager manager = new
IncrementalClusterManager(table, null, appended);
+ Map<BinaryRow, CompactUnit> units = manager.createCompactUnits(true);
+ assertThat(units).containsOnlyKeys(partition(table, "a"));
+ assertThat(units.get(partition(table, "a")).files())
+ .hasSize(5)
+ .extracting(DataFileMeta::fileName)
+ .containsAll(files);
+ commitCluster(table, appended, rewriteClusterUnits(table, manager,
units, true));
+
+ assertThat(readRows(table))
+ .containsExactly(
+ "a|10|updated-10|15",
+ "a|20|updated-20|16",
+ "a|30|updated-30|17",
+ "a|40|updated-40|18",
+ "a|50|updated-50|19",
+ "a|60|updated-60|20",
+ "a|70|base-70|21",
+ "a|80|base-80|22",
+ "a|90|base-90|23");
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(24L);
+ assertIndexRows(
+ table,
+ new PredicateBuilder(table.rowType())
+ .equal(table.rowType().getFieldIndex("id"), 20),
+ new Range(16, 16));
+ }
+
@Test
void testIncrementalClusterWaitsForCompactionThreshold() throws Exception {
FileStoreTable table = createTable(false);
@@ -292,6 +362,7 @@ class DataEvolutionClusterTest extends TableTestBase {
List<String> expected = new ArrayList<>(readRows(table).subList(3, 6));
updatePayload(table, "a", 6L, 10, 20, 30);
+ updateSortKeys(table, "a", 6L, 10, 20, 30);
append(table, "a", 90, 70, 80);
Snapshot snapshot = table.latestSnapshot().get();
IncrementalClusterManager manager = new
IncrementalClusterManager(table, null, snapshot);
@@ -304,7 +375,7 @@ class DataEvolutionClusterTest extends TableTestBase {
.doesNotContain(untouchedBase.fileName());
assertThat(unit.files())
.filteredOn(file -> file.nonNullRowIdRange().equals(new
Range(6, 8)))
- .hasSize(2);
+ .hasSize(3);
commitCluster(table, snapshot, rewriteClusterUnits(table, manager,
units));
expected.addAll(
@@ -1324,6 +1395,8 @@ class DataEvolutionClusterTest extends TableTestBase {
updatePayload(table, "a", 9L, 10, 20, 30);
updatePayload(table, "a", 15L, 70, 80, 90);
+ updateSortKeys(table, "a", 9L, 10, 20, 30);
+ updateSortKeys(table, "a", 15L, 70, 80, 90);
append(table, "a", 120, 100, 110);
if (deletionVectors) {
delete(table, "a", new Range(9, 11), 9L);
@@ -1340,10 +1413,12 @@ class DataEvolutionClusterTest extends TableTestBase {
assertThat(selected)
.extracting(DataFileMeta::nonNullRowIdRange)
.containsExactlyInAnyOrder(
+ new Range(9, 11),
new Range(9, 11),
new Range(9, 11),
new Range(15, 17),
new Range(15, 17),
+ new Range(15, 17),
new Range(18, 20));
assertThat(selected)
.extracting(DataFileMeta::fileName)
@@ -1416,6 +1491,21 @@ class DataEvolutionClusterTest extends TableTestBase {
commit(table, preparePayloadUpdate(table, partition, firstRowId,
values));
}
+ private void updateSortKeys(
+ FileStoreTable table, String partition, long firstRowId, int...
values)
+ throws Exception {
+ RowType writeType = table.rowType().project(Arrays.asList("pt", "id"));
+ try (BatchTableWrite write =
+
table.newBatchWriteBuilder().newWrite().withWriteType(writeType)) {
+ for (int value : values) {
+ write.write(GenericRow.of(BinaryString.fromString(partition),
value));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ assignFirstRowId(messages, firstRowId);
+ commit(table, messages);
+ }
+ }
+
private List<CommitMessage> preparePayloadUpdate(
FileStoreTable table, String partition, long firstRowId, int...
values)
throws Exception {
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 4569ff31df..01e17a2bd9 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
@@ -272,6 +272,83 @@ public class DataEvolutionDeleteSqlITCase extends
CatalogITCaseBase {
.containsExactlyElementsOf(negativeMatches);
}
+ @ParameterizedTest
+ @CsvSource({"btree,global-sort", "bitmap,local-sort"})
+ public void testIncrementalClusteringSkipsNonClusteringColumnUpdates(
+ String indexType, String mode) throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ createTable();
+ sql(
+ "ALTER TABLE T SET ("
+ + "'write-only' = 'true', "
+ + "'global-index.column-update-action' = 'ignore', "
+ + "'clustering.incremental' = 'true', "
+ + "'clustering.strategy' = 'order', "
+ + "'clustering.columns' = 'id', "
+ + "'clustering.incremental.mode' = '"
+ + mode
+ + "', 'scan.parallelism' = '1', 'sink.parallelism' =
'1')");
+ sql("INSERT INTO T VALUES (3, 'three', 'A'), (1, 'one', 'A'), (2,
'two', 'A')");
+ sql("CALL sys.compact(`table` => 'default.T', compact_strategy =>
'full')");
+ sql(
+ "CALL sys.create_global_index(`table` => 'default.T', "
+ + "index_column => 'id', index_type => '"
+ + indexType
+ + "')");
+ FileStoreTable table = paimonTable("T");
+ List<String> indexes = scalarIndexFiles(table, indexType, "A");
+ assertThat(indexes).isNotEmpty();
+ List<String> clusteredFiles = plannedFiles(table);
+ assertThat(clusteredFiles).hasSize(1);
+ long firstRowId =
table.store().newScan().plan().files().get(0).file().nonNullFirstRowId();
+ List<Row> originalIds = sql("SELECT id, _ROW_ID FROM T$row_tracking
ORDER BY id");
+
+ writePartialGroup(table, "name", firstRowId, "updated-one",
"updated-two", "updated-three");
+ Snapshot payloadSnapshot = table.latestSnapshot().get();
+ List<String> payloadFiles = plannedFiles(table);
+ assertThat(payloadFiles).hasSize(2);
+
assertThat(payloadSnapshot.commitKind()).isEqualTo(Snapshot.CommitKind.APPEND);
+
+ sql("CALL sys.compact(`table` => 'default.T', compact_strategy =>
'full')");
+
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(payloadSnapshot.id());
+
assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(payloadSnapshot.nextRowId());
+ assertThat(plannedFiles(table)).isEqualTo(payloadFiles);
+ assertThat(scalarIndexFiles(table, indexType, "A")).isEqualTo(indexes);
+ assertThat(sql("SELECT id, _ROW_ID FROM T$row_tracking ORDER BY id"))
+ .containsExactlyElementsOf(originalIds);
+ // The unchanged id index must still find rows whose other columns
have newer versions.
+ sql("ALTER TABLE T SET ('scalar-index.search-mode' = 'fast')");
+ assertThat(sql("SELECT id, name FROM T WHERE id IN (1, 3) ORDER BY
id"))
+ .containsExactly(Row.of(1, "updated-one"), Row.of(3,
"updated-three"));
+
+ writePartialGroup(table, "id", firstRowId, 30, 10, 20);
+ Snapshot sortKeySnapshot = table.latestSnapshot().get();
+ List<String> beforeRewrite = plannedFiles(table);
+ assertThat(beforeRewrite).hasSize(3);
+ long nextRowId = sortKeySnapshot.nextRowId();
+
+ sql("CALL sys.compact(`table` => 'default.T', compact_strategy =>
'full')");
+
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(sortKeySnapshot.id() +
1);
+ assertThat(table.latestSnapshot().get().commitKind())
+ .isEqualTo(Snapshot.CommitKind.COMPACT);
+
assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(nextRowId + 3);
+
assertThat(plannedFiles(table)).doesNotContainAnyElementsOf(beforeRewrite);
+ assertThat(sql("SELECT id, name, dt FROM T ORDER BY id"))
+ .containsExactly(
+ Row.of(10, "updated-two", "A"),
+ Row.of(20, "updated-three", "A"),
+ Row.of(30, "updated-one", "A"));
+ assertThat(sql("SELECT id, _ROW_ID FROM T$row_tracking ORDER BY id"))
+ .containsExactly(
+ Row.of(10, nextRowId),
+ Row.of(20, nextRowId + 1),
+ Row.of(30, nextRowId + 2));
+ // The sort-key update also changed indexed values, so old keys cannot
be reused.
+ assertThat(scalarIndexFiles(table, indexType, "A")).isEmpty();
+ }
+
@ParameterizedTest
@ValueSource(strings = {"global-sort", "local-sort"})
public void testIncrementalClusteringPreservesInlineBlobViews(String mode)
throws Exception {
@@ -583,24 +660,36 @@ 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 {
+ List<String> names = new ArrayList<>();
+ for (int id = 39; id >= 1; id -= 2) {
+ names.add("updated-" + id);
+ }
+ writePartialGroup(table, "name", 0L, names.toArray());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static void writePartialGroup(
+ FileStoreTable table, String column, long firstRowId, Object...
values)
+ 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")));
+
table.rowType().project(Arrays.asList(column, "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) {
+ for (Object value : values) {
fileStoreWrite.write(
partition,
0,
GenericRow.of(
- BinaryString.fromString("updated-" + id),
+ value instanceof String
+ ? BinaryString.fromString((String)
value)
+ : value,
BinaryString.fromString("A")));
}
List<CommitMessage> messages = write.prepareCommit();
@@ -609,7 +698,7 @@ public class DataEvolutionDeleteSqlITCase extends
CatalogITCaseBase {
((CommitMessageImpl)
message).newFilesIncrement().newFiles();
List<DataFileMeta> assigned =
files.stream()
- .map(file -> file.assignFirstRowId(0L))
+ .map(file -> file.assignFirstRowId(firstRowId))
.collect(Collectors.toList());
files.clear();
files.addAll(assigned);
