zghong opened a new issue, #68156: URL: https://github.com/apache/doris/issues/68156
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/doris/issues?q=is%3Aissue) and found no similar issues. ### Description Tablet-ID Shuffle is used by `OlapTableSink` to route rows targeting the same tablet to a smaller and more stable set of sink instances. This can reduce writer and MemTable fan-out when an `INSERT` writes to many tablets. For DUP tables, Tablet-ID Shuffle is a performance optimization rather than a correctness requirement. Doris therefore skips the shuffle when the estimated number of target tablets is smaller than: ```java // default: 64 Config.min_tablets_for_dup_table_shuffle ``` However, the current implementation does not estimate the tablets actually targeted by the current `INSERT`. Instead, it uses the number of all historical partitions in the table multiplied by the current default bucket number: ```java if (targetTable.getKeysType() == KeysType.DUP_KEYS) { final long partitionNums = Math.max(targetTable.getPartitionInfo().getAllPartitions().size(), 1); final long tabletNums = partitionNums * distributionInfo.getBucketNum(); if (tabletNums < Config.min_tablets_for_dup_table_shuffle) { return PhysicalProperties.ANY; } } return PhysicalProperties.TABLET_ID_SHUFFLE; ``` In other words, the current estimation is: ```text all historical partitions × current default bucket number ``` instead of: ```text tablets actually targeted by the current INSERT ``` This can significantly overestimate the target tablet count for tables with many historical partitions. For example: ```text Historical partitions: 100 Default buckets: 32 Partitions targeted by INSERT: 1 Actual target tablets: 32 ``` The current code estimates: ```text 100 × 32 = 3200 tablets ``` Therefore, it enables Tablet-ID Shuffle because: ```text 3200 >= min_tablets_for_dup_table_shuffle ``` However, the current `INSERT` actually targets only: ```text 1 × 32 = 32 tablets ``` and should not be forced to use Tablet-ID Shuffle under the existing threshold policy. ### Problems in the current logic The current implementation has the following issues: 1. It includes unrelated historical partitions Partitions that cannot be written by the current `INSERT` are still included in the estimation. 2. It ignores explicitly specified target partitions Even for an `INSERT` such as: ```sql INSERT INTO target_table PARTITION(p1) SELECT ...; ``` the current implementation still counts all partitions in the table. 3. It assumes that all partitions use the current default bucket number After `MODIFY DISTRIBUTION`, existing partitions may have different bucket numbers from newly created partitions. Therefore: ```text partition count × current default bucket number ``` may not represent the actual tablet count of either the table or the current `INSERT`. 4. It cannot distinguish an unknown target from all historical partitions When the target partitions cannot be derived, the current implementation treats all historical partitions as if they were the target of the `INSERT`. This can force an expensive Exchange without reliable evidence that it is beneficial. 5. It does not derive target partitions from constant input For `LIST` or automatic partition tables, the target partition may be determinable from constant partition-column values or VALUES input, but the current implementation does not use this information. ### Production impact We observed this issue in production on a large DUP table using automatic `LIST` partitioning and hash distribution. The table had many historical partitions, while each `INSERT` only wrote to one new partition with 32 buckets. The current estimation incorrectly exceeded the 64-tablet threshold and generated a Tablet-ID Shuffle for the entire input. For representative failed queries, the execution statistics included: ```text Scanned rows: 91–108 million Shuffle rows: 90–107 million Shuffle bytes: 190–228 GB ``` The additional Exchange also reduced the effective sink parallelism to approximately 100 receivers globally, with about one local sink on each participating BE (we have 120 BEs in our cluster). For rows containing `Bitmap`, `Array`, and other wide columns, this caused: - a large amount of network transfer; - expensive serialization and deserialization; - Exchange backpressure; - data concentration on a small number of sink instances; - increased MemTable and flush memory pressure; - significantly longer import latency; - BE memory limit errors in some cases. As a comparison, inserting the same data into a table with only one 32-bucket partition did not generate Tablet-ID Shuffle: ```text Shuffle rows: 0 Shuffle bytes: 0 ``` ### Solution Determine the tablet count based on the partitions that may actually be targeted by the current `INSERT`, instead of using all historical partitions in the table. The estimation can use the following sources in priority order: 1. Explicit target partitions If the `INSERT` explicitly specifies partition IDs, calculate the tablet count by summing the actual bucket number of each target partition: ```text targetTabletNum = sum(actual bucket number of each explicitly targeted partition) ``` This also handles partitions with different bucket numbers. 2. Unpartitioned tables An unpartitioned table has one physical partition. Use the actual bucket number of that partition instead of multiplying the default bucket number by a derived partition count. 3. Statically derivable `LIST` partitions For constant rows, VALUES, or other inputs whose complete `LIST` partition keys can be safely folded to literals: - derive the target partition keys; - remove duplicate partition keys; - match existing `LIST`/default partitions; - use the actual bucket number for existing partitions; - for an automatic partition that has not yet been created, use the current default bucket number. 4. Reliable statistics When the target partitions cannot be derived statically but reliable partition-column statistics are available, estimate the number of target partitions using NDV and input row count, then convert it to an estimated tablet count. 5. Unknown target If the target partitions cannot be safely derived or reliably estimated, do not use the number of all historical partitions as a substitute. For DUP tables, return: ```java PhysicalProperties.ANY ``` rather than forcing Tablet-ID Shuffle based on unrelated historical partitions. The existing configuration and threshold comparison should remain unchanged: ```java Config.min_tablets_for_dup_table_shuffle ``` This issue only proposes correcting how the target tablet count of the current `INSERT` is determined. Whether the existing tablet threshold should later be replaced by a more complete cost model considering input bytes, row width, source parallelism, receiver parallelism, and writer/MemTable fan-out can be evaluated separately. ### Are you willing to submit PR? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
