查询数据

已完成

本单元介绍特定于 PostgreSQL 的查询功能和模式，这些功能对于 AI 应用程序至关重要。 本单元重点介绍 PostgreSQL 的独特功能和高级模式，用于管理聊天历史记录、处理状态和应用程序元数据。

本单元假定熟悉标准 SQL 概念，例如 SELECT、 WHERE联接和聚合。 如果你需要复习，学习路径使用 Transact-SQL 进行查询入门 是一个很好的资源。

查询执行和别名范围

即使是经验丰富的 SQL 开发人员也会被执行顺序困扰，尤其是在使用列别名时。 SQL 语句以与写入方式不同的特定逻辑顺序执行，理解此顺序有助于排查神秘的“列不存在”错误。

下单	第	目的
1	FROM	标识源表
2	WHERE	筛选行
3	GROUP BY	对行进行聚合分组
4	HAVING	过滤器组
5	SELECT	选择列和计算表达式
6	ORDER BY	对结果进行排序
7	LIMIT / OFFSET	限制结果数量

别名范围规则： 在 中 SELECT 定义的列别名仅对在子句 之后SELECT执行的子句可见，即 ORDER BY 和 LIMIT/OFFSET。 较早执行的子句（WHERE、GROUP BY、HAVING）无法引用这些别名，因为它们尚未定义。

这意味着在 WHERE 时不能按别名筛选，但可以在 ORDER BY 时按别名排序。

-- This FAILS: WHERE executes before SELECT, so 'msg_date' doesn't exist yet
SELECT DATE(created_at) AS msg_date, content
FROM messages
WHERE msg_date > '2024-01-01';  -- Error: column "msg_date" does not exist

-- This WORKS: repeat the expression in WHERE
SELECT DATE(created_at) AS msg_date, content
FROM messages
WHERE DATE(created_at) > '2024-01-01';

-- This also WORKS: ORDER BY executes after SELECT, so aliases are available
SELECT DATE(created_at) AS msg_date, content
FROM messages
WHERE DATE(created_at) > '2024-01-01'
ORDER BY msg_date;

特定于 PostgreSQL 的筛选

PostgreSQL 使用简化常见筛选任务的运算符扩展标准 SQL。 这些功能对于需要灵活的文本搜索和结构化元数据查询的 AI 应用程序非常有用。

运算符 ILIKE 提供不区分大小写的模式匹配，而无需像这样的 LOWER()函数。 这对于要匹配的面向用户的搜索功能非常有用，无论大小写如何： SELECT * FROM messages WHERE content ILIKE '%error%'

PostgreSQL 允许你使用NULLS FIRST或NULLS LAST控制值在排序结果中的显示位置。 默认情况下，NULL 值排序时仿佛比任何其他值都大。 使用 ORDER BY ended_at NULLS LAST 将未完成的对话保留在结尾，或使用 ORDER BY completed_at NULLS FIRST 优先显示未处理的任务。

该 COALESCE 函数从其参数中返回第一个非 null 值。 使用它在查询结果中提供默认值（COALESCE(title, 'Untitled')）或处理表达式中可为 null 的列。

查询 JSONB 数据

PostgreSQL 的 JSONB 类型存储不适合固定架构的结构化数据。 AI 应用程序通常将 JSONB 用于元数据、配置、模型参数和变量响应结构。 PostgreSQL 提供用于提取值、检查结构和基于 JSON 内容的筛选的专用运算符。

用于 -> 将 JSON 元素提取为 JSON，或 ->> 将其提取为文本（用于比较和显示）。 对于嵌套路径，请使用 #> （返回 JSON）或 #>> （返回文本）。 例如，metadata->>'status' 将状态字段提取为文本，而 checkpoint_data#>>'{results,0,score}' 导航嵌套路径以获取特定值。

存在和包含运算符可实现高效的筛选。 ?运算符检查密钥是否存在（WHERE metadata ? 'priority'），而@>测试包含 （WHERE checkpoint_data @> '{"status": "completed"}'）。 这些运算符可以使用 GIN 索引对大型表进行高效筛选。

当 JSONB 列包含数组时，用于 jsonb_array_elements 展开它们以筛选或聚合：

-- Find conversations tagged with 'support'
SELECT DISTINCT c.*
FROM conversations c,
     jsonb_array_elements_text(c.metadata->'tags') AS tag
WHERE tag = 'support';

使用键集分页进行高效分页

基于传统OFFSET的分页在大型表上变得缓慢，因为 PostgreSQL 必须扫描并丢弃所有跳过的行。 第1,000页，每页20行，需要扫描20,000行并丢弃19,980行。 键集分页（也称为基于游标的分页）使用 WHERE 子句跳过行，无论分页深度如何，这些子句都会一致地执行。

不跟踪页码，而是跟踪您看到的最后一个值并从那里进行筛选。 此方法需要唯一、可排序的列（或列的组合）：

-- First page: get the 20 most recent messages
SELECT id, conversation_id, content, created_at
FROM messages
ORDER BY created_at DESC, id DESC
LIMIT 20;

-- Next page: filter by the last seen timestamp and id
SELECT id, conversation_id, content, created_at
FROM messages
WHERE (created_at, id) < ('2024-06-15 10:30:00', 12345)
ORDER BY created_at DESC, id DESC
LIMIT 20;

在多个行具有相同时间戳时，通过在ORDER BY和WHERE子句中都包含id来处理冲突。 应用程序存储最后一行的排序值，并将其传递给下一个查询。 要实现升序，请将<更改为>，并将DESC更改为ASC。

公用表表达式 (CTE)

CTE 定义仅在执行查询期间存在的命名临时结果集。 它们通过逐步生成复杂查询来提高可读性，并为分层数据启用递归查询。

使用 CTE 将复杂查询分解为逻辑步骤。 每个 CTE 都可以引用以前定义的 CTE，从而创建转换流程：

WITH recent_conversations AS (
    SELECT id, user_id, started_at
    FROM conversations
    WHERE started_at > CURRENT_DATE - INTERVAL '7 days'
),
message_stats AS (
    SELECT conversation_id, COUNT(*) AS message_count, MAX(created_at) AS last_message_at
    FROM messages
    GROUP BY conversation_id
)
SELECT rc.user_id, rc.started_at, COALESCE(ms.message_count, 0) AS message_count
FROM recent_conversations rc
LEFT JOIN message_stats ms ON rc.id = ms.conversation_id;

递归 CTE 查询树结构，例如任务层次结构、组织图表或线性对话。 它们由基事例（定位点）和引用 CTE 本身的递归事例组成。 如果数据包含周期，始终包含深度限制或其他终止条件，以防止无限循环：

WITH RECURSIVE task_tree AS (
    -- Base case: start with the parent task
    SELECT id, parent_id, title, 1 AS depth
    FROM tasks WHERE id = 1
    UNION ALL
    -- Recursive case: find children of current level
    SELECT t.id, t.parent_id, t.title, tt.depth + 1
    FROM tasks t
    INNER JOIN task_tree tt ON t.parent_id = tt.id
    WHERE tt.depth < 10
)
SELECT * FROM task_tree ORDER BY depth, id;

AI 应用程序通常需要检索消息引用父消息的会话线程：

WITH RECURSIVE thread AS (
    -- Start with the root message
    SELECT id, parent_id, content, role, 0 AS depth
    FROM messages
    WHERE id = :root_message_id

    UNION ALL

    -- Get all replies
    SELECT m.id, m.parent_id, m.content, m.role, t.depth + 1
    FROM messages m
    INNER JOIN thread t ON m.parent_id = t.id
    WHERE t.depth < 50
)
SELECT * FROM thread ORDER BY depth, id;

使用 RETURNING 插入

PostgreSQL 中的 RETURNING 子句在一次往返中从插入、更新或删除的行里检索值。 这对于在不单独查询的情况下获取自动生成的 ID、时间戳或计算默认值至关重要。

-- Get the generated ID after inserting a conversation
INSERT INTO conversations (user_id, session_id)
VALUES ('user123', 'sess_abc')
RETURNING id;

-- Get multiple generated values
INSERT INTO messages (conversation_id, role, content)
VALUES (1, 'user', 'Hello')
RETURNING id, created_at;

-- Use RETURNING with UPDATE
UPDATE tasks SET status = 'completed', completed_at = CURRENT_TIMESTAMP
WHERE id = 5
RETURNING id, status, completed_at;

ON CONFLICT 的 Upserts

处理唯一约束违反的 INSERT ... ON CONFLICT 子句，允许执行插入新行或更新现有行的“upsert”操作。 此模式对于 AI 应用程序中的幂等操作和状态管理非常有价值。

在唯一约束上发生冲突时，可以使用 DO UPDATE 将现有行更新为新值。 EXCLUDED伪表引用已插入的值：

INSERT INTO user_preferences (user_id, preference_key, preference_value)
VALUES ('user123', 'theme', 'dark')
ON CONFLICT (user_id, preference_key)
DO UPDATE SET
    preference_value = EXCLUDED.preference_value,
    updated_at = CURRENT_TIMESTAMP;

使用DO NOTHING来静默跳过违反约束的行：INSERT INTO tags (name) VALUES ('important') ON CONFLICT (name) DO NOTHING

您可以向DO UPDATE添加WHERE子句，用于条件逻辑，仅在新值与现有值不同时进行更新。 将ON CONFLICT和RETURNING结合使用，以便了解一行是插入还是更新。表达式(xmax = 0)对新插入的行返回true，对更新的行返回false。

AI 应用程序的实用模式

这些模式将前面介绍的功能合并为常见 AI 应用程序要求的解决方案。 以下查询使用用于上下文生成的元数据检索会话历史记录：

WITH conversation_context AS (
    SELECT c.id, c.session_id, c.metadata->>'model' AS model
    FROM conversations c
    WHERE c.session_id = :session_id AND c.ended_at IS NULL
)
SELECT cc.session_id, cc.model, m.role, m.content, m.created_at
FROM conversation_context cc
INNER JOIN messages m ON cc.id = m.conversation_id
ORDER BY m.created_at
LIMIT 50;

以下示例演示如何使用 upsert 通过状态管理记录任务检查点：

INSERT INTO task_checkpoints (task_id, step_number, checkpoint_data)
VALUES (:task_id, :step_number, :checkpoint_json::jsonb)
ON CONFLICT (task_id, step_number)
DO UPDATE SET checkpoint_data = EXCLUDED.checkpoint_data, updated_at = CURRENT_TIMESTAMP
RETURNING id, created_at, (xmax = 0) AS is_new;

以下示例显示了包含 JSONB 筛选器的分页搜索：

SELECT c.id, c.session_id, c.started_at, c.metadata->>'status' AS status
FROM conversations c
WHERE c.user_id = :user_id
  AND c.metadata @> :filter_json::jsonb
  AND (c.started_at, c.id) < (:last_started_at, :last_id)
ORDER BY c.started_at DESC, c.id DESC
LIMIT 20;

其他资源

• PostgreSQL JSON 函数和运算符
• PostgreSQL INSERT 文档
• PostgreSQL WITH 查询 （CTE）