You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by di...@apache.org on 2026/04/07 06:06:05 UTC
(doris-flink-connector) branch master updated: [Feature] enable gz compression by default for StreamLoad (#648)
This is an automated email from the ASF dual-hosted git repository.
diwu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-flink-connector.git
The following commit(s) were added to refs/heads/master by this push:
new 06edd0c3 [Feature] enable gz compression by default for StreamLoad (#648)
06edd0c3 is described below
commit 06edd0c3d773c6f0f42c5cfb53c543e1cbb8aa57
Author: wudi <67...@qq.com>
AuthorDate: Tue Apr 7 14:05:52 2026 +0800
[Feature] enable gz compression by default for StreamLoad (#648)
Enable gz compression by default for StreamLoad writes to reduce network transfer and improve write performance
---
.github/workflows/checkstyle.yaml | 4 +-
.../doris/flink/cfg/DorisExecutionOptions.java | 7 ++
.../doris/flink/cfg/DorisExecutionOptionsTest.java | 4 +-
.../doris/flink/container/AbstractE2EService.java | 3 +
.../apache/doris/flink/sink/DorisSinkITCase.java | 97 ++++++++++++++++++++++
5 files changed, 112 insertions(+), 3 deletions(-)
diff --git a/.github/workflows/checkstyle.yaml b/.github/workflows/checkstyle.yaml
index 083729fc..368d24e0 100644
--- a/.github/workflows/checkstyle.yaml
+++ b/.github/workflows/checkstyle.yaml
@@ -40,7 +40,7 @@ jobs:
- name: Run java checkstyle
run:
- cd flink-doris-connector && mvn clean checkstyle:check -Pflink1 -pl flink-doris-connector-flink1 -am
+ cd flink-doris-connector && mvn clean install checkstyle:check -DskipTests -Pflink1 -pl flink-doris-connector-flink1 -am
- name: Setup java 17
uses: actions/setup-java@v2
@@ -50,4 +50,4 @@ jobs:
- name: Run java checkstyle
run:
- cd flink-doris-connector && mvn clean checkstyle:check -Pflink2 -pl flink-doris-connector-flink2 -am
\ No newline at end of file
+ cd flink-doris-connector && mvn clean install checkstyle:check -DskipTests -Pflink2 -pl flink-doris-connector-flink2 -am
\ No newline at end of file
diff --git a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/DorisExecutionOptions.java b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/DorisExecutionOptions.java
index 99379ff8..1d33d21a 100644
--- a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/DorisExecutionOptions.java
+++ b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/DorisExecutionOptions.java
@@ -25,6 +25,8 @@ import java.io.Serializable;
import java.util.Objects;
import java.util.Properties;
+import static org.apache.doris.flink.sink.writer.LoadConstants.COMPRESS_TYPE;
+import static org.apache.doris.flink.sink.writer.LoadConstants.COMPRESS_TYPE_GZ;
import static org.apache.doris.flink.sink.writer.LoadConstants.FORMAT_KEY;
import static org.apache.doris.flink.sink.writer.LoadConstants.JSON;
import static org.apache.doris.flink.sink.writer.LoadConstants.READ_JSON_BY_LINE;
@@ -531,6 +533,11 @@ public class DorisExecutionOptions implements Serializable {
streamLoadProp.put(READ_JSON_BY_LINE, true);
}
+ // Enable gz compression by default
+ if (streamLoadProp != null && !streamLoadProp.containsKey(COMPRESS_TYPE)) {
+ streamLoadProp.put(COMPRESS_TYPE, COMPRESS_TYPE_GZ);
+ }
+
Preconditions.checkArgument(
bufferFlushIntervalMs >= 1000,
"bufferFlushIntervalMs must be greater than or equal to 1 second");
diff --git a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/cfg/DorisExecutionOptionsTest.java b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/cfg/DorisExecutionOptionsTest.java
index 97c2868a..d1aaf422 100644
--- a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/cfg/DorisExecutionOptionsTest.java
+++ b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/cfg/DorisExecutionOptionsTest.java
@@ -36,10 +36,12 @@ public class DorisExecutionOptionsTest {
Properties expected = new Properties();
expected.put("format", "json");
expected.put("read_json_by_line", true);
- Assert.assertTrue(actual.size() == 2);
+ expected.put("compress_type", "gz");
+ Assert.assertTrue(actual.size() == 3);
Assert.assertTrue(actual.get("format").equals(expected.get("format")));
Assert.assertTrue(
actual.get("read_json_by_line").equals(expected.get("read_json_by_line")));
+ Assert.assertTrue(actual.get("compress_type").equals(expected.get("compress_type")));
}
@Test
diff --git a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/container/AbstractE2EService.java b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/container/AbstractE2EService.java
index f0c027a0..669f8abd 100644
--- a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/container/AbstractE2EService.java
+++ b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/container/AbstractE2EService.java
@@ -142,6 +142,9 @@ public abstract class AbstractE2EService extends AbstractContainerTestBase {
argList.add(JDBC_URL + "=" + getDorisQueryUrl());
argList.add(SINK_CONF);
argList.add(SINK_LABEL_PREFIX + "=" + "label");
+ // disable gz compression for json format, doris 2.1 does not support json compression
+ argList.add(SINK_CONF);
+ argList.add("sink.properties.compress_type=");
}
public static void closeE2EContainers() {
diff --git a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/sink/DorisSinkITCase.java b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/sink/DorisSinkITCase.java
index 3d1345b8..f3374f31 100644
--- a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/sink/DorisSinkITCase.java
+++ b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/sink/DorisSinkITCase.java
@@ -77,6 +77,8 @@ public class DorisSinkITCase extends AbstractITCaseService {
static final String TABLE_GROUP_COMMIT = "tbl_group_commit";
static final String TABLE_OVERWRITE = "tbl_overwrite";
static final String TABLE_GZ_FORMAT = "tbl_gz_format";
+ static final String TABLE_GZ_FORMAT_DEFAULT = "tbl_gz_format_default";
+ static final String TABLE_NO_COMPRESS = "tbl_no_compress";
static final String TABLE_CSV_JM = "tbl_csv_jm";
static final String TABLE_CSV_TM = "tbl_csv_tm";
static final String TABLE_UNICODE_COLUMN = "tbl_unicode_column";
@@ -137,6 +139,7 @@ public class DorisSinkITCase extends AbstractITCaseService {
Properties properties = new Properties();
properties.setProperty("read_json_by_line", "true");
properties.setProperty("format", "json");
+ properties.setProperty("compress_type", "");
// mock data
Map<String, Object> row1 = new HashMap<>();
@@ -223,6 +226,7 @@ public class DorisSinkITCase extends AbstractITCaseService {
+ " 'sink.ignore.update-before' = 'true',"
+ " 'sink.properties.format' = 'json',"
+ " 'sink.properties.read_json_by_line' = 'true',"
+ + " 'sink.properties.compress_type' = '',"
+ " 'sink.label-prefix' = 'doris_sink"
+ UUID.randomUUID()
+ "'"
@@ -481,6 +485,98 @@ public class DorisSinkITCase extends AbstractITCaseService {
ContainerUtils.checkResult(getDorisQueryConnection(), LOG, expected, query, 2);
}
+ @Test
+ public void testTableDefaultGzFormat() throws Exception {
+ initializeTable(TABLE_GZ_FORMAT_DEFAULT, DataModel.UNIQUE);
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(DEFAULT_PARALLELISM);
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+ final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ String sinkDDL =
+ String.format(
+ "CREATE TABLE doris_gz_default_sink ("
+ + " name STRING,"
+ + " age INT"
+ + ") WITH ("
+ + " 'connector' = '"
+ + DorisConfigOptions.IDENTIFIER
+ + "',"
+ + " 'fenodes' = '%s',"
+ + " 'table.identifier' = '%s',"
+ + " 'username' = '%s',"
+ + " 'password' = '%s',"
+ + " 'sink.enable.batch-mode' = '%s',"
+ + " 'sink.enable-delete' = 'false',"
+ + " 'sink.label-prefix' = '"
+ + UUID.randomUUID()
+ + "',"
+ + " 'sink.properties.column_separator' = '\\x01',"
+ + " 'sink.properties.line_delimiter' = '\\x02'"
+ + ")",
+ getFenodes(),
+ DATABASE + "." + TABLE_GZ_FORMAT_DEFAULT,
+ getDorisUsername(),
+ getDorisPassword(),
+ batchMode);
+ tEnv.executeSql(sinkDDL);
+ tEnv.executeSql(
+ "INSERT INTO doris_gz_default_sink SELECT 'doris',1 union all SELECT 'flink',2");
+
+ Thread.sleep(25000);
+ List<String> expected = Arrays.asList("doris,1", "flink,2");
+ String query =
+ String.format(
+ "select name,age from %s.%s order by 1", DATABASE, TABLE_GZ_FORMAT_DEFAULT);
+ ContainerUtils.checkResult(getDorisQueryConnection(), LOG, expected, query, 2);
+ }
+
+ @Test
+ public void testTableNoCompressFormat() throws Exception {
+ initializeTable(TABLE_NO_COMPRESS, DataModel.UNIQUE);
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(DEFAULT_PARALLELISM);
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+ final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ String sinkDDL =
+ String.format(
+ "CREATE TABLE doris_no_compress_sink ("
+ + " name STRING,"
+ + " age INT"
+ + ") WITH ("
+ + " 'connector' = '"
+ + DorisConfigOptions.IDENTIFIER
+ + "',"
+ + " 'fenodes' = '%s',"
+ + " 'table.identifier' = '%s',"
+ + " 'username' = '%s',"
+ + " 'password' = '%s',"
+ + " 'sink.enable.batch-mode' = '%s',"
+ + " 'sink.enable-delete' = 'false',"
+ + " 'sink.label-prefix' = '"
+ + UUID.randomUUID()
+ + "',"
+ + " 'sink.properties.column_separator' = '\\x01',"
+ + " 'sink.properties.line_delimiter' = '\\x02',"
+ + " 'sink.properties.compress_type' = ''"
+ + ")",
+ getFenodes(),
+ DATABASE + "." + TABLE_NO_COMPRESS,
+ getDorisUsername(),
+ getDorisPassword(),
+ batchMode);
+ tEnv.executeSql(sinkDDL);
+ tEnv.executeSql(
+ "INSERT INTO doris_no_compress_sink SELECT 'doris',1 union all SELECT 'flink',2");
+
+ Thread.sleep(25000);
+ List<String> expected = Arrays.asList("doris,1", "flink,2");
+ String query =
+ String.format("select name,age from %s.%s order by 1", DATABASE, TABLE_NO_COMPRESS);
+ ContainerUtils.checkResult(getDorisQueryConnection(), LOG, expected, query, 2);
+ }
+
@Test
public void testJobManagerFailoverSink() throws Exception {
LOG.info("start to test JobManagerFailoverSink.");
@@ -780,6 +876,7 @@ public class DorisSinkITCase extends AbstractITCaseService {
Properties properties = new Properties();
properties.setProperty("read_json_by_line", "true");
properties.setProperty("format", "json");
+ properties.setProperty("compress_type", "");
// mock data
Map<String, Object> row1 = new HashMap<>();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org