はじめに
MySQL 8.0で追加されたCTE（Common Table Expression）を実際に触ってみました。これまでMySQLでは困難だった階層データの処理や複雑なクエリの可読性向上が実現できるので、実例とともに紹介します。
環境
- Google Colab
- MySQL 8.0
- Python 3.x + PyMySQL
サンプルデータの準備
組織管理システムを想定したデータで学習します。
# MySQL環境構築
!apt update && apt install mysql-server -y
!service mysql start
# rootパスワード設定（初回はパスワードなし）
!mysql -u root -e "ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY 'password';"
# 動作確認
!mysql -u root -p'password' -e "SHOW DATABASES;"
!pip install pymysql
import pymysql
connection = pymysql.connect(host='localhost', user='root', password='password', charset='utf8mb4')
with connection.cursor() as cursor:
cursor.execute("CREATE DATABASE IF NOT EXISTS org_db")
cursor.execute("USE org_db")
# 組織テーブル（階層構造）
cursor.execute("""
CREATE TABLE organizations (
id INT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
parent_id INT,
org_type VARCHAR(50),
INDEX(parent_id)
)
""")
# 組織データ投入
org_data = [
(1, '本社', None, 'headquarters'),
(2, '営業本部', 1, 'division'),
(3, '技術本部', 1, 'division'),
(4, '東京営業部', 2, 'department'),
(5, '大阪営業部', 2, 'department'),
(6, '開発部', 3, 'department'),
(7, 'QA部', 3, 'department'),
(8, '第1営業課', 4, 'section'),
(9, '第2営業課', 4, 'section'),
(10, 'UI/UXチーム', 6, 'team'),
(11, 'バックエンドチーム', 6, 'team')
]
cursor.executemany(
"INSERT INTO organizations (id, name, parent_id, org_type) VALUES (%s, %s, %s, %s)",
org_data
)
connection.commit()
print("サンプルデータの準備完了")
connection.close()
CTEの基本：階層構造の展開
再帰CTEによる組織階層の取得
import pymysql
connection = pymysql.connect(
host='localhost',
user='root',
password='password',
database='org_db',
charset='utf8mb4'
)
with connection.cursor() as cursor:
# 組織階層を全て展開
hierarchy_query = """
WITH RECURSIVE org_hierarchy AS (
-- 最上位組織から開始
SELECT
id,
name,
parent_id,
org_type,
0 as level,
CAST(name AS CHAR(500)) as hierarchy_path
FROM organizations
WHERE parent_id IS NULL
UNION ALL
-- 子組織を順次追加
SELECT
o.id,
o.name,
o.parent_id,
o.org_type,
oh.level + 1,
CONCAT(oh.hierarchy_path, ' > ', o.name) as hierarchy_path
FROM organizations o
INNER JOIN org_hierarchy oh ON o.parent_id = oh.id
)
SELECT
level,
REPEAT(' ', level) as indent,
name,
org_type,
hierarchy_path
FROM org_hierarchy
ORDER BY hierarchy_path
"""
cursor.execute(hierarchy_query)
results = cursor.fetchall()
print("組織階層:")
for row in results:
level, indent, name, org_type, path = row
print(f"レベル{level}: {indent}{name} ({org_type})")
# パスも表示する場合
# print(f" → {path}")
connection.close()
期待される出力例:
組織階層:
レベル0: 本社 (headquarters)
レベル1: 営業本部 (division)
レベル1: 技術本部 (division)
レベル2: 東京営業部 (department)
レベル2: 大阪営業部 (department)
レベル2: 開発部 (department)
レベル2: QA部 (department)
レベル3: 第1営業課 (section)
レベル3: 第2営業課 (section)
レベル3: UI/UXチーム (team)
レベル3: バックエンドチーム (team)
特定組織の配下組織を取得
# 「営業本部」配下の全組織を取得
connection = pymysql.connect(
host='localhost',
user='root',
password='password',
database='org_db',
charset='utf8mb4'
)
with connection.cursor() as cursor:
sales_org_query = """
WITH RECURSIVE sales_org AS (
-- 営業本部から開始
SELECT id, name, parent_id, 0 as depth
FROM organizations
WHERE name = '営業本部'
UNION ALL
-- 配下組織を再帰的に取得
SELECT o.id, o.name, o.parent_id, so.depth + 1
FROM organizations o
INNER JOIN sales_org so ON o.parent_id = so.id
)
SELECT
REPEAT('├─ ', depth) as tree_mark,
name,
depth
FROM sales_org
ORDER BY depth, name
"""
cursor.execute(sales_org_query)
results = cursor.fetchall()
print("営業本部配下の組織:")
for row in results:
tree_mark, name, depth = row
print(f"レベル{depth}: {tree_mark}{name}")
connection.close()
CTEの応用：複雑なクエリの整理
組織タイプ別統計の取得
# 新しい接続を作成
connection = pymysql.connect(
host='localhost',
user='root',
password='password',
database='org_db',
charset='utf8mb4'
)
# 組織タイプ別の統計
with connection.cursor() as cursor:
stats_query = """
WITH org_stats AS (
SELECT
org_type,
COUNT(*) as org_count,
GROUP_CONCAT(name ORDER BY name) as org_list
FROM organizations
GROUP BY org_type
),
hierarchy_depth AS (
WITH RECURSIVE depth_calc AS (
SELECT id, name, 0 as max_depth
FROM organizations
WHERE parent_id IS NULL
UNION ALL
SELECT o.id, o.name, dc.max_depth + 1
FROM organizations o
INNER JOIN depth_calc dc ON o.parent_id = dc.id
)
SELECT MAX(max_depth) as total_depth
FROM depth_calc
)
SELECT
os.org_type,
os.org_count,
os.org_list,
hd.total_depth as max_hierarchy_depth
FROM org_stats os
CROSS JOIN hierarchy_depth hd
ORDER BY os.org_count DESC
"""
cursor.execute(stats_query)
results = cursor.fetchall()
print("組織統計:")
for row in results:
org_type, count, org_list, max_depth = row
print(f"{org_type}: {count}個 (最大階層: {max_depth})")
print(f" 組織: {org_list}")
print()
connection.close()
複数CTEの組み合わせ
# 新しい接続で階層レベル別の組織情報を取得
connection = pymysql.connect(
host='localhost',
user='root',
password='password',
database='org_db',
charset='utf8mb4'
)
with connection.cursor() as cursor:
complex_query = """
WITH RECURSIVE level_info AS (
SELECT id, name, parent_id, org_type, 0 as level
FROM organizations WHERE parent_id IS NULL
UNION ALL
SELECT o.id, o.name, o.parent_id, o.org_type, li.level + 1
FROM organizations o
JOIN level_info li ON o.parent_id = li.id
),
level_summary AS (
SELECT
level,
COUNT(*) as count_per_level,
GROUP_CONCAT(DISTINCT org_type) as types_in_level
FROM level_info
GROUP BY level
)
SELECT
li.level,
li.name,
li.org_type,
ls.count_per_level,
ls.types_in_level
FROM level_info li
JOIN level_summary ls ON li.level = ls.level
ORDER BY li.level, li.name
"""
cursor.execute(complex_query)
results = cursor.fetchall()
print("階層レベル別組織情報:")
current_level = -1
for row in results:
level, name, org_type, count_per_level, types_in_level = row
if level != current_level:
print(f"\nレベル {level} ({count_per_level}個, タイプ: {types_in_level}):")
current_level = level
indent = " " * (level + 1)
print(f"{indent}{name} ({org_type})")
connection.close()
まとめ
MySQL 8.0で導入されたCTE（共通テーブル式）により、階層構造を持つデータの処理が格段に効率化されました。これにより、組織図や商品カテゴリ、メニュー構造、BOM（部品構成表）などの再帰的なデータを扱うシステムにおいて、複雑な自己結合を用いることなく、可読性の高いクエリを記述できます。
また、アプリケーション側で行っていた再帰処理をデータベース側に移すことで、開発効率の向上にもつながります。適切なインデックス設計と組み合わせれば、実用的なパフォーマンスも確保できるため、CTEは階層データ処理の強力な選択肢となります。
参考情報