OQL 解析器实现：从文本到 AST
Tags: #OQL #Parser #AST #Lexer #RecursiveDescent #智策平台
Coomia发布于 2025年7月16日19 分钟阅读
分享本文Twitter / X
“系列：S3 数据基座 · 第 7 篇 | 难度：高级 | 阅读时间：20 分钟
OQL 解析器实现：从文本到 AST
Tags: #OQL #Parser #AST #Lexer #RecursiveDescent #智策平台
#TL;DR
上一篇定义了 OQL 的完整语法，本文深入解析器实现：如何将一段 OQL 文本转换为结构化的抽象语法树（AST）。我们采用手写递归下降解析器而非 ANTLR/PEG 生成器，以获得更精细的错误恢复和更好的性能。全文覆盖词法分析器（Lexer）的 Token 设计、递归下降解析器（Parser）的核心算法、AST 节点类型体系、语义分析阶段（类型检查与名称解析）、以及友好的错误报告机制。通过完整的代码示例展示从 FETCH Person WHERE age > 30 到可执行查询计划的全链路。
#1. 解析器架构总览
#1.1 三阶段处理流水线
Code
OQL 解析器处理流水线：
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ OQL 文本 │───→│ 词法分析 │───→│ 语法分析 │───→│ 语义分析 │
│ │ │ (Lexer) │ │ (Parser) │ │ (Analyzer)│
└──────────┘ └──────────┘ └──────────┘ └──────────┘
│ │ │
▼ ▼ ▼
Token Stream Raw AST Typed AST
│
▼
┌──────────┐
│ 查询计划 │
│ (Plan) │
└──────────┘
#1.2 为什么手写而非生成
Code
手写解析器 vs 生成器对比：
┌─────────────────┬──────────────────┬──────────────────┐
│ 维度 │ 手写递归下降 │ ANTLR/PEG 生成 │
├─────────────────┼──────────────────┼──────────────────┤
│ 错误恢复 │ 精细控制 │ 有限 │
│ 错误消息质量 │ 高度定制 │ 模板化 │
│ 性能 │ 极致优化 │ 通用开销 │
│ 调试难度 │ 直接 step-through │ 生成代码难读 │
│ 增量解析 │ 容易实现 │ 需要额外框架 │
│ 依赖 │ 零依赖 │ 运行时依赖 │
│ 学习曲线 │ 需要编译原理基础 │ 需要学习工具链 │
│ 维护成本 │ 中等 │ 低（语法文件即文档） │
└─────────────────┴──────────────────┴──────────────────┘
结论：OQL 语法复杂度适中（~50 产生式），手写解析器在
错误体验和性能方面的优势远超维护成本。
#2. 词法分析器（Lexer）
#2.1 Token 类型定义
Python
from enum import Enum, auto
from dataclasses import dataclass
from typing import Optional
class TokenType(Enum):
"""OQL 词法单元类型"""
# 关键字
FETCH = auto()
TRAVERSE = auto()
AGGREGATE = auto()
TIMELINE = auto()
DIFF = auto()
WHERE = auto()
WITH = auto()
METRIC = auto()
AT = auto()
TIME = auto()
BRANCH = auto()
ORDER = auto()
BY = auto()
LIMIT = auto()
OFFSET = auto()
AS = auto()
AND = auto()
OR = auto()
NOT = auto()
IN = auto()
BETWEEN = auto()
LIKE = auto()
IS = auto()
NULL = auto()
TRUE = auto()
FALSE = auto()
SELECT = auto()
GROUP = auto()
HAVING = auto()
# 标识符和字面量
IDENTIFIER = auto()
STRING_LITERAL = auto()
NUMBER_LITERAL = auto()
FLOAT_LITERAL = auto()
DATETIME_LITERAL = auto()
# 运算符
EQ = auto() # =
NEQ = auto() # !=
LT = auto() # <
GT = auto() # >
LTE = auto() # <=
GTE = auto() # >=
ARROW = auto() # ->
DOT = auto() # .
STAR = auto() # *
PLUS = auto() # +
MINUS = auto() # -
SLASH = auto() # /
PERCENT = auto() # %
# 分隔符
LPAREN = auto() # (
RPAREN = auto() # )
LBRACKET = auto() # [
RBRACKET = auto() # ]
COMMA = auto() # ,
SEMICOLON = auto() # ;
COLON = auto() # :
# 特殊
EOF = auto()
ERROR = auto()
@dataclass(frozen=True)
class Token:
"""词法单元"""
type: TokenType
value: str
line: int
column: int
offset: int # 在源文本中的字节偏移
@property
def span(self) -> tuple[int, int]:
return (self.offset, self.offset + len(self.value))
#2.2 Lexer 核心实现
Python
class OQLLexer:
"""OQL 词法分析器 — 零依赖、单遍扫描"""
KEYWORDS: dict[str, TokenType] = {
'FETCH': TokenType.FETCH,
'TRAVERSE': TokenType.TRAVERSE,
'AGGREGATE': TokenType.AGGREGATE,
'TIMELINE': TokenType.TIMELINE,
'DIFF': TokenType.DIFF,
'WHERE': TokenType.WHERE,
'WITH': TokenType.WITH,
'METRIC': TokenType.METRIC,
'AT': TokenType.AT,
'TIME': TokenType.TIME,
'BRANCH': TokenType.BRANCH,
'ORDER': TokenType.ORDER,
'BY': TokenType.BY,
'LIMIT': TokenType.LIMIT,
'OFFSET': TokenType.OFFSET,
'AS': TokenType.AS,
'AND': TokenType.AND,
'OR': TokenType.OR,
'NOT': TokenType.NOT,
'IN': TokenType.IN,
'BETWEEN': TokenType.BETWEEN,
'LIKE': TokenType.LIKE,
'IS': TokenType.IS,
'NULL': TokenType.NULL,
'TRUE': TokenType.TRUE,
'FALSE': TokenType.FALSE,
'SELECT': TokenType.SELECT,
'GROUP': TokenType.GROUP,
'HAVING': TokenType.HAVING,
}
def __init__(self, source: str):
self._source = source
self._pos = 0
self._line = 1
self._column = 1
self._tokens: list[Token] = []
def tokenize(self) -> list[Token]:
while self._pos < len(self._source):
self._skip_whitespace_and_comments()
if self._pos >= len(self._source):
break
ch = self._source[self._pos]
if ch.isalpha() or ch == '_':
self._read_identifier_or_keyword()
elif ch.isdigit():
self._read_number()
elif ch == "'":
self._read_string()
elif ch == '-' and self._peek(1) == '>':
self._emit(TokenType.ARROW, '->', 2)
elif ch == '!' and self._peek(1) == '=':
self._emit(TokenType.NEQ, '!=', 2)
elif ch == '<' and self._peek(1) == '=':
self._emit(TokenType.LTE, '<=', 2)
elif ch == '>' and self._peek(1) == '=':
self._emit(TokenType.GTE, '>=', 2)
else:
self._read_single_char(ch)
self._tokens.append(Token(
TokenType.EOF, '', self._line, self._column, self._pos
))
return self._tokens
#2.3 关键字大小写不敏感
Python
def _read_identifier_or_keyword(self):
start = self._pos
start_col = self._column
while self._pos < len(self._source) and (
self._source[self._pos].isalnum() or self._source[self._pos] == '_'
):
self._advance()
value = self._source[start:self._pos]
upper = value.upper()
# 关键字匹配（大小写不敏感）
if upper in self.KEYWORDS:
token_type = self.KEYWORDS[upper]
else:
token_type = TokenType.IDENTIFIER
self._tokens.append(Token(token_type, value, self._line, start_col, start))
#3. 递归下降解析器（Parser）
#3.1 AST 节点类型体系
Python
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Union
class ASTNode(ABC):
"""AST 节点基类"""
@abstractmethod
def accept(self, visitor: 'ASTVisitor') -> any:
...
@dataclass
class FetchStatement(ASTNode):
"""FETCH 语句节点"""
entity_type: str
properties: list[str] | None # None 表示 SELECT *
where_clause: 'Expression | None'
with_metrics: list['MetricRef']
at_clause: 'AtClause | None'
order_by: list['OrderByItem']
limit: int | None
offset: int | None
def accept(self, visitor):
return visitor.visit_fetch(self)
@dataclass
class TraverseStatement(ASTNode):
"""TRAVERSE 语句节点"""
start_type: str
path_segments: list['PathSegment']
where_clause: 'Expression | None'
max_depth: int | None
return_mode: str # 'NODES' | 'PATHS' | 'SUBGRAPH'
def accept(self, visitor):
return visitor.visit_traverse(self)
@dataclass
class AggregateStatement(ASTNode):
"""AGGREGATE 语句节点"""
entity_type: str
group_by: list[str]
aggregations: list['AggregationExpr']
where_clause: 'Expression | None'
having_clause: 'Expression | None'
def accept(self, visitor):
return visitor.visit_aggregate(self)
@dataclass
class TimelineStatement(ASTNode):
"""TIMELINE 语句节点"""
entity_type: str
entity_filter: 'Expression | None'
time_range: 'TimeRange | None'
event_types: list[str]
granularity: str | None
def accept(self, visitor):
return visitor.visit_timeline(self)
@dataclass
class DiffStatement(ASTNode):
"""DIFF 语句节点"""
entity_type: str
from_ref: 'BranchOrTimeRef'
to_ref: 'BranchOrTimeRef'
properties: list[str] | None
def accept(self, visitor):
return visitor.visit_diff(self)
#3.2 表达式节点
Python
@dataclass
class BinaryExpr(ASTNode):
"""二元表达式"""
left: 'Expression'
operator: str # '=', '!=', '<', '>', '<=', '>=', 'AND', 'OR'
right: 'Expression'
def accept(self, visitor):
return visitor.visit_binary(self)
@dataclass
class UnaryExpr(ASTNode):
"""一元表达式"""
operator: str # 'NOT', '-'
operand: 'Expression'
def accept(self, visitor):
return visitor.visit_unary(self)
@dataclass
class PropertyRef(ASTNode):
"""属性引用"""
entity_type: str | None # 可选限定
property_name: str
def accept(self, visitor):
return visitor.visit_property_ref(self)
@dataclass
class LiteralExpr(ASTNode):
"""字面量"""
value: str | int | float | bool | None
literal_type: str # 'string', 'int', 'float', 'bool', 'null', 'datetime'
def accept(self, visitor):
return visitor.visit_literal(self)
@dataclass
class FunctionCall(ASTNode):
"""函数调用"""
function_name: str
arguments: list['Expression']
def accept(self, visitor):
return visitor.visit_function_call(self)
# 类型别名
Expression = Union[BinaryExpr, UnaryExpr, PropertyRef, LiteralExpr, FunctionCall]
#3.3 解析器核心逻辑
Python
class OQLParser:
"""OQL 递归下降解析器"""
def __init__(self, tokens: list[Token]):
self._tokens = tokens
self._pos = 0
self._errors: list[ParseError] = []
def parse(self) -> ASTNode:
"""解析入口"""
token = self._current()
if token.type == TokenType.FETCH:
return self._parse_fetch()
elif token.type == TokenType.TRAVERSE:
return self._parse_traverse()
elif token.type == TokenType.AGGREGATE:
return self._parse_aggregate()
elif token.type == TokenType.TIMELINE:
return self._parse_timeline()
elif token.type == TokenType.DIFF:
return self._parse_diff()
else:
raise self._error(
f"Expected statement keyword (FETCH, TRAVERSE, AGGREGATE, "
f"TIMELINE, DIFF), found '{token.value}'",
token
)
def _parse_fetch(self) -> FetchStatement:
"""
FETCH EntityType
[SELECT prop1, prop2, ...]
[WHERE condition]
[WITH METRIC metric1, metric2, ...]
[AT TIME datetime | AT BRANCH name]
[ORDER BY prop [ASC|DESC], ...]
[LIMIT n [OFFSET m]]
"""
self._expect(TokenType.FETCH)
entity_type = self._expect(TokenType.IDENTIFIER).value
# 可选 SELECT 子句
properties = None
if self._match(TokenType.SELECT):
properties = self._parse_property_list()
# 可选 WHERE 子句
where_clause = None
if self._match(TokenType.WHERE):
where_clause = self._parse_expression()
# 可选 WITH METRIC 子句
with_metrics = []
if self._match(TokenType.WITH):
self._expect(TokenType.METRIC)
with_metrics = self._parse_metric_list()
# 可选 AT 子句
at_clause = None
if self._match(TokenType.AT):
at_clause = self._parse_at_clause()
# 可选 ORDER BY 子句
order_by = []
if self._match(TokenType.ORDER):
self._expect(TokenType.BY)
order_by = self._parse_order_by_list()
# 可选 LIMIT / OFFSET
limit = None
offset = None
if self._match(TokenType.LIMIT):
limit = int(self._expect(TokenType.NUMBER_LITERAL).value)
if self._match(TokenType.OFFSET):
offset = int(self._expect(TokenType.NUMBER_LITERAL).value)
return FetchStatement(
entity_type=entity_type,
properties=properties,
where_clause=where_clause,
with_metrics=with_metrics,
at_clause=at_clause,
order_by=order_by,
limit=limit,
offset=offset,
)
#3.4 表达式解析（Pratt Parsing）
Python
def _parse_expression(self, min_precedence: int = 0) -> Expression:
"""Pratt 解析法处理运算符优先级"""
left = self._parse_primary()
while True:
token = self._current()
prec = self._get_precedence(token)
if prec <= min_precedence:
break
operator = self._advance().value
right = self._parse_expression(prec)
left = BinaryExpr(left=left, operator=operator, right=right)
return left
PRECEDENCE_TABLE = {
'OR': 1,
'AND': 2,
'NOT': 3,
'=': 4, '!=': 4,
'<': 5, '>': 5, '<=': 5, '>=': 5,
'IN': 5, 'BETWEEN': 5, 'LIKE': 5, 'IS': 5,
'+': 6, '-': 6,
'*': 7, '/': 7, '%': 7,
}
def _parse_primary(self) -> Expression:
"""解析原子表达式"""
token = self._current()
if token.type == TokenType.LPAREN:
self._advance()
expr = self._parse_expression()
self._expect(TokenType.RPAREN)
return expr
if token.type == TokenType.NOT:
self._advance()
operand = self._parse_expression(self.PRECEDENCE_TABLE['NOT'])
return UnaryExpr(operator='NOT', operand=operand)
if token.type == TokenType.MINUS:
self._advance()
operand = self._parse_primary()
return UnaryExpr(operator='-', operand=operand)
if token.type == TokenType.NUMBER_LITERAL:
self._advance()
return LiteralExpr(value=int(token.value), literal_type='int')
if token.type == TokenType.FLOAT_LITERAL:
self._advance()
return LiteralExpr(value=float(token.value), literal_type='float')
if token.type == TokenType.STRING_LITERAL:
self._advance()
return LiteralExpr(value=token.value, literal_type='string')
if token.type in (TokenType.TRUE, TokenType.FALSE):
self._advance()
return LiteralExpr(value=token.type == TokenType.TRUE, literal_type='bool')
if token.type == TokenType.NULL:
self._advance()
return LiteralExpr(value=None, literal_type='null')
if token.type == TokenType.IDENTIFIER:
return self._parse_identifier_or_function()
raise self._error(f"Unexpected token '{token.value}'", token)
#4. 语义分析阶段
#4.1 名称解析
Python
class NameResolver(ASTVisitor):
"""名称解析器：将标识符绑定到 Ontology Schema"""
def __init__(self, schema_registry: SchemaRegistry):
self._registry = schema_registry
self._errors: list[SemanticError] = []
def visit_fetch(self, node: FetchStatement):
# 验证 Entity Type 存在
entity_def = self._registry.get_entity_type(node.entity_type)
if entity_def is None:
candidates = self._registry.suggest_entity_type(node.entity_type)
self._errors.append(SemanticError(
f"Unknown entity type '{node.entity_type}'",
suggestions=candidates,
node=node
))
return
# 验证属性存在
if node.properties:
for prop in node.properties:
if prop not in entity_def.properties:
candidates = entity_def.suggest_property(prop)
self._errors.append(SemanticError(
f"Unknown property '{prop}' on type '{node.entity_type}'",
suggestions=candidates,
node=node
))
# 验证 WHERE 子句中的属性引用
if node.where_clause:
self._resolve_expression(node.where_clause, entity_def)
# 验证 METRIC 引用
for metric in node.with_metrics:
metric_def = self._registry.get_metric(metric.name)
if metric_def is None:
self._errors.append(SemanticError(
f"Unknown metric '{metric.name}'",
node=node
))
#4.2 类型检查
Python
class TypeChecker(ASTVisitor):
"""类型检查器：验证表达式类型兼容性"""
def visit_binary(self, node: BinaryExpr) -> OQLType:
left_type = node.left.accept(self)
right_type = node.right.accept(self)
if node.operator in ('AND', 'OR'):
if left_type != OQLType.BOOLEAN or right_type != OQLType.BOOLEAN:
self._error(
f"Operator '{node.operator}' requires boolean operands, "
f"got {left_type} and {right_type}",
node
)
return OQLType.BOOLEAN
if node.operator in ('=', '!='):
if not self._types_comparable(left_type, right_type):
self._error(
f"Cannot compare {left_type} with {right_type}",
node
)
return OQLType.BOOLEAN
if node.operator in ('<', '>', '<=', '>='):
if not self._types_orderable(left_type, right_type):
self._error(
f"Cannot order-compare {left_type} with {right_type}",
node
)
return OQLType.BOOLEAN
if node.operator in ('+', '-', '*', '/'):
return self._numeric_promotion(left_type, right_type, node)
raise ValueError(f"Unknown operator: {node.operator}")
#4.3 Schema 验证示例
Code
语义分析完整示例：
输入 OQL:
FETCH Person
SELECT name, age, department.name
WHERE age > 30 AND status = 'active'
WITH METRIC direct_reports, total_revenue
AT BRANCH 'feature-branch'
ORDER BY age DESC
LIMIT 100
名称解析结果:
✅ Person → EntityType(id='person', namespace='core')
✅ name → Property(name='name', type=STRING)
✅ age → Property(name='age', type=INT)
✅ department.name → Traversal(Person->BelongsTo->Department).name
✅ status → Property(name='status', type=STRING)
✅ direct_reports → Metric(id='direct_reports', return_type=INT)
✅ total_revenue → Metric(id='total_revenue', return_type=DECIMAL)
✅ 'feature-branch' → Branch(name='feature-branch')
类型检查结果:
✅ age > 30 : INT > INT → BOOLEAN
✅ status = 'active' : STRING = STRING → BOOLEAN
✅ ... AND ... : BOOLEAN AND BOOLEAN → BOOLEAN
✅ ORDER BY age : INT is orderable ✓
#5. 错误恢复机制
#5.1 恐慌模式恢复
Python
class ErrorRecovery:
"""解析器错误恢复策略"""
# 同步点：遇到这些 Token 停止跳过
SYNC_TOKENS = {
TokenType.FETCH, TokenType.TRAVERSE, TokenType.AGGREGATE,
TokenType.TIMELINE, TokenType.DIFF,
TokenType.WHERE, TokenType.WITH, TokenType.AT,
TokenType.ORDER, TokenType.LIMIT,
TokenType.SEMICOLON, TokenType.EOF,
}
@staticmethod
def synchronize(parser: 'OQLParser'):
"""恐慌模式：跳过 Token 直到找到同步点"""
parser._advance() # 跳过导致错误的 Token
while not parser._is_at_end():
if parser._current().type in ErrorRecovery.SYNC_TOKENS:
return
parser._advance()
#5.2 友好错误消息
Code
错误消息设计原则：
1. 指出错误位置（行、列、上下文）
2. 说明期望什么
3. 给出修正建议
示例输出：
Error at line 1, column 14:
FETCH Person WERE age > 30
^^^^
Expected 'WHERE', found 'WERE'.
Did you mean: WHERE
Error at line 2, column 3:
FETCH Person
WHERE ages > 30
^^^^
Unknown property 'ages' on type 'Person'.
Available properties: age (Int), name (String), email (String)
Did you mean: age
Error at line 1, column 26:
FETCH Person WHERE age > 'thirty'
^^^^^^^^
Type mismatch: cannot compare Int with String.
Property 'age' is of type Int, but got String literal 'thirty'.
#5.3 多错误收集
Python
class MultiErrorCollector:
"""收集所有错误而非在第一个错误时停止"""
def __init__(self, max_errors: int = 20):
self._errors: list[OQLError] = []
self._max_errors = max_errors
def report(self, error: OQLError):
self._errors.append(error)
if len(self._errors) >= self._max_errors:
raise TooManyErrorsException(self._errors)
def has_errors(self) -> bool:
return len(self._errors) > 0
def format_all(self, source: str) -> str:
"""格式化所有错误为用户友好的输出"""
lines = source.splitlines()
output = []
for error in self._errors:
output.append(f"\nError at line {error.line}, column {error.column}:")
if 0 < error.line <= len(lines):
output.append(f" {lines[error.line - 1]}")
output.append(f" {' ' * (error.column - 1)}{'^' * error.length}")
output.append(f" {error.message}")
if error.suggestions:
output.append(f" Did you mean: {', '.join(error.suggestions)}")
return '\n'.join(output)
#6. AST 到查询计划的转换
#6.1 逻辑计划生成
Python
class LogicalPlanGenerator(ASTVisitor):
"""将 Typed AST 转换为逻辑查询计划"""
def visit_fetch(self, node: FetchStatement) -> LogicalPlan:
# 基础扫描
plan = EntityScan(entity_type=node.entity_type)
# WHERE 过滤
if node.where_clause:
predicate = self._compile_expression(node.where_clause)
plan = Filter(child=plan, predicate=predicate)
# 属性投影
if node.properties:
plan = Project(child=plan, columns=node.properties)
# 指标展开
for metric in node.with_metrics:
plan = MetricExpansion(child=plan, metric=metric)
# 时间旅行/分支
if node.at_clause:
plan = TimeTravel(child=plan, ref=node.at_clause)
# 排序
if node.order_by:
plan = Sort(child=plan, order_by=node.order_by)
# 分页
if node.limit is not None:
plan = Limit(child=plan, limit=node.limit, offset=node.offset)
return plan
#6.2 逻辑计划到 SQL
Python
class DorisSQL Compiler(LogicalPlanVisitor):
"""将逻辑计划编译为 Doris SQL"""
def visit_entity_scan(self, node: EntityScan) -> str:
return (
f"SELECT * FROM entity_common "
f"WHERE entity_type = '{node.entity_type}'"
)
def visit_filter(self, node: Filter) -> str:
child_sql = node.child.accept(self)
predicate_sql = self._compile_predicate(node.predicate)
return f"SELECT * FROM ({child_sql}) t WHERE {predicate_sql}"
def visit_metric_expansion(self, node: MetricExpansion) -> str:
child_sql = node.child.accept(self)
metric_def = self._registry.get_metric(node.metric.name)
metric_sql = metric_def.compile_to_sql()
return (
f"SELECT t.*, ({metric_sql}) AS {node.metric.name} "
f"FROM ({child_sql}) t"
)
#6.3 完整编译示例
Code
编译流水线完整示例：
OQL 输入:
FETCH Person
WHERE age > 30 AND department.name = 'Engineering'
WITH METRIC direct_reports
ORDER BY age DESC
LIMIT 50
↓ Lexer
Token Stream:
[FETCH] [Person:ID] [WHERE] [age:ID] [>] [30:NUM]
[AND] [department:ID] [.] [name:ID] [=] ['Engineering':STR]
[WITH] [METRIC] [direct_reports:ID]
[ORDER] [BY] [age:ID] [DESC:ID]
[LIMIT] [50:NUM] [EOF]
↓ Parser
AST:
FetchStatement(
entity_type='Person',
where=BinaryExpr(AND,
BinaryExpr(>, PropertyRef('age'), Literal(30)),
BinaryExpr(=, PropertyRef('department.name'), Literal('Engineering'))
),
metrics=[MetricRef('direct_reports')],
order_by=[OrderBy('age', DESC)],
limit=50
)
↓ Semantic Analysis (Name Resolution + Type Check)
Typed AST:
(同上，但属性引用已绑定到 Schema 定义，类型已验证)
↓ Logical Plan
Limit(50)
└─ Sort(age DESC)
└─ MetricExpansion(direct_reports)
└─ Filter(age > 30 AND department.name = 'Engineering')
└─ EntityScan(Person)
↓ SQL Compilation
SELECT t2.*, (
SELECT COUNT(*) FROM entity_edge
WHERE source_id = t2.entity_id AND edge_type = 'Manages'
) AS direct_reports
FROM (
SELECT ec.* FROM entity_common ec
JOIN entity_edge ee ON ec.entity_id = ee.source_id
JOIN entity_common ec2 ON ee.target_id = ec2.entity_id
WHERE ec.entity_type = 'Person'
AND JSON_EXTRACT(ec.properties, '$.age') > 30
AND ee.edge_type = 'BelongsTo'
AND ec2.entity_type = 'Department'
AND JSON_EXTRACT(ec2.properties, '$.name') = 'Engineering'
) t2
ORDER BY JSON_EXTRACT(t2.properties, '$.age') DESC
LIMIT 50
#7. 增量解析与 IDE 支持
#7.1 增量解析策略
Code
增量解析：用户输入实时反馈
策略：
1. 将文档按语句分割为独立解析单元
2. 编辑时只重新解析受影响的语句
3. Token 级别的变更检测，最小化重解析范围
┌─────────────────────────────────────────┐
│ 文档 │
│ │
│ FETCH Person WHERE age > 30 ← 已缓存 │
│ ; │
│ TRAVERSE Person -> Company ← 正在编辑 │
│ ; │
│ AGGREGATE Device GROUP BY type ← 已缓存 │
└─────────────────────────────────────────┘
只重新解析第 2 条语句，其他使用缓存 AST
#7.2 自动补全支持
Python
class OQLCompletionProvider:
"""OQL 自动补全提供者"""
def complete(self, source: str, cursor_pos: int) -> list[Completion]:
# 解析到光标位置
tokens = self._lexer.tokenize(source[:cursor_pos])
context = self._analyze_context(tokens)
if context == CompletionContext.STATEMENT_START:
return [
Completion('FETCH', 'Query entities'),
Completion('TRAVERSE', 'Graph traversal'),
Completion('AGGREGATE', 'Aggregation query'),
Completion('TIMELINE', 'Event timeline'),
Completion('DIFF', 'Branch diff'),
]
if context == CompletionContext.ENTITY_TYPE:
return [
Completion(et.name, et.description)
for et in self._registry.list_entity_types()
]
if context == CompletionContext.PROPERTY:
entity_type = self._resolve_current_entity_type(tokens)
if entity_type:
return [
Completion(p.name, f"{p.type} - {p.description}")
for p in entity_type.properties.values()
]
if context == CompletionContext.METRIC:
return [
Completion(m.name, m.description)
for m in self._registry.list_metrics()
]
return []
#8. 性能基准测试
#8.1 解析性能
Code
OQL 解析器性能基准（单线程，Apple M2）：
┌───────────────────────────────┬──────────┬──────────┐
│ 测试用例 │ 解析时间 │ 吞吐量 │
├───────────────────────────────┼──────────┼──────────┤
│ 简单 FETCH（30 字符） │ 12 μs │ 83K qps │
│ 带 WHERE 的 FETCH（100 字符） │ 28 μs │ 36K qps │
│ 复杂 TRAVERSE（200 字符） │ 45 μs │ 22K qps │
│ 完整查询（500 字符） │ 89 μs │ 11K qps │
│ 大型聚合（1000 字符） │ 156 μs │ 6.4K qps │
└───────────────────────────────┴──────────┴──────────┘
对比（相同查询）：
ANTLR 4 生成的解析器：慢 3-5 倍
Python lark 库：慢 8-12 倍
手写 Lexer + Parser：最快
#8.2 内存占用
Code
AST 内存占用：
┌───────────────────┬──────────┬──────────┐
│ AST 节点类型 │ 单节点大小 │ 典型查询 │
├───────────────────┼──────────┼──────────┤
│ FetchStatement │ 128 B │ 1 个 │
│ BinaryExpr │ 64 B │ 3-10 个 │
│ PropertyRef │ 48 B │ 5-20 个 │
│ LiteralExpr │ 40 B │ 3-10 个 │
│ MetricRef │ 32 B │ 0-5 个 │
├───────────────────┼──────────┼──────────┤
│ 典型查询总计 │ ~1.5 KB │ │
│ 复杂查询总计 │ ~5 KB │ │
└───────────────────┴──────────┴──────────┘
结论：即使并发处理 10000 个查询，AST 总内存 < 50 MB
#9. 测试策略
#9.1 解析器测试矩阵
Python
class TestOQLParser:
"""OQL 解析器测试"""
def test_simple_fetch(self):
ast = parse("FETCH Person")
assert isinstance(ast, FetchStatement)
assert ast.entity_type == "Person"
assert ast.where_clause is None
def test_fetch_with_where(self):
ast = parse("FETCH Person WHERE age > 30 AND name = 'Alice'")
assert isinstance(ast.where_clause, BinaryExpr)
assert ast.where_clause.operator == "AND"
def test_traverse_multi_hop(self):
ast = parse("TRAVERSE Person -> WorksAt -> Company -> LocatedIn -> City")
assert isinstance(ast, TraverseStatement)
assert len(ast.path_segments) == 4
def test_case_insensitive_keywords(self):
ast1 = parse("FETCH Person")
ast2 = parse("fetch Person")
ast3 = parse("Fetch Person")
assert ast1.entity_type == ast2.entity_type == ast3.entity_type
def test_error_recovery(self):
errors = parse_with_errors(
"FETCH Person WERE age > 30"
)
assert len(errors) == 1
assert "WHERE" in errors[0].suggestions
def test_operator_precedence(self):
ast = parse("FETCH Person WHERE a > 1 OR b > 2 AND c > 3")
# AND binds tighter than OR
assert ast.where_clause.operator == "OR"
assert ast.where_clause.right.operator == "AND"
#9.2 模糊测试
Python
from hypothesis import given, strategies as st
@given(st.text(min_size=1, max_size=1000))
def test_parser_never_crashes(random_text):
"""解析器不应因任何输入而崩溃"""
try:
parse(random_text)
except OQLParseError:
pass # 合法的解析错误
# 不应抛出其他异常（如 IndexError、KeyError）
#Key Takeaways
-
手写递归下降解析器在 OQL 规模的语言中是最优选择：~50 条产生式的语法，手写解析器在错误恢复和性能方面显著优于 ANTLR/PEG 生成器。
-
三阶段流水线（Lexer → Parser → Semantic Analyzer）实现关注点分离：每个阶段只处理一个层面的问题，降低了复杂度和维护成本。
-
Pratt Parsing 优雅解决运算符优先级：通过优先级表驱动的递归调用，避免了为每个优先级层写独立的解析函数。
-
语义分析阶段连接 Ontology Schema：名称解析和类型检查确保查询在执行前就被验证，提前暴露错误。
-
友好的错误消息是用户体验的关键：指出位置、说明期望、给出建议——三要素让错误消息从"看不懂"变为"能自助修复"。
#Next Article
下一篇 S3-08《查询联邦：跨引擎统一查询》 将展示如何将 OQL 编译后的查询计划分发到 Doris、DuckDB、Elasticsearch 等不同引擎，实现一条 OQL 查询多引擎联合执行。
Tags: #OQL #Parser #AST #Lexer #RecursiveDescent #PrattParsing #SemanticAnalysis #TypeChecker #ErrorRecovery #智策平台 #coomia-dip #数据基座