[compact]support compact unaware bucket for combine_mode #2858

wg1026688210 · 2024-02-05T13:30:47Z

Refact the compaction of combine mode for supporting compacting unaware bucket table

Purpose

Linked issue: close #2670

Tests

CompactDatabaseActionITCase.java for IT case in stream and batch mode
CompactionTaskSerializerTest.java for the test of serializing compaction task
AppendOnlyMultiTableCompactionWorkerOperatorTest.java for the test of AppendOnlyMultiTableCompactionWorkerOperator

API and Format

Documentation

leaves12138 · 2024-05-07T02:45:50Z

...mmon/src/main/java/org/apache/paimon/flink/sink/CombinedUmawareCompactionWorkerOperator.java

+    }
+
+    @VisibleForTesting
+    Iterable<Future<CommitMessage>> result() {


Is there any place using this method?

Done ,this has been removed.

this has been finished at #3175

leaves12138 · 2024-05-07T02:48:15Z

...ommon/src/main/java/org/apache/paimon/flink/sink/DividedUnawareCompactionWorkerOperator.java


    private final FileStoreTable table;
    private final String commitUser;

-    private transient AppendOnlyFileStoreWrite write;
+    private UnwareBucketCompactionHelper compactionHelper;


done,this has been finished at #3175

leaves12138 · 2024-05-07T03:01:52Z

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

+    }
+
+    public AppendOnlyCompactionTask(
+            BinaryRow partition, List<DataFileMeta> files, Identifier identifier) {


I think a new class MultiTableAppendOnlyCompactionTask which extends AppendOnlyCompactionTask seems better, in this way, we don't need Identifier.EMPTY which may cause hidden bug.

done ,this change has been finished at #3174

leaves12138

You can remove MultiTablesCompactorSourceFunction if you have renamed it

leaves12138 · 2024-05-07T06:08:52Z

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

+ *           unaware bucket table.
+ *     </ol>
+ */
+public abstract class AbstractBucketScanLogic<T> {


Maybe another name? Like MultiTableScanBase

done ,this has been finished at #3179

leaves12138 · 2024-05-07T06:09:38Z

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

+ * This class is responsible for implementing the scanning logic {@link AbstractBucketScanLogic} for
+ * the table with multi bucket such as dynamic or fixed bucket table.
+ */
+public class MultiBucketScanLogic extends AbstractBucketScanLogic<Tuple2<Split, String>> {


MultiAwareBucketTableScan?

done ,this has been finished at #3179

leaves12138 · 2024-05-07T06:10:32Z

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

+ * This class is responsible for implementing the scanning logic {@link AbstractBucketScanLogic} for
+ * the table with fix single bucket such as unaware bucket table.
+ */
+public class UnwareBucketScanLogic extends AbstractBucketScanLogic<AppendOnlyCompactionTask> {


MultiUnawareBuckeTableScan?

done ,this has been finished at #3179

leaves12138 · 2024-05-07T06:16:00Z

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

+ *     </ol>
+ */
+public abstract class CompactionFileScanner<T> {
+    protected final AtomicBoolean isRunning;


可以在 CompactionFileScanner 和 AtomicBoolean 之间加个换行

其他的类也一样

done ,this has been finished at #3179

leaves12138 · 2024-05-07T06:28:58Z

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

+    @Override
+    public void scan(SourceFunction.SourceContext<T> ctx) throws Exception {
+        while (isRunning.get()) {
+            Boolean isEmpty = tableScanLogic.collectFiles(ctx);


Kind of weird here, use 'null' to tag 'end', can we find a better way?

a simple Enum was added for a more clear expression at #3179

leaves12138 · 2024-05-15T10:25:16Z

.../src/main/java/org/apache/paimon/flink/source/operator/CombinedAwareBatchSourceFunction.java

-                    splits.forEach(ctx::collect);
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        tableScanLogic =


just tableScan?

leaves12138 · 2024-05-15T10:26:25Z

.../src/main/java/org/apache/paimon/flink/source/operator/CombinedAwareBatchSourceFunction.java

+/** It is responsible for monitoring compactor source of aware bucket table in batch mode. */
+public class CombinedAwareBatchSourceFunction
+        extends CombinedCompactorSourceFunction<Tuple2<Split, String>> {
+    private static final Logger LOGGER =


leaves12138 · 2024-05-15T10:26:46Z

.../main/java/org/apache/paimon/flink/source/operator/CombinedAwareStreamingSourceFunction.java

-                    splits.forEach(ctx::collect);
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        tableScanLogic =


just tableScan

leaves12138 · 2024-05-15T10:27:16Z

...rc/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareBatchSourceFunction.java

+ */
+public class CombinedUnawareBatchSourceFunction
+        extends CombinedCompactorSourceFunction<MultiTableAppendOnlyCompactionTask> {
+    private static final Logger LOGGER =


leaves12138 · 2024-05-15T10:27:33Z

...rc/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareBatchSourceFunction.java

+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        tableScanLogic =


leaves12138 · 2024-05-15T10:27:45Z

...ain/java/org/apache/paimon/flink/source/operator/CombinedUnawareStreamingSourceFunction.java

+ */
+public class CombinedUnawareStreamingSourceFunction
+        extends CombinedCompactorSourceFunction<MultiTableAppendOnlyCompactionTask> {
+    private final long monitorInterval;


leaves12138 · 2024-05-15T10:27:53Z

...ain/java/org/apache/paimon/flink/source/operator/CombinedUnawareStreamingSourceFunction.java

+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        tableScanLogic =


leaves12138 · 2024-05-15T10:28:19Z

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

+
+/** Compaction task for multi table . */
+public class MultiTableAppendOnlyCompactionTask extends AppendOnlyCompactionTask {
+    private final Identifier tableIdentifier;


leaves12138 · 2024-05-15T10:28:34Z

paimon-core/src/main/java/org/apache/paimon/table/sink/MultiTableCompactionTaskSerializer.java

+/** Serializer for {@link MultiTableAppendOnlyCompactionTask}. */
+public class MultiTableCompactionTaskSerializer
+        implements VersionedSerializer<MultiTableAppendOnlyCompactionTask> {
+    private static final int CURRENT_VERSION = 1;


leaves12138 · 2024-05-15T10:29:00Z

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

+ * table with multi bucket such as dynamic or fixed bucket table.
+ */
+public class MultiAwareBucketTableScan extends MultiTableScanBase<Tuple2<Split, String>> {
+    protected transient Map<Identifier, BucketsTable> tablesMap;


line feed (new line)

leaves12138 · 2024-05-15T10:29:09Z

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

+ *     </ol>
+ */
+public abstract class MultiTableScanBase<T> {
+    private static final Logger LOG = LoggerFactory.getLogger(MultiTableScanBase.class);


leaves12138 · 2024-05-15T10:29:23Z

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

+
+/** The Compactor of unaware bucket table to execute {@link AppendOnlyCompactionTask}. */
+public class UnawareBucketCompactor {
+    private final FileStoreTable table;


…of scan table.

leaves12138

+1

wg1026688210 force-pushed the compaction/support_unaware_mode branch 7 times, most recently from f6274ef to a43c51e Compare February 6, 2024 10:46

wg1026688210 changed the title ~~poc/support_compact_unaware_bucket_for_combine_mode~~ poc/support compact unaware bucket for combine_mode Feb 6, 2024

wg1026688210 force-pushed the compaction/support_unaware_mode branch 3 times, most recently from 65f0781 to cc99417 Compare February 6, 2024 14:00

wg1026688210 marked this pull request as ready for review February 25, 2024 14:55

wg1026688210 changed the title ~~poc/support compact unaware bucket for combine_mode~~ compact/support compact unaware bucket for combine_mode Feb 26, 2024

wg1026688210 changed the title ~~compact/support compact unaware bucket for combine_mode~~ [compact]/support compact unaware bucket for combine_mode Feb 27, 2024

wg1026688210 changed the title ~~[compact]/support compact unaware bucket for combine_mode~~ [compact]support compact unaware bucket for combine_mode Feb 27, 2024

yuzelin self-requested a review February 28, 2024 07:40

wg1026688210 force-pushed the compaction/support_unaware_mode branch from 1d80aac to c4ef352 Compare March 31, 2024 10:30

leaves12138 reviewed May 7, 2024

View reviewed changes

wg1026688210 force-pushed the compaction/support_unaware_mode branch from c4ef352 to 5366f01 Compare May 10, 2024 10:03

wg1026688210 requested a review from leaves12138 May 10, 2024 10:08

wg1026688210 force-pushed the compaction/support_unaware_mode branch 2 times, most recently from 39ef4b0 to ad6e873 Compare May 12, 2024 05:42

leaves12138 reviewed May 15, 2024

View reviewed changes

wg1026688210 added 9 commits May 15, 2024 21:08

support specify database and table for AppendOnlyCompactionTask

6e0dd45

add MultiTableAppendOnlyCompactionTask

d17dd29

refactor unaware bucket compaction work operator

1120779

add unaware bucket compaction worker

40bd9c5

check style

ec37857

refact the scan logic of combined mode compaction to reuse the logic …

50ceefb

…of scan table.

remove jbrain code

721b4a4

rebase

050be0e

improve

fafc6b0

wg1026688210 force-pushed the compaction/support_unaware_mode branch from 6ecddb2 to fafc6b0 Compare May 15, 2024 14:50

wg1026688210 requested a review from leaves12138 May 15, 2024 14:57

leaves12138 approved these changes May 16, 2024

View reviewed changes

leaves12138 merged commit 29d13fd into apache:master May 16, 2024
9 checks passed

sunxiaojian pushed a commit to sunxiaojian/paimon that referenced this pull request May 28, 2024

[compact] Support compact unaware bucket for combine_mode (apache#2858)

ccc34fe

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[compact]support compact unaware bucket for combine_mode #2858

[compact]support compact unaware bucket for combine_mode #2858

wg1026688210 commented Feb 5, 2024 •

edited

leaves12138 May 7, 2024

wg1026688210 May 9, 2024

wg1026688210 May 9, 2024

leaves12138 May 7, 2024

wg1026688210 May 9, 2024 •

edited

leaves12138 May 7, 2024

wg1026688210 May 9, 2024

leaves12138 left a comment

leaves12138 May 7, 2024

wg1026688210 May 10, 2024

leaves12138 May 7, 2024

wg1026688210 May 10, 2024

leaves12138 May 7, 2024

wg1026688210 May 10, 2024

leaves12138 May 7, 2024

leaves12138 May 7, 2024

wg1026688210 May 10, 2024

leaves12138 May 7, 2024

wg1026688210 May 10, 2024 •

edited

leaves12138 May 15, 2024

leaves12138 May 15, 2024

leaves12138 May 15, 2024

leaves12138 May 15, 2024

leaves12138 May 15, 2024

leaves12138 May 15, 2024

leaves12138 May 15, 2024

leaves12138 May 15, 2024

leaves12138 May 15, 2024

leaves12138 May 15, 2024

leaves12138 May 15, 2024

leaves12138 May 15, 2024

leaves12138 left a comment

[compact]support compact unaware bucket for combine_mode #2858

[compact]support compact unaware bucket for combine_mode #2858

Conversation

wg1026688210 commented Feb 5, 2024 • edited

Purpose

Tests

API and Format

Documentation

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wg1026688210 May 9, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leaves12138 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wg1026688210 May 10, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leaves12138 left a comment

Choose a reason for hiding this comment

wg1026688210 commented Feb 5, 2024 •

edited

wg1026688210 May 9, 2024 •

edited

wg1026688210 May 10, 2024 •

edited