arg_types
+
+ // Return type for the function.
+ 4: required TTypeDesc ret_type
+
+ // If true, this function takes var args.
+ 5: required bool has_var_args
+
+ // Optional comment to attach to the function
+ 6: optional string comment
+
+ 7: optional string signature
+
+ // HDFS path for the function binary. This binary must exist at the time the
+ // function is created.
+ 8: optional string hdfs_location
+
+ // One of these should be set.
+ 9: optional TScalarFunction scalar_fn
+ 10: optional TAggregateFunction aggregate_fn
+
+ 11: optional i64 id
+ 12: optional string checksum
+}
+
+enum TLoadJobState {
+ PENDING,
+ ETL,
+ LOADING,
+ FINISHED,
+ CANCELLED
+}
+
+enum TEtlState {
+ RUNNING,
+ FINISHED,
+ CANCELLED,
+ UNKNOWN
+}
+
+enum TTableType {
+ MYSQL_TABLE,
+ OLAP_TABLE,
+ SCHEMA_TABLE,
+ KUDU_TABLE, // Deprecated
+ BROKER_TABLE,
+ ES_TABLE
+}
+
+enum TKeysType {
+ PRIMARY_KEYS,
+ DUP_KEYS,
+ UNIQUE_KEYS,
+ AGG_KEYS
+}
+
+enum TPriority {
+ NORMAL,
+ HIGH
+}
+
+struct TBackend {
+ 1: required string host
+ 2: required TPort be_port
+ 3: required TPort http_port
+}
+
+struct TResourceInfo {
+ 1: required string user
+ 2: required string group
+}
+
+enum TExportState {
+ RUNNING,
+ FINISHED,
+ CANCELLED,
+ UNKNOWN
+}
+
+enum TFileType {
+ FILE_LOCAL,
+ FILE_BROKER,
+ FILE_STREAM, // file content is streaming in the buffer
+}
+
+struct TTabletCommitInfo {
+ 1: required i64 tabletId
+ 2: required i64 backendId
+}
+
+enum TLoadType {
+ MANUL_LOAD,
+ ROUTINE_LOAD,
+ MINI_LOAD
+}
+
+enum TLoadSourceType {
+ RAW,
+ KAFKA,
+}
diff --git a/bitsail-connectors/connector-doris/src/test/java/com/bytedance/bitsail/connector/doris/source/DorisSourceITCase.java b/bitsail-connectors/connector-doris/src/test/java/com/bytedance/bitsail/connector/doris/source/DorisSourceITCase.java
new file mode 100644
index 000000000..2418d8178
--- /dev/null
+++ b/bitsail-connectors/connector-doris/src/test/java/com/bytedance/bitsail/connector/doris/source/DorisSourceITCase.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2022 Bytedance Ltd. and/or its affiliates.
+ *
+ * Licensed 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 com.bytedance.bitsail.connector.doris.source;
+
+import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
+import com.bytedance.bitsail.connector.doris.option.DorisReaderOptions;
+import com.bytedance.bitsail.test.connector.test.EmbeddedFlinkCluster;
+import com.bytedance.bitsail.test.connector.test.utils.JobConfUtils;
+
+import org.junit.Ignore;
+import org.junit.Test;
+
+@Ignore
+public class DorisSourceITCase {
+
+ @Test
+ public void test() throws Exception {
+ BitSailConfiguration jobConf = JobConfUtils.fromClasspath("doris_to_print.json");
+ addDorisInfo(jobConf);
+ EmbeddedFlinkCluster.submitJob(jobConf);
+ }
+
+ /**
+ * Add your doris setting to job configuration.
+ * Below codes are just example.
+ *
+ * select id, bigint_type, string_type, double_type from doris_table where id = 1
+ */
+ public void addDorisInfo(BitSailConfiguration jobConf) {
+ jobConf.set(DorisReaderOptions.FE_HOSTS, "127.0.0.1:8030");
+ jobConf.set(DorisReaderOptions.MYSQL_HOSTS, "127.0.0.1:9030");
+ jobConf.set(DorisReaderOptions.USER, "root");
+ jobConf.set(DorisReaderOptions.PASSWORD, "");
+ jobConf.set(DorisReaderOptions.DB_NAME, "test");
+ jobConf.set(DorisReaderOptions.TABLE_NAME, "doris_table");
+ jobConf.set(DorisReaderOptions.SQL_FILTER, "id=1");
+ }
+
+}
diff --git a/bitsail-dist/src/main/resources/examples/Doris_Print_Example.json b/bitsail-dist/src/main/resources/examples/Doris_Print_Example.json
new file mode 100644
index 000000000..e69de29bb
diff --git a/bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/resources/doris_to_print.json b/bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/resources/doris_to_print.json
new file mode 100644
index 000000000..e69de29bb
diff --git a/website/en/documents/connectors/doris/doris-example.md b/website/en/documents/connectors/doris/doris-example.md
index 0f1e42de3..05be8a790 100644
--- a/website/en/documents/connectors/doris/doris-example.md
+++ b/website/en/documents/connectors/doris/doris-example.md
@@ -42,6 +42,41 @@ PROPERTIES
)
```
+## Doris Reader
+Assuming there is a test doris cluster, then we can use the following configuration to read `test_doris_table` table.
+```json
+{
+ "job": {
+ "reader": {
+ "class": "com.bytedance.bitsail.connector.doris.source.DorisSource",
+ "fe_hosts": "127.0.0.1:1234",
+ "mysql_hosts": "127.0.0.1:4321",
+ "user": "test_user",
+ "password": "1234567",
+ "db_name": "test_db",
+ "table_name": "test_doris_table",
+ "columns": [
+ {
+ "name": "id",
+ "type": "bigint"
+ },
+ {
+ "name": "bigint_type",
+ "type": "bigint"
+ },
+ {
+ "name": "string_type",
+ "type": "string"
+ },
+ {
+ "name": "double_type",
+ "type": "double"
+ }
+ ]
+ }
+ }
+}
+```
## Doris writer
diff --git a/website/en/documents/connectors/doris/doris.md b/website/en/documents/connectors/doris/doris.md
index 324a4e511..2da6a795b 100644
--- a/website/en/documents/connectors/doris/doris.md
+++ b/website/en/documents/connectors/doris/doris.md
@@ -17,6 +17,67 @@ Parent document: [Connectors](../README.md)
```
+## Doris Reader
+### Supported data types
+Doris read connector parses according to the data segment mapping and supports the following data types:
+
+- CHAR
+- VARCHAR
+- BOOLEAN
+- BINARY
+- VARBINARY
+- INT
+- TINYINT
+- SMALLINT
+- INTEGER
+- BIGINT
+- FLOAT
+- DOUBLE
+
+### Parameters
+The following mentioned parameters should be added to `job.reader` block when using, for example:
+```json
+{
+ "job": {
+ "writer": {
+ "class": "com.bytedance.bitsail.connector.doris.source.DorisSource",
+ "fe_hosts": "127.0.0.1:8030",
+ "mysql_hosts": "127.0.0.1:9030",
+ "user": "root",
+ "password": "",
+ "db_name": "test",
+ "table_name": "test_doris_table"
+ }
+ }
+}
+```
+
+#### Necessary parameters
+| Param name | Required | Default Value | Description |
+|:----------------|:---------|:--------------|:------------------------------------------------------------------------------------|
+| class | yes | -- | Doris writer class name, `com.bytedance.bitsail.connector.doris.source.DorisSource` |
+| fe_hosts | yes | -- | Doris FE address, multi addresses separated by comma |
+| mysql_hosts | yes | -- | Doris jdbc query address , multi addresses separated by comma |
+| user | yes | -- | Doris account user |
+| password | yes | -- | Doris account password, can be empty |
+| db_name | yes | -- | database to read |
+| table_name | yes | -- | table to read |
+| columns | yes | -- | The name and type of columns to read |
+
+#### Optional parameters
+| Param name | Required | Default Value | Description |
+|:------------------------------|:---------|:------------------|:----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| reader_parallelism_num | no | 1 | reader parallelism |
+| sql_filter | no | -- | Column value conditions that need to be queried and filtered |
+| tablet_size | no | Integer.MAX_VALUE | The number of Doris Tablets corresponding to an Partition. The smaller this value is set, the more partitions will be generated. This will increase the parallelism on the bitSail side, but at the same time will cause greater pressure on Doris. |
+| exec_mem_limit | no | 2147483648 | Memory limit for a single query. The default is 2GB, in bytes. |
+| request_query_timeout_s | no | 3600 | Query the timeout time of doris, the default is 1 hour, -1 means no timeout limit |
+| request_batch_size | no | 1024 | The maximum number of rows to read data from BE at one time. Increasing this value can reduce the number of connections between bitSail and Doris. Thereby reducing the extra time overhead caused by network delay. |
+| request_connect_timeouts | no | 30 * 1000 | Connection timeout for sending requests to Doris |
+| request_read_timeouts | no | 30 * 1000 | Read timeout for sending request to Doris |
+| request_retries | no | 3 | Number of retries to send requests to Doris |
+
+
## Doris Writer
### Supported data type
diff --git a/website/en/documents/start/env_setup.md b/website/en/documents/start/env_setup.md
index 0d3f51c53..21d1e3463 100644
--- a/website/en/documents/start/env_setup.md
+++ b/website/en/documents/start/env_setup.md
@@ -15,6 +15,34 @@ English | [简体中文](../../../zh/documents/start/env_setup.md)
- JDK1.8
- maven 3.6+
- [Docker desktop](https://www.docker.com/products/docker-desktop/)
+- thrift
+```bash
+install thrift
+ Windows:
+ 1. Download: `http://archive.apache.org/dist/thrift/0.13.0/thrift-0.13.0.exe`
+ 2. Modify thrift-0.13.0.exe to thrift
+
+ MacOS:
+ 1. Download: `brew install thrift@0.13.0`
+ 2. default address: /opt/homebrew/Cellar/thrift@0.13.0/0.13.0/bin/thrift
+
+ Note: Executing `brew install thrift@0.13.0` on MacOS may report an error that the version cannot be found. The solution is as follows, execute it in the terminal:
+ 1. `brew tap-new $USER/local-tap`
+ 2. `brew extract --version='0.13.0' thrift $USER/local-tap`
+ 3. `brew install thrift@0.13.0`
+ Reference link: `https://gist.github.com/tonydeng/02e571f273d6cce4230dc8d5f394493c`
+
+ Linux:
+ 1.Download source package：`wget https://archive.apache.org/dist/thrift/0.13.0/thrift-0.13.0.tar.gz`
+ 2.Install dependencies：`yum install -y autoconf automake libtool cmake ncurses-devel openssl-devel lzo-devel zlib-devel gcc gcc-c++`
+ 3.`tar zxvf thrift-0.13.0.tar.gz`
+ 4.`cd thrift-0.13.0`
+ 5.`./configure --without-tests`
+ 6.`make`
+ 7.`make install`
+ Check the version after installation is complete：thrift --version
+ Note: If you have compiled Doris, you do not need to install thrift, you can directly use $DORIS_HOME/thirdparty/installed/bin/thrift
+```
After correctly installing the above required components, we are able to run integration tests on your local IDE.
diff --git a/website/zh/documents/connectors/doris/doris-example.md b/website/zh/documents/connectors/doris/doris-example.md
index b4b091263..6bef28ef1 100644
--- a/website/zh/documents/connectors/doris/doris-example.md
+++ b/website/zh/documents/connectors/doris/doris-example.md
@@ -42,6 +42,41 @@ PROPERTIES
)
```
+## Doris读连接器
+假设当前有一个测试doris集群，则可以通过如下配置读取`test_doris_table`表。
+```json
+{
+ "job": {
+ "reader": {
+ "class": "com.bytedance.bitsail.connector.doris.source.DorisSource",
+ "fe_hosts": "127.0.0.1:1234",
+ "mysql_hosts": "127.0.0.1:4321",
+ "user": "test_user",
+ "password": "1234567",
+ "db_name": "test_db",
+ "table_name": "test_doris_table",
+ "columns": [
+ {
+ "name": "id",
+ "type": "bigint"
+ },
+ {
+ "name": "bigint_type",
+ "type": "bigint"
+ },
+ {
+ "name": "string_type",
+ "type": "string"
+ },
+ {
+ "name": "double_type",
+ "type": "double"
+ }
+ ]
+ }
+ }
+}
+```
## Doris写连接器
diff --git a/website/zh/documents/connectors/doris/doris.md b/website/zh/documents/connectors/doris/doris.md
index 3aa9a1055..b7178a8c2 100644
--- a/website/zh/documents/connectors/doris/doris.md
+++ b/website/zh/documents/connectors/doris/doris.md
@@ -19,6 +19,67 @@
```
+## Doris读取
+### 支持的数据类型
+Doris读连接器根据字段映射进行解析，支持以下数据类型:
+
+- CHAR
+- VARCHAR
+- BOOLEAN
+- BINARY
+- VARBINARY
+- INT
+- TINYINT
+- SMALLINT
+- INTEGER
+- BIGINT
+- FLOAT
+- DOUBLE
+
+### 主要参数
+写连接器参数在`job.reader`中配置，实际使用时请注意路径前缀。示例:
+```json
+{
+ "job": {
+ "writer": {
+ "class": "com.bytedance.bitsail.connector.doris.source.DorisSource",
+ "fe_hosts": "127.0.0.1:8030",
+ "mysql_hosts": "127.0.0.1:9030",
+ "user": "root",
+ "password": "",
+ "db_name": "test",
+ "table_name": "test_doris_table"
+ }
+ }
+}
+```
+#### 必需参数
+| 参数名称 | 是否必填 | 默认值 | 参数含义 |
+|:------------|:-------|:----|:------------------------------------------------------------------------|
+| class | 是 | -- | Doris读连接器类型, `com.bytedance.bitsail.connector.doris.source.DorisSource` |
+| fe_hosts | 是 | -- | Doris FE地址, 多个地址用逗号分隔 |
+| mysql_hosts | 是 | -- | JDBC连接Doris的地址, 多个地址用逗号分隔 |
+| user | 是 | -- | Doris账户名 |
+| password | 是 | -- | Doris密码，可为空 |
+| db_name | 是 | -- | 要读取的Doris库 |
+| table_name | 是 | -- | 要读取的Doris表 |
+| columns | 是 | -- | 要读取的数据列的列名和类型 |
+
+#### 可选参数
+| 参数名称 | 是否必填 | 默认值 | 参数含义 |
+|:------------------------------|:-----|:------------------|:----------------------------------------------------------------------------------------------------|
+| reader_parallelism_num | 否 | 1 | 指定doris读并发 |
+| sql_filter | 否 | -- | 需要查询过滤的列值条件 |
+| tablet_size | 否 | Integer.MAX_VALUE | 一个 Partition 对应的 Doris Tablet 个数。 此数值设置越小，则会生成越多的 Partition。从而提升 bitSail 侧的并行度，但同时会对 Doris 造成更大的压力。 |
+| exec_mem_limit | 否 | 2147483648 | 单个查询的内存限制。默认为 2GB，单位为字节 |
+| request_query_timeout_s | 否 | 3600 | 查询 Doris 的超时时间，默认值为1小时，-1表示无超时限制 |
+| request_batch_size | 否 | 1024 | 一次从 Doris BE 读取数据的最大行数。增大此数值可减少 bitSail 与 Doris 之间建立连接的次数。 从而减轻网络延迟所带来的额外时间开销 |
+| request_connect_timeouts | 否 | 30 * 1000 | 向 Doris 发送请求的连接超时时间 |
+| request_read_timeouts | 否 | 30 * 1000 | 向 Doris 发送请求的读取超时时间 |
+| request_retries | 否 | 3 | 向 Doris 发送请求的重试次数 |
+
+
+
## Doris写入
### 支持的数据类型
diff --git a/website/zh/documents/start/env_setup.md b/website/zh/documents/start/env_setup.md
index c187aabf4..3e95c0503 100644
--- a/website/zh/documents/start/env_setup.md
+++ b/website/zh/documents/start/env_setup.md
@@ -15,6 +15,34 @@ order: 2
- JDK1.8
- maven 3.6+
- [Docker desktop](https://www.docker.com/products/docker-desktop/)
+- thrift
+```bash
+安装 thrift
+ Windows:
+ 1.下载：`http://archive.apache.org/dist/thrift/0.13.0/thrift-0.13.0.exe`(下载目录自己指定)
+ 2.修改thrift-0.13.0.exe 为 thrift
+
+ MacOS:
+ 1. 下载：`brew install thrift@0.13.0`
+ 2. 默认下载地址：/opt/homebrew/Cellar/thrift@0.13.0/0.13.0/bin/thrift
+
+ 注：MacOS执行 `brew install thrift@0.13.0` 可能会报找不到版本的错误，解决方法如下，在终端执行：
+ 1. `brew tap-new $USER/local-tap`
+ 2. `brew extract --version='0.13.0' thrift $USER/local-tap`
+ 3. `brew install thrift@0.13.0`
+ 参考链接: `https://gist.github.com/tonydeng/02e571f273d6cce4230dc8d5f394493c`
+
+ Linux:
+ 1.下载源码包：`wget https://archive.apache.org/dist/thrift/0.13.0/thrift-0.13.0.tar.gz`
+ 2.安装依赖：`yum install -y autoconf automake libtool cmake ncurses-devel openssl-devel lzo-devel zlib-devel gcc gcc-c++`
+ 3.`tar zxvf thrift-0.13.0.tar.gz`
+ 4.`cd thrift-0.13.0`
+ 5.`./configure --without-tests`
+ 6.`make`
+ 7.`make install`
+ 安装完成后查看版本：thrift --version
+ 注：如果编译过Doris，则不需要安装thrift,可以直接使用 $DORIS_HOME/thirdparty/installed/bin/thrift
+```
在安装上述必需组件后，您可以在本地的IDE上直接运行已有的集成测试。
From 98f25a7e6c84dc8879e042f66f1dc517a9350da6 Mon Sep 17 00:00:00 2001
From: DongLiang-0 <1747644936@qq.com>
Date: Wed, 1 Feb 2023 20:16:52 +0800
Subject: [PATCH 2/3] fix doc
---
README.md | 2 +-
README_zh.md | 2 +-
website/en/documents/introduce.md | 2 +-
website/zh/documents/introduce.md | 2 +-
4 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/README.md b/README.md
index 4697cb9da..df6de81d9 100644
--- a/README.md
+++ b/README.md
@@ -112,7 +112,7 @@ In the Runtime layer, it supports multiple execution modes, such as yarn, local,
| Doris |
- |
- |
+ ✅ |
✅ |
diff --git a/README_zh.md b/README_zh.md
index 99e9f5d95..ce906e1a5 100644
--- a/README_zh.md
+++ b/README_zh.md
@@ -96,7 +96,7 @@ BitSail目前已被广泛使用,并支持数百万亿的大流量场景。同时
| Doris |
- |
- |
+ ✅ |
✅ |
diff --git a/website/en/documents/introduce.md b/website/en/documents/introduce.md
index c289fafe6..9543cb766 100644
--- a/website/en/documents/introduce.md
+++ b/website/en/documents/introduce.md
@@ -120,7 +120,7 @@ In the Runtime layer, it supports multiple execution modes, such as yarn, local,
| Doris |
- |
- |
+ ✅ |
✅ |
diff --git a/website/zh/documents/introduce.md b/website/zh/documents/introduce.md
index 984ab82c5..887d73476 100644
--- a/website/zh/documents/introduce.md
+++ b/website/zh/documents/introduce.md
@@ -104,7 +104,7 @@ BitSail目前已被广泛使用,并支持数百万亿的大流量场景。同时
| Doris |
- |
- |
+ ✅ |
✅ |
From 96c1799835be31522d7a50e73a18ccb2c855ae11 Mon Sep 17 00:00:00 2001
From: DongLiang-0 <1747644936@qq.com>
Date: Wed, 10 May 2023 18:04:14 +0800
Subject: [PATCH 3/3] add doris-sdk to replace thrift compile
---
bitsail-connectors/connector-doris/pom.xml | 31 +-
.../doris/backend/BackendClient.java | 33 +-
.../doris/backend/model/RowBatch.java | 4 +-
.../source/split/DorisSourceSplitReader.java | 12 +-
.../DorisSourceSplitCoordinator.java | 2 +-
.../thrift/doris/DorisExternalService.thrift | 121 ------
.../src/main/thrift/doris/Status.thrift | 65 ---
.../src/main/thrift/doris/Types.thrift | 374 ------------------
.../doris/{ => sink}/DorisSinkITCase.java | 14 +-
.../doris/source/DorisSourceITCase.java | 14 +-
.../src/test/resources/doris_to_print.json | 40 ++
website/en/documents/start/env_setup.md | 28 --
website/zh/documents/start/env_setup.md | 29 +-
13 files changed, 88 insertions(+), 679 deletions(-)
delete mode 100644 bitsail-connectors/connector-doris/src/main/thrift/doris/DorisExternalService.thrift
delete mode 100644 bitsail-connectors/connector-doris/src/main/thrift/doris/Status.thrift
delete mode 100644 bitsail-connectors/connector-doris/src/main/thrift/doris/Types.thrift
rename bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/java/com/bytedance/bitsail/test/integration/doris/{ => sink}/DorisSinkITCase.java (85%)
rename {bitsail-connectors/connector-doris/src/test/java/com/bytedance/bitsail/connector => bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/java/com/bytedance/bitsail/test/integration}/doris/source/DorisSourceITCase.java (78%)
diff --git a/bitsail-connectors/connector-doris/pom.xml b/bitsail-connectors/connector-doris/pom.xml
index 548b9cc40..5e98d7d7a 100644
--- a/bitsail-connectors/connector-doris/pom.xml
+++ b/bitsail-connectors/connector-doris/pom.xml
@@ -32,10 +32,16 @@
8
5.1.49
5.0.0
- 0.13.0
+ 0.16.0
+ 1.0.0
+
+ org.apache.doris
+ thrift-service
+ ${thrift-service.version}
+
mysql
mysql-connector-java
@@ -118,27 +124,4 @@
test
-
-
-
-
- org.apache.thrift.tools
- maven-thrift-plugin
- 0.1.11
-
- ${thrift.binary}
- java:fullcamel
-
-
-
- thrift-sources
- generate-sources
-
- compile
-
-
-
-
-
-
\ No newline at end of file
diff --git a/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/backend/BackendClient.java b/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/backend/BackendClient.java
index bfe6d15bb..a77f72239 100644
--- a/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/backend/BackendClient.java
+++ b/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/backend/BackendClient.java
@@ -20,15 +20,16 @@
import com.bytedance.bitsail.connector.doris.backend.model.Routing;
import com.bytedance.bitsail.connector.doris.config.DorisExecutionOptions;
import com.bytedance.bitsail.connector.doris.error.DorisErrorCode;
-import com.bytedance.bitsail.connector.doris.thrift.TDorisExternalService;
-import com.bytedance.bitsail.connector.doris.thrift.TScanBatchResult;
-import com.bytedance.bitsail.connector.doris.thrift.TScanCloseParams;
-import com.bytedance.bitsail.connector.doris.thrift.TScanCloseResult;
-import com.bytedance.bitsail.connector.doris.thrift.TScanNextBatchParams;
-import com.bytedance.bitsail.connector.doris.thrift.TScanOpenParams;
-import com.bytedance.bitsail.connector.doris.thrift.TScanOpenResult;
-import com.bytedance.bitsail.connector.doris.thrift.TStatusCode;
+import org.apache.doris.sdk.thrift.TDorisExternalService;
+import org.apache.doris.sdk.thrift.TScanBatchResult;
+import org.apache.doris.sdk.thrift.TScanCloseParams;
+import org.apache.doris.sdk.thrift.TScanCloseResult;
+import org.apache.doris.sdk.thrift.TScanNextBatchParams;
+import org.apache.doris.sdk.thrift.TScanOpenParams;
+import org.apache.doris.sdk.thrift.TScanOpenResult;
+import org.apache.doris.sdk.thrift.TStatusCode;
+import org.apache.thrift.TConfiguration;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
@@ -66,15 +67,15 @@ private void open() {
TException ex = null;
for (int attempt = 0; !isConnected && attempt < retries; ++attempt) {
LOGGER.debug("Attempt {} to connect {}.", attempt, routing);
- TBinaryProtocol.Factory factory = new TBinaryProtocol.Factory();
- transport = new TSocket(routing.getHost(), routing.getPort(), socketTimeout, connectTimeout);
- TProtocol protocol = factory.getProtocol(transport);
- client = new TDorisExternalService.Client(protocol);
- if (isConnected) {
- LOGGER.info("Success connect to {}.", routing);
- return;
- }
try {
+ TBinaryProtocol.Factory factory = new TBinaryProtocol.Factory();
+ transport = new TSocket(new TConfiguration(), routing.getHost(), routing.getPort(), socketTimeout, connectTimeout);
+ TProtocol protocol = factory.getProtocol(transport);
+ client = new TDorisExternalService.Client(protocol);
+ if (isConnected) {
+ LOGGER.info("Success connect to {}.", routing);
+ return;
+ }
LOGGER.trace("Connect status before open transport to {} is '{}'.", routing, isConnected);
if (!transport.isOpen()) {
transport.open();
diff --git a/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/backend/model/RowBatch.java b/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/backend/model/RowBatch.java
index 6a2c2ecdb..b11af762d 100644
--- a/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/backend/model/RowBatch.java
+++ b/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/backend/model/RowBatch.java
@@ -17,9 +17,9 @@
package com.bytedance.bitsail.connector.doris.backend.model;
import com.bytedance.bitsail.common.BitSailException;
+import com.bytedance.bitsail.common.util.Preconditions;
import com.bytedance.bitsail.connector.doris.error.DorisErrorCode;
import com.bytedance.bitsail.connector.doris.rest.model.Schema;
-import com.bytedance.bitsail.connector.doris.thrift.TScanBatchResult;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.BigIntVector;
@@ -36,7 +36,7 @@
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowStreamReader;
import org.apache.arrow.vector.types.Types;
-import org.apache.flink.util.Preconditions;
+import org.apache.doris.sdk.thrift.TScanBatchResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/source/split/DorisSourceSplitReader.java b/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/source/split/DorisSourceSplitReader.java
index 8ab9df72a..2e8a05a3a 100644
--- a/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/source/split/DorisSourceSplitReader.java
+++ b/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/source/split/DorisSourceSplitReader.java
@@ -24,13 +24,13 @@
import com.bytedance.bitsail.connector.doris.rest.model.Field;
import com.bytedance.bitsail.connector.doris.rest.model.PartitionDefinition;
import com.bytedance.bitsail.connector.doris.rest.model.Schema;
-import com.bytedance.bitsail.connector.doris.thrift.TScanBatchResult;
-import com.bytedance.bitsail.connector.doris.thrift.TScanCloseParams;
-import com.bytedance.bitsail.connector.doris.thrift.TScanColumnDesc;
-import com.bytedance.bitsail.connector.doris.thrift.TScanNextBatchParams;
-import com.bytedance.bitsail.connector.doris.thrift.TScanOpenParams;
-import com.bytedance.bitsail.connector.doris.thrift.TScanOpenResult;
+import org.apache.doris.sdk.thrift.TScanBatchResult;
+import org.apache.doris.sdk.thrift.TScanCloseParams;
+import org.apache.doris.sdk.thrift.TScanColumnDesc;
+import org.apache.doris.sdk.thrift.TScanNextBatchParams;
+import org.apache.doris.sdk.thrift.TScanOpenParams;
+import org.apache.doris.sdk.thrift.TScanOpenResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/source/split/coordinator/DorisSourceSplitCoordinator.java b/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/source/split/coordinator/DorisSourceSplitCoordinator.java
index 46da687f7..187d54b1f 100644
--- a/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/source/split/coordinator/DorisSourceSplitCoordinator.java
+++ b/bitsail-connectors/connector-doris/src/main/java/com/bytedance/bitsail/connector/doris/source/split/coordinator/DorisSourceSplitCoordinator.java
@@ -118,7 +118,7 @@ public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname
}
@Override
- public EmptyState snapshotState() throws Exception {
+ public EmptyState snapshotState(long checkpoint) throws Exception {
return new EmptyState();
}
diff --git a/bitsail-connectors/connector-doris/src/main/thrift/doris/DorisExternalService.thrift b/bitsail-connectors/connector-doris/src/main/thrift/doris/DorisExternalService.thrift
deleted file mode 100644
index ba4ea6ffd..000000000
--- a/bitsail-connectors/connector-doris/src/main/thrift/doris/DorisExternalService.thrift
+++ /dev/null
@@ -1,121 +0,0 @@
-/*
- * Copyright 2022 Bytedance Ltd. and/or its affiliates.
- *
- * Licensed 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.
- */
-
-namespace java com.bytedance.bitsail.connector.doris.thrift
-namespace cpp doris
-
-include "Types.thrift"
-include "Status.thrift"
-
-
-// Parameters to open().
-struct TScanOpenParams {
-
- 1: required string cluster
-
- 2: required string database
-
- 3: required string table
-
- // tablets to scan
- 4: required list tablet_ids
-
- // base64 encoded binary plan fragment
- 5: required string opaqued_query_plan
-
- // A string specified for the table that is passed to the external data source.
- // Always set, may be an empty string.
- 6: optional i32 batch_size
-
- // reserved params for use
- 7: optional map properties
-
- // The query limit, if specified.
- 8: optional i64 limit
-
- // The authenticated user name. Always set.
- // maybe usefullless
- 9: optional string user
-
- 10: optional string passwd
- // max keep alive time min
- 11: optional i16 keep_alive_min
-
- 12: optional i32 query_timeout
-
- // memory limit for a single query
- 13: optional i64 mem_limit
-}
-
-struct TScanColumnDesc {
- // The column name
- 1: optional string name
- // The column type. Always set.
- 2: optional Types.TPrimitiveType type
-}
-
-// Returned by open().
-struct TScanOpenResult {
- 1: required Status.TStatus status
- // An opaque context_id used in subsequent getNext()/close() calls. Required.
- 2: optional string context_id
- // selected fields
- 3: optional list selected_columns
-
-}
-
-// Parameters to getNext()
-struct TScanNextBatchParams {
- // The opaque handle returned by the previous open() call. Always set.
- 1: optional string context_id // doris olap engine context id
- 2: optional i64 offset // doris should check the offset to prevent duplicate rpc calls
-}
-
-// Returned by getNext().
-struct TScanBatchResult {
- 1: required Status.TStatus status
-
- // If true, reached the end of the result stream; subsequent calls to
- // getNext() won’t return any more results. Required.
- 2: optional bool eos
-
- // A batch of rows of arrow format to return, if any exist. The number of rows in the batch
- // should be less than or equal to the batch_size specified in TOpenParams.
- 3: optional binary rows
-}
-
-// Parameters to close()
-struct TScanCloseParams {
- // The opaque handle returned by the previous open() call. Always set.
- 1: optional string context_id
-}
-
-// Returned by close().
-struct TScanCloseResult {
- 1: required Status.TStatus status
-}
-
-// scan service expose ability of scanning data ability to other compute system
-service TDorisExternalService {
- // doris will build a scan context for this session, context_id returned if success
- TScanOpenResult open_scanner(1: TScanOpenParams params);
-
- // return the batch_size of data
- TScanBatchResult get_next(1: TScanNextBatchParams params);
-
- // release the context resource associated with the context_id
- TScanCloseResult close_scanner(1: TScanCloseParams params);
-}
diff --git a/bitsail-connectors/connector-doris/src/main/thrift/doris/Status.thrift b/bitsail-connectors/connector-doris/src/main/thrift/doris/Status.thrift
deleted file mode 100644
index 5308a6f9f..000000000
--- a/bitsail-connectors/connector-doris/src/main/thrift/doris/Status.thrift
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Copyright 2022 Bytedance Ltd. and/or its affiliates.
- *
- * Licensed 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.
- */
-
-namespace java com.bytedance.bitsail.connector.doris.thrift
-namespace cpp doris
-
-enum TStatusCode {
- OK,
- CANCELLED,
- ANALYSIS_ERROR,
- NOT_IMPLEMENTED_ERROR,
- RUNTIME_ERROR,
- MEM_LIMIT_EXCEEDED,
- INTERNAL_ERROR,
- THRIFT_RPC_ERROR,
- TIMEOUT,
- KUDU_NOT_ENABLED, // Deprecated
- KUDU_NOT_SUPPORTED_ON_OS, // Deprecated
- MEM_ALLOC_FAILED,
- BUFFER_ALLOCATION_FAILED,
- MINIMUM_RESERVATION_UNAVAILABLE,
- PUBLISH_TIMEOUT,
- LABEL_ALREADY_EXISTS,
- ES_INTERNAL_ERROR,
- ES_INDEX_NOT_FOUND,
- ES_SHARD_NOT_FOUND,
- ES_INVALID_CONTEXTID,
- ES_INVALID_OFFSET,
- ES_REQUEST_ERROR,
-
- // end of file
- END_OF_FILE = 30,
- NOT_FOUND = 31,
- CORRUPTION = 32,
- INVALID_ARGUMENT = 33,
- IO_ERROR = 34,
- ALREADY_EXIST = 35,
- NETWORK_ERROR = 36,
- ILLEGAL_STATE = 37,
- NOT_AUTHORIZED = 38,
- ABORTED = 39,
- REMOTE_ERROR = 40,
- SERVICE_UNAVAILABLE = 41,
- UNINITIALIZED = 42,
- CONFIGURATION_ERROR = 43,
- INCOMPLETE = 44
-}
-
-struct TStatus {
- 1: required TStatusCode status_code
- 2: optional list error_msgs
-}
diff --git a/bitsail-connectors/connector-doris/src/main/thrift/doris/Types.thrift b/bitsail-connectors/connector-doris/src/main/thrift/doris/Types.thrift
deleted file mode 100644
index fa48c8e20..000000000
--- a/bitsail-connectors/connector-doris/src/main/thrift/doris/Types.thrift
+++ /dev/null
@@ -1,374 +0,0 @@
-/*
- * Copyright 2022 Bytedance Ltd. and/or its affiliates.
- *
- * Licensed 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.
- */
-
-namespace java com.bytedance.bitsail.connector.doris.thrift
-namespace cpp doris
-
-typedef i64 TTimestamp
-typedef i32 TPlanNodeId
-typedef i32 TTupleId
-typedef i32 TSlotId
-typedef i64 TTableId
-typedef i64 TTabletId
-typedef i64 TVersion
-typedef i64 TVersionHash
-typedef i32 TSchemaHash
-typedef i32 TPort
-typedef i64 TCount
-typedef i64 TSize
-typedef i32 TClusterId
-typedef i64 TEpoch
-
-// add for real time load, partitionid is not defined previously, define it here
-typedef i64 TTransactionId
-typedef i64 TPartitionId
-
-enum TStorageType {
- ROW,
- COLUMN,
-}
-
-enum TStorageMedium {
- HDD,
- SSD,
-}
-
-enum TVarType {
- SESSION,
- GLOBAL
-}
-
-enum TPrimitiveType {
- INVALID_TYPE,
- NULL_TYPE,
- BOOLEAN,
- TINYINT,
- SMALLINT,
- INT,
- BIGINT,
- FLOAT,
- DOUBLE,
- DATE,
- DATETIME,
- BINARY,
- DECIMAL,
- // CHAR(n). Currently only supported in UDAs
- CHAR,
- LARGEINT,
- VARCHAR,
- HLL,
- DECIMALV2,
- TIME,
- OBJECT,
- ARRAY,
- MAP,
- STRUCT,
- STRING,
- ALL
-}
-
-enum TTypeNodeType {
- SCALAR,
- ARRAY,
- MAP,
- STRUCT
-}
-
-struct TScalarType {
- 1: required TPrimitiveType type
-
- // Only set if type == CHAR or type == VARCHAR
- 2: optional i32 len
-
- // Only set for DECIMAL
- 3: optional i32 precision
- 4: optional i32 scale
-}
-
-// Represents a field in a STRUCT type.
-// TODO: Model column stats for struct fields.
-struct TStructField {
- 1: required string name
- 2: optional string comment
-}
-
-struct TTypeNode {
- 1: required TTypeNodeType type
-
- // only set for scalar types
- 2: optional TScalarType scalar_type
-
- // only used for structs; has struct_fields.size() corresponding child types
- 3: optional list struct_fields
-}
-
-// A flattened representation of a tree of column types obtained by depth-first
-// traversal. Complex types such as map, array and struct have child types corresponding
-// to the map key/value, array item type, and struct fields, respectively.
-// For scalar types the list contains only a single node.
-// Note: We cannot rename this to TType because it conflicts with Thrift's internal TType
-// and the generated Python thrift files will not work.
-// Note: TTypeDesc in impala is TColumnType, but we already use TColumnType, so we name this
-// to TTypeDesc. In future, we merge these two to one
-struct TTypeDesc {
- 1: list types
-}
-
-enum TAggregationType {
- SUM,
- MAX,
- MIN,
- REPLACE,
- HLL_UNION,
- NONE
-}
-
-enum TPushType {
- LOAD,
- DELETE,
- LOAD_DELETE
-}
-
-enum TTaskType {
- CREATE,
- DROP,
- PUSH,
- CLONE,
- STORAGE_MEDIUM_MIGRATE,
- ROLLUP,
- SCHEMA_CHANGE,
- CANCEL_DELETE, // Deprecated
- MAKE_SNAPSHOT,
- RELEASE_SNAPSHOT,
- CHECK_CONSISTENCY,
- UPLOAD,
- DOWNLOAD,
- CLEAR_REMOTE_FILE,
- MOVE
- REALTIME_PUSH,
- PUBLISH_VERSION,
- CLEAR_ALTER_TASK,
- CLEAR_TRANSACTION_TASK,
- RECOVER_TABLET,
- STREAM_LOAD,
- UPDATE_TABLET_META_INFO,
- ALTER_TASK
-}
-
-enum TStmtType {
- QUERY,
- DDL, // Data definition, e.g. CREATE TABLE (includes read-only functions e.g. SHOW)
- DML, // Data modification e.g. INSERT
- EXPLAIN // EXPLAIN
-}
-
-// level of verboseness for "explain" output
-// TODO: should this go somewhere else?
-enum TExplainLevel {
- NORMAL,
- VERBOSE
-}
-
-struct TColumnType {
- 1: required TPrimitiveType type
- // Only set if type == CHAR_ARRAY
- 2: optional i32 len
- 3: optional i32 index_len
- 4: optional i32 precision
- 5: optional i32 scale
-}
-
-// A TNetworkAddress is the standard host, port representation of a
-// network address. The hostname field must be resolvable to an IPv4
-// address.
-struct TNetworkAddress {
- 1: required string hostname
- 2: required i32 port
-}
-
-// Wire format for UniqueId
-struct TUniqueId {
- 1: required i64 hi
- 2: required i64 lo
-}
-
-enum QueryState {
- CREATED,
- INITIALIZED,
- COMPILED,
- RUNNING,
- FINISHED,
- EXCEPTION
-}
-
-enum TFunctionType {
- SCALAR,
- AGGREGATE,
-}
-
-enum TFunctionBinaryType {
- // Palo builtin. We can either run this interpreted or via codegen
- // depending on the query option.
- BUILTIN,
-
- // Hive UDFs, loaded from *.jar
- HIVE,
-
- // Native-interface, precompiled UDFs loaded from *.so
- NATIVE,
-
- // Native-interface, precompiled to IR; loaded from *.ll
- IR,
-}
-
-// Represents a fully qualified function name.
-struct TFunctionName {
- // Name of the function's parent database. Not set if in global
- // namespace (e.g. builtins)
- 1: optional string db_name
-
- // Name of the function
- 2: required string function_name
-}
-
-struct TScalarFunction {
- // Symbol for the function
- 1: required string symbol
- 2: optional string prepare_fn_symbol
- 3: optional string close_fn_symbol
-}
-
-struct TAggregateFunction {
- 1: required TTypeDesc intermediate_type
- 2: optional string update_fn_symbol
- 3: optional string init_fn_symbol
- 4: optional string serialize_fn_symbol
- 5: optional string merge_fn_symbol
- 6: optional string finalize_fn_symbol
- 8: optional string get_value_fn_symbol
- 9: optional string remove_fn_symbol
- 10: optional bool is_analytic_only_fn = false
-}
-
-// Represents a function in the Catalog.
-struct TFunction {
- // Fully qualified function name.
- 1: required TFunctionName name
-
- // Type of the udf. e.g. hive, native, ir
- 2: required TFunctionBinaryType binary_type
-
- // The types of the arguments to the function
- 3: required list arg_types
-
- // Return type for the function.
- 4: required TTypeDesc ret_type
-
- // If true, this function takes var args.
- 5: required bool has_var_args
-
- // Optional comment to attach to the function
- 6: optional string comment
-
- 7: optional string signature
-
- // HDFS path for the function binary. This binary must exist at the time the
- // function is created.
- 8: optional string hdfs_location
-
- // One of these should be set.
- 9: optional TScalarFunction scalar_fn
- 10: optional TAggregateFunction aggregate_fn
-
- 11: optional i64 id
- 12: optional string checksum
-}
-
-enum TLoadJobState {
- PENDING,
- ETL,
- LOADING,
- FINISHED,
- CANCELLED
-}
-
-enum TEtlState {
- RUNNING,
- FINISHED,
- CANCELLED,
- UNKNOWN
-}
-
-enum TTableType {
- MYSQL_TABLE,
- OLAP_TABLE,
- SCHEMA_TABLE,
- KUDU_TABLE, // Deprecated
- BROKER_TABLE,
- ES_TABLE
-}
-
-enum TKeysType {
- PRIMARY_KEYS,
- DUP_KEYS,
- UNIQUE_KEYS,
- AGG_KEYS
-}
-
-enum TPriority {
- NORMAL,
- HIGH
-}
-
-struct TBackend {
- 1: required string host
- 2: required TPort be_port
- 3: required TPort http_port
-}
-
-struct TResourceInfo {
- 1: required string user
- 2: required string group
-}
-
-enum TExportState {
- RUNNING,
- FINISHED,
- CANCELLED,
- UNKNOWN
-}
-
-enum TFileType {
- FILE_LOCAL,
- FILE_BROKER,
- FILE_STREAM, // file content is streaming in the buffer
-}
-
-struct TTabletCommitInfo {
- 1: required i64 tabletId
- 2: required i64 backendId
-}
-
-enum TLoadType {
- MANUL_LOAD,
- ROUTINE_LOAD,
- MINI_LOAD
-}
-
-enum TLoadSourceType {
- RAW,
- KAFKA,
-}
diff --git a/bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/java/com/bytedance/bitsail/test/integration/doris/DorisSinkITCase.java b/bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/java/com/bytedance/bitsail/test/integration/doris/sink/DorisSinkITCase.java
similarity index 85%
rename from bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/java/com/bytedance/bitsail/test/integration/doris/DorisSinkITCase.java
rename to bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/java/com/bytedance/bitsail/test/integration/doris/sink/DorisSinkITCase.java
index 05c841ca0..6ca4b2650 100644
--- a/bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/java/com/bytedance/bitsail/test/integration/doris/DorisSinkITCase.java
+++ b/bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/java/com/bytedance/bitsail/test/integration/doris/sink/DorisSinkITCase.java
@@ -14,7 +14,7 @@
* limitations under the License.
*/
-package com.bytedance.bitsail.test.integration.doris;
+package com.bytedance.bitsail.test.integration.doris.sink;
import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
import com.bytedance.bitsail.common.option.CommonOptions;
@@ -48,12 +48,12 @@ public void test() throws Exception {
* Below codes are just example.
*/
public void addDorisInfo(BitSailConfiguration jobConf) {
- jobConf.set(DorisWriterOptions.FE_HOSTS, "127.0.0.1:1234");
- jobConf.set(DorisWriterOptions.MYSQL_HOSTS, "127.0.0.1:4321");
- jobConf.set(DorisWriterOptions.USER, "test_user");
- jobConf.set(DorisWriterOptions.PASSWORD, "password");
- jobConf.set(DorisWriterOptions.DB_NAME, "test_db");
- jobConf.set(DorisWriterOptions.TABLE_NAME, "test_table");
+ jobConf.set(DorisWriterOptions.FE_HOSTS, "127.0.0.1:8030");
+ jobConf.set(DorisWriterOptions.MYSQL_HOSTS, "127.0.0.1:9030");
+ jobConf.set(DorisWriterOptions.USER, "root");
+ jobConf.set(DorisWriterOptions.PASSWORD, "");
+ jobConf.set(DorisWriterOptions.DB_NAME, "test");
+ jobConf.set(DorisWriterOptions.TABLE_NAME, "test_bitsails");
jobConf.set(CommonOptions.CheckPointOptions.CHECKPOINT_ENABLE, true);
jobConf.set(CommonOptions.CheckPointOptions.CHECKPOINT_INTERVAL, 5000L);
jobConf.set(DorisWriterOptions.SINK_ENABLE_2PC, false);
diff --git a/bitsail-connectors/connector-doris/src/test/java/com/bytedance/bitsail/connector/doris/source/DorisSourceITCase.java b/bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/java/com/bytedance/bitsail/test/integration/doris/source/DorisSourceITCase.java
similarity index 78%
rename from bitsail-connectors/connector-doris/src/test/java/com/bytedance/bitsail/connector/doris/source/DorisSourceITCase.java
rename to bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/java/com/bytedance/bitsail/test/integration/doris/source/DorisSourceITCase.java
index 2418d8178..3f6f0d716 100644
--- a/bitsail-connectors/connector-doris/src/test/java/com/bytedance/bitsail/connector/doris/source/DorisSourceITCase.java
+++ b/bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/java/com/bytedance/bitsail/test/integration/doris/source/DorisSourceITCase.java
@@ -14,24 +14,24 @@
* limitations under the License.
*/
-package com.bytedance.bitsail.connector.doris.source;
+package com.bytedance.bitsail.test.integration.doris.source;
import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
import com.bytedance.bitsail.connector.doris.option.DorisReaderOptions;
-import com.bytedance.bitsail.test.connector.test.EmbeddedFlinkCluster;
-import com.bytedance.bitsail.test.connector.test.utils.JobConfUtils;
+import com.bytedance.bitsail.test.integration.AbstractIntegrationTest;
+import com.bytedance.bitsail.test.integration.utils.JobConfUtils;
import org.junit.Ignore;
import org.junit.Test;
@Ignore
-public class DorisSourceITCase {
+public class DorisSourceITCase extends AbstractIntegrationTest {
@Test
public void test() throws Exception {
BitSailConfiguration jobConf = JobConfUtils.fromClasspath("doris_to_print.json");
addDorisInfo(jobConf);
- EmbeddedFlinkCluster.submitJob(jobConf);
+ submitJob(jobConf);
}
/**
@@ -46,8 +46,8 @@ public void addDorisInfo(BitSailConfiguration jobConf) {
jobConf.set(DorisReaderOptions.USER, "root");
jobConf.set(DorisReaderOptions.PASSWORD, "");
jobConf.set(DorisReaderOptions.DB_NAME, "test");
- jobConf.set(DorisReaderOptions.TABLE_NAME, "doris_table");
- jobConf.set(DorisReaderOptions.SQL_FILTER, "id=1");
+ jobConf.set(DorisReaderOptions.TABLE_NAME, "test_bitsail");
+// jobConf.set(DorisReaderOptions.SQL_FILTER, "id=1");
}
}
diff --git a/bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/resources/doris_to_print.json b/bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/resources/doris_to_print.json
index e69de29bb..113ca2fc1 100644
--- a/bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/resources/doris_to_print.json
+++ b/bitsail-test/bitsail-test-integration/bitsail-test-integration-doris/src/test/resources/doris_to_print.json
@@ -0,0 +1,40 @@
+{
+ "job": {
+ "common": {
+ "job_id": -31198,
+ "instance_id": -31198,
+ "job_name": "bitsail_doris_to_print_test",
+ "user_name": "test"
+ },
+ "reader": {
+ "class": "com.bytedance.bitsail.connector.doris.source.DorisSource",
+ "fe_hosts": "127.0.0.1:8030",
+ "mysql_hosts": "127.0.0.1:9030",
+ "user": "root",
+ "password": "",
+ "db_name": "test",
+ "table_name": "doris_table",
+ "columns": [
+ {
+ "name": "id",
+ "type": "int"
+ },
+ {
+ "name": "bigint_type",
+ "type": "bigint"
+ },
+ {
+ "name": "string_type",
+ "type": "string"
+ },
+ {
+ "name": "double_type",
+ "type": "double"
+ }
+ ]
+ },
+ "writer": {
+ "class": "com.bytedance.bitsail.connector.legacy.print.sink.PrintSink"
+ }
+ }
+}
\ No newline at end of file
diff --git a/website/en/documents/start/env_setup.md b/website/en/documents/start/env_setup.md
index 21d1e3463..0d3f51c53 100644
--- a/website/en/documents/start/env_setup.md
+++ b/website/en/documents/start/env_setup.md
@@ -15,34 +15,6 @@ English | [简体中文](../../../zh/documents/start/env_setup.md)
- JDK1.8
- maven 3.6+
- [Docker desktop](https://www.docker.com/products/docker-desktop/)
-- thrift
-```bash
-install thrift
- Windows:
- 1. Download: `http://archive.apache.org/dist/thrift/0.13.0/thrift-0.13.0.exe`
- 2. Modify thrift-0.13.0.exe to thrift
-
- MacOS:
- 1. Download: `brew install thrift@0.13.0`
- 2. default address: /opt/homebrew/Cellar/thrift@0.13.0/0.13.0/bin/thrift
-
- Note: Executing `brew install thrift@0.13.0` on MacOS may report an error that the version cannot be found. The solution is as follows, execute it in the terminal:
- 1. `brew tap-new $USER/local-tap`
- 2. `brew extract --version='0.13.0' thrift $USER/local-tap`
- 3. `brew install thrift@0.13.0`
- Reference link: `https://gist.github.com/tonydeng/02e571f273d6cce4230dc8d5f394493c`
-
- Linux:
- 1.Download source package：`wget https://archive.apache.org/dist/thrift/0.13.0/thrift-0.13.0.tar.gz`
- 2.Install dependencies：`yum install -y autoconf automake libtool cmake ncurses-devel openssl-devel lzo-devel zlib-devel gcc gcc-c++`
- 3.`tar zxvf thrift-0.13.0.tar.gz`
- 4.`cd thrift-0.13.0`
- 5.`./configure --without-tests`
- 6.`make`
- 7.`make install`
- Check the version after installation is complete：thrift --version
- Note: If you have compiled Doris, you do not need to install thrift, you can directly use $DORIS_HOME/thirdparty/installed/bin/thrift
-```
After correctly installing the above required components, we are able to run integration tests on your local IDE.
diff --git a/website/zh/documents/start/env_setup.md b/website/zh/documents/start/env_setup.md
index 3e95c0503..41651e941 100644
--- a/website/zh/documents/start/env_setup.md
+++ b/website/zh/documents/start/env_setup.md
@@ -15,34 +15,7 @@ order: 2
- JDK1.8
- maven 3.6+
- [Docker desktop](https://www.docker.com/products/docker-desktop/)
-- thrift
-```bash
-安装 thrift
- Windows:
- 1.下载：`http://archive.apache.org/dist/thrift/0.13.0/thrift-0.13.0.exe`(下载目录自己指定)
- 2.修改thrift-0.13.0.exe 为 thrift
-
- MacOS:
- 1. 下载：`brew install thrift@0.13.0`
- 2. 默认下载地址：/opt/homebrew/Cellar/thrift@0.13.0/0.13.0/bin/thrift
-
- 注：MacOS执行 `brew install thrift@0.13.0` 可能会报找不到版本的错误，解决方法如下，在终端执行：
- 1. `brew tap-new $USER/local-tap`
- 2. `brew extract --version='0.13.0' thrift $USER/local-tap`
- 3. `brew install thrift@0.13.0`
- 参考链接: `https://gist.github.com/tonydeng/02e571f273d6cce4230dc8d5f394493c`
-
- Linux:
- 1.下载源码包：`wget https://archive.apache.org/dist/thrift/0.13.0/thrift-0.13.0.tar.gz`
- 2.安装依赖：`yum install -y autoconf automake libtool cmake ncurses-devel openssl-devel lzo-devel zlib-devel gcc gcc-c++`
- 3.`tar zxvf thrift-0.13.0.tar.gz`
- 4.`cd thrift-0.13.0`
- 5.`./configure --without-tests`
- 6.`make`
- 7.`make install`
- 安装完成后查看版本：thrift --version
- 注：如果编译过Doris，则不需要安装thrift,可以直接使用 $DORIS_HOME/thirdparty/installed/bin/thrift
-```
+
在安装上述必需组件后，您可以在本地的IDE上直接运行已有的集成测试。