This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 2714e6e25c [Improve][Connector-V2][Doris] Support partition cleanup
for DROP_DATA (#11917)
2714e6e25c is described below
commit 2714e6e25c67adb0f26eff7327368749041d1e5b
Author: Goutam Adwant <[email protected]>
AuthorDate: Sun Aug 23 06:27:04 2026 -0700
[Improve][Connector-V2][Doris] Support partition cleanup for DROP_DATA
(#11917)
Signed-off-by: goutamadwant <[email protected]>
---
docs/en/connectors/sink/Doris.md | 4 +-
docs/zh/connectors/sink/Doris.md | 4 +-
.../connectors/doris/catalog/DorisCatalog.java | 14 ++-
.../connectors/doris/config/DorisSinkConfig.java | 49 ++++++++
.../seatunnel/connectors/doris/sink/DorisSink.java | 13 ++-
.../doris/sink/savemode/DorisSaveModeHandler.java | 59 ++++++++++
.../connectors/doris/util/DorisCatalogUtil.java | 19 +++
.../doris/config/DorisSinkConfigTest.java | 66 +++++++++++
.../sink/savemode/DorisSaveModeHandlerTest.java | 128 +++++++++++++++++++++
.../doris/util/DorisCatalogUtilTest.java | 32 ++++++
10 files changed, 378 insertions(+), 10 deletions(-)
diff --git a/docs/en/connectors/sink/Doris.md b/docs/en/connectors/sink/Doris.md
index 5afacfc0bd..9db57c496d 100644
--- a/docs/en/connectors/sink/Doris.md
+++ b/docs/en/connectors/sink/Doris.md
@@ -65,7 +65,7 @@ The internal implementation of Doris sink connector is cached
and imported by st
| data_save_mode | Enum | no | APPEND_DATA
| the data save mode, please refer to `data_save_mode` below
|
| save_mode_create_template | string | no | see below
| see below
|
| custom_sql | String | no | -
| When data_save_mode selects CUSTOM_PROCESSING, you should fill in the
CUSTOM_SQL parameter. This parameter usually fills in a SQL that can be
executed. SQL will be executed before synchronization tasks.
|
-| doris.config | map | yes | -
| Stream Load data description parameters passed to Doris. The most
common keys are `format` (`json` or `csv`), `read_json_by_line`
(`true`/`false`), `column_separator`, and `row_delimiter`. See the Doris Stream
Load documentation for the full key set. |
+| doris.config | map | yes | -
| Stream Load data description parameters passed to Doris. The most
common keys are `format` (`json` or `csv`), `read_json_by_line`
(`true`/`false`), `column_separator`, `row_delimiter`, and `partitions`. See
the Doris Stream Load documentation for the full key set. |
## Redirect Behavior
@@ -93,7 +93,7 @@ Option introduction：
Before the synchronous task is turned on, different processing schemes are
selected for data existing data on the target side.
Option introduction：
-`DROP_DATA`： Preserve database structure and delete data
+`DROP_DATA`： Preserve database structure and delete data. By default the
complete target table is truncated. When `doris.config.partitions` contains a
comma-separated list of formal Doris partition names, only those partitions are
truncated and subsequent Stream Load writes are restricted to the same
partitions. Every target in a multi-table job must contain the configured
partition names. Doris rejects cleanup when a partition does not exist.
Partition values and temporary partitions ar [...]
`APPEND_DATA`：Preserve database structure, preserve data
`CUSTOM_PROCESSING`：User defined processing
`ERROR_WHEN_DATA_EXISTS`：When there is data, an error is reported
diff --git a/docs/zh/connectors/sink/Doris.md b/docs/zh/connectors/sink/Doris.md
index 70af62651b..627f2fe66f 100644
--- a/docs/zh/connectors/sink/Doris.md
+++ b/docs/zh/connectors/sink/Doris.md
@@ -65,7 +65,7 @@ Doris Sink连接器的内部实现是通过stream load批量缓存和导入的
| data_save_mode | Enum | no | APPEND_DATA
| 数据保存模式，请参考下面的`data_save_mode`。
|
| save_mode_create_template | string | no | see below
| 见下文。
|
| custom_sql | String | no | -
| 当data_save_mode选择CUSTOM_PROCESSING时，需要填写CUSTOM_SQL参数。
该参数通常填写一条可以执行的SQL。 SQL将在同步任务之前执行。
|
-| doris.config | map | yes | -
| 该选项用于支持自动生成sql时的insert、delete、update等操作，以及支持的格式。
|
+| doris.config | map | yes | -
| 传递给 Doris Stream Load 的数据描述参数。常用参数包括
`format`、`read_json_by_line`、`column_separator`、`row_delimiter` 和 `partitions`。
|
## Redirect 行为说明
@@ -93,7 +93,7 @@ Doris Sink连接器的内部实现是通过stream load批量缓存和导入的
在开启同步任务之前，针对目标端已有的数据选择不同的处理方案。
选项介绍：
-`DROP_DATA`： 保留数据库结构并删除数据。
+`DROP_DATA`： 保留数据库结构并删除数据。默认清空整个目标表。当 `doris.config.partitions` 配置了以逗号分隔的
Doris 正式分区名称时，只清空这些分区，后续 Stream Load 也只写入相同分区。多表任务中的每个目标表都必须包含这些分区；分区不存在时 Doris
会拒绝清理。不支持分区值或临时分区。
`APPEND_DATA`：保留数据库结构，保留数据。
`CUSTOM_PROCESSING`：用户自定义处理。
`ERROR_WHEN_DATA_EXISTS`：有数据时报错。
diff --git
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/catalog/DorisCatalog.java
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/catalog/DorisCatalog.java
index 20e25e6cc6..f06f382bcf 100644
---
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/catalog/DorisCatalog.java
+++
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/catalog/DorisCatalog.java
@@ -471,9 +471,21 @@ public class DorisCatalog implements Catalog {
public void truncateTable(TablePath tablePath, boolean ignoreIfNotExists)
throws TableNotExistException, CatalogException {
+ truncateTable(tablePath, ignoreIfNotExists, Collections.emptyList());
+ }
+
+ /**
+ * Truncates the requested partitions, or the complete table when no
partitions are supplied.
+ */
+ public void truncateTable(
+ TablePath tablePath, boolean ignoreIfNotExists, List<String>
partitions)
+ throws TableNotExistException, CatalogException {
try {
if (ignoreIfNotExists) {
-
conn.createStatement().execute(DorisCatalogUtil.getTruncateTableQuery(tablePath));
+ try (Statement statement = conn.createStatement()) {
+ statement.execute(
+ DorisCatalogUtil.getTruncateTableQuery(tablePath,
partitions));
+ }
}
} catch (Exception e) {
throw new CatalogException(
diff --git
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/config/DorisSinkConfig.java
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/config/DorisSinkConfig.java
index 26294865d3..7bfde8acfe 100644
---
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/config/DorisSinkConfig.java
+++
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/config/DorisSinkConfig.java
@@ -20,7 +20,10 @@ package org.apache.seatunnel.connectors.doris.config;
import org.apache.seatunnel.shade.com.typesafe.config.Config;
import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.sink.DataSaveMode;
+import org.apache.seatunnel.connectors.doris.exception.DorisConnectorException;
import lombok.Getter;
import lombok.Setter;
@@ -28,8 +31,13 @@ import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.Set;
import static
org.apache.seatunnel.connectors.doris.config.DorisBaseOptions.DATABASE;
import static
org.apache.seatunnel.connectors.doris.config.DorisBaseOptions.DORIS_BATCH_SIZE;
@@ -40,6 +48,7 @@ import static
org.apache.seatunnel.connectors.doris.config.DorisBaseOptions.TABL
import static
org.apache.seatunnel.connectors.doris.config.DorisBaseOptions.USERNAME;
import static
org.apache.seatunnel.connectors.doris.config.DorisSinkOptions.BENODES;
import static
org.apache.seatunnel.connectors.doris.config.DorisSinkOptions.CASE_SENSITIVE;
+import static
org.apache.seatunnel.connectors.doris.config.DorisSinkOptions.DATA_SAVE_MODE;
import static
org.apache.seatunnel.connectors.doris.config.DorisSinkOptions.DIRECT_TO_BE;
import static
org.apache.seatunnel.connectors.doris.config.DorisSinkOptions.DORIS_SINK_CONFIG_PREFIX;
import static
org.apache.seatunnel.connectors.doris.config.DorisSinkOptions.NEEDS_UNSUPPORTED_TYPE_CASTING;
@@ -77,6 +86,7 @@ public class DorisSinkConfig implements Serializable {
private Integer bufferSize;
private Integer bufferCount;
private Properties streamLoadProps;
+ private List<String> partitions = Collections.emptyList();
private boolean directToBe;
private boolean needsUnsupportedTypeCasting;
private boolean caseSensitive;
@@ -99,6 +109,15 @@ public class DorisSinkConfig implements Serializable {
dorisSinkConfig.setPassword(config.get(PASSWORD));
dorisSinkConfig.setQueryPort(config.get(QUERY_PORT));
dorisSinkConfig.setStreamLoadProps(parseStreamLoadProperties(config));
+ if (config.get(DATA_SAVE_MODE) == DataSaveMode.DROP_DATA) {
+ List<String> partitions =
parseDropDataPartitions(dorisSinkConfig.getStreamLoadProps());
+ dorisSinkConfig.setPartitions(partitions);
+ if (!partitions.isEmpty()) {
+ dorisSinkConfig
+ .getStreamLoadProps()
+ .setProperty("partitions", String.join(",",
partitions));
+ }
+ }
dorisSinkConfig.setDatabase(config.get(DATABASE));
dorisSinkConfig.setTable(config.get(TABLE));
dorisSinkConfig.setBatchSize(config.get(DORIS_BATCH_SIZE));
@@ -136,4 +155,34 @@ public class DorisSinkConfig implements Serializable {
}
return streamLoadProps;
}
+ /**
+ * Parses a comma-separated partition list for partition-scoped DROP_DATA
cleanup.
+ *
+ * @throws DorisConnectorException if a partition name is blank or
duplicated
+ */
+ private static List<String> parseDropDataPartitions(Properties
streamLoadProperties) {
+ String configuredPartitions =
streamLoadProperties.getProperty("partitions");
+ if (configuredPartitions == null) {
+ return Collections.emptyList();
+ }
+
+ String[] values = configuredPartitions.split(",", -1);
+ List<String> partitions = new ArrayList<>(values.length);
+ Set<String> uniquePartitions = new HashSet<>();
+ for (String value : values) {
+ String partition = value.trim();
+ if (partition.isEmpty()) {
+ throw new DorisConnectorException(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+ "PluginName: Doris, Message: 'doris.config.partitions'
cannot contain blank partition names.");
+ }
+ if (!uniquePartitions.add(partition)) {
+ throw new DorisConnectorException(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+ "PluginName: Doris, Message: 'doris.config.partitions'
cannot contain duplicate partition names.");
+ }
+ partitions.add(partition);
+ }
+ return Collections.unmodifiableList(partitions);
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/DorisSink.java
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/DorisSink.java
index 88bee59012..b592d38972 100644
---
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/DorisSink.java
+++
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/DorisSink.java
@@ -21,7 +21,6 @@ import org.apache.seatunnel.api.common.JobContext;
import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.serialization.Serializer;
-import org.apache.seatunnel.api.sink.DefaultSaveModeHandler;
import org.apache.seatunnel.api.sink.SaveModeHandler;
import org.apache.seatunnel.api.sink.SeaTunnelSink;
import org.apache.seatunnel.api.sink.SinkCommitter;
@@ -29,18 +28,19 @@ import org.apache.seatunnel.api.sink.SinkWriter;
import org.apache.seatunnel.api.sink.SupportMultiTableSink;
import org.apache.seatunnel.api.sink.SupportSaveMode;
import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSink;
-import org.apache.seatunnel.api.table.catalog.Catalog;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.factory.CatalogFactory;
import org.apache.seatunnel.api.table.schema.SchemaChangeType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.common.constants.PluginType;
+import org.apache.seatunnel.connectors.doris.catalog.DorisCatalog;
import org.apache.seatunnel.connectors.doris.config.DorisSinkConfig;
import org.apache.seatunnel.connectors.doris.config.DorisSinkOptions;
import org.apache.seatunnel.connectors.doris.exception.DorisConnectorException;
import org.apache.seatunnel.connectors.doris.sink.committer.DorisCommitInfo;
import
org.apache.seatunnel.connectors.doris.sink.committer.DorisCommitInfoSerializer;
import org.apache.seatunnel.connectors.doris.sink.committer.DorisCommitter;
+import
org.apache.seatunnel.connectors.doris.sink.savemode.DorisSaveModeHandler;
import org.apache.seatunnel.connectors.doris.sink.writer.DorisSinkState;
import
org.apache.seatunnel.connectors.doris.sink.writer.DorisSinkStateSerializer;
import org.apache.seatunnel.connectors.doris.sink.writer.DorisSinkWriter;
@@ -155,14 +155,17 @@ public class DorisSink
getPluginName(), PluginType.SINK, "Cannot find
Doris catalog factory"));
}
- Catalog catalog =
catalogFactory.createCatalog(catalogFactory.factoryIdentifier(), config);
+ DorisCatalog catalog =
+ (DorisCatalog)
+
catalogFactory.createCatalog(catalogFactory.factoryIdentifier(), config);
return Optional.of(
- new DefaultSaveModeHandler(
+ new DorisSaveModeHandler(
config.get(DorisSinkOptions.SCHEMA_SAVE_MODE),
config.get(DorisSinkOptions.DATA_SAVE_MODE),
catalog,
catalogTable,
- config.get(DorisSinkOptions.CUSTOM_SQL)));
+ config.get(DorisSinkOptions.CUSTOM_SQL),
+ dorisSinkConfig.getPartitions()));
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/savemode/DorisSaveModeHandler.java
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/savemode/DorisSaveModeHandler.java
new file mode 100644
index 0000000000..093c70de20
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/savemode/DorisSaveModeHandler.java
@@ -0,0 +1,59 @@
+/*
+ * 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.seatunnel.connectors.doris.sink.savemode;
+
+import org.apache.seatunnel.api.sink.DataSaveMode;
+import org.apache.seatunnel.api.sink.DefaultSaveModeHandler;
+import org.apache.seatunnel.api.sink.SchemaSaveMode;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.connectors.doris.catalog.DorisCatalog;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.List;
+
+/** Applies Doris-specific partition cleanup for the DROP_DATA save mode. */
+@Slf4j
+public class DorisSaveModeHandler extends DefaultSaveModeHandler {
+
+ private final List<String> partitions;
+
+ private final DorisCatalog dorisCatalog;
+
+ public DorisSaveModeHandler(
+ SchemaSaveMode schemaSaveMode,
+ DataSaveMode dataSaveMode,
+ DorisCatalog catalog,
+ CatalogTable catalogTable,
+ String customSql,
+ List<String> partitions) {
+ super(schemaSaveMode, dataSaveMode, catalog, catalogTable, customSql);
+ this.dorisCatalog = catalog;
+ this.partitions = partitions;
+ }
+
+ @Override
+ protected void truncateTable() {
+ if (partitions.isEmpty()) {
+ super.truncateTable();
+ return;
+ }
+ log.info("Truncating partitions {} in table {}", partitions,
tablePath);
+ dorisCatalog.truncateTable(tablePath, true, partitions);
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/DorisCatalogUtil.java
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/DorisCatalogUtil.java
index 0a169a6d74..2ca95a0dd3 100644
---
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/DorisCatalogUtil.java
+++
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/DorisCatalogUtil.java
@@ -102,6 +102,25 @@ public class DorisCatalogUtil {
return "TRUNCATE TABLE " + tablePath.getFullName();
}
+ /** Builds a table-wide or partition-scoped Doris TRUNCATE TABLE
statement. */
+ public static String getTruncateTableQuery(TablePath tablePath,
List<String> partitions) {
+ String query = getTruncateTableQuery(tablePath);
+ if (partitions == null || partitions.isEmpty()) {
+ return query;
+ }
+ return query
+ + " PARTITION ("
+ + partitions.stream()
+ .map(DorisCatalogUtil::quoteIdentifier)
+ .collect(Collectors.joining(", "))
+ + ")";
+ }
+
+ /** Quotes a Doris identifier and escapes embedded backticks. */
+ private static String quoteIdentifier(String identifier) {
+ return "`" + identifier.replace("`", "``") + "`";
+ }
+
/**
* @param createTableTemplate create table template
* @param catalogTable catalog table
diff --git
a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/config/DorisSinkConfigTest.java
b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/config/DorisSinkConfigTest.java
index 7024bb5070..bb6cc34109 100644
---
a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/config/DorisSinkConfigTest.java
+++
b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/config/DorisSinkConfigTest.java
@@ -20,16 +20,73 @@ package org.apache.seatunnel.connectors.doris.config;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.configuration.util.ConfigValidator;
import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+import org.apache.seatunnel.connectors.doris.exception.DorisConnectorException;
import org.apache.seatunnel.connectors.doris.sink.DorisSinkFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
public class DorisSinkConfigTest {
+ @Test
+ public void testDropDataPartitionsAreNormalized() {
+ Map<String, Object> options = new HashMap<>();
+ options.put("data_save_mode", "DROP_DATA");
+ options.put("doris.config", Collections.singletonMap("partitions", "
p1 , p2 "));
+
+ DorisSinkConfig sinkConfig = DorisSinkConfig.of(createConfig(options));
+
+ Assertions.assertEquals(Arrays.asList("p1", "p2"),
sinkConfig.getPartitions());
+ Assertions.assertEquals("p1,p2",
sinkConfig.getStreamLoadProps().getProperty("partitions"));
+ }
+
+ @Test
+ public void testDropDataRejectsBlankPartitionNames() {
+ Map<String, Object> options = new HashMap<>();
+ options.put("data_save_mode", "DROP_DATA");
+ options.put("doris.config", Collections.singletonMap("partitions",
"p1,,p2"));
+ ReadonlyConfig config = createConfig(options);
+
+ DorisConnectorException exception =
+ Assertions.assertThrows(
+ DorisConnectorException.class, () ->
DorisSinkConfig.of(config));
+
+ Assertions.assertTrue(exception.getMessage().contains("blank partition
names"));
+ }
+
+ @Test
+ public void testDropDataRejectsDuplicatePartitionNames() {
+ Map<String, Object> options = new HashMap<>();
+ options.put("data_save_mode", "DROP_DATA");
+ options.put("doris.config", Collections.singletonMap("partitions",
"p1,p1"));
+ ReadonlyConfig config = createConfig(options);
+
+ DorisConnectorException exception =
+ Assertions.assertThrows(
+ DorisConnectorException.class, () ->
DorisSinkConfig.of(config));
+
+ Assertions.assertTrue(exception.getMessage().contains("duplicate
partition names"));
+ }
+
+ @Test
+ public void testAppendDataDoesNotInterpretStreamLoadPartitions() {
+ DorisSinkConfig sinkConfig =
+ DorisSinkConfig.of(
+ createConfig(
+ Collections.singletonMap(
+ "doris.config",
+ Collections.singletonMap("partitions",
"p1,,p2"))));
+
+ Assertions.assertTrue(sinkConfig.getPartitions().isEmpty());
+ Assertions.assertEquals(
+ "p1,,p2",
sinkConfig.getStreamLoadProps().getProperty("partitions"));
+ }
+
@Test
public void testDirectToBeRequiresBenodes() {
Map<String, Object> configMap = baseConfig();
@@ -78,4 +135,13 @@ public class DorisSinkConfigTest {
config.put("doris.config", new HashMap<String, String>());
return config;
}
+
+ private static ReadonlyConfig createConfig(Map<String, Object>
extraOptions) {
+ Map<String, Object> config = baseConfig();
+ config.put("database", "test_db");
+ config.put("table", "test_table");
+ config.put("sink.label-prefix", "test_job");
+ config.putAll(extraOptions);
+ return ReadonlyConfig.fromMap(config);
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/sink/savemode/DorisSaveModeHandlerTest.java
b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/sink/savemode/DorisSaveModeHandlerTest.java
new file mode 100644
index 0000000000..ea295ae363
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/sink/savemode/DorisSaveModeHandlerTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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.seatunnel.connectors.doris.sink.savemode;
+
+import org.apache.seatunnel.api.sink.DataSaveMode;
+import org.apache.seatunnel.api.sink.SchemaSaveMode;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.connectors.doris.catalog.DorisCatalog;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class DorisSaveModeHandlerTest {
+
+ private static final TableIdentifier TABLE_IDENTIFIER =
+ TableIdentifier.of("doris", "test_db", "test_table");
+ private static final TablePath TABLE_PATH = TABLE_IDENTIFIER.toTablePath();
+
+ @Test
+ void dropDataTruncatesConfiguredPartitions() {
+ DorisCatalog catalog = mock(DorisCatalog.class);
+ CatalogTable catalogTable = mock(CatalogTable.class);
+ when(catalogTable.getTableId()).thenReturn(TABLE_IDENTIFIER);
+ when(catalog.tableExists(TABLE_PATH)).thenReturn(true);
+ DorisSaveModeHandler handler =
+ new DorisSaveModeHandler(
+ SchemaSaveMode.IGNORE,
+ DataSaveMode.DROP_DATA,
+ catalog,
+ catalogTable,
+ null,
+ Arrays.asList("p1", "p2"));
+
+ handler.handleDataSaveMode();
+
+ verify(catalog).truncateTable(TABLE_PATH, true, Arrays.asList("p1",
"p2"));
+ verify(catalog, never()).truncateTable(TABLE_PATH, true);
+ }
+
+ @Test
+ void dropDataWithoutPartitionsPreservesWholeTableCleanup() {
+ DorisCatalog catalog = mock(DorisCatalog.class);
+ CatalogTable catalogTable = mock(CatalogTable.class);
+ when(catalogTable.getTableId()).thenReturn(TABLE_IDENTIFIER);
+ when(catalog.tableExists(TABLE_PATH)).thenReturn(true);
+ DorisSaveModeHandler handler =
+ new DorisSaveModeHandler(
+ SchemaSaveMode.IGNORE,
+ DataSaveMode.DROP_DATA,
+ catalog,
+ catalogTable,
+ null,
+ Collections.emptyList());
+
+ handler.handleDataSaveMode();
+
+ verify(catalog).truncateTable(TABLE_PATH, true);
+ verify(catalog, never()).truncateTable(TABLE_PATH, true,
Collections.emptyList());
+ }
+
+ @Test
+ void appendDataDoesNotTruncatePartitions() {
+ DorisCatalog catalog = mock(DorisCatalog.class);
+ CatalogTable catalogTable = mock(CatalogTable.class);
+ when(catalogTable.getTableId()).thenReturn(TABLE_IDENTIFIER);
+ DorisSaveModeHandler handler =
+ new DorisSaveModeHandler(
+ SchemaSaveMode.IGNORE,
+ DataSaveMode.APPEND_DATA,
+ catalog,
+ catalogTable,
+ null,
+ Collections.singletonList("p1"));
+
+ handler.handleDataSaveMode();
+
+ verify(catalog, never()).truncateTable(TABLE_PATH, true);
+ verify(catalog, never()).truncateTable(TABLE_PATH, true,
Collections.singletonList("p1"));
+ }
+
+ @Test
+ void dropDataDoesNotTruncateNewlyCreatedTable() {
+ DorisCatalog catalog = mock(DorisCatalog.class);
+ CatalogTable catalogTable = mock(CatalogTable.class);
+ when(catalogTable.getTableId()).thenReturn(TABLE_IDENTIFIER);
+
when(catalog.databaseExists(TABLE_PATH.getDatabaseName())).thenReturn(true);
+ when(catalog.tableExists(TABLE_PATH)).thenReturn(false, true);
+ DorisSaveModeHandler handler =
+ new DorisSaveModeHandler(
+ SchemaSaveMode.CREATE_SCHEMA_WHEN_NOT_EXIST,
+ DataSaveMode.DROP_DATA,
+ catalog,
+ catalogTable,
+ null,
+ Collections.singletonList("p1"));
+
+ handler.handleSchemaSaveMode();
+ handler.handleDataSaveMode();
+
+ verify(catalog).createTable(TABLE_PATH, catalogTable, true);
+ verify(catalog, never()).truncateTable(TABLE_PATH, true);
+ verify(catalog, never()).truncateTable(TABLE_PATH, true,
Collections.singletonList("p1"));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/util/DorisCatalogUtilTest.java
b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/util/DorisCatalogUtilTest.java
index 4228a943cc..d28b7fae44 100644
---
a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/util/DorisCatalogUtilTest.java
+++
b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/util/DorisCatalogUtilTest.java
@@ -19,6 +19,7 @@ package org.apache.seatunnel.connectors.doris.util;
import org.apache.seatunnel.api.table.catalog.Column;
import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
import org.apache.seatunnel.api.table.converter.TypeConverter;
import org.apache.seatunnel.api.table.type.BasicType;
@@ -28,12 +29,43 @@ import
org.apache.seatunnel.connectors.doris.datatype.DorisTypeConverterV2;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
+import java.util.Collections;
+
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class DorisCatalogUtilTest {
+ @Test
+ void buildsWholeTableTruncateQueryWithoutPartitions() {
+ TablePath tablePath = TablePath.of("test_db", "test_table");
+
+ assertEquals(
+ "TRUNCATE TABLE test_db.test_table",
+ DorisCatalogUtil.getTruncateTableQuery(tablePath,
Collections.emptyList()));
+ }
+
+ @Test
+ void buildsPartitionTruncateQueryWithQuotedIdentifiers() {
+ TablePath tablePath = TablePath.of("test_db", "test_table");
+
+ assertEquals(
+ "TRUNCATE TABLE test_db.test_table PARTITION (`p1`, `p``2`)",
+ DorisCatalogUtil.getTruncateTableQuery(tablePath,
Arrays.asList("p1", "p`2")));
+ }
+
+ @Test
+ void keepsPartitionContentInsideQuotedIdentifier() {
+ TablePath tablePath = TablePath.of("test_db", "test_table");
+
+ assertEquals(
+ "TRUNCATE TABLE test_db.test_table PARTITION (`p1``) DROP
TABLE test_db.other; --`)",
+ DorisCatalogUtil.getTruncateTableQuery(
+ tablePath, Collections.singletonList("p1`) DROP TABLE
test_db.other; --")));
+ }
+
@Test
void returnsReconvertedTypeWhenSinkTypeNotNull() {
Column column = mock(Column.class);
