[Bug] Data skew in Hash Bucketing with CRC32

[zghong]

Search before asking

• I had searched in the issues and found no similar issues.

Version

master / 4.0.5

What's Wrong?

We have a table with 484,862,245,170 rows, and the data is evenly divided according to shard_num.

CREATE TABLE user_label_double_v1
(
    `shard_num`   SMALLINT    NOT NULL
    `label_name`  VARCHAR(64) NOT NULL
    `label_value` DOUBLE      NOT NULL
    `uid`         INT         NOT NULL
    `dt`          Date        NOT NULL
)
DUPLICATE KEY(shard_num, label_name)
AUTO PARTITION BY RANGE (date_trunc(`dt`, 'day')) ()
DISTRIBUTED BY HASH(shard_num) BUCKETS 512
PROPERTIES (
    "replication_num" = "3",
    "compression" = "zstd"
);

Define DISTRIBUTED BY HASH(shard_num) BUCKETS 512 of the table in Doris, but we found there are some empty tablets after importing all data.

Doris root@6.94.183.233:datamill_test2> select shard_num, count() from user_label_double_v1 group by shard_num order by shard_num;
+-----------+-----------+
| shard_num |  count(*) |
+-----------+-----------+
|         0 | 947087198 |
|         1 | 947550670 |
|         2 | 947106194 |
-- ...
|       510 | 947408237 |
|       511 | 946735634 |
+-----------+-----------+
512 rows in set
Time: 46.628s

What You Expected?

Each data bucket is evenly distributed.

How to Reproduce?

No response

Anything Else?

When defining Hash Bucketing, we want to specify a custom hash function instead of using the fixed CRC32.

For example, we can use crc32(col) % bucket_num, or use xxhash64(col) % bucket_num, or even col % bucket_num.

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

[Doris-Breakwater]

Initial analysis

This is reproducible and the empty tablets are explained by the current distribution algorithm. I would classify it as a systematic hash-quality/design limitation, not an ingestion failure or random statistical skew. There are currently no labels or earlier comments on the issue.

Root cause

For hash-distributed OLAP writes, Doris initializes the hash to zero, feeds the raw bytes of each distribution column to zlib-compatible CRC32, and selects hash % num_buckets (4.0.5 routing code). A SMALLINT contributes exactly two bytes (4.0.5 type handling). Current master retains the same routing rule (master routing code).

For the exact little-endian SMALLINT domain 0..511 and 512 buckets, the lower nine CRC32 bits have a deterministic collision:

bucket(x) = bucket(x XOR 0x117), for every x in [0, 511]

Consequently, the 512 values reach only 256 distinct bucket indexes, with two values per reachable bucket. For example:

shard_num 0   -> bucket 255
shard_num 279 -> bucket 255
shard_num 1   -> bucket 446
shard_num 278 -> bucket 446

Thus, even if every shard_num has the same row count, half of the 512 logical tablets can never receive a row. The zero-row replica groups in the screenshot are consistent with this result. DISTRIBUTED BY HASH does not generally promise a one-to-one mapping when key NDV equals the bucket count, but this 50% unreachable pattern is a particularly poor interaction between CRC32's low bits, a two-byte sequential domain, and a power-of-two bucket count.

A small Doris-side validation is:

SELECT
    COUNT(DISTINCT crc32_internal(CAST(number AS SMALLINT)) % 512)
        AS smallint_used_buckets,
    COUNT(DISTINCT crc32_internal(CAST(number AS INT)) % 512)
        AS int_used_buckets
FROM numbers("number" = "512");

With the current algorithm, the results are 256 and 512, respectively. crc32_internal is appropriate here because it is implemented specifically to match tablet routing; ordinary SQL crc32() is not the routing function.

Compatibility and PR direction

Changing the default hash or adding a final mixing step in place would be unsafe. Existing rows must continue to map to their original tablets; otherwise an upgraded table could route new rows differently, and FE tablet pruning could select a different tablet from the one containing old rows. FE independently computes CRC32 and applies the same modulo during pruning (pruner, CRC32 calculation).

The requested custom hash is therefore best treated as a versioned table feature rather than a BE-only fix:

1. Persist a distribution-hash algorithm identifier in table metadata and in the FE-to-BE sink descriptor; existing tables must default to legacy CRC32.
2. Initially support a closed set of deterministic algorithms (for example, legacy CRC32 and XXH64) rather than arbitrary SQL expressions. The current grammar and metadata carry only distribution-column names, not an expression or algorithm (DDL grammar, sink descriptor).
3. Apply the selected algorithm consistently in BE write routing, FE tablet pruning, bucket-shuffle/colocation compatibility checks, schema changes, rollups/materialized indexes, and all load paths.
4. Add regression coverage for SMALLINT 0..511 / 512 buckets, pruning correctness, restart/metadata replay, mixed-version upgrade, and rejection of incompatible colocate definitions.
5. Require a rewrite/rebuild to change the algorithm of a table that already contains data.

For this specific schema, practical workarounds for a newly created table are:

• Use 256 buckets: the same 512 SMALLINT values occupy all 256 buckets, two shard values per bucket.
• If 512-way placement and one-tablet-per-shard behavior are required, use an INT physical distribution key; for this exact 0..511 domain, the current four-byte CRC32 mapping reaches all 512 buckets. Verify with the query above before migration.
• Add a higher-cardinality second distribution column if keeping all rows for one shard_num in one tablet is not required.

Clarifications useful for final validation

The root cause does not require logs or a query profile. To confirm the observed scope and prepare a regression case, please provide:

• whether the affected build is exactly 4.0.5, current master, or both (the version field is ambiguous);
• valid SHOW CREATE TABLE output (the pasted column definitions omit commas);
• for one affected partition, the number of distinct zero-row TabletId values rather than replica rows, plus whether all 512 shard_num values occur in that partition.

Breakwater-GitHub-Analysis-Slot: slot_be381027fb5f