refactor(query): csv reader support prefetch

[youngsofun]

I hereby agree to the terms of the CLA available at: https://docs.databend.com/dev/policies/cla/

Summary

• Fixes copy tsv.gz 3X faster than csv.gz #14973

so for a large file read of next batch and processing of the prev can overlap.
make the whole time shorter especially for relative slow storage.

add PrefetchAsyncSourcer, keep return Async (instead of NEED_CONSUME) until prefetch goal is reached or finished.
rely on some implementation details (correct me if I am wrong @zhang2014 ): trigger of downstream processor dependent on push_data, not NEED_CONSUME event, NEED_CONSUME is only used to change processor state to IDLE

update

the reason for #14973 is simple: the num of processors is not correctly set.

fixed.

but the prefetching still make sense on slow backend.

Tests

• Unit Test
• Logic Test
• Benchmark Test
• No Test - Explain why

Type of change

• Bug Fix (non-breaking change which fixes an issue)
• New Feature (non-breaking change which adds functionality)
• Breaking Change (fix or feature that could cause existing functionality not to work as expected)
• Documentation Update
• Refactoring
• Performance Improvement
• Other (please describe):

---

This change is 

[github-actions]

Docker Image for PR

• tag: pr-14983-400e01f

note: this image tag is only available for internal use,
please check the internal doc for more details.

[youngsofun]

I have test this PR, the csv copy speed from 0.06M/s to 0.09M/s, but still slow than the tsv(0.16M/s):

It seems rows/s for the CSV is not correct, now the COPY of CSV cost is same as TSV now, about 10minutes.

how many rows in this file？

[BohuTANG]

Update:
CSV COPY is also slow than TSV:

TSV will cost 9 minute:

[BohuTANG]

It seems this PR is not help for CSV:

[BohuTANG]

how many rows in this file？

You can try follow this issue: #14973

[youngsofun]

@zhang2014 description updated. plz review again.

[github-actions]

Docker Image for PR

• tag: pr-14983-d047095

note: this image tag is only available for internal use,
please check the internal doc for more details.

[BohuTANG]

Still not working:/

[youngsofun]

@BohuTANG

now cost 11m36, much faster than before (24 - 32min), but still slower than expected 9min12 (my last run with enable_new_copy_for_text_formats=0)

this image is right before ending, the progress is correct now (99997497 rows, 15.47G)

[BohuTANG]

now cost 11m36,

I think the cost time is ok.

Two questions:

1. Why we should do this setting: enable_new_copy_for_text_formats=0, does it mean the new copy impl is still need to improve?
2. This PR is helpful but is one part of the improvement(Means we can merge this PR first)?

[youngsofun]

Why we should do this setting: enable_new_copy_for_text_formats=0, does it mean the new copy impl is still need to improve?

the new impl is much better to maintain, logic is infact the same, but nearly half of the codes are rewritten, I`m not sure there are no problem.

This PR is helpful but is one part of the improvement(Means we can merge this PR first)?

yes, we can merge first

---

it is strange that the Duraition in the history is diff from the time. it is longer :

if it is not acceptable, we can set enable_new_copy_for_text_formats=0 by default until it is improved.

[BohuTANG]

Why we should do this setting: enable_new_copy_for_text_formats=0, does it mean the new copy impl is still need to improve?

the new impl is much better to maintain, logic is infact the same, but nearly half of the codes are rewritten, I`m not sure there are no problem.

This PR is helpful but is one part of the improvement(Means we can merge this PR first)?

yes, we can merge first

it is strange that the Duraition in the history is diff from the time. it is longer :

This is the real time which cost.

if it is not acceptable, we can set enable_new_copy_for_text_formats=0 by default until it is improved.