flink-cdc Documentation

Flink CDC is a streaming data integration tool

6,452 stars Java Java 1 file · ~24,414 tokens #batch#cdc#change-data-capture#data-integration#data-pipeline#distributed#elt#etl
RAW Doc

Docker & Self-Hosted Deployment Verified OCI

Deploy flink-cdc on any server, homelab, or cloud container platform.

1-Click Terminal Command
bash (docker run) 
docker run -d --name flink-cdc --restart always apache/flink-cdc:latest

File: README.md

Flink CDC is a distributed data integration tool for real-time data and batch data, built on top of Apache Flink. It prioritizes efficient end-to-end data integration and offers enhanced functionalities such as full database synchronization, sharding table synchronization, schema evolution and data transformation.

API Layers

Flink CDC provides three API layers for different usage scenarios:

1. YAML API (Pipeline API)

The YAML API provides a declarative, zero-code approach to define data pipelines. Users describe the source, sink, routing, transformation, and schema evolution rules in a YAML file and submit it via the flink-cdc.sh CLI.

Please refer to the Quickstart Guide for detailed setup instructions.

yaml
source:
  type: mysql
  hostname: localhost
  port: 3306
  username: root
  password: 123456
  tables: app_db.\.*

sink:
  type: doris
  fenodes: 127.0.0.1:8030
  username: root
  password: ""

# Transform data on-the-fly
transform:
  - source-table: app_db.orders
    projection: id, order_id, UPPER(product_name) as product_name
    filter: id > 10 AND order_id > 100
    
# Route source tables to different sink tables
route:
  - source-table: app_db.orders
    sink-table: ods_db.ods_orders
  - source-table: app_db.shipments
    sink-table: ods_db.ods_shipments
  - source-table: app_db.\.*
    sink-table: ods_db.others

pipeline:
  name: Sync MySQL Database to Doris
  parallelism: 2
  schema.change.behavior: evolve  # Support schema evolution

Pipeline connectors:

Doris | Elasticsearch | Fluss | Hudi | Iceberg | Kafka | MaxCompute | MySQL | OceanBase | Oracle | Paimon | PostgreSQL | StarRocks

See the connector overview for a full list and configurations.

2. SQL API (Table/SQL API)

The SQL API integrates with Flink SQL, allowing users to define CDC sources using SQL DDL statements. Deploy the SQL connector JAR to FLINK_HOME/lib/ and use it directly in Flink SQL Client:

sql
CREATE TABLE mysql_binlog (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3),
  PRIMARY KEY(id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'flinkuser',
  'password' = 'flinkpw',
  'database-name' = 'inventory',
  'table-name' = 'products'
);

SELECT id, UPPER(name), description, weight FROM mysql_binlog;

Available SQL connectors (dependencies bundled):

MySQL | PostgreSQL | Oracle | SQL Server | MongoDB | OceanBase | TiDB | Db2 | Vitess

See the source connector overview for a full list and configurations.

3. DataStream API

The DataStream API provides programmatic access for building custom Flink streaming applications. Add the corresponding connector as a Maven dependency:

xml
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-mysql-cdc</artifactId>
  <version>${flink-cdc.version}</version>
</dependency>

Available source connectors:

MySQL | PostgreSQL | Oracle | SQL Server | MongoDB | OceanBase | TiDB | Db2 | Vitess

All artifacts use group ID org.apache.flink. See the DataStream API packaging guide for a complete pom.xml example.

Flink CDC Supported Flink Versions Notes
3.6 1.20, 2.2
3.5 1.19, 1.20
3.4 1.19, 1.20
3.3 1.19, 1.20
3.2 1.17, 1.18, 1.19
3.1 1.16, 1.17, 1.18, 1.19 Only Flink CDC 3.1.1 supports Flink 1.19
3.0 1.14, 1.15, 1.16, 1.17, 1.18 Pipeline API requires Flink 1.17 and above
2.4 1.13, 1.14, 1.15, 1.16, 1.17 Flink CDC 1.x and 2.x does not support Pipeline API, same below
2.3 1.13, 1.14, 1.15, 1.16
2.2 1.13, 1.14
2.1 1.13
2.0 1.13
1.4 1.13
1.3 1.12
1.2 1.12
1.1 1.11
1.0 1.11

See the Pipeline connector overview and source connector overview for details.

Join the Community

There are many ways to participate in the Apache Flink CDC community. The
mailing lists are the primary place where all Flink
committers are present. For user support and questions use the user mailing list. If you've found a problem of Flink CDC,
please create a Flink JIRA and tag it with the Flink CDC tag.
Bugs and feature requests can either be discussed on the dev mailing list or on Jira.

Contributing

Welcome to contribute to Flink CDC, please see our Developer Guide
and APIs Guide.

License

Apache 2.0 License.

Special Thanks

The Flink CDC community welcomes everyone who is willing to contribute, whether it's through submitting bug reports,
enhancing the documentation, or submitting code contributions for bug fixes, test additions, or new feature development.
Thanks to all contributors for their enthusiastic contributions.

File: docs/content.zh/docs/get-started/_index.md

title: "入门指南" bold: true bookCollapseSection: true weight: 1

File: docs/content.zh/docs/get-started/introduction.md

title: "项目介绍"
weight: 1
type: docs
aliases:

  • /get-started/introduction/

Flink CDC 是一个基于流的数据集成工具，旨在为用户提供一套功能更加全面的编程接口（API）。
该工具使得用户能够以 YAML 配置文件的形式，优雅地定义其 ETL（Extract, Transform, Load）流程，并协助用户自动化生成定制化的 Flink 算子并且提交 Flink 作业。
Flink CDC 在任务提交过程中进行了优化，并且增加了一些高级特性，如表结构变更自动同步（Schema Evolution）、数据转换（Data Transformation）、整库同步（Full Database Synchronization）以及 精确一次（Exactly-once）语义。

Flink CDC 深度集成并由 Apache Flink 驱动，提供以下核心功能：

  • ✅ 端到端的数据集成框架
  • ✅ 为数据集成的用户提供了易于构建作业的 API
  • ✅ 支持在 Source 和 Sink 中处理多个表
  • ✅ 整库同步
  • ✅具备表结构变更自动同步的能力（Schema Evolution），

环境要求

Flink CDC 有以下环境要求：

  • JDK：JDK 11 或更高版本（Flink CDC 从 3.6.0 版本开始基于 JDK 11 构建）
  • Apache Flink：Flink 1.20.x 或 Flink 2.2.x

{{< hint info >}}
在运行 Flink CDC 之前，请确保已安装正确的 JDK 版本。您可以使用 java -version 命令验证 Java 版本。
{{< /hint >}}

支持的连接器

Flink CDC 提供了丰富的连接器生态系统，用于与各种外部系统进行交互：

连接器 类型
MySQL [Source Connector]({{< ref "docs/connectors/flink-sources/mysql-cdc" >}}) / [Pipeline Source Connector]({{< ref "docs/connectors/pipeline-connectors/mysql" >}})
Oracle [Source Connector]({{< ref "docs/connectors/flink-sources/oracle-cdc" >}}) / [Pipeline Source Connector]({{< ref "docs/connectors/pipeline-connectors/oracle" >}})
PostgreSQL [Source Connector]({{< ref "docs/connectors/flink-sources/postgres-cdc" >}}) / [Pipeline Source Connector]({{< ref "docs/connectors/pipeline-connectors/postgres" >}})
Db2 [Source Connector]({{< ref "docs/connectors/flink-sources/db2-cdc" >}})
MongoDB [Source Connector]({{< ref "docs/connectors/flink-sources/mongodb-cdc" >}})
SQL Server [Source Connector]({{< ref "docs/connectors/flink-sources/sqlserver-cdc" >}})
TiDB [Source Connector]({{< ref "docs/connectors/flink-sources/tidb-cdc" >}})
Vitess [Source Connector]({{< ref "docs/connectors/flink-sources/vitess-cdc" >}})
Apache Doris [Pipeline Sink Connector]({{< ref "docs/connectors/pipeline-connectors/doris" >}})
Elasticsearch [Pipeline Sink Connector]({{< ref "docs/connectors/pipeline-connectors/elasticsearch" >}})
Fluss [Pipeline Sink Connector]({{< ref "docs/connectors/pipeline-connectors/fluss" >}})
Hudi [Pipeline Sink Connector]({{< ref "docs/connectors/pipeline-connectors/hudi" >}})
Iceberg [Pipeline Sink Connector]({{< ref "docs/connectors/pipeline-connectors/iceberg" >}})
Kafka [Pipeline Sink Connector]({{< ref "docs/connectors/pipeline-connectors/kafka" >}})
MaxCompute [Pipeline Sink Connector]({{< ref "docs/connectors/pipeline-connectors/maxcompute" >}})
OceanBase [Pipeline Sink Connector]({{< ref "docs/connectors/pipeline-connectors/oceanbase" >}})
Paimon [Pipeline Sink Connector]({{< ref "docs/connectors/pipeline-connectors/paimon" >}})
StarRocks [Pipeline Sink Connector]({{< ref "docs/connectors/pipeline-connectors/starrocks" >}})

关于连接器下载链接，请访问 [Flink Source Connectors]({{< ref "docs/connectors/flink-sources/overview#supported-connectors" >}}) 和 [Pipeline Connectors]({{< ref "docs/connectors/pipeline-connectors/overview#supported-connectors" >}}) 页面。

Flink CDC 提供了基于 YAML 格式的用户 API，更适合于数据集成场景。以下是一个 YAML 文件的示例，它定义了一个数据管道(Pipeline)，该Pipeline从 MySQL 捕获实时变更，并将它们同步到 Apache Doris：

yaml
source:
  type: mysql
  hostname: localhost
  port: 3306
  username: root
  password: 123456
  tables: app_db.\.*
  server-id: 5400-5404
  server-time-zone: UTC

sink:
  type: doris
  fenodes: 127.0.0.1:8030
  username: root
  password: ""
  table.create.properties.light_schema_change: true
  table.create.properties.replication_num: 1

pipeline:
  name: Sync MySQL Database to Doris
  parallelism: 2

通过使用 flink-cdc.sh 提交 YAML 文件，一个 Flink 作业将会被编译并部署到指定的 Flink 集群。
请参考 [核心概念]({{<ref "docs/core-concept/data-pipeline" >}}) 以获取 Pipeline 支持的所有功能的完整文档说明。

浏览 Flink CDC 文档，开始创建您的第一个实时数据集成管道(Pipeline)。

快速开始

查看快速入门指南，了解如何建立一个 Flink CDC Pipeline：

示例 版本
MySQL to Apache Doris [1.20.x]({{< ref "docs/get-started/quickstart-for-1.20/mysql-to-doris" >}}) / [2.2.x]({{< ref "docs/get-started/quickstart-for-2.2/mysql-to-doris" >}})
MySQL to StarRocks [1.20.x]({{< ref "docs/get-started/quickstart-for-1.20/mysql-to-starrocks" >}}) / [2.2.x]({{< ref "docs/get-started/quickstart-for-2.2/mysql-to-starrocks" >}})
MySQL to Kafka [1.20.x]({{< ref "docs/get-started/quickstart-for-1.20/mysql-to-kafka" >}}) / [2.2.x]({{< ref "docs/get-started/quickstart-for-2.2/mysql-to-kafka" >}})
PostgreSQL to Fluss [1.20.x]({{< ref "docs/get-started/quickstart-for-1.20/postgres-to-fluss" >}}) / [2.2.x]({{< ref "docs/get-started/quickstart-for-2.2/postgres-to-fluss" >}})

理解核心概念

熟悉我们在 Flink CDC 中引入的核心概念，并尝试构建更复杂的数据Pipeline：

  • [Data Pipeline]({{< ref "docs/core-concept/data-pipeline" >}})
  • [Data Source]({{< ref "docs/core-concept/data-source" >}})
  • [Data Sink]({{< ref "docs/core-concept/data-sink" >}})
  • [Table ID]({{< ref "docs/core-concept/table-id" >}})
  • [Transform]({{< ref "docs/core-concept/transform" >}})
  • [Route]({{< ref "docs/core-concept/route" >}})

了解如何将 Pipeline 提交到运行在不同部署模式下的 Flink 集群：

  • [standalone]({{< ref "docs/deployment/standalone" >}})
  • [Kubernetes]({{< ref "docs/deployment/kubernetes" >}})
  • [YARN]({{< ref "docs/deployment/yarn" >}})

开发与贡献

如果您想要将 Flink CDC 连接到您定制化的外部系统，或者想要为框架本身做出贡献，以下这些部分可能会有所帮助：

  • [理解 Flink CDC API]({{< ref "docs/developer-guide/understand-flink-cdc-api" >}})，开发您自己的Flink CDC 连接器。
  • 了解如何[向 Flink CDC 提交贡献]({{< ref "docs/developer-guide/contribute-to-flink-cdc" >}})
  • 查看 Flink CDC 使用的[许可证]({{< ref "docs/developer-guide/licenses" >}})

File: docs/content.zh/docs/faq/_index.md

title: "常见问题"

bold: true

bookCollapseSection: true weight: 6

File: docs/content.zh/docs/faq/faq.md

title: "通用FAQ"
weight: 1
type: docs
aliases:

  • /faq/faq

通用FAQ

和主流的 maven 项目版本管理相同，xxx-SNAPSHOT 版本都是对应开发分支的代码，需要用户自己下载源码并编译对应的jar， 用户应该使用已经 release 过的版本，比如 flink-sql-connector-mysql-cdc-2.1.0.jar，release 过的版本maven中心仓库才会有。

Flink CDC 项目中各个connector的依赖管理和Flink 项目中 connector 保持一致。flink-sql-connector-xx 是胖包，除了connector的代码外，还把 connector 依赖的所有三方包 shade 后打入，提供给 SQL 作业使用，用户只需要在 lib目录下添加该胖包即可。flink-connector-xx 只有该 connector 的代码，不包含其所需的依赖，提供 datastream 作业使用，用户需要自己管理所需的三方包依赖，有冲突的依赖需要自己做 exclude, shade 处理。

Flink CDC 项目 从 2.0.0 版本将 group id 从com.alibaba.ververica 改成 com.ververica, 自 3.1 版本从将 group id 从 com.ververica 改成 org.apache.flink。
这是为了让项目更加社区中立，让各个公司的开发者共建时更方便。所以在maven仓库找 2.x 的包时，路径是 /com/ververica；找3.1及以上版本的包时，路径是/org/apache/flink

MySQL CDC FAQ

Q1: 使用CDC 2.x版本，只能读取全量数据，无法读取增量（binlog） 数据，怎么回事？

CDC 2.0 支持了无锁算法，支持并发读取，为了保证全量数据 + 增量数据的顺序性，依赖Flink 的 checkpoint机制，所以作业需要配置 checkpoint。 SQL 作业中配置方式：

sql
Flink SQL> SET 'execution.checkpointing.interval' = '3s';

DataStream 作业配置方式：

java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(3000);

此外，如果某些数据库的只读实例是简化过binlog的，比如阿里云RDS MySQL 5.6 只读实例，其binlog不含有变更数据，自然无法获得所需增量数据

Q2: 使用 MySQL CDC，增量阶段读取出来的 timestamp 字段时区相差8小时，怎么回事呢？

在解析binlog数据中的timestamp字段时，cdc 会使用到作业里配置的server-time-zone信息，也就是MySQL服务器的时区，如果这个时区没有和你的MySQL服务器时区一致，就会出现这个问题。

此外，如果是在DataStream作业中自定义列化器如 MyDeserializer implements DebeziumDeserializationSchema, 自定义的序列化器在解析 timestamp 类型数据时，需要参考下 RowDataDebeziumDeserializeSchema 中对 timestamp 类型的解析，用时给定的时区信息。

text
private TimestampData convertToTimestamp(Object dbzObj, Schema schema) {
        if (dbzObj instanceof Long) {
            switch (schema.name()) {
                case Timestamp.SCHEMA_NAME:
                    return TimestampData.fromEpochMillis((Long) dbzObj);
                case MicroTimestamp.SCHEMA_NAME:
                    long micro = (long) dbzObj;
                    return TimestampData.fromEpochMillis(micro / 1000, (int) (micro % 1000 * 1000));
                case NanoTimestamp.SCHEMA_NAME:
                    long nano = (long) dbzObj;
                    return TimestampData.fromEpochMillis(nano / 1000_000, (int) (nano % 1000_000));
            }
        }
        LocalDateTime localDateTime = TemporalConversions.toLocalDateTime(dbzObj, serverTimeZone);
        return TimestampData.fromLocalDateTime(localDateTime);
    }

Q3: mysql cdc支持监听从库吗？从库需要如何配置？

支持的，从库需要配置 log-slave-updates = 1 使从实例也能将从主实例同步的数据写入从库的 binlog 文件中，如果主库开启了gtid mode，从库也需要开启。

text
log-slave-updates = 1
gtid_mode = on 
enforce_gtid_consistency = on

Q4: 我想同步分库分表，应该如何配置？

通过 mysql cdc 表的with参数中，表名和库名均支持正则配置，比如 'table-name' ='user_.' 可以匹配表名 user_1, user_2,user_a表，注意正则匹配任意字符是'.' 而不是 '*', 其中点号表示任意字符，星号表示0个或多个，database-name也如此。

Q5: 我想跳过存量读取阶段，只读取 binlog 数据，怎么配置？

在 mysql cdc 表的 with 参数中指定 'scan.startup.mode' = 'latest-offset' 即可。

text
'scan.startup.mode' = 'latest-offset'.

Q6: 我想获取数据库中的 DDL事件，怎么办，有demo吗？

CDC 2.1 版本提供了 DataStream API： MysqlSource， 用户可以配置 includeSchemaChanges 表示是否需要DDL 事件，获取到 DDL 事件后自己写代码处理。

java
public void consumingAllEvents() throws Exception {
        inventoryDatabase.createAndInitialize();
        MySqlSource<String> mySqlSource =
                MySqlSource.<String>builder()
                        .hostname(MYSQL_CONTAINER.getHost())
                        .port(MYSQL_CONTAINER.getDatabasePort())
                        .databaseList(inventoryDatabase.getDatabaseName())
                        .tableList(inventoryDatabase.getDatabaseName() + ".products")
                        .username(inventoryDatabase.getUsername())
                        .password(inventoryDatabase.getPassword())
                        .serverId("5401-5404")
                        .deserializer(new JsonDebeziumDeserializationSchema())
                        .includeSchemaChanges(true) // Configure here and output DDL events
                        .build();
				... // Other processing logic                        
    }

Flink CDC 支持的.

  1. Q6 中 提供的 DataStream API 已经可以让用户获取 DDL 变更事件和数据变更事件，用户需要在此基础上，根据自己的业务逻辑和下游存储进行 DataStream 作业开发。
  2. Flink CDC 3.0以上版本支持以Pipeline的形式对Mysql整库同步。

Q8: 同一个实例下，某个库的表无法同步增量数据，其他库都可以，这是为啥？

这个问题是因为 mysql 服务器 可以配置 binlog 过滤器，忽略了某些库的 binlog。用户可以通过 show master status 命令查看 Binlog_Ignore_DB 和 Binlog_Do_DB。

mysql
mysql> show master status;
+------------------+----------+--------------+------------------+----------------------+
| File             | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set    |
+------------------+----------+--------------+------------------+----------------------+
| mysql-bin.000006 |     4594 |              |                  | xxx:1-15             |
+------------------+----------+--------------+------------------+----------------------+

Q9: 作业报错 The connector is trying to read binlog starting at GTIDs xxx and binlog file 'binlog.000064', pos=89887992, skipping 4 events plus 1 rows, but this is no longer available on the server. Reconfigure the connector to use a snapshot when needed，怎么办呢？

出现这种错误是 作业正在读取的binlog文件在 MySQL 服务器已经被清理掉，这种情况一般是 MySQL 服务器上保留的 binlog 文件过期时间太短，可以将该值设置大一点，比如7天。

mysql
mysql> show variables like 'expire_logs_days';
mysql> set global expire_logs_days=7;

还有种情况是 flink cdc 作业消费binlog 太慢，这种一般分配足够的资源即可。

Q10: 作业报错 ConnectException: A slave with the same server_uuid/server_id as this slave has connected to the master，怎么办呢？

出现这种错误是 作业里使用的 server id 和其他作业或其他同步工具使用的server id 冲突了，server id 需要全局唯一，server id 是一个int类型整数。 在 CDC 2.x 版本中，source 的每个并发都需要一个server id，建议合理规划好server id，比如作业的 source 设置成了四个并发，可以配置 'server-id' = '5001-5004', 这样每个 source task 就不会冲突了。

Q11: 作业报错 ConnectException: Received DML ‘…’ for processing, binlog probably contains events generated with statement or mixed based replication format，怎么办呢？

出现这种错误是 MySQL 服务器配置不对，需要检查下 binlog_format 是不是 ROW? 可以通过下面的命令查看

mysql
mysql> show variables like '%binlog_format%';

Q12: 作业报错 Mysql8.0 Public Key Retrieval is not allowed， 怎么办呢?

这是因为用户配置的 MySQL 用户 使用的是 sha256 密码认证，需要 TLS 等协议传输密码。一种简单的方法是使允许 MySQL用户 支持原始密码方式访问。

mysql
mysql> ALTER USER 'username'@'localhost' IDENTIFIED WITH mysql_native_password BY 'password';
mysql> FLUSH PRIVILEGES;

Q13: 作业报错 EventDataDeserializationException: Failed to deserialize data of EventHeaderV4 .... Caused by: java.net.SocketException: Connection reset， 怎么办呢 ?

这个问题一般是网络原因或者数据库繁忙引起，首先排查flink 集群 到 数据库之间的网络情况，其次可以调大 MySQL 服务器的网络参数。

mysql
mysql> set global slave_net_timeout = 120; 
mysql> set global thread_pool_idle_timeout = 120;

或者采用下面的Flink配置：

text
execution.checkpointing.interval=10min
execution.checkpointing.tolerable-failed-checkpoints=100
restart-strategy=fixed-delay
restart-strategy.fixed-delay.attempts=2147483647
restart-strategy.fixed-delay.delay= 30s

如果作业存在反压，也可能出现这个问题。你需要先处理作业的反压。

Q14: 作业报错 The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1, but the master has purged binary logs containing GTIDs that the slave requires. 怎么办呢 ?

出现这个问题的原因是的作业全量阶段读取太慢，在全量阶段读完后，之前记录的全量阶段开始时的 gtid 位点已经被 mysql 清理掉了。这种可以增大 mysql 服务器上 binlog 文件的保存时间，也可以调大 source 的并发，让全量阶段读取更快。

Q15: 在 DataStream API中构建MySQL CDC源时如何配置tableList选项？

  1. tableList选项要求表名使用数据库名，而不是DataStream API中的表名。对于MySQL CDC源代码，tableList选项值应该类似于‘my_db.my_table’。
  2. 如果要同步排除products和orders表之外的整个my_db库，tableList选项值应该类似于‘my_db.(?!products｜orders).*’。

Q16: MySQL源表中存在TINYINT(1)类型的列，且部分行的数值>1，Pipeline作业下游接收到的数据却是true/false，为什么？

这是由于MySQL连接参数tinyInt1isBit默认值为true，Flink CDC 3.3.0之前的版本未处理该参数，导致TINYINT(1)类型的数据被解析为布尔值。
若需将其转换为实际值，请将CDC升级至3.3.0+，并在source节点添加配置treat-tinyint1-as-boolean.enabled: false。
例如：

yaml
source:
  type: mysql
  ...
  treat-tinyint1-as-boolean.enabled: false

sink:
  type: ...

Postgres CDC FAQ

Q1: 发现 PG 服务器磁盘使用率高，WAL 不释放 是什么原因？

Flink Postgres CDC 只会在 checkpoint 完成的时候更新 Postgres slot 中的 LSN。因此如果发现磁盘使用率高的情况下，请先确认 checkpoint 是否开启。

Flink 中如果收到数据的 precision 大于在 Flink 中声明的类型的 precision 时，会将数据处理成 NULL。此时可以配置相应'debezium.decimal.handling.mode' = 'string' 将读取的数据用 STRING 类型 来处理。

请先确保 REPLICA IDENTITY 是 FULL。 TOAST 的数据比较大，为了节省 wal 的大小，如果 TOAST 数据没有变更，那么 wal2json plugin 就不会在更新后的数据中带上 toast 数据。为了避免这个问题，可以通过 'debezium.schema.refresh.mode'='columns_diff_exclude_unchanged_toast'来解决。

Q4: 作业报错 Replication slot "xxxx" is active， 怎么办？

当前 Flink Postgres CDC 在作业退出后并不会手动释放 slot。前往 Postgres 中手动执行以下命令：

text
select pg_drop_replication_slot('rep_slot');
    ERROR:  replication slot "rep_slot" is active for PID 162564
select pg_terminate_backend(162564); select pg_drop_replication_slot('rep_slot');

Q5: 作业有脏数据，比如非法的日期，有参数可以配置可以过滤吗？

可以的，可以在 Flink CDC 表的with 参数里 加下 'debezium.event.deserialization.failure.handling.mode'='warn' 参数，跳过脏数据，将脏数据打印到WARN日志里。 也可以配置 'debezium.event.deserialization.failure.handling.mode'='ignore'， 直接跳过脏数据，不打印脏数据到日志。

Q6: 在DataStream API中构建Postgres CDC源时如何配置tableList选项？

tableList选项要求表名使用架构名，而不是DataStream API中的表名。对于Postgres CDC source，tableList选项值应为‘my_schema.my_table’。

MongoDB CDC FAQ

Q1: MongoDB CDC 支持 全量+增量读 和 只读增量吗？

支持，默认为 全量+增量 读取；使用 'scan.startup.mode' = 'latest-offset' 参数设置为只读增量。

Q2: MongoDB CDC 支持从 checkpoint 恢复吗? 原理是怎么样的呢？

支持，checkpoint 会记录 ChangeStream 的 resumeToken，恢复的时候可以通过resumeToken重新恢复ChangeStream。其中 resumeToken 对应 oplog.rs (MongoDB 变更日志collection) 的位置，oplog.rs 是一个固定容量的 collection。当 resumeToken 对应的记录在 oplog.rs 中不存在的时候，可能会出现 Invalid resumeToken 的异常。这种情况，在使用时可以设置合适oplog.rs的集合大小，避免oplog.rs保留时间过短，可以参考 https://docs.mongodb.com/manual/tutorial/change-oplog-size/ 。另外，resumeToken 可以通过新到的变更记录和 heartbeat 记录来刷新。

Q3: MongoDB CDC 支持输出 -U（update_before，更新前镜像值）消息吗？

在 MongoDB 6.0 及以上版本，若数据库开启了前像或后像功能，可以在SQL作业中配置参数 'scan.full-changelog' = 'true'，使得数据源能够输出-U 消息，从而省去ChangelogNormalize。

在 MongoDB 6.0 版本前，MongoDB 原始的 oplog.rs 只有 INSERT, UPDATE, REPLACE, DELETE 这几种操作类型，没有保留更新前的信息，不能输出-U 消息，在 Flink 中只能实现 UPSERT 语义。在使用MongoDBTableSource 时，Flink planner 会自动进行 ChangelogNormalize 优化，补齐缺失的 -U 消息，输出完整的 +I, -U， +U， -D 四种消息， 代价是 ChangelogNormalize 优化的代价是该节点会保存之前所有 key 的状态。所以，如果是 DataStream 作业直接使用 MongoDBSource，如果没有 Flink planner 的优化，将不会自动进行 ChangelogNormalize，所以不能直接获取 —U 消息。想要获取更新前镜像值，需要自己管理状态，如果不希望自己管理状态，可以将 MongoDBTableSource 转换为 ChangelogStream 或者 RetractStream，借助 Flink planner 的优化能力补齐更新前镜像值，示例如下：

text
tEnv.executeSql("CREATE TABLE orders ( ... ) WITH ( 'connector'='mongodb-cdc',... )");

    Table table = tEnv.from("orders")
            .select($("*"));

    tEnv.toChangelogStream(table)
            .print()
            .setParallelism(1);

    env.execute();

Q4: MongoDB CDC 支持订阅多个集合吗？

支持订阅整库的 collection，例如配置 database 为 'mgdb'，并且配置 collection 为空字符串，则会订阅 'mgdb' 库下所有 collection。

也支持通过正则表达式匹配 collection，如果要监控的集合名称中包含正则表达式特殊字符，则 collection 参数必须配置为完全限定的名字空间（数据库名称.集合名称），否则无法捕获对应 collection 的变更。

Q5: MongoDB CDC 支持 MongoDB 的版本是哪些？

MongoDB CDC 基于 ChangeStream 特性实现，ChangeStream 是 MongoDB 3.6 推出的新特性。MongoDB CDC 理论上支持 3.6 及以上版本，建议运行版本 >= 4.0, 在低于3.6版本执行时，会出现错误: Unrecognized pipeline stage name: '$changeStream' 。

Q6: MongoDB CDC 支持 MongoDB 的运行模式是什么？

ChangeStream 需要 MongoDB 以副本集或者分片模式运行，本地测试可以使用单机版副本集 rs.initiate() 。在 standalone 模式下会出现错误：The $changestage is only supported on replica sets.

Q7: MongoDB CDC 报错用户名密码错误, 但其他组件使用该用户名密码都能正常连接，这是什么原因？

如果用户不是在默认的admin数据库下创建的，需要在with参数里加下 'connection.options' = 'authSource=用户所在的db'。

Q8: MongoDB CDC 是否支持 debezium 相关的参数？

不支持的，因为 MongoDB CDC 连接器是在 Flink CDC 项目中独立开发，并不依赖Debezium项目，所以不支持。

Oracle CDC FAQ

Q1: Oracle CDC 的归档日志增长很快，且读取 log 慢？

可以使用在线挖掘的模式，不写入数据字典到 redo log 中，但是这样无法处理 DDL 语句。 生产环境默认策略读取 log 较慢，且默认策略会写入数据字典信息到 redo log 中导致日志量增加较多，可以添加如下 debezium 的配置项。 'log.mining.strategy' = 'online_catalog','log.mining.continuous.mine' = 'true'。如果使用 SQL 的方式，则需要在配置项中加上前缀 'debezium.',即：

text
'debezium.log.mining.strategy' = 'online_catalog',
'debezium.log.mining.continuous.mine' = 'true'

Q2: 作业报错 Caused by: io.debezium.DebeziumException: Supplemental logging not configured for table xxx. Use command: ALTER TABLE xxx ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS， 怎么办呢？

对于 oracle11 版本，debezium 会默认把 tableIdCaseInsensitive 设置为true, 导致表名被更新为小写，因此在oracle中查询不到 这个表补全日志设置，导致误报这个Supplemental logging not configured for table 错误”。 添加 debezium 的配置项 'database.tablename.case.insensitive' = 'false'， 如果使用 SQL 的方式，则在表的 option 中添加配置项 'debezium.database.tablename.case.insensitive' = 'false'

Q3: Oracle CDC 如何切换成 XStream 的方式？

添加 debezium 的配置项 'database.connection.adapter' = 'xstream'， 如果使用 SQL 的方式，则在表的 option 中添加配置项 'debezium.database.connection.adapter' = 'xstream'

Q4: Oracle CDC 的 database-name 和 schema-name 分别是什么?

database-name 是数据库示例的名字，也就是 Oracle 的 SID schema-name 是表对应的 schema，一般而言，一个用户就对应一个 schema, 该用户的 schema 名等于用户名，并作为该用户缺省 schema。所以 schema-name 一般都是创建这个表的用户名，但是如果创建表的时候指定了 schema，则指定的 schema 则为 schema-name。 比如用 CREATE TABLE aaaa.testtable(xxxx) 的方式成功创建了表 testtable， 则 aaaa 为 schema-name。

File: docs/content.zh/docs/developer-guide/_index.md

title: 开发者指南 bold: true bookCollapseSection: true weight: 5

2. Official Technical Reference & Guides (apache/docs)

File: README.md

Apache HoraeDB (incubating) Website

[!IMPORTANT]
Apache HoraeDB (incubating) is an effort undergoing incubation at the Apache
Software Foundation (ASF), sponsored by the Apache Incubator PMC.

Please read the DISCLAIMER and a full explanation of "incubating".

This repository stores all the source files of HoraeDB website, it's built with hugo and docsy.

How to run

bash
# Start a local server for preview
hugo serve

By default hugo will listen on http://localhost:1313

Contributing

Welcome any contributions from the community. You can

  • Open an issue with any suggestions, or
  • Submit a pull request to improve the website (trivial fixes are welcome).

File: content/en/blog/2024/release-2.0.0.md

title: Release 2.0.0 date: 2024-04-23

Upgrade from 1.x.x to 2.0.0

The transition from CeresDB to Apache HoraeDB introduces several breaking changes. To facilitate upgrading from older versions to v2.0.0, specific alterations are necessary.

Upgrade Steps

  1. Setup required envs
text
export HORAEDB_DEFAULT_CATALOG=ceresdb
  1. Update config

Etcd's root should be configured both in horaedb and horaemeta

For horaedb

text
[cluster_deployment.etcd_client]
server_addrs = ['127.0.0.1:2379']
root_path = "/rootPath"

For horaemeta

text
storage-root-path = "/rootPath"
  1. Upgrade horaemeta

Horaedb will throw following errors, which is expected

text
2024-01-23 14:37:57.726 ERRO [src/cluster/src/cluster_impl.rs:136] Send heartbeat to meta failed, err:Failed to send heartbeat, cluster:defaultCluster, err:status: Unimplemented, message: "unknown service meta_service.MetaRpcService", details: [], metadata: MetadataMap { headers: {"content-type": "application/grpc"} }
  1. Upgrade horaedb

After all server upgraded, the cluster should be ready for read/write, and old data could be queried like before.

What's Changed

Breaking Changes

Features

Refactor

Fixed

Docs

Chore

New Contributors

Full Changelog: https://github.com/apache/horaedb/compare/v1.2.7...v2.0.0

File: content/en/blog/2024/release-2.1.0.md

title: Release 2.1.0
date: 2024-11-18
tags:

  • release

The Apache HoraeDB(incubating) team are pleased to announce that v2.1.0 is released, which has closed over 60 issues, including two major features:

1. New WAL implementation based on local disk.

In previous version, there is a RocksDB-based WAL. Although it works well in most cases, it has following issues:

  • Compiling from source can be a challenging task, especially since RocksDB is primarily written in C++.
  • For WAL, RocksDB can be somewhat overkill. If you are not familiar with RocksDB, tuning it can be very challenging.

With this new WAL, the above two problems are solved very well, and in performance test, the new WAL slightly outperforms the previous implementation, giving a solid foundation for future optimizations.

Interested readers can refer to the design documentation [here]({{< ref "wal_on_disk" >}}) for more details on this feature.

How to enable

text
[analytic.wal]
type = "Local"
data_dir = "/path/to/local/wal"

2. Access object store with [Apache OpenDAL](https://github.com/apache/opendal)

OpenDAL (Open Data Access Layer) is a project that provides a unified API for accessing various data storage backends.
It offers several advantages for developers and organizations. Here are some key benefits:

  • Unified API. OpenDAL provides a consistent and unified API for accessing different storage backends, such as AWS S3, Azure Blob Storage, and local file systems.
  • Optimized for Efficiency: OpenDAL is built with performance in mind. It includes optimizations to ensure efficient data access and manipulation, making it suitable for high-performance applications.
  • Comprehensive Documentation: The project provides detailed documentation, making it easier for developers to get started and understand how to use the library effectively.

In newer versions of OpenDAL, object_store integration is provided, which is very beneficial for HoraeDB code migration, as the APIs used by the upper layers remain virtually unchanged, and only the object store part needs to be abstracted to a unified OpenDAL operator:

rust
// Create a new operator
let operator = Operator::new(S3::default())?.finish();

// Create a new object store
let object_store = Arc::new(OpendalStore::new(operator));

Additionally, since the Apache OpenDAL implementation of object_store is based on the latest version of the object_store, which has breaking changes from the previous version that HoraeDB is using, we've chosen to make it compatible in order to keep the scope of this upgrade as manageable as possible.

In the process of adapting to the new API, the put_multipart interface has changed the most, so the main adaptation logic is also here, HoraeDB's approach is to encapsulate the underlying put_multipart interface to ensure that the upper layer code is not modified, the details can be found in the reference:

https://github.com/apache/horaedb/blob/v2.1.0/src/components/object_store/src/multi_part.rs

Note: The adaptation logic is only practical when parquet version < 52.0.0.

Download

Go to download pages.

Conclusion

Other bug fixes and improvements can be seen here:

As always, we warmly welcome you to join our community and share your insights.

File: content/en/blog/_index.md

title: "Blog"
weight: 1
simple_list: true
menu:
main:
weight: 30
pre:

File: content/en/community/_index.md

title: "Community"
params:
contributingUrl: https://github.com/apache/horaedb/blob/main/CONTRIBUTING.md
menu:
main:
weight: 40

{{% blocks/section color="white" %}}

How to Subscribe Mail List

Mailing lists are a form of communication used by the Apache community. Generally speaking, many things in the Apache community are hosted by mailing lists, For example: project q&A, technical discussion, transaction decision, release vote, etc，By subscribing, you can get the first update of the HoraeDB community and keep up with the community. The following list is the main list we are using:

In order to subscribe to the [email protected] mailing list, the steps are as follows：

  1. Send an email without any content or subject: [email protected]
  2. Wait until you receive an email with the subject line confirm subscribe to [email protected] (if you have not received it for a long time, please confirm whether the email is blocked by your email, if you have not been blocked and will receive a reply for a long time, return to step 1)
  3. Reply directly to the email without changing the subject line or adding the email content.
  4. Wait until you receive an email with the subject line WELCOME to [email protected] .
  5. If you receive an email from dev, you have successfully subscribed to the email. To initiate a discussion, you can send an email directly to [email protected], which will be sent to everyone who subscribed to the mailing list.

Unsubscribe from the mail list

The steps for unsubscribing to a mailing list are similar to those for subscribing to a mailing list:

  1. Send an email without any content or subject to: [email protected]
  2. Wait until you receive an email with the subject line confirm unsubscribe from [email protected]
  3. Reply directly to the email without changing the subject line or adding the email content
  4. Wait until you receive an email with the subject line GOODBYE from [email protected]
  5. Unsubscribe success

{{% /blocks/section %}}

File: content/en/docs/design/architecture.md

title: "Introduction to HoraeDB's Architecture"

Target

  • Provide the overview of HoraeDB to the developers who want to know more about HoraeDB but have no idea where to start.
  • Make a brief introduction to the important modules of HoraeDB and the connections between these modules but details about their implementations are not be involved.

Motivation

HoraeDB is a timeseries database (TSDB). However, HoraeDB's goal is to handle both timeseries and analytic workloads compared with the classic TSDB, which usually have a poor performance in handling analytic workloads.

In the classic timeseries database, the Tag columns (InfluxDB calls them Tag and Prometheus calls them Label) are normally indexed by generating an inverted index. However, it is found that the cardinality of Tag varies in different scenarios. And in some scenarios the cardinality of Tag is very high (we name this case after analytic workload), and it takes a very high cost to store and retrieve the inverted index. On the other hand, it is observed that scanning+pruning often used by the analytical databases can do a good job to handle such analytic workload.

The basic design idea of HoraeDB is to adopt a hybrid storage format and the corresponding query method for a better performance in processing both timeseries and analytic workloads.

Architecture

plaintext
┌──────────────────────────────────────────┐
│       RPC Layer (HTTP/gRPC/MySQL)        │
└──────────────────────────────────────────┘
┌──────────────────────────────────────────┐
│                 SQL Layer                │
│ ┌─────────────────┐  ┌─────────────────┐ │
│ │     Parser      │  │     Planner     │ │
│ └─────────────────┘  └─────────────────┘ │
└──────────────────────────────────────────┘
┌───────────────────┐  ┌───────────────────┐
│    Interpreter    │  │      Catalog      │
└───────────────────┘  └───────────────────┘
┌──────────────────────────────────────────┐
│               Query Engine               │
│ ┌─────────────────┐  ┌─────────────────┐ │
│ │    Optimizer    │  │    Executor     │ │
│ └─────────────────┘  └─────────────────┘ │
└──────────────────────────────────────────┘
┌──────────────────────────────────────────┐
│         Pluggable Table Engine           │
│  ┌────────────────────────────────────┐  │
│  │              Analytic              │  │
│  │┌────────────────┐┌────────────────┐│  │
│  ││      Wal       ││    Memtable    ││  │
│  │└────────────────┘└────────────────┘│  │
│  │┌────────────────┐┌────────────────┐│  │
│  ││     Flush      ││   Compaction   ││  │
│  │└────────────────┘└────────────────┘│  │
│  │┌────────────────┐┌────────────────┐│  │
│  ││    Manifest    ││  Object Store  ││  │
│  │└────────────────┘└────────────────┘│  │
│  └────────────────────────────────────┘  │
│  ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─   │
│           Another Table Engine        │  │
│  └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─   │
└──────────────────────────────────────────┘

The figure above shows the architecture of HoraeDB stand-alone service and the details of some important modules will be described in the following part.

RPC Layer

module path: https://github.com/apache/incubator-horaedb/tree/main/server

The current RPC supports multiple protocols including HTTP, gRPC, MySQL.

Basically, HTTP and MySQL are used to debug HoraeDB, query manually and perform DDL operations (such as creating, deleting tables, etc.). And gRPC protocol can be regarded as a customized protocol for high-performance, which is suitable for massive reading and writing operations.

SQL Layer

module path: https://github.com/apache/incubator-horaedb/tree/main/query_frontend

SQL layer takes responsibilities for parsing sql and generating the query plan.

Based on sqlparser a sql dialect, which introduces some key concepts including Tag and Timestamp, is provided for processing timeseries data. And by utilizing DataFusion the planner is able to generate both regular logical plans and tailored ones which is used to implement the special operators defined by timeseries queries, e.g PromQL.

Interpreter

module path: https://github.com/apache/incubator-horaedb/tree/main/interpreters

The Interpreter module encapsulates the SQL CRUD operations. In the query procedure, a sql received by HoraeDB is parsed, converted into the query plan and then executed in some specific interpreter, such as SelectInterpreter, InsertInterpreter and etc.

Catalog

module path: https://github.com/apache/incubator-horaedb/tree/main/catalog_impls

Catalog is actually the module managing metadata and the levels of metadata adopted by HoraeDB is similar to PostgreSQL: Catalog > Schema > Table, but they are only used as namespace.

At present, Catalog and Schema have two different kinds of implementation for standalone and distributed mode because some strategies to generate ids and ways to persist metadata differ in different mode.

Query Engine

module path: https://github.com/apache/incubator-horaedb/tree/main/query_engine

Query Engine is responsible for optimizing and executing query plan given a basic SQL plan provided by SQL layer and now such work is mainly delegated to DataFusion.

In addition to the basic functions of SQL, HoraeDB also defines some customized query protocols and optimization rules for some specific query plans by utilizing the extensibility provided by DataFusion. For example, the implementation of PromQL is implemented in this way and read it if you are interested.

Pluggable Table Engine

module path: https://github.com/apache/incubator-horaedb/tree/main/table_engine

Table Engine is actually a storage engine for managing tables in HoraeDB and the pluggability of Table Engine is a core design of HoraeDB which matters in achieving our long-term target, e.g supporting handle log or tracing workload by implementing new storage engines. HoraeDB will have multiple kinds of Table Engine for different workloads and the most appropriate one should be chosen as the storage engine according to the workload pattern.

Now the requirements for a Table Engine are:

  • Manage all the shared resources under the engine:
    • Memory
    • Storage
    • CPU
  • Manage metadata of tables such as table schema and table options;
  • Provide Table instances which provides read and write methods;
  • Take responsibilities for creating, opening, dropping and closing Table instance;
  • ....

Actually the things that a Table Engine needs to process are a little complicated. And now in HoraeDB only one Table Engine called Analytic is provided and does a good job in processing analytical workload, but it is not ready yet to handle the timeseries workload (we plan to enhance it for a better performance by adding some indexes which help handle timeseries workload).

The following part gives a description about details of Analytic Table Engine.

WAL

module path: https://github.com/apache/incubator-horaedb/tree/main/wal

The model of HoraeDB processing data is WAL + MemTable that the recent written data is written to WAL first and then to MemTable and after a certain amount of data in MemTable is accumulated, the data will be organized in a query-friendly form to persistent devices.

Now three implementations of WAL are provided for standalone and distributed mode:

  • For standalone mode, WAL is based on RocksDB and data is persisted on the local disk.
  • For distributed mode, WAL is required as a distributed component and to be responsible for durability of the newly written data, so now we provide an implementation based on OceanBase.
  • For distributed mode, in addition to OceanBase, we also provide a more lightweight implementation based on Apache Kafka.

MemTable

module path: https://github.com/apache/incubator-horaedb/tree/main/analytic_engine/src/memtable

For WAL can't provide efficient data retrieval, the newly written data is also stored in Memtable for efficient data retrieval, after a certain amount of data is reached, HoraeDB organizes the data in MemTable into a query-friendly storage format (SST) and stores it to the persistent device.

The current implementation of MemTable is based on agatedb's skiplist. It allows concurrent reads and writes and can control memory usage based on Arena.

Flush

module path: https://github.com/apache/incubator-horaedb/blob/main/analytic_engine/src/instance/flush_compaction.rs

What Flush does is that when the memory usage of MemTable reaches the threshold, some MemTables are selected for flushing into query-friendly SSTs saved on persistent device.

During the flushing procedure, the data will be divided by a certain time range (which is configured by table option Segment Duration), and any SST is ensured that the timestamps of the data in it are in the same Segment. Actually this is also a common operation in most timeseries databases which organizes data in the time dimension to speed up subsequent time-related operations, such as querying data over a time range and assisting purge data outside the TTL.

Compaction

module path: https://github.com/apache/incubator-horaedb/tree/main/analytic_engine/src/compaction

The data of MemTable is flushed as SSTs, but the file size of recently flushed SST may be very small. And too small or too many SSTs lead to the poor query performance. Therefore, Compaction is then introduced to rearrange the SSTs so that the multiple smaller SST files can be compacted into a larger SST file.

Manifest

module path: https://github.com/apache/incubator-horaedb/tree/main/analytic_engine/src/meta

Manifest takes responsibilities for managing tables' metadata of Analytic Engine including:

  • Table schema and table options;
  • The sequence number where the newest flush finishes;
  • The information of all the SSTs belonging to the table.

Now the Manifest is based on WAL and Object Storage. The newly written updates on the Manifest are persisted as logs in WAL, and in order to avoid infinite expansion of Manifest (actually every Flush leads to an update), Snapshot is also introduced to clean up the history of metadata updates, and the generated Snapshot will be saved to Object Storage.

Object Storage

module path: https://github.com/apache/incubator-horaedb/tree/main/components/object_store

The SST generated by Flush needs to be persisted and the abstraction of the persistent storage device is ObjectStore including multiple implementations:

The distributed architecture of HoraeDB separates storage and computing, which requires Object Store needs to be a highly available and reliable service independent of HoraeDB. Therefore, storage systems like Amazon S3, Alibaba Cloud OSS is a good choice and in the future implementations on storage systems of some other cloud service providers is planned to provide.

SST

module path: https://github.com/apache/incubator-horaedb/tree/main/analytic_engine/src/sst

SST is actually an abstraction that can have multiple specific implementations. The current implementation is based on Parquet, which is a column-oriented data file format designed for efficient data storage and retrieval.

The format of SST is very critical for retrieving data and is also the most important part to perform well in handling both timeseries and analytic workloads. At present, our Parquet-based implementation is good at processing analytic workload but is poor at processing timeseries workload. In our roadmap, we will explore more storage formats in order to achieve a good performance in both workloads.

Space

module path: https://github.com/apache/incubator-horaedb/blob/main/analytic_engine/src/space.rs

In Analytic Engine, there is a concept called space and here is an explanation for it to resolve some ambiguities when read source code. Actually Analytic Engine does not have the concept of catalog and schema and only provides two levels of relationship: space and table. And in the implementation, the schema id (which should be unique across all catalogs) on the upper layer is actually mapped to space id.

The space in Analytic Engine serves mainly for isolation of resources for different tenants, such as the usage of memory.

Critical Path

After a brief introduction to some important modules of HoraeDB, we will give a description for some critical paths in code, hoping to provide interested developers with a guide for reading the code.

Query

plaintext
┌───────┐      ┌───────┐      ┌───────┐
│       │──1──▶│       │──2──▶│       │
│Server │      │  SQL  │      │Catalog│
│       │◀─10──│       │◀─3───│       │
└───────┘      └───────┘      └───────┘
                │    ▲
               4│   9│
                │    │
                ▼    │
┌─────────────────────────────────────┐
│                                     │
│             Interpreter             │
│                                     │
└─────────────────────────────────────┘
                           │    ▲
                          5│   8│
                           │    │
                           ▼    │
                   ┌──────────────────┐
                   │                  │
                   │   Query Engine   │
                   │                  │
                   └──────────────────┘
                           │    ▲
                          6│   7│
                           │    │
                           ▼    │
 ┌─────────────────────────────────────┐
 │                                     │
 │            Table Engine             │
 │                                     │
 └─────────────────────────────────────┘

Take SELECT SQL as an example. The figure above shows the query procedure and the numbers in it indicates the order of calling between the modules.

Here are the details:

  • Server module chooses a proper rpc module (it may be HTTP, gRPC or mysql) to process the requests according the protocol used by the requests;
  • Parse SQL in the request by the parser;
  • With the parsed sql and the information provided by catalog/schema module, DataFusion can generate the logical plan;
  • With the logical plan, the corresponding Interpreter is created and logical plan will be executed by it;
  • For the logical plan of normal Select SQL, it will be executed through SelectInterpreter;
  • In the SelectInterpreter the specific query logic is executed by the Query Engine:
    • Optimize the logical plan;
    • Generate the physical plan;
    • Optimize the physical plan;
    • Execute the physical plan;
  • The execution of physical plan involves Analytic Engine:
    • Data is obtained by read method of Table instance provided by Analytic Engine;
    • The source of the table data is SST and Memtable, and the data can be filtered by the pushed down predicates;
    • After retrieving the table data, Query Engine will complete the specific computation and generate the final results;
  • SelectInterpreter gets the results and feeds them to the protocol module;
  • After the protocol layer converts the results, the server module responds to the client with them.

The following is the flow of function calls in version v1.2.2:

text
/* Detailed source-code truncated for AI context efficiency. */
  1. The received request will be forwarded to handle_sql after various protocol conversions, and since the request may not be processed by this node, it may need to be forwarded to maybe_forward_sql_query to handle the forwarding logic.
  2. After constructing the ForwardRequest in maybe_forward_sql_query, call forward
  3. After constructing the RouteRequest in forward, call route
  4. Use route to get the destination node endpoint and return to forward.
  5. Call forward_with_endpoint to forward the request
  6. return forward
  7. return maybe_forward_sql_query
  8. return handle_sql
  9. If this is a Local request, call fetch_sql_query_output to process it
  10. Call parse_sql to parse sql into Statment
  11. return fetch_sql_query_output
  12. Call statement_to_plan with Statment
  13. Construct Planner with ctx and Statment, and call the statement_to_plan method of Planner
  14. The planner will call the corresponding planner method for the requested category, at this point our sql is a query and will call sql_statement_to_plan
  15. Call sql_statement_to_datafusion_plan , which will generate the datafusion object, and then call SqlToRel::sql_statement_to_plan
  16. The generated logical plan is returned from SqlToRel::sql_statement_to_plan
  17. return
  18. return
  19. return
  20. Call execute_plan_involving_partition_table (in the default configuration) for subsequent optimization and execution of this logical plan
  21. Call build_interpreter to generate Interpreter
  22. return
  23. Call Interpreter's interpreter_execute_plan method for logical plan execution.
  24. The corresponding execute function is called, at this time the sql is a query, so the execute of the SelectInterpreter will be called
  25. call execute_logical_plan , which will call build_df_session_ctx to generate the optimizer
  26. build_df_session_ctx will use the config information to generate the corresponding context, first using datafusion and some custom optimization rules (in logical_optimize_rules()) to generate the logical plan optimizer, using apply_adapters_for_physical_optimize_rules to generate the physical plan optimizer
  27. return optimizer
  28. Call optimize_plan, using the optimizer just generated to first optimize the logical plan and then the physical plan
  29. Return to optimized physical plan
  30. execute physical plan
  31. returned after execution
  32. After collecting the results of all slices, return
  33. return
  34. return
  35. return
  36. Return to the upper layer for network protocol conversion and finally return to the request sender

Write

plaintext
┌───────┐      ┌───────┐      ┌───────┐
│       │──1──▶│       │──2──▶│       │
│Server │      │  SQL  │      │Catalog│
│       │◀─8───│       │◀─3───│       │
└───────┘      └───────┘      └───────┘
                │    ▲
               4│   7│
                │    │
                ▼    │
┌─────────────────────────────────────┐
│                                     │
│             Interpreter             │
│                                     │
└─────────────────────────────────────┘
      │    ▲
      │    │
      │    │
      │    │
      │    │       ┌──────────────────┐
      │    │       │                  │
     5│   6│       │   Query Engine   │
      │    │       │                  │
      │    │       └──────────────────┘
      │    │
      │    │
      │    │
      ▼    │
 ┌─────────────────────────────────────┐
 │                                     │
 │            Table Engine             │
 │                                     │
 └─────────────────────────────────────┘

Take INSERT SQL as an example. The figure above shows the query procedure and the numbers in it indicates the order of calling between the modules.

Here are the details:

  • Server module chooses a proper rpc module (it may be HTTP, gRPC or mysql) to process the requests according the protocol used by the requests;
  • Parse SQL in the request by the parser;
  • With the parsed sql and the catalog/schema module, DataFusion can generate the logical plan;
  • With the logical plan, the corresponding Interpreter is created and logical plan will be executed by it;
  • For the logical plan of normal INSERT SQL, it will be executed through InsertInterpreter;
  • In the InsertInterpreter, write method of Table provided Analytic Engine is called:
    • Write the data into WAL first;
    • Write the data into MemTable then;
  • Before writing to MemTable, the memory usage will be checked. If the memory usage is too high, the flush process will be triggered:
    • Persist some old MemTables as SSTs;
    • Store updates about the new SSTs and the flushed sequence number of WAL to Manifest;
    • Delete the corresponding WAL entries;
  • Server module responds to the client with the execution result.

File: content/en/docs/design/clustering.md

title: "Introduction to Architecture of HoraeDB Cluster"
---Note: Some of the features mentioned in the article have not yet been implemented.

Overview

text
/* Detailed source-code truncated for AI context efficiency. */

The diagram above describes the architecture of a HoraeDB cluster, where some key concepts need to be explained:

  • HoraeMeta Cluster: Takes responsibilities for managing the metadata and resource scheduling of the cluster;
  • Shard(L)/Shard(F): Leader shard and follower shard consisting of multiple tables;
  • HoraeDB: One HoraeDB instance consisting of multiple shards;
  • WAL Service: Write-ahead log service for storing new-written real-time data;
  • Object Storage: Object storage service for storing SST converted from memtable;

From the architecture diagram above, it can be concluded that the compute and storage are separated in the HoraeDB cluster, which makes it easy to implement useful distributed features, such as elastic autoscaling of compute/storage resources, high availability, load balancing, and so on.

Let's dive into some of the key components mentioned above before explaining how these features are implemented.

Shard

Shard is the basic scheduling unit in the cluster, which consists of a group of tables. And the tables in a shard share the same region for better storage locality in the WAL Service, and because of this, it is efficient to recover the data of all tables in the shard by scanning the entire WAL region. For most of implementations of WAL Service, without the shard concept, it costs a lot to recover the table data one by one due to massive random IO, and this case will deteriorate sharply when the number of tables grows to a certain level.

A specific role, Leader or Follower, should be assigned to a shard. A pair of leader-follower shards share the same set of tables, and the leader shard can serve the write and query requests from the client while the follower shard can only serve the read-only requests, and must synchronize the newly written data from the WAL service in order to provide the latest snapshot for data retrieval. Actually, the follower is not needed if the high availability is not required, while with at least one follower, it takes only a short time to resume service by simply switching the Follower to Leader when the HoraeDB instance on which the leader shard exists crashes.

The diagram below concludes the relationship between HoraeDB instance, Shard, Table. As shown in the diagram, the leader and follower shards are interleaved on the HoraeDB instance.

plaintext
┌─HoraeDB Instance0──────┐     ┌─HoraeDB Instance1──────┐
│  ┌─Shard0(L)────────┐  │     │  ┌─Shard0(F)────────┐  │
│  │ ┌────┬────┬────┐ │  │     │  │ ┌────┬────┬────┐ │  │
│  │ │ T0 │ T1 │ T2 │ │  │     │  │ │ T0 │ T1 │ T2 │ │  │
│  │ └────┴────┴────┘ │  │     │  │ └────┴────┴────┘ │  │
│  └──────────────────┘  │     │  └──────────────────┘  │
│                        │     │                        │
│  ┌─Shard1(F)────────┐  │     │  ┌─Shard1(L)────────┐  │
│  │ ┌────┬────┬────┐ │  │     │  │ ┌────┬────┬────┐ │  │
│  │ │ T0 │ T1 │ T2 │ │  │     │  │ │ T0 │ T1 │ T2 │ │  │
│  │ └────┴────┴────┘ │  │     │  │ └────┴────┴────┘ │  │
│  └──────────────────┘  │     │  └──────────────────┘  │
└────────────────────────┘     └────────────────────────┘

Since Shard is the basic scheduling unit, it is natural to introduce some basic shard operations:

  • Create/Drop table to/from a shard;
  • Open/Close a shard;
  • Split one shard into two shards;
  • Merge two shards into one shard;
  • Switch the role of a shard;

With these basic shard operations, some complex scheduling logic can be implemented, e.g. perform an expansion by splitting one shard into two shards and migrating one of them to the new HoraeDB instance.

HoraeMeta

HoraeMeta is implemented by embedding an ETCD inside to ensure consistency and takes responsibilities for cluster metadata management and scheduling.

The cluster metadata includes:

  • Table information, such as table name, table ID, and which cluster the table belongs to;
  • The mapping between table and shard and between shard and HoraeDB instance;
  • ...

As for the cluster scheduling work, it mainly includes:

  • Receiving the heartbeats from the HoraeDB instances and determining the online status of these registered instances;
  • Assigning specific role shards to the registered HoraeDB instances;
  • Participating in table creation by assigning a unique table ID and the most appropriate shard to the table;
  • Performing load balancing through shard operations according to the load information sent with the heartbeats;
  • Performing expansion through shard operations when new instances are registered;
  • Initiating failover through shard operations when old instances go offline;

Route

In order to avoid the overhead of forwarding requests, the communication between clients and the HoraeDB instances is peer-to-peer, that is to say, the client should retrieve routing information from the server before sending any specific write/query requests.

Actually, the routing information is decided by the HoraeMeta, but clients are only allowed the access to it through the HoraeDB instances rather than HoraeMeta, to avoid potential performance issues on the HoraeMeta.

WAL Service & Object Storage

In the HoraeDB cluster, WAL Service and Object Storage exist as separate distributed systems featured with HA, data replication and scalability. Current distributed implementations for WAL Service includes Kafka and OBKV (access OceanBase by its table api), and the implementations for Object Storage include popular object storage services, such as AWS S3, Azure object storage and Aliyun OSS.

The two components are similar in that they are introduced to serve as the underlying storage layer for separating compute and storage, while the difference between two components is obvious that WAL Service is used to store the newly written data from the real-time write requests whose individual size is small but quantity is large, and Object Storage is used to store the read-friendly data files (SST) organized in the background, whose individual size is large and aggregate size is much larger.

The two components make it much easier to implement the horaedb cluster, which features horizontal scalability, high availability and load balancing.

Scalability

Scalability is an important feature for a distributed system. Let's take a look at to how the horizontal scalability of the HoraeDB cluster is achieved.

First, the two storage components (WAL Service and Object Storage) should be horizontally scalable when deciding on the actual implementations for them, so the two storage services can be expanded separately if the storage capacity is not sufficient.

It will be a little bit complex when discussing the scalability of the compute service. Basically, these cases will bring the capacity problem:

  • Massive queries on massive tables;
  • Massive queries on a single large table;
  • Massive queries on a normal table;

For the first case, it is easy to achieve horizontal scalability just by assigning shards that are created or split from old shards to expanded HoraeDB instances.

For the second case, the table partitioning is proposed and after partitioning, massive queries are distributed across multiple HoraeDB instances.

And the last case is the most important and the most difficult. Actually, the follower shard can handle part of the queries, but the number of follower shards is limited by the throughput threshold of the synchronization from the WAL regions. As shown in the diagram below, a pure compute shard can be introduced if the followers are not enough to handle the massive queries. Such a shard is not required to synchronize data with the leader shard, and retrieves the newly written data from the leader/follower shard only when the query comes. As for the SSTs required by the query, they will be downloaded from Object Storage and cached afterwards. With the two parts of the data, the compute resources are fully utilized to execute the CPU-intensive query plan. As we can see, such a shard can be added with only a little overhead (retrieving some data from the leader/follower shard when it needs), so to some extent, the horizontal scalability is achieved.

plaintext
┌HoraeDB─────┬┬─┐
                            ┌──newly written─│  │  │TableN││ │
                            ▼                └──Shard(L/F)┴┴─┘
┌───────┐  Query  ┌HoraeDB─────┬┬─┐
│client │────────▶│  │  │TableN││ │
└───────┘         └──Shard─────┴┴─┘          ┌───────────────┐
                            ▲                │    Object     │
                            └───old SST──────│    Storage    │
                                             └───────────────┘

High Availability

Assuming that WAL service and Object Storage are highly available, the high availability of the HoraeDB cluster can be achieved by such a procedure:

  • When detecting that the heartbeat is broken, HoraeMeta determines that the HoraeDB instance is offline;
  • The follower shards whose paired leader shards exist on the offline instance are switched to leader shards for fast failover;
  • A slow failover can be achieved by opening the crashed shards on another instance if such follower shards don't exist.
text
/* Detailed source-code truncated for AI context efficiency. */

Load Balancing

HoraeMeta collects the instance load information contained in the received heartbeats to create a load overview of the whole cluster, according to which the load balancing can be implemented as an automatic mechanism:

  • Pick a shard on a low-load instance for the newly created table;
  • Migrate a shard from a high-load instance load to another low-load instance;
  • Split the large shard on the high-load instance and migrate the split shards to other low-load instances;

File: content/en/docs/design/compaction_offload.md

title: "Compaction Offload"

Note: This feature is still in development.

This chapter discusses compaction offload, which is designed to separate the compaction workload from the local horaedb nodes and delegate it to external compaction nodes.

Overview

plaintext
┌─────────────────────────────────────────────────────────────────────────┐
│                                                                         │
│                           HoraeMeta Cluster                             │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘
    ▲                ▲                           |                   |
    │                │1.Fetch compaction         │(Monitor compaction│
    │                │  node  info               │node)              │
    |                │                           ▼                   ▼
┌────────────┐  ┌────────────┐              ┌────────────┐   ┌────────────┐
│            │  │            │2.Offload Task│            │   │            │
│  HoraeDB   │  │  HoraeDB   │  ─────────▶  │ Compaction │   │ Compaction │
│            │  │            │  ◀─────────  │ Node       │   │ Node       │
└────────────┘  └────────────┘ 4.Ret Result └────────────┘   └────────────┘
    |                |                            |                 |
    │  5.Update the  │                            │    3.Compact    │
    │    SSTable     │                            │                 │
    ▼                ▼                            ▼                 ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                                                ┌─────────────────────┐  │
│                            Object Storage      │ Temporary Workspace │  │
│                                                └─────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────┘

The diagram above describes the architecture of cluster for compaction offload, where some key concepts need to be explained:

  • Compaction Node: Takes responsibility to handle offloaded compaction tasks. The compaction node receives the compaction task and performs the actual merging of SSTables, then sends back the task result to HoraeDB.
  • HoraeMeta Cluster: HoraeMeta acts as a compaction nodes manager in the compaction offload scenario. It monitors the compaction nodes cluster and schedule the compaction nodes.

The procedure of remote compaction based above architecture diagram is:

  1. HoraeDB servers fetch the information of suitable compaction nodes from the HoraeMeta.
  2. HoraeDB submit the compaction task to the remote compaction node, according to the information fetched from HoraeMeta.
  3. Compaction node executes the task and write results to the temporary workspace.
  4. Compaction node sends compaction results back to HoraeDB.
  5. HoraeDB receives the result, installs the data in temporary workspace and purges compaction input files.

The architecture above makes it easy to implement some wonderful features like load balancing and high availability. Let's dive into the key components in the architecture and talking about how these features are implemented.

Compaction Node

Compaction Node runs the main logic of compaction. It is implemented based on HoraeDB and distinguished by:

  • NodeType: A config parameter used to distinguished the HoraeDB and CompactionNode. This info would be sent to HoraeMeta through heartbeat.

The compaction service is implemented as grpc service.

HoraeMeta

HoraeMeta manages the compaction nodes cluster with CompactionNodeManager, which takes responsibilities for compaction nodes metadata management and scheduling.

The compaction nodes metadata includes:

  • Compaction node information, such as node name, node state;
  • A compaction node name list, used as the key to access compaction node info, for better scheduling with round-robin strategy;
  • ...

As for the compaction nodes scheduling work, it mainly includes:

  • Receiving the heartbeats from the compaction node and determining the online status of these registered nodes.
  • Performing load balancing according to the compaction nodes cluster info.
  • Providing the info of suitable compaction node for HoraeDB when remote compaction execution is needed.

Load Balancing

Load Balancing is critical for compaction nodes cluster to make their overall processing more efficient. The effect of load balancing mainly based on the schedule algorithm for compaction nodes impl in CompactionNodeManager.

(ps: The current implementation of schedule algorithm is round-robin strategy for easiness.)

The main process for the schedule algorithm based on real load is:

  • HoraeMeta collects the compaction nodes load information through the heartbeats to create a load overview of the compaction nodes cluster.
  • Pick a compaction node with low load according to the load overview.

High Availability

The fault tolerance of above architecture can be achieved by such a procedure:

  • When detecting that the heartbeat is broken, HoraeMeta determines that the compaction node is offline.
  • When HoraeMeta can not provide suitable compaction node for HoraeDB or compaction node doesn't return the task result successfully, HoraeDB would switches to run compaction task locally.

File: content/en/docs/design/storage.md

title: "Storage"

The storage engine mainly provides the following two functions：

  1. Persistence of data
  2. Under the premise of ensuring the correctness of the data, organize the data in the most reasonable way to meet the query needs of different scenarios.

This document will introduce the internal implementation of the storage engine in HoraeDB. Readers can refer to the content here to explore how to use HoraeDB efficiently.

Overall Structure

HoraeDB is a distributed storage system based on the share-nothing architecture.

Data between different servers is isolated from each other and does not affect each other. The storage engine in each stand-alone machine is a variant of log-structured merge-tree, which is optimized for time-series scenarios. The following figure shows its core components:

Write Ahead Log (WAL)

A write request will be written to

  1. memtable in memory
  2. WAL in durable storage

Since memtable is not persisted to the underlying storage system in real time, so WAL is required to ensure the reliability of the data in memtable.

On the other hand, due to the design of the distributed architecture, WAL itself is required to be highly available. Now there are following implementations in HoraeDB:

Memtable

Memtable is a memory data structure used to hold recently written table data. Different tables have its corresponding memtable.

Memtable is read-write by default (aka active), and when the write reaches some threshold, it will become read-only and be replaced by a new memtable.

The read-only memtable will be flushed to the underlying storage system in SST format by background thread. After flush is completed, the read-only memtable can be destroyed, and the corresponding data in WAL can also be deleted.

Sorted String Table（SST）

SST is a persistent format for data, which is stored in the order of primary keys of table. Currently, HoraeDB uses parquet format for this.

For HoraeDB, SST has an important option: segment_duration, only SST within the same segment can be merged, which is benefical for time-series data. And it is also convenient to eliminate expired data.

In addition to storing the original data, the statistical information of the data will also be stored in the SST to speed up the query, such as the maximum value, the minimum value, etc.

Compactor

Compactor can merge multiple small SST files into one, which is used to solve the problem of too many small files. In addition, Compactor will also delete expired data and duplicate data during the compaction. In future, compaction maybe add more task, such as downsample.

The current compaction strategy in HoraeDB reference Cassandra:

Manifest

Manifest records metadata of table, SST file, such as: the minimum and maximum timestamps of the data in an SST.

Due to the design of the distributed architecture, the manifest itself is required to be highly available. Now in HoraeDB, there are mainly the following implementations：

  • WAL
  • ObjectStore

ObjectStore

ObjectStore is place where data (i.e. SST) is persisted.

Generally speaking major cloud vendors should provide corresponding services, such as Alibaba Cloud's OSS and AWS's S3.

Explore Related Technologies & Topics

View All 9,111+ Repositories →

Discover similar high-velocity repositories, agent skills, and OpenAPI specifications across the ecosystem.

#batch Browse batch catalog → #cdc Browse cdc catalog → #change-data-capture Browse change-data-capture catalog → #data-integration Browse data-integration catalog → #data-pipeline Browse data-pipeline catalog → #distributed Browse distributed catalog → ⚡ MCP Servers Model Context Protocol → 🧠 Agent Skills Prompt & rule files → 🔌 OpenAPI Specs REST & Swagger schemas →