graph-database-expert

グラフデータベースのエキスパート

1. 概要

リスクレベル：中（データモデリングとクエリのパフォーマンス）

グラフデータベースに関する深い専門知識を持つ、優れたエキスパートです。

• グラフ理論：ノード、エッジ、パス、サイクル、グラフアルゴリズム
• グラフモデリング：エンティティとリレーションシップのマッピング、スキーマ設計、非正規化戦略
• クエリ言語：SurrealQL、Cypher、Gremlin、SPARQL のパターン
• グラフのトラバーサル：深さ優先、幅優先、最短経路、パターンマッチング
• リレーションシップ設計：双方向エッジ、型付きリレーションシップ、エッジ上のプロパティ
• パフォーマンス：インデックス戦略、クエリの最適化、トラバーサル深度の制限
• マルチモデル：グラフと並行したドキュメントストレージ、時系列、キーバリュー
• SurrealDB：RELATE 文、グラフ演算子、レコードリンク

次の特性を備えたグラフデータベースを設計します。

• 直感的：つながりのあるデータとリレーションシップを自然にモデリング
• 高性能：最適化されたインデックス、効率的なトラバーサル、上限を設けたクエリ
• 柔軟：スキーマの進化、動的なリレーションシップ、マルチモデルのサポート
• スケーラブル：適切なインデックス作成、クエリプランニング、接続管理

グラフデータベースを使用するケース：

• ソーシャルネットワーク（友人、フォロワー、つながり）
• ナレッジグラフ（エンティティ、概念、リレーションシップ）
• レコメンデーションエンジン（ユーザーの好み、類似アイテム）
• 不正検出（トランザクションパターン、ネットワーク分析）
• アクセス制御（ロール階層、権限の継承）
• ネットワークトポロジ（インフラストラクチャ、依存関係、経路）
• コンテンツ管理（分類体系、参照、バージョン）

グラフデータベースを使用しないケース：

• リレーションシップがほとんどない単純な CRUD
• 集計や分析が中心のワークロード（OLAP を使用してください）
• トラバーサルの必要がない、つながりのないデータ
• 大規模な時系列データ（専用の TSDB を使用してください）

グラフデータベースの概要：

• Neo4j：市場をリードする製品、Cypher クエリ言語、ACID 準拠
• SurrealDB：マルチモデル、グラフとドキュメント、SurrealQL
• ArangoDB：マルチモデル、AQL クエリ言語、分散型
• Amazon Neptune：マネージドサービス、Gremlin と SPARQL
• JanusGraph：分散型、スケーラブル、複数のバックエンド

---

2. 基本原則

TDD を最優先

• 実装の前にグラフクエリのテストを作成します
• トラバーサルの結果が期待されるパターンと一致することを検証します
• サイクル、深いトラバーサル、存在しないノードなどのエッジケースをテストします
• 一貫したグラフの状態を保つためにテストフィクスチャを使用します

パフォーマンスを意識

• すべてのクエリを実行計画でプロファイリングします
• すべてのトラバーサルに深度制限を設定します
• ボトルネックになる前にプロパティへインデックスを作成します
• 大きな結果セットのメモリ使用量を監視します

セキュリティを重視

• 常にパラメーター化クエリを使用します
• ノードとエッジに行レベルセキュリティを実装します
• トラバーサル結果で公開するデータを制限します
• クエリを構築する前に、すべてのユーザー入力を検証します

スキーマの進化に備える

• リレーションシップの種類を追加できるように設計します
• ノードとエッジのプロパティ変更に備えます
• 監査証跡にはバージョン管理を使用します
• スキーマの変更を記録します

クエリパターンを基準にする

• アクセスパターンに基づいてスキーマをモデリングします
• 最も頻繁に行われるトラバーサルを最適化します
• 一般的なクエリに合わせてリレーションシップの方向を設計します
• 正規化とクエリパフォーマンスのバランスを取ります

---

3. 主な責務

1. グラフスキーマの設計

最適なグラフスキーマを設計します。

• エンティティを、適切なプロパティを持つノード／頂点としてモデリングします
• 意味のあるリレーションシップをエッジとして定義します
• アクセスパターンに基づいて、埋め込みとリンクのどちらを使うか選択します
• 必要に応じて双方向のリレーションシップを設計します
• 異なる種類のリレーションシップには型付きエッジを使用します
• リレーションシップのメタデータとしてエッジにプロパティを追加します
• クエリパフォーマンスを考慮し、正規化と非正規化のバランスを取ります
• スキーマの進化やリレーションシップの変更に備えます
• 詳細なパターンについては references/modeling-guide.md を参照してください

2. クエリの最適化

グラフクエリのパフォーマンスを最適化します。

• 頻繁にクエリされるノードプロパティにインデックスを作成します
• エッジタイプとリレーションシッププロパティにインデックスを付けます
• 適切なトラバーサルアルゴリズム（BFS、DFS、最短経路）を使用します
• クエリが際限なく実行されないように深さ制限を設定します
• パターンマッチングで直積を避けます
• クエリヒントと実行計画を使用します
• 大規模な結果セットにはページネーションを実装します
• 頻繁に実行されるトラバーサル結果をキャッシュします
• 戦略については、references/query-optimization.md を参照してください

3. リレーションシップのモデリング

効果的なリレーションシップパターンを設計します。

• クエリパターンに基づいてリレーションシップの方向を選択します
• 多対多の関係は中間エッジでモデル化します
• 階層（ツリー、DAG）を効率的に実装します
• 時間ベースのリレーションシップ（有効開始日時／終了日時）を設計します
• リレーションシップの多重度（一対一、一対多、多対多）を処理します
• エッジにメタデータ（重み、タイムスタンプ、プロパティ）を追加します
• リレーションシップにソフトデリートを実装します
• 監査証跡のためにリレーションシップをバージョン管理します

4. パフォーマンスとスケーラビリティ

グラフデータベースのパフォーマンスを確保します。

• クエリ実行計画を監視します
• 遅いトラバーサルを特定して最適化します
• コネクションプーリングを使用します
• 適切なキャッシュ戦略を実装します
• 妥当なトラバーサル深さの上限を設定します
• 可能な場合は操作をバッチ処理します
• 大規模なトラバーサルでメモリ使用量を監視します
• 大規模な結果セットにはページネーションとカーソルを使用します

---

4. 実装ワークフロー（TDD）

ステップ 1：最初に失敗するテストを書く

# tests/test_graph_queries.py
import pytest
from surrealdb import Surreal

@pytest.fixture
async def db():
    """Setup test database with graph schema."""
    db = Surreal("ws://localhost:8000/rpc")
    await db.connect()
    await db.signin({"user": "root", "pass": "root"})
    await db.use("test", "test")

    # Setup schema
    await db.query("""
        DEFINE TABLE person SCHEMAFULL;
        DEFINE FIELD name ON TABLE person TYPE string;
        DEFINE INDEX person_name ON TABLE person COLUMNS name;

        DEFINE TABLE follows SCHEMAFULL;
        DEFINE FIELD in ON TABLE follows TYPE record<person>;
        DEFINE FIELD out ON TABLE follows TYPE record<person>;
    """)

    yield db

    # Cleanup
    await db.query("REMOVE TABLE person; REMOVE TABLE follows;")
    await db.close()

@pytest.mark.asyncio
async def test_multi_hop_traversal(db):
    """Test that multi-hop traversal returns correct results."""
    # Arrange: Create test graph
    await db.query("""
        CREATE person:alice SET name = 'Alice';
        CREATE person:bob SET name = 'Bob';
        CREATE person:charlie SET name = 'Charlie';
        RELATE person:alice->follows->person:bob;
        RELATE person:bob->follows->person:charlie;
    """)

    # Act: Traverse 2 hops
    result = await db.query(
        "SELECT ->follows[..2]->person.name FROM person:alice"
    )

    # Assert: Should find Bob and Charlie
    names = result[0]['result'][0]['name']
    assert 'Bob' in names
    assert 'Charlie' in names

@pytest.mark.asyncio
async def test_depth_limit_respected(db):
    """Test that traversal depth limits are enforced."""
    # Arrange: Create chain of 5 nodes
    await db.query("""
        CREATE person:a SET name = 'A';
        CREATE person:b SET name = 'B';
        CREATE person:c SET name = 'C';
        CREATE person:d SET name = 'D';
        CREATE person:e SET name = 'E';
        RELATE person:a->follows->person:b;
        RELATE person:b->follows->person:c;
        RELATE person:c->follows->person:d;
        RELATE person:d->follows->person:e;
    """)

    # Act: Traverse only 2 hops
    result = await db.query(
        "SELECT ->follows[..2]->person.name FROM person:a"
    )

    # Assert: Should NOT include D or E
    names = result[0]['result'][0]['name']
    assert 'D' not in names
    assert 'E' not in names

@pytest.mark.asyncio
async def test_bidirectional_relationship(db):
    """Test querying in both directions."""
    # Arrange
    await db.query("""
        CREATE person:alice SET name = 'Alice';
        CREATE person:bob SET name = 'Bob';
        RELATE person:alice->follows->person:bob;
    """)

    # Act: Query both directions
    forward = await db.query(
        "SELECT ->follows->person.name FROM person:alice"
    )
    backward = await db.query(
        "SELECT <-follows<-person.name FROM person:bob"
    )

    # Assert
    assert 'Bob' in str(forward)
    assert 'Alice' in str(backward)

@pytest.mark.asyncio
async def test_weighted_edge_filter(db):
    """Test filtering edges by weight."""
    # Setup weighted edges
    await db.query("""
        DEFINE TABLE connected SCHEMAFULL;
        DEFINE FIELD in ON TABLE connected TYPE record<person>;
        DEFINE FIELD out ON TABLE connected TYPE record<person>;
        DEFINE FIELD weight ON TABLE connected TYPE float;

        CREATE person:alice SET name = 'Alice';
        CREATE person:bob SET name = 'Bob';
        CREATE person:charlie SET name = 'Charlie';
        RELATE person:alice->connected->person:bob SET weight = 0.9;
        RELATE person:alice->connected->person:charlie SET weight = 0.3;
    """)

    # Act: Filter by weight
    result = await db.query(
        "SELECT ->connected[WHERE weight > 0.5]->person.name FROM person:alice"
    )

    # Assert: Only Bob (high weight)
    assert 'Bob' in str(result)
    assert 'Charlie' not in str(result)

ステップ 2：テストに合格するための最小限の実装を行う

# src/graph/queries.py
from surrealdb import Surreal

class GraphQueryService:
    def __init__(self, db: Surreal):
        self.db = db

    async def get_connections(
        self,
        node_id: str,
        relationship: str,
        depth: int = 2,
        min_weight: float | None = None
    ) -> list[dict]:
        """Get connected nodes with depth limit."""
        if depth > 5:
            raise ValueError("Maximum depth is 5 to prevent runaway queries")

        # Build query with parameterization
        if min_weight is not None:
            query = f"""
                SELECT ->{relationship}[..{depth}][WHERE weight > $min_weight]->*.*
                FROM $node_id
            """
            params = {"node_id": node_id, "min_weight": min_weight}
        else:
            query = f"""
                SELECT ->{relationship}[..{depth}]->*.*
                FROM $node_id
            """
            params = {"node_id": node_id}

        result = await self.db.query(query, params)
        return result[0]['result']

    async def find_path(
        self,
        from_id: str,
        to_id: str,
        relationship: str,
        max_depth: int = 5
    ) -> list[str] | None:
        """Find shortest path between two nodes."""
        # BFS implementation with depth limit
        visited = set()
        queue = [(from_id, [from_id])]

        while queue and len(visited) < 1000:  # Safety limit
            current, path = queue.pop(0)
            if len(path) > max_depth:
                continue

            if current == to_id:
                return path

            if current in visited:
                continue
            visited.add(current)

            # Get neighbors
            result = await self.db.query(
                f"SELECT ->{relationship}->*.id FROM $node",
                {"node": current}
            )

            for neighbor in result[0]['result']:
                if neighbor not in visited:
                    queue.append((neighbor, path + [neighbor]))

        return None

ステップ 3：必要に応じてリファクタリングする

# After tests pass, refactor for better performance
class GraphQueryService:
    def __init__(self, db: Surreal):
        self.db = db
        self._cache = {}  # Add caching

    async def get_connections_cached(
        self,
        node_id: str,
        relationship: str,
        depth: int = 2
    ) -> list[dict]:
        """Get connections with caching."""
        cache_key = f"{node_id}:{relationship}:{depth}"

        if cache_key in self._cache:
            return self._cache[cache_key]

        result = await self.get_connections(node_id, relationship, depth)
        self._cache[cache_key] = result

        return result

    def invalidate_cache(self, node_id: str = None):
        """Clear cache entries."""
        if node_id:
            self._cache = {
                k: v for k, v in self._cache.items()
                if not k.startswith(node_id)
            }
        else:
            self._cache.clear()

ステップ 4：完全な検証を実行する

# Run all graph database tests
pytest tests/test_graph_queries.py -v

# Run with coverage
pytest tests/test_graph_queries.py --cov=src/graph --cov-report=term-missing

# Run performance tests
pytest tests/test_graph_performance.py -v --benchmark-only

# Check for slow queries (custom marker)
pytest tests/test_graph_queries.py -m slow -v

---

5. パフォーマンスパターン

パターン 1：インデックス戦略

良い例：クエリで必要になる前にインデックスを作成する

-- Index frequently queried properties
DEFINE INDEX person_email ON TABLE person COLUMNS email UNIQUE;
DEFINE INDEX person_name ON TABLE person COLUMNS name;

-- Index edge properties used in filters
DEFINE INDEX follows_weight ON TABLE follows COLUMNS weight;
DEFINE INDEX employment_role ON TABLE employment COLUMNS role;
DEFINE INDEX employment_dates ON TABLE employment COLUMNS valid_from, valid_to;

-- Composite index for common filter combinations
DEFINE INDEX person_status_created ON TABLE person COLUMNS status, created_at;

悪い例：インデックスなしでクエリする

-- Full table scan on every query!
SELECT * FROM person WHERE email = '[email protected]';
SELECT ->follows[WHERE weight > 0.5]->person.* FROM person:alice;

パターン 2：クエリの最適化

良い例：上限を設けたトラバーサル

-- Always set depth limits
SELECT ->follows[..3]->person.name FROM person:alice;

-- Use pagination for large results
SELECT ->follows->person.* FROM person:alice LIMIT 50 START 0;

-- Filter early to reduce traversal
SELECT ->follows[WHERE weight > 0.5][..2]->person.name
FROM person:alice
LIMIT 100;

悪い例：上限のないクエリ

-- Can traverse entire graph!
SELECT ->follows->person.* FROM person:alice;

-- No limits on results
SELECT * FROM person WHERE status = 'active';

パターン 3：頻繁に実行されるトラバーサルのキャッシュ

良い例：コストの高いトラバーサルをキャッシュする

from functools import lru_cache
from datetime import datetime, timedelta

class GraphCache:
    def __init__(self, ttl_seconds: int = 300):
        self.cache = {}
        self.ttl = timedelta(seconds=ttl_seconds)

    async def get_followers_cached(
        self,
        db: Surreal,
        person_id: str
    ) -> list[dict]:
        cache_key = f"followers:{person_id}"

        if cache_key in self.cache:
            entry = self.cache[cache_key]
            if datetime.now() - entry['time'] < self.ttl:
                return entry['data']

        # Execute query
        result = await db.query(
            "SELECT <-follows<-person.* FROM $person LIMIT 100",
            {"person": person_id}
        )

        # Cache result
        self.cache[cache_key] = {
            'data': result[0]['result'],
            'time': datetime.now()
        }

        return result[0]['result']

    def invalidate(self, person_id: str):
        """Invalidate cache when graph changes."""
        keys_to_remove = [
            k for k in self.cache
            if person_id in k
        ]
        for key in keys_to_remove:
            del self.cache[key]

悪い例：繰り返し実行するクエリをキャッシュしない

# Every call hits the database
async def get_followers(db, person_id):
    return await db.query(
        "SELECT <-follows<-person.* FROM $person",
        {"person": person_id}
    )

パターン 4：バッチ操作

良い例：複数の操作をバッチ処理する

-- Batch create nodes
CREATE person CONTENT [
    { id: 'person:alice', name: 'Alice' },
    { id: 'person:bob', name: 'Bob' },
    { id: 'person:charlie', name: 'Charlie' }
];

-- Batch create relationships
LET $relations = [
    { from: 'person:alice', to: 'person:bob' },
    { from: 'person:bob', to: 'person:charlie' }
];
FOR $rel IN $relations {
    RELATE type::thing('person', $rel.from)->follows->type::thing('person', $rel.to);
};

# Python batch operations
async def batch_create_relationships(
    db: Surreal,
    relationships: list[dict]
) -> None:
    """Create multiple relationships in one transaction."""
    queries = []
    for rel in relationships:
        queries.append(
            f"RELATE {rel['from']}->follows->{rel['to']};"
        )

    # Execute as single transaction
    await db.query("BEGIN TRANSACTION; " + " ".join(queries) + " COMMIT;")

悪い例：個別に操作する

# N database round trips!
async def create_relationships_slow(db, relationships):
    for rel in relationships:
        await db.query(
            f"RELATE {rel['from']}->follows->{rel['to']};"
        )

パターン 5：コネクションプーリング

良い例：コネクションプールを使用する

from contextlib import asynccontextmanager
import asyncio

class SurrealPool:
    def __init__(self, url: str, pool_size: int = 10):
        self.url = url
        self.pool_size = pool_size
        self._pool = asyncio.Queue(maxsize=pool_size)
        self._created = 0

    async def initialize(self):
        """Pre-create connections."""
        for _ in range(self.pool_size):
            conn = await self._create_connection()
            await self._pool.put(conn)

    async def _create_connection(self) -> Surreal:
        db = Surreal(self.url)
        await db.connect()
        await db.signin({"user": "root", "pass": "root"})
        await db.use("jarvis", "main")
        self._created += 1
        return db

    @asynccontextmanager
    async def acquire(self):
        """Get connection from pool."""
        conn = await self._pool.get()
        try:
            yield conn
        finally:
            await self._pool.put(conn)

    async def close(self):
        """Close all connections."""
        while not self._pool.empty():
            conn = await self._pool.get()
            await conn.close()

# Usage
pool = SurrealPool("ws://localhost:8000/rpc")
await pool.initialize()

async with pool.acquire() as db:
    result = await db.query("SELECT * FROM person LIMIT 10")

悪い例：クエリごとに接続を作成する

# Connection overhead on every query!
async def query_slow(query: str):
    db = Surreal("ws://localhost:8000/rpc")
    await db.connect()
    await db.signin({"user": "root", "pass": "root"})
    result = await db.query(query)
    await db.close()
    return result

---

6. グラフモデリングの主要パターン 7 選

パターン 1：型付きリレーションシップを持つエンティティノード（SurrealDB）

-- Define entity tables
DEFINE TABLE person SCHEMAFULL;
DEFINE FIELD name ON TABLE person TYPE string;
DEFINE FIELD email ON TABLE person TYPE string;
DEFINE FIELD created_at ON TABLE person TYPE datetime DEFAULT time::now();

DEFINE TABLE company SCHEMAFULL;
DEFINE FIELD name ON TABLE company TYPE string;
DEFINE FIELD industry ON TABLE company TYPE string;

-- Define relationship tables (typed edges)
DEFINE TABLE works_at SCHEMAFULL;
DEFINE FIELD in ON TABLE works_at TYPE record<person>;
DEFINE FIELD out ON TABLE works_at TYPE record<company>;
DEFINE FIELD role ON TABLE works_at TYPE string;
DEFINE FIELD start_date ON TABLE works_at TYPE datetime;
DEFINE FIELD end_date ON TABLE works_at TYPE option<datetime>;

-- Create relationships
RELATE person:alice->works_at->company:acme SET
    role = 'Engineer',
    start_date = time::now();

-- Forward traversal: Who works at this company?
SELECT <-works_at<-person.* FROM company:acme;

-- Backward traversal: Where does this person work?
SELECT ->works_at->company.* FROM person:alice;

-- Filter on edge properties
SELECT ->works_at[WHERE role = 'Engineer']->company.*
FROM person:alice;

一般的な概念：エンティティをノード、リレーションシップをプロパティ付きのエッジとしてモデル化します。方向はクエリの効率に影響します。

---

パターン 2：マルチホップのグラフトラバーサル

-- Schema: person -> follows -> person -> likes -> post
DEFINE TABLE follows SCHEMAFULL;
DEFINE FIELD in ON TABLE follows TYPE record<person>;
DEFINE FIELD out ON TABLE follows TYPE record<person>;

DEFINE TABLE likes SCHEMAFULL;
DEFINE FIELD in ON TABLE likes TYPE record<person>;
DEFINE FIELD out ON TABLE likes TYPE record<post>;

-- Multi-hop: Posts liked by people I follow
SELECT ->follows->person->likes->post.* FROM person:alice;

-- Depth limit to prevent runaway queries
SELECT ->follows[..3]->person.name FROM person:alice;

-- Variable depth traversal
SELECT ->follows[1..2]->person.* FROM person:alice;

-- DON'T: Unbounded traversal (dangerous!)
-- SELECT ->follows->person.* FROM person:alice; -- Could traverse entire graph!

一般的な概念：グラフトラバーサルでは、エッジをたどって接続されたノードを見つけます。パフォーマンス上の問題を防ぐため、必ず深さ制限を設定してください。

Neo4j 相当の例：

// Multi-hop in Cypher
MATCH (alice:Person {id: 'alice'})-[:FOLLOWS*1..2]->(person:Person)
RETURN person

---

パターン 3：双方向リレーションシップ

-- Model friendship (symmetric relationship)
DEFINE TABLE friendship SCHEMAFULL;
DEFINE FIELD in ON TABLE friendship TYPE record<person>;
DEFINE FIELD out ON TABLE friendship TYPE record<person>;
DEFINE FIELD created_at ON TABLE friendship TYPE datetime DEFAULT time::now();

-- Create both directions for friendship
RELATE person:alice->friendship->person:bob;
RELATE person:bob->friendship->person:alice;

-- Query friends in either direction
SELECT ->friendship->person.* FROM person:alice;
SELECT <-friendship<-person.* FROM person:alice;

-- Alternative: Single edge with bidirectional query
-- Query both incoming and outgoing
SELECT ->friendship->person.*, <-friendship<-person.*
FROM person:alice;

一般的な概念：対称的なリレーションシップには慎重な設計が必要です。双方向のエッジを作成するか、両方向にクエリします。

設計上の選択肢：

• エッジを複製する：クエリは高速になりますが、ストレージを多く使用します
• 単一のエッジ＋双方向クエリ：ストレージの使用量は少なくなりますが、やや遅くなります
• 無向グラフフラグ：データベース固有の機能です

---

パターン 4：階層データ（ツリーと DAG）

-- Organization hierarchy
DEFINE TABLE org_unit SCHEMAFULL;
DEFINE FIELD name ON TABLE org_unit TYPE string;
DEFINE FIELD level ON TABLE org_unit TYPE string;

DEFINE TABLE reports_to SCHEMAFULL;
DEFINE FIELD in ON TABLE reports_to TYPE record<org_unit>;
DEFINE FIELD out ON TABLE reports_to TYPE record<org_unit>;

-- Create hierarchy
RELATE org_unit:eng->reports_to->org_unit:cto;
RELATE org_unit:product->reports_to->org_unit:cto;
RELATE org_unit:cto->reports_to->org_unit:ceo;

-- Get all ancestors (upward traversal)
SELECT ->reports_to[..10]->org_unit.* FROM org_unit:eng;

-- Get all descendants (downward traversal)
SELECT <-reports_to[..10]<-org_unit.* FROM org_unit:ceo;

-- Add materialized path for faster ancestor queries
DEFINE FIELD path ON TABLE org_unit TYPE string;
-- Store as: '/ceo/cto/eng' for fast LIKE queries

-- Add level for depth queries
UPDATE org_unit:eng SET level = 3;
SELECT * FROM org_unit WHERE level = 3;

一般的な概念：ツリーと階層は特別なグラフパターンです。複雑なクエリには、マテリアライズドパスやネストセットを検討してください。

---

パターン 5：時間ベースのリレーションシップ（時間ベースのエッジ）

-- Track relationship validity periods
DEFINE TABLE employment SCHEMAFULL;
DEFINE FIELD in ON TABLE employment TYPE record<person>;
DEFINE FIELD out ON TABLE employment TYPE record<company>;
DEFINE FIELD role ON TABLE employment TYPE string;
DEFINE FIELD valid_from ON TABLE employment TYPE datetime;
DEFINE FIELD valid_to ON TABLE employment TYPE option<datetime>;

-- Create temporal relationship
RELATE person:alice->employment->company:acme SET
    role = 'Engineer',
    valid_from = d'2020-01-01T00:00:00Z',
    valid_to = d'2023-12-31T23:59:59Z';

-- Query current relationships
LET $now = time::now();
SELECT ->employment[WHERE valid_from <= $now AND (valid_to = NONE OR valid_to >= $now)]->company.*
FROM person:alice;

-- Query historical relationships
SELECT ->employment[WHERE valid_from <= d'2021-06-01']->company.*
FROM person:alice;

-- Index temporal fields
DEFINE INDEX employment_valid_from ON TABLE employment COLUMNS valid_from;
DEFINE INDEX employment_valid_to ON TABLE employment COLUMNS valid_to;

一般的な概念：時間ベースのクエリのために、エッジにタイムスタンプを追加します。監査証跡、履歴分析、バージョン管理に不可欠です。

---

パターン 6：重み付きリレーションシップ（グラフアルゴリズム）

-- Social network with relationship strength
DEFINE TABLE connected_to SCHEMAFULL;
DEFINE FIELD in ON TABLE connected_to TYPE record<person>;
DEFINE FIELD out ON TABLE connected_to TYPE record<person>;
DEFINE FIELD weight ON TABLE connected_to TYPE float;
DEFINE FIELD interaction_count ON TABLE connected_to TYPE int DEFAULT 0;

-- Create weighted edges
RELATE person:alice->connected_to->person:bob SET
    weight = 0.8,
    interaction_count = 45;

-- Filter by weight threshold
SELECT ->connected_to[WHERE weight > 0.5]->person.* FROM person:alice;

-- Sort by relationship strength
SELECT ->connected_to->person.*, ->connected_to.weight AS strength
FROM person:alice
ORDER BY strength DESC;

-- Use cases:
-- - Shortest weighted path algorithms
-- - Recommendation scoring
-- - Fraud detection patterns
-- - Network flow analysis

一般的な概念：エッジプロパティを使うと、グラフアルゴリズムを利用できます。重みは経路探索、レコメンデーション、ネットワーク分析の基本要素です。

---

パターン 7：グラフトラバーサルによる N+1 クエリの回避

-- N+1 ANTI-PATTERN: Multiple queries
-- First query
SELECT * FROM person;
-- Then for each person (N queries)
SELECT * FROM company WHERE id = (SELECT ->works_at->company FROM person:alice);
SELECT * FROM company WHERE id = (SELECT ->works_at->company FROM person:bob);

-- CORRECT: Single graph traversal
SELECT
    *,
    ->works_at->company.* AS companies
FROM person;

-- With FETCH to include related data
SELECT * FROM person FETCH ->works_at->company;

-- Complex traversal in one query
SELECT
    name,
    ->works_at->company.name AS company_name,
    ->follows->person.name AS following,
    <-follows<-person.name AS followers
FROM person:alice;

一般的な概念：グラフデータベースは結合処理に優れています。複数回の往復クエリではなく、トラバーサル演算子を使用します。

---

7. テスト

グラフクエリの単体テスト

# tests/test_graph_service.py
import pytest
from unittest.mock import AsyncMock, MagicMock

@pytest.fixture
def mock_db():
    """Create mock database for unit tests."""
    db = AsyncMock()
    return db

@pytest.mark.asyncio
async def test_get_connections_enforces_depth_limit(mock_db):
    """Test that depth limit is enforced."""
    from src.graph.queries import GraphQueryService

    service = GraphQueryService(mock_db)

    with pytest.raises(ValueError) as exc_info:
        await service.get_connections("person:alice", "follows", depth=10)

    assert "Maximum depth is 5" in str(exc_info.value)

@pytest.mark.asyncio
async def test_cache_invalidation(mock_db):
    """Test cache invalidation works correctly."""
    from src.graph.queries import GraphQueryService

    mock_db.query.return_value = [{'result': [{'name': 'Bob'}]}]

    service = GraphQueryService(mock_db)

    # First call
    result1 = await service.get_connections_cached("person:alice", "follows")
    # Second call (should use cache)
    result2 = await service.get_connections_cached("person:alice", "follows")

    # Only one DB call
    assert mock_db.query.call_count == 1

    # Invalidate and call again
    service.invalidate_cache("person:alice")
    result3 = await service.get_connections_cached("person:alice", "follows")

    # Should hit DB again
    assert mock_db.query.call_count == 2

実データベースを使用した統合テスト

# tests/integration/test_graph_integration.py
import pytest
from surrealdb import Surreal

@pytest.fixture(scope="module")
async def test_db():
    """Setup test database."""
    db = Surreal("ws://localhost:8000/rpc")
    await db.connect()
    await db.signin({"user": "root", "pass": "root"})
    await db.use("test", "graph_test")

    yield db

    # Cleanup
    await db.query("REMOVE DATABASE graph_test;")
    await db.close()

@pytest.mark.integration
@pytest.mark.asyncio
async def test_full_graph_workflow(test_db):
    """Test complete graph workflow."""
    # Setup schema
    await test_db.query("""
        DEFINE TABLE person SCHEMAFULL;
        DEFINE FIELD name ON TABLE person TYPE string;
        DEFINE INDEX person_name ON TABLE person COLUMNS name;

        DEFINE TABLE follows SCHEMAFULL;
        DEFINE FIELD in ON TABLE follows TYPE record<person>;
        DEFINE FIELD out ON TABLE follows TYPE record<person>;
    """)

    # Create nodes
    await test_db.query("""
        CREATE person:alice SET name = 'Alice';
        CREATE person:bob SET name = 'Bob';
    """)

    # Create relationship
    await test_db.query(
        "RELATE person:alice->follows->person:bob"
    )

    # Query relationship
    result = await test_db.query(
        "SELECT ->follows->person.name FROM person:alice"
    )

    assert 'Bob' in str(result)

パフォーマンステスト

# tests/performance/test_graph_performance.py
import pytest
import time

@pytest.mark.slow
@pytest.mark.asyncio
async def test_traversal_performance(test_db):
    """Test that traversal completes within time limit."""
    # Setup large graph
    await test_db.query("""
        FOR $i IN 1..100 {
            CREATE person SET name = $i;
        };
        FOR $i IN 1..99 {
            RELATE type::thing('person', $i)->follows->type::thing('person', $i + 1);
        };
    """)

    start = time.time()

    # Run bounded traversal
    result = await test_db.query(
        "SELECT ->follows[..5]->person.* FROM person:1"
    )

    elapsed = time.time() - start

    # Should complete in under 100ms
    assert elapsed < 0.1, f"Traversal took {elapsed}s"

    # Should return limited results
    assert len(result[0]['result']) <= 5

---

8. セキュリティ

8.1 アクセス制御

-- Row-level security on nodes
DEFINE TABLE document SCHEMAFULL
    PERMISSIONS
        FOR select WHERE public = true OR owner = $auth.id
        FOR create WHERE $auth.id != NONE
        FOR update, delete WHERE owner = $auth.id;

-- Relationship permissions
DEFINE TABLE friendship SCHEMAFULL
    PERMISSIONS
        FOR select WHERE in = $auth.id OR out = $auth.id
        FOR create WHERE in = $auth.id
        FOR delete WHERE in = $auth.id OR out = $auth.id;

-- Prevent unauthorized traversal
DEFINE TABLE follows SCHEMAFULL
    PERMISSIONS
        FOR select WHERE in.public = true OR in.id = $auth.id;

8.2 インジェクションの防止

-- SECURE: Parameterized queries
LET $person_id = "person:alice";
SELECT ->follows->person.* FROM $person_id;

-- With SDK
const result = await db.query(
    'SELECT ->follows->person.* FROM $person',
    { person: `person:${userId}` }
);

-- VULNERABLE: String concatenation
-- const query = `SELECT * FROM person:${userInput}`;

8.3 クエリの深さ制限

-- SAFE: Bounded traversal
SELECT ->follows[..3]->person.* FROM person:alice;

-- SAFE: Limit results
SELECT ->follows->person.* FROM person:alice LIMIT 100;

-- DANGEROUS: Unbounded traversal
-- SELECT ->follows->person.* FROM person:alice;
-- Could traverse millions of nodes!

8.4 データの露出

-- Filter sensitive data in traversals
SELECT
    name,
    ->follows->person.{name, public_bio} AS following
FROM person:alice;

-- DON'T: Expose all fields in traversal
-- SELECT ->follows->person.* FROM person:alice;
-- May include email, phone, private data

---

9. よくある間違い

間違い 1：上限のないグラフ走査

-- DON'T: No depth limit
SELECT ->follows->person.* FROM person:alice;
-- Could traverse entire social network!

-- DO: Set depth limits
SELECT ->follows[..2]->person.* FROM person:alice;
SELECT ->follows[1..3]->person.* FROM person:alice LIMIT 100;

---

間違い 2：走査経路にインデックスがない

-- DON'T: Query without indexes
SELECT * FROM person WHERE email = '[email protected]';
-- Full table scan!

-- DO: Create indexes
DEFINE INDEX email_idx ON TABLE person COLUMNS email UNIQUE;
DEFINE INDEX name_idx ON TABLE person COLUMNS name;

-- Index edge properties used in filters
DEFINE INDEX works_at_role ON TABLE works_at COLUMNS role;

---

間違い 3：リレーションシップの方向が間違っている

-- Inefficient: Traversing against primary direction
SELECT <-authored<-post WHERE author = person:alice;

-- Better: Traverse with primary direction
SELECT ->authored->post.* FROM person:alice;

-- Design rule: Model edges in the direction of common queries

---

間違い 4：グラフでの N+1 クエリパターン

-- DON'T: Multiple round trips
SELECT * FROM person;
-- Then for each person:
SELECT * FROM post WHERE author = person:1;

-- DO: Single graph traversal
SELECT *, ->authored->post.* FROM person;

---

間違い 5：リレーションシップデータの過剰な正規化

-- DON'T: Over-normalize simple properties
-- Separate table for single property
DEFINE TABLE person_email;

-- DO: Embed simple properties
DEFINE TABLE person;
DEFINE FIELD email ON TABLE person TYPE string;

-- Use relationships for:
-- - Many-to-many associations
-- - Entities with independent lifecycle
-- - Rich metadata on relationships

---

間違い 6：循環を処理していない

-- Circular references can cause issues
-- Example: A follows B, B follows C, C follows A

-- Set depth limit to prevent infinite loops
SELECT ->follows[..5]->person.* FROM person:alice;

-- Track visited nodes in application logic
-- Use cycle detection in graph algorithms

---

間違い 7：クエリの実行計画を無視している

-- Always check query plans for slow queries
-- (Database-specific syntax)

-- SurrealDB: Monitor query performance
-- Neo4j: EXPLAIN / PROFILE
-- EXPLAIN SELECT ->follows->person.* FROM person:alice;

-- Look for:
-- - Full table scans
-- - Missing indexes
-- - Cartesian products
-- - Excessive traversal depth

---

10. 実装前チェックリスト

フェーズ 1：コードを書く前

• グラフ要件について PRD のセクションを読む
• エンティティ（ノード）とリレーションシップ（エッジ）を特定する
• クエリパターンに基づいてスキーマを設計する
• 頻繁にクエリされるプロパティのインデックスを計画する
• 走査の深さ制限を決める
• セキュリティ要件（権限、データの露出）を確認する
• 期待されるクエリ動作を検証する失敗テストを書く

フェーズ 2：実装中

• パラメーター化クエリを使用する（インジェクションを防止）
• すべての走査に深さ制限を設定する
• 大きな結果セットにはページネーションを実装する
• 頻繁に実行されるクエリにキャッシュを追加する
• 一括挿入にはバッチ操作を使用する
• 実行計画を使ってクエリのパフォーマンスを監視する
• 走査結果から機密フィールドを除外する

フェーズ 3：コミット前

• すべてのグラフクエリテストに合格している
• 実データベースを使用した統合テストに合格している
• パフォーマンステストがレイテンシ要件を満たしている
• コードベースに上限のない走査がない
• クエリ対象のすべてのプロパティにインデックスがある
• データの露出に関するセキュリティレビューが完了している
• スキーマの変更に合わせてドキュメントが更新されている

---

12. まとめ

グラフデータベースのエキスパートとして、次の点に注力してください。

1. グラフモデリング - エンティティをノード、リレーションシップをエッジ、接続を型付きで表現する
2. クエリの最適化 - インデックス、深さ制限、実行計画、効率的な走査
3. リレーションシップの設計 - 双方向エッジ、時系列データ、重み付き接続
4. パフォーマンス - N+1 の回避、上限付きの走査、適切なインデックス設定
5. セキュリティ - 行レベルの権限、インジェクションの防止、データの露出

主な原則：

• まずクエリをモデル化し、その後でグラフスキーマを設計する
• 再帰的な走査には必ず深さ制限を設定する
• JOIN や複数のクエリの代わりにグラフ走査を使用する
• ノードとエッジの両方のプロパティにインデックスを付ける
• エッジにメタデータ（タイムスタンプ、重み、プロパティ）を追加する
• 一般的なクエリに基づいてリレーションシップの方向を設計する
• 実行計画を使ってクエリのパフォーマンスを監視する

グラフデータベースのリソース：

• SurrealDB のドキュメント：https://surrealdb.com/docs
• Neo4j Graph Academy：https://neo4j.com/graphacademy/
• グラフデータベースの理論：https://neo4j.com/docs/getting-started/appendix/graphdb-concepts/

参考ドキュメント：

• クエリの最適化：references/query-optimization.md を参照
• モデリングガイド：references/modeling-guide.md を参照

グラフデータベースは、つながりのあるデータを扱うことに優れています。リレーションシップを第一級の要素としてモデル化し、走査演算子を活用することで、強力で効率的なクエリを実現できます。