You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/07/14 07:05:42 UTC
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #214: [FLINK-28108] Support "ALTER TABLE ... COMPACT" for append-only table
JingsongLi commented on code in PR #214:
URL: https://github.com/apache/flink-table-store/pull/214#discussion_r920820896
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/data/AppendOnlyCompactManager.java:
##########
@@ -97,31 +108,90 @@ LinkedList<DataFileMeta> getToCompact() {
return toCompact;
}
- /** A {@link CompactTask} impl for append-only table. */
- public static class AppendOnlyCompactTask extends CompactTask {
+ /** A {@link CompactTask} impl for ALTER TABLE COMPACT of append-only table. */
+ public static class RollingCompactTask extends CompactTask {
+ private final long targetFileSize;
+ private final int minFileNum;
+ private final int maxFileNum;
private final CompactRewriter rewriter;
- public AppendOnlyCompactTask(List<DataFileMeta> toCompact, CompactRewriter rewriter) {
- super(toCompact);
+ public RollingCompactTask(
+ List<DataFileMeta> inputs,
+ long targetFileSize,
+ int minFileNum,
+ int maxFileNum,
+ CompactRewriter rewriter) {
+ super(inputs);
+ this.targetFileSize = targetFileSize;
+ this.minFileNum = minFileNum;
+ this.maxFileNum = maxFileNum;
this.rewriter = rewriter;
}
@Override
protected CompactResult doCompact(List<DataFileMeta> inputs) throws Exception {
- List<DataFileMeta> compactAfter = rewriter.rewrite(inputs);
- return new CompactResult() {
- @Override
- public List<DataFileMeta> before() {
- return inputs;
+ LinkedList<DataFileMeta> toCompact = new LinkedList<>(inputs);
+ Set<DataFileMeta> compactBefore = new HashSet<>();
+ List<DataFileMeta> compactAfter = new ArrayList<>();
+ while (!toCompact.isEmpty()) {
+ Optional<List<DataFileMeta>> candidates =
+ AppendOnlyCompactManager.pick(
+ toCompact, targetFileSize, minFileNum, maxFileNum);
+ if (candidates.isPresent()) {
+ List<DataFileMeta> before = candidates.get();
+ compactBefore.addAll(before);
+ List<DataFileMeta> after = rewriter.rewrite(before);
+ compactAfter.addAll(after);
+ DataFileMeta lastFile = after.get(after.size() - 1);
+ if (lastFile.fileSize() < targetFileSize) {
+ toCompact.offerFirst(lastFile);
+ }
+ } else {
+ break;
}
-
- @Override
- public List<DataFileMeta> after() {
- return compactAfter;
+ }
+ // remove intermediate files
+ Iterator<DataFileMeta> iterator = compactAfter.iterator();
+ while (iterator.hasNext()) {
+ DataFileMeta file = iterator.next();
+ if (compactBefore.contains(file)) {
+ compactBefore.remove(file);
+ iterator.remove();
Review Comment:
Do we need to delete file?
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/data/AppendOnlyCompactManager.java:
##########
@@ -97,31 +108,90 @@ LinkedList<DataFileMeta> getToCompact() {
return toCompact;
}
- /** A {@link CompactTask} impl for append-only table. */
- public static class AppendOnlyCompactTask extends CompactTask {
+ /** A {@link CompactTask} impl for ALTER TABLE COMPACT of append-only table. */
+ public static class RollingCompactTask extends CompactTask {
Review Comment:
Can we add more description comments?
It dose not look like a `Rolling`, maybe `IterativeCompactTask`?
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/data/AppendOnlyCompactManager.java:
##########
@@ -97,31 +108,90 @@ LinkedList<DataFileMeta> getToCompact() {
return toCompact;
}
- /** A {@link CompactTask} impl for append-only table. */
- public static class AppendOnlyCompactTask extends CompactTask {
+ /** A {@link CompactTask} impl for ALTER TABLE COMPACT of append-only table. */
+ public static class RollingCompactTask extends CompactTask {
+ private final long targetFileSize;
+ private final int minFileNum;
+ private final int maxFileNum;
private final CompactRewriter rewriter;
- public AppendOnlyCompactTask(List<DataFileMeta> toCompact, CompactRewriter rewriter) {
- super(toCompact);
+ public RollingCompactTask(
+ List<DataFileMeta> inputs,
+ long targetFileSize,
+ int minFileNum,
+ int maxFileNum,
+ CompactRewriter rewriter) {
+ super(inputs);
+ this.targetFileSize = targetFileSize;
+ this.minFileNum = minFileNum;
+ this.maxFileNum = maxFileNum;
this.rewriter = rewriter;
}
@Override
protected CompactResult doCompact(List<DataFileMeta> inputs) throws Exception {
- List<DataFileMeta> compactAfter = rewriter.rewrite(inputs);
- return new CompactResult() {
- @Override
- public List<DataFileMeta> before() {
- return inputs;
+ LinkedList<DataFileMeta> toCompact = new LinkedList<>(inputs);
+ Set<DataFileMeta> compactBefore = new HashSet<>();
Review Comment:
Do not use `HashSet`, at least, we can use `LinkedHashSet`.
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/data/AppendOnlyCompactManager.java:
##########
@@ -97,31 +108,90 @@ LinkedList<DataFileMeta> getToCompact() {
return toCompact;
}
- /** A {@link CompactTask} impl for append-only table. */
- public static class AppendOnlyCompactTask extends CompactTask {
+ /** A {@link CompactTask} impl for ALTER TABLE COMPACT of append-only table. */
+ public static class RollingCompactTask extends CompactTask {
+ private final long targetFileSize;
+ private final int minFileNum;
+ private final int maxFileNum;
private final CompactRewriter rewriter;
- public AppendOnlyCompactTask(List<DataFileMeta> toCompact, CompactRewriter rewriter) {
- super(toCompact);
+ public RollingCompactTask(
+ List<DataFileMeta> inputs,
+ long targetFileSize,
+ int minFileNum,
+ int maxFileNum,
+ CompactRewriter rewriter) {
+ super(inputs);
+ this.targetFileSize = targetFileSize;
+ this.minFileNum = minFileNum;
+ this.maxFileNum = maxFileNum;
this.rewriter = rewriter;
}
@Override
protected CompactResult doCompact(List<DataFileMeta> inputs) throws Exception {
- List<DataFileMeta> compactAfter = rewriter.rewrite(inputs);
- return new CompactResult() {
- @Override
- public List<DataFileMeta> before() {
- return inputs;
+ LinkedList<DataFileMeta> toCompact = new LinkedList<>(inputs);
+ Set<DataFileMeta> compactBefore = new HashSet<>();
+ List<DataFileMeta> compactAfter = new ArrayList<>();
+ while (!toCompact.isEmpty()) {
+ Optional<List<DataFileMeta>> candidates =
+ AppendOnlyCompactManager.pick(
+ toCompact, targetFileSize, minFileNum, maxFileNum);
+ if (candidates.isPresent()) {
+ List<DataFileMeta> before = candidates.get();
+ compactBefore.addAll(before);
+ List<DataFileMeta> after = rewriter.rewrite(before);
+ compactAfter.addAll(after);
+ DataFileMeta lastFile = after.get(after.size() - 1);
+ if (lastFile.fileSize() < targetFileSize) {
+ toCompact.offerFirst(lastFile);
+ }
+ } else {
+ break;
}
-
- @Override
- public List<DataFileMeta> after() {
- return compactAfter;
+ }
+ // remove intermediate files
+ Iterator<DataFileMeta> iterator = compactAfter.iterator();
Review Comment:
afterIterator?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org