Skip to content

[Flink] Fix dedicated streaming compaction after snapshot expiration #9548

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lilei1128 wants to merge 1 commit into
base: master
Choose a base branch
from
+2,907 −134
Open

[Flink] Fix dedicated streaming compaction after snapshot expiration #9548

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
143 changes: 134 additions & 9 deletions paimon-core/src/main/java/org/apache/paimon/operation/FileSystemWriteRestore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,12 @@
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.table.source.OutOfRangeException;
import org.apache.paimon.utils.SnapshotManager;

import javax.annotation.Nullable;

import java.io.FileNotFoundException;
import java.util.ArrayList;
import java.util.List;

Expand All @@ -41,13 +43,16 @@ public class FileSystemWriteRestore implements WriteRestore {
private final FileStoreScan scan;
private final IndexFileHandler indexFileHandler;
private final @Nullable Long snapshotId;
private final boolean fallbackToLatest;
/** Snapshot selected after an expiration fallback; shared by all buckets of this restore. */
private @Nullable Snapshot fallbackSnapshot;

public FileSystemWriteRestore(
CoreOptions options,
SnapshotManager snapshotManager,
FileStoreScan scan,
IndexFileHandler indexFileHandler) {
this(options, snapshotManager, scan, indexFileHandler, null);
this(options, snapshotManager, scan, indexFileHandler, null, false);
}

public FileSystemWriteRestore(
Expand All @@ -56,19 +61,41 @@ public FileSystemWriteRestore(
FileStoreScan scan,
IndexFileHandler indexFileHandler,
long snapshotId) {
this(options, snapshotManager, scan, indexFileHandler, Long.valueOf(snapshotId));
this(options, snapshotManager, scan, indexFileHandler, Long.valueOf(snapshotId), false);
}

/**
* Creates a restore which prefers {@code snapshotId}, but falls back to the latest filesystem
* snapshot when expiration races with writer initialization.
*/
public FileSystemWriteRestore(
CoreOptions options,
SnapshotManager snapshotManager,
FileStoreScan scan,
IndexFileHandler indexFileHandler,
long snapshotId,
boolean fallbackToLatest) {
this(
options,
snapshotManager,
scan,
indexFileHandler,
Long.valueOf(snapshotId),
fallbackToLatest);
}

private FileSystemWriteRestore(
CoreOptions options,
SnapshotManager snapshotManager,
FileStoreScan scan,
IndexFileHandler indexFileHandler,
@Nullable Long snapshotId) {
@Nullable Long snapshotId,
boolean fallbackToLatest) {
this.snapshotManager = snapshotManager;
this.scan = scan;
this.indexFileHandler = indexFileHandler;
this.snapshotId = snapshotId;
this.fallbackToLatest = fallbackToLatest;
if (options.manifestDeleteFileDropStats()) {
if (this.scan != null) {
this.scan.dropStats();
Expand All @@ -91,16 +118,75 @@ public RestoreFiles restoreFiles(
boolean scanDynamicBucketIndex,
boolean scanDeleteVectorsIndex,
boolean scanSourceIndexPayloads) {
// NOTE: don't use snapshotManager.latestSnapshot() here,
// because we don't want to flood the catalog with high concurrency
Snapshot snapshot =
snapshotId == null
? snapshotManager.latestSnapshotFromFileSystem()
: snapshotManager.snapshot(snapshotId);
Snapshot snapshot = snapshot();
if (snapshot == null) {
return RestoreFiles.empty();
}

// Expiration can remove the manifest of the selected snapshot and then remove the
// replacement snapshot while it is being read. When fallback is explicitly enabled for
// a rebase, advance monotonically through a small bounded number of newer snapshots.
// Ordinary restores retain their original fail-fast behavior.
for (int attempt = 0; ; attempt++) {
try {
return restoreFiles(
snapshot,
partition,
bucket,
scanDynamicBucketIndex,
scanDeleteVectorsIndex,
scanSourceIndexPayloads);
} catch (RuntimeException e) {
if (!shouldFallback(e, snapshot) || attempt >= 2) {
throw e;
}

Snapshot latest = snapshotManager.latestSnapshotFromFileSystem();
if (latest == null || latest.id() <= snapshot.id()) {
throw e;
}
snapshot = latest;
fallbackSnapshot = latest;
}
}
}

private Snapshot snapshot() {
// NOTE: don't use snapshotManager.latestSnapshot() here,
// because we don't want to flood the catalog with high concurrency
if (fallbackSnapshot != null) {
return fallbackSnapshot;
}
try {
return snapshotId == null
? snapshotManager.latestSnapshotFromFileSystem()
: snapshotManager.snapshot(snapshotId);
} catch (RuntimeException e) {
if (!fallbackToLatest
|| snapshotId == null
|| !containsExpiredSnapshotFailure(e)
|| !isExpiredOrSuperseded(snapshotId)) {
throw e;
}
Snapshot latest = snapshotManager.latestSnapshotFromFileSystem();
// A missing latest snapshot means the table is empty or the filesystem is still
// inconsistent. Do not turn that ambiguous condition into an empty restore and lose
// the original failure.
if (latest == null) {
throw e;
}
fallbackSnapshot = latest;
return latest;
}
}

private RestoreFiles restoreFiles(
Snapshot snapshot,
BinaryRow partition,
int bucket,
boolean scanDynamicBucketIndex,
boolean scanDeleteVectorsIndex,
boolean scanSourceIndexPayloads) {
List<DataFileMeta> restoreFiles = new ArrayList<>();
List<ManifestEntry> entries =
scan.withSnapshot(snapshot).withPartitionBucket(partition, bucket).plan().files();
Expand Down Expand Up @@ -131,4 +217,43 @@ public RestoreFiles restoreFiles(
deleteVectorsIndex,
sourceIndexPayloads);
}

private boolean shouldFallback(RuntimeException failure, Snapshot snapshot) {
return fallbackToLatest
&& snapshotId != null
&& containsExpiredSnapshotFailure(failure)
&& isExpiredOrSuperseded(snapshot.id());
}

private boolean isExpiredOrSuperseded(long requestedSnapshotId) {
if (snapshotManager.isSnapshotExpired(requestedSnapshotId)) {
return true;
}

// Expiration cleans manifests before deleting the snapshot file and advancing the earliest
// hint. Therefore a requested snapshot can still exist while one of its manifests or
// indexes is already gone. A newer complete snapshot is sufficient to identify that race;
// failures without one are left untouched.
try {
Snapshot latest = snapshotManager.latestSnapshotFromFileSystem();
return latest != null && latest.id() > requestedSnapshotId;
} catch (RuntimeException ignored) {
return false;
}
}

private boolean containsExpiredSnapshotFailure(Throwable failure) {
Throwable current = failure;
while (current != null) {
// A restore reads concrete snapshot/manifest files. Only explicit expiration signals
// can use this opt-in fallback; retained snapshots and unrelated failures must still
// fail the job.
if (current instanceof FileNotFoundException
|| current instanceof OutOfRangeException) {
return true;
}
current = current.getCause();
}
return false;
}
}
38 changes: 34 additions & 4 deletions paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,8 @@ public class DataSplit implements Split {

private boolean isStreaming = false;
private boolean rawConvertible;
/** Whether this split is a control split used to rebase a dedicated compaction writer. */
private transient boolean rebase;

public DataSplit() {}

Expand Down Expand Up @@ -122,6 +124,18 @@ public boolean rawConvertible() {
return rawConvertible;
}

public boolean isRebase() {
return rebase;
}

/** Returns a copy of this split marked as a dedicated compaction rebase split. */
public DataSplit asRebase() {
DataSplit copy = new DataSplit();
copy.assign(this);
copy.rebase = true;
return copy;
}

public OptionalLong earliestFileCreationEpochMillis() {
return this.dataFiles.stream().mapToLong(DataFileMeta::creationTimeEpochMillis).min();
}
Expand Down Expand Up @@ -349,6 +363,7 @@ public boolean equals(Object o) {
&& bucket == dataSplit.bucket
&& isStreaming == dataSplit.isStreaming
&& rawConvertible == dataSplit.rawConvertible
&& rebase == dataSplit.rebase
&& Objects.equals(partition, dataSplit.partition)
&& Objects.equals(bucketPath, dataSplit.bucketPath)
&& Objects.equals(totalBuckets, dataSplit.totalBuckets)
Expand All @@ -367,7 +382,8 @@ public int hashCode() {
dataFiles,
dataDeletionFiles,
isStreaming,
rawConvertible);
rawConvertible,
rebase);
}

@Override
Expand Down Expand Up @@ -404,11 +420,14 @@ protected void assign(DataSplit other) {
this.dataDeletionFiles = other.dataDeletionFiles;
this.isStreaming = other.isStreaming;
this.rawConvertible = other.rawConvertible;
this.rebase = other.rebase;
}

public void serialize(DataOutputView out) throws IOException {
out.writeLong(MAGIC);
out.writeInt(VERSION);
// Keep the wire format byte-for-byte compatible for ordinary splits. Rebase splits are an
// internal extension and carry the extra marker in version 10.
out.writeInt(rebase ? VERSION + 1 : VERSION);
out.writeLong(snapshotId);
SerializationUtils.serializeBinaryRow(partition, out);
out.writeInt(bucket);
Expand Down Expand Up @@ -436,6 +455,10 @@ public void serialize(DataOutputView out) throws IOException {
out.writeBoolean(isStreaming);

out.writeBoolean(rawConvertible);

if (rebase) {
out.writeBoolean(true);
}
}

public static DataSplit deserialize(DataInputView in) throws IOException {
Expand Down Expand Up @@ -473,6 +496,7 @@ public static DataSplit deserialize(DataInputView in) throws IOException {

boolean isStreaming = in.readBoolean();
boolean rawConvertible = in.readBoolean();
boolean rebase = version >= 10 && in.readBoolean();

DataSplit.Builder builder =
builder()
Expand All @@ -483,7 +507,8 @@ public static DataSplit deserialize(DataInputView in) throws IOException {
.withTotalBuckets(totalBuckets)
.withDataFiles(dataFiles)
.isStreaming(isStreaming)
.rawConvertible(rawConvertible);
.rawConvertible(rawConvertible)
.rebase(rebase);

if (dataDeletionFiles != null) {
builder.withDataDeletionFiles(dataDeletionFiles);
Expand Down Expand Up @@ -513,7 +538,7 @@ private static FunctionWithIOException<DataInputView, DataFileMeta> getFileMetaS
DataFileMetaWriteColsLegacySerializer serializer =
new DataFileMetaWriteColsLegacySerializer();
return serializer::deserialize;
} else if (version == 9) {
} else if (version == 9 || version == VERSION + 1) {
DataFileMetaSerializer serializer = new DataFileMetaSerializer();
return serializer::deserialize;
} else {
Expand Down Expand Up @@ -586,6 +611,11 @@ public Builder rawConvertible(boolean rawConvertible) {
return this;
}

public Builder rebase(boolean rebase) {
this.split.rebase = rebase;
return this;
}

public DataSplit build() {
checkArgument(split.partition != null);
checkArgument(split.bucket != -1);
Expand Down
Loading
Loading