ApsaraDB for SelectDB:Control data updates by using a sequence column

Prerequisites

Before you begin, ensure that:

• The target table uses the Unique Key model. For more information, see Data models.

How it works

ApsaraDB for SelectDB adds a hidden column named __DORIS_SEQUENCE_COL__ to the table. The data type is set at table creation time.

During import: The frontend (FE) reads the sequence value for each incoming row — from the ORDER BY clause in Broker Load or Routine Load jobs, or from the function_column.sequence_col header in Stream Load jobs — and stores it in __DORIS_SEQUENCE_COL__.

During reads and compaction: When resolving conflicts among rows with the same primary key, the system always keeps the row with the largest __DORIS_SEQUENCE_COL__ value. Base compaction and cumulative compaction follow the same rule.

The __DORIS_SEQUENCE_COL__ column can map to an existing column in the table schema or to a column in the data source, depending on which property you use to enable the feature.

Enable the sequence column feature

PROPERTIES ("function_column.sequence_col" = 'column_name')

PROPERTIES ("function_column.sequence_type" = 'column_name')

ALTER TABLE db_name.table_name
ENABLE FEATURE "SEQUENCE_LOAD"
WITH PROPERTIES ("function_column.sequence_type" = "column_name")

Verify that the feature is enabled

1. Run SET show_hidden_columns = true to make hidden columns visible.

2. Run DESC table_name.

3. If __DORIS_SEQUENCE_COL__ appears in the output, the feature is enabled.

Specify a sequence column during import

The examples below use a column named source_sequence from the data source as the sequence column.

curl --location-trusted -u root \
  -H "columns: k1,k2,source_sequence,v1,v2" \
  -H "function_column.sequence_col: source_sequence" \
  -T testData \
  http://host:port/api/testDb/testTbl/_stream_load

LOAD LABEL db1.label1
(
    DATA INFILE("hdfs://host:port/user/data/*/test.txt")
    INTO TABLE `tbl1`
    COLUMNS TERMINATED BY ","
    (k1, k2, source_sequence, v1, v2)
    ORDER BY source_sequence
)
WITH BROKER 'broker'
(
    "username" = "user",
    "password" = "pass"
)
PROPERTIES
(
    "timeout" = "3600"
);

CREATE ROUTINE LOAD example_db.test1 ON example_tbl
[WITH MERGE|APPEND|DELETE]
COLUMNS(k1, k2, source_sequence, v1, v2),
WHERE k1 > 100 and k2 like "%doris%"
[ORDER BY source_sequence]
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"
);

Example

This example uses Stream Load to show how the sequence column controls which row survives across multiple import batches.

Step 1: Create a table with modify_date as the sequence column.

CREATE TABLE test_db.test_table
(
    user_id   BIGINT,
    date      DATE,
    group_id  BIGINT,
    modify_date DATE,
    keyword   VARCHAR(128)
)
UNIQUE KEY(user_id, date, group_id)
DISTRIBUTED BY HASH(user_id) BUCKETS 32
PROPERTIES(
    "function_column.sequence_col" = 'modify_date',
    "enable_unique_key_merge_on_write" = "true"
);

Step 2: Check the table schema.

DESC test_table;

+-------------+--------------+------+-------+---------+---------+
| Field       | Type         | Null | Key   | Default | Extra   |
+-------------+--------------+------+-------+---------+---------+
| user_id     | BIGINT       | No   | true  | NULL    |         |
| date        | DATE         | No   | true  | NULL    |         |
| group_id    | BIGINT       | No   | true  | NULL    |         |
| modify_date | DATE         | No   | false | NULL    | REPLACE |
| keyword     | VARCHAR(128) | No   | false | NULL    | REPLACE |
+-------------+--------------+------+-------+---------+---------+

Step 3: Import the first batch. All six rows share the same primary key (1, 2020-02-22, 1).

1,2020-02-22,1,2020-02-21,a
1,2020-02-22,1,2020-02-22,b
1,2020-02-22,1,2020-03-05,c
1,2020-02-22,1,2020-02-26,d
1,2020-02-22,1,2020-02-23,e
1,2020-02-22,1,2020-02-24,b

curl --location-trusted -u root: -T data.csv \
  -H "expect:100-continue" \
  -H "column_separator:," \
  http://host:port/api/test_db/test_table/_stream_load

Query the result. 2020-03-05 is the largest modify_date value, so row c is retained:

SELECT * FROM test_table;

+---------+------------+----------+-------------+---------+
| user_id | date       | group_id | modify_date | keyword |
+---------+------------+----------+-------------+---------+
|       1 | 2020-02-22 |        1 | 2020-03-05  | c       |
+---------+------------+----------+-------------+---------+

Step 4: Import a second batch with smaller modify_date values than the current maximum.

1,2020-02-22,1,2020-02-22,a
1,2020-02-22,1,2020-02-23,b

Query the result. The existing row is unchanged because 2020-03-05 is still the largest value across all batches:

+---------+------------+----------+-------------+---------+
| user_id | date       | group_id | modify_date | keyword |
+---------+------------+----------+-------------+---------+
|       1 | 2020-02-22 |        1 | 2020-03-05  | c       |
+---------+------------+----------+-------------+---------+

Step 5: Import a third batch with a larger modify_date value.

1,2020-02-22,1,2020-02-22,a
1,2020-02-22,1,2020-03-23,w

Query the result. 2020-03-23 is now the largest value, so row w replaces c:

+---------+------------+----------+-------------+---------+
| user_id | date       | group_id | modify_date | keyword |
+---------+------------+----------+-------------+---------+
|       1 | 2020-02-22 |        1 | 2020-03-23  | w       |
+---------+------------+----------+-------------+---------+

The system compares sequence column values across all batches. The row with the largest sequence value persists in the table.

Usage notes

• Sequence column required in most import jobs: For Stream Load, Broker Load, and INSERT statement-based row updates, you must explicitly specify a sequence column. If you omit it, the import fails with: Table test_table has sequence column, need to specify the sequence column. The only exception is when the sequence column has a default value of CURRENT_TIMESTAMP.

• Partial column updates: When updating only a subset of columns, you do not need to specify a sequence column. If you specify one, the import proceeds normally. If you do not specify one, ApsaraDB for SelectDB uses the sequence value from the matching historical row. If no matching historical row exists, the sequence column is filled with NULL or the column's default value.