[Proposal] Doris support version column for REPLACE aggregate type

[Youngwb]

BackGround

Doris currently use REPLACE to update data, but the replacement order cannot be guaranteed for the data import of the same batch. The user needs to guarantee that there is no same key column in the imported data of the same batch to guarantee the replacement order, which is very inconvenient for the user. To solve this problem, we can use a version column to specify the replacement order.

Goal

The user specifies a version column when creating the table. Doris relies on this column to update the data of REPLACE type. The larger version column data can REPLACE the data of the smaller version column, while the data of the smaller version column cannot REPLACE the larger version column data.

Create Table Interface

CREATE TABLE `test` (
`id` bigint(20) NOT NULL,
`date` date NOT NULL,
`group_id` bigint(20) NOT NULL,
`version` int MAX NOT NULL,
`keyword` varchar(128) REPLACE NOT NULL,
`clicks` bigint(20) SUM NULL DEFAULT "0" ,
`cost` bigint(20) SUM NULL DEFAULT "0" 
) ENGINE=OLAP
AGGREGATE KEY(`id`, `date`, `group_id`)
DISTRIBUTED BY HASH(`id`) BUCKETS 16
PROPERTIES (
  "replace_version_column" = "version"
);

When creating a table, the user simply adds the replace_version_column attribute in PROPERTIES to identify the version column, which requires a MAX aggregation type to ensure that only the largest version column is retained for the same key column.

Query

When a user's query does not contain the REPLACE column, the original logic follows. When a user's query contains REPLACE columns, BE needs to extend the Version column on which the REPLACE column depends, and compare the value column when it is aggregated. These operations can be done by extending Reader return columns, and in FE，the isPreAggregation is OFF because of the REPLACE column is value column in StorageEngine
，which means the storage engine needs to aggregate the data before returning to scan node，so we can guarantee that the same key columns will be aggregated in Reader.

Compaction

Base and Cumulative Compaction use Reader to aggregate data, and it use all tablet columns as return columns, so similar to the query processing, we can use Reader for replace based on version columns.

Load

With the same batch of data load, Doris uses one or more MemTable. We need to ensure that the same key column in one MemTable, columns of REPLACE type are replaced with version column, while the data in different MemTable is not guaranteed in LOAD because Query and Compaction guarantee the order of replacement.

RollUp

If rollup contains a column of REPLACE type, we need the user to add the Replace version column or extend the column automatically.

[morningman]

How to generate the value of the version column? Is is auto generated? Or it has to be in the user's origin load data?

[Youngwb]

How to generate the value of the version column? Is is auto generated? Or it has to be in the user's origin load data?

it's user's origin load data

[yangzhg]

I suggest that the columns used for keeping order should not exist in the create table statement, but as a hidden column. When creating a unique table, add a property such as sequence=true, so that be can automatically add a hidden one after the key column. The hidden key column is only used for sorting, so that it can be used as a sorting basis when doing compaction and read. A column may be configed as a sequence column during load

[morningman]

I suggest that the columns used for keeping order should not exist in the create table statement, but as a hidden column. When creating a unique table, add a property such as sequence=true, so that be can automatically add a hidden one after the key column. The hidden key column is only used for sorting, so that it can be used as a sorting basis when doing compaction and read. A column may be configed as a sequence column during load

+1.

The only function of the version column is to retain the order information of the data in the business logic. So if the version column appears in the table schema explicitly, we cannot control the user's query behavior for this column.

[Youngwb]

According to @morningman @yangzhg 's suggest, I made some corrections

name

use sequence column instead of version column for user understand easy.

Create table

Use UNIQUE_KEYS instead of AGG_KEYS. Because sequence column is a hidden column, there is no need to create a version column with MAX AGG_TYPE.

CREATE TABLE `test_1` (
`pin_id` bigint(20) NOT NULL COMMENT "",
`date` date NOT NULL COMMENT "",
`group_id` bigint(20) NOT NULL COMMENT "",
`keyword` varchar(128)  NOT NULL
) ENGINE=OLAP
UNIQUE KEY(`pin_id`, `date`, `group_id`)
PROPERTIES (
  "function_column.sequence_type" = "int"
);

like such example, user need to add sequence_type to Identify the sequence column type. It only support the Integer types (int, bigint, largeint) and time types(date, datetime). User can't query the sequence_column hidden in table , but can add one column which value is equal to sequence_column. like this

CREATE TABLE `test_2` (
`pin_id` bigint(20) NOT NULL COMMENT "",
`date` date NOT NULL COMMENT "",
`group_id` bigint(20) NOT NULL COMMENT "",
`sequence_visiable` int NOT NULL,
`keyword` varchar(128)  NOT NULL
) ENGINE=OLAP
UNIQUE KEY(`pin_id`, `date`, `group_id`)
PROPERTIES (
"function_column.sequence_type" = "int"
);

Column names are not necessarily "sequence_visiable", this is just an example. The user ensures that the values are same by specifying parameters at LOAD time

LOAD

Stream Load

curl --location-trusted -u root -H "columns: pin_id,date,group_id,source_sequence,keyword" -H "function_column.sequence_col: source_sequence" -T test_load http://127.0.01:8030/api/test/test_1/_stream_load

Broker Load

LOAD LABEL test.test11
(
    DATA INFILE("hdfs://path/to/load_file")
    INTO TABLE `test_1`
    FORMAT AS "parquet"
    (pin_id,date,group_id,source_sequence,keyword)
    ORDER BY source_sequence
) with BROKER broker_name (...)

Routine Load

CREATE ROUTINE LOAD test_1_job ON test_1
COLUMNS TERMINATED BY ",",
(pin_id,date,group_id,source_sequence,keyword)
ORDER BY source_sequence
PROPERTIES
(
    "desired_concurrent_number"="3",
    "max_batch_interval" = "30",
    "max_batch_rows" = "1000000",
    "max_batch_size" = "509715200"
) FROM KAFKA
(
    "kafka_broker_list" = "...",
    "kafka_topic" = "...",
    "property.client.id" = "...",
    "property.group.id" = "..."
);

I added a parameter sequence_col to identify the source data for the sequence column at load, because it's hidden column, user need to identify the source column in columns_mapping.

For table test_2 which has column sequence_visiable, user can set "sequence_col" = "sequence_visiable" at properties, which means the hidden column "sequence_col" is same with the "sequence_visiable" in table, user can query the the column "sequence_visiable" instead of "sequence_col"

[morningman]

If user create sequence_visible column in table, does it still has a hidden sequence column?
In my point, it should has.

[Youngwb]

If user create sequence_visible column in table, does it still has a hidden sequence column?
In my point, it should has.

yes, it has.

[chaoyli]

I think it's better to handle the problem in load.
If you want to preserve the ordering, you can speculate the behavior in the load command.
CreateTable should not catch this logic, otherwise it's confused to people to understand the create table.

[Youngwb]

I think it's better to handle the problem in load.
If you want to preserve the ordering, you can speculate the behavior in the load command.
CreateTable should not catch this logic, otherwise it's confused to people to understand the create table.

If a sequence column is generated during the Load command phase, we need to consider the case where the user first load and specified a sequence column and then did not, or did not specify a sequence column the first time and then specified a sequence column. This will make the analysis of the load phase more complicated

[chaoyli]

I think it's better to handle the problem in load.
If you want to preserve the ordering, you can speculate the behavior in the load command.
CreateTable should not catch this logic, otherwise it's confused to people to understand the create table.

If a sequence column is generated during the Load command phase, we need to consider the case where the user first load and specified a sequence column and then did not, or did not specify a sequence column the first time and then specified a sequence column. This will make the analysis of the load phase more complicated

I think it's a problem about stable sort. If we can ensure the data received lately been flushed lately.
It seems can solve the problem rightly?