Top-N join query still scans 6,249,580 rows on fact table even after HASH distribution and colocate optimization

[junit]

Hi Doris team,

We reproduced this issue on Apache Doris 2.1.11.

We are seeing poor performance on a simple JOIN + ORDER BY + LIMIT query in Doris.
We have already optimized the table design by:

• distributing both tables by dynamic_id
• putting both tables in the same colocate_with group
• adding inverted indexes on filter/sort related columns
• adjusting key design of the tag relation table

However, the query plan still shows that Doris scans 6,249,580 rows from table e_dynamic, although the final result only needs LIMIT 15.

This is very different from MySQL, where the same SQL only reads a small number of rows from the driving table and then does PK lookup on the joined table.

We would like to know:

1. Why Doris still scans the whole filtered e_dynamic side (cardinality=6249580) in this case.
2. Why colocate/distribution optimization does not help this query enough.
3. Whether Doris can push down ORDER BY ... LIMIT better in this join pattern.
4. Whether there are recommended table design or query rewrite patterns for this scenario.

---

Table definitions

e_dynamic_new

CREATE TABLE `e_dynamic_new` (
  `DYNAMIC_ID` int NULL COMMENT "动态ID，主键",
  `USER_ID` int NULL COMMENT "动态的用户ID",
  `USER_TYPE` int NULL COMMENT "用户类型，0普通用户，1,2分别为个人和企业机构加V",
  `TITLE` varchar(765) NULL COMMENT "动态标题",
  `CONTENT` text NULL COMMENT "动态内容",
  `PICTURE_URLS` text NULL COMMENT "动态的图片的URL",
  `DYNAMIC_TYPE` int NULL COMMENT "动态的类型，0:热门动态;   1:普通动态;  2首席圈动态",
  `RECOMMEND_STATUS` int NULL COMMENT "是否推荐",
  `SOURCE` int NULL COMMENT "动态来源，1:原创;   2:转发/分享的;  3:提问；4:回答",
  `STATUS` boolean NULL COMMENT "状态，默认为1开启，0为关闭",
  `CREATE_TIMESTAMP` bigint NULL COMMENT "创建时间",
  `REVIEW_STATUS` int NULL COMMENT "审核状态，0暂未审核，1审核通过，-1不通过",
  `FEATURED` boolean NULL COMMENT "是否精选视频【0否；1是】",
  `FEATURED_TIMESTAMP` bigint NULL COMMENT "设置为精选的时间",
  `ADJUST_TIMESTAMP` bigint NULL COMMENT "调仓记录时间戳",
  `answer_user_id` int NULL COMMENT "向谁提问",
  `ANSWER_DYNAMIC_ID` int NULL COMMENT "回答哪条动态的ID号",
  `question_type` varchar(60) NULL COMMENT "提问的方式：anonymity匿名；public公开",
  `ANSWER_TYPE` varchar(60) NULL COMMENT "回答的方式：public公开；private仅推送给提问者",
  `reply_status` int NULL COMMENT "回复状态 0:未回复 1:已回复 2:已忽略",
  `APP` varchar(60) NULL COMMENT "应用 GLH-格隆汇，DXS-东新社",
  `reason` varchar(765) NULL COMMENT "审核被拒绝的原因",
  INDEX idx_status (`STATUS`) USING INVERTED,
  INDEX idx_review (`REVIEW_STATUS`) USING INVERTED,
  INDEX idx_create_time (`CREATE_TIMESTAMP`) USING INVERTED
) ENGINE=OLAP
UNIQUE KEY(`DYNAMIC_ID`)
DISTRIBUTED BY HASH(`DYNAMIC_ID`) BUCKETS 10
PROPERTIES (
  "replication_allocation" = "tag.location.default: 3",
  "min_load_replica_num" = "-1",
  "is_being_synced" = "false",
  "colocate_with" = "dynamic_group",
  "storage_medium" = "hdd",
  "storage_format" = "V2",
  "inverted_index_storage_format" = "V1",
  "enable_unique_key_merge_on_write" = "true",
  "light_schema_change" = "true",
  "disable_auto_compaction" = "false",
  "enable_single_replica_compaction" = "false",
  "group_commit_interval_ms" = "10000",
  "group_commit_data_bytes" = "134217728",
  "enable_mow_light_delete" = "false"
);

r_hot_picture_tag_new

CREATE TABLE `r_hot_picture_tag_new` (
  `dynamic_id` int NULL COMMENT "动态id号",
  `id` int NULL COMMENT "主键id号",
  `tag_id` int NULL COMMENT "标签id号",
  `create_timestamp` bigint NULL COMMENT "创建时间",
  `status` tinyint NULL COMMENT "有效状态[0无效；1有效]",
  INDEX idx_dynamic_id (`dynamic_id`) USING INVERTED,
  INDEX idx_tag_id (`tag_id`) USING INVERTED,
  INDEX idx_create_time (`create_timestamp`) USING INVERTED
) ENGINE=OLAP
UNIQUE KEY(`dynamic_id`, `id`)
COMMENT '热图标签关联表-优化版'
DISTRIBUTED BY HASH(`dynamic_id`) BUCKETS 10
PROPERTIES (
  "replication_allocation" = "tag.location.default: 3",
  "min_load_replica_num" = "-1",
  "is_being_synced" = "false",
  "colocate_with" = "dynamic_group",
  "storage_medium" = "hdd",
  "storage_format" = "V2",
  "inverted_index_storage_format" = "V1",
  "enable_unique_key_merge_on_write" = "true",
  "light_schema_change" = "true",
  "disable_auto_compaction" = "false",
  "enable_single_replica_compaction" = "false",
  "group_commit_interval_ms" = "10000",
  "group_commit_data_bytes" = "134217728",
  "enable_mow_light_delete" = "false"
);

---

Query

SELECT
    t1.*
FROM r_hot_picture_tag t1
JOIN e_dynamic t2
    ON t1.DYNAMIC_ID = t2.DYNAMIC_ID
WHERE
    t1.tag_id = 115
    AND t2.STATUS = true
    AND t2.REVIEW_STATUS = 1
ORDER BY
    t2.CREATE_TIMESTAMP DESC
LIMIT 0, 15;

---

Doris EXPLAIN

PLAN FRAGMENT 0
  OUTPUT EXPRS:
    dynamic_id[#57]
    id[#58]
    tag_id[#59]
    create_timestamp[#60]
    status[#61]
  PARTITION: UNPARTITIONED

  HAS_COLO_PLAN_NODE: false

  VRESULT SINK
     MYSQL_PROTOCAL

  4:VMERGING-EXCHANGE
     offset: 0
     limit: 15
     final projections: dynamic_id[#51], id[#52], tag_id[#53], create_timestamp[#54], status[#55]
     final project output tuple id: 7
     distribute expr lists: 

PLAN FRAGMENT 1

  PARTITION: HASH_PARTITIONED: dynamic_id[#26]

  HAS_COLO_PLAN_NODE: true

  STREAM DATA SINK
    EXCHANGE ID: 04
    UNPARTITIONED

  3:VTOP-N(533)
  |  order by: CREATE_TIMESTAMP[#56] DESC
  |  TOPN OPT
  |  offset: 0
  |  limit: 15
  |  distribute expr lists: dynamic_id[#45]
  |  
  2:VHASH JOIN(523)
  |  join op: INNER JOIN(COLOCATE[])[]
  |  equal join conjunct: (dynamic_id[#33] = DYNAMIC_ID[#24])
  |  runtime filters: RF000[min_max] <- DYNAMIC_ID[#24](1/1/1048576), RF001[in_or_bloom] <- DYNAMIC_ID[#24](1/1/1048576)
  |  cardinality=667
  |  vec output tuple id: 5
  |  output tuple id: 5
  |  vIntermediate tuple ids: 4 
  |  hash output slot ids: 33 34 35 36 37 25 
  |  final projections: dynamic_id[#38], id[#39], tag_id[#40], create_timestamp[#41], status[#42], CREATE_TIMESTAMP[#44]
  |  final project output tuple id: 5
  |  distribute expr lists: dynamic_id[#33]
  |  distribute expr lists: DYNAMIC_ID[#24]
  |  
  |----0:VOlapScanNode(508)
  |       TABLE: glhs.e_dynamic_new(e_dynamic_new), PREAGGREGATION: ON
  |       TOPN OPT:3
  |       PREDICATES: (((STATUS[#9] = TRUE) AND (REVIEW_STATUS[#11] = 1)) AND (__DORIS_DELETE_SIGN__[#22] = 0))
  |       partitions=1/1 (e_dynamic_new)
  |       tablets=10/10, tabletList=161267786,161267790,161267794 ...
  |       cardinality=6249334, avgRowSize=0.0, numNodes=1
  |       pushAggOp=NONE
  |       final projections: DYNAMIC_ID[#0], CREATE_TIMESTAMP[#10]
  |       final project output tuple id: 1
  |    
  1:VOlapScanNode(497)
     TABLE: glhs.r_hot_picture_tag_new(r_hot_picture_tag_new), PREAGGREGATION: ON
     PREDICATES: ((tag_id[#28] = 115) AND (__DORIS_DELETE_SIGN__[#31] = 0))
     runtime filters: RF000[min_max] -> dynamic_id[#26], RF001[in_or_bloom] -> dynamic_id[#26]
     partitions=1/1 (r_hot_picture_tag_new)
     tablets=10/10, tabletList=161266563,161266567,161266571 ...
     cardinality=1334, avgRowSize=0.0, numNodes=1
     pushAggOp=NONE
     final projections: dynamic_id[#26], id[#27], tag_id[#28], create_timestamp[#29], status[#30]
     final project output tuple id: 3

---

Key observation

Even after:

• DISTRIBUTED BY HASH(DYNAMIC_ID)
• same BUCKETS
• same colocate_with = "dynamic_group"
• joining on dynamic_id

the plan still shows:

• HAS_COLO_PLAN_NODE: false
• INNER JOIN(BROADCAST)
• e_dynamic scan cardinality = 6249580

So the main issue is:

• Doris still scans a very large amount of rows from e_dynamic
• ORDER BY t2.CREATE_TIMESTAMP DESC LIMIT 15 does not effectively reduce the scan range before join
• colocate optimization is apparently not triggered

---

MySQL execution plan for the same SQL

id  select_type  table  type   possible_keys                   key         key_len  ref                  rows  Extra
1   SIMPLE       t1     ref    idx_tag_id,idx_dynamic_id_status idx_tag_id 4        const                175   Using temporary; Using filesort
1   SIMPLE       t2     eq_ref PRIMARY                         PRIMARY      4        glhs.t1.dynamic_id   1     Using where

In MySQL:

• the driving table is t1
• only about 175 rows are read from t1
• t2 is accessed by PK lookup (eq_ref)
• total work is much smaller

---

Expected behavior

We expect Doris to avoid scanning millions of rows from e_dynamic for this kind of query pattern, especially after colocate/hash distribution optimization.

At minimum, we hope to understand:

• why colocate join is not chosen
• why the optimizer still prefers scanning the large e_dynamic side
• whether there is a better schema/index/sort-key design for this pattern
• whether this is an optimizer limitation for JOIN + ORDER BY on right table + LIMIT

---

Additional questions

1. Does UNIQUE KEY table with current design prevent colocate join from being used here?
2. Is inverted index on CREATE_TIMESTAMP ineffective for ORDER BY ... DESC LIMIT?
3. Would Doris require a different data model (for example duplicate key / aggregate key / materialized view) for this pattern?
4. Is there any known limitation that ORDER BY column from join table cannot be used to prune scan early?

---

Environment

Please help us confirm what additional info is needed. We can provide:

• Doris version
• FE/BE config
• EXPLAIN SHAPE PLAN
• profile
• table row counts
• column statistics / analyzed stats

Thanks!

---

[lsy3993]

hello @junit we will analyze this issue. you can join our discord(https://discord.gg/JemTbMKj), we are always online.

[lsy3993]

hello @junit we have found that:

1. your join is COLOCATE join, the explain tells us that join op: INNER JOIN(COLOCATE[])[]
2. the cardinality=6249334, avgRowSize=0.0, numNodes=1 is not the scan row number, it's just the table's row number. if you want to know the scan number of each table, you should find it at the profile. or you can share with us the profile, we will analyze it for you.

so, please feel free, your join is COLOCATE join and the 6249334 is not the scan row number. if you want to figure out the scan number of each table, you can share with us the profile, we will analyze it for you.