“

系列：S12 开发者教程 · 第 6 篇 | 难度：中级 | 阅读时间：15 分钟

OQL 查询指南：20 个真实示例

#TL;DR

OQL（Ontology Query Language）是 coomia-dip 平台的声明式查询语言，专为 Ontology 数据模型设计。它比 SQL 更贴近业务语义，支持对象、关系、属性的联合查询，以及图遍历、聚合和推理集成。本文通过 20 个真实示例，从基础到高级全面覆盖 OQL 的查询能力。

#1. OQL 概述

#1.1 什么是 OQL？

OQL 是 coomia-dip 平台的原生查询语言，运行在 Control Layer（Control Layer）之上。与传统 SQL 不同，OQL 以 Ontology 为核心抽象：你查询的是"对象"和"关系"，而不是"表"和"行"。

Code

SQL:   SELECT * FROM projects WHERE status = 'active'
OQL:   FIND Project WHERE status = 'active'

#1.2 OQL 与 SQL 的关键差异

特性	SQL	OQL
查询目标	表/行	对象/关系
关联方式	JOIN	TRAVERSE
数据模型	关系型	本体(Ontology)
图遍历	多级 JOIN	原生 TRAVERSE
权限过滤	手动 WHERE	自动 ABAC 集成
类型安全	弱	Schema 强校验

#1.3 OQL 执行架构

Code

Client (SDK/API)
    │
    ▼
Control Layer (gRPC)
    │  ① 解析 OQL → AST
    │  ② Schema 校验
    │  ③ 权限过滤注入
    │  ④ 查询优化
    ▼
Data Layer (Doris/Iceberg)
    │  ⑤ 执行查询
    ▼
Result Set → Object 映射

#2. 环境准备

#2.1 安装 SDK

Bash

pip install ontology-sdk>=1.0.0

#2.2 初始化连接

Python

from ontology_sdk import OntoPlatform

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    data_plane_url="localhost:50052",
    auth_token="your-token"
)

# 获取 OQL 客户端
oql = platform.oql

#2.3 示例数据模型

本文所有示例基于以下 Ontology 模型：

YAML

object_types:
  - Employee:
      properties: [name, email, department, salary, hire_date, level]
  - Project:
      properties: [name, status, priority, budget, start_date, end_date]
  - Department:
      properties: [name, code, location, head_count]
  - Task:
      properties: [title, status, assignee_rid, due_date, story_points]
  - Customer:
      properties: [name, industry, region, annual_revenue, tier]

relations:
  - works_in: Employee -> Department
  - participates_in: Employee -> Project
  - owns_project: Department -> Project
  - has_task: Project -> Task
  - assigned_to: Task -> Employee
  - serves: Project -> Customer

#3. 基础查询（示例 1-5）

#示例 1：简单对象查询

查找所有进行中的项目：

Python

results = oql.execute("""
    FIND Project
    WHERE status = 'in_progress'
    SELECT name, priority, budget
    ORDER BY priority DESC
""")

for project in results:
    print(f"{project.name} | 优先级: {project.priority} | 预算: {project.budget}")

输出：

Code

智慧工厂监控系统 | 优先级: critical | 预算: 5000000
供应链优化平台 | 优先级: high | 预算: 3200000
客户360画像 | 优先级: medium | 预算: 1800000

#示例 2：多条件组合过滤

查找高薪资的高级员工：

Python

results = oql.execute("""
    FIND Employee
    WHERE salary > 30000
      AND level IN ('senior', 'staff', 'principal')
      AND department != 'HR'
    SELECT name, level, salary, department
    ORDER BY salary DESC
    LIMIT 20
""")

OQL 支持的操作符：

比较：=, !=, >, <, >=, <=
集合：IN, NOT IN
模糊：LIKE, CONTAINS, STARTS_WITH
空值：IS NULL, IS NOT NULL
逻辑：AND, OR, NOT

#示例 3：属性投影与别名

Python

results = oql.execute("""
    FIND Project
    WHERE status = 'in_progress'
    SELECT
        name AS project_name,
        budget AS total_budget,
        (end_date - start_date) AS duration_days,
        priority
""")

#示例 4：分页查询

Python

# 第一页
page1 = oql.execute("""
    FIND Employee
    WHERE department = 'Engineering'
    SELECT name, email, level
    ORDER BY hire_date ASC
    LIMIT 20 OFFSET 0
""")

# 第二页
page2 = oql.execute("""
    FIND Employee
    WHERE department = 'Engineering'
    SELECT name, email, level
    ORDER BY hire_date ASC
    LIMIT 20 OFFSET 20
""")

# SDK 封装的分页方式
paginator = oql.paginate(
    query="FIND Employee WHERE department = 'Engineering'",
    page_size=20
)

for page in paginator:
    for emp in page.results:
        print(emp.name)
    print(f"--- 第 {page.page_number} 页，共 {page.total_pages} 页 ---")

#示例 5：模糊搜索与全文检索

Python

# 属性模糊匹配
results = oql.execute("""
    FIND Project
    WHERE name LIKE '%供应链%'
       OR description CONTAINS '优化'
    SELECT name, description
""")

# 全文检索模式
results = oql.execute("""
    FIND Project
    SEARCH '供应链风险预警'
    SELECT name, description, _score
    ORDER BY _score DESC
    LIMIT 10
""")

#4. 关系遍历查询（示例 6-10）

#示例 6：单级关系遍历

查找某个部门下的所有员工：

Python

results = oql.execute("""
    FIND Employee
    TRAVERSE works_in -> Department
    WHERE Department.name = '工程部'
    SELECT Employee.name, Employee.level, Employee.salary
""")

等效的 gRPC 调用：

Python

# 先通过 SDK 对象 API
dept = platform.objects.get("Department", filters={"name": "工程部"})
employees = platform.relations.get_related(
    source_rid=dept.rid,
    relation_type="works_in",
    direction="inbound"
)

#示例 7：多级关系遍历

查找参与了某客户项目的所有员工：

Python

results = oql.execute("""
    FIND Employee
    TRAVERSE participates_in -> Project
    TRAVERSE serves -> Customer
    WHERE Customer.name = '华为技术有限公司'
    SELECT
        Employee.name AS employee_name,
        Project.name AS project_name,
        Customer.name AS customer_name
""")

#示例 8：反向关系遍历

查找某个员工负责的所有任务及其所属项目：

Python

results = oql.execute("""
    FIND Task
    TRAVERSE assigned_to -> Employee
    WHERE Employee.name = '张三'
    INCLUDE
        TRAVERSE has_task <- Project
        SELECT Project.name AS project_name
    SELECT Task.title, Task.status, Task.due_date, project_name
    ORDER BY Task.due_date ASC
""")

#示例 9：关系属性过滤

Python

results = oql.execute("""
    FIND Employee
    TRAVERSE participates_in -> Project
    WHERE participates_in.role = 'tech_lead'
      AND Project.status = 'in_progress'
    SELECT
        Employee.name,
        participates_in.role,
        participates_in.joined_date,
        Project.name AS project_name
""")

#示例 10：图路径查询

查找两个对象之间的连接路径：

Python

results = oql.execute("""
    PATH FROM Employee WHERE name = '张三'
    TO Customer WHERE name = '华为技术有限公司'
    MAX_DEPTH 4
    VIA [participates_in, serves, owns_project, works_in]
""")

for path in results:
    print(" -> ".join([node.type_name + ":" + node.name for node in path.nodes]))

输出：

Code

Employee:张三 -> Project:智慧工厂监控系统 -> Customer:华为技术有限公司
Employee:张三 -> Department:工程部 -> Project:供应链优化平台 -> Customer:华为技术有限公司

#5. 聚合与分析查询（示例 11-15）

#示例 11：基础聚合

Python

results = oql.execute("""
    FIND Employee
    GROUP BY department
    AGGREGATE
        COUNT(*) AS total_count,
        AVG(salary) AS avg_salary,
        MAX(salary) AS max_salary,
        MIN(salary) AS min_salary
    ORDER BY avg_salary DESC
""")

for row in results:
    print(f"{row.department}: {row.total_count}人, "
          f"平均薪资: {row.avg_salary:.0f}")

#示例 12：多维度聚合

Python

results = oql.execute("""
    FIND Task
    TRAVERSE has_task <- Project
    GROUP BY Project.name, Task.status
    AGGREGATE
        COUNT(*) AS task_count,
        SUM(Task.story_points) AS total_points
    ORDER BY Project.name, Task.status
""")

#示例 13：时间序列聚合

Python

results = oql.execute("""
    FIND Employee
    GROUP BY DATE_TRUNC('month', hire_date) AS hire_month
    AGGREGATE
        COUNT(*) AS new_hires
    WHERE hire_date >= '2025-01-01'
    ORDER BY hire_month ASC
""")

# 生成入职趋势报表
for row in results:
    bar = '█' * row.new_hires
    print(f"{row.hire_month}: {bar} ({row.new_hires})")

#示例 14：HAVING 条件过滤

Python

results = oql.execute("""
    FIND Project
    TRAVERSE has_task -> Task
    GROUP BY Project.name
    AGGREGATE
        COUNT(*) AS total_tasks,
        COUNT(CASE WHEN Task.status = 'done' THEN 1 END) AS done_tasks,
        ROUND(
            COUNT(CASE WHEN Task.status = 'done' THEN 1 END) * 100.0
            / COUNT(*), 1
        ) AS completion_rate
    HAVING total_tasks > 5 AND completion_rate < 50
    ORDER BY completion_rate ASC
""")

print("=== 进度落后的项目 ===")
for row in results:
    print(f"{row.project_name}: {row.completion_rate}% "
          f"({row.done_tasks}/{row.total_tasks})")

#示例 15：跨对象统计

Python

results = oql.execute("""
    FIND Department
    INCLUDE
        TRAVERSE works_in <- Employee
        AGGREGATE COUNT(*) AS emp_count, AVG(Employee.salary) AS avg_salary
    INCLUDE
        TRAVERSE owns_project <- Project
        WHERE Project.status = 'in_progress'
        AGGREGATE COUNT(*) AS active_projects, SUM(Project.budget) AS total_budget
    SELECT
        Department.name,
        Department.location,
        emp_count,
        avg_salary,
        active_projects,
        total_budget
    ORDER BY total_budget DESC
""")

#6. 高级查询（示例 16-20）

#示例 16：子查询

Python

results = oql.execute("""
    FIND Employee
    WHERE salary > (
        FIND Employee
        WHERE department = $self.department
        AGGREGATE AVG(salary)
    )
    SELECT name, department, salary
    ORDER BY department, salary DESC
""")

#示例 17：WITH 预计算（CTE 等价）

Python

results = oql.execute("""
    WITH dept_stats AS (
        FIND Employee
        GROUP BY department
        AGGREGATE
            AVG(salary) AS avg_salary,
            COUNT(*) AS emp_count
    )
    FIND Employee
    JOIN dept_stats ON Employee.department = dept_stats.department
    WHERE Employee.salary > dept_stats.avg_salary * 1.5
    SELECT
        Employee.name,
        Employee.department,
        Employee.salary,
        dept_stats.avg_salary AS dept_avg,
        ROUND(Employee.salary / dept_stats.avg_salary * 100, 1) AS ratio_pct
    ORDER BY ratio_pct DESC
""")

#示例 18：条件表达式

Python

results = oql.execute("""
    FIND Project
    SELECT
        name,
        budget,
        CASE
            WHEN budget > 5000000 THEN '大型项目'
            WHEN budget > 2000000 THEN '中型项目'
            WHEN budget > 500000 THEN '小型项目'
            ELSE '微型项目'
        END AS project_scale,
        CASE
            WHEN end_date < NOW() AND status != 'done' THEN '已逾期'
            WHEN end_date < NOW() + INTERVAL '7 days' THEN '即将到期'
            ELSE '正常'
        END AS deadline_status
    ORDER BY budget DESC
""")

#示例 19：参数化查询与防注入

Python

# 参数化查询 — 防止 OQL 注入
results = oql.execute(
    """
    FIND Employee
    WHERE department = :dept
      AND level IN :levels
      AND hire_date >= :start_date
    SELECT name, level, salary
    ORDER BY salary DESC
    LIMIT :page_size OFFSET :offset
    """,
    params={
        "dept": "Engineering",
        "levels": ["senior", "staff", "principal"],
        "start_date": "2024-01-01",
        "page_size": 20,
        "offset": 0
    }
)

# 批量参数化
batch_results = oql.execute_batch(
    "FIND Employee WHERE department = :dept AGGREGATE COUNT(*) AS cnt",
    params_list=[
        {"dept": "Engineering"},
        {"dept": "Product"},
        {"dept": "Design"},
    ]
)

for dept_result in batch_results:
    print(f"{dept_result.params['dept']}: {dept_result.results[0].cnt}人")

#示例 20：推理集成查询

OQL 可以与 Intelligence Layer 的推理引擎联动：

Python

# 基于规则推理的查询
results = oql.execute("""
    FIND Employee
    WHERE INFERRED risk_level = 'high'
    SELECT
        name, department,
        INFERRED risk_level,
        INFERRED risk_factors,
        INFERRED recommended_actions
""")

# INFERRED 关键字触发推理引擎
# Intelligence Layer 会基于规则和知识图谱计算衍生属性

# 图谱推理查询
results = oql.execute("""
    FIND Customer
    WHERE INFERRED churn_probability > 0.7
    TRAVERSE serves <- Project
    SELECT
        Customer.name,
        Customer.tier,
        INFERRED Customer.churn_probability,
        INFERRED Customer.churn_reason,
        Project.name AS related_project
    ORDER BY INFERRED Customer.churn_probability DESC
""")

for row in results:
    print(f"⚠ {row.customer_name} (流失概率: {row.churn_probability:.1%})")
    print(f"  原因: {row.churn_reason}")
    print(f"  关联项目: {row.related_project}")

#7. OQL 最佳实践

#7.1 性能优化

Python

# 差：全量扫描后过滤
oql.execute("FIND Employee SELECT * WHERE salary > 50000")

# 好：只选必要字段、先过滤
oql.execute("""
    FIND Employee
    WHERE salary > 50000
    SELECT name, salary, level
    LIMIT 100
""")

# 差：N+1 查询
for project in projects:
    tasks = oql.execute(f"FIND Task TRAVERSE has_task <- Project WHERE Project.rid = '{project.rid}'")

# 好：批量遍历
results = oql.execute("""
    FIND Project
    WHERE status = 'in_progress'
    INCLUDE
        TRAVERSE has_task -> Task
        SELECT Task.title, Task.status
    SELECT Project.name, tasks
""")

#7.2 查询计划分析

Python

# 使用 EXPLAIN 查看查询计划
plan = oql.explain("""
    FIND Employee
    TRAVERSE works_in -> Department
    WHERE Department.location = '上海'
    SELECT Employee.name, Employee.salary
""")

print(plan.to_tree())
# ├── ObjectScan: Employee (estimated: 5000 rows)
# ├── RelationLookup: works_in (index: btree)
# ├── Filter: Department.location = '上海' (selectivity: 0.15)
# └── Project: [Employee.name, Employee.salary]
# Estimated cost: 750 | Uses indexes: [works_in_idx, dept_location_idx]

#7.3 常见错误与排查

错误	原因	解决方案
`ObjectTypeNotFound`	对象类型名拼写错误	检查 Schema Registry
`PropertyNotFound`	属性名不存在	使用 `platform.schema.describe("Employee")`
`PermissionDenied`	ABAC 策略阻止访问	检查当前用户的权限策略
`TraversalDepthExceeded`	图遍历深度超限	调整 `MAX_DEPTH` 或优化查询
`QueryTimeout`	查询执行超时	添加索引或重构查询

#8. 完整实战：项目健康度仪表盘

Python

from ontology_sdk import OntoPlatform

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    data_plane_url="localhost:50052"
)
oql = platform.oql

# 项目总览
overview = oql.execute("""
    FIND Project
    GROUP BY status
    AGGREGATE COUNT(*) AS count, SUM(budget) AS total_budget
""")

print("=== 项目状态分布 ===")
for row in overview:
    print(f"  {row.status}: {row.count} 个项目, 预算合计: ¥{row.total_budget:,.0f}")

# 资源利用率
utilization = oql.execute("""
    FIND Employee
    INCLUDE
        TRAVERSE participates_in -> Project
        WHERE Project.status = 'in_progress'
        AGGREGATE COUNT(*) AS active_projects
    SELECT
        name, department, level, active_projects,
        CASE
            WHEN active_projects > 3 THEN '过载'
            WHEN active_projects = 0 THEN '空闲'
            ELSE '正常'
        END AS workload_status
    ORDER BY active_projects DESC
""")

print("\n=== 资源负载 ===")
overloaded = [r for r in utilization if r.workload_status == '过载']
idle = [r for r in utilization if r.workload_status == '空闲']
print(f"  过载: {len(overloaded)} 人")
print(f"  空闲: {len(idle)} 人")

# 延期风险
at_risk = oql.execute("""
    FIND Project
    WHERE status = 'in_progress'
      AND end_date < NOW() + INTERVAL '14 days'
    INCLUDE
        TRAVERSE has_task -> Task
        WHERE Task.status != 'done'
        AGGREGATE COUNT(*) AS remaining_tasks
    SELECT name, end_date, remaining_tasks, priority
    HAVING remaining_tasks > 0
    ORDER BY end_date ASC
""")

print("\n=== 延期风险项目 ===")
for row in at_risk:
    print(f"  {row.name}: 剩余 {row.remaining_tasks} 个任务, "
          f"截止 {row.end_date}")

#Key Takeaways

OQL 以 Ontology 为核心：查询的是对象和关系，而非表和行，业务语义更直观
TRAVERSE 替代 JOIN：图遍历是 OQL 的核心优势，多级关系查询无需复杂 JOIN
内置安全：ABAC 权限策略自动注入查询，无需手动过滤
推理集成：INFERRED 关键字无缝对接 Intelligence Layer 的推理引擎
参数化防注入：始终使用参数化查询，避免 OQL 注入风险
性能优先：善用 EXPLAIN、合理选择字段、避免 N+1 查询

#Next Article

下一篇：S12-07 Pipeline 开发指南 — 学习如何构建端到端的数据管道，从数据接入到 Ontology 写入。

Tags: OQL 查询语言 Ontology 图遍历 聚合分析 推理查询 coomia-dip