As above said, breakup table should not be distributed by fid. As you can see from the query plan of the join statement you mentioned. Compared to the breakup table, the KV table (kv_attr_name) here is very small. Therefore, this table will be broadcasted according to the plan. If the number of attribute is less than the number of segments, then the data will be clustered to some of those segments and only those segments will participate in the broadcast motion. The following code shows the query plans when we distributed the KV table by fid/(fid, fval). Thanks.

** the table test_connect_dt_ed_col is distributed by (fid, fval).
```sql
testdb=# explain analyze INSERT INTO test.test_connect_dt_ed2(id, fid, fval, is_cont, class)
testdb-#              SELECT p.id AS id, p.fid AS fid,
testdb-#                  CASE WHEN (p.is_cont AND p.fval IS NOT NULL) THEN
testdb-#                     p.fval::FLOAT8
testdb-#                  ELSE 
testdb-#                     m.code::FLOAT8 
testdb-#                  END AS fval,
testdb-#                  p.is_cont AS is_cont,
testdb-#                  p.class::INT AS class
testdb-#             FROM
testdb-#                 tmp_breakup_table p LEFT JOIN test.test_connect_dt_ed_col m
testdb-#             ON
testdb-#                 m.fid = p.fid AND
testdb-#                 (coalesce(m.fval, '') = (coalesce(p.fval, '')));
                                                                             QUERY PLAN                                                                             
--------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Insert (slice0; segments: 4)  (rows=10971 width=57)
   ->  Hash Left Join  (cost=54.04..1367.24 rows=10971 width=57)
         Hash Cond: p.fid = m.fid AND COALESCE(p.fval, ''::text) = COALESCE(m.fval, ''::text)
         Rows out:  Avg 709348.5 rows x 4 workers.  Max 709380 rows (seg2) with 7.007 ms to first row, 1464 ms to end, start offset by 25 ms.
         Executor memory:  4K bytes avg, 4K bytes max (seg0).
         Work_mem used:  4K bytes avg, 4K bytes max (seg0).
         (seg2)   Hash chain length 1.0 avg, 1 max, using 126 of 262151 buckets.
         ->  Append-only Columnar Scan on tmp_breakup_table p  (cost=0.00..544.84 rows=10971 width=49)
               Rows out:  Avg 709348.5 rows x 4 workers.  Max 709380 rows (seg2) with 0.464 ms to first row, 455 ms to end, start offset by 31 ms.
         ->  Hash  (cost=27.52..27.52 rows=442 width=44)
               Rows in:  Avg 126.0 rows x 4 workers.  Max 126 rows (seg0) with 7.227 ms to end, start offset by 24 ms.
               ->  Broadcast Motion 4:4  (slice1; segments: 4)  (cost=0.00..27.52 rows=442 width=44)
                     Rows out:  Avg 126.0 rows x 4 workers at destination.  Max 126 rows (seg0) with 4.988 ms to first row, 7.112 ms to end, start offset by 24 ms.
                     ->  Seq Scan on test_connect_dt_ed_col m  (cost=0.00..5.42 rows=111 width=44)
                           Rows out:  Avg 31.5 rows x 4 workers.  Max 32 rows (seg1) with 0.216 ms to first row, 0.245 ms to end, start offset by 30 ms.
 Slice statistics:
   (slice0)    Executor memory: 4640K bytes avg x 4 workers, 4640K bytes max (seg0).  Work_mem: 4K bytes max.
   (slice1)    Executor memory: 256K bytes avg x 4 workers, 256K bytes max (seg0).
 Statement statistics:
   Memory used: 128000K bytes
 Total runtime: 3448.242 ms
(21 rows)

```
** the table connect_kv is distributed by fid.
```sql
testdb=# explain analyze INSERT INTO test.test_connect_dt_ed2(id, fid, fval, is_cont, class)
testdb-#              SELECT p.id AS id, p.fid AS fid,
testdb-#                  CASE WHEN (p.is_cont AND p.fval IS NOT NULL) THEN
testdb-#                     p.fval::FLOAT8
testdb-#                  ELSE 
testdb-#                     m.code::FLOAT8 
testdb-#                  END AS fval,
testdb-#                  p.is_cont AS is_cont,
testdb-#                  p.class::INT AS class
testdb-#             FROM
testdb-#                 tmp_breakup_table p LEFT JOIN connect_kv m
testdb-#             ON
testdb-#                 m.fid = p.fid AND
testdb-#                 (coalesce(m.fval, '') = (coalesce(p.fval, '')));
                                                                              QUERY PLAN                                                                               
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Insert (slice0; segments: 4)  (rows=10971 width=57)
   ->  Hash Left Join  (cost=18.12..1331.04 rows=10971 width=57)
         Hash Cond: p.fid = m.fid AND COALESCE(p.fval, ''::text) = COALESCE(m.fval, ''::text)
         Rows out:  Avg 709348.5 rows x 4 workers.  Max 709380 rows (seg2) with 7.820 ms to first row, 1449 ms to end, start offset by 2.919 ms.
         Executor memory:  4K bytes avg, 4K bytes max (seg0).
         Work_mem used:  4K bytes avg, 4K bytes max (seg0).
         (seg2)   Hash chain length 1.0 avg, 1 max, using 126 of 524341 buckets.
         ->  Append-only Columnar Scan on tmp_breakup_table p  (cost=0.00..544.84 rows=10971 width=49)
               Rows out:  Avg 709348.5 rows x 4 workers.  Max 709380 rows (seg2) with 0.522 ms to first row, 432 ms to end, start offset by 10 ms.
         ->  Hash  (cost=10.56..10.56 rows=126 width=14)
               Rows in:  Avg 126.0 rows x 4 workers.  Max 126 rows (seg0) with 2.469 ms to end, start offset by 7.492 ms.
               ->  Broadcast Motion 4:4  (slice1; segments: 4)  (cost=0.00..10.56 rows=126 width=14)
                     Rows out:  Avg 126.0 rows x 4 workers at destination.  Max 126 rows (seg0) with 0.038 ms to first row, 2.388 ms to end, start offset by 7.493 ms.
                     ->  Seq Scan on connect_kv m  (cost=0.00..4.26 rows=32 width=14)
                           Rows out:  Avg 42.0 rows x 3 workers.  Max 42 rows (seg0) with 0.129 ms to first row, 0.148 ms to end, start offset by 2.274 ms.
 Slice statistics:
   (slice0)    Executor memory: 7713K bytes avg x 4 workers, 7713K bytes max (seg0).  Work_mem: 4K bytes max.
   (slice1)    Executor memory: 256K bytes avg x 4 workers, 256K bytes max (seg0).
 Statement statistics:
   Memory used: 128000K bytes
 Total runtime: 3012.207 ms
(21 rows)

```