diff --git a/code_review_graph/cli.py b/code_review_graph/cli.py
index 6e9349b6..638ef9e5 100644
--- a/code_review_graph/cli.py
+++ b/code_review_graph/cli.py
@@ -14,6 +14,8 @@
     code-review-graph register <path> [--alias name]
     code-review-graph unregister <path_or_alias>
     code-review-graph repos
+    code-review-graph query --sql <SELECT> [options]
+    code-review-graph query --file <path> [options]
 """
 
 from __future__ import annotations
@@ -103,6 +105,7 @@ def _print_banner() -> None:
     {g}postprocess{r} Run post-processing {d}(flows, communities, FTS){r}
     {g}eval{r}        Run evaluation benchmarks
     {g}serve{r}       Start MCP server {d}(stdio, or {g}--http{r} on localhost:5555){r}
+    {g}query{r}       Execute a raw SQL SELECT against the graph database
 
   {d}Run{r} {b}code-review-graph <command> --help{r} {d}for details{r}
 """)
@@ -285,6 +288,70 @@ def _cli_post_process(store: GraphStore) -> None:
         print(f"Communities: {pp['communities_detected']}")
 
 
+def _handle_query(args: argparse.Namespace) -> None:
+    """Execute a raw SQL SELECT against the graph database."""
+    from .incremental import find_project_root, get_db_path
+    from .query import _MAX_LIMIT, QueryError, format_table, run_query
+
+    # Resolve SQL source
+    if args.sql:
+        sql = args.sql
+    elif args.file:
+        try:
+            sql = Path(args.file).read_text(encoding="utf-8")
+        except OSError as exc:
+            print(json.dumps({"error": str(exc), "type": "file_error"}), file=sys.stderr)
+            sys.exit(1)
+    elif not sys.stdin.isatty():
+        sql = sys.stdin.read()
+    else:
+        print(
+            json.dumps(
+                {"error": "provide --sql, --file, or pipe SQL via stdin",
+                 "type": "missing_input"}
+            ),
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    # Parse --param key=value flags
+    params: dict[str, str] = {}
+    for p in args.param or []:
+        if "=" not in p:
+            print(
+                json.dumps(
+                    {"error": f"--param must be key=value format, got: {p!r}",
+                     "type": "invalid_param"}
+                ),
+                file=sys.stderr,
+            )
+            sys.exit(1)
+        k, _, v = p.partition("=")
+        params[k] = v
+
+    limit = min(max(1, args.limit), _MAX_LIMIT)
+
+    repo_root = Path(args.repo) if args.repo else find_project_root()
+    db_path = get_db_path(repo_root)
+
+    try:
+        rows = run_query(
+            db_path,
+            sql,
+            params=params or None,
+            limit=limit,
+            timeout=args.timeout,
+        )
+    except QueryError as exc:
+        print(json.dumps(exc.to_dict()), file=sys.stderr)
+        sys.exit(1)
+
+    if args.output_format == "table":
+        print(format_table(rows))
+    else:
+        print(json.dumps(rows, indent=2, default=str))
+
+
 def main() -> None:
     """Main CLI entry point."""
     ap = argparse.ArgumentParser(
@@ -486,6 +553,43 @@ def main() -> None:
     detect_cmd.add_argument("--brief", action="store_true", help="Show brief summary only")
     detect_cmd.add_argument("--repo", default=None, help="Repository root (auto-detected)")
 
+    # query
+    query_cmd = sub.add_parser(
+        "query", help="Execute a raw SQL SELECT against the graph database"
+    )
+    query_sql_group = query_cmd.add_mutually_exclusive_group()
+    query_sql_group.add_argument("--sql", default=None, help="SQL SELECT statement")
+    query_sql_group.add_argument("--file", default=None, help="Path to a .sql file")
+    query_cmd.add_argument(
+        "--param",
+        action="append",
+        default=[],
+        metavar="key=value",
+        help="Bind a named parameter (:key) in the SQL (repeatable)",
+    )
+    query_cmd.add_argument(
+        "--limit",
+        type=int,
+        default=100,
+        help="Maximum rows returned (default: 100, max: 1000)",
+    )
+    query_cmd.add_argument(
+        "--format",
+        choices=["json", "table"],
+        default="json",
+        dest="output_format",
+        help="Output format: json (default) | table",
+    )
+    query_cmd.add_argument(
+        "--timeout",
+        type=int,
+        default=10,
+        help="Query timeout in seconds (default: 10)",
+    )
+    query_cmd.add_argument(
+        "--repo", default=None, help="Repository root (default: current directory)"
+    )
+
     # serve
     serve_cmd = sub.add_parser(
         "serve",
@@ -624,6 +728,11 @@ def main() -> None:
                     print(f"  {entry['path']}{alias_str}")
         return
 
+    if args.command == "query":
+        logging.basicConfig(level=logging.WARNING, format="%(levelname)s: %(message)s")
+        _handle_query(args)
+        return
+
     logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
 
     from .graph import GraphStore
diff --git a/code_review_graph/query.py b/code_review_graph/query.py
new file mode 100644
index 00000000..2b07e7f2
--- /dev/null
+++ b/code_review_graph/query.py
@@ -0,0 +1,134 @@
+"""Read-only raw SQL query against the graph database."""
+
+from __future__ import annotations
+
+import sqlite3
+import threading
+from pathlib import Path
+from typing import Any
+from urllib.parse import quote
+
+import sqlglot
+from sqlglot import expressions as exp
+
+_DEFAULT_LIMIT = 100
+_MAX_LIMIT = 1000
+_DEFAULT_TIMEOUT = 10
+
+
+class QueryError(Exception):
+    """Structured error raised when a query violates safety rules or fails at runtime."""
+
+    def __init__(self, message: str, error_type: str) -> None:
+        super().__init__(message)
+        self.error_type = error_type
+
+    def to_dict(self) -> dict[str, str]:
+        return {"error": str(self), "type": self.error_type}
+
+
+def _validate_sql(sql: str) -> None:
+    """Raise QueryError if the SQL statement violates safety constraints.
+
+    Uses sqlglot to parse into an AST so string literals, double-quoted
+    identifiers, and comments can never fool the keyword check.
+    """
+    try:
+        statements = sqlglot.parse(sql, dialect="sqlite",
+                                   error_level=sqlglot.ErrorLevel.RAISE)
+    except sqlglot.errors.ParseError as exc:
+        raise QueryError(str(exc), "sql_syntax_error") from exc
+
+    if not statements:
+        raise QueryError("empty query", "disallowed_statement")
+
+    if len(statements) > 1:
+        raise QueryError("only a single SQL statement is permitted",
+                         "disallowed_statement")
+
+    stmt = statements[0]
+
+    if isinstance(stmt, exp.Pragma):
+        raise QueryError("PRAGMA statements are not permitted", "disallowed_pragma")
+
+    # exp.Select covers both plain SELECT and CTEs (WITH ... SELECT).
+    if not isinstance(stmt, exp.Select):
+        raise QueryError("only SELECT statements are permitted", "disallowed_statement")
+
+
+def run_query(
+    db_path: Path,
+    sql: str,
+    params: dict[str, Any] | None = None,
+    limit: int = _DEFAULT_LIMIT,
+    timeout: int = _DEFAULT_TIMEOUT,
+) -> list[dict[str, Any]]:
+    """Execute a read-only SELECT against the graph database.
+
+    Opens the database in read-only URI mode so writes are rejected at the
+    connection level, not just by statement inspection.
+
+    Raises:
+        QueryError: on safety violations, syntax errors, timeout, or missing DB.
+    """
+    limit = min(max(1, limit), _MAX_LIMIT)
+
+    _validate_sql(sql)
+
+    db_uri = f"file:{quote(str(db_path))}?mode=ro"
+    try:
+        conn = sqlite3.connect(db_uri, uri=True, check_same_thread=False)
+    except sqlite3.OperationalError as exc:
+        raise QueryError(str(exc), "db_error") from exc
+
+    conn.row_factory = sqlite3.Row
+    timed_out = threading.Event()
+
+    def _on_timeout() -> None:
+        timed_out.set()
+        conn.interrupt()
+
+    timer = threading.Timer(timeout, _on_timeout)
+    timer.start()
+    try:
+        cur = conn.execute(sql, params or {})
+        # Pull one extra row to detect truncation. We cap at the driver level
+        # rather than appending "LIMIT N" to the SQL so the user's query runs
+        # unmodified — appending LIMIT would break queries that already contain one.
+        rows = cur.fetchmany(limit + 1)
+        return [dict(row) for row in rows[:limit]]
+    except sqlite3.OperationalError as exc:
+        if timed_out.is_set():
+            raise QueryError(f"query exceeded timeout of {timeout}s", "timeout") from exc
+        raise QueryError(str(exc), "sql_error") from exc
+    except sqlite3.DatabaseError as exc:
+        raise QueryError(str(exc), "sql_syntax_error") from exc
+    finally:
+        timer.cancel()
+        conn.close()
+
+
+def format_table(rows: list[dict[str, Any]]) -> str:
+    """Format a list of row dicts as a pipe-delimited ASCII table."""
+    if not rows:
+        return "(no results)"
+
+    columns = list(rows[0].keys())
+
+    def _str(v: Any) -> str:
+        return "NULL" if v is None else str(v)
+
+    str_rows = [{col: _str(row.get(col)) for col in columns} for row in rows]
+    col_widths = {
+        col: max(len(col), max(len(r[col]) for r in str_rows))
+        for col in columns
+    }
+
+    header = " | ".join(col.ljust(col_widths[col]) for col in columns)
+    separator = "-+-".join("-" * col_widths[col] for col in columns)
+    data_lines = [
+        " | ".join(r[col].ljust(col_widths[col]) for col in columns)
+        for r in str_rows
+    ]
+
+    return "\n".join([header, separator, *data_lines])
diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md
index 7caae26a..15e94ae4 100644
--- a/docs/COMMANDS.md
+++ b/docs/COMMANDS.md
@@ -250,6 +250,15 @@ code-review-graph register <path> [--alias name]  # Register a repository
 code-review-graph unregister <path_or_alias>       # Remove from registry
 code-review-graph repos                            # List registered repositories
 
+# Raw SQL query
+code-review-graph query --sql "SELECT name, kind FROM nodes LIMIT 10"
+code-review-graph query --file query.sql                       # Read SQL from file
+code-review-graph query --sql "SELECT * FROM nodes WHERE name = :n" --param n=main
+code-review-graph query --sql "..." --format table             # ASCII table output
+code-review-graph query --sql "..." --limit 500                # Up to 1000 rows
+code-review-graph query --sql "..." --timeout 30               # Custom timeout (seconds)
+code-review-graph query --sql "..." --repo /path/to/repo       # Explicit repo root
+
 # Evaluation
 code-review-graph eval                         # Run evaluation benchmarks
 
diff --git a/pyproject.toml b/pyproject.toml
index d07edf19..28852ae0 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -34,6 +34,7 @@ dependencies = [
     "tree-sitter-language-pack>=0.3.0,<1",
     "networkx>=3.2,<4",
     "watchdog>=4.0.0,<6",
+    "sqlglot>=30.4.3",
 ]
 
 [project.urls]
diff --git a/tests/test_query.py b/tests/test_query.py
new file mode 100644
index 00000000..069fc3ec
--- /dev/null
+++ b/tests/test_query.py
@@ -0,0 +1,384 @@
+"""Tests for the raw SQL query command (code_review_graph/query.py)."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sqlite3
+import sys
+from pathlib import Path
+
+import pytest
+
+from code_review_graph.query import (
+    QueryError,
+    _MAX_LIMIT,
+    _validate_sql,
+    format_table,
+    run_query,
+)
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def fixture_db(tmp_path: Path) -> Path:
+    """Minimal graph DB with nodes and edges tables for query tests."""
+    db_path = tmp_path / "graph.db"
+    conn = sqlite3.connect(str(db_path))
+    conn.executescript("""
+        CREATE TABLE nodes (
+            id INTEGER PRIMARY KEY,
+            kind TEXT,
+            name TEXT,
+            qualified_name TEXT,
+            file_path TEXT,
+            line_start INTEGER,
+            line_end INTEGER,
+            language TEXT,
+            parent_name TEXT,
+            is_test INTEGER DEFAULT 0
+        );
+        CREATE TABLE edges (
+            id INTEGER PRIMARY KEY,
+            kind TEXT,
+            source_qualified TEXT,
+            target_qualified TEXT,
+            file_path TEXT,
+            line INTEGER
+        );
+    """)
+    conn.executemany(
+        """INSERT INTO nodes
+           (kind, name, qualified_name, file_path, line_start, line_end,
+            language, parent_name, is_test)
+           VALUES (?,?,?,?,?,?,?,?,?)""",
+        [
+            ("Function", "foo", "/a.py::foo", "/a.py", 1, 10, "python", None, 0),
+            ("Function", "bar", "/a.py::bar", "/a.py", 11, 20, "python", None, 0),
+            ("Function", "test_foo", "/test_a.py::test_foo", "/test_a.py", 1, 5,
+             "python", None, 1),
+        ],
+    )
+    conn.executemany(
+        """INSERT INTO edges (kind, source_qualified, target_qualified, file_path, line)
+           VALUES (?,?,?,?,?)""",
+        [
+            ("CALLS", "/a.py::bar", "/a.py::foo", "/a.py", 15),
+            ("CALLS", "/test_a.py::test_foo", "/a.py::foo", "/test_a.py", 3),
+        ],
+    )
+    conn.commit()
+    conn.close()
+    return db_path
+
+
+# ---------------------------------------------------------------------------
+# _validate_sql
+# ---------------------------------------------------------------------------
+
+
+class TestValidateSql:
+    def test_select_allowed(self):
+        _validate_sql("SELECT 1")
+
+    def test_select_with_whitespace_allowed(self):
+        _validate_sql("  \n  SELECT * FROM nodes")
+
+    def test_cte_allowed(self):
+        _validate_sql("WITH cte AS (SELECT 1) SELECT * FROM cte")
+
+    def test_insert_rejected(self):
+        with pytest.raises(QueryError) as exc:
+            _validate_sql("INSERT INTO nodes VALUES (1)")
+        assert exc.value.error_type == "disallowed_statement"
+
+    def test_update_rejected(self):
+        with pytest.raises(QueryError) as exc:
+            _validate_sql("UPDATE nodes SET name = 'x'")
+        assert exc.value.error_type == "disallowed_statement"
+
+    def test_delete_rejected(self):
+        with pytest.raises(QueryError):
+            _validate_sql("DELETE FROM nodes")
+
+    def test_create_rejected(self):
+        with pytest.raises(QueryError):
+            _validate_sql("CREATE TABLE t (id INTEGER)")
+
+    def test_drop_rejected(self):
+        with pytest.raises(QueryError):
+            _validate_sql("DROP TABLE nodes")
+
+    def test_alter_rejected(self):
+        with pytest.raises(QueryError):
+            _validate_sql("ALTER TABLE nodes ADD COLUMN x TEXT")
+
+    def test_attach_rejected(self):
+        with pytest.raises(QueryError):
+            _validate_sql("ATTACH DATABASE 'other.db' AS other")
+
+    def test_pragma_rejected(self):
+        with pytest.raises(QueryError) as exc:
+            _validate_sql("PRAGMA journal_mode")
+        assert exc.value.error_type == "disallowed_pragma"
+
+    def test_keyword_in_string_literal_allowed(self):
+        _validate_sql("SELECT name FROM nodes WHERE name = 'CREATE TABLE'")
+
+    def test_keyword_in_line_comment_allowed(self):
+        _validate_sql("SELECT 1 -- DROP TABLE here")
+
+    def test_keyword_in_block_comment_allowed(self):
+        _validate_sql("SELECT 1 /* INSERT INTO t VALUES (1) */")
+
+    def test_keyword_as_double_quoted_identifier_allowed(self):
+        # Double-quoted identifiers are valid SQLite column aliases.
+        # The regex approach would have false-positived here.
+        _validate_sql('SELECT name AS "CREATE" FROM nodes')
+
+
+# ---------------------------------------------------------------------------
+# run_query
+# ---------------------------------------------------------------------------
+
+
+class TestRunQuery:
+    def test_basic_select(self, fixture_db: Path):
+        rows = run_query(fixture_db, "SELECT name FROM nodes ORDER BY name")
+        assert len(rows) == 3
+        names = [r["name"] for r in rows]
+        assert "foo" in names and "bar" in names and "test_foo" in names
+
+    def test_column_names_present(self, fixture_db: Path):
+        rows = run_query(fixture_db, "SELECT name, kind FROM nodes LIMIT 1")
+        assert set(rows[0].keys()) == {"name", "kind"}
+
+    def test_param_binding(self, fixture_db: Path):
+        rows = run_query(
+            fixture_db,
+            "SELECT name FROM nodes WHERE kind = :kind AND is_test = 0",
+            params={"kind": "Function"},
+        )
+        names = [r["name"] for r in rows]
+        assert "test_foo" not in names
+        assert set(names) == {"foo", "bar"}
+
+    def test_limit_caps_rows(self, fixture_db: Path):
+        rows = run_query(fixture_db, "SELECT * FROM nodes", limit=2)
+        assert len(rows) == 2
+
+    def test_limit_max_enforced(self, fixture_db: Path):
+        rows = run_query(fixture_db, "SELECT * FROM nodes", limit=_MAX_LIMIT + 500)
+        assert len(rows) <= _MAX_LIMIT
+
+    def test_default_limit_applied(self, fixture_db: Path):
+        # 3 rows in fixture, all returned within default limit of 100
+        rows = run_query(fixture_db, "SELECT * FROM nodes")
+        assert len(rows) == 3
+
+    def test_cte_query_allowed(self, fixture_db: Path):
+        rows = run_query(
+            fixture_db,
+            "WITH fns AS (SELECT * FROM nodes WHERE kind = 'Function') "
+            "SELECT name FROM fns ORDER BY name",
+        )
+        assert len(rows) == 3
+
+    def test_join_query(self, fixture_db: Path):
+        rows = run_query(
+            fixture_db,
+            """
+            SELECT DISTINCT n.name
+            FROM nodes n
+            JOIN edges e ON e.target_qualified = n.qualified_name
+            WHERE e.kind = 'CALLS'
+            """,
+        )
+        assert any(r["name"] == "foo" for r in rows)
+
+    def test_insert_rejected(self, fixture_db: Path):
+        with pytest.raises(QueryError) as exc:
+            run_query(fixture_db, "INSERT INTO nodes (name) VALUES ('x')")
+        assert exc.value.error_type == "disallowed_statement"
+
+    def test_pragma_rejected(self, fixture_db: Path):
+        with pytest.raises(QueryError) as exc:
+            run_query(fixture_db, "PRAGMA journal_mode")
+        assert exc.value.error_type == "disallowed_pragma"
+
+    def test_nonexistent_db_raises_db_error(self, tmp_path: Path):
+        bad_path = tmp_path / "missing.db"
+        with pytest.raises(QueryError) as exc:
+            run_query(bad_path, "SELECT 1")
+        assert exc.value.error_type == "db_error"
+
+    def test_sql_syntax_error(self, fixture_db: Path):
+        with pytest.raises(QueryError) as exc:
+            run_query(fixture_db, "SELEKT * FROM nodes")
+        assert exc.value.error_type == "sql_syntax_error"
+
+    def test_runtime_sql_error(self, fixture_db: Path):
+        # Valid SELECT start but references a column that does not exist
+        with pytest.raises(QueryError) as exc:
+            run_query(fixture_db, "SELECT nonexistent_column FROM nodes")
+        assert exc.value.error_type == "sql_error"
+
+    def test_keyword_in_where_clause_allowed(self, fixture_db: Path):
+        rows = run_query(
+            fixture_db,
+            "SELECT name FROM nodes WHERE name = 'CREATE TABLE'",
+        )
+        assert rows == []
+
+    def test_empty_result(self, fixture_db: Path):
+        rows = run_query(fixture_db, "SELECT * FROM nodes WHERE 1 = 0")
+        assert rows == []
+
+    def test_null_value_in_row(self, fixture_db: Path):
+        # parent_name is NULL for top-level functions
+        rows = run_query(
+            fixture_db,
+            "SELECT name, parent_name FROM nodes WHERE name = 'foo'",
+        )
+        assert rows[0]["parent_name"] is None
+
+    def test_query_error_to_dict(self):
+        err = QueryError("oops", "sql_error")
+        d = err.to_dict()
+        assert d == {"error": "oops", "type": "sql_error"}
+
+
+# ---------------------------------------------------------------------------
+# format_table
+# ---------------------------------------------------------------------------
+
+
+class TestFormatTable:
+    def test_empty_returns_no_results(self):
+        assert format_table([]) == "(no results)"
+
+    def test_contains_column_names(self):
+        rows = [{"name": "foo", "kind": "Function"}]
+        out = format_table(rows)
+        assert "name" in out
+        assert "kind" in out
+
+    def test_contains_row_values(self):
+        rows = [{"name": "foo", "kind": "Function"}]
+        out = format_table(rows)
+        assert "foo" in out
+        assert "Function" in out
+
+    def test_null_rendered_as_NULL(self):
+        rows = [{"name": None, "kind": "Function"}]
+        out = format_table(rows)
+        assert "NULL" in out
+
+    def test_header_and_separator_same_width(self):
+        rows = [
+            {"col_a": "short", "col_b": "x"},
+            {"col_a": "a_much_longer_value", "col_b": "y"},
+        ]
+        lines = format_table(rows).split("\n")
+        # header, separator, then data rows
+        assert len(lines) == 4
+        assert len(lines[0]) == len(lines[1])
+
+    def test_multi_row(self):
+        rows = [{"n": "a"}, {"n": "b"}, {"n": "c"}]
+        lines = format_table(rows).split("\n")
+        # header + separator + 3 data rows = 5 lines
+        assert len(lines) == 5
+
+    def test_output_structure(self):
+        rows = [{"a": "1", "b": "hello"}]
+        lines = format_table(rows).split("\n")
+        # header row contains column names
+        assert "a" in lines[0] and "b" in lines[0]
+        # separator row contains only dashes and -+- joins, no letters
+        assert all(c in "-+ " for c in lines[1])
+        # data row contains the actual values
+        assert "1" in lines[2] and "hello" in lines[2]
+
+
+# ---------------------------------------------------------------------------
+# _handle_query (CLI layer)
+# ---------------------------------------------------------------------------
+
+
+def _make_args(repo_root: Path, sql: str, output_format: str = "json", **kwargs) -> argparse.Namespace:
+    defaults = dict(
+        sql=sql,
+        file=None,
+        param=[],
+        limit=100,
+        timeout=10,
+        repo=str(repo_root),
+        output_format=output_format,
+    )
+    defaults.update(kwargs)
+    return argparse.Namespace(**defaults)
+
+
+@pytest.fixture
+def cli_db(tmp_path: Path) -> Path:
+    """Fixture DB placed where get_db_path() expects it: <root>/.code-review-graph/graph.db."""
+    data_dir = tmp_path / ".code-review-graph"
+    data_dir.mkdir()
+    db_path = data_dir / "graph.db"
+    conn = sqlite3.connect(str(db_path))
+    conn.executescript("""
+        CREATE TABLE nodes (
+            id INTEGER PRIMARY KEY,
+            kind TEXT, name TEXT, qualified_name TEXT,
+            file_path TEXT, line_start INTEGER, line_end INTEGER,
+            language TEXT, parent_name TEXT, is_test INTEGER DEFAULT 0
+        );
+    """)
+    conn.executemany(
+        "INSERT INTO nodes (kind, name, qualified_name, file_path, line_start, line_end, "
+        "language, parent_name, is_test) VALUES (?,?,?,?,?,?,?,?,?)",
+        [
+            ("Function", "foo", "/a.py::foo", "/a.py", 1, 10, "python", None, 0),
+            ("Function", "bar", "/a.py::bar", "/a.py", 11, 20, "python", None, 0),
+        ],
+    )
+    conn.commit()
+    conn.close()
+    return tmp_path
+
+
+class TestHandleQuery:
+    def test_json_output_is_valid_json(self, cli_db: Path, capsys):
+        from code_review_graph.cli import _handle_query
+
+        _handle_query(_make_args(cli_db, "SELECT name FROM nodes ORDER BY id"))
+        out = capsys.readouterr().out
+        data = json.loads(out)
+        assert isinstance(data, list)
+
+    def test_json_output_contains_row_values(self, cli_db: Path, capsys):
+        from code_review_graph.cli import _handle_query
+
+        _handle_query(_make_args(cli_db, "SELECT name FROM nodes WHERE name = 'foo'"))
+        data = json.loads(capsys.readouterr().out)
+        assert data == [{"name": "foo"}]
+
+    def test_json_output_all_rows(self, cli_db: Path, capsys):
+        from code_review_graph.cli import _handle_query
+
+        _handle_query(_make_args(cli_db, "SELECT name FROM nodes ORDER BY id"))
+        data = json.loads(capsys.readouterr().out)
+        assert [r["name"] for r in data] == ["foo", "bar"]
+
+    def test_table_output_renders_values(self, cli_db: Path, capsys):
+        from code_review_graph.cli import _handle_query
+
+        _handle_query(_make_args(cli_db, "SELECT name FROM nodes WHERE name = 'foo'",
+                                 output_format="table"))
+        out = capsys.readouterr().out
+        assert "name" in out and "foo" in out
diff --git a/uv.lock b/uv.lock
index 4f8e3192..979ac933 100644
--- a/uv.lock
+++ b/uv.lock
@@ -338,6 +338,7 @@ dependencies = [
     { name = "mcp" },
     { name = "networkx", version = "3.4.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
     { name = "networkx", version = "3.6.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
+    { name = "sqlglot" },
     { name = "tree-sitter" },
     { name = "tree-sitter-language-pack" },
     { name = "watchdog" },
@@ -410,6 +411,7 @@ requires-dist = [
     { name = "pyyaml", marker = "extra == 'eval'", specifier = ">=6.0" },
     { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.3.0,<1" },
     { name = "sentence-transformers", marker = "extra == 'embeddings'", specifier = ">=3.0.0,<4" },
+    { name = "sqlglot", specifier = ">=30.4.3" },
     { name = "tomli", marker = "python_full_version < '3.11' and extra == 'dev'", specifier = ">=2.0" },
     { name = "tree-sitter", specifier = ">=0.23.0,<1" },
     { name = "tree-sitter-language-pack", specifier = ">=0.3.0,<1" },
@@ -3438,6 +3440,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" },
 ]
 
+[[package]]
+name = "sqlglot"
+version = "30.4.3"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/9b/af8fcaca1b0821349ef4f88e2775e059bcd7640900bca6533832f1fb845d/sqlglot-30.4.3.tar.gz", hash = "sha256:3a4e9a1e1dd47f8e536ba822d77cb784681704da5e4a3e1a07d2ef86b6067826", size = 5827662, upload-time = "2026-04-13T17:05:15.456Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ee/19/7df8b292accba3bc0de92c611c1e89423b25c08c82c18b14ca1fdbcf6e44/sqlglot-30.4.3-py3-none-any.whl", hash = "sha256:58ea8e723444569da5cec91e4c8f16e385bce3f0ce0374b8c722c3088e1c1c7a", size = 670965, upload-time = "2026-04-13T17:05:13.128Z" },
+]
+
 [[package]]
 name = "sse-starlette"
 version = "3.3.3"