这个问题有两个不同的部分：

• 第一个问题是如何生成具有公共引用的用户组
• 第二部分是如何在考虑孤儿资产的情况下，在集团内部进行资产分配

第一部分：图遍历问题

基于公共引用识别用户集群看起来像是一个图行走问题。这在SQL中是一个复杂的任务，需要递归查询。该模式是取消透视用户的引用以生成节点，然后标识边（具有公共引用的节点），最后遍历图（无循环）以生成组。
在Postgres中，数组可以方便地聚合节点：

with recursive 
    nodes as (
        select u.id, r.ref
        from users u 
        cross join lateral ( values (u.ref1), (u.ref2) ) r(ref)
    ),
    edges as (
        select distinct n1.id as id1, n2.id as id2
        from nodes n1 
        inner join nodes n2 on n1.ref = n2.ref
    ),
    rcte as (
        select id1, id2, array[id1] as visited from edges where id1 = id2
        union all
        select r.id1, e.id2, r.visited || e.id2
        from rcte r
        inner join edges e on e.id1 = r.id2
        where e.id2 <> all(r.visited) 
    ),
    groups as (
        select id1 as id, array_agg(distinct id2 order by id2) as ids
        from rcte
        group by id1
    )
select * from groups order by id

| id| IDS|
| - -----|- -----|
| 1| {1，3，4，5}|
| 2| {2}|
| 3| {1，3，4，5}|
| 4| {1，3，4，5}|
| 5个|{1，3，4，5}|

第二部分：left join和聚合

既然我们已经确定了组，我们可以检查资产。由于您希望结果中包含所有资产，因此我们从assets表开始，然后使用left join s引入用户和组。我们仍然可以group by用户组，这将所有孤立资产放在同一个组中（组为null）-这正是我们想要的。
最后一步是数组聚合;孤儿资产的所有者到ref2的“传播”可以用case表达式来处理。

with recursive 
    nodes  as (...),
    edges  as (...),
    rcte   as (...),
    groups as (...)
select g.ids,
    array_agg(distinct u.username) as usernames,
    array_agg(distinct u.ref1) as refs1,
    case when g.ids is null then array_agg(distinct a.owner) else array_agg(distinct u.ref2) end as refs2,
    array_agg(distinct a.id) as asset_ids,
    count(distinct a.id) as asset_count
from assets a
left join users u on a.owner in (u.ref1, u.ref2)
left join groups g on g.id = u.id
group by g.ids

| IDS|用户名|参考文献1|参考文献2|资产ID|资产计数|
| - -----|- -----|- -----|- -----|- -----|- -----|
| {1，3，4，5}|{bobo，lolo，momo，popo}|{a，c}|{d，f}|（一）（二）（三）（一）（二）（三）（四）（四）（五）（五）（六）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（八）（|八|
| {2} |{toto}|{b}|{e}|[2，7]| 2|
| 联系我们|联系我们|联系我们|{z}|{11}| 1|

• • Demo on DB Fiddlde**

插件：图行走性能

在有很多边的网络上，性能会受到影响，特别是如果图中只有几个集群，每个集群包含许多用户。我们可以尝试优化这种情况下的查询;想法是通过在每次迭代 * 聚合每个用户 * 的所有路径来尝试和限制需要行走的路径的数量。
这个查询**passes your test case**有200个用户，都属于同一个集群（而第一个查询耗尽了DB Fiddle资源）：

with recursive 
    nodes as (
        select u.id, r.ref
        from users u 
        cross join lateral ( values (u.ref1), (u.ref2) ) r(ref)
    ),
    edges as (
        select distinct n1.id as id1, n2.id as id2
        from nodes n1 
        inner join nodes n2 on n1.ref = n2.ref
    ),
    rcte as (
        select id1 as id, array[id1] as visited from edges where id1 = id2
        union all
        select id, array_agg(distinct visited) as visited
        from (
            select r.id, v.visited
            from rcte r
            inner join edges e on e.id1 = any(r.visited)
            cross join lateral unnest(r.visited || e.id2) v(visited)
            where e.id2 <> all(r.visited)
        ) as x
        group by id
    ),
    groups as (
        select distinct on (id) id, visited as ids 
        from rcte 
        order by id, array_length(visited, 1) desc 
    )
select * from groups g order by id

这很难（通常非常慢）在SQL中仅使用基于集合的操作来解决。所以我在PL/pgSQL函数中循环：

CREATE OR REPLACE FUNCTION f_merged_users()
  RETURNS TABLE (
     ids int[]             -- adapt to your actual data types !!!
   , usernames text[]
   , refs1 text[]
   , refs2 text[]
   , asset_ids int[]
   , asset_count int
  )
  LANGUAGE plpgsql AS
$func$
BEGIN
   /*
   -- set enough temp buffers, so temp tables don't spill to disk (optional)
   -- NOTE: this can't be used after temporary tables have been accessed in the session
   -- using 2x size of users table. adapt to your needs!
   -- only fires when the new size is bigger, and it has not been set manually, yet
   PERFORM set_config('temp_buffers', (pg_table_size('users') * 2 / 8096)::text, true)
   FROM   pg_settings s
   WHERE  s.name = 'temp_buffers'
   AND    pg_table_size('users') * 2 / 8096 > s.setting::bigint
   AND    s.source = 'default';
   */

   -- create two temp tables: one for ref1, one for ref2
   -- based on your information that ref1 & ref2 don't overlap
   CREATE TEMP TABLE r1 ON COMMIT DROP AS
   SELECT ARRAY[ref1] AS r, array_agg(id) AS i
   FROM  (TABLE users ORDER BY id) t
   GROUP  BY ref1
   ORDER  BY ref1;

   CREATE TEMP TABLE r2 ON COMMIT DROP AS
   SELECT ARRAY[ref2] AS r, array_agg(id) AS i
   FROM  (TABLE users ORDER BY id) t
   GROUP  BY ref2
   ORDER  BY ref2;

   -- fold rows where a common link in the other table exists
   -- achieve that by deleting rows and inserting the merged results
   -- loop back and forth until no more rows can be folded
   LOOP
      WITH d AS (
         DELETE FROM r2
         USING  r1
         WHERE  EXISTS (
            SELECT FROM r2 r0
            WHERE  r1.i && r2.i
            AND    r1.i && r0.i
            AND    r2.ctid <> r0.ctid
            )
         RETURNING r1.ctid, r2.*
         )
      INSERT INTO r2
      SELECT r.r, i.i
      FROM  (
         SELECT ctid, array_agg(ref ORDER BY ref) AS r
         FROM   d, unnest(d.r) ref
         GROUP  BY 1
         ) r
      JOIN  (
         SELECT ctid, array_agg(id ORDER BY id) AS i
         FROM   d, unnest(d.i) id
         GROUP  BY 1
         )  i USING (ctid);

      EXIT WHEN NOT FOUND;  -- no folding happened, stop loop

      WITH d AS (
         DELETE FROM r1
         USING  r2
         WHERE  EXISTS (
            SELECT FROM r1 r0
            WHERE  r2.i && r1.i
            AND    r2.i && r0.i
            AND    r1.ctid <> r0.ctid
            )
         RETURNING r2.ctid, r1.*
         )
      INSERT INTO r1
      SELECT r.r, i.i
      FROM  (
         SELECT ctid, array_agg(ref ORDER BY ref) AS r
         FROM   d, unnest(d.r) ref
         GROUP  BY 1
         ) r
      JOIN  (
         SELECT ctid, array_agg(id ORDER BY id) AS i
         FROM   d, unnest(d.i) id
         GROUP  BY 1
         ) i USING (ctid);
         
      EXIT WHEN NOT FOUND;     -- no folding happened, stop loop
   END LOOP;

   -- output result
   RETURN QUERY
   (
   SELECT i                         -- AS ids
        , u.usernames
        , r1.r                      -- AS refs1
        , r2.r                      -- AS refs2
        , a.asset_ids
        , cardinality(a.asset_ids)  -- AS asset_count
   FROM   r1
   JOIN   r2 USING (i)
   CROSS  JOIN LATERAL (
      SELECT ARRAY (
         SELECT username
         FROM   users u
         WHERE  u.id = ANY (i)
         ORDER  BY u.id
         ) AS usernames
      ) u
   CROSS  JOIN LATERAL (
      SELECT ARRAY (
         SELECT a.id
         FROM   assets a
         WHERE  a.owner = ANY (r1.r || r2.r)
         ORDER  BY a.id
         ) AS asset_ids
      ) a
   ORDER  BY i
   )
   UNION ALL  --  add "missing entries in the user table"
   (
   SELECT null             -- AS ids
        , null             -- AS usernames
        , null             -- AS refs1
        , ARRAY[a.owner]   -- AS refs2
        , ARRAY[a.id]      -- AS ids
        , 1                -- AS asset_count
   FROM   assets a
   WHERE  NOT EXISTS (
      SELECT FROM users u
      WHERE  a.owner IN (u.ref1, u.ref2)
      )
   ORDER  BY a.id
   );
END
$func$;

fiddle
电话：

SELECT * FROM f_merged_users();

见内联注解和解释。
一般的想法是避免一步一步地遍历可能很长的图，并尽可能多地进行基于集合的工作。最佳解决方案在很大程度上取决于实际的数据分布。
根据ref1和ref2不重叠的信息，我创建了两个临时表：一个用于ref1，一个用于ref2。初始查询聚合同一资源上的所有用户。这可以立即处理同一资源（同一个人）上的许多用户。你说你有很多这样的。
在循环的每次迭代中，我合并 other 临时表中的所有行，这些行被 this 临时表中的同一行链接在一起，反之亦然。这两个查询是完全对称的。（从一个或另一个开始可能更有效。）只要一个查询没有找到任何要合并的行，我们就完成了。
然后返回包含所有详细信息的最终集合，并像您在问题中添加的那样附加“用户表中缺少的条目”。
我已经简化了工作流程，但是DELETE和INSERT是相对昂贵的操作。确保您允许足够的临时缓冲区，以便两个临时表可以在RAM中独占操作。我在顶部添加了一点巧妙的自动设置。但如果您的temp_buffer设置足够高，您可能不需要它。在同一会话中访问任何临时表后，不能使用自动设置。参见：

• 在SQL中创建临时表
• 按ID删除数百万行的最佳方法

这两个临时表在 transaction 结束时自动删除，因为是用ON COMMIT DROP创建的。
使用临时表上的索引和更多的微调，性能可能会得到进一步优化。在这种情况下，在临时表上运行ANALYZE（一次或多次）也会有所帮助。参见：

• Are regular VACUUM ANALYZE still recommended under 9.1?

关于我使用的ctid：

• 从小表中删除重复行

您所显示的已经是一个有效的图表。使用pgRouting，您可以直接查询它：

select chr(component::int)       as "the_group",
       array_agg(chr(node::int)) as "ids"
from pgr_connectedComponents(
  'SELECT id, 
          ascii(ref1) as source, 
          ascii(ref2) as target, 
          1 as cost, 
          1 as reverse_cost 
   FROM users') 
group by component;
-- the_group |    ids
-------------+-----------
-- a         | {a,c,d,f}
-- b         | {b,e}

ascii()和chr()可以将1个字符的ref转换为扩展所期望的integer，然后转换回来：ascii('a')=97、chr('97')='a'。在数百万用户组成的不同数量和长度的链接组合上进行了测试，它需要几秒钟的时间，并且在单个CPU上占用的内存可以忽略不计。在引擎盖下，它是一个Boost C++ DFS。
db<>fiddle，db-fiddle和sqlfiddle不提供pgrouting，这就是为什么我不能附加一个在线演示，但here的测试数据生成器，我折磨这个。
一旦SQL:2023的SQL/PGQ进入PostgreSQL发行版，您可以期待一个内置的机制。

请看下一个解决方案：

WITH agg1 as (
    SELECT 
      string_agg(distinct users.id::text, ',') id, 
      string_agg(distinct username, ',') username, 
      string_agg(distinct ref2, ',') ref2,
      string_agg(distinct assets.id::text, ',') assets_id,
      sum(assets_count) assets_count,
      ref1
    FROM users
    JOIN (
      SELECT string_agg(id::text, ',') id, count(*) assets_count, owner 
      FROM assets GROUP BY owner
    ) assets ON assets.owner in (ref1, ref2)
    GROUP BY ref1
) SELECT 
    string_to_array(string_agg(id::text, ','),',') ids, 
    string_to_array(string_agg(username, ','),',') usernames, 
    string_to_array(string_agg(ref1, ','),',') refs1,
    string_to_array(ref2,',') refs2,
    string_to_array(string_agg(assets_id, ','),',') assets_id,
    min(assets_count) assets_count
FROM agg1
GROUP BY agg1.ref2
;

https://sqlize.online/sql/psql11/88ab227ab4a34c532fc711cff533f272/
它返回几乎期望的结果。您只需要unique assets_id数组