Skip to content
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

[clone]refactor the clone action as we introduced external path #4844

Open
wants to merge 8 commits into
base: master
Choose a base branch
from
Open

[clone]refactor the clone action as we introduced external path #4844

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
644594d
fix the clone tests
neuyilan Dec 25, 2024
aeeb30f
refactor the clone job
neuyilan Jan 10, 2025
31bf4e7
add clone it for external path table
neuyilan Jan 10, 2025
80a034f
remove useless codes
neuyilan Jan 10, 2025
0f96789
remove FileType
neuyilan Jan 14, 2025
3e47968
merge master
neuyilan Jan 16, 2025
bb5523a
refactor the clone action
neuyilan Jan 17, 2025
902e659
ignore the testCloneTableWithExpiration
neuyilan Jan 20, 2025
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
22 changes: 22 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -484,6 +484,28 @@ public DataFileMeta copy(List<String> newExtraFiles) {
externalPath);
}

public DataFileMeta newExternalPath(String newExternalPath) {
return new DataFileMeta(
fileName,
fileSize,
rowCount,
minKey,
maxKey,
keyStats,
valueStats,
minSequenceNumber,
maxSequenceNumber,
schemaId,
level,
extraFiles,
creationTime,
deleteRowCount,
embeddedIndex,
fileSource,
valueStatsCols,
newExternalPath);
}

public DataFileMeta copy(byte[] newEmbeddedIndex) {
return new DataFileMeta(
fileName,
Expand Down
7 changes: 6 additions & 1 deletion paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,12 @@ public RollingFileWriter<ManifestEntry, ManifestFileMeta> createRollingWriter()
suggestedFileSize);
}

private class ManifestEntryWriter extends SingleFileWriter<ManifestEntry, ManifestFileMeta> {
public ManifestEntryWriter createManifestEntryWriter(Path manifestPath) {
return new ManifestEntryWriter(writerFactory, manifestPath, compression);
}

/** Writer for {@link ManifestEntry}. */
public class ManifestEntryWriter extends SingleFileWriter<ManifestEntry, ManifestFileMeta> {

private final SimpleStatsCollector partitionStatsCollector;
private final SimpleStatsConverter partitionStatsSerializer;
Expand Down
54 changes: 46 additions & 8 deletions ...n-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,10 @@
import org.apache.paimon.flink.clone.CloneFileInfo;
import org.apache.paimon.flink.clone.CloneSourceBuilder;
import org.apache.paimon.flink.clone.CopyFileOperator;
import org.apache.paimon.flink.clone.PickFilesForCloneOperator;
import org.apache.paimon.flink.clone.CopyManifestFileOperator;
import org.apache.paimon.flink.clone.PickDataFileForCloneOperator;
import org.apache.paimon.flink.clone.PickManifestFileForCloneOperator;
import org.apache.paimon.flink.clone.PickSchemaFilesForCloneOperator;
import org.apache.paimon.flink.clone.SnapshotHintChannelComputer;
import org.apache.paimon.flink.clone.SnapshotHintOperator;
import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
Expand Down Expand Up @@ -105,20 +108,55 @@ private void buildCloneFlinkJob(StreamExecutionEnvironment env) throws Exception
targetTableName)
.build();

SingleOutputStreamOperator<CloneFileInfo> pickFilesForClone =
SingleOutputStreamOperator<CloneFileInfo> pickSchemaFilesForClone =
cloneSource
.transform(
"Pick Files",
"Pick Schema Files",
TypeInformation.of(CloneFileInfo.class),
new PickFilesForCloneOperator(
new PickSchemaFilesForCloneOperator(
sourceCatalogConfig, targetCatalogConfig))
.forceNonParallel();
.setParallelism(1);

SingleOutputStreamOperator<CloneFileInfo> copySchemaFile =
pickSchemaFilesForClone
.transform(
"Copy Schema Files",
TypeInformation.of(CloneFileInfo.class),
new CopyFileOperator(sourceCatalogConfig, targetCatalogConfig))
.setParallelism(1);

SingleOutputStreamOperator<CloneFileInfo> pickManifestFile =
copySchemaFile
.transform(
"Pick Manifest Files",
TypeInformation.of(CloneFileInfo.class),
new PickManifestFileForCloneOperator(
sourceCatalogConfig, targetCatalogConfig))
.setParallelism(1);

SingleOutputStreamOperator<CloneFileInfo> copyManifestFile =
pickManifestFile
.transform(
"Copy Manifest Files",
TypeInformation.of(CloneFileInfo.class),
new CopyManifestFileOperator(
sourceCatalogConfig, targetCatalogConfig))
.setParallelism(parallelism);

SingleOutputStreamOperator<CloneFileInfo> pickDataFile =
copyManifestFile
.transform(
"Pick Data Files",
TypeInformation.of(CloneFileInfo.class),
new PickDataFileForCloneOperator(
sourceCatalogConfig, targetCatalogConfig))
.setParallelism(1);

SingleOutputStreamOperator<CloneFileInfo> copyFiles =
pickFilesForClone
.rebalance()
pickDataFile
.keyBy(CloneFileInfo::getSourceIdentifier) // key by source identifier
.transform(
"Copy Files",
"Copy Data Files",
TypeInformation.of(CloneFileInfo.class),
new CopyFileOperator(sourceCatalogConfig, targetCatalogConfig))
.setParallelism(parallelism);
Expand Down
31 changes: 23 additions & 8 deletions ...-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/CloneFileInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,29 +18,35 @@

package org.apache.paimon.flink.clone;

import javax.annotation.Nullable;

/** The information of copy file. */
public class CloneFileInfo {

private final String sourceFilePath;
private final String filePathExcludeTableRoot;
@Nullable private final String sourceFilePath;
@Nullable private final String filePathExcludeTableRoot;
private final String sourceIdentifier;
private final String targetIdentifier;
private final long snapshotId;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Where will this variable be used?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This variable is used to transfer selections between upstream and downstream. For example, when creating the schema file, the latest snapshotid of the snapshot is determined. When selecting the data file later, the snapshot id is used for selection. Because in this process, the latest snapshot may change. If the snapshot id is not passed, the cloned schema file and data file may not be from the same snapshot.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please remove this field, see my comments for operators.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we can not remove this filed. because if do not provided the snapshotId. when we do job in CopyManifestFilesOperator, we can not just pick the data files in manifest file, because the data files maybe delete in another mainifest file.

for example, in snapshot1, we add one data file data-file1.parquet in manifest-file1 ; in snapshot 2, we add one data file data-file2.parquet and delete data-file1.parquet in manifest-file2. And these two manifest files were processed in two separate tasks, when processing manifest-file1 and copy data-file1.parquet, the job will fail.

So we can not just pick the data files in manifest file. I think we still need the snapshot id. What do you think?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we don't need to worry about deleting files in the manifest anymore. We just need to see which data files we need. Although we may copy extra files, it won't cause any problems

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The problem is that, when snapshot1 is expired, then the data-file1.parquet will be deleted. when we read the manifest-file1 and copy data-file1.parquet, it will always fail, because the file do not exist.


public CloneFileInfo(
String sourceFilePath,
String filePathExcludeTableRoot,
@Nullable String sourceFilePath,
@Nullable String filePathExcludeTableRoot,
String sourceIdentifier,
String targetIdentifier) {
String targetIdentifier,
long snapshotId) {
this.sourceFilePath = sourceFilePath;
this.filePathExcludeTableRoot = filePathExcludeTableRoot;
this.sourceIdentifier = sourceIdentifier;
this.targetIdentifier = targetIdentifier;
this.snapshotId = snapshotId;
}

@Nullable
public String getSourceFilePath() {
return sourceFilePath;
}

@Nullable
public String getFilePathExcludeTableRoot() {
return filePathExcludeTableRoot;
}
Expand All @@ -53,10 +59,19 @@ public String getTargetIdentifier() {
return targetIdentifier;
}

public long getSnapshotId() {
return snapshotId;
}

@Override
public String toString() {
return String.format(
"{ sourceFilePath: %s, filePathExcludeTableRoot: %s, sourceIdentifier: %s, targetIdentifier: %s }",
sourceFilePath, filePathExcludeTableRoot, sourceIdentifier, targetIdentifier);
"{ sourceFilePath: %s, filePathExcludeTableRoot: %s, sourceIdentifier: %s, targetIdentifier: %s, "
+ "snapshotId: %d}",
sourceFilePath,
filePathExcludeTableRoot,
sourceIdentifier,
targetIdentifier,
snapshotId);
}
}
Loading
Loading