[Enhancement] Significantly improve DUP table load performance by avoiding unnecessary Tablet-ID Shuffle

[zghong]

Search before asking

• I had searched in the issues 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:

// 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:

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:

all historical partitions × current default bucket number

instead of:

tablets actually targeted by the current INSERT

This can significantly overestimate the target tablet count for tables with many historical partitions.

For example:

Historical partitions:        100
Default buckets:               32
Partitions targeted by INSERT: 1
Actual target tablets:         32

The current code estimates:

100 × 32 = 3200 tablets

Therefore, it enables Tablet-ID Shuffle because:

3200 >= min_tablets_for_dup_table_shuffle

However, the current INSERT actually targets only:

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:

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:

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:

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:

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:

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:

PhysicalProperties.ANY

rather than forcing Tablet-ID Shuffle based on unrelated historical partitions.

The existing configuration and threshold comparison should remain unchanged:

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?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct