s 在大模型（LLM）驱动的问答系统中，RAG（Retrieval-Augmented Generation）架构正迅速成为主流；然而在实际应用中，即便接入了如 GPT-4 或 Claude 等先进模型，但生成结果仍然不够理想。

问题的根源往往并不在于模型本身，而在于——它没有检索到相关信息，这就引出了评估 RAG 检索质量的核心指标：召回率（Recall）。

本文将深入探讨召回率的本质，以及如何通过构建一个结构化、丰富且高质量的知识库，显著提升 RAG 系统的召回效果，从而增强问答系统的准确性与实用性。

召回率是什么？​

在 RAG 检索系统中，召回率指的是在所有真正相关的文档中，有多少被成功地检索了出来。

计算方式：

召回率 = 检索到的相关内容数量 ÷ 所有相关内容的总数量

举个例子，假设你有一个技术文档知识库，里面记录着产品安装、配置、调优等各类信息。

某个用户提问：“如何在 Kubernetes 上部署？”

假设知识库中有 6 条与这个问题高度相关的内容，但系统只返回了其中 3 条。那么，召回率就是 3 ÷ 6 = 50%。

在 RAG 系统中，召回率尤为关键，因为大模型能不能“答对”，很大程度取决于有没有拿到相关内容；召回率越高，LLM 在生成答案时能参考的有效信息就越多，回答的质量和准确性也就越有保障。

召回不准的原因​

很多人将注意力放在向量数据库、查询优化、模型推理上，却忽视了最根本的基础设施 —— 知识库构建。

召回失败，往往源于三个层面：

1. 数据覆盖不足：知识来源单一，未能全面汇聚 FAQ、产品文档、技术手册、历史工单等高价值内容。
2. 语义表达偏差：不合理的 分块策略（如按固定字数切割）会割裂上下文；**Embedding 模型 **选择不当则会无法精准捕捉文本的深层语义，导致向量表达失真。
3. 结构策略粗糙：没有上下文信息、缺少结构化字段或文档元数据。

如何构建高召回率的知识库​

提高数据覆盖率​

任何检索的前提是知识库中有相关信息。因此在构建 RAG 专属知识库时，需要聚焦以下能力：

• 汇聚多渠道内容：FAQ、文档、部署手册、工单记录等
• 支持多种接入方式：数据库、OSS/S3、Google Docs、语雀、本地文件等

提高语义嵌入质量​

选择合适的 Embedding 模型，决定了 用户问题 能否成功匹配到 知识块。

目前业界有许多优秀的 Embedding 模型，以下是一个简单对比：

分块策略合理​

分块（Chunking） 是指将长文档切割成适合 RAG 检索的、更小的文本单元：

• 若分块太小：上下文缺失，回答不准确。
• 若分块太大：Embedding 过于抽象，无法命中具体问题。

在具体实践中，应考虑：

• 按语义、标题、段落切块，避免语义断层。
• 支持 Chunk Overlap，每块有一定重叠，如每 300 个 Token 滑动切，同时根据语义分段，召回命中率更高。

结构化向量库​

传统向量检索仅依赖 Embedding 相似度，虽具备语义匹配能力，但仍存在明显短板：向量相似但语义不相关的内容易被误召回。

结构化向量库在此基础上引入了丰富的元信息结构字段，进一步提升了召回的准确性。

• 使用大模型提取 FAQ、摘要、标签、时间字段等，可有效补充语义缺失的上下文信息。
• 将 Markdown、客服记录等非结构化内容转为统一格式，显著提升整体检索命中率。
• 利用结构化 Schema 支撑后续精准检索、过滤、排序。

构建高质量知识库​

在实际落地过程中，CloudCanal 提供了一套面向企业的自动化知识库构建能力，支持：

• 多源文档采集（OSS、S3、SSH、Google Docs、语雀）
• LLM 提取结构化字段，支持自定义 Prompt
• 段落分块 + 重叠控制 + 元信息附带
• 向量化写入 StarRocks 等向量库，支持 Qwen 等主流模型接入

下面将使用 StarRocks 作为目标向量库，展示如何在 CloudCanal 中快速构建高质量的知识库。

添加数据源​

登录 CloudCanal 控制台，点击 「数据源管理」>「新增数据源」，添加以下数据源：

文件数据源（SshFile）​

用于读取本地或远程服务器中的 Markdown 文档。添加步骤如下：

• 类型选择：自建 > SshFile（同理可配置 S3、OSS 等）
• 基础配置：填写服务器 IP、端口、用户名和密码。
  • 网络地址：localhost:22
  • 用户名：root****
  • 密码：***
• 额外参数：
  • fileSuffixArray：填写 **.md**，仅处理 Markdown 文件
  • enableLLMExtraction：true，启用 LLM 提取 额外信息 的功能。
  • defaultLineSchemaJson：定义需要 LLM 提取的 结构化字段（第一行 line 表示原文）

[
    {
        "column": "line",
        "jdbcType": 12,
        "typeName": "TEXT"
    },
    {
        "column": "doc_title",
        "jdbcType": 12,
        "typeName": "TEXT",
        "desc": "从文本中提取出文档的明确标题"
    },
    {
        "column": "summary",
        "jdbcType": 12,
        "typeName": "TEXT",
        "desc": "为这段文本生成一个不超过100字的简明摘要"
    },
    {
        "column": "keywords",
        "jdbcType": 12,
        "typeName": "TEXT",
        "desc": "提取5个最能代表文本核心内容的关键词，以逗号分隔"
    },
    {
        "column": "faq_pairs",
        "jdbcType": 12,
        "typeName": "TEXT",
        "desc": "从文本中提取出潜在的问答对（FAQ），以[{'question':'...', 'answer':'...'}]的JSON数组格式返回"
    }
]

• dbsJson：用于指定要同步的文档目录，你可以将其中的 **schema 字段 **修改为你实际存放 Markdown 文件的根目录路径。

[
  {
    "db":"cc_virtual_fs",
    "schemas":[
      {
        "schema":"/Users/barry/source/cloudcanal-doc-v2",
        "tables":[]
      }
    ]
   }
]

• db：虚拟文件库的逻辑名称，默认保持为 cc_virtual_fs 即可，无需修改。
• schema：表示你希望读取的本地或远程目录路径，CloudCanal 会以该路径作为文档扫描入口。例如：/Users/barry/source/cloudcanal-doc-v2。该字段必须填写为绝对路径。
• tables：用于指定目录中要处理的具体文件名，若设置为空数组（[]），则表示自动抓取该目录下所有符合后缀规则（如 .md）的文件，无需逐一列出文件名。

向量数据库（StarRocks）​

用于存储通过大模型编码后的文档向量，是整个 RAG 检索流程的核心数据源。

• 类型选择：自建 > StarRocks
• 准备工作：
  • 部署 StarRocks：参考文档
  • 版本要求：3.4 及以上
  • 打开 Vector Index

ADMIN SET FRONTEND CONFIG ("enable_experimental_vector" = "true");

• 配置说明：
  • 网络地址：localhost:9030
  • 用户名：root** **
• 额外参数：
  • privateHttpHost：填写 localhost:8030。用于 Stream Load 写入，可填写 FE 或 BE 的地址。

大模型平台（Ollama）​

CloudCanal 支持通过 Ollama 提供完整的向量生成与推理能力，适用于完全私有化的 RAG 场景。

• 类型选择：自建 > Ollama
• 配置说明：
  • 部署 Ollama：参考文档
  • 网络地址：localhost:11434
  • **额外参数 **
    • llmEmbedding：嵌入大模型配置

{
  "qwen3:8b": {
    "dimension": 4096
  }
}

• **llmChat：**对话大模型配置

{
  "deepseek-r1": {
    "temperature": 0.9,
    "topP": 0.9,
    "showReasoning": false
  }
}

创建任务：构建知识库​

1. 点击 同步任务 > 创建任务。
2. 选择以下数据源，并点击 测试连接 确认网络与权限正常。
   • 源端：SshFile
   • 目标端：StarRocks

3. 在 功能配置 页面，任务类型选择 全量迁移，任务规格选择默认 2 GB 即可。

4. 在 表&action过滤 页面，进行以下配置：

   1. 点击 配置大模型 > Ollama，选择刚添加的大模型实例

      1. 嵌入模型选择 qwen3:8b（用于数据嵌入）
      2. 对话模型选择 deepseek:r1（用于结构化信息提取）

   2. 选择需要定时数据迁移的文件，可同时选择多个。

   3. 点击 批量修改目标名称 > 统一表名 > 填写表名（如 my_knowledge），并确认，方便将不同文件向量化并写入同一个表。

5. 在 数据处理 页面，点击 批量操作 > 大模型嵌入：

   1. 设置数据分隔长度 1000，分隔重叠 100。
   2. 选择需要嵌入的字段，并全选表。

6. 在 创建确认 页面，点击 创建任务，开始运行。任务会自动根据源端定义的格式，自动在 StarRocks 中创建向量表，并把源端文件处理分块后，嵌入，最终导入到 StarRocks

7. 到 StarRocks 中查询知识库数据。

select 
  doc_title, -- 标题
  summary,   -- 总结
  keywords,  -- 关键词
  faq_pairs, -- 问答
  __content  -- 原始内容
from my_knowledge limit 3 \G

至此，我们就构建好了一个结构化、丰富且高质量内容的知识库，后续可在 CloudCanal 中继续创建任务，构建 RAG API，直接与你的知识库对话。

总结​

召回率决定了 RAG 系统的检索表现，也影响最终生成结果的准确性，只有策略得当、工具可靠，才能构建真正智能的问答能力。

通过 CloudCanal RAG,可在 更短周期内,以 更简单的方式 实现更高检索质量的知识库，为企业打造实用、可信的 AI 应用打下坚实基础。

在企业级应用中，RAG（Retrieval-Augmented Generation）技术正在逐步从探索走向落地。与面向个人使用者的轻量级问答系统不同，企业对 RAG 的要求更高：它必须可靠、可控、可扩展，最重要的是——安全。许多企业对于数据上传至在线大模型或公有云向量数据库持谨慎甚至禁止态度，因为这可能导致敏感信息暴露给外部服务。

CloudCanal 近期已支持通过 Ollama 本地部署 RAG 服务，有效解决了传统方案中存在的数据安全隐患。本文将聚焦于企业级 RAG 服务，介绍如何通过本地部署实现一个不依赖公网的企业级 RAG 构建方案。

什么是企业级 RAG 服务？​

企业级 RAG 服务强调端到端的集成能力、安全可控以及对业务系统的适配性。它不仅能实现基于知识的智能问答，还能提供高可靠、高安全、高扩展的服务。在保护企业数据的前提下，真正服务于业务流程的自动化与智能化。

相比于面向个人或实验场景的轻量 RAG 项目，企业级 RAG 服务具有以下核心特征：

• 技术栈全私有，部署可控
  服务能够完全运行在本地数据中心或私有云环境中，避免数据外泄风险，满足对数据合规性要求较高的行业需求。
• 数据来源多样，格式丰富
  不局限于文本文件类型，还支持数据库等多种数据来源，实现真正的“全域知识接入”。
• 支持增量数据同步，确保时效性与一致性
  在企业级场景中，知识信息更新频繁，服务需支持高效的增量同步能力，确保 RAG 索引内容始终与业务系统保持同步。
• 可与多种工具链协同，完成复杂任务链路（类MCP能力）
  企业级 RAG 服务不仅要检索和生成，还要与函数调用（Function Calling）、工具执行（如 SQL 查询）等能力打通，构建更完整的智能任务执行流程。

那么，如何在不依赖任何在线服务的前提下，安全地构建一个完全私有的企业级 RAG 服务呢？

CloudCanal RAG​

CloudCanal 推出的 RagApi 封装了向量检索与模型问答能力，并兼容 MCP 协议，让用户能快速上手搭建属于自己的 RAG 服务。

相比传统 RAG 架构手动部署流程，CloudCanal 提供的 RAG 服务具有以下独特优势：

• 双任务完成全流程：文档导入 + API 发布。
• 零代码部署：无需开发，自定义配置即可构建 API 服务。
• 参数可调：支持设置向量 Top-K 数量、匹配阈值、Prompt 模板、模型温度等核心参数。
• 多模型与平台适配：支持阿里云 DashScope、OpenAI、DeepSeek 等主流模型与 API 平台。
• OpenAI API 兼容接口：直接接入现有 Chat 应用或工具链，无需额外适配。

实例演示​

下面将使用以下组件，展示如何快速搭建一个完全离线、数据可控、安全可靠的企业级 RAG 服务。

• Ollama：用于本地运行大语言模型
• PostgreSQL 向量数据库：作为本地向量检索存储引擎
• CloudCanal RagApi：用于构建本地化的 RAG 问答服务

整体工作流程如下：

前置部署​

部署本地 Ollama​

Ollama 可以让你在本地轻松部署和管理大语言模型。在本方案中，Ollama 不仅用于文档向量化阶段生成嵌入向量 (Embeddings)，也作为对话模型用于最终的问答交互。CloudCanal 支持通过 Ollama 提供完整的向量生成与推理能力，完美契合纯私有化 RAG 场景。

1. 下载与安装 Ollama

访问 Ollama 官方网站下载对应操作系统的安装包：https://ollama.com/download

2. 拉取并运行模型:

安装完成后，打开终端，执行以下命令来拉取并运行一个适合嵌入和推理的模型，例如 qwen3:8b (也可以根据硬件情况选择其他模型，如 qwen3:14b 等。请注意，部分大型模型对硬件资源要求较高)：

ollama run qwen3:8b

模型下载并首次运行后，可以按 Ctrl + D 退出当前模型的交互模式。

部署本地 PostgreSQL 向量数据库​

1. 安装 Docker（如已安装可跳过）

不同操作系统可参考以下步骤进行安装：

• MacOS：参考官方安装文档：Docker Desktop for Mac
• CentOS / RHEL：参考下面脚本

## centos / rhel
sudo yum-config-manager --add-repo https://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo

sudo yum install -y docker-ce-20.10.9-3.* docker-ce-cli-20.10.9-3.*

sudo systemctl start docker

sudo echo '{"exec-opts": ["native.cgroupdriver=systemd"]}' > /etc/docker/daemon.json

sudo systemctl restart docker

• Ubuntu：参考下面脚本

## ubuntu
curl -fsSL https://mirrors.aliyun.com/docker-ce/linux/ubuntu/gpg | sudo apt-key add -

sudo add-apt-repository "deb [arch=amd64] https://mirrors.aliyun.com/docker-ce/linux/ubuntu $(lsb_release -cs) stable"

sudo apt-get update

sudo apt-get -y install docker-ce=5:20.10.24~3-0~ubuntu-* docker-ce-cli=5:20.10.24~3-0~ubuntu-*

sudo systemctl start docker

sudo echo '{"exec-opts": ["native.cgroupdriver=systemd"]}' > /etc/docker/daemon.json

sudo systemctl restart docker

2. 启动 PostgreSQL + pgvector 容器服务

打开终端，执行以下命令，一键完成 PostgreSQL 环境部署：

cat <<'EOF' > init_pgvector.sh
#!/bin/bash

# 创建 docker-compose.yml 文件
cat <<YML > docker-compose.yml
version: "3"
services:
  db:
    container_name: pgvector-db
    hostname: 127.0.0.1
    image: pgvector/pgvector:pg16
    ports:
      - 5432:5432
    restart: always
    environment:
      - POSTGRES_DB=api
      - POSTGRES_USER=root
      - POSTGRES_PASSWORD=123456
YML

# 启动容器服务（后台运行）
docker-compose up --build -d

# 等待容器启动后，进入数据库并启用向量扩展
echo "等待容器启动中..."
sleep 5

docker exec -it pgvector-db psql -U root -d api -c "CREATE EXTENSION IF NOT EXISTS vector;"
EOF

# 赋予执行权限并运行脚本
chmod +x init_pgvector.sh
./init_pgvector.sh

执行完毕后，本地 PostgreSQL 将自动启用 pgvector 插件，作为私有化向量检索引擎，为 CloudCanal RAG 提供底层存储支持。

部署 CloudCanal 私有化版本​

下载并解压 CloudCanal 私有部署版本，完成全新安装。

RAG 服务构建​

添加数据源​

登录 CloudCanal 控制台，点击 「数据源管理」>「新增数据源」，添加以下数据源：

文件数据源（SshFile）​

用于读取本地或远程服务器中的 Markdown 文档。添加步骤如下：

• 类型选择：自建 > SshFile
• 配置说明：
  • 网络地址：填写目标服务器的 IP 地址（本机如** 127.0.0.1**）和 SSH 端口（默认端口为 22）。
  • 账号密码：填写用于 SSH 登录该服务器的用户名和密码。

如果你在 Mac 或 Linux 上操作，可以直接填写当前用户名和本地密码即可。

• 额外参数 fileSuffixArray：填写 .md，用于筛选 Markdown 文件，避免无关文件被加载。
• 额外参数 dbsJson：用于指定要同步的文档目录。你可以复制系统提供的默认值，并将其中的 schema 字段修改为你实际存放 Markdown 文件的根目录路径。

[
  {
    "db":"cc_virtual_fs",
    "schemas":[
      {
        "schema":"/Users/barry/source/cloudcanal-doc-v2",
        "tables":[]
      }
    ]
   }
]

• db：虚拟文件库的逻辑名称，默认保持为 cc_virtual_fs 即可，无需修改。
• schema：表示你希望读取的本地或远程目录路径，CloudCanal 会以该路径作为文档扫描入口。例如：/Users/barry/source/cloudcanal-doc-v2。该字段必须填写为绝对路径。
• tables：用于指定目录中要处理的具体文件名，若设置为空数组（[]），则表示自动抓取该目录下所有符合后缀规则（如 .md）的文件，无需逐一列出文件名。

向量数据库（PostgreSQL）​

用于存储通过大模型编码后的文档向量，是整个 RAG 检索流程的核心数据源。

• 类型选择：自建 > PostgreSQL
• 配置说明：
  • 网络地址：localhost:5432
  • 用户名：root** **
  • 密码：123456

大模型平台（Ollama）​

CloudCanal 支持通过 Ollama 提供完整的向量生成与推理能力，适用于完全私有化的 RAG 场景。

• 类型选择：自建 > Ollama
• 配置说明：
  • 网络地址：localhost:11434
  • 额外参数 llmEmbedding：

{
  "qwen3:8b": {
    "dimension": 4096
  }
}

• 额外参数 llmChat：

{
  "qwen3:8b": {
    "temperature": 1,
    "topP": 0.9,
    "showReasoning": false
  }
}

RagApi 服务（CloudCanal）​

RagApi 是最终面向用户提供问答接口的服务模块，用于对接 Chat 界面或上层应用。

• 类型选择：自建 > RagApi
• 配置说明：
  • 网络地址：localhost:18089
  • API 密钥：自定义一个字符串（例如 my-cc-rag-key），用于后续调用 RagApi 接口时进行身份验证。

创建任务 1：数据向量化​

1. 点击 同步任务 > 创建任务。
2. 选择以下数据源，并点击 测试连接 确认网络与权限正常。
   • 源端：SshFile
   • 目标端：PostgreSQL

3. 在 功能配置 页面，任务类型选择 全量迁移，任务规格选择默认 2 GB 即可。

4. 在 表&action过滤 页面，进行以下配置：

   1. 选择需要定时数据迁移的文件，可同时选择多个。
   2. 点击 批量修改目标名称 > 统一表名 > 填写表名（如 knowledge_base），并确认，方便将不同文件向量化并写入同一个表。

5. 在 数据处理 页面，进行以下配置：

   1. 点击 配置大模型 > Ollama，选择刚添加的大模型实例，并选择 qwen3:8b。
   2. 点击 批量操作 > 大模型嵌入，选择需要嵌入的字段，并全选表。

6. 在 创建确认 页面，点击 创建任务，开始运行。

创建任务 2：RagApi 服务​

1. 点击 同步任务 > 创建任务。
2. 选择以下数据源，并点击 测试连接 确认网络与权限正常。
   • 源端：已配置的 PostgreSQL（向量表所在库）
   • 目标端：RagApi

3. 在 功能配置 页面，任务类型选择 全量迁移，任务规格选择默认 2 GB 即可。
4. 在 表&action过滤 页面，选择要使用的向量表（可多选）。

5. 在 数据处理 页面，配置大模型：

   1. 嵌入模型：选择 Ollama。

      注意: PostgreSQL 中的向量维度需与选定嵌入模型一致。

   2. 聊天模型：选择 Ollama。

6. 在 创建确认 页面，点击 创建任务，系统将自动完成 RagApi 服务构建。

7. 构建完成后，可使用以下命令进行简单测试：

curl http://localhost:18089/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer my-cc-rag-key" \
  -d '{
        "messages": [
          {"role": "system", "content": "You are a helpful assistant."},
          {"role": "user", "content": "Hello!"}
        ],
        "stream": false
      }'

效果测试​

RagApi 支持通过可视化工具 CherryStudio 进行交互测试。CherryStudio 兼容 OpenAI 接口标准，适合用于接口联调、上下文调试和模型效果验证。

1. 打开 CherryStudio，点击左下角 设置图标。
2. 在 模型服务 中搜索 OpenAI，并配置如下参数：
   • API 密钥：填写在 CloudCanal 中 RagApi 设置的密钥：my-cc-rag-key
   • API 地址：http://localhost:18089

• 模型名称：CC_RAG

3. 回到对话页面：
   • 点击 添加助手 > Default Assistant。
   • 右键点击 Default Assistant > 编辑助手 > 模型设置，绑定上一步添加的模型。

4. 在对话框输入：MySQL 源端权限需要什么？，RagApi 将根据向量数据检索相关内容，并通过对话模型生成响应。

总结​

企业级 RAG 服务对数据安全有极高的要求。通过 CloudCanal 与 Ollama 的组合，可以轻松实现全私有部署 RAG 服务，打造一个真正不依赖公网、稳定可靠的企业级 RAG 解决方案。

在实时数仓的浪潮下，企业越来越重视如何以低延迟、高一致性的方式将数据写入数据湖，并结合下游分析引擎完成统一分析。Apache Paimon 作为新一代流批一体的数据湖存储引擎，因其高效写入、实时更新等能力，成为构建实时湖仓架构的理想选型。

本文将介绍如何基于 Paimon、CloudCanal、StarRocks 快速构建一套真正实时、灵活、高可维护的数据湖仓架构。

Apache Paimon 介绍​

Apache Paimon 是一个开源的、面向流计算的湖仓存储格式，源于 Apache Flink 社区（原 Flink Table Store）。它的核心创新在于将 湖存储格式 与 日志结构合并树（LSM-tree） 技术融合，为数据湖带来了强大的实时流式更新能力，这使得高吞吐、低延迟的数据摄取、变更日志追踪和高效的即时分析成为可能。

核心能力：​

• 流批一体处理：支持 Streaming Write / Snapshot Read
• 主键支持：高效 UPSERT / DELETE
• Schema 演进：新增、删除、修改列，无需重写旧数据
• ACID 事务保障：支持并发读写一致性
• 生态兼容：支持 StarRocks、Flink、Spark 等引擎
• 对象存储兼容：支持 S3、OSS 等文件系统

举个例子：数据湖中的实时订单状态变更​

以大型电商平台的实时销售大屏为例，业务团队希望能够每秒更新订单状态，例如订单从“已支付”变为“已发货”，并立即体现在可视化大屏中。

• 传统方案的挑战（Merge-on-Read）

在传统的数据湖架构中，变更数据通常会先追加写入日志区，再依赖定期的 Merge-on-Read 批处理任务将新旧数据合并成最新快照。

• 实时写入采用 追加方式，变更记录不会立即反映在查询结果中，需等待合并操作完成后才可见。
• 读旧、写新、重写文件的过程通常以分钟为单位，难以满足秒级迟需求。

• Paimon 的解决方式（LSM-tree）

Paimon 从架构层面对这一痛点进行了优化，引入类似数据库的主键更新能力。

• 订单在交易库（如 MySQL）中状态变更。
• 变更事件（如 UPDATE orders SET status = '已发货' WHERE order_id = '123'）实时写入 Paimon。
• Paimon 基于 LSM-tree，更新后读取，通常可在 秒级 内完成。

• 实际效果

下游如 StarRocks 等分析引擎可以秒级查询到更新后的状态，从而确保实时大屏始终反映业务的最新动态。

Paimon vs. Iceberg：关键特性对比​

Apache Paimon 与 Apache Iceberg 是当前数据湖领域的两大主流方案，分别在实时处理和批量分析方向各有侧重，体现了两种不同的技术演进路径。

Paimon 设计之初就面向流式数据处理，底层采用 LSM-tree 架构，原生支持主键和高效变更，特别适合 高频更新、CDC等行级数据实时入湖场景。

Iceberg 以快照机制为核心，强调 强一致性，同时也逐步增强对流式写入与变更处理的支持，具备实现近实时入湖的能力。

总体来看，Paimon 在实时更新、变更频繁的场景中具备更直接的能力优势，而 Iceberg 则在批量处理、版本治理和生态成熟度方面表现更稳健。

构建湖仓一体架构​

尽管 Paimon 可以通过 Flink CDC 写入数据，但管理作业状态、处理容错和检查点等工作，对多数团队而言仍有不小的技术门槛和运维成本。CloudCanal 提供了一种更轻量、更自动化的方式实现数据入湖。

• 外部数据源：企业的核心交易库（如 MySQL, PostgreSQL）、日志数据（Kafka）等。
• CloudCanal：
  • 基于日志实时捕获数据的变更，流式批量写入，实现秒级延迟。
  • 支持 **结构自动迁移 **和 DDL 同步。
  • 支持数据迁移同步、数据校验与订正、断点续传、任务监控、告警。
• Apache Paimon：
  • 作为湖仓底座，承接 CloudCanal 写入的实时数据流。
  • 采用 LSM-tree 架构自动去重合并，管理数据分区与后台 Compaction。
  • 基于 **S3、OSS **等其他对象存储，构建存算分离架构。
• StarRocks：直接进行实时查询分析，无需额外的数据导入或转换。

实操演示​

接下来将以 MySQL -> Paimon 链路为例，展示如何快速将数据实时同步至 Paimon，并使用 StarRocks 外表实时查询入湖数据。

步骤 1: 安装 CloudCanal​

请参考 全新安装(Docker Linux/MacOS)，下载安装 CloudCanal 私有部署版本。

步骤 2: 添加数据源​

1. 登录 CloudCanal 平台，点击 数据源管理 > 添加数据源，分别添加 MySQL 和 Paimon 数据源。

2. 添加 Paimon 数据源时，需配置额外参数，具体可参考：Paimon 数据源配置

步骤 3: 创建任务​

1. 点击 同步任务 > 创建任务。
2. 选择源和目标实例，并分别点击 测试连接。

3. 在 功能配置 页面，选择 增量同步，并勾选 全量初始化。

4. 在 表和操作过滤 页面，选择需要迁移同步的表，可同时选择多张。

5. 在 数据处理 页面，保持默认配置。

6. 在 创建确认 页面，点击 创建任务，开始运行。

任务启动后，CloudCanal 会自动完成全量数据的初始化，并实时捕获增量变更写入 Paimon。

步骤 4: 使用 StarRocks 查询数据​

Paimon 中的数据最终需要被下游分析引擎查询，StarRocks 原生支持 Paimon Catalog，可通过外部 Catalog 的方式直接查询 Paimon 中的实时数据，无需额外导入或数据转换。

1. 在 StarRocks 中创建 Paimon Catalog

在 StarRocks 中执行 CREATE EXTERNAL CATALOG 语句，即可将整个 Paimon 仓库映射进 StarRocks：

CREATE EXTERNAL CATALOG paimon_catalog
PROPERTIES
(
    "type" = "paimon",
    "paimon.catalog.type" = "filesystem",
    "paimon.catalog.warehouse" = "<s3_paimon_warehouse_path>",
    "aws.s3.use_instance_profile" = "true",
    "aws.s3.endpoint" = "<s3_endpoint>"
);

具体配置可参考：https://docs.starrocks.io/docs/data_source/catalog/paimon_catalog/#examples

2. 实时查询数据

Catalog 创建成功后，可以像 操作本地表 一样查询 Paimon 中的数据：

-- 查看 Catalog 中有哪些数据库
SHOW DATABASES FROM paimon_catalog;

-- 切换 Catalog
SET CATALOG paimon_catalog;

-- 切换至指定数据库
USE your_database;

-- 查询表数据
SELECT COUNT(*) FROM your_table LIMIT 10;

当 MySQL 中的数据发生变化后，CloudCanal 会将变更实时同步至 Paimon，StarRocks 查询到的数据也将同步更新，从而实现真正的端到端 秒级实时分析能力，无需构建复杂的 ETL 流程。

总结​

Apache Paimon 以其创新的 实时更新 能力，为现代数据湖仓架构注入了新的活力，而 CloudCanal + Paimon 的组合，则通过零代码、自动化的方式，快速落地高性能的实时湖仓，最后与 StarRocks 等下游引擎配合，打通了从数据同步到实时查询的完整闭环。

在企业级数据场景中，一个报表查询往往需要 3 张以上表的 JOIN，这类查询在数据量较大的场景下，需要数分钟甚至个把小时才能返回。

本文将简要探讨宽表技术的来龙去脉，以及它如何帮助解决多表关联带来的性能瓶颈，并结合 CloudCanal 最新推出的可视化宽表构建功能，无痛实现跨表数据的实时整合。

数据分析困境​

在结构化数据系统中，随着业务模型的复杂化，数据表之间的关联不断增多。以电商系统为例，订单、商品、用户等表结构天然具有关联性：

• 订单数据：商品 ID (关联 商品数据)、数量、总价、优惠信息、买家 ID (关联 用户数据)等
• 商品数据：名称、颜色、质地、库存、商家 ID (关联 用户数据)等
• 用户数据：账号、密码、昵称、手机、邮箱等

关系型数据库通过建立关系范式实现了存储效率最大化（冗余信息少），并优化了事务型操作的性能。但一旦进入数据分析阶段，这种表结构会给查询带来极大挑战。

在进行批量聚合、筛选等复杂分析时，如“计算过去 1 个月商品销售额 Top 10”，往往需要多表 JOIN 操作。而随着关联表数量增加，可能的执行计划（搜索空间）也呈几何级增长：

对于一个 ERP 或 CRM 系统而言，5 张表以上的关联是常态。面对如此庞大的执行计划可能性，如何高效地找到最优解 成为数据库分析查询的核心难题。

面对这一挑战，行业演进出了两种技术路线：查询优化 和 预计算。而 打宽表 是预计算的核心能力之一。

查询优化 vs 预计算​

查询优化​

面对这么多种可能性，解决方式之一就是通过减少可能性加快搜索，这就是数据库优化器术语——剪枝。其中衍生出两种技术：基于规则的优化器（RBO: Rule-Based Optimizer） 和 基于代价的优化器（CBO: Cost-Based Optimizer）。

• 基于规则的优化器（RBO）：RBO不考虑数据实际分布，而是依据预设的静态规则对 SQL 执行计划进行调整。市面上的数据库产品，都有一些通用的优化规则，如谓词下推等。不同数据库产品基于其业务属性和架构特点，还拥有独特的优化规则，如 SAP Hana 承载 SAP ERP 业务，针对全内存和大量 JOIN 的场景，其优化器规则便与其他数据库存在显著差异。

• 基于代价的优化器（CBO）：相较于 RBO，CBO 通过评估不同执行计划的 I/O、CPU 等资源消耗，选择代价最低的计划。这种优化方式会通过数据的具体分布与查询 SQL 的特点来动态调整，即使两条一模一样的 SQL，因为设置的参数值不一样，可能最后的执行计划大相径庭。CBO 一般具备一个精密复杂的统计信息子系统，包括各个表的数据量、基于主键或者关联维度的数据分布直方图等信息。

现代数据库内核一般会联合使用 RBO 和 CBO 对 SQL 进行优化。

预计算​

预计算认为 关联表之间的数据关系是确定的，所以在数据生成时就进行聚合与关联，形成一张“宽”的表结构，从而避免后续每次查询都重新进行关联。数据发生变化时，仅需增量更新对应的宽表数据。它的早期实现可以追溯到关系型数据库的 物化视图。

从计算量消耗角度而言，预计算大幅优于即时查询优化。然而，其局限性亦不容忽视：

• JOIN 语义实现受限：预计算对于非 Left Join 语义的实现逻辑复杂且代价巨大，通常难以在流式处理中高效支持。
• 大量数据更新影响稳定性：在 1 对 N 的数据关系中，若“1”端数据发生变化，可能导致宽表数据大规模更新，对服务稳定性构成挑战。
• 功能与性能权衡：预计算通常无法像即时查询那样支持各种聚合、过滤条件及函数，部分是功能实现难度，部分是性能代价考量。

最佳实践​

在真实场景中，业界普遍倾向于 “双管齐下” 的策略：预计算生成中间数据，而即时查询则在此基础上完成最终的聚合与过滤，实现功能与性能的平衡。

• 预计算：当前最流行的方案即 流计算，近些年出现的流式数据库也在尝试更好地解决这类问题，传统关系型数据库或者数据仓库实现的物化视图也是一种很好的解决方案。

• 即时查询：随着 列式、行列混合 数据结构普及，AVX 512 等新指令集落地，FPGA、GPU 等专用或高性能计算硬件尝试，以及 知识网格、分布式计算 等软件手段应用，实时分析数据库在数据筛选和聚合性能上获得了显著提升。

两者结合使用，为在线数据分析场景带来了巨大的助益。

CloudCanal 的宽表历程​

CloudCanal 在宽表构建方面的演进，实现了从“高代码门槛”到“零代码构建”的转变。早期版本中通过自定义代码支持用户在数据迁移同步过程中，自行查询关联表数据并构建宽表，再回写到目标端。尽管功能可行，但用户需要写代码，难度太高，因而未形成大规模应用。

为此，CloudCanal 实现了 可视化宽表构建。用户仅需在界面中选择需要打宽表的表，并选择关联的列，即可构建出一个既支持历史数据迁移，又支持实时更新的宽表任务。

目前，该功能已支持链路包括：

• MySQL -> MySQL/StarRocks/Doris/SelectDB
• PostgreSQL/SQL Server/Oracle/MySQL -> MySQL
• PostgreSQL -> StarRocks/Doris/SelectDB

未来将持续开放更多链路。

CloudCanal 可视化构建宽表​

宽表定义​

CloudCanal 宽表构建中的核心角色包括：

• 驱动表：仅可选择一张，作为宽表的基准数据源。
• 关联表：可选择多张，提供驱动表需要补充的字段信息。

CloudCanal 默认实现 Left Join 语义，确保返回驱动表所有数据，即使关联表中无匹配数据。

目前，CloudCanal 支持 2 种关联拓扑结构：

• 线形 关系：A.b_id = B.id and B.c_id = C.id and ...，其中所有表均不可重复选择，且不支持形成环状关联。
• 星形 关系：A.b_id = B.id and A.c_id = C.id and ...，不支持形成环状关联。

其中 A 表定义为驱动表，B、C 表定义为关联表。

数据变更处理策略​

对端为关系型数据库(如 MySQL)​

• 驱动表 INSERT：自动补全关联表字段。
• 驱动表 UPDATE/DELETE：不做关联表字段关联。
• 关联表 INSERT：如有下一层关联表，则做关联表字段关联，INSERT 操作转换为 UPDATE。
• 关联表 UPDATE：如有下一层关联表，不做关联表字段关联。
• 关联表 DELETE：如有下一层关联表，则做关联表字段关联，DELETE 操作转换为全字段为 NULL 的 UPDATE 操作。

对端为整行覆盖类数据库(如 StarRocks/Doris)​

• 驱动表 INSERT/UPDATE/DELETE：自动补全关联表字段。

• 关联表 INSERT/UPDATE/DELETE：自动忽略。

  信息

  若需实现到覆盖类数据库 关联表变更，可以分阶段完成。

  首先创建 源库 > 关系型数据库宽表 迁移同步任务，再创建 关系型数据库宽表 > 覆盖类数据库 迁移同步任务。

操作示例​

1. 进入 CloudCanal 控制台，点击 同步任务 > 创建任务。
2. 进入 选择表 步骤，选择需要打宽表的表，对端均选择宽表进行映射。
3. 进入 数据处理 步骤。

   1. 页面左侧选择 驱动表，并点击 操作 > 宽表，设定宽表定义。

      • 关联列 选择和设置，支持多个。

      • 关联表 其他列 选择，以及宽表映射字段填写，此操作主要防止多张来源表字段名在目标宽表中重复。

        信息

        1. 关联表如果级联关联其他表，务必选择其对应关联列。如 A.b_id = B.id and B.c_id = C.id，则选择 B 的其他列时，必须选择 c_id。

        2. 在复杂关联关系中，不同驱动表或关联表存在相同列名，故在选择其他列时，映射到不同列以规避冲突。

   2. 点击 确定 完成设置。

   3. 页面左侧选择 关联表，确认列和宽表列映射正确。

4. 继续完成任务创建流程，即可启动宽表同步任务。

总结​

宽表作为一种核心的数据预计算策略，有效解决了关系数据库范式化设计在 OLAP 场景下多表关联查询的性能瓶颈。CloudCanal 提供的可视化宽表构建能力，不仅大幅简化了构建流程，而且显著降低了技术门槛。它为构建具备实时性与灵活性的数据分析平台提供了关键工具，助力数据架构师和 DBA 高效构建实时数据分析层。

在之前的文章中，我们介绍了如何使用 CloudCanal 和 Ollama 搭建全栈私有的 RAG 问答服务，为企业级 RAG 应用提供了部署简单、安全可靠的解决方案。

最近，我们用这套方案，在 CloudCanal 官网上线了基于 知识库 的智能问答机器人。在发布前的测试过程中，却发现它似乎没有想象中那么“智能”，出现了答非所问、检索失焦、上下文理解有误等问题。

今天，我们将从 CloudCanal 官网问答机器人工程实践出发，深入分析传统 RAG 的瓶颈，并介绍 CloudCanal RAG 的针对性优化策略，展示如何构建一个真实可用的智能问答机器人。

传统 RAG 模式的问题​

传统 RAG 流程简化如下：

1. 原始文档 → 切分 → 向量化 → 存入向量数据库
2. 用户提问 → 转换为查询向量 → 相似度检索 → 拼接上下文
3. 构造 Prompt → 输入大模型 → 大模型推理并回答

这一流程看似十分合理，但实际操作时却会发现各种各样的问题：

数据处理粗糙，信息密度低​

• 问题表现：直接对原始文档分块向量化，缺乏摘要、关键词、标签等关键元信息，导致向量无法精准表达文本核心内容。
• 典型场景：一篇关于各数据库版本支持情况的文档，若无摘要或关键词提炼，模型在检索时很难精准匹配到“版本支持”这一核心概念，导致召回失败。

知识库未分类，检索范围模糊​

• 问题表现：将 FAQ、操作指南、更新日志等不同类型的文档混合存储在同一向量空间，未进行有效分类，容易导致检索范围定位错误。
• 典型场景：当用户提问“社区版支持哪些数据源？”时，系统可能因无法区分“产品功能”与“操作步骤”，错误地返回一篇关于“如何配置数据源连接”的指南，导致答非所问。

对话历史理解脱节，无法有效追溯​

• 问题表现：简单拼接历史对话作为上下文，缺乏结构化标注，导致模型难以理解指代关系和焦点变化。
• 典型场景：用户先问“CloudCanal 免费吗？”，再问“Hana 呢？”。若缺乏上下文关联，模型可能将第二问理解为“什么是 Hana 数据库”，而不是“CloudCanal 对 Hana 数据源的支持在社区版是否免费”，无法准确理解用户问题的含义。

用户意图识别不清，向量检索偏离目标​

• 问题表现：用户提问往往简洁甚至模糊，如果模型不能主动推断其真实意图，就会导致向量检索偏离甚至完全失效。
• 典型场景：对于一个孤立的问题“Hana 免费吗？”，若系统未能结合潜在上下文（如用户正在浏览 CloudCanal 文档）推断出“Hana”是指“CloudCanal 的数据源”，“免费”是指“社区版功能”，那么检索结果很可能跑偏，返回无关的 Hana 数据库介绍。

CloudCanal RAG：从“检索驱动”走向“理解驱动”​

传统 RAG 模型的问题在于 检索驱动、理解缺失，而 CloudCanal RAG 对以上问题做了针对性的优化。

1. 查询重构：将用户的模糊提问补全为精准问题；
2. 查询扩展：生成多种等价问法，提升向量召回率；
3. 查询路由：定位到最相关的知识模块，避免全库检索；
4. 智能检索：并行将多种查询送入向量库，智能判断最优查询方式后，结合 **分表召回 **与 相关性重排（Rerank），过滤掉弱相关或无关内容；
5. 结构化提示：用“权威事实＋任务指令”框定模型输出，杜绝幻觉。

这些能力均内置于 CloudCanal，用户只需根据自身业务场景，稍作提示词调整，即可快速实现高精度、可落地的智能问答服务。

查询案例​

下面通过一个真实的模糊查询案例来说明：

用户先问：CloudCanal 免费吗？

紧接着又问：Hana 支持吗？

在传统 RAG 中，第二问往往被理解为“什么是 Hana？”，没有关联到对话历史中的“免费”，最终返回 Hana 数据库的相关介绍，而非“CloudCanal 社区版是否支持 SAP Hana 同步”。

在 CloudCanal RAG 中，整个执行流程如下：

第一步：查询重构​

CloudCanal RAG 首先进行上下文分析，将模糊问题补全并精准化。

• 对话历史分析：检查到用户之前的对话提及了 社区版。
• 实体关联：Hana 识别为 SAP Hana 数据源，关联到核心内容 数据同步 和 社区版。
• 重构结果：Hana 支持吗？=> CloudCanal 的社区版是否支持免费同步 SAP Hana 数据源？

这一步彻底解决了传统 RAG 中 用户提问模糊、对话历史关联弱 的问题，为后续操作提供了清晰而精准的输入。

第二步：查询扩展​

为了避免单一查询方式在向量检索中的失败，CloudCanal RAG 会将重构后的问题扩展为多个语义等价的问法，提升召回率。

• 社区版 CloudCanal 能连接 SAP Hana 吗？
• 使用 CloudCanal 社区版可以免费同步 Hana 数据吗？
• CloudCanal 社区版本包含对 SAP Hana 数据源的支持吗？

这一步有效应对了 数据处理粗糙 的问题，即使原始文档的措辞与用户提问不完全一致，也能精准命中相关知识。

第三步：查询路由​

在进入具体向量检索前，CloudCanal RAG 会先判断当前问题应当查阅哪些类别的知识文档，以实现精确查询。

• 结合问题关键词和意图，判断是属于 FAQ、产品功能说明、价格策略、版本日志还是安装手册。
• 针对问题 “CloudCanal 社区版支持 SAP Hana 吗？”，系统判断属于“产品功能 + 产品定价”模块，只在相关知识表中查询。

这一步解决了 检索范围模糊 的问题，避免在全局知识库中进行大海捞针式的检索，提升了检索效率和准确性，防止不相关的知识干扰结果。

第四步：智能检索​

将多个查询并行送入向量数据库，并对返回的结果进行智能筛选。

• 分表检索：从“产品功能”和“产品定价”知识表中召回初步匹配的片段。例如：
  • 召回片段 1：...当前社区版默认支持 MySQL、PostgreSQL...
  • 召回片段 2：...SAP Hana 数据源仅在商业版中开放...
  • 召回片段 n：...
• 相关性重排 (Rerank)：对所有召回片段进行细致评估，判断其与用户核心意图的关联度，剔除不相关或弱相关的部分。这一步至关重要，它能有效过滤掉噪声信息，确保最终用于生成答案的上下文是高度准确且相关的，避免大模型被无关信息误导。

这一环节确保了最终用于生成答案的上下文是高度准确且相关的。

第五步：结构化提示词生成​

最后，构建一个高度结构化的提示词（用户可自定义），让大模型更好地推理。下面为简单的例子（实际会更复杂）：

## 角色设定
“你是 CloudCanal 的官方技术支持助手。”

## 权威材料 (事实)：
  - “SAP Hana 数据源同步功能仅在 CloudCanal 企业版中支持。”
  - “CloudCanal 社区版暂不包含对该数据源的支持。”

## 用户原始问题：
“CloudCanal 的 Hana 社区版免费吗？”

## 任务指令
“请基于权威材料，直接回答用户问题，并提供清晰指引。”

通过这种严谨的结构化方式，大模型的回答被严格限定在已验证的事实范围内，从根本上杜绝了“幻觉”的产生，保证了回答的权威性与可信度。

实际应用​

构建好 CloudCanal RAG API 后，将其转化为面向用户的智能问答服务变得轻而易举。

通过整合 CloudCanal 自己开发的机器人应用，可以将 RAG 能力快速对接至多种企业协作平台，真正实现智能问答的落地应用。

上线后，CloudCanal 官网的智能问答机器人在多种复杂场景下均能基于文档准确解答用户问题，哪怕面对模糊表达、多轮追问或不规范术语，依然能提供清晰可信的答案。

总结​

传统 RAG 偏向“被动检索”，而 CloudCanal RAG 通过一系列优化措施，让 RAG 应用走向 “主动理解 + 智能编排”。

• 从模糊提问中提炼用户真实意图
• 从上下文中构建精准查询路径
• 从海量信息中筛选最关键事实

它不仅能找答案，更能理解问题、规划流程、控制输出，为企业级 RAG 问答系统提供真正可落地的技术方案。

在之前的文章中，我们已经厘清了 GenAI 的关键概念：RAG、Function Calling、MCP、AI Agent。接下来的问题在于，如何从概念到实操？

目前，网上可以搜到很多 RAG 构建教程，但大部分教程都基于 LangChain 等，对小白来说仍有一定的入门门槛。

CloudCanal 本身作为数据同步平台，已经具备多源异构数据的接入与加工能力，在 RAG 系统构建语义搜索基础方面具备天然优势。近期 CloudCanal 推出的 RagApi 封装了向量检索与模型问答能力，为用户提供一个即插即用的智能查询接口。只需在 CloudCanal 中创建两个任务，即可获得你的专属 RAG 服务，全程无需使用代码。

CloudCanal RagApi 优势​

相比传统 RAG 架构手动部署流程，CloudCanal 提供的 RagApi 服务具有以下独特优势：

• 双任务完成全流程：文档导入 + API 发布。
• 零代码部署：无需开发，自定义配置即可构建 API 服务。
• 参数可调：支持设置向量 Top-K 数量、匹配阈值、Prompt 模板、模型温度等核心参数。
• 多模型与平台适配：支持阿里云 DashScope、OpenAI、DeepSeek 等主流模型与 API 平台。
• OpenAI API 兼容接口：直接接入现有 Chat 应用或工具链，无需额外适配。

实例演示​

本文将以 CloudCanal 官方文档为知识库，构建关于 CloudCanal 产品的 RAG 问答服务。

先展示效果：

RagApi showoff.mp4

创建这样一个 RAG 私有知识问答服务，需要用到：

• CloudCanal：自动创建 RagApi 服务
• PostgreSQL：向量数据库
• 嵌入模型：阿里云百炼平台（DashScope）的 text-embedding-v3
• 对话模型：阿里云百炼平台（DashScope）的 qwq-plus

整体工作流程如下：

操作步骤​

下载 CloudCanal​

下载安装 CloudCanal 私有部署版本。

准备资源​

1. 登录 阿里云百炼 并创建 API-KEY。
2. 本地安装免费的 PostgreSQL 数据库

#!/bin/bash

# 创建 docker-compose.yml 文件
cat <<EOF > docker-compose.yml
version: "3"
services:
  db:
    container_name: pgvector-db
    hostname: 127.0.0.1
    image: pgvector/pgvector:pg16
    ports:
      - 5432:5432
    restart: always
    environment:
      - POSTGRES_DB=api
      - POSTGRES_USER=root
      - POSTGRES_PASSWORD=123456
    volumes:
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
EOF

# 自动执行 docker-compose 启动
docker-compose up --build

# 进入 PG 命令行
docker exec -it pgvector-db psql -U root -d api

3. 创建高权限账号并登录。
4. 切换到需要建表的目标 schema (如public)。
5. 执行以下 SQL 开启向量能力。

CREATE EXTENSION IF NOT EXISTS vector;

添加数据源​

登录 CloudCanal 平台，点击 数据源管理 > 新增数据源。

添加文件：

选择 自建 > SshFile 数据源，可设定额外参数。

• 网络地址：填写目标文件所在机器和 SSH 端口（默认 22）。
• 账号密码：即登录目标机器的用户名、密码。
• 参数 fileSuffixArray：填写 .md 以过滤出所有 markdown 文件。
• 参数 dbsJson：复制默认值并修改 schema 值(即目标文件所在根目录)。

[
  {
    "db":"cc_virtual_fs",
    "schemas":[
      {
        "schema":"/Users/johnli/source/cloudcanal-doc-v2",
        "tables":[]
      }
    ]
   }
]

添加向量数据库：

选择 自建 > PostgreSQL，获取数据源并添加。

添加大模型：

选择 阿里云 > 手动填写 > DashScope 数据源，填写之前步骤获取的 API-KEY。

添加 RagApi 服务：

选择 自建 > RagApi。

• 网络地址：填写为 localhost，端口默认使用 18089。
• API 密钥：自定义一个 API-KEY，用于后续调用 RagApi 接口。

创建任务 1：数据向量化​

1. 点击 同步任务 > 创建任务。
2. 选择以下数据源，并点击 测试连接 确认网络与权限正常。
   • 源端：SshFile
   • 目标端：PostgreSQL

3. 在 功能配置 页面，任务类型选择 全量迁移，任务规格选择默认 2 GB 即可。

4. 在 表&action过滤 页面，进行以下配置：

   1. 选择需要定时数据迁移的文件，可同时选择多个。
   2. 点击 批量修改目标名称 > 统一表名 > 填写表名（如 file_vector），并确认，方便将不同文件向量化并写入同一个表。

5. 在 数据处理 页面，进行以下配置：

   1. 点击 配置大模型 > DashScope，选择刚添加的大模型实例，并选择某一个嵌入模型（如 text-embedding-v3）。
   2. 点击 批量操作 > 大模型嵌入，选择需要嵌入的字段，并全选表。

6. 在 创建确认 页面，点击 创建任务，开始运行。

创建任务 2：RagApi 服务​

1. 点击 同步任务 > 创建任务。
2. 选择以下数据源，并点击 测试连接 确认网络与权限正常。
   • 源端：已配置的 PostgreSQL（向量表所在库）
   • 目标端：RagApi

3. 在 功能配置 页面，任务类型选择 全量迁移，任务规格选择默认 2 GB 即可。
4. 在 表&action过滤 页面，选择要使用的向量表（可多选）。

5. 在 数据处理 页面，配置大模型：

   1. 嵌入模型：选择 DashScope 实例与向量数据使用的嵌入模型（如 text-embedding-v3）。

      注意: PostgreSQL 中的向量维度需与选定嵌入模型一致。

   2. 聊天模型：选择 DashScope 实例与对话模型（如 qwq-plus）。

6. 在 创建确认 页面，点击 创建任务，系统将自动完成 RagApi 服务构建。

效果测试​

RagApi 支持通过可视化工具 CherryStudio 进行交互测试。CherryStudio 兼容 OpenAI 接口标准，适合用于接口联调、上下文调试和模型效果验证。

1. 打开 CherryStudio，点击左下角 设置图标。
2. 在 模型服务 中搜索 openai，并配置如下参数：
   • API 密钥：填写在 CloudCanal 中配置的 RagApi API Key
   • API 地址：http://localhost:18089

• 模型名称：CC_RAG

3. 回到对话页面：
   • 点击 添加助手 > Default Assistant。
   • 右键点击 Default Assistant > 编辑助手 > 模型设置，绑定上一步添加的模型。

4. 在对话框输入：CloudCanal 增量同步任务延迟是什么原因？应该怎么处理？，RagApi 将根据向量数据检索相关内容，并通过对话模型生成响应。

总结​

经过简单的几步，我们完成了从零构建 RagApi 服务的全过程：从数据向量化、接入向量库、配置大模型、构建 Prompt，到部署兼容 OpenAI 接口的对话服务 RagApi。

整个过程无需编写任何代码，借助 CloudCanal 提供的可视化平台和多模型支持，企业可以快速构建具备私有知识问答能力的智能服务。

简介​

随着大语言模型飞速演进，其在知识时效、生成准确性以及与外部系统交互方面的局限也愈发显现。

为此，检索增强生成（RAG）、函数调用（Function Calling）、模型上下文协议（MCP）与 AI 智能体（AI Agent）等一系列技术相继涌现，为模型补足“知识新鲜度”与“操作执行力”。

近期 CloudCanal 也推出了 RagApi 功能，并引入了 MCP 协议。本文将聚焦 RAG、Function Calling、MCP、AI Agent 等核心概念，并介绍 CloudCanal 在 RAG 架构上的具体实现。

RAG：检索增强生成​

RAG（Retrieval-Augmented Generation） 是一种将“检索”与“生成”结合的 AI 架构。与传统大模型直接回答问题不同，RAG 会先从外部知识库（如文档库、数据库、向量数据库）中找到与用户问题相关的上下文信息，再将这些内容作为提示输入给大语言模型，从而生成更加准确的回答。

RAG 的优势​

• 模型不再完全依赖预训练知识，可结合实时或特定领域的信息；
• 对私有数据支持更强，安全性与定制性更高；
• 减少模型“胡编乱造”的情况，提高回答可靠性。

RAG 的工作流程​

1. 构建知识库：准备大量文本资料，并将其向量化，存储到向量数据库中（如 PGVector）。
2. 相似度检索：用户提问时，问题也会被向量化，然后通过相似度计算检索出最相关的文本段落。
3. 生成回答：将这些相关内容作为上下文，提供给生成模型，用于回答用户问题。

什么是向量？​

在 RAG 的工作流程中，数据向量化是第一步。那么，什么是向量呢？

为了方便理解，让我们来举个例子。对于“苹果”这个概念，人类靠经验理解，但计算机不懂“苹果”，它需要一种可以量化的方式来表示这个词。

于是，AI 会用一种叫做嵌入的方式，把“苹果”变成一个高维度的向量（Embedding），比如：

[0.12, 0.85, -0.33, ..., 0.07]（假设有 768 维）

你可以理解为：计算机试图用很多个“语义维度”来描述“苹果”这件事。例如：

• 第 12 维可能代表“是不是水果”
• 第 47 维可能代表“是不是食物”
• 第 202 维可能代表“是不是公司名字”
• 第 588 维可能代表“颜色偏红”

每一维都像是在回答一个隐形的问题，而这个维度上的数值就是模型给出的“打分”，越高表示这个特征越明显。

不同的词在这些语义维度上的“打分”不同，最终就构成了不一样的向量。

相似度如何计算？​

虽然“苹果”和“香蕉”的词面不同，但它们在语义向量空间中的表示非常相近——因为在很多语义维度上，它们的“打分”都很接近，这就是语义相似性。

我们可以用向量来描述这些词的语义特征。例如，每个词用 [类别, 可食性, 颜色] 三个维度表示如下：

在语义向量中，我们判断两个词是否相似，看的不是它们的数值大小，而是它们“指向的方向”是否一致。为此，我们通常使用 余弦相似度。

cos(θ) = (A · B) / (||A|| × ||B||)

它的核心思想是：比较两个向量之间的夹角。

• 夹角越小 → 方向越一致 → 语义越相似（cos θ 接近 1）
• 夹角越大 → 方向越偏离 → 语义差异越大（cos θ 接近 0，甚至为负）

Function Calling：让模型具备调用工具的能力​

在日常对话中，大模型通常只需返回文字答案。但当用户提出诸如“帮我查一下明天北京的天气”这类超出模型内置知识范围的问题时，就需要借助 Function Calling，即让 AI 调用外部工具来完成任务。

Function Calling 的核心作用在于让模型具备以下能力：

1. 判断当前问题是否需要使用工具
2. 自动提取参数，并以结构化 JSON 形式生成调用指令
3. 将调用交由程序执行，并接收返回结果，用于后续生成回复。

Function Calling 操作示例​

举个例子：用户说

“我明天要去北京旅游，请帮我查天气”

AI 会这样处理：

• 提取参数：城市 “北京”，时间 “明天”
• 制定计划：调用 get_weather 工具获取天气信息
• 生成调用指令：输出包含一次对 get_weather 的 tool_call，并传入所需参数

Function Calling 快速演示​

为了让你更直观地理解 Function Calling 的原理和流程，我们准备了一份演示用的 Prompt 模板。你只需将其复制到 Cherry Studio，即可观察模型如何分析用户请求、提取参数，并生成工具调用指令。

{
  "role": "AI Assistant",
  "description": "You are an AI assistant. Your primary goal is to analyze user queries and respond in a structured JSON format. If a query requires a tool and all necessary parameters are present, prepare for tool use. If a query requires a tool but essential parameters are missing, you MUST ask the user for clarification. If no tool is needed, answer directly. Your entire output MUST be a single JSON object at the root level, strictly adhering to the 'response_format'. Ensure all required fields from the schema (like 'requires_tools') are always present in your JSON output.",
  "capabilities": [
    "Analyzing user queries for intent and necessary parameters.",
    "Identifying when required parameters for a tool are missing.",
    "Strictly following instructions to set 'requires_tools' to false and use 'direct_response' to ask *only* for the specific missing information required by the tool.",
    "Remembering the initial query context (e.g., 'weather' intent) when a user provides previously missing information, and then proceeding to tool use if all tool requirements are met.",
    "Preparing and executing tool calls when the query intent matches a tool and all its defined required parameters are satisfied. Do not ask for details beyond the tool's documented capabilities.",
    "Formulating direct answers for non-tool queries or clarification questions.",
    "Detailing internal reasoning in 'thought' and, if calling a tool, a step-by-step plan in 'plan' (as an array of strings)."
  ],
  "instructions": [
    "1. Analyze the user's query and any relevant preceding conversation turns to understand the full context and intent.",
    "2. **Scenario 1: No tool needed (e.g., greeting, general knowledge).**",
    "    a. Set 'requires_tools': false.",
    "    b. Populate 'direct_response' with your answer.",
    "    c. Omit 'thought', 'plan', 'tool_calls'. Ensure 'requires_tools' and 'direct_response' are present.",
    "3. **Scenario 2: Tool seems needed, but *required* parameters are missing (e.g., 'city' for weather).**",
    "    a. **You MUST set 'requires_tools': false.** (Because you cannot call the tool yet).",
    "    b. **You MUST populate 'direct_response' with a clear question to the user asking *only* for the specific missing information required by the tool's parameters.** (e.g., if 'city' is missing for 'get_weather', ask for the city. Do not ask for additional details not specified in the tool's parameters like 'which aspect of weather').",
    "    c. Your 'thought' should explain that information is missing, what that information is, and that you are asking the user for it.",
    "    d. **You MUST Omit 'plan' and 'tool_calls'.** Ensure 'requires_tools', 'thought', and 'direct_response' are present.",
    "    e. **Do NOT make assumptions** for missing required parameters.",
    "4. **Scenario 3: Tool needed, and ALL required parameters are available (this includes cases where the user just provided a missing parameter in response to your clarification request from Scenario 2).**",
    "    a. Set 'requires_tools': true.",
    "    b. Populate 'thought' with your reasoning for tool use, acknowledging how all parameters were met (e.g., 'User confirmed city for weather query.').",
    "    c. Populate 'plan' (array of strings) with your intended steps (e.g., ['Initial query was for weather.', 'User specified city: Hangzhou.', 'Call get_weather tool for Hangzhou.']).",
    "    d. Populate 'tool_calls' with the tool call object(s).",
    "    e. **If the user just provided a missing parameter, combine this new information with the original intent (e.g., 'weather'). If all parameters for the relevant tool are now met, proceed DIRECTLY to using the tool. Do NOT ask for further, unrelated, or overly specific clarifications if the tool's defined requirements are satisfied by the information at hand.** (e.g., if tool gets 'current weather', don't ask 'which aspect of current weather').",
    "    f. Omit 'direct_response'. Ensure 'requires_tools', 'thought', 'plan', and 'tool_calls' are present.",
    "5. **Schema and Output Integrity:** Your entire output *must* be a single, valid JSON object provided directly at the root level (no wrappers). This JSON object must strictly follow the 'response_format' schema, ensuring ALL non-optional fields defined in the schema for the chosen scenario are present (especially 'requires_tools'). Respond in the language of the user's query for 'direct_response'."
  ],
  "tools": [
    {
      "name": "get_weather",
      "description": "获取指定城市当前天气 (Gets current weather for a specified city). This tool provides a general overview of the current weather. It takes only the city name as a parameter and does not support queries for more specific facets of weather (e.g., asking for only humidity or only wind speed). Assume it provides a standard, comprehensive current weather report.",
      "parameters": {
        "city": {
          "type": "string",
          "description": "城市名称 (City name)",
          "required": true
        }
      }
    }
  ],
  "response_format": {
    "type": "json",
    "schema": {
      "requires_tools": {
        "type": "boolean",
        "description": "MUST be false if asking for clarification on missing parameters (Scenario 2) or if no tool is needed (Scenario 1). True only if a tool is being called with all required parameters (Scenario 3)."
      },
      "direct_response": {
        "type": "string",
        "description": "The textual response to the user. Used when 'requires_tools' is false (Scenario 1 or 2). This field MUST be omitted if 'requires_tools' is true (Scenario 3).",
        "optional": true // Optional because it's not present in Scenario 3
      },
      "thought": {
        "type": "string",
        "description": "Your internal reasoning. Explain parameter absence if asking for clarification, or tool choice if calling a tool. Generally present unless it's an extremely simple Scenario 1 case.",
        "optional": true // Optional for very simple direct answers
      },
      "plan": {
        "type": "array",
        "items": {
          "type": "string"
        },
        "description": "Your internal step-by-step plan (array of strings) when 'requires_tools' is true (Scenario 3). Omit if 'requires_tools' is false. Each item MUST be a string.",
        "optional": true // Optional because it's not present in Scenario 1 or 2
      },
      "tool_calls": {
        "type": "array",
        "items": {
          "type": "object",
          "properties": {
            "tool": { "type": "string", "description": "Name of the tool." },
            "args": { "type": "object", "description": "Arguments for the tool." }
          },
          "required": ["tool", "args"]
        },
        "description": "Tool calls to be made. Used only when 'requires_tools' is true (Scenario 3). Omit if 'requires_tools' is false.",
        "optional": true // Optional because it's not present in Scenario 1 or 2
      }
    }
  },
  "examples": [
    // Example for Scenario 3 (direct tool use)
    {
      "query": "今天北京天气如何？",
      "response": {
        "requires_tools": true,
        "thought": "User wants current weather for Beijing. City is specified. Use 'get_weather'.",
        "plan": ["Identified city: 北京.", "Tool 'get_weather' is appropriate.", "Prepare 'get_weather' tool call."],
        "tool_calls": [{"tool": "get_weather", "args": {"city": "北京"}}]
      }
    },
    // Multi-turn example demonstrating Scenario 2 then Scenario 3
    {
      "query": "天气如何？", // Turn 1: User asks for weather, no city
      "response": { // AI asks for city (Scenario 2)
        "requires_tools": false,
        "thought": "用户询问天气但未指定城市。'get_weather'工具需要城市名。因此，我必须询问用户城市。",
        "direct_response": "请问您想查询哪个城市的天气呢？"
      }
    },
    {
      "query": "杭州", // Turn 2: User provides city "Hangzhou"
      "response": { // AI uses tool (Scenario 3)
        "requires_tools": true,
        "thought": "The user previously asked for weather and has now provided the city '杭州'. All required parameters for 'get_weather' are met. The tool provides a general current weather report.",
        "plan": ["Initial query was for weather.", "User specified city: 杭州.", "Call 'get_weather' tool for Hangzhou."],
        "tool_calls": [{"tool": "get_weather", "args": {"city": "杭州"}}]
      }
    },
    // Another multi-turn example (English)
    {
      "query": "What's the weather like today?", // Turn 1
      "response": { // AI asks for city (Scenario 2)
        "requires_tools": false,
        "thought": "User wants today's weather but no city provided. 'get_weather' tool needs the city parameter. I must ask for clarification.",
        "direct_response": "Sure, I can get the weather for you! Which city are you interested in?"
      }
    },
    {
      "query": "London", // Turn 2
      "response": { // AI uses tool (Scenario 3)
        "requires_tools": true,
        "thought": "The user asked about weather and has now specified 'London'. The 'get_weather' tool takes city and provides current weather. Proceeding with tool call.",
        "plan": ["Original intent was weather.", "City provided: London.", "Call 'get_weather' tool."],
        "tool_calls": [{"tool": "get_weather", "args": {"city": "London"}}]
      }
    },
    // Example for Scenario 1 (direct answer)
    {
      "query": "你好",
      "response": {
        "requires_tools": false,
        "direct_response": "你好！有什么我可以帮你的吗？"
        // 'thought' is optional for very simple cases
      }
    },
    // Example of user being too vague even after specifying city for a non-weather query, but we want to avoid this for weather.
    // This example is more to ensure the AI understands the "don't ask for unnecessary details for the get_weather tool" instruction.
    {
      "query": "告诉我关于巴黎的事情。", // "Tell me about Paris." (General query)
      "response": {
        "requires_tools": false, // Assuming no general knowledge tool for this example, or AI should ask for specifics.
        "thought": "User is asking for general information about Paris. This is too broad. I should ask for more specific interests if no general info tool is available.",
        "direct_response": "巴黎有很多方面，比如历史、景点、美食等等。您对哪个方面比较感兴趣呢？ (Paris has many aspects, like history, attractions, food, etc. Which aspect are you interested in?)"
      }
    },
    {
      "query": "查询杭州的天气", // User explicitly states "Hangzhou's weather"
      "response": { // AI should directly use the tool
        "requires_tools": true,
        "thought": "User explicitly asked for Hangzhou's weather. City is clear. The 'get_weather' tool is appropriate and provides a general current weather report.",
        "plan": ["User query: Hangzhou's weather.", "City: Hangzhou.", "Call 'get_weather' tool."],
        "tool_calls": [{"tool": "get_weather", "args": {"city": "杭州"}}]
      }
    }
  ]
}

Function Calling 多轮对话流程​

• 用户提问：“天气如何？”由于未明确城市信息，AI 无法直接调用工具，此时应追问用户所在城市。

• 用户回复：“杭州”。AI 获取了查询所需的关键信息，提取参数后生成 tool_calls。此时，应用程序识别到 requires_tools: true，根据 tool_calls 调用相应工具函数。

• 工具执行完成后，结果返回给 AI，AI 再基于结果进行总结并回复用户。

本质上，大模型通过自然语言理解用户意图：要完成什么任务、需要哪些信息。它会自动从对话中提取出关键参数。随后，用户的程序可根据这些参数调用对应的函数完成任务，并将执行结果返回给模型，由模型生成最终回复。

MCP：让模型更好地调用工具​

Function Calling 解决了“模型怎么调用自定义函数”，但在实际使用中还面临一些问题：

• 多个工具组成的调用链（先查天气、再发邮件）
• 工具参数结构的规范与自动注册
• 不同调用方式的适配（HTTP、本地插件等）
• 在不同模型间复用统一的工具体系

什么是 MCP？​

MCP 是由 Anthropic 推出的开放标准协议，旨在为大模型和外部工具之间的通信提供通用接口。

它不是 Function Calling 的替代，而是对其在执行层面的进一步规范和封装，使工具系统更易接入、更易管理、更易复用。

MCP 核心角色​

MCP Client​

• 向 MCP Server 请求工具列表
• 使用 HTTP 或 stdio 协议发起工具调用请求

MCP Server​

• 接收 tool_calls，根据调用内容执行对应工具
• 返回统一格式的结构化结果

MCP Server 调用方式​

HTTP 模式（StreamableHttp）​

MCP Server 作为 Web 服务运行，暴露如下接口：

• /mcp：用于接收工具调用或列出工具列表
• 支持 Event Stream（流式响应）与 JSON-RPC 协议

以下是一个天气服务的 HTTP 模式演示：

cat > streamable_weather.mjs << 'EOF'
#!/usr/bin/env node
import express from "express";
import { McpServer, ResourceTemplate } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js";
import { randomUUID } from "node:crypto";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const app = express();
app.use(express.json());

function getServer() {
    const server = new McpServer({
        name: "Weather",
        version: "1.0.0"
    });

    server.resource(
        "get_weather",
        new ResourceTemplate("weather://{city}", { list: undefined }),
        async (uri, { city }) => ({
            contents: [{
                uri: uri.href,
                text: `Resource weather for ${city}: 晴，24°C`
            }]
        })
    );

    server.tool(
        "get_weather",
        { city: z.string() },
        async ({ city }) => ({
            content: [{ type: "text", text: `Tool weather for ${city}: 明天晴，最高24°C，微风3km/h` }]
        })
    );

    server.prompt(
        "get_weather",
        { city: z.string() },
        ({ city }) => ({
            messages: [{
                role: "user",
                content: {
                    type: "text",
                    text: `请告诉我 ${city} 的天气情况`
                }
            }]
        })
    );
    return server;
}

app.post("/mcp", async (req, res) => {
    try {
        const server = getServer();
        const transport = new StreamableHTTPServerTransport({
            sessionIdGenerator: undefined,
        });

        res.on("close", () => {
            console.log("Request closed");
            transport.close();
            server.close();
        });

        await server.connect(transport);
        await transport.handleRequest(req, res, req.body);
    } catch (error) {
        console.error("Error handling MCP request:", error);
        if (!res.headersSent) {
            res.status(500).json({
                jsonrpc: "2.0",
                error: {
                    code: -32603,
                    message: "Internal server error",
                },
                id: null,
            });
        }
    }
});

app.get("/mcp", (req, res) => {
    console.log("Received GET MCP request");
    res.status(405).json({
        jsonrpc: "2.0",
        error: {
            code: -32000,
            message: "Method not allowed.",
        },
        id: null,
    });
});

app.delete("/mcp", (req, res) => {
    console.log("Received DELETE MCP request");
    res.status(405).json({
        jsonrpc: "2.0",
        error: {
            code: -32000,
            message: "Method not allowed.",
        },
        id: null,
    });
});

const PORT = process.env.PORT || 30001;
app.listen(PORT, () => {
    console.log(
        `MCP Stateless Streamable HTTP Server listening on http://localhost:${PORT}/mcp`
    );
});

EOF

# 安装依赖
npm install express @modelcontextprotocol/sdk zod

# 启动服务
node streamable_weather.mjs 

# 获取工具列表
curl -N -X POST http://localhost:30001/mcp \
  -H 'Accept: application/json, text/event-stream' \
  -H 'Content-Type: application/json' \
  -d '{
    "jsonrpc":"2.0",
    "id":1,
    "method":"tools/list",
    "params":{}
  }'

# > 返回工具
event: message
data: {"result":{"tools":[{"name":"get_weather","inputSchema":{"type":"object","properties":{"city":{"type":"string"}},"required":["city"],"additionalProperties":false,"$schema":"http://json-schema.org/draft-07/schema#"}}]},"jsonrpc":"2.0","id":1}

# 执行工具调用链
curl -N -X POST http://localhost:30001/mcp \
  -H 'Accept: application/json, text/event-stream' \
  -H 'Content-Type: application/json' \
  -d '{
    "jsonrpc":"2.0",
    "id":2,
    "method":"tools/call",
    "params":{
      "name":"get_weather",
      "arguments":{ "city":"北京" }
    }
  }'

# > 返回执行结果
event: message
data: {"result":{"content":[{"type":"text","text":"Tool weather for 北京: 明天晴，最高24°C，微风3km/h"}]},"jsonrpc":"2.0","id":2}

Stdio 模式（本地插件）​

Stdio 模式适用于本地运行的插件程序。模型与 MCP Server 通过标准输入输出进行通信，不依赖网络，适合部署在受限环境下。

以下是一个天气服务的 Stdio 模式演示：

cat > weather_stdio.mjs << 'EOF'
#!/usr/bin/env node
import { McpServer, ResourceTemplate } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
    name: "Weather",
    version: "1.0.0"
});

server.resource(
    "get_weather",
    new ResourceTemplate("weather://{city}", { list: undefined }),
    async (uri, { city }) => ({
        contents: [{
            uri: uri.href,
            text: `Resource weather for ${city}: 晴，24°C`
        }]
    })
);

server.tool(
    "get_weather",
    { city: z.string() },
    async ({ city }) => ({
        content: [{ type: "text", text: `Tool weather for ${city}: 明天晴，最高24°C，微风3km/h` }]
    })
);

server.prompt(
    "get_weather",
    { city: z.string() },
    ({ city }) => ({
        messages: [{
            role: "user",
            content: {
                type: "text",
                text: `请告诉我 ${city} 的天气情况`
            }
        }]
    })
);

const transport = new StdioServerTransport();
await server.connect(transport);
EOF

# 获取工具列表
printf '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}\n' | node weather_stdio.mjs

# > 返回工具
{"result":{"tools":[{"name":"get_weather","inputSchema":{"type":"object","properties":{"city":{"type":"string"}},"required":["city"],"additionalProperties":false,"$schema":"http://json-schema.org/draft-07/schema#"}}]},"jsonrpc":"2.0","id":1}

# 执行工具调用链：调用 get_weather
printf '{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"get_weather","arguments":{"city":"北京"}}}\n' | node weather_stdio.mjs

# > 返回执行结果
{"result":{"content":[{"type":"text","text":"Tool weather for 北京: 明天晴，最高24°C，微风3km/h"}]},"jsonrpc":"2.0","id":4}

MCP 多轮对话流程​

• 在多轮对话中，当用户输入：“北京天气如何？”

• AI 会识别出用户意图需要使用工具 get_weather，并生成如下结构化调用指令：

{
  "tool": "get_weather",
  "args": {
    "city": "北京"
  }
}

• 用户的程序会将该调用指令转发至 MCP Server，MCP Server 接收到该调用请求后，执行对应工具，并返回如下结构化结果：

• 用户的程序从 result.content 中提取文本字段，即 content.text，再进行总结和自然语言生成，最终回复用户：

“明天北京天气晴朗，最高气温24°C，微风3km/h。感谢您的咨询！如果还有其他问题，请随时提出。”

MCP 为大模型对接外部世界提供了统一且可扩展的执行框架，具备以下优势：

• 支持多种通信方式（HTTP、Stdio）；
• 支持统一的工具注册与声明；
• 可复用的跨模型调用协议；
• 易于本地或远程部署。

它与 Function Calling 搭配使用，为构建模块化、可编排、可维护的 AI Agent 系统打下了基础。

AI Agent：具备认知与行动能力的智能体​

AI Agent 是一个具备认知、行动、反思能力的完整智能系统，通常整合了 RAG（检索增强生成）、Function Calling 和 MCP：

• 用 RAG 获取知识；
• 用 Function Calling 执行操作；
• 用 MCP 统一工具调用标准。

一个成熟的 AI Agent 能够：

• 理解目标：通过自然语言指令（如“帮我查下北京的天气”）识别用户意图；
• 主动拆解任务：将复杂任务拆分为多个可执行步骤，按序执行；
• 调用外部工具：自动连接 API、数据库、搜索引擎等外部系统；
• 记忆上下文：理解当前对话历史与任务进展；
• 自我反思：在执行失败后尝试重试、重新规划或变更路径（部分 Agent 支持）。

相较于传统的聊天式 AI，AI Agent 更像一个“可指挥、可编排”的执行者，具备在真实应用中解决复杂问题的能力，广泛适用于客服、数据处理、自动化办公、个人助理等场景。

概念对比一览​

CloudCanal RAG​

近期 CloudCanal 支持了构建 RagApi 服务，基于标准的 RAG 架构，同时引入了 MCP 协议，实现了向量化、检索、问答生成与工具链调用的端到端闭环。

构建流程​

CloudCanal 构建的 RagApi 对外暴露为 OpenAI 格式的 API 接口，可直接对接业务系统或调用方。整体流程分为两个阶段：

阶段一：数据准备与嵌入（File → PostgreSQL 向量库）​

1. 数据采集与准备
   企业知识来源包括 Markdown、TXT、数据库、内部文档等。用户通过 CloudCanal 创建嵌入任务，配置数据源、模型、目标表等信息。
2. 数据切分与向量化
   CloudCanal 自动处理原始文档并生成向量嵌入，写入 pgvector 扩展的向量字段中（如 __vector 列）。

阶段二：API 构建与服务发布（PostgreSQL → RagApi）​

1. 查询向量化与语义优化
   用户问题进入对话接口后，系统首先会使用相同的嵌入模型将问题向量化。此过程中可启用以下能力模块：
   • 压缩查询（QUERY_COMPRESS）：对原始提问进行语义压缩，去除冗余、聚焦核心内容，提高向量匹配精度。
   • 扩展查询（QUERY_EXTEND）：自动引入近义词、相关概念或补充说明，扩大匹配范围，提高召回率。
2. 向量检索与知识片段选择
   在向量库中进行相似度搜索，检索结果可进一步通过 知识片段选择（KNOWLEDGE_SELECT） 进行筛选，支持多个知识库场景，系统会根据语义相关性自动选择最匹配的知识片段（支持跨表路由）。
3. Prompt 构造与上下文拼接
   系统根据用户配置的 Prompt 模板，将问题与召回内容结合，构造出最终用于模型推理的 Prompt 输入。
4. 模型推理与回答生成

生成的 Prompt 被送入指定的 Chat 模型进行推理（如 deepseek r1、qwq-plus、GPT-4o 等），模型返回最终回答内容。

5. MCP 工具链集成（可选）

如需执行任务类问题（如“查 GitHub PR 状态”、“调用企业 API”），可启用 MCP 工具链调用。

支持标准化注册的 MCP 工具（HTTP / stdio），通过 Function Calling 调用链执行外部任务，补全答案或直接完成任务。

具体操作步骤可参考：

• 1.《使用大模型将数据嵌入到 PostgreSQL 向量》https://www.clougence.com/cc-doc/bestPractice/file_to_aliyun_pg_vector
• 2.《基于 PostgreSQL 向量构建 RAG API 服务》https://www.clougence.com/cc-doc/bestPractice/pg_vector_to_rag_api

任务创建成功后，将对外暴露一个具备内置知识库支持、查询语义优化能力、可选 MCP 工具链执行的 RAG 服务，且兼容 OpenAI 的 API 协议。

相当于将用户原有的大模型 API 接口进行了增强——无需更改客户端代码，即可接入增强后的智能问答与任务执行服务。

这里可以通过可视化工具 CherryStudio 进行交互测试。CherryStudio 兼容 OpenAI 接口标准，适合用于接口联调、上下文调试和模型效果验证。

Cherry Studio 配置步骤​

1. 打开 Cherry Studio，在“模型服务”中搜索 openai。
2. 配置参数如下：
   • API 密钥：填写在 CloudCanal 中配置的 RagApi API Key
   • API 地址：http://localhost:18089

• 模型名称：填写 CC_RAG

3. 回到对话页面：

• 添加助手 → Default Assistant。
• 右键点击 Default Assistant → 编辑助手 → 模型设置，绑定上一步添加的模型。

4. 在对话窗口中输入：

CloudCanal 增量同步任务延迟是什么原因？应该怎么处理？

RagApi 将根据向量数据自动检索相关知识内容（本例使用 CloudCanal 文档知识库），结合模型生成自然语言回答。

MCP 工具链集成示例​

如果 RagApi 配置了 MCP 工具服务（如网页抓取、GitHub 查询等），模型可自动生成工具调用。

步骤如下：

1. 进入 CloudCanal，点击任务详情页右上角「功能列表」>「参数修改」。
2. 进入「目标数据源配置」，找到 mcpServers，将如下配置粘贴进去。
3. 点击右上角「生效配置」，确认参数修改内容。
4. 点击「确认」。如部分参数需重启任务生效，系统会自动提示是否重启。

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-github"
      ],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "<YOUR_TOKEN>"
      }
    },
    "mcp-server-firecrawl": {
      "command": "npx",
      "args": [
        "-y",
        "firecrawl-mcp"
      ],
      "env": {
        "FIRECRAWL_API_KEY": "<YOUR_API_KEY_HERE>"
      }
    }
  }
}

CloudCanal 将自动触发 MCP 工具执行，并进行多轮调用与总结，最终返回结果：

总结​

RAG、Function Calling、MCP 和 AI Agent 不是孤立存在的技术，而是在现实应用中彼此协同、互为补充。CloudCanal 近期支持的 RagApi 服务，融合了这些 AI 底层能力，可以零代码傻瓜式完成 RAG 服务的构建，大大降低了使用智能 AI 的门槛。

简述​

CloudCanal 4.4.0.0 版本开始支持 OceanBase (Oracle 租户) 作为源/对端的数据迁移能力。本篇文章主要介绍如何使用 CloudCanal 构建一条 OceanBase（Oracle 租户）到 OceanBase（MySQL 租户）的数据同步链路（以下简称 ObForOracle 和 OceanBase）。

技术点​

订阅增量日志​

ObForOracle 增量数据变更的订阅依赖其官方提供的日志代理组件 oblogproxy。CloudCanal 不断接收并解析 oblogproxy 推送的增量事件，最终写入对端。

前提条件​

ObForOracle 版本为 3.0、3.1、3.2。

操作示例​

步骤 1: 安装 CloudCanal​

请参考 全新安装(Docker Linux/MacOS)，下载安装 CloudCanal 私有部署版本。

步骤 2: 添加数据源​

1. 登录 CloudCanal 控制台，点击 数据源管理 > 新增数据源。

2. 数据库类型选择 ObForOracle，并填写网络地址和认证信息。

   信息

   • 网络地址格式为 ip1:sql_port1;ip2:sql_port2。
   • 多个 Root Server 用英文分号分隔。
   • 支持填写 OBProxy 地址，格式为 proxy_ip:proxy_port。

3. 配置 额外参数：

   参数	描述
   obLogProxyHost	oblogproxy server 的地址，格式为 ip:port，默认端口统一为 2983。如果需要订阅增量数据，该参数不可为空。
   clusterUrl	可以为空。为空时订阅增量数据时会使用 root server list。不为空时订阅增量数据会优先使用 cluster url。
   rpcPortList	订阅增量数据时，该参数不可为空，默认端口为 2882。如果网络地址包含多个 Root Server（假设为 3 个），此处填写格式为 2882:2882:2882。
   tenant	OceanBase Oracle 模式的租户名称，不可为空。若为空，使用 OceanBase 驱动在连接时会默认连接到 sys 租户（MySQL 模式）。
   clusterName	OceanBase 集群名称，可以为空。用于拼接连接串（用户名@租户名#集群名）以通过 OceanBase 官方驱动连接数据库。

步骤 3: 创建任务​

1. 点击 同步任务 > 创建任务。

2. 配置源和目标数据源，并分别点击 测试连接。

3. 选择 数据同步 并勾选 全量初始化。

4. 选择需要同步的表。

5. 选择表对应的列。

   信息

   如果需要选择同步的列，可先行在对端创建好表结构。

6. 点击 确认创建。

   信息

   任务创建过程将会进行一系列操作，点击 同步设置 > 异步任务，找到任务的创建记录并点击 详情 即可查看。

   ObForOracle 源端的任务创建会有以下几个步骤：

   • 结构迁移
   • 分配任务执行机器
   • 创建任务状态机
   • 完成任务创建

7. 等待任务自动流转。

   信息

   当任务创建完成，CloudCanal 会自动进行任务流转，其中的步骤包括：

   • 结构迁移: ObForOracle 源端的表定义将会迁移到对端，如果同名表在对端已经存在，则会忽略。
   • 全量数据迁移: 已存在的存量数据将会完整迁移到对端。
   • 增量数据同步: 增量数据将会持续地同步到对端数据库，并且保持实时（秒级别延迟）。

总结​

本文简单介绍了如何使用 CloudCanal 进行 ObForOracle 到 OceanBase 数据迁移同步，打通数据流动的渠道，实现端到端的精准数据传输。

简述​

SAP HANA 是一款列式存储的内存数据库系统，相比传统的硬盘存储数据库，数据处理速度更快，支持联机分析处理（OLAP）和联机事务处理（OLTP），常常用于实时分析处理、应用程序开发等场景。

MySQL 是在全球广泛使用的开源关系型数据库，历史悠久，运行稳定可靠，简便易用，灵活可扩展，因而受到许多组织的青睐。常用于 Web 应用的后端数据库、企业资源规划（ERP）系统的数据库、开发和测试数据库等。

本篇文章主要介绍如何使用 CloudCanal 构建一条 HANA 到 MySQL 的数据同步链路。

技术点​

数据同步整体流程​

CloudCanal 实现 HANA 源端增量数据同步，主要使用其触发器捕获变更事件。整体流程如下：

• 安装触发器，通过触发器捕获增量变更数据
• 记录位点，记录增量数据同步的起点
• 执行全量数据迁移
• 执行增量数据同步

表级别 CDC 表​

CloudCanal 实现了表级别的 CDC 表设计，每张源表都对应一张 CDC 表，CDC 表的结构仅在原表结构的基础上增加了几个位点字段，用于增量同步。

相比于所有数据写入单一 CDC 表，表级别的 CDC 表更加独立，方便多次订阅表。此外，触发器只需要执行 INSERT 语句，因此对于字段较多的表也能够快速执行。扫描消费 CDC 数据时，不需要做额外的处理，消费更简单。

原表：

CREATE COLUMN TABLE "SYSTEM"."TABLE_TWO_PK" (
  "ORDERID" INTEGER NOT NULL ,
  "PRODUCTID" INTEGER NOT NULL ,
  "QUANTITY" INTEGER,
  CONSTRAINT "FANQIE_pkey_for_TA_171171268" PRIMARY KEY ("ORDERID", "PRODUCTID")
)

CDC 表：

CREATE COLUMN TABLE "SYSTEM"."SYSTEMDB_FANQIE_TABLE_TWO_PK_CDC_TABLE" (
  "ORDERID" INTEGER,
  "PRODUCTID" INTEGER,
  "QUANTITY" INTEGER,
  "__$DATA_ID" BIGINT NOT NULL ,
  "__$TRIGGER_ID" INTEGER NOT NULL ,
  "__$TRANSACTION_ID" BIGINT NOT NULL ,
  "__$CREATE_TIME" TIMESTAMP,
  "__$OPERATION" INTEGER NOT NULL 
);
-- other index

触发器 (INSERT)：

CREATE TRIGGER "FANQIE"."CLOUD_CANAL_ON_I_TABLE_TWO_PK_TRIGGER_104" AFTER INSERT ON "SYSTEM"."TABLE_TWO_PK" REFERENCING NEW ROW NEW FOR EACH ROW 
BEGIN 
  DECLARE EXIT HANDLER FOR SQLEXCEPTION BEGIN  END; 
  IF 1=1 THEN 
    INSERT INTO "SYSTEM"."SYSTEMDB_FANQIE_TABLE_TWO_PK_CDC_TABLE" (__$DATA_ID, __$TRIGGER_ID, __$TRANSACTION_ID, __$CREATE_TIME, __$OPERATION, "ORDERID","PRODUCTID","QUANTITY") 
    VALUES( 
      "SYSTEM"."CC_TRIGGER_SEQ".NEXTVAL, 
      433, 
      CURRENT_UPDATE_TRANSACTION(), 
      CURRENT_UTCTIMESTAMP, 
      2, 
      :NEW."ORDERID" ,
      :NEW."PRODUCTID" ,
      :NEW."QUANTITY"  
    ); 
  END IF; 
END;

表级别任务位点​

在表级别 CDC 表模式下，同步增量数据时，每个表都有自己的位点，原有的单一位点无法满足这种同步需求。

因此，CloudCanal 引入了表级别的增量同步位点，确保每个表能够消费各自对应的增量同步位点。位点的具体体现为：

[
  {
    "db": "SYSTEMDB",
    "schema": "FANQIE",
    "table": "TABLE_TWO_PK",
    "dataId": 352,
    "txId": 442441,
    "timestamp": 1715828416114
  },
  {
    "db": "SYSTEMDB",
    "schema": "FANQIE",
    "table": "TABLE_TWO_PK_2",
    "dataId": 97,
    "txId": 11212,
    "timestamp": 1715828311123
  },
  ...
]

这样的设计有以下好处：

• 位点精细控制：每个表都有自己的增量同步位点，在增量任务中可以重新消费特定表中的增量数据，而无需消费所有表的数据，实现更加精细的控制，减少不必要的数据传输和处理，提高同步效率。

• 数据并行处理：由于每个表有自己的位点，可以实现表级别的并行处理。不同表的增量数据可以同时处理，避免了单一位点导致的串行处理瓶颈，从而加快了同步速度。

操作示例​

步骤 1: 安装 CloudCanal​

请参考 全新安装(Docker Linux/MacOS)，下载安装 CloudCanal 私有部署版本。

步骤 2: 添加数据源​

登录 CloudCanal 控制台，点击 数据源管理 > 新增数据源 。

步骤 3: 创建任务​

1. 点击 同步任务 > 创建任务。

2. 配置源和目标数据源。

   1. 选择源和目标实例，并分别点击 测试连接。
   2. 在目标实例下方 高级配置 中选择源端 CDC 表模式：单 CDC 表 / 表级 CDC 表。

3. 选择 数据同步 并勾选 全量初始化。

4. 选择需要同步的表。

5. 选择表对应的列。

   信息

   如果需要选择同步的列，可先行在对端创建好表结构。

6. 点击 确认创建。

   信息

   任务创建过程将会进行一系列操作，点击 同步设置 > 异步任务，找到任务的创建记录并点击 详情 即可查看。

   HANA 源端的任务创建会有以下几个步骤：

   • 结构迁移
   • 初始化 HANA CDC 表以及对应触发器
   • 分配任务执行机器
   • 创建任务状态机
   • 完成任务创建

7. 等待任务自动流转。

   信息

   当任务创建完成，CloudCanal 会自动进行任务流转，其中的步骤包括：

   • 结构迁移: HANA 源端的表定义将会迁移到对端，如果同名表在对端已经存在，则会忽略。
   • 全量数据迁移: 已存在的存量数据将会完整迁移到对端。
   • 增量数据同步: 增量数据将会持续地同步到对端数据库，并且保持实时（秒级别延迟）。

总结​

本文简单介绍了如何使用 CloudCanal 进行 HANA 到 MySQL 数据迁移同步，打通数据流动的渠道，实现端到端的精准数据传输。

简述​

Kafka 为处理实时数据提供了一个统一、高吞吐、低延迟的平台，其持久化层本质上是一个“按照分布式事务日志架构的大规模发布/订阅消息队列”，这使它作为企业级基础设施来处理流式数据非常有价值。因此实现 Kafka 到 Kafka 的数据同步也成了一项重要工作。

本篇文章主要介绍如何使用 CloudCanal 构建一条 Kafka 到 Kafka 的数据同步链路。

技术点​

消费者消息推送​

在任务创建后，CloudCanal 会自动创建消费组，并订阅需要同步消息的 Topic。CloudCanal 从源端拉取到消息后，会将消息推送到目标端。

心跳机制​

在源端 Kafka 未产生消息时，CloudCanal 便无法正常感知消息的延时时间。

我们采用心跳模式解决这个问题。在 打开 Kafka 源端心跳 后，CloudCanal 会监测所有分区的消费位点，若所有分区的最新的位点与当前位点差值均小于设定的最长位点间隔（通过 dbHeartbeatToleranceStep 参数设置），则会产生一条包含当前系统时间的心跳记录。CloudCanal 在消费到该记录后，会根据其包含的时间计算延迟。

操作示例​

步骤1: 配置 Kafka 权限​

参考 Kafka 需要的权限 文档，设置 CloudCanal 需要的账号权限。

步骤2: 安装 CloudCanal​

请参考 全新安装(Docker Linux/MacOS)，下载安装 CloudCanal 私有部署版本。

步骤 3: 添加数据源​

登录 CloudCanal 控制台，点击 数据源管理 > 新增数据源 。

步骤4: 创建任务​

1. 点击 同步任务 > 创建任务。

2. 选择源和目标数据源，并分别点击 测试连接。

3. 选择同步的 消息格式。

   信息

   倘若没有特定的消息格式，请选择 原始消息格式。

4. 选择 增量同步。

5. 选择需要同步的 Topic。

6. 点击 确认创建。

   信息

   任务创建过程将会进行一系列操作，点击 同步设置 > 异步任务，找到任务的创建记录并点击 详情 即可查看。

   Kafka 源端的任务创建会有以下几个步骤：

   • 结构迁移
   • 分配任务执行机器
   • 创建任务状态机
   • 完成任务创建

7. 等待任务自动流转。

   信息

   当任务创建完成，CloudCanal 会自动进行任务流转，其中的步骤包括：

   • 结构迁移: Kafka 源端会自动为对端创建 Topic，如果目标 Topic 在对端已存在，则会忽略。
   • 增量数据同步: 增量数据将会持续地同步到对端数据库，并且保持实时（秒级别延迟）。

总结​

本文简单介绍了如何使用 CloudCanal 进行 Kakfa 到 Kafka 数据迁移同步，助力企业快速构建数据管道，增强数据分析能力。

简述​

PostgreSQL 是一个历史悠久且广泛使用的数据库，不仅具备标准的关系型数据库能力，还具有相当不错的复杂 SQL 执行能力。用户常常会将 PostgreSQL 应用于在线事务型业务，以及部分数据分析工作，所以 PostgreSQL 到 PostgreSQL 数据迁移同步成为了一个重要工作。

本文主要介绍如何通过 CloudCanal 实现 PostgreSQL 到 PostgreSQL 数据迁移同步。

技术点​

使用 PostgreSQL 逻辑复制机制​

CloudCanal 使用 PostgreSQL 逻辑复制 (Logical) 机制实现对其增量数据的订阅。

发布（Publication）和同步任务一一关联，任务修改订阅后也会自动增减 Publication 中的表。

Trigger 方式实现 DDL 同步​

DDL 同步对于在线数据库灾备等场景必不可少，但 PostgreSQL 逻辑复制并未提供 DDL 操作。

为此，我们采用了行业通用的 Trigger 方式捕获 DDL 操作，并且自动写入一张普通表 cc_pg_ddl_capture_tab。CloudCanal 正常订阅该表即可获取 DDL 操作，和普通 DML 增量订阅机制一致，可准确地保障 DDL 和涉及表 DML 事件顺序。

双向同步防循环能力​

数据库多活能力要求常常出现在一些在线数据库核心应用场景中，对于数据同步工具，主要需要做到同步数据无循环。

对于 PostgreSQL 之间的多活同步防循环，我们采用同一事务标记操作方式实现。

通过一张额外的标记表，同步任务写入正常业务数据到对端时，在同一个事务中对标记表做操作。当任务从 PostgreSQL 中获取到该标记表事件时，则忽略当前事务所有操作，即达到防循环目的。

操作示例​

步骤 1: 修改 PostgreSQL 日志级别​

1. 参考 PostgreSQL 需要的权限 文档创建用户和授予权限。
2. 设置 PostgreSQL wal_level 为 logical。

3. 设置同步账号网络权限。

4. 重启 PostgreSQL 生效。

步骤 2: 安装 CloudCanal​

请参考 全新安装(Docker Linux/MacOS)，下载安装 CloudCanal 私有部署版本。

步骤 3: 添加数据源​

登录 CloudCanal 控制台，点击 数据源管理 > 新增数据源。

步骤 4: 创建任务​

1. 点击 同步任务 > 创建任务。

2. 选择源和目标数据源，并分别点击 测试连接。

3. 选择 数据同步 并勾选 全量初始化。

   信息

   勾选 同步 DDL 将会自动创建对应的 DDL 捕获 trigger 和 event，需要较高权限。

4. 选择需要同步的索引。

5. 选择索引对应的列。

   信息

   如果需要选择同步的列，可先行在对端创建好表即可。

6. 点击 确认创建。

   信息

   任务创建过程将会进行一系列操作，点击 同步设置 > 异步任务，找到任务的创建记录并点击 详情 即可查看。

   PostgreSQL 源端的任务创建会有以下几个步骤：

   • 结构迁移
   • 初始化 DDL 捕获触发器和表
   • 初始化 PostgreSQL 增量复制位点
   • 分配任务执行机器
   • 创建任务状态机
   • 完成任务创建

7. 等待任务自动步骤流转。

   信息

   当任务创建完成，CloudCanal 会自动进行任务流转，其中的步骤包括：

   • 结构迁移: 将 PostgreSQL 源端的表结构迁移到对端，如果同名表在对端已存在，则忽略。
   • 全量数据迁移: 已存在的存量数据将会完整迁移到对端，支持断点续传。
   • 增量数据同步: 增量数据将会持续地同步到对端数据库，并且保持实时（秒级别延迟）。

总结​

本文简单介绍了如何使用 CloudCanal 进行 PostgreSQL 到 PostgreSQL 数据迁移同步。

PostgreSQL 作为流行的关系型数据库，通过 CloudCanal 数据迁移同步加持，让数据进出更加便利和顺畅。

简述​

本篇文章主要介绍如何使用 CloudCanal 构建一条 Hana 到 PostgreSQL 的数据同步链路。

技术点​

表级别 CDC 表​

CloudCanal 在实现 Hana 源端增量同步时，最初采用的是单 CDC 表的模式，即所有订阅表的增量数据（插入、更新、删除）通过触发器统一写入同一张 CDC 表。这样设计的初衷是简化架构和实现方式，但是同时也带来了一些问题。

• 触发器执行效率低：采用单个 CDC 表时，我们将订阅表的字段值拼接成 JSON 字符串。虽然这种方式统一，但增加了触发器的复杂性。当字段数量超过 300 个时，触发器效率显著下降，影响同步性能。

• 增量数据积压：所有订阅表的变更数据集中写入单个 CDC 表，当 A 表增量数据较多而 B 表较少时，混合写入会导致无法及时处理 B 表数据，造成 B 表数据积压，影响同步及时性。

后续我们对于单 CDC 表模式进行了优化。本次优化实现了表级别的 CDC 表设计，每张源表都对应一张 CDC 表，CDC 表的结构仅在原表结构的基础上增加了几个位点字段，用于增量同步。

原表：

CREATE COLUMN TABLE "SYSTEM"."TEST" (
  "TEST1" INTEGER NOT NULL ,
  "TEST2" INTEGER NOT NULL ,
  "TEST3" INTEGER,
  CONSTRAINT "TEST_KEY" PRIMARY KEY ("TEST1", "TEST2")
)

CDC 表：

CREATE COLUMN TABLE "SYSTEM"."SYSTEM_TEST_CDC_TABLE" (
  "TEST1" INTEGER,
  "TEST2" INTEGER,
  "TEST3" INTEGER,
  "__$DATA_ID" BIGINT NOT NULL ,
  "__$TRIGGER_ID" INTEGER NOT NULL ,
  "__$TRANSACTION_ID" BIGINT NOT NULL ,
  "__$CREATE_TIME" TIMESTAMP,
  "__$OPERATION" INTEGER NOT NULL 
);
-- other index

触发器 (INSERT)：

CREATE TRIGGER "SYSTEM"."CLOUD_CANAL_ON_I_TEST_TRIGGER_TEST" AFTER INSERT ON "SYSTEM"."TEST" REFERENCING NEW ROW NEW FOR EACH ROW 
BEGIN 
  DECLARE EXIT HANDLER FOR SQLEXCEPTION BEGIN END; 
  IF 1=1 THEN 
    INSERT INTO "SYSTEM"."SYSTEM_TEST_CDC_TABLE" ("__$DATA_ID", "__$TRIGGER_ID", "__$TRANSACTION_ID", "__$CREATE_TIME", "__$OPERATION", "TEST1", "TEST2", "TEST3") 
    VALUES( 
      "SYSTEM"."CC_TRIGGER_SEQ".NEXTVAL, 
      433, 
      CURRENT_UPDATE_TRANSACTION(), 
      CURRENT_UTCTIMESTAMP, 
      2, 
      :NEW."TEST1" ,
      :NEW."TEST2" ,
      :NEW."TEST3"  
    ); 
  END IF; 
END;