Feature: make executor.finish() respect abort signal

[sundy-li]

Summary

Description for this feature.

Related #5448 #5472

For example NumbersSource, we should respect the abortable signal from context.

impl SyncSource for NumbersSource {
    const NAME: &'static str = "numbers";

    fn generate(&mut self) -> Result<Option<DataBlock>> {
        let source_remain_size = self.end - self.begin;

        match source_remain_size {
            0 => Ok(None),
            remain_size => {
                let step = std::cmp::min(remain_size, self.step);
                let column_data = (self.begin..self.begin + step).collect();

                self.begin += step;
                let column = UInt64Column::new_from_vec(column_data);
                Ok(Some(DataBlock::create(self.schema.clone(), vec![
                    Arc::new(column),
                ])))
            }
        }
    }
}

[zhang2014]

We don't need to add it. We can abort executor(executor.finish() ).

[TCeason]

I make a test on my local pc:

not use new processor

In session 1:

run a long query and then let it execute a few seconds,

next, try to interpret it.

root@mysqldb 12:44:19 [(none)]> SET enable_new_processor_framework=0;
Query OK, 0 rows affected (0.05 sec)
Read 0 rows, 0.00 B in 0.002 sec., 0 rows/sec., 0.00 B/sec.

root@mysqldb 12:45:05 [(none)]> select * from numbers(9999999999999);
^C^C -- query aborted
^C^C -- query aborted

In session 2:

execute show processlist to query the state. session 1 is Idle.

root@mysqldb 12:45:10 [(none)]> SET enable_new_processor_framework=0;
Query OK, 0 rows affected (0.06 sec)
Read 0 rows, 0.00 B in 0.002 sec., 0 rows/sec., 0.00 B/sec.

root@mysqldb 12:45:14 [(none)]> show processlist;
+--------------------------------------+-------+-----------------+------+-------+----------+--------------------------------------+--------------+------------------------+-------------------------+-------------------------+--------------------------+---------------------+
| id                                   | type  | host            | user | state | database | extra_info                           | memory_usage | dal_metrics_read_bytes | dal_metrics_write_bytes | scan_progress_read_rows | scan_progress_read_bytes | mysql_connection_id |
+--------------------------------------+-------+-----------------+------+-------+----------+--------------------------------------+--------------+------------------------+-------------------------+-------------------------+--------------------------+---------------------+
| 3b9fa9c1-7a74-4375-a98f-70987d215e55 | MySQL | 127.0.0.1:57648 | root | Query | default  | select * from numbers(9999999999999) |   4295000390 |                      0 |                       0 |                       0 |                        0 |                  10 |
| e2f7dcf3-b543-40b2-a357-5c544c875c20 | MySQL | 127.0.0.1:57646 | root | Query | default  | show processlist                     |            0 |                      0 |                       0 |                       0 |                        0 |                   9 |
+--------------------------------------+-------+-----------------+------+-------+----------+--------------------------------------+--------------+------------------------+-------------------------+-------------------------+--------------------------+---------------------+
2 rows in set (0.09 sec)
Read 2 rows, 452.00 B in 0.034 sec., 58 rows/sec., 12.80 KiB/sec.

-- After Press Ctrl C in session 1

root@mysqldb 12:45:22 [(none)]> show processlist;
+--------------------------------------+-------+-----------------+------+-------+----------+------------------+--------------+------------------------+-------------------------+-------------------------+--------------------------+---------------------+
| id                                   | type  | host            | user | state | database | extra_info       | memory_usage | dal_metrics_read_bytes | dal_metrics_write_bytes | scan_progress_read_rows | scan_progress_read_bytes | mysql_connection_id |
+--------------------------------------+-------+-----------------+------+-------+----------+------------------+--------------+------------------------+-------------------------+-------------------------+--------------------------+---------------------+
| 3b9fa9c1-7a74-4375-a98f-70987d215e55 | MySQL | 127.0.0.1:57648 | root | Idle  | default  | NULL             |            0 |                   NULL |                    NULL |                    NULL |                     NULL |                  10 |
| e2f7dcf3-b543-40b2-a357-5c544c875c20 | MySQL | 127.0.0.1:57646 | root | Query | default  | show processlist |            0 |                      0 |                       0 |                       0 |                        0 |                   9 |
+--------------------------------------+-------+-----------------+------+-------+----------+------------------+--------------+------------------------+-------------------------+-------------------------+--------------------------+---------------------+
2 rows in set (0.09 sec)
Read 2 rows, 415.00 B in 0.033 sec., 60.48 rows/sec., 12.25 KiB/sec.

So could we still need to support the abortable source?@sundy-li

[sundy-li]

We already confirmed that it's related to enable_new_processor_framework =1, so let's support the kill session in this case.

[TCeason]

NumbersSource

I think I got this reason:

https://github.com/datafuselabs/databend/blob/2bd1f211f92511779d5cb4fb23f8361ba55953cd/query/src/table_functions/numbers_stream.rs#L115

numbers will generate a big vec to store the begin..end.

for part in partitions {
                let numbers_part = NumbersPartInfo::from_part(&part)?;
                let (begin, end) =
                    self.try_apply_top_n(numbers_part.part_start, numbers_part.part_end);

                let diff = end - begin;
                let block_nums = diff / block_size;
                let block_remain = diff % block_size;

                if block_nums == 0 {
                    blocks.push(BlockRange { begin, end });
                } else {
                    for r in 0..block_nums {
                        let range_begin = begin + block_size * r;
                        let mut range_end = range_begin + block_size;
                        if r == (block_nums - 1) && block_remain > 0 {
                            range_end += block_remain;
                        }
                        blocks.push(BlockRange {
                            begin: range_begin,
                            end: range_end,
                        });
                    }
                }
            }

But the kill query just set an abort tag not really kill the executing query.

It will wait for the next poll and then judge the is_abort tag, if true, really abort.

So in this case, if send a query select * from numbers(n) to databend server and n is so large.

Although I send kill query on other session or press Ctrl C in the current session. It looks like not useful.

And because the n is so large and then generates a very large vector that will make databend be oom killed.

[TCeason]

So maybe we can close this issue?

In the new processor we can abort executor(executor.finish()).

In the old processor, we find the query hang is generated a big meta vector in one poll, so the query can not be killed.

[zhang2014]

Let's to close.