[Proposal] Support `SELECT INTO OUTFILE` feature.

[morningman]

Motivation

Doris' current data export method is relatively simple, and data can only be exported through the Export feature. The Export feature can only export data in partition level, and does not support advanced SQL functions such as Expressions, Joins, and Group By.

In some cases, Doris users will have the function to export the query results:

1. Export the query results, provide other users to download.
2. Export query results for further processing by other systems.
3. Bitmap type export: Sometimes users store a large number of ids (100 million +) through bitmap, and need to return these ids. At present, Doris does not support UDTF. And If it is returned through the MySQL protocol, it will not be processed because the result set is too large.

Alternatives

Support SELECT ... FROM tbl INTO OUTFILE xxx ...

This is the most intuitive and user-friendly way. The grammar is referenced from the MySQL grammar manual. In the implementation of Mysql, the exported file will be written to the local file system of the MySQL server. Since Doris is a distributed system, query results may be produced by any node. So here are two options to export:

1. Write the result to the local file system of a BE. And provide users with an http connection for downloading.

   This solution is relatively simple. But some problems are involved:

   1. File cleanup problem. As a result, the file may be very large and it will take up too much BE disk space. And the file cannot exist for a long time, it needs a mechanism to clean it regularly.
   2. Connection problems. The user may not have permission or cannot directly access a BE node, so the file cannot be downloaded.

   For the first problem, it can be satisfied by limiting the size of a single file, monitoring disk space, and increasing the file timeout setting and regularly cleaning up. The second problem cannot be solved, but it can be used as a temporary solution to meet the needs of some users.

2. Write the result to a remote storage system, such as hdfs.

   This solution can use the Broker to write files to the remote system. This solves the two problems in Option 1.

I plan to implement Option 1 in the first phase so that users who do not have remote storage in time can use this feature. And then only need to make less changes, add different write file interface, we can support Option 2.

I plan to implement Option 2. Because for local file, we have to care about all messy things, such as the authentication, GC, disk space, etc.. And for remote storage, all things become simple.

Other system

For AWS Redshift, it provides a UPLOAD command to satisfy this requirement. And it acts same as SELECT INTO OUTFILE in MySQL.

For Vertica and Snowflake, they provides shell tools to export the query result. Vertica, Snowflake. And Snowflake also support a Copy Into command.

Export format

CSV is a common way to export the query result. But some columns are in binary type, such as BITMAP_UNION and HLL. So that we should find a way to save these kind of types.
Parquet, I think, is a good format to do this. And for binary type, we can use BINARY_ARRAY type in Parquet to save.
But Parquet does not support Largeint. I will think about it later. But for most case, Parquet works.

Implementation

Syntax

query_stmt
INTO OUFILE "file:///path/to/file_prefix"
FORMAT AS CSV|PARQUET
WITH BROKER broker_name
(broker_properties)
PROPERTIES
(other_properties);

Example:

SELECT a, sum(b) FROM tbl1 GROUP BY a LIMIT 10
INTO OUFILE "file:///path/to/file_prefix"
FORMAT AS CSV
WITH BROKER my_broker
("username" = "user", "password" = "pwd")
PROPERTIES
("column_separator" = ",");

In MySQL, the grammar is slightly different that it has, for example,FIELD TERMINATED BY... clause at the end. But I think this is not good for further extension. So I decide to use a property map instead.

The file_prefix is the prefix of the exported files. The final file name will be assembled as
prefix + seq no + format suffix. Like

file_prefix_0.csv
file_prefix_1.csv
file_prefix_2.csv

file_prefix_0.parquet
file_prefix_1.parquet
file_prefix_2.parquet

Result Sink

The ResultSink class in BE now has a ResultWriter and a Sender inside it. the ResultWriter is to writer the batch of results to Sender, And the Sender is to send the batch of results back to FE client.

I will make ResultWriter an abstract class, and implements 2 subclass: MysqlResultWriter and FileResultWriter.
The MysqlResultWriter act same as the old ResultWriter. And the FileResultWriter is a new class. This class is to writer the results to file.

Query Plan

There is almost no need to change the query plan, just add additional information to the TResultSink to explain what ResultWriter to use.

Statement return value

In MySQL, the SELECT INTO OUTFILE will return with number of rows exported.

mysql> select * from tbl1 into outfile "/tmp/file_prefix";
Query OK, 100 row affected (0.06 sec)

We keep consistent with it.

Options

Some options can be specified in properties of the statement:

1. column_separator

   The column separator of the output format. Only for CSV format.

2. line_delimiter

   The line delimiter of the output format. Only for CSV format.

3. split_size_bytes
3. max_file_size_bytes

If the size of export file is large, it will be splited by this size.

Remaining problem

1. Whether to support asynchronous export operations

   Some queries may take a long time or have a large result set, which may cause the request to occupy the connection for a long time.
   We can consider supporting asynchronous commands. And through the Session Variable to control whether to use asynchronous commands.

Scheduling Plan

1. Implement export file with CSV format
2. Support PARQUET format
3. Support split files

[imay]

@morningman
Wonderful proposal.
I have some questions about this.

Write the result to the local file system of a BE. And provide users with an http connection for downloading.

If the result file will be downloaded through HTTP, what's the authentication information will be attached for the download request. If there is an authentication check for the download request, the relation between user and result file will be saved. Maybe through rule, such as all result belong to one user locates in the same directory. Or FE will store all the authentication information.

Besides that, it is not a clever idea to store huge data in one single local file. So we should have a solution to support split a huge result into many files in local filesystem. And if it is supported, what user gets is not a single URL, he will get a list of URLs.

And there is another choice that we don't support unload to local file, we only can unload file to remote storage, such as S3, HDFS.

{"download_url" : "http://be:http_port/file.."}

1. I think download operation should be routed by FE. And we can't give user the absolute path in the URL. It is a good that FE can rewrite the path.
2. Usually unload is a heavy operation, if user lose the returned download URL, what left for him/her is to do this unload again, which is not efficient. If we make sure to support unload to local file, what I suggest is that we should support a command for users to list all unload result information.
3. As far as I know, it is hard to get information from OK message of MySQL protocol. It is not a good design to return load-id for insert into select command. So I think we should avoid this in unload operation.

result_size_in_bytes
I think "max_result_bytes" is better.

 result_file_expiration_time
result_expiration_second. Better to give unit information in name.

file name
fix typo to name_prefix and I think there is no need to add TTL in file name.

After all, I think it is a good way to support remote FS first, because it will avoid us to do many things unrelated.

[morningman]

Hi @imay, actually, I totally agree your consideration about the local file upload. We have to care about all messy things, such as the authentication, GC, disk space, etc..

So I may directly implement the remote storage function, after all, they are not much different.

The local file function may only be used as a test case for convenient testing of the rest of the processes.

[imay]

It is OK for test, and remember to disable it when releasing. Better to update proposal for others to review it.

[morningman]

Updated

[imay]

For the updated proposal, lack naming rule for the exported files.

[morningman]

For the updated proposal, lack naming rule for the exported files.

Hi @imay , for the new design, the export file name depends on user's specification in OUTFILE clause, such as:

select * from tbl1 INTO OUTFILE "/path/to/export_file_name.csv"

[imay]

@morningman
If the exported file name is specified in OUTFILE, how do you want to handle when the output will be split into many files.
In Redshift, user can specify the prefix of the output files. It has some benefits, it is easy to handle when the output will be split. And the file extension can be deferred from output type.

And split_file_bytes can be renamed to max_file_size.

[morningman]

@morningman
If the exported file name is specified in OUTFILE, how do you want to handle when the output will be split into many files.
In Redshift, user can specify the prefix of the output files. It has some benefits, it is easy to handle when the output will be split. And the file extension can be deferred from output type.

And split_file_bytes can be renamed to max_file_size.

Updated

[imay]

For the example when the prefix is "file:///path/to/file_prefix", the result should be "file:///path/to/file_prefix0.csv" rather than "file:///path/to/file_prefix_0.csv".

max_file_size may be better, because we can allow user to write size like "512MB", "1GB".