如何在SQL中实现基于id3唯一值的多列数据表扁平化？

这是一个典型的**行转列（Pivot）**需求，而且要适配任意数量的value列，下面分两种场景给出具体实现方案：

一、静态场景（已知id3的所有可能取值） #

如果已经明确id3只会取固定值（比如1、2、3），用CASE WHEN的写法是跨数据库兼容的，也是最直观的实现方式：

SELECT
  id1,
  id2,
  -- 为每个value列按id3拆分生成新列
  MAX(CASE WHEN id3 = 1 THEN value1 END) AS value1_id3=1,
  MAX(CASE WHEN id3 = 1 THEN value2 END) AS value2_id3=1,
  MAX(CASE WHEN id3 = 2 THEN value1 END) AS value1_id3=2,
  MAX(CASE WHEN id3 = 2 THEN value2 END) AS value2_id3=2,
  MAX(CASE WHEN id3 = 3 THEN value1 END) AS value1_id3=3,
  MAX(CASE WHEN id3 = 3 THEN value2 END) AS value2_id3=3
FROM source_table
GROUP BY id1, id2;

为什么用MAX？ #

因为id1-id2-id3是唯一标识，每个组合只有一条数据，用MAX/MIN/ANY_VALUE（MySQL专属）这类聚合函数，只是为了满足GROUP BY后的语法要求，不会改变实际取值。

如果是SQL Server/Oracle这类支持专用PIVOT语法的数据库，也可以用如下写法，但一次只能处理一个value列，多列需要嵌套PIVOT，不如CASE WHEN直观：

-- SQL Server示例
SELECT
  id1,
  id2,
  value1_1 AS value1_id3=1,
  value2_1 AS value2_id3=1,
  value1_2 AS value1_id3=2,
  value2_2 AS value2_id3=2,
  value1_3 AS value1_id3=3,
  value2_3 AS value2_id3=3
FROM (
  SELECT
    id1,
    id2,
    CONCAT('value1_', id3) AS value1_col,
    value1,
    CONCAT('value2_', id3) AS value2_col,
    value2
  FROM source_table
) AS src
PIVOT (
  MAX(value1) FOR value1_col IN ([value1_1], [value1_2], [value1_3])
) AS p1
PIVOT (
  MAX(value2) FOR value2_col IN ([value2_1], [value2_2], [value2_3])
) AS p2;

二、动态场景（支持任意id3取值和任意数量value列） #

如果id3的取值不确定，或者后续可能新增value列，就需要用动态SQL自动生成转换逻辑，完全适配任意场景：

1. MySQL/MariaDB实现 #

-- 1. 获取所有唯一的id3值
SET @id3_list = (SELECT GROUP_CONCAT(DISTINCT id3) FROM source_table);
-- 2. 获取所有value列（假设列名以value开头，可根据实际调整规则）
SET @value_cols = (SELECT GROUP_CONCAT(DISTINCT COLUMN_NAME) 
                   FROM INFORMATION_SCHEMA.COLUMNS 
                   WHERE TABLE_SCHEMA = DATABASE() 
                     AND TABLE_NAME = 'source_table' 
                     AND COLUMN_NAME LIKE 'value%');

-- 3. 拼接完整的SQL语句
SET @sql = CONCAT('SELECT id1, id2, ',
                  (SELECT GROUP_CONCAT(
                     CONCAT('MAX(CASE WHEN id3 = ', id3, ' THEN ', col, ' END) AS ', col, '_id3=', id3)
                   ) FROM (SELECT DISTINCT id3 FROM source_table) AS t1,
                        (SELECT COLUMN_NAME AS col FROM INFORMATION_SCHEMA.COLUMNS 
                         WHERE TABLE_SCHEMA = DATABASE() 
                           AND TABLE_NAME = 'source_table' 
                           AND COLUMN_NAME LIKE 'value%') AS t2),
                  ' FROM source_table GROUP BY id1, id2;');

-- 4. 执行动态生成的SQL
PREPARE stmt FROM @sql;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

2. PostgreSQL实现 #

用PL/pgSQL编写自定义函数：

CREATE OR REPLACE FUNCTION pivot_source_table() 
RETURNS SETOF RECORD AS $$
DECLARE
  id3_values INT[];
  value_cols TEXT[];
  sql TEXT;
BEGIN
  -- 获取所有唯一id3值
  SELECT ARRAY_AGG(DISTINCT id3) INTO id3_values FROM source_table;
  -- 获取所有value列
  SELECT ARRAY_AGG(column_name) INTO value_cols 
  FROM information_schema.columns
  WHERE table_schema = current_schema() 
    AND table_name = 'source_table' 
    AND column_name LIKE 'value%';

  -- 拼接SELECT子句
  sql := 'SELECT id1, id2, ';
  FOR col IN SELECT unnest(value_cols) LOOP
    FOR id3_val IN SELECT unnest(id3_values) LOOP
      sql := sql || format('MAX(CASE WHEN id3 = %L THEN %I END) AS %I, ', 
                           id3_val, col, col || '_id3=' || id3_val);
    END LOOP;
  END LOOP;
  -- 移除末尾多余的逗号
  sql := rtrim(sql, ', ');
  sql := sql || ' FROM source_table GROUP BY id1, id2;';

  RETURN QUERY EXECUTE sql;
END;
$$ LANGUAGE plpgsql;

-- 调用函数获取结果
SELECT * FROM pivot_source_table() 
AS (id1 INT, id2 INT, value1_id3=1 TEXT, value2_id3=1 TEXT, value1_id3=2 TEXT, value2_id3=2 TEXT);

3. SQL Server实现 #

DECLARE @id3_list NVARCHAR(MAX), @value_cols NVARCHAR(MAX), @sql NVARCHAR(MAX);

-- 获取所有唯一id3值
SELECT @id3_list = STRING_AGG(DISTINCT CAST(id3 AS NVARCHAR), ', ') FROM source_table;
-- 获取所有value列
SELECT @value_cols = STRING_AGG(DISTINCT COLUMN_NAME, ', ') 
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = 'dbo' 
  AND TABLE_NAME = 'source_table' 
  AND COLUMN_NAME LIKE 'value%';

-- 拼接完整SQL
SET @sql = 'SELECT id1, id2, ' +
           (SELECT STRING_AGG(
              CONCAT('MAX(CASE WHEN id3 = ', id3, ' THEN ', col, ' END) AS ', QUOTENAME(col + '_id3=' + CAST(id3 AS NVARCHAR))),
              ', '
            ) FROM (SELECT DISTINCT id3 FROM source_table) AS t1
            CROSS JOIN (SELECT COLUMN_NAME AS col FROM INFORMATION_SCHEMA.COLUMNS 
                        WHERE TABLE_SCHEMA = 'dbo' 
                          AND TABLE_NAME = 'source_table' 
                          AND COLUMN_NAME LIKE 'value%') AS t2) +
           ' FROM source_table GROUP BY id1, id2;';

-- 执行动态SQL
EXEC sp_executesql @sql;

关键注意事项 #

• 如果数据库不允许列名包含=，可以把生成的列名改成value1_id3_1这类格式，只需要修改拼接列名的部分即可。
• 动态SQL方案会自动识别所有value列和id3取值，后续新增value列或id3值时，不需要修改代码，完全满足“支持任意数量value列”的需求。

内容的提问来源于stack exchange，提问作者rvdaele