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 cdb3e502b3 [flink] Maintain scalar indexes during Data Evolution
clustering (#10682)
cdb3e502b3 is described below
commit cdb3e502b3dc7a9e621e00af2b0dfbdc506c1e93
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 20 12:48:38 2026 +0800
[flink] Maintain scalar indexes during Data Evolution clustering (#10682)
---
.../data-evolution-maintenance.mdx | 20 +-
.../dataevolution/ClusteringIndexRewriter.java | 469 ++++++++++
.../paimon/operation/FileStoreCommitImpl.java | 59 +-
.../dataevolution/DataEvolutionClusterTest.java | 945 ++++++++++++++++++++-
.../paimon/operation/FileStoreCommitTest.java | 32 +-
.../cluster/IncrementalClusterSplitSource.java | 16 +-
.../flink/compact/IncrementalClusterCompact.java | 47 +-
.../paimon/flink/DataEvolutionDeleteSqlITCase.java | 103 ++-
8 files changed, 1643 insertions(+), 48 deletions(-)
diff --git a/docs/docs/multimodal-table/data-evolution-maintenance.mdx
b/docs/docs/multimodal-table/data-evolution-maintenance.mdx
index 682334e99b..9c71595773 100644
--- a/docs/docs/multimodal-table/data-evolution-maintenance.mdx
+++ b/docs/docs/multimodal-table/data-evolution-maintenance.mdx
@@ -201,10 +201,22 @@ 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.
+Clustering removes the applied deletion vectors. Flink clustering also rewrites
+B-tree and bitmap index files with the new row IDs, reusing their stored keys.
+Indexes outside the rewritten row-ID ranges are preserved, including those in
+the same partition. Partially indexed data retains its exact coverage after
sorting.
+The current contiguous-range index protocol limits rewriting to 64 new coverage
+fragments per index definition, and 64 retained plus new fragments per source
file.
+If sorting would exceed either bound, clustering fails before committing;
rebuild
+complete indexes with smaller shards first.
+The job records the old row IDs alongside each output file so that concurrent
+index creation and optimistic commit retries use the final assigned row IDs.
+
+Indexes with stale keys after indexed-column updates, and affected index types
+without RowID rewrite support, are dropped. Spark clustering currently drops
+global indexes in the affected partitions. Rebuild dropped indexes, or use
+`scalar-index.search-mode = full` for complete scalar query results while part
+of the table remains unindexed.
Concurrent updates or deletes that overlap the selected row-ID groups cause
commit to fail, preventing stale rewrites from replacing newer values.
Historical snapshots and tags retain
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringIndexRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringIndexRewriter.java
new file mode 100644
index 0000000000..3cf1f2f687
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringIndexRewriter.java
@@ -0,0 +1,469 @@
+/*
+ * 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.data.BinaryRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner;
+import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
+import org.apache.paimon.globalindex.GlobalIndexFileReadWrite;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RowRangeIndex;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Rewrites scalar index postings after clustering output files receive their
final row IDs. */
+public final class ClusteringIndexRewriter {
+
+ // The current metadata protocol represents coverage with contiguous
ranges. Bound both
+ // tiny-file fan-out and repeated source scans until sparse coverage has
its own protocol.
+ private static final int MAX_COVERAGE_FRAGMENTS = 64;
+
+ private ClusteringIndexRewriter() {}
+
+ public static List<IndexManifestEntry> rewrite(
+ FileIO fileIO,
+ FileStorePathFactory pathFactory,
+ SchemaManager schemaManager,
+ CoreOptions options,
+ @Nullable IOManager ioManager,
+ List<ManifestEntry> dataFiles,
+ List<IndexManifestEntry> indexFiles,
+ Set<String> mappedClusteringOutputs,
+ long newSnapshotId,
+ List<Path> createdFiles)
+ throws IOException {
+ if (indexFiles.stream()
+ .noneMatch(
+ entry ->
+ entry.kind() == FileKind.DELETE
+ && entry.indexFile().globalIndexMeta()
!= null)) {
+ return indexFiles;
+ }
+ Map<Pair<BinaryRow, Integer>, List<ManifestEntry>> groups =
+ dataFiles.stream()
+ .collect(Collectors.groupingBy(e ->
Pair.of(e.partition(), e.bucket())));
+ Map<Pair<BinaryRow, Integer>, List<IndexManifestEntry>> deletes = new
HashMap<>();
+ List<IndexManifestEntry> result = new ArrayList<>();
+ for (IndexManifestEntry entry : indexFiles) {
+ if (entry.kind() == FileKind.DELETE &&
entry.indexFile().globalIndexMeta() != null) {
+ deletes.computeIfAbsent(
+ Pair.of(entry.partition(), entry.bucket()), k
-> new ArrayList<>())
+ .add(entry);
+ } else {
+ result.add(entry);
+ }
+ }
+ for (Map.Entry<Pair<BinaryRow, Integer>, List<IndexManifestEntry>>
group :
+ deletes.entrySet()) {
+ List<ManifestEntry> files =
+ groups.getOrDefault(group.getKey(),
Collections.emptyList());
+ List<DataFileMeta> after =
+ files.stream()
+ .filter(e -> e.kind() == FileKind.ADD)
+ .map(ManifestEntry::file)
+ .collect(Collectors.toList());
+ // Other materialization callers do not supply a permutation and
retain their existing
+ // invalidation behavior. A mixed mapping must never silently
retain a stale posting.
+ if (after.stream()
+ .anyMatch(file ->
!mappedClusteringOutputs.contains(file.fileName()))) {
+ result.addAll(group.getValue());
+ continue;
+ }
+ List<ManifestEntry> before =
+ files.stream()
+ .filter(e -> e.kind() == FileKind.DELETE)
+ .map(
+ e ->
+ ManifestEntry.create(
+ FileKind.ADD,
+ e.partition(),
+ e.bucket(),
+ e.totalBuckets(),
+ e.file()))
+ .collect(Collectors.toList());
+ if (before.isEmpty()) {
+ result.addAll(group.getValue());
+ continue;
+ }
+
after.sort(Comparator.comparingLong(DataFileMeta::nonNullFirstRowId));
+ long firstAssignedRowId =
+ after.isEmpty() ? Long.MAX_VALUE :
after.get(0).nonNullFirstRowId();
+ List<Range> removed =
+ Range.sortAndMergeOverlap(
+ before.stream()
+ .map(ManifestEntry::file)
+ .map(DataFileMeta::nonNullRowIdRange)
+ .collect(Collectors.toList()),
+ true);
+ DataFilePathFactory dataPath =
+ pathFactory.createDataFilePathFactory(
+ group.getKey().getLeft(),
group.getKey().getRight());
+ try (DiskMapping mapping = new DiskMapping(removed, ioManager)) {
+ for (DataFileMeta file : after) {
+ readMapping(
+ fileIO,
+ dataPath,
+ file,
+ (ordinal, oldId) ->
+ mapping.put(oldId,
file.nonNullFirstRowId() + ordinal));
+ }
+ RowType rowType =
+ schemaManager
+ .latestOrThrow("Missing clustering table
schema")
+ .logicalRowType();
+ Map<Pair<String, List<Integer>>, List<IndexManifestEntry>>
validSources =
+ new LinkedHashMap<>();
+ for (IndexManifestEntry entry : group.getValue()) {
+ GlobalIndexMeta meta = entry.indexFile().globalIndexMeta();
+ if (Range.and(Collections.singletonList(meta.rowRange()),
removed).isEmpty()) {
+ continue;
+ }
+ result.add(entry);
+ String type = entry.indexFile().indexType();
+ if (!type.equals("btree") && !type.equals("bitmap")) {
+ continue;
+ }
+ // Dropped indexed columns can leave old index files in
the manifest.
+ if
(!meta.getIndexedFieldIds().stream().allMatch(rowType::containsField)) {
+ continue;
+ }
+ List<DataField> fields = meta.getIndexedFields(rowType);
+ IndexManifestEntry source =
+ new IndexManifestEntry(
+ FileKind.ADD,
+ entry.partition(),
+ entry.bucket(),
+ entry.indexFile());
+ // Updating row IDs cannot repair keys from an older
indexed-column value.
+ if
(!DataEvolutionIndexSourceMeta.isDataEvolutionMeta(meta.sourceMeta())
+ ||
!DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh(
+ schemaManager,
+ before,
+ Collections.singletonList(source),
+ fields)
+ .isEmpty()) {
+ continue;
+ }
+ validSources
+ .computeIfAbsent(
+ Pair.of(type, meta.getIndexedFieldIds()),
+ ignored -> new ArrayList<>())
+ .add(entry);
+ }
+ for (Map.Entry<Pair<String, List<Integer>>,
List<IndexManifestEntry>> sourceGroup :
+ validSources.entrySet()) {
+ List<IndexManifestEntry> sources = sourceGroup.getValue();
+ String type = sourceGroup.getKey().getLeft();
+ GlobalIndexMeta firstMeta =
sources.get(0).indexFile().globalIndexMeta();
+ List<DataField> fields =
firstMeta.getIndexedFields(rowType);
+ RowRangeIndex coverage =
+ RowRangeIndex.create(
+ sources.stream()
+ .map(
+ source ->
+ source.indexFile()
+
.globalIndexMeta()
+
.rowRange())
+ .collect(Collectors.toList()),
+ false);
+ String sourceName = sources.get(0).indexFile().fileName();
+ List<Range> freshFragments = new ArrayList<>();
+ // Source shards can interleave after sorting. Their
union, rather than each
+ // original shard, defines the coverage shared by the
rewritten payload files.
+ for (DataFileMeta file : after) {
+ final long[] start = {-1};
+ final long[] end = {-1};
+ readMapping(
+ fileIO,
+ dataPath,
+ file,
+ (ordinal, oldId) -> {
+ long newId = file.nonNullFirstRowId() +
ordinal;
+ if (coverage.intersects(oldId, oldId)) {
+ if (start[0] < 0) {
+ start[0] = newId;
+ }
+ end[0] = newId;
+ } else if (start[0] >= 0) {
+ addFragment(
+ freshFragments,
+ new Range(start[0], end[0]),
+ sourceName,
+ firstAssignedRowId);
+ start[0] = -1;
+ }
+ });
+ if (start[0] >= 0) {
+ addFragment(
+ freshFragments,
+ new Range(start[0], end[0]),
+ sourceName,
+ firstAssignedRowId);
+ }
+ }
+ GlobalIndexer indexer =
+ GlobalIndexer.create(type, fields,
Options.fromMap(options.toMap()));
+ IndexPathFactory indexPath =
pathFactory.globalIndexFileFactory();
+ GlobalIndexFileReadWrite indexIO =
+ trackingIndexIO(fileIO, indexPath, createdFiles);
+ for (IndexManifestEntry source : sources) {
+ GlobalIndexMeta meta =
source.indexFile().globalIndexMeta();
+ List<Range> fragments = new ArrayList<>();
+ for (Range untouched :
meta.rowRange().exclude(removed)) {
+ addFragment(
+ fragments,
+ untouched,
+ source.indexFile().fileName(),
+ firstAssignedRowId);
+ }
+ for (Range fresh : freshFragments) {
+ addFragment(
+ fragments,
+ fresh,
+ source.indexFile().fileName(),
+ firstAssignedRowId);
+ }
+ GlobalIndexIOMeta input =
+ new GlobalIndexIOMeta(
+ indexPath.toPath(source.indexFile()),
+ source.indexFile().fileSize(),
+ source.indexFile().rowCount(),
+ meta.indexMeta());
+ for (Range fragment : fragments) {
+ List<ResultEntry> rewritten =
+ indexer.rewriteRowIds(
+ indexIO,
+ Collections.singletonList(input),
+ meta.rowRange().count(),
+ fragment.count(),
+ relative -> {
+ try {
+ long target =
+ mapping.get(
+ relative
+ +
meta.rowRangeStart());
+ return target >=
fragment.from
+ && target
<= fragment.to
+ ? target -
fragment.from
+ : -1;
+ } catch (IOException e) {
+ throw new
UncheckedIOException(e);
+ }
+ },
+ indexIO);
+ // Only newly assigned coverage shares a fresh
baseline. Old fragments
+ // keep their source's baseline, even when another
shard is older.
+ byte[] sourceMeta =
+ fragment.from >= firstAssignedRowId
+ ? new
DataEvolutionIndexSourceMeta(newSnapshotId)
+ .serialize()
+ : meta.sourceMeta();
+ for (IndexFileMeta rewrittenFile :
+ GlobalIndexBuilderUtils.toIndexFileMetas(
+ fileIO,
+ indexPath,
+ options,
+ fragment,
+ fields,
+ type,
+ rewritten,
+ sourceMeta)) {
+ result.add(
+ new IndexManifestEntry(
+ FileKind.ADD,
+ source.partition(),
+ source.bucket(),
+ rewrittenFile));
+ }
+ }
+ }
+ }
+ }
+ }
+ return result;
+ }
+
+ private static void addFragment(
+ List<Range> fragments, Range fragment, String fileName, long
firstAssignedRowId) {
+ if (!fragments.isEmpty()) {
+ Range previous = fragments.get(fragments.size() - 1);
+ if (previous.to == fragment.from - 1
+ && (previous.from >= firstAssignedRowId)
+ == (fragment.from >= firstAssignedRowId)) {
+ fragments.set(fragments.size() - 1, new Range(previous.from,
fragment.to));
+ return;
+ }
+ }
+ checkArgument(
+ fragments.size() < MAX_COVERAGE_FRAGMENTS,
+ "Clustering exceeds %s index coverage fragments for %s. "
+ + "Build smaller, fully covered index shards before
clustering this range.",
+ MAX_COVERAGE_FRAGMENTS,
+ fileName);
+ fragments.add(fragment);
+ }
+
+ private static GlobalIndexFileReadWrite trackingIndexIO(
+ FileIO fileIO, IndexPathFactory indexPath, List<Path>
createdFiles) {
+ return new GlobalIndexFileReadWrite(fileIO, indexPath) {
+ @Override
+ public PositionOutputStream newOutputStream(String fileName)
throws IOException {
+ // Track before opening: failed writes can leave a partial
file without a result.
+ createdFiles.add(indexPath.toPath(fileName));
+ return super.newOutputStream(fileName);
+ }
+ };
+ }
+
+ private static void readMapping(
+ FileIO fileIO,
+ DataFilePathFactory pathFactory,
+ DataFileMeta file,
+ ClusteringRowIdMapping.RowIdConsumer consumer)
+ throws IOException {
+ ClusteringRowIdMapping.read(
+ fileIO,
+
pathFactory.toAlignedPath(ClusteringRowIdMapping.fileName(file.fileName()),
file),
+ file.rowCount(),
+ consumer);
+ }
+
+ /**
+ * Disk-backed inverse permutation; memory is proportional to selected
ranges, not row count.
+ */
+ private static final class DiskMapping implements Closeable {
+ private final NavigableMap<Long, Pair<Range, Long>> offsets = new
TreeMap<>();
+ private final File path;
+ private final RandomAccessFile file;
+
+ private DiskMapping(List<Range> ranges, @Nullable IOManager ioManager)
throws IOException {
+ long rows = 0;
+ for (Range range : ranges) {
+ offsets.put(range.from, Pair.of(range, rows));
+ rows = Math.addExact(rows, range.count());
+ }
+ long length = Math.multiplyExact(rows, Long.BYTES);
+ path =
+ ioManager == null
+ ? File.createTempFile("paimon-cluster-row-id-",
".mapping")
+ : ioManager.createChannel().getPathFile();
+ RandomAccessFile opened = null;
+ try {
+ opened = new RandomAccessFile(path, "rw");
+ opened.setLength(length);
+ file = opened;
+ } catch (IOException | RuntimeException | Error failure) {
+ if (opened != null) {
+ try {
+ opened.close();
+ } catch (IOException closeFailure) {
+ failure.addSuppressed(closeFailure);
+ }
+ }
+ try {
+ Files.deleteIfExists(path.toPath());
+ } catch (IOException cleanupFailure) {
+ failure.addSuppressed(cleanupFailure);
+ }
+ throw failure;
+ }
+ }
+
+ @Nullable
+ private Long offset(long oldId) {
+ Map.Entry<Long, Pair<Range, Long>> entry =
offsets.floorEntry(oldId);
+ if (entry == null || oldId > entry.getValue().getLeft().to) {
+ return null;
+ }
+ return Math.multiplyExact(
+ entry.getValue().getRight() + oldId - entry.getKey(),
Long.BYTES);
+ }
+
+ private void put(long oldId, long newId) throws IOException {
+ Long offset = offset(oldId);
+ checkArgument(
+ offset != null, "Clustering output refers to unselected
row ID %s", oldId);
+ file.seek(offset);
+ checkArgument(file.readLong() == 0, "Duplicate clustering row ID
%s", oldId);
+ file.seek(offset);
+ file.writeLong(Math.addExact(newId, 1));
+ }
+
+ private long get(long oldId) throws IOException {
+ Long offset = offset(oldId);
+ if (offset == null) {
+ return oldId;
+ }
+ file.seek(offset);
+ return file.readLong() - 1;
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ file.close();
+ } finally {
+ Files.deleteIfExists(path.toPath());
+ }
+ }
+ }
+}
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 d1573b9412..5ff13ba2f1 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
@@ -22,6 +22,8 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.Snapshot.CommitKind;
import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.append.dataevolution.ClusteringIndexRewriter;
+import org.apache.paimon.append.dataevolution.ClusteringRowIdMapping;
import org.apache.paimon.catalog.SnapshotCommit;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
@@ -888,8 +890,8 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
.collect(Collectors.toList());
// This provider is invoked again after every optimistic-commit
conflict. Scanning the
- // latest snapshot here guarantees that an index committed
concurrently is either
- // deleted by this attempt or makes this attempt retry and is
deleted by the next one.
+ // latest snapshot includes concurrently created indexes in this
attempt's changes.
+ // Clustering can then remap their postings after the final row
IDs have been assigned.
if (latestSnapshot != null && latestSnapshot.indexManifest() !=
null) {
for (IndexManifestEntry entry :
indexManifestFile.read(latestSnapshot.indexManifest())) {
@@ -1153,6 +1155,23 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
List<ManifestEntry> deltaFiles = changes.tableFiles;
List<ManifestEntry> changelogFiles = changes.changelogFiles;
List<IndexManifestEntry> indexFiles = changes.indexFiles;
+ // Side files remain attached to committed data. Only outputs awaiting
their first row-ID
+ // assignment carry a permutation for this commit; later metadata
rewrites must ignore it.
+ Set<String> mappedClusteringOutputs =
+ commitKind == CommitKind.COMPACT &&
options.dataEvolutionEnabled()
+ ? deltaFiles.stream()
+ .filter(entry -> entry.kind() == FileKind.ADD)
+ .filter(entry -> entry.file().firstRowId() ==
null)
+ .filter(
+ entry ->
+ entry.file()
+ .extraFiles()
+ .contains(
+
ClusteringRowIdMapping.fileName(
+
entry.file().fileName())))
+ .map(entry -> entry.file().fileName())
+ .collect(Collectors.toSet())
+ : Collections.emptySet();
Set<Long> rebasedReassignments =
changesProvider.rebasedReassignments();
long newSnapshotId = Snapshot.FIRST_SNAPSHOT_ID;
@@ -1261,6 +1280,7 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
List<ManifestFileMeta> mergeAfterManifests = new ArrayList<>();
boolean skipManifestMerge = false;
long nextRowIdStart = firstRowIdStart;
+ List<Path> clusteringIndexOutputs = new ArrayList<>();
try {
long previousTotalRecordCount = 0L;
Long currentWatermark = watermark;
@@ -1326,6 +1346,23 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
deltaFiles = assigned.assignedEntries;
}
+ if (commitKind == CommitKind.COMPACT
+ && options.dataEvolutionEnabled()
+ && deltaFiles.stream().anyMatch(entry -> entry.kind() ==
FileKind.DELETE)) {
+ indexFiles =
+ ClusteringIndexRewriter.rewrite(
+ fileIO,
+ pathFactory,
+ schemaManager,
+ options,
+ ioManager,
+ deltaFiles,
+ indexFiles,
+ mappedClusteringOutputs,
+ newSnapshotId,
+ clusteringIndexOutputs);
+ }
+
if (options.snapshotSequenceOrdering()) {
deltaFiles = stampSequenceWithSnapshotId(newSnapshotId,
commitKind, deltaFiles);
}
@@ -1406,6 +1443,7 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
operation);
} catch (Throwable e) {
// fails when preparing for commit, we should clean up
+ clusteringIndexOutputs.forEach(fileIO::deleteQuietly);
commitCleaner.cleanUpReuseTmpManifests(
deltaManifestList, changelogManifestList,
oldIndexManifest, indexManifest);
commitCleaner.cleanUpNoReuseTmpManifests(
@@ -1421,13 +1459,22 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
boolean success;
final List<SimpleFileEntry> finalBaseFiles = baseDataFiles;
final List<ManifestEntry> finalDeltaFiles = deltaFiles;
- commitPreCallbacks.forEach(
- callback ->
- callback.call(finalBaseFiles, finalDeltaFiles,
indexFiles, newSnapshot));
+ final List<IndexManifestEntry> finalIndexFiles = indexFiles;
+ try {
+ commitPreCallbacks.forEach(
+ callback ->
+ callback.call(
+ finalBaseFiles, finalDeltaFiles,
finalIndexFiles, newSnapshot));
+ } catch (RuntimeException | Error failure) {
+ // No snapshot publication has been attempted, so these new
payloads are unreferenced.
+ clusteringIndexOutputs.forEach(fileIO::deleteQuietly);
+ throw failure;
+ }
try {
success = commitSnapshotImpl(latestSnapshot, newSnapshot,
deltaPartitionEntries);
} catch (Exception e) {
// commit exception, not sure about the situation and should not
clean up the files
+ // This also retains index payloads generated for this possibly
published candidate.
LOG.warn(
"Retry commit for exception when committing snapshot #{}
for table {} by user {}.",
newSnapshotId,
@@ -1438,6 +1485,8 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
}
if (!success) {
+ // Some committers can publish the snapshot and still report
false. Retain the index
+ // payloads together with the manifests until the next attempt
reconciles publication.
long commitTime = (System.currentTimeMillis() - startMillis) /
1000;
LOG.warn(
"Atomic commit failed for snapshot #{} for table {} by
user {} "
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 51e79a1c31..48b60d3ab8 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
@@ -35,13 +35,17 @@ 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.DataEvolutionGlobalIndexScanner;
import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
+import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
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.DataFilePathFactory;
import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
@@ -49,6 +53,8 @@ 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.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.schema.Schema;
@@ -69,13 +75,16 @@ 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.ProjectedRow;
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.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -83,6 +92,7 @@ import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@@ -415,6 +425,689 @@ class DataEvolutionClusterTest extends TableTestBase {
.containsExactly("a|10|base-10|3", "a|20|base-20|4",
"a|30|base-30|5");
}
+ @ParameterizedTest
+ @CsvSource({"btree,false", "btree,true", "bitmap,false", "bitmap,true"})
+ void testClusterRemapsScalarIndexPostings(String indexType, boolean
deletionVectors)
+ throws Exception {
+ FileStoreTable table = createTable(deletionVectors);
+ appendIndexRows(
+ table,
+ new int[] {30, 10, 20, 40, 50},
+ new String[] {"x", "x", null, "deleted", "y"});
+ commit(
+ table,
+ prepareGlobalIndex(
+ table, table.latestSnapshot().get(), "payload",
indexType, null));
+ if (deletionVectors) {
+ delete(table, "a", new Range(0, 4), 3L);
+ }
+ table =
table.copy(Collections.singletonMap(CoreOptions.TARGET_FILE_ROW_NUM.key(),
"2"));
+ Snapshot snapshot = table.latestSnapshot().get();
+ commitCluster(table, snapshot, prepareMappedCluster(table, snapshot,
true));
+
+ PredicateBuilder predicates = new PredicateBuilder(table.rowType());
+ int payload = table.rowType().getFieldIndex("payload");
+ assertIndexRows(
+ table,
+ predicates.equal(payload, BinaryString.fromString("x")),
+ new Range(5, 5),
+ new Range(7, 7));
+ assertIndexRows(table, predicates.isNull(payload), new Range(6, 6));
+ assertIndexRows(
+ table,
+ predicates.equal(payload, BinaryString.fromString("y")),
+ new Range(deletionVectors ? 8 : 9, deletionVectors ? 8 : 9));
+ assertIndexRows(
+ table,
+ predicates.equal(payload, BinaryString.fromString("deleted")),
+ deletionVectors ? new Range[0] : new Range[] {new Range(8,
8)});
+ assertThat(readRows(table))
+ .containsExactlyElementsOf(
+ deletionVectors
+ ? Arrays.asList("a|10|x|5", "a|20|null|6",
"a|30|x|7", "a|50|y|8")
+ : Arrays.asList(
+ "a|10|x|5",
+ "a|20|null|6",
+ "a|30|x|7",
+ "a|40|deleted|8",
+ "a|50|y|9"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"btree", "bitmap"})
+ void testClusterPreservesExactPartialIndexCoverage(String indexType)
throws Exception {
+ FileStoreTable table = createTable(false);
+ append(table, "a", 90, 10, 50);
+ append(table, "a", 80, 20, 40);
+ commit(
+ table,
+ prepareGlobalIndex(
+ table, table.latestSnapshot().get(), "id", indexType,
new Range(0, 2)));
+ table =
table.copy(Collections.singletonMap(CoreOptions.TARGET_FILE_ROW_NUM.key(),
"2"));
+ Snapshot snapshot = table.latestSnapshot().get();
+ commitCluster(table, snapshot, prepareMappedCluster(table, snapshot,
true));
+
+ PredicateBuilder predicates = new PredicateBuilder(table.rowType());
+ int id = table.rowType().getFieldIndex("id");
+ assertIndexRows(table, predicates.equal(id, 10), new Range(6, 6));
+ assertIndexRows(table, predicates.equal(id, 50), new Range(9, 9));
+ assertIndexRows(table, predicates.equal(id, 90), new Range(11, 11));
+ assertIndexRows(table, predicates.equal(id, 20));
+ List<Range> coverage =
+ table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry -> entry.indexFile().globalIndexMeta()
!= null)
+ .map(entry ->
entry.indexFile().globalIndexMeta().rowRange())
+ .collect(Collectors.toList());
+ assertThat(Range.sortAndMergeOverlap(coverage, true))
+ .containsExactly(new Range(6, 6), new Range(9, 9), new
Range(11, 11));
+ FileStoreTable detail =
+ table.copy(
+ Collections.singletonMap(
+ CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(),
"detail"));
+ assertThat(readRows(detail, predicates.equal(id,
20))).containsExactly("a|20|base-20|7");
+ assertThat(readRows(detail, predicates.equal(id,
80))).containsExactly("a|80|base-80|10");
+ }
+
+ @Test
+ void testClusterRejectsTooManyIndexCoverageFragments() throws Exception {
+ FileStoreTable table = createTable(false);
+ int[] indexed = new int[65];
+ int[] unindexed = new int[65];
+ for (int i = 0; i < indexed.length; i++) {
+ indexed[i] = i * 2 + 1;
+ unindexed[i] = i * 2 + 2;
+ }
+ append(table, "a", indexed);
+ append(table, "a", unindexed);
+ commit(
+ table,
+ prepareGlobalIndex(
+ table, table.latestSnapshot().get(), "id", "btree",
new Range(0, 64)));
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<String> beforeFiles = fileNames(table, "a");
+ List<String> beforeIndexes = globalIndexNames(table);
+ List<String> beforeRows = readRows(table);
+ List<CommitMessage> messages = prepareMappedCluster(table, snapshot,
true);
+ assertThat(messages).hasSize(1);
+ assertThat(((CommitMessageImpl)
messages.get(0)).compactIncrement().compactAfter())
+ .hasSize(1);
+
+ // One indexed row alternates with one unindexed row: 65 exact
coverage fragments.
+ assertThatThrownBy(() -> commitCluster(table, snapshot, messages))
+ .isInstanceOf(RuntimeException.class)
+ .hasStackTraceContaining("coverage fragments")
+ .hasStackTraceContaining("64");
+ assertThat(table.latestSnapshot().get().id()).isEqualTo(snapshot.id());
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(130L);
+ assertThat(fileNames(table, "a")).isEqualTo(beforeFiles);
+ assertThat(globalIndexNames(table)).isEqualTo(beforeIndexes);
+ assertThat(readRows(table)).isEqualTo(beforeRows);
+ assertIndexRows(
+ table,
+ new
PredicateBuilder(table.rowType()).equal(table.rowType().getFieldIndex("id"), 1),
+ new Range(0, 0));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"btree", "bitmap"})
+ void
testClusterCombinesFullyIndexedSourceRangesBeforeCountingFragments(String
indexType)
+ throws Exception {
+ FileStoreTable table = createTable(false);
+ int[] odd = new int[65];
+ int[] even = new int[65];
+ for (int i = 0; i < odd.length; i++) {
+ odd[i] = i * 2 + 1;
+ even[i] = i * 2 + 2;
+ }
+ append(table, "a", odd);
+ append(table, "a", even);
+ Snapshot indexed = table.latestSnapshot().get();
+ List<CommitMessage> indexes =
+ new ArrayList<>(
+ prepareGlobalIndex(table, indexed, "id", indexType,
new Range(0, 64)));
+ indexes.addAll(prepareGlobalIndex(table, indexed, "id", indexType, new
Range(65, 129)));
+ commit(table, indexes);
+ assertThat(globalIndexNames(table)).hasSize(2);
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareMappedCluster(table, snapshot,
true);
+ assertThat(messages).hasSize(1);
+ assertThat(((CommitMessageImpl)
messages.get(0)).compactIncrement().compactAfter())
+ .hasSize(1);
+
+ // Each old range maps to 65 alternating positions, but their union
has no coverage gap.
+ commitCluster(table, snapshot, messages);
+
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(260L);
+ assertThat(
+
table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry ->
entry.indexFile().globalIndexMeta() != null)
+ .map(entry ->
entry.indexFile().globalIndexMeta().rowRange()))
+ .isNotEmpty()
+ .allSatisfy(range -> assertThat(range).isEqualTo(new
Range(130, 259)));
+ PredicateBuilder predicates = new PredicateBuilder(table.rowType());
+ int id = table.rowType().getFieldIndex("id");
+ List<String> expectedRows = new ArrayList<>();
+ for (int value = 1; value <= 130; value++) {
+ long rowId = 129L + value;
+ assertIndexRows(table, predicates.equal(id, value), new
Range(rowId, rowId));
+ expectedRows.add("a|" + value + "|base-" + value + "|" + rowId);
+ }
+ assertIndexRows(table, predicates.equal(id, 0));
+ assertIndexRows(table, predicates.isNull(id));
+ assertThat(readRows(table)).containsExactlyElementsOf(expectedRows);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testMinorClusterRemapsIndexesAcrossDisjointRanges(boolean
deletionVectors)
+ throws Exception {
+ FileStoreTable table = createMinorClusterTable(deletionVectors);
+ List<String> untouchedRows = rowsInRange(table, new Range(12, 14));
+ List<String> untouchedFiles = filesInRange(table, new Range(12, 14));
+ commit(table, prepareGlobalIndex(table, table.latestSnapshot().get(),
"id"));
+ List<String> oldIndexFiles = globalIndexNames(table);
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareMinorCluster(table, snapshot);
+ messages =
+ IncrementalClusterRewriter.create(table,
snapshot).prepareCommitMessages(messages);
+ List<CommitMessage> preparedGapIndex =
+ prepareGlobalIndex(table, snapshot, "payload", "bitmap", new
Range(12, 14));
+ commit(table, prepareGlobalIndex(table, snapshot, "payload"));
+ commitChecked(table, messages);
+
+
assertThat(globalIndexNames(table)).isNotEmpty().doesNotContainAnyElementsOf(oldIndexFiles);
+ PredicateBuilder predicates = new PredicateBuilder(table.rowType());
+ int id = table.rowType().getFieldIndex("id");
+ int payload = table.rowType().getFieldIndex("payload");
+ assertIndexRows(table, predicates.equal(id, 40), new Range(12, 12));
+ assertIndexRows(table, predicates.equal(id, 50), new Range(13, 13));
+ assertIndexRows(table, predicates.equal(id, 60), new Range(14, 14));
+ long seventy = deletionVectors ? 23 : 24;
+ assertIndexRows(table, predicates.equal(id, 70), new Range(seventy,
seventy));
+ assertIndexRows(
+ table,
+ predicates.equal(payload,
BinaryString.fromString("updated-70")),
+ new Range(seventy, seventy));
+ assertMinorClusterResult(table, deletionVectors, untouchedRows,
untouchedFiles, 21L, 4);
+
+ // A pre-clustering index can still commit for an unchanged row-ID
group.
+ commit(table, preparedGapIndex);
+ assertIndexRows(table, predicates.equal(id, 40), new Range(12, 12));
+ assertMinorClusterResult(table, deletionVectors, untouchedRows,
untouchedFiles, 21L, 4);
+ }
+
+ @Test
+ void testMinorClusterRemapsIndexWhenAllSelectedRowsAreDeleted() throws
Exception {
+ FileStoreTable table = createMinorClusterTable(true);
+ List<String> untouchedRows = rowsInRange(table, new Range(12, 14));
+ List<String> untouchedFiles = filesInRange(table, new Range(12, 14));
+ commit(table, prepareGlobalIndex(table, table.latestSnapshot().get(),
"id"));
+ delete(table, "a", new Range(9, 11), 9L, 10L, 11L);
+ delete(table, "a", new Range(15, 17), 15L, 16L, 17L);
+ delete(table, "a", new Range(18, 20), 18L, 19L, 20L);
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareMinorCluster(table, snapshot);
+ assertThat(messages)
+ .allSatisfy(
+ message ->
+ assertThat(
+ ((CommitMessageImpl) message)
+ .compactIncrement()
+ .compactAfter())
+ .isEmpty());
+ commitCluster(table, snapshot, messages);
+
+ PredicateBuilder predicates = new PredicateBuilder(table.rowType());
+ int id = table.rowType().getFieldIndex("id");
+ assertIndexRows(table, predicates.equal(id, 40), new Range(12, 12));
+ assertIndexRows(table, predicates.equal(id, 20));
+ assertIndexRows(table, predicates.equal(id, 100));
+ assertThat(
+
table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry ->
entry.indexFile().globalIndexMeta() != null)
+ .map(entry ->
entry.indexFile().globalIndexMeta().rowRange()))
+ .containsExactly(new Range(12, 14));
+ assertThat(readRows(table)).containsExactlyElementsOf(untouchedRows);
+ assertThat(fileNames(table,
"a")).containsExactlyElementsOf(untouchedFiles);
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(21L);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testMinorClusterRetryRemapsConcurrentIndexAfterAppend(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 =
createMinorClusterTable(deletionVectors).copy(retryOptions);
+ List<String> untouchedRows = rowsInRange(original, new Range(12, 14));
+ List<String> untouchedFiles = filesInRange(original, new Range(12,
14));
+ commit(original, prepareGlobalIndex(original,
original.latestSnapshot().get(), "id"));
+ Snapshot scanSnapshot = original.latestSnapshot().get();
+ List<CommitMessage> messages = prepareMinorCluster(original,
scanSnapshot);
+ 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();
+ assertThat(
+ original.store()
+
.indexManifestFileFactory()
+ .create()
+
.read(snapshot.indexManifest()))
+ .isNotEmpty();
+ if (attempt == 1) {
+
assertThat(original.latestSnapshot().get().id())
+ .isEqualTo(scanSnapshot.id());
+ DataEvolutionClusterTest.this.commit(
+ original,
+ prepareGlobalIndex(original,
scanSnapshot, "payload"));
+ append(original, "a", 200, 180, 190);
+ }
+ boolean committed =
+ delegate.commit(baseUuid, snapshot,
branch, statistics);
+ assertThat(committed).isEqualTo(attempt == 2);
+ return committed;
+ }
+
+ @Override
+ protected void closeImpl() throws Exception {
+ delegate.close();
+ }
+ };
+ }
+ };
+ FileStoreTable committingTable =
+ FileStoreTableFactory.create(
+ original.fileIO(), original.location(),
original.schema(), environment);
+ commitCluster(committingTable, scanSnapshot, messages);
+
+ assertThat(attempts).hasValue(2);
+
assertThat(original.latestSnapshot().get().id()).isEqualTo(scanSnapshot.id() +
3);
+ long seventy = deletionVectors ? 26 : 27;
+ PredicateBuilder predicates = new PredicateBuilder(original.rowType());
+ assertIndexRows(
+ original,
+ predicates.equal(original.rowType().getFieldIndex("id"), 70),
+ new Range(seventy, seventy));
+ assertIndexRows(
+ original,
+ predicates.equal(
+ original.rowType().getFieldIndex("payload"),
+ BinaryString.fromString("updated-70")),
+ new Range(seventy, seventy));
+ untouchedRows.addAll(
+ Arrays.asList("a|200|base-200|21", "a|180|base-180|22",
"a|190|base-190|23"));
+ assertMinorClusterResult(original, deletionVectors, untouchedRows,
untouchedFiles, 24L, 5);
+ FileStoreTable detail =
+ original.copy(
+ Collections.singletonMap(
+ CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(),
"detail"));
+ assertThat(readRows(detail,
predicates.equal(original.rowType().getFieldIndex("id"), 180)))
+ .containsExactly("a|180|base-180|22");
+ }
+
+ @Test
+ void testClusterDoesNotMakeStaleIndexKeysFresh() throws Exception {
+ FileStoreTable table =
+ createTable(false)
+ .copy(
+ Collections.singletonMap(
+
CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION.key(),
+ "ignore"));
+ append(table, "a", 30, 10, 20);
+ Snapshot indexed = table.latestSnapshot().get();
+ List<CommitMessage> indexes =
+ new ArrayList<>(prepareGlobalIndex(table, indexed, "payload"));
+ indexes.addAll(prepareGlobalIndex(table, indexed, "id"));
+ commit(table, indexes);
+ updatePayload(table, "a", 0L, 30, 10, 20);
+ Snapshot snapshot = table.latestSnapshot().get();
+ commitCluster(table, snapshot, prepareMappedCluster(table, snapshot,
true));
+
+ PredicateBuilder predicates = new PredicateBuilder(table.rowType());
+ assertIndexRows(
+ table, predicates.equal(table.rowType().getFieldIndex("id"),
20), new Range(4, 4));
+ assertThat(
+
table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry ->
entry.indexFile().globalIndexMeta() != null)
+ .flatMap(
+ entry ->
+
entry.indexFile().globalIndexMeta()
+
.getIndexedFieldIds().stream()))
+ .doesNotContain(table.rowType().getField("payload").id());
+ FileStoreTable detail =
+ table.copy(
+ Collections.singletonMap(
+ CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(),
"detail"));
+ assertThat(
+ readRows(
+ detail,
+ predicates.equal(
+
table.rowType().getFieldIndex("payload"),
+
BinaryString.fromString("updated-20"))))
+ .containsExactly("a|20|updated-20|4");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testClusterKeepsIndexPayloadsAfterUncertainPublication(boolean
publicationThrows)
+ throws Exception {
+ FileStoreTable original = createTable(false);
+ append(original, "a", 30, 10, 20);
+ commit(original, prepareGlobalIndex(original,
original.latestSnapshot().get(), "id"));
+ Snapshot scanSnapshot = original.latestSnapshot().get();
+ List<CommitMessage> messages = prepareMappedCluster(original,
scanSnapshot, true);
+ AtomicInteger publications = 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 {
+ publications.incrementAndGet();
+ assertThat(delegate.commit(baseUuid, snapshot,
branch, statistics))
+ .isTrue();
+ if (publicationThrows) {
+ throw new IOException("Publication
response lost");
+ }
+ return false;
+ }
+
+ @Override
+ protected void closeImpl() throws Exception {
+ delegate.close();
+ }
+ };
+ }
+ };
+ FileStoreTable committingTable =
+ FileStoreTableFactory.create(
+ original.fileIO(), original.location(),
original.schema(), environment);
+ commitCluster(committingTable, scanSnapshot, messages);
+
+ assertThat(publications).hasValue(1);
+
assertThat(original.latestSnapshot().get().id()).isEqualTo(scanSnapshot.id() +
1);
+ assertThat(original.latestSnapshot().get().nextRowId()).isEqualTo(6L);
+ List<IndexFileMeta> indexes =
+ original.store().newIndexFileHandler().scanEntries().stream()
+ .map(IndexManifestEntry::indexFile)
+ .filter(file -> file.globalIndexMeta() != null)
+ .collect(Collectors.toList());
+ assertThat(indexes).isNotEmpty();
+ for (IndexFileMeta index : indexes) {
+ assertThat(
+ original.fileIO()
+ .exists(
+ original.store()
+ .pathFactory()
+ .globalIndexFileFactory()
+ .toPath(index)))
+ .isTrue();
+ }
+ PredicateBuilder predicates = new PredicateBuilder(original.rowType());
+ int id = original.rowType().getFieldIndex("id");
+ assertIndexRows(original, predicates.equal(id, 10), new Range(3, 3));
+ assertIndexRows(original, predicates.equal(id, 20), new Range(4, 4));
+ assertIndexRows(original, predicates.equal(id, 30), new Range(5, 5));
+ assertThat(readRows(original))
+ .containsExactly("a|10|base-10|3", "a|20|base-20|4",
"a|30|base-30|5");
+ }
+
+ @Test
+ void testClusterRemapsCompositeBTreeIndex() throws Exception {
+ FileStoreTable table = createTable(false);
+ appendIndexRows(table, new int[] {30, 10, 20, 40}, new String[] {"x",
"x", null, "y"});
+ Snapshot indexed = table.latestSnapshot().get();
+ List<String> fields = Arrays.asList("id", "payload");
+ List<CommitMessage> indexes = new ArrayList<>();
+ for (DataSplit split :
+ GlobalIndexBuilderUtils.splitByContiguousRowRange(
+ new SortedGlobalIndexScanner(table, "btree")
+ .withIndexFields(fields)
+ .withSnapshot(indexed)
+ .scan()
+ .get()
+ .entries())) {
+ indexes.addAll(
+ SortedGlobalIndexTestUtils.buildIndex(
+ table, "btree", fields, split, indexed.id()));
+ }
+ assertThat(indexes).isNotEmpty();
+ commit(table, indexes);
+ table =
table.copy(Collections.singletonMap(CoreOptions.TARGET_FILE_ROW_NUM.key(),
"2"));
+ Snapshot snapshot = table.latestSnapshot().get();
+ commitCluster(table, snapshot, prepareMappedCluster(table, snapshot,
true));
+
+ PredicateBuilder predicates = new PredicateBuilder(table.rowType());
+ int id = table.rowType().getFieldIndex("id");
+ int payload = table.rowType().getFieldIndex("payload");
+ assertIndexRows(
+ table,
+ PredicateBuilder.and(
+ predicates.equal(id, 10),
+ predicates.equal(payload,
BinaryString.fromString("x"))),
+ new Range(4, 4));
+ assertIndexRows(
+ table,
+ PredicateBuilder.and(predicates.equal(id, 20),
predicates.isNull(payload)),
+ new Range(5, 5));
+ assertIndexRows(
+ table,
+ PredicateBuilder.and(
+ predicates.equal(id, 30),
+ predicates.equal(payload,
BinaryString.fromString("x"))),
+ new Range(6, 6));
+ assertIndexRows(
+ table,
+ PredicateBuilder.and(
+ predicates.equal(id, 20),
+ predicates.equal(payload,
BinaryString.fromString("y"))));
+ }
+
+ @Test
+ void testClusterPreservesIndexOnlyDeleteForAnotherPartition() throws
Exception {
+ FileStoreTable table = createTable(false);
+ append(table, "a", 30, 10, 20);
+ append(table, "b", 60, 40, 50);
+ commit(table, prepareGlobalIndex(table, table.latestSnapshot().get(),
"id"));
+ List<String> untouchedFiles = fileNames(table, "b");
+ List<IndexFileMeta> deletedIndexes =
+ table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry ->
entry.partition().equals(partition(table, "b")))
+ .map(IndexManifestEntry::indexFile)
+ .filter(file -> file.globalIndexMeta() != null)
+ .collect(Collectors.toList());
+ assertThat(deletedIndexes).hasSize(1);
+ Snapshot snapshot = table.latestSnapshot().get();
+ IncrementalClusterManager manager =
+ new IncrementalClusterManager(table, partitionPredicate(table,
"a"), snapshot);
+ Map<BinaryRow, CompactUnit> units = manager.createCompactUnits(true);
+ assertThat(units).containsOnlyKeys(partition(table, "a"));
+ List<CommitMessage> messages =
+ manager.prepareCommitMessages(rewriteClusterUnits(table,
manager, units, true));
+ messages.add(
+ new CommitMessageImpl(
+ partition(table, "b"),
+ UNAWARE_BUCKET,
+ null,
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ deletedIndexes))
+ .withCheckFromSnapshot(snapshot.id()));
+ commitChecked(table, messages);
+
+ assertThat(table.store().newIndexFileHandler().scanEntries())
+ .allSatisfy(
+ entry ->
assertThat(entry.partition()).isEqualTo(partition(table, "a")));
+ PredicateBuilder predicates = new PredicateBuilder(table.rowType());
+ int id = table.rowType().getFieldIndex("id");
+ assertIndexRows(table, predicates.equal(id, 10), new Range(6, 6));
+ assertIndexRows(table, predicates.equal(id, 20), new Range(7, 7));
+ assertIndexRows(table, predicates.equal(id, 30), new Range(8, 8));
+ assertIndexRows(table, predicates.equal(id, 40));
+ 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 testAdjacentIndexCoverageKeepsSeparateFreshnessBaselines() throws
Exception {
+ FileStoreTable table = createTable(false);
+ append(table, "a", 30, 10, 20);
+ append(table, "a", 60, 40, 50);
+ Snapshot indexed = table.latestSnapshot().get();
+ commit(table, prepareGlobalIndex(table, indexed, "id"));
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<DataFileMeta> selected =
+
table.store().newScan().withSnapshot(snapshot).plan().files().stream()
+ .map(entry -> entry.file())
+ .filter(file -> file.nonNullRowIdRange().equals(new
Range(0, 2)))
+ .collect(Collectors.toList());
+ assertThat(selected).hasSize(1);
+ List<String> untouchedFiles = filesInRange(table, new Range(3, 5));
+ IncrementalClusterManager manager = new
IncrementalClusterManager(table, null, snapshot);
+ Map<BinaryRow, CompactUnit> units =
+ Collections.singletonMap(
+ partition(table, "a"), CompactUnit.fromFiles(1,
selected, true));
+ commitCluster(table, snapshot, rewriteClusterUnits(table, manager,
units, true));
+
+ Snapshot clustered = table.latestSnapshot().get();
+ Map<Range, Long> baselines =
+ table.store().newIndexFileHandler().scanEntries().stream()
+ .map(IndexManifestEntry::indexFile)
+ .filter(file -> file.globalIndexMeta() != null)
+ .collect(
+ Collectors.toMap(
+ file ->
file.globalIndexMeta().rowRange(),
+ file ->
+
DataEvolutionIndexSourceMeta.fromIndexFile(file)
+ .scanSnapshotId()));
+ assertThat(baselines)
+ .hasSize(2)
+ .containsEntry(new Range(3, 5), indexed.id())
+ .containsEntry(new Range(6, 8), clustered.id());
+ assertThat(filesInRange(table, new Range(3,
5))).isEqualTo(untouchedFiles);
+ PredicateBuilder predicates = new PredicateBuilder(table.rowType());
+ int id = table.rowType().getFieldIndex("id");
+ assertIndexRows(table, predicates.equal(id, 10), new Range(6, 6));
+ assertIndexRows(table, predicates.equal(id, 60), new Range(3, 3));
+
+ append(table, "a", 90, 70, 80);
+ Snapshot next = table.latestSnapshot().get();
+ commitCluster(table, next, prepareMappedCluster(table, next, true));
+
+ // Neither the retained coverage nor the first rewrite became stale on
the second pass.
+ assertIndexRows(table, predicates.equal(id, 10), new Range(12, 12));
+ assertIndexRows(table, predicates.equal(id, 20), new Range(13, 13));
+ assertIndexRows(table, predicates.equal(id, 40), new Range(15, 15));
+ assertIndexRows(table, predicates.equal(id, 60), new Range(17, 17));
+ assertIndexRows(table, predicates.equal(id, 70));
+ assertThat(readRows(table))
+ .containsExactly(
+ "a|10|base-10|12",
+ "a|20|base-20|13",
+ "a|30|base-30|14",
+ "a|40|base-40|15",
+ "a|50|base-50|16",
+ "a|60|base-60|17",
+ "a|70|base-70|18",
+ "a|80|base-80|19",
+ "a|90|base-90|20");
+ }
+
+ @Test
+ void testMetadataOnlyCompactDoesNotReplayOldClusteringMapping() throws
Exception {
+ FileStoreTable table = createTable(false);
+ append(table, "a", 30, 10, 20);
+ commit(table, prepareGlobalIndex(table, table.latestSnapshot().get(),
"id"));
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<DataFileMeta> initialFiles =
+
table.store().newScan().withSnapshot(snapshot).plan().files().stream()
+ .map(entry -> entry.file())
+ .collect(Collectors.toList());
+ assertThat(initialFiles).hasSize(1);
+ IncrementalClusterManager manager = new
IncrementalClusterManager(table, null, snapshot);
+ Map<BinaryRow, CompactUnit> units =
+ Collections.singletonMap(
+ partition(table, "a"), CompactUnit.fromFiles(1,
initialFiles, true));
+ commitCluster(table, snapshot, rewriteClusterUnits(table, manager,
units, true));
+
+ Snapshot clustered = table.latestSnapshot().get();
+ DataFileMeta clusteredFile =
+
table.store().newScan().withSnapshot(clustered).plan().files().get(0).file();
+ assertThat(clusteredFile.nonNullRowIdRange()).isEqualTo(new Range(3,
5));
+
assertThat(clusteredFile.extraFiles()).anyMatch(ClusteringRowIdMapping::isMappingFile);
+ List<IndexFileMeta> deletedIndexes =
+ table.store().newIndexFileHandler().scanEntries().stream()
+ .map(IndexManifestEntry::indexFile)
+ .filter(file -> file.globalIndexMeta() != null)
+ .collect(Collectors.toList());
+ assertThat(deletedIndexes).isNotEmpty();
+ commit(
+ table,
+ new CommitMessageImpl(
+ partition(table, "a"),
+ UNAWARE_BUCKET,
+ null,
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.singletonList(clusteredFile),
+
Collections.singletonList(clusteredFile.upgrade(2)),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ deletedIndexes)));
+
+ assertThat(table.latestSnapshot().get().id()).isEqualTo(clustered.id()
+ 1);
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(6L);
+ assertThat(fileNames(table,
"a")).containsExactly(clusteredFile.fileName());
+ DataFileMeta upgraded =
+ table.store()
+ .newScan()
+ .withSnapshot(table.latestSnapshot().get())
+ .plan()
+ .files()
+ .get(0)
+ .file();
+ assertThat(upgraded.level()).isEqualTo(2);
+ assertThat(upgraded.nonNullRowIdRange()).isEqualTo(new Range(3, 5));
+
assertThat(upgraded.extraFiles()).isEqualTo(clusteredFile.extraFiles());
+ 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)
@@ -589,6 +1282,119 @@ class DataEvolutionClusterTest extends TableTestBase {
return getTableDefault();
}
+ private FileStoreTable createMinorClusterTable(boolean deletionVectors)
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(deletionVectors).copy(options);
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int id : new int[] {90, 10, 60, 30, 80, 40, 20, 70, 50}) {
+ Random random = new Random(id);
+ int length = id >= 40 && id <= 60 ? 262144 : 8192;
+ 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)
+ .extracting(DataFileMeta::nonNullRowIdRange)
+ .containsExactly(new Range(9, 11), new Range(12, 14), new
Range(15, 17));
+ assertThat(clusteredFiles.get(1).fileSize())
+ .isGreaterThan(
+ (clusteredFiles.get(0).fileSize() +
clusteredFiles.get(2).fileSize()) * 8);
+
+ updatePayload(table, "a", 9L, 10, 20, 30);
+ updatePayload(table, "a", 15L, 70, 80, 90);
+ append(table, "a", 120, 100, 110);
+ if (deletionVectors) {
+ delete(table, "a", new Range(9, 11), 9L);
+ }
+ return table;
+ }
+
+ private List<CommitMessage> prepareMinorCluster(FileStoreTable table,
Snapshot snapshot)
+ throws Exception {
+ IncrementalClusterManager manager = new
IncrementalClusterManager(table, null, snapshot);
+ Map<BinaryRow, CompactUnit> units = manager.createCompactUnits(false);
+ assertThat(units).hasSize(1);
+ List<DataFileMeta> selected = units.get(partition(table, "a")).files();
+ assertThat(selected)
+ .extracting(DataFileMeta::nonNullRowIdRange)
+ .containsExactlyInAnyOrder(
+ new Range(9, 11),
+ new Range(9, 11),
+ new Range(15, 17),
+ new Range(15, 17),
+ new Range(18, 20));
+ assertThat(selected)
+ .extracting(DataFileMeta::fileName)
+ .doesNotContainAnyElementsOf(filesInRange(table, new Range(12,
14)));
+ return rewriteClusterUnits(table, manager, units, true);
+ }
+
+ private void assertMinorClusterResult(
+ FileStoreTable table,
+ boolean deletionVectors,
+ List<String> untouchedRows,
+ List<String> untouchedFiles,
+ long firstAssignedRowId,
+ int expectedFileCount)
+ throws Exception {
+ List<String> expected = new ArrayList<>(untouchedRows);
+ long rowId = firstAssignedRowId;
+ for (int id : new int[] {10, 20, 30, 70, 80, 90, 100, 110, 120}) {
+ if (deletionVectors && id == 10) {
+ continue;
+ }
+ expected.add("a|" + id + "|" + (id < 100 ? "updated-" : "base-") +
id + "|" + rowId++);
+ }
+ assertThat(readRows(table)).containsExactlyElementsOf(expected);
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(rowId);
+ assertThat(filesInRange(table, new Range(12, 14)))
+ .containsExactlyElementsOf(untouchedFiles);
+ assertThat(fileNames(table,
"a")).hasSize(expectedFileCount).containsAll(untouchedFiles);
+ }
+
+ private List<String> rowsInRange(FileStoreTable table, Range range) throws
Exception {
+ return readRows(table).stream()
+ .filter(
+ row -> {
+ long rowId =
Long.parseLong(row.substring(row.lastIndexOf('|') + 1));
+ return rowId >= range.from && rowId <= range.to;
+ })
+ .collect(Collectors.toList());
+ }
+
+ private List<String> filesInRange(FileStoreTable table, Range range) {
+ return
table.store().newScan().withSnapshot(table.latestSnapshot().get()).plan().files()
+ .stream()
+ .filter(entry -> entry.partition().equals(partition(table,
"a")))
+ .map(entry -> entry.file())
+ .filter(file -> file.nonNullRowIdRange().equals(range))
+ .map(DataFileMeta::fileName)
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
private void append(FileStoreTable table, String partition, int... values)
throws Exception {
BatchWriteBuilder builder = table.newBatchWriteBuilder();
try (BatchTableWrite write = builder.newWrite();
@@ -651,6 +1457,13 @@ class DataEvolutionClusterTest extends TableTestBase {
return rewriteClusterUnits(table, manager,
manager.createCompactUnits(fullCompaction));
}
+ private List<CommitMessage> prepareMappedCluster(
+ FileStoreTable table, Snapshot snapshot, boolean fullCompaction)
throws Exception {
+ IncrementalClusterManager manager = new
IncrementalClusterManager(table, null, snapshot);
+ return rewriteClusterUnits(
+ table, manager, manager.createCompactUnits(fullCompaction),
true);
+ }
+
// 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(
@@ -658,6 +1471,15 @@ class DataEvolutionClusterTest extends TableTestBase {
IncrementalClusterManager manager,
Map<BinaryRow, CompactUnit> units)
throws Exception {
+ return rewriteClusterUnits(table, manager, units, false);
+ }
+
+ private List<CommitMessage> rewriteClusterUnits(
+ FileStoreTable table,
+ IncrementalClusterManager manager,
+ Map<BinaryRow, CompactUnit> units,
+ boolean preserveIndexes)
+ throws Exception {
Map<BinaryRow, Pair<List<DataSplit>, CommitMessage>> planned =
manager.toSplitsAndRewriteDvFiles(units);
List<CommitMessage> messages = new ArrayList<>();
@@ -667,10 +1489,17 @@ class DataEvolutionClusterTest extends TableTestBase {
Pair<List<DataSplit>, CommitMessage> partitionPlan =
planned.get(entry.getKey());
assertThat(partitionPlan.getRight()).isNull();
List<DataFileMeta> after = new ArrayList<>();
+ List<Long> oldRowIds = new ArrayList<>();
+ RowType readType =
+ preserveIndexes
+ ? SpecialFields.rowTypeWithRowId(table.rowType())
+ : table.rowType();
+ ProjectedRow dataRow = ProjectedRow.from(table.rowType(),
readType);
try (IOManager ioManager = IOManager.create(tempPath.toString());
RecordReaderIterator<InternalRow> reader =
new RecordReaderIterator<>(
table.newReadBuilder()
+ .withReadType(readType)
.newRead()
.createReader(
new
ArrayList<>(partitionPlan.getLeft())));
@@ -681,13 +1510,16 @@ class DataEvolutionClusterTest extends TableTestBase {
.withWriteType(table.rowType())) {
if (reader.hasNext()) {
Sorter sorter =
- Sorter.getSorter(
- reader, ioManager, table.rowType(),
table.coreOptions());
+ Sorter.getSorter(reader, ioManager, readType,
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));
+ InternalRow sortedRow = sorter.removeSortKey(row);
+ if (preserveIndexes) {
+
oldRowIds.add(sortedRow.getLong(readType.getFieldCount() - 1));
+ }
+ write.write(dataRow.replaceRow(sortedRow));
}
} finally {
sorter.close();
@@ -697,6 +1529,24 @@ class DataEvolutionClusterTest extends TableTestBase {
after.addAll(((CommitMessageImpl)
output).newFilesIncrement().newFiles());
}
}
+ DataFilePathFactory pathFactory =
+
table.store().pathFactory().createDataFilePathFactory(entry.getKey(), 0);
+ int rowOffset = 0;
+ for (int i = 0; preserveIndexes && i < after.size(); i++) {
+ DataFileMeta file = after.get(i);
+ String mappingName =
ClusteringRowIdMapping.fileName(file.fileName());
+ try (ClusteringRowIdMapping.Writer mapping =
+ ClusteringRowIdMapping.createWriter(
+ table.fileIO(),
pathFactory.toAlignedPath(mappingName, file))) {
+ for (long row = 0; row < file.rowCount(); row++) {
+ mapping.write(oldRowIds.get(rowOffset++));
+ }
+ }
+ List<String> extraFiles = new ArrayList<>(file.extraFiles());
+ extraFiles.add(mappingName);
+ after.set(i, file.copy(extraFiles));
+ }
+ assertThat(rowOffset).isEqualTo(oldRowIds.size());
messages.add(
new CommitMessageImpl(
entry.getKey(),
@@ -774,6 +1624,11 @@ class DataEvolutionClusterTest extends TableTestBase {
private void addGlobalIndex(
FileStoreTable table, String value, String fileName, Range range,
String column)
throws Exception {
+ commit(table, globalIndexMessage(table, value, fileName, range,
column));
+ }
+
+ private CommitMessage globalIndexMessage(
+ FileStoreTable table, String value, String fileName, Range range,
String column) {
IndexFileMeta index =
new IndexFileMeta(
"test-global-index",
@@ -787,14 +1642,12 @@ class DataEvolutionClusterTest extends TableTestBase {
null,
null),
null);
- commit(
- table,
- new CommitMessageImpl(
- partition(table, value),
- UNAWARE_BUCKET,
- null,
-
DataIncrement.indexIncrement(Collections.singletonList(index)),
- CompactIncrement.emptyIncrement()));
+ return new CommitMessageImpl(
+ partition(table, value),
+ UNAWARE_BUCKET,
+ null,
+ DataIncrement.indexIncrement(Collections.singletonList(index)),
+ CompactIncrement.emptyIncrement());
}
private void commit(FileStoreTable table, CommitMessage message) throws
Exception {
@@ -809,40 +1662,68 @@ class DataEvolutionClusterTest extends TableTestBase {
private List<CommitMessage> prepareGlobalIndex(
FileStoreTable table, Snapshot snapshot, String column) throws
Exception {
+ return prepareGlobalIndex(table, snapshot, column, "btree", null);
+ }
+
+ private List<CommitMessage> prepareGlobalIndex(
+ FileStoreTable table,
+ Snapshot snapshot,
+ String column,
+ String indexType,
+ Range fileRange)
+ throws Exception {
List<CommitMessage> messages = new ArrayList<>();
for (DataSplit split :
GlobalIndexBuilderUtils.splitByContiguousRowRange(
- new SortedGlobalIndexScanner(table, "btree")
+ new SortedGlobalIndexScanner(table, indexType)
.withIndexFields(Collections.singletonList(column))
.withSnapshot(snapshot)
.scan()
.get()
.entries())) {
+ if (fileRange != null) {
+ Optional<DataSplit> filtered =
+ split.filterDataFile(file ->
file.nonNullRowIdRange().equals(fileRange));
+ if (!filtered.isPresent()) {
+ continue;
+ }
+ split = filtered.get();
+ }
messages.addAll(
SortedGlobalIndexTestUtils.buildIndex(
- table, "btree", column, split, snapshot.id()));
+ table, indexType, column, split, snapshot.id()));
}
assertThat(messages).isNotEmpty();
return messages;
}
private List<String> readRows(FileStoreTable table) throws Exception {
+ return readRows(table, null);
+ }
+
+ private List<String> readRows(FileStoreTable table, Predicate predicate)
throws Exception {
ReadBuilder builder =
table.newReadBuilder()
.withReadType(SpecialFields.rowTypeWithRowId(table.rowType()));
+ if (predicate != null) {
+ builder.withFilter(predicate);
+ }
List<String> rows = new ArrayList<>();
try (RecordReader<InternalRow> reader =
builder.newRead().createReader(builder.newScan().plan())) {
reader.forEachRemaining(
- row ->
+ row -> {
+ if (predicate == null || predicate.test(row)) {
rows.add(
row.getString(0)
+ "|"
+ row.getInt(1)
+ "|"
- + row.getString(2)
+ + (row.isNullAt(2) ? "null" :
row.getString(2))
+ "|"
- + row.getLong(3)));
+ + row.getLong(3));
+ }
+ });
}
rows.sort(
Comparator.comparingLong(
@@ -850,6 +1731,38 @@ class DataEvolutionClusterTest extends TableTestBase {
return rows;
}
+ private void appendIndexRows(FileStoreTable table, int[] ids, String[]
payloads)
+ throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < ids.length; i++) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString("a"),
+ ids[i],
+ payloads[i] == null ? null :
BinaryString.fromString(payloads[i])));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private void assertIndexRows(FileStoreTable table, Predicate predicate,
Range... expected)
+ throws Exception {
+ Optional<DataEvolutionGlobalIndexScanner> scanner =
+ DataEvolutionGlobalIndexScanner.create(
+ table, PartitionPredicate.ALWAYS_TRUE, predicate);
+ if (!scanner.isPresent() && expected.length == 0) {
+ return;
+ }
+ assertThat(scanner).isPresent();
+ try (DataEvolutionGlobalIndexScanner index = scanner.get()) {
+ Optional<GlobalIndexResult> result = index.scan(predicate);
+ assertThat(result).isPresent();
+
assertThat(result.get().results().toRangeList()).containsExactly(expected);
+ }
+ }
+
private List<String> fileNames(FileStoreTable table, String value) {
BinaryRow partition = partition(table, value);
return
table.store().newScan().withSnapshot(table.latestSnapshot().get()).plan().files()
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index 99fd1f6532..4c9291b25b 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -77,6 +77,7 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FailingFileIO;
import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TraceableFileIO;
@@ -999,14 +1000,26 @@ public class FileStoreCommitTest {
TestFileStore store = createStore(false, 2);
BinaryRow partition = gen.getPartition(gen.next());
IndexManifestEntry materializedBucketDelete =
- globalIndexDeleteEntry(partition, 0,
"materialized-bucket-index");
+ globalIndexDeleteEntry(partition, 0,
"materialized-bucket-index", new Range(0, 0));
+ IndexManifestEntry disjointRangeDelete =
+ globalIndexDeleteEntry(partition, 0, "disjoint-range-index",
new Range(1, 2));
IndexManifestEntry otherBucketDelete =
- globalIndexDeleteEntry(partition, 1, "other-bucket-index");
+ globalIndexDeleteEntry(partition, 1, "other-bucket-index", new
Range(0, 0));
+ IndexManifestEntry nonGlobalDelete =
+ new IndexManifestEntry(
+ FileKind.DELETE,
+ partition,
+ 0,
+ new IndexFileMeta(HASH_INDEX, "non-global-index", 1,
1, null, null, null));
CommitChanges changes =
new CommitChanges(
Collections.emptyList(),
Collections.emptyList(),
- Arrays.asList(materializedBucketDelete,
otherBucketDelete));
+ Arrays.asList(
+ materializedBucketDelete,
+ disjointRangeDelete,
+ otherBucketDelete,
+ nonGlobalDelete));
try (FileStoreCommitImpl commit = store.newCommit()) {
CommitChanges refreshed =
@@ -1014,15 +1027,22 @@ public class FileStoreCommitTest {
changes,
Collections.singleton(Pair.of(partition, 0)))
.provide(null);
-
assertThat(refreshed.indexFiles).containsExactly(otherBucketDelete);
+ // Global-index removals from earlier preparation are replaced for
the entire bucket,
+ // even if their ranges do not overlap the materialized input.
+
assertThat(refreshed.indexFiles).containsExactly(otherBucketDelete,
nonGlobalDelete);
}
}
private static IndexManifestEntry globalIndexDeleteEntry(
- BinaryRow partition, int bucket, String fileName) {
+ BinaryRow partition, int bucket, String fileName, Range range) {
IndexFileMeta file =
new IndexFileMeta(
- "btree", fileName, 1, 1, new GlobalIndexMeta(0, 0, 0,
null, null), null);
+ "btree",
+ fileName,
+ 1,
+ range.count(),
+ new GlobalIndexMeta(range.from, range.to, 0, null,
null),
+ null);
return new IndexManifestEntry(FileKind.DELETE, partition, bucket,
file);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/cluster/IncrementalClusterSplitSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/cluster/IncrementalClusterSplitSource.java
index 25fe923df5..967d862553 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/cluster/IncrementalClusterSplitSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/cluster/IncrementalClusterSplitSource.java
@@ -27,9 +27,11 @@ import org.apache.paimon.flink.source.SimpleSourceSplit;
import org.apache.paimon.flink.source.operator.ReadOperator;
import org.apache.paimon.flink.utils.JavaTypeInfo;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Pair;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
@@ -108,15 +110,23 @@ public class IncrementalClusterSplitSource extends
AbstractNonCoordinatedSource<
partitioned.setParallelism(parallelism);
}
+ RowType readType =
+ table.coreOptions().dataEvolutionEnabled()
+ ? SpecialFields.rowTypeWithRowId(table.rowType())
+ : table.rowType();
+ ReadOperator reader =
+ table.coreOptions().dataEvolutionEnabled()
+ ? new ReadOperator(() ->
table.newRead().withReadType(readType), null, null)
+ : new ReadOperator(table::newRead, null, null);
+
return Pair.of(
new DataStream<>(source.getExecutionEnvironment(), partitioned)
.transform(
String.format(
"Incremental-cluster reader: %s - %s",
table.fullName(), partitionSpec),
- InternalTypeInfo.of(
-
LogicalTypeConversion.toLogicalType(table.rowType())),
- new ReadOperator(table::newRead, null, null))
+
InternalTypeInfo.of(LogicalTypeConversion.toLogicalType(readType)),
+ reader)
.setParallelism(partitioned.getParallelism()),
source.forward()
.transform(
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 ef5a58632b..b848f2467e 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
@@ -28,6 +28,7 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.JoinedRow;
import org.apache.paimon.flink.FlinkRowData;
import org.apache.paimon.flink.FlinkRowWrapper;
+import org.apache.paimon.flink.cluster.DataEvolutionClusterWriteOperator;
import org.apache.paimon.flink.cluster.IncrementalClusterSplitSource;
import
org.apache.paimon.flink.cluster.RewriteIncrementalClusterCommittableOperator;
import org.apache.paimon.flink.sink.Committable;
@@ -43,6 +44,7 @@ import org.apache.paimon.flink.utils.InternalTypeInfo;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.types.RowType;
@@ -71,6 +73,9 @@ 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.flink.FlinkConnectorOptions.SINK_MANAGED_WRITER_BUFFER_MEMORY;
+import static
org.apache.paimon.flink.FlinkConnectorOptions.SINK_USE_MANAGED_MEMORY;
+import static
org.apache.paimon.flink.utils.ManagedMemoryUtils.declareManagedMemory;
import static org.apache.paimon.table.PrimaryKeyTableUtils.addKeyNamePrefix;
/** Compact for incremental clustering. */
@@ -169,9 +174,13 @@ public class IncrementalClusterCompact {
}
DataStream<RowData> sorted;
+ RowType clusterRowType =
+ table.coreOptions().dataEvolutionEnabled()
+ ? SpecialFields.rowTypeWithRowId(table.rowType())
+ : table.rowType();
CoreOptions.ClusteringIncrementalMode mode =
clusterManager.clusteringIncrementalMode();
if (mode == CoreOptions.ClusteringIncrementalMode.LOCAL_SORT) {
- sorted = localSort(sourcePair.getLeft(), sinkParallelism);
+ sorted = localSort(sourcePair.getLeft(), sinkParallelism,
clusterRowType);
} else {
// GLOBAL_SORT: range shuffle + local sort (default)
TableSortInfo sortInfo =
@@ -188,22 +197,34 @@ public class IncrementalClusterCompact {
env,
sourcePair.getLeft(),
table.coreOptions(),
- table.rowType(),
+ clusterRowType,
sortInfo)
.sort();
}
// 2.3 write and then reorganize the committable
// set parallelism to null, and it'll forward parallelism when
doWrite()
- RowAppendTableSink sink = new RowAppendTableSink(table, null, null);
- DataStream<Committable> written =
- sink.doWrite(
- FlinkSinkBuilder.mapToInternalRow(
- sorted,
- table.rowType(),
- table.catalogEnvironment().catalogContext()),
- commitUser,
- null);
+ DataStream<InternalRow> rows =
+ FlinkSinkBuilder.mapToInternalRow(
+ sorted, clusterRowType,
table.catalogEnvironment().catalogContext());
+ DataStream<Committable> written;
+ if (table.coreOptions().dataEvolutionEnabled()) {
+ written =
+ rows.forward()
+ .transform(
+ "Write data evolution cluster row-id
mappings",
+ new CommittableTypeInfo(),
+ new DataEvolutionClusterWriteOperator(
+ table, partition, commitUser))
+ .setParallelism(sinkParallelism);
+ Options writeOptions = Options.fromMap(table.options());
+ if (writeOptions.get(SINK_USE_MANAGED_MEMORY)) {
+ declareManagedMemory(written,
writeOptions.get(SINK_MANAGED_WRITER_BUFFER_MEMORY));
+ }
+ } else {
+ RowAppendTableSink sink = new RowAppendTableSink(table, null,
null);
+ written = sink.doWrite(rows, commitUser, null);
+ }
DataStream<Committable> clusterCommittable =
written.forward()
.transform(
@@ -233,9 +254,9 @@ public class IncrementalClusterCompact {
*
* <p>Pipeline: add key prefix → SortOperator → remove key prefix → back
to RowData
*/
- private DataStream<RowData> localSort(DataStream<RowData> input, int
sinkParallelism) {
+ private DataStream<RowData> localSort(
+ DataStream<RowData> input, int sinkParallelism, RowType rowType) {
CoreOptions options = table.coreOptions();
- RowType rowType = table.rowType();
List<String> clusterKeys = clusterManager.clusterKeys();
final int[] keyProjectionMap = rowType.projectIndexes(clusterKeys);
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 e0504d20bd..4569ff31df 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
@@ -175,11 +175,103 @@ public class DataEvolutionDeleteSqlITCase extends
CatalogITCaseBase {
.containsExactlyElementsOf(expectedRowIds);
assertClusteredFileOrder(table, mode);
assertThat(deletionVectorCardinality(table)).isZero();
- assertThat(globalIndexPartitions(table)).containsExactly("B");
+ assertThat(globalIndexPartitions(table)).containsExactly("A", "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
+ @CsvSource({"btree,global-sort", "btree,local-sort", "bitmap,global-sort",
"bitmap,local-sort"})
+ public void testIncrementalClusteringRemapsScalarIndexRowIds(String
indexType, String mode)
+ throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ createTable();
+ sql(
+ "ALTER TABLE T SET ("
+ + "'write-only' = 'true', "
+ + "'clustering.incremental' = 'true', "
+ + "'clustering.strategy' = 'order', "
+ + "'clustering.columns' = 'id', "
+ + "'clustering.incremental.mode' = '"
+ + mode
+ + "', 'scan.parallelism' = '2', "
+ + "'source.split.target-size' = '1 b', "
+ + "'sink.parallelism' = '1', 'target-file-row-num' =
'7')");
+ List<String> odds = new ArrayList<>();
+ List<String> evens = new ArrayList<>();
+ for (int id = 39; id >= 1; id -= 2) {
+ odds.add(
+ "("
+ + id
+ + ", "
+ + (id == 9 ? "CAST(NULL AS STRING)" : "'key-" + id
+ "'")
+ + ", 'A')");
+ evens.add("(" + (id + 1) + ", 'key-" + (id + 1) + "', 'A')");
+ }
+ sql("INSERT INTO T VALUES " + String.join(", ", odds));
+ sql("INSERT INTO T VALUES " + String.join(", ", evens));
+ sql("INSERT INTO T VALUES (41, 'key-41', 'B')");
+ sql(
+ "CALL sys.create_global_index(`table` => 'default.T', "
+ + "index_column => 'name', index_type => '"
+ + indexType
+ + "')");
+ // The old postings still contain these rows; the clustering mapping
must omit them.
+ sql("DELETE FROM T WHERE id IN (2, 11)");
+ FileStoreTable table = paimonTable("T");
+ List<String> oldAIndexes = scalarIndexFiles(table, indexType, "A");
+ List<String> oldBIndexes = scalarIndexFiles(table, indexType, "B");
+ assertThat(oldAIndexes).isNotEmpty();
+ assertThat(oldBIndexes).isNotEmpty();
+ long beforeSnapshot = table.latestSnapshot().get().id();
+ long firstNewRowId = table.latestSnapshot().get().nextRowId();
+
+ sql(
+ "CALL sys.compact(`table` => 'default.T', partitions =>
'dt=A', "
+ + "compact_strategy => 'full')");
+
+ // Data and remapped indexes become visible atomically, without a
second index-build job.
+ assertThat(table.latestSnapshot().get().id()).isEqualTo(beforeSnapshot
+ 1);
+ assertThat(table.latestSnapshot().get().commitKind())
+ .isEqualTo(Snapshot.CommitKind.COMPACT);
+ assertThat(scalarIndexFiles(table, indexType, "A"))
+ .isNotEmpty()
+ .doesNotContainAnyElementsOf(oldAIndexes);
+ assertThat(scalarIndexFiles(table, indexType,
"B")).isEqualTo(oldBIndexes);
+ assertThat(
+
table.store().newIndexFileHandler().scan(indexType).stream()
+ .filter(
+ entry ->
+ entry.partition()
+ .getString(0)
+ .toString()
+ .equals("A")))
+ .allSatisfy(
+ entry ->
+
assertThat(entry.indexFile().globalIndexMeta().rowRangeStart())
+
.isGreaterThanOrEqualTo(firstNewRowId));
+ assertThat(sql("SELECT _ROW_ID FROM T$row_tracking WHERE dt = 'A'"))
+ .hasSize(38)
+ .allSatisfy(
+ row ->
+ assertThat((Long) row.getField(0))
+
.isGreaterThanOrEqualTo(firstNewRowId));
+
+ // FAST searches only indexed rows, so data-scan fallback cannot hide
missing postings.
+ sql("ALTER TABLE T SET ('scalar-index.search-mode' = 'fast')");
+ assertThat(sql("SELECT id FROM T WHERE name IN ('key-5', 'key-2',
'key-11') ORDER BY id"))
+ .containsExactly(Row.of(5));
+ assertThat(sql("SELECT id FROM T WHERE name IS
NULL")).containsExactly(Row.of(9));
+ List<Row> negativeMatches = new ArrayList<>();
+ for (int id = 1; id <= 40; id++) {
+ if (id != 2 && id != 11 && id != 5 && id != 9) {
+ negativeMatches.add(Row.of(id));
+ }
+ }
+ assertThat(sql("SELECT id FROM T WHERE dt = 'A' AND name <> 'key-5'
ORDER BY id"))
+ .containsExactlyElementsOf(negativeMatches);
+ }
+
@ParameterizedTest
@ValueSource(strings = {"global-sort", "local-sort"})
public void testIncrementalClusteringPreservesInlineBlobViews(String mode)
throws Exception {
@@ -614,6 +706,15 @@ public class DataEvolutionDeleteSqlITCase extends
CatalogITCaseBase {
.collect(Collectors.toList());
}
+ private static List<String> scalarIndexFiles(
+ FileStoreTable table, String indexType, String partition) {
+ return table.store().newIndexFileHandler().scan(indexType).stream()
+ .filter(entry ->
entry.partition().getString(0).toString().equals(partition))
+ .map(entry -> entry.indexFile().fileName())
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
private static void assertDeleteSnapshot(
FileStoreTable table, List<String> originalFiles, long
deletedRows) {
assertThat(table.latestSnapshot().get().operation()).isEqualTo(Snapshot.Operation.DELETE);
