[Bug] Batch write report error when the second time
- 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:
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