[flink] Introduce precommit compact for newly created files in unaware bucket tables

docs/layouts/shortcodes/generated/flink_connector_configuration.html

@@ -26,12 +26,6 @@
         </tr>
     </thead>
     <tbody>
-        <tr>
-            <td><h5>changelog.precommit-compact</h5></td>
-            <td style="word-wrap: break-word;">false</td>
-            <td>Boolean</td>
-            <td>If true, it will add a changelog compact coordinator and worker operator after the writer operator,in order to compact several changelog files from the same partition into large ones, which can decrease the number of small files. </td>
-        </tr>
         <tr>
             <td><h5>end-input.watermark</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
@@ -110,6 +104,12 @@
             <td>Duration</td>
             <td>You can specify time interval for partition, for example, daily partition is '1 d', hourly partition is '1 h'.</td>
         </tr>
+        <tr>
+            <td><h5>precommit-compact</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>If true, it will add a compact coordinator and worker operator after the writer operator,in order to compact several changelog files (for primary key tables) or newly created data files (for unaware bucket tables) from the same partition into large ones, which can decrease the number of small files. </td>
+        </tr>
         <tr>
             <td><h5>scan.infer-parallelism</h5></td>
             <td style="word-wrap: break-word;">true</td>

paimon-core/src/main/java/org/apache/paimon/append/UnawareBucketNewFilesCompactionCoordinator.java

@@ -0,0 +1,74 @@
+/*
+ * 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.append;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.utils.Pair;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** Buffer files from the same partition, until their total size reaches {@code targetFileSize}. */
+public class UnawareBucketNewFilesCompactionCoordinator {
+
+    private final long targetFileSize;
+    private final Map<BinaryRow, PartitionFiles> partitions;
+
+    public UnawareBucketNewFilesCompactionCoordinator(long targetFileSize) {
+        this.targetFileSize = targetFileSize;
+        this.partitions = new LinkedHashMap<>();
+    }
+
+    public Optional<Pair<BinaryRow, List<DataFileMeta>>> addFile(
+            BinaryRow partition, DataFileMeta file) {
+        PartitionFiles files =
+                partitions.computeIfAbsent(partition, ignore -> new PartitionFiles());
+        files.addFile(file);
+        if (files.totalSize >= targetFileSize) {
+            partitions.remove(partition);
+            return Optional.of(Pair.of(partition, files.files));
+        } else {
+            return Optional.empty();
+        }
+    }
+
+    public List<Pair<BinaryRow, List<DataFileMeta>>> emitAll() {
+        List<Pair<BinaryRow, List<DataFileMeta>>> result =
+                partitions.entrySet().stream()
+                        .map(e -> Pair.of(e.getKey(), e.getValue().files))
+                        .collect(Collectors.toList());
+        partitions.clear();
+        return result;
+    }
+
+    private static class PartitionFiles {
+        private final List<DataFileMeta> files = new ArrayList<>();
+        private long totalSize = 0;
+
+        private void addFile(DataFileMeta file) {
+            files.add(file);
+            totalSize += file.fileSize();
+        }
+    }
+}

paimon-core/src/main/java/org/apache/paimon/table/source/DataTableStreamScan.java

@@ -164,13 +164,20 @@ private Plan tryFirstPlan() {
             }
             isFullPhaseEnd =
                     boundedChecker.shouldEndInput(snapshotManager.snapshot(currentSnapshotId));
+            LOG.debug(
+                    "Starting snapshot is {}, next snapshot will be {}.",
+                    scannedResult.plan().snapshotId(),
+                    nextSnapshotId);
             return scannedResult.plan();
         } else if (result instanceof StartingScanner.NextSnapshot) {
             nextSnapshotId = ((StartingScanner.NextSnapshot) result).nextSnapshotId();
             isFullPhaseEnd =
                     snapshotManager.snapshotExists(nextSnapshotId - 1)
                             && boundedChecker.shouldEndInput(
                                     snapshotManager.snapshot(nextSnapshotId - 1));
+            LOG.debug("There is no starting snapshot. Next snapshot will be {}.", nextSnapshotId);
+        } else if (result instanceof StartingScanner.NoSnapshot) {
+            LOG.debug("There is no starting snapshot and currently there is no next snapshot.");
         }
         return SnapshotNotExistPlan.INSTANCE;
     }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java

@@ -414,13 +414,16 @@ public class FlinkConnectorOptions {
                     .withDescription(
                             "Optional endInput watermark used in case of batch mode or bounded stream.");
 
-    public static final ConfigOption<Boolean> CHANGELOG_PRECOMMIT_COMPACT =
-            key("changelog.precommit-compact")
+    public static final ConfigOption<Boolean> PRECOMMIT_COMPACT =
+            key("precommit-compact")
                     .booleanType()
                     .defaultValue(false)
+                    .withFallbackKeys("changelog.precommit-compact")
                     .withDescription(
-                            "If true, it will add a changelog compact coordinator and worker operator after the writer operator,"
-                                    + "in order to compact several changelog files from the same partition into large ones, "
+                            "If true, it will add a compact coordinator and worker operator after the writer operator,"
+                                    + "in order to compact several changelog files (for primary key tables) "
+                                    + "or newly created data files (for unaware bucket tables) "
+                                    + "from the same partition into large ones, "
                                     + "which can decrease the number of small files. ");
 
     public static final ConfigOption<String> SOURCE_OPERATOR_UID_SUFFIX =

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketNewFilesCompactionCoordinatorOperator.java

@@ -0,0 +1,179 @@
+/*
+ * 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;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.append.UnawareAppendCompactionTask;
+import org.apache.paimon.append.UnawareBucketNewFilesCompactionCoordinator;
+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.BucketMode;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+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.List;
+import java.util.Optional;
+
+/**
+ * Coordinator operator for compacting newly created files for unaware bucket tables.
+ *
+ * <p>{@link UnawareBucketNewFilesCompactionCoordinatorOperator} calculates the file size of newly
+ * created files contained in all buckets within each partition from {@link Committable} message
+ * emitted from writer operator. And emit {@link UnawareAppendCompactionTask} to {@link
+ * UnawareBucketNewFilesCompactionWorkerOperator}.
+ */
+public class UnawareBucketNewFilesCompactionCoordinatorOperator
+        extends AbstractStreamOperator<
+                Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>>
+        implements OneInputStreamOperator<
+                        Committable,
+                        Either<Committable, Tuple2<Long, UnawareAppendCompactionTask>>>,
+                BoundedOneInput {
+
+    private final long targetFileSize;
+    private final long compactionFileSize;
+
+    private transient UnawareBucketNewFilesCompactionCoordinator coordinator;
+    private transient long checkpointId;
+
+    public UnawareBucketNewFilesCompactionCoordinatorOperator(CoreOptions options) {
+        this.targetFileSize = options.targetFileSize(false);
+        this.compactionFileSize = options.compactionFileSize(false);
+    }
+
+    @Override
+    public void open() throws Exception {
+        super.open();
+
+        coordinator = new UnawareBucketNewFilesCompactionCoordinator(targetFileSize);
+        checkpointId = Long.MIN_VALUE;
+    }
+
+    @Override
+    public void processElement(StreamRecord<Committable> record) throws Exception {
+        Committable committable = record.getValue();
+        checkpointId = Math.max(checkpointId, committable.checkpointId());
+        if (committable.kind() != Committable.Kind.FILE) {
+            output.collect(new StreamRecord<>(Either.Left(committable)));
+            return;
+        }
+
+        CommitMessageImpl message = (CommitMessageImpl) committable.wrappedCommittable();
+        if (message.newFilesIncrement().newFiles().isEmpty()) {
+            output.collect(new StreamRecord<>(Either.Left(committable)));
+            return;
+        }
+
+        BinaryRow partition = message.partition();
+        List<DataFileMeta> skippedFiles = new ArrayList<>();
+        for (DataFileMeta meta : message.newFilesIncrement().newFiles()) {
+            if (meta.fileSize() >= compactionFileSize) {
+                skippedFiles.add(meta);
+                continue;
+            }
+
+            Optional<Pair<BinaryRow, List<DataFileMeta>>> optionalPair =
+                    coordinator.addFile(partition, meta);
+            if (optionalPair.isPresent()) {
+                Pair<BinaryRow, List<DataFileMeta>> p = optionalPair.get();
+                Preconditions.checkArgument(!p.getValue().isEmpty());
+                if (p.getValue().size() > 1) {
+                    output.collect(
+                            new StreamRecord<>(
+                                    Either.Right(
+                                            Tuple2.of(
+                                                    checkpointId,
+                                                    new UnawareAppendCompactionTask(
+                                                            p.getKey(), p.getValue())))));
+                } else {
+                    skippedFiles.add(p.getValue().get(0));
+                }
+            }
+        }
+
+        CommitMessageImpl newMessage =
+                new CommitMessageImpl(
+                        message.partition(),
+                        message.bucket(),
+                        new DataIncrement(
+                                skippedFiles,
+                                message.newFilesIncrement().deletedFiles(),
+                                message.newFilesIncrement().changelogFiles()),
+                        message.compactIncrement(),
+                        message.indexIncrement());
+        if (!newMessage.isEmpty()) {
+            Committable newCommittable =
+                    new Committable(committable.checkpointId(), Committable.Kind.FILE, newMessage);
+            output.collect(new StreamRecord<>(Either.Left(newCommittable)));
+        }
+    }
+
+    @Override
+    public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+        emitAll();
+    }
+
+    @Override
+    public void endInput() throws Exception {
+        emitAll();
+    }
+
+    private void emitAll() {
+        for (Pair<BinaryRow, List<DataFileMeta>> p : coordinator.emitAll()) {
+            Preconditions.checkArgument(!p.getValue().isEmpty());
+            if (p.getValue().size() > 1) {
+                output.collect(
+                        new StreamRecord<>(
+                                Either.Right(
+                                        Tuple2.of(
+                                                checkpointId,
+                                                new UnawareAppendCompactionTask(
+                                                        p.getKey(), p.getValue())))));
+            } else {
+                CommitMessageImpl message =
+                        new CommitMessageImpl(
+                                p.getKey(),
+                                BucketMode.UNAWARE_BUCKET,
+                                new DataIncrement(
+                                        Collections.singletonList(p.getValue().get(0)),
+                                        Collections.emptyList(),
+                                        Collections.emptyList()),
+                                CompactIncrement.emptyIncrement());
+                output.collect(
+                        new StreamRecord<>(
+                                Either.Left(
+                                        new Committable(
+                                                checkpointId, Committable.Kind.FILE, message))));
+            }
+        }
+    }
+}