Delete data in batches to improve query performance - ApsaraDB for SelectDB

Frequent DELETE statements generate a new data version on every execution and accumulate delete conditions that must be filtered on every read, which degrades query performance over time. The batch delete feature lets you delete data through import jobs—avoiding this overhead and making mixed-write/delete workloads efficient. It is especially effective in Change Data Capture (CDC) scenarios where inserts and deletes are interleaved.

Use cases

CDC synchronization: When using Flink CDC to replicate data from an OLTP database, inserts and deletes in the binlog are interleaved. Handling these with individual DELETE statements is inefficient. The batch delete feature processes both operations through a single import job.
Bulk primary key deletion: Deleting a large number of rows with repeated DELETE statements generates a new data version each time and accumulates delete conditions that must be filtered on every read. The batch delete feature handles bulk deletions in a single import job.

How it works

ApsaraDB for SelectDB adds a hidden column __DORIS_DELETE_SIGN__ to each table built on the Unique Key model. The column has aggregation type REPLACE and defaults to 0.

During import: The hidden column is processed like any other column. When merge_type is set to MERGE, its value is determined by the DELETE ON expression. When set to DELETE, all imported rows are treated as delete markers.

During reads: The frontend (FE) automatically excludes __DORIS_DELETE_SIGN__ from wildcard (*) queries and adds the filter __DORIS_DELETE_SIGN__ != true, so deleted rows are invisible by default.

During compaction:

Cumulative compaction treats the hidden column as a regular column; the compaction logic is unchanged.
Base compaction physically removes rows marked as deleted to reclaim storage.

Prerequisites

Before you begin, make sure that you have:

An ApsaraDB for SelectDB instance with a table using the Unique Key model
Sufficient permissions to run DDL statements and import jobs

Enable batch delete

Batch delete is enabled by default for all new tables (enable_batch_delete_by_default = true in the FE configuration file). All tables that are created after the FE is restarted support the batch delete feature. For tables that were created before the feature was enabled, run:

ALTER TABLE <table_name> ENABLE FEATURE "BATCH_DELETE";

This triggers a schema change. To check whether the operation has completed, run:

SHOW ALTER TABLE COLUMN;

Check whether a table supports batch delete

Run the following commands to confirm that the __DORIS_DELETE_SIGN__ hidden column is present:

SET show_hidden_columns = true;
DESC <table_name>;

If __DORIS_DELETE_SIGN__ appears in the output, the table supports batch delete.

Configure the batch delete syntax

All import methods use the merge_type parameter to control delete behavior. Choose the value that matches your data file:

`merge_type` value	Behavior
`APPEND`	Appends all imported rows (default; same as a regular import)
`DELETE`	Deletes all rows whose key column values match the imported rows
`MERGE`	Applies the DELETE ON expression: matching rows are deleted, others are appended

The delete marker column and the DELETE ON expression are defined differently depending on the import method.

Stream Load

Add the delete marker column to the columns header and specify the DELETE ON expression in the delete header:

curl --location-trusted -u admin:admin_123 \
  -H "expect:100-continue" \
  -H "columns: k1,k2,delete_sign" \
  -H "merge_type: [MERGE|APPEND|DELETE]" \
  -H "delete: delete_sign=1" \
  -T data.csv \
  http://host:port/api/test_db/test_table/_stream_load

OSS Load

Add the delete marker column to the COLUMNS clause and specify the DELETE ON expression in the load statement:

LOAD LABEL test_db.test_label_1
(
    [MERGE|APPEND|DELETE] DATA INFILE("s3://your_bucket_name/your_file.txt")
    INTO TABLE test_table
    COLUMNS TERMINATED BY ","
    (tmp_c1, tmp_c2, delete_sign)
    SET
    (
        id = tmp_c2,
        name = tmp_c1,
    )
    [DELETE ON delete_sign=true]
)
WITH S3
(
    "AWS_PROVIDER" = "OSS",
    "AWS_REGION" = "oss-cn-beijing",
    "AWS_ENDPOINT" = "oss-cn-beijing-internal.aliyuncs.com",
    "AWS_ACCESS_KEY" = "<your_access_key>",
    "AWS_SECRET_KEY" = "<your_secret_key>"
)
PROPERTIES
(
    "timeout" = "3600"
);

Routine Load

Add the delete marker column to the COLUMNS clause:

CREATE ROUTINE LOAD test_db.test1 ON test_table
    [WITH MERGE|APPEND|DELETE]
    COLUMNS(k1, k2, k3, v1, v2, delete_sign),
    WHERE k1 > 100 and k2 like "%doris%"
    [DELETE ON delete_sign=true]
    PROPERTIES
    (
        "desired_concurrent_number" = "3",
        "max_batch_interval" = "20",
        "max_batch_rows" = "300000",
        "max_batch_size" = "209715200",
        "strict_mode" = "false"
    )
    FROM KAFKA
    (
        "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
        "kafka_topic" = "my_topic",
        "kafka_partitions" = "0,1,2,3",
        "kafka_offsets" = "101,0,0,200"
    );

Important

When using an import method other than Stream Load with merge_type: MERGE, you must specify a sequence column because data may arrive out of order. For more information, see Update data by using a sequence column.

Usage notes

The DELETE ON expression is only valid with merge_type: MERGE.
After running a DELETE or MERGE import, reset the session variable before querying the table. Otherwise, deleted rows may appear in query results:
```
SET show_hidden_columns = false;
```
Alternatively, open a new session.

Examples

The following examples use Stream Load.

Delete from a table without a sequence column

Create the table:

CREATE TABLE `test_table` (
    `siteid`   varchar(100) NULL,
    `citycode` varchar(100) NULL,
    `username` varchar(100) NULL,
    `pv`       int NULL
) ENGINE=OLAP
UNIQUE KEY(`siteid`, `citycode`, `username`)
COMMENT 'OLAP'
DISTRIBUTED BY HASH(`siteid`) BUCKETS 16;

Verify that batch delete is enabled:

SET show_hidden_columns = true;
DESC test_table;

Expected output:

+-----------------------+--------------+------+-------+---------+---------+
| Field                 | Type         | Null | Key   | Default | Extra   |
+-----------------------+--------------+------+-------+---------+---------+
| siteid                | VARCHAR(100) | No   | true  | NULL    |         |
| citycode              | VARCHAR(100) | No   | true  | NULL    |         |
| username              | VARCHAR(100) | No   | true  | NULL    |         |
| pv                    | INT          | Yes  | false | NULL    | REPLACE |
| __DORIS_DELETE_SIGN__ | TINYINT      | No   | false | 0       | REPLACE |
+-----------------------+--------------+------+-------+---------+---------+

Insert the initial data:

+--------+----------+----------+------+
| siteid | citycode | username | pv   |
+--------+----------+----------+------+
|      3 |        2 | tom      |    2 |
|      4 |        3 | bush     |    3 |
|      5 |        3 | helen    |    3 |
+--------+----------+----------+------+

Append data using merge_type: APPEND (equivalent to a regular import):

Omitting merge_type has the same effect as setting it to APPEND.

curl --location-trusted -u root: \
  -H "expect:100-continue" \
  -H "column_separator:," \
  -H "columns: siteid, citycode, username, pv" \
  -H "merge_type: APPEND" \
  -T ~/data.csv \
  http://127.0.0.1:8130/api/test_db/test_table/_stream_load

Delete all rows whose key column values match the imported data using merge_type: DELETE:

curl --location-trusted -u root: \
  -H "expect:100-continue" \
  -H "column_separator:," \
  -H "columns: siteid, citycode, username, pv" \
  -H "merge_type: DELETE" \
  -T ~/data.csv \
  http://127.0.0.1:8130/api/test_db/test_table/_stream_load

Import the following data:

3,2,tom,0

After the import, the row where siteid=3 is deleted:

+--------+----------+----------+------+
| siteid | citycode | username | pv   |
+--------+----------+----------+------+
|      4 |        3 | bush     |    3 |
|      5 |        3 | helen    |    3 |
+--------+----------+----------+------+

Use merge_type: MERGE to delete specific rows and append others. The following example deletes rows where siteid=1 and appends all other rows:

curl --location-trusted -u root: \
  -H "expect:100-continue" \
  -H "column_separator:," \
  -H "columns: siteid, citycode, username, pv" \
  -H "merge_type: MERGE" \
  -H "delete: siteid=1" \
  -T ~/data.csv \
  http://127.0.0.1:8130/api/test_db/test_table/_stream_load

Before the import, the table contains:

+--------+----------+----------+------+
| siteid | citycode | username | pv   |
+--------+----------+----------+------+
|      4 |        3 | bush     |    3 |
|      5 |        3 | helen    |    3 |
|      1 |        1 | jim      |    2 |
+--------+----------+----------+------+

Import the following data:

2,1,grace,2
3,2,tom,2
1,1,jim,2

After the import, the row where siteid=1 is deleted and the others are appended:

+--------+----------+----------+------+
| siteid | citycode | username | pv   |
+--------+----------+----------+------+
|      4 |        3 | bush     |    3 |
|      2 |        1 | grace    |    2 |
|      3 |        2 | tom      |    2 |
|      5 |        3 | helen    |    3 |
+--------+----------+----------+------+

Delete from a table with a sequence column

When a Unique Key model table has a sequence column, delete operations are order-sensitive: only rows whose sequence column value is less than or equal to the imported value are deleted.

Create a table with a sequence column:

CREATE TABLE `test_table` (
    `name`   varchar(100) NULL,
    `gender` varchar(10)  NULL,
    `age`    INT NULL
) ENGINE=OLAP
UNIQUE KEY(`name`)
COMMENT 'OLAP'
DISTRIBUTED BY HASH(`name`) BUCKETS 16
PROPERTIES(
    "function_column.sequence_col" = 'age'
);

Verify the table schema:

SET show_hidden_columns = true;
DESC test_table;

Expected output:

+------------------------+--------------+------+-------+---------+---------+
| Field                  | Type         | Null | Key   | Default | Extra   |
+------------------------+--------------+------+-------+---------+---------+
| name                   | VARCHAR(100) | No   | true  | NULL    |         |
| gender                 | VARCHAR(10)  | Yes  | false | NULL    | REPLACE |
| age                    | INT          | Yes  | false | NULL    | REPLACE |
| __DORIS_DELETE_SIGN__  | TINYINT      | No   | false | 0       | REPLACE |
| __DORIS_SEQUENCE_COL__ | INT          | Yes  | false | NULL    | REPLACE |
+------------------------+--------------+------+-------+---------+---------+

The table initially contains:

+-------+--------+------+
| name  | gender | age  |
+-------+--------+------+
| li    | male   |   10 |
| wang  | male   |   14 |
| zhang | male   |   12 |
+-------+--------+------+

Run a DELETE import with function_column.sequence_col set to age:

curl --location-trusted -u root: \
  -H "expect:100-continue" \
  -H "column_separator:," \
  -H "columns: name, gender, age" \
  -H "function_column.sequence_col: age" \
  -H "merge_type: DELETE" \
  -T ~/data.csv \
  http://127.0.0.1:8130/api/test_db/test_table/_stream_load

Case 1: Import li,male,10. The imported age=10 matches the existing row's sequence value (10 <= 10), so the row is deleted:

+-------+--------+------+
| name  | gender | age  |
+-------+--------+------+
| wang  | male   |   14 |
| zhang | male   |   12 |
+-------+--------+------+

Case 2: Import li,male,9. The imported age=9 is less than the existing row's sequence value (9 < 10), so the row is not deleted. The system finds the row with the greater sequence value, checks that __DORIS_DELETE_SIGN__ is 0, and keeps the row visible:

+-------+--------+------+
| name  | gender | age  |
+-------+--------+------+
| li    | male   |   10 |
| wang  | male   |   14 |
| zhang | male   |   12 |
+-------+--------+------+

In Flink CDC scenarios where data may arrive out of order, the sequence column ensures consistency: rows with a greater sequence value always take precedence, regardless of arrival order.

What's next

Update data by using a sequence column