apache / apache/paimon

[Bug] Batch write report error when the second time

Open
#1,217 2 comments 0 reactions 0 assignees View on GitHub
bug
Dominant language
Java
Stars
3.4k
Forks
1.4k
Avg merge
1d 11h
Merged PRs (30d)
396

Description

### Search before asking

- [X] I searched in the [issues](https://github.com/apache/incubator-paimon/issues) and found nothing similar.

### Paimon version

0.5-SNAPSHOT

### Compute Engine

Flink-1.17.0

### Minimal reproduce step

1. new a flink batch write task with the following parameters:

parallelism.default=200
execution.batch.adaptive.auto-parallelism.enabled=true ( the default value of Flink-17 is true)

2. The first time ,the task runs successfully。the operator execution graph is:
image

3. The second time,the task report error:This is a bug! The partition org.apache.paimon.data.BinaryRow@23546dee and bucket 0 is filtered!

4. when we change execution.batch.adaptive.auto-parallelism.enabled to false or remove parallelism.default parameter, the task runs successfully

### What doesn't meet your expectations?

Caused by: java.lang.IllegalArgumentException: This is a bug! The partition org.apache.paimon.data.BinaryRow@23546dee and bucket 0 is filtered!
at org.apache.paimon.utils.Preconditions.checkArgument(Preconditions.java:127)
at org.apache.paimon.operation.AbstractFileStoreScan.withPartitionBucket(AbstractFileStoreScan.java:156)
at org.apache.paimon.operation.AbstractFileStoreWrite.scanExistingFileMetas(AbstractFileStoreWrite.java:332)
at org.apache.paimon.operation.AbstractFileStoreWrite.createWriterContainer(AbstractFileStoreWrite.java:314)
at org.apache.paimon.operation.AbstractFileStoreWrite.lambda$getWriterWrapper$2(AbstractFileStoreWrite.java:289)
at java.util.HashMap.computeIfAbsent(HashMap.java:1127)
at org.apache.paimon.operation.AbstractFileStoreWrite.getWriterWrapper(AbstractFileStoreWrite.java:288)
at org.apache.paimon.operation.AbstractFileStoreWrite.write(AbstractFileStoreWrite.java:98)
at org.apache.paimon.table.sink.TableWriteImpl.writeAndReturn(TableWriteImpl.java:105)
at org.apache.paimon.flink.sink.StoreSinkWriteImpl.write(StoreSinkWriteImpl.java:83)
at org.apache.paimon.flink.sink.GlobalFullCompactionSinkWrite.write(GlobalFullCompactionSinkWrite.java:96)
at org.apache.paimon.flink.sink.RowDataStoreWriteOperator.processElement(RowDataStoreWriteOperator.java:174)
... 13 more

### Anything else?

The root cause is :
1) the channel num of BucketingStreamPartitioner is the value of parallelism.default,that is 200 。
2)While the the channel num of RowDataChannelComputer is setted up by RowDataStoreWriteOperator.initializeState(),the value is getRuntimeContext().getNumberOfParallelSubtasks(),that is 7
3)the different channel cause the calculated channel is different

### Are you willing to submit a PR?

- [X] I'm willing to submit a PR!

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with AbstractFileStoreScan.withPartitionBucket and AbstractFileStoreWrite.scanExistingFileMetas, then inspect RowDataStoreWriteOperator.initializeState and the channel setup described in the report. Reproduce with parallelism.default=200 and adaptive auto-parallelism enabled, comparing the first and second batch runs. Done means the second run no longer reports that the partition and bucket are filtered.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend, data-engineering
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.