Improve the COPY from external location performance

[BohuTANG]

Summary
If we COPY a s3 file and insert into a table, the progresses are:
S1. Read s3 file by blocks from s3 location
S2. Write blocks stream to table t1
S3. Commit

https://github.com/datafuselabs/databend/blob/16e06e414c4680f0d640abada631af89369be877/query/src/interpreters/interpreter_copy.rs#L83-L102

S1 and S2 is in the same thread, looks we can make them in parallel.

[GrapeBaBa]

/assignme

[GrapeBaBa]

@BohuTANG Since I am not quite familiar with rust threads, have questions about this issue. I feel the source stream items will produced by pipeline executor, it seems already in other threads, right? The consumer is in current thread?

[sundy-li]

I think if we transform the Copy Plan into insert select Plan, then the pipeline engine will make it on parallel.
We can do it after #4345

[sundy-li]

@GrapeBaBa

It's ready to do it now.

We can build pipeline through new processor, example

Feel free to contact me if you need any help.

[GrapeBaBa]

@sundy-li Got it.

[GrapeBaBa]

@sundy-li Should we use a sink pipe and a source pipe to create a complete pipeline for refactoring?

[sundy-li]

Should we use a sink pipe and a source pipe to create a complete pipeline for refactoring?

Yes!

[GrapeBaBa]

@BohuTANG @sundy-li Do we have an approach to test copy into feature local?

[BohuTANG]

Run databend-query with disk:

1. create a named internal stage
2. Streaming a file into the stage via the http streaming_load api
3. Do COPY from internal stage

[GrapeBaBa]

Why is there

Run databend-query with disk:

1. create a named internal stage
2. Streaming a file into the stage via the http streaming_load api
3. Do COPY from internal stage

OK. I also find some tests in the codes which seems using minio without detail steps. And why is interpreter_copy unit test is missing? Is it difficult to mock or something else?

[sundy-li]

interpreter_copy unit test
Yes, it's hard to make a unit stateful test. So tests must run using minio.

[GrapeBaBa]

Upload file to stage using

curl -H "stage_name:my_internal_stage" -F "upload=@./books.csv" -XPUT http://localhost:8081/v1/upload_to_stage

Got this error

unexpected: (op: write, path: /Users/kaichen/Documents/projects/databend/target/debug/benddata/datas/stage/my_internal_stage, source: File exists (os error 17))%   

[GrapeBaBa]

The log shows

^[[2m2022-04-03T12:47:38.884447Z^[[0m ^[[31mERROR^[[0m ^[[2mopendal::services::fs::backend^[[0m^[[2m:^[[0m object /Users/kaichen/Documents/projects/databend/target/debug/benddata/datas/stage/my_internal_stage/books.csv create_dir_all for parent /Users/kaichen/Documents/projects/databend/target/debug/benddata/datas/stage/my_internal_stage: Custom { kind: AlreadyExists, error: ObjectError { op: "write", path: "/Users/kaichen/Documents/projects/databend/target/debug/benddata/datas/stage/my_internal_stage", source: File exists (os error 17) } }
^[[2m2022-04-03T12:47:38.884878Z^[[0m ^[[33m WARN^[[0m ^[[2mdatabend_query::servers::http::middleware^[[0m^[[2m:^[[0m http request error: status=500 Internal Server Error, msg=op: write, path: /Users/kaichen/Documents/projects/databend/target/debug/benddata/datas/stage/my_internal_stage, source: File exists (os error 17)

[GrapeBaBa]

After

create stage xxx;

there will be a xxx file generated in local dir, is it expected? @BohuTANG @sundy-li

[GrapeBaBa]

-rw-r--r--  1 kaichen  staff    0 Apr  3 19:47 my_internal_stage
-rw-r--r--  1 kaichen  staff    0 Apr  3 20:12 test
-rw-r--r--  1 kaichen  staff    0 Apr  3 20:26 test1
-rw-r--r--  1 kaichen  staff    0 Apr  3 20:54 test2

[BohuTANG]

xxx should be a folder under the stage directory not a file, this should be a bug:
https://github.com/datafuselabs/databend/blob/cf6da3ead9111b8bfa893cd5c0957cdabf149e2b/query/src/interpreters/interpreter_user_stage_create.rs#L55-L67

[GrapeBaBa]

example

Yes, it should be a bug since I already tested successfully using minio. Let me try to fix it.

[GrapeBaBa]

#4783 (comment) @zhang2014 @sundy-li @BohuTANG let me back here, I looked at the MemoryTable code and know we can use one pipe to implement it. However about if using S3StageTable, I am not quite sure. If I understand correct, this copy operation should do internal stage as well which is not using S3 storage. What is your suggestion?

[GrapeBaBa]

I finally got actually S3StageTable may handle both internal and external stage source. Maybe just name it StageTable more make sense.