This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 36c2756c6 [core] Remove EMIT_PER_BATCH in
AppendBypassCoordinateOperator to avoid slow compaction
36c2756c6 is described below
commit 36c2756c66800059ee2fa9a48895606bee511fea
Author: Jingsong <[email protected]>
AuthorDate: Wed Sep 25 13:31:04 2024 +0800
[core] Remove EMIT_PER_BATCH in AppendBypassCoordinateOperator to avoid
slow compaction
---
.../source/AppendBypassCoordinateOperator.java | 21 ++++++---------------
.../AppendBypassCoordinateOperatorFactory.java | 9 +++------
2 files changed, 9 insertions(+), 21 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperator.java
index 2d32bb885..668aa24c1 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperator.java
@@ -23,14 +23,12 @@ import
org.apache.paimon.append.UnawareAppendTableCompactionCoordinator;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.ExecutorUtils;
-import org.apache.flink.api.common.operators.MailboxExecutor;
import
org.apache.flink.api.common.operators.ProcessingTimeService.ProcessingTimeCallback;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
-import org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl;
import org.apache.flink.types.Either;
import java.util.List;
@@ -52,21 +50,16 @@ public class AppendBypassCoordinateOperator<CommitT>
ProcessingTimeCallback {
private static final long MAX_PENDING_TASKS = 5000;
- private static final long EMIT_PER_BATCH = 100;
private final FileStoreTable table;
- private final MailboxExecutorImpl mailbox;
private transient ScheduledExecutorService executorService;
private transient LinkedBlockingQueue<UnawareAppendCompactionTask>
compactTasks;
public AppendBypassCoordinateOperator(
- FileStoreTable table,
- ProcessingTimeService processingTimeService,
- MailboxExecutor mailbox) {
+ FileStoreTable table, ProcessingTimeService processingTimeService)
{
this.table = table;
this.processingTimeService = processingTimeService;
- this.mailbox = (MailboxExecutorImpl) mailbox;
this.chainingStrategy = ChainingStrategy.HEAD;
}
@@ -100,14 +93,12 @@ public class AppendBypassCoordinateOperator<CommitT>
@Override
public void onProcessingTime(long time) {
- while (mailbox.isIdle()) {
- for (int i = 0; i < EMIT_PER_BATCH; i++) {
- UnawareAppendCompactionTask task = compactTasks.poll();
- if (task == null) {
- return;
- }
- output.collect(new StreamRecord<>(Either.Right(task)));
+ while (true) {
+ UnawareAppendCompactionTask task = compactTasks.poll();
+ if (task == null) {
+ return;
}
+ output.collect(new StreamRecord<>(Either.Right(task)));
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperatorFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperatorFactory.java
index 416865a98..7c53e01b4 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperatorFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperatorFactory.java
@@ -25,15 +25,13 @@ import
org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
-import org.apache.flink.streaming.api.operators.YieldingOperatorFactory;
import org.apache.flink.types.Either;
/** Factory of {@link AppendBypassCoordinateOperator}. */
public class AppendBypassCoordinateOperatorFactory<CommitT>
extends AbstractStreamOperatorFactory<Either<CommitT,
UnawareAppendCompactionTask>>
- implements YieldingOperatorFactory<Either<CommitT,
UnawareAppendCompactionTask>>,
- OneInputStreamOperatorFactory<
- CommitT, Either<CommitT, UnawareAppendCompactionTask>>
{
+ implements OneInputStreamOperatorFactory<
+ CommitT, Either<CommitT, UnawareAppendCompactionTask>> {
private final FileStoreTable table;
@@ -47,8 +45,7 @@ public class AppendBypassCoordinateOperatorFactory<CommitT>
StreamOperatorParameters<Either<CommitT,
UnawareAppendCompactionTask>>
parameters) {
AppendBypassCoordinateOperator<CommitT> operator =
- new AppendBypassCoordinateOperator<>(
- table, processingTimeService, getMailboxExecutor());
+ new AppendBypassCoordinateOperator<>(table,
processingTimeService);
operator.setup(
parameters.getContainingTask(),
parameters.getStreamConfig(),
