forked from apache/paimon
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[core] Introduce ChangelogDeletion to Expiration (apache#3383)
- Loading branch information
Showing
18 changed files
with
596 additions
and
275 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
166 changes: 166 additions & 0 deletions
166
paimon-common/src/main/java/org/apache/paimon/options/ExpireConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,166 @@ | ||
| /* | ||
| * 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.options; | ||
|
|
||
| import java.time.Duration; | ||
|
|
||
| /** Config of snapshot and changelog expiration. */ | ||
| public class ExpireConfig { | ||
| private final int snapshotRetainMax; | ||
| private final int snapshotRetainMin; | ||
| private final Duration snapshotTimeRetain; | ||
| private final int snapshotMaxDeletes; | ||
| private final int changelogRetainMax; | ||
| private final int changelogRetainMin; | ||
| private final Duration changelogTimeRetain; | ||
| private final int changelogMaxDeletes; | ||
| private final boolean changelogDecoupled; | ||
|
|
||
| public ExpireConfig( | ||
| int snapshotRetainMax, | ||
| int snapshotRetainMin, | ||
| Duration snapshotTimeRetain, | ||
| int snapshotMaxDeletes, | ||
| int changelogRetainMax, | ||
| int changelogRetainMin, | ||
| Duration changelogTimeRetain, | ||
| int changelogMaxDeletes) { | ||
| this.snapshotRetainMax = snapshotRetainMax; | ||
| this.snapshotRetainMin = snapshotRetainMin; | ||
| this.snapshotTimeRetain = snapshotTimeRetain; | ||
| this.snapshotMaxDeletes = snapshotMaxDeletes; | ||
| this.changelogRetainMax = changelogRetainMax; | ||
| this.changelogRetainMin = changelogRetainMin; | ||
| this.changelogTimeRetain = changelogTimeRetain; | ||
| this.changelogMaxDeletes = changelogMaxDeletes; | ||
| this.changelogDecoupled = | ||
| changelogRetainMax > snapshotRetainMax | ||
| || changelogRetainMin > snapshotRetainMin | ||
| || changelogTimeRetain.compareTo(snapshotTimeRetain) > 0; | ||
| } | ||
|
|
||
| public int getSnapshotRetainMax() { | ||
| return snapshotRetainMax; | ||
| } | ||
|
|
||
| public int getSnapshotRetainMin() { | ||
| return snapshotRetainMin; | ||
| } | ||
|
|
||
| public Duration getSnapshotTimeRetain() { | ||
| return snapshotTimeRetain; | ||
| } | ||
|
|
||
| public int getChangelogRetainMax() { | ||
| return changelogRetainMax; | ||
| } | ||
|
|
||
| public int getChangelogRetainMin() { | ||
| return changelogRetainMin; | ||
| } | ||
|
|
||
| public Duration getChangelogTimeRetain() { | ||
| return changelogTimeRetain; | ||
| } | ||
|
|
||
| public int getSnapshotMaxDeletes() { | ||
| return snapshotMaxDeletes; | ||
| } | ||
|
|
||
| public int getChangelogMaxDeletes() { | ||
| return changelogMaxDeletes; | ||
| } | ||
|
|
||
| public boolean isChangelogDecoupled() { | ||
| return changelogDecoupled; | ||
| } | ||
|
|
||
| public static Builder builder() { | ||
| return new Builder(); | ||
| } | ||
|
|
||
| /** The builder for {@link ExpireConfig}. */ | ||
| public static final class Builder { | ||
| private int snapshotRetainMax = Integer.MAX_VALUE; | ||
| private int snapshotRetainMin = 1; | ||
| private Duration snapshotTimeRetain = Duration.ofMillis(Long.MAX_VALUE); | ||
| private int snapshotMaxDeletes = Integer.MAX_VALUE; | ||
| // If changelog config is not set, use the snapshot's by default | ||
| private Integer changelogRetainMax = null; | ||
| private Integer changelogRetainMin = null; | ||
| private Duration changelogTimeRetain = null; | ||
| private Integer changelogMaxDeletes = null; | ||
|
|
||
| public static Builder builder() { | ||
| return new Builder(); | ||
| } | ||
|
|
||
| public Builder snapshotRetainMax(int snapshotRetainMax) { | ||
| this.snapshotRetainMax = snapshotRetainMax; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder snapshotRetainMin(int snapshotRetainMin) { | ||
| this.snapshotRetainMin = snapshotRetainMin; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder snapshotTimeRetain(Duration snapshotTimeRetain) { | ||
| this.snapshotTimeRetain = snapshotTimeRetain; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder snapshotMaxDeletes(int snapshotMaxDeletes) { | ||
| this.snapshotMaxDeletes = snapshotMaxDeletes; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder changelogRetainMax(Integer changelogRetainMax) { | ||
| this.changelogRetainMax = changelogRetainMax; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder changelogRetainMin(Integer changelogRetainMin) { | ||
| this.changelogRetainMin = changelogRetainMin; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder changelogTimeRetain(Duration changelogTimeRetain) { | ||
| this.changelogTimeRetain = changelogTimeRetain; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder changelogMaxDeletes(Integer changelogMaxDeletes) { | ||
| this.changelogMaxDeletes = changelogMaxDeletes; | ||
| return this; | ||
| } | ||
|
|
||
| public ExpireConfig build() { | ||
| return new ExpireConfig( | ||
| snapshotRetainMax, | ||
| snapshotRetainMin, | ||
| snapshotTimeRetain, | ||
| snapshotMaxDeletes, | ||
| changelogRetainMax == null ? snapshotRetainMax : changelogRetainMax, | ||
| changelogRetainMin == null ? snapshotRetainMin : changelogRetainMin, | ||
| changelogTimeRetain == null ? snapshotTimeRetain : changelogTimeRetain, | ||
| changelogMaxDeletes == null ? snapshotMaxDeletes : changelogMaxDeletes); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
117 changes: 117 additions & 0 deletions
117
paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| /* | ||
| * 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.operation; | ||
|
|
||
| import org.apache.paimon.Changelog; | ||
| import org.apache.paimon.Snapshot; | ||
| import org.apache.paimon.fs.FileIO; | ||
| import org.apache.paimon.index.IndexFileHandler; | ||
| import org.apache.paimon.index.IndexFileMeta; | ||
| import org.apache.paimon.manifest.IndexManifestEntry; | ||
| import org.apache.paimon.manifest.ManifestEntry; | ||
| import org.apache.paimon.manifest.ManifestFile; | ||
| import org.apache.paimon.manifest.ManifestFileMeta; | ||
| import org.apache.paimon.manifest.ManifestList; | ||
| import org.apache.paimon.stats.StatsFileHandler; | ||
| import org.apache.paimon.utils.FileStorePathFactory; | ||
|
|
||
| import java.util.HashSet; | ||
| import java.util.List; | ||
| import java.util.Set; | ||
| import java.util.function.Predicate; | ||
|
|
||
| /** Delete changelog files. */ | ||
| public class ChangelogDeletion extends FileDeletionBase<Changelog> { | ||
| public ChangelogDeletion( | ||
| FileIO fileIO, | ||
| FileStorePathFactory pathFactory, | ||
| ManifestFile manifestFile, | ||
| ManifestList manifestList, | ||
| IndexFileHandler indexFileHandler, | ||
| StatsFileHandler statsFileHandler) { | ||
| super(fileIO, pathFactory, manifestFile, manifestList, indexFileHandler, statsFileHandler); | ||
| } | ||
|
|
||
| @Override | ||
| public void cleanUnusedDataFiles(Changelog changelog, Predicate<ManifestEntry> skipper) { | ||
| if (changelog.changelogManifestList() != null) { | ||
| deleteAddedDataFiles(changelog.changelogManifestList()); | ||
| } | ||
|
|
||
| if (manifestList.exists(changelog.deltaManifestList())) { | ||
| cleanUnusedDataFiles(changelog.deltaManifestList(), skipper); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void cleanUnusedManifests(Changelog changelog, Set<String> skippingSet) { | ||
| if (changelog.changelogManifestList() != null) { | ||
| cleanUnusedManifestList(changelog.changelogManifestList(), skippingSet); | ||
| } | ||
|
|
||
| if (manifestList.exists(changelog.deltaManifestList())) { | ||
| cleanUnusedManifestList(changelog.deltaManifestList(), skippingSet); | ||
| } | ||
|
|
||
| if (manifestList.exists(changelog.baseManifestList())) { | ||
| cleanUnusedManifestList(changelog.baseManifestList(), skippingSet); | ||
| } | ||
|
|
||
| // the index and statics manifest list should handle by snapshot deletion. | ||
| } | ||
|
|
||
| public Set<String> manifestSkippingSet(List<Snapshot> skippingSnapshots) { | ||
| Set<String> skippingSet = new HashSet<>(); | ||
|
|
||
| for (Snapshot skippingSnapshot : skippingSnapshots) { | ||
| // base manifests | ||
| if (manifestList.exists(skippingSnapshot.baseManifestList())) { | ||
| skippingSet.add(skippingSnapshot.baseManifestList()); | ||
| manifestList.read(skippingSnapshot.baseManifestList()).stream() | ||
| .map(ManifestFileMeta::fileName) | ||
| .forEach(skippingSet::add); | ||
| } | ||
|
|
||
| // delta manifests | ||
| if (manifestList.exists(skippingSnapshot.deltaManifestList())) { | ||
| skippingSet.add(skippingSnapshot.deltaManifestList()); | ||
| manifestList.read(skippingSnapshot.deltaManifestList()).stream() | ||
| .map(ManifestFileMeta::fileName) | ||
| .forEach(skippingSet::add); | ||
| } | ||
|
|
||
| // index manifests | ||
| String indexManifest = skippingSnapshot.indexManifest(); | ||
| if (indexManifest != null) { | ||
| skippingSet.add(indexManifest); | ||
| indexFileHandler.readManifest(indexManifest).stream() | ||
| .map(IndexManifestEntry::indexFile) | ||
| .map(IndexFileMeta::fileName) | ||
| .forEach(skippingSet::add); | ||
| } | ||
|
|
||
| // statistics | ||
| if (skippingSnapshot.statistics() != null) { | ||
| skippingSet.add(skippingSnapshot.statistics()); | ||
| } | ||
| } | ||
|
|
||
| return skippingSet; | ||
| } | ||
| } |
Oops, something went wrong.