Unifies load flow

[EmmyMiao87]

What is problem of old load framework
There are different type of load use the different framework of load.

• The most directly problem is that the same feature of load has been implemented many times in different type of load. The duplicated code is too complex to maintenance and enhance.

• Also, the plan of broker and mini load cost more disk I/O and network transform. The performance of stream load is better then broker and mini load

• Due to the non-streaming plan of broker load, it could not support to load the large size of file in limited memory.

What is the solution
Add a new framework of load which uses the unified load plan(streaming load plan) in be and quickly schedule logic in fe.

The design of solution

1. According to the new framework of load in fe:
   The new stage of load consists of PENDING -> LOADING -> FINISHED which removes the ETL stage. Also, the new framework uses the TaskCallback and TxnStateChangeCallback instead of the LoadChecker which is used to trigger the next step of load.
   In the new framework
   Step1: New scheduler named LoadScheduler only support the broker load now. It will pick the PENDING jobs and call the execute function of jobs which is performed to begin txn and submit tasks. After this step, the state of job will be changed to LOADING.
   Step2.0~2.N: The OnTaskFinished will be invoked after the task is finished. It is used to update the progress of job or trigger the next step of load. If all of tasks are finished, the txn will be committed.
   Step3: The AfterVisible will be invoked after txn is visible which is used to change job state to FINISHED.

2. According to the broker load
   Step1: The pending task will be submit by execute function of broker load job. There are preparations that need to be done on pending task.
   Step2.1: OnPendingTaskFinished which is sub-function of OnTaskFinsihed will be invoked after pending task is finished. It is used to submit the loading task which is created based on attachment of pending task.
   Step2.2: OnLoadingTaskFinished which is sub-function of OnTaskFinsihed too will be invoked after loding task is finished. It is used to record the commit info and commit txn when all of task has been finished.
   Step3: The AfterVisible will be invoked after txn is visible which is used to change job state to FINISHED.

3. According to the new load plan:
   The broker load, mini load, multi mini load will use the BrokerScanNode and OlapTableSink instead of CSVScanNode and DataSplitSink. The new load plan is streaming which support to load the huge size of file without Exceeding memory limit.

[EmmyMiao87]

##1076

[EmmyMiao87]

https://github.com/apache/incubator-doris/pull/1134 Change the label of broker load and fix the LabelAlreadyUsedException

[EmmyMiao87]

#1076

[EmmyMiao87]

https://github.com/apache/incubator-doris/pull/1191

[EmmyMiao87]

#1191 1. Add persistence of new load 2. Change non-streaming insert to streaming insert

[EmmyMiao87]

#1323 The streaming mini load has been implemented in this commit

[EmmyMiao87]

#1338 Fix the hung bug in streaming broker load

[EmmyMiao87]

#1335 Add the streaming broker load in load jobs proc

[EmmyMiao87]

I propose to test the performance of streaming broker load and mini load. This test will ignore the effect of rollup and query. The result of test will only depends on the performance of new load framework.

The environment:
Node 1:

• Memory: about 3G
• Disk: SATA about 2T
• Network: about 10000Mb/s

Node 2,3,4:

• Memory: about 27G
• Disk: SSD about 369G
• Network: about 10000Mb/s

The prepare of test:

1. Install doris:

• One frontend and three backends.
• Three backends are located in different node(2,3,4).
• The frontend is located Node1.

2. Create table:

• Situation 1: The table has only one tablet and 3 replicas.
  SQL: create table xxx (about 23 columns: 12 sum aggregation columns) ENGINE=OLAP AGGREGATE KEY(2 columns) DISTRIBUTED BY HASH(2 columns) BUCKETS 1 PROPERTIES ("storage_type" = "COLUMN");
• Situation 2: The table has 100 tablet and every tablet has 3 replicas.
  SQL: create table xxx (about 23 columns: sum 12) ENGINE=OLAP AGGREGATE KEY(2 columns) DISTRIBUTED BY HASH(2 columns) BUCKETS 100 PROPERTIES ("storage_type" = "COLUMN");

The data of test:
Load different size (2K, 1G, 20G, 100G, 500G) of file: store_sales.dat

The step of test:

1. Load data by non-streaming broker load
2. Load data by streaming broker load
3. Load data by non-streaming mini load
4. Load data by streaming mini load

[chaoyli]

You can use store_sales in TPC-DS. It is may be better.
The create table statement of store_sales can be elaborated in place.