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/13 09:24:01 UTC
[GitHub] [flink-table-store] LadyForest opened a new pull request, #214: [FLINK-28108] Support "ALTER TABLE ... COMPACT" for append-only table
LadyForest opened a new pull request, #214:
URL: https://github.com/apache/flink-table-store/pull/214
Compact tasks differ between auto-compact and "ALTER TABLE ... COMPACT"
- auto-compact task will pick the first qualified candidates to rewrite, and the AppendOnlyWriter must invoke the next compaction.
- "ALTER TABLE ... COMPACT" will perform a whole scan of the input, and within one task, there might be multiple times of rewrite happens.
--
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
[GitHub] [flink-table-store] JingsongLi merged pull request #214: [FLINK-28108] Support "ALTER TABLE ... COMPACT" for append-only table
Posted by GitBox <gi...@apache.org>.
JingsongLi merged PR #214:
URL: https://github.com/apache/flink-table-store/pull/214
--
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
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #214: [FLINK-28108] Support "ALTER TABLE ... COMPACT" for append-only table
Posted by GitBox <gi...@apache.org>.
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