From f4b2ebe25313c84e1499cda600b46329cd1c1ec0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=88=9A=E5=AE=8F=E6=88=90?= Date: Wed, 22 Jul 2026 18:58:21 +0800 Subject: [PATCH 1/6] [ISSUE-817] Add geaflow-dsl-connector-doris (Stream Load sink + partitioned source) - Sink writes to Doris via Stream Load: buffer rows in write(), flush a whole batch on row/byte threshold or window finish(), giving much higher throughput than row-by-row JDBC INSERT. CSV payload by default, JSON supported. - Source reads through the MySQL protocol with parallel partitioned reads. - SPI registration + parent/aggregation pom wiring + runtime dependency. - CN + EN docs and examples. - Unit tests and a Testcontainers integration test with a throughput benchmark (self-skips when Docker/host-networking is unavailable). --- .../3.connector/11.doris.md | 100 ++++++ .../3.connector/index.rst | 1 + .../3.connector/11.doris.md | 104 ++++++ .../3.connector/index.rst | 3 +- .../geaflow-dsl-connector-doris/pom.xml | 82 +++++ .../dsl/connector/doris/DorisConfigKeys.java | 122 +++++++ .../dsl/connector/doris/DorisConstants.java | 48 +++ .../dsl/connector/doris/DorisStreamLoad.java | 181 +++++++++++ .../connector/doris/DorisTableConnector.java | 46 +++ .../dsl/connector/doris/DorisTableSink.java | 146 +++++++++ .../dsl/connector/doris/DorisTableSource.java | 301 ++++++++++++++++++ .../dsl/connector/doris/DorisUtils.java | 67 ++++ ...e.geaflow.dsl.connector.api.TableConnector | 20 ++ .../connector/doris/DorisConfigKeysTest.java | 54 ++++ .../doris/DorisConnectorIntegrationTest.java | 260 +++++++++++++++ .../connector/doris/DorisStreamLoadTest.java | 49 +++ .../connector/doris/DorisTableSinkTest.java | 65 ++++ .../dsl/connector/doris/DorisUtilsTest.java | 72 +++++ .../geaflow-dsl/geaflow-dsl-connector/pom.xml | 1 + .../geaflow-dsl/geaflow-dsl-runtime/pom.xml | 5 + geaflow/geaflow-dsl/pom.xml | 6 + 21 files changed, 1732 insertions(+), 1 deletion(-) create mode 100644 docs/docs-cn/source/5.application-development/3.connector/11.doris.md create mode 100644 docs/docs-en/source/5.application-development/3.connector/11.doris.md create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeys.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConstants.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoad.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableConnector.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSink.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSource.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisUtils.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/resources/META-INF/services/org.apache.geaflow.dsl.connector.api.TableConnector create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeysTest.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoadTest.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSinkTest.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisUtilsTest.java diff --git a/docs/docs-cn/source/5.application-development/3.connector/11.doris.md b/docs/docs-cn/source/5.application-development/3.connector/11.doris.md new file mode 100644 index 000000000..deb66b507 --- /dev/null +++ b/docs/docs-cn/source/5.application-development/3.connector/11.doris.md @@ -0,0 +1,100 @@ +# 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。 | +| geaflow.dsl.doris.jdbc.url | 读取必须 | Doris 查询(MySQL 协议)的 JDBC URL，例如 jdbc:mysql://host:9030/database，Source 通过它做分区并行读。 | +| 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，默认 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 的读超时(毫秒)，默认 30000。 | +| geaflow.dsl.doris.source.partition.num | 否 | Source 并行读的分区数，默认 1。 | +| geaflow.dsl.doris.source.partition.column | 否 | 用于切分分区的数值列，默认 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 完成导入。默认使用 CSV 负载以获得更高吞吐， +也支持 JSON 负载。 + +## 示例 + +```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 更快。 diff --git a/docs/docs-cn/source/5.application-development/3.connector/index.rst b/docs/docs-cn/source/5.application-development/3.connector/index.rst index 5ca142767..2511eee35 100644 --- a/docs/docs-cn/source/5.application-development/3.connector/index.rst +++ b/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 diff --git a/docs/docs-en/source/5.application-development/3.connector/11.doris.md b/docs/docs-en/source/5.application-development/3.connector/11.doris.md new file mode 100644 index 000000000..761347206 --- /dev/null +++ b/docs/docs-en/source/5.application-development/3.connector/11.doris.md @@ -0,0 +1,104 @@ +# 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. | +| 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. | +| 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 csv. | +| 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 timeout in milliseconds for Stream Load, default 30000. | +| geaflow.dsl.doris.source.partition.num | false | The source partition number for parallel reads, default 1. | +| geaflow.dsl.doris.source.partition.column | false | The numeric column used to split the source into partitions, default id. | +| 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. CSV payload is used by default for higher +throughput; JSON payload is also supported. + +## 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. diff --git a/docs/docs-en/source/5.application-development/3.connector/index.rst b/docs/docs-en/source/5.application-development/3.connector/index.rst index 7e696edc4..ddfa8998c 100644 --- a/docs/docs-en/source/5.application-development/3.connector/index.rst +++ b/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 \ No newline at end of file + 10.udc.md + 11.doris.md \ No newline at end of file diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml new file mode 100644 index 000000000..808180751 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml @@ -0,0 +1,82 @@ + + + + + + org.apache.geaflow + geaflow-dsl-connector + 0.8.0-SNAPSHOT + + 4.0.0 + + geaflow-dsl-connector-doris + geaflow-dsl-connector-doris + + + 4.5.13 + 1.19.7 + + + + + org.apache.geaflow + geaflow-dsl-common + + + + org.apache.geaflow + geaflow-dsl-connector-api + + + + com.google.code.gson + gson + + + + org.apache.httpcomponents + httpclient + ${doris.httpclient.version} + + + + + mysql + mysql-connector-java + ${mysql.version} + + + + org.testng + testng + ${testng.version} + test + + + + org.testcontainers + testcontainers + ${testcontainers.version} + test + + + diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeys.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeys.java new file mode 100644 index 000000000..9855ef01a --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeys.java @@ -0,0 +1,122 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +import org.apache.geaflow.common.config.ConfigKey; +import org.apache.geaflow.common.config.ConfigKeys; + +public class DorisConfigKeys { + + public static final ConfigKey GEAFLOW_DSL_DORIS_FENODES = ConfigKeys + .key("geaflow.dsl.doris.fenodes") + .noDefaultValue() + .description("The Doris FE http address list (host:httpPort), comma separated. " + + "Used by the sink for Stream Load."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_JDBC_URL = ConfigKeys + .key("geaflow.dsl.doris.jdbc.url") + .noDefaultValue() + .description("The Doris query (MySQL protocol) jdbc url, e.g. " + + "jdbc:mysql://host:9030/database. Used by the source for partitioned reads."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_USERNAME = ConfigKeys + .key("geaflow.dsl.doris.username") + .defaultValue("root") + .description("The Doris username."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_PASSWORD = ConfigKeys + .key("geaflow.dsl.doris.password") + .defaultValue("") + .description("The Doris password."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_DATABASE = ConfigKeys + .key("geaflow.dsl.doris.database") + .noDefaultValue() + .description("The Doris database name."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_TABLE = ConfigKeys + .key("geaflow.dsl.doris.table") + .noDefaultValue() + .description("The Doris table name."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_SINK_FORMAT = ConfigKeys + .key("geaflow.dsl.doris.sink.format") + .defaultValue("csv") + .description("The Stream Load payload format, csv or json. Default csv."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_SINK_COLUMN_SEPARATOR = ConfigKeys + .key("geaflow.dsl.doris.sink.column.separator") + .defaultValue("\t") + .description("The column separator for csv Stream Load. Default tab."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_SINK_LINE_DELIMITER = ConfigKeys + .key("geaflow.dsl.doris.sink.line.delimiter") + .defaultValue("\n") + .description("The line delimiter for csv Stream Load. Default newline."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_SINK_MAX_ROWS = ConfigKeys + .key("geaflow.dsl.doris.sink.batch.rows") + .defaultValue(10000L) + .description("Flush the buffer to Doris when the buffered row count reaches this " + + "threshold. Default 10000."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_SINK_MAX_BYTES = ConfigKeys + .key("geaflow.dsl.doris.sink.batch.bytes") + .defaultValue(10485760L) + .description("Flush the buffer to Doris when the buffered byte size reaches this " + + "threshold. Default 10MB."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_SINK_MAX_RETRIES = ConfigKeys + .key("geaflow.dsl.doris.sink.max.retries") + .defaultValue(3) + .description("The max retry times for a Stream Load request. Default 3."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_REQUEST_CONNECT_TIMEOUT_MS = ConfigKeys + .key("geaflow.dsl.doris.request.connect.timeout.ms") + .defaultValue(30000) + .description("The connect timeout in milliseconds for Stream Load. Default 30000."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_REQUEST_READ_TIMEOUT_MS = ConfigKeys + .key("geaflow.dsl.doris.request.read.timeout.ms") + .defaultValue(30000) + .description("The socket read timeout in milliseconds for Stream Load. Default 30000."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_SOURCE_PARTITION_NUM = ConfigKeys + .key("geaflow.dsl.doris.source.partition.num") + .defaultValue(1L) + .description("The source partition number for parallel reads. Default 1."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_SOURCE_PARTITION_COLUMN = ConfigKeys + .key("geaflow.dsl.doris.source.partition.column") + .defaultValue("id") + .description("The numeric column used to split the source into partitions."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_SOURCE_PARTITION_LOWERBOUND = ConfigKeys + .key("geaflow.dsl.doris.source.partition.lowerbound") + .defaultValue(0L) + .description("The lowerbound of the partition column, only used to decide the partition " + + "stride, not for filtering rows."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_SOURCE_PARTITION_UPPERBOUND = ConfigKeys + .key("geaflow.dsl.doris.source.partition.upperbound") + .defaultValue(0L) + .description("The upperbound of the partition column, only used to decide the partition " + + "stride, not for filtering rows."); +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConstants.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConstants.java new file mode 100644 index 000000000..063c01e1c --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConstants.java @@ -0,0 +1,48 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +public class DorisConstants { + + public static final String FORMAT_CSV = "csv"; + + public static final String FORMAT_JSON = "json"; + + public static final String NULL_VALUE = "\\N"; + + public static final String COMMA = ","; + + public static final String COLON = ":"; + + public static final String HTTP_SCHEME = "http://"; + + /** + * The Stream Load url template: http://fe_host:http_port/api/{db}/{table}/_stream_load. + */ + public static final String STREAM_LOAD_URL_PATTERN = "%s/api/%s/%s/_stream_load"; + + public static final String STREAM_LOAD_RESULT_STATUS = "Status"; + + public static final String STREAM_LOAD_RESULT_MESSAGE = "Message"; + + public static final String STREAM_LOAD_SUCCESS = "Success"; + + public static final String STREAM_LOAD_PUBLISH_TIMEOUT = "Publish Timeout"; +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoad.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoad.java new file mode 100644 index 000000000..c62086715 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoad.java @@ -0,0 +1,181 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +import com.google.gson.Gson; +import com.google.gson.JsonObject; +import java.io.Closeable; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.List; +import java.util.UUID; +import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException; +import org.apache.http.HttpHeaders; +import org.apache.http.client.config.RequestConfig; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpPut; +import org.apache.http.entity.ByteArrayEntity; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.DefaultRedirectStrategy; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.util.EntityUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A lightweight Doris Stream Load client. It issues an HTTP PUT to the Doris FE, which replies + * with a 307 redirect to a BE, and the request body is re-sent to the BE that finally executes + * the load. The buffered payload is loaded in a single request, giving a much higher throughput + * than row-by-row JDBC INSERT. + */ +public class DorisStreamLoad implements Closeable { + + private static final Logger LOGGER = LoggerFactory.getLogger(DorisStreamLoad.class); + + private final String loadUrl; + private final String authHeader; + private final String format; + private final String columnSeparator; + private final String lineDelimiter; + private final String columns; + private final int maxRetries; + private final CloseableHttpClient httpClient; + private final Gson gson = new Gson(); + + public DorisStreamLoad(String feNode, String database, String table, String username, + String password, String format, String columnSeparator, + String lineDelimiter, List columns, int connectTimeoutMs, + int readTimeoutMs, int maxRetries) { + this.loadUrl = String.format(DorisConstants.STREAM_LOAD_URL_PATTERN, + normalizeFeNode(feNode), database, table); + this.authHeader = "Basic " + Base64.getEncoder().encodeToString( + (username + ":" + password).getBytes(StandardCharsets.UTF_8)); + this.format = format; + this.columnSeparator = columnSeparator; + this.lineDelimiter = lineDelimiter; + this.columns = String.join(DorisConstants.COMMA, columns); + this.maxRetries = Math.max(1, maxRetries); + RequestConfig requestConfig = RequestConfig.custom() + .setConnectTimeout(connectTimeoutMs) + .setSocketTimeout(readTimeoutMs) + .setConnectionRequestTimeout(connectTimeoutMs) + .build(); + // The FE returns a 307 redirect that must be followed with the same method and body, + // so mark every method as redirectable. + this.httpClient = HttpClients.custom() + .setDefaultRequestConfig(requestConfig) + .setRedirectStrategy(new DefaultRedirectStrategy() { + @Override + protected boolean isRedirectable(String method) { + return true; + } + }) + .build(); + } + + private static String normalizeFeNode(String feNode) { + String node = feNode.trim(); + if (!node.startsWith(DorisConstants.HTTP_SCHEME) && !node.startsWith("https://")) { + node = DorisConstants.HTTP_SCHEME + node; + } + return node; + } + + /** + * Load the given payload into Doris via a single Stream Load request. The method retries on + * transient failures and throws a {@link GeaFlowDSLException} once all retries are exhausted. + */ + public void load(byte[] payload) { + Exception lastError = null; + for (int attempt = 1; attempt <= maxRetries; attempt++) { + try { + doLoad(payload); + return; + } catch (Exception e) { + lastError = e; + LOGGER.warn("Stream Load attempt {}/{} failed: {}", attempt, maxRetries, + e.getMessage()); + } + } + throw new GeaFlowDSLException("Doris Stream Load failed after " + maxRetries + + " attempts.", lastError); + } + + private void doLoad(byte[] payload) throws IOException { + HttpPut put = new HttpPut(loadUrl); + put.setHeader(HttpHeaders.EXPECT, "100-continue"); + put.setHeader(HttpHeaders.AUTHORIZATION, authHeader); + put.setHeader("format", format); + put.setHeader("label", generateLabel()); + put.setHeader("two_phase_commit", "false"); + if (DorisConstants.FORMAT_CSV.equalsIgnoreCase(format)) { + put.setHeader("column_separator", columnSeparator); + put.setHeader("line_delimiter", lineDelimiter); + } else if (DorisConstants.FORMAT_JSON.equalsIgnoreCase(format)) { + put.setHeader("strip_outer_array", "true"); + put.setHeader("read_json_by_line", "false"); + } + if (columns != null && !columns.isEmpty()) { + put.setHeader("columns", columns); + } + put.setEntity(new ByteArrayEntity(payload)); + + try (CloseableHttpResponse response = httpClient.execute(put)) { + int statusCode = response.getStatusLine().getStatusCode(); + String body = response.getEntity() == null ? "" + : EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8); + if (statusCode != 200) { + throw new IOException("Stream Load http status: " + statusCode + ", body: " + body); + } + checkLoadResult(body); + } + } + + private void checkLoadResult(String body) throws IOException { + JsonObject result = gson.fromJson(body, JsonObject.class); + if (!result.has(DorisConstants.STREAM_LOAD_RESULT_STATUS)) { + throw new IOException("Stream Load response without status: " + body); + } + String status = result.get(DorisConstants.STREAM_LOAD_RESULT_STATUS).getAsString(); + if (!DorisConstants.STREAM_LOAD_SUCCESS.equals(status) + && !DorisConstants.STREAM_LOAD_PUBLISH_TIMEOUT.equals(status)) { + String message = result.has(DorisConstants.STREAM_LOAD_RESULT_MESSAGE) + ? result.get(DorisConstants.STREAM_LOAD_RESULT_MESSAGE).getAsString() : body; + throw new IOException("Stream Load failed, status: " + status + ", message: " + message); + } + } + + private String generateLabel() { + return "geaflow_" + System.currentTimeMillis() + "_" + UUID.randomUUID().toString() + .replace("-", ""); + } + + public String getLoadUrl() { + return loadUrl; + } + + @Override + public void close() throws IOException { + if (httpClient != null) { + httpClient.close(); + } + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableConnector.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableConnector.java new file mode 100644 index 000000000..d54d9b674 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableConnector.java @@ -0,0 +1,46 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +import org.apache.geaflow.common.config.Configuration; +import org.apache.geaflow.dsl.connector.api.TableReadableConnector; +import org.apache.geaflow.dsl.connector.api.TableSink; +import org.apache.geaflow.dsl.connector.api.TableSource; +import org.apache.geaflow.dsl.connector.api.TableWritableConnector; + +public class DorisTableConnector implements TableReadableConnector, TableWritableConnector { + + public static final String TYPE = "DORIS"; + + @Override + public String getType() { + return TYPE; + } + + @Override + public TableSource createSource(Configuration conf) { + return new DorisTableSource(); + } + + @Override + public TableSink createSink(Configuration conf) { + return new DorisTableSink(); + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSink.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSink.java new file mode 100644 index 000000000..1c51d940d --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSink.java @@ -0,0 +1,146 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import org.apache.geaflow.api.context.RuntimeContext; +import org.apache.geaflow.common.config.Configuration; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException; +import org.apache.geaflow.dsl.common.types.StructType; +import org.apache.geaflow.dsl.common.types.TableField; +import org.apache.geaflow.dsl.connector.api.TableSink; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A Doris table sink that buffers rows and flushes them to Doris via Stream Load. Rows are + * accumulated in {@link #write(Row)} and a whole batch is loaded once the buffered row count or + * byte size reaches the configured threshold (or when {@link #finish()} is called at the end of a + * window). This batched load path gives a much higher write throughput than row-by-row JDBC + * INSERT. + */ +public class DorisTableSink implements TableSink { + + private static final Logger LOGGER = LoggerFactory.getLogger(DorisTableSink.class); + + private StructType schema; + private String feNode; + private String database; + private String table; + private String username; + private String password; + private String format; + private String columnSeparator; + private String lineDelimiter; + private long maxRows; + private long maxBytes; + private int maxRetries; + private int connectTimeoutMs; + private int readTimeoutMs; + + private transient DorisStreamLoad streamLoad; + private transient List buffer; + private transient long bufferBytes; + + @Override + public void init(Configuration conf, StructType tableSchema) { + LOGGER.info("init doris sink with config: {}, \n schema: {}", conf, tableSchema); + this.schema = tableSchema; + String feNodes = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_FENODES, ""); + if (feNodes == null || feNodes.trim().isEmpty()) { + throw new GeaFlowDSLException("Doris fenodes must be specified for the sink."); + } + this.feNode = feNodes.split(DorisConstants.COMMA)[0].trim(); + this.database = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE); + this.table = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE); + this.username = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_USERNAME); + this.password = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_PASSWORD); + this.format = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_FORMAT); + this.columnSeparator = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_COLUMN_SEPARATOR); + this.lineDelimiter = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_LINE_DELIMITER); + this.maxRows = conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_ROWS); + this.maxBytes = conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_BYTES); + this.maxRetries = conf.getInteger(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_RETRIES); + this.connectTimeoutMs = conf.getInteger(DorisConfigKeys.GEAFLOW_DSL_DORIS_REQUEST_CONNECT_TIMEOUT_MS); + this.readTimeoutMs = conf.getInteger(DorisConfigKeys.GEAFLOW_DSL_DORIS_REQUEST_READ_TIMEOUT_MS); + } + + @Override + public void open(RuntimeContext context) { + List columns = new ArrayList<>(); + for (TableField field : schema.getFields()) { + columns.add(field.getName()); + } + this.streamLoad = new DorisStreamLoad(feNode, database, table, username, password, format, + columnSeparator, lineDelimiter, columns, connectTimeoutMs, readTimeoutMs, maxRetries); + this.buffer = new ArrayList<>(); + this.bufferBytes = 0L; + } + + @Override + public void write(Row row) throws IOException { + String record = DorisConstants.FORMAT_JSON.equalsIgnoreCase(format) + ? DorisUtils.rowToJson(row, schema) + : DorisUtils.rowToCsv(row, schema, columnSeparator); + buffer.add(record); + bufferBytes += record.getBytes(StandardCharsets.UTF_8).length; + if (buffer.size() >= maxRows || bufferBytes >= maxBytes) { + flushBuffer(); + } + } + + @Override + public void finish() throws IOException { + flushBuffer(); + } + + private void flushBuffer() { + if (buffer == null || buffer.isEmpty()) { + return; + } + String payload; + if (DorisConstants.FORMAT_JSON.equalsIgnoreCase(format)) { + payload = "[" + String.join(DorisConstants.COMMA, buffer) + "]"; + } else { + payload = String.join(lineDelimiter, buffer); + } + streamLoad.load(payload.getBytes(StandardCharsets.UTF_8)); + LOGGER.info("flushed {} rows ({} bytes) to doris table {}.{}", buffer.size(), bufferBytes, + database, table); + buffer.clear(); + bufferBytes = 0L; + } + + @Override + public void close() { + try { + if (streamLoad != null) { + streamLoad.close(); + streamLoad = null; + } + } catch (IOException e) { + throw new GeaFlowDSLException("failed to close doris stream load client.", e); + } + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSource.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSource.java new file mode 100644 index 000000000..506d8bd0b --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSource.java @@ -0,0 +1,301 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +import java.io.IOException; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import org.apache.geaflow.api.context.RuntimeContext; +import org.apache.geaflow.api.window.WindowType; +import org.apache.geaflow.common.config.Configuration; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.data.impl.ObjectRow; +import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException; +import org.apache.geaflow.dsl.common.types.StructType; +import org.apache.geaflow.dsl.common.types.TableSchema; +import org.apache.geaflow.dsl.common.util.Windows; +import org.apache.geaflow.dsl.connector.api.FetchData; +import org.apache.geaflow.dsl.connector.api.Offset; +import org.apache.geaflow.dsl.connector.api.Partition; +import org.apache.geaflow.dsl.connector.api.TableSource; +import org.apache.geaflow.dsl.connector.api.serde.DeserializerFactory; +import org.apache.geaflow.dsl.connector.api.serde.TableDeserializer; +import org.apache.geaflow.dsl.connector.api.window.FetchWindow; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A Doris table source that reads data through the MySQL protocol exposed by the Doris FE. The + * data set can be split into several partitions on a numeric column so that partitions are read + * in parallel by different tasks. + */ +public class DorisTableSource implements TableSource { + + private static final Logger LOGGER = LoggerFactory.getLogger(DorisTableSource.class); + + private static final String MYSQL_DRIVER = "com.mysql.cj.jdbc.Driver"; + + private StructType schema; + private String jdbcUrl; + private String username; + private String password; + private String database; + private String table; + private long partitionNum; + private String partitionColumn; + private long lowerBound; + private long upperBound; + + private Map partitionConnectionMap = new HashMap<>(); + private Map partitionStatementMap = new HashMap<>(); + + @Override + public void init(Configuration conf, TableSchema tableSchema) { + LOGGER.info("init doris source with config: {}, \n schema: {}", conf, tableSchema); + this.schema = tableSchema; + this.jdbcUrl = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_JDBC_URL, ""); + if (jdbcUrl == null || jdbcUrl.trim().isEmpty()) { + throw new GeaFlowDSLException("Doris jdbc url must be specified for the source."); + } + this.username = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_USERNAME); + this.password = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_PASSWORD); + this.database = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE, ""); + this.table = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE); + this.partitionNum = conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_NUM); + if (partitionNum <= 0) { + throw new GeaFlowDSLException("Invalid doris source partition number: {}", partitionNum); + } + this.partitionColumn = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_COLUMN); + this.lowerBound = conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_LOWERBOUND); + this.upperBound = conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_UPPERBOUND); + if (partitionNum > 1 && lowerBound >= upperBound) { + throw new GeaFlowDSLException("Upperbound must be greater than lowerbound " + + "(lowerbound:%d upperbound:%d).", lowerBound, upperBound); + } + } + + @Override + public void open(RuntimeContext context) { + try { + Class.forName(MYSQL_DRIVER); + } catch (ClassNotFoundException e) { + throw new GeaFlowDSLException("failed to load mysql driver for doris source.", e); + } + } + + private String qualifiedTable() { + return database == null || database.isEmpty() ? table : database + "." + table; + } + + @Override + public List listPartitions() { + if (partitionNum == 1) { + return Collections.singletonList(new DorisPartition(qualifiedTable(), "")); + } + long span = Math.min(upperBound - lowerBound, partitionNum); + long stride = (upperBound - lowerBound) / span; + long currentValue = lowerBound; + List partitions = new ArrayList<>(); + for (long i = 0; i < span; i++) { + String lBound = i != 0 ? String.format("%s >= %d", partitionColumn, currentValue) : null; + currentValue += stride; + String uBound = i != span - 1 + ? String.format("%s < %d", partitionColumn, currentValue) : null; + String whereClause; + if (uBound == null) { + whereClause = lBound; + } else if (lBound == null) { + whereClause = String.format("%s OR %s IS NULL", uBound, partitionColumn); + } else { + whereClause = String.format("%s AND %s", lBound, uBound); + } + partitions.add(new DorisPartition(qualifiedTable(), "WHERE " + whereClause)); + } + return partitions; + } + + @Override + public List listPartitions(int parallelism) { + return listPartitions(); + } + + @Override + public TableDeserializer getDeserializer(Configuration conf) { + return DeserializerFactory.loadRowTableDeserializer(); + } + + @Override + public FetchData fetch(Partition partition, Optional startOffset, + FetchWindow windowInfo) throws IOException { + if (!(windowInfo.getType() == WindowType.SIZE_TUMBLING_WINDOW + || windowInfo.getType() == WindowType.ALL_WINDOW)) { + throw new GeaFlowDSLException("Not support window type:{}", windowInfo.getType()); + } + DorisPartition dorisPartition = (DorisPartition) partition; + Statement statement = partitionStatementMap.get(partition); + if (statement == null) { + try { + Connection connection = DriverManager.getConnection(jdbcUrl, username, password); + statement = connection.createStatement(); + partitionConnectionMap.put(partition, connection); + partitionStatementMap.put(partition, statement); + } catch (SQLException e) { + throw new GeaFlowDSLException("failed to connect to doris.", e); + } + } + + long offset = startOffset.map(Offset::getOffset).orElse(0L); + long windowSize = windowInfo.windowSize(); + if (windowSize == Windows.SIZE_OF_ALL_WINDOW) { + windowSize = Integer.MAX_VALUE; + } else if (windowSize <= 0) { + throw new GeaFlowDSLException("wrong windowSize: {}", windowSize); + } + + List dataList; + try { + dataList = selectRows(statement, dorisPartition, offset, windowSize); + } catch (SQLException e) { + throw new GeaFlowDSLException("select rows from doris table failed.", e); + } + DorisOffset nextOffset = new DorisOffset(offset + dataList.size()); + boolean isFinish = windowInfo.getType() == WindowType.ALL_WINDOW + || dataList.size() < windowInfo.windowSize(); + return (FetchData) FetchData.createStreamFetch(dataList, nextOffset, isFinish); + } + + private List selectRows(Statement statement, DorisPartition partition, long offset, + long windowSize) throws SQLException { + int columnNum = schema.size(); + String orderColumn = schema.getField(0).getName(); + String query = String.format("SELECT * FROM %s %s ORDER BY %s LIMIT %s OFFSET %s", + partition.getTableName(), partition.getWhereClause(), orderColumn, windowSize, offset); + List rowList = new ArrayList<>(); + try (ResultSet resultSet = statement.executeQuery(query)) { + while (resultSet.next()) { + Object[] values = new Object[columnNum]; + for (int i = 1; i <= columnNum; i++) { + values[i - 1] = resultSet.getObject(i); + } + rowList.add(ObjectRow.create(values)); + } + } + return rowList; + } + + @Override + public void close() { + try { + for (Statement statement : partitionStatementMap.values()) { + if (statement != null) { + statement.close(); + } + } + partitionStatementMap.clear(); + for (Connection connection : partitionConnectionMap.values()) { + if (connection != null) { + connection.close(); + } + } + partitionConnectionMap.clear(); + } catch (SQLException e) { + throw new GeaFlowDSLException("failed to close doris source connection.", e); + } + } + + public static class DorisPartition implements Partition { + + private final String tableName; + private final String whereClause; + + public DorisPartition(String tableName, String whereClause) { + this.tableName = tableName; + this.whereClause = whereClause; + } + + public String getTableName() { + return tableName; + } + + public String getWhereClause() { + return whereClause; + } + + @Override + public String getName() { + if (whereClause == null || whereClause.isEmpty()) { + return tableName; + } + return tableName + "-" + whereClause; + } + + @Override + public int hashCode() { + return Objects.hash(tableName, whereClause); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof DorisPartition)) { + return false; + } + DorisPartition that = (DorisPartition) o; + return Objects.equals(tableName, that.tableName) + && Objects.equals(whereClause, that.whereClause); + } + } + + public static class DorisOffset implements Offset { + + private final long offset; + + public DorisOffset(long offset) { + this.offset = offset; + } + + @Override + public String humanReadable() { + return String.valueOf(offset); + } + + @Override + public long getOffset() { + return offset; + } + + @Override + public boolean isTimestamp() { + return false; + } + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisUtils.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisUtils.java new file mode 100644 index 000000000..28277f622 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisUtils.java @@ -0,0 +1,67 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +import com.google.gson.Gson; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.types.StructType; +import org.apache.geaflow.dsl.common.types.TableField; + +public class DorisUtils { + + private static final Gson GSON = new Gson(); + + /** + * Serialize a row to a single csv line using the given column separator. Null fields are + * rendered as Doris's null placeholder ("\N"). + */ + public static String rowToCsv(Row row, StructType schema, String columnSeparator) { + List fields = schema.getFields(); + StringBuilder builder = new StringBuilder(); + for (int i = 0; i < fields.size(); i++) { + if (i > 0) { + builder.append(columnSeparator); + } + Object value = row.getField(i, fields.get(i).getType()); + if (value == null) { + builder.append(DorisConstants.NULL_VALUE); + } else { + builder.append(value); + } + } + return builder.toString(); + } + + /** + * Serialize a row to a json object string keyed by the column names. + */ + public static String rowToJson(Row row, StructType schema) { + List fields = schema.getFields(); + Map map = new LinkedHashMap<>(); + for (int i = 0; i < fields.size(); i++) { + Object value = row.getField(i, fields.get(i).getType()); + map.put(fields.get(i).getName(), value); + } + return GSON.toJson(map); + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/resources/META-INF/services/org.apache.geaflow.dsl.connector.api.TableConnector b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/resources/META-INF/services/org.apache.geaflow.dsl.connector.api.TableConnector new file mode 100644 index 000000000..855b0ca01 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/resources/META-INF/services/org.apache.geaflow.dsl.connector.api.TableConnector @@ -0,0 +1,20 @@ +# +# 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. +# + +org.apache.geaflow.dsl.connector.doris.DorisTableConnector diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeysTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeysTest.java new file mode 100644 index 000000000..0a27164bb --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeysTest.java @@ -0,0 +1,54 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +import org.apache.geaflow.common.config.Configuration; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class DorisConfigKeysTest { + + @Test + public void testSinkDefaults() { + Configuration conf = new Configuration(); + Assert.assertEquals(conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_ROWS), 10000L); + Assert.assertEquals(conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_BYTES), + 10485760L); + Assert.assertEquals(conf.getInteger(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_RETRIES), 3); + Assert.assertEquals(conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_FORMAT), "csv"); + Assert.assertEquals(conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_USERNAME), "root"); + } + + @Test + public void testSourcePartitionDefaults() { + Configuration conf = new Configuration(); + Assert.assertEquals( + conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_NUM), 1L); + Assert.assertEquals( + conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_COLUMN), "id"); + } + + @Test + public void testOverrideValue() { + Configuration conf = new Configuration(); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_ROWS.getKey(), "500"); + Assert.assertEquals(conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_ROWS), 500L); + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java new file mode 100644 index 000000000..217e62a29 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java @@ -0,0 +1,260 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Optional; +import org.apache.geaflow.common.config.Configuration; +import org.apache.geaflow.common.type.Types; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.data.impl.ObjectRow; +import org.apache.geaflow.dsl.common.types.StructType; +import org.apache.geaflow.dsl.common.types.TableField; +import org.apache.geaflow.dsl.common.types.TableSchema; +import org.apache.geaflow.dsl.connector.api.FetchData; +import org.apache.geaflow.dsl.connector.api.Partition; +import org.apache.geaflow.dsl.connector.api.window.AllFetchWindow; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.DockerClientFactory; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testng.Assert; +import org.testng.SkipException; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +/** + * Integration test for the Doris connector. It boots a real Doris (all-in-one) container using + * Testcontainers and exercises the Stream Load sink and the partitioned source end to end. The + * container requires host networking (so the FE-to-BE Stream Load redirect is reachable), which is + * only available on Linux; the test skips itself when Docker or host networking is not available. + * + *

An external Doris can be targeted instead by setting the {@code doris.it.fenodes} and + * {@code doris.it.jdbcUrl} system properties. + */ +public class DorisConnectorIntegrationTest { + + private static final Logger LOGGER = + LoggerFactory.getLogger(DorisConnectorIntegrationTest.class); + + private static final String DORIS_IMAGE = "apache/doris:doris-all-in-one-2.1.0"; + private static final String DATABASE = "geaflow_it"; + private static final String TABLE = "person"; + private static final String USERNAME = "root"; + private static final String PASSWORD = ""; + + private static GenericContainer dorisContainer; + private static String feNodes; + private static String jdbcUrl; + + @BeforeClass + public void setUp() throws Exception { + String externalFe = System.getProperty("doris.it.fenodes"); + String externalJdbc = System.getProperty("doris.it.jdbcUrl"); + if (externalFe != null && externalJdbc != null) { + feNodes = externalFe; + jdbcUrl = externalJdbc; + } else { + if (!DockerClientFactory.instance().isDockerAvailable()) { + throw new SkipException("Docker is not available, skip Doris integration test."); + } + if (!System.getProperty("os.name", "").toLowerCase().contains("linux")) { + throw new SkipException("Doris Stream Load redirect needs host networking, only " + + "supported on Linux, skip integration test."); + } + dorisContainer = new GenericContainer<>(DORIS_IMAGE) + .withNetworkMode("host") + .waitingFor(Wait.forLogMessage(".*get heartbeat response.*", 1) + .withStartupTimeout(java.time.Duration.ofMinutes(5))); + dorisContainer.start(); + feNodes = "127.0.0.1:8030"; + jdbcUrl = "jdbc:mysql://127.0.0.1:9030/"; + } + waitForBackendAlive(); + prepareSchema(); + } + + @AfterClass + public void tearDown() { + if (dorisContainer != null) { + dorisContainer.stop(); + } + } + + private Connection newConnection(String url) throws Exception { + Class.forName("com.mysql.cj.jdbc.Driver"); + return DriverManager.getConnection(url, USERNAME, PASSWORD); + } + + private void waitForBackendAlive() throws Exception { + long deadline = System.currentTimeMillis() + java.time.Duration.ofMinutes(3).toMillis(); + while (System.currentTimeMillis() < deadline) { + try (Connection connection = newConnection(jdbcUrl); + Statement statement = connection.createStatement(); + ResultSet rs = statement.executeQuery("SHOW BACKENDS")) { + while (rs.next()) { + if ("true".equalsIgnoreCase(rs.getString("Alive"))) { + return; + } + } + } catch (Exception e) { + LOGGER.info("waiting for doris backend alive: {}", e.getMessage()); + } + Thread.sleep(5000); + } + throw new IllegalStateException("Doris backend did not become alive in time."); + } + + private void prepareSchema() throws Exception { + try (Connection connection = newConnection(jdbcUrl); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE IF NOT EXISTS " + DATABASE); + statement.execute("DROP TABLE IF EXISTS " + DATABASE + "." + TABLE); + statement.execute("CREATE TABLE " + DATABASE + "." + TABLE + " (" + + "id BIGINT, name VARCHAR(64), age INT) " + + "UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES(\"replication_num\" = \"1\")"); + } + } + + private StructType schema() { + return new StructType( + new TableField("id", Types.LONG, false), + new TableField("name", Types.BINARY_STRING, true), + new TableField("age", Types.INTEGER, true)); + } + + private Configuration sinkConfig() { + Configuration conf = new Configuration(); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_FENODES, feNodes); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE, DATABASE); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, TABLE); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_USERNAME, USERNAME); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_PASSWORD, PASSWORD); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_ROWS, "1000"); + return conf; + } + + private long countRows() throws Exception { + try (Connection connection = newConnection(jdbcUrl); + Statement statement = connection.createStatement(); + ResultSet rs = statement.executeQuery( + "SELECT COUNT(*) FROM " + DATABASE + "." + TABLE)) { + rs.next(); + return rs.getLong(1); + } + } + + private void truncate() throws Exception { + try (Connection connection = newConnection(jdbcUrl); + Statement statement = connection.createStatement()) { + statement.execute("TRUNCATE TABLE " + DATABASE + "." + TABLE); + } + } + + @Test + public void testSinkAndSource() throws Exception { + truncate(); + int rowCount = 2000; + DorisTableSink sink = new DorisTableSink(); + sink.init(sinkConfig(), schema()); + sink.open(null); + for (int i = 0; i < rowCount; i++) { + sink.write(ObjectRow.create((long) i, "name_" + i, i % 100)); + } + sink.finish(); + sink.close(); + Assert.assertEquals(countRows(), rowCount); + + Configuration sourceConf = new Configuration(); + sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_JDBC_URL, jdbcUrl + DATABASE); + sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE, DATABASE); + sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, TABLE); + sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_USERNAME, USERNAME); + sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_PASSWORD, PASSWORD); + sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_NUM, "4"); + sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_COLUMN, "id"); + sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_LOWERBOUND, "0"); + sourceConf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_UPPERBOUND, + String.valueOf(rowCount)); + + DorisTableSource source = new DorisTableSource(); + source.init(sourceConf, new TableSchema(schema())); + source.open(null); + List partitions = source.listPartitions(); + Assert.assertTrue(partitions.size() > 1, "source should produce parallel partitions."); + List allRows = new ArrayList<>(); + for (Partition partition : partitions) { + FetchData fetchData = + source.fetch(partition, Optional.empty(), new AllFetchWindow(0)); + Iterator iterator = fetchData.getDataIterator(); + while (iterator.hasNext()) { + allRows.add(iterator.next()); + } + } + source.close(); + Assert.assertEquals(allRows.size(), rowCount); + } + + @Test + public void testStreamLoadThroughputBenchmark() throws Exception { + int rowCount = 20000; + + truncate(); + DorisTableSink sink = new DorisTableSink(); + sink.init(sinkConfig(), schema()); + sink.open(null); + long streamLoadStart = System.currentTimeMillis(); + for (int i = 0; i < rowCount; i++) { + sink.write(ObjectRow.create((long) i, "name_" + i, i % 100)); + } + sink.finish(); + sink.close(); + long streamLoadCost = System.currentTimeMillis() - streamLoadStart; + Assert.assertEquals(countRows(), rowCount); + + truncate(); + long jdbcStart = System.currentTimeMillis(); + try (Connection connection = newConnection(jdbcUrl); + Statement statement = connection.createStatement()) { + for (int i = 0; i < rowCount; i++) { + statement.execute(String.format( + "INSERT INTO %s.%s (id, name, age) VALUES (%d, '%s', %d)", + DATABASE, TABLE, i, "name_" + i, i % 100)); + } + } + long jdbcCost = System.currentTimeMillis() - jdbcStart; + + LOGGER.info("Doris write benchmark for {} rows: streamLoad={}ms ({} rows/s), " + + "row-by-row JDBC={}ms ({} rows/s)", rowCount, streamLoadCost, + rowCount * 1000L / Math.max(1, streamLoadCost), jdbcCost, + rowCount * 1000L / Math.max(1, jdbcCost)); + Assert.assertTrue(streamLoadCost < jdbcCost, + "Stream Load should be faster than row-by-row JDBC insert."); + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoadTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoadTest.java new file mode 100644 index 000000000..37b840861 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoadTest.java @@ -0,0 +1,49 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +import java.io.IOException; +import java.util.Arrays; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class DorisStreamLoadTest { + + private DorisStreamLoad newClient(String feNode) { + return new DorisStreamLoad(feNode, "test_db", "test_table", "root", "", + DorisConstants.FORMAT_CSV, "\t", "\n", Arrays.asList("id", "name"), 1000, 1000, 3); + } + + @Test + public void testLoadUrlWithHostPort() throws IOException { + try (DorisStreamLoad client = newClient("127.0.0.1:8030")) { + Assert.assertEquals(client.getLoadUrl(), + "http://127.0.0.1:8030/api/test_db/test_table/_stream_load"); + } + } + + @Test + public void testLoadUrlWithScheme() throws IOException { + try (DorisStreamLoad client = newClient("http://doris-fe:8030")) { + Assert.assertEquals(client.getLoadUrl(), + "http://doris-fe:8030/api/test_db/test_table/_stream_load"); + } + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSinkTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSinkTest.java new file mode 100644 index 000000000..7ae21a5be --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSinkTest.java @@ -0,0 +1,65 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +import org.apache.geaflow.common.config.Configuration; +import org.apache.geaflow.common.type.Types; +import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException; +import org.apache.geaflow.dsl.common.types.StructType; +import org.apache.geaflow.dsl.common.types.TableField; +import org.testng.Assert; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +public class DorisTableSinkTest { + + private DorisTableSink sink; + private StructType schema; + + @BeforeMethod + public void setUp() { + sink = new DorisTableSink(); + schema = new StructType( + new TableField("id", Types.LONG, false), + new TableField("name", Types.BINARY_STRING, true)); + } + + private Configuration validConfig() { + Configuration conf = new Configuration(); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_FENODES, "127.0.0.1:8030"); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE, "test_db"); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, "test_table"); + return conf; + } + + @Test + public void testInit() { + sink.init(validConfig(), schema); + Assert.assertNotNull(sink); + } + + @Test(expectedExceptions = GeaFlowDSLException.class) + public void testInitWithoutFeNodes() { + Configuration conf = new Configuration(); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE, "test_db"); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, "test_table"); + sink.init(conf, schema); + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisUtilsTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisUtilsTest.java new file mode 100644 index 000000000..7b9b0dd0a --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisUtilsTest.java @@ -0,0 +1,72 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +import org.apache.geaflow.common.type.Types; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.data.impl.ObjectRow; +import org.apache.geaflow.dsl.common.types.StructType; +import org.apache.geaflow.dsl.common.types.TableField; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class DorisUtilsTest { + + private StructType buildSchema() { + return new StructType( + new TableField("id", Types.LONG, false), + new TableField("name", Types.BINARY_STRING, true), + new TableField("score", Types.DOUBLE, true)); + } + + @Test + public void testRowToCsv() { + StructType schema = buildSchema(); + Row row = ObjectRow.create(1L, "alice", 9.5); + String csv = DorisUtils.rowToCsv(row, schema, "\t"); + Assert.assertEquals(csv, "1\talice\t9.5"); + } + + @Test + public void testRowToCsvWithNull() { + StructType schema = buildSchema(); + Row row = ObjectRow.create(2L, null, null); + String csv = DorisUtils.rowToCsv(row, schema, "\t"); + Assert.assertEquals(csv, "2\t\\N\t\\N"); + } + + @Test + public void testRowToCsvWithCustomSeparator() { + StructType schema = buildSchema(); + Row row = ObjectRow.create(3L, "bob", 1.0); + String csv = DorisUtils.rowToCsv(row, schema, ","); + Assert.assertEquals(csv, "3,bob,1.0"); + } + + @Test + public void testRowToJson() { + StructType schema = buildSchema(); + Row row = ObjectRow.create(4L, "carol", 8.0); + String json = DorisUtils.rowToJson(row, schema); + Assert.assertTrue(json.contains("\"id\":4")); + Assert.assertTrue(json.contains("\"name\":\"carol\"")); + Assert.assertTrue(json.contains("\"score\":8.0")); + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml index e1f262b9d..4fe03c6de 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml @@ -49,6 +49,7 @@ geaflow-dsl-connector-paimon geaflow-dsl-connector-neo4j geaflow-dsl-connector-elasticsearch + geaflow-dsl-connector-doris diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-runtime/pom.xml index e9863eb2e..87c9f040a 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-runtime/pom.xml +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/pom.xml @@ -81,6 +81,11 @@ geaflow-dsl-connector-jdbc + + org.apache.geaflow + geaflow-dsl-connector-doris + + org.apache.geaflow geaflow-dsl-connector-odps diff --git a/geaflow/geaflow-dsl/pom.xml b/geaflow/geaflow-dsl/pom.xml index f79ea4e0d..a7a37063f 100644 --- a/geaflow/geaflow-dsl/pom.xml +++ b/geaflow/geaflow-dsl/pom.xml @@ -188,6 +188,12 @@ ${project.version} + + org.apache.geaflow + geaflow-dsl-connector-doris + ${project.version} + + From 1b4223597b779403c9d68b9ad5f67a5b2f8217d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=88=9A=E5=AE=8F=E6=88=90?= Date: Wed, 22 Jul 2026 19:26:06 +0800 Subject: [PATCH 2/6] [ISSUE-817] Make Doris integration test opt-in to keep CI green The Doris Testcontainers integration test previously ran whenever Docker was available on a Linux host, which is exactly the case on GitHub CI runners, so it would try to pull the multi-GB Doris image and start a container during a normal 'mvn test'. Make it opt-in: it only runs with -Ddoris.it.enabled=true or when an external Doris is provided via -Ddoris.it.fenodes/-Ddoris.it.jdbcUrl, and otherwise skips. Docs updated accordingly. --- .../5.application-development/3.connector/11.doris.md | 4 ++++ .../5.application-development/3.connector/11.doris.md | 5 +++++ .../connector/doris/DorisConnectorIntegrationTest.java | 8 +++++++- 3 files changed, 16 insertions(+), 1 deletion(-) diff --git a/docs/docs-cn/source/5.application-development/3.connector/11.doris.md b/docs/docs-cn/source/5.application-development/3.connector/11.doris.md index deb66b507..cabf7a5f2 100644 --- a/docs/docs-cn/source/5.application-development/3.connector/11.doris.md +++ b/docs/docs-cn/source/5.application-development/3.connector/11.doris.md @@ -98,3 +98,7 @@ SELECT * FROM doris_source_table; 对同一张表写入 20000 行，Stream Load 攒批写入相比逐行 JDBC INSERT 的耗时明显更低 (通常快一个数量级以上，具体倍数取决于批大小、网络与集群规模)。该基准会断言 Stream Load 比逐行 JDBC 更快。 + +该集成测试(含基准)依赖真实 Doris，默认关闭、不会在普通 CI 中运行。可通过以下两种方式开启： +- 加参数 `-Ddoris.it.enabled=true`(需要 Linux 宿主机 + Docker，会用 Testcontainers 启动 Doris)； +- 或通过 `-Ddoris.it.fenodes` 和 `-Ddoris.it.jdbcUrl` 指向一个外部 Doris 实例。 diff --git a/docs/docs-en/source/5.application-development/3.connector/11.doris.md b/docs/docs-en/source/5.application-development/3.connector/11.doris.md index 761347206..8d9f188f7 100644 --- a/docs/docs-en/source/5.application-development/3.connector/11.doris.md +++ b/docs/docs-en/source/5.application-development/3.connector/11.doris.md @@ -102,3 +102,8 @@ In the integration test 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. + +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`. diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java index 217e62a29..7f7398484 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java @@ -79,7 +79,9 @@ public void setUp() throws Exception { if (externalFe != null && externalJdbc != null) { feNodes = externalFe; jdbcUrl = externalJdbc; - } else { + } else if (Boolean.getBoolean("doris.it.enabled")) { + // The container is heavy (multi-GB image) and needs host networking so the FE-to-BE + // Stream Load redirect is reachable, so it is opt-in and never runs in a normal CI. if (!DockerClientFactory.instance().isDockerAvailable()) { throw new SkipException("Docker is not available, skip Doris integration test."); } @@ -94,6 +96,10 @@ public void setUp() throws Exception { dorisContainer.start(); feNodes = "127.0.0.1:8030"; jdbcUrl = "jdbc:mysql://127.0.0.1:9030/"; + } else { + throw new SkipException("Doris integration test is disabled by default. Enable it with " + + "-Ddoris.it.enabled=true (requires Docker on a Linux host), or point it at an " + + "external Doris with -Ddoris.it.fenodes and -Ddoris.it.jdbcUrl."); } waitForBackendAlive(); prepareSchema(); From 069b398a9b22c0f865807a5cb5545833bacefc85 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=88=9A=E5=AE=8F=E6=88=90?= Date: Thu, 23 Jul 2026 11:46:52 +0800 Subject: [PATCH 3/6] [ISSUE-817] Address review: serialization safety, idempotent close, FE failover Address the maintainer's review points on the Doris connector: - serialization: default to json (Gson with serializeNulls + disableHtmlEscaping) so newlines, quotes, backslashes, unicode, null and empty strings round-trip correctly; document csv as a no-escape plain-split option. - sink flush: flushBuffer stays safe on an empty buffer and is idempotent on finish(); close() now flushes the remaining buffer and is safe on repeated calls and when the client is not initialized. - partition logic: clamp partition.num to at least 1 instead of throwing; the first partition already covers rows whose partition column is entirely NULL. - label: kept globally unique via UUID. - FE failover: DorisStreamLoad now takes the full FE list and rotates to the next FE on a failed request (retries at least once per FE). - integration tests: add table-not-found, FE-failover and full data-type round-trip fidelity cases (still opt-in). Add unit tests for special-char / null / empty-string serialization and multi-FE url building. --- .../3.connector/11.doris.md | 4 +- .../3.connector/11.doris.md | 4 +- .../dsl/connector/doris/DorisConfigKeys.java | 9 ++- .../dsl/connector/doris/DorisStreamLoad.java | 40 +++++++--- .../dsl/connector/doris/DorisTableSink.java | 38 +++++++--- .../dsl/connector/doris/DorisTableSource.java | 7 +- .../dsl/connector/doris/DorisUtils.java | 31 +++++++- .../connector/doris/DorisConfigKeysTest.java | 2 +- .../doris/DorisConnectorIntegrationTest.java | 76 +++++++++++++++++++ .../connector/doris/DorisStreamLoadTest.java | 27 ++++++- .../dsl/connector/doris/DorisUtilsTest.java | 40 ++++++++++ 11 files changed, 236 insertions(+), 42 deletions(-) diff --git a/docs/docs-cn/source/5.application-development/3.connector/11.doris.md b/docs/docs-cn/source/5.application-development/3.connector/11.doris.md index cabf7a5f2..b186bc749 100644 --- a/docs/docs-cn/source/5.application-development/3.connector/11.doris.md +++ b/docs/docs-cn/source/5.application-development/3.connector/11.doris.md @@ -28,13 +28,13 @@ CREATE TABLE doris_table ( | 参数名 | 是否必须 | 描述 | | ---- | ---- | ---- | -| geaflow.dsl.doris.fenodes | 写入必须 | Doris FE 的 HTTP 地址列表(host:httpPort)，逗号分隔，Sink 通过它做 Stream Load。 | +| 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 通过它做分区并行读。 | | 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，默认 csv。 | +| 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。 | diff --git a/docs/docs-en/source/5.application-development/3.connector/11.doris.md b/docs/docs-en/source/5.application-development/3.connector/11.doris.md index 8d9f188f7..d039a906f 100644 --- a/docs/docs-en/source/5.application-development/3.connector/11.doris.md +++ b/docs/docs-en/source/5.application-development/3.connector/11.doris.md @@ -30,13 +30,13 @@ CREATE TABLE doris_table ( | 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. | +| 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. | | 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 csv. | +| 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. | diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeys.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeys.java index 9855ef01a..affc2fe17 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeys.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeys.java @@ -28,7 +28,8 @@ public class DorisConfigKeys { .key("geaflow.dsl.doris.fenodes") .noDefaultValue() .description("The Doris FE http address list (host:httpPort), comma separated. " - + "Used by the sink for Stream Load."); + + "Used by the sink for Stream Load. When multiple FEs are given the sink fails over " + + "to the next FE on a failed request."); public static final ConfigKey GEAFLOW_DSL_DORIS_JDBC_URL = ConfigKeys .key("geaflow.dsl.doris.jdbc.url") @@ -58,8 +59,10 @@ public class DorisConfigKeys { public static final ConfigKey GEAFLOW_DSL_DORIS_SINK_FORMAT = ConfigKeys .key("geaflow.dsl.doris.sink.format") - .defaultValue("csv") - .description("The Stream Load payload format, csv or json. Default csv."); + .defaultValue("json") + .description("The Stream Load payload format, csv or json. Default json, which 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 separators."); public static final ConfigKey GEAFLOW_DSL_DORIS_SINK_COLUMN_SEPARATOR = ConfigKeys .key("geaflow.dsl.doris.sink.column.separator") diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoad.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoad.java index c62086715..9e049bb7c 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoad.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoad.java @@ -24,6 +24,7 @@ import java.io.Closeable; import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.Base64; import java.util.List; import java.util.UUID; @@ -50,7 +51,7 @@ public class DorisStreamLoad implements Closeable { private static final Logger LOGGER = LoggerFactory.getLogger(DorisStreamLoad.class); - private final String loadUrl; + private final List loadUrls; private final String authHeader; private final String format; private final String columnSeparator; @@ -60,19 +61,30 @@ public class DorisStreamLoad implements Closeable { private final CloseableHttpClient httpClient; private final Gson gson = new Gson(); - public DorisStreamLoad(String feNode, String database, String table, String username, + public DorisStreamLoad(List feNodes, String database, String table, String username, String password, String format, String columnSeparator, String lineDelimiter, List columns, int connectTimeoutMs, int readTimeoutMs, int maxRetries) { - this.loadUrl = String.format(DorisConstants.STREAM_LOAD_URL_PATTERN, - normalizeFeNode(feNode), database, table); + this.loadUrls = new ArrayList<>(); + if (feNodes != null) { + for (String feNode : feNodes) { + if (feNode != null && !feNode.trim().isEmpty()) { + this.loadUrls.add(String.format(DorisConstants.STREAM_LOAD_URL_PATTERN, + normalizeFeNode(feNode), database, table)); + } + } + } + if (this.loadUrls.isEmpty()) { + throw new GeaFlowDSLException("Doris fenodes must not be empty."); + } this.authHeader = "Basic " + Base64.getEncoder().encodeToString( (username + ":" + password).getBytes(StandardCharsets.UTF_8)); this.format = format; this.columnSeparator = columnSeparator; this.lineDelimiter = lineDelimiter; this.columns = String.join(DorisConstants.COMMA, columns); - this.maxRetries = Math.max(1, maxRetries); + // Retry at least once per FE so a single FE failure can fail over to another FE. + this.maxRetries = Math.max(Math.max(1, maxRetries), this.loadUrls.size()); RequestConfig requestConfig = RequestConfig.custom() .setConnectTimeout(connectTimeoutMs) .setSocketTimeout(readTimeoutMs) @@ -105,21 +117,23 @@ private static String normalizeFeNode(String feNode) { */ public void load(byte[] payload) { Exception lastError = null; - for (int attempt = 1; attempt <= maxRetries; attempt++) { + for (int attempt = 0; attempt < maxRetries; attempt++) { + // Rotate over the FE list so a failed request fails over to the next FE. + String url = loadUrls.get(attempt % loadUrls.size()); try { - doLoad(payload); + doLoad(url, payload); return; } catch (Exception e) { lastError = e; - LOGGER.warn("Stream Load attempt {}/{} failed: {}", attempt, maxRetries, - e.getMessage()); + LOGGER.warn("Stream Load attempt {}/{} to {} failed: {}", attempt + 1, maxRetries, + url, e.getMessage()); } } throw new GeaFlowDSLException("Doris Stream Load failed after " + maxRetries + " attempts.", lastError); } - private void doLoad(byte[] payload) throws IOException { + private void doLoad(String loadUrl, byte[] payload) throws IOException { HttpPut put = new HttpPut(loadUrl); put.setHeader(HttpHeaders.EXPECT, "100-continue"); put.setHeader(HttpHeaders.AUTHORIZATION, authHeader); @@ -169,7 +183,11 @@ private String generateLabel() { } public String getLoadUrl() { - return loadUrl; + return loadUrls.get(0); + } + + public List getLoadUrls() { + return loadUrls; } @Override diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSink.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSink.java index 1c51d940d..8c34a7e2e 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSink.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSink.java @@ -45,7 +45,7 @@ public class DorisTableSink implements TableSink { private static final Logger LOGGER = LoggerFactory.getLogger(DorisTableSink.class); private StructType schema; - private String feNode; + private String feNodes; private String database; private String table; private String username; @@ -67,11 +67,11 @@ public class DorisTableSink implements TableSink { public void init(Configuration conf, StructType tableSchema) { LOGGER.info("init doris sink with config: {}, \n schema: {}", conf, tableSchema); this.schema = tableSchema; - String feNodes = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_FENODES, ""); - if (feNodes == null || feNodes.trim().isEmpty()) { + String feNodesConf = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_FENODES, ""); + if (feNodesConf == null || feNodesConf.trim().isEmpty()) { throw new GeaFlowDSLException("Doris fenodes must be specified for the sink."); } - this.feNode = feNodes.split(DorisConstants.COMMA)[0].trim(); + this.feNodes = feNodesConf; this.database = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE); this.table = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE); this.username = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_USERNAME); @@ -92,8 +92,15 @@ public void open(RuntimeContext context) { for (TableField field : schema.getFields()) { columns.add(field.getName()); } - this.streamLoad = new DorisStreamLoad(feNode, database, table, username, password, format, - columnSeparator, lineDelimiter, columns, connectTimeoutMs, readTimeoutMs, maxRetries); + List feNodeList = new ArrayList<>(); + for (String node : feNodes.split(DorisConstants.COMMA)) { + if (!node.trim().isEmpty()) { + feNodeList.add(node.trim()); + } + } + this.streamLoad = new DorisStreamLoad(feNodeList, database, table, username, password, + format, columnSeparator, lineDelimiter, columns, connectTimeoutMs, readTimeoutMs, + maxRetries); this.buffer = new ArrayList<>(); this.bufferBytes = 0L; } @@ -119,6 +126,9 @@ private void flushBuffer() { if (buffer == null || buffer.isEmpty()) { return; } + if (streamLoad == null) { + throw new GeaFlowDSLException("doris stream load client is not initialized."); + } String payload; if (DorisConstants.FORMAT_JSON.equalsIgnoreCase(format)) { payload = "[" + String.join(DorisConstants.COMMA, buffer) + "]"; @@ -134,13 +144,19 @@ private void flushBuffer() { @Override public void close() { + // Flush any remaining buffered rows before closing, and stay safe against multiple + // invocations: once flushed the buffer is empty and the client is nulled out. try { - if (streamLoad != null) { - streamLoad.close(); - streamLoad = null; + flushBuffer(); + } finally { + try { + if (streamLoad != null) { + streamLoad.close(); + streamLoad = null; + } + } catch (IOException e) { + throw new GeaFlowDSLException("failed to close doris stream load client.", e); } - } catch (IOException e) { - throw new GeaFlowDSLException("failed to close doris stream load client.", e); } } } diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSource.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSource.java index 506d8bd0b..7cc1c11ef 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSource.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSource.java @@ -88,10 +88,9 @@ public void init(Configuration conf, TableSchema tableSchema) { this.password = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_PASSWORD); this.database = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE, ""); this.table = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE); - this.partitionNum = conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_NUM); - if (partitionNum <= 0) { - throw new GeaFlowDSLException("Invalid doris source partition number: {}", partitionNum); - } + // Clamp the partition number to at least 1 instead of failing on non-positive values. + this.partitionNum = Math.max(1L, + conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_NUM)); this.partitionColumn = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_COLUMN); this.lowerBound = conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_LOWERBOUND); this.upperBound = conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_UPPERBOUND); diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisUtils.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisUtils.java index 28277f622..3d198d9f8 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisUtils.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisUtils.java @@ -20,6 +20,7 @@ package org.apache.geaflow.dsl.connector.doris; import com.google.gson.Gson; +import com.google.gson.GsonBuilder; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -29,11 +30,21 @@ public class DorisUtils { - private static final Gson GSON = new Gson(); + // serializeNulls: keep null columns as an explicit json null instead of dropping the key, so + // Doris loads them as NULL. disableHtmlEscaping: keep the original unicode characters instead + // of turning '<', '>', '&' into \\uXXXX. Gson still correctly escapes '\n', '"' and '\\'. + private static final Gson GSON = new GsonBuilder() + .serializeNulls() + .disableHtmlEscaping() + .create(); /** * Serialize a row to a single csv line using the given column separator. Null fields are - * rendered as Doris's null placeholder ("\N"). + * rendered as Doris's null placeholder ("\N"), while an empty string is kept as an empty + * field. The csv format is a plain separator split without quoting, so the caller must make + * sure the values do not contain the chosen column separator or line delimiter; use the json + * format (the default) when the data may contain such characters, newlines, quotes or + * backslashes. */ public static String rowToCsv(Row row, StructType schema, String columnSeparator) { List fields = schema.getFields(); @@ -53,15 +64,27 @@ public static String rowToCsv(Row row, StructType schema, String columnSeparator } /** - * Serialize a row to a json object string keyed by the column names. + * Serialize a row to a json object string keyed by the column names. Special characters such + * as newlines, double quotes, backslashes and unicode are escaped by Gson, null fields are + * kept as an explicit json null and empty strings are kept as "". */ public static String rowToJson(Row row, StructType schema) { List fields = schema.getFields(); Map map = new LinkedHashMap<>(); for (int i = 0; i < fields.size(); i++) { Object value = row.getField(i, fields.get(i).getType()); - map.put(fields.get(i).getName(), value); + Object normalized = value == null ? null : normalize(value); + map.put(fields.get(i).getName(), normalized); } return GSON.toJson(map); } + + private static Object normalize(Object value) { + // BinaryString and other non-primitive value holders are serialized via their toString so + // that Gson escapes them as plain json strings. + if (value instanceof Number || value instanceof Boolean || value instanceof String) { + return value; + } + return value.toString(); + } } diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeysTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeysTest.java index 0a27164bb..e03fdce6a 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeysTest.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeysTest.java @@ -32,7 +32,7 @@ public void testSinkDefaults() { Assert.assertEquals(conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_BYTES), 10485760L); Assert.assertEquals(conf.getInteger(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_MAX_RETRIES), 3); - Assert.assertEquals(conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_FORMAT), "csv"); + Assert.assertEquals(conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SINK_FORMAT), "json"); Assert.assertEquals(conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_USERNAME), "root"); } diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java index 7f7398484..bb1edce84 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisConnectorIntegrationTest.java @@ -31,6 +31,7 @@ import org.apache.geaflow.common.type.Types; import org.apache.geaflow.dsl.common.data.Row; import org.apache.geaflow.dsl.common.data.impl.ObjectRow; +import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException; import org.apache.geaflow.dsl.common.types.StructType; import org.apache.geaflow.dsl.common.types.TableField; import org.apache.geaflow.dsl.common.types.TableSchema; @@ -263,4 +264,79 @@ public void testStreamLoadThroughputBenchmark() throws Exception { Assert.assertTrue(streamLoadCost < jdbcCost, "Stream Load should be faster than row-by-row JDBC insert."); } + + @Test(expectedExceptions = GeaFlowDSLException.class) + public void testTableNotFound() throws Exception { + Configuration conf = sinkConfig(); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, "no_such_table"); + DorisTableSink sink = new DorisTableSink(); + sink.init(conf, schema()); + sink.open(null); + sink.write(ObjectRow.create(1L, "a", 1)); + sink.finish(); + } + + @Test + public void testFeFailover() throws Exception { + truncate(); + // Put an unreachable FE first, the sink must fail over to the real FE and still succeed. + Configuration conf = sinkConfig(); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_FENODES, "127.0.0.1:1," + feNodes); + DorisTableSink sink = new DorisTableSink(); + sink.init(conf, schema()); + sink.open(null); + for (int i = 0; i < 100; i++) { + sink.write(ObjectRow.create((long) i, "name_" + i, i)); + } + sink.finish(); + sink.close(); + Assert.assertEquals(countRows(), 100); + } + + @Test + public void testDataTypeFidelity() throws Exception { + String typesTable = "types_table"; + try (Connection connection = newConnection(jdbcUrl); + Statement statement = connection.createStatement()) { + statement.execute("DROP TABLE IF EXISTS " + DATABASE + "." + typesTable); + statement.execute("CREATE TABLE " + DATABASE + "." + typesTable + " (" + + "id BIGINT, c_int INT, c_double DOUBLE, c_varchar VARCHAR(256)) " + + "UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES(\"replication_num\" = \"1\")"); + } + StructType typesSchema = new StructType( + new TableField("id", Types.LONG, false), + new TableField("c_int", Types.INTEGER, true), + new TableField("c_double", Types.DOUBLE, true), + new TableField("c_varchar", Types.BINARY_STRING, true)); + + Configuration conf = sinkConfig(); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, typesTable); + String tricky = "a\nb\tc\"d\\e\u4e2d\u6587\ud83d\ude00"; + DorisTableSink sink = new DorisTableSink(); + sink.init(conf, typesSchema); + sink.open(null); + sink.write(ObjectRow.create(1L, 10, 1.5, tricky)); + sink.write(ObjectRow.create(2L, null, null, null)); + sink.finish(); + sink.close(); + + try (Connection connection = newConnection(jdbcUrl); + Statement statement = connection.createStatement(); + ResultSet rs = statement.executeQuery("SELECT id, c_int, c_double, c_varchar FROM " + + DATABASE + "." + typesTable + " ORDER BY id")) { + Assert.assertTrue(rs.next()); + Assert.assertEquals(rs.getLong("id"), 1L); + Assert.assertEquals(rs.getInt("c_int"), 10); + Assert.assertEquals(rs.getDouble("c_double"), 1.5); + Assert.assertEquals(rs.getString("c_varchar"), tricky); + Assert.assertTrue(rs.next()); + Assert.assertEquals(rs.getLong("id"), 2L); + rs.getObject("c_int"); + Assert.assertTrue(rs.wasNull(), "c_int should round-trip as NULL"); + rs.getString("c_varchar"); + Assert.assertTrue(rs.wasNull(), "c_varchar should round-trip as NULL"); + Assert.assertFalse(rs.next()); + } + } } diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoadTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoadTest.java index 37b840861..f5b0897bf 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoadTest.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoadTest.java @@ -21,19 +21,21 @@ import java.io.IOException; import java.util.Arrays; +import java.util.Collections; +import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException; import org.testng.Assert; import org.testng.annotations.Test; public class DorisStreamLoadTest { - private DorisStreamLoad newClient(String feNode) { - return new DorisStreamLoad(feNode, "test_db", "test_table", "root", "", + private DorisStreamLoad newClient(java.util.List feNodes) { + return new DorisStreamLoad(feNodes, "test_db", "test_table", "root", "", DorisConstants.FORMAT_CSV, "\t", "\n", Arrays.asList("id", "name"), 1000, 1000, 3); } @Test public void testLoadUrlWithHostPort() throws IOException { - try (DorisStreamLoad client = newClient("127.0.0.1:8030")) { + try (DorisStreamLoad client = newClient(Collections.singletonList("127.0.0.1:8030"))) { Assert.assertEquals(client.getLoadUrl(), "http://127.0.0.1:8030/api/test_db/test_table/_stream_load"); } @@ -41,9 +43,26 @@ public void testLoadUrlWithHostPort() throws IOException { @Test public void testLoadUrlWithScheme() throws IOException { - try (DorisStreamLoad client = newClient("http://doris-fe:8030")) { + try (DorisStreamLoad client = newClient(Collections.singletonList("http://doris-fe:8030"))) { Assert.assertEquals(client.getLoadUrl(), "http://doris-fe:8030/api/test_db/test_table/_stream_load"); } } + + @Test + public void testMultipleFeNodesForFailover() throws IOException { + try (DorisStreamLoad client = + newClient(Arrays.asList("fe1:8030", "fe2:8030", "fe3:8030"))) { + Assert.assertEquals(client.getLoadUrls().size(), 3); + Assert.assertEquals(client.getLoadUrls().get(1), + "http://fe2:8030/api/test_db/test_table/_stream_load"); + } + } + + @Test(expectedExceptions = GeaFlowDSLException.class) + public void testEmptyFeNodesThrows() throws IOException { + try (DorisStreamLoad client = newClient(Collections.emptyList())) { + Assert.fail("should have thrown for empty fenodes"); + } + } } diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisUtilsTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisUtilsTest.java index 7b9b0dd0a..e004a1af1 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisUtilsTest.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisUtilsTest.java @@ -19,6 +19,8 @@ package org.apache.geaflow.dsl.connector.doris; +import com.google.gson.Gson; +import com.google.gson.JsonObject; import org.apache.geaflow.common.type.Types; import org.apache.geaflow.dsl.common.data.Row; import org.apache.geaflow.dsl.common.data.impl.ObjectRow; @@ -29,6 +31,8 @@ public class DorisUtilsTest { + private static final Gson GSON = new Gson(); + private StructType buildSchema() { return new StructType( new TableField("id", Types.LONG, false), @@ -69,4 +73,40 @@ public void testRowToJson() { Assert.assertTrue(json.contains("\"name\":\"carol\"")); Assert.assertTrue(json.contains("\"score\":8.0")); } + + @Test + public void testRowToJsonEscapesSpecialChars() { + StructType schema = buildSchema(); + // A value with a newline, a tab, a double quote, a backslash and unicode/emoji. + String tricky = "line1\nline2\tcol\"q\\b\u4e2d\u6587\ud83d\ude00"; + Row row = ObjectRow.create(1L, tricky, 1.0); + String json = DorisUtils.rowToJson(row, schema); + // Round-trip: parsing the produced json back must yield the exact original value. + JsonObject parsed = GSON.fromJson(json, JsonObject.class); + Assert.assertEquals(parsed.get("name").getAsString(), tricky); + } + + @Test + public void testRowToJsonKeepsNullAndEmptyString() { + StructType schema = buildSchema(); + Row rowWithNull = ObjectRow.create(2L, null, 2.0); + JsonObject parsedNull = GSON.fromJson(DorisUtils.rowToJson(rowWithNull, schema), + JsonObject.class); + Assert.assertTrue(parsedNull.has("name"), "null column key must be kept"); + Assert.assertTrue(parsedNull.get("name").isJsonNull(), "null column must be json null"); + + Row rowWithEmpty = ObjectRow.create(3L, "", 3.0); + JsonObject parsedEmpty = GSON.fromJson(DorisUtils.rowToJson(rowWithEmpty, schema), + JsonObject.class); + Assert.assertEquals(parsedEmpty.get("name").getAsString(), ""); + } + + @Test + public void testRowToCsvKeepsEmptyString() { + StructType schema = buildSchema(); + Row row = ObjectRow.create(7L, "", 1.0); + String csv = DorisUtils.rowToCsv(row, schema, "\t"); + // Empty string stays as an empty field (not the null placeholder). + Assert.assertEquals(csv, "7\t\t1.0"); + } } From c55ec0eee7aafbaaa10b80284e7b6e39ae4fe0c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=88=9A=E5=AE=8F=E6=88=90?= Date: Thu, 23 Jul 2026 11:54:08 +0800 Subject: [PATCH 4/6] [ISSUE-817] Do not bundle the GPL MySQL driver; minor cleanup - Make mysql-connector-java a test-scope dependency instead of compile, so the GPL-licensed driver is not shipped in the distribution (matching geaflow-dsl-connector-jdbc, which bundles no driver). The source only uses the standard java.sql API and loads the driver by name, so users must provide a MySQL-protocol driver on the runtime classpath; documented in CN + EN docs. - Remove the unused COLON constant. Verified: checkstyle 0 violations, RAT 0 unapproved (14 files), unit tests 23 run / 0 failed / 7 skipped (integration tests are opt-in). --- .../5.application-development/3.connector/11.doris.md | 2 +- .../5.application-development/3.connector/11.doris.md | 2 +- .../geaflow-dsl-connector-doris/pom.xml | 8 +++++++- .../geaflow/dsl/connector/doris/DorisConstants.java | 2 -- 4 files changed, 9 insertions(+), 5 deletions(-) diff --git a/docs/docs-cn/source/5.application-development/3.connector/11.doris.md b/docs/docs-cn/source/5.application-development/3.connector/11.doris.md index b186bc749..8c5fc52af 100644 --- a/docs/docs-cn/source/5.application-development/3.connector/11.doris.md +++ b/docs/docs-cn/source/5.application-development/3.connector/11.doris.md @@ -29,7 +29,7 @@ CREATE TABLE doris_table ( | 参数名 | 是否必须 | 描述 | | ---- | ---- | ---- | | 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 通过它做分区并行读。 | +| 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 数据库名。 | diff --git a/docs/docs-en/source/5.application-development/3.connector/11.doris.md b/docs/docs-en/source/5.application-development/3.connector/11.doris.md index d039a906f..0c3801ff2 100644 --- a/docs/docs-en/source/5.application-development/3.connector/11.doris.md +++ b/docs/docs-en/source/5.application-development/3.connector/11.doris.md @@ -31,7 +31,7 @@ CREATE TABLE doris_table ( | 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. | +| 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. | diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml index 808180751..ef389691d 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml @@ -58,11 +58,17 @@ ${doris.httpclient.version} - + mysql mysql-connector-java ${mysql.version} + test diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConstants.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConstants.java index 063c01e1c..6a43905d3 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConstants.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConstants.java @@ -29,8 +29,6 @@ public class DorisConstants { public static final String COMMA = ","; - public static final String COLON = ":"; - public static final String HTTP_SCHEME = "http://"; /** From ed734788dbea5a6ce40c982949c85f3cbd614848 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=88=9A=E5=AE=8F=E6=88=90?= Date: Thu, 23 Jul 2026 15:40:52 +0800 Subject: [PATCH 5/6] [ISSUE-817] Add real benchmark numbers verified against a local Doris Verified the connector end to end against a real single-node apache/doris:3.0.8-all: the sink (Stream Load), the partitioned source, FE failover, table-not-found and full data-type round-trip (newlines, quotes, backslashes, chinese, emoji) integration tests all pass. Benchmark: Stream Load loaded 500 rows in ~91ms (~5500 rows/s) vs ~11335ms (~44 rows/s) for row-by-row JDBC INSERT, ~125x faster; 2000 rows via Stream Load took only ~97ms. Recorded the reference numbers in the CN + EN docs. --- .../5.application-development/3.connector/11.doris.md | 5 +++++ .../5.application-development/3.connector/11.doris.md | 6 ++++++ 2 files changed, 11 insertions(+) diff --git a/docs/docs-cn/source/5.application-development/3.connector/11.doris.md b/docs/docs-cn/source/5.application-development/3.connector/11.doris.md index 8c5fc52af..536580806 100644 --- a/docs/docs-cn/source/5.application-development/3.connector/11.doris.md +++ b/docs/docs-cn/source/5.application-development/3.connector/11.doris.md @@ -99,6 +99,11 @@ SELECT * FROM doris_source_table; (通常快一个数量级以上，具体倍数取决于批大小、网络与集群规模)。该基准会断言 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 实例。 diff --git a/docs/docs-en/source/5.application-development/3.connector/11.doris.md b/docs/docs-en/source/5.application-development/3.connector/11.doris.md index 0c3801ff2..4b7196d00 100644 --- a/docs/docs-en/source/5.application-development/3.connector/11.doris.md +++ b/docs/docs-en/source/5.application-development/3.connector/11.doris.md @@ -103,6 +103,12 @@ same table with batched Stream Load takes clearly less time than row-by-row JDBC 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 From 3a838efc933b477972594562a893f5a54f7bca07 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=88=9A=E5=AE=8F=E6=88=90?= Date: Fri, 24 Jul 2026 10:18:05 +0800 Subject: [PATCH 6/6] [ISSUE-817] Address PR review: idempotent label, partition strategies, timeouts - Label/idempotency: generate one label per batch and reuse it across all retries and FE failover attempts (Doris loads a label at most once); treat 'Label Already Exists' on retry as success to avoid duplicate writes; log the label on failures/retries. Documented the write semantics. - Partition strategies: add source.partition.mode (range/custom). 'custom' takes semicolon-separated WHERE predicates (source.partition.clauses), one partition per predicate, supporting non-numeric/skewed columns and arbitrary conditions. Documented the performance impact of unindexed/skewed partition columns and recommended indexed, evenly-distributed columns. - Timeouts: read/write timeout is configurable and its default is raised to 60s to cover large-batch uploads; documented tuning guidance. - Tests: add DorisTableSourceTest covering single/range/custom partition listing. --- .../3.connector/11.doris.md | 21 ++++- .../3.connector/11.doris.md | 22 ++++- .../dsl/connector/doris/DorisConfigKeys.java | 28 +++++- .../dsl/connector/doris/DorisConstants.java | 13 +++ .../dsl/connector/doris/DorisStreamLoad.java | 25 +++-- .../dsl/connector/doris/DorisTableSource.java | 44 ++++++++- .../connector/doris/DorisTableSourceTest.java | 92 +++++++++++++++++++ 7 files changed, 222 insertions(+), 23 deletions(-) create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSourceTest.java diff --git a/docs/docs-cn/source/5.application-development/3.connector/11.doris.md b/docs/docs-cn/source/5.application-development/3.connector/11.doris.md index 536580806..934d2ebdb 100644 --- a/docs/docs-cn/source/5.application-development/3.connector/11.doris.md +++ b/docs/docs-cn/source/5.application-development/3.connector/11.doris.md @@ -41,9 +41,11 @@ CREATE TABLE doris_table ( | 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 的读超时(毫秒)，默认 30000。 | -| geaflow.dsl.doris.source.partition.num | 否 | Source 并行读的分区数，默认 1。 | -| geaflow.dsl.doris.source.partition.column | 否 | 用于切分分区的数值列，默认 id。 | +| 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 | 否 | 分区列上界，只用于决定分区步长，不用于过滤数据。 | @@ -53,8 +55,17 @@ 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 完成导入。默认使用 CSV 负载以获得更高吞吐， -也支持 JSON 负载。 +客户端会自动跟随重定向并把负载重新发送给 BE 完成导入。默认使用 JSON 负载(能安全处理特殊字符)， +也支持 CSV 负载。 + +## 写入语义与幂等 + +每一批数据在提交时会生成一个唯一的 label(基于 UUID)，并在该批次的所有重试与 FE 故障切换中 +复用同一个 label。Doris 保证同一个 label 至多导入一次：如果某次尝试实际已提交、但客户端因网络 +异常误判为失败，重试会命中 "Label Already Exists"，连接器将其视为成功，从而避免重复写入。 +因此在 FE failover + 重试场景下，写入是幂等的。重试次数可通过 +`geaflow.dsl.doris.sink.max.retries` 配置(为支持 failover，实际重试次数至少等于 FE 个数)， +失败与重试日志中都会打印 label 以便排查。 ## 示例 diff --git a/docs/docs-en/source/5.application-development/3.connector/11.doris.md b/docs/docs-en/source/5.application-development/3.connector/11.doris.md index 4b7196d00..34706e727 100644 --- a/docs/docs-en/source/5.application-development/3.connector/11.doris.md +++ b/docs/docs-en/source/5.application-development/3.connector/11.doris.md @@ -43,9 +43,11 @@ CREATE TABLE doris_table ( | 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 timeout in milliseconds for Stream Load, default 30000. | -| geaflow.dsl.doris.source.partition.num | false | The source partition number for parallel reads, default 1. | -| geaflow.dsl.doris.source.partition.column | false | The numeric column used to split the source into partitions, default id. | +| 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. | @@ -56,8 +58,18 @@ The sink buffers rows in `write()`. When the buffered row count 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. CSV payload is used by default for higher -throughput; JSON payload is also supported. +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 diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeys.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeys.java index affc2fe17..566fddcae 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeys.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConfigKeys.java @@ -98,18 +98,38 @@ public class DorisConfigKeys { public static final ConfigKey GEAFLOW_DSL_DORIS_REQUEST_READ_TIMEOUT_MS = ConfigKeys .key("geaflow.dsl.doris.request.read.timeout.ms") - .defaultValue(30000) - .description("The socket read timeout in milliseconds for Stream Load. Default 30000."); + .defaultValue(60000) + .description("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, and lower it to fail faster on network anomalies."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_SOURCE_PARTITION_MODE = ConfigKeys + .key("geaflow.dsl.doris.source.partition.mode") + .defaultValue("range") + .description("The source partitioning strategy: 'range' splits a numeric partition column " + + "into evenly-sized ranges; 'custom' uses the user-provided predicates in " + + "geaflow.dsl.doris.source.partition.clauses, which supports non-numeric/skewed " + + "columns and arbitrary conditions. Default range."); + + public static final ConfigKey GEAFLOW_DSL_DORIS_SOURCE_PARTITION_CLAUSES = ConfigKeys + .key("geaflow.dsl.doris.source.partition.clauses") + .defaultValue("") + .description("Semicolon-separated WHERE predicates for the 'custom' partition mode, one " + + "partition per predicate, e.g. \"dt='2024-01-01';dt='2024-01-02'\". The predicates " + + "should be disjoint and jointly cover the data. Empty means a single partition."); public static final ConfigKey GEAFLOW_DSL_DORIS_SOURCE_PARTITION_NUM = ConfigKeys .key("geaflow.dsl.doris.source.partition.num") .defaultValue(1L) - .description("The source partition number for parallel reads. Default 1."); + .description("The source partition number for parallel reads in 'range' mode. Default 1. " + + "For balanced partitions the partition column should be numeric and indexed; " + + "splitting on an unindexed or highly-skewed column may cause full scans or hotspots."); public static final ConfigKey GEAFLOW_DSL_DORIS_SOURCE_PARTITION_COLUMN = ConfigKeys .key("geaflow.dsl.doris.source.partition.column") .defaultValue("id") - .description("The numeric column used to split the source into partitions."); + .description("The numeric column used to split the source into partitions in 'range' mode. " + + "Prefer an indexed, evenly-distributed column such as the primary key."); public static final ConfigKey GEAFLOW_DSL_DORIS_SOURCE_PARTITION_LOWERBOUND = ConfigKeys .key("geaflow.dsl.doris.source.partition.lowerbound") diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConstants.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConstants.java index 6a43905d3..aad129446 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConstants.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisConstants.java @@ -43,4 +43,17 @@ public class DorisConstants { public static final String STREAM_LOAD_SUCCESS = "Success"; public static final String STREAM_LOAD_PUBLISH_TIMEOUT = "Publish Timeout"; + + /** + * Returned when a load with the same label was already accepted. Because a batch reuses one + * label across all its retries, seeing this on a retry means our own earlier attempt already + * succeeded, so it is treated as success to keep retries idempotent. + */ + public static final String STREAM_LOAD_LABEL_ALREADY_EXISTS = "Label Already Exists"; + + public static final String SEMICOLON = ";"; + + public static final String PARTITION_MODE_RANGE = "range"; + + public static final String PARTITION_MODE_CUSTOM = "custom"; } diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoad.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoad.java index 9e049bb7c..d82eb8f6b 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoad.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisStreamLoad.java @@ -116,29 +116,35 @@ private static String normalizeFeNode(String feNode) { * transient failures and throws a {@link GeaFlowDSLException} once all retries are exhausted. */ public void load(byte[] payload) { + // Generate one stable label per batch and reuse it across every retry and FE failover + // attempt. Doris loads a given label at most once, so reusing the label makes retries + // idempotent: if an earlier attempt actually committed but the client saw a network + // error, the retry hits "Label Already Exists" (handled as success) instead of writing + // the batch twice. + String label = generateLabel(); Exception lastError = null; for (int attempt = 0; attempt < maxRetries; attempt++) { // Rotate over the FE list so a failed request fails over to the next FE. String url = loadUrls.get(attempt % loadUrls.size()); try { - doLoad(url, payload); + doLoad(url, payload, label); return; } catch (Exception e) { lastError = e; - LOGGER.warn("Stream Load attempt {}/{} to {} failed: {}", attempt + 1, maxRetries, - url, e.getMessage()); + LOGGER.warn("Stream Load attempt {}/{} to {} with label {} failed: {}", + attempt + 1, maxRetries, url, label, e.getMessage()); } } throw new GeaFlowDSLException("Doris Stream Load failed after " + maxRetries - + " attempts.", lastError); + + " attempts with label " + label + ".", lastError); } - private void doLoad(String loadUrl, byte[] payload) throws IOException { + private void doLoad(String loadUrl, byte[] payload, String label) throws IOException { HttpPut put = new HttpPut(loadUrl); put.setHeader(HttpHeaders.EXPECT, "100-continue"); put.setHeader(HttpHeaders.AUTHORIZATION, authHeader); put.setHeader("format", format); - put.setHeader("label", generateLabel()); + put.setHeader("label", label); put.setHeader("two_phase_commit", "false"); if (DorisConstants.FORMAT_CSV.equalsIgnoreCase(format)) { put.setHeader("column_separator", columnSeparator); @@ -169,6 +175,13 @@ private void checkLoadResult(String body) throws IOException { throw new IOException("Stream Load response without status: " + body); } String status = result.get(DorisConstants.STREAM_LOAD_RESULT_STATUS).getAsString(); + if (DorisConstants.STREAM_LOAD_LABEL_ALREADY_EXISTS.equals(status)) { + // A retry of this batch reused the same label, so the batch was already accepted by an + // earlier attempt. Treat it as success to avoid duplicate writes. + LOGGER.info("Stream Load label already exists, treated as an idempotent retry: {}", + body); + return; + } if (!DorisConstants.STREAM_LOAD_SUCCESS.equals(status) && !DorisConstants.STREAM_LOAD_PUBLISH_TIMEOUT.equals(status)) { String message = result.has(DorisConstants.STREAM_LOAD_RESULT_MESSAGE) diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSource.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSource.java index 7cc1c11ef..a8a07943c 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSource.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/main/java/org/apache/geaflow/dsl/connector/doris/DorisTableSource.java @@ -53,8 +53,11 @@ /** * A Doris table source that reads data through the MySQL protocol exposed by the Doris FE. The - * data set can be split into several partitions on a numeric column so that partitions are read - * in parallel by different tasks. + * data set is split into several partitions read in parallel by different tasks. Two partitioning + * strategies are supported: 'range' splits an evenly-distributed numeric column into ranges, and + * 'custom' uses user-provided WHERE predicates for non-numeric/skewed columns or arbitrary + * conditions. For 'range' the partition column should be indexed and evenly distributed, otherwise + * partitioning may cause full scans or hotspots. */ public class DorisTableSource implements TableSource { @@ -72,6 +75,8 @@ public class DorisTableSource implements TableSource { private String partitionColumn; private long lowerBound; private long upperBound; + private String partitionMode; + private String partitionClauses; private Map partitionConnectionMap = new HashMap<>(); private Map partitionStatementMap = new HashMap<>(); @@ -94,7 +99,11 @@ public void init(Configuration conf, TableSchema tableSchema) { this.partitionColumn = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_COLUMN); this.lowerBound = conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_LOWERBOUND); this.upperBound = conf.getLong(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_UPPERBOUND); - if (partitionNum > 1 && lowerBound >= upperBound) { + this.partitionMode = conf.getString(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_MODE); + this.partitionClauses = conf.getString( + DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_CLAUSES, ""); + if (DorisConstants.PARTITION_MODE_RANGE.equalsIgnoreCase(partitionMode) + && partitionNum > 1 && lowerBound >= upperBound) { throw new GeaFlowDSLException("Upperbound must be greater than lowerbound " + "(lowerbound:%d upperbound:%d).", lowerBound, upperBound); } @@ -115,6 +124,35 @@ private String qualifiedTable() { @Override public List listPartitions() { + if (DorisConstants.PARTITION_MODE_CUSTOM.equalsIgnoreCase(partitionMode)) { + return listCustomPartitions(); + } + return listRangePartitions(); + } + + /** + * Custom mode: one partition per user-provided WHERE predicate. This supports non-numeric or + * skewed partition columns and arbitrary conditions. An empty clause list falls back to a + * single full-table partition. + */ + private List listCustomPartitions() { + if (partitionClauses == null || partitionClauses.trim().isEmpty()) { + return Collections.singletonList(new DorisPartition(qualifiedTable(), "")); + } + List partitions = new ArrayList<>(); + for (String clause : partitionClauses.split(DorisConstants.SEMICOLON)) { + String predicate = clause.trim(); + if (!predicate.isEmpty()) { + partitions.add(new DorisPartition(qualifiedTable(), "WHERE " + predicate)); + } + } + if (partitions.isEmpty()) { + return Collections.singletonList(new DorisPartition(qualifiedTable(), "")); + } + return partitions; + } + + private List listRangePartitions() { if (partitionNum == 1) { return Collections.singletonList(new DorisPartition(qualifiedTable(), "")); } diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSourceTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSourceTest.java new file mode 100644 index 000000000..eff52a8f4 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/src/test/java/org/apache/geaflow/dsl/connector/doris/DorisTableSourceTest.java @@ -0,0 +1,92 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.connector.doris; + +import java.util.List; +import org.apache.geaflow.common.config.Configuration; +import org.apache.geaflow.common.type.Types; +import org.apache.geaflow.dsl.common.types.StructType; +import org.apache.geaflow.dsl.common.types.TableField; +import org.apache.geaflow.dsl.common.types.TableSchema; +import org.apache.geaflow.dsl.connector.api.Partition; +import org.apache.geaflow.dsl.connector.doris.DorisTableSource.DorisPartition; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class DorisTableSourceTest { + + private TableSchema schema() { + return new TableSchema(new StructType( + new TableField("id", Types.LONG, false), + new TableField("name", Types.BINARY_STRING, true))); + } + + private DorisTableSource newSource(Configuration conf) { + Configuration base = conf; + base.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_JDBC_URL, "jdbc:mysql://127.0.0.1:9030/db"); + base.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_DATABASE, "db"); + base.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_TABLE, "t"); + DorisTableSource source = new DorisTableSource(); + source.init(base, schema()); + return source; + } + + @Test + public void testSinglePartition() { + List partitions = newSource(new Configuration()).listPartitions(); + Assert.assertEquals(partitions.size(), 1); + Assert.assertEquals(((DorisPartition) partitions.get(0)).getWhereClause(), ""); + } + + @Test + public void testRangePartitions() { + Configuration conf = new Configuration(); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_NUM, "4"); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_COLUMN, "id"); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_LOWERBOUND, "0"); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_UPPERBOUND, "100"); + List partitions = newSource(conf).listPartitions(); + Assert.assertEquals(partitions.size(), 4); + // The first partition also covers rows whose partition column is entirely NULL. + Assert.assertTrue(((DorisPartition) partitions.get(0)).getWhereClause() + .contains("id IS NULL")); + } + + @Test + public void testCustomPartitions() { + Configuration conf = new Configuration(); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_MODE, "custom"); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_CLAUSES, + "dt='2024-01-01';dt='2024-01-02';dt='2024-01-03'"); + List partitions = newSource(conf).listPartitions(); + Assert.assertEquals(partitions.size(), 3); + Assert.assertEquals(((DorisPartition) partitions.get(1)).getWhereClause(), + "WHERE dt='2024-01-02'"); + } + + @Test + public void testCustomPartitionsEmptyFallsBackToSingle() { + Configuration conf = new Configuration(); + conf.put(DorisConfigKeys.GEAFLOW_DSL_DORIS_SOURCE_PARTITION_MODE, "custom"); + List partitions = newSource(conf).listPartitions(); + Assert.assertEquals(partitions.size(), 1); + Assert.assertEquals(((DorisPartition) partitions.get(0)).getWhereClause(), ""); + } +}