You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "LsomeYeah (via GitHub)" <gi...@apache.org> on 2024/10/25 06:17:35 UTC
[PR] [flink] add coordinate and worker operator for small changelog files compaction [paimon]
LsomeYeah opened a new pull request, #4380:
URL: https://github.com/apache/paimon/pull/4380
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #xxx
<!-- What is the purpose of the change -->
Add a Coordinator node to small changelog files compaction pipeline to decide how to concatenate it into a target file size result file, which can be one or multiple files, and add a worker node to merge those small files.
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] add coordinate and worker operator for small changelog files compaction [paimon]
Posted by "tsreaper (via GitHub)" <gi...@apache.org>.
tsreaper commented on code in PR #4380:
URL: https://github.com/apache/paimon/pull/4380#discussion_r1821989618
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java:
##########
@@ -335,6 +335,199 @@ public void testBatchJobWithConflictAndRestart() throws Exception {
}
}
+ @Test
+ @Timeout(120)
+ public void testChangelogCompactInBatchWrite() throws Exception {
+ TableEnvironment bEnv = tableEnvironmentBuilder().batchMode().build();
+ String catalogDdl =
+ "CREATE CATALOG mycat WITH ( 'type' = 'paimon', 'warehouse' = '" + path + "' )";
+ bEnv.executeSql(catalogDdl);
+ bEnv.executeSql("USE CATALOG mycat");
+ bEnv.executeSql(
+ "CREATE TABLE t ( pt INT, k INT, v INT, PRIMARY KEY (pt, k) NOT ENFORCED ) "
+ + "PARTITIONED BY (pt) "
+ + "WITH ("
+ + " 'bucket' = '10',\n"
+ + " 'changelog-producer' = 'lookup',\n"
+ + " 'changelog.precommit-compact' = 'true',\n"
+ + " 'snapshot.num-retained.min' = '3',\n"
+ + " 'snapshot.num-retained.max' = '3'\n"
+ + ")");
+
+ TableEnvironment sEnv =
+ tableEnvironmentBuilder().streamingMode().checkpointIntervalMs(1000).build();
+ sEnv.executeSql(catalogDdl);
+ sEnv.executeSql("USE CATALOG mycat");
+
+ List<String> values = new ArrayList<>();
+ for (int i = 0; i < 1000; i++) {
+ values.add(String.format("(0, %d, %d)", i, i));
+ values.add(String.format("(1, %d, %d)", i, i));
+ }
+ bEnv.executeSql("INSERT INTO t VALUES " + String.join(", ", values)).await();
+
+ List<String> compactedChangelogs2 = listAllFilesWithPrefix("compacted-changelog-");
+ assertThat(compactedChangelogs2).hasSize(2);
+ assertThat(listAllFilesWithPrefix("changelog-")).isEmpty();
+
+ List<Row> expected = new ArrayList<>();
+ for (int i = 0; i < 1000; i++) {
+ expected.add(Row.ofKind(RowKind.INSERT, 0, i, i));
+ expected.add(Row.ofKind(RowKind.INSERT, 1, i, i));
+ }
+ assertStreamingResult(
+ sEnv.executeSql("SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '1') */"),
+ expected);
+
+ values.clear();
+ for (int i = 0; i < 1000; i++) {
+ values.add(String.format("(0, %d, %d)", i, i + 1));
+ values.add(String.format("(1, %d, %d)", i, i + 1));
+ }
+ bEnv.executeSql("INSERT INTO t VALUES " + String.join(", ", values)).await();
+
+ assertThat(listAllFilesWithPrefix("compacted-changelog-")).hasSize(4);
+ assertThat(listAllFilesWithPrefix("changelog-")).isEmpty();
+
+ for (int i = 0; i < 1000; i++) {
+ expected.add(Row.ofKind(RowKind.UPDATE_BEFORE, 0, i, i));
+ expected.add(Row.ofKind(RowKind.UPDATE_BEFORE, 1, i, i));
+ expected.add(Row.ofKind(RowKind.UPDATE_AFTER, 0, i, i + 1));
+ expected.add(Row.ofKind(RowKind.UPDATE_AFTER, 1, i, i + 1));
+ }
+ assertStreamingResult(
+ sEnv.executeSql("SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '1') */"),
+ expected);
+
+ values.clear();
+ for (int i = 0; i < 1000; i++) {
+ values.add(String.format("(0, %d, %d)", i, i + 2));
+ values.add(String.format("(1, %d, %d)", i, i + 2));
+ }
+ bEnv.executeSql("INSERT INTO t VALUES " + String.join(", ", values)).await();
+
+ assertThat(listAllFilesWithPrefix("compacted-changelog-")).hasSize(4);
+ assertThat(listAllFilesWithPrefix("changelog-")).isEmpty();
+ LocalFileIO fileIO = LocalFileIO.create();
+ for (String p : compactedChangelogs2) {
+ assertThat(fileIO.exists(new Path(p))).isFalse();
+ }
+
+ expected = expected.subList(2000, 6000);
+ for (int i = 0; i < 1000; i++) {
+ expected.add(Row.ofKind(RowKind.UPDATE_BEFORE, 0, i, i + 1));
+ expected.add(Row.ofKind(RowKind.UPDATE_BEFORE, 1, i, i + 1));
+ expected.add(Row.ofKind(RowKind.UPDATE_AFTER, 0, i, i + 2));
+ expected.add(Row.ofKind(RowKind.UPDATE_AFTER, 1, i, i + 2));
+ }
+ assertStreamingResult(
+ sEnv.executeSql("SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '1') */"),
+ expected);
+ }
+
+ @Test
+ @Timeout(120)
+ public void testChangelogCompactInStreamWrite() throws Exception {
+ TableEnvironment sEnv =
+ tableEnvironmentBuilder()
+ .streamingMode()
+ .checkpointIntervalMs(2000)
+ .parallelism(4)
+ .build();
+
+ sEnv.executeSql(createCatalogSql("testCatalog", path + "/warehouse"));
+ sEnv.executeSql("USE CATALOG testCatalog");
+ sEnv.executeSql(
+ "CREATE TABLE t ( pt INT, k INT, v INT, PRIMARY KEY (pt, k) NOT ENFORCED ) "
+ + "PARTITIONED BY (pt) "
+ + "WITH ("
+ + " 'bucket' = '10',\n"
+ + " 'changelog-producer' = 'lookup',\n"
+ + " 'changelog.precommit-compact' = 'true'\n"
+ + ")");
+
+ Path inputPath = new Path(path, "input");
+ LocalFileIO.create().mkdirs(inputPath);
+ sEnv.executeSql(
+ "CREATE TABLE `default_catalog`.`default_database`.`s` ( pt INT, k INT, v INT, PRIMARY KEY (pt, k) NOT ENFORCED) "
+ + "WITH ( 'connector' = 'filesystem', 'format' = 'testcsv', 'path' = '"
+ + inputPath
+ + "', 'source.monitor-interval' = '500ms' )");
+
+ sEnv.executeSql("INSERT INTO t SELECT * FROM `default_catalog`.`default_database`.`s`");
+ CloseableIterator<Row> it = sEnv.executeSql("SELECT * FROM t").collect();
+
+ // write initial data
+ List<String> values = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ values.add(String.format("(0, %d, %d)", i, i));
+ values.add(String.format("(1, %d, %d)", i, i));
+ }
+ sEnv.executeSql(
+ "INSERT INTO `default_catalog`.`default_database`.`s` VALUES "
+ + String.join(", ", values))
+ .await();
+
+ List<Row> expected = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ expected.add(Row.ofKind(RowKind.INSERT, 0, i, i));
+ expected.add(Row.ofKind(RowKind.INSERT, 1, i, i));
+ }
+ assertStreamingResult(it, expected);
+
+ List<String> compactedChangelogs2 = listAllFilesWithPrefix("compacted-changelog-");
+ assertThat(compactedChangelogs2).hasSize(2);
+ assertThat(listAllFilesWithPrefix("changelog-")).isEmpty();
+
+ // write update data
Review Comment:
nit: extra spaces
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTask.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.compact.changelog.format.CompactedChangelogReadOnlyFormat;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+
+/**
+ * {@link ChangelogCompactTask} to compact several changelog files from the same partition into one
+ * file, in order to reduce the number of small files.
+ */
+public class ChangelogCompactTask implements Serializable {
+ private final long checkpointId;
+ private final BinaryRow partition;
+ private final Map<Integer, List<DataFileMeta>> newFileChangelogFiles;
+ private final Map<Integer, List<DataFileMeta>> compactChangelogFiles;
+
+ private transient OutputStream outputStream;
+ private final transient List<Result> results = new ArrayList<>();
+
+ public ChangelogCompactTask(
+ long checkpointId,
+ BinaryRow partition,
+ Map<Integer, List<DataFileMeta>> newFileChangelogFiles,
+ Map<Integer, List<DataFileMeta>> compactChangelogFiles) {
+ this.checkpointId = checkpointId;
+ this.partition = partition;
+ this.newFileChangelogFiles = newFileChangelogFiles;
+ this.compactChangelogFiles = compactChangelogFiles;
+ }
+
+ public long checkpointId() {
+ return checkpointId;
+ }
+
+ public BinaryRow partition() {
+ return partition;
+ }
+
+ public Map<Integer, List<DataFileMeta>> newFileChangelogFiles() {
+ return newFileChangelogFiles;
+ }
+
+ public Map<Integer, List<DataFileMeta>> compactChangelogFiles() {
+ return compactChangelogFiles;
+ }
+
+ public List<Committable> doCompact(FileStoreTable table) throws Exception {
+ FileStorePathFactory pathFactory = table.store().pathFactory();
+
+ // copy all changelog files to a new big file
+ for (Map.Entry<Integer, List<DataFileMeta>> entry : newFileChangelogFiles.entrySet()) {
+ Integer bucket = entry.getKey();
+ DataFilePathFactory dataFilePathFactory =
+ pathFactory.createDataFilePathFactory(partition, bucket);
+ for (DataFileMeta meta : entry.getValue()) {
+ copyFile(table, dataFilePathFactory.toPath(meta.fileName()), bucket, false, meta);
+ }
+ }
+ for (Map.Entry<Integer, List<DataFileMeta>> entry : compactChangelogFiles.entrySet()) {
+ Integer bucket = entry.getKey();
+ DataFilePathFactory dataFilePathFactory =
+ pathFactory.createDataFilePathFactory(partition, bucket);
+ for (DataFileMeta meta : entry.getValue()) {
+ copyFile(table, dataFilePathFactory.toPath(meta.fileName()), bucket, true, meta);
+ }
+ }
+ outputStream.out.close();
+
+ return produceNewCommittables(table, pathFactory);
+ }
+
+ private void copyFile(
+ FileStoreTable table, Path path, int bucket, boolean isCompactResult, DataFileMeta meta)
+ throws Exception {
+ if (outputStream == null) {
Review Comment:
`copyFile` is only called in `doCompact`, so `outputStream` can be a local variable instead of a class member.
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java:
##########
@@ -490,13 +683,16 @@ private void testLookupChangelogProducerRandom(
enableFailure,
"'bucket' = '4',"
+ String.format(
- "'write-buffer-size' = '%s',"
- + "'changelog-producer' = 'lookup',"
- + "'lookup-wait' = '%s',"
- + "'deletion-vectors.enabled' = '%s'",
- random.nextBoolean() ? "4mb" : "8mb",
+ "'bucket' = '4', "
+ + "'writer-buffer-size' = '%s', "
+ + "'changelog-producer' = 'lookup', "
+ + "'lookup-wait' = '%s', "
+ + "'deletion-vectors.enabled' = '%s', "
+ + "'changelog.compact.parallelism' = '%s'",
Review Comment:
What is this table option? Also why do you change write buffer size?
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] add coordinate and worker operator for small changelog files compaction [paimon]
Posted by "wwj6591812 (via GitHub)" <gi...@apache.org>.
wwj6591812 commented on code in PR #4380:
URL: https://github.com/apache/paimon/pull/4380#discussion_r1816585623
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactCoordinateOperator.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinator operator for compacting changelog files.
+ *
+ * <p>{@link UnawareAppendTableCompactionCoordinator} calculates the file size of changelog files
+ * contained in all buckets within each partition from {@link Committable} message emitted from
+ * writer operator. And emit {@link ChangelogCompactTask} to {@link ChangelogCompactWorkerOperator}.
+ */
+public class ChangelogCompactCoordinateOperator extends AbstractStreamOperator<ChangelogCompactTask>
+ implements OneInputStreamOperator<Committable, ChangelogCompactTask>, BoundedOneInput {
+ private final FileStoreTable table;
+ private final OutputTag<Committable> committableOutputTag;
+
+ private transient long checkpointId;
+ private transient Map<BinaryRow, PartitionChangelog> partitionChangelogs;
+
+ public ChangelogCompactCoordinateOperator(
+ FileStoreTable table, OutputTag<Committable> committableOutputTag) {
+ this.table = table;
+ this.committableOutputTag = committableOutputTag;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+
+ checkpointId = Long.MIN_VALUE;
+ partitionChangelogs = new HashMap<>();
+ }
+
+ public void processElement(StreamRecord<Committable> record) {
+ Committable committable = record.getValue();
+ checkpointId = Math.max(checkpointId, committable.checkpointId());
+ if (committable.kind() != Committable.Kind.FILE) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ CommitMessageImpl message = (CommitMessageImpl) committable.wrappedCommittable();
+ if (message.newFilesIncrement().changelogFiles().isEmpty()
+ && message.compactIncrement().changelogFiles().isEmpty()) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ BinaryRow partition = message.partition();
+ Integer bucket = message.bucket();
+ for (DataFileMeta meta : message.newFilesIncrement().changelogFiles()) {
+ partitionChangelogs
+ .computeIfAbsent(partition, k -> new PartitionChangelog())
+ .addNewChangelogFile(bucket, meta);
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ if (partitionChangelog.totalFileSize >= table.coreOptions().targetFileSize(false)) {
+ emitPartitionChangelogCompactTask(partition);
+ }
+ }
+ for (DataFileMeta meta : message.compactIncrement().changelogFiles()) {
+ partitionChangelogs
+ .computeIfAbsent(partition, k -> new PartitionChangelog())
+ .addCompactChangelogFile(bucket, meta);
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ if (partitionChangelog.totalFileSize >= table.coreOptions().targetFileSize(false)) {
+ emitPartitionChangelogCompactTask(partition);
+ }
+ }
+
+ CommitMessageImpl newMessage =
+ new CommitMessageImpl(
+ message.partition(),
+ message.bucket(),
+ new DataIncrement(
+ message.newFilesIncrement().newFiles(),
+ message.newFilesIncrement().deletedFiles(),
+ Collections.emptyList()),
+ new CompactIncrement(
+ message.compactIncrement().compactBefore(),
+ message.compactIncrement().compactAfter(),
+ Collections.emptyList()),
+ message.indexIncrement());
+ Committable newCommittable =
+ new Committable(committable.checkpointId(), Committable.Kind.FILE, newMessage);
+ output.collect(committableOutputTag, new StreamRecord<>(newCommittable));
+ }
+
+ public void prepareSnapshotPreBarrier(long checkpointId) {
+ emitAllPartitionsChanglogCompactTask();
+ }
+
+ public void endInput() {
+ emitAllPartitionsChanglogCompactTask();
+ }
+
+ private void emitPartitionChangelogCompactTask(BinaryRow partition) {
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
Review Comment:
partitionChangelog may be null or not?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactCoordinateOperator.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinator operator for compacting changelog files.
+ *
+ * <p>{@link UnawareAppendTableCompactionCoordinator} calculates the file size of changelog files
+ * contained in all buckets within each partition from {@link Committable} message emitted from
+ * writer operator. And emit {@link ChangelogCompactTask} to {@link ChangelogCompactWorkerOperator}.
+ */
+public class ChangelogCompactCoordinateOperator extends AbstractStreamOperator<ChangelogCompactTask>
+ implements OneInputStreamOperator<Committable, ChangelogCompactTask>, BoundedOneInput {
+ private final FileStoreTable table;
+ private final OutputTag<Committable> committableOutputTag;
+
+ private transient long checkpointId;
+ private transient Map<BinaryRow, PartitionChangelog> partitionChangelogs;
+
+ public ChangelogCompactCoordinateOperator(
+ FileStoreTable table, OutputTag<Committable> committableOutputTag) {
+ this.table = table;
+ this.committableOutputTag = committableOutputTag;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+
+ checkpointId = Long.MIN_VALUE;
+ partitionChangelogs = new HashMap<>();
+ }
+
+ public void processElement(StreamRecord<Committable> record) {
+ Committable committable = record.getValue();
+ checkpointId = Math.max(checkpointId, committable.checkpointId());
+ if (committable.kind() != Committable.Kind.FILE) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ CommitMessageImpl message = (CommitMessageImpl) committable.wrappedCommittable();
+ if (message.newFilesIncrement().changelogFiles().isEmpty()
+ && message.compactIncrement().changelogFiles().isEmpty()) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ BinaryRow partition = message.partition();
+ Integer bucket = message.bucket();
+ for (DataFileMeta meta : message.newFilesIncrement().changelogFiles()) {
+ partitionChangelogs
+ .computeIfAbsent(partition, k -> new PartitionChangelog())
+ .addNewChangelogFile(bucket, meta);
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ if (partitionChangelog.totalFileSize >= table.coreOptions().targetFileSize(false)) {
+ emitPartitionChangelogCompactTask(partition);
+ }
+ }
+ for (DataFileMeta meta : message.compactIncrement().changelogFiles()) {
+ partitionChangelogs
+ .computeIfAbsent(partition, k -> new PartitionChangelog())
+ .addCompactChangelogFile(bucket, meta);
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ if (partitionChangelog.totalFileSize >= table.coreOptions().targetFileSize(false)) {
+ emitPartitionChangelogCompactTask(partition);
+ }
+ }
+
+ CommitMessageImpl newMessage =
+ new CommitMessageImpl(
+ message.partition(),
+ message.bucket(),
+ new DataIncrement(
+ message.newFilesIncrement().newFiles(),
+ message.newFilesIncrement().deletedFiles(),
+ Collections.emptyList()),
+ new CompactIncrement(
+ message.compactIncrement().compactBefore(),
+ message.compactIncrement().compactAfter(),
+ Collections.emptyList()),
+ message.indexIncrement());
+ Committable newCommittable =
+ new Committable(committable.checkpointId(), Committable.Kind.FILE, newMessage);
+ output.collect(committableOutputTag, new StreamRecord<>(newCommittable));
+ }
+
+ public void prepareSnapshotPreBarrier(long checkpointId) {
+ emitAllPartitionsChanglogCompactTask();
+ }
+
+ public void endInput() {
+ emitAllPartitionsChanglogCompactTask();
+ }
+
+ private void emitPartitionChangelogCompactTask(BinaryRow partition) {
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ output.collect(
+ new StreamRecord<>(
+ new ChangelogCompactTask(
+ checkpointId,
+ partition,
+ partitionChangelog.newFileChangelogFiles,
+ partitionChangelog.compactChangelogFiles)));
+ partitionChangelogs.remove(partition);
+ }
+
+ private void emitAllPartitionsChanglogCompactTask() {
+ List<BinaryRow> partitions = new ArrayList<>(partitionChangelogs.keySet());
+ for (BinaryRow partition : partitions) {
+ emitPartitionChangelogCompactTask(partition);
+ }
+ }
+
+ private static class PartitionChangelog {
+ private long totalFileSize;
+ private final Map<Integer, List<DataFileMeta>> newFileChangelogFiles;
+ private final Map<Integer, List<DataFileMeta>> compactChangelogFiles;
+
+ public long totalFileSize() {
Review Comment:
This mothod has not be called, delete it?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactCoordinateOperator.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinator operator for compacting changelog files.
+ *
+ * <p>{@link UnawareAppendTableCompactionCoordinator} calculates the file size of changelog files
Review Comment:
UnawareAppendTableCompactionCoordinator?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactOperator.java:
##########
@@ -0,0 +1,299 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.compact.changelog.format.CompactedChangelogReadOnlyFormat;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * Operator to compact several changelog files from the same partition into one file, in order to
+ * reduce the number of small files.
+ */
+public class ChangelogCompactOperator extends AbstractStreamOperator<Committable>
Review Comment:
Where this operator use?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTask.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.compact.changelog.format.CompactedChangelogReadOnlyFormat;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+
+/**
+ * {@link ChangelogCompactTask} to compact several changelog files from the same partition into one
+ * file, in order to reduce the number of small files.
+ */
+public class ChangelogCompactTask implements Serializable {
+ private final long checkpointId;
+ private final BinaryRow partition;
+ private final Map<Integer, List<DataFileMeta>> newFileChangelogFiles;
+ private final Map<Integer, List<DataFileMeta>> compactChangelogFiles;
+
+ private transient OutputStream outputStream;
+ private final transient List<Result> results = new ArrayList<>();
+
+ public ChangelogCompactTask(
+ long checkpointId,
+ BinaryRow partition,
+ Map<Integer, List<DataFileMeta>> newFileChangelogFiles,
+ Map<Integer, List<DataFileMeta>> compactChangelogFiles) {
+ this.checkpointId = checkpointId;
+ this.partition = partition;
+ this.newFileChangelogFiles = newFileChangelogFiles;
+ this.compactChangelogFiles = compactChangelogFiles;
+ }
+
+ public long checkpointId() {
+ return checkpointId;
+ }
+
+ public BinaryRow partition() {
+ return partition;
+ }
+
+ public Map<Integer, List<DataFileMeta>> newFileChangelogFiles() {
+ return newFileChangelogFiles;
+ }
+
+ public Map<Integer, List<DataFileMeta>> compactChangelogFiles() {
+ return compactChangelogFiles;
+ }
+
+ public List<Committable> doCompact(FileStoreTable table) throws Exception {
+ FileStorePathFactory pathFactory = table.store().pathFactory();
+
+ // copy all changelog files to a new big file
+ for (Map.Entry<Integer, List<DataFileMeta>> entry : newFileChangelogFiles.entrySet()) {
+ Integer bucket = entry.getKey();
Review Comment:
int
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogTaskTypeInfo.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.flink.sink.NoneCopyVersionedSerializerTypeSerializerProxy;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+/** Type information for {@link ChangelogCompactTask}. */
+public class ChangelogTaskTypeInfo extends TypeInformation<ChangelogCompactTask> {
Review Comment:
add a blank line
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactCoordinateOperator.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinator operator for compacting changelog files.
+ *
+ * <p>{@link UnawareAppendTableCompactionCoordinator} calculates the file size of changelog files
+ * contained in all buckets within each partition from {@link Committable} message emitted from
+ * writer operator. And emit {@link ChangelogCompactTask} to {@link ChangelogCompactWorkerOperator}.
+ */
+public class ChangelogCompactCoordinateOperator extends AbstractStreamOperator<ChangelogCompactTask>
+ implements OneInputStreamOperator<Committable, ChangelogCompactTask>, BoundedOneInput {
+ private final FileStoreTable table;
+ private final OutputTag<Committable> committableOutputTag;
+
+ private transient long checkpointId;
+ private transient Map<BinaryRow, PartitionChangelog> partitionChangelogs;
+
+ public ChangelogCompactCoordinateOperator(
+ FileStoreTable table, OutputTag<Committable> committableOutputTag) {
+ this.table = table;
+ this.committableOutputTag = committableOutputTag;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+
+ checkpointId = Long.MIN_VALUE;
+ partitionChangelogs = new HashMap<>();
+ }
+
+ public void processElement(StreamRecord<Committable> record) {
+ Committable committable = record.getValue();
+ checkpointId = Math.max(checkpointId, committable.checkpointId());
+ if (committable.kind() != Committable.Kind.FILE) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ CommitMessageImpl message = (CommitMessageImpl) committable.wrappedCommittable();
+ if (message.newFilesIncrement().changelogFiles().isEmpty()
+ && message.compactIncrement().changelogFiles().isEmpty()) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ BinaryRow partition = message.partition();
+ Integer bucket = message.bucket();
+ for (DataFileMeta meta : message.newFilesIncrement().changelogFiles()) {
+ partitionChangelogs
+ .computeIfAbsent(partition, k -> new PartitionChangelog())
+ .addNewChangelogFile(bucket, meta);
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ if (partitionChangelog.totalFileSize >= table.coreOptions().targetFileSize(false)) {
+ emitPartitionChangelogCompactTask(partition);
+ }
+ }
+ for (DataFileMeta meta : message.compactIncrement().changelogFiles()) {
+ partitionChangelogs
+ .computeIfAbsent(partition, k -> new PartitionChangelog())
+ .addCompactChangelogFile(bucket, meta);
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ if (partitionChangelog.totalFileSize >= table.coreOptions().targetFileSize(false)) {
+ emitPartitionChangelogCompactTask(partition);
+ }
+ }
+
+ CommitMessageImpl newMessage =
+ new CommitMessageImpl(
+ message.partition(),
+ message.bucket(),
+ new DataIncrement(
+ message.newFilesIncrement().newFiles(),
+ message.newFilesIncrement().deletedFiles(),
+ Collections.emptyList()),
+ new CompactIncrement(
+ message.compactIncrement().compactBefore(),
+ message.compactIncrement().compactAfter(),
+ Collections.emptyList()),
+ message.indexIncrement());
+ Committable newCommittable =
+ new Committable(committable.checkpointId(), Committable.Kind.FILE, newMessage);
+ output.collect(committableOutputTag, new StreamRecord<>(newCommittable));
+ }
+
+ public void prepareSnapshotPreBarrier(long checkpointId) {
+ emitAllPartitionsChanglogCompactTask();
+ }
+
+ public void endInput() {
+ emitAllPartitionsChanglogCompactTask();
+ }
+
+ private void emitPartitionChangelogCompactTask(BinaryRow partition) {
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ output.collect(
+ new StreamRecord<>(
+ new ChangelogCompactTask(
+ checkpointId,
+ partition,
+ partitionChangelog.newFileChangelogFiles,
+ partitionChangelog.compactChangelogFiles)));
+ partitionChangelogs.remove(partition);
+ }
+
+ private void emitAllPartitionsChanglogCompactTask() {
Review Comment:
partitionChangelogs.keySet().forEach(this::emitPartitionChangelogCompactTask);
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactCoordinateOperator.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinator operator for compacting changelog files.
+ *
+ * <p>{@link UnawareAppendTableCompactionCoordinator} calculates the file size of changelog files
+ * contained in all buckets within each partition from {@link Committable} message emitted from
+ * writer operator. And emit {@link ChangelogCompactTask} to {@link ChangelogCompactWorkerOperator}.
+ */
+public class ChangelogCompactCoordinateOperator extends AbstractStreamOperator<ChangelogCompactTask>
+ implements OneInputStreamOperator<Committable, ChangelogCompactTask>, BoundedOneInput {
+ private final FileStoreTable table;
+ private final OutputTag<Committable> committableOutputTag;
+
+ private transient long checkpointId;
+ private transient Map<BinaryRow, PartitionChangelog> partitionChangelogs;
+
+ public ChangelogCompactCoordinateOperator(
+ FileStoreTable table, OutputTag<Committable> committableOutputTag) {
+ this.table = table;
+ this.committableOutputTag = committableOutputTag;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+
+ checkpointId = Long.MIN_VALUE;
+ partitionChangelogs = new HashMap<>();
+ }
+
+ public void processElement(StreamRecord<Committable> record) {
+ Committable committable = record.getValue();
+ checkpointId = Math.max(checkpointId, committable.checkpointId());
+ if (committable.kind() != Committable.Kind.FILE) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ CommitMessageImpl message = (CommitMessageImpl) committable.wrappedCommittable();
+ if (message.newFilesIncrement().changelogFiles().isEmpty()
+ && message.compactIncrement().changelogFiles().isEmpty()) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
Review Comment:
long targetFileSize = table.coreOptions().targetFileSize(false);
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTask.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.compact.changelog.format.CompactedChangelogReadOnlyFormat;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+
+/**
+ * {@link ChangelogCompactTask} to compact several changelog files from the same partition into one
+ * file, in order to reduce the number of small files.
+ */
+public class ChangelogCompactTask implements Serializable {
+ private final long checkpointId;
+ private final BinaryRow partition;
+ private final Map<Integer, List<DataFileMeta>> newFileChangelogFiles;
+ private final Map<Integer, List<DataFileMeta>> compactChangelogFiles;
+
+ private transient OutputStream outputStream;
+ private final transient List<Result> results = new ArrayList<>();
+
+ public ChangelogCompactTask(
+ long checkpointId,
+ BinaryRow partition,
+ Map<Integer, List<DataFileMeta>> newFileChangelogFiles,
+ Map<Integer, List<DataFileMeta>> compactChangelogFiles) {
+ this.checkpointId = checkpointId;
+ this.partition = partition;
+ this.newFileChangelogFiles = newFileChangelogFiles;
+ this.compactChangelogFiles = compactChangelogFiles;
+ }
+
+ public long checkpointId() {
+ return checkpointId;
+ }
+
+ public BinaryRow partition() {
+ return partition;
+ }
+
+ public Map<Integer, List<DataFileMeta>> newFileChangelogFiles() {
+ return newFileChangelogFiles;
+ }
+
+ public Map<Integer, List<DataFileMeta>> compactChangelogFiles() {
+ return compactChangelogFiles;
+ }
+
+ public List<Committable> doCompact(FileStoreTable table) throws Exception {
+ FileStorePathFactory pathFactory = table.store().pathFactory();
+
+ // copy all changelog files to a new big file
Review Comment:
Two for statement has lots of some code, you can avoid this.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactCoordinateOperator.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinator operator for compacting changelog files.
+ *
+ * <p>{@link UnawareAppendTableCompactionCoordinator} calculates the file size of changelog files
+ * contained in all buckets within each partition from {@link Committable} message emitted from
+ * writer operator. And emit {@link ChangelogCompactTask} to {@link ChangelogCompactWorkerOperator}.
+ */
+public class ChangelogCompactCoordinateOperator extends AbstractStreamOperator<ChangelogCompactTask>
+ implements OneInputStreamOperator<Committable, ChangelogCompactTask>, BoundedOneInput {
+ private final FileStoreTable table;
+ private final OutputTag<Committable> committableOutputTag;
+
+ private transient long checkpointId;
+ private transient Map<BinaryRow, PartitionChangelog> partitionChangelogs;
+
+ public ChangelogCompactCoordinateOperator(
+ FileStoreTable table, OutputTag<Committable> committableOutputTag) {
+ this.table = table;
+ this.committableOutputTag = committableOutputTag;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+
+ checkpointId = Long.MIN_VALUE;
+ partitionChangelogs = new HashMap<>();
+ }
+
+ public void processElement(StreamRecord<Committable> record) {
+ Committable committable = record.getValue();
+ checkpointId = Math.max(checkpointId, committable.checkpointId());
+ if (committable.kind() != Committable.Kind.FILE) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ CommitMessageImpl message = (CommitMessageImpl) committable.wrappedCommittable();
+ if (message.newFilesIncrement().changelogFiles().isEmpty()
+ && message.compactIncrement().changelogFiles().isEmpty()) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ BinaryRow partition = message.partition();
+ Integer bucket = message.bucket();
+ for (DataFileMeta meta : message.newFilesIncrement().changelogFiles()) {
+ partitionChangelogs
+ .computeIfAbsent(partition, k -> new PartitionChangelog())
+ .addNewChangelogFile(bucket, meta);
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ if (partitionChangelog.totalFileSize >= table.coreOptions().targetFileSize(false)) {
+ emitPartitionChangelogCompactTask(partition);
+ }
+ }
+ for (DataFileMeta meta : message.compactIncrement().changelogFiles()) {
+ partitionChangelogs
+ .computeIfAbsent(partition, k -> new PartitionChangelog())
+ .addCompactChangelogFile(bucket, meta);
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ if (partitionChangelog.totalFileSize >= table.coreOptions().targetFileSize(false)) {
+ emitPartitionChangelogCompactTask(partition);
+ }
+ }
+
+ CommitMessageImpl newMessage =
+ new CommitMessageImpl(
+ message.partition(),
+ message.bucket(),
+ new DataIncrement(
+ message.newFilesIncrement().newFiles(),
+ message.newFilesIncrement().deletedFiles(),
+ Collections.emptyList()),
+ new CompactIncrement(
+ message.compactIncrement().compactBefore(),
+ message.compactIncrement().compactAfter(),
+ Collections.emptyList()),
+ message.indexIncrement());
+ Committable newCommittable =
+ new Committable(committable.checkpointId(), Committable.Kind.FILE, newMessage);
+ output.collect(committableOutputTag, new StreamRecord<>(newCommittable));
+ }
+
+ public void prepareSnapshotPreBarrier(long checkpointId) {
+ emitAllPartitionsChanglogCompactTask();
+ }
+
+ public void endInput() {
+ emitAllPartitionsChanglogCompactTask();
+ }
+
+ private void emitPartitionChangelogCompactTask(BinaryRow partition) {
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ output.collect(
+ new StreamRecord<>(
+ new ChangelogCompactTask(
+ checkpointId,
+ partition,
+ partitionChangelog.newFileChangelogFiles,
+ partitionChangelog.compactChangelogFiles)));
+ partitionChangelogs.remove(partition);
+ }
+
+ private void emitAllPartitionsChanglogCompactTask() {
+ List<BinaryRow> partitions = new ArrayList<>(partitionChangelogs.keySet());
+ for (BinaryRow partition : partitions) {
+ emitPartitionChangelogCompactTask(partition);
+ }
+ }
+
+ private static class PartitionChangelog {
+ private long totalFileSize;
+ private final Map<Integer, List<DataFileMeta>> newFileChangelogFiles;
Review Comment:
newChangelogFiles
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTask.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.compact.changelog.format.CompactedChangelogReadOnlyFormat;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+
+/**
+ * {@link ChangelogCompactTask} to compact several changelog files from the same partition into one
+ * file, in order to reduce the number of small files.
+ */
+public class ChangelogCompactTask implements Serializable {
+ private final long checkpointId;
+ private final BinaryRow partition;
+ private final Map<Integer, List<DataFileMeta>> newFileChangelogFiles;
Review Comment:
newChangelogFiles
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactOperator.java:
##########
@@ -0,0 +1,299 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.compact.changelog.format.CompactedChangelogReadOnlyFormat;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * Operator to compact several changelog files from the same partition into one file, in order to
+ * reduce the number of small files.
+ */
+public class ChangelogCompactOperator extends AbstractStreamOperator<Committable>
+ implements OneInputStreamOperator<Committable, Committable>, BoundedOneInput {
+
+ private final FileStoreTable table;
+
+ private transient FileStorePathFactory pathFactory;
+ private transient long checkpointId;
+ private transient Map<BinaryRow, OutputStream> outputStreams;
+ private transient Map<BinaryRow, List<Result>> results;
+
+ public ChangelogCompactOperator(FileStoreTable table) {
Review Comment:
where call here?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactCoordinateOperator.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinator operator for compacting changelog files.
+ *
+ * <p>{@link UnawareAppendTableCompactionCoordinator} calculates the file size of changelog files
+ * contained in all buckets within each partition from {@link Committable} message emitted from
+ * writer operator. And emit {@link ChangelogCompactTask} to {@link ChangelogCompactWorkerOperator}.
+ */
+public class ChangelogCompactCoordinateOperator extends AbstractStreamOperator<ChangelogCompactTask>
+ implements OneInputStreamOperator<Committable, ChangelogCompactTask>, BoundedOneInput {
+ private final FileStoreTable table;
+ private final OutputTag<Committable> committableOutputTag;
+
+ private transient long checkpointId;
+ private transient Map<BinaryRow, PartitionChangelog> partitionChangelogs;
+
+ public ChangelogCompactCoordinateOperator(
+ FileStoreTable table, OutputTag<Committable> committableOutputTag) {
+ this.table = table;
+ this.committableOutputTag = committableOutputTag;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+
+ checkpointId = Long.MIN_VALUE;
+ partitionChangelogs = new HashMap<>();
+ }
+
+ public void processElement(StreamRecord<Committable> record) {
+ Committable committable = record.getValue();
+ checkpointId = Math.max(checkpointId, committable.checkpointId());
+ if (committable.kind() != Committable.Kind.FILE) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ CommitMessageImpl message = (CommitMessageImpl) committable.wrappedCommittable();
+ if (message.newFilesIncrement().changelogFiles().isEmpty()
+ && message.compactIncrement().changelogFiles().isEmpty()) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ BinaryRow partition = message.partition();
+ Integer bucket = message.bucket();
+ for (DataFileMeta meta : message.newFilesIncrement().changelogFiles()) {
+ partitionChangelogs
+ .computeIfAbsent(partition, k -> new PartitionChangelog())
+ .addNewChangelogFile(bucket, meta);
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ if (partitionChangelog.totalFileSize >= table.coreOptions().targetFileSize(false)) {
+ emitPartitionChangelogCompactTask(partition);
+ }
+ }
+ for (DataFileMeta meta : message.compactIncrement().changelogFiles()) {
+ partitionChangelogs
+ .computeIfAbsent(partition, k -> new PartitionChangelog())
+ .addCompactChangelogFile(bucket, meta);
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
+ if (partitionChangelog.totalFileSize >= table.coreOptions().targetFileSize(false)) {
+ emitPartitionChangelogCompactTask(partition);
+ }
+ }
+
+ CommitMessageImpl newMessage =
+ new CommitMessageImpl(
+ message.partition(),
+ message.bucket(),
+ new DataIncrement(
+ message.newFilesIncrement().newFiles(),
+ message.newFilesIncrement().deletedFiles(),
+ Collections.emptyList()),
+ new CompactIncrement(
+ message.compactIncrement().compactBefore(),
+ message.compactIncrement().compactAfter(),
+ Collections.emptyList()),
+ message.indexIncrement());
+ Committable newCommittable =
+ new Committable(committable.checkpointId(), Committable.Kind.FILE, newMessage);
+ output.collect(committableOutputTag, new StreamRecord<>(newCommittable));
+ }
+
+ public void prepareSnapshotPreBarrier(long checkpointId) {
+ emitAllPartitionsChanglogCompactTask();
+ }
+
+ public void endInput() {
+ emitAllPartitionsChanglogCompactTask();
+ }
+
+ private void emitPartitionChangelogCompactTask(BinaryRow partition) {
+ PartitionChangelog partitionChangelog = partitionChangelogs.get(partition);
Review Comment:
partitionChangelog may be null or not?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTask.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.compact.changelog.format.CompactedChangelogReadOnlyFormat;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+
+/**
+ * {@link ChangelogCompactTask} to compact several changelog files from the same partition into one
+ * file, in order to reduce the number of small files.
+ */
+public class ChangelogCompactTask implements Serializable {
+ private final long checkpointId;
+ private final BinaryRow partition;
+ private final Map<Integer, List<DataFileMeta>> newFileChangelogFiles;
+ private final Map<Integer, List<DataFileMeta>> compactChangelogFiles;
+
+ private transient OutputStream outputStream;
+ private final transient List<Result> results = new ArrayList<>();
+
+ public ChangelogCompactTask(
+ long checkpointId,
+ BinaryRow partition,
+ Map<Integer, List<DataFileMeta>> newFileChangelogFiles,
+ Map<Integer, List<DataFileMeta>> compactChangelogFiles) {
+ this.checkpointId = checkpointId;
+ this.partition = partition;
+ this.newFileChangelogFiles = newFileChangelogFiles;
+ this.compactChangelogFiles = compactChangelogFiles;
+ }
+
+ public long checkpointId() {
+ return checkpointId;
+ }
+
+ public BinaryRow partition() {
+ return partition;
+ }
+
+ public Map<Integer, List<DataFileMeta>> newFileChangelogFiles() {
+ return newFileChangelogFiles;
+ }
+
+ public Map<Integer, List<DataFileMeta>> compactChangelogFiles() {
+ return compactChangelogFiles;
+ }
+
+ public List<Committable> doCompact(FileStoreTable table) throws Exception {
+ FileStorePathFactory pathFactory = table.store().pathFactory();
+
+ // copy all changelog files to a new big file
+ for (Map.Entry<Integer, List<DataFileMeta>> entry : newFileChangelogFiles.entrySet()) {
+ Integer bucket = entry.getKey();
+ DataFilePathFactory dataFilePathFactory =
+ pathFactory.createDataFilePathFactory(partition, bucket);
+ for (DataFileMeta meta : entry.getValue()) {
+ copyFile(table, dataFilePathFactory.toPath(meta.fileName()), bucket, false, meta);
+ }
+ }
+ for (Map.Entry<Integer, List<DataFileMeta>> entry : compactChangelogFiles.entrySet()) {
+ Integer bucket = entry.getKey();
+ DataFilePathFactory dataFilePathFactory =
+ pathFactory.createDataFilePathFactory(partition, bucket);
+ for (DataFileMeta meta : entry.getValue()) {
+ copyFile(table, dataFilePathFactory.toPath(meta.fileName()), bucket, true, meta);
+ }
+ }
+ outputStream.out.close();
+
+ return produceNewCommittables(table, pathFactory);
+ }
+
+ private void copyFile(
+ FileStoreTable table, Path path, int bucket, boolean isCompactResult, DataFileMeta meta)
+ throws Exception {
+ if (outputStream == null) {
+ Path outputPath =
+ new Path(path.getParent(), "tmp-compacted-changelog-" + UUID.randomUUID());
+ outputStream =
+ new OutputStream(outputPath, table.fileIO().newOutputStream(outputPath, false));
+ }
+ long offset = outputStream.out.getPos();
+ try (SeekableInputStream in = table.fileIO().newInputStream(path)) {
+ IOUtils.copyBytes(in, outputStream.out, IOUtils.BLOCKSIZE, false);
+ }
+ table.fileIO().deleteQuietly(path);
+ results.add(
+ new Result(
+ bucket, isCompactResult, meta, offset, outputStream.out.getPos() - offset));
+ }
+
+ private List<Committable> produceNewCommittables(
+ FileStoreTable table, FileStorePathFactory pathFactory) throws IOException {
+ Result baseResult = results.get(0);
+ Preconditions.checkArgument(baseResult.offset == 0);
+ DataFilePathFactory dataFilePathFactory =
+ pathFactory.createDataFilePathFactory(partition, baseResult.bucket);
+ // see Java docs of `CompactedChangelogFormatReaderFactory`
+ String realName =
+ "compacted-changelog-"
+ + UUID.randomUUID()
+ + "$"
+ + baseResult.bucket
+ + "-"
+ + baseResult.length;
+ table.fileIO()
+ .rename(
+ outputStream.path,
+ dataFilePathFactory.toPath(
+ realName
+ + "."
+ + CompactedChangelogReadOnlyFormat.getIdentifier(
+ baseResult.meta.fileFormat())));
+
+ List<Committable> newCommittables = new ArrayList<>();
Review Comment:
List<Committable> newCommittables = new ArrayList<>(bucketedResults.entrySet().size());
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactCoordinateOperator.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinator operator for compacting changelog files.
+ *
+ * <p>{@link UnawareAppendTableCompactionCoordinator} calculates the file size of changelog files
+ * contained in all buckets within each partition from {@link Committable} message emitted from
+ * writer operator. And emit {@link ChangelogCompactTask} to {@link ChangelogCompactWorkerOperator}.
+ */
+public class ChangelogCompactCoordinateOperator extends AbstractStreamOperator<ChangelogCompactTask>
+ implements OneInputStreamOperator<Committable, ChangelogCompactTask>, BoundedOneInput {
+ private final FileStoreTable table;
+ private final OutputTag<Committable> committableOutputTag;
+
+ private transient long checkpointId;
+ private transient Map<BinaryRow, PartitionChangelog> partitionChangelogs;
+
+ public ChangelogCompactCoordinateOperator(
+ FileStoreTable table, OutputTag<Committable> committableOutputTag) {
+ this.table = table;
+ this.committableOutputTag = committableOutputTag;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+
+ checkpointId = Long.MIN_VALUE;
+ partitionChangelogs = new HashMap<>();
+ }
+
+ public void processElement(StreamRecord<Committable> record) {
+ Committable committable = record.getValue();
+ checkpointId = Math.max(checkpointId, committable.checkpointId());
+ if (committable.kind() != Committable.Kind.FILE) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ CommitMessageImpl message = (CommitMessageImpl) committable.wrappedCommittable();
+ if (message.newFilesIncrement().changelogFiles().isEmpty()
+ && message.compactIncrement().changelogFiles().isEmpty()) {
+ output.collect(committableOutputTag, record);
+ return;
+ }
+
+ BinaryRow partition = message.partition();
+ Integer bucket = message.bucket();
+ for (DataFileMeta meta : message.newFilesIncrement().changelogFiles()) {
Review Comment:
Two for statement has lots of some code, you can avoid this.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] add coordinate and worker operator for small changelog files compaction [paimon]
Posted by "tsreaper (via GitHub)" <gi...@apache.org>.
tsreaper commented on code in PR #4380:
URL: https://github.com/apache/paimon/pull/4380#discussion_r1816084555
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactCoordinateOperator.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.types.Either;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinator operator for compacting changelog files.
+ *
+ * <p>{@link UnawareAppendTableCompactionCoordinator} calculates the file size of changelog files
+ * contained in all buckets within each partition from {@link Committable} message emitted from
+ * writer operator. And emit {@link ChangelogCompactTask} to {@link ChangelogCompactWorkerOperator}.
+ */
+public class ChangelogCompactCoordinateOperator
+ extends AbstractStreamOperator<Either<Committable, ChangelogCompactTask>>
Review Comment:
Use side output to send `Committable` directly to committers.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] add coordinate and worker operator for small changelog files compaction [paimon]
Posted by "tsreaper (via GitHub)" <gi...@apache.org>.
tsreaper merged PR #4380:
URL: https://github.com/apache/paimon/pull/4380
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] add coordinate and worker operator for small changelog files compaction [paimon]
Posted by "tsreaper (via GitHub)" <gi...@apache.org>.
tsreaper commented on code in PR #4380:
URL: https://github.com/apache/paimon/pull/4380#discussion_r1816084555
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactCoordinateOperator.java:
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.flink.compact.changelog;
+
+import org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.types.Either;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinator operator for compacting changelog files.
+ *
+ * <p>{@link UnawareAppendTableCompactionCoordinator} calculates the file size of changelog files
+ * contained in all buckets within each partition from {@link Committable} message emitted from
+ * writer operator. And emit {@link ChangelogCompactTask} to {@link ChangelogCompactWorkerOperator}.
+ */
+public class ChangelogCompactCoordinateOperator
+ extends AbstractStreamOperator<Either<Committable, ChangelogCompactTask>>
Review Comment:
Use side output to send `Committable` directly to committers.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org