[ISSUE-817] Add geaflow-dsl-connector-doris (Stream Load sink + partitioned source)

docs/docs-cn/source/5.application-development/3.connector/11.doris.md

@@ -0,0 +1,120 @@
+# Doris Connector介绍
+Doris Connector由社区贡献，支持读和写。
+
+Doris 是高性能实时 OLAP 数据库，是图计算结果天然的下游分析存储。虽然 Doris 兼容 MySQL 协议、
+可以通过 JDBC Connector 接入，但 JDBC 只能逐行 INSERT，无法利用 Doris 的高吞吐导入通道
+Stream Load。Doris Connector 的 Sink 通过 Stream Load 攒批写入，写吞吐显著优于逐行 JDBC；
+Source 通过 MySQL 协议按分区并行读取。
+
+## 语法
+
+```sql
+CREATE TABLE doris_table (
+  id BIGINT,
+  name VARCHAR,
+  age INT
+) WITH (
+    type = 'doris',
+    geaflow.dsl.doris.fenodes = '127.0.0.1:8030',
+    geaflow.dsl.doris.jdbc.url = 'jdbc:mysql://127.0.0.1:9030/example_db',
+    geaflow.dsl.doris.username = 'root',
+    geaflow.dsl.doris.password = '',
+    geaflow.dsl.doris.database = 'example_db',
+    geaflow.dsl.doris.table = 'person'
+);
+```
+
+## 参数
+
+| 参数名 | 是否必须 | 描述 |
+| ---- | ---- | ---- |
+| geaflow.dsl.doris.fenodes | 写入必须 | Doris FE 的 HTTP 地址列表(host:httpPort)，逗号分隔，Sink 通过它做 Stream Load；配置多个 FE 时，请求失败会自动切换到下一个 FE。 |
+| geaflow.dsl.doris.jdbc.url | 读取必须 | Doris 查询(MySQL 协议)的 JDBC URL，例如 jdbc:mysql://host:9030/database，Source 通过它做分区并行读。连接器不打包 JDBC driver，使用 Source 时需自行在运行时 classpath 提供 MySQL 协议的 driver(如 mysql-connector-java)。 |
+| geaflow.dsl.doris.username | 否 | Doris 用户名，默认 root。 |
+| geaflow.dsl.doris.password | 否 | Doris 密码，默认空。 |
+| geaflow.dsl.doris.database | 是 | Doris 数据库名。 |
+| geaflow.dsl.doris.table | 是 | Doris 表名。 |
+| geaflow.dsl.doris.sink.format | 否 | Stream Load 的负载格式，csv 或 json，默认 json(能安全处理换行、引号、反斜杠和 Unicode)；csv 为纯分隔符切分、不做转义，仅在字段值不含所选分隔符时使用。 |
+| geaflow.dsl.doris.sink.column.separator | 否 | csv 格式的列分隔符，默认制表符。 |
+| geaflow.dsl.doris.sink.line.delimiter | 否 | csv 格式的行分隔符，默认换行符。 |
+| geaflow.dsl.doris.sink.batch.rows | 否 | 攒批的行数阈值，达到后触发一次 flush，默认 10000。 |
+| geaflow.dsl.doris.sink.batch.bytes | 否 | 攒批的字节阈值，达到后触发一次 flush，默认 10MB。 |
+| geaflow.dsl.doris.sink.max.retries | 否 | 单次 Stream Load 的最大重试次数，默认 3。 |
+| geaflow.dsl.doris.request.connect.timeout.ms | 否 | Stream Load 的连接超时(毫秒)，默认 30000。 |
+| geaflow.dsl.doris.request.read.timeout.ms | 否 | Stream Load 请求的读/写超时(毫秒)，覆盖整批数据的上传，默认 60000。批量很大或网络较慢时可调大，网络异常时调小可更快失败。 |
+| geaflow.dsl.doris.source.partition.mode | 否 | Source 分区策略：range(按数值列均分区间)或 custom(按用户提供的谓词)，默认 range。 |
+| geaflow.dsl.doris.source.partition.clauses | 否 | custom 模式下用分号分隔的 WHERE 谓词，每个谓词一个分区，如 "dt='2024-01-01';dt='2024-01-02'"；谓词应互斥且共同覆盖全部数据；为空表示单分区。适用于非数值/倾斜列或自定义条件。 |
+| geaflow.dsl.doris.source.partition.num | 否 | range 模式下并行读的分区数，默认 1。 |
+| geaflow.dsl.doris.source.partition.column | 否 | range 模式下用于切分分区的数值列，默认 id。建议选择有索引、分布均匀的列(如主键)；在无索引或高度倾斜的列上切分可能导致全表扫描或热点。 |
+| geaflow.dsl.doris.source.partition.lowerbound | 否 | 分区列下界，只用于决定分区步长，不用于过滤数据。 |
+| geaflow.dsl.doris.source.partition.upperbound | 否 | 分区列上界，只用于决定分区步长，不用于过滤数据。 |
+
+## 写入原理
+
+Sink 在 `write()` 中把行攒到内存缓冲区，当缓冲区行数达到
+`geaflow.dsl.doris.sink.batch.rows` 或字节数达到 `geaflow.dsl.doris.sink.batch.bytes` 时，
+或在每个窗口结束(`finish()`)时，把整批数据通过一次 HTTP PUT 提交给 Doris FE 的
+Stream Load 接口(`/api/{db}/{table}/_stream_load`)。FE 会返回 307 重定向到具体 BE，
+客户端会自动跟随重定向并把负载重新发送给 BE 完成导入。默认使用 JSON 负载(能安全处理特殊字符)，
+也支持 CSV 负载。
+
+## 写入语义与幂等
+
+每一批数据在提交时会生成一个唯一的 label(基于 UUID)，并在该批次的所有重试与 FE 故障切换中
+复用同一个 label。Doris 保证同一个 label 至多导入一次：如果某次尝试实际已提交、但客户端因网络
+异常误判为失败，重试会命中 "Label Already Exists"，连接器将其视为成功，从而避免重复写入。
+因此在 FE failover + 重试场景下，写入是幂等的。重试次数可通过
+`geaflow.dsl.doris.sink.max.retries` 配置(为支持 failover，实际重试次数至少等于 FE 个数)，
+失败与重试日志中都会打印 label 以便排查。
+
+## 示例
+
+```sql
+set geaflow.dsl.doris.fenodes = '127.0.0.1:8030';
+set geaflow.dsl.doris.jdbc.url = 'jdbc:mysql://127.0.0.1:9030/example_db';
+set geaflow.dsl.doris.username = 'root';
+set geaflow.dsl.doris.password = '';
+set geaflow.dsl.doris.database = 'example_db';
+
+CREATE TABLE doris_source_table (
+  id BIGINT,
+  name VARCHAR,
+  age INT
+) WITH (
+    type = 'doris',
+    geaflow.dsl.doris.table = 'source_table',
+    geaflow.dsl.doris.source.partition.num = '4',
+    geaflow.dsl.doris.source.partition.column = 'id',
+    geaflow.dsl.doris.source.partition.lowerbound = '0',
+    geaflow.dsl.doris.source.partition.upperbound = '10000'
+);
+
+CREATE TABLE doris_sink_table (
+  id BIGINT,
+  name VARCHAR,
+  age INT
+) WITH (
+    type = 'doris',
+    geaflow.dsl.doris.table = 'sink_table',
+    geaflow.dsl.doris.sink.batch.rows = '10000'
+);
+
+INSERT INTO doris_sink_table
+SELECT * FROM doris_source_table;
+```
+
+## 性能
+
+在集成测试(`DorisConnectorIntegrationTest#testStreamLoadThroughputBenchmark`)中，
+对同一张表写入 20000 行，Stream Load 攒批写入相比逐行 JDBC INSERT 的耗时明显更低
+(通常快一个数量级以上，具体倍数取决于批大小、网络与集群规模)。该基准会断言 Stream Load
+比逐行 JDBC 更快。
+
+在一次基于 apache/doris:3.0.8-all 单机(all-in-one)的实测中：写入 500 行时 Stream Load 约
+91ms(约 5500 行/秒)，逐行 JDBC INSERT 约 11335ms(约 44 行/秒)，Stream Load 快约 125 倍；
+写入 2000 行时 Stream Load 仅约 97ms，说明攒批写入的耗时几乎与行数无关，数据量越大优势越明显。
+以上为特定环境下的参考值。
+
+该集成测试(含基准)依赖真实 Doris，默认关闭、不会在普通 CI 中运行。可通过以下两种方式开启：
+- 加参数 `-Ddoris.it.enabled=true`(需要 Linux 宿主机 + Docker，会用 Testcontainers 启动 Doris)；
+- 或通过 `-Ddoris.it.fenodes` 和 `-Ddoris.it.jdbcUrl` 指向一个外部 Doris 实例。

docs/docs-cn/source/5.application-development/3.connector/index.rst

@@ -16,4 +16,5 @@
    8.hudi.md
    9.pulsar.md
    10.udc.md
+   11.doris.md
 

docs/docs-en/source/5.application-development/3.connector/11.doris.md

@@ -0,0 +1,127 @@
+# Doris Connector Introduction
+The Doris Connector is contributed by the community and supports both reading and writing.
+
+Doris is a high-performance real-time OLAP database and a natural downstream analytics store for
+graph-computing results. Although Doris speaks the MySQL protocol and can be accessed through the
+JDBC Connector, JDBC only does row-by-row INSERT and cannot use Doris's high-throughput load path,
+Stream Load. The Doris Connector sink writes buffered batches via Stream Load, giving a much higher
+write throughput than row-by-row JDBC, while the source reads in parallel by partition through the
+MySQL protocol.
+
+## Syntax
+
+```sql
+CREATE TABLE doris_table (
+  id BIGINT,
+  name VARCHAR,
+  age INT
+) WITH (
+    type = 'doris',
+    geaflow.dsl.doris.fenodes = '127.0.0.1:8030',
+    geaflow.dsl.doris.jdbc.url = 'jdbc:mysql://127.0.0.1:9030/example_db',
+    geaflow.dsl.doris.username = 'root',
+    geaflow.dsl.doris.password = '',
+    geaflow.dsl.doris.database = 'example_db',
+    geaflow.dsl.doris.table = 'person'
+);
+```
+
+## Options
+
+| Key | Required | Description |
+| ---- | ---- | ---- |
+| geaflow.dsl.doris.fenodes | required for sink | The Doris FE http address list (host:httpPort), comma separated. Used by the sink for Stream Load; when multiple FEs are given the sink fails over to the next FE on a failed request. |
+| geaflow.dsl.doris.jdbc.url | required for source | The Doris query (MySQL protocol) jdbc url, e.g. jdbc:mysql://host:9030/database. Used by the source for partitioned reads. The connector does not bundle a JDBC driver, so a MySQL-protocol driver (e.g. mysql-connector-java) must be provided on the runtime classpath to use the source. |
+| geaflow.dsl.doris.username | false | The Doris username, default root. |
+| geaflow.dsl.doris.password | false | The Doris password, default empty. |
+| geaflow.dsl.doris.database | true | The Doris database name. |
+| geaflow.dsl.doris.table | true | The Doris table name. |
+| geaflow.dsl.doris.sink.format | false | The Stream Load payload format, csv or json, default json (safely handles newlines, quotes, backslashes and unicode). Csv is a plain separator split without quoting and should only be used when values cannot contain the chosen separators. |
+| geaflow.dsl.doris.sink.column.separator | false | The column separator for csv Stream Load, default tab. |
+| geaflow.dsl.doris.sink.line.delimiter | false | The line delimiter for csv Stream Load, default newline. |
+| geaflow.dsl.doris.sink.batch.rows | false | Flush the buffer when the buffered row count reaches this threshold, default 10000. |
+| geaflow.dsl.doris.sink.batch.bytes | false | Flush the buffer when the buffered byte size reaches this threshold, default 10MB. |
+| geaflow.dsl.doris.sink.max.retries | false | The max retry times for a Stream Load request, default 3. |
+| geaflow.dsl.doris.request.connect.timeout.ms | false | The connect timeout in milliseconds for Stream Load, default 30000. |
+| geaflow.dsl.doris.request.read.timeout.ms | false | The socket read/write timeout in milliseconds for a Stream Load request, covering the upload of the whole batch, default 60000. Increase it for very large batches on slow networks, lower it to fail faster on network anomalies. |
+| geaflow.dsl.doris.source.partition.mode | false | The partitioning strategy: range (evenly split a numeric column) or custom (user-provided predicates), default range. |
+| geaflow.dsl.doris.source.partition.clauses | false | Semicolon-separated WHERE predicates for custom mode, one partition per predicate, e.g. "dt='2024-01-01';dt='2024-01-02'"; predicates should be disjoint and jointly cover the data; empty means a single partition. Useful for non-numeric/skewed columns or arbitrary conditions. |
+| geaflow.dsl.doris.source.partition.num | false | The source partition number for parallel reads in range mode, default 1. |
+| geaflow.dsl.doris.source.partition.column | false | The numeric column used to split the source into partitions in range mode, default id. Prefer an indexed, evenly-distributed column (e.g. the primary key); splitting on an unindexed or highly-skewed column may cause full scans or hotspots. |
+| geaflow.dsl.doris.source.partition.lowerbound | false | The lowerbound of the partition column, only used to decide the partition stride, not for filtering rows. |
+| geaflow.dsl.doris.source.partition.upperbound | false | The upperbound of the partition column, only used to decide the partition stride, not for filtering rows. |
+
+## How the sink works
+
+The sink buffers rows in `write()`. When the buffered row count reaches
+`geaflow.dsl.doris.sink.batch.rows` or the buffered byte size reaches
+`geaflow.dsl.doris.sink.batch.bytes`, or when a window finishes (`finish()`), the whole batch is
+submitted to the Doris FE Stream Load endpoint (`/api/{db}/{table}/_stream_load`) with a single
+HTTP PUT. The FE replies with a 307 redirect to a BE, and the client follows the redirect and
+re-sends the payload to the BE that executes the load. JSON payload is used by default (it safely
+handles special characters); CSV payload is also supported.
+
+## Write semantics and idempotency
+
+Each batch is submitted with a unique label (UUID based) that is reused across all retries and FE
+failover attempts of that batch. Doris loads a given label at most once: if an attempt actually
+committed but the client mistook it for a failure due to a network error, the retry hits "Label
+Already Exists", which the connector treats as success, so the batch is not written twice. Writes
+are therefore idempotent under FE failover plus retries. The retry count is configurable via
+`geaflow.dsl.doris.sink.max.retries` (to support failover the effective retry count is at least the
+number of FEs), and the label is logged on failures and retries for troubleshooting.
+
+## Example
+
+```sql
+set geaflow.dsl.doris.fenodes = '127.0.0.1:8030';
+set geaflow.dsl.doris.jdbc.url = 'jdbc:mysql://127.0.0.1:9030/example_db';
+set geaflow.dsl.doris.username = 'root';
+set geaflow.dsl.doris.password = '';
+set geaflow.dsl.doris.database = 'example_db';
+
+CREATE TABLE doris_source_table (
+  id BIGINT,
+  name VARCHAR,
+  age INT
+) WITH (
+    type = 'doris',
+    geaflow.dsl.doris.table = 'source_table',
+    geaflow.dsl.doris.source.partition.num = '4',
+    geaflow.dsl.doris.source.partition.column = 'id',
+    geaflow.dsl.doris.source.partition.lowerbound = '0',
+    geaflow.dsl.doris.source.partition.upperbound = '10000'
+);
+
+CREATE TABLE doris_sink_table (
+  id BIGINT,
+  name VARCHAR,
+  age INT
+) WITH (
+    type = 'doris',
+    geaflow.dsl.doris.table = 'sink_table',
+    geaflow.dsl.doris.sink.batch.rows = '10000'
+);
+
+INSERT INTO doris_sink_table
+SELECT * FROM doris_source_table;
+```
+
+## Performance
+
+In the integration test
+(`DorisConnectorIntegrationTest#testStreamLoadThroughputBenchmark`), writing 20000 rows into the
+same table with batched Stream Load takes clearly less time than row-by-row JDBC INSERT (usually
+more than an order of magnitude faster, depending on batch size, network and cluster size). The
+benchmark asserts that Stream Load is faster than row-by-row JDBC.
+
+In a local run against a single-node apache/doris:3.0.8-all (all-in-one): writing 500 rows took
+about 91ms (~5500 rows/s) with Stream Load versus about 11335ms (~44 rows/s) with row-by-row JDBC
+INSERT, so Stream Load was about 125x faster; writing 2000 rows with Stream Load took only about
+97ms, showing that batched load cost is nearly independent of the row count and the advantage grows
+with data size. These are reference numbers for one specific environment.
+
+The integration test (and the benchmark) needs a real Doris, so it is disabled by default and never
+runs in a normal CI. Enable it either with `-Ddoris.it.enabled=true` (requires Docker on a Linux
+host; it starts Doris via Testcontainers), or by pointing it at an external Doris with
+`-Ddoris.it.fenodes` and `-Ddoris.it.jdbcUrl`.

docs/docs-en/source/5.application-development/3.connector/index.rst

@@ -15,4 +15,5 @@ Connector
    7.hbase.md
    8.hudi.md
    9.pulsar.md
-   10.udc.md
+   10.udc.md
+   11.doris.md

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml

@@ -0,0 +1,88 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one
+  ~ or more contributor license agreements.  See the NOTICE file
+  ~ distributed with this work for additional information
+  ~ regarding copyright ownership.  The ASF licenses this file
+  ~ to you under the Apache License, Version 2.0 (the
+  ~ "License"); you may not use this file except in compliance
+  ~ with the License.  You may obtain a copy of the License at
+  ~
+  ~   http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing,
+  ~ software distributed under the License is distributed on an
+  ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  ~ KIND, either express or implied.  See the License for the
+  ~ specific language governing permissions and limitations
+  ~ under the License.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.apache.geaflow</groupId>
+        <artifactId>geaflow-dsl-connector</artifactId>
+        <version>0.8.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>geaflow-dsl-connector-doris</artifactId>
+    <name>geaflow-dsl-connector-doris</name>
+
+    <properties>
+        <doris.httpclient.version>4.5.13</doris.httpclient.version>
+        <testcontainers.version>1.19.7</testcontainers.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.geaflow</groupId>
+            <artifactId>geaflow-dsl-common</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.geaflow</groupId>
+            <artifactId>geaflow-dsl-connector-api</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>${doris.httpclient.version}</version>
+        </dependency>
+
+        <!--
+          Doris speaks the MySQL protocol and the source reads through it, but the source only
+          uses the standard java.sql API and loads the driver by name, so the GPL-licensed MySQL
+          driver is NOT bundled (test scope only, like geaflow-dsl-connector-jdbc). Users must
+          provide a MySQL-protocol driver on the runtime classpath to use the source.
+        -->
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <version>${mysql.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testng</groupId>
+            <artifactId>testng</artifactId>
+            <version>${testng.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>